ResultSetManager Class Reference

#include <ResultSet.h>

Public Member Functions
ResultSet *	reduce (std::vector< ResultSet * > &, const size_t executor_id)
std::shared_ptr< ResultSet >	getOwnResultSet ()
void	rewriteVarlenAggregates (ResultSet *)

Private Attributes
std::shared_ptr< ResultSet >	rs_

Detailed Description

Definition at line 1033 of file ResultSet.h.

Member Function Documentation

std::shared_ptr< ResultSet > ResultSetManager::getOwnResultSet

(

)

Definition at line 1133 of file ResultSetReduction.cpp.

References rs_.

                                                           {
  return rs_;
}

ResultSet * ResultSetManager::reduce	(	std::vector< ResultSet * > &	result_sets,
		const size_t	executor_id
	)

Definition at line 1054 of file ResultSetReduction.cpp.

References gpu_enabled::accumulate(), CHECK, CHECK_EQ, ResultSetReductionJIT::codegen(), CPU, GroupByBaselineHash, logger::init(), query_mem_desc, run_benchmark_import::result, and rs_.

                                                              {
  CHECK(!result_sets.empty());
  auto result_rs = result_sets.front();
  CHECK(result_rs->storage_);
  auto& first_result = *result_rs->storage_;
  auto result = &first_result;
  const auto row_set_mem_owner = result_rs->row_set_mem_owner_;
  for (const auto result_set : result_sets) {
    CHECK_EQ(row_set_mem_owner, result_set->row_set_mem_owner_);
  }
  if (first_result.query_mem_desc_.getQueryDescriptionType() ==
      QueryDescriptionType::GroupByBaselineHash) {
    const auto total_entry_count =
        std::accumulate(result_sets.begin(),
                        result_sets.end(),
                        size_t(0),
                        [](const size_t init, const ResultSet* rs) {
                          return init + rs->query_mem_desc_.getEntryCount();
                        });
    CHECK(total_entry_count);
    auto query_mem_desc = first_result.query_mem_desc_;
    query_mem_desc.setEntryCount(total_entry_count);
    rs_.reset(new ResultSet(first_result.targets_,
                            ExecutorDeviceType::CPU,
                            query_mem_desc,
                            row_set_mem_owner,
                            0,
                            0));
    auto result_storage = rs_->allocateStorage(first_result.target_init_vals_);
    rs_->initializeStorage();
    switch (query_mem_desc.getEffectiveKeyWidth()) {
      case 4:
        first_result.moveEntriesToBuffer<int32_t>(result_storage->getUnderlyingBuffer(),
                                                  query_mem_desc.getEntryCount());
        break;
      case 8:
        first_result.moveEntriesToBuffer<int64_t>(result_storage->getUnderlyingBuffer(),
                                                  query_mem_desc.getEntryCount());
        break;
      default:
        CHECK(false);
    }
    result = rs_->storage_.get();
    result_rs = rs_.get();
  }

  auto& serialized_varlen_buffer = result_sets.front()->serialized_varlen_buffer_;
  if (!serialized_varlen_buffer.empty()) {
    result->rewriteAggregateBufferOffsets(serialized_varlen_buffer.front());
    for (auto result_it = result_sets.begin() + 1; result_it != result_sets.end();
         ++result_it) {
      auto& result_serialized_varlen_buffer = (*result_it)->serialized_varlen_buffer_;
      CHECK_EQ(result_serialized_varlen_buffer.size(), size_t(1));
      serialized_varlen_buffer.emplace_back(
          std::move(result_serialized_varlen_buffer.front()));
    }
  }

  ResultSetReductionJIT reduction_jit(result_rs->getQueryMemDesc(),
                                      result_rs->getTargetInfos(),
                                      result_rs->getTargetInitVals(),
                                      executor_id);
  auto reduction_code = reduction_jit.codegen();
  size_t ctr = 1;
  for (auto result_it = result_sets.begin() + 1; result_it != result_sets.end();
       ++result_it) {
    if (!serialized_varlen_buffer.empty()) {
      result->reduce(*((*result_it)->storage_),
                     serialized_varlen_buffer[ctr++],
                     reduction_code,
                     executor_id);
    } else {
      result->reduce(*((*result_it)->storage_), {}, reduction_code, executor_id);
    }
  }
  return result_rs;
}

Here is the call graph for this function:

void ResultSetManager::rewriteVarlenAggregates

(

ResultSet *

result_rs

)

Definition at line 1137 of file ResultSetReduction.cpp.

                                                                   {
  auto& result_storage = result_rs->storage_;
  result_storage->rewriteAggregateBufferOffsets(
      result_rs->serialized_varlen_buffer_.front());
}

Member Data Documentation

std::shared_ptr<ResultSet> ResultSetManager::rs_

private

Definition at line 1042 of file ResultSet.h.

Referenced by getOwnResultSet(), and reduce().

---

The documentation for this class was generated from the following files:

• /home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/ResultSet.h
• /home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/ResultSetReduction.cpp