OmniSciDB  0fdbebe030
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
ColumnarResults Class Reference

#include <ColumnarResults.h>

Public Types

using ReadFunction = std::function< int64_t(const ResultSet &, const size_t, const size_t, const size_t)>
 
using WriteFunction = std::function< void(const ResultSet &, const size_t, const size_t, const size_t, const size_t, const ReadFunction &)>
 

Public Member Functions

 ColumnarResults (const std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner, const ResultSet &rows, const size_t num_columns, const std::vector< SQLTypeInfo > &target_types, const bool is_parallel_execution_enforced=false)
 
 ColumnarResults (const std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner, const int8_t *one_col_buffer, const size_t num_rows, const SQLTypeInfo &target_type)
 
const std::vector< int8_t * > & getColumnBuffers () const
 
const size_t size () const
 
const SQLTypeInfogetColumnType (const int col_id) const
 
bool isParallelConversion () const
 
bool isDirectColumnarConversionPossible () const
 

Static Public Member Functions

static std::unique_ptr
< ColumnarResults
mergeResults (const std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner, const std::vector< std::unique_ptr< ColumnarResults >> &sub_results)
 

Protected Attributes

std::vector< int8_t * > column_buffers_
 
size_t num_rows_
 

Private Member Functions

 ColumnarResults (const size_t num_rows, const std::vector< SQLTypeInfo > &target_types)
 
void writeBackCell (const TargetValue &col_val, const size_t row_idx, const size_t column_idx)
 
void materializeAllColumnsDirectly (const ResultSet &rows, const size_t num_columns)
 
void materializeAllColumnsThroughIteration (const ResultSet &rows, const size_t num_columns)
 
void materializeAllColumnsGroupBy (const ResultSet &rows, const size_t num_columns)
 
void materializeAllColumnsProjection (const ResultSet &rows, const size_t num_columns)
 
void copyAllNonLazyColumns (const std::vector< ColumnLazyFetchInfo > &lazy_fetch_info, const ResultSet &rows, const size_t num_columns)
 
void materializeAllLazyColumns (const std::vector< ColumnLazyFetchInfo > &lazy_fetch_info, const ResultSet &rows, const size_t num_columns)
 
void locateAndCountEntries (const ResultSet &rows, ColumnBitmap &bitmap, std::vector< size_t > &non_empty_per_thread, const size_t entry_count, const size_t num_threads, const size_t size_per_thread) const
 
void compactAndCopyEntries (const ResultSet &rows, const ColumnBitmap &bitmap, const std::vector< size_t > &non_empty_per_thread, const size_t num_columns, const size_t entry_count, const size_t num_threads, const size_t size_per_thread)
 
void compactAndCopyEntriesWithTargetSkipping (const ResultSet &rows, const ColumnBitmap &bitmap, const std::vector< size_t > &non_empty_per_thread, const std::vector< size_t > &global_offsets, const std::vector< bool > &targets_to_skip, const std::vector< size_t > &slot_idx_per_target_idx, const size_t num_columns, const size_t entry_count, const size_t num_threads, const size_t size_per_thread)
 
void compactAndCopyEntriesWithoutTargetSkipping (const ResultSet &rows, const ColumnBitmap &bitmap, const std::vector< size_t > &non_empty_per_thread, const std::vector< size_t > &global_offsets, const std::vector< size_t > &slot_idx_per_target_idx, const size_t num_columns, const size_t entry_count, const size_t num_threads, const size_t size_per_thread)
 
template<typename DATA_TYPE >
void writeBackCellDirect (const ResultSet &rows, const size_t input_buffer_entry_idx, const size_t output_buffer_entry_idx, const size_t target_idx, const size_t slot_idx, const ReadFunction &read_function)
 
std::vector< WriteFunctioninitWriteFunctions (const ResultSet &rows, const std::vector< bool > &targets_to_skip={})
 
template<QueryDescriptionType QUERY_TYPE, bool COLUMNAR_OUTPUT>
std::vector< ReadFunctioninitReadFunctions (const ResultSet &rows, const std::vector< size_t > &slot_idx_per_target_idx, const std::vector< bool > &targets_to_skip={})
 
std::tuple< std::vector
< WriteFunction >, std::vector
< ReadFunction > > 
initAllConversionFunctions (const ResultSet &rows, const std::vector< size_t > &slot_idx_per_target_idx, const std::vector< bool > &targets_to_skip={})
 
template<>
void writeBackCellDirect (const ResultSet &rows, const size_t input_buffer_entry_idx, const size_t output_buffer_entry_idx, const size_t target_idx, const size_t slot_idx, const ReadFunction &read_from_function)
 
template<>
void writeBackCellDirect (const ResultSet &rows, const size_t input_buffer_entry_idx, const size_t output_buffer_entry_idx, const size_t target_idx, const size_t slot_idx, const ReadFunction &read_from_function)
 

Private Attributes

const std::vector< SQLTypeInfotarget_types_
 
bool parallel_conversion_
 
bool direct_columnar_conversion_
 

Detailed Description

Definition at line 61 of file ColumnarResults.h.

Member Typedef Documentation

using ColumnarResults::ReadFunction = std::function<int64_t(const ResultSet&, const size_t, const size_t, const size_t)>

Definition at line 94 of file ColumnarResults.h.

using ColumnarResults::WriteFunction = std::function<void(const ResultSet&, const size_t, const size_t, const size_t, const size_t, const ReadFunction&)>

Definition at line 103 of file ColumnarResults.h.

Constructor & Destructor Documentation

ColumnarResults::ColumnarResults ( const std::shared_ptr< RowSetMemoryOwner row_set_mem_owner,
const ResultSet rows,
const size_t  num_columns,
const std::vector< SQLTypeInfo > &  target_types,
const bool  is_parallel_execution_enforced = false 
)

Definition at line 43 of file ColumnarResults.cpp.

References checked_malloc(), column_buffers_, DEBUG_TIMER, isDirectColumnarConversionPossible(), kENCODING_NONE, materializeAllColumnsDirectly(), materializeAllColumnsThroughIteration(), and num_rows_.

Referenced by mergeResults().

49  : column_buffers_(num_columns)
50  , num_rows_(use_parallel_algorithms(rows) || rows.isDirectColumnarConversionPossible()
51  ? rows.entryCount()
52  : rows.rowCount())
53  , target_types_(target_types)
54  , parallel_conversion_(is_parallel_execution_enforced ? true
56  , direct_columnar_conversion_(rows.isDirectColumnarConversionPossible()) {
57  auto timer = DEBUG_TIMER(__func__);
58  column_buffers_.resize(num_columns);
59  for (size_t i = 0; i < num_columns; ++i) {
60  const bool is_varlen = target_types[i].is_array() ||
61  (target_types[i].is_string() &&
62  target_types[i].get_compression() == kENCODING_NONE) ||
63  target_types[i].is_geometry();
64  if (is_varlen) {
66  }
68  !rows.isZeroCopyColumnarConversionPossible(i)) {
69  column_buffers_[i] = reinterpret_cast<int8_t*>(
70  checked_malloc(num_rows_ * target_types[i].get_size()));
71  row_set_mem_owner->addColBuffer(column_buffers_[i]);
72  }
73  }
74 
75  if (isDirectColumnarConversionPossible() && rows.entryCount() > 0) {
76  materializeAllColumnsDirectly(rows, num_columns);
77  } else {
78  materializeAllColumnsThroughIteration(rows, num_columns);
79  }
80 }
std::vector< int8_t * > column_buffers_
bool use_parallel_algorithms(const ResultSet &rows)
Definition: ResultSet.cpp:899
bool direct_columnar_conversion_
void materializeAllColumnsThroughIteration(const ResultSet &rows, const size_t num_columns)
void * checked_malloc(const size_t size)
Definition: checked_alloc.h:40
bool isDirectColumnarConversionPossible() const
void materializeAllColumnsDirectly(const ResultSet &rows, const size_t num_columns)
#define DEBUG_TIMER(name)
Definition: Logger.h:313
const std::vector< SQLTypeInfo > target_types_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ColumnarResults::ColumnarResults ( const std::shared_ptr< RowSetMemoryOwner row_set_mem_owner,
const int8_t *  one_col_buffer,
const size_t  num_rows,
const SQLTypeInfo target_type 
)

Definition at line 82 of file ColumnarResults.cpp.

87  : column_buffers_(1)
89  , target_types_{target_type}
90  , parallel_conversion_(false)
92  auto timer = DEBUG_TIMER(__func__);
93  const bool is_varlen =
94  target_type.is_array() ||
95  (target_type.is_string() && target_type.get_compression() == kENCODING_NONE) ||
96  target_type.is_geometry();
97  if (is_varlen) {
99  }
100  const auto buf_size = num_rows * target_type.get_size();
101  column_buffers_[0] = reinterpret_cast<int8_t*>(checked_malloc(buf_size));
102  memcpy(((void*)column_buffers_[0]), one_col_buffer, buf_size);
103  row_set_mem_owner->addColBuffer(column_buffers_[0]);
104 }
std::vector< int8_t * > column_buffers_
HOST DEVICE int get_size() const
Definition: sqltypes.h:258
const int8_t const int64_t * num_rows
bool direct_columnar_conversion_
void * checked_malloc(const size_t size)
Definition: checked_alloc.h:40
HOST DEVICE EncodingType get_compression() const
Definition: sqltypes.h:256
bool is_geometry() const
Definition: sqltypes.h:411
#define DEBUG_TIMER(name)
Definition: Logger.h:313
bool is_string() const
Definition: sqltypes.h:399
const std::vector< SQLTypeInfo > target_types_
bool is_array() const
Definition: sqltypes.h:407
ColumnarResults::ColumnarResults ( const size_t  num_rows,
const std::vector< SQLTypeInfo > &  target_types 
)
inlineprivate

Definition at line 110 of file ColumnarResults.h.

111  : num_rows_(num_rows), target_types_(target_types) {}
const int8_t const int64_t * num_rows
const std::vector< SQLTypeInfo > target_types_

Member Function Documentation

void ColumnarResults::compactAndCopyEntries ( const ResultSet rows,
const ColumnBitmap bitmap,
const std::vector< size_t > &  non_empty_per_thread,
const size_t  num_columns,
const size_t  entry_count,
const size_t  num_threads,
const size_t  size_per_thread 
)
private

This function goes through all non-empty elements marked in the bitmap data structure, and store them back into output column buffers. The output column buffers are compacted without any holes in it.

TODO(Saman): if necessary, we can look into the distribution of non-empty entries and choose a different load-balanced strategy (assigning equal number of non-empties to each thread) as opposed to equal partitioning of the bitmap

Definition at line 546 of file ColumnarResults.cpp.

References CHECK(), CHECK_EQ, compactAndCopyEntriesWithoutTargetSkipping(), compactAndCopyEntriesWithTargetSkipping(), GroupByBaselineHash, GroupByPerfectHash, and isDirectColumnarConversionPossible().

Referenced by materializeAllColumnsGroupBy().

553  {
555  CHECK(rows.getQueryDescriptionType() == QueryDescriptionType::GroupByPerfectHash ||
556  rows.getQueryDescriptionType() == QueryDescriptionType::GroupByBaselineHash);
557  CHECK_EQ(num_threads, non_empty_per_thread.size());
558 
559  // compute the exclusive scan over all non-empty totals
560  std::vector<size_t> global_offsets(num_threads + 1, 0);
561  std::partial_sum(non_empty_per_thread.begin(),
562  non_empty_per_thread.end(),
563  std::next(global_offsets.begin()));
564 
565  const auto slot_idx_per_target_idx = rows.getSlotIndicesForTargetIndices();
566  const auto [single_slot_targets_to_skip, num_single_slot_targets] =
567  rows.getSupportedSingleSlotTargetBitmap();
568 
569  // We skip multi-slot targets (e.g., AVG). These skipped targets are treated
570  // differently and accessed through result set's iterator
571  if (num_single_slot_targets < num_columns) {
573  bitmap,
574  non_empty_per_thread,
575  global_offsets,
576  single_slot_targets_to_skip,
577  slot_idx_per_target_idx,
578  num_columns,
579  entry_count,
580  num_threads,
581  size_per_thread);
582  } else {
584  bitmap,
585  non_empty_per_thread,
586  global_offsets,
587  slot_idx_per_target_idx,
588  num_columns,
589  entry_count,
590  num_threads,
591  size_per_thread);
592  }
593 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
CHECK(cgen_state)
bool isDirectColumnarConversionPossible() const
void compactAndCopyEntriesWithTargetSkipping(const ResultSet &rows, const ColumnBitmap &bitmap, const std::vector< size_t > &non_empty_per_thread, const std::vector< size_t > &global_offsets, const std::vector< bool > &targets_to_skip, const std::vector< size_t > &slot_idx_per_target_idx, const size_t num_columns, const size_t entry_count, const size_t num_threads, const size_t size_per_thread)
void compactAndCopyEntriesWithoutTargetSkipping(const ResultSet &rows, const ColumnBitmap &bitmap, const std::vector< size_t > &non_empty_per_thread, const std::vector< size_t > &global_offsets, const std::vector< size_t > &slot_idx_per_target_idx, const size_t num_columns, const size_t entry_count, const size_t num_threads, const size_t size_per_thread)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void ColumnarResults::compactAndCopyEntriesWithoutTargetSkipping ( const ResultSet rows,
const ColumnBitmap bitmap,
const std::vector< size_t > &  non_empty_per_thread,
const std::vector< size_t > &  global_offsets,
const std::vector< size_t > &  slot_idx_per_target_idx,
const size_t  num_columns,
const size_t  entry_count,
const size_t  num_threads,
const size_t  size_per_thread 
)
private

This functions takes a bitmap of non-empty entries within the result set's storage and compact and copy those contents back into the output column_buffers_. In this variation, all targets are assumed to be single-slot and thus can be directly columnarized.

Definition at line 690 of file ColumnarResults.cpp.

References CHECK(), CHECK_EQ, GroupByBaselineHash, GroupByPerfectHash, initAllConversionFunctions(), and isDirectColumnarConversionPossible().

Referenced by compactAndCopyEntries().

699  {
701  CHECK(rows.getQueryDescriptionType() == QueryDescriptionType::GroupByPerfectHash ||
702  rows.getQueryDescriptionType() == QueryDescriptionType::GroupByBaselineHash);
703 
704  const auto [write_functions, read_functions] =
705  initAllConversionFunctions(rows, slot_idx_per_target_idx);
706  CHECK_EQ(write_functions.size(), num_columns);
707  CHECK_EQ(read_functions.size(), num_columns);
708 
709  auto compact_buffer_func = [&rows,
710  &bitmap,
711  &global_offsets,
712  &non_empty_per_thread,
713  &num_columns,
714  &slot_idx_per_target_idx,
715  &write_functions = write_functions,
716  &read_functions = read_functions](const size_t start_index,
717  const size_t end_index,
718  const size_t thread_idx) {
719  const size_t total_non_empty = non_empty_per_thread[thread_idx];
720  size_t non_empty_idx = 0;
721  size_t local_idx = 0;
722  for (size_t entry_idx = start_index; entry_idx < end_index;
723  entry_idx++, local_idx++) {
724  if (non_empty_idx >= total_non_empty) {
725  // all non-empty entries has been written back
726  break;
727  }
728  const size_t output_buffer_row_idx = global_offsets[thread_idx] + non_empty_idx;
729  if (bitmap.get(local_idx, thread_idx)) {
730  for (size_t column_idx = 0; column_idx < num_columns; column_idx++) {
731  write_functions[column_idx](rows,
732  entry_idx,
733  output_buffer_row_idx,
734  column_idx,
735  slot_idx_per_target_idx[column_idx],
736  read_functions[column_idx]);
737  }
738  non_empty_idx++;
739  } else {
740  continue;
741  }
742  }
743  };
744 
745  std::vector<std::future<void>> compaction_threads;
746  for (size_t thread_idx = 0; thread_idx < num_threads; thread_idx++) {
747  const size_t start_entry = thread_idx * size_per_thread;
748  const size_t end_entry = std::min(start_entry + size_per_thread, entry_count);
749  compaction_threads.push_back(std::async(
750  std::launch::async, compact_buffer_func, start_entry, end_entry, thread_idx));
751  }
752 
753  for (auto& child : compaction_threads) {
754  child.wait();
755  }
756 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
CHECK(cgen_state)
std::tuple< std::vector< WriteFunction >, std::vector< ReadFunction > > initAllConversionFunctions(const ResultSet &rows, const std::vector< size_t > &slot_idx_per_target_idx, const std::vector< bool > &targets_to_skip={})
bool isDirectColumnarConversionPossible() const

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void ColumnarResults::compactAndCopyEntriesWithTargetSkipping ( const ResultSet rows,
const ColumnBitmap bitmap,
const std::vector< size_t > &  non_empty_per_thread,
const std::vector< size_t > &  global_offsets,
const std::vector< bool > &  targets_to_skip,
const std::vector< size_t > &  slot_idx_per_target_idx,
const size_t  num_columns,
const size_t  entry_count,
const size_t  num_threads,
const size_t  size_per_thread 
)
private

This functions takes a bitmap of non-empty entries within the result set's storage and compact and copy those contents back into the output column_buffers_. In this variation, multi-slot targets (e.g., AVG) are treated with the existing result set's iterations, but everything else is directly columnarized.

Definition at line 601 of file ColumnarResults.cpp.

References CHECK(), CHECK_EQ, GroupByBaselineHash, GroupByPerfectHash, initAllConversionFunctions(), isDirectColumnarConversionPossible(), and writeBackCell().

Referenced by compactAndCopyEntries().

611  {
613  CHECK(rows.getQueryDescriptionType() == QueryDescriptionType::GroupByPerfectHash ||
614  rows.getQueryDescriptionType() == QueryDescriptionType::GroupByBaselineHash);
615 
616  const auto [write_functions, read_functions] =
617  initAllConversionFunctions(rows, slot_idx_per_target_idx, targets_to_skip);
618  CHECK_EQ(write_functions.size(), num_columns);
619  CHECK_EQ(read_functions.size(), num_columns);
620 
621  auto compact_buffer_func = [this,
622  &rows,
623  &bitmap,
624  &global_offsets,
625  &non_empty_per_thread,
626  &num_columns,
627  &targets_to_skip,
628  &slot_idx_per_target_idx,
629  &write_functions = write_functions,
630  &read_functions = read_functions](const size_t start_index,
631  const size_t end_index,
632  const size_t thread_idx) {
633  const size_t total_non_empty = non_empty_per_thread[thread_idx];
634  size_t non_empty_idx = 0;
635  size_t local_idx = 0;
636  for (size_t entry_idx = start_index; entry_idx < end_index;
637  entry_idx++, local_idx++) {
638  if (non_empty_idx >= total_non_empty) {
639  // all non-empty entries has been written back
640  break;
641  }
642  const size_t output_buffer_row_idx = global_offsets[thread_idx] + non_empty_idx;
643  if (bitmap.get(local_idx, thread_idx)) {
644  // targets that are recovered from the result set iterators:
645  const auto crt_row = rows.getRowAtNoTranslations(entry_idx, targets_to_skip);
646  for (size_t column_idx = 0; column_idx < num_columns; ++column_idx) {
647  if (!targets_to_skip.empty() && !targets_to_skip[column_idx]) {
648  writeBackCell(crt_row[column_idx], output_buffer_row_idx, column_idx);
649  }
650  }
651  // targets that are copied directly without any translation/decoding from
652  // result set
653  for (size_t column_idx = 0; column_idx < num_columns; column_idx++) {
654  if (!targets_to_skip.empty() && !targets_to_skip[column_idx]) {
655  continue;
656  }
657  write_functions[column_idx](rows,
658  entry_idx,
659  output_buffer_row_idx,
660  column_idx,
661  slot_idx_per_target_idx[column_idx],
662  read_functions[column_idx]);
663  }
664  non_empty_idx++;
665  } else {
666  continue;
667  }
668  }
669  };
670 
671  std::vector<std::future<void>> compaction_threads;
672  for (size_t thread_idx = 0; thread_idx < num_threads; thread_idx++) {
673  const size_t start_entry = thread_idx * size_per_thread;
674  const size_t end_entry = std::min(start_entry + size_per_thread, entry_count);
675  compaction_threads.push_back(std::async(
676  std::launch::async, compact_buffer_func, start_entry, end_entry, thread_idx));
677  }
678 
679  for (auto& child : compaction_threads) {
680  child.wait();
681  }
682 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
CHECK(cgen_state)
std::tuple< std::vector< WriteFunction >, std::vector< ReadFunction > > initAllConversionFunctions(const ResultSet &rows, const std::vector< size_t > &slot_idx_per_target_idx, const std::vector< bool > &targets_to_skip={})
bool isDirectColumnarConversionPossible() const
void writeBackCell(const TargetValue &col_val, const size_t row_idx, const size_t column_idx)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void ColumnarResults::copyAllNonLazyColumns ( const std::vector< ColumnLazyFetchInfo > &  lazy_fetch_info,
const ResultSet rows,
const size_t  num_columns 
)
private

Definition at line 352 of file ColumnarResults.cpp.

References CHECK(), column_buffers_, isDirectColumnarConversionPossible(), num_rows_, and target_types_.

Referenced by materializeAllColumnsProjection().

355  {
357  const auto is_column_non_lazily_fetched = [&lazy_fetch_info](const size_t col_idx) {
358  // Saman: make sure when this lazy_fetch_info is empty
359  if (lazy_fetch_info.empty()) {
360  return true;
361  } else {
362  return !lazy_fetch_info[col_idx].is_lazily_fetched;
363  }
364  };
365 
366  // parallelized by assigning each column to a thread
367  std::vector<std::future<void>> direct_copy_threads;
368  for (size_t col_idx = 0; col_idx < num_columns; col_idx++) {
369  if (rows.isZeroCopyColumnarConversionPossible(col_idx)) {
370  CHECK(!column_buffers_[col_idx]);
371  column_buffers_[col_idx] = const_cast<int8_t*>(rows.getColumnarBuffer(col_idx));
372  } else if (is_column_non_lazily_fetched(col_idx)) {
373  direct_copy_threads.push_back(std::async(
374  std::launch::async,
375  [&rows, this](const size_t column_index) {
376  const size_t column_size = num_rows_ * target_types_[column_index].get_size();
377  rows.copyColumnIntoBuffer(
378  column_index, column_buffers_[column_index], column_size);
379  },
380  col_idx));
381  }
382  }
383 
384  for (auto& child : direct_copy_threads) {
385  child.wait();
386  }
387 }
std::vector< int8_t * > column_buffers_
CHECK(cgen_state)
bool isDirectColumnarConversionPossible() const
const std::vector< SQLTypeInfo > target_types_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

const std::vector<int8_t*>& ColumnarResults::getColumnBuffers ( ) const
inline

Definition at line 78 of file ColumnarResults.h.

References column_buffers_.

Referenced by ColumnFetcher::transferColumnIfNeeded().

78 { return column_buffers_; }
std::vector< int8_t * > column_buffers_

+ Here is the caller graph for this function:

const SQLTypeInfo& ColumnarResults::getColumnType ( const int  col_id) const
inline

Definition at line 82 of file ColumnarResults.h.

References CHECK_GE, CHECK_LT, and target_types_.

Referenced by ColumnFetcher::transferColumnIfNeeded().

82  {
83  CHECK_GE(col_id, 0);
84  CHECK_LT(static_cast<size_t>(col_id), target_types_.size());
85  return target_types_[col_id];
86  }
#define CHECK_GE(x, y)
Definition: Logger.h:210
#define CHECK_LT(x, y)
Definition: Logger.h:207
const std::vector< SQLTypeInfo > target_types_

+ Here is the caller graph for this function:

std::tuple< std::vector< ColumnarResults::WriteFunction >, std::vector< ColumnarResults::ReadFunction > > ColumnarResults::initAllConversionFunctions ( const ResultSet rows,
const std::vector< size_t > &  slot_idx_per_target_idx,
const std::vector< bool > &  targets_to_skip = {} 
)
private

This function goes through all target types in the output, and chooses appropriate write and read functions per target. The goal is then to simply use these functions for each row and per target. Read functions are used to read each cell's data content (particular target in a row), and write functions are used to properly write back the cell's content into the output column buffers.

Definition at line 1049 of file ColumnarResults.cpp.

References CHECK(), GroupByBaselineHash, GroupByPerfectHash, initWriteFunctions(), and isDirectColumnarConversionPossible().

Referenced by compactAndCopyEntriesWithoutTargetSkipping(), and compactAndCopyEntriesWithTargetSkipping().

1052  {
1054  (rows.getQueryDescriptionType() == QueryDescriptionType::GroupByPerfectHash ||
1055  rows.getQueryDescriptionType() == QueryDescriptionType::GroupByBaselineHash));
1056 
1057  const auto write_functions = initWriteFunctions(rows, targets_to_skip);
1058  if (rows.getQueryDescriptionType() == QueryDescriptionType::GroupByPerfectHash) {
1059  if (rows.didOutputColumnar()) {
1060  return std::make_tuple(
1061  std::move(write_functions),
1062  initReadFunctions<QueryDescriptionType::GroupByPerfectHash, true>(
1063  rows, slot_idx_per_target_idx, targets_to_skip));
1064  } else {
1065  return std::make_tuple(
1066  std::move(write_functions),
1067  initReadFunctions<QueryDescriptionType::GroupByPerfectHash, false>(
1068  rows, slot_idx_per_target_idx, targets_to_skip));
1069  }
1070  } else {
1071  if (rows.didOutputColumnar()) {
1072  return std::make_tuple(
1073  std::move(write_functions),
1074  initReadFunctions<QueryDescriptionType::GroupByBaselineHash, true>(
1075  rows, slot_idx_per_target_idx, targets_to_skip));
1076  } else {
1077  return std::make_tuple(
1078  std::move(write_functions),
1079  initReadFunctions<QueryDescriptionType::GroupByBaselineHash, false>(
1080  rows, slot_idx_per_target_idx, targets_to_skip));
1081  }
1082  }
1083 }
CHECK(cgen_state)
std::vector< WriteFunction > initWriteFunctions(const ResultSet &rows, const std::vector< bool > &targets_to_skip={})
bool isDirectColumnarConversionPossible() const

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

template<QueryDescriptionType QUERY_TYPE, bool COLUMNAR_OUTPUT>
std::vector< ColumnarResults::ReadFunction > ColumnarResults::initReadFunctions ( const ResultSet rows,
const std::vector< size_t > &  slot_idx_per_target_idx,
const std::vector< bool > &  targets_to_skip = {} 
)
private

Initializes a set of read funtions to properly access the contents of the result set's storage buffer. Each particular read function is chosen based on the data type and data size used to store that target in the result set's storage buffer. These functions are then used for each row in the result set.

Definition at line 951 of file ColumnarResults.cpp.

References CHECK(), CHECK_EQ, GroupByBaselineHash, anonymous_namespace{ColumnarResults.cpp}::invalid_read_func(), isDirectColumnarConversionPossible(), kDOUBLE, kFLOAT, target_types_, and UNREACHABLE.

954  {
956  CHECK(COLUMNAR_OUTPUT == rows.didOutputColumnar());
957  CHECK(QUERY_TYPE == rows.getQueryDescriptionType());
958 
959  std::vector<ReadFunction> read_functions;
960  read_functions.reserve(target_types_.size());
961 
962  for (size_t target_idx = 0; target_idx < target_types_.size(); target_idx++) {
963  if (!targets_to_skip.empty() && !targets_to_skip[target_idx]) {
964  // for targets that should be skipped, we use a placeholder function that should
965  // never be called. The CHECKs inside it make sure that never happens.
966  read_functions.emplace_back(invalid_read_func);
967  continue;
968  }
969 
970  if (QUERY_TYPE == QueryDescriptionType::GroupByBaselineHash) {
971  if (rows.getPaddedSlotWidthBytes(slot_idx_per_target_idx[target_idx]) == 0) {
972  // for key columns only
973  CHECK(rows.query_mem_desc_.getTargetGroupbyIndex(target_idx) >= 0);
974  if (target_types_[target_idx].is_fp()) {
975  CHECK_EQ(size_t(8), rows.query_mem_desc_.getEffectiveKeyWidth());
976  switch (target_types_[target_idx].get_type()) {
977  case kFLOAT:
978  read_functions.emplace_back(
979  read_float_key_baseline<QUERY_TYPE, COLUMNAR_OUTPUT>);
980  break;
981  case kDOUBLE:
982  read_functions.emplace_back(read_double_func<QUERY_TYPE, COLUMNAR_OUTPUT>);
983  break;
984  default:
985  UNREACHABLE()
986  << "Invalid data type encountered (BaselineHash, floating point key).";
987  break;
988  }
989  } else {
990  switch (rows.query_mem_desc_.getEffectiveKeyWidth()) {
991  case 8:
992  read_functions.emplace_back(read_int64_func<QUERY_TYPE, COLUMNAR_OUTPUT>);
993  break;
994  case 4:
995  read_functions.emplace_back(read_int32_func<QUERY_TYPE, COLUMNAR_OUTPUT>);
996  break;
997  default:
998  UNREACHABLE()
999  << "Invalid data type encountered (BaselineHash, integer key).";
1000  }
1001  }
1002  continue;
1003  }
1004  }
1005  if (target_types_[target_idx].is_fp()) {
1006  switch (rows.getPaddedSlotWidthBytes(slot_idx_per_target_idx[target_idx])) {
1007  case 8:
1008  read_functions.emplace_back(read_double_func<QUERY_TYPE, COLUMNAR_OUTPUT>);
1009  break;
1010  case 4:
1011  read_functions.emplace_back(read_float_func<QUERY_TYPE, COLUMNAR_OUTPUT>);
1012  break;
1013  default:
1014  UNREACHABLE() << "Invalid data type encountered (floating point agg column).";
1015  break;
1016  }
1017  } else {
1018  switch (rows.getPaddedSlotWidthBytes(slot_idx_per_target_idx[target_idx])) {
1019  case 8:
1020  read_functions.emplace_back(read_int64_func<QUERY_TYPE, COLUMNAR_OUTPUT>);
1021  break;
1022  case 4:
1023  read_functions.emplace_back(read_int32_func<QUERY_TYPE, COLUMNAR_OUTPUT>);
1024  break;
1025  case 2:
1026  read_functions.emplace_back(read_int16_func<QUERY_TYPE, COLUMNAR_OUTPUT>);
1027  break;
1028  case 1:
1029  read_functions.emplace_back(read_int8_func<QUERY_TYPE, COLUMNAR_OUTPUT>);
1030  break;
1031  default:
1032  UNREACHABLE() << "Invalid data type encountered (integer agg column).";
1033  break;
1034  }
1035  }
1036  }
1037  return read_functions;
1038 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
#define UNREACHABLE()
Definition: Logger.h:241
CHECK(cgen_state)
bool isDirectColumnarConversionPossible() const
int64_t invalid_read_func(const ResultSet &rows, const size_t input_buffer_entry_idx, const size_t target_idx, const size_t slot_idx)
const std::vector< SQLTypeInfo > target_types_

+ Here is the call graph for this function:

std::vector< ColumnarResults::WriteFunction > ColumnarResults::initWriteFunctions ( const ResultSet rows,
const std::vector< bool > &  targets_to_skip = {} 
)
private

Initialize a set of write functions per target (i.e., column). Target types' logical size are used to categorize the correct write function per target. These functions are then used for every row in the result set.

Definition at line 763 of file ColumnarResults.cpp.

References CHECK(), GroupByBaselineHash, GroupByPerfectHash, isDirectColumnarConversionPossible(), run_benchmark_import::result, target_types_, and UNREACHABLE.

Referenced by initAllConversionFunctions().

765  {
767  CHECK(rows.getQueryDescriptionType() == QueryDescriptionType::GroupByPerfectHash ||
768  rows.getQueryDescriptionType() == QueryDescriptionType::GroupByBaselineHash);
769 
770  std::vector<WriteFunction> result;
771  result.reserve(target_types_.size());
772 
773  for (size_t target_idx = 0; target_idx < target_types_.size(); target_idx++) {
774  if (!targets_to_skip.empty() && !targets_to_skip[target_idx]) {
775  result.emplace_back([](const ResultSet& rows,
776  const size_t input_buffer_entry_idx,
777  const size_t output_buffer_entry_idx,
778  const size_t target_idx,
779  const size_t slot_idx,
780  const ReadFunction& read_function) {
781  UNREACHABLE() << "Invalid write back function used.";
782  });
783  continue;
784  }
785 
786  if (target_types_[target_idx].is_fp()) {
787  switch (target_types_[target_idx].get_size()) {
788  case 8:
789  result.emplace_back(std::bind(&ColumnarResults::writeBackCellDirect<double>,
790  this,
791  std::placeholders::_1,
792  std::placeholders::_2,
793  std::placeholders::_3,
794  std::placeholders::_4,
795  std::placeholders::_5,
796  std::placeholders::_6));
797  break;
798  case 4:
799  result.emplace_back(std::bind(&ColumnarResults::writeBackCellDirect<float>,
800  this,
801  std::placeholders::_1,
802  std::placeholders::_2,
803  std::placeholders::_3,
804  std::placeholders::_4,
805  std::placeholders::_5,
806  std::placeholders::_6));
807  break;
808  default:
809  UNREACHABLE() << "Invalid target type encountered.";
810  break;
811  }
812  } else {
813  switch (target_types_[target_idx].get_size()) {
814  case 8:
815  result.emplace_back(std::bind(&ColumnarResults::writeBackCellDirect<int64_t>,
816  this,
817  std::placeholders::_1,
818  std::placeholders::_2,
819  std::placeholders::_3,
820  std::placeholders::_4,
821  std::placeholders::_5,
822  std::placeholders::_6));
823  break;
824  case 4:
825  result.emplace_back(std::bind(&ColumnarResults::writeBackCellDirect<int32_t>,
826  this,
827  std::placeholders::_1,
828  std::placeholders::_2,
829  std::placeholders::_3,
830  std::placeholders::_4,
831  std::placeholders::_5,
832  std::placeholders::_6));
833  break;
834  case 2:
835  result.emplace_back(std::bind(&ColumnarResults::writeBackCellDirect<int16_t>,
836  this,
837  std::placeholders::_1,
838  std::placeholders::_2,
839  std::placeholders::_3,
840  std::placeholders::_4,
841  std::placeholders::_5,
842  std::placeholders::_6));
843  break;
844  case 1:
845  result.emplace_back(std::bind(&ColumnarResults::writeBackCellDirect<int8_t>,
846  this,
847  std::placeholders::_1,
848  std::placeholders::_2,
849  std::placeholders::_3,
850  std::placeholders::_4,
851  std::placeholders::_5,
852  std::placeholders::_6));
853  break;
854  default:
855  UNREACHABLE() << "Invalid target type encountered.";
856  break;
857  }
858  }
859  }
860  return result;
861 }
#define UNREACHABLE()
Definition: Logger.h:241
CHECK(cgen_state)
std::function< int64_t(const ResultSet &, const size_t, const size_t, const size_t)> ReadFunction
bool isDirectColumnarConversionPossible() const
const std::vector< SQLTypeInfo > target_types_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

bool ColumnarResults::isDirectColumnarConversionPossible ( ) const
inline
bool ColumnarResults::isParallelConversion ( ) const
inline

Definition at line 88 of file ColumnarResults.h.

References parallel_conversion_.

Referenced by materializeAllColumnsGroupBy(), and materializeAllColumnsThroughIteration().

88 { return parallel_conversion_; }

+ Here is the caller graph for this function:

void ColumnarResults::locateAndCountEntries ( const ResultSet rows,
ColumnBitmap bitmap,
std::vector< size_t > &  non_empty_per_thread,
const size_t  entry_count,
const size_t  num_threads,
const size_t  size_per_thread 
) const
private

This function goes through all the keys in the result set, and count the total number of non-empty keys. It also store the location of non-empty keys in a bitmap data structure for later faster access.

Definition at line 499 of file ColumnarResults.cpp.

References CHECK(), CHECK_EQ, GroupByBaselineHash, GroupByPerfectHash, and isDirectColumnarConversionPossible().

Referenced by materializeAllColumnsGroupBy().

504  {
506  CHECK(rows.getQueryDescriptionType() == QueryDescriptionType::GroupByPerfectHash ||
507  rows.getQueryDescriptionType() == QueryDescriptionType::GroupByBaselineHash);
508  CHECK_EQ(num_threads, non_empty_per_thread.size());
509  auto locate_and_count_func =
510  [&rows, &bitmap, &non_empty_per_thread](
511  size_t start_index, size_t end_index, size_t thread_idx) {
512  size_t total_non_empty = 0;
513  size_t local_idx = 0;
514  for (size_t entry_idx = start_index; entry_idx < end_index;
515  entry_idx++, local_idx++) {
516  if (!rows.isRowAtEmpty(entry_idx)) {
517  total_non_empty++;
518  bitmap.set(local_idx, thread_idx, true);
519  }
520  }
521  non_empty_per_thread[thread_idx] = total_non_empty;
522  };
523 
524  std::vector<std::future<void>> conversion_threads;
525  for (size_t thread_idx = 0; thread_idx < num_threads; thread_idx++) {
526  const size_t start_entry = thread_idx * size_per_thread;
527  const size_t end_entry = std::min(start_entry + size_per_thread, entry_count);
528  conversion_threads.push_back(std::async(
529  std::launch::async, locate_and_count_func, start_entry, end_entry, thread_idx));
530  }
531 
532  for (auto& child : conversion_threads) {
533  child.wait();
534  }
535 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
CHECK(cgen_state)
bool isDirectColumnarConversionPossible() const

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void ColumnarResults::materializeAllColumnsDirectly ( const ResultSet rows,
const size_t  num_columns 
)
private

This function materializes all columns from the main storage and all appended storages and form a single continguous column for each output column. Depending on whether the column is lazily fetched or not, it will treat them differently.

NOTE: this function should only be used when the result set is columnar and completely compacted (e.g., in columnar projections).

Definition at line 304 of file ColumnarResults.cpp.

References CHECK(), GroupByBaselineHash, GroupByPerfectHash, isDirectColumnarConversionPossible(), materializeAllColumnsGroupBy(), materializeAllColumnsProjection(), Projection, and UNREACHABLE.

Referenced by ColumnarResults().

305  {
307  switch (rows.getQueryDescriptionType()) {
309  materializeAllColumnsProjection(rows, num_columns);
310  break;
311  }
314  materializeAllColumnsGroupBy(rows, num_columns);
315  break;
316  }
317  default:
318  UNREACHABLE()
319  << "Direct columnar conversion for this query type is not supported yet.";
320  }
321 }
#define UNREACHABLE()
Definition: Logger.h:241
CHECK(cgen_state)
void materializeAllColumnsGroupBy(const ResultSet &rows, const size_t num_columns)
bool isDirectColumnarConversionPossible() const
void materializeAllColumnsProjection(const ResultSet &rows, const size_t num_columns)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void ColumnarResults::materializeAllColumnsGroupBy ( const ResultSet rows,
const size_t  num_columns 
)
private

This function is to directly columnarize a result set for group by queries. Its main difference with the traditional alternative is that it directly reads non-empty entries from the result set, and then writes them into output column buffers, rather than using the result set's iterators.

Definition at line 464 of file ColumnarResults.cpp.

References CHECK(), compactAndCopyEntries(), cpu_threads(), GroupByBaselineHash, GroupByPerfectHash, isDirectColumnarConversionPossible(), isParallelConversion(), and locateAndCountEntries().

Referenced by materializeAllColumnsDirectly().

465  {
467  CHECK(rows.getQueryDescriptionType() == QueryDescriptionType::GroupByPerfectHash ||
468  rows.getQueryDescriptionType() == QueryDescriptionType::GroupByBaselineHash);
469 
470  const size_t num_threads = isParallelConversion() ? cpu_threads() : 1;
471  const size_t entry_count = rows.entryCount();
472  const size_t size_per_thread = (entry_count + num_threads - 1) / num_threads;
473 
474  // step 1: compute total non-empty elements and store a bitmap per thread
475  std::vector<size_t> non_empty_per_thread(num_threads,
476  0); // number of non-empty entries per thread
477 
478  ColumnBitmap bitmap(size_per_thread, num_threads);
479 
481  rows, bitmap, non_empty_per_thread, entry_count, num_threads, size_per_thread);
482 
483  // step 2: go through the generated bitmap and copy/decode corresponding entries
484  // into the output buffer
486  bitmap,
487  non_empty_per_thread,
488  num_columns,
489  entry_count,
490  num_threads,
491  size_per_thread);
492 }
bool isParallelConversion() const
void locateAndCountEntries(const ResultSet &rows, ColumnBitmap &bitmap, std::vector< size_t > &non_empty_per_thread, const size_t entry_count, const size_t num_threads, const size_t size_per_thread) const
void compactAndCopyEntries(const ResultSet &rows, const ColumnBitmap &bitmap, const std::vector< size_t > &non_empty_per_thread, const size_t num_columns, const size_t entry_count, const size_t num_threads, const size_t size_per_thread)
CHECK(cgen_state)
bool isDirectColumnarConversionPossible() const
int cpu_threads()
Definition: thread_count.h:25

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void ColumnarResults::materializeAllColumnsProjection ( const ResultSet rows,
const size_t  num_columns 
)
private

This function handles materialization for two types of columns in columnar projections:

  1. for all non-lazy columns, it directly copies the results from the result set's storage into the output column buffers
  2. for all lazy fetched columns, it uses result set's iterators to decode the proper values before storing them into the output column buffers

Definition at line 330 of file ColumnarResults.cpp.

References CHECK(), copyAllNonLazyColumns(), isDirectColumnarConversionPossible(), materializeAllLazyColumns(), and Projection.

Referenced by materializeAllColumnsDirectly().

331  {
332  CHECK(rows.query_mem_desc_.didOutputColumnar());
334  rows.query_mem_desc_.getQueryDescriptionType() ==
336 
337  const auto& lazy_fetch_info = rows.getLazyFetchInfo();
338 
339  // We can directly copy each non-lazy column's content
340  copyAllNonLazyColumns(lazy_fetch_info, rows, num_columns);
341 
342  // Only lazy columns are iterated through first and then materialized
343  materializeAllLazyColumns(lazy_fetch_info, rows, num_columns);
344 }
CHECK(cgen_state)
bool isDirectColumnarConversionPossible() const
void copyAllNonLazyColumns(const std::vector< ColumnLazyFetchInfo > &lazy_fetch_info, const ResultSet &rows, const size_t num_columns)
void materializeAllLazyColumns(const std::vector< ColumnLazyFetchInfo > &lazy_fetch_info, const ResultSet &rows, const size_t num_columns)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void ColumnarResults::materializeAllColumnsThroughIteration ( const ResultSet rows,
const size_t  num_columns 
)
private

This function iterates through the result set (using the getRowAtNoTranslation and getNextRow family of functions) and writes back the results into output column buffers.

Definition at line 152 of file ColumnarResults.cpp.

References cpu_threads(), isParallelConversion(), makeIntervals(), num_rows_, Asio::start(), and writeBackCell().

Referenced by ColumnarResults().

153  {
154  std::atomic<size_t> row_idx{0};
155  const auto do_work = [num_columns, this](const std::vector<TargetValue>& crt_row,
156  const size_t row_idx) {
157  for (size_t i = 0; i < num_columns; ++i) {
158  writeBackCell(crt_row[i], row_idx, i);
159  }
160  };
161  if (isParallelConversion()) {
162  const size_t worker_count = cpu_threads();
163  std::vector<std::future<void>> conversion_threads;
164  for (auto interval : makeIntervals(size_t(0), rows.entryCount(), worker_count)) {
165  conversion_threads.push_back(std::async(
166  std::launch::async,
167  [&rows, &do_work, &row_idx](const size_t start, const size_t end) {
168  for (size_t i = start; i < end; ++i) {
169  const auto crt_row = rows.getRowAtNoTranslations(i);
170  if (!crt_row.empty()) {
171  do_work(crt_row, row_idx.fetch_add(1));
172  }
173  }
174  },
175  interval.begin,
176  interval.end));
177  }
178  for (auto& child : conversion_threads) {
179  child.wait();
180  }
181 
182  num_rows_ = row_idx;
183  rows.setCachedRowCount(num_rows_);
184  return;
185  }
186  while (true) {
187  const auto crt_row = rows.getNextRow(false, false);
188  if (crt_row.empty()) {
189  break;
190  }
191  do_work(crt_row, row_idx);
192  ++row_idx;
193  }
194  rows.moveToBegin();
195 }
bool isParallelConversion() const
Intervals< T > makeIntervals(T begin, T end, std::size_t n_workers)
Definition: Intervals.h:108
void start()
Definition: Asio.cpp:33
void writeBackCell(const TargetValue &col_val, const size_t row_idx, const size_t column_idx)
int cpu_threads()
Definition: thread_count.h:25

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void ColumnarResults::materializeAllLazyColumns ( const std::vector< ColumnLazyFetchInfo > &  lazy_fetch_info,
const ResultSet rows,
const size_t  num_columns 
)
private

For all lazy fetched columns, we should iterate through the column's content and properly materialize it.

This function is parallelized through dividing total rows among all existing threads. Since there's no invalid element in the result set (e.g., columnar projections), the output buffer will have as many rows as there are in the result set, removing the need for atomicly incrementing the output buffer position.

Definition at line 398 of file ColumnarResults.cpp.

References CHECK(), CHECK_EQ, cpu_threads(), isDirectColumnarConversionPossible(), makeIntervals(), Asio::start(), use_parallel_algorithms(), and writeBackCell().

Referenced by materializeAllColumnsProjection().

401  {
403  const auto do_work_just_lazy_columns = [num_columns, this](
404  const std::vector<TargetValue>& crt_row,
405  const size_t row_idx,
406  const std::vector<bool>& targets_to_skip) {
407  for (size_t i = 0; i < num_columns; ++i) {
408  if (!targets_to_skip.empty() && !targets_to_skip[i]) {
409  writeBackCell(crt_row[i], row_idx, i);
410  }
411  }
412  };
413 
414  const auto contains_lazy_fetched_column =
415  [](const std::vector<ColumnLazyFetchInfo>& lazy_fetch_info) {
416  for (auto& col_info : lazy_fetch_info) {
417  if (col_info.is_lazily_fetched) {
418  return true;
419  }
420  }
421  return false;
422  };
423 
424  // parallelized by assigning a chunk of rows to each thread)
425  const bool skip_non_lazy_columns = rows.isPermutationBufferEmpty();
426  if (contains_lazy_fetched_column(lazy_fetch_info)) {
427  const size_t worker_count = use_parallel_algorithms(rows) ? cpu_threads() : 1;
428  std::vector<std::future<void>> conversion_threads;
429  std::vector<bool> targets_to_skip;
430  if (skip_non_lazy_columns) {
431  CHECK_EQ(lazy_fetch_info.size(), size_t(num_columns));
432  targets_to_skip.reserve(num_columns);
433  for (size_t i = 0; i < num_columns; i++) {
434  // we process lazy columns (i.e., skip non-lazy columns)
435  targets_to_skip.push_back(!lazy_fetch_info[i].is_lazily_fetched);
436  }
437  }
438  for (auto interval : makeIntervals(size_t(0), rows.entryCount(), worker_count)) {
439  conversion_threads.push_back(std::async(
440  std::launch::async,
441  [&rows, &do_work_just_lazy_columns, &targets_to_skip](const size_t start,
442  const size_t end) {
443  for (size_t i = start; i < end; ++i) {
444  const auto crt_row = rows.getRowAtNoTranslations(i, targets_to_skip);
445  do_work_just_lazy_columns(crt_row, i, targets_to_skip);
446  }
447  },
448  interval.begin,
449  interval.end));
450  }
451 
452  for (auto& child : conversion_threads) {
453  child.wait();
454  }
455  }
456 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
bool use_parallel_algorithms(const ResultSet &rows)
Definition: ResultSet.cpp:899
Intervals< T > makeIntervals(T begin, T end, std::size_t n_workers)
Definition: Intervals.h:108
void start()
Definition: Asio.cpp:33
CHECK(cgen_state)
bool isDirectColumnarConversionPossible() const
void writeBackCell(const TargetValue &col_val, const size_t row_idx, const size_t column_idx)
int cpu_threads()
Definition: thread_count.h:25

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

std::unique_ptr< ColumnarResults > ColumnarResults::mergeResults ( const std::shared_ptr< RowSetMemoryOwner row_set_mem_owner,
const std::vector< std::unique_ptr< ColumnarResults >> &  sub_results 
)
static

Definition at line 106 of file ColumnarResults.cpp.

References CHECK_EQ, checked_malloc(), ColumnarResults(), logger::init(), run_benchmark_import::result, and target_types_.

Referenced by ColumnFetcher::getAllTableColumnFragments().

108  {
109  if (sub_results.empty()) {
110  return nullptr;
111  }
112  const auto total_row_count = std::accumulate(
113  sub_results.begin(),
114  sub_results.end(),
115  size_t(0),
116  [](const size_t init, const std::unique_ptr<ColumnarResults>& result) {
117  return init + result->size();
118  });
119  std::unique_ptr<ColumnarResults> merged_results(
120  new ColumnarResults(total_row_count, sub_results[0]->target_types_));
121  const auto col_count = sub_results[0]->column_buffers_.size();
122  const auto nonempty_it = std::find_if(
123  sub_results.begin(),
124  sub_results.end(),
125  [](const std::unique_ptr<ColumnarResults>& needle) { return needle->size(); });
126  if (nonempty_it == sub_results.end()) {
127  return nullptr;
128  }
129  for (size_t col_idx = 0; col_idx < col_count; ++col_idx) {
130  const auto byte_width = (*nonempty_it)->getColumnType(col_idx).get_size();
131  auto write_ptr =
132  reinterpret_cast<int8_t*>(checked_malloc(byte_width * total_row_count));
133  merged_results->column_buffers_.push_back(write_ptr);
134  row_set_mem_owner->addColBuffer(write_ptr);
135  for (auto& rs : sub_results) {
136  CHECK_EQ(col_count, rs->column_buffers_.size());
137  if (!rs->size()) {
138  continue;
139  }
140  CHECK_EQ(byte_width, rs->getColumnType(col_idx).get_size());
141  memcpy(write_ptr, rs->column_buffers_[col_idx], rs->size() * byte_width);
142  write_ptr += rs->size() * byte_width;
143  }
144  }
145  return merged_results;
146 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
void init(LogOptions const &log_opts)
Definition: Logger.cpp:276
void * checked_malloc(const size_t size)
Definition: checked_alloc.h:40
ColumnarResults(const std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner, const ResultSet &rows, const size_t num_columns, const std::vector< SQLTypeInfo > &target_types, const bool is_parallel_execution_enforced=false)
const std::vector< SQLTypeInfo > target_types_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

const size_t ColumnarResults::size ( ) const
inline

Definition at line 80 of file ColumnarResults.h.

References num_rows_.

Referenced by ColumnFetcher::transferColumnIfNeeded().

80 { return num_rows_; }

+ Here is the caller graph for this function:

void ColumnarResults::writeBackCell ( const TargetValue col_val,
const size_t  row_idx,
const size_t  column_idx 
)
inlineprivate

Definition at line 205 of file ColumnarResults.cpp.

References CHECK(), column_buffers_, anonymous_namespace{ColumnarResults.cpp}::fixed_encoding_nullable_val(), kDOUBLE, kFLOAT, and target_types_.

Referenced by compactAndCopyEntriesWithTargetSkipping(), materializeAllColumnsThroughIteration(), and materializeAllLazyColumns().

207  {
208  const auto scalar_col_val = boost::get<ScalarTargetValue>(&col_val);
209  CHECK(scalar_col_val);
210  auto i64_p = boost::get<int64_t>(scalar_col_val);
211  const auto& type_info = target_types_[column_idx];
212  if (i64_p) {
213  const auto val = fixed_encoding_nullable_val(*i64_p, type_info);
214  switch (target_types_[column_idx].get_size()) {
215  case 1:
216  ((int8_t*)column_buffers_[column_idx])[row_idx] = static_cast<int8_t>(val);
217  break;
218  case 2:
219  ((int16_t*)column_buffers_[column_idx])[row_idx] = static_cast<int16_t>(val);
220  break;
221  case 4:
222  ((int32_t*)column_buffers_[column_idx])[row_idx] = static_cast<int32_t>(val);
223  break;
224  case 8:
225  ((int64_t*)column_buffers_[column_idx])[row_idx] = val;
226  break;
227  default:
228  CHECK(false);
229  }
230  } else {
231  CHECK(target_types_[column_idx].is_fp());
232  switch (target_types_[column_idx].get_type()) {
233  case kFLOAT: {
234  auto float_p = boost::get<float>(scalar_col_val);
235  ((float*)column_buffers_[column_idx])[row_idx] = static_cast<float>(*float_p);
236  break;
237  }
238  case kDOUBLE: {
239  auto double_p = boost::get<double>(scalar_col_val);
240  ((double*)column_buffers_[column_idx])[row_idx] = static_cast<double>(*double_p);
241  break;
242  }
243  default:
244  CHECK(false);
245  }
246  }
247 }
std::vector< int8_t * > column_buffers_
CHECK(cgen_state)
int64_t fixed_encoding_nullable_val(const int64_t val, const SQLTypeInfo &type_info)
const std::vector< SQLTypeInfo > target_types_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

template<typename DATA_TYPE >
void ColumnarResults::writeBackCellDirect ( const ResultSet rows,
const size_t  input_buffer_entry_idx,
const size_t  output_buffer_entry_idx,
const size_t  target_idx,
const size_t  slot_idx,
const ReadFunction read_from_function 
)
private

A set of write functions to be used to directly write into final column_buffers_. The read_from_function is used to read from the input result set's storage NOTE: currently only used for direct columnarizations

Definition at line 255 of file ColumnarResults.cpp.

References column_buffers_, anonymous_namespace{ColumnarResults.cpp}::fixed_encoding_nullable_val(), and target_types_.

260  {
261  const auto val = static_cast<DATA_TYPE>(fixed_encoding_nullable_val(
262  read_from_function(rows, input_buffer_entry_idx, target_idx, slot_idx),
263  target_types_[target_idx]));
264  reinterpret_cast<DATA_TYPE*>(column_buffers_[target_idx])[output_buffer_entry_idx] =
265  val;
266 }
std::vector< int8_t * > column_buffers_
int64_t fixed_encoding_nullable_val(const int64_t val, const SQLTypeInfo &type_info)
const std::vector< SQLTypeInfo > target_types_

+ Here is the call graph for this function:

template<>
void ColumnarResults::writeBackCellDirect ( const ResultSet rows,
const size_t  input_buffer_entry_idx,
const size_t  output_buffer_entry_idx,
const size_t  target_idx,
const size_t  slot_idx,
const ReadFunction read_from_function 
)
private

Definition at line 269 of file ColumnarResults.cpp.

274  {
275  const int32_t ival =
276  read_from_function(rows, input_buffer_entry_idx, target_idx, slot_idx);
277  const float fval = *reinterpret_cast<const float*>(may_alias_ptr(&ival));
278  reinterpret_cast<float*>(column_buffers_[target_idx])[output_buffer_entry_idx] = fval;
279 }
std::vector< int8_t * > column_buffers_
template<>
void ColumnarResults::writeBackCellDirect ( const ResultSet rows,
const size_t  input_buffer_entry_idx,
const size_t  output_buffer_entry_idx,
const size_t  target_idx,
const size_t  slot_idx,
const ReadFunction read_from_function 
)
private

Definition at line 282 of file ColumnarResults.cpp.

288  {
289  const int64_t ival =
290  read_from_function(rows, input_buffer_entry_idx, target_idx, slot_idx);
291  const double dval = *reinterpret_cast<const double*>(may_alias_ptr(&ival));
292  reinterpret_cast<double*>(column_buffers_[target_idx])[output_buffer_entry_idx] = dval;
293 }
std::vector< int8_t * > column_buffers_

Member Data Documentation

std::vector<int8_t*> ColumnarResults::column_buffers_
protected
bool ColumnarResults::direct_columnar_conversion_
private

Definition at line 193 of file ColumnarResults.h.

Referenced by isDirectColumnarConversionPossible().

size_t ColumnarResults::num_rows_
protected
bool ColumnarResults::parallel_conversion_
private

Definition at line 191 of file ColumnarResults.h.

Referenced by isParallelConversion().

const std::vector<SQLTypeInfo> ColumnarResults::target_types_
private

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