OmniSciDB  c1a53651b2
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
InputMetadata.cpp File Reference
#include "InputMetadata.h"
#include "Execute.h"
#include "../Fragmenter/Fragmenter.h"
#include <tbb/parallel_for.h>
#include <tbb/task_arena.h>
#include <future>
+ Include dependency graph for InputMetadata.cpp:

Go to the source code of this file.

Namespaces

 

Functions

Fragmenter_Namespace::TableInfo anonymous_namespace{InputMetadata.cpp}::copy_table_info (const Fragmenter_Namespace::TableInfo &table_info)
 
Fragmenter_Namespace::TableInfo build_table_info (const std::vector< const TableDescriptor * > &shard_tables)
 
bool anonymous_namespace{InputMetadata.cpp}::uses_int_meta (const SQLTypeInfo &col_ti)
 
Fragmenter_Namespace::TableInfo anonymous_namespace{InputMetadata.cpp}::synthesize_table_info (const ResultSetPtr &rows)
 
void anonymous_namespace{InputMetadata.cpp}::collect_table_infos (std::vector< InputTableInfo > &table_infos, const std::vector< InputDescriptor > &input_descs, Executor *executor)
 
template<typename T >
void compute_table_function_col_chunk_stats (std::shared_ptr< ChunkMetadata > &chunk_metadata, const T *values_buffer, const size_t values_count, const T null_val)
 
ChunkMetadataMap synthesize_metadata_table_function (const ResultSet *rows)
 
ChunkMetadataMap synthesize_metadata (const ResultSet *rows)
 
size_t get_frag_count_of_table (const shared::TableKey &table_key, Executor *executor)
 
std::vector< InputTableInfoget_table_infos (const std::vector< InputDescriptor > &input_descs, Executor *executor)
 
std::vector< InputTableInfoget_table_infos (const RelAlgExecutionUnit &ra_exe_unit, Executor *executor)
 

Variables

bool g_enable_data_recycler
 
bool g_use_chunk_metadata_cache
 

Function Documentation

Fragmenter_Namespace::TableInfo build_table_info ( const std::vector< const TableDescriptor * > &  shard_tables)

Definition at line 44 of file InputMetadata.cpp.

References CHECK, Fragmenter_Namespace::TableInfo::fragments, and Fragmenter_Namespace::TableInfo::setPhysicalNumTuples().

Referenced by InputTableInfoCache::getTableInfo().

45  {
46  size_t total_number_of_tuples{0};
47  Fragmenter_Namespace::TableInfo table_info_all_shards;
48  for (const TableDescriptor* shard_table : shard_tables) {
49  CHECK(shard_table->fragmenter);
50  const auto& shard_metainfo = shard_table->fragmenter->getFragmentsForQuery();
51  total_number_of_tuples += shard_metainfo.getPhysicalNumTuples();
52  table_info_all_shards.fragments.reserve(table_info_all_shards.fragments.size() +
53  shard_metainfo.fragments.size());
54  table_info_all_shards.fragments.insert(table_info_all_shards.fragments.end(),
55  shard_metainfo.fragments.begin(),
56  shard_metainfo.fragments.end());
57  }
58  table_info_all_shards.setPhysicalNumTuples(total_number_of_tuples);
59  return table_info_all_shards;
60 }
std::vector< FragmentInfo > fragments
Definition: Fragmenter.h:171
#define CHECK(condition)
Definition: Logger.h:291
void setPhysicalNumTuples(const size_t physNumTuples)
Definition: Fragmenter.h:166

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

template<typename T >
void compute_table_function_col_chunk_stats ( std::shared_ptr< ChunkMetadata > &  chunk_metadata,
const T *  values_buffer,
const size_t  values_count,
const T  null_val 
)

Definition at line 142 of file InputMetadata.cpp.

References max_inputs_per_thread, threading_serial::parallel_for(), and heavydb.dtypes::T.

Referenced by synthesize_metadata_table_function().

146  {
147  T min_val{std::numeric_limits<T>::max()};
148  T max_val{std::numeric_limits<T>::lowest()};
149  bool has_nulls{false};
150  constexpr size_t parallel_stats_compute_threshold = 20000UL;
151  if (values_count < parallel_stats_compute_threshold) {
152  for (size_t row_idx = 0; row_idx < values_count; ++row_idx) {
153  const T cell_val = values_buffer[row_idx];
154  if (cell_val == null_val) {
155  has_nulls = true;
156  continue;
157  }
158  if (cell_val < min_val) {
159  min_val = cell_val;
160  }
161  if (cell_val > max_val) {
162  max_val = cell_val;
163  }
164  }
165  } else {
166  const size_t max_thread_count = std::thread::hardware_concurrency();
167  const size_t max_inputs_per_thread = 20000;
168  const size_t min_grain_size = max_inputs_per_thread / 2;
169  const size_t num_threads =
170  std::min(max_thread_count,
171  ((values_count + max_inputs_per_thread - 1) / max_inputs_per_thread));
172 
173  std::vector<T> threads_local_mins(num_threads, std::numeric_limits<T>::max());
174  std::vector<T> threads_local_maxes(num_threads, std::numeric_limits<T>::lowest());
175  std::vector<bool> threads_local_has_nulls(num_threads, false);
176  tbb::task_arena limited_arena(num_threads);
177 
178  limited_arena.execute([&] {
180  tbb::blocked_range<size_t>(0, values_count, min_grain_size),
181  [&](const tbb::blocked_range<size_t>& r) {
182  const size_t start_idx = r.begin();
183  const size_t end_idx = r.end();
184  T local_min_val = std::numeric_limits<T>::max();
185  T local_max_val = std::numeric_limits<T>::lowest();
186  bool local_has_nulls = false;
187  for (size_t row_idx = start_idx; row_idx < end_idx; ++row_idx) {
188  const T cell_val = values_buffer[row_idx];
189  if (cell_val == null_val) {
190  local_has_nulls = true;
191  continue;
192  }
193  if (cell_val < local_min_val) {
194  local_min_val = cell_val;
195  }
196  if (cell_val > local_max_val) {
197  local_max_val = cell_val;
198  }
199  }
200  size_t thread_idx = tbb::this_task_arena::current_thread_index();
201  if (local_min_val < threads_local_mins[thread_idx]) {
202  threads_local_mins[thread_idx] = local_min_val;
203  }
204  if (local_max_val > threads_local_maxes[thread_idx]) {
205  threads_local_maxes[thread_idx] = local_max_val;
206  }
207  if (local_has_nulls) {
208  threads_local_has_nulls[thread_idx] = true;
209  }
210  },
211  tbb::simple_partitioner());
212  });
213 
214  for (size_t thread_idx = 0; thread_idx < num_threads; ++thread_idx) {
215  if (threads_local_mins[thread_idx] < min_val) {
216  min_val = threads_local_mins[thread_idx];
217  }
218  if (threads_local_maxes[thread_idx] > max_val) {
219  max_val = threads_local_maxes[thread_idx];
220  }
221  has_nulls |= threads_local_has_nulls[thread_idx];
222  }
223  }
224  chunk_metadata->fillChunkStats(min_val, max_val, has_nulls);
225 }
const size_t max_inputs_per_thread
void parallel_for(const blocked_range< Int > &range, const Body &body, const Partitioner &p=Partitioner())

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

size_t get_frag_count_of_table ( const shared::TableKey table_key,
Executor executor 
)

Definition at line 456 of file InputMetadata.cpp.

References CHECK, CHECK_GE, and shared::TableKey::table_id.

Referenced by RelAlgExecutor::getOuterFragmentCount().

456  {
457  const auto temporary_tables = executor->getTemporaryTables();
458  CHECK(temporary_tables);
459  auto it = temporary_tables->find(table_key.table_id);
460  if (it != temporary_tables->end()) {
461  CHECK_GE(int(0), table_key.table_id);
462  return size_t(1);
463  } else {
464  const auto table_info = executor->getTableInfo(table_key);
465  return table_info.fragments.size();
466  }
467 }
#define CHECK_GE(x, y)
Definition: Logger.h:306
#define CHECK(condition)
Definition: Logger.h:291

+ Here is the caller graph for this function:

std::vector<InputTableInfo> get_table_infos ( const std::vector< InputDescriptor > &  input_descs,
Executor executor 
)

Definition at line 469 of file InputMetadata.cpp.

References anonymous_namespace{InputMetadata.cpp}::collect_table_infos().

Referenced by RelAlgExecutor::computeWindow(), RelAlgExecutor::createAggregateWorkUnit(), RelAlgExecutor::createCompoundWorkUnit(), RelAlgExecutor::createFilterWorkUnit(), RelAlgExecutor::createProjectWorkUnit(), RelAlgExecutor::createTableFunctionWorkUnit(), RelAlgExecutor::createUnionWorkUnit(), RelAlgExecutor::executeDelete(), RelAlgExecutor::executeTableFunction(), RelAlgExecutor::executeUpdate(), RelAlgExecutor::executeWorkUnit(), TableOptimizer::getDeletedColumnStats(), RelAlgExecutor::getFilteredCountAll(), RelAlgExecutor::getFilterSelectivity(), RelAlgExecutor::getNDVEstimation(), RelAlgExecutor::handleOutOfMemoryRetry(), TableOptimizer::recomputeColumnMetadata(), and RelAlgExecutor::selectFiltersToBePushedDown().

471  {
472  std::vector<InputTableInfo> table_infos;
473  collect_table_infos(table_infos, input_descs, executor);
474  return table_infos;
475 }
void collect_table_infos(std::vector< InputTableInfo > &table_infos, const std::vector< InputDescriptor > &input_descs, Executor *executor)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

std::vector<InputTableInfo> get_table_infos ( const RelAlgExecutionUnit ra_exe_unit,
Executor executor 
)

Definition at line 477 of file InputMetadata.cpp.

References anonymous_namespace{InputMetadata.cpp}::collect_table_infos(), and RelAlgExecutionUnit::input_descs.

478  {
479  std::vector<InputTableInfo> table_infos;
480  collect_table_infos(table_infos, ra_exe_unit.input_descs, executor);
481  return table_infos;
482 }
std::vector< InputDescriptor > input_descs
void collect_table_infos(std::vector< InputTableInfo > &table_infos, const std::vector< InputDescriptor > &input_descs, Executor *executor)

+ Here is the call graph for this function:

ChunkMetadataMap synthesize_metadata ( const ResultSet rows)

Definition at line 332 of file InputMetadata.cpp.

References threading_serial::async(), CHECK, CHECK_LT, cpu_threads(), Encoder::Create(), DEBUG_TIMER, inline_fp_null_val(), inline_int_null_val(), kDOUBLE, kFLOAT, synthesize_metadata_table_function(), TableFunction, result_set::use_parallel_algorithms(), and anonymous_namespace{InputMetadata.cpp}::uses_int_meta().

Referenced by Fragmenter_Namespace::FragmentInfo::getChunkMetadataMap().

332  {
333  auto timer = DEBUG_TIMER(__func__);
334  ChunkMetadataMap metadata_map;
335 
336  if (rows->definitelyHasNoRows()) {
337  // resultset has no valid storage, so we fill dummy metadata and return early
338  std::vector<std::unique_ptr<Encoder>> decoders;
339  for (size_t i = 0; i < rows->colCount(); ++i) {
340  decoders.emplace_back(Encoder::Create(nullptr, rows->getColType(i)));
341  const auto it_ok =
342  metadata_map.emplace(i, decoders.back()->getMetadata(rows->getColType(i)));
343  CHECK(it_ok.second);
344  }
345  return metadata_map;
346  }
347 
348  std::vector<std::vector<std::unique_ptr<Encoder>>> dummy_encoders;
349  const size_t worker_count =
351  for (size_t worker_idx = 0; worker_idx < worker_count; ++worker_idx) {
352  dummy_encoders.emplace_back();
353  for (size_t i = 0; i < rows->colCount(); ++i) {
354  const auto& col_ti = rows->getColType(i);
355  dummy_encoders.back().emplace_back(Encoder::Create(nullptr, col_ti));
356  }
357  }
358 
359  if (rows->getQueryMemDesc().getQueryDescriptionType() ==
362  }
363  rows->moveToBegin();
364  const auto do_work = [rows](const std::vector<TargetValue>& crt_row,
365  std::vector<std::unique_ptr<Encoder>>& dummy_encoders) {
366  for (size_t i = 0; i < rows->colCount(); ++i) {
367  const auto& col_ti = rows->getColType(i);
368  const auto& col_val = crt_row[i];
369  const auto scalar_col_val = boost::get<ScalarTargetValue>(&col_val);
370  CHECK(scalar_col_val);
371  if (uses_int_meta(col_ti)) {
372  const auto i64_p = boost::get<int64_t>(scalar_col_val);
373  CHECK(i64_p);
374  dummy_encoders[i]->updateStats(*i64_p, *i64_p == inline_int_null_val(col_ti));
375  } else if (col_ti.is_fp()) {
376  switch (col_ti.get_type()) {
377  case kFLOAT: {
378  const auto float_p = boost::get<float>(scalar_col_val);
379  CHECK(float_p);
380  dummy_encoders[i]->updateStats(*float_p,
381  *float_p == inline_fp_null_val(col_ti));
382  break;
383  }
384  case kDOUBLE: {
385  const auto double_p = boost::get<double>(scalar_col_val);
386  CHECK(double_p);
387  dummy_encoders[i]->updateStats(*double_p,
388  *double_p == inline_fp_null_val(col_ti));
389  break;
390  }
391  default:
392  CHECK(false);
393  }
394  } else {
395  throw std::runtime_error(col_ti.get_type_name() +
396  " is not supported in temporary table.");
397  }
398  }
399  };
401  const size_t worker_count = cpu_threads();
402  std::vector<std::future<void>> compute_stats_threads;
403  const auto entry_count = rows->entryCount();
404  for (size_t i = 0,
405  start_entry = 0,
406  stride = (entry_count + worker_count - 1) / worker_count;
407  i < worker_count && start_entry < entry_count;
408  ++i, start_entry += stride) {
409  const auto end_entry = std::min(start_entry + stride, entry_count);
410  compute_stats_threads.push_back(std::async(
412  [rows, &do_work, &dummy_encoders](
413  const size_t start, const size_t end, const size_t worker_idx) {
414  for (size_t i = start; i < end; ++i) {
415  const auto crt_row = rows->getRowAtNoTranslations(i);
416  if (!crt_row.empty()) {
417  do_work(crt_row, dummy_encoders[worker_idx]);
418  }
419  }
420  },
421  start_entry,
422  end_entry,
423  i));
424  }
425  for (auto& child : compute_stats_threads) {
426  child.wait();
427  }
428  for (auto& child : compute_stats_threads) {
429  child.get();
430  }
431  } else {
432  while (true) {
433  auto crt_row = rows->getNextRow(false, false);
434  if (crt_row.empty()) {
435  break;
436  }
437  do_work(crt_row, dummy_encoders[0]);
438  }
439  }
440  rows->moveToBegin();
441  for (size_t worker_idx = 1; worker_idx < worker_count; ++worker_idx) {
442  CHECK_LT(worker_idx, dummy_encoders.size());
443  const auto& worker_encoders = dummy_encoders[worker_idx];
444  for (size_t i = 0; i < rows->colCount(); ++i) {
445  dummy_encoders[0][i]->reduceStats(*worker_encoders[i]);
446  }
447  }
448  for (size_t i = 0; i < rows->colCount(); ++i) {
449  const auto it_ok =
450  metadata_map.emplace(i, dummy_encoders[0][i]->getMetadata(rows->getColType(i)));
451  CHECK(it_ok.second);
452  }
453  return metadata_map;
454 }
ChunkMetadataMap synthesize_metadata_table_function(const ResultSet *rows)
static Encoder * Create(Data_Namespace::AbstractBuffer *buffer, const SQLTypeInfo sqlType)
Definition: Encoder.cpp:26
double inline_fp_null_val(const SQL_TYPE_INFO &ti)
std::map< int, std::shared_ptr< ChunkMetadata >> ChunkMetadataMap
bool use_parallel_algorithms(const ResultSet &rows)
Definition: ResultSet.cpp:1577
future< Result > async(Fn &&fn, Args &&...args)
bool uses_int_meta(const SQLTypeInfo &col_ti)
#define CHECK_LT(x, y)
Definition: Logger.h:303
#define CHECK(condition)
Definition: Logger.h:291
#define DEBUG_TIMER(name)
Definition: Logger.h:411
int64_t inline_int_null_val(const SQL_TYPE_INFO &ti)
int cpu_threads()
Definition: thread_count.h:25

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ChunkMetadataMap synthesize_metadata_table_function ( const ResultSet rows)

Definition at line 227 of file InputMetadata.cpp.

References CHECK, CHECK_EQ, compute_table_function_col_chunk_stats(), FlatBufferManager::flatbufferSize(), inline_fixed_encoding_null_val(), inline_fp_null_value< double >(), inline_fp_null_value< float >(), FlatBufferManager::isFlatBuffer(), kBIGINT, kBOOLEAN, kDOUBLE, kENCODING_DICT, kENCODING_NONE, kFLOAT, kINT, kSMALLINT, kTEXT, kTIMESTAMP, kTINYINT, TableFunction, and UNREACHABLE.

Referenced by synthesize_metadata().

227  {
228  CHECK(rows->getQueryMemDesc().getQueryDescriptionType() ==
230  CHECK(rows->didOutputColumnar());
231  CHECK(!(rows->areAnyColumnsLazyFetched()));
232  const size_t col_count = rows->colCount();
233  const auto row_count = rows->entryCount();
234 
235  ChunkMetadataMap chunk_metadata_map;
236 
237  for (size_t col_idx = 0; col_idx < col_count; ++col_idx) {
238  std::shared_ptr<ChunkMetadata> chunk_metadata = std::make_shared<ChunkMetadata>();
239  const int8_t* columnar_buffer = const_cast<int8_t*>(rows->getColumnarBuffer(col_idx));
240  const auto col_sql_type_info = rows->getColType(col_idx);
241  // Here, min/max of a column of arrays, col, is defined as
242  // min/max(unnest(col)). That is, if is_array is true, the
243  // metadata is supposed to be syntesized for a query like `SELECT
244  // UNNEST(col_of_arrays) ... GROUP BY ...`. How can we verify that
245  // here?
246  bool is_array = col_sql_type_info.is_array();
247  const auto col_type =
248  (is_array ? col_sql_type_info.get_subtype() : col_sql_type_info.get_type());
249  const auto col_type_info =
250  (is_array ? col_sql_type_info.get_elem_type() : col_sql_type_info);
251 
252  chunk_metadata->sqlType = col_type_info;
253  chunk_metadata->numElements = row_count;
254 
255  const int8_t* values_buffer;
256  size_t values_count;
257  if (is_array) {
258  CHECK(FlatBufferManager::isFlatBuffer(columnar_buffer));
259  FlatBufferManager m{const_cast<int8_t*>(columnar_buffer)};
260  chunk_metadata->numBytes = m.flatbufferSize();
261  values_count = m.VarlenArray_nof_values();
262  values_buffer = m.VarlenArray_values();
263  } else {
264  chunk_metadata->numBytes = row_count * col_type_info.get_size();
265  values_count = row_count;
266  values_buffer = columnar_buffer;
267  }
268 
269  if (col_type != kTEXT) {
270  CHECK(col_type_info.get_compression() == kENCODING_NONE);
271  } else {
272  CHECK(col_type_info.get_compression() == kENCODING_DICT);
273  CHECK_EQ(col_type_info.get_size(), sizeof(int32_t));
274  }
275 
276  switch (col_type) {
277  case kBOOLEAN:
278  case kTINYINT:
280  chunk_metadata,
281  values_buffer,
282  values_count,
283  static_cast<int8_t>(inline_fixed_encoding_null_val(col_type_info)));
284  break;
285  case kSMALLINT:
287  chunk_metadata,
288  reinterpret_cast<const int16_t*>(values_buffer),
289  values_count,
290  static_cast<int16_t>(inline_fixed_encoding_null_val(col_type_info)));
291  break;
292  case kINT:
293  case kTEXT:
295  chunk_metadata,
296  reinterpret_cast<const int32_t*>(values_buffer),
297  values_count,
298  static_cast<int32_t>(inline_fixed_encoding_null_val(col_type_info)));
299  break;
300  case kBIGINT:
301  case kTIMESTAMP:
303  chunk_metadata,
304  reinterpret_cast<const int64_t*>(values_buffer),
305  values_count,
306  static_cast<int64_t>(inline_fixed_encoding_null_val(col_type_info)));
307  break;
308  case kFLOAT:
309  // For float use the typed null accessor as the generic one converts to double,
310  // and do not want to risk loss of precision
312  chunk_metadata,
313  reinterpret_cast<const float*>(values_buffer),
314  values_count,
316  break;
317  case kDOUBLE:
319  chunk_metadata,
320  reinterpret_cast<const double*>(values_buffer),
321  values_count,
323  break;
324  default:
325  UNREACHABLE();
326  }
327  chunk_metadata_map.emplace(col_idx, chunk_metadata);
328  }
329  return chunk_metadata_map;
330 }
#define CHECK_EQ(x, y)
Definition: Logger.h:301
#define UNREACHABLE()
Definition: Logger.h:337
std::map< int, std::shared_ptr< ChunkMetadata >> ChunkMetadataMap
Definition: sqltypes.h:69
constexpr float inline_fp_null_value< float >()
constexpr double inline_fp_null_value< double >()
int64_t flatbufferSize() const
Definition: FlatBuffer.h:219
#define CHECK(condition)
Definition: Logger.h:291
void compute_table_function_col_chunk_stats(std::shared_ptr< ChunkMetadata > &chunk_metadata, const T *values_buffer, const size_t values_count, const T null_val)
int64_t inline_fixed_encoding_null_val(const SQL_TYPE_INFO &ti)
Definition: sqltypes.h:62
HOST static DEVICE bool isFlatBuffer(const void *buffer)
Definition: FlatBuffer.h:186

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

Variable Documentation

bool g_enable_data_recycler

Definition at line 146 of file Execute.cpp.

bool g_use_chunk_metadata_cache

Definition at line 149 of file Execute.cpp.