OmniSciDB  7bf56492aa
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
RelAlgExecutor Class Reference

#include <RelAlgExecutor.h>

+ Inheritance diagram for RelAlgExecutor:
+ Collaboration diagram for RelAlgExecutor:

Classes

struct  TableFunctionWorkUnit
 
struct  WorkUnit
 

Public Types

using TargetInfoList = std::vector< TargetInfo >
 

Public Member Functions

 RelAlgExecutor (Executor *executor, const Catalog_Namespace::Catalog &cat, std::shared_ptr< const query_state::QueryState > query_state=nullptr)
 
 RelAlgExecutor (Executor *executor, const Catalog_Namespace::Catalog &cat, const std::string &query_ra, std::shared_ptr< const query_state::QueryState > query_state=nullptr)
 
 RelAlgExecutor (Executor *executor, const Catalog_Namespace::Catalog &cat, std::unique_ptr< RelAlgDagBuilder > query_dag, std::shared_ptr< const query_state::QueryState > query_state=nullptr)
 
size_t getOuterFragmentCount (const CompilationOptions &co, const ExecutionOptions &eo)
 
ExecutionResult executeRelAlgQuery (const CompilationOptions &co, const ExecutionOptions &eo, const bool just_explain_plan, RenderInfo *render_info)
 
ExecutionResult executeRelAlgQueryWithFilterPushDown (const RaExecutionSequence &seq, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const int64_t queue_time_ms)
 
void prepareLeafExecution (const AggregatedColRange &agg_col_range, const StringDictionaryGenerations &string_dictionary_generations, const TableGenerations &table_generations)
 
ExecutionResult executeRelAlgSeq (const RaExecutionSequence &seq, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const int64_t queue_time_ms, const bool with_existing_temp_tables=false)
 
ExecutionResult executeRelAlgSubSeq (const RaExecutionSequence &seq, const std::pair< size_t, size_t > interval, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const int64_t queue_time_ms)
 
FirstStepExecutionResult executeRelAlgQuerySingleStep (const RaExecutionSequence &seq, const size_t step_idx, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info)
 
void addLeafResult (const unsigned id, const AggregatedResult &result)
 
const RelAlgNodegetRootRelAlgNode () const
 
const std::vector
< std::shared_ptr< RexSubQuery > > & 
getSubqueries () const noexcept
 
ExecutionResult executeSimpleInsert (const Analyzer::Query &insert_query)
 
AggregatedColRange computeColRangesCache ()
 
StringDictionaryGenerations computeStringDictionaryGenerations ()
 
TableGenerations computeTableGenerations ()
 
ExecutorgetExecutor () const
 
void cleanupPostExecution ()
 

Static Public Member Functions

static std::string getErrorMessageFromCode (const int32_t error_code)
 

Private Member Functions

ExecutionResult executeRelAlgQueryNoRetry (const CompilationOptions &co, const ExecutionOptions &eo, const bool just_explain_plan, RenderInfo *render_info)
 
void executeRelAlgStep (const RaExecutionSequence &seq, const size_t step_idx, const CompilationOptions &, const ExecutionOptions &, RenderInfo *, const int64_t queue_time_ms)
 
void executeUpdate (const RelAlgNode *node, const CompilationOptions &co, const ExecutionOptions &eo, const int64_t queue_time_ms)
 
void executeDelete (const RelAlgNode *node, const CompilationOptions &co, const ExecutionOptions &eo_in, const int64_t queue_time_ms)
 
ExecutionResult executeCompound (const RelCompound *, const CompilationOptions &, const ExecutionOptions &, RenderInfo *, const int64_t queue_time_ms)
 
ExecutionResult executeAggregate (const RelAggregate *aggregate, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const int64_t queue_time_ms)
 
ExecutionResult executeProject (const RelProject *, const CompilationOptions &, const ExecutionOptions &, RenderInfo *, const int64_t queue_time_ms, const ssize_t previous_count)
 
ExecutionResult executeTableFunction (const RelTableFunction *, const CompilationOptions &, const ExecutionOptions &, const int64_t queue_time_ms)
 
void computeWindow (const RelAlgExecutionUnit &ra_exe_unit, const CompilationOptions &co, const ExecutionOptions &eo, ColumnCacheMap &column_cache_map, const int64_t queue_time_ms)
 
std::unique_ptr
< WindowFunctionContext
createWindowFunctionContext (const Analyzer::WindowFunction *window_func, const std::shared_ptr< Analyzer::BinOper > &partition_key_cond, const RelAlgExecutionUnit &ra_exe_unit, const std::vector< InputTableInfo > &query_infos, const CompilationOptions &co, ColumnCacheMap &column_cache_map)
 
ExecutionResult executeFilter (const RelFilter *, const CompilationOptions &, const ExecutionOptions &, RenderInfo *, const int64_t queue_time_ms)
 
ExecutionResult executeSort (const RelSort *, const CompilationOptions &, const ExecutionOptions &, RenderInfo *, const int64_t queue_time_ms)
 
ExecutionResult executeLogicalValues (const RelLogicalValues *, const ExecutionOptions &)
 
ExecutionResult executeModify (const RelModify *modify, const ExecutionOptions &eo)
 
WorkUnit createSortInputWorkUnit (const RelSort *, const ExecutionOptions &eo)
 
ExecutionResult executeWorkUnit (const WorkUnit &work_unit, const std::vector< TargetMetaInfo > &targets_meta, const bool is_agg, const CompilationOptions &co_in, const ExecutionOptions &eo, RenderInfo *, const int64_t queue_time_ms, const ssize_t previous_count=-1)
 
size_t getNDVEstimation (const WorkUnit &work_unit, const bool is_agg, const CompilationOptions &co, const ExecutionOptions &eo)
 
ssize_t getFilteredCountAll (const WorkUnit &work_unit, const bool is_agg, const CompilationOptions &co, const ExecutionOptions &eo)
 
FilterSelectivity getFilterSelectivity (const std::vector< std::shared_ptr< Analyzer::Expr >> &filter_expressions, const CompilationOptions &co, const ExecutionOptions &eo)
 
std::vector< PushedDownFilterInfoselectFiltersToBePushedDown (const RelAlgExecutor::WorkUnit &work_unit, const CompilationOptions &co, const ExecutionOptions &eo)
 
bool isRowidLookup (const WorkUnit &work_unit)
 
ExecutionResult handleOutOfMemoryRetry (const RelAlgExecutor::WorkUnit &work_unit, const std::vector< TargetMetaInfo > &targets_meta, const bool is_agg, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const bool was_multifrag_kernel_launch, const int64_t queue_time_ms)
 
WorkUnit createWorkUnit (const RelAlgNode *, const SortInfo &, const ExecutionOptions &eo)
 
WorkUnit createCompoundWorkUnit (const RelCompound *, const SortInfo &, const ExecutionOptions &eo)
 
WorkUnit createAggregateWorkUnit (const RelAggregate *, const SortInfo &, const bool just_explain)
 
WorkUnit createProjectWorkUnit (const RelProject *, const SortInfo &, const ExecutionOptions &eo)
 
WorkUnit createFilterWorkUnit (const RelFilter *, const SortInfo &, const bool just_explain)
 
WorkUnit createJoinWorkUnit (const RelJoin *, const SortInfo &, const bool just_explain)
 
TableFunctionWorkUnit createTableFunctionWorkUnit (const RelTableFunction *table_func, const bool just_explain)
 
void addTemporaryTable (const int table_id, const ResultSetPtr &result)
 
void eraseFromTemporaryTables (const int table_id)
 
void handleNop (RaExecutionDesc &ed)
 
JoinQualsPerNestingLevel translateLeftDeepJoinFilter (const RelLeftDeepInnerJoin *join, const std::vector< InputDescriptor > &input_descs, const std::unordered_map< const RelAlgNode *, int > &input_to_nest_level, const bool just_explain)
 
std::list< std::shared_ptr
< Analyzer::Expr > > 
makeJoinQuals (const RexScalar *join_condition, const std::vector< JoinType > &join_types, const std::unordered_map< const RelAlgNode *, int > &input_to_nest_level, const bool just_explain) const
 
- Private Member Functions inherited from StorageIOFacility< RelAlgExecutorTraits >
 StorageIOFacility (ExecutorType *executor, CatalogType const &catalog)
 
UpdateCallback yieldUpdateCallback (UpdateTransactionParameters &update_parameters)
 
UpdateCallback yieldDeleteCallback (DeleteTransactionParameters &delete_parameters)
 

Static Private Member Functions

static void handlePersistentError (const int32_t error_code)
 

Private Attributes

Executorexecutor_
 
const Catalog_Namespace::Catalogcat_
 
std::unique_ptr< RelAlgDagBuilderquery_dag_
 
std::shared_ptr< const
query_state::QueryState
query_state_
 
TemporaryTables temporary_tables_
 
time_t now_
 
std::vector< std::shared_ptr
< Analyzer::Expr > > 
target_exprs_owned_
 
std::unordered_map< unsigned,
AggregatedResult
leaf_results_
 
int64_t queue_time_ms_
 

Static Private Attributes

static SpeculativeTopNBlacklist speculative_topn_blacklist_
 
static const size_t max_groups_buffer_entry_default_guess {16384}
 

Friends

class PendingExecutionClosure
 

Additional Inherited Members

- Private Types inherited from StorageIOFacility< RelAlgExecutorTraits >
using ExecutorType = typename RelAlgExecutorTraits::ExecutorType
 
using CatalogType = typename RelAlgExecutorTraits::CatalogType
 
using FragmentUpdaterType = UpdateLogForFragment
 
using UpdateCallback = typename FragmentUpdaterType::Callback
 
using TableDescriptorType = typename RelAlgExecutorTraits::TableDescriptorType
 
using DeleteVictimOffsetList = std::vector< uint64_t >
 
using UpdateTargetOffsetList = std::vector< uint64_t >
 
using UpdateTargetTypeList = std::vector< TargetMetaInfo >
 
using UpdateTargetColumnNamesList = std::vector< std::string >
 
using FragmenterType = Fragmenter_Namespace::InsertOrderFragmenter
 
using TransactionLog = typename FragmenterType::ModifyTransactionTracker
 
using TransactionLogPtr = std::unique_ptr< TransactionLog >
 
using ColumnValidationFunction = std::function< bool(std::string const &)>
 

Detailed Description

Definition at line 54 of file RelAlgExecutor.h.

Member Typedef Documentation

Definition at line 56 of file RelAlgExecutor.h.

Constructor & Destructor Documentation

RelAlgExecutor::RelAlgExecutor ( Executor executor,
const Catalog_Namespace::Catalog cat,
std::shared_ptr< const query_state::QueryState query_state = nullptr 
)
inline

Definition at line 58 of file RelAlgExecutor.h.

61  : StorageIOFacility(executor, cat)
62  , executor_(executor)
63  , cat_(cat)
64  , query_state_(std::move(query_state))
65  , now_(0)
66  , queue_time_ms_(0) {}
StorageIOFacility(ExecutorType *executor, CatalogType const &catalog)
int64_t queue_time_ms_
const Catalog_Namespace::Catalog & cat_
std::shared_ptr< const query_state::QueryState > query_state_
Executor * executor_
RelAlgExecutor::RelAlgExecutor ( Executor executor,
const Catalog_Namespace::Catalog cat,
const std::string &  query_ra,
std::shared_ptr< const query_state::QueryState query_state = nullptr 
)
inline

Definition at line 68 of file RelAlgExecutor.h.

72  : StorageIOFacility(executor, cat)
73  , executor_(executor)
74  , cat_(cat)
75  , query_dag_(std::make_unique<RelAlgDagBuilder>(query_ra, cat_, nullptr))
76  , query_state_(std::move(query_state))
77  , now_(0)
78  , queue_time_ms_(0) {}
StorageIOFacility(ExecutorType *executor, CatalogType const &catalog)
int64_t queue_time_ms_
const Catalog_Namespace::Catalog & cat_
std::shared_ptr< const query_state::QueryState > query_state_
std::unique_ptr< RelAlgDagBuilder > query_dag_
Executor * executor_
RelAlgExecutor::RelAlgExecutor ( Executor executor,
const Catalog_Namespace::Catalog cat,
std::unique_ptr< RelAlgDagBuilder query_dag,
std::shared_ptr< const query_state::QueryState query_state = nullptr 
)
inline

Definition at line 80 of file RelAlgExecutor.h.

84  : StorageIOFacility(executor, cat)
85  , executor_(executor)
86  , cat_(cat)
87  , query_dag_(std::move(query_dag))
88  , query_state_(std::move(query_state))
89  , now_(0)
90  , queue_time_ms_(0) {}
StorageIOFacility(ExecutorType *executor, CatalogType const &catalog)
int64_t queue_time_ms_
const Catalog_Namespace::Catalog & cat_
std::shared_ptr< const query_state::QueryState > query_state_
std::unique_ptr< RelAlgDagBuilder > query_dag_
Executor * executor_

Member Function Documentation

void RelAlgExecutor::addLeafResult ( const unsigned  id,
const AggregatedResult result 
)
inline

Definition at line 130 of file RelAlgExecutor.h.

References CHECK(), and leaf_results_.

130  {
131  const auto it_ok = leaf_results_.emplace(id, result);
132  CHECK(it_ok.second);
133  }
std::unordered_map< unsigned, AggregatedResult > leaf_results_
CHECK(cgen_state)

+ Here is the call graph for this function:

void RelAlgExecutor::addTemporaryTable ( const int  table_id,
const ResultSetPtr result 
)
inlineprivate

Definition at line 320 of file RelAlgExecutor.h.

References CHECK(), CHECK_LT, and temporary_tables_.

Referenced by executeRelAlgStep(), and handleNop().

320  {
321  CHECK_LT(size_t(0), result->colCount());
322  CHECK_LT(table_id, 0);
323  const auto it_ok = temporary_tables_.emplace(table_id, result);
324  CHECK(it_ok.second);
325  }
TemporaryTables temporary_tables_
CHECK(cgen_state)
#define CHECK_LT(x, y)
Definition: Logger.h:207

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void RelAlgExecutor::cleanupPostExecution ( )

Definition at line 328 of file RelAlgExecutor.cpp.

References CHECK(), and executor_.

Referenced by executeRelAlgQueryNoRetry(), and getOuterFragmentCount().

328  {
329  CHECK(executor_);
330  executor_->row_set_mem_owner_ = nullptr;
331  executor_->lit_str_dict_proxy_ = nullptr;
332 }
CHECK(cgen_state)
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

AggregatedColRange RelAlgExecutor::computeColRangesCache ( )

Definition at line 308 of file RelAlgExecutor.cpp.

References cat_, executor_, get_physical_inputs(), and getRootRelAlgNode().

308  {
309  AggregatedColRange agg_col_range_cache;
310  const auto phys_inputs = get_physical_inputs(cat_, &getRootRelAlgNode());
311  return executor_->computeColRangesCache(phys_inputs);
312 }
const Catalog_Namespace::Catalog & cat_
const RelAlgNode & getRootRelAlgNode() const
std::unordered_set< PhysicalInput > get_physical_inputs(const RelAlgNode *ra)
Executor * executor_

+ Here is the call graph for this function:

StringDictionaryGenerations RelAlgExecutor::computeStringDictionaryGenerations ( )

Definition at line 314 of file RelAlgExecutor.cpp.

References cat_, executor_, get_physical_inputs(), and getRootRelAlgNode().

314  {
315  const auto phys_inputs = get_physical_inputs(cat_, &getRootRelAlgNode());
316  return executor_->computeStringDictionaryGenerations(phys_inputs);
317 }
const Catalog_Namespace::Catalog & cat_
const RelAlgNode & getRootRelAlgNode() const
std::unordered_set< PhysicalInput > get_physical_inputs(const RelAlgNode *ra)
Executor * executor_

+ Here is the call graph for this function:

TableGenerations RelAlgExecutor::computeTableGenerations ( )

Definition at line 319 of file RelAlgExecutor.cpp.

References executor_, get_physical_table_inputs(), and getRootRelAlgNode().

319  {
320  const auto phys_table_ids = get_physical_table_inputs(&getRootRelAlgNode());
321  return executor_->computeTableGenerations(phys_table_ids);
322 }
const RelAlgNode & getRootRelAlgNode() const
Executor * executor_
std::unordered_set< int > get_physical_table_inputs(const RelAlgNode *ra)

+ Here is the call graph for this function:

void RelAlgExecutor::computeWindow ( const RelAlgExecutionUnit ra_exe_unit,
const CompilationOptions co,
const ExecutionOptions eo,
ColumnCacheMap column_cache_map,
const int64_t  queue_time_ms 
)
private

Definition at line 1568 of file RelAlgExecutor.cpp.

References CHECK_EQ, WindowProjectNodeContext::create(), createWindowFunctionContext(), executor_, ExecutionOptions::executor_type, Extern, get_table_infos(), Analyzer::WindowFunction::getPartitionKeys(), RelAlgExecutionUnit::input_descs, kBOOLEAN, kBW_EQ, kONE, RelAlgExecutionUnit::target_exprs, and anonymous_namespace{RelAlgExecutor.cpp}::transform_to_inner().

Referenced by executeWorkUnit().

1572  {
1573  auto query_infos = get_table_infos(ra_exe_unit.input_descs, executor_);
1574  CHECK_EQ(query_infos.size(), size_t(1));
1575  if (query_infos.front().info.fragments.size() != 1) {
1576  throw std::runtime_error(
1577  "Only single fragment tables supported for window functions for now");
1578  }
1579  if (eo.executor_type == ::ExecutorType::Extern) {
1580  return;
1581  }
1582  query_infos.push_back(query_infos.front());
1583  auto window_project_node_context = WindowProjectNodeContext::create();
1584  for (size_t target_index = 0; target_index < ra_exe_unit.target_exprs.size();
1585  ++target_index) {
1586  const auto& target_expr = ra_exe_unit.target_exprs[target_index];
1587  const auto window_func = dynamic_cast<const Analyzer::WindowFunction*>(target_expr);
1588  if (!window_func) {
1589  continue;
1590  }
1591  // Always use baseline layout hash tables for now, make the expression a tuple.
1592  const auto& partition_keys = window_func->getPartitionKeys();
1593  std::shared_ptr<Analyzer::Expr> partition_key_tuple;
1594  if (partition_keys.size() > 1) {
1595  partition_key_tuple = makeExpr<Analyzer::ExpressionTuple>(partition_keys);
1596  } else {
1597  if (partition_keys.empty()) {
1598  throw std::runtime_error(
1599  "Empty window function partitions are not supported yet");
1600  }
1601  CHECK_EQ(partition_keys.size(), size_t(1));
1602  partition_key_tuple = partition_keys.front();
1603  }
1604  // Creates a tautology equality with the partition expression on both sides.
1605  const auto partition_key_cond =
1606  makeExpr<Analyzer::BinOper>(kBOOLEAN,
1607  kBW_EQ,
1608  kONE,
1609  partition_key_tuple,
1610  transform_to_inner(partition_key_tuple.get()));
1611  auto context = createWindowFunctionContext(
1612  window_func, partition_key_cond, ra_exe_unit, query_infos, co, column_cache_map);
1613  context->compute();
1614  window_project_node_context->addWindowFunctionContext(std::move(context),
1615  target_index);
1616  }
1617 }
std::vector< Analyzer::Expr * > target_exprs
#define CHECK_EQ(x, y)
Definition: Logger.h:205
const std::vector< InputDescriptor > input_descs
std::shared_ptr< Analyzer::Expr > transform_to_inner(const Analyzer::Expr *expr)
ExecutorType executor_type
static WindowProjectNodeContext * create()
Definition: sqldefs.h:69
std::vector< InputTableInfo > get_table_infos(const std::vector< InputDescriptor > &input_descs, Executor *executor)
Definition: sqldefs.h:31
std::unique_ptr< WindowFunctionContext > createWindowFunctionContext(const Analyzer::WindowFunction *window_func, const std::shared_ptr< Analyzer::BinOper > &partition_key_cond, const RelAlgExecutionUnit &ra_exe_unit, const std::vector< InputTableInfo > &query_infos, const CompilationOptions &co, ColumnCacheMap &column_cache_map)
const std::vector< std::shared_ptr< Analyzer::Expr > > & getPartitionKeys() const
Definition: Analyzer.h:1400
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

RelAlgExecutor::WorkUnit RelAlgExecutor::createAggregateWorkUnit ( const RelAggregate aggregate,
const SortInfo sort_info,
const bool  just_explain 
)
private

Definition at line 3301 of file RelAlgExecutor.cpp.

References cat_, CHECK_EQ, executor_, anonymous_namespace{RelAlgExecutor.cpp}::get_input_desc(), anonymous_namespace{RelAlgExecutor.cpp}::get_input_nest_levels(), anonymous_namespace{RelAlgExecutor.cpp}::get_join_type(), anonymous_namespace{RelAlgExecutor.cpp}::get_targets_meta(), RelAlgNode::getInput(), RelAlgNode::getOutputMetainfo(), RelAlgNode::inputCount(), max_groups_buffer_entry_default_guess, now_, query_state_, RelAlgNode::setOutputMetainfo(), anonymous_namespace{RelAlgExecutor.cpp}::synthesize_inputs(), target_exprs_owned_, anonymous_namespace{RelAlgExecutor.cpp}::translate_groupby_exprs(), and anonymous_namespace{RelAlgExecutor.cpp}::translate_targets().

Referenced by createWorkUnit(), and executeAggregate().

3304  {
3305  std::vector<InputDescriptor> input_descs;
3306  std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
3307  std::vector<std::shared_ptr<RexInput>> used_inputs_owned;
3308  const auto input_to_nest_level = get_input_nest_levels(aggregate, {});
3309  std::tie(input_descs, input_col_descs, used_inputs_owned) =
3310  get_input_desc(aggregate, input_to_nest_level, {}, cat_);
3311  const auto join_type = get_join_type(aggregate);
3312  QueryFeatureDescriptor query_features;
3313  RelAlgTranslator translator(cat_,
3314  executor_,
3315  input_to_nest_level,
3316  {join_type},
3317  now_,
3318  just_explain,
3319  query_features);
3320  CHECK_EQ(size_t(1), aggregate->inputCount());
3321  const auto source = aggregate->getInput(0);
3322  const auto& in_metainfo = source->getOutputMetainfo();
3323  const auto scalar_sources =
3324  synthesize_inputs(aggregate, size_t(0), in_metainfo, input_to_nest_level);
3325  const auto groupby_exprs = translate_groupby_exprs(aggregate, scalar_sources);
3326  const auto target_exprs = translate_targets(
3327  target_exprs_owned_, scalar_sources, groupby_exprs, aggregate, translator);
3328  const auto targets_meta = get_targets_meta(aggregate, target_exprs);
3329  aggregate->setOutputMetainfo(targets_meta);
3330  return {RelAlgExecutionUnit{input_descs,
3331  input_col_descs,
3332  {},
3333  {},
3334  {},
3335  groupby_exprs,
3336  target_exprs,
3337  nullptr,
3338  sort_info,
3339  0,
3340  query_features,
3341  false,
3342  query_state_},
3343  aggregate,
3345  nullptr};
3346 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
std::unordered_map< const RelAlgNode *, int > get_input_nest_levels(const RelAlgNode *ra_node, const std::vector< size_t > &input_permutation)
std::vector< std::shared_ptr< Analyzer::Expr > > synthesize_inputs(const RelAlgNode *ra_node, const size_t nest_level, const std::vector< TargetMetaInfo > &in_metainfo, const std::unordered_map< const RelAlgNode *, int > &input_to_nest_level)
static const size_t max_groups_buffer_entry_default_guess
std::vector< Analyzer::Expr * > translate_targets(std::vector< std::shared_ptr< Analyzer::Expr >> &target_exprs_owned, const std::vector< std::shared_ptr< Analyzer::Expr >> &scalar_sources, const std::list< std::shared_ptr< Analyzer::Expr >> &groupby_exprs, const RelCompound *compound, const RelAlgTranslator &translator, const ExecutorType executor_type)
std::vector< std::shared_ptr< Analyzer::Expr > > target_exprs_owned_
std::list< std::shared_ptr< Analyzer::Expr > > translate_groupby_exprs(const RelCompound *compound, const std::vector< std::shared_ptr< Analyzer::Expr >> &scalar_sources)
const Catalog_Namespace::Catalog & cat_
const RelAlgNode * getInput(const size_t idx) const
JoinType get_join_type(const RelAlgNode *ra)
std::vector< TargetMetaInfo > get_targets_meta(const RA *ra_node, const std::vector< Analyzer::Expr * > &target_exprs)
std::shared_ptr< const query_state::QueryState > query_state_
void setOutputMetainfo(const std::vector< TargetMetaInfo > &targets_metainfo) const
std::tuple< std::vector< InputDescriptor >, std::list< std::shared_ptr< const InputColDescriptor > >, std::vector< std::shared_ptr< RexInput > > > get_input_desc(const RA *ra_node, const std::unordered_map< const RelAlgNode *, int > &input_to_nest_level, const std::vector< size_t > &input_permutation, const Catalog_Namespace::Catalog &cat)
const size_t inputCount() const
const std::vector< TargetMetaInfo > & getOutputMetainfo() const
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

RelAlgExecutor::WorkUnit RelAlgExecutor::createCompoundWorkUnit ( const RelCompound compound,
const SortInfo sort_info,
const ExecutionOptions eo 
)
private

Definition at line 3021 of file RelAlgExecutor.cpp.

References cat_, CHECK_EQ, anonymous_namespace{RelAlgExecutor.cpp}::do_table_reordering(), executor_, ExecutionOptions::executor_type, g_from_table_reordering, anonymous_namespace{RelAlgExecutor.cpp}::get_input_desc(), anonymous_namespace{RelAlgExecutor.cpp}::get_input_nest_levels(), anonymous_namespace{RelAlgExecutor.cpp}::get_join_type(), anonymous_namespace{RelAlgExecutor.cpp}::get_left_deep_join_input_sizes(), get_table_infos(), anonymous_namespace{RelAlgExecutor.cpp}::get_targets_meta(), RelAlgNode::getInput(), RelAlgNode::inputCount(), ExecutionOptions::just_explain, LEFT, anonymous_namespace{RelAlgExecutor.cpp}::left_deep_join_types(), max_groups_buffer_entry_default_guess, now_, query_state_, anonymous_namespace{RelAlgExecutor.cpp}::rewrite_quals(), RelAlgNode::setOutputMetainfo(), RelAlgExecutionUnit::simple_quals, RelCompound::size(), target_exprs_owned_, anonymous_namespace{RelAlgExecutor.cpp}::translate_groupby_exprs(), anonymous_namespace{RelAlgExecutor.cpp}::translate_quals(), anonymous_namespace{RelAlgExecutor.cpp}::translate_scalar_sources(), anonymous_namespace{RelAlgExecutor.cpp}::translate_targets(), and translateLeftDeepJoinFilter().

Referenced by createWorkUnit(), executeCompound(), executeDelete(), executeUpdate(), and getOuterFragmentCount().

3024  {
3025  std::vector<InputDescriptor> input_descs;
3026  std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
3027  auto input_to_nest_level = get_input_nest_levels(compound, {});
3028  std::tie(input_descs, input_col_descs, std::ignore) =
3029  get_input_desc(compound, input_to_nest_level, {}, cat_);
3030  const auto query_infos = get_table_infos(input_descs, executor_);
3031  CHECK_EQ(size_t(1), compound->inputCount());
3032  const auto left_deep_join =
3033  dynamic_cast<const RelLeftDeepInnerJoin*>(compound->getInput(0));
3034  JoinQualsPerNestingLevel left_deep_join_quals;
3035  const auto join_types = left_deep_join ? left_deep_join_types(left_deep_join)
3036  : std::vector<JoinType>{get_join_type(compound)};
3037  std::vector<size_t> input_permutation;
3038  std::vector<size_t> left_deep_join_input_sizes;
3039  if (left_deep_join) {
3040  left_deep_join_input_sizes = get_left_deep_join_input_sizes(left_deep_join);
3041  left_deep_join_quals = translateLeftDeepJoinFilter(
3042  left_deep_join, input_descs, input_to_nest_level, eo.just_explain);
3044  std::find(join_types.begin(), join_types.end(), JoinType::LEFT) ==
3045  join_types.end()) {
3046  input_permutation = do_table_reordering(input_descs,
3047  input_col_descs,
3048  left_deep_join_quals,
3049  input_to_nest_level,
3050  compound,
3051  query_infos,
3052  executor_);
3053  input_to_nest_level = get_input_nest_levels(compound, input_permutation);
3054  std::tie(input_descs, input_col_descs, std::ignore) =
3055  get_input_desc(compound, input_to_nest_level, input_permutation, cat_);
3056  left_deep_join_quals = translateLeftDeepJoinFilter(
3057  left_deep_join, input_descs, input_to_nest_level, eo.just_explain);
3058  }
3059  }
3060  QueryFeatureDescriptor query_features;
3061  RelAlgTranslator translator(cat_,
3062  executor_,
3063  input_to_nest_level,
3064  join_types,
3065  now_,
3066  eo.just_explain,
3067  query_features);
3068  const auto scalar_sources =
3069  translate_scalar_sources(compound, translator, eo.executor_type);
3070  const auto groupby_exprs = translate_groupby_exprs(compound, scalar_sources);
3071  const auto quals_cf = translate_quals(compound, translator);
3072  const auto target_exprs = translate_targets(target_exprs_owned_,
3073  scalar_sources,
3074  groupby_exprs,
3075  compound,
3076  translator,
3077  eo.executor_type);
3078  CHECK_EQ(compound->size(), target_exprs.size());
3079  const RelAlgExecutionUnit exe_unit = {input_descs,
3080  input_col_descs,
3081  quals_cf.simple_quals,
3082  rewrite_quals(quals_cf.quals),
3083  left_deep_join_quals,
3084  groupby_exprs,
3085  target_exprs,
3086  nullptr,
3087  sort_info,
3088  0,
3089  query_features,
3090  false,
3091  query_state_};
3092  auto query_rewriter = std::make_unique<QueryRewriter>(query_infos, executor_);
3093  const auto rewritten_exe_unit = query_rewriter->rewrite(exe_unit);
3094  const auto targets_meta = get_targets_meta(compound, rewritten_exe_unit.target_exprs);
3095  compound->setOutputMetainfo(targets_meta);
3096  return {rewritten_exe_unit,
3097  compound,
3099  std::move(query_rewriter),
3100  input_permutation,
3101  left_deep_join_input_sizes};
3102 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
std::unordered_map< const RelAlgNode *, int > get_input_nest_levels(const RelAlgNode *ra_node, const std::vector< size_t > &input_permutation)
std::vector< JoinType > left_deep_join_types(const RelLeftDeepInnerJoin *left_deep_join)
JoinType
Definition: sqldefs.h:107
std::vector< size_t > do_table_reordering(std::vector< InputDescriptor > &input_descs, std::list< std::shared_ptr< const InputColDescriptor >> &input_col_descs, const JoinQualsPerNestingLevel &left_deep_join_quals, std::unordered_map< const RelAlgNode *, int > &input_to_nest_level, const RA *node, const std::vector< InputTableInfo > &query_infos, const Executor *executor)
size_t size() const override
static const size_t max_groups_buffer_entry_default_guess
std::vector< Analyzer::Expr * > translate_targets(std::vector< std::shared_ptr< Analyzer::Expr >> &target_exprs_owned, const std::vector< std::shared_ptr< Analyzer::Expr >> &scalar_sources, const std::list< std::shared_ptr< Analyzer::Expr >> &groupby_exprs, const RelCompound *compound, const RelAlgTranslator &translator, const ExecutorType executor_type)
std::vector< JoinCondition > JoinQualsPerNestingLevel
std::vector< std::shared_ptr< Analyzer::Expr > > target_exprs_owned_
bool g_from_table_reordering
Definition: Execute.cpp:78
ExecutorType executor_type
std::list< std::shared_ptr< Analyzer::Expr > > translate_groupby_exprs(const RelCompound *compound, const std::vector< std::shared_ptr< Analyzer::Expr >> &scalar_sources)
const Catalog_Namespace::Catalog & cat_
const RelAlgNode * getInput(const size_t idx) const
JoinType get_join_type(const RelAlgNode *ra)
std::vector< TargetMetaInfo > get_targets_meta(const RA *ra_node, const std::vector< Analyzer::Expr * > &target_exprs)
std::shared_ptr< const query_state::QueryState > query_state_
std::vector< std::shared_ptr< Analyzer::Expr > > translate_scalar_sources(const RA *ra_node, const RelAlgTranslator &translator, const ::ExecutorType executor_type)
std::vector< InputTableInfo > get_table_infos(const std::vector< InputDescriptor > &input_descs, Executor *executor)
void setOutputMetainfo(const std::vector< TargetMetaInfo > &targets_metainfo) const
QualsConjunctiveForm translate_quals(const RelCompound *compound, const RelAlgTranslator &translator)
std::tuple< std::vector< InputDescriptor >, std::list< std::shared_ptr< const InputColDescriptor > >, std::vector< std::shared_ptr< RexInput > > > get_input_desc(const RA *ra_node, const std::unordered_map< const RelAlgNode *, int > &input_to_nest_level, const std::vector< size_t > &input_permutation, const Catalog_Namespace::Catalog &cat)
const size_t inputCount() const
Executor * executor_
std::list< std::shared_ptr< Analyzer::Expr > > simple_quals
JoinQualsPerNestingLevel translateLeftDeepJoinFilter(const RelLeftDeepInnerJoin *join, const std::vector< InputDescriptor > &input_descs, const std::unordered_map< const RelAlgNode *, int > &input_to_nest_level, const bool just_explain)
std::vector< size_t > get_left_deep_join_input_sizes(const RelLeftDeepInnerJoin *left_deep_join)
std::list< std::shared_ptr< Analyzer::Expr > > rewrite_quals(const std::list< std::shared_ptr< Analyzer::Expr >> &quals)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

RelAlgExecutor::WorkUnit RelAlgExecutor::createFilterWorkUnit ( const RelFilter filter,
const SortInfo sort_info,
const bool  just_explain 
)
private

Definition at line 3539 of file RelAlgExecutor.cpp.

References cat_, CHECK_EQ, executor_, fold_expr(), get_exprs_not_owned(), anonymous_namespace{RelAlgExecutor.cpp}::get_input_desc(), anonymous_namespace{RelAlgExecutor.cpp}::get_input_nest_levels(), anonymous_namespace{RelAlgExecutor.cpp}::get_inputs_meta(), anonymous_namespace{RelAlgExecutor.cpp}::get_join_type(), RelFilter::getCondition(), RelAlgNode::inputCount(), max_groups_buffer_entry_default_guess, now_, rewrite_expr(), RelAlgNode::setOutputMetainfo(), and target_exprs_owned_.

Referenced by createWorkUnit(), and executeFilter().

3541  {
3542  CHECK_EQ(size_t(1), filter->inputCount());
3543  std::vector<InputDescriptor> input_descs;
3544  std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
3545  std::vector<TargetMetaInfo> in_metainfo;
3546  std::vector<std::shared_ptr<RexInput>> used_inputs_owned;
3547  std::vector<std::shared_ptr<Analyzer::Expr>> target_exprs_owned;
3548 
3549  const auto input_to_nest_level = get_input_nest_levels(filter, {});
3550  std::tie(input_descs, input_col_descs, used_inputs_owned) =
3551  get_input_desc(filter, input_to_nest_level, {}, cat_);
3552  const auto join_type = get_join_type(filter);
3553  QueryFeatureDescriptor query_features;
3554  RelAlgTranslator translator(cat_,
3555  executor_,
3556  input_to_nest_level,
3557  {join_type},
3558  now_,
3559  just_explain,
3560  query_features);
3561  std::tie(in_metainfo, target_exprs_owned) =
3562  get_inputs_meta(filter, translator, used_inputs_owned, input_to_nest_level);
3563  const auto filter_expr = translator.translateScalarRex(filter->getCondition());
3564  const auto qual = fold_expr(filter_expr.get());
3565  target_exprs_owned_.insert(
3566  target_exprs_owned_.end(), target_exprs_owned.begin(), target_exprs_owned.end());
3567  const auto target_exprs = get_exprs_not_owned(target_exprs_owned);
3568  filter->setOutputMetainfo(in_metainfo);
3569  const auto rewritten_qual = rewrite_expr(qual.get());
3570  return {{input_descs,
3571  input_col_descs,
3572  {},
3573  {rewritten_qual ? rewritten_qual : qual},
3574  {},
3575  {nullptr},
3576  target_exprs,
3577  nullptr,
3578  sort_info,
3579  0},
3580  filter,
3582  nullptr};
3583 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
std::unordered_map< const RelAlgNode *, int > get_input_nest_levels(const RelAlgNode *ra_node, const std::vector< size_t > &input_permutation)
std::pair< std::vector< TargetMetaInfo >, std::vector< std::shared_ptr< Analyzer::Expr > > > get_inputs_meta(const RelFilter *filter, const RelAlgTranslator &translator, const std::vector< std::shared_ptr< RexInput >> &inputs_owned, const std::unordered_map< const RelAlgNode *, int > &input_to_nest_level)
static const size_t max_groups_buffer_entry_default_guess
const RexScalar * getCondition() const
std::vector< Analyzer::Expr * > get_exprs_not_owned(const std::vector< std::shared_ptr< Analyzer::Expr >> &exprs)
Definition: Execute.h:214
Analyzer::ExpressionPtr rewrite_expr(const Analyzer::Expr *expr)
std::vector< std::shared_ptr< Analyzer::Expr > > target_exprs_owned_
const Catalog_Namespace::Catalog & cat_
JoinType get_join_type(const RelAlgNode *ra)
void setOutputMetainfo(const std::vector< TargetMetaInfo > &targets_metainfo) const
std::tuple< std::vector< InputDescriptor >, std::list< std::shared_ptr< const InputColDescriptor > >, std::vector< std::shared_ptr< RexInput > > > get_input_desc(const RA *ra_node, const std::unordered_map< const RelAlgNode *, int > &input_to_nest_level, const std::vector< size_t > &input_permutation, const Catalog_Namespace::Catalog &cat)
const size_t inputCount() const
Executor * executor_
std::shared_ptr< Analyzer::Expr > fold_expr(const Analyzer::Expr *expr)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

WorkUnit RelAlgExecutor::createJoinWorkUnit ( const RelJoin ,
const SortInfo ,
const bool  just_explain 
)
private
RelAlgExecutor::WorkUnit RelAlgExecutor::createProjectWorkUnit ( const RelProject project,
const SortInfo sort_info,
const ExecutionOptions eo 
)
private

Definition at line 3348 of file RelAlgExecutor.cpp.

References cat_, anonymous_namespace{RelAlgExecutor.cpp}::do_table_reordering(), executor_, ExecutionOptions::executor_type, g_from_table_reordering, get_exprs_not_owned(), anonymous_namespace{RelAlgExecutor.cpp}::get_input_desc(), anonymous_namespace{RelAlgExecutor.cpp}::get_input_nest_levels(), anonymous_namespace{RelAlgExecutor.cpp}::get_join_type(), anonymous_namespace{RelAlgExecutor.cpp}::get_left_deep_join_input_sizes(), get_table_infos(), anonymous_namespace{RelAlgExecutor.cpp}::get_targets_meta(), RelAlgNode::getInput(), ExecutionOptions::just_explain, anonymous_namespace{RelAlgExecutor.cpp}::left_deep_join_types(), max_groups_buffer_entry_default_guess, now_, query_state_, RelAlgNode::setOutputMetainfo(), target_exprs_owned_, anonymous_namespace{RelAlgExecutor.cpp}::translate_scalar_sources(), and translateLeftDeepJoinFilter().

Referenced by createWorkUnit(), executeDelete(), executeProject(), executeUpdate(), and getOuterFragmentCount().

3351  {
3352  std::vector<InputDescriptor> input_descs;
3353  std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
3354  auto input_to_nest_level = get_input_nest_levels(project, {});
3355  std::tie(input_descs, input_col_descs, std::ignore) =
3356  get_input_desc(project, input_to_nest_level, {}, cat_);
3357  const auto query_infos = get_table_infos(input_descs, executor_);
3358 
3359  const auto left_deep_join =
3360  dynamic_cast<const RelLeftDeepInnerJoin*>(project->getInput(0));
3361  JoinQualsPerNestingLevel left_deep_join_quals;
3362  const auto join_types = left_deep_join ? left_deep_join_types(left_deep_join)
3363  : std::vector<JoinType>{get_join_type(project)};
3364  std::vector<size_t> input_permutation;
3365  std::vector<size_t> left_deep_join_input_sizes;
3366  if (left_deep_join) {
3367  left_deep_join_input_sizes = get_left_deep_join_input_sizes(left_deep_join);
3368  const auto query_infos = get_table_infos(input_descs, executor_);
3369  left_deep_join_quals = translateLeftDeepJoinFilter(
3370  left_deep_join, input_descs, input_to_nest_level, eo.just_explain);
3372  input_permutation = do_table_reordering(input_descs,
3373  input_col_descs,
3374  left_deep_join_quals,
3375  input_to_nest_level,
3376  project,
3377  query_infos,
3378  executor_);
3379  input_to_nest_level = get_input_nest_levels(project, input_permutation);
3380  std::tie(input_descs, input_col_descs, std::ignore) =
3381  get_input_desc(project, input_to_nest_level, input_permutation, cat_);
3382  left_deep_join_quals = translateLeftDeepJoinFilter(
3383  left_deep_join, input_descs, input_to_nest_level, eo.just_explain);
3384  }
3385  }
3386 
3387  QueryFeatureDescriptor query_features;
3388  RelAlgTranslator translator(cat_,
3389  executor_,
3390  input_to_nest_level,
3391  join_types,
3392  now_,
3393  eo.just_explain,
3394  query_features);
3395  const auto target_exprs_owned =
3396  translate_scalar_sources(project, translator, eo.executor_type);
3397  target_exprs_owned_.insert(
3398  target_exprs_owned_.end(), target_exprs_owned.begin(), target_exprs_owned.end());
3399  const auto target_exprs = get_exprs_not_owned(target_exprs_owned);
3400  const RelAlgExecutionUnit exe_unit = {input_descs,
3401  input_col_descs,
3402  {},
3403  {},
3404  left_deep_join_quals,
3405  {nullptr},
3406  target_exprs,
3407  nullptr,
3408  sort_info,
3409  0,
3410  query_features,
3411  false,
3412  query_state_};
3413  auto query_rewriter = std::make_unique<QueryRewriter>(query_infos, executor_);
3414  const auto rewritten_exe_unit = query_rewriter->rewrite(exe_unit);
3415  const auto targets_meta = get_targets_meta(project, rewritten_exe_unit.target_exprs);
3416  project->setOutputMetainfo(targets_meta);
3417  return {rewritten_exe_unit,
3418  project,
3420  std::move(query_rewriter),
3421  input_permutation,
3422  left_deep_join_input_sizes};
3423 }
std::unordered_map< const RelAlgNode *, int > get_input_nest_levels(const RelAlgNode *ra_node, const std::vector< size_t > &input_permutation)
std::vector< JoinType > left_deep_join_types(const RelLeftDeepInnerJoin *left_deep_join)
JoinType
Definition: sqldefs.h:107
std::vector< size_t > do_table_reordering(std::vector< InputDescriptor > &input_descs, std::list< std::shared_ptr< const InputColDescriptor >> &input_col_descs, const JoinQualsPerNestingLevel &left_deep_join_quals, std::unordered_map< const RelAlgNode *, int > &input_to_nest_level, const RA *node, const std::vector< InputTableInfo > &query_infos, const Executor *executor)
static const size_t max_groups_buffer_entry_default_guess
std::vector< JoinCondition > JoinQualsPerNestingLevel
std::vector< Analyzer::Expr * > get_exprs_not_owned(const std::vector< std::shared_ptr< Analyzer::Expr >> &exprs)
Definition: Execute.h:214
std::vector< std::shared_ptr< Analyzer::Expr > > target_exprs_owned_
bool g_from_table_reordering
Definition: Execute.cpp:78
ExecutorType executor_type
const Catalog_Namespace::Catalog & cat_
const RelAlgNode * getInput(const size_t idx) const
JoinType get_join_type(const RelAlgNode *ra)
std::vector< TargetMetaInfo > get_targets_meta(const RA *ra_node, const std::vector< Analyzer::Expr * > &target_exprs)
std::shared_ptr< const query_state::QueryState > query_state_
std::vector< std::shared_ptr< Analyzer::Expr > > translate_scalar_sources(const RA *ra_node, const RelAlgTranslator &translator, const ::ExecutorType executor_type)
std::vector< InputTableInfo > get_table_infos(const std::vector< InputDescriptor > &input_descs, Executor *executor)
void setOutputMetainfo(const std::vector< TargetMetaInfo > &targets_metainfo) const
std::tuple< std::vector< InputDescriptor >, std::list< std::shared_ptr< const InputColDescriptor > >, std::vector< std::shared_ptr< RexInput > > > get_input_desc(const RA *ra_node, const std::unordered_map< const RelAlgNode *, int > &input_to_nest_level, const std::vector< size_t > &input_permutation, const Catalog_Namespace::Catalog &cat)
Executor * executor_
JoinQualsPerNestingLevel translateLeftDeepJoinFilter(const RelLeftDeepInnerJoin *join, const std::vector< InputDescriptor > &input_descs, const std::unordered_map< const RelAlgNode *, int > &input_to_nest_level, const bool just_explain)
std::vector< size_t > get_left_deep_join_input_sizes(const RelLeftDeepInnerJoin *left_deep_join)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

RelAlgExecutor::WorkUnit RelAlgExecutor::createSortInputWorkUnit ( const RelSort sort,
const ExecutionOptions eo 
)
private

Definition at line 2259 of file RelAlgExecutor.cpp.

References RelSort::collationCount(), SpeculativeTopNBlacklist::contains(), createWorkUnit(), Default, anonymous_namespace{RelAlgExecutor.cpp}::first_oe_is_desc(), anonymous_namespace{RelAlgExecutor.cpp}::get_order_entries(), anonymous_namespace{RelAlgExecutor.cpp}::get_scan_limit(), RelAlgNode::getInput(), RelSort::getLimit(), RelSort::getOffset(), RelAlgExecutionUnit::input_descs, max_groups_buffer_entry_default_guess, RelAlgNode::setOutputMetainfo(), speculative_topn_blacklist_, SpeculativeTopN, and StreamingTopN.

Referenced by executeRelAlgQuerySingleStep(), and executeSort().

2261  {
2262  const auto source = sort->getInput(0);
2263  const size_t limit = sort->getLimit();
2264  const size_t offset = sort->getOffset();
2265  const size_t scan_limit = sort->collationCount() ? 0 : get_scan_limit(source, limit);
2266  const size_t scan_total_limit =
2267  scan_limit ? get_scan_limit(source, scan_limit + offset) : 0;
2268  size_t max_groups_buffer_entry_guess{
2269  scan_total_limit ? scan_total_limit : max_groups_buffer_entry_default_guess};
2271  const auto order_entries = get_order_entries(sort);
2272  SortInfo sort_info{order_entries, sort_algorithm, limit, offset};
2273  auto source_work_unit = createWorkUnit(source, sort_info, eo);
2274  const auto& source_exe_unit = source_work_unit.exe_unit;
2275  if (source_exe_unit.groupby_exprs.size() == 1) {
2276  if (!source_exe_unit.groupby_exprs.front()) {
2277  sort_algorithm = SortAlgorithm::StreamingTopN;
2278  } else {
2279  if (speculative_topn_blacklist_.contains(source_exe_unit.groupby_exprs.front(),
2280  first_oe_is_desc(order_entries))) {
2281  sort_algorithm = SortAlgorithm::Default;
2282  }
2283  }
2284  }
2285 
2286  sort->setOutputMetainfo(source->getOutputMetainfo());
2287  // NB: the `body` field of the returned `WorkUnit` needs to be the `source` node,
2288  // not the `sort`. The aggregator needs the pre-sorted result from leaves.
2289  return {RelAlgExecutionUnit{source_exe_unit.input_descs,
2290  std::move(source_exe_unit.input_col_descs),
2291  source_exe_unit.simple_quals,
2292  source_exe_unit.quals,
2293  source_exe_unit.join_quals,
2294  source_exe_unit.groupby_exprs,
2295  source_exe_unit.target_exprs,
2296  nullptr,
2297  {sort_info.order_entries, sort_algorithm, limit, offset},
2298  scan_total_limit,
2299  source_exe_unit.query_features,
2300  source_exe_unit.use_bump_allocator,
2301  source_exe_unit.query_state},
2302  source,
2303  max_groups_buffer_entry_guess,
2304  std::move(source_work_unit.query_rewriter),
2305  source_work_unit.input_permutation,
2306  source_work_unit.left_deep_join_input_sizes};
2307 }
size_t getOffset() const
bool contains(const std::shared_ptr< Analyzer::Expr > expr, const bool desc) const
size_t get_scan_limit(const RelAlgNode *ra, const size_t limit)
static SpeculativeTopNBlacklist speculative_topn_blacklist_
static const size_t max_groups_buffer_entry_default_guess
const std::vector< InputDescriptor > input_descs
WorkUnit createWorkUnit(const RelAlgNode *, const SortInfo &, const ExecutionOptions &eo)
SortAlgorithm
const RelAlgNode * getInput(const size_t idx) const
size_t collationCount() const
size_t getLimit() const
void setOutputMetainfo(const std::vector< TargetMetaInfo > &targets_metainfo) const
std::list< Analyzer::OrderEntry > get_order_entries(const RelSort *sort)
bool first_oe_is_desc(const std::list< Analyzer::OrderEntry > &order_entries)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

RelAlgExecutor::TableFunctionWorkUnit RelAlgExecutor::createTableFunctionWorkUnit ( const RelTableFunction table_func,
const bool  just_explain 
)
private

Definition at line 3425 of file RelAlgExecutor.cpp.

References cat_, CHECK_EQ, CHECK_GT, executor_, table_functions::TableFunctionsFactory::get(), get_exprs_not_owned(), anonymous_namespace{RelAlgExecutor.cpp}::get_input_desc(), anonymous_namespace{RelAlgExecutor.cpp}::get_input_nest_levels(), get_table_infos(), anonymous_namespace{RelAlgExecutor.cpp}::get_targets_meta(), RelTableFunction::getColInputsSize(), RelTableFunction::getFunctionName(), table_functions::TableFunction::getOutputSQLType(), RelTableFunction::getTableFuncInputAt(), RelAlgNode::inputCount(), Native, now_, RelAlgNode::setOutputMetainfo(), TableFunctionExecutionUnit::target_exprs, target_exprs_owned_, to_string(), and anonymous_namespace{RelAlgExecutor.cpp}::translate_scalar_sources().

Referenced by executeTableFunction().

3427  {
3428  std::vector<InputDescriptor> input_descs;
3429  std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
3430  auto input_to_nest_level = get_input_nest_levels(table_func, {});
3431  std::tie(input_descs, input_col_descs, std::ignore) =
3432  get_input_desc(table_func, input_to_nest_level, {}, cat_);
3433  const auto query_infos = get_table_infos(input_descs, executor_);
3434  CHECK_EQ(size_t(1), table_func->inputCount());
3435 
3436  QueryFeatureDescriptor query_features; // TODO(adb): remove/make optional
3437  RelAlgTranslator translator(
3438  cat_, executor_, input_to_nest_level, {}, now_, just_explain, query_features);
3439  const auto input_exprs_owned =
3440  translate_scalar_sources(table_func, translator, ::ExecutorType::Native);
3441  target_exprs_owned_.insert(
3442  target_exprs_owned_.end(), input_exprs_owned.begin(), input_exprs_owned.end());
3443  const auto input_exprs = get_exprs_not_owned(input_exprs_owned);
3444 
3445  std::vector<Analyzer::ColumnVar*> input_col_exprs;
3446  for (auto input_expr : input_exprs) {
3447  if (auto col_var = dynamic_cast<Analyzer::ColumnVar*>(input_expr)) {
3448  input_col_exprs.push_back(col_var);
3449  }
3450  }
3451  CHECK_EQ(input_col_exprs.size(), table_func->getColInputsSize());
3452 
3453  const auto& table_function_impl =
3455 
3456  std::vector<Analyzer::Expr*> table_func_outputs;
3457  for (size_t i = 0; i < table_function_impl.getOutputsSize(); i++) {
3458  const auto ti = table_function_impl.getOutputSQLType(i);
3459  target_exprs_owned_.push_back(std::make_shared<Analyzer::ColumnVar>(ti, 0, i, -1));
3460  table_func_outputs.push_back(target_exprs_owned_.back().get());
3461  }
3462 
3463  std::optional<size_t> output_row_multiplier;
3464  if (table_function_impl.hasUserSpecifiedOutputMultiplier()) {
3465  const auto parameter_index = table_function_impl.getOutputRowParameter();
3466  CHECK_GT(parameter_index, size_t(0));
3467  const auto parameter_expr = table_func->getTableFuncInputAt(parameter_index - 1);
3468  const auto parameter_expr_literal = dynamic_cast<const RexLiteral*>(parameter_expr);
3469  if (!parameter_expr_literal) {
3470  throw std::runtime_error(
3471  "Provided output buffer multiplier parameter is not a literal. Only literal "
3472  "values are supported with output buffer multiplier configured table "
3473  "functions.");
3474  }
3475  int64_t literal_val = parameter_expr_literal->getVal<int64_t>();
3476  if (literal_val < 0) {
3477  throw std::runtime_error("Provided output row multiplier " +
3478  std::to_string(literal_val) +
3479  " is not valid for table functions.");
3480  }
3481  output_row_multiplier = static_cast<size_t>(literal_val);
3482  }
3483 
3484  const TableFunctionExecutionUnit exe_unit = {
3485  input_descs,
3486  input_col_descs,
3487  input_exprs, // table function inputs
3488  input_col_exprs, // table function column inputs (duplicates w/ above)
3489  table_func_outputs, // table function projected exprs
3490  output_row_multiplier, // output buffer multiplier
3491  table_func->getFunctionName()};
3492  const auto targets_meta = get_targets_meta(table_func, exe_unit.target_exprs);
3493  table_func->setOutputMetainfo(targets_meta);
3494  return {exe_unit, table_func};
3495 }
SQLTypeInfo getOutputSQLType(const size_t idx) const
#define CHECK_EQ(x, y)
Definition: Logger.h:205
std::unordered_map< const RelAlgNode *, int > get_input_nest_levels(const RelAlgNode *ra_node, const std::vector< size_t > &input_permutation)
static const TableFunction & get(const std::string &name)
std::vector< Analyzer::Expr * > get_exprs_not_owned(const std::vector< std::shared_ptr< Analyzer::Expr >> &exprs)
Definition: Execute.h:214
#define CHECK_GT(x, y)
Definition: Logger.h:209
std::string to_string(char const *&&v)
size_t getColInputsSize() const
std::vector< std::shared_ptr< Analyzer::Expr > > target_exprs_owned_
const Catalog_Namespace::Catalog & cat_
std::vector< TargetMetaInfo > get_targets_meta(const RA *ra_node, const std::vector< Analyzer::Expr * > &target_exprs)
std::vector< std::shared_ptr< Analyzer::Expr > > translate_scalar_sources(const RA *ra_node, const RelAlgTranslator &translator, const ::ExecutorType executor_type)
const RexScalar * getTableFuncInputAt(const size_t idx) const
std::string getFunctionName() const
std::vector< InputTableInfo > get_table_infos(const std::vector< InputDescriptor > &input_descs, Executor *executor)
void setOutputMetainfo(const std::vector< TargetMetaInfo > &targets_metainfo) const
std::vector< Analyzer::Expr * > target_exprs
std::tuple< std::vector< InputDescriptor >, std::list< std::shared_ptr< const InputColDescriptor > >, std::vector< std::shared_ptr< RexInput > > > get_input_desc(const RA *ra_node, const std::unordered_map< const RelAlgNode *, int > &input_to_nest_level, const std::vector< size_t > &input_permutation, const Catalog_Namespace::Catalog &cat)
const size_t inputCount() const
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

std::unique_ptr< WindowFunctionContext > RelAlgExecutor::createWindowFunctionContext ( const Analyzer::WindowFunction window_func,
const std::shared_ptr< Analyzer::BinOper > &  partition_key_cond,
const RelAlgExecutionUnit ra_exe_unit,
const std::vector< InputTableInfo > &  query_infos,
const CompilationOptions co,
ColumnCacheMap column_cache_map 
)
private

Definition at line 1619 of file RelAlgExecutor.cpp.

References CHECK(), CHECK_EQ, Data_Namespace::CPU_LEVEL, CompilationOptions::device_type, executor_, ColumnFetcher::getOneColumnFragment(), Analyzer::WindowFunction::getOrderKeys(), GPU, Data_Namespace::GPU_LEVEL, and JoinHashTableInterface::OneToMany.

Referenced by computeWindow().

1625  {
1626  const auto memory_level = co.device_type == ExecutorDeviceType::GPU
1629  const auto join_table_or_err =
1630  executor_->buildHashTableForQualifier(partition_key_cond,
1631  query_infos,
1632  memory_level,
1634  column_cache_map);
1635  if (!join_table_or_err.fail_reason.empty()) {
1636  throw std::runtime_error(join_table_or_err.fail_reason);
1637  }
1638  CHECK(join_table_or_err.hash_table->getHashType() ==
1640  const auto& order_keys = window_func->getOrderKeys();
1641  std::vector<std::shared_ptr<Chunk_NS::Chunk>> chunks_owner;
1642  const size_t elem_count = query_infos.front().info.fragments.front().getNumTuples();
1643  auto context = std::make_unique<WindowFunctionContext>(
1644  window_func, join_table_or_err.hash_table, elem_count, co.device_type);
1645  for (const auto& order_key : order_keys) {
1646  const auto order_col =
1647  std::dynamic_pointer_cast<const Analyzer::ColumnVar>(order_key);
1648  if (!order_col) {
1649  throw std::runtime_error("Only order by columns supported for now");
1650  }
1651  const int8_t* column;
1652  size_t join_col_elem_count;
1653  std::tie(column, join_col_elem_count) =
1655  *order_col,
1656  query_infos.front().info.fragments.front(),
1657  memory_level,
1658  0,
1659  chunks_owner,
1660  column_cache_map);
1661  CHECK_EQ(join_col_elem_count, elem_count);
1662  context->addOrderColumn(column, order_col.get(), chunks_owner);
1663  }
1664  return context;
1665 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
const std::vector< std::shared_ptr< Analyzer::Expr > > & getOrderKeys() const
Definition: Analyzer.h:1404
CHECK(cgen_state)
static std::pair< const int8_t *, size_t > getOneColumnFragment(Executor *executor, const Analyzer::ColumnVar &hash_col, const Fragmenter_Namespace::FragmentInfo &fragment, const Data_Namespace::MemoryLevel effective_mem_lvl, const int device_id, std::vector< std::shared_ptr< Chunk_NS::Chunk >> &chunks_owner, ColumnCacheMap &column_cache)
Gets one chunk&#39;s pointer and element count on either CPU or GPU.
ExecutorDeviceType device_type
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

RelAlgExecutor::WorkUnit RelAlgExecutor::createWorkUnit ( const RelAlgNode node,
const SortInfo sort_info,
const ExecutionOptions eo 
)
private

Definition at line 2852 of file RelAlgExecutor.cpp.

References CHECK(), createAggregateWorkUnit(), createCompoundWorkUnit(), createFilterWorkUnit(), createProjectWorkUnit(), and ExecutionOptions::just_explain.

Referenced by createSortInputWorkUnit().

2854  {
2855  const auto compound = dynamic_cast<const RelCompound*>(node);
2856  if (compound) {
2857  return createCompoundWorkUnit(compound, sort_info, eo);
2858  }
2859  const auto project = dynamic_cast<const RelProject*>(node);
2860  if (project) {
2861  return createProjectWorkUnit(project, sort_info, eo);
2862  }
2863  const auto aggregate = dynamic_cast<const RelAggregate*>(node);
2864  if (aggregate) {
2865  return createAggregateWorkUnit(aggregate, sort_info, eo.just_explain);
2866  }
2867  const auto filter = dynamic_cast<const RelFilter*>(node);
2868  CHECK(filter);
2869  return createFilterWorkUnit(filter, sort_info, eo.just_explain);
2870 }
WorkUnit createProjectWorkUnit(const RelProject *, const SortInfo &, const ExecutionOptions &eo)
WorkUnit createAggregateWorkUnit(const RelAggregate *, const SortInfo &, const bool just_explain)
WorkUnit createCompoundWorkUnit(const RelCompound *, const SortInfo &, const ExecutionOptions &eo)
CHECK(cgen_state)
WorkUnit createFilterWorkUnit(const RelFilter *, const SortInfo &, const bool just_explain)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void RelAlgExecutor::eraseFromTemporaryTables ( const int  table_id)
inlineprivate

Definition at line 327 of file RelAlgExecutor.h.

References temporary_tables_.

327 { temporary_tables_.erase(table_id); }
TemporaryTables temporary_tables_
ExecutionResult RelAlgExecutor::executeAggregate ( const RelAggregate aggregate,
const CompilationOptions co,
const ExecutionOptions eo,
RenderInfo render_info,
const int64_t  queue_time_ms 
)
private

Definition at line 1439 of file RelAlgExecutor.cpp.

References createAggregateWorkUnit(), DEBUG_TIMER, Default, executeWorkUnit(), RelAlgNode::getOutputMetainfo(), and ExecutionOptions::just_explain.

Referenced by executeRelAlgStep().

1443  {
1444  auto timer = DEBUG_TIMER(__func__);
1445  const auto work_unit = createAggregateWorkUnit(
1446  aggregate, {{}, SortAlgorithm::Default, 0, 0}, eo.just_explain);
1447  return executeWorkUnit(work_unit,
1448  aggregate->getOutputMetainfo(),
1449  true,
1450  co,
1451  eo,
1452  render_info,
1453  queue_time_ms);
1454 }
WorkUnit createAggregateWorkUnit(const RelAggregate *, const SortInfo &, const bool just_explain)
ExecutionResult executeWorkUnit(const WorkUnit &work_unit, const std::vector< TargetMetaInfo > &targets_meta, const bool is_agg, const CompilationOptions &co_in, const ExecutionOptions &eo, RenderInfo *, const int64_t queue_time_ms, const ssize_t previous_count=-1)
#define DEBUG_TIMER(name)
Definition: Logger.h:313
const std::vector< TargetMetaInfo > & getOutputMetainfo() const

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeCompound ( const RelCompound compound,
const CompilationOptions co,
const ExecutionOptions eo,
RenderInfo render_info,
const int64_t  queue_time_ms 
)
private

Definition at line 1421 of file RelAlgExecutor.cpp.

References createCompoundWorkUnit(), DEBUG_TIMER, Default, executeWorkUnit(), RelAlgNode::getOutputMetainfo(), and RelCompound::isAggregate().

Referenced by executeRelAlgStep().

1425  {
1426  auto timer = DEBUG_TIMER(__func__);
1427  const auto work_unit =
1428  createCompoundWorkUnit(compound, {{}, SortAlgorithm::Default, 0, 0}, eo);
1429  CompilationOptions co_compound = co;
1430  return executeWorkUnit(work_unit,
1431  compound->getOutputMetainfo(),
1432  compound->isAggregate(),
1433  co_compound,
1434  eo,
1435  render_info,
1436  queue_time_ms);
1437 }
WorkUnit createCompoundWorkUnit(const RelCompound *, const SortInfo &, const ExecutionOptions &eo)
bool isAggregate() const
ExecutionResult executeWorkUnit(const WorkUnit &work_unit, const std::vector< TargetMetaInfo > &targets_meta, const bool is_agg, const CompilationOptions &co_in, const ExecutionOptions &eo, RenderInfo *, const int64_t queue_time_ms, const ssize_t previous_count=-1)
#define DEBUG_TIMER(name)
Definition: Logger.h:313
const std::vector< TargetMetaInfo > & getOutputMetainfo() const

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void RelAlgExecutor::executeDelete ( const RelAlgNode node,
const CompilationOptions co,
const ExecutionOptions eo_in,
const int64_t  queue_time_ms 
)
private

Definition at line 1336 of file RelAlgExecutor.cpp.

References CompilationOptions::add_delete_column, cat_, CHECK(), CHECK_EQ, createCompoundWorkUnit(), createProjectWorkUnit(), DEBUG_TIMER, Default, executor_, get_table_infos(), get_temporary_table(), Catalog_Namespace::Catalog::getDeletedColumn(), CacheInvalidator< CACHE_HOLDING_TYPES >::invalidateCaches(), CompilationOptions::makeCpuOnly(), ExecutionOptions::output_columnar_hint, table_is_temporary(), temporary_tables_, and StorageIOFacility< RelAlgExecutorTraits >::yieldDeleteCallback().

Referenced by executeRelAlgStep().

1339  {
1340  CHECK(node);
1341  auto timer = DEBUG_TIMER(__func__);
1342 
1344 
1345  auto execute_delete_for_node = [this, &co, &eo_in](const auto node,
1346  auto& work_unit,
1347  const bool is_aggregate) {
1348  auto* table_descriptor = node->getModifiedTableDescriptor();
1349  CHECK(table_descriptor);
1350  if (!table_descriptor->hasDeletedCol) {
1351  throw std::runtime_error(
1352  "DELETE only supported on tables with the vacuum attribute set to 'delayed'");
1353  }
1354 
1355  const auto table_infos = get_table_infos(work_unit.exe_unit, executor_);
1356 
1357  auto execute_delete_ra_exe_unit =
1358  [this, &table_infos, &table_descriptor, &eo_in, &co](const auto& exe_unit,
1359  const bool is_aggregate) {
1360  DeleteTransactionParameters delete_params(table_is_temporary(table_descriptor));
1361  auto delete_callback = yieldDeleteCallback(delete_params);
1363 
1364  auto eo = eo_in;
1365  if (delete_params.tableIsTemporary()) {
1366  eo.output_columnar_hint = true;
1367  co_delete.add_delete_column =
1368  false; // project the entire delete column for columnar update
1369  } else {
1370  CHECK_EQ(exe_unit.target_exprs.size(), size_t(1));
1371  }
1372 
1373  executor_->executeUpdate(exe_unit,
1374  table_infos,
1375  co_delete,
1376  eo,
1377  cat_,
1378  executor_->row_set_mem_owner_,
1379  delete_callback,
1380  is_aggregate);
1381  delete_params.finalizeTransaction();
1382  };
1383 
1384  if (table_is_temporary(table_descriptor)) {
1385  auto query_rewrite = std::make_unique<QueryRewriter>(table_infos, executor_);
1386  auto cd = cat_.getDeletedColumn(table_descriptor);
1387  CHECK(cd);
1388  auto delete_column_expr = makeExpr<Analyzer::ColumnVar>(
1389  cd->columnType, table_descriptor->tableId, cd->columnId, 0);
1390  const auto rewritten_exe_unit =
1391  query_rewrite->rewriteColumnarDelete(work_unit.exe_unit, delete_column_expr);
1392  execute_delete_ra_exe_unit(rewritten_exe_unit, is_aggregate);
1393  } else {
1394  execute_delete_ra_exe_unit(work_unit.exe_unit, is_aggregate);
1395  }
1396  };
1397 
1398  if (auto compound = dynamic_cast<const RelCompound*>(node)) {
1399  const auto work_unit =
1400  createCompoundWorkUnit(compound, {{}, SortAlgorithm::Default, 0, 0}, eo_in);
1401  execute_delete_for_node(compound, work_unit, compound->isAggregate());
1402  } else if (auto project = dynamic_cast<const RelProject*>(node)) {
1403  auto work_unit =
1404  createProjectWorkUnit(project, {{}, SortAlgorithm::Default, 0, 0}, eo_in);
1405  if (project->isSimple()) {
1406  CHECK_EQ(size_t(1), project->inputCount());
1407  const auto input_ra = project->getInput(0);
1408  if (dynamic_cast<const RelSort*>(input_ra)) {
1409  const auto& input_table =
1410  get_temporary_table(&temporary_tables_, -input_ra->getId());
1411  CHECK(input_table);
1412  work_unit.exe_unit.scan_limit = input_table->rowCount();
1413  }
1414  }
1415  execute_delete_for_node(project, work_unit, false);
1416  } else {
1417  throw std::runtime_error("Unsupported parent node for delete: " + node->toString());
1418  }
1419 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
WorkUnit createProjectWorkUnit(const RelProject *, const SortInfo &, const ExecutionOptions &eo)
const ColumnDescriptor * getDeletedColumn(const TableDescriptor *td) const
Definition: Catalog.cpp:2539
TemporaryTables temporary_tables_
static void invalidateCaches()
WorkUnit createCompoundWorkUnit(const RelCompound *, const SortInfo &, const ExecutionOptions &eo)
const ResultSetPtr & get_temporary_table(const TemporaryTables *temporary_tables, const int table_id)
Definition: Execute.h:179
static CompilationOptions makeCpuOnly(const CompilationOptions &in)
CHECK(cgen_state)
UpdateCallback yieldDeleteCallback(DeleteTransactionParameters &delete_parameters)
const Catalog_Namespace::Catalog & cat_
bool table_is_temporary(const TableDescriptor *const td)
std::vector< InputTableInfo > get_table_infos(const std::vector< InputDescriptor > &input_descs, Executor *executor)
#define DEBUG_TIMER(name)
Definition: Logger.h:313
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeFilter ( const RelFilter filter,
const CompilationOptions co,
const ExecutionOptions eo,
RenderInfo render_info,
const int64_t  queue_time_ms 
)
private

Definition at line 1667 of file RelAlgExecutor.cpp.

References createFilterWorkUnit(), DEBUG_TIMER, Default, executeWorkUnit(), RelAlgNode::getOutputMetainfo(), and ExecutionOptions::just_explain.

Referenced by executeRelAlgStep().

1671  {
1672  auto timer = DEBUG_TIMER(__func__);
1673  const auto work_unit =
1674  createFilterWorkUnit(filter, {{}, SortAlgorithm::Default, 0, 0}, eo.just_explain);
1675  return executeWorkUnit(
1676  work_unit, filter->getOutputMetainfo(), false, co, eo, render_info, queue_time_ms);
1677 }
ExecutionResult executeWorkUnit(const WorkUnit &work_unit, const std::vector< TargetMetaInfo > &targets_meta, const bool is_agg, const CompilationOptions &co_in, const ExecutionOptions &eo, RenderInfo *, const int64_t queue_time_ms, const ssize_t previous_count=-1)
#define DEBUG_TIMER(name)
Definition: Logger.h:313
const std::vector< TargetMetaInfo > & getOutputMetainfo() const
WorkUnit createFilterWorkUnit(const RelFilter *, const SortInfo &, const bool just_explain)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeLogicalValues ( const RelLogicalValues logical_values,
const ExecutionOptions eo 
)
private

Definition at line 1679 of file RelAlgExecutor.cpp.

References CHECK(), CHECK_EQ, CHECK_GE, CPU, DEBUG_TIMER, EMPTY_KEY_64, executor_, RelLogicalValues::getNumRows(), RelLogicalValues::getRowsSize(), RelLogicalValues::getTupleType(), RelLogicalValues::getValueAt(), RelLogicalValues::hasRows(), inline_int_null_val(), ExecutionOptions::just_explain, kBIGINT, kCOUNT, kNULLT, Projection, query_mem_desc, RelAlgNode::setOutputMetainfo(), RelLogicalValues::size(), and RelAlgTranslator::translateLiteral().

Referenced by executeRelAlgStep().

1681  {
1682  auto timer = DEBUG_TIMER(__func__);
1683  if (eo.just_explain) {
1684  throw std::runtime_error("EXPLAIN not supported for LogicalValues");
1685  }
1686 
1688  logical_values->getNumRows(),
1690  /*is_table_function=*/false);
1691 
1692  auto tuple_type = logical_values->getTupleType();
1693  for (size_t i = 0; i < tuple_type.size(); ++i) {
1694  auto& target_meta_info = tuple_type[i];
1695  if (target_meta_info.get_type_info().is_varlen()) {
1696  throw std::runtime_error("Variable length types not supported in VALUES yet.");
1697  }
1698  if (target_meta_info.get_type_info().get_type() == kNULLT) {
1699  // replace w/ bigint
1700  tuple_type[i] =
1701  TargetMetaInfo(target_meta_info.get_resname(), SQLTypeInfo(kBIGINT, false));
1702  }
1703  query_mem_desc.addColSlotInfo(
1704  {std::make_tuple(tuple_type[i].get_type_info().get_size(), 8)});
1705  }
1706  logical_values->setOutputMetainfo(tuple_type);
1707 
1708  std::vector<TargetInfo> target_infos;
1709  for (const auto& tuple_type_component : tuple_type) {
1710  target_infos.emplace_back(TargetInfo{false,
1711  kCOUNT,
1712  tuple_type_component.get_type_info(),
1713  SQLTypeInfo(kNULLT, false),
1714  false,
1715  false});
1716  }
1717  auto rs = std::make_shared<ResultSet>(target_infos,
1720  executor_->getRowSetMemoryOwner(),
1721  executor_);
1722 
1723  if (logical_values->hasRows()) {
1724  CHECK_EQ(logical_values->getRowsSize(), logical_values->size());
1725 
1726  auto storage = rs->allocateStorage();
1727  auto buff = storage->getUnderlyingBuffer();
1728 
1729  for (size_t i = 0; i < logical_values->getNumRows(); i++) {
1730  std::vector<std::shared_ptr<Analyzer::Expr>> row_literals;
1731  int8_t* ptr = buff + i * query_mem_desc.getRowSize();
1732 
1733  for (size_t j = 0; j < logical_values->getRowsSize(); j++) {
1734  auto rex_literal =
1735  dynamic_cast<const RexLiteral*>(logical_values->getValueAt(i, j));
1736  CHECK(rex_literal);
1737  const auto expr = RelAlgTranslator::translateLiteral(rex_literal);
1738  const auto constant = std::dynamic_pointer_cast<Analyzer::Constant>(expr);
1739  CHECK(constant);
1740 
1741  if (constant->get_is_null()) {
1742  CHECK(!target_infos[j].sql_type.is_varlen());
1743  *reinterpret_cast<int64_t*>(ptr) =
1744  inline_int_null_val(target_infos[j].sql_type);
1745  } else {
1746  const auto ti = constant->get_type_info();
1747  const auto datum = constant->get_constval();
1748 
1749  // Initialize the entire 8-byte slot
1750  *reinterpret_cast<int64_t*>(ptr) = EMPTY_KEY_64;
1751 
1752  const auto sz = ti.get_size();
1753  CHECK_GE(sz, int(0));
1754  std::memcpy(ptr, &datum, sz);
1755  }
1756  ptr += 8;
1757  }
1758  }
1759  }
1760  return {rs, tuple_type};
1761 }
bool hasRows() const
#define CHECK_EQ(x, y)
Definition: Logger.h:205
#define EMPTY_KEY_64
size_t size() const override
size_t getNumRows() const
const std::vector< TargetMetaInfo > getTupleType() const
#define CHECK_GE(x, y)
Definition: Logger.h:210
size_t getRowsSize() const
static std::shared_ptr< Analyzer::Expr > translateLiteral(const RexLiteral *)
CHECK(cgen_state)
Definition: sqldefs.h:76
const RexScalar * getValueAt(const size_t row_idx, const size_t col_idx) const
#define DEBUG_TIMER(name)
Definition: Logger.h:313
void setOutputMetainfo(const std::vector< TargetMetaInfo > &targets_metainfo) const
int64_t inline_int_null_val(const SQL_TYPE_INFO &ti)
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeModify ( const RelModify modify,
const ExecutionOptions eo 
)
private

Definition at line 1811 of file RelAlgExecutor.cpp.

References CPU, DEBUG_TIMER, executor_, and ExecutionOptions::just_explain.

Referenced by executeRelAlgStep().

1812  {
1813  auto timer = DEBUG_TIMER(__func__);
1814  if (eo.just_explain) {
1815  throw std::runtime_error("EXPLAIN not supported for ModifyTable");
1816  }
1817 
1818  auto rs = std::make_shared<ResultSet>(TargetInfoList{},
1821  executor_->getRowSetMemoryOwner(),
1822  executor_);
1823 
1824  std::vector<TargetMetaInfo> empty_targets;
1825  return {rs, empty_targets};
1826 }
std::vector< TargetInfo > TargetInfoList
#define DEBUG_TIMER(name)
Definition: Logger.h:313
Executor * executor_

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeProject ( const RelProject project,
const CompilationOptions co,
const ExecutionOptions eo,
RenderInfo render_info,
const int64_t  queue_time_ms,
const ssize_t  previous_count 
)
private

Definition at line 1469 of file RelAlgExecutor.cpp.

References CHECK(), CHECK_EQ, CPU, createProjectWorkUnit(), DEBUG_TIMER, Default, executeWorkUnit(), get_temporary_table(), RelAlgNode::getInput(), RelAlgNode::getOutputMetainfo(), RelAlgNode::inputCount(), RelProject::isSimple(), and temporary_tables_.

Referenced by executeRelAlgStep().

1474  {
1475  auto timer = DEBUG_TIMER(__func__);
1476  auto work_unit = createProjectWorkUnit(project, {{}, SortAlgorithm::Default, 0, 0}, eo);
1477  CompilationOptions co_project = co;
1478  if (project->isSimple()) {
1479  CHECK_EQ(size_t(1), project->inputCount());
1480  const auto input_ra = project->getInput(0);
1481  if (dynamic_cast<const RelSort*>(input_ra)) {
1482  co_project.device_type = ExecutorDeviceType::CPU;
1483  const auto& input_table =
1484  get_temporary_table(&temporary_tables_, -input_ra->getId());
1485  CHECK(input_table);
1486  work_unit.exe_unit.scan_limit =
1487  std::min(input_table->getLimit(), input_table->rowCount());
1488  }
1489  }
1490  return executeWorkUnit(work_unit,
1491  project->getOutputMetainfo(),
1492  false,
1493  co_project,
1494  eo,
1495  render_info,
1496  queue_time_ms,
1497  previous_count);
1498 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
WorkUnit createProjectWorkUnit(const RelProject *, const SortInfo &, const ExecutionOptions &eo)
TemporaryTables temporary_tables_
const ResultSetPtr & get_temporary_table(const TemporaryTables *temporary_tables, const int table_id)
Definition: Execute.h:179
CHECK(cgen_state)
const RelAlgNode * getInput(const size_t idx) const
bool isSimple() const
ExecutionResult executeWorkUnit(const WorkUnit &work_unit, const std::vector< TargetMetaInfo > &targets_meta, const bool is_agg, const CompilationOptions &co_in, const ExecutionOptions &eo, RenderInfo *, const int64_t queue_time_ms, const ssize_t previous_count=-1)
#define DEBUG_TIMER(name)
Definition: Logger.h:313
const size_t inputCount() const
const std::vector< TargetMetaInfo > & getOutputMetainfo() const

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeRelAlgQuery ( const CompilationOptions co,
const ExecutionOptions eo,
const bool  just_explain_plan,
RenderInfo render_info 
)

Definition at line 151 of file RelAlgExecutor.cpp.

References CHECK(), DEBUG_TIMER, executeRelAlgQueryNoRetry(), g_allow_cpu_retry, logger::INFO, INJECT_TIMER, LOG, CompilationOptions::makeCpuOnly(), query_dag_, and RenderInfo::setForceNonInSituData().

154  {
155  CHECK(query_dag_);
156  auto timer = DEBUG_TIMER(__func__);
158  try {
159  return executeRelAlgQueryNoRetry(co, eo, just_explain_plan, render_info);
160  } catch (const QueryMustRunOnCpu&) {
161  if (!g_allow_cpu_retry) {
162  throw;
163  }
164  }
165  LOG(INFO) << "Query unable to run in GPU mode, retrying on CPU";
166  auto co_cpu = CompilationOptions::makeCpuOnly(co);
167 
168  if (render_info) {
169  render_info->setForceNonInSituData();
170  }
171  return executeRelAlgQueryNoRetry(co_cpu, eo, just_explain_plan, render_info);
172 }
void setForceNonInSituData()
Definition: RenderInfo.cpp:42
#define LOG(tag)
Definition: Logger.h:188
static CompilationOptions makeCpuOnly(const CompilationOptions &in)
CHECK(cgen_state)
#define INJECT_TIMER(DESC)
Definition: measure.h:91
ExecutionResult executeRelAlgQueryNoRetry(const CompilationOptions &co, const ExecutionOptions &eo, const bool just_explain_plan, RenderInfo *render_info)
ExecutionResult executeRelAlgQuery(const CompilationOptions &co, const ExecutionOptions &eo, const bool just_explain_plan, RenderInfo *render_info)
#define DEBUG_TIMER(name)
Definition: Logger.h:313
std::unique_ptr< RelAlgDagBuilder > query_dag_
bool g_allow_cpu_retry
Definition: Execute.cpp:75

+ Here is the call graph for this function:

ExecutionResult RelAlgExecutor::executeRelAlgQueryNoRetry ( const CompilationOptions co,
const ExecutionOptions eo,
const bool  just_explain_plan,
RenderInfo render_info 
)
private

Definition at line 174 of file RelAlgExecutor.cpp.

References cat_, CHECK(), cleanupPostExecution(), DEBUG_TIMER, RenderInfo::disallow_in_situ_only_if_final_ED_is_aggregate, executeRelAlgQueryWithFilterPushDown(), executeRelAlgSeq(), executor_, ExecutionOptions::find_push_down_candidates, g_enable_dynamic_watchdog, g_enable_runtime_query_interrupt, g_runtime_query_interrupt_frequency, get_physical_inputs(), get_physical_table_inputs(), getSubqueries(), INJECT_TIMER, query_dag_, query_state_, run_benchmark_import::result, RenderInfo::setInSituDataIfUnset(), timer_start(), timer_stop(), to_string(), and VLOG.

Referenced by executeRelAlgQuery().

177  {
179  auto timer = DEBUG_TIMER(__func__);
180  auto timer_setup = DEBUG_TIMER("Query pre-execution steps");
181 
182  query_dag_->resetQueryExecutionState();
183  const auto& ra = query_dag_->getRootNode();
184 
185  // capture the lock acquistion time
186  auto clock_begin = timer_start();
188  executor_->resetInterrupt();
189  }
190 
191  std::string query_session = "";
193  // a request of query execution without session id can happen, i.e., test query
194  // if so, we turn back to the original way: a runtime query interrupt
195  // without per-session management (as similar to dynamic watchdog)
196  if (query_state_ != nullptr && query_state_->getConstSessionInfo() != nullptr) {
197  query_session = query_state_->getConstSessionInfo()->get_session_id();
198  } else if (executor_->getCurrentQuerySession() != query_session) {
199  query_session = executor_->getCurrentQuerySession();
200  }
201  if (query_session != "") {
202  // if session is valid, then we allow per-session runtime query interrupt
203  executor_->addToQuerySessionList(query_session);
204  // hybrid spinlock. if it fails to acquire a lock, then
205  // it sleeps {g_runtime_query_interrupt_frequency} millisecond.
206  while (executor_->execute_spin_lock_.test_and_set(std::memory_order_acquire)) {
207  // failed to get the spinlock: check whether query is interrupted
208  if (executor_->checkIsQuerySessionInterrupted(query_session)) {
209  executor_->removeFromQuerySessionList(query_session);
210  VLOG(1) << "Kill the Interrupted pending query.";
211  throw std::runtime_error(
212  "Query execution has been interrupted (pending query).");
213  }
214  // here it fails to acquire the lock
215  std::this_thread::sleep_for(
216  std::chrono::milliseconds(g_runtime_query_interrupt_frequency));
217  };
218  }
219  // currently, atomic_flag does not provide a way to get its current status,
220  // i.e., spinlock.is_locked(), so we additionally lock the execute_mutex_
221  // right after acquiring spinlock to let other part of the code can know
222  // whether there exists a running query on the executor
223  }
224  std::lock_guard<std::mutex> lock(executor_->execute_mutex_);
225 
226  ScopeGuard clearRuntimeInterruptStatus = [this] {
227  // reset the runtime query interrupt status
229  executor_->removeFromQuerySessionList(executor_->getCurrentQuerySession());
230  executor_->invalidateQuerySession();
231  executor_->resetInterrupt();
232  executor_->execute_spin_lock_.clear(std::memory_order_acquire);
233  VLOG(1) << "RESET runtime query interrupt status of Executor " << this;
234  }
235  };
236 
238  // make sure to set the running session ID
239  executor_->invalidateQuerySession();
240  executor_->setCurrentQuerySession(query_session);
241  }
242 
243  int64_t queue_time_ms = timer_stop(clock_begin);
244  ScopeGuard row_set_holder = [this] { cleanupPostExecution(); };
245  const auto phys_inputs = get_physical_inputs(cat_, &ra);
246  const auto phys_table_ids = get_physical_table_inputs(&ra);
247  executor_->setCatalog(&cat_);
248  executor_->setupCaching(phys_inputs, phys_table_ids);
249 
250  ScopeGuard restore_metainfo_cache = [this] { executor_->clearMetaInfoCache(); };
251  auto ed_seq = RaExecutionSequence(&ra);
252 
253  if (just_explain_plan) {
254  std::stringstream ss;
255  std::vector<const RelAlgNode*> nodes;
256  for (size_t i = 0; i < ed_seq.size(); i++) {
257  nodes.emplace_back(ed_seq.getDescriptor(i)->getBody());
258  }
259  size_t ctr = nodes.size();
260  size_t tab_ctr = 0;
261  for (auto& body : boost::adaptors::reverse(nodes)) {
262  const auto index = ctr--;
263  const auto tabs = std::string(tab_ctr++, '\t');
264  CHECK(body);
265  ss << tabs << std::to_string(index) << " : " << body->toString() << "\n";
266  if (auto sort = dynamic_cast<const RelSort*>(body)) {
267  ss << tabs << " : " << sort->getInput(0)->toString() << "\n";
268  }
269  }
270  auto rs = std::make_shared<ResultSet>(ss.str());
271  return {rs, {}};
272  }
273 
274  if (render_info) {
275  // set render to be non-insitu in certain situations.
277  ed_seq.size() > 1) {
278  // old logic
279  // disallow if more than one ED
280  render_info->setInSituDataIfUnset(false);
281  }
282  }
283 
284  if (eo.find_push_down_candidates) {
285  // this extra logic is mainly due to current limitations on multi-step queries
286  // and/or subqueries.
288  ed_seq, co, eo, render_info, queue_time_ms);
289  }
290  timer_setup.stop();
291 
292  // Dispatch the subqueries first
293  for (auto subquery : getSubqueries()) {
294  const auto subquery_ra = subquery->getRelAlg();
295  CHECK(subquery_ra);
296  if (subquery_ra->hasContextData()) {
297  continue;
298  }
299  // Execute the subquery and cache the result.
300  RelAlgExecutor ra_executor(executor_, cat_, query_state_);
301  RaExecutionSequence subquery_seq(subquery_ra);
302  auto result = ra_executor.executeRelAlgSeq(subquery_seq, co, eo, nullptr, 0);
303  subquery->setExecutionResult(std::make_shared<ExecutionResult>(result));
304  }
305  return executeRelAlgSeq(ed_seq, co, eo, render_info, queue_time_ms);
306 }
ExecutionResult executeRelAlgSeq(const RaExecutionSequence &seq, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const int64_t queue_time_ms, const bool with_existing_temp_tables=false)
ExecutionResult executeRelAlgQueryWithFilterPushDown(const RaExecutionSequence &seq, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const int64_t queue_time_ms)
bool setInSituDataIfUnset(const bool is_in_situ_data)
Definition: RenderInfo.cpp:95
TypeR::rep timer_stop(Type clock_begin)
Definition: measure.h:46
bool g_enable_dynamic_watchdog
Definition: Execute.cpp:72
const std::vector< std::shared_ptr< RexSubQuery > > & getSubqueries() const noexcept
std::string to_string(char const *&&v)
bool disallow_in_situ_only_if_final_ED_is_aggregate
Definition: RenderInfo.h:41
const bool find_push_down_candidates
CHECK(cgen_state)
#define INJECT_TIMER(DESC)
Definition: measure.h:91
A container for relational algebra descriptors defining the execution order for a relational algebra ...
const Catalog_Namespace::Catalog & cat_
ExecutionResult executeRelAlgQueryNoRetry(const CompilationOptions &co, const ExecutionOptions &eo, const bool just_explain_plan, RenderInfo *render_info)
unsigned g_runtime_query_interrupt_frequency
Definition: Execute.cpp:105
std::shared_ptr< const query_state::QueryState > query_state_
#define DEBUG_TIMER(name)
Definition: Logger.h:313
std::unique_ptr< RelAlgDagBuilder > query_dag_
void cleanupPostExecution()
std::unordered_set< PhysicalInput > get_physical_inputs(const RelAlgNode *ra)
Executor * executor_
bool g_enable_runtime_query_interrupt
Definition: Execute.cpp:104
std::unordered_set< int > get_physical_table_inputs(const RelAlgNode *ra)
#define VLOG(n)
Definition: Logger.h:291
Type timer_start()
Definition: measure.h:40

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

FirstStepExecutionResult RelAlgExecutor::executeRelAlgQuerySingleStep ( const RaExecutionSequence seq,
const size_t  step_idx,
const CompilationOptions co,
const ExecutionOptions eo,
RenderInfo render_info 
)

Definition at line 346 of file RelAlgExecutor.cpp.

References CHECK(), CHECK_EQ, anonymous_namespace{RelAlgExecutor.cpp}::check_sort_node_source_constraint(), createSortInputWorkUnit(), executeRelAlgSubSeq(), executor_, RaExecutionSequence::getDescriptor(), INJECT_TIMER, anonymous_namespace{RelAlgExecutor.cpp}::node_is_aggregate(), queue_time_ms_, Reduce, GroupByAndAggregate::shard_count_for_top_groups(), and Union.

351  {
352  INJECT_TIMER(executeRelAlgQueryStep);
353  auto exe_desc_ptr = seq.getDescriptor(step_idx);
354  CHECK(exe_desc_ptr);
355  const auto sort = dynamic_cast<const RelSort*>(exe_desc_ptr->getBody());
356 
357  size_t shard_count{0};
358  auto merge_type = [&shard_count](const RelAlgNode* body) -> MergeType {
359  return node_is_aggregate(body) && !shard_count ? MergeType::Reduce : MergeType::Union;
360  };
361 
362  if (sort) {
364  const auto source_work_unit = createSortInputWorkUnit(sort, eo);
366  source_work_unit.exe_unit, *executor_->getCatalog());
367  if (!shard_count) {
368  // No point in sorting on the leaf, only execute the input to the sort node.
369  CHECK_EQ(size_t(1), sort->inputCount());
370  const auto source = sort->getInput(0);
371  if (sort->collationCount() || node_is_aggregate(source)) {
372  auto temp_seq = RaExecutionSequence(std::make_unique<RaExecutionDesc>(source));
373  CHECK_EQ(temp_seq.size(), size_t(1));
374  // Use subseq to avoid clearing existing temporary tables
375  return {executeRelAlgSubSeq(temp_seq, std::make_pair(0, 1), co, eo, nullptr, 0),
376  merge_type(source),
377  source->getId(),
378  false};
379  }
380  }
381  }
382  return {executeRelAlgSubSeq(seq,
383  std::make_pair(step_idx, step_idx + 1),
384  co,
385  eo,
386  render_info,
388  merge_type(exe_desc_ptr->getBody()),
389  exe_desc_ptr->getBody()->getId(),
390  false};
391 }
int64_t queue_time_ms_
#define CHECK_EQ(x, y)
Definition: Logger.h:205
RaExecutionDesc * getDescriptor(size_t idx) const
void check_sort_node_source_constraint(const RelSort *sort)
CHECK(cgen_state)
#define INJECT_TIMER(DESC)
Definition: measure.h:91
MergeType
A container for relational algebra descriptors defining the execution order for a relational algebra ...
ExecutionResult executeRelAlgSubSeq(const RaExecutionSequence &seq, const std::pair< size_t, size_t > interval, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const int64_t queue_time_ms)
bool node_is_aggregate(const RelAlgNode *ra)
Executor * executor_
WorkUnit createSortInputWorkUnit(const RelSort *, const ExecutionOptions &eo)
static size_t shard_count_for_top_groups(const RelAlgExecutionUnit &ra_exe_unit, const Catalog_Namespace::Catalog &catalog)

+ Here is the call graph for this function:

ExecutionResult RelAlgExecutor::executeRelAlgQueryWithFilterPushDown ( const RaExecutionSequence seq,
const CompilationOptions co,
const ExecutionOptions eo,
RenderInfo render_info,
const int64_t  queue_time_ms 
)

Definition at line 154 of file JoinFilterPushDown.cpp.

References ExecutionOptions::allow_loop_joins, ExecutionOptions::allow_multifrag, ExecutionOptions::allow_runtime_query_interrupt, cat_, CHECK(), ExecutionOptions::dynamic_watchdog_time_limit, executeRelAlgSeq(), executor_, ExecutionOptions::find_push_down_candidates, getSubqueries(), ExecutionOptions::gpu_input_mem_limit_percent, ExecutionOptions::jit_debug, ExecutionOptions::just_calcite_explain, ExecutionOptions::just_explain, ExecutionOptions::just_validate, ExecutionOptions::output_columnar_hint, run_benchmark_import::result, RaExecutionSequence::size(), ExecutionOptions::with_dynamic_watchdog, and ExecutionOptions::with_watchdog.

Referenced by executeRelAlgQueryNoRetry().

159  {
160  // we currently do not fully support filter push down with
161  // multi-step execution and/or with subqueries
162  // TODO(Saman): add proper caching to enable filter push down for all cases
163  const auto& subqueries = getSubqueries();
164  if (seq.size() > 1 || !subqueries.empty()) {
165  if (eo.just_calcite_explain) {
166  return ExecutionResult(std::vector<PushedDownFilterInfo>{},
168  }
169  const ExecutionOptions eo_modified{eo.output_columnar_hint,
170  eo.allow_multifrag,
171  eo.just_explain,
172  eo.allow_loop_joins,
173  eo.with_watchdog,
174  eo.jit_debug,
175  eo.just_validate,
178  /*find_push_down_candidates=*/false,
179  /*just_calcite_explain=*/false,
182 
183  // Dispatch the subqueries first
184  for (auto subquery : subqueries) {
185  // Execute the subquery and cache the result.
186  RelAlgExecutor ra_executor(executor_, cat_);
187  const auto subquery_ra = subquery->getRelAlg();
188  CHECK(subquery_ra);
189  RaExecutionSequence subquery_seq(subquery_ra);
190  auto result = ra_executor.executeRelAlgSeq(subquery_seq, co, eo, nullptr, 0);
191  subquery->setExecutionResult(std::make_shared<ExecutionResult>(result));
192  }
193  return executeRelAlgSeq(seq, co, eo_modified, render_info, queue_time_ms);
194  }
195  // else
196 
197  // Dispatch the subqueries first
198  for (auto subquery : subqueries) {
199  // Execute the subquery and cache the result.
200  RelAlgExecutor ra_executor(executor_, cat_);
201  const auto subquery_ra = subquery->getRelAlg();
202  CHECK(subquery_ra);
203  RaExecutionSequence subquery_seq(subquery_ra);
204  auto result = ra_executor.executeRelAlgSeq(subquery_seq, co, eo, nullptr, 0);
205  subquery->setExecutionResult(std::make_shared<ExecutionResult>(result));
206  }
207  return executeRelAlgSeq(seq, co, eo, render_info, queue_time_ms);
208 }
ExecutionResult executeRelAlgSeq(const RaExecutionSequence &seq, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const int64_t queue_time_ms, const bool with_existing_temp_tables=false)
const std::vector< std::shared_ptr< RexSubQuery > > & getSubqueries() const noexcept
const bool allow_multifrag
const bool find_push_down_candidates
CHECK(cgen_state)
const bool just_validate
const bool with_dynamic_watchdog
const double gpu_input_mem_limit_percent
A container for relational algebra descriptors defining the execution order for a relational algebra ...
const Catalog_Namespace::Catalog & cat_
const bool just_calcite_explain
const bool allow_loop_joins
const bool allow_runtime_query_interrupt
const unsigned dynamic_watchdog_time_limit
Executor * executor_
const bool with_watchdog

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeRelAlgSeq ( const RaExecutionSequence seq,
const CompilationOptions co,
const ExecutionOptions eo,
RenderInfo render_info,
const int64_t  queue_time_ms,
const bool  with_existing_temp_tables = false 
)

Definition at line 409 of file RelAlgExecutor.cpp.

References cat_, CHECK(), DEBUG_TIMER, RaExecutionSequence::empty(), executeRelAlgStep(), executor_, ExecutionOptions::executor_type, g_enable_interop, RaExecutionDesc::getBody(), RaExecutionSequence::getDescriptor(), logger::INFO, INJECT_TIMER, ExecutionOptions::just_explain, LOG, now_, RaExecutionSequence::size(), target_exprs_owned_, and temporary_tables_.

Referenced by executeRelAlgQueryNoRetry(), and executeRelAlgQueryWithFilterPushDown().

414  {
416  auto timer = DEBUG_TIMER(__func__);
417  if (!with_existing_temp_tables) {
418  decltype(temporary_tables_)().swap(temporary_tables_);
419  }
420  decltype(target_exprs_owned_)().swap(target_exprs_owned_);
421  executor_->catalog_ = &cat_;
422  executor_->temporary_tables_ = &temporary_tables_;
423 
424  time(&now_);
425  CHECK(!seq.empty());
426  const auto exec_desc_count = eo.just_explain ? size_t(1) : seq.size();
427 
428  for (size_t i = 0; i < exec_desc_count; i++) {
429  // only render on the last step
430  try {
431  executeRelAlgStep(seq,
432  i,
433  co,
434  eo,
435  (i == exec_desc_count - 1) ? render_info : nullptr,
436  queue_time_ms);
437  } catch (const NativeExecutionError&) {
438  if (!g_enable_interop) {
439  throw;
440  }
441  auto eo_extern = eo;
442  eo_extern.executor_type = ::ExecutorType::Extern;
443  auto exec_desc_ptr = seq.getDescriptor(i);
444  const auto body = exec_desc_ptr->getBody();
445  const auto compound = dynamic_cast<const RelCompound*>(body);
446  if (compound && (compound->getGroupByCount() || compound->isAggregate())) {
447  LOG(INFO) << "Also failed to run the query using interoperability";
448  throw;
449  }
450  executeRelAlgStep(seq,
451  i,
452  co,
453  eo_extern,
454  (i == exec_desc_count - 1) ? render_info : nullptr,
455  queue_time_ms);
456  }
457  }
458 
459  return seq.getDescriptor(exec_desc_count - 1)->getResult();
460 }
RaExecutionDesc * getDescriptor(size_t idx) const
ExecutionResult executeRelAlgSeq(const RaExecutionSequence &seq, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const int64_t queue_time_ms, const bool with_existing_temp_tables=false)
#define LOG(tag)
Definition: Logger.h:188
TemporaryTables temporary_tables_
std::vector< std::shared_ptr< Analyzer::Expr > > target_exprs_owned_
CHECK(cgen_state)
ExecutorType executor_type
#define INJECT_TIMER(DESC)
Definition: measure.h:91
const Catalog_Namespace::Catalog & cat_
void executeRelAlgStep(const RaExecutionSequence &seq, const size_t step_idx, const CompilationOptions &, const ExecutionOptions &, RenderInfo *, const int64_t queue_time_ms)
bool g_enable_interop
#define DEBUG_TIMER(name)
Definition: Logger.h:313
const RelAlgNode * getBody() const
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void RelAlgExecutor::executeRelAlgStep ( const RaExecutionSequence seq,
const size_t  step_idx,
const CompilationOptions co,
const ExecutionOptions eo,
RenderInfo render_info,
const int64_t  queue_time_ms 
)
private

Definition at line 489 of file RelAlgExecutor.cpp.

References addTemporaryTable(), ExecutionOptions::allow_loop_joins, ExecutionOptions::allow_multifrag, ExecutionOptions::allow_runtime_query_interrupt, CHECK(), DEBUG_TIMER, ExecutionOptions::dynamic_watchdog_time_limit, executeAggregate(), executeCompound(), executeDelete(), executeFilter(), executeLogicalValues(), executeModify(), executeProject(), executeSort(), executeTableFunction(), executeUpdate(), ExecutionOptions::executor_type, ExecutionOptions::find_push_down_candidates, g_cluster, g_skip_intermediate_count, RaExecutionSequence::getDescriptor(), ExecutionOptions::gpu_input_mem_limit_percent, handleNop(), INJECT_TIMER, ExecutionOptions::jit_debug, ExecutionOptions::just_calcite_explain, ExecutionOptions::just_explain, ExecutionOptions::just_validate, ExecutionOptions::outer_fragment_indices, ExecutionOptions::output_columnar_hint, WindowProjectNodeContext::reset(), ExecutionOptions::runtime_query_interrupt_frequency, ExecutionOptions::with_dynamic_watchdog, and ExecutionOptions::with_watchdog.

Referenced by executeRelAlgSeq(), and executeRelAlgSubSeq().

494  {
496  auto timer = DEBUG_TIMER(__func__);
498  auto exec_desc_ptr = seq.getDescriptor(step_idx);
499  CHECK(exec_desc_ptr);
500  auto& exec_desc = *exec_desc_ptr;
501  const auto body = exec_desc.getBody();
502  if (body->isNop()) {
503  handleNop(exec_desc);
504  return;
505  }
506  const ExecutionOptions eo_work_unit{
508  eo.allow_multifrag,
509  eo.just_explain,
510  eo.allow_loop_joins,
511  eo.with_watchdog && (step_idx == 0 || dynamic_cast<const RelProject*>(body)),
512  eo.jit_debug,
513  eo.just_validate,
521  eo.executor_type,
522  step_idx == 0 ? eo.outer_fragment_indices : std::vector<size_t>()};
523 
524  const auto compound = dynamic_cast<const RelCompound*>(body);
525  if (compound) {
526  if (compound->isDeleteViaSelect()) {
527  executeDelete(compound, co, eo_work_unit, queue_time_ms);
528  } else if (compound->isUpdateViaSelect()) {
529  executeUpdate(compound, co, eo_work_unit, queue_time_ms);
530  } else {
531  exec_desc.setResult(
532  executeCompound(compound, co, eo_work_unit, render_info, queue_time_ms));
533  if (exec_desc.getResult().isFilterPushDownEnabled()) {
534  return;
535  }
536  addTemporaryTable(-compound->getId(), exec_desc.getResult().getDataPtr());
537  }
538  return;
539  }
540  const auto project = dynamic_cast<const RelProject*>(body);
541  if (project) {
542  if (project->isDeleteViaSelect()) {
543  executeDelete(project, co, eo_work_unit, queue_time_ms);
544  } else if (project->isUpdateViaSelect()) {
545  executeUpdate(project, co, eo_work_unit, queue_time_ms);
546  } else {
547  ssize_t prev_count = -1;
548  // Disabling the intermediate count optimization in distributed, as the previous
549  // execution descriptor will likely not hold the aggregated result.
550  if (g_skip_intermediate_count && step_idx > 0 && !g_cluster) {
551  auto prev_exec_desc = seq.getDescriptor(step_idx - 1);
552  CHECK(prev_exec_desc);
553  // If the previous node produced a reliable count, skip the pre-flight count
554  if (dynamic_cast<const RelCompound*>(prev_exec_desc->getBody()) ||
555  dynamic_cast<const RelLogicalValues*>(prev_exec_desc->getBody())) {
556  const auto& prev_exe_result = prev_exec_desc->getResult();
557  const auto prev_result = prev_exe_result.getRows();
558  if (prev_result) {
559  prev_count = static_cast<ssize_t>(prev_result->rowCount());
560  }
561  }
562  }
563  exec_desc.setResult(executeProject(
564  project, co, eo_work_unit, render_info, queue_time_ms, prev_count));
565  if (exec_desc.getResult().isFilterPushDownEnabled()) {
566  return;
567  }
568  addTemporaryTable(-project->getId(), exec_desc.getResult().getDataPtr());
569  }
570  return;
571  }
572  const auto aggregate = dynamic_cast<const RelAggregate*>(body);
573  if (aggregate) {
574  exec_desc.setResult(
575  executeAggregate(aggregate, co, eo_work_unit, render_info, queue_time_ms));
576  addTemporaryTable(-aggregate->getId(), exec_desc.getResult().getDataPtr());
577  return;
578  }
579  const auto filter = dynamic_cast<const RelFilter*>(body);
580  if (filter) {
581  exec_desc.setResult(
582  executeFilter(filter, co, eo_work_unit, render_info, queue_time_ms));
583  addTemporaryTable(-filter->getId(), exec_desc.getResult().getDataPtr());
584  return;
585  }
586  const auto sort = dynamic_cast<const RelSort*>(body);
587  if (sort) {
588  exec_desc.setResult(executeSort(sort, co, eo_work_unit, render_info, queue_time_ms));
589  if (exec_desc.getResult().isFilterPushDownEnabled()) {
590  return;
591  }
592  addTemporaryTable(-sort->getId(), exec_desc.getResult().getDataPtr());
593  return;
594  }
595  const auto logical_values = dynamic_cast<const RelLogicalValues*>(body);
596  if (logical_values) {
597  exec_desc.setResult(executeLogicalValues(logical_values, eo_work_unit));
598  addTemporaryTable(-logical_values->getId(), exec_desc.getResult().getDataPtr());
599  return;
600  }
601  const auto modify = dynamic_cast<const RelModify*>(body);
602  if (modify) {
603  exec_desc.setResult(executeModify(modify, eo_work_unit));
604  return;
605  }
606  const auto table_func = dynamic_cast<const RelTableFunction*>(body);
607  if (table_func) {
608  exec_desc.setResult(
609  executeTableFunction(table_func, co, eo_work_unit, queue_time_ms));
610  addTemporaryTable(-table_func->getId(), exec_desc.getResult().getDataPtr());
611  return;
612  }
613  CHECK(false);
614 }
ExecutionResult executeAggregate(const RelAggregate *aggregate, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const int64_t queue_time_ms)
RaExecutionDesc * getDescriptor(size_t idx) const
ExecutionResult executeProject(const RelProject *, const CompilationOptions &, const ExecutionOptions &, RenderInfo *, const int64_t queue_time_ms, const ssize_t previous_count)
const unsigned runtime_query_interrupt_frequency
ExecutionResult executeModify(const RelModify *modify, const ExecutionOptions &eo)
void addTemporaryTable(const int table_id, const ResultSetPtr &result)
ExecutionResult executeLogicalValues(const RelLogicalValues *, const ExecutionOptions &)
ExecutionResult executeFilter(const RelFilter *, const CompilationOptions &, const ExecutionOptions &, RenderInfo *, const int64_t queue_time_ms)
void handleNop(RaExecutionDesc &ed)
const bool allow_multifrag
const bool find_push_down_candidates
CHECK(cgen_state)
const bool just_validate
ExecutorType executor_type
#define INJECT_TIMER(DESC)
Definition: measure.h:91
const bool with_dynamic_watchdog
const double gpu_input_mem_limit_percent
void executeUpdate(const RelAlgNode *node, const CompilationOptions &co, const ExecutionOptions &eo, const int64_t queue_time_ms)
void executeRelAlgStep(const RaExecutionSequence &seq, const size_t step_idx, const CompilationOptions &, const ExecutionOptions &, RenderInfo *, const int64_t queue_time_ms)
ExecutionResult executeTableFunction(const RelTableFunction *, const CompilationOptions &, const ExecutionOptions &, const int64_t queue_time_ms)
const std::vector< size_t > outer_fragment_indices
const bool just_calcite_explain
ExecutionResult executeSort(const RelSort *, const CompilationOptions &, const ExecutionOptions &, RenderInfo *, const int64_t queue_time_ms)
const bool allow_loop_joins
bool g_skip_intermediate_count
#define DEBUG_TIMER(name)
Definition: Logger.h:313
ExecutionResult executeCompound(const RelCompound *, const CompilationOptions &, const ExecutionOptions &, RenderInfo *, const int64_t queue_time_ms)
bool g_cluster
const bool allow_runtime_query_interrupt
const unsigned dynamic_watchdog_time_limit
void executeDelete(const RelAlgNode *node, const CompilationOptions &co, const ExecutionOptions &eo_in, const int64_t queue_time_ms)
const bool with_watchdog

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeRelAlgSubSeq ( const RaExecutionSequence seq,
const std::pair< size_t, size_t >  interval,
const CompilationOptions co,
const ExecutionOptions eo,
RenderInfo render_info,
const int64_t  queue_time_ms 
)

Definition at line 462 of file RelAlgExecutor.cpp.

References cat_, CHECK(), executeRelAlgStep(), executor_, RaExecutionSequence::getDescriptor(), INJECT_TIMER, ExecutionOptions::just_explain, now_, and temporary_tables_.

Referenced by executeRelAlgQuerySingleStep().

468  {
470  executor_->catalog_ = &cat_;
471  executor_->temporary_tables_ = &temporary_tables_;
472 
473  time(&now_);
474  CHECK(!eo.just_explain);
475 
476  for (size_t i = interval.first; i < interval.second; i++) {
477  // only render on the last step
478  executeRelAlgStep(seq,
479  i,
480  co,
481  eo,
482  (i == interval.second - 1) ? render_info : nullptr,
483  queue_time_ms);
484  }
485 
486  return seq.getDescriptor(interval.second - 1)->getResult();
487 }
RaExecutionDesc * getDescriptor(size_t idx) const
TemporaryTables temporary_tables_
CHECK(cgen_state)
#define INJECT_TIMER(DESC)
Definition: measure.h:91
const Catalog_Namespace::Catalog & cat_
ExecutionResult executeRelAlgSubSeq(const RaExecutionSequence &seq, const std::pair< size_t, size_t > interval, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const int64_t queue_time_ms)
void executeRelAlgStep(const RaExecutionSequence &seq, const size_t step_idx, const CompilationOptions &, const ExecutionOptions &, RenderInfo *, const int64_t queue_time_ms)
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeSimpleInsert ( const Analyzer::Query insert_query)

Definition at line 1828 of file RelAlgExecutor.cpp.

References appendDatum(), DataBlockPtr::arraysPtr, cat_, CHECK(), CHECK_EQ, checked_malloc(), Fragmenter_Namespace::InsertData::columnIds, CPU, Fragmenter_Namespace::InsertData::data, Fragmenter_Namespace::InsertData::databaseId, Catalog_Namespace::DBMetadata::dbId, executor_, get_column_descriptor(), SQLTypeInfo::get_compression(), SQLTypeInfo::get_elem_type(), Analyzer::Query::get_result_col_list(), Analyzer::Query::get_result_table_id(), SQLTypeInfo::get_size(), Analyzer::Query::get_targetlist(), SQLTypeInfo::get_type(), Catalog_Namespace::Catalog::getCurrentDB(), Catalog_Namespace::Catalog::getMetadataForDict(), Catalog_Namespace::Catalog::getMetadataForTable(), Catalog_Namespace::Catalog::getPhysicalTablesDescriptors(), inline_fixed_encoding_null_val(), anonymous_namespace{RelAlgExecutor.cpp}::insert_one_dict_str(), anonymous_namespace{TypedDataAccessors.h}::is_null(), SQLTypeInfo::is_string(), kARRAY, kBIGINT, kBOOLEAN, kCAST, kCHAR, kDATE, kDECIMAL, kDOUBLE, kENCODING_DICT, kENCODING_NONE, kFLOAT, kINT, kLINESTRING, kMULTIPOLYGON, kNUMERIC, kPOINT, kPOLYGON, kSMALLINT, kTEXT, kTIME, kTIMESTAMP, kTINYINT, kVARCHAR, DataBlockPtr::numbersPtr, Fragmenter_Namespace::InsertData::numRows, anonymous_namespace{TypedDataAccessors.h}::put_null(), anonymous_namespace{TypedDataAccessors.h}::put_null_array(), SHARD_FOR_KEY, DataBlockPtr::stringsPtr, Fragmenter_Namespace::InsertData::tableId, and to_string().

Referenced by Parser::InsertValuesStmt::execute().

1828  {
1829  // Note: We currently obtain an executor for this method, but we do not need it.
1830  // Therefore, we skip the executor state setup in the regular execution path. In the
1831  // future, we will likely want to use the executor to evaluate expressions in the insert
1832  // statement.
1833 
1834  const auto& targets = query.get_targetlist();
1835  const int table_id = query.get_result_table_id();
1836  const auto& col_id_list = query.get_result_col_list();
1837 
1838  std::vector<const ColumnDescriptor*> col_descriptors;
1839  std::vector<int> col_ids;
1840  std::unordered_map<int, std::unique_ptr<uint8_t[]>> col_buffers;
1841  std::unordered_map<int, std::vector<std::string>> str_col_buffers;
1842  std::unordered_map<int, std::vector<ArrayDatum>> arr_col_buffers;
1843 
1844  const auto table_descriptor = cat_.getMetadataForTable(table_id);
1845  CHECK(table_descriptor);
1846  const auto shard_tables = cat_.getPhysicalTablesDescriptors(table_descriptor);
1847  const TableDescriptor* shard{nullptr};
1848 
1849  for (const int col_id : col_id_list) {
1850  const auto cd = get_column_descriptor(col_id, table_id, cat_);
1851  const auto col_enc = cd->columnType.get_compression();
1852  if (cd->columnType.is_string()) {
1853  switch (col_enc) {
1854  case kENCODING_NONE: {
1855  auto it_ok =
1856  str_col_buffers.insert(std::make_pair(col_id, std::vector<std::string>{}));
1857  CHECK(it_ok.second);
1858  break;
1859  }
1860  case kENCODING_DICT: {
1861  const auto dd = cat_.getMetadataForDict(cd->columnType.get_comp_param());
1862  CHECK(dd);
1863  const auto it_ok = col_buffers.emplace(
1864  col_id, std::make_unique<uint8_t[]>(cd->columnType.get_size()));
1865  CHECK(it_ok.second);
1866  break;
1867  }
1868  default:
1869  CHECK(false);
1870  }
1871  } else if (cd->columnType.is_geometry()) {
1872  auto it_ok =
1873  str_col_buffers.insert(std::make_pair(col_id, std::vector<std::string>{}));
1874  CHECK(it_ok.second);
1875  } else if (cd->columnType.is_array()) {
1876  auto it_ok =
1877  arr_col_buffers.insert(std::make_pair(col_id, std::vector<ArrayDatum>{}));
1878  CHECK(it_ok.second);
1879  } else {
1880  const auto it_ok = col_buffers.emplace(
1881  col_id,
1882  std::unique_ptr<uint8_t[]>(
1883  new uint8_t[cd->columnType.get_logical_size()]())); // changed to zero-init
1884  // the buffer
1885  CHECK(it_ok.second);
1886  }
1887  col_descriptors.push_back(cd);
1888  col_ids.push_back(col_id);
1889  }
1890  size_t col_idx = 0;
1892  insert_data.databaseId = cat_.getCurrentDB().dbId;
1893  insert_data.tableId = table_id;
1894  int64_t int_col_val{0};
1895  for (auto target_entry : targets) {
1896  auto col_cv = dynamic_cast<const Analyzer::Constant*>(target_entry->get_expr());
1897  if (!col_cv) {
1898  auto col_cast = dynamic_cast<const Analyzer::UOper*>(target_entry->get_expr());
1899  CHECK(col_cast);
1900  CHECK_EQ(kCAST, col_cast->get_optype());
1901  col_cv = dynamic_cast<const Analyzer::Constant*>(col_cast->get_operand());
1902  }
1903  CHECK(col_cv);
1904  const auto cd = col_descriptors[col_idx];
1905  auto col_datum = col_cv->get_constval();
1906  auto col_type = cd->columnType.get_type();
1907  uint8_t* col_data_bytes{nullptr};
1908  if (!cd->columnType.is_array() && !cd->columnType.is_geometry() &&
1909  (!cd->columnType.is_string() ||
1910  cd->columnType.get_compression() == kENCODING_DICT)) {
1911  const auto col_data_bytes_it = col_buffers.find(col_ids[col_idx]);
1912  CHECK(col_data_bytes_it != col_buffers.end());
1913  col_data_bytes = col_data_bytes_it->second.get();
1914  }
1915  switch (col_type) {
1916  case kBOOLEAN: {
1917  auto col_data = col_data_bytes;
1918  *col_data = col_cv->get_is_null() ? inline_fixed_encoding_null_val(cd->columnType)
1919  : (col_datum.boolval ? 1 : 0);
1920  break;
1921  }
1922  case kTINYINT: {
1923  auto col_data = reinterpret_cast<int8_t*>(col_data_bytes);
1924  *col_data = col_cv->get_is_null() ? inline_fixed_encoding_null_val(cd->columnType)
1925  : col_datum.tinyintval;
1926  int_col_val = col_datum.tinyintval;
1927  break;
1928  }
1929  case kSMALLINT: {
1930  auto col_data = reinterpret_cast<int16_t*>(col_data_bytes);
1931  *col_data = col_cv->get_is_null() ? inline_fixed_encoding_null_val(cd->columnType)
1932  : col_datum.smallintval;
1933  int_col_val = col_datum.smallintval;
1934  break;
1935  }
1936  case kINT: {
1937  auto col_data = reinterpret_cast<int32_t*>(col_data_bytes);
1938  *col_data = col_cv->get_is_null() ? inline_fixed_encoding_null_val(cd->columnType)
1939  : col_datum.intval;
1940  int_col_val = col_datum.intval;
1941  break;
1942  }
1943  case kBIGINT:
1944  case kDECIMAL:
1945  case kNUMERIC: {
1946  auto col_data = reinterpret_cast<int64_t*>(col_data_bytes);
1947  *col_data = col_cv->get_is_null() ? inline_fixed_encoding_null_val(cd->columnType)
1948  : col_datum.bigintval;
1949  int_col_val = col_datum.bigintval;
1950  break;
1951  }
1952  case kFLOAT: {
1953  auto col_data = reinterpret_cast<float*>(col_data_bytes);
1954  *col_data = col_datum.floatval;
1955  break;
1956  }
1957  case kDOUBLE: {
1958  auto col_data = reinterpret_cast<double*>(col_data_bytes);
1959  *col_data = col_datum.doubleval;
1960  break;
1961  }
1962  case kTEXT:
1963  case kVARCHAR:
1964  case kCHAR: {
1965  switch (cd->columnType.get_compression()) {
1966  case kENCODING_NONE:
1967  str_col_buffers[col_ids[col_idx]].push_back(
1968  col_datum.stringval ? *col_datum.stringval : "");
1969  break;
1970  case kENCODING_DICT: {
1971  switch (cd->columnType.get_size()) {
1972  case 1:
1973  int_col_val = insert_one_dict_str(
1974  reinterpret_cast<uint8_t*>(col_data_bytes), cd, col_cv, cat_);
1975  break;
1976  case 2:
1977  int_col_val = insert_one_dict_str(
1978  reinterpret_cast<uint16_t*>(col_data_bytes), cd, col_cv, cat_);
1979  break;
1980  case 4:
1981  int_col_val = insert_one_dict_str(
1982  reinterpret_cast<int32_t*>(col_data_bytes), cd, col_cv, cat_);
1983  break;
1984  default:
1985  CHECK(false);
1986  }
1987  break;
1988  }
1989  default:
1990  CHECK(false);
1991  }
1992  break;
1993  }
1994  case kTIME:
1995  case kTIMESTAMP:
1996  case kDATE: {
1997  auto col_data = reinterpret_cast<int64_t*>(col_data_bytes);
1998  *col_data = col_cv->get_is_null() ? inline_fixed_encoding_null_val(cd->columnType)
1999  : col_datum.bigintval;
2000  break;
2001  }
2002  case kARRAY: {
2003  const auto is_null = col_cv->get_is_null();
2004  const auto size = cd->columnType.get_size();
2005  const SQLTypeInfo elem_ti = cd->columnType.get_elem_type();
2006  // POINT coords: [un]compressed coords always need to be encoded, even if NULL
2007  const auto is_point_coords = (cd->isGeoPhyCol && elem_ti.get_type() == kTINYINT);
2008  if (is_null && !is_point_coords) {
2009  if (size > 0) {
2010  // NULL fixlen array: NULL_ARRAY sentinel followed by NULL sentinels
2011  if (elem_ti.is_string() && elem_ti.get_compression() == kENCODING_DICT) {
2012  throw std::runtime_error("Column " + cd->columnName +
2013  " doesn't accept NULL values");
2014  }
2015  int8_t* buf = (int8_t*)checked_malloc(size);
2016  put_null_array(static_cast<void*>(buf), elem_ti, "");
2017  for (int8_t* p = buf + elem_ti.get_size(); (p - buf) < size;
2018  p += elem_ti.get_size()) {
2019  put_null(static_cast<void*>(p), elem_ti, "");
2020  }
2021  arr_col_buffers[col_ids[col_idx]].emplace_back(size, buf, is_null);
2022  } else {
2023  arr_col_buffers[col_ids[col_idx]].emplace_back(0, nullptr, is_null);
2024  }
2025  break;
2026  }
2027  const auto l = col_cv->get_value_list();
2028  size_t len = l.size() * elem_ti.get_size();
2029  if (size > 0 && static_cast<size_t>(size) != len) {
2030  throw std::runtime_error("Array column " + cd->columnName + " expects " +
2031  std::to_string(size / elem_ti.get_size()) +
2032  " values, " + "received " + std::to_string(l.size()));
2033  }
2034  if (elem_ti.is_string()) {
2035  CHECK(kENCODING_DICT == elem_ti.get_compression());
2036  CHECK(4 == elem_ti.get_size());
2037 
2038  int8_t* buf = (int8_t*)checked_malloc(len);
2039  int32_t* p = reinterpret_cast<int32_t*>(buf);
2040 
2041  int elemIndex = 0;
2042  for (auto& e : l) {
2043  auto c = std::dynamic_pointer_cast<Analyzer::Constant>(e);
2044  CHECK(c);
2045 
2046  int_col_val = insert_one_dict_str(
2047  &p[elemIndex], cd->columnName, elem_ti, c.get(), cat_);
2048 
2049  elemIndex++;
2050  }
2051  arr_col_buffers[col_ids[col_idx]].push_back(ArrayDatum(len, buf, is_null));
2052 
2053  } else {
2054  int8_t* buf = (int8_t*)checked_malloc(len);
2055  int8_t* p = buf;
2056  for (auto& e : l) {
2057  auto c = std::dynamic_pointer_cast<Analyzer::Constant>(e);
2058  CHECK(c);
2059  p = appendDatum(p, c->get_constval(), elem_ti);
2060  }
2061  arr_col_buffers[col_ids[col_idx]].push_back(ArrayDatum(len, buf, is_null));
2062  }
2063  break;
2064  }
2065  case kPOINT:
2066  case kLINESTRING:
2067  case kPOLYGON:
2068  case kMULTIPOLYGON:
2069  str_col_buffers[col_ids[col_idx]].push_back(
2070  col_datum.stringval ? *col_datum.stringval : "");
2071  break;
2072  default:
2073  CHECK(false);
2074  }
2075  ++col_idx;
2076  if (col_idx == static_cast<size_t>(table_descriptor->shardedColumnId)) {
2077  const auto shard_count = shard_tables.size();
2078  const size_t shard_idx = SHARD_FOR_KEY(int_col_val, shard_count);
2079  shard = shard_tables[shard_idx];
2080  }
2081  }
2082  for (const auto& kv : col_buffers) {
2083  insert_data.columnIds.push_back(kv.first);
2084  DataBlockPtr p;
2085  p.numbersPtr = reinterpret_cast<int8_t*>(kv.second.get());
2086  insert_data.data.push_back(p);
2087  }
2088  for (auto& kv : str_col_buffers) {
2089  insert_data.columnIds.push_back(kv.first);
2090  DataBlockPtr p;
2091  p.stringsPtr = &kv.second;
2092  insert_data.data.push_back(p);
2093  }
2094  for (auto& kv : arr_col_buffers) {
2095  insert_data.columnIds.push_back(kv.first);
2096  DataBlockPtr p;
2097  p.arraysPtr = &kv.second;
2098  insert_data.data.push_back(p);
2099  }
2100  insert_data.numRows = 1;
2101  if (shard) {
2102  shard->fragmenter->insertData(insert_data);
2103  } else {
2104  table_descriptor->fragmenter->insertData(insert_data);
2105  }
2106 
2107  auto rs = std::make_shared<ResultSet>(TargetInfoList{},
2110  executor_->getRowSetMemoryOwner(),
2111  executor_);
2112  std::vector<TargetMetaInfo> empty_targets;
2113  return {rs, empty_targets};
2114 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
HOST DEVICE int get_size() const
Definition: sqltypes.h:258
Definition: sqltypes.h:50
std::vector< std::string > * stringsPtr
Definition: sqltypes.h:139
std::vector< ArrayDatum > * arraysPtr
Definition: sqltypes.h:140
std::vector< const TableDescriptor * > getPhysicalTablesDescriptors(const TableDescriptor *logicalTableDesc) const
Definition: Catalog.cpp:3287
int64_t insert_one_dict_str(T *col_data, const std::string &columnName, const SQLTypeInfo &columnType, const Analyzer::Constant *col_cv, const Catalog_Namespace::Catalog &catalog)
Definition: sqldefs.h:49
std::vector< TargetInfo > TargetInfoList
HOST DEVICE SQLTypes get_type() const
Definition: sqltypes.h:248
std::string to_string(char const *&&v)
int tableId
identifies the database into which the data is being inserted
Definition: Fragmenter.h:61
size_t numRows
a vector of column ids for the row(s) being inserted
Definition: Fragmenter.h:63
CHECK(cgen_state)
const DBMetadata & getCurrentDB() const
Definition: Catalog.h:182
void * checked_malloc(const size_t size)
Definition: checked_alloc.h:40
void put_null_array(void *ndptr, const SQLTypeInfo &ntype, const std::string col_name)
const Catalog_Namespace::Catalog & cat_
const DictDescriptor * getMetadataForDict(int dict_ref, bool loadDict=true) const
Definition: Catalog.cpp:1444
void put_null(void *ndptr, const SQLTypeInfo &ntype, const std::string col_name)
Definition: sqltypes.h:53
Definition: sqltypes.h:54
int8_t * appendDatum(int8_t *buf, Datum d, const SQLTypeInfo &ti)
Definition: sqltypes.h:842
HOST DEVICE EncodingType get_compression() const
Definition: sqltypes.h:256
bool is_null(const T &v, const SQLTypeInfo &t)
std::vector< DataBlockPtr > data
the number of rows being inserted
Definition: Fragmenter.h:64
Definition: sqltypes.h:42
int64_t inline_fixed_encoding_null_val(const SQL_TYPE_INFO &ti)
The data to be inserted using the fragment manager.
Definition: Fragmenter.h:59
Definition: sqltypes.h:46
const TableDescriptor * getMetadataForTable(const std::string &tableName, const bool populateFragmenter=true) const
Returns a pointer to a const TableDescriptor struct matching the provided tableName.
bool is_string() const
Definition: sqltypes.h:399
specifies the content in-memory of a row in the table metadata table
int8_t * numbersPtr
Definition: sqltypes.h:138
SQLTypeInfo get_elem_type() const
Definition: sqltypes.h:603
std::vector< int > columnIds
identifies the table into which the data is being inserted
Definition: Fragmenter.h:62
Executor * executor_
#define SHARD_FOR_KEY(key, num_shards)
Definition: shard_key.h:20
std::conditional_t< isCudaCC(), DeviceArrayDatum, HostArrayDatum > ArrayDatum
Definition: sqltypes.h:120
const ColumnDescriptor * get_column_descriptor(const int col_id, const int table_id, const Catalog_Namespace::Catalog &cat)
Definition: Execute.h:142

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeSort ( const RelSort sort,
const CompilationOptions co,
const ExecutionOptions eo,
RenderInfo render_info,
const int64_t  queue_time_ms 
)
private

Definition at line 2146 of file RelAlgExecutor.cpp.

References SpeculativeTopNBlacklist::add(), GroupByAndAggregate::addTransientStringLiterals(), ExecutionOptions::allow_loop_joins, ExecutionOptions::allow_multifrag, ExecutionOptions::allow_runtime_query_interrupt, CHECK(), CHECK_EQ, anonymous_namespace{RelAlgExecutor.cpp}::check_sort_node_source_constraint(), RelSort::collationCount(), createSortInputWorkUnit(), DEBUG_TIMER, CompilationOptions::device_type, ExecutionOptions::dynamic_watchdog_time_limit, executeWorkUnit(), executor_, ExecutionOptions::executor_type, ExecutionOptions::find_push_down_candidates, anonymous_namespace{RelAlgExecutor.cpp}::first_oe_is_desc(), g_cluster, anonymous_namespace{RelAlgExecutor.cpp}::get_order_entries(), RelAlgNode::getId(), RelAlgNode::getInput(), RelSort::getLimit(), RelSort::getOffset(), ExecutionResult::getRows(), ExecutionOptions::gpu_input_mem_limit_percent, RelSort::isEmptyResult(), ExecutionOptions::jit_debug, ExecutionOptions::just_calcite_explain, ExecutionOptions::just_explain, ExecutionOptions::just_validate, leaf_results_, anonymous_namespace{RelAlgExecutor.cpp}::node_is_aggregate(), ExecutionOptions::output_columnar_hint, run_benchmark_import::result, ExecutionOptions::runtime_query_interrupt_frequency, RelAlgNode::setOutputMetainfo(), speculative_topn_blacklist_, use_speculative_top_n(), ExecutionOptions::with_dynamic_watchdog, and ExecutionOptions::with_watchdog.

Referenced by executeRelAlgStep().

2150  {
2151  auto timer = DEBUG_TIMER(__func__);
2153  const auto source = sort->getInput(0);
2154  const bool is_aggregate = node_is_aggregate(source);
2155  auto it = leaf_results_.find(sort->getId());
2156  if (it != leaf_results_.end()) {
2157  // Add any transient string literals to the sdp on the agg
2158  const auto source_work_unit = createSortInputWorkUnit(sort, eo);
2160  source_work_unit.exe_unit, executor_, executor_->row_set_mem_owner_);
2161  // Handle push-down for LIMIT for multi-node
2162  auto& aggregated_result = it->second;
2163  auto& result_rows = aggregated_result.rs;
2164  const size_t limit = sort->getLimit();
2165  const size_t offset = sort->getOffset();
2166  const auto order_entries = get_order_entries(sort);
2167  if (limit || offset) {
2168  if (!order_entries.empty()) {
2169  result_rows->sort(order_entries, limit + offset);
2170  }
2171  result_rows->dropFirstN(offset);
2172  if (limit) {
2173  result_rows->keepFirstN(limit);
2174  }
2175  }
2176  ExecutionResult result(result_rows, aggregated_result.targets_meta);
2177  sort->setOutputMetainfo(aggregated_result.targets_meta);
2178  return result;
2179  }
2180  while (true) {
2181  std::list<std::shared_ptr<Analyzer::Expr>> groupby_exprs;
2182  bool is_desc{false};
2183  try {
2184  const auto source_work_unit = createSortInputWorkUnit(sort, eo);
2185  is_desc = first_oe_is_desc(source_work_unit.exe_unit.sort_info.order_entries);
2186  ExecutionOptions eo_copy = {
2188  eo.allow_multifrag,
2189  eo.just_explain,
2190  eo.allow_loop_joins,
2191  eo.with_watchdog,
2192  eo.jit_debug,
2193  eo.just_validate || sort->isEmptyResult(),
2201  eo.executor_type,
2202  };
2203 
2204  groupby_exprs = source_work_unit.exe_unit.groupby_exprs;
2205  auto source_result = executeWorkUnit(source_work_unit,
2206  source->getOutputMetainfo(),
2207  is_aggregate,
2208  co,
2209  eo_copy,
2210  render_info,
2211  queue_time_ms);
2212  if (render_info && render_info->isPotentialInSituRender()) {
2213  return source_result;
2214  }
2215  if (source_result.isFilterPushDownEnabled()) {
2216  return source_result;
2217  }
2218  auto rows_to_sort = source_result.getRows();
2219  if (eo.just_explain) {
2220  return {rows_to_sort, {}};
2221  }
2222  const size_t limit = sort->getLimit();
2223  const size_t offset = sort->getOffset();
2224  if (sort->collationCount() != 0 && !rows_to_sort->definitelyHasNoRows() &&
2225  !use_speculative_top_n(source_work_unit.exe_unit,
2226  rows_to_sort->getQueryMemDesc())) {
2227  rows_to_sort->sort(source_work_unit.exe_unit.sort_info.order_entries,
2228  limit + offset);
2229  }
2230  if (limit || offset) {
2231  if (g_cluster && sort->collationCount() == 0) {
2232  if (offset >= rows_to_sort->rowCount()) {
2233  rows_to_sort->dropFirstN(offset);
2234  } else {
2235  rows_to_sort->keepFirstN(limit + offset);
2236  }
2237  } else {
2238  rows_to_sort->dropFirstN(offset);
2239  if (limit) {
2240  rows_to_sort->keepFirstN(limit);
2241  }
2242  }
2243  }
2244  return {rows_to_sort, source_result.getTargetsMeta()};
2245  } catch (const SpeculativeTopNFailed&) {
2246  CHECK_EQ(size_t(1), groupby_exprs.size());
2247  speculative_topn_blacklist_.add(groupby_exprs.front(), is_desc);
2248  }
2249  }
2250  CHECK(false);
2251  return {std::make_shared<ResultSet>(std::vector<TargetInfo>{},
2252  co.device_type,
2254  nullptr,
2255  executor_),
2256  {}};
2257 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
size_t getOffset() const
const unsigned runtime_query_interrupt_frequency
static SpeculativeTopNBlacklist speculative_topn_blacklist_
bool use_speculative_top_n(const RelAlgExecutionUnit &ra_exe_unit, const QueryMemoryDescriptor &query_mem_desc)
void check_sort_node_source_constraint(const RelSort *sort)
std::unordered_map< unsigned, AggregatedResult > leaf_results_
const bool allow_multifrag
const bool find_push_down_candidates
CHECK(cgen_state)
const bool just_validate
unsigned getId() const
ExecutorType executor_type
const bool with_dynamic_watchdog
const double gpu_input_mem_limit_percent
const std::shared_ptr< ResultSet > & getRows() const
const RelAlgNode * getInput(const size_t idx) const
bool node_is_aggregate(const RelAlgNode *ra)
bool isEmptyResult() const
const bool just_calcite_explain
size_t collationCount() const
size_t getLimit() const
const bool allow_loop_joins
ExecutionResult executeWorkUnit(const WorkUnit &work_unit, const std::vector< TargetMetaInfo > &targets_meta, const bool is_agg, const CompilationOptions &co_in, const ExecutionOptions &eo, RenderInfo *, const int64_t queue_time_ms, const ssize_t previous_count=-1)
#define DEBUG_TIMER(name)
Definition: Logger.h:313
void setOutputMetainfo(const std::vector< TargetMetaInfo > &targets_metainfo) const
std::list< Analyzer::OrderEntry > get_order_entries(const RelSort *sort)
bool g_cluster
void add(const std::shared_ptr< Analyzer::Expr > expr, const bool desc)
const bool allow_runtime_query_interrupt
const unsigned dynamic_watchdog_time_limit
Executor * executor_
WorkUnit createSortInputWorkUnit(const RelSort *, const ExecutionOptions &eo)
const bool with_watchdog
bool first_oe_is_desc(const std::list< Analyzer::OrderEntry > &order_entries)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeTableFunction ( const RelTableFunction table_func,
const CompilationOptions co_in,
const ExecutionOptions eo,
const int64_t  queue_time_ms 
)
private

Definition at line 1500 of file RelAlgExecutor.cpp.

References cat_, CHECK(), createTableFunctionWorkUnit(), DEBUG_TIMER, CompilationOptions::device_type, Executor::ERR_OUT_OF_GPU_MEM, executor_, g_cluster, g_enable_table_functions, get_table_infos(), QueryExecutionError::getErrorCode(), handlePersistentError(), INJECT_TIMER, ExecutionOptions::just_explain, and run_benchmark_import::result.

Referenced by executeRelAlgStep().

1503  {
1505  auto timer = DEBUG_TIMER(__func__);
1506 
1507  auto co = co_in;
1508 
1509  if (g_cluster) {
1510  throw std::runtime_error("Table functions not supported in distributed mode yet");
1511  }
1512  if (!g_enable_table_functions) {
1513  throw std::runtime_error("Table function support is disabled");
1514  }
1515 
1516  auto table_func_work_unit = createTableFunctionWorkUnit(table_func, eo.just_explain);
1517  const auto body = table_func_work_unit.body;
1518  CHECK(body);
1519 
1520  const auto table_infos =
1521  get_table_infos(table_func_work_unit.exe_unit.input_descs, executor_);
1522 
1523  ExecutionResult result{std::make_shared<ResultSet>(std::vector<TargetInfo>{},
1524  co.device_type,
1526  nullptr,
1527  executor_),
1528  {}};
1529 
1530  try {
1531  result = {executor_->executeTableFunction(
1532  table_func_work_unit.exe_unit, table_infos, co, eo, cat_),
1533  body->getOutputMetainfo()};
1534  } catch (const QueryExecutionError& e) {
1537  throw std::runtime_error("Table function ran out of memory during execution");
1538  }
1539  result.setQueueTime(queue_time_ms);
1540  return result;
1541 }
int32_t getErrorCode() const
Definition: ErrorHandling.h:55
TableFunctionWorkUnit createTableFunctionWorkUnit(const RelTableFunction *table_func, const bool just_explain)
CHECK(cgen_state)
#define INJECT_TIMER(DESC)
Definition: measure.h:91
const Catalog_Namespace::Catalog & cat_
static const int32_t ERR_OUT_OF_GPU_MEM
Definition: Execute.h:980
ExecutionResult executeTableFunction(const RelTableFunction *, const CompilationOptions &, const ExecutionOptions &, const int64_t queue_time_ms)
bool g_enable_table_functions
Definition: Execute.cpp:95
static void handlePersistentError(const int32_t error_code)
std::vector< InputTableInfo > get_table_infos(const std::vector< InputDescriptor > &input_descs, Executor *executor)
#define DEBUG_TIMER(name)
Definition: Logger.h:313
bool g_cluster
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void RelAlgExecutor::executeUpdate ( const RelAlgNode node,
const CompilationOptions co,
const ExecutionOptions eo,
const int64_t  queue_time_ms 
)
private

Definition at line 1235 of file RelAlgExecutor.cpp.

References CompilationOptions::allow_lazy_fetch, cat_, CHECK(), CHECK_EQ, createCompoundWorkUnit(), createProjectWorkUnit(), DEBUG_TIMER, Default, executor_, get_table_infos(), get_temporary_table(), Catalog_Namespace::Catalog::getMetadataForColumn(), CompilationOptions::hoist_literals, CacheInvalidator< CACHE_HOLDING_TYPES >::invalidateCaches(), CompilationOptions::makeCpuOnly(), ExecutionOptions::output_columnar_hint, temporary_tables_, and StorageIOFacility< RelAlgExecutorTraits >::yieldUpdateCallback().

Referenced by executeRelAlgStep().

1238  {
1239  CHECK(node);
1240  auto timer = DEBUG_TIMER(__func__);
1241 
1243 
1244  auto co = co_in;
1245  co.hoist_literals = false; // disable literal hoisting as it interferes with dict
1246  // encoded string updates
1247 
1248  auto execute_update_for_node =
1249  [this, &co, &eo_in](const auto node, auto& work_unit, const bool is_aggregate) {
1250  UpdateTransactionParameters update_params(node->getModifiedTableDescriptor(),
1251  node->getTargetColumns(),
1252  node->getOutputMetainfo(),
1253  node->isVarlenUpdateRequired());
1254 
1255  const auto table_infos = get_table_infos(work_unit.exe_unit, executor_);
1256 
1257  auto execute_update_ra_exe_unit =
1258  [this, &co, &eo_in, &table_infos, &update_params](
1259  const RelAlgExecutionUnit& ra_exe_unit, const bool is_aggregate) {
1261 
1262  auto eo = eo_in;
1263  if (update_params.tableIsTemporary()) {
1264  eo.output_columnar_hint = true;
1265  co_project.allow_lazy_fetch = false;
1266  }
1267 
1268  auto update_callback = yieldUpdateCallback(update_params);
1269  executor_->executeUpdate(ra_exe_unit,
1270  table_infos,
1271  co_project,
1272  eo,
1273  cat_,
1274  executor_->row_set_mem_owner_,
1275  update_callback,
1276  is_aggregate);
1277  update_params.finalizeTransaction();
1278  };
1279 
1280  if (update_params.tableIsTemporary()) {
1281  // hold owned target exprs during execution if rewriting
1282  auto query_rewrite = std::make_unique<QueryRewriter>(table_infos, executor_);
1283  // rewrite temp table updates to generate the full column by moving the where
1284  // clause into a case if such a rewrite is not possible, bail on the update
1285  // operation build an expr for the update target
1286  const auto td = update_params.getTableDescriptor();
1287  CHECK(td);
1288  const auto update_column_names = update_params.getUpdateColumnNames();
1289  if (update_column_names.size() > 1) {
1290  throw std::runtime_error(
1291  "Multi-column update is not yet supported for temporary tables.");
1292  }
1293 
1294  auto cd = cat_.getMetadataForColumn(td->tableId, update_column_names.front());
1295  CHECK(cd);
1296  auto projected_column_to_update =
1297  makeExpr<Analyzer::ColumnVar>(cd->columnType, td->tableId, cd->columnId, 0);
1298  const auto rewritten_exe_unit = query_rewrite->rewriteColumnarUpdate(
1299  work_unit.exe_unit, projected_column_to_update);
1300  if (rewritten_exe_unit.target_exprs.front()->get_type_info().is_varlen()) {
1301  throw std::runtime_error(
1302  "Variable length updates not yet supported on temporary tables.");
1303  }
1304  execute_update_ra_exe_unit(rewritten_exe_unit, is_aggregate);
1305  } else {
1306  execute_update_ra_exe_unit(work_unit.exe_unit, is_aggregate);
1307  }
1308  };
1309 
1310  if (auto compound = dynamic_cast<const RelCompound*>(node)) {
1311  auto work_unit =
1312  createCompoundWorkUnit(compound, {{}, SortAlgorithm::Default, 0, 0}, eo_in);
1313 
1314  execute_update_for_node(compound, work_unit, compound->isAggregate());
1315  } else if (auto project = dynamic_cast<const RelProject*>(node)) {
1316  auto work_unit =
1317  createProjectWorkUnit(project, {{}, SortAlgorithm::Default, 0, 0}, eo_in);
1318 
1319  if (project->isSimple()) {
1320  CHECK_EQ(size_t(1), project->inputCount());
1321  const auto input_ra = project->getInput(0);
1322  if (dynamic_cast<const RelSort*>(input_ra)) {
1323  const auto& input_table =
1324  get_temporary_table(&temporary_tables_, -input_ra->getId());
1325  CHECK(input_table);
1326  work_unit.exe_unit.scan_limit = input_table->rowCount();
1327  }
1328  }
1329 
1330  execute_update_for_node(project, work_unit, false);
1331  } else {
1332  throw std::runtime_error("Unsupported parent node for update: " + node->toString());
1333  }
1334 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
UpdateCallback yieldUpdateCallback(UpdateTransactionParameters &update_parameters)
WorkUnit createProjectWorkUnit(const RelProject *, const SortInfo &, const ExecutionOptions &eo)
TemporaryTables temporary_tables_
static void invalidateCaches()
WorkUnit createCompoundWorkUnit(const RelCompound *, const SortInfo &, const ExecutionOptions &eo)
const ResultSetPtr & get_temporary_table(const TemporaryTables *temporary_tables, const int table_id)
Definition: Execute.h:179
static CompilationOptions makeCpuOnly(const CompilationOptions &in)
CHECK(cgen_state)
const Catalog_Namespace::Catalog & cat_
const ColumnDescriptor * getMetadataForColumn(int tableId, const std::string &colName) const
std::vector< InputTableInfo > get_table_infos(const std::vector< InputDescriptor > &input_descs, Executor *executor)
#define DEBUG_TIMER(name)
Definition: Logger.h:313
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeWorkUnit ( const WorkUnit work_unit,
const std::vector< TargetMetaInfo > &  targets_meta,
const bool  is_agg,
const CompilationOptions co_in,
const ExecutionOptions eo,
RenderInfo render_info,
const int64_t  queue_time_ms,
const ssize_t  previous_count = -1 
)
private

Definition at line 2445 of file RelAlgExecutor.cpp.

References GroupByAndAggregate::addTransientStringLiterals(), CompilationOptions::allow_lazy_fetch, RelAlgExecutor::WorkUnit::body, anonymous_namespace{RelAlgExecutor.cpp}::build_render_targets(), anonymous_namespace{RelAlgExecutor.cpp}::can_use_bump_allocator(), cat_, CHECK(), CHECK_EQ, CHECK_GT, anonymous_namespace{RelAlgExecutor.cpp}::compute_output_buffer_size(), computeWindow(), CPU, DEBUG_TIMER, anonymous_namespace{RelAlgExecutor.cpp}::decide_approx_count_distinct_implementation(), CompilationOptions::device_type, RelAlgExecutor::WorkUnit::exe_unit, anonymous_namespace{RelAlgExecutor.cpp}::exe_unit_has_quals(), executor_, ExecutionOptions::executor_type, Extern, ExecutionOptions::find_push_down_candidates, g_big_group_threshold, g_enable_window_functions, get_table_infos(), QueryExecutionError::getErrorCode(), getFilteredCountAll(), getNDVEstimation(), anonymous_namespace{RelAlgExecutor.cpp}::groups_approx_upper_bound(), handleOutOfMemoryRetry(), handlePersistentError(), INJECT_TIMER, anonymous_namespace{RelAlgExecutor.cpp}::is_agg(), anonymous_namespace{RelAlgExecutor.cpp}::is_window_execution_unit(), RenderInfo::isPotentialInSituRender(), isRowidLookup(), ExecutionOptions::just_calcite_explain, ExecutionOptions::just_explain, leaf_results_, RelAlgExecutor::WorkUnit::max_groups_buffer_entry_guess, run_benchmark_import::result, selectFiltersToBePushedDown(), RelAlgExecutionUnit::target_exprs, target_exprs_owned_, and QueryExecutionError::wasMultifragKernelLaunch().

Referenced by executeAggregate(), executeCompound(), executeFilter(), executeProject(), and executeSort().

2453  {
2455  auto timer = DEBUG_TIMER(__func__);
2456 
2457  auto co = co_in;
2458  ColumnCacheMap column_cache;
2459  if (is_window_execution_unit(work_unit.exe_unit)) {
2461  throw std::runtime_error("Window functions support is disabled");
2462  }
2463  co.device_type = ExecutorDeviceType::CPU;
2464  computeWindow(work_unit.exe_unit, co, eo, column_cache, queue_time_ms);
2465  }
2466  if (!eo.just_explain && eo.find_push_down_candidates) {
2467  // find potential candidates:
2468  auto selected_filters = selectFiltersToBePushedDown(work_unit, co, eo);
2469  if (!selected_filters.empty() || eo.just_calcite_explain) {
2470  return ExecutionResult(selected_filters, eo.find_push_down_candidates);
2471  }
2472  }
2473  if (render_info && render_info->isPotentialInSituRender()) {
2474  co.allow_lazy_fetch = false;
2475  }
2476  const auto body = work_unit.body;
2477  CHECK(body);
2478  auto it = leaf_results_.find(body->getId());
2479  if (it != leaf_results_.end()) {
2481  work_unit.exe_unit, executor_, executor_->row_set_mem_owner_);
2482  auto& aggregated_result = it->second;
2483  auto& result_rows = aggregated_result.rs;
2484  ExecutionResult result(result_rows, aggregated_result.targets_meta);
2485  body->setOutputMetainfo(aggregated_result.targets_meta);
2486  if (render_info) {
2487  build_render_targets(*render_info, work_unit.exe_unit.target_exprs, targets_meta);
2488  }
2489  return result;
2490  }
2491  const auto table_infos = get_table_infos(work_unit.exe_unit, executor_);
2492 
2494  work_unit.exe_unit, table_infos, executor_, co.device_type, target_exprs_owned_);
2495  auto max_groups_buffer_entry_guess = work_unit.max_groups_buffer_entry_guess;
2496  if (is_window_execution_unit(ra_exe_unit)) {
2497  CHECK_EQ(table_infos.size(), size_t(1));
2498  CHECK_EQ(table_infos.front().info.fragments.size(), size_t(1));
2499  max_groups_buffer_entry_guess =
2500  table_infos.front().info.fragments.front().getNumTuples();
2501  ra_exe_unit.scan_limit = max_groups_buffer_entry_guess;
2502  } else if (compute_output_buffer_size(ra_exe_unit) && !isRowidLookup(work_unit)) {
2503  if (previous_count > 0 && !exe_unit_has_quals(ra_exe_unit)) {
2504  ra_exe_unit.scan_limit = static_cast<size_t>(previous_count);
2505  } else {
2506  // TODO(adb): enable bump allocator path for render queries
2507  if (can_use_bump_allocator(ra_exe_unit, co, eo) && !render_info) {
2508  ra_exe_unit.scan_limit = 0;
2509  ra_exe_unit.use_bump_allocator = true;
2510  } else if (eo.executor_type == ::ExecutorType::Extern) {
2511  ra_exe_unit.scan_limit = 0;
2512  } else if (!eo.just_explain) {
2513  const auto filter_count_all = getFilteredCountAll(work_unit, true, co, eo);
2514  if (filter_count_all >= 0) {
2515  ra_exe_unit.scan_limit = std::max(filter_count_all, ssize_t(1));
2516  }
2517  }
2518  }
2519  }
2520 
2521  ExecutionResult result{std::make_shared<ResultSet>(std::vector<TargetInfo>{},
2522  co.device_type,
2524  nullptr,
2525  executor_),
2526  {}};
2527 
2528  auto execute_and_handle_errors =
2529  [&](const auto max_groups_buffer_entry_guess_in,
2530  const bool has_cardinality_estimation) -> ExecutionResult {
2531  // Note that the groups buffer entry guess may be modified during query execution.
2532  // Create a local copy so we can track those changes if we need to attempt a retry
2533  // due to OOM
2534  auto local_groups_buffer_entry_guess = max_groups_buffer_entry_guess_in;
2535  try {
2536  return {executor_->executeWorkUnit(local_groups_buffer_entry_guess,
2537  is_agg,
2538  table_infos,
2539  ra_exe_unit,
2540  co,
2541  eo,
2542  cat_,
2543  executor_->row_set_mem_owner_,
2544  render_info,
2545  has_cardinality_estimation,
2546  column_cache),
2547  targets_meta};
2548  } catch (const QueryExecutionError& e) {
2550  return handleOutOfMemoryRetry(
2551  {ra_exe_unit, work_unit.body, local_groups_buffer_entry_guess},
2552  targets_meta,
2553  is_agg,
2554  co,
2555  eo,
2556  render_info,
2558  queue_time_ms);
2559  }
2560  };
2561 
2562  try {
2563  result = execute_and_handle_errors(
2564  max_groups_buffer_entry_guess,
2566  } catch (const CardinalityEstimationRequired&) {
2567  const auto estimated_groups_buffer_entry_guess =
2568  2 * std::min(groups_approx_upper_bound(table_infos),
2569  getNDVEstimation(work_unit, is_agg, co, eo));
2570  CHECK_GT(estimated_groups_buffer_entry_guess, size_t(0));
2571  result = execute_and_handle_errors(estimated_groups_buffer_entry_guess, true);
2572  }
2573 
2574  result.setQueueTime(queue_time_ms);
2575  if (render_info) {
2576  build_render_targets(*render_info, work_unit.exe_unit.target_exprs, targets_meta);
2577  if (render_info->isPotentialInSituRender()) {
2578  // return an empty result (with the same queue time, and zero render time)
2579  return {std::make_shared<ResultSet>(
2580  queue_time_ms,
2581  0,
2582  executor_->row_set_mem_owner_
2583  ? executor_->row_set_mem_owner_->cloneStrDictDataOnly()
2584  : nullptr),
2585  {}};
2586  }
2587  }
2588  return result;
2589 }
bool is_agg(const Analyzer::Expr *expr)
#define CHECK_EQ(x, y)
Definition: Logger.h:205
int32_t getErrorCode() const
Definition: ErrorHandling.h:55
bool can_use_bump_allocator(const RelAlgExecutionUnit &ra_exe_unit, const CompilationOptions &co, const ExecutionOptions &eo)
std::vector< PushedDownFilterInfo > selectFiltersToBePushedDown(const RelAlgExecutor::WorkUnit &work_unit, const CompilationOptions &co, const ExecutionOptions &eo)
std::unordered_map< unsigned, AggregatedResult > leaf_results_
size_t getNDVEstimation(const WorkUnit &work_unit, const bool is_agg, const CompilationOptions &co, const ExecutionOptions &eo)
#define CHECK_GT(x, y)
Definition: Logger.h:209
bool is_window_execution_unit(const RelAlgExecutionUnit &ra_exe_unit)
void computeWindow(const RelAlgExecutionUnit &ra_exe_unit, const CompilationOptions &co, const ExecutionOptions &eo, ColumnCacheMap &column_cache_map, const int64_t queue_time_ms)
std::vector< std::shared_ptr< Analyzer::Expr > > target_exprs_owned_
const bool find_push_down_candidates
CHECK(cgen_state)
ExecutorType executor_type
#define INJECT_TIMER(DESC)
Definition: measure.h:91
const Catalog_Namespace::Catalog & cat_
size_t g_big_group_threshold
Definition: Execute.cpp:93
ExecutionResult handleOutOfMemoryRetry(const RelAlgExecutor::WorkUnit &work_unit, const std::vector< TargetMetaInfo > &targets_meta, const bool is_agg, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const bool was_multifrag_kernel_launch, const int64_t queue_time_ms)
size_t groups_approx_upper_bound(const std::vector< InputTableInfo > &table_infos)
void build_render_targets(RenderInfo &render_info, const std::vector< Analyzer::Expr * > &work_unit_target_exprs, const std::vector< TargetMetaInfo > &targets_meta)
std::unordered_map< int, std::unordered_map< int, std::shared_ptr< const ColumnarResults >>> ColumnCacheMap
bool g_enable_window_functions
Definition: Execute.cpp:94
ssize_t getFilteredCountAll(const WorkUnit &work_unit, const bool is_agg, const CompilationOptions &co, const ExecutionOptions &eo)
const bool just_calcite_explain
bool isRowidLookup(const WorkUnit &work_unit)
bool exe_unit_has_quals(const RelAlgExecutionUnit ra_exe_unit)
static void handlePersistentError(const int32_t error_code)
bool compute_output_buffer_size(const RelAlgExecutionUnit &ra_exe_unit)
bool wasMultifragKernelLaunch() const
Definition: ErrorHandling.h:57
ExecutionResult executeWorkUnit(const WorkUnit &work_unit, const std::vector< TargetMetaInfo > &targets_meta, const bool is_agg, const CompilationOptions &co_in, const ExecutionOptions &eo, RenderInfo *, const int64_t queue_time_ms, const ssize_t previous_count=-1)
bool isPotentialInSituRender() const
Definition: RenderInfo.cpp:61
std::vector< InputTableInfo > get_table_infos(const std::vector< InputDescriptor > &input_descs, Executor *executor)
#define DEBUG_TIMER(name)
Definition: Logger.h:313
RelAlgExecutionUnit decide_approx_count_distinct_implementation(const RelAlgExecutionUnit &ra_exe_unit_in, const std::vector< InputTableInfo > &table_infos, const Executor *executor, const ExecutorDeviceType device_type_in, std::vector< std::shared_ptr< Analyzer::Expr >> &target_exprs_owned)
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

std::string RelAlgExecutor::getErrorMessageFromCode ( const int32_t  error_code)
static

Definition at line 2817 of file RelAlgExecutor.cpp.

References Executor::ERR_COLUMNAR_CONVERSION_NOT_SUPPORTED, Executor::ERR_DIV_BY_ZERO, Executor::ERR_INTERRUPTED, Executor::ERR_OUT_OF_CPU_MEM, Executor::ERR_OUT_OF_GPU_MEM, Executor::ERR_OUT_OF_RENDER_MEM, Executor::ERR_OUT_OF_TIME, Executor::ERR_OVERFLOW_OR_UNDERFLOW, Executor::ERR_SINGLE_VALUE_FOUND_MULTIPLE_VALUES, Executor::ERR_STREAMING_TOP_N_NOT_SUPPORTED_IN_RENDER_QUERY, Executor::ERR_STRING_CONST_IN_RESULTSET, Executor::ERR_TOO_MANY_LITERALS, Executor::ERR_UNSUPPORTED_SELF_JOIN, and to_string().

Referenced by getNDVEstimation(), and handlePersistentError().

2817  {
2818  if (error_code < 0) {
2819  return "Ran out of slots in the query output buffer";
2820  }
2821  switch (error_code) {
2823  return "Division by zero";
2825  return "Query couldn't keep the entire working set of columns in GPU memory";
2827  return "Self joins not supported yet";
2829  return "Not enough host memory to execute the query";
2831  return "Overflow or underflow";
2833  return "Query execution has exceeded the time limit";
2835  return "Query execution has been interrupted";
2837  return "Columnar conversion not supported for variable length types";
2839  return "Too many literals in the query";
2841  return "NONE ENCODED String types are not supported as input result set.";
2843  return "Not enough OpenGL memory to render the query results";
2845  return "Streaming-Top-N not supported in Render Query";
2847  return "Multiple distinct values encountered";
2848  }
2849  return "Other error: code " + std::to_string(error_code);
2850 }
static const int32_t ERR_INTERRUPTED
Definition: Execute.h:988
static const int32_t ERR_TOO_MANY_LITERALS
Definition: Execute.h:990
std::string to_string(char const *&&v)
static const int32_t ERR_STRING_CONST_IN_RESULTSET
Definition: Execute.h:991
static const int32_t ERR_STREAMING_TOP_N_NOT_SUPPORTED_IN_RENDER_QUERY
Definition: Execute.h:992
static const int32_t ERR_COLUMNAR_CONVERSION_NOT_SUPPORTED
Definition: Execute.h:989
static const int32_t ERR_DIV_BY_ZERO
Definition: Execute.h:979
static const int32_t ERR_OUT_OF_RENDER_MEM
Definition: Execute.h:983
const int8_t const int64_t const uint64_t const int32_t const int64_t int64_t uint32_t const int64_t int32_t * error_code
static const int32_t ERR_OVERFLOW_OR_UNDERFLOW
Definition: Execute.h:985
static const int32_t ERR_OUT_OF_TIME
Definition: Execute.h:987
static const int32_t ERR_UNSUPPORTED_SELF_JOIN
Definition: Execute.h:982
static const int32_t ERR_SINGLE_VALUE_FOUND_MULTIPLE_VALUES
Definition: Execute.h:993
static const int32_t ERR_OUT_OF_GPU_MEM
Definition: Execute.h:980
static const int32_t ERR_OUT_OF_CPU_MEM
Definition: Execute.h:984

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

Executor * RelAlgExecutor::getExecutor ( ) const

Definition at line 324 of file RelAlgExecutor.cpp.

References executor_.

324  {
325  return executor_;
326 }
Executor * executor_
ssize_t RelAlgExecutor::getFilteredCountAll ( const WorkUnit work_unit,
const bool  is_agg,
const CompilationOptions co,
const ExecutionOptions eo 
)
private

Definition at line 2591 of file RelAlgExecutor.cpp.

References cat_, CHECK(), CHECK_EQ, CHECK_GE, create_count_all_execution_unit(), RelAlgExecutor::WorkUnit::exe_unit, executor_, g_bigint_count, get_table_infos(), kBIGINT, kCOUNT, kINT, LOG, and logger::WARNING.

Referenced by executeWorkUnit().

2594  {
2595  const auto count =
2596  makeExpr<Analyzer::AggExpr>(SQLTypeInfo(g_bigint_count ? kBIGINT : kINT, false),
2597  kCOUNT,
2598  nullptr,
2599  false,
2600  nullptr);
2601  const auto count_all_exe_unit =
2602  create_count_all_execution_unit(work_unit.exe_unit, count);
2603  size_t one{1};
2604  ResultSetPtr count_all_result;
2605  try {
2606  ColumnCacheMap column_cache;
2607  count_all_result =
2608  executor_->executeWorkUnit(one,
2609  is_agg,
2610  get_table_infos(work_unit.exe_unit, executor_),
2611  count_all_exe_unit,
2612  co,
2613  eo,
2614  cat_,
2615  executor_->row_set_mem_owner_,
2616  nullptr,
2617  false,
2618  column_cache);
2619  } catch (const std::exception& e) {
2620  LOG(WARNING) << "Failed to run pre-flight filtered count with error " << e.what();
2621  return -1;
2622  }
2623  const auto count_row = count_all_result->getNextRow(false, false);
2624  CHECK_EQ(size_t(1), count_row.size());
2625  const auto& count_tv = count_row.front();
2626  const auto count_scalar_tv = boost::get<ScalarTargetValue>(&count_tv);
2627  CHECK(count_scalar_tv);
2628  const auto count_ptr = boost::get<int64_t>(count_scalar_tv);
2629  CHECK(count_ptr);
2630  CHECK_GE(*count_ptr, 0);
2631  auto count_upper_bound = static_cast<size_t>(*count_ptr);
2632  return std::max(count_upper_bound, size_t(1));
2633 }
bool is_agg(const Analyzer::Expr *expr)
#define CHECK_EQ(x, y)
Definition: Logger.h:205
#define LOG(tag)
Definition: Logger.h:188
#define CHECK_GE(x, y)
Definition: Logger.h:210
std::shared_ptr< ResultSet > ResultSetPtr
RelAlgExecutionUnit create_count_all_execution_unit(const RelAlgExecutionUnit &ra_exe_unit, std::shared_ptr< Analyzer::Expr > replacement_target)
CHECK(cgen_state)
const Catalog_Namespace::Catalog & cat_
bool g_bigint_count
std::unordered_map< int, std::unordered_map< int, std::shared_ptr< const ColumnarResults >>> ColumnCacheMap
Definition: sqldefs.h:76
std::vector< InputTableInfo > get_table_infos(const std::vector< InputDescriptor > &input_descs, Executor *executor)
Definition: sqltypes.h:46
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

FilterSelectivity RelAlgExecutor::getFilterSelectivity ( const std::vector< std::shared_ptr< Analyzer::Expr >> &  filter_expressions,
const CompilationOptions co,
const ExecutionOptions eo 
)
private

Given a set of filter expressions for a table, it launches a new COUNT query to compute the number of passing rows, and then generates a set of statistics related to those filters. Later, these stats are used to decide whether a filter should be pushed down or not.

Definition at line 57 of file JoinFilterPushDown.cpp.

References cat_, CHECK(), CHECK_EQ, Default, executor_, g_bigint_count, get_table_infos(), kBIGINT, kCOUNT, and kINT.

Referenced by selectFiltersToBePushedDown().

60  {
61  CollectInputColumnsVisitor input_columns_visitor;
62  std::list<std::shared_ptr<Analyzer::Expr>> quals;
63  std::unordered_set<InputColDescriptor> input_column_descriptors;
64  BindFilterToOutermostVisitor bind_filter_to_outermost;
65  for (const auto& filter_expr : filter_expressions) {
66  input_column_descriptors = input_columns_visitor.aggregateResult(
67  input_column_descriptors, input_columns_visitor.visit(filter_expr.get()));
68  quals.push_back(bind_filter_to_outermost.visit(filter_expr.get()));
69  }
70  std::vector<InputDescriptor> input_descs;
71  std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
72  for (const auto& input_col_desc : input_column_descriptors) {
73  if (input_descs.empty()) {
74  input_descs.push_back(input_col_desc.getScanDesc());
75  } else {
76  CHECK(input_col_desc.getScanDesc() == input_descs.front());
77  }
78  input_col_descs.push_back(std::make_shared<const InputColDescriptor>(input_col_desc));
79  }
80  const auto count_expr =
81  makeExpr<Analyzer::AggExpr>(SQLTypeInfo(g_bigint_count ? kBIGINT : kINT, false),
82  kCOUNT,
83  nullptr,
84  false,
85  nullptr);
86  RelAlgExecutionUnit ra_exe_unit{input_descs,
87  input_col_descs,
88  {},
89  quals,
90  {},
91  {},
92  {count_expr.get()},
93  nullptr,
94  {{}, SortAlgorithm::Default, 0, 0},
95  0};
96  size_t one{1};
97  ResultSetPtr filtered_result;
98  const auto table_infos = get_table_infos(input_descs, executor_);
99  CHECK_EQ(size_t(1), table_infos.size());
100  const size_t total_rows_upper_bound = table_infos.front().info.getNumTuplesUpperBound();
101  try {
102  ColumnCacheMap column_cache;
103  filtered_result = executor_->executeWorkUnit(one,
104  true,
105  table_infos,
106  ra_exe_unit,
107  co,
108  eo,
109  cat_,
110  executor_->getRowSetMemoryOwner(),
111  nullptr,
112  false,
113  column_cache);
114  } catch (...) {
115  return {false, 1.0, 0};
116  }
117  const auto count_row = filtered_result->getNextRow(false, false);
118  CHECK_EQ(size_t(1), count_row.size());
119  const auto& count_tv = count_row.front();
120  const auto count_scalar_tv = boost::get<ScalarTargetValue>(&count_tv);
121  CHECK(count_scalar_tv);
122  const auto count_ptr = boost::get<int64_t>(count_scalar_tv);
123  CHECK(count_ptr);
124  const auto rows_passing = *count_ptr;
125  const auto rows_total = std::max(total_rows_upper_bound, size_t(1));
126  return {true, static_cast<float>(rows_passing) / rows_total, total_rows_upper_bound};
127 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
std::shared_ptr< ResultSet > ResultSetPtr
CHECK(cgen_state)
const Catalog_Namespace::Catalog & cat_
bool g_bigint_count
std::unordered_map< int, std::unordered_map< int, std::shared_ptr< const ColumnarResults >>> ColumnCacheMap
Definition: sqldefs.h:76
std::vector< InputTableInfo > get_table_infos(const std::vector< InputDescriptor > &input_descs, Executor *executor)
Definition: sqltypes.h:46
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

size_t RelAlgExecutor::getNDVEstimation ( const WorkUnit work_unit,
const bool  is_agg,
const CompilationOptions co,
const ExecutionOptions eo 
)
private

Definition at line 36 of file CardinalityEstimator.cpp.

References cat_, create_ndv_execution_unit(), Executor::ERR_INTERRUPTED, Executor::ERR_OUT_OF_TIME, RelAlgExecutor::WorkUnit::exe_unit, executor_, get_table_infos(), QueryExecutionError::getErrorCode(), getErrorMessageFromCode(), and UNREACHABLE.

Referenced by executeWorkUnit().

39  {
40  const auto estimator_exe_unit = create_ndv_execution_unit(work_unit.exe_unit);
41  size_t one{1};
42  ColumnCacheMap column_cache;
43  try {
44  const auto estimator_result =
45  executor_->executeWorkUnit(one,
46  is_agg,
47  get_table_infos(work_unit.exe_unit, executor_),
48  estimator_exe_unit,
49  co,
50  eo,
51  cat_,
52  executor_->row_set_mem_owner_,
53  nullptr,
54  false,
55  column_cache);
56  if (!estimator_result) {
57  return 1;
58  }
59  return std::max(estimator_result->getNDVEstimator(), size_t(1));
60  } catch (const QueryExecutionError& e) {
62  throw std::runtime_error("Cardinality estimation query ran out of time");
63  }
65  throw std::runtime_error("Cardinality estimation query has been interrupted");
66  }
67  throw std::runtime_error("Failed to run the cardinality estimation query: " +
69  }
70  UNREACHABLE();
71  return 1;
72 }
bool is_agg(const Analyzer::Expr *expr)
int32_t getErrorCode() const
Definition: ErrorHandling.h:55
static const int32_t ERR_INTERRUPTED
Definition: Execute.h:988
#define UNREACHABLE()
Definition: Logger.h:241
RelAlgExecutionUnit create_ndv_execution_unit(const RelAlgExecutionUnit &ra_exe_unit)
const Catalog_Namespace::Catalog & cat_
static const int32_t ERR_OUT_OF_TIME
Definition: Execute.h:987
std::unordered_map< int, std::unordered_map< int, std::shared_ptr< const ColumnarResults >>> ColumnCacheMap
std::vector< InputTableInfo > get_table_infos(const std::vector< InputDescriptor > &input_descs, Executor *executor)
static std::string getErrorMessageFromCode(const int32_t error_code)
Executor * executor_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

size_t RelAlgExecutor::getOuterFragmentCount ( const CompilationOptions co,
const ExecutionOptions eo 
)

Definition at line 72 of file RelAlgExecutor.cpp.

References cat_, CHECK(), cleanupPostExecution(), createCompoundWorkUnit(), createProjectWorkUnit(), Default, executor_, ExecutionOptions::find_push_down_candidates, get_frag_count_of_table(), get_physical_inputs(), get_physical_table_inputs(), getSubqueries(), ExecutionOptions::just_explain, query_dag_, WindowProjectNodeContext::reset(), target_exprs_owned_, and temporary_tables_.

Referenced by Parser::LocalConnector::getOuterFragmentCount().

73  {
75  return 0;
76  }
77 
78  if (eo.just_explain) {
79  return 0;
80  }
81 
83 
84  query_dag_->resetQueryExecutionState();
85  const auto& ra = query_dag_->getRootNode();
86 
87  std::lock_guard<std::mutex> lock(executor_->execute_mutex_);
88  ScopeGuard row_set_holder = [this] { cleanupPostExecution(); };
89  const auto phys_inputs = get_physical_inputs(cat_, &ra);
90  const auto phys_table_ids = get_physical_table_inputs(&ra);
91  executor_->setCatalog(&cat_);
92  executor_->setupCaching(phys_inputs, phys_table_ids);
93 
94  ScopeGuard restore_metainfo_cache = [this] { executor_->clearMetaInfoCache(); };
95  auto ed_seq = RaExecutionSequence(&ra);
96 
97  if (!getSubqueries().empty()) {
98  return 0;
99  }
100 
101  CHECK(!ed_seq.empty());
102  if (ed_seq.size() > 1) {
103  return 0;
104  }
105 
106  decltype(temporary_tables_)().swap(temporary_tables_);
107  decltype(target_exprs_owned_)().swap(target_exprs_owned_);
108  executor_->catalog_ = &cat_;
109  executor_->temporary_tables_ = &temporary_tables_;
110 
112  auto exec_desc_ptr = ed_seq.getDescriptor(0);
113  CHECK(exec_desc_ptr);
114  auto& exec_desc = *exec_desc_ptr;
115  const auto body = exec_desc.getBody();
116  if (body->isNop()) {
117  return 0;
118  }
119 
120  const auto project = dynamic_cast<const RelProject*>(body);
121  if (project) {
122  auto work_unit =
123  createProjectWorkUnit(project, {{}, SortAlgorithm::Default, 0, 0}, eo);
124 
125  return get_frag_count_of_table(work_unit.exe_unit.input_descs[0].getTableId(),
126  executor_);
127  }
128 
129  const auto compound = dynamic_cast<const RelCompound*>(body);
130  if (compound) {
131  if (compound->isDeleteViaSelect()) {
132  return 0;
133  } else if (compound->isUpdateViaSelect()) {
134  return 0;
135  } else {
136  if (compound->isAggregate()) {
137  return 0;
138  }
139 
140  const auto work_unit =
141  createCompoundWorkUnit(compound, {{}, SortAlgorithm::Default, 0, 0}, eo);
142 
143  return get_frag_count_of_table(work_unit.exe_unit.input_descs[0].getTableId(),
144  executor_);
145  }
146  }
147 
148  return 0;
149 }
WorkUnit createProjectWorkUnit(const RelProject *, const SortInfo &, const ExecutionOptions &eo)
TemporaryTables temporary_tables_
const std::vector< std::shared_ptr< RexSubQuery > > & getSubqueries() const noexcept
WorkUnit createCompoundWorkUnit(const RelCompound *, const SortInfo &, const ExecutionOptions &eo)
std::vector< std::shared_ptr< Analyzer::Expr > > target_exprs_owned_
const bool find_push_down_candidates
CHECK(cgen_state)
A container for relational algebra descriptors defining the execution order for a relational algebra ...
const Catalog_Namespace::Catalog & cat_
std::unique_ptr< RelAlgDagBuilder > query_dag_
void cleanupPostExecution()
std::unordered_set< PhysicalInput > get_physical_inputs(const RelAlgNode *ra)
size_t get_frag_count_of_table(const int table_id, Executor *executor)
Executor * executor_
std::unordered_set< int > get_physical_table_inputs(const RelAlgNode *ra)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

const RelAlgNode& RelAlgExecutor::getRootRelAlgNode ( ) const
inline

Definition at line 135 of file RelAlgExecutor.h.

References CHECK(), and query_dag_.

Referenced by computeColRangesCache(), computeStringDictionaryGenerations(), and computeTableGenerations().

135  {
136  CHECK(query_dag_);
137  return query_dag_->getRootNode();
138  }
CHECK(cgen_state)
std::unique_ptr< RelAlgDagBuilder > query_dag_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

const std::vector<std::shared_ptr<RexSubQuery> >& RelAlgExecutor::getSubqueries ( ) const
inlinenoexcept

Definition at line 139 of file RelAlgExecutor.h.

References CHECK(), and query_dag_.

Referenced by executeRelAlgQueryNoRetry(), executeRelAlgQueryWithFilterPushDown(), and getOuterFragmentCount().

139  {
140  CHECK(query_dag_);
141  return query_dag_->getSubqueries();
142  };
CHECK(cgen_state)
std::unique_ptr< RelAlgDagBuilder > query_dag_

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void RelAlgExecutor::handleNop ( RaExecutionDesc ed)
private

Definition at line 616 of file RelAlgExecutor.cpp.

References addTemporaryTable(), CHECK(), CHECK_EQ, RaExecutionDesc::getBody(), and temporary_tables_.

Referenced by executeRelAlgStep().

616  {
617  // just set the result of the previous node as the result of no op
618  auto body = ed.getBody();
619  CHECK(dynamic_cast<const RelAggregate*>(body));
620  CHECK_EQ(size_t(1), body->inputCount());
621  const auto input = body->getInput(0);
622  body->setOutputMetainfo(input->getOutputMetainfo());
623  const auto it = temporary_tables_.find(-input->getId());
624  CHECK(it != temporary_tables_.end());
625  // set up temp table as it could be used by the outer query or next step
626  addTemporaryTable(-body->getId(), it->second);
627 
628  ed.setResult({it->second, input->getOutputMetainfo()});
629 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
TemporaryTables temporary_tables_
void addTemporaryTable(const int table_id, const ResultSetPtr &result)
CHECK(cgen_state)
const RelAlgNode * getBody() const

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::handleOutOfMemoryRetry ( const RelAlgExecutor::WorkUnit work_unit,
const std::vector< TargetMetaInfo > &  targets_meta,
const bool  is_agg,
const CompilationOptions co,
const ExecutionOptions eo,
RenderInfo render_info,
const bool  was_multifrag_kernel_launch,
const int64_t  queue_time_ms 
)
private

Definition at line 2670 of file RelAlgExecutor.cpp.

References ExecutionOptions::allow_loop_joins, cat_, CHECK(), anonymous_namespace{RelAlgExecutor.cpp}::decide_approx_count_distinct_implementation(), CompilationOptions::device_type, ExecutionOptions::dynamic_watchdog_time_limit, RelAlgExecutor::WorkUnit::exe_unit, executor_, g_enable_watchdog, get_table_infos(), QueryExecutionError::getErrorCode(), ExecutionOptions::gpu_input_mem_limit_percent, handlePersistentError(), ExecutionOptions::jit_debug, LOG, CompilationOptions::makeCpuOnly(), RelAlgExecutor::WorkUnit::max_groups_buffer_entry_guess, ExecutionOptions::output_columnar_hint, run_benchmark_import::result, RenderInfo::setForceNonInSituData(), target_exprs_owned_, RelAlgExecutionUnit::use_bump_allocator, VLOG, logger::WARNING, ExecutionOptions::with_dynamic_watchdog, and ExecutionOptions::with_watchdog.

Referenced by executeWorkUnit().

2678  {
2679  // Disable the bump allocator
2680  // Note that this will have basically the same affect as using the bump allocator for
2681  // the kernel per fragment path. Need to unify the max_groups_buffer_entry_guess = 0
2682  // path and the bump allocator path for kernel per fragment execution.
2683  auto ra_exe_unit_in = work_unit.exe_unit;
2684  ra_exe_unit_in.use_bump_allocator = false;
2685 
2686  auto result = ExecutionResult{std::make_shared<ResultSet>(std::vector<TargetInfo>{},
2687  co.device_type,
2689  nullptr,
2690  executor_),
2691  {}};
2692 
2693  const auto table_infos = get_table_infos(ra_exe_unit_in, executor_);
2694  auto max_groups_buffer_entry_guess = work_unit.max_groups_buffer_entry_guess;
2695  ExecutionOptions eo_no_multifrag{eo.output_columnar_hint,
2696  false,
2697  false,
2698  eo.allow_loop_joins,
2699  eo.with_watchdog,
2700  eo.jit_debug,
2701  false,
2704  false,
2705  false,
2707  false};
2708 
2709  if (was_multifrag_kernel_launch) {
2710  try {
2711  // Attempt to retry using the kernel per fragment path. The smaller input size
2712  // required may allow the entire kernel to execute in GPU memory.
2713  LOG(WARNING) << "Multifrag query ran out of memory, retrying with multifragment "
2714  "kernels disabled.";
2715  const auto ra_exe_unit = decide_approx_count_distinct_implementation(
2716  ra_exe_unit_in, table_infos, executor_, co.device_type, target_exprs_owned_);
2717  ColumnCacheMap column_cache;
2718  result = {executor_->executeWorkUnit(max_groups_buffer_entry_guess,
2719  is_agg,
2720  table_infos,
2721  ra_exe_unit,
2722  co,
2723  eo_no_multifrag,
2724  cat_,
2725  executor_->row_set_mem_owner_,
2726  nullptr,
2727  true,
2728  column_cache),
2729  targets_meta};
2730  result.setQueueTime(queue_time_ms);
2731  } catch (const QueryExecutionError& e) {
2733  LOG(WARNING) << "Kernel per fragment query ran out of memory, retrying on CPU.";
2734  }
2735  }
2736 
2737  if (render_info) {
2738  render_info->setForceNonInSituData();
2739  }
2740 
2741  const auto co_cpu = CompilationOptions::makeCpuOnly(co);
2742  // Only reset the group buffer entry guess if we ran out of slots, which
2743  // suggests a
2744  // highly pathological input which prevented a good estimation of distinct tuple
2745  // count. For projection queries, this will force a per-fragment scan limit, which is
2746  // compatible with the CPU path
2747  VLOG(1) << "Resetting max groups buffer entry guess.";
2748  max_groups_buffer_entry_guess = 0;
2749 
2750  int iteration_ctr = -1;
2751  while (true) {
2752  iteration_ctr++;
2754  ra_exe_unit_in, table_infos, executor_, co_cpu.device_type, target_exprs_owned_);
2755  ColumnCacheMap column_cache;
2756  try {
2757  result = {executor_->executeWorkUnit(max_groups_buffer_entry_guess,
2758  is_agg,
2759  table_infos,
2760  ra_exe_unit,
2761  co_cpu,
2762  eo_no_multifrag,
2763  cat_,
2764  executor_->row_set_mem_owner_,
2765  nullptr,
2766  true,
2767  column_cache),
2768  targets_meta};
2769  } catch (const QueryExecutionError& e) {
2770  // Ran out of slots
2771  if (e.getErrorCode() < 0) {
2772  // Even the conservative guess failed; it should only happen when we group
2773  // by a huge cardinality array. Maybe we should throw an exception instead?
2774  // Such a heavy query is entirely capable of exhausting all the host memory.
2775  CHECK(max_groups_buffer_entry_guess);
2776  // Only allow two iterations of increasingly large entry guesses up to a maximum
2777  // of 512MB per column per kernel
2778  if (g_enable_watchdog || iteration_ctr > 1) {
2779  throw std::runtime_error("Query ran out of output slots in the result");
2780  }
2781  max_groups_buffer_entry_guess *= 2;
2782  LOG(WARNING) << "Query ran out of slots in the output buffer, retrying with max "
2783  "groups buffer entry "
2784  "guess equal to "
2785  << max_groups_buffer_entry_guess;
2786  } else {
2788  }
2789  continue;
2790  }
2791  result.setQueueTime(queue_time_ms);
2792  return result;
2793  }
2794  return result;
2795 }
bool is_agg(const Analyzer::Expr *expr)
int32_t getErrorCode() const
Definition: ErrorHandling.h:55
void setForceNonInSituData()
Definition: RenderInfo.cpp:42
#define LOG(tag)
Definition: Logger.h:188
RelAlgExecutionUnit exe_unit
bool g_enable_watchdog
std::vector< std::shared_ptr< Analyzer::Expr > > target_exprs_owned_
static CompilationOptions makeCpuOnly(const CompilationOptions &in)
CHECK(cgen_state)
const bool with_dynamic_watchdog
const double gpu_input_mem_limit_percent
const Catalog_Namespace::Catalog & cat_
ExecutorDeviceType device_type
std::unordered_map< int, std::unordered_map< int, std::shared_ptr< const ColumnarResults >>> ColumnCacheMap
static void handlePersistentError(const int32_t error_code)
const size_t max_groups_buffer_entry_guess
const bool allow_loop_joins
std::vector< InputTableInfo > get_table_infos(const std::vector< InputDescriptor > &input_descs, Executor *executor)
RelAlgExecutionUnit decide_approx_count_distinct_implementation(const RelAlgExecutionUnit &ra_exe_unit_in, const std::vector< InputTableInfo > &table_infos, const Executor *executor, const ExecutorDeviceType device_type_in, std::vector< std::shared_ptr< Analyzer::Expr >> &target_exprs_owned)
const unsigned dynamic_watchdog_time_limit
Executor * executor_
#define VLOG(n)
Definition: Logger.h:291
const bool with_watchdog

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

void RelAlgExecutor::handlePersistentError ( const int32_t  error_code)
staticprivate

Definition at line 2797 of file RelAlgExecutor.cpp.

References Executor::ERR_OUT_OF_GPU_MEM, Executor::ERR_SPECULATIVE_TOP_OOM, logger::ERROR, g_allow_cpu_retry, getErrorMessageFromCode(), logger::INFO, and LOG.

Referenced by executeTableFunction(), executeWorkUnit(), and handleOutOfMemoryRetry().

2797  {
2798  LOG(ERROR) << "Query execution failed with error "
2801  throw SpeculativeTopNFailed();
2802  }
2804  // We ran out of GPU memory, this doesn't count as an error if the query is
2805  // allowed to continue on CPU because retry on CPU is explicitly allowed through
2806  // --allow-cpu-retry.
2807  LOG(INFO) << "Query ran out of GPU memory, attempting punt to CPU";
2808  if (!g_allow_cpu_retry) {
2809  throw std::runtime_error(
2810  "Query ran out of GPU memory, unable to automatically retry on CPU");
2811  }
2812  return;
2813  }
2814  throw std::runtime_error(getErrorMessageFromCode(error_code));
2815 }
#define LOG(tag)
Definition: Logger.h:188
const int8_t const int64_t const uint64_t const int32_t const int64_t int64_t uint32_t const int64_t int32_t * error_code
static const int32_t ERR_OUT_OF_GPU_MEM
Definition: Execute.h:980
static const int32_t ERR_SPECULATIVE_TOP_OOM
Definition: Execute.h:986
bool g_allow_cpu_retry
Definition: Execute.cpp:75
static std::string getErrorMessageFromCode(const int32_t error_code)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

bool RelAlgExecutor::isRowidLookup ( const WorkUnit work_unit)
private

Definition at line 2635 of file RelAlgExecutor.cpp.

References cat_, CHECK_EQ, RelAlgExecutor::WorkUnit::exe_unit, get_column_descriptor(), Analyzer::BinOper::get_left_operand(), RelAlgExecutionUnit::input_descs, kEQ, and TABLE.

Referenced by executeWorkUnit().

2635  {
2636  const auto& ra_exe_unit = work_unit.exe_unit;
2637  if (ra_exe_unit.input_descs.size() != 1) {
2638  return false;
2639  }
2640  const auto& table_desc = ra_exe_unit.input_descs.front();
2641  if (table_desc.getSourceType() != InputSourceType::TABLE) {
2642  return false;
2643  }
2644  const int table_id = table_desc.getTableId();
2645  for (const auto simple_qual : ra_exe_unit.simple_quals) {
2646  const auto comp_expr =
2647  std::dynamic_pointer_cast<const Analyzer::BinOper>(simple_qual);
2648  if (!comp_expr || comp_expr->get_optype() != kEQ) {
2649  return false;
2650  }
2651  const auto lhs = comp_expr->get_left_operand();
2652  const auto lhs_col = dynamic_cast<const Analyzer::ColumnVar*>(lhs);
2653  if (!lhs_col || !lhs_col->get_table_id() || lhs_col->get_rte_idx()) {
2654  return false;
2655  }
2656  const auto rhs = comp_expr->get_right_operand();
2657  const auto rhs_const = dynamic_cast<const Analyzer::Constant*>(rhs);
2658  if (!rhs_const) {
2659  return false;
2660  }
2661  auto cd = get_column_descriptor(lhs_col->get_column_id(), table_id, cat_);
2662  if (cd->isVirtualCol) {
2663  CHECK_EQ("rowid", cd->columnName);
2664  return true;
2665  }
2666  }
2667  return false;
2668 }
#define CHECK_EQ(x, y)
Definition: Logger.h:205
Definition: sqldefs.h:30
const Catalog_Namespace::Catalog & cat_
const Expr * get_left_operand() const
Definition: Analyzer.h:436
const ColumnDescriptor * get_column_descriptor(const int col_id, const int table_id, const Catalog_Namespace::Catalog &cat)
Definition: Execute.h:142

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

std::list< std::shared_ptr< Analyzer::Expr > > RelAlgExecutor::makeJoinQuals ( const RexScalar join_condition,
const std::vector< JoinType > &  join_types,
const std::unordered_map< const RelAlgNode *, int > &  input_to_nest_level,
const bool  just_explain 
) const
private

Definition at line 3198 of file RelAlgExecutor.cpp.

References cat_, combine_equi_join_conditions(), executor_, anonymous_namespace{RelAlgExecutor.cpp}::get_bitwise_equals_conjunction(), now_, qual_to_conjunctive_form(), anonymous_namespace{RelAlgExecutor.cpp}::reverse_logical_distribution(), anonymous_namespace{RelAlgExecutor.cpp}::rex_to_conjunctive_form(), and RelAlgTranslator::translateScalarRex().

Referenced by translateLeftDeepJoinFilter().

3202  {
3203  QueryFeatureDescriptor query_features;
3204  RelAlgTranslator translator(cat_,
3205  executor_,
3206  input_to_nest_level,
3207  join_types,
3208  now_,
3209  just_explain,
3210  query_features);
3211  const auto rex_condition_cf = rex_to_conjunctive_form(join_condition);
3212  std::list<std::shared_ptr<Analyzer::Expr>> join_condition_quals;
3213  for (const auto rex_condition_component : rex_condition_cf) {
3214  const auto bw_equals = get_bitwise_equals_conjunction(rex_condition_component);
3215  const auto join_condition =
3216  reverse_logical_distribution(translator.translateScalarRex(
3217  bw_equals ? bw_equals.get() : rex_condition_component));
3218  auto join_condition_cf = qual_to_conjunctive_form(join_condition);
3219  join_condition_quals.insert(join_condition_quals.end(),
3220  join_condition_cf.quals.begin(),
3221  join_condition_cf.quals.end());
3222  join_condition_quals.insert(join_condition_quals.end(),
3223  join_condition_cf.simple_quals.begin(),
3224  join_condition_cf.simple_quals.end());
3225  }
3226  return combine_equi_join_conditions(join_condition_quals);
3227 }
std::unique_ptr< const RexOperator > get_bitwise_equals_conjunction(const RexScalar *scalar)
QualsConjunctiveForm qual_to_conjunctive_form(const std::shared_ptr< Analyzer::Expr > qual_expr)
std::vector< const RexScalar * > rex_to_conjunctive_form(const RexScalar *qual_expr)
const Catalog_Namespace::Catalog & cat_
std::shared_ptr< Analyzer::Expr > reverse_logical_distribution(const std::shared_ptr< Analyzer::Expr > &expr)
Executor * executor_
std::list< std::shared_ptr< Analyzer::Expr > > combine_equi_join_conditions(const std::list< std::shared_ptr< Analyzer::Expr >> &join_quals)

+ Here is the call graph for this function: