#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, std::shared_ptr< const query_state::QueryState > query_state=nullptr)

	RelAlgExecutor (Executor *executor, const std::string &query_ra, std::shared_ptr< const query_state::QueryState > query_state=nullptr)

	RelAlgExecutor (Executor *executor, std::unique_ptr< RelAlgDag > 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, const bool explain_verbose, 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)

QueryStepExecutionResult	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)

std::unique_ptr< RelAlgDag >	getOwnedRelAlgDag ()

RelAlgDag *	getRelAlgDag ()

const RelAlgNode &	getRootRelAlgNode () const

void	prepareForeignTables ()

std::shared_ptr< const RelAlgNode >	getRootRelAlgNodeShPtr () const

std::pair< std::vector < unsigned > , std::unordered_map< unsigned, JoinQualsPerNestingLevel > >	getJoinInfo (const RelAlgNode *root_node)

std::shared_ptr< RelAlgTranslator >	getRelAlgTranslator (const RelAlgNode *root_node)

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

std::optional < RegisteredQueryHint >	getParsedQueryHint (const RelAlgNode *node)

std::optional < std::unordered_map< size_t, std::unordered_map< unsigned, RegisteredQueryHint > > >	getParsedQueryHints ()

std::optional < RegisteredQueryHint >	getGlobalQueryHint ()

ExecutionResult	executeSimpleInsert (const Analyzer::Query &insert_query, Fragmenter_Namespace::InsertDataLoader &inserter, const Catalog_Namespace::SessionInfo &session)

AggregatedColRange	computeColRangesCache ()

StringDictionaryGenerations	computeStringDictionaryGenerations ()

TableGenerations	computeTableGenerations ()

Executor *	getExecutor () const

void	cleanupPostExecution ()

void	executePostExecutionCallback ()

RaExecutionSequence	getRaExecutionSequence (const RelAlgNode root_node, Executor executor)

void	prepareForeignTable ()

std::unordered_set < shared::TableKey >	getPhysicalTableIds () const

void	prepareForSystemTableExecution (const CompilationOptions &co) const

Static Public Member Functions
static std::string	getErrorMessageFromCode (const int32_t error_code)

Private Member Functions
void	initializeParallelismHints ()

ExecutionResult	executeRelAlgQueryNoRetry (const CompilationOptions &co, const ExecutionOptions &eo, const bool just_explain_plan, const bool explain_verbose, 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 std::optional< size_t > previous_count)

ExecutionResult	executeTableFunction (const RelTableFunction *, const CompilationOptions &, const ExecutionOptions &, const int64_t queue_time_ms)

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)

ExecutionResult	executeUnion (const RelLogicalUnion , const RaExecutionSequence &, const CompilationOptions &, const ExecutionOptions &, RenderInfo , const int64_t queue_time_ms)

WorkUnit	createSortInputWorkUnit (const RelSort *, std::list< Analyzer::OrderEntry > &order_entries, 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_in, RenderInfo *, const int64_t queue_time_ms, const std::optional< size_t > previous_count=std::nullopt)

void	computeWindow (const WorkUnit &work_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, std::unordered_map< QueryPlanHash, std::shared_ptr< HashJoin >> &partition_cache, std::unordered_map< QueryPlanHash, size_t > &sorted_partition_key_ref_count_map, const WorkUnit &work_unit, const std::vector< InputTableInfo > &query_infos, const CompilationOptions &co, ColumnCacheMap &column_cache_map, std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner)

size_t	getNDVEstimation (const WorkUnit &work_unit, const int64_t range, const bool is_agg, const CompilationOptions &co, const ExecutionOptions &eo)

bool	hasDeletedRowInQuery (std::vector< InputTableInfo > const &) const

std::optional< size_t >	getFilteredCountAll (const RelAlgExecutionUnit &ra_exe_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< PushedDownFilterInfo >	selectFiltersToBePushedDown (const RelAlgExecutor::WorkUnit &work_unit, const CompilationOptions &co, const ExecutionOptions &eo)

bool	isRowidLookup (const WorkUnit &work_unit)

ExecutionResult	handleOutOfMemoryRetry (const RelAlgExecutor::WorkUnit &work_unit, ColumnCacheMap &column_cache, const std::vector< TargetMetaInfo > &targets_meta, const bool is_agg, const CompilationOptions &co, const ExecutionOptions &eo, RenderInfo *render_info, const QueryExecutionError &e, 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)

WorkUnit	createUnionWorkUnit (const RelLogicalUnion *, const SortInfo &, const ExecutionOptions &eo)

TableFunctionWorkUnit	createTableFunctionWorkUnit (const RelTableFunction *table_func, const bool just_explain, const bool is_gpu)

void	addTemporaryTable (const int table_id, const ResultSetPtr &result)

void	eraseFromTemporaryTables (const int table_id)

void	handleNop (RaExecutionDesc &ed)

std::unordered_map< unsigned, JoinQualsPerNestingLevel > &	getLeftDeepJoinTreesInfo ()

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

void	setHasStepForUnion (bool flag)

bool	hasStepForUnion () const

bool	canUseResultsetCache (const ExecutionOptions &eo, RenderInfo *render_info) const

void	setupCaching (const RelAlgNode *ra)

Private Member Functions inherited from StorageIOFacility
	StorageIOFacility (Executor *executor)

StorageIOFacility::UpdateCallback	yieldUpdateCallback (UpdateTransactionParameters &update_parameters)

StorageIOFacility::UpdateCallback	yieldDeleteCallback (DeleteTransactionParameters &delete_parameters)

Static Private Member Functions
static void	handlePersistentError (const int32_t error_code)

Private Attributes
Executor *	executor_

std::unique_ptr< RelAlgDag >	query_dag_

std::shared_ptr< const query_state::QueryState >	query_state_

TemporaryTables	temporary_tables_

time_t	now_

std::unordered_map< unsigned, JoinQualsPerNestingLevel >	left_deep_join_info_

std::vector< std::shared_ptr < Analyzer::Expr > >	target_exprs_owned_

std::unordered_map< unsigned, AggregatedResult >	leaf_results_

int64_t	queue_time_ms_

bool	has_step_for_union_

std::unique_ptr < TransactionParameters >	dml_transaction_parameters_

std::optional< std::function < void()> >	post_execution_callback_

Static Private Attributes
static SpeculativeTopNBlacklist	speculative_topn_blacklist_

Friends
class	PendingExecutionClosure

Additional Inherited Members
Private Types inherited from StorageIOFacility
using	UpdateCallback = UpdateLogForFragment::Callback

using	TableDescriptorType = TableDescriptor

using	DeleteVictimOffsetList = std::vector< uint64_t >

using	UpdateTargetOffsetList = std::vector< uint64_t >

using	UpdateTargetTypeList = std::vector< TargetMetaInfo >

using	UpdateTargetColumnNamesList = std::vector< std::string >

using	TransactionLog = Fragmenter_Namespace::InsertOrderFragmenter::ModifyTransactionTracker

using	TransactionLogPtr = std::unique_ptr< TransactionLog >

using	ColumnValidationFunction = std::function< bool(std::string const &)>

Detailed Description

Definition at line 53 of file RelAlgExecutor.h.

Member Typedef Documentation

using RelAlgExecutor::TargetInfoList = std::vector<TargetInfo>

Definition at line 55 of file RelAlgExecutor.h.

Constructor & Destructor Documentation

RelAlgExecutor::RelAlgExecutor	(	Executor *	executor,
		std::shared_ptr< const query_state::QueryState >	query_state = `nullptr`
	)

inline

Definition at line 57 of file RelAlgExecutor.h.

References initializeParallelismHints().

       : StorageIOFacility(executor)
       , executor_(executor)
       , query_state_(std::move(query_state))
       , now_(0)
       , queue_time_ms_(0) {
     initializeParallelismHints();
   }

Here is the call graph for this function:

RelAlgExecutor::RelAlgExecutor	(	Executor *	executor,
		const std::string &	query_ra,
		std::shared_ptr< const query_state::QueryState >	query_state = `nullptr`
	)

inline

Definition at line 67 of file RelAlgExecutor.h.

References initializeParallelismHints().

       : StorageIOFacility(executor)
       , executor_(executor)
       , query_dag_(RelAlgDagBuilder::buildDag(query_ra, true))
       , query_state_(std::move(query_state))
       , now_(0)
       , queue_time_ms_(0) {
     initializeParallelismHints();
   }

Here is the call graph for this function:

RelAlgExecutor::RelAlgExecutor	(	Executor *	executor,
		std::unique_ptr< RelAlgDag >	query_dag,
		std::shared_ptr< const query_state::QueryState >	query_state = `nullptr`
	)

inline

Definition at line 79 of file RelAlgExecutor.h.

References initializeParallelismHints().

       : StorageIOFacility(executor)
       , executor_(executor)
       , query_dag_(std::move(query_dag))
       , query_state_(std::move(query_state))
       , now_(0)
       , queue_time_ms_(0) {
     initializeParallelismHints();
   }

Here is the call graph for this function:

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_.

                                                                         {
     const auto it_ok = leaf_results_.emplace(id, result);
     CHECK(it_ok.second);
   }

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

inlineprivate

Definition at line 402 of file RelAlgExecutor.h.

References CHECK, CHECK_LT, and temporary_tables_.

Referenced by executeRelAlgSeq(), executeRelAlgStep(), executeSort(), executeTableFunction(), and handleNop().

                                                                          {
     CHECK_LT(size_t(0), result->colCount());
     CHECK_LT(table_id, 0);
     auto it_ok = temporary_tables_.emplace(table_id, result);
     CHECK(it_ok.second);
   }

Here is the caller graph for this function:

bool RelAlgExecutor::canUseResultsetCache	(	const ExecutionOptions &	eo,
		RenderInfo *	render_info
	)		const

private

Definition at line 483 of file RelAlgExecutor.cpp.

References g_cluster, g_enable_data_recycler, g_use_query_resultset_cache, hasStepForUnion(), is_validate_or_explain_query(), heavyai::InSituFlagsOwnerInterface::isInSitu(), and ExecutionOptions::outer_fragment_indices.

Referenced by executeRelAlgStep(), executeSort(), executeTableFunction(), and executeWorkUnit().

                                                                          {
   auto validate_or_explain_query = is_validate_or_explain_query(eo);
   auto query_for_partial_outer_frag = !eo.outer_fragment_indices.empty();
   return g_enable_data_recycler && g_use_query_resultset_cache && !g_cluster &&
          !validate_or_explain_query && !hasStepForUnion() &&
          !query_for_partial_outer_frag &&
          (!render_info || (render_info && !render_info->isInSitu()));
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void RelAlgExecutor::cleanupPostExecution ( )

Definition at line 776 of file RelAlgExecutor.cpp.

References CHECK, and executor_.

Referenced by executeRelAlgQueryNoRetry(), and getOuterFragmentCount().

                                           {
   CHECK(executor_);
   executor_->row_set_mem_owner_ = nullptr;
 }

Here is the caller graph for this function:

AggregatedColRange RelAlgExecutor::computeColRangesCache ( )

Definition at line 756 of file RelAlgExecutor.cpp.

References executor_, anonymous_namespace{RelAlgExecutor.cpp}::get_physical_inputs_with_spi_col_id(), and getRootRelAlgNode().

                                                          {
   AggregatedColRange agg_col_range_cache;
   const auto phys_inputs = get_physical_inputs_with_spi_col_id(&getRootRelAlgNode());
   return executor_->computeColRangesCache(phys_inputs);
 }

Here is the call graph for this function:

StringDictionaryGenerations RelAlgExecutor::computeStringDictionaryGenerations ( )

Definition at line 762 of file RelAlgExecutor.cpp.

References executor_, anonymous_namespace{RelAlgExecutor.cpp}::get_physical_inputs_with_spi_col_id(), and getRootRelAlgNode().

                                                                                {
   const auto phys_inputs = get_physical_inputs_with_spi_col_id(&getRootRelAlgNode());
   return executor_->computeStringDictionaryGenerations(phys_inputs);
 }

Here is the call graph for this function:

TableGenerations RelAlgExecutor::computeTableGenerations ( )

Definition at line 767 of file RelAlgExecutor.cpp.

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

                                                          {
   const auto phys_table_ids = get_physical_table_inputs(&getRootRelAlgNode());
   return executor_->computeTableGenerations(phys_table_ids);
 }

Here is the call graph for this function:

void RelAlgExecutor::computeWindow	(	const WorkUnit &	work_unit,
		const CompilationOptions &	co,
		const ExecutionOptions &	eo,
		ColumnCacheMap &	column_cache_map,
		const int64_t	queue_time_ms
	)

private

Definition at line 2492 of file RelAlgExecutor.cpp.

References CHECK, CHECK_EQ, WindowProjectNodeContext::create(), createWindowFunctionContext(), RelAlgExecutor::WorkUnit::exe_unit, 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 executeUpdate(), and executeWorkUnit().

                                                                 {
   auto query_infos = get_table_infos(work_unit.exe_unit.input_descs, executor_);
   CHECK_EQ(query_infos.size(), size_t(1));
   if (query_infos.front().info.fragments.size() != 1) {
     throw std::runtime_error(
         "Only single fragment tables supported for window functions for now");
   }
   if (eo.executor_type == ::ExecutorType::Extern) {
     return;
   }
   query_infos.push_back(query_infos.front());
   auto window_project_node_context = WindowProjectNodeContext::create(executor_);
   // a query may hold multiple window functions having the same partition by condition
   // then after building the first hash partition we can reuse it for the rest of
   // the window functions
   // here, a cached partition can be shared via multiple window function contexts as is
   // but sorted partition should be copied to reuse since we use it for (intermediate)
   // output buffer
   // todo (yoonmin) : support recycler for window function computation?
   std::unordered_map<QueryPlanHash, std::shared_ptr<HashJoin>> partition_cache;
   std::unordered_map<QueryPlanHash, std::shared_ptr<std::vector<int64_t>>>
       sorted_partition_cache;
   std::unordered_map<QueryPlanHash, size_t> sorted_partition_key_ref_count_map;
   std::unordered_map<size_t, std::unique_ptr<WindowFunctionContext>>
       window_function_context_map;
   std::unordered_map<QueryPlanHash, AggregateTreeForWindowFraming> aggregate_tree_map;
   for (size_t target_index = 0; target_index < work_unit.exe_unit.target_exprs.size();
        ++target_index) {
     const auto& target_expr = work_unit.exe_unit.target_exprs[target_index];
     const auto window_func = dynamic_cast<const Analyzer::WindowFunction*>(target_expr);
     if (!window_func) {
       continue;
     }
     // Always use baseline layout hash tables for now, make the expression a tuple.
     const auto& partition_keys = window_func->getPartitionKeys();
     std::shared_ptr<Analyzer::BinOper> partition_key_cond;
     if (partition_keys.size() >= 1) {
       std::shared_ptr<Analyzer::Expr> partition_key_tuple;
       if (partition_keys.size() > 1) {
         partition_key_tuple = makeExpr<Analyzer::ExpressionTuple>(partition_keys);
       } else {
         CHECK_EQ(partition_keys.size(), size_t(1));
         partition_key_tuple = partition_keys.front();
       }
       // Creates a tautology equality with the partition expression on both sides.
       partition_key_cond =
           makeExpr<Analyzer::BinOper>(kBOOLEAN,
                                       kBW_EQ,
                                       kONE,
                                       partition_key_tuple,
                                       transform_to_inner(partition_key_tuple.get()));
     }
     auto context =
         createWindowFunctionContext(window_func,
                                     partition_key_cond /*nullptr if no partition key*/,
                                     partition_cache,
                                     sorted_partition_key_ref_count_map,
                                     work_unit,
                                     query_infos,
                                     co,
                                     column_cache_map,
                                     executor_->getRowSetMemoryOwner());
     CHECK(window_function_context_map.emplace(target_index, std::move(context)).second);
   }
 
   for (auto& kv : window_function_context_map) {
     kv.second->compute(
         sorted_partition_key_ref_count_map, sorted_partition_cache, aggregate_tree_map);
     window_project_node_context->addWindowFunctionContext(std::move(kv.second), kv.first);
   }
 }

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 4849 of file RelAlgExecutor.cpp.

References QueryPlanDagExtractor::applyLimitClauseToCacheKey(), CHECK_EQ, RegisteredQueryHint::defaults(), executor_, QueryPlanDagExtractor::extractJoinInfo(), g_default_max_groups_buffer_entry_guess, anonymous_namespace{RelAlgExecutor.cpp}::get_input_desc(), anonymous_namespace{RelAlgExecutor.cpp}::get_input_nest_levels(), anonymous_namespace{RelAlgExecutor.cpp}::get_join_type(), get_table_infos(), anonymous_namespace{RelAlgExecutor.cpp}::get_targets_meta(), RelAlgNode::getInput(), getLeftDeepJoinTreesInfo(), RelAlgNode::getOutputMetainfo(), RelAlgNode::getQueryPlanDagHash(), RelAlgNode::inputCount(), now_, query_dag_, 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().

                              {
   std::vector<InputDescriptor> input_descs;
   std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
   std::vector<std::shared_ptr<RexInput>> used_inputs_owned;
   const auto input_to_nest_level = get_input_nest_levels(aggregate, {});
   std::tie(input_descs, input_col_descs, used_inputs_owned) =
       get_input_desc(aggregate, input_to_nest_level, {});
   const auto join_type = get_join_type(aggregate);
 
   RelAlgTranslator translator(
       query_state_, executor_, input_to_nest_level, {join_type}, now_, just_explain);
   CHECK_EQ(size_t(1), aggregate->inputCount());
   const auto source = aggregate->getInput(0);
   const auto& in_metainfo = source->getOutputMetainfo();
   const auto scalar_sources =
       synthesize_inputs(aggregate, size_t(0), in_metainfo, input_to_nest_level);
   const auto groupby_exprs = translate_groupby_exprs(aggregate, scalar_sources);
   std::unordered_map<size_t, SQLTypeInfo> target_exprs_type_infos;
   const auto target_exprs = translate_targets(target_exprs_owned_,
                                               target_exprs_type_infos,
                                               scalar_sources,
                                               groupby_exprs,
                                               aggregate,
                                               translator);
 
   const auto query_infos = get_table_infos(input_descs, executor_);
 
   const auto targets_meta = get_targets_meta(aggregate, target_exprs);
   aggregate->setOutputMetainfo(targets_meta);
   auto query_hint = RegisteredQueryHint::defaults();
   if (query_dag_) {
     auto candidate = query_dag_->getQueryHint(aggregate);
     if (candidate) {
       query_hint = *candidate;
     }
   }
   auto join_info = QueryPlanDagExtractor::extractJoinInfo(
       aggregate, std::nullopt, getLeftDeepJoinTreesInfo(), executor_);
   return {RelAlgExecutionUnit{input_descs,
                               input_col_descs,
                               {},
                               {},
                               {},
                               groupby_exprs,
                               target_exprs,
                               target_exprs_type_infos,
                               nullptr,
                               sort_info,
                               0,
                               query_hint,
                               QueryPlanDagExtractor::applyLimitClauseToCacheKey(
                                   aggregate->getQueryPlanDagHash(), sort_info),
                               join_info.hash_table_plan_dag,
                               join_info.table_id_to_node_map,
                               false,
                               std::nullopt,
                               query_state_},
           aggregate,
           g_default_max_groups_buffer_entry_guess,
           nullptr};
 }

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 4513 of file RelAlgExecutor.cpp.

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

                                 {
   std::vector<InputDescriptor> input_descs;
   std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
   auto input_to_nest_level = get_input_nest_levels(compound, {});
   std::tie(input_descs, input_col_descs, std::ignore) =
       get_input_desc(compound, input_to_nest_level, {});
   VLOG(3) << "input_descs=" << shared::printContainer(input_descs);
   VLOG(3) << "input_col_descs=" << shared::printContainer(input_col_descs);
   auto query_infos = get_table_infos(input_descs, executor_);
   CHECK_EQ(size_t(1), compound->inputCount());
   const auto left_deep_join =
       dynamic_cast<const RelLeftDeepInnerJoin*>(compound->getInput(0));
   JoinQualsPerNestingLevel left_deep_join_quals;
   const auto join_types = left_deep_join ? left_deep_join_types(left_deep_join)
                                          : std::vector<JoinType>{get_join_type(compound)};
   std::vector<size_t> input_permutation;
   std::vector<size_t> left_deep_join_input_sizes;
   std::optional<unsigned> left_deep_tree_id;
   if (left_deep_join) {
     left_deep_tree_id = left_deep_join->getId();
     left_deep_join_input_sizes = get_left_deep_join_input_sizes(left_deep_join);
     left_deep_join_quals = translateLeftDeepJoinFilter(
         left_deep_join, input_descs, input_to_nest_level, eo.just_explain);
     if (eo.table_reordering &&
         std::find(join_types.begin(), join_types.end(), JoinType::LEFT) ==
             join_types.end()) {
       input_permutation = do_table_reordering(input_descs,
                                               input_col_descs,
                                               left_deep_join_quals,
                                               input_to_nest_level,
                                               compound,
                                               query_infos,
                                               executor_);
       input_to_nest_level = get_input_nest_levels(compound, input_permutation);
       std::tie(input_descs, input_col_descs, std::ignore) =
           get_input_desc(compound, input_to_nest_level, input_permutation);
       left_deep_join_quals = translateLeftDeepJoinFilter(
           left_deep_join, input_descs, input_to_nest_level, eo.just_explain);
     }
   }
   auto const bbox_intersect_qual_info = convert_bbox_intersect_join(left_deep_join_quals,
                                                                     input_descs,
                                                                     input_to_nest_level,
                                                                     input_permutation,
                                                                     input_col_descs,
                                                                     executor_);
   if (bbox_intersect_qual_info.is_reordered) {
     query_infos = get_table_infos(input_descs, executor_);
     VLOG(3) << "input_descs=" << shared::printContainer(input_descs);
     VLOG(3) << "input_col_descs=" << shared::printContainer(input_col_descs);
   }
   if (bbox_intersect_qual_info.has_bbox_intersect_join) {
     left_deep_join_quals = bbox_intersect_qual_info.join_quals;
   }
   RelAlgTranslator translator(
       query_state_, executor_, input_to_nest_level, join_types, now_, eo.just_explain);
   const auto quals_cf = translate_quals(compound, translator);
   const auto quals = rewrite_quals(quals_cf.quals);
   const auto scalar_sources =
       translate_scalar_sources(compound, translator, eo.executor_type);
   const auto groupby_exprs = translate_groupby_exprs(compound, scalar_sources);
   std::unordered_map<size_t, SQLTypeInfo> target_exprs_type_infos;
   const auto target_exprs = translate_targets(target_exprs_owned_,
                                               target_exprs_type_infos,
                                               scalar_sources,
                                               groupby_exprs,
                                               compound,
                                               translator,
                                               eo.executor_type);
 
   auto query_hint = RegisteredQueryHint::defaults();
   if (query_dag_) {
     auto candidate = query_dag_->getQueryHint(compound);
     if (candidate) {
       query_hint = *candidate;
     }
   }
   CHECK_EQ(compound->size(), target_exprs.size());
   const RelAlgExecutionUnit exe_unit = {input_descs,
                                         input_col_descs,
                                         quals_cf.simple_quals,
                                         quals,
                                         left_deep_join_quals,
                                         groupby_exprs,
                                         target_exprs,
                                         target_exprs_type_infos,
                                         nullptr,
                                         sort_info,
                                         0,
                                         query_hint,
                                         QueryPlanDagExtractor::applyLimitClauseToCacheKey(
                                             compound->getQueryPlanDagHash(), sort_info),
                                         {},
                                         {},
                                         false,
                                         std::nullopt,
                                         query_state_};
   auto query_rewriter = std::make_unique<QueryRewriter>(query_infos, executor_);
   auto rewritten_exe_unit = query_rewriter->rewrite(exe_unit);
   const auto targets_meta = get_targets_meta(compound, rewritten_exe_unit.target_exprs);
   compound->setOutputMetainfo(targets_meta);
   auto& left_deep_trees_info = getLeftDeepJoinTreesInfo();
   if (left_deep_tree_id && left_deep_tree_id.has_value()) {
     left_deep_trees_info.emplace(left_deep_tree_id.value(),
                                  rewritten_exe_unit.join_quals);
   }
   if (has_valid_query_plan_dag(compound)) {
     auto join_info = QueryPlanDagExtractor::extractJoinInfo(
         compound, left_deep_tree_id, left_deep_trees_info, executor_);
     rewritten_exe_unit.hash_table_build_plan_dag = join_info.hash_table_plan_dag;
     rewritten_exe_unit.table_id_to_node_map = join_info.table_id_to_node_map;
   }
   return {rewritten_exe_unit,
           compound,
           g_default_max_groups_buffer_entry_guess,
           std::move(query_rewriter),
           input_permutation,
           left_deep_join_input_sizes};
 }

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 5400 of file RelAlgExecutor.cpp.

References QueryPlanDagExtractor::applyLimitClauseToCacheKey(), CHECK_EQ, RegisteredQueryHint::defaults(), executor_, QueryPlanDagExtractor::extractJoinInfo(), fold_expr(), g_default_max_groups_buffer_entry_guess, 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(), anonymous_namespace{RelAlgExecutor.cpp}::get_raw_pointers(), get_table_infos(), RelFilter::getCondition(), getLeftDeepJoinTreesInfo(), RelAlgNode::getQueryPlanDagHash(), RelAlgNode::inputCount(), now_, query_dag_, query_state_, rewrite_expr(), RelAlgNode::setOutputMetainfo(), and target_exprs_owned_.

Referenced by createWorkUnit(), and executeFilter().

                                                                                        {
   CHECK_EQ(size_t(1), filter->inputCount());
   std::vector<InputDescriptor> input_descs;
   std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
   std::vector<TargetMetaInfo> in_metainfo;
   std::vector<std::shared_ptr<RexInput>> used_inputs_owned;
   std::vector<std::shared_ptr<Analyzer::Expr>> target_exprs_owned;
 
   const auto input_to_nest_level = get_input_nest_levels(filter, {});
   std::tie(input_descs, input_col_descs, used_inputs_owned) =
       get_input_desc(filter, input_to_nest_level, {});
   const auto join_type = get_join_type(filter);
   RelAlgTranslator translator(
       query_state_, executor_, input_to_nest_level, {join_type}, now_, just_explain);
   std::tie(in_metainfo, target_exprs_owned) =
       get_inputs_meta(filter, translator, used_inputs_owned, input_to_nest_level);
   const auto filter_expr = translator.translate(filter->getCondition());
   const auto query_infos = get_table_infos(input_descs, executor_);
 
   const auto qual = fold_expr(filter_expr.get());
   target_exprs_owned_.insert(
       target_exprs_owned_.end(), target_exprs_owned.begin(), target_exprs_owned.end());
 
   const auto target_exprs = get_raw_pointers(target_exprs_owned);
   filter->setOutputMetainfo(in_metainfo);
   const auto rewritten_qual = rewrite_expr(qual.get());
   auto query_hint = RegisteredQueryHint::defaults();
   if (query_dag_) {
     auto candidate = query_dag_->getQueryHint(filter);
     if (candidate) {
       query_hint = *candidate;
     }
   }
   auto join_info = QueryPlanDagExtractor::extractJoinInfo(
       filter, std::nullopt, getLeftDeepJoinTreesInfo(), executor_);
   return {{input_descs,
            input_col_descs,
            {},
            {rewritten_qual ? rewritten_qual : qual},
            {},
            {nullptr},
            target_exprs,
            {},
            nullptr,
            sort_info,
            0,
            query_hint,
            QueryPlanDagExtractor::applyLimitClauseToCacheKey(
                filter->getQueryPlanDagHash(), sort_info),
            join_info.hash_table_plan_dag,
            join_info.table_id_to_node_map},
           filter,
           g_default_max_groups_buffer_entry_guess,
           nullptr};
 }

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 4914 of file RelAlgExecutor.cpp.

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

                                 {
   std::vector<InputDescriptor> input_descs;
   std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
   auto input_to_nest_level = get_input_nest_levels(project, {});
   std::tie(input_descs, input_col_descs, std::ignore) =
       get_input_desc(project, input_to_nest_level, {});
   VLOG(3) << "input_descs=" << shared::printContainer(input_descs);
   VLOG(3) << "input_col_descs=" << shared::printContainer(input_col_descs);
   auto query_infos = get_table_infos(input_descs, executor_);
   const auto left_deep_join =
       dynamic_cast<const RelLeftDeepInnerJoin*>(project->getInput(0));
   JoinQualsPerNestingLevel left_deep_join_quals;
   const auto join_types = left_deep_join ? left_deep_join_types(left_deep_join)
                                          : std::vector<JoinType>{get_join_type(project)};
   std::vector<size_t> input_permutation;
   std::vector<size_t> left_deep_join_input_sizes;
   std::optional<unsigned> left_deep_tree_id;
   if (left_deep_join) {
     left_deep_tree_id = left_deep_join->getId();
     left_deep_join_input_sizes = get_left_deep_join_input_sizes(left_deep_join);
     left_deep_join_quals = translateLeftDeepJoinFilter(
         left_deep_join, input_descs, input_to_nest_level, eo.just_explain);
     if (eo.table_reordering) {
       input_permutation = do_table_reordering(input_descs,
                                               input_col_descs,
                                               left_deep_join_quals,
                                               input_to_nest_level,
                                               project,
                                               query_infos,
                                               executor_);
       input_to_nest_level = get_input_nest_levels(project, input_permutation);
       std::tie(input_descs, input_col_descs, std::ignore) =
           get_input_desc(project, input_to_nest_level, input_permutation);
       left_deep_join_quals = translateLeftDeepJoinFilter(
           left_deep_join, input_descs, input_to_nest_level, eo.just_explain);
     }
   }
   auto const bbox_intersect_qual_info = convert_bbox_intersect_join(left_deep_join_quals,
                                                                     input_descs,
                                                                     input_to_nest_level,
                                                                     input_permutation,
                                                                     input_col_descs,
                                                                     executor_);
   if (bbox_intersect_qual_info.is_reordered) {
     query_infos = get_table_infos(input_descs, executor_);
     VLOG(3) << "input_descs=" << shared::printContainer(input_descs);
     VLOG(3) << "input_col_descs=" << shared::printContainer(input_col_descs);
   }
   if (bbox_intersect_qual_info.has_bbox_intersect_join) {
     left_deep_join_quals = bbox_intersect_qual_info.join_quals;
   }
   RelAlgTranslator translator(
       query_state_, executor_, input_to_nest_level, join_types, now_, eo.just_explain);
   const auto target_exprs_owned =
       translate_scalar_sources(project, translator, eo.executor_type);
 
   target_exprs_owned_.insert(
       target_exprs_owned_.end(), target_exprs_owned.begin(), target_exprs_owned.end());
   const auto target_exprs = get_raw_pointers(target_exprs_owned);
   auto query_hint = RegisteredQueryHint::defaults();
   if (query_dag_) {
     auto candidate = query_dag_->getQueryHint(project);
     if (candidate) {
       query_hint = *candidate;
     }
   }
   const RelAlgExecutionUnit exe_unit = {input_descs,
                                         input_col_descs,
                                         {},
                                         {},
                                         left_deep_join_quals,
                                         {nullptr},
                                         target_exprs,
                                         {},
                                         nullptr,
                                         sort_info,
                                         0,
                                         query_hint,
                                         QueryPlanDagExtractor::applyLimitClauseToCacheKey(
                                             project->getQueryPlanDagHash(), sort_info),
                                         {},
                                         {},
                                         false,
                                         std::nullopt,
                                         query_state_};
   auto query_rewriter = std::make_unique<QueryRewriter>(query_infos, executor_);
   auto rewritten_exe_unit = query_rewriter->rewrite(exe_unit);
   const auto targets_meta = get_targets_meta(project, rewritten_exe_unit.target_exprs);
   project->setOutputMetainfo(targets_meta);
   auto& left_deep_trees_info = getLeftDeepJoinTreesInfo();
   if (left_deep_tree_id && left_deep_tree_id.has_value()) {
     left_deep_trees_info.emplace(left_deep_tree_id.value(),
                                  rewritten_exe_unit.join_quals);
   }
   if (has_valid_query_plan_dag(project)) {
     auto join_info = QueryPlanDagExtractor::extractJoinInfo(
         project, left_deep_tree_id, left_deep_trees_info, executor_);
     rewritten_exe_unit.hash_table_build_plan_dag = join_info.hash_table_plan_dag;
     rewritten_exe_unit.table_id_to_node_map = join_info.table_id_to_node_map;
   }
   return {rewritten_exe_unit,
           project,
           g_default_max_groups_buffer_entry_guess,
           std::move(query_rewriter),
           input_permutation,
           left_deep_join_input_sizes};
 }

Here is the call graph for this function:

Here is the caller graph for this function:

RelAlgExecutor::WorkUnit RelAlgExecutor::createSortInputWorkUnit	(	const RelSort *	sort,
		std::list< Analyzer::OrderEntry > &	order_entries,
		const ExecutionOptions &	eo
	)

private

Definition at line 3394 of file RelAlgExecutor.cpp.

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

Referenced by executeRelAlgQuerySingleStep(), and executeSort().

                                 {
   const auto source = sort->getInput(0);
   auto limit = sort->getLimit();
   const size_t offset = sort->getOffset();
   const size_t scan_limit = sort->collationCount() ? 0 : get_scan_limit(source, limit);
   const size_t scan_total_limit =
       scan_limit ? get_scan_limit(source, scan_limit + offset) : 0;
   size_t max_groups_buffer_entry_guess{
       scan_total_limit ? scan_total_limit : g_default_max_groups_buffer_entry_guess};
   SortAlgorithm sort_algorithm{SortAlgorithm::SpeculativeTopN};
   SortInfo sort_info{order_entries, sort_algorithm, limit, offset};
   auto source_work_unit = createWorkUnit(source, sort_info, eo);
   const auto& source_exe_unit = source_work_unit.exe_unit;
 
   // we do not allow sorting geometry or array types
   for (auto order_entry : order_entries) {
     CHECK_GT(order_entry.tle_no, 0);  // tle_no is a 1-base index
     const auto& te = source_exe_unit.target_exprs[order_entry.tle_no - 1];
     const auto& ti = get_target_info(te, false);
     if (ti.sql_type.is_geometry() || ti.sql_type.is_array()) {
       throw std::runtime_error(
           "Columns with geometry or array types cannot be used in an ORDER BY clause.");
     }
   }
 
   if (source_exe_unit.groupby_exprs.size() == 1) {
     if (!source_exe_unit.groupby_exprs.front()) {
       sort_algorithm = SortAlgorithm::StreamingTopN;
     } else {
       if (speculative_topn_blacklist_.contains(source_exe_unit.groupby_exprs.front(),
                                                first_oe_is_desc(order_entries))) {
         sort_algorithm = SortAlgorithm::Default;
       }
     }
   }
 
   sort->setOutputMetainfo(source->getOutputMetainfo());
   // NB: the `body` field of the returned `WorkUnit` needs to be the `source` node,
   // not the `sort`. The aggregator needs the pre-sorted result from leaves.
   return {RelAlgExecutionUnit{source_exe_unit.input_descs,
                               std::move(source_exe_unit.input_col_descs),
                               source_exe_unit.simple_quals,
                               source_exe_unit.quals,
                               source_exe_unit.join_quals,
                               source_exe_unit.groupby_exprs,
                               source_exe_unit.target_exprs,
                               source_exe_unit.target_exprs_original_type_infos,
                               nullptr,
                               {sort_info.order_entries, sort_algorithm, limit, offset},
                               scan_total_limit,
                               source_exe_unit.query_hint,
                               source_exe_unit.query_plan_dag_hash,
                               source_exe_unit.hash_table_build_plan_dag,
                               source_exe_unit.table_id_to_node_map,
                               source_exe_unit.use_bump_allocator,
                               source_exe_unit.union_all,
                               source_exe_unit.query_state},
           source,
           max_groups_buffer_entry_guess,
           std::move(source_work_unit.query_rewriter),
           source_work_unit.input_permutation,
           source_work_unit.left_deep_join_input_sizes};
 }

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,
		const bool	is_gpu
	)

private

Definition at line 5156 of file RelAlgExecutor.cpp.

References bind_table_function(), CHECK, CHECK_EQ, CHECK_GT, CHECK_LT, RelTableFunction::countRexLiteralArgs(), DEFAULT_ROW_MULTIPLIER_VALUE, executor_, ext_arg_type_to_type_info(), anonymous_namespace{RelAlgExecutor.cpp}::get_input_desc(), anonymous_namespace{RelAlgExecutor.cpp}::get_input_nest_levels(), anonymous_namespace{RelAlgExecutor.cpp}::get_raw_pointers(), get_table_infos(), anonymous_namespace{RelAlgExecutor.cpp}::get_targets_meta(), RelTableFunction::getColInputsSize(), RelTableFunction::getFunctionName(), RelTableFunction::getTableFuncInputAt(), IS_GEO, kENCODING_ARRAY, kENCODING_NONE, kINT, shared::StringDictKey::kTransientDictKey, LOG, now_, query_state_, RelAlgNode::setOutputMetainfo(), TableFunctions, TableFunctionExecutionUnit::target_exprs, target_exprs_owned_, to_string(), TRANSIENT_DICT_ID, anonymous_namespace{RelAlgExecutor.cpp}::translate_scalar_sources(), UNREACHABLE, and logger::WARNING.

Referenced by executeTableFunction().

                        {
   std::vector<InputDescriptor> input_descs;
   std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
   auto input_to_nest_level = get_input_nest_levels(rel_table_func, {});
   std::tie(input_descs, input_col_descs, std::ignore) =
       get_input_desc(rel_table_func, input_to_nest_level, {});
   const auto query_infos = get_table_infos(input_descs, executor_);
   RelAlgTranslator translator(
       query_state_, executor_, input_to_nest_level, {}, now_, just_explain);
   auto input_exprs_owned = translate_scalar_sources(
       rel_table_func, translator, ::ExecutorType::TableFunctions);
   target_exprs_owned_.insert(
       target_exprs_owned_.end(), input_exprs_owned.begin(), input_exprs_owned.end());
 
   const auto table_function_impl_and_type_infos = [=]() {
     if (is_gpu) {
       try {
         return bind_table_function(
             rel_table_func->getFunctionName(), input_exprs_owned, is_gpu);
       } catch (ExtensionFunctionBindingError& e) {
         LOG(WARNING) << "createTableFunctionWorkUnit[GPU]: " << e.what()
                      << " Redirecting " << rel_table_func->getFunctionName()
                      << " step to run on CPU.";
         throw QueryMustRunOnCpu();
       }
     } else {
       try {
         return bind_table_function(
             rel_table_func->getFunctionName(), input_exprs_owned, is_gpu);
       } catch (ExtensionFunctionBindingError& e) {
         LOG(WARNING) << "createTableFunctionWorkUnit[CPU]: " << e.what();
         throw;
       }
     }
   }();
   const auto& table_function_impl = std::get<0>(table_function_impl_and_type_infos);
   const auto& table_function_type_infos = std::get<1>(table_function_impl_and_type_infos);
   size_t output_row_sizing_param = 0;
   if (table_function_impl
           .hasUserSpecifiedOutputSizeParameter()) {  // constant and row multiplier
     const auto parameter_index =
         table_function_impl.getOutputRowSizeParameter(table_function_type_infos);
     CHECK_GT(parameter_index, size_t(0));
     if (rel_table_func->countRexLiteralArgs() == table_function_impl.countScalarArgs()) {
       const auto parameter_expr =
           rel_table_func->getTableFuncInputAt(parameter_index - 1);
       const auto parameter_expr_literal = dynamic_cast<const RexLiteral*>(parameter_expr);
       if (!parameter_expr_literal) {
         throw std::runtime_error(
             "Provided output buffer sizing parameter is not a literal. Only literal "
             "values are supported with output buffer sizing configured table "
             "functions.");
       }
       int64_t literal_val = parameter_expr_literal->getVal<int64_t>();
       if (literal_val < 0) {
         throw std::runtime_error("Provided output sizing parameter " +
                                  std::to_string(literal_val) +
                                  " must be positive integer.");
       }
       output_row_sizing_param = static_cast<size_t>(literal_val);
     } else {
       // RowMultiplier not specified in the SQL query. Set it to 1
       output_row_sizing_param = 1;  // default value for RowMultiplier
       static Datum d = {DEFAULT_ROW_MULTIPLIER_VALUE};
       static Analyzer::ExpressionPtr DEFAULT_ROW_MULTIPLIER_EXPR =
           makeExpr<Analyzer::Constant>(kINT, false, d);
       // Push the constant 1 to input_exprs
       input_exprs_owned.insert(input_exprs_owned.begin() + parameter_index - 1,
                                DEFAULT_ROW_MULTIPLIER_EXPR);
     }
   } else if (table_function_impl.hasNonUserSpecifiedOutputSize()) {
     output_row_sizing_param = table_function_impl.getOutputRowSizeParameter();
   } else {
     UNREACHABLE();
   }
 
   std::vector<Analyzer::ColumnVar*> input_col_exprs;
   size_t input_index = 0;
   size_t arg_index = 0;
   const auto table_func_args = table_function_impl.getInputArgs();
   CHECK_EQ(table_func_args.size(), table_function_type_infos.size());
   for (const auto& ti : table_function_type_infos) {
     if (ti.is_column_list()) {
       for (int i = 0; i < ti.get_dimension(); i++) {
         auto& input_expr = input_exprs_owned[input_index];
         auto col_var = dynamic_cast<Analyzer::ColumnVar*>(input_expr.get());
         CHECK(col_var);
 
         // avoid setting type info to ti here since ti doesn't have all the
         // properties correctly set
         auto type_info = input_expr->get_type_info();
         if (ti.is_column_array()) {
           type_info.set_compression(kENCODING_ARRAY);
           type_info.set_subtype(type_info.get_subtype());  // set type to be subtype
         } else {
           type_info.set_subtype(type_info.get_type());  // set type to be subtype
         }
         type_info.set_type(ti.get_type());  // set type to column list
         type_info.set_dimension(ti.get_dimension());
         type_info.setUsesFlatBuffer(type_info.get_elem_type().supportsFlatBuffer());
         input_expr->set_type_info(type_info);
 
         input_col_exprs.push_back(col_var);
         input_index++;
       }
     } else if (ti.is_column()) {
       auto& input_expr = input_exprs_owned[input_index];
       auto col_var = dynamic_cast<Analyzer::ColumnVar*>(input_expr.get());
       CHECK(col_var);
       // same here! avoid setting type info to ti since it doesn't have all the
       // properties correctly set
       auto type_info = input_expr->get_type_info();
       if (ti.is_column_array()) {
         type_info.set_compression(kENCODING_ARRAY);
         type_info.set_subtype(type_info.get_subtype());  // set type to be subtype
       } else {
         type_info.set_subtype(type_info.get_type());  // set type to be subtype
       }
       type_info.set_type(ti.get_type());  // set type to column
       type_info.setUsesFlatBuffer(type_info.get_elem_type().supportsFlatBuffer());
       input_expr->set_type_info(type_info);
       input_col_exprs.push_back(col_var);
       input_index++;
     } else {
       auto input_expr = input_exprs_owned[input_index];
       auto ext_func_arg_ti = ext_arg_type_to_type_info(table_func_args[arg_index]);
       if (ext_func_arg_ti != input_expr->get_type_info()) {
         input_exprs_owned[input_index] = input_expr->add_cast(ext_func_arg_ti);
       }
       input_index++;
     }
     arg_index++;
   }
   CHECK_EQ(input_col_exprs.size(), rel_table_func->getColInputsSize());
   std::vector<Analyzer::Expr*> table_func_outputs;
   constexpr int32_t transient_pos{-1};
   for (size_t i = 0; i < table_function_impl.getOutputsSize(); i++) {
     auto ti = table_function_impl.getOutputSQLType(i);
     ti.setUsesFlatBuffer(ti.supportsFlatBuffer());
     if (ti.is_geometry()) {
       auto p = table_function_impl.getInputID(i);
       int32_t input_pos = p.first;
       if (input_pos != transient_pos) {
         CHECK(!ti.is_column_list());  // see QE-472
         CHECK_LT(input_pos, input_exprs_owned.size());
         const auto& reference_ti = input_exprs_owned[input_pos]->get_type_info();
         CHECK(IS_GEO(reference_ti.get_type()) || IS_GEO(reference_ti.get_subtype()));
         ti.set_input_srid(reference_ti.get_input_srid());
         ti.set_output_srid(reference_ti.get_output_srid());
         ti.set_compression(reference_ti.get_compression());
         ti.set_comp_param(reference_ti.get_comp_param());
       } else {
         ti.set_input_srid(0);
         ti.set_output_srid(0);
         ti.set_compression(kENCODING_NONE);
         ti.set_comp_param(0);
       }
     } else if (ti.is_dict_encoded_string() || ti.is_text_encoding_dict_array()) {
       auto p = table_function_impl.getInputID(i);
 
       int32_t input_pos = p.first;
       if (input_pos == transient_pos) {
         ti.set_comp_param(TRANSIENT_DICT_ID);
         ti.setStringDictKey(shared::StringDictKey::kTransientDictKey);
       } else {
         // Iterate over the list of arguments to compute the offset. Use this offset to
         // get the corresponding input
         int32_t offset = 0;
         for (int j = 0; j < input_pos; j++) {
           const auto ti = table_function_type_infos[j];
           offset += ti.is_column_list() ? ti.get_dimension() : 1;
         }
         input_pos = offset + p.second;
 
         CHECK_LT(input_pos, input_exprs_owned.size());
         const auto& dict_key =
             input_exprs_owned[input_pos]->get_type_info().getStringDictKey();
         ti.set_comp_param(dict_key.dict_id);
         ti.setStringDictKey(dict_key);
       }
     }
     target_exprs_owned_.push_back(std::make_shared<Analyzer::ColumnVar>(
         ti, shared::ColumnKey{0, 0, int32_t(i)}, -1));
     table_func_outputs.push_back(target_exprs_owned_.back().get());
   }
   auto input_exprs = get_raw_pointers(input_exprs_owned);
   const TableFunctionExecutionUnit exe_unit = {
       input_descs,
       input_col_descs,
       input_exprs,              // table function inputs
       input_col_exprs,          // table function column inputs (duplicates w/ above)
       table_func_outputs,       // table function projected exprs
       output_row_sizing_param,  // output buffer sizing param
       table_function_impl};
   const auto targets_meta = get_targets_meta(rel_table_func, exe_unit.target_exprs);
   rel_table_func->setOutputMetainfo(targets_meta);
   return {exe_unit, rel_table_func};
 }

Here is the call graph for this function:

Here is the caller graph for this function:

RelAlgExecutor::WorkUnit RelAlgExecutor::createUnionWorkUnit	(	const RelLogicalUnion *	logical_union,
		const SortInfo &	sort_info,
		const ExecutionOptions &	eo
	)

private

Definition at line 5049 of file RelAlgExecutor.cpp.

References gpu_enabled::accumulate(), shared::append_move(), CHECK, RelRexToStringConfig::defaults(), RegisteredQueryHint::defaults(), EMPTY_HASHED_PLAN_DAG_KEY, executor_, g_default_max_groups_buffer_entry_guess, anonymous_namespace{RelAlgExecutor.cpp}::get_input_desc(), anonymous_namespace{RelAlgExecutor.cpp}::get_input_nest_levels(), anonymous_namespace{RelAlgExecutor.cpp}::get_raw_pointers(), get_table_infos(), anonymous_namespace{RelAlgExecutor.cpp}::get_targets_meta(), RelAlgNode::getInput(), RelAlgNode::getOutputMetainfo(), RelLogicalUnion::isAll(), shared::printContainer(), query_state_, RelAlgNode::setOutputMetainfo(), anonymous_namespace{RelAlgExecutor.cpp}::target_exprs_for_union(), target_exprs_owned_, RelAlgNode::toString(), and VLOG.

Referenced by executeUnion().

                                 {
   std::vector<InputDescriptor> input_descs;
   std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
   // Map ra input ptr to index (0, 1).
   auto input_to_nest_level = get_input_nest_levels(logical_union, {});
   std::tie(input_descs, input_col_descs, std::ignore) =
       get_input_desc(logical_union, input_to_nest_level, {});
   const auto query_infos = get_table_infos(input_descs, executor_);
   auto const max_num_tuples =
       std::accumulate(query_infos.cbegin(),
                       query_infos.cend(),
                       size_t(0),
                       [](auto max, auto const& query_info) {
                         return std::max(max, query_info.info.getNumTuples());
                       });
 
   VLOG(3) << "input_to_nest_level.size()=" << input_to_nest_level.size() << " Pairs are:";
   for (auto& pair : input_to_nest_level) {
     VLOG(3) << "  (" << pair.first->toString(RelRexToStringConfig::defaults()) << ", "
             << pair.second << ')';
   }
 
   // For UNION queries, we need to keep the target_exprs from both subqueries since they
   // may differ on StringDictionaries.
   std::vector<Analyzer::Expr*> target_exprs_pair[2];
   for (unsigned i = 0; i < 2; ++i) {
     auto input_exprs_owned = target_exprs_for_union(logical_union->getInput(i));
     CHECK(!input_exprs_owned.empty())
         << "No metainfo found for input node(" << i << ") "
         << logical_union->getInput(i)->toString(RelRexToStringConfig::defaults());
     VLOG(3) << "i(" << i << ") input_exprs_owned.size()=" << input_exprs_owned.size();
     for (auto& input_expr : input_exprs_owned) {
       VLOG(3) << "  " << input_expr->toString();
     }
     target_exprs_pair[i] = get_raw_pointers(input_exprs_owned);
     shared::append_move(target_exprs_owned_, std::move(input_exprs_owned));
   }
 
   VLOG(3) << "input_descs=" << shared::printContainer(input_descs)
           << " input_col_descs=" << shared::printContainer(input_col_descs)
           << " target_exprs.size()=" << target_exprs_pair[0].size()
           << " max_num_tuples=" << max_num_tuples;
 
   const RelAlgExecutionUnit exe_unit = {input_descs,
                                         input_col_descs,
                                         {},  // quals_cf.simple_quals,
                                         {},  // rewrite_quals(quals_cf.quals),
                                         {},
                                         {nullptr},
                                         target_exprs_pair[0],
                                         {},
                                         nullptr,
                                         sort_info,
                                         max_num_tuples,
                                         RegisteredQueryHint::defaults(),
                                         EMPTY_HASHED_PLAN_DAG_KEY,
                                         {},
                                         {},
                                         false,
                                         logical_union->isAll(),
                                         query_state_,
                                         target_exprs_pair[1]};
   auto query_rewriter = std::make_unique<QueryRewriter>(query_infos, executor_);
   const auto rewritten_exe_unit = query_rewriter->rewrite(exe_unit);
 
   RelAlgNode const* input0 = logical_union->getInput(0);
   if (auto const* node = dynamic_cast<const RelCompound*>(input0)) {
     logical_union->setOutputMetainfo(
         get_targets_meta(node, rewritten_exe_unit.target_exprs));
   } else if (auto const* node = dynamic_cast<const RelProject*>(input0)) {
     logical_union->setOutputMetainfo(
         get_targets_meta(node, rewritten_exe_unit.target_exprs));
   } else if (auto const* node = dynamic_cast<const RelLogicalUnion*>(input0)) {
     logical_union->setOutputMetainfo(
         get_targets_meta(node, rewritten_exe_unit.target_exprs));
   } else if (auto const* node = dynamic_cast<const RelAggregate*>(input0)) {
     logical_union->setOutputMetainfo(
         get_targets_meta(node, rewritten_exe_unit.target_exprs));
   } else if (auto const* node = dynamic_cast<const RelScan*>(input0)) {
     logical_union->setOutputMetainfo(
         get_targets_meta(node, rewritten_exe_unit.target_exprs));
   } else if (auto const* node = dynamic_cast<const RelFilter*>(input0)) {
     logical_union->setOutputMetainfo(
         get_targets_meta(node, rewritten_exe_unit.target_exprs));
   } else if (auto const* node = dynamic_cast<const RelLogicalValues*>(input0)) {
     logical_union->setOutputMetainfo(
         get_targets_meta(node, rewritten_exe_unit.target_exprs));
   } else if (dynamic_cast<const RelSort*>(input0)) {
     throw QueryNotSupported("LIMIT and OFFSET are not currently supported with UNION.");
   } else {
     throw QueryNotSupported("Unsupported input type: " +
                             input0->toString(RelRexToStringConfig::defaults()));
   }
   VLOG(3) << "logical_union->getOutputMetainfo()="
           << shared::printContainer(logical_union->getOutputMetainfo())
           << " rewritten_exe_unit.input_col_descs.front()->getScanDesc().getTableKey()="
           << rewritten_exe_unit.input_col_descs.front()->getScanDesc().getTableKey();
 
   return {rewritten_exe_unit,
           logical_union,
           g_default_max_groups_buffer_entry_guess,
           std::move(query_rewriter)};
 }

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,
		std::unordered_map< QueryPlanHash, std::shared_ptr< HashJoin >> &	partition_cache,
		std::unordered_map< QueryPlanHash, size_t > &	sorted_partition_key_ref_count_map,
		const WorkUnit &	work_unit,
		const std::vector< InputTableInfo > &	query_infos,
		const CompilationOptions &	co,
		ColumnCacheMap &	column_cache_map,
		std::shared_ptr< RowSetMemoryOwner >	row_set_mem_owner
	)

private

Definition at line 2568 of file RelAlgExecutor.cpp.

References RegisteredQueryHint::aggregate_tree_fanout, QueryPlanDagExtractor::applyLimitClauseToCacheKey(), RelAlgExecutor::WorkUnit::body, CHECK, CHECK_EQ, Data_Namespace::CPU_LEVEL, CompilationOptions::device_type, RelAlgExecutor::WorkUnit::exe_unit, executor_, g_window_function_aggregation_tree_fanout, Analyzer::WindowFunction::getArgs(), Analyzer::WindowFunction::getCollation(), ColumnFetcher::getOneColumnFragment(), Analyzer::WindowFunction::getOrderKeys(), RelAlgNode::getQueryPlanDagHash(), GPU, Data_Namespace::GPU_LEVEL, RelAlgExecutionUnit::hash_table_build_plan_dag, hash_value(), Analyzer::WindowFunction::isFrameNavigateWindowFunction(), OneToMany, RelAlgExecutionUnit::query_hint, RelAlgExecutionUnit::sort_info, RelAlgExecutionUnit::table_id_to_node_map, VLOG, WINDOW_FUNCTION, and WINDOW_FUNCTION_FRAMING.

Referenced by computeWindow().

                                                         {
   const size_t elem_count = query_infos.front().info.fragments.front().getNumTuples();
   const auto memory_level = co.device_type == ExecutorDeviceType::GPU
                                 ? MemoryLevel::GPU_LEVEL
                                 : MemoryLevel::CPU_LEVEL;
   std::unique_ptr<WindowFunctionContext> context;
   auto partition_cache_key = QueryPlanDagExtractor::applyLimitClauseToCacheKey(
       work_unit.body->getQueryPlanDagHash(), work_unit.exe_unit.sort_info);
   JoinType window_partition_type = window_func->isFrameNavigateWindowFunction()
                                        ? JoinType::WINDOW_FUNCTION_FRAMING
                                        : JoinType::WINDOW_FUNCTION;
   if (partition_key_cond) {
     auto partition_cond_str = partition_key_cond->toString();
     auto partition_key_hash = boost::hash_value(partition_cond_str);
     boost::hash_combine(partition_cache_key, partition_key_hash);
     boost::hash_combine(partition_cache_key, static_cast<int>(window_partition_type));
     std::shared_ptr<HashJoin> partition_ptr;
     auto cached_hash_table_it = partition_cache.find(partition_cache_key);
     if (cached_hash_table_it != partition_cache.end()) {
       partition_ptr = cached_hash_table_it->second;
       VLOG(1) << "Reuse a hash table to compute window function context (key: "
               << partition_cache_key << ", partition condition: " << partition_cond_str
               << ")";
     } else {
       const auto hash_table_or_err = executor_->buildHashTableForQualifier(
           partition_key_cond,
           query_infos,
           memory_level,
           window_partition_type,
           HashType::OneToMany,
           column_cache_map,
           work_unit.exe_unit.hash_table_build_plan_dag,
           work_unit.exe_unit.query_hint,
           work_unit.exe_unit.table_id_to_node_map);
       if (!hash_table_or_err.fail_reason.empty()) {
         throw std::runtime_error(hash_table_or_err.fail_reason);
       }
       CHECK(hash_table_or_err.hash_table->getHashType() == HashType::OneToMany);
       partition_ptr = hash_table_or_err.hash_table;
       CHECK(partition_cache.insert(std::make_pair(partition_cache_key, partition_ptr))
                 .second);
       VLOG(1) << "Put a generated hash table for computing window function context to "
                  "cache (key: "
               << partition_cache_key << ", partition condition: " << partition_cond_str
               << ")";
     }
     CHECK(partition_ptr);
     auto aggregate_tree_fanout = g_window_function_aggregation_tree_fanout;
     if (work_unit.exe_unit.query_hint.aggregate_tree_fanout != aggregate_tree_fanout) {
       aggregate_tree_fanout = work_unit.exe_unit.query_hint.aggregate_tree_fanout;
       VLOG(1) << "Aggregate tree's fanout is set to " << aggregate_tree_fanout;
     }
     context = std::make_unique<WindowFunctionContext>(window_func,
                                                       partition_cache_key,
                                                       partition_ptr,
                                                       elem_count,
                                                       co.device_type,
                                                       row_set_mem_owner,
                                                       aggregate_tree_fanout);
   } else {
     context = std::make_unique<WindowFunctionContext>(
         window_func, elem_count, co.device_type, row_set_mem_owner);
   }
   const auto& order_keys = window_func->getOrderKeys();
   if (!order_keys.empty()) {
     auto sorted_partition_cache_key = partition_cache_key;
     for (auto& order_key : order_keys) {
       boost::hash_combine(sorted_partition_cache_key, order_key->toString());
     }
     for (auto& collation : window_func->getCollation()) {
       boost::hash_combine(sorted_partition_cache_key, collation.toString());
     }
     context->setSortedPartitionCacheKey(sorted_partition_cache_key);
     auto cache_key_cnt_it =
         sorted_partition_key_ref_count_map.try_emplace(sorted_partition_cache_key, 1);
     if (!cache_key_cnt_it.second) {
       sorted_partition_key_ref_count_map[sorted_partition_cache_key] =
           cache_key_cnt_it.first->second + 1;
     }
 
     std::vector<std::shared_ptr<Chunk_NS::Chunk>> chunks_owner;
     for (const auto& order_key : order_keys) {
       const auto order_col =
           std::dynamic_pointer_cast<const Analyzer::ColumnVar>(order_key);
       if (!order_col) {
         throw std::runtime_error("Only order by columns supported for now");
       }
       auto const [column, col_elem_count] =
           ColumnFetcher::getOneColumnFragment(executor_,
                                               *order_col,
                                               query_infos.front().info.fragments.front(),
                                               memory_level,
                                               0,
                                               nullptr,
                                               /*thread_idx=*/0,
                                               chunks_owner,
                                               column_cache_map);
 
       CHECK_EQ(col_elem_count, elem_count);
       context->addOrderColumn(column, order_col->get_type_info(), chunks_owner);
     }
   }
   if (context->getWindowFunction()->hasFraming() ||
       context->getWindowFunction()->isMissingValueFillingFunction()) {
     // todo (yoonmin) : if we try to support generic window function expression without
     // extra project node, we need to revisit here b/c the current logic assumes that
     // window function expression has a single input source
     auto& window_function_expression_args = window_func->getArgs();
     std::vector<std::shared_ptr<Chunk_NS::Chunk>> chunks_owner;
     for (auto& expr : window_function_expression_args) {
       if (const auto arg_col_var =
               std::dynamic_pointer_cast<const Analyzer::ColumnVar>(expr)) {
         auto const [column, col_elem_count] = ColumnFetcher::getOneColumnFragment(
             executor_,
             *arg_col_var,
             query_infos.front().info.fragments.front(),
             memory_level,
             0,
             nullptr,
             /*thread_idx=*/0,
             chunks_owner,
             column_cache_map);
         CHECK_EQ(col_elem_count, elem_count);
         context->addColumnBufferForWindowFunctionExpression(column, chunks_owner);
       }
     }
   }
   return context;
 }

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 4332 of file RelAlgExecutor.cpp.

References createAggregateWorkUnit(), createCompoundWorkUnit(), createFilterWorkUnit(), createProjectWorkUnit(), RelRexToStringConfig::defaults(), logger::FATAL, ExecutionOptions::just_explain, LOG, and RelAlgNode::toString().

Referenced by createSortInputWorkUnit(), and getJoinInfo().

                                                                                     {
   const auto compound = dynamic_cast<const RelCompound*>(node);
   if (compound) {
     return createCompoundWorkUnit(compound, sort_info, eo);
   }
   const auto project = dynamic_cast<const RelProject*>(node);
   if (project) {
     return createProjectWorkUnit(project, sort_info, eo);
   }
   const auto aggregate = dynamic_cast<const RelAggregate*>(node);
   if (aggregate) {
     return createAggregateWorkUnit(aggregate, sort_info, eo.just_explain);
   }
   const auto filter = dynamic_cast<const RelFilter*>(node);
   if (filter) {
     return createFilterWorkUnit(filter, sort_info, eo.just_explain);
   }
   LOG(FATAL) << "Unhandled node type: "
              << node->toString(RelRexToStringConfig::defaults());
   return {};
 }

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 409 of file RelAlgExecutor.h.

References temporary_tables_.

409 { temporary_tables_.erase(table_id); }

RelAlgExecutor::temporary_tables_

TemporaryTables temporary_tables_

Definition: RelAlgExecutor.h:442

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 2302 of file RelAlgExecutor.cpp.

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

Referenced by executeRelAlgStep().

                                                                               {
   auto timer = DEBUG_TIMER(__func__);
   const auto work_unit = createAggregateWorkUnit(aggregate, SortInfo(), eo.just_explain);
   return executeWorkUnit(work_unit,
                          aggregate->getOutputMetainfo(),
                          true,
                          co,
                          eo,
                          render_info,
                          queue_time_ms);
 }

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 2285 of file RelAlgExecutor.cpp.

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

Referenced by executeRelAlgStep().

                                                                              {
   auto timer = DEBUG_TIMER(__func__);
   const auto work_unit = createCompoundWorkUnit(compound, SortInfo(), eo);
   CompilationOptions co_compound = co;
   return executeWorkUnit(work_unit,
                          compound->getOutputMetainfo(),
                          compound->isAggregate(),
                          co_compound,
                          eo,
                          render_info,
                          queue_time_ms);
 }

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 2180 of file RelAlgExecutor.cpp.

References CHECK, CHECK_EQ, Executor::clearExternalCaches(), createCompoundWorkUnit(), createProjectWorkUnit(), DEBUG_TIMER, RelRexToStringConfig::defaults(), dml_transaction_parameters_, executor_, CompilationOptions::filter_on_deleted_column, get_table_infos(), get_temporary_table(), QueryExecutionError::getErrorCode(), getErrorMessageFromCode(), CompilationOptions::makeCpuOnly(), post_execution_callback_, table_is_temporary(), temporary_tables_, and StorageIOFacility::yieldDeleteCallback().

Referenced by executeRelAlgStep().

                                                                 {
   CHECK(node);
   auto timer = DEBUG_TIMER(__func__);
 
   auto execute_delete_for_node = [this, &co, &eo_in](const auto node,
                                                      auto& work_unit,
                                                      const bool is_aggregate) {
     auto* table_descriptor = node->getModifiedTableDescriptor();
     CHECK(table_descriptor);
     if (!table_descriptor->hasDeletedCol) {
       throw std::runtime_error(
           "DELETE queries are only supported on tables with the vacuum attribute set to "
           "'delayed'");
     }
 
     const auto catalog = node->getModifiedTableCatalog();
     CHECK(catalog);
     Executor::clearExternalCaches(false, table_descriptor, catalog->getDatabaseId());
 
     const auto table_infos = get_table_infos(work_unit.exe_unit, executor_);
 
     auto execute_delete_ra_exe_unit =
         [this, &table_infos, &table_descriptor, &eo_in, &co, catalog](
             const auto& exe_unit, const bool is_aggregate) {
           dml_transaction_parameters_ =
               std::make_unique<DeleteTransactionParameters>(table_descriptor, *catalog);
           auto delete_params = dynamic_cast<DeleteTransactionParameters*>(
               dml_transaction_parameters_.get());
           CHECK(delete_params);
           auto delete_callback = yieldDeleteCallback(*delete_params);
           CompilationOptions co_delete = CompilationOptions::makeCpuOnly(co);
 
           auto eo = eo_in;
           if (dml_transaction_parameters_->tableIsTemporary()) {
             eo.output_columnar_hint = true;
             co_delete.filter_on_deleted_column =
                 false;  // project the entire delete column for columnar update
           } else {
             CHECK_EQ(exe_unit.target_exprs.size(), size_t(1));
           }
 
           try {
             auto table_update_metadata =
                 executor_->executeUpdate(exe_unit,
                                          table_infos,
                                          table_descriptor,
                                          co_delete,
                                          eo,
                                          *catalog,
                                          executor_->row_set_mem_owner_,
                                          delete_callback,
                                          is_aggregate);
             post_execution_callback_ = [table_update_metadata, this, catalog]() {
               dml_transaction_parameters_->finalizeTransaction(*catalog);
               TableOptimizer table_optimizer{
                   dml_transaction_parameters_->getTableDescriptor(), executor_, *catalog};
               table_optimizer.vacuumFragmentsAboveMinSelectivity(table_update_metadata);
             };
           } catch (const QueryExecutionError& e) {
             throw std::runtime_error(getErrorMessageFromCode(e.getErrorCode()));
           }
         };
 
     if (table_is_temporary(table_descriptor)) {
       auto query_rewrite = std::make_unique<QueryRewriter>(table_infos, executor_);
       const auto cd = catalog->getDeletedColumn(table_descriptor);
       CHECK(cd);
       auto delete_column_expr = makeExpr<Analyzer::ColumnVar>(
           cd->columnType,
           shared::ColumnKey{
               catalog->getDatabaseId(), table_descriptor->tableId, cd->columnId},
           0);
       const auto rewritten_exe_unit =
           query_rewrite->rewriteColumnarDelete(work_unit.exe_unit, delete_column_expr);
       execute_delete_ra_exe_unit(rewritten_exe_unit, is_aggregate);
     } else {
       execute_delete_ra_exe_unit(work_unit.exe_unit, is_aggregate);
     }
   };
 
   if (auto compound = dynamic_cast<const RelCompound*>(node)) {
     const auto work_unit = createCompoundWorkUnit(compound, SortInfo(), eo_in);
     execute_delete_for_node(compound, work_unit, compound->isAggregate());
   } else if (auto project = dynamic_cast<const RelProject*>(node)) {
     auto work_unit = createProjectWorkUnit(project, SortInfo(), eo_in);
     if (project->isSimple()) {
       CHECK_EQ(size_t(1), project->inputCount());
       const auto input_ra = project->getInput(0);
       if (dynamic_cast<const RelSort*>(input_ra)) {
         const auto& input_table =
             get_temporary_table(&temporary_tables_, -input_ra->getId());
         CHECK(input_table);
         work_unit.exe_unit.scan_limit = input_table->rowCount();
       }
     }
     execute_delete_for_node(project, work_unit, false);
   } else {
     throw std::runtime_error("Unsupported parent node for delete: " +
                              node->toString(RelRexToStringConfig::defaults()));
   }
 }

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 2707 of file RelAlgExecutor.cpp.

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

Referenced by executeRelAlgStep().

                                                                            {
   auto timer = DEBUG_TIMER(__func__);
   const auto work_unit = createFilterWorkUnit(filter, SortInfo(), eo.just_explain);
   return executeWorkUnit(
       work_unit, filter->getOutputMetainfo(), false, co, eo, render_info, queue_time_ms);
 }

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 2760 of file RelAlgExecutor.cpp.

References CPU, DEBUG_TIMER, executor_, RelLogicalValues::getNumRows(), RelLogicalValues::getTupleType(), kBIGINT, kCOUNT, kNULLT, heavyai::Projection, query_mem_desc, and RelAlgNode::setOutputMetainfo().

Referenced by executeRelAlgStep().

                                 {
   auto timer = DEBUG_TIMER(__func__);
   QueryMemoryDescriptor query_mem_desc(
       executor_, logical_values->getNumRows(), QueryDescriptionType::Projection);
 
   auto tuple_type = logical_values->getTupleType();
   for (size_t i = 0; i < tuple_type.size(); ++i) {
     auto& target_meta_info = tuple_type[i];
     if (target_meta_info.get_type_info().get_type() == kNULLT) {
       // replace w/ bigint
       tuple_type[i] =
           TargetMetaInfo(target_meta_info.get_resname(), SQLTypeInfo(kBIGINT, false));
     }
     query_mem_desc.addColSlotInfo(
         {std::make_tuple(tuple_type[i].get_type_info().get_size(), 8)});
   }
   logical_values->setOutputMetainfo(tuple_type);
 
   std::vector<TargetInfo> target_infos;
   for (const auto& tuple_type_component : tuple_type) {
     target_infos.emplace_back(TargetInfo{false,
                                          kCOUNT,
                                          tuple_type_component.get_type_info(),
                                          SQLTypeInfo(kNULLT, false),
                                          false,
                                          false,
                                          /*is_varlen_projection=*/false});
   }
   std::shared_ptr<ResultSet> rs{
       ResultSetLogicalValuesBuilder{logical_values,
                                     target_infos,
                                     ExecutorDeviceType::CPU,
                                     query_mem_desc,
                                     executor_->getRowSetMemoryOwner(),
                                     executor_}
           .build()};
 
   return {rs, tuple_type};
 }

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 2850 of file RelAlgExecutor.cpp.

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

Referenced by executeRelAlgStep().

                                                                           {
   auto timer = DEBUG_TIMER(__func__);
   if (eo.just_explain) {
     throw std::runtime_error("EXPLAIN not supported for ModifyTable");
   }
 
   auto rs = std::make_shared<ResultSet>(TargetInfoList{},
                                         ExecutorDeviceType::CPU,
                                         QueryMemoryDescriptor(),
                                         executor_->getRowSetMemoryOwner(),
                                         executor_->blockSize(),
                                         executor_->gridSize());
 
   std::vector<TargetMetaInfo> empty_targets;
   return {rs, empty_targets};
 }

Here is the caller graph for this function:

void RelAlgExecutor::executePostExecutionCallback ( )

Definition at line 4325 of file RelAlgExecutor.cpp.

References post_execution_callback_, and VLOG.

                                                   {
   if (post_execution_callback_) {
     VLOG(1) << "Running post execution callback.";
     (*post_execution_callback_)();
   }
 }

ExecutionResult RelAlgExecutor::executeProject	(	const RelProject *	project,
		const CompilationOptions &	co,
		const ExecutionOptions &	eo,
		RenderInfo *	render_info,
		const int64_t	queue_time_ms,
		const std::optional< size_t >	previous_count
	)

private

Definition at line 2331 of file RelAlgExecutor.cpp.

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

Referenced by executeRelAlgStep().

                                               {
   auto timer = DEBUG_TIMER(__func__);
   auto work_unit = createProjectWorkUnit(project, SortInfo(), eo);
   CompilationOptions co_project = co;
   if (project->isSimple()) {
     CHECK_EQ(size_t(1), project->inputCount());
     const auto input_ra = project->getInput(0);
     if (dynamic_cast<const RelSort*>(input_ra)) {
       co_project.device_type = ExecutorDeviceType::CPU;
       const auto& input_table =
           get_temporary_table(&temporary_tables_, -input_ra->getId());
       CHECK(input_table);
       work_unit.exe_unit.scan_limit =
           std::min(input_table->getLimit(), input_table->rowCount());
     }
   }
   return executeWorkUnit(work_unit,
                          project->getOutputMetainfo(),
                          false,
                          co_project,
                          eo,
                          render_info,
                          queue_time_ms,
                          previous_count);
 }

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,
		const bool	explain_verbose,
		RenderInfo *	render_info
	)

Definition at line 564 of file RelAlgExecutor.cpp.

References CHECK, DEBUG_TIMER, executeRelAlgQueryNoRetry(), RenderInfo::forceNonInSitu(), g_allow_cpu_retry, logger::INFO, INJECT_TIMER, RelAlgDag::kBuiltOptimized, LOG, CompilationOptions::makeCpuOnly(), post_execution_callback_, query_dag_, run_benchmark::run_query(), and VLOG.

                                                                             {
   CHECK(query_dag_);
   CHECK(query_dag_->getBuildState() == RelAlgDag::BuildState::kBuiltOptimized)
       << static_cast<int>(query_dag_->getBuildState());
 
   auto timer = DEBUG_TIMER(__func__);
   INJECT_TIMER(executeRelAlgQuery);
 
   auto run_query = [&](const CompilationOptions& co_in) {
     auto execution_result = executeRelAlgQueryNoRetry(
         co_in, eo, just_explain_plan, explain_verbose, render_info);
 
     constexpr bool vlog_result_set_summary{false};
     if constexpr (vlog_result_set_summary) {
       VLOG(1) << execution_result.getRows()->summaryToString();
     }
 
     if (post_execution_callback_) {
       VLOG(1) << "Running post execution callback.";
       (*post_execution_callback_)();
     }
     return execution_result;
   };
 
   try {
     return run_query(co);
   } catch (const QueryMustRunOnCpu&) {
     if (!g_allow_cpu_retry) {
       throw;
     }
   }
   LOG(INFO) << "Query unable to run in GPU mode, retrying on CPU";
   auto co_cpu = CompilationOptions::makeCpuOnly(co);
 
   if (render_info) {
     render_info->forceNonInSitu();
   }
   return run_query(co_cpu);
 }

Here is the call graph for this function:

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

private

Definition at line 608 of file RelAlgExecutor.cpp.

References ExecutionOptions::allow_runtime_query_interrupt, CHECK, cleanupPostExecution(), DEBUG_TIMER, executeRelAlgQueryWithFilterPushDown(), executeRelAlgSeq(), executor_, ExecutionOptions::find_push_down_candidates, g_enable_dynamic_watchdog, getGlobalQueryHint(), RelAlgDag::getNodes(), getParsedQueryHint(), getRelAlgDag(), getRootRelAlgNode(), getSubqueries(), RelAlgDagViewer::handleQueryEngineVector(), QueryExecutionError::hasErrorCode(), INJECT_TIMER, ExecutionOptions::just_calcite_explain, ExecutionOptions::just_explain, ExecutionOptions::just_validate, anonymous_namespace{RelAlgExecutor.cpp}::prepare_for_system_table_execution(), anonymous_namespace{RelAlgExecutor.cpp}::prepare_foreign_table_for_execution(), query_dag_, query_state_, run_benchmark_import::result, RelAlgDag::setGlobalQueryHints(), setupCaching(), timer_start(), and timer_stop().

Referenced by executeRelAlgQuery().

                                                                                    {
   INJECT_TIMER(executeRelAlgQueryNoRetry);
   auto timer = DEBUG_TIMER(__func__);
   auto timer_setup = DEBUG_TIMER("Query pre-execution steps");
 
   query_dag_->resetQueryExecutionState();
   const auto& ra = query_dag_->getRootNode();
 
   // capture the lock acquistion time
   auto clock_begin = timer_start();
   if (g_enable_dynamic_watchdog) {
     executor_->resetInterrupt();
   }
   std::string query_session{""};
   std::string query_str{"N/A"};
   std::string query_submitted_time{""};
   // gather necessary query's info
   if (query_state_ != nullptr && query_state_->getConstSessionInfo() != nullptr) {
     query_session = query_state_->getConstSessionInfo()->get_session_id();
     query_str = query_state_->getQueryStr();
     query_submitted_time = query_state_->getQuerySubmittedTime();
   }
 
   auto validate_or_explain_query =
       just_explain_plan || eo.just_validate || eo.just_explain || eo.just_calcite_explain;
   auto interruptable = !render_info && !query_session.empty() &&
                        eo.allow_runtime_query_interrupt && !validate_or_explain_query;
   if (interruptable) {
     // if we reach here, the current query which was waiting an idle executor
     // within the dispatch queue is now scheduled to the specific executor
     // (not UNITARY_EXECUTOR)
     // so we update the query session's status with the executor that takes this query
     std::tie(query_session, query_str) = executor_->attachExecutorToQuerySession(
         query_session, query_str, query_submitted_time);
 
     // now the query is going to be executed, so update the status as
     // "RUNNING_QUERY_KERNEL"
     executor_->updateQuerySessionStatus(
         query_session,
         query_submitted_time,
         QuerySessionStatus::QueryStatus::RUNNING_QUERY_KERNEL);
   }
 
   // so it should do cleanup session info after finishing its execution
   ScopeGuard clearQuerySessionInfo =
       [this, &query_session, &interruptable, &query_submitted_time] {
         // reset the runtime query interrupt status after the end of query execution
         if (interruptable) {
           // cleanup running session's info
           executor_->clearQuerySessionStatus(query_session, query_submitted_time);
         }
       };
 
   auto acquire_execute_mutex = [](Executor * executor) -> auto{
     auto ret = executor->acquireExecuteMutex();
     return ret;
   };
   // now we acquire executor lock in here to make sure that this executor holds
   // all necessary resources and at the same time protect them against other executor
   auto lock = acquire_execute_mutex(executor_);
 
   if (interruptable) {
     // check whether this query session is "already" interrupted
     // this case occurs when there is very short gap between being interrupted and
     // taking the execute lock
     // if so we have to remove "all" queries initiated by this session and we do in here
     // without running the query
     try {
       executor_->checkPendingQueryStatus(query_session);
     } catch (QueryExecutionError& e) {
       if (e.hasErrorCode(ErrorCode::INTERRUPTED)) {
         throw std::runtime_error("Query execution has been interrupted (pending query)");
       }
       throw e;
     } catch (...) {
       throw std::runtime_error("Checking pending query status failed: unknown error");
     }
   }
   int64_t queue_time_ms = timer_stop(clock_begin);
 
   prepare_for_system_table_execution(ra, co);
 
   // Notify foreign tables to load prior to caching
   prepare_foreign_table_for_execution(ra);
 
   ScopeGuard row_set_holder = [this] { cleanupPostExecution(); };
   setupCaching(&ra);
 
   ScopeGuard restore_metainfo_cache = [this] { executor_->clearMetaInfoCache(); };
   auto ed_seq = RaExecutionSequence(&ra, executor_);
 
   if (just_explain_plan) {
     const auto& ra = getRootRelAlgNode();
     auto ed_seq = RaExecutionSequence(&ra, executor_);
     std::vector<const RelAlgNode*> steps;
     for (size_t i = 0; i < ed_seq.size(); i++) {
       steps.emplace_back(ed_seq.getDescriptor(i)->getBody());
       steps.back()->setStepNumber(i + 1);
     }
     std::stringstream ss;
     auto& nodes = getRelAlgDag()->getNodes();
     RelAlgDagViewer dagv(false, explain_verbose);
     dagv.handleQueryEngineVector(nodes);
     ss << dagv;
     auto rs = std::make_shared<ResultSet>(ss.str());
     return {rs, {}};
   }
 
   if (eo.find_push_down_candidates) {
     // this extra logic is mainly due to current limitations on multi-step queries
     // and/or subqueries.
     return executeRelAlgQueryWithFilterPushDown(
         ed_seq, co, eo, render_info, queue_time_ms);
   }
   timer_setup.stop();
 
   // Dispatch the subqueries first
   const auto global_hints = getGlobalQueryHint();
   for (auto subquery : getSubqueries()) {
     const auto subquery_ra = subquery->getRelAlg();
     CHECK(subquery_ra);
     if (subquery_ra->hasContextData()) {
       continue;
     }
     // Execute the subquery and cache the result.
     RelAlgExecutor subquery_executor(executor_, query_state_);
     // Propagate global and local query hint if necessary
     const auto local_hints = getParsedQueryHint(subquery_ra);
     if (global_hints || local_hints) {
       const auto subquery_rel_alg_dag = subquery_executor.getRelAlgDag();
       if (global_hints) {
         subquery_rel_alg_dag->setGlobalQueryHints(*global_hints);
       }
       if (local_hints) {
         subquery_rel_alg_dag->registerQueryHint(subquery_ra, *local_hints);
       }
     }
     RaExecutionSequence subquery_seq(subquery_ra, executor_);
     auto result = subquery_executor.executeRelAlgSeq(subquery_seq, co, eo, nullptr, 0);
     subquery->setExecutionResult(std::make_shared<ExecutionResult>(result));
   }
   return executeRelAlgSeq(ed_seq, co, eo, render_info, queue_time_ms);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

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

Definition at line 806 of file RelAlgExecutor.cpp.

References CHECK, CHECK_EQ, anonymous_namespace{RelAlgExecutor.cpp}::check_sort_node_source_constraint(), CPU, createSortInputWorkUnit(), CompilationOptions::device_type, executeRelAlgSubSeq(), RaExecutionSequence::getDescriptor(), GPU, logger::INFO, INJECT_TIMER, ExecutionOptions::just_validate, LOG, anonymous_namespace{RelAlgExecutor.cpp}::node_is_aggregate(), post_execution_callback_, queue_time_ms_, Reduce, GroupByAndAggregate::shard_count_for_top_groups(), gpu_enabled::sort(), Union, and VLOG.

                              {
   INJECT_TIMER(executeRelAlgQueryStep);
 
   auto exe_desc_ptr = seq.getDescriptor(step_idx);
   CHECK(exe_desc_ptr);
   const auto sort = dynamic_cast<const RelSort*>(exe_desc_ptr->getBody());
 
   size_t shard_count{0};
   auto merge_type = [&shard_count](const RelAlgNode* body) -> MergeType {
     return node_is_aggregate(body) && !shard_count ? MergeType::Reduce : MergeType::Union;
   };
 
   if (sort) {
     check_sort_node_source_constraint(sort);
     auto order_entries = sort->getOrderEntries();
     const auto source_work_unit = createSortInputWorkUnit(sort, order_entries, eo);
     shard_count =
         GroupByAndAggregate::shard_count_for_top_groups(source_work_unit.exe_unit);
     if (!shard_count) {
       // No point in sorting on the leaf, only execute the input to the sort node.
       CHECK_EQ(size_t(1), sort->inputCount());
       const auto source = sort->getInput(0);
       if (sort->collationCount() || node_is_aggregate(source)) {
         auto temp_seq = RaExecutionSequence(std::make_unique<RaExecutionDesc>(source));
         CHECK_EQ(temp_seq.size(), size_t(1));
         ExecutionOptions eo_copy = eo;
         eo_copy.just_validate = eo.just_validate || sort->isEmptyResult();
         // Use subseq to avoid clearing existing temporary tables
         return {
             executeRelAlgSubSeq(temp_seq, std::make_pair(0, 1), co, eo_copy, nullptr, 0),
             merge_type(source),
             source->getId(),
             false};
       }
     }
   }
   QueryStepExecutionResult query_step_result{ExecutionResult{},
                                              merge_type(exe_desc_ptr->getBody()),
                                              exe_desc_ptr->getBody()->getId(),
                                              false};
   try {
     query_step_result.result = executeRelAlgSubSeq(
         seq, std::make_pair(step_idx, step_idx + 1), co, eo, render_info, queue_time_ms_);
   } catch (QueryMustRunOnCpu const& e) {
     CHECK_EQ(co.device_type, ExecutorDeviceType::GPU);
     auto copied_co = co;
     copied_co.device_type = ExecutorDeviceType::CPU;
     LOG(INFO) << "Retry the query via CPU mode";
     query_step_result.result = executeRelAlgSubSeq(seq,
                                                    std::make_pair(step_idx, step_idx + 1),
                                                    copied_co,
                                                    eo,
                                                    render_info,
                                                    queue_time_ms_);
   }
   if (post_execution_callback_) {
     VLOG(1) << "Running post execution callback.";
     (*post_execution_callback_)();
   }
   return query_step_result;
 }

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 148 of file JoinFilterPushDown.cpp.

References CHECK, executeRelAlgSeq(), executor_, ExecutionOptions::find_push_down_candidates, getSubqueries(), ExecutionOptions::just_calcite_explain, run_benchmark_import::result, and RaExecutionSequence::size().

Referenced by executeRelAlgQueryNoRetry().

                                  {
   // we currently do not fully support filter push down with
   // multi-step execution and/or with subqueries
   // TODO(Saman): add proper caching to enable filter push down for all cases
   const auto& subqueries = getSubqueries();
   if (seq.size() > 1 || !subqueries.empty()) {
     if (eo.just_calcite_explain) {
       return ExecutionResult(std::vector<PushedDownFilterInfo>{},
                              eo.find_push_down_candidates);
     }
     auto eo_modified = eo;
     eo_modified.find_push_down_candidates = false;
     eo_modified.just_calcite_explain = false;
 
     // Dispatch the subqueries first
     for (auto subquery : subqueries) {
       // Execute the subquery and cache the result.
       RelAlgExecutor ra_executor(executor_);
       const auto subquery_ra = subquery->getRelAlg();
       CHECK(subquery_ra);
       RaExecutionSequence subquery_seq(subquery_ra, executor_);
       auto result =
           ra_executor.executeRelAlgSeq(subquery_seq, co, eo_modified, nullptr, 0);
       subquery->setExecutionResult(std::make_shared<ExecutionResult>(result));
     }
     return executeRelAlgSeq(seq, co, eo_modified, render_info, queue_time_ms);
   }
   // else
   return executeRelAlgSeq(seq, co, eo, render_info, queue_time_ms);
 }

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 890 of file RelAlgExecutor.cpp.

References addTemporaryTable(), CHECK, CHECK_GE, DEBUG_TIMER, CompilationOptions::device_type, RaExecutionSequence::empty(), executeRelAlgStep(), executor_, RenderInfo::forceNonInSitu(), g_allow_query_step_cpu_retry, g_allow_query_step_skipping, g_cluster, g_enable_interop, RaExecutionDesc::getBody(), RaExecutionSequence::getDescriptor(), RaExecutionSequence::getSkippedQueryStepCacheKeys(), GPU, RaExecutionSequence::hasQueryStepForUnion(), logger::INFO, INJECT_TIMER, ExecutionOptions::just_explain, ExecutionOptions::keep_result, left_deep_join_info_, LOG, CompilationOptions::makeCpuOnly(), now_, RaExecutionSequence::size(), gpu_enabled::swap(), target_exprs_owned_, temporary_tables_, and VLOG.

Referenced by executeRelAlgQueryNoRetry(), and executeRelAlgQueryWithFilterPushDown().

                                                                                        {
   INJECT_TIMER(executeRelAlgSeq);
   auto timer = DEBUG_TIMER(__func__);
   if (!with_existing_temp_tables) {
     decltype(temporary_tables_)().swap(temporary_tables_);
   }
   decltype(target_exprs_owned_)().swap(target_exprs_owned_);
   decltype(left_deep_join_info_)().swap(left_deep_join_info_);
   executor_->temporary_tables_ = &temporary_tables_;
 
   time(&now_);
   CHECK(!seq.empty());
 
   auto get_descriptor_count = [&seq, &eo]() -> size_t {
     if (eo.just_explain) {
       if (dynamic_cast<const RelLogicalValues*>(seq.getDescriptor(0)->getBody())) {
         // run the logical values descriptor to generate the result set, then the next
         // descriptor to generate the explain
         CHECK_GE(seq.size(), size_t(2));
         return 2;
       } else {
         return 1;
       }
     } else {
       return seq.size();
     }
   };
 
   const auto exec_desc_count = get_descriptor_count();
   auto eo_copied = eo;
   if (seq.hasQueryStepForUnion()) {
     // we currently do not support resultset recycling when an input query
     // contains union (all) operation
     eo_copied.keep_result = false;
   }
 
   // we have to register resultset(s) of the skipped query step(s) as temporary table
   // before executing the remaining query steps
   // since they may be required during the query processing
   // i.e., get metadata of the target expression from the skipped query step
   if (g_allow_query_step_skipping) {
     for (const auto& kv : seq.getSkippedQueryStepCacheKeys()) {
       const auto cached_res =
           executor_->getResultSetRecyclerHolder().getCachedQueryResultSet(kv.second);
       CHECK(cached_res);
       addTemporaryTable(kv.first, cached_res);
     }
   }
 
   const auto num_steps = exec_desc_count - 1;
   for (size_t i = 0; i < exec_desc_count; i++) {
     VLOG(1) << "Executing query step " << i << " / " << num_steps;
     try {
       executeRelAlgStep(
           seq, i, co, eo_copied, (i == num_steps) ? render_info : nullptr, queue_time_ms);
     } catch (const QueryMustRunOnCpu&) {
       // Do not allow per-step retry if flag is off or in distributed mode
       // TODO(todd): Determine if and when we can relax this restriction
       // for distributed
       CHECK(co.device_type == ExecutorDeviceType::GPU);
       if (!g_allow_query_step_cpu_retry || g_cluster) {
         throw;
       }
       LOG(INFO) << "Retrying current query step " << i << " / " << num_steps << " on CPU";
       const auto co_cpu = CompilationOptions::makeCpuOnly(co);
       if (render_info && i == num_steps) {
         // only render on the last step
         render_info->forceNonInSitu();
       }
       executeRelAlgStep(seq,
                         i,
                         co_cpu,
                         eo_copied,
                         (i == num_steps) ? render_info : nullptr,
                         queue_time_ms);
     } catch (const NativeExecutionError&) {
       if (!g_enable_interop) {
         throw;
       }
       auto eo_extern = eo_copied;
       eo_extern.executor_type = ::ExecutorType::Extern;
       auto exec_desc_ptr = seq.getDescriptor(i);
       const auto body = exec_desc_ptr->getBody();
       const auto compound = dynamic_cast<const RelCompound*>(body);
       if (compound && (compound->getGroupByCount() || compound->isAggregate())) {
         LOG(INFO) << "Also failed to run the query using interoperability";
         throw;
       }
       executeRelAlgStep(
           seq, i, co, eo_extern, (i == num_steps) ? render_info : nullptr, queue_time_ms);
     }
   }
 
   return seq.getDescriptor(num_steps)->getResult();
 }

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 1154 of file RelAlgExecutor.cpp.

References addTemporaryTable(), QueryPlanDagExtractor::applyLimitClauseToCacheKey(), build_render_targets(), canUseResultsetCache(), CHECK, SortInfo::createFromSortNode(), DEBUG_TIMER, RelRexToStringConfig::defaults(), executeAggregate(), executeCompound(), executeDelete(), executeFilter(), executeLogicalValues(), executeModify(), executeProject(), executeSort(), executeTableFunction(), executeUnion(), executeUpdate(), executor_, logger::FATAL, g_cluster, g_skip_intermediate_count, RelAlgNode::getContextData(), RaExecutionSequence::getDescriptor(), RaExecutionSequence::getDescriptorByBodyId(), RelAlgNode::getId(), RelAlgNode::getInput(), getParsedQueryHint(), anonymous_namespace{RelAlgDag.cpp}::handle_query_hint(), handleNop(), anonymous_namespace{RelAlgExecutor.cpp}::has_valid_query_plan_dag(), RelAlgNode::hasContextData(), RaExecutionSequence::hasQueryStepForUnion(), INJECT_TIMER, LOG, ExecutionOptions::outer_fragment_indices, setHasStepForUnion(), gpu_enabled::sort(), VLOG, and ExecutionOptions::with_watchdog.

Referenced by executeRelAlgSeq(), and executeRelAlgSubSeq().

                                                                     {
   INJECT_TIMER(executeRelAlgStep);
   auto timer = DEBUG_TIMER(__func__);
   auto exec_desc_ptr = seq.getDescriptor(step_idx);
   CHECK(exec_desc_ptr);
   auto& exec_desc = *exec_desc_ptr;
   const auto body = exec_desc.getBody();
   if (body->isNop()) {
     handleNop(exec_desc);
     return;
   }
   ExecutionOptions eo_copied = eo;
   CompilationOptions co_copied = co;
   eo_copied.with_watchdog =
       eo.with_watchdog && (step_idx == 0 || dynamic_cast<const RelProject*>(body));
   eo_copied.outer_fragment_indices =
       step_idx == 0 ? eo.outer_fragment_indices : std::vector<size_t>();
 
   auto target_node = body;
   auto query_plan_dag_hash = body->getQueryPlanDagHash();
   if (auto sort_body = dynamic_cast<const RelSort*>(body)) {
     target_node = sort_body->getInput(0);
     query_plan_dag_hash = QueryPlanDagExtractor::applyLimitClauseToCacheKey(
         target_node->getQueryPlanDagHash(), SortInfo::createFromSortNode(sort_body));
   } else {
     query_plan_dag_hash = QueryPlanDagExtractor::applyLimitClauseToCacheKey(
         target_node->getQueryPlanDagHash(), SortInfo());
   }
   auto query_hints = getParsedQueryHint(target_node);
   if (query_hints) {
     handle_query_hint(*query_hints, eo_copied, co_copied);
   }
 
   setHasStepForUnion(seq.hasQueryStepForUnion());
   if (canUseResultsetCache(eo, render_info) && has_valid_query_plan_dag(body)) {
     if (auto cached_resultset =
             executor_->getResultSetRecyclerHolder().getCachedQueryResultSet(
                 query_plan_dag_hash)) {
       VLOG(1) << "recycle resultset of the root node " << body->getRelNodeDagId()
               << " from resultset cache";
       body->setOutputMetainfo(cached_resultset->getTargetMetaInfo());
       if (render_info) {
         std::vector<std::shared_ptr<Analyzer::Expr>>& cached_target_exprs =
             executor_->getResultSetRecyclerHolder().getTargetExprs(query_plan_dag_hash);
         std::vector<Analyzer::Expr*> copied_target_exprs;
         for (const auto& expr : cached_target_exprs) {
           copied_target_exprs.push_back(expr.get());
         }
         build_render_targets(
             *render_info, copied_target_exprs, cached_resultset->getTargetMetaInfo());
       }
       exec_desc.setResult({cached_resultset, cached_resultset->getTargetMetaInfo()});
       addTemporaryTable(-body->getId(), exec_desc.getResult().getDataPtr());
       return;
     }
   }
 
   const auto compound = dynamic_cast<const RelCompound*>(body);
   if (compound) {
     if (compound->isDeleteViaSelect()) {
       executeDelete(compound, co_copied, eo_copied, queue_time_ms);
     } else if (compound->isUpdateViaSelect()) {
       executeUpdate(compound, co_copied, eo_copied, queue_time_ms);
     } else {
       exec_desc.setResult(
           executeCompound(compound, co_copied, eo_copied, render_info, queue_time_ms));
       VLOG(3) << "Returned from executeCompound(), addTemporaryTable("
               << static_cast<int>(-compound->getId()) << ", ...)"
               << " exec_desc.getResult().getDataPtr()->rowCount()="
               << exec_desc.getResult().getDataPtr()->rowCount();
       if (exec_desc.getResult().isFilterPushDownEnabled()) {
         return;
       }
       addTemporaryTable(-compound->getId(), exec_desc.getResult().getDataPtr());
     }
     return;
   }
   const auto project = dynamic_cast<const RelProject*>(body);
   if (project) {
     if (project->isDeleteViaSelect()) {
       executeDelete(project, co_copied, eo_copied, queue_time_ms);
     } else if (project->isUpdateViaSelect()) {
       executeUpdate(project, co_copied, eo_copied, queue_time_ms);
     } else {
       std::optional<size_t> prev_count;
       // Disabling the intermediate count optimization in distributed, as the previous
       // execution descriptor will likely not hold the aggregated result.
       if (g_skip_intermediate_count && step_idx > 0 && !g_cluster) {
         // If the previous node produced a reliable count, skip the pre-flight count.
         RelAlgNode const* const prev_body = project->getInput(0);
         if (shared::dynamic_castable_to_any<RelCompound, RelLogicalValues>(prev_body)) {
           if (RaExecutionDesc const* const prev_exec_desc =
                   prev_body->hasContextData()
                       ? prev_body->getContextData()
                       : seq.getDescriptorByBodyId(prev_body->getId(), step_idx - 1)) {
             const auto& prev_exe_result = prev_exec_desc->getResult();
             const auto prev_result = prev_exe_result.getRows();
             if (prev_result) {
               prev_count = prev_result->rowCount();
               VLOG(3) << "Setting output row count for projection node to previous node ("
                       << prev_exec_desc->getBody()->toString(
                              RelRexToStringConfig::defaults())
                       << ") to " << *prev_count;
             }
           }
         }
       }
       exec_desc.setResult(executeProject(
           project, co_copied, eo_copied, render_info, queue_time_ms, prev_count));
       VLOG(3) << "Returned from executeProject(), addTemporaryTable("
               << static_cast<int>(-project->getId()) << ", ...)"
               << " exec_desc.getResult().getDataPtr()->rowCount()="
               << exec_desc.getResult().getDataPtr()->rowCount();
       if (exec_desc.getResult().isFilterPushDownEnabled()) {
         return;
       }
       addTemporaryTable(-project->getId(), exec_desc.getResult().getDataPtr());
     }
     return;
   }
   const auto aggregate = dynamic_cast<const RelAggregate*>(body);
   if (aggregate) {
     exec_desc.setResult(
         executeAggregate(aggregate, co_copied, eo_copied, render_info, queue_time_ms));
     addTemporaryTable(-aggregate->getId(), exec_desc.getResult().getDataPtr());
     return;
   }
   const auto filter = dynamic_cast<const RelFilter*>(body);
   if (filter) {
     exec_desc.setResult(
         executeFilter(filter, co_copied, eo_copied, render_info, queue_time_ms));
     addTemporaryTable(-filter->getId(), exec_desc.getResult().getDataPtr());
     return;
   }
   const auto sort = dynamic_cast<const RelSort*>(body);
   if (sort) {
     exec_desc.setResult(
         executeSort(sort, co_copied, eo_copied, render_info, queue_time_ms));
     if (exec_desc.getResult().isFilterPushDownEnabled()) {
       return;
     }
     addTemporaryTable(-sort->getId(), exec_desc.getResult().getDataPtr());
     return;
   }
   const auto logical_values = dynamic_cast<const RelLogicalValues*>(body);
   if (logical_values) {
     exec_desc.setResult(executeLogicalValues(logical_values, eo_copied));
     addTemporaryTable(-logical_values->getId(), exec_desc.getResult().getDataPtr());
     return;
   }
   const auto modify = dynamic_cast<const RelModify*>(body);
   if (modify) {
     exec_desc.setResult(executeModify(modify, eo_copied));
     return;
   }
   const auto logical_union = dynamic_cast<const RelLogicalUnion*>(body);
   if (logical_union) {
     exec_desc.setResult(executeUnion(
         logical_union, seq, co_copied, eo_copied, render_info, queue_time_ms));
     addTemporaryTable(-logical_union->getId(), exec_desc.getResult().getDataPtr());
     return;
   }
   const auto table_func = dynamic_cast<const RelTableFunction*>(body);
   if (table_func) {
     exec_desc.setResult(
         executeTableFunction(table_func, co_copied, eo_copied, queue_time_ms));
     addTemporaryTable(-table_func->getId(), exec_desc.getResult().getDataPtr());
     return;
   }
   LOG(FATAL) << "Unhandled body type: "
              << body->toString(RelRexToStringConfig::defaults());
 }

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 991 of file RelAlgExecutor.cpp.

References CHECK, CompilationOptions::device_type, executeRelAlgStep(), executor_, RenderInfo::forceNonInSitu(), g_allow_query_step_cpu_retry, g_cluster, RaExecutionSequence::getDescriptor(), GPU, logger::INFO, INJECT_TIMER, left_deep_join_info_, LOG, CompilationOptions::makeCpuOnly(), now_, gpu_enabled::swap(), and temporary_tables_.

Referenced by executeRelAlgQuerySingleStep().

                                  {
   INJECT_TIMER(executeRelAlgSubSeq);
   executor_->temporary_tables_ = &temporary_tables_;
   decltype(left_deep_join_info_)().swap(left_deep_join_info_);
   time(&now_);
   for (size_t i = interval.first; i < interval.second; i++) {
     // only render on the last step
     try {
       executeRelAlgStep(seq,
                         i,
                         co,
                         eo,
                         (i == interval.second - 1) ? render_info : nullptr,
                         queue_time_ms);
     } catch (const QueryMustRunOnCpu&) {
       // Do not allow per-step retry if flag is off or in distributed mode
       // TODO(todd): Determine if and when we can relax this restriction
       // for distributed
       CHECK(co.device_type == ExecutorDeviceType::GPU);
       if (!g_allow_query_step_cpu_retry || g_cluster) {
         throw;
       }
       LOG(INFO) << "Retrying current query step " << i << " on CPU";
       const auto co_cpu = CompilationOptions::makeCpuOnly(co);
       if (render_info && i == interval.second - 1) {
         render_info->forceNonInSitu();
       }
       executeRelAlgStep(seq,
                         i,
                         co_cpu,
                         eo,
                         (i == interval.second - 1) ? render_info : nullptr,
                         queue_time_ms);
     }
   }
 
   return seq.getDescriptor(interval.second - 1)->getResult();
 }

Here is the call graph for this function:

Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeSimpleInsert	(	const Analyzer::Query &	insert_query,
		Fragmenter_Namespace::InsertDataLoader &	inserter,
		const Catalog_Namespace::SessionInfo &	session
	)

Definition at line 2868 of file RelAlgExecutor.cpp.

References append_datum(), DataBlockPtr::arraysPtr, CHECK, CHECK_EQ, checked_malloc(), Executor::clearExternalCaches(), Fragmenter_Namespace::InsertData::columnIds, ColumnDescriptor::columnType, CPU, Fragmenter_Namespace::InsertData::data, Fragmenter_Namespace::InsertData::databaseId, executor_, import_export::fill_missing_columns(), 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(), SQLTypeInfo::get_type(), Analyzer::Query::get_values_lists(), Catalog_Namespace::SessionInfo::getCatalog(), Fragmenter_Namespace::InsertDataLoader::getLeafCount(), Catalog_Namespace::Catalog::getMetadataForTable(), inline_fixed_encoding_null_val(), anonymous_namespace{RelAlgExecutor.cpp}::insert_one_dict_str(), Fragmenter_Namespace::InsertDataLoader::insertData(), is_null(), SQLTypeInfo::is_string(), kARRAY, kBIGINT, kBOOLEAN, kCAST, kCHAR, kDATE, kDECIMAL, kDOUBLE, kENCODING_DICT, kENCODING_NONE, kFLOAT, kINT, kLINESTRING, kMULTILINESTRING, kMULTIPOINT, 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(), DataBlockPtr::stringsPtr, Fragmenter_Namespace::InsertData::tableId, and to_string().

                                                  {
   // Note: We currently obtain an executor for this method, but we do not need it.
   // Therefore, we skip the executor state setup in the regular execution path. In the
   // future, we will likely want to use the executor to evaluate expressions in the insert
   // statement.
 
   const auto& values_lists = query.get_values_lists();
   const int table_id = query.get_result_table_id();
   const auto& col_id_list = query.get_result_col_list();
   size_t rows_number = values_lists.size();
   size_t leaf_count = inserter.getLeafCount();
   const auto& catalog = session.getCatalog();
   const auto td = catalog.getMetadataForTable(table_id);
   CHECK(td);
   size_t rows_per_leaf = rows_number;
   if (td->nShards == 0) {
     rows_per_leaf =
         ceil(static_cast<double>(rows_number) / static_cast<double>(leaf_count));
   }
   auto max_number_of_rows_per_package =
       std::max(size_t(1), std::min(rows_per_leaf, size_t(64 * 1024)));
 
   std::vector<const ColumnDescriptor*> col_descriptors;
   std::vector<int> col_ids;
   std::unordered_map<int, std::unique_ptr<uint8_t[]>> col_buffers;
   std::unordered_map<int, std::vector<std::string>> str_col_buffers;
   std::unordered_map<int, std::vector<ArrayDatum>> arr_col_buffers;
   std::unordered_map<int, int> sequential_ids;
 
   for (const int col_id : col_id_list) {
     const auto cd = get_column_descriptor({catalog.getDatabaseId(), table_id, col_id});
     const auto col_enc = cd->columnType.get_compression();
     if (cd->columnType.is_string()) {
       switch (col_enc) {
         case kENCODING_NONE: {
           auto it_ok =
               str_col_buffers.insert(std::make_pair(col_id, std::vector<std::string>{}));
           CHECK(it_ok.second);
           break;
         }
         case kENCODING_DICT: {
           const auto dd = catalog.getMetadataForDict(cd->columnType.get_comp_param());
           CHECK(dd);
           const auto it_ok = col_buffers.emplace(
               col_id,
               std::make_unique<uint8_t[]>(cd->columnType.get_size() *
                                           max_number_of_rows_per_package));
           CHECK(it_ok.second);
           break;
         }
         default:
           CHECK(false);
       }
     } else if (cd->columnType.is_geometry()) {
       auto it_ok =
           str_col_buffers.insert(std::make_pair(col_id, std::vector<std::string>{}));
       CHECK(it_ok.second);
     } else if (cd->columnType.is_array()) {
       auto it_ok =
           arr_col_buffers.insert(std::make_pair(col_id, std::vector<ArrayDatum>{}));
       CHECK(it_ok.second);
     } else {
       const auto it_ok = col_buffers.emplace(
           col_id,
           std::unique_ptr<uint8_t[]>(new uint8_t[cd->columnType.get_logical_size() *
                                                  max_number_of_rows_per_package]()));
       CHECK(it_ok.second);
     }
     col_descriptors.push_back(cd);
     sequential_ids[col_id] = col_ids.size();
     col_ids.push_back(col_id);
   }
 
   Executor::clearExternalCaches(true, td, catalog.getDatabaseId());
 
   size_t start_row = 0;
   size_t rows_left = rows_number;
   while (rows_left != 0) {
     // clear the buffers
     for (const auto& kv : col_buffers) {
       memset(kv.second.get(), 0, max_number_of_rows_per_package);
     }
     for (auto& kv : str_col_buffers) {
       kv.second.clear();
     }
     for (auto& kv : arr_col_buffers) {
       kv.second.clear();
     }
 
     auto package_size = std::min(rows_left, max_number_of_rows_per_package);
     // Note: if there will be use cases with batch inserts with lots of rows, it might be
     // more efficient to do the loops below column by column instead of row by row.
     // But for now I consider such a refactoring not worth investigating, as we have more
     // efficient ways to insert many rows anyway.
     for (size_t row_idx = 0; row_idx < package_size; ++row_idx) {
       const auto& values_list = values_lists[row_idx + start_row];
       for (size_t col_idx = 0; col_idx < col_descriptors.size(); ++col_idx) {
         CHECK(values_list.size() == col_descriptors.size());
         auto col_cv =
             dynamic_cast<const Analyzer::Constant*>(values_list[col_idx]->get_expr());
         if (!col_cv) {
           auto col_cast =
               dynamic_cast<const Analyzer::UOper*>(values_list[col_idx]->get_expr());
           CHECK(col_cast);
           CHECK_EQ(kCAST, col_cast->get_optype());
           col_cv = dynamic_cast<const Analyzer::Constant*>(col_cast->get_operand());
         }
         CHECK(col_cv);
         const auto cd = col_descriptors[col_idx];
         auto col_datum = col_cv->get_constval();
         auto col_type = cd->columnType.get_type();
         uint8_t* col_data_bytes{nullptr};
         if (!cd->columnType.is_array() && !cd->columnType.is_geometry() &&
             (!cd->columnType.is_string() ||
              cd->columnType.get_compression() == kENCODING_DICT)) {
           const auto col_data_bytes_it = col_buffers.find(col_ids[col_idx]);
           CHECK(col_data_bytes_it != col_buffers.end());
           col_data_bytes = col_data_bytes_it->second.get();
         }
         switch (col_type) {
           case kBOOLEAN: {
             auto col_data = reinterpret_cast<int8_t*>(col_data_bytes);
             auto null_bool_val =
                 col_datum.boolval == inline_fixed_encoding_null_val(cd->columnType);
             col_data[row_idx] = col_cv->get_is_null() || null_bool_val
                                     ? inline_fixed_encoding_null_val(cd->columnType)
                                     : (col_datum.boolval ? 1 : 0);
             break;
           }
           case kTINYINT: {
             auto col_data = reinterpret_cast<int8_t*>(col_data_bytes);
             col_data[row_idx] = col_cv->get_is_null()
                                     ? inline_fixed_encoding_null_val(cd->columnType)
                                     : col_datum.tinyintval;
             break;
           }
           case kSMALLINT: {
             auto col_data = reinterpret_cast<int16_t*>(col_data_bytes);
             col_data[row_idx] = col_cv->get_is_null()
                                     ? inline_fixed_encoding_null_val(cd->columnType)
                                     : col_datum.smallintval;
             break;
           }
           case kINT: {
             auto col_data = reinterpret_cast<int32_t*>(col_data_bytes);
             col_data[row_idx] = col_cv->get_is_null()
                                     ? inline_fixed_encoding_null_val(cd->columnType)
                                     : col_datum.intval;
             break;
           }
           case kBIGINT:
           case kDECIMAL:
           case kNUMERIC: {
             auto col_data = reinterpret_cast<int64_t*>(col_data_bytes);
             col_data[row_idx] = col_cv->get_is_null()
                                     ? inline_fixed_encoding_null_val(cd->columnType)
                                     : col_datum.bigintval;
             break;
           }
           case kFLOAT: {
             auto col_data = reinterpret_cast<float*>(col_data_bytes);
             col_data[row_idx] = col_datum.floatval;
             break;
           }
           case kDOUBLE: {
             auto col_data = reinterpret_cast<double*>(col_data_bytes);
             col_data[row_idx] = col_datum.doubleval;
             break;
           }
           case kTEXT:
           case kVARCHAR:
           case kCHAR: {
             switch (cd->columnType.get_compression()) {
               case kENCODING_NONE:
                 str_col_buffers[col_ids[col_idx]].push_back(
                     col_datum.stringval ? *col_datum.stringval : "");
                 break;
               case kENCODING_DICT: {
                 switch (cd->columnType.get_size()) {
                   case 1:
                     insert_one_dict_str(
                         &reinterpret_cast<uint8_t*>(col_data_bytes)[row_idx],
                         cd,
                         col_cv,
                         catalog);
                     break;
                   case 2:
                     insert_one_dict_str(
                         &reinterpret_cast<uint16_t*>(col_data_bytes)[row_idx],
                         cd,
                         col_cv,
                         catalog);
                     break;
                   case 4:
                     insert_one_dict_str(
                         &reinterpret_cast<int32_t*>(col_data_bytes)[row_idx],
                         cd,
                         col_cv,
                         catalog);
                     break;
                   default:
                     CHECK(false);
                 }
                 break;
               }
               default:
                 CHECK(false);
             }
             break;
           }
           case kTIME:
           case kTIMESTAMP:
           case kDATE: {
             auto col_data = reinterpret_cast<int64_t*>(col_data_bytes);
             col_data[row_idx] = col_cv->get_is_null()
                                     ? inline_fixed_encoding_null_val(cd->columnType)
                                     : col_datum.bigintval;
             break;
           }
           case kARRAY: {
             const auto is_null = col_cv->get_is_null();
             const auto size = cd->columnType.get_size();
             const SQLTypeInfo elem_ti = cd->columnType.get_elem_type();
             // POINT coords: [un]compressed coords always need to be encoded, even if NULL
             const auto is_point_coords =
                 (cd->isGeoPhyCol && elem_ti.get_type() == kTINYINT);
             if (is_null && !is_point_coords) {
               if (size > 0) {
                 int8_t* buf = (int8_t*)checked_malloc(size);
                 put_null_array(static_cast<void*>(buf), elem_ti, "");
                 for (int8_t* p = buf + elem_ti.get_size(); (p - buf) < size;
                      p += elem_ti.get_size()) {
                   put_null(static_cast<void*>(p), elem_ti, "");
                 }
                 arr_col_buffers[col_ids[col_idx]].emplace_back(size, buf, is_null);
               } else {
                 arr_col_buffers[col_ids[col_idx]].emplace_back(0, nullptr, is_null);
               }
               break;
             }
             const auto l = col_cv->get_value_list();
             size_t len = l.size() * elem_ti.get_size();
             if (size > 0 && static_cast<size_t>(size) != len) {
               throw std::runtime_error("Array column " + cd->columnName + " expects " +
                                        std::to_string(size / elem_ti.get_size()) +
                                        " values, " + "received " +
                                        std::to_string(l.size()));
             }
             if (elem_ti.is_string()) {
               CHECK(kENCODING_DICT == elem_ti.get_compression());
               CHECK(4 == elem_ti.get_size());
 
               int8_t* buf = (int8_t*)checked_malloc(len);
               int32_t* p = reinterpret_cast<int32_t*>(buf);
 
               int elemIndex = 0;
               for (auto& e : l) {
                 auto c = std::dynamic_pointer_cast<Analyzer::Constant>(e);
                 CHECK(c);
                 insert_one_dict_str(
                     &p[elemIndex], cd->columnName, elem_ti, c.get(), catalog);
                 elemIndex++;
               }
               arr_col_buffers[col_ids[col_idx]].push_back(ArrayDatum(len, buf, is_null));
 
             } else {
               int8_t* buf = (int8_t*)checked_malloc(len);
               int8_t* p = buf;
               for (auto& e : l) {
                 auto c = std::dynamic_pointer_cast<Analyzer::Constant>(e);
                 CHECK(c);
                 p = append_datum(p, c->get_constval(), elem_ti);
                 CHECK(p);
               }
               arr_col_buffers[col_ids[col_idx]].push_back(ArrayDatum(len, buf, is_null));
             }
             break;
           }
           case kPOINT:
           case kMULTIPOINT:
           case kLINESTRING:
           case kMULTILINESTRING:
           case kPOLYGON:
           case kMULTIPOLYGON:
             if (col_datum.stringval && col_datum.stringval->empty()) {
               throw std::runtime_error(
                   "Empty values are not allowed for geospatial column \"" +
                   cd->columnName + "\"");
             } else {
               str_col_buffers[col_ids[col_idx]].push_back(
                   col_datum.stringval ? *col_datum.stringval : "");
             }
             break;
           default:
             CHECK(false);
         }
       }
     }
     start_row += package_size;
     rows_left -= package_size;
 
     Fragmenter_Namespace::InsertData insert_data;
     insert_data.databaseId = catalog.getCurrentDB().dbId;
     insert_data.tableId = table_id;
     insert_data.data.resize(col_ids.size());
     insert_data.columnIds = col_ids;
     for (const auto& kv : col_buffers) {
       DataBlockPtr p;
       p.numbersPtr = reinterpret_cast<int8_t*>(kv.second.get());
       insert_data.data[sequential_ids[kv.first]] = p;
     }
     for (auto& kv : str_col_buffers) {
       DataBlockPtr p;
       p.stringsPtr = &kv.second;
       insert_data.data[sequential_ids[kv.first]] = p;
     }
     for (auto& kv : arr_col_buffers) {
       DataBlockPtr p;
       p.arraysPtr = &kv.second;
       insert_data.data[sequential_ids[kv.first]] = p;
     }
     insert_data.numRows = package_size;
     auto data_memory_holder = import_export::fill_missing_columns(&catalog, insert_data);
     inserter.insertData(session, insert_data);
   }
 
   auto rs = std::make_shared<ResultSet>(TargetInfoList{},
                                         ExecutorDeviceType::CPU,
                                         QueryMemoryDescriptor(),
                                         executor_->getRowSetMemoryOwner(),
                                         0,
                                         0);
   std::vector<TargetMetaInfo> empty_targets;
   return {rs, empty_targets};
 }

Here is the call 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 3232 of file RelAlgExecutor.cpp.

References SpeculativeTopNBlacklist::add(), addTemporaryTable(), anonymous_namespace{QueryMemoryDescriptor.cpp}::any_of(), QueryPlanDagExtractor::applyLimitClauseToCacheKey(), build_render_targets(), canUseResultsetCache(), CHECK, CHECK_EQ, anonymous_namespace{RelAlgExecutor.cpp}::check_sort_node_source_constraint(), RelSort::collationCount(), CPU, createSortInputWorkUnit(), DEBUG_TIMER, CompilationOptions::device_type, executeWorkUnit(), executor_, anonymous_namespace{RelAlgExecutor.cpp}::first_oe_is_desc(), g_cluster, anonymous_namespace{RelAlgExecutor.cpp}::get_limit_value(), ExecutionResult::getDataPtr(), RelAlgNode::getId(), RelAlgNode::getInput(), RelSort::getLimit(), RelSort::getOffset(), RelSort::getOrderEntries(), GPU, anonymous_namespace{RelAlgExecutor.cpp}::has_valid_query_plan_dag(), hasStepForUnion(), anonymous_namespace{RelAlgExecutor.cpp}::is_none_encoded_text(), RelSort::isEmptyResult(), ExecutionOptions::just_validate, leaf_results_, anonymous_namespace{RelAlgExecutor.cpp}::node_is_aggregate(), run_benchmark_import::result, RelAlgNode::setOutputMetainfo(), gpu_enabled::sort(), speculative_topn_blacklist_, SpeculativeTopN, temporary_tables_, use_speculative_top_n(), and VLOG.

Referenced by executeRelAlgStep().

                                                                          {
   auto timer = DEBUG_TIMER(__func__);
   check_sort_node_source_constraint(sort);
   const auto source = sort->getInput(0);
   const bool is_aggregate = node_is_aggregate(source);
   auto it = leaf_results_.find(sort->getId());
   auto order_entries = sort->getOrderEntries();
   if (it != leaf_results_.end()) {
     // Add any transient string literals to the sdp on the agg
     const auto source_work_unit = createSortInputWorkUnit(sort, order_entries, eo);
     executor_->addTransientStringLiterals(source_work_unit.exe_unit,
                                           executor_->row_set_mem_owner_);
     // Handle push-down for LIMIT for multi-node
     auto& aggregated_result = it->second;
     auto& result_rows = aggregated_result.rs;
     auto limit = sort->getLimit();
     const size_t offset = sort->getOffset();
     if (limit || offset) {
       if (!order_entries.empty()) {
         result_rows->sort(
             order_entries, get_limit_value(limit) + offset, co.device_type, executor_);
       }
       result_rows->dropFirstN(offset);
       if (limit) {
         result_rows->keepFirstN(get_limit_value(limit));
       }
     }
 
     if (render_info) {
       // We've hit a sort step that is the very last step
       // in a distributed render query. We'll fill in the render targets
       // since we have all that data needed to do so. This is normally
       // done in executeWorkUnit, but that is bypassed in this case.
       build_render_targets(*render_info,
                            source_work_unit.exe_unit.target_exprs,
                            aggregated_result.targets_meta);
     }
 
     ExecutionResult result(result_rows, aggregated_result.targets_meta);
     sort->setOutputMetainfo(aggregated_result.targets_meta);
 
     return result;
   }
 
   std::list<std::shared_ptr<Analyzer::Expr>> groupby_exprs;
   bool is_desc{false};
   bool use_speculative_top_n_sort{false};
 
   auto execute_sort_query = [this,
                              sort,
                              &source,
                              &is_aggregate,
                              &eo,
                              &co,
                              render_info,
                              queue_time_ms,
                              &groupby_exprs,
                              &is_desc,
                              &order_entries,
                              &use_speculative_top_n_sort]() -> ExecutionResult {
     std::optional<size_t> limit = sort->getLimit();
     const size_t offset = sort->getOffset();
     // check whether sort's input is cached
     auto source_node = sort->getInput(0);
     CHECK(source_node);
     ExecutionResult source_result{nullptr, {}};
     SortAlgorithm sort_algorithm{SortAlgorithm::SpeculativeTopN};
     SortInfo sort_info{order_entries, sort_algorithm, limit, offset};
     auto source_query_plan_dag = QueryPlanDagExtractor::applyLimitClauseToCacheKey(
         source_node->getQueryPlanDagHash(), sort_info);
     if (canUseResultsetCache(eo, render_info) && has_valid_query_plan_dag(source_node) &&
         !sort->isEmptyResult()) {
       if (auto cached_resultset =
               executor_->getResultSetRecyclerHolder().getCachedQueryResultSet(
                   source_query_plan_dag)) {
         CHECK(cached_resultset->canUseSpeculativeTopNSort());
         VLOG(1) << "recycle resultset of the root node " << source_node->getRelNodeDagId()
                 << " from resultset cache";
         source_result =
             ExecutionResult{cached_resultset, cached_resultset->getTargetMetaInfo()};
         if (temporary_tables_.find(-source_node->getId()) == temporary_tables_.end()) {
           addTemporaryTable(-source_node->getId(), cached_resultset);
         }
         use_speculative_top_n_sort = *cached_resultset->canUseSpeculativeTopNSort() &&
                                      co.device_type == ExecutorDeviceType::GPU;
         source_node->setOutputMetainfo(cached_resultset->getTargetMetaInfo());
         sort->setOutputMetainfo(source_node->getOutputMetainfo());
       }
     }
     if (!source_result.getDataPtr()) {
       const auto source_work_unit = createSortInputWorkUnit(sort, order_entries, eo);
       is_desc = first_oe_is_desc(order_entries);
       ExecutionOptions eo_copy = eo;
       CompilationOptions co_copy = co;
       eo_copy.just_validate = eo.just_validate || sort->isEmptyResult();
       if (hasStepForUnion() &&
           boost::algorithm::any_of(source->getOutputMetainfo(), is_none_encoded_text)) {
         co_copy.device_type = ExecutorDeviceType::CPU;
         VLOG(1) << "Punt sort's input query to CPU: detect union(-all) of none-encoded "
                    "text column";
       }
 
       groupby_exprs = source_work_unit.exe_unit.groupby_exprs;
       source_result = executeWorkUnit(source_work_unit,
                                       source->getOutputMetainfo(),
                                       is_aggregate,
                                       co_copy,
                                       eo_copy,
                                       render_info,
                                       queue_time_ms);
       use_speculative_top_n_sort =
           source_result.getDataPtr() && source_result.getRows()->hasValidBuffer() &&
           use_speculative_top_n(source_work_unit.exe_unit,
                                 source_result.getRows()->getQueryMemDesc());
     }
     if (render_info && render_info->isInSitu()) {
       return source_result;
     }
     if (source_result.isFilterPushDownEnabled()) {
       return source_result;
     }
     auto rows_to_sort = source_result.getRows();
     if (eo.just_explain) {
       return {rows_to_sort, {}};
     }
     auto const limit_val = get_limit_value(limit);
     if (sort->collationCount() != 0 && !rows_to_sort->definitelyHasNoRows() &&
         !use_speculative_top_n_sort) {
       const size_t top_n = limit_val + offset;
       rows_to_sort->sort(order_entries, top_n, co.device_type, executor_);
     }
     if (limit || offset) {
       if (g_cluster && sort->collationCount() == 0) {
         if (offset >= rows_to_sort->rowCount()) {
           rows_to_sort->dropFirstN(offset);
         } else {
           rows_to_sort->keepFirstN(limit_val + offset);
         }
       } else {
         rows_to_sort->dropFirstN(offset);
         if (limit) {
           rows_to_sort->keepFirstN(limit_val);
         }
       }
     }
     return {rows_to_sort, source_result.getTargetsMeta()};
   };
 
   try {
     return execute_sort_query();
   } catch (const SpeculativeTopNFailed& e) {
     CHECK_EQ(size_t(1), groupby_exprs.size());
     CHECK(groupby_exprs.front());
     speculative_topn_blacklist_.add(groupby_exprs.front(), is_desc);
     return execute_sort_query();
   }
 }

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 2363 of file RelAlgExecutor.cpp.

References addTemporaryTable(), canUseResultsetCache(), CHECK, createTableFunctionWorkUnit(), DEBUG_TIMER, executor_, g_allow_auto_resultset_caching, g_auto_resultset_caching_threshold, g_cluster, g_enable_table_functions, get_table_infos(), QueryExecutionError::getErrorCode(), getGlobalQueryHint(), RelAlgNode::getQueryPlanDagHash(), RelAlgNode::getRelNodeDagId(), ScanNodeTableKeyCollector::getScanNodeTableKey(), GPU, handlePersistentError(), anonymous_namespace{RelAlgExecutor.cpp}::has_valid_query_plan_dag(), QueryExecutionError::hasErrorCode(), hasStepForUnion(), INJECT_TIMER, is_validate_or_explain_query(), ExecutionOptions::just_explain, ExecutionOptions::keep_result, kKeepTableFuncResult, run_benchmark_import::result, target_exprs_owned_, timer_start(), timer_stop(), and VLOG.

Referenced by executeRelAlgStep().

                                                                                   {
   INJECT_TIMER(executeTableFunction);
   auto timer = DEBUG_TIMER(__func__);
 
   auto co = co_in;
 
   if (g_cluster) {
     throw std::runtime_error("Table functions not supported in distributed mode yet");
   }
   if (!g_enable_table_functions) {
     throw std::runtime_error("Table function support is disabled");
   }
   auto table_func_work_unit = createTableFunctionWorkUnit(
       table_func,
       eo.just_explain,
       /*is_gpu = */ co.device_type == ExecutorDeviceType::GPU);
   const auto body = table_func_work_unit.body;
   CHECK(body);
 
   const auto table_infos =
       get_table_infos(table_func_work_unit.exe_unit.input_descs, executor_);
 
   ExecutionResult result{std::make_shared<ResultSet>(std::vector<TargetInfo>{},
                                                      co.device_type,
                                                      QueryMemoryDescriptor(),
                                                      nullptr,
                                                      executor_->blockSize(),
                                                      executor_->gridSize()),
                          {}};
 
   auto const global_hint = getGlobalQueryHint();
   auto const use_resultset_recycler =
       canUseResultsetCache(eo, nullptr) && has_valid_query_plan_dag(table_func);
   if (use_resultset_recycler) {
     auto cached_resultset =
         executor_->getResultSetRecyclerHolder().getCachedQueryResultSet(
             table_func->getQueryPlanDagHash());
     if (cached_resultset) {
       VLOG(1) << "recycle table function's resultset of the root node "
               << table_func->getRelNodeDagId() << " from resultset cache";
       result = {cached_resultset, cached_resultset->getTargetMetaInfo()};
       addTemporaryTable(-body->getId(), result.getDataPtr());
       return result;
     }
   }
 
   auto query_exec_time_begin = timer_start();
   try {
     result = {executor_->executeTableFunction(
                   table_func_work_unit.exe_unit, table_infos, co, eo),
               body->getOutputMetainfo()};
   } catch (const QueryExecutionError& e) {
     handlePersistentError(e.getErrorCode());
     CHECK(e.hasErrorCode(ErrorCode::OUT_OF_GPU_MEM)) << e.getErrorCode();
     throw std::runtime_error("Table function ran out of memory during execution");
   }
   auto query_exec_time = timer_stop(query_exec_time_begin);
   result.setQueueTime(queue_time_ms);
   auto resultset_ptr = result.getDataPtr();
   if (use_resultset_recycler) {
     resultset_ptr->setExecTime(query_exec_time);
     resultset_ptr->setQueryPlanHash(table_func_work_unit.exe_unit.query_plan_dag_hash);
     resultset_ptr->setTargetMetaInfo(body->getOutputMetainfo());
     auto input_table_keys = ScanNodeTableKeyCollector::getScanNodeTableKey(body);
     resultset_ptr->setInputTableKeys(std::move(input_table_keys));
     auto allow_auto_caching_resultset =
         resultset_ptr && resultset_ptr->hasValidBuffer() &&
         g_allow_auto_resultset_caching &&
         resultset_ptr->getBufferSizeBytes(co.device_type) <=
             g_auto_resultset_caching_threshold;
     if (global_hint->isHintRegistered(QueryHint::kKeepTableFuncResult) ||
         allow_auto_caching_resultset) {
       if (allow_auto_caching_resultset) {
         VLOG(1) << "Automatically keep table function's query resultset to recycler";
       }
       executor_->getResultSetRecyclerHolder().putQueryResultSetToCache(
           table_func_work_unit.exe_unit.query_plan_dag_hash,
           resultset_ptr->getInputTableKeys(),
           resultset_ptr,
           resultset_ptr->getBufferSizeBytes(co.device_type),
           target_exprs_owned_);
     }
   } else {
     if (eo.keep_result) {
       if (g_cluster) {
         VLOG(1) << "Query hint \'keep_table_function_result\' is ignored since we do not "
                    "support resultset recycling on distributed mode";
       } else if (hasStepForUnion()) {
         VLOG(1) << "Query hint \'keep_table_function_result\' is ignored since a query "
                    "has union-(all) operator";
       } else if (is_validate_or_explain_query(eo)) {
         VLOG(1) << "Query hint \'keep_table_function_result\' is ignored since a query "
                    "is either validate or explain query";
       } else {
         VLOG(1) << "Query hint \'keep_table_function_result\' is ignored";
       }
     }
   }
 
   return result;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::executeUnion	(	const RelLogicalUnion *	logical_union,
		const RaExecutionSequence &	seq,
		const CompilationOptions &	co,
		const ExecutionOptions &	eo,
		RenderInfo *	render_info,
		const int64_t	queue_time_ms
	)

private

Definition at line 2735 of file RelAlgExecutor.cpp.

References anonymous_namespace{QueryMemoryDescriptor.cpp}::any_of(), createUnionWorkUnit(), DEBUG_TIMER, executeWorkUnit(), RelLogicalUnion::getCompatibleMetainfoTypes(), RelAlgNode::getOutputMetainfo(), RelLogicalUnion::isAll(), isGeometry(), CompilationOptions::makeCpuOnly(), and RelAlgNode::setOutputMetainfo().

Referenced by executeRelAlgStep().

                                                                           {
   auto timer = DEBUG_TIMER(__func__);
   if (!logical_union->isAll()) {
     throw std::runtime_error("UNION without ALL is not supported yet.");
   }
   // Will throw a std::runtime_error if types don't match.
   logical_union->setOutputMetainfo(logical_union->getCompatibleMetainfoTypes());
   if (boost::algorithm::any_of(logical_union->getOutputMetainfo(), isGeometry)) {
     throw std::runtime_error("UNION does not support subqueries with geo-columns.");
   }
   auto work_unit = createUnionWorkUnit(logical_union, SortInfo(), eo);
   return executeWorkUnit(work_unit,
                          logical_union->getOutputMetainfo(),
                          false,
                          CompilationOptions::makeCpuOnly(co),
                          eo,
                          render_info,
                          queue_time_ms);
 }

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 2022 of file RelAlgExecutor.cpp.

References CompilationOptions::allow_lazy_fetch, CHECK, CHECK_EQ, Executor::clearExternalCaches(), computeWindow(), CPU, createCompoundWorkUnit(), createProjectWorkUnit(), DEBUG_TIMER, RelRexToStringConfig::defaults(), dml_transaction_parameters_, executor_, CompilationOptions::filter_on_deleted_column, get_table_infos(), get_temporary_table(), QueryExecutionError::getErrorCode(), getErrorMessageFromCode(), CompilationOptions::hoist_literals, leaf_results_, CompilationOptions::makeCpuOnly(), post_execution_callback_, StorageIOFacility::TransactionParameters::setInputSourceNode(), temporary_tables_, and StorageIOFacility::yieldUpdateCallback().

Referenced by executeRelAlgStep().

                                                                 {
   CHECK(node);
   auto timer = DEBUG_TIMER(__func__);
 
   auto co = co_in;
   co.hoist_literals = false;  // disable literal hoisting as it interferes with dict
                               // encoded string updates
 
   auto execute_update_for_node = [this, &co, &eo_in](const auto node,
                                                      auto& work_unit,
                                                      const bool is_aggregate) {
     auto table_descriptor = node->getModifiedTableDescriptor();
     CHECK(table_descriptor);
     if (node->isVarlenUpdateRequired() && !table_descriptor->hasDeletedCol) {
       throw std::runtime_error(
           "UPDATE queries involving variable length columns are only supported on tables "
           "with the vacuum attribute set to 'delayed'");
     }
 
     auto catalog = node->getModifiedTableCatalog();
     CHECK(catalog);
     Executor::clearExternalCaches(true, table_descriptor, catalog->getDatabaseId());
 
     dml_transaction_parameters_ =
         std::make_unique<UpdateTransactionParameters>(table_descriptor,
                                                       *catalog,
                                                       node->getTargetColumns(),
                                                       node->getOutputMetainfo(),
                                                       node->isVarlenUpdateRequired());
 
     const auto table_infos = get_table_infos(work_unit.exe_unit, executor_);
 
     auto execute_update_ra_exe_unit =
         [this, &co, &eo_in, &table_infos, &table_descriptor, &node, catalog](
             const RelAlgExecutionUnit& ra_exe_unit, const bool is_aggregate) {
           CompilationOptions co_project = CompilationOptions::makeCpuOnly(co);
 
           auto eo = eo_in;
           if (dml_transaction_parameters_->tableIsTemporary()) {
             eo.output_columnar_hint = true;
             co_project.allow_lazy_fetch = false;
             co_project.filter_on_deleted_column =
                 false;  // project the entire delete column for columnar update
           }
 
           auto update_transaction_parameters = dynamic_cast<UpdateTransactionParameters*>(
               dml_transaction_parameters_.get());
           update_transaction_parameters->setInputSourceNode(node);
           CHECK(update_transaction_parameters);
           auto update_callback = yieldUpdateCallback(*update_transaction_parameters);
           try {
             auto table_update_metadata =
                 executor_->executeUpdate(ra_exe_unit,
                                          table_infos,
                                          table_descriptor,
                                          co_project,
                                          eo,
                                          *catalog,
                                          executor_->row_set_mem_owner_,
                                          update_callback,
                                          is_aggregate);
             post_execution_callback_ = [table_update_metadata, this, catalog]() {
               dml_transaction_parameters_->finalizeTransaction(*catalog);
               TableOptimizer table_optimizer{
                   dml_transaction_parameters_->getTableDescriptor(), executor_, *catalog};
               table_optimizer.vacuumFragmentsAboveMinSelectivity(table_update_metadata);
             };
           } catch (const QueryExecutionError& e) {
             throw std::runtime_error(getErrorMessageFromCode(e.getErrorCode()));
           }
         };
 
     if (dml_transaction_parameters_->tableIsTemporary()) {
       // hold owned target exprs during execution if rewriting
       auto query_rewrite = std::make_unique<QueryRewriter>(table_infos, executor_);
       // rewrite temp table updates to generate the full column by moving the where
       // clause into a case if such a rewrite is not possible, bail on the update
       // operation build an expr for the update target
       auto update_transaction_params =
           dynamic_cast<UpdateTransactionParameters*>(dml_transaction_parameters_.get());
       CHECK(update_transaction_params);
       const auto td = update_transaction_params->getTableDescriptor();
       CHECK(td);
       const auto update_column_names = update_transaction_params->getUpdateColumnNames();
       if (update_column_names.size() > 1) {
         throw std::runtime_error(
             "Multi-column update is not yet supported for temporary tables.");
       }
 
       const auto cd =
           catalog->getMetadataForColumn(td->tableId, update_column_names.front());
       CHECK(cd);
       auto projected_column_to_update = makeExpr<Analyzer::ColumnVar>(
           cd->columnType,
           shared::ColumnKey{catalog->getDatabaseId(), td->tableId, cd->columnId},
           0);
       const auto rewritten_exe_unit = query_rewrite->rewriteColumnarUpdate(
           work_unit.exe_unit, projected_column_to_update);
       if (rewritten_exe_unit.target_exprs.front()->get_type_info().is_varlen()) {
         throw std::runtime_error(
             "Variable length updates not yet supported on temporary tables.");
       }
       execute_update_ra_exe_unit(rewritten_exe_unit, is_aggregate);
     } else {
       execute_update_ra_exe_unit(work_unit.exe_unit, is_aggregate);
     }
   };
 
   if (auto compound = dynamic_cast<const RelCompound*>(node)) {
     auto work_unit = createCompoundWorkUnit(compound, SortInfo(), eo_in);
 
     execute_update_for_node(compound, work_unit, compound->isAggregate());
   } else if (auto project = dynamic_cast<const RelProject*>(node)) {
     auto work_unit = createProjectWorkUnit(project, SortInfo(), eo_in);
 
     if (project->isSimple()) {
       CHECK_EQ(size_t(1), project->inputCount());
       const auto input_ra = project->getInput(0);
       if (dynamic_cast<const RelSort*>(input_ra)) {
         const auto& input_table =
             get_temporary_table(&temporary_tables_, -input_ra->getId());
         CHECK(input_table);
         work_unit.exe_unit.scan_limit = input_table->rowCount();
       }
     }
     if (project->hasWindowFunctionExpr() || project->hasPushedDownWindowExpr()) {
       // the first condition means this project node has at least one window function
       // and the second condition indicates that this project node falls into
       // one of the following cases:
       // 1) window function expression on a multi-fragmented table
       // 2) window function expression is too complex to evaluate without codegen:
       // i.e., sum(x+y+z) instead of sum(x) -> we currently do not support codegen to
       // evaluate such a complex window function expression
       // 3) nested window function expression
       // but currently we do not support update on a multi-fragmented table having
       // window function, so the second condition only refers to non-fragmented table with
       // cases 2) or 3)
       // if at least one of two conditions satisfy, we must compute corresponding window
       // context before entering `execute_update_for_node` to properly update the table
       if (!leaf_results_.empty()) {
         throw std::runtime_error(
             "Update query having window function is not yet supported in distributed "
             "mode.");
       }
       ColumnCacheMap column_cache;
       co.device_type = ExecutorDeviceType::CPU;
       computeWindow(work_unit, co, eo_in, column_cache, queue_time_ms);
     }
     execute_update_for_node(project, work_unit, false);
   } else {
     throw std::runtime_error("Unsupported parent node for update: " +
                              node->toString(RelRexToStringConfig::defaults()));
   }
 }

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_in,
		RenderInfo *	render_info,
		const int64_t	queue_time_ms,
		const std::optional< size_t >	previous_count = `std::nullopt`
	)

private

Definition at line 3654 of file RelAlgExecutor.cpp.

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

                                               {
   INJECT_TIMER(executeWorkUnit);
   auto timer = DEBUG_TIMER(__func__);
   auto query_exec_time_begin = timer_start();
 
   const auto query_infos = get_table_infos(work_unit.exe_unit.input_descs, executor_);
   check_none_encoded_string_cast_tuple_limit(query_infos, work_unit.exe_unit);
 
   auto co = co_in;
   auto eo = eo_in;
   ColumnCacheMap column_cache;
   ScopeGuard clearWindowContextIfNecessary = [&]() {
     if (is_window_execution_unit(work_unit.exe_unit)) {
       WindowProjectNodeContext::reset(executor_);
     }
   };
   if (is_window_execution_unit(work_unit.exe_unit)) {
     if (!g_enable_window_functions) {
       throw std::runtime_error("Window functions support is disabled");
     }
     co.device_type = ExecutorDeviceType::CPU;
     co.allow_lazy_fetch = false;
     computeWindow(work_unit, co, eo, column_cache, queue_time_ms);
   }
   if (!eo.just_explain && eo.find_push_down_candidates) {
     // find potential candidates:
     VLOG(1) << "Try to find filter predicate push-down candidate.";
     auto selected_filters = selectFiltersToBePushedDown(work_unit, co, eo);
     if (!selected_filters.empty() || eo.just_calcite_explain) {
       VLOG(1) << "Found " << selected_filters.size()
               << " filter(s) to be pushed down. Re-create a query plan based on pushed "
                  "filter predicate(s).";
       return ExecutionResult(selected_filters, eo.find_push_down_candidates);
     }
     VLOG(1) << "Continue with the current query plan";
   }
   if (render_info && render_info->isInSitu()) {
     co.allow_lazy_fetch = false;
   }
   const auto body = work_unit.body;
   CHECK(body);
   auto it = leaf_results_.find(body->getId());
   VLOG(3) << "body->getId()=" << body->getId()
           << " body->toString()=" << body->toString(RelRexToStringConfig::defaults())
           << " it==leaf_results_.end()=" << (it == leaf_results_.end());
   if (it != leaf_results_.end()) {
     executor_->addTransientStringLiterals(work_unit.exe_unit,
                                           executor_->row_set_mem_owner_);
     auto& aggregated_result = it->second;
     auto& result_rows = aggregated_result.rs;
     ExecutionResult result(result_rows, aggregated_result.targets_meta);
     body->setOutputMetainfo(aggregated_result.targets_meta);
     if (render_info) {
       build_render_targets(*render_info, work_unit.exe_unit.target_exprs, targets_meta);
     }
     return result;
   }
   const auto table_infos = get_table_infos(work_unit.exe_unit, executor_);
 
   auto ra_exe_unit = decide_approx_count_distinct_implementation(
       work_unit.exe_unit, table_infos, executor_, co.device_type, target_exprs_owned_);
 
   // register query hint if query_dag_ is valid
   ra_exe_unit.query_hint = RegisteredQueryHint::defaults();
   if (query_dag_) {
     auto candidate = query_dag_->getQueryHint(body);
     if (candidate) {
       ra_exe_unit.query_hint = *candidate;
     }
   }
 
   const auto& query_hints = ra_exe_unit.query_hint;
   ScopeGuard reset_cuda_block_grid_sizes = [&,
                                             orig_block_size = executor_->blockSize(),
                                             orig_grid_size = executor_->gridSize()]() {
     if (executor_->getDataMgr()->getCudaMgr()) {
       if (query_hints.isHintRegistered(QueryHint::kCudaBlockSize)) {
         if (orig_block_size) {
           executor_->setBlockSize(orig_block_size);
         } else {
           executor_->resetBlockSize();
         }
       }
       if (query_hints.isHintRegistered(QueryHint::kCudaGridSize)) {
         if (orig_grid_size) {
           executor_->setGridSize(orig_grid_size);
         } else {
           executor_->resetGridSize();
         }
       }
     }
   };
 
   if (co.device_type == ExecutorDeviceType::GPU) {
     if (query_hints.isHintRegistered(QueryHint::kCudaGridSize)) {
       if (!executor_->getDataMgr()->getCudaMgr()) {
         VLOG(1) << "Skip CUDA grid size query hint: cannot detect CUDA device";
       } else {
         const auto num_sms = executor_->cudaMgr()->getMinNumMPsForAllDevices();
         const auto new_grid_size = static_cast<unsigned>(
             std::max(1.0, std::round(num_sms * query_hints.cuda_grid_size_multiplier)));
         const auto default_grid_size = executor_->gridSize();
         if (new_grid_size != default_grid_size) {
           VLOG(1) << "Change CUDA grid size: " << default_grid_size
                   << " (default_grid_size) -> " << new_grid_size << " (# SMs * "
                   << query_hints.cuda_grid_size_multiplier << ")";
           // todo (yoonmin): do we need to check a hard limit?
           executor_->setGridSize(new_grid_size);
         } else {
           VLOG(1) << "Skip CUDA grid size query hint: invalid grid size";
         }
       }
     }
     if (query_hints.isHintRegistered(QueryHint::kCudaBlockSize)) {
       if (!executor_->getDataMgr()->getCudaMgr()) {
         VLOG(1) << "Skip CUDA block size query hint: cannot detect CUDA device";
       } else {
         int cuda_block_size = query_hints.cuda_block_size;
         int warp_size = executor_->warpSize();
         if (cuda_block_size >= warp_size) {
           cuda_block_size = (cuda_block_size + warp_size - 1) / warp_size * warp_size;
           VLOG(1) << "Change CUDA block size w.r.t warp size (" << warp_size
                   << "): " << executor_->blockSize() << " -> " << cuda_block_size;
         } else {
           VLOG(1) << "Change CUDA block size: " << executor_->blockSize() << " -> "
                   << cuda_block_size;
         }
         executor_->setBlockSize(cuda_block_size);
       }
     }
   }
 
   auto max_groups_buffer_entry_guess = work_unit.max_groups_buffer_entry_guess;
   if (is_window_execution_unit(ra_exe_unit)) {
     CHECK_EQ(table_infos.size(), size_t(1));
     CHECK_EQ(table_infos.front().info.fragments.size(), size_t(1));
     max_groups_buffer_entry_guess =
         table_infos.front().info.fragments.front().getNumTuples();
     ra_exe_unit.scan_limit = max_groups_buffer_entry_guess;
   } else if (compute_output_buffer_size(ra_exe_unit) && !isRowidLookup(work_unit)) {
     if (previous_count && !exe_unit_has_quals(ra_exe_unit)) {
       ra_exe_unit.scan_limit = *previous_count;
     } else {
       // TODO(adb): enable bump allocator path for render queries
       if (can_use_bump_allocator(ra_exe_unit, co, eo) && !render_info) {
         ra_exe_unit.scan_limit = 0;
         ra_exe_unit.use_bump_allocator = true;
       } else if (eo.executor_type == ::ExecutorType::Extern) {
         ra_exe_unit.scan_limit = 0;
       } else if (!eo.just_explain) {
         const auto filter_count_all = getFilteredCountAll(ra_exe_unit, true, co, eo);
         if (filter_count_all) {
           ra_exe_unit.scan_limit = std::max(*filter_count_all, size_t(1));
           VLOG(1) << "Set a new scan limit from filtered_count_all: "
                   << ra_exe_unit.scan_limit;
           auto const has_limit_value = ra_exe_unit.sort_info.limit.has_value();
           auto const top_k_sort_query =
               has_limit_value && !ra_exe_unit.sort_info.order_entries.empty();
           // top-k sort query needs to get a global result before sorting, so we cannot
           // apply LIMIT value at this point
           if (has_limit_value && !top_k_sort_query &&
               ra_exe_unit.scan_limit > ra_exe_unit.sort_info.limit.value()) {
             ra_exe_unit.scan_limit = ra_exe_unit.sort_info.limit.value();
             VLOG(1) << "Override scan limit to LIMIT value: " << ra_exe_unit.scan_limit;
           }
         }
       }
     }
   }
 
   // when output_columnar_hint is true here, it means either 1) columnar output
   // configuration is on or 2) a user hint is given but we have to disable it if some
   // requirements are not satisfied
   if (can_output_columnar(ra_exe_unit, render_info, body)) {
     if (!eo.output_columnar_hint && should_output_columnar(ra_exe_unit)) {
       VLOG(1) << "Using columnar layout for projection as output size of "
               << ra_exe_unit.scan_limit << " rows exceeds threshold of "
               << g_columnar_large_projections_threshold
               << " or some target uses FlatBuffer memory layout.";
       eo.output_columnar_hint = true;
     }
   } else {
     eo.output_columnar_hint = false;
   }
 
   ExecutionResult result{std::make_shared<ResultSet>(std::vector<TargetInfo>{},
                                                      co.device_type,
                                                      QueryMemoryDescriptor(),
                                                      nullptr,
                                                      executor_->blockSize(),
                                                      executor_->gridSize()),
                          {}};
 
   auto execute_and_handle_errors = [&](const auto max_groups_buffer_entry_guess_in,
                                        const bool has_cardinality_estimation,
                                        const bool has_ndv_estimation) -> ExecutionResult {
     // Note that the groups buffer entry guess may be modified during query execution.
     // Create a local copy so we can track those changes if we need to attempt a retry
     // due to OOM
     auto local_groups_buffer_entry_guess = max_groups_buffer_entry_guess_in;
     try {
       return {executor_->executeWorkUnit(local_groups_buffer_entry_guess,
                                          is_agg,
                                          table_infos,
                                          ra_exe_unit,
                                          co,
                                          eo,
                                          render_info,
                                          has_cardinality_estimation,
                                          column_cache),
               targets_meta};
     } catch (const QueryExecutionError& e) {
       if (!has_ndv_estimation && e.getErrorCode() < 0) {
         throw CardinalityEstimationRequired(/*range=*/0);
       }
       handlePersistentError(e.getErrorCode());
       return handleOutOfMemoryRetry(
           {ra_exe_unit, work_unit.body, local_groups_buffer_entry_guess},
           column_cache,
           targets_meta,
           is_agg,
           co,
           eo,
           render_info,
           e,
           queue_time_ms);
     }
   };
 
   auto use_resultset_cache = canUseResultsetCache(eo, render_info);
   for (const auto& table_info : table_infos) {
     const auto db_id = table_info.table_key.db_id;
     if (db_id > 0) {
       const auto td = Catalog_Namespace::get_metadata_for_table(table_info.table_key);
       if (td && (td->isTemporaryTable() || td->isView)) {
         use_resultset_cache = false;
         if (eo.keep_result) {
           VLOG(1) << "Query hint \'keep_result\' is ignored since a query has either "
                      "temporary table or view";
         }
       }
     }
   }
 
   CardinalityCacheKey cache_key{ra_exe_unit};
   try {
     auto cached_cardinality = executor_->getCachedCardinality(cache_key);
     auto card = cached_cardinality.second;
     if (cached_cardinality.first && card >= 0) {
       VLOG(1) << "Use cached cardinality for max_groups_buffer_entry_guess: " << card;
       result = execute_and_handle_errors(
           card, /*has_cardinality_estimation=*/true, /*has_ndv_estimation=*/false);
     } else {
       VLOG(1) << "Use default cardinality for max_groups_buffer_entry_guess: "
               << max_groups_buffer_entry_guess;
       result = execute_and_handle_errors(
           max_groups_buffer_entry_guess,
           groups_approx_upper_bound(table_infos).first <= g_big_group_threshold,
           /*has_ndv_estimation=*/false);
     }
   } catch (const CardinalityEstimationRequired& e) {
     // check the cardinality cache
     auto cached_cardinality = executor_->getCachedCardinality(cache_key);
     auto card = cached_cardinality.second;
     if (cached_cardinality.first && card >= 0) {
       VLOG(1) << "CardinalityEstimationRequired, Use cached cardinality for "
                  "max_groups_buffer_entry_guess: "
               << card;
       result = execute_and_handle_errors(card, true, /*has_ndv_estimation=*/true);
     } else {
       const auto ndv_groups_estimation =
           getNDVEstimation(work_unit, e.range(), is_agg, co, eo);
       auto ndv_groups_estimator_multiplier = g_ndv_groups_estimator_multiplier;
       if (query_hints.isHintRegistered(QueryHint::kNDVGroupsEstimatorMultiplier)) {
         ndv_groups_estimator_multiplier = query_hints.ndv_groups_estimator_multiplier;
         VLOG(1) << "Modify NDV groups estimator multiplier: "
                 << g_ndv_groups_estimator_multiplier << " -> "
                 << ndv_groups_estimator_multiplier;
       }
       const auto estimated_groups_buffer_entry_guess =
           ndv_groups_estimation > 0
               ? static_cast<size_t>(static_cast<double>(ndv_groups_estimation) *
                                     ndv_groups_estimator_multiplier)
               : std::min(groups_approx_upper_bound(table_infos).first,
                          g_estimator_failure_max_groupby_size);
       CHECK_GT(estimated_groups_buffer_entry_guess, size_t(0));
       VLOG(1) << "CardinalityEstimationRequired, Use ndv_estimation: "
               << ndv_groups_estimation
               << ", cardinality for estimated_groups_buffer_entry_guess: "
               << estimated_groups_buffer_entry_guess;
       result = execute_and_handle_errors(
           estimated_groups_buffer_entry_guess, true, /*has_ndv_estimation=*/true);
       if (!(eo.just_validate || eo.just_explain)) {
         executor_->addToCardinalityCache(cache_key, estimated_groups_buffer_entry_guess);
       }
     }
   }
 
   result.setQueueTime(queue_time_ms);
   if (render_info) {
     build_render_targets(*render_info, work_unit.exe_unit.target_exprs, targets_meta);
     if (render_info->isInSitu()) {
       // return an empty result (with the same queue time, and zero render time)
       return {std::make_shared<ResultSet>(
                   queue_time_ms,
                   0,
                   executor_->row_set_mem_owner_
                       ? executor_->row_set_mem_owner_->cloneStrDictDataOnly()
                       : nullptr),
               {}};
     }
   }
 
   for (auto& target_info : result.getTargetsMeta()) {
     if (target_info.get_type_info().is_string() &&
         !target_info.get_type_info().is_dict_encoded_string()) {
       // currently, we do not support resultset caching if non-encoded string is projected
       use_resultset_cache = false;
       if (eo.keep_result) {
         VLOG(1) << "Query hint \'keep_result\' is ignored since a query has non-encoded "
                    "string column projection";
       }
     }
   }
 
   const auto res = result.getDataPtr();
   if (use_resultset_cache && has_valid_query_plan_dag(body)) {
     auto query_exec_time = timer_stop(query_exec_time_begin);
     res->setExecTime(query_exec_time);
     res->setQueryPlanHash(ra_exe_unit.query_plan_dag_hash);
     res->setTargetMetaInfo(body->getOutputMetainfo());
     auto input_table_keys = ScanNodeTableKeyCollector::getScanNodeTableKey(body);
     res->setInputTableKeys(std::move(input_table_keys));
     auto allow_auto_caching_resultset =
         res && res->hasValidBuffer() && g_allow_auto_resultset_caching &&
         res->getBufferSizeBytes(co.device_type) <= g_auto_resultset_caching_threshold;
     if (eo.keep_result || allow_auto_caching_resultset) {
       if (allow_auto_caching_resultset) {
         VLOG(1) << "Automatically keep query resultset to recycler";
       }
       res->setUseSpeculativeTopNSort(
           use_speculative_top_n(ra_exe_unit, res->getQueryMemDesc()));
       executor_->getResultSetRecyclerHolder().putQueryResultSetToCache(
           ra_exe_unit.query_plan_dag_hash,
           res->getInputTableKeys(),
           res,
           res->getBufferSizeBytes(co.device_type),
           target_exprs_owned_);
     }
   } else {
     if (eo.keep_result) {
       if (g_cluster) {
         VLOG(1) << "Query hint \'keep_result\' is ignored since we do not support "
                    "resultset recycling on distributed mode";
       } else if (hasStepForUnion()) {
         VLOG(1) << "Query hint \'keep_result\' is ignored since a query has union-(all) "
                    "operator";
       } else if (render_info && render_info->isInSitu()) {
         VLOG(1) << "Query hint \'keep_result\' is ignored since a query is classified as "
                    "a in-situ rendering query";
       } else if (is_validate_or_explain_query(eo)) {
         VLOG(1) << "Query hint \'keep_result\' is ignored since a query is either "
                    "validate or explain query";
       } else {
         VLOG(1) << "Query hint \'keep_result\' is ignored";
       }
     }
   }
 
   return result;
 }

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 4312 of file RelAlgExecutor.cpp.

References anonymous_namespace{RelAlgExecutor.cpp}::getErrorDescription(), and to_string().

Referenced by executeDelete(), executeUpdate(), getNDVEstimation(), and handlePersistentError().

                                                                           {
   if (error_code < 0) {
     return "Ran out of slots in the query output buffer";
   }
   const auto errorInfo = getErrorDescription(error_code);
 
   if (errorInfo.code) {
     return errorInfo.code + ": "s + errorInfo.description;
   } else {
     return "Other error: code "s + std::to_string(error_code);
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Executor * RelAlgExecutor::getExecutor ( ) const

Definition at line 772 of file RelAlgExecutor.cpp.

References executor_.

                                             {
   return executor_;
 }

std::optional< size_t > RelAlgExecutor::getFilteredCountAll	(	const RelAlgExecutionUnit &	ra_exe_unit,
		const bool	is_agg,
		const CompilationOptions &	co,
		const ExecutionOptions &	eo
	)

private

Definition at line 4067 of file RelAlgExecutor.cpp.

References CHECK, CHECK_EQ, CHECK_GE, RelAlgExecutionUnit::createCountAllExecutionUnit(), ExecutionOptions::estimate_output_cardinality, executor_, g_bigint_count, get_table_infos(), anonymous_namespace{RelAlgExecutor.cpp}::get_table_name_from_table_key(), anonymous_namespace{RelAlgExecutor.cpp}::groups_approx_upper_bound(), hasDeletedRowInQuery(), anonymous_namespace{RelAlgExecutor.cpp}::is_projection(), RelAlgExecutionUnit::join_quals, kBIGINT, kCOUNT, kINT, LOG, RelAlgExecutionUnit::per_device_cardinality, RelAlgExecutionUnit::quals, RelAlgExecutionUnit::simple_quals, VLOG, and logger::WARNING.

Referenced by executeWorkUnit().

                                 {
   auto const input_tables_info = get_table_infos(ra_exe_unit, executor_);
   if (is_projection(ra_exe_unit) && ra_exe_unit.simple_quals.empty() &&
       ra_exe_unit.quals.empty() && ra_exe_unit.join_quals.empty() &&
       !hasDeletedRowInQuery(input_tables_info)) {
     auto const max_row_info = groups_approx_upper_bound(input_tables_info);
     auto const num_rows = max_row_info.first;
     auto const table_name = get_table_name_from_table_key(max_row_info.second);
     VLOG(1) << "Short-circuiting filtered count query for the projection query "
                "containing input table "
             << table_name << ": return its table cardinality " << num_rows << " instead";
     return std::make_optional<size_t>(num_rows);
   }
   const auto count =
       makeExpr<Analyzer::AggExpr>(SQLTypeInfo(g_bigint_count ? kBIGINT : kINT, false),
                                   kCOUNT,
                                   nullptr,
                                   false,
                                   nullptr);
   const auto count_all_exe_unit = ra_exe_unit.createCountAllExecutionUnit(count.get());
   size_t one{1};
   ResultSetPtr count_all_result;
   try {
     VLOG(1) << "Try to execute pre-flight counts query";
     ColumnCacheMap column_cache;
     ExecutionOptions copied_eo = eo;
     copied_eo.estimate_output_cardinality = true;
     count_all_result = executor_->executeWorkUnit(one,
                                                   is_agg,
                                                   input_tables_info,
                                                   count_all_exe_unit,
                                                   co,
                                                   copied_eo,
                                                   nullptr,
                                                   false,
                                                   column_cache);
     ra_exe_unit.per_device_cardinality = count_all_exe_unit.per_device_cardinality;
   } catch (const foreign_storage::ForeignStorageException& error) {
     throw;
   } catch (const QueryMustRunOnCpu&) {
     // force a retry of the top level query on CPU
     throw;
   } catch (const std::exception& e) {
     LOG(WARNING) << "Failed to run pre-flight filtered count with error " << e.what();
     return std::nullopt;
   }
   const auto count_row = count_all_result->getNextRow(false, false);
   CHECK_EQ(size_t(1), count_row.size());
   const auto& count_tv = count_row.front();
   const auto count_scalar_tv = boost::get<ScalarTargetValue>(&count_tv);
   CHECK(count_scalar_tv);
   const auto count_ptr = boost::get<int64_t>(count_scalar_tv);
   CHECK(count_ptr);
   CHECK_GE(*count_ptr, 0);
   auto count_upper_bound = static_cast<size_t>(*count_ptr);
   return std::max(count_upper_bound, size_t(1));
 }

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 59 of file JoinFilterPushDown.cpp.

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

Referenced by selectFiltersToBePushedDown().

                                 {
   CollectInputColumnsVisitor input_columns_visitor;
   std::list<std::shared_ptr<Analyzer::Expr>> quals;
   std::unordered_set<InputColDescriptor> input_column_descriptors;
   BindFilterToOutermostVisitor bind_filter_to_outermost;
   for (const auto& filter_expr : filter_expressions) {
     input_column_descriptors = input_columns_visitor.aggregateResult(
         input_column_descriptors, input_columns_visitor.visit(filter_expr.get()));
     quals.push_back(bind_filter_to_outermost.visit(filter_expr.get()));
   }
   std::vector<InputDescriptor> input_descs;
   std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
   for (const auto& input_col_desc : input_column_descriptors) {
     if (input_descs.empty()) {
       input_descs.push_back(input_col_desc.getScanDesc());
     } else {
       CHECK(input_col_desc.getScanDesc() == input_descs.front());
     }
     input_col_descs.push_back(std::make_shared<const InputColDescriptor>(input_col_desc));
   }
   const auto count_expr =
       makeExpr<Analyzer::AggExpr>(SQLTypeInfo(g_bigint_count ? kBIGINT : kINT, false),
                                   kCOUNT,
                                   nullptr,
                                   false,
                                   nullptr);
   RelAlgExecutionUnit ra_exe_unit{input_descs,
                                   input_col_descs,
                                   {},
                                   quals,
                                   {},
                                   {},
                                   {count_expr.get()},
                                   {},
                                   nullptr,
                                   SortInfo(),
                                   0};
   size_t one{1};
   ResultSetPtr filtered_result;
   const auto table_infos = get_table_infos(input_descs, executor_);
   CHECK_EQ(size_t(1), table_infos.size());
   const size_t total_rows_upper_bound = table_infos.front().info.getNumTuplesUpperBound();
   try {
     ColumnCacheMap column_cache;
     filtered_result = executor_->executeWorkUnit(
         one, true, table_infos, ra_exe_unit, co, eo, nullptr, false, column_cache);
   } catch (...) {
     return {false, 1.0, 0};
   }
   const auto count_row = filtered_result->getNextRow(false, false);
   CHECK_EQ(size_t(1), count_row.size());
   const auto& count_tv = count_row.front();
   const auto count_scalar_tv = boost::get<ScalarTargetValue>(&count_tv);
   CHECK(count_scalar_tv);
   const auto count_ptr = boost::get<int64_t>(count_scalar_tv);
   CHECK(count_ptr);
   const auto rows_passing = *count_ptr;
   const auto rows_total = std::max(total_rows_upper_bound, size_t(1));
   return {true, static_cast<float>(rows_passing) / rows_total, total_rows_upper_bound};
 }

Here is the call graph for this function:

Here is the caller graph for this function:

std::optional<RegisteredQueryHint> RelAlgExecutor::getGlobalQueryHint ( )

inline

Definition at line 179 of file RelAlgExecutor.h.

References query_dag_.

Referenced by executeRelAlgQueryNoRetry(), executeTableFunction(), and QueryRunner::QueryRunner::getParsedGlobalQueryHints().

                                                         {
     return query_dag_ ? std::make_optional(query_dag_->getGlobalHints()) : std::nullopt;
   }

Here is the caller graph for this function:

std::pair< std::vector< unsigned >, std::unordered_map< unsigned, JoinQualsPerNestingLevel > > RelAlgExecutor::getJoinInfo ( const RelAlgNode * root_node )

Definition at line 782 of file RelAlgExecutor.cpp.

References createWorkUnit(), ExecutionOptions::defaults(), getLeftDeepJoinTreesInfo(), and RelAlgVisitor< T >::visit().

                                                        {
   auto sort_node = dynamic_cast<const RelSort*>(root_node);
   if (sort_node) {
     // we assume that test query that needs join info does not contain any sort node
     return {};
   }
   auto work_unit = createWorkUnit(root_node, {}, ExecutionOptions::defaults());
   RelLeftDeepTreeIdsCollector visitor;
   auto left_deep_tree_ids = visitor.visit(root_node);
   return {left_deep_tree_ids, getLeftDeepJoinTreesInfo()};
 }

Here is the call graph for this function:

std::unordered_map<unsigned, JoinQualsPerNestingLevel>& RelAlgExecutor::getLeftDeepJoinTreesInfo ( )

inlineprivate

Definition at line 413 of file RelAlgExecutor.h.

References left_deep_join_info_.

Referenced by createAggregateWorkUnit(), createCompoundWorkUnit(), createFilterWorkUnit(), createProjectWorkUnit(), and getJoinInfo().

                                                                                    {
     return left_deep_join_info_;
   }

Here is the caller graph for this function:

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

private

Definition at line 54 of file CardinalityEstimator.cpp.

References RelAlgExecutionUnit::createNdvExecutionUnit(), RelAlgExecutor::WorkUnit::exe_unit, executor_, get_table_infos(), QueryExecutionError::getErrorCode(), getErrorMessageFromCode(), QueryExecutionError::hasErrorCode(), and UNREACHABLE.

Referenced by executeWorkUnit().

                                                                     {
   const auto estimator_exe_unit = work_unit.exe_unit.createNdvExecutionUnit(range);
   size_t one{1};
   ColumnCacheMap column_cache;
   try {
     const auto estimator_result =
         executor_->executeWorkUnit(one,
                                    is_agg,
                                    get_table_infos(work_unit.exe_unit, executor_),
                                    estimator_exe_unit,
                                    co,
                                    eo,
                                    nullptr,
                                    false,
                                    column_cache);
     if (!estimator_result) {
       return 1;  // empty row set, only needs one slot
     }
     return estimator_result->getNDVEstimator();
   } catch (const QueryExecutionError& e) {
     if (e.hasErrorCode(ErrorCode::OUT_OF_TIME)) {
       throw std::runtime_error("Cardinality estimation query ran out of time");
     }
     if (e.hasErrorCode(ErrorCode::INTERRUPTED)) {
       throw std::runtime_error("Cardinality estimation query has been interrupted");
     }
     throw std::runtime_error("Failed to run the cardinality estimation query: " +
                              getErrorMessageFromCode(e.getErrorCode()));
   }
   UNREACHABLE();
   return 0;
 }

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 493 of file RelAlgExecutor.cpp.

References CHECK, cleanupPostExecution(), createCompoundWorkUnit(), createProjectWorkUnit(), executor_, ExecutionOptions::find_push_down_candidates, get_frag_count_of_table(), getSubqueries(), ExecutionOptions::just_explain, query_dag_, setupCaching(), gpu_enabled::swap(), target_exprs_owned_, and temporary_tables_.

                                                                          {
   if (eo.find_push_down_candidates) {
     return 0;
   }
 
   if (eo.just_explain) {
     return 0;
   }
 
   CHECK(query_dag_);
 
   query_dag_->resetQueryExecutionState();
   const auto& ra = query_dag_->getRootNode();
 
   auto lock = executor_->acquireExecuteMutex();
   ScopeGuard row_set_holder = [this] { cleanupPostExecution(); };
   setupCaching(&ra);
 
   ScopeGuard restore_metainfo_cache = [this] { executor_->clearMetaInfoCache(); };
   auto ed_seq = RaExecutionSequence(&ra, executor_);
 
   if (!getSubqueries().empty()) {
     return 0;
   }
 
   CHECK(!ed_seq.empty());
   if (ed_seq.size() > 1) {
     return 0;
   }
 
   decltype(temporary_tables_)().swap(temporary_tables_);
   decltype(target_exprs_owned_)().swap(target_exprs_owned_);
   executor_->temporary_tables_ = &temporary_tables_;
 
   auto exec_desc_ptr = ed_seq.getDescriptor(0);
   CHECK(exec_desc_ptr);
   auto& exec_desc = *exec_desc_ptr;
   const auto body = exec_desc.getBody();
   if (body->isNop()) {
     return 0;
   }
 
   const auto project = dynamic_cast<const RelProject*>(body);
   if (project) {
     auto work_unit = createProjectWorkUnit(project, SortInfo(), eo);
     return get_frag_count_of_table(work_unit.exe_unit.input_descs[0].getTableKey(),
                                    executor_);
   }
 
   const auto compound = dynamic_cast<const RelCompound*>(body);
   if (compound) {
     if (compound->isDeleteViaSelect()) {
       return 0;
     } else if (compound->isUpdateViaSelect()) {
       return 0;
     } else {
       if (compound->isAggregate()) {
         return 0;
       }
 
       const auto work_unit = createCompoundWorkUnit(compound, SortInfo(), eo);
 
       return get_frag_count_of_table(work_unit.exe_unit.input_descs[0].getTableKey(),
                                      executor_);
     }
   }
 
   return 0;
 }

Here is the call graph for this function:

std::unique_ptr<RelAlgDag> RelAlgExecutor::getOwnedRelAlgDag ( )

inline

Definition at line 135 of file RelAlgExecutor.h.

References CHECK, and query_dag_.

Referenced by QueryRunner::QueryRunner::getRelAlgDag().

                                                {
     CHECK(query_dag_);
     return std::move(query_dag_);
   }

Here is the caller graph for this function:

std::optional<RegisteredQueryHint> RelAlgExecutor::getParsedQueryHint ( const RelAlgNode * node )

inline

Definition at line 169 of file RelAlgExecutor.h.

References query_dag_.

Referenced by executeRelAlgQueryNoRetry(), executeRelAlgStep(), and QueryRunner::QueryRunner::getParsedQueryHint().

                                                                               {
     return query_dag_ ? query_dag_->getQueryHint(node) : std::nullopt;
   }

Here is the caller graph for this function:

std::optional< std::unordered_map<size_t, std::unordered_map<unsigned, RegisteredQueryHint> > > RelAlgExecutor::getParsedQueryHints ( )

inline

Definition at line 175 of file RelAlgExecutor.h.

References query_dag_.

Referenced by QueryRunner::QueryRunner::getParsedQueryHints().

                         {
     return query_dag_ ? std::make_optional(query_dag_->getQueryHints()) : std::nullopt;
   }

Here is the caller graph for this function:

std::unordered_set< shared::TableKey > RelAlgExecutor::getPhysicalTableIds ( ) const

Definition at line 5482 of file RelAlgExecutor.cpp.

References get_physical_table_inputs(), and getRootRelAlgNode().

                                                                            {
   return get_physical_table_inputs(&getRootRelAlgNode());
 }

Here is the call graph for this function:

RaExecutionSequence RelAlgExecutor::getRaExecutionSequence	(	const RelAlgNode *	root_node,
		Executor *	executor
	)

inline

Definition at line 200 of file RelAlgExecutor.h.

References CHECK.

Referenced by QueryRunner::QueryRunner::getRaExecutionSequence().

                                                                  {
     CHECK(executor);
     CHECK(root_node);
     return RaExecutionSequence(root_node, executor);
   }

Here is the caller graph for this function:

RelAlgDag* RelAlgExecutor::getRelAlgDag ( )

inline

Definition at line 140 of file RelAlgExecutor.h.

References query_dag_.

Referenced by executeRelAlgQueryNoRetry().

                             {
     if (!query_dag_) {
       query_dag_ = std::make_unique<RelAlgDag>();
     }
     return query_dag_.get();
   }

Here is the caller graph for this function:

std::shared_ptr< RelAlgTranslator > RelAlgExecutor::getRelAlgTranslator ( const RelAlgNode * root_node )

Definition at line 4636 of file RelAlgExecutor.cpp.

References executor_, anonymous_namespace{RelAlgExecutor.cpp}::get_input_nest_levels(), anonymous_namespace{RelAlgExecutor.cpp}::get_join_type(), RelAlgNode::getInput(), anonymous_namespace{RelAlgExecutor.cpp}::left_deep_join_types(), now_, and query_state_.

                             {
   auto input_to_nest_level = get_input_nest_levels(node, {});
   const auto left_deep_join =
       dynamic_cast<const RelLeftDeepInnerJoin*>(node->getInput(0));
   const auto join_types = left_deep_join ? left_deep_join_types(left_deep_join)
                                          : std::vector<JoinType>{get_join_type(node)};
   return std::make_shared<RelAlgTranslator>(
       query_state_, executor_, input_to_nest_level, join_types, now_, false);
 }

Here is the call graph for this function:

const RelAlgNode& RelAlgExecutor::getRootRelAlgNode ( ) const

inline

Definition at line 147 of file RelAlgExecutor.h.

References CHECK, and query_dag_.

Referenced by computeColRangesCache(), computeStringDictionaryGenerations(), computeTableGenerations(), executeRelAlgQueryNoRetry(), getPhysicalTableIds(), and prepareForSystemTableExecution().

                                               {
     CHECK(query_dag_);
     return query_dag_->getRootNode();
   }

Here is the caller graph for this function:

std::shared_ptr<const RelAlgNode> RelAlgExecutor::getRootRelAlgNodeShPtr ( ) const

inline

Definition at line 154 of file RelAlgExecutor.h.

References CHECK, and query_dag_.

Referenced by QueryRunner::QueryRunner::getQueryInfoForDataRecyclerTest(), and QueryRunner::QueryRunner::getRootNodeFromParsedQuery().

                                                                  {
     CHECK(query_dag_);
     return query_dag_->getRootNodeShPtr();
   }

Here is the caller graph for this function:

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

inlinenoexcept

Definition at line 164 of file RelAlgExecutor.h.

References CHECK, and query_dag_.

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

                                                                               {
     CHECK(query_dag_);
     return query_dag_->getSubqueries();
   };

Here is the caller graph for this function:

void RelAlgExecutor::handleNop ( RaExecutionDesc & ed )

private

Definition at line 1332 of file RelAlgExecutor.cpp.

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

Referenced by executeRelAlgStep().

                                                   {
   // just set the result of the previous node as the result of no op
   auto body = ed.getBody();
   CHECK(dynamic_cast<const RelAggregate*>(body));
   CHECK_EQ(size_t(1), body->inputCount());
   const auto input = body->getInput(0);
   body->setOutputMetainfo(input->getOutputMetainfo());
   const auto it = temporary_tables_.find(-input->getId());
   CHECK(it != temporary_tables_.end());
   // set up temp table as it could be used by the outer query or next step
   addTemporaryTable(-body->getId(), it->second);
 
   ed.setResult({it->second, input->getOutputMetainfo()});
 }

Here is the call graph for this function:

Here is the caller graph for this function:

ExecutionResult RelAlgExecutor::handleOutOfMemoryRetry	(	const RelAlgExecutor::WorkUnit &	work_unit,
		ColumnCacheMap &	column_cache,
		const std::vector< TargetMetaInfo > &	targets_meta,
		const bool	is_agg,
		const CompilationOptions &	co,
		const ExecutionOptions &	eo,
		RenderInfo *	render_info,
		const QueryExecutionError &	e,
		const int64_t	queue_time_ms
	)

private

Definition at line 4165 of file RelAlgExecutor.cpp.

References CHECK, anonymous_namespace{RelAlgExecutor.cpp}::decide_approx_count_distinct_implementation(), CompilationOptions::device_type, RelAlgExecutor::WorkUnit::exe_unit, executor_, RenderInfo::forceNonInSitu(), g_enable_watchdog, get_table_infos(), QueryExecutionError::getErrorCode(), handlePersistentError(), LOG, CompilationOptions::makeCpuOnly(), RelAlgExecutor::WorkUnit::max_groups_buffer_entry_guess, run_benchmark_import::result, ExecutionOptions::setNoExplainExecutionOptions(), target_exprs_owned_, RelAlgExecutionUnit::use_bump_allocator, VLOG, logger::WARNING, and QueryExecutionError::wasMultifragKernelLaunch().

Referenced by executeWorkUnit().

                                  {
   // Disable the bump allocator
   // Note that this will have basically the same affect as using the bump allocator for
   // the kernel per fragment path. Need to unify the max_groups_buffer_entry_guess = 0
   // path and the bump allocator path for kernel per fragment execution.
   auto ra_exe_unit_in = work_unit.exe_unit;
   ra_exe_unit_in.use_bump_allocator = false;
 
   auto result = ExecutionResult{std::make_shared<ResultSet>(std::vector<TargetInfo>{},
                                                             co.device_type,
                                                             QueryMemoryDescriptor(),
                                                             nullptr,
                                                             executor_->blockSize(),
                                                             executor_->gridSize()),
                                 {}};
 
   const auto table_infos = get_table_infos(ra_exe_unit_in, executor_);
   auto max_groups_buffer_entry_guess = work_unit.max_groups_buffer_entry_guess;
   auto eo_no_multifrag = eo;
   eo_no_multifrag.setNoExplainExecutionOptions(true);
   eo_no_multifrag.allow_multifrag = false;
   eo_no_multifrag.find_push_down_candidates = false;
   if (e.wasMultifragKernelLaunch()) {
     try {
       // Attempt to retry using the kernel per fragment path. The smaller input size
       // required may allow the entire kernel to execute in GPU memory.
       LOG(WARNING) << "Multifrag query ran out of memory, retrying with multifragment "
                       "kernels disabled.";
       const auto ra_exe_unit = decide_approx_count_distinct_implementation(
           ra_exe_unit_in, table_infos, executor_, co.device_type, target_exprs_owned_);
       result = {executor_->executeWorkUnit(max_groups_buffer_entry_guess,
                                            is_agg,
                                            table_infos,
                                            ra_exe_unit,
                                            co,
                                            eo_no_multifrag,
                                            nullptr,
                                            true,
                                            column_cache),
                 targets_meta};
       result.setQueueTime(queue_time_ms);
     } catch (const QueryExecutionError& new_e) {
       handlePersistentError(new_e.getErrorCode());
       LOG(WARNING) << "Kernel per fragment query ran out of memory, retrying on CPU.";
     }
   }
 
   if (render_info) {
     render_info->forceNonInSitu();
   }
 
   const auto co_cpu = CompilationOptions::makeCpuOnly(co);
   if (e.getErrorCode() < 0) {
     // Only reset the group buffer entry guess if we ran out of slots, which
     // suggests a highly pathological input which prevented a good estimation of distinct
     // tuple count. For projection queries, this will force a per-fragment scan limit,
     // which is compatible with the CPU path
     VLOG(1) << "Resetting max groups buffer entry guess.";
     max_groups_buffer_entry_guess = 0;
   }
 
   int iteration_ctr = -1;
   while (true) {
     iteration_ctr++;
     auto ra_exe_unit = decide_approx_count_distinct_implementation(
         ra_exe_unit_in, table_infos, executor_, co_cpu.device_type, target_exprs_owned_);
     try {
       result = {executor_->executeWorkUnit(max_groups_buffer_entry_guess,
                                            is_agg,
                                            table_infos,
                                            ra_exe_unit,
                                            co_cpu,
                                            eo_no_multifrag,
                                            nullptr,
                                            true,
                                            column_cache),
                 targets_meta};
     } catch (const QueryExecutionError& new_e) {
       // Ran out of slots
       if (new_e.getErrorCode() < 0) {
         // Even the conservative guess failed; it should only happen when we group
         // by a huge cardinality array. Maybe we should throw an exception instead?
         // Such a heavy query is entirely capable of exhausting all the host memory.
         CHECK(max_groups_buffer_entry_guess);
         // Only allow two iterations of increasingly large entry guesses up to a maximum
         // of 512MB per column per kernel
         if (g_enable_watchdog || iteration_ctr > 1) {
           throw std::runtime_error("Query ran out of output slots in the result");
         }
         max_groups_buffer_entry_guess *= 2;
         LOG(WARNING) << "Query ran out of slots in the output buffer, retrying with max "
                         "groups buffer entry "
                         "guess equal to "
                      << max_groups_buffer_entry_guess;
       } else {
         handlePersistentError(new_e.getErrorCode());
       }
       continue;
     }
     result.setQueueTime(queue_time_ms);
     return result;
   }
   return result;
 }

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 4279 of file RelAlgExecutor.cpp.

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

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

                                                                    {
   LOG(ERROR) << "Query execution failed with error "
              << getErrorMessageFromCode(error_code);
   if (error_code == int32_t(ErrorCode::OUT_OF_GPU_MEM)) {
     // We ran out of GPU memory, this doesn't count as an error if the query is
     // allowed to continue on CPU because retry on CPU is explicitly allowed through
     // --allow-cpu-retry.
     LOG(INFO) << "Query ran out of GPU memory, attempting punt to CPU";
     if (!g_allow_cpu_retry) {
       throw std::runtime_error(
           "Query ran out of GPU memory, unable to automatically retry on CPU");
     }
     return;
   }
   throw std::runtime_error(getErrorMessageFromCode(error_code));
 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool RelAlgExecutor::hasDeletedRowInQuery ( std::vector< InputTableInfo > const & input_tables_info ) const

private

Definition at line 4034 of file RelAlgExecutor.cpp.

References anonymous_namespace{QueryMemoryDescriptor.cpp}::any_of(), CHECK, Catalog_Namespace::SysCatalog::getCatalog(), and Catalog_Namespace::SysCatalog::instance().

Referenced by getFilteredCountAll().

                                                               {
   return std::any_of(
       input_tables_info.begin(), input_tables_info.end(), [](InputTableInfo const& info) {
         auto const& table_key = info.table_key;
         if (table_key.db_id > 0) {
           auto catalog =
               Catalog_Namespace::SysCatalog::instance().getCatalog(table_key.db_id);
           CHECK(catalog);
           auto td = catalog->getMetadataForTable(table_key.table_id);
           CHECK(td);
           if (catalog->getDeletedColumnIfRowsDeleted(td)) {
             return true;
           }
         }
         return false;
       });
 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool RelAlgExecutor::hasStepForUnion ( ) const

inlineprivate

Definition at line 433 of file RelAlgExecutor.h.

References has_step_for_union_.

Referenced by canUseResultsetCache(), executeSort(), executeTableFunction(), and executeWorkUnit().

433 { return has_step_for_union_; }

RelAlgExecutor::has_step_for_union_

bool has_step_for_union_

Definition: RelAlgExecutor.h:448

Here is the caller graph for this function:

void RelAlgExecutor::initializeParallelismHints ( )

private

Definition at line 5460 of file RelAlgExecutor.cpp.

References executor_.

Referenced by RelAlgExecutor().

                                                 {
   if (auto foreign_storage_mgr =
           executor_->getDataMgr()->getPersistentStorageMgr()->getForeignStorageMgr()) {
     // Parallelism hints need to be reset to empty so that we don't accidentally re-use
     // them.  This can cause attempts to fetch strings that do not shard to the correct
     // node in distributed mode.
     foreign_storage_mgr->setParallelismHints({});
   }
 }

Here is the caller graph for this function:

bool RelAlgExecutor::isRowidLookup ( const WorkUnit & work_unit )

private

Definition at line 4129 of file RelAlgExecutor.cpp.

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

Referenced by executeWorkUnit().

                                                             {
   const auto& ra_exe_unit = work_unit.exe_unit;
   if (ra_exe_unit.input_descs.size() != 1) {
     return false;
   }
   const auto& table_desc = ra_exe_unit.input_descs.front();
   if (table_desc.getSourceType() != InputSourceType::TABLE) {
     return false;
   }
   const auto& table_key = table_desc.getTableKey();
   for (const auto& simple_qual : ra_exe_unit.simple_quals) {
     const auto comp_expr =
         std::dynamic_pointer_cast<const Analyzer::BinOper>(simple_qual);
     if (!comp_expr || comp_expr->get_optype() != kEQ) {
       return false;
     }
     const auto lhs = comp_expr->get_left_operand();
     const auto lhs_col = dynamic_cast<const Analyzer::ColumnVar*>(lhs);
     if (!lhs_col || !lhs_col->getTableKey().table_id || lhs_col->get_rte_idx()) {
       return false;
     }
     const auto rhs = comp_expr->get_right_operand();
     const auto rhs_const = dynamic_cast<const Analyzer::Constant*>(rhs);
     if (!rhs_const) {
       return false;
     }
     const auto cd = get_column_descriptor(
         {table_key.db_id, table_key.table_id, lhs_col->getColumnKey().column_id});
     if (cd->isVirtualCol) {
       CHECK_EQ("rowid", cd->columnName);
       return true;
     }
   }
   return false;
 }

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 4741 of file RelAlgExecutor.cpp.

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

Referenced by translateLeftDeepJoinFilter().

                                    {
   RelAlgTranslator translator(
       query_state_, executor_, input_to_nest_level, join_types, now_, just_explain);
   const auto rex_condition_cf = rex_to_conjunctive_form(join_condition);
   std::list<std::shared_ptr<Analyzer::Expr>> join_condition_quals;
   for (const auto rex_condition_component : rex_condition_cf) {
     const auto bw_equals = get_bitwise_equals_conjunction(rex_condition_component);
     const auto join_condition = reverse_logical_distribution(
         translator.translate(bw_equals ? bw_equals.get() : rex_condition_component));
     auto join_condition_cf = qual_to_conjunctive_form(join_condition);
 
     auto append_folded_cf_quals = [&join_condition_quals](const auto& cf_quals) {
       for (const auto& cf_qual : cf_quals) {
         join_condition_quals.emplace_back(fold_expr(cf_qual.get()));
       }
     };
 
     append_folded_cf_quals(join_condition_cf.quals);
     append_folded_cf_quals(join_condition_cf.simple_quals);
   }
   return combine_equi_join_conditions(join_condition_quals);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void RelAlgExecutor::prepareForeignTable ( )

void RelAlgExecutor::prepareForeignTables ( )

Definition at line 5477 of file RelAlgExecutor.cpp.

References anonymous_namespace{RelAlgExecutor.cpp}::prepare_foreign_table_for_execution(), and query_dag_.

                                           {
   const auto& ra = query_dag_->getRootNode();
   prepare_foreign_table_for_execution(ra);
 }

Here is the call graph for this function:

void RelAlgExecutor::prepareForSystemTableExecution ( const CompilationOptions & co ) const

Definition at line 5486 of file RelAlgExecutor.cpp.

References getRootRelAlgNode(), and anonymous_namespace{RelAlgExecutor.cpp}::prepare_for_system_table_execution().

                                                                                       {
   prepare_for_system_table_execution(getRootRelAlgNode(), co);
 }

Here is the call graph for this function:

void RelAlgExecutor::prepareLeafExecution	(	const AggregatedColRange &	agg_col_range,
		const StringDictionaryGenerations &	string_dictionary_generations,
		const TableGenerations &	table_generations
	)

Definition at line 873 of file RelAlgExecutor.cpp.

References executor_, g_enable_dynamic_watchdog, Executor::getArenaBlockSize(), queue_time_ms_, timer_start(), and timer_stop().

                                                {
   // capture the lock acquistion time
   auto clock_begin = timer_start();
   if (g_enable_dynamic_watchdog) {
     executor_->resetInterrupt();
   }
   queue_time_ms_ = timer_stop(clock_begin);
   executor_->row_set_mem_owner_ = std::make_shared<RowSetMemoryOwner>(
       Executor::getArenaBlockSize(), executor_->executor_id_);
   executor_->row_set_mem_owner_->setDictionaryGenerations(string_dictionary_generations);
   executor_->table_generations_ = table_generations;
   executor_->agg_col_range_cache_ = agg_col_range;
 }

Here is the call graph for this function:

std::vector< PushedDownFilterInfo > RelAlgExecutor::selectFiltersToBePushedDown	(	const RelAlgExecutor::WorkUnit &	work_unit,
		const CompilationOptions &	co,
		const ExecutionOptions &	eo
	)

private

Goes through all candidate filters and evaluate whether they pass the selectivity criteria or not.

Definition at line 127 of file JoinFilterPushDown.cpp.

References RelAlgExecutor::WorkUnit::exe_unit, executor_, find_push_down_filters(), get_table_infos(), getFilterSelectivity(), RelAlgExecutionUnit::input_descs, RelAlgExecutor::WorkUnit::input_permutation, RelAlgExecutor::WorkUnit::left_deep_join_input_sizes, and to_gather_info_for_filter_selectivity().

Referenced by executeWorkUnit().

                                 {
   const auto all_push_down_candidates =
       find_push_down_filters(work_unit.exe_unit,
                              work_unit.input_permutation,
                              work_unit.left_deep_join_input_sizes);
   std::vector<PushedDownFilterInfo> selective_push_down_candidates;
   const auto ti = get_table_infos(work_unit.exe_unit.input_descs, executor_);
   if (to_gather_info_for_filter_selectivity(ti)) {
     for (const auto& candidate : all_push_down_candidates) {
       const auto selectivity = getFilterSelectivity(candidate.filter_expressions, co, eo);
       if (selectivity.is_valid && selectivity.isFilterSelectiveEnough()) {
         selective_push_down_candidates.push_back(candidate);
       }
     }
   }
   return selective_push_down_candidates;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void RelAlgExecutor::setHasStepForUnion ( bool flag )

inlineprivate

Definition at line 431 of file RelAlgExecutor.h.

References has_step_for_union_.

Referenced by executeRelAlgStep().

431 { has_step_for_union_ = flag; }

RelAlgExecutor::has_step_for_union_

bool has_step_for_union_

Definition: RelAlgExecutor.h:448

Here is the caller graph for this function:

void RelAlgExecutor::setupCaching ( const RelAlgNode * ra )

private

Definition at line 5470 of file RelAlgExecutor.cpp.

References CHECK, executor_, anonymous_namespace{RelAlgExecutor.cpp}::get_physical_inputs_with_spi_col_id(), and get_physical_table_inputs().

Referenced by executeRelAlgQueryNoRetry(), and getOuterFragmentCount().

                                                       {
   CHECK(executor_);
   const auto phys_inputs = get_physical_inputs_with_spi_col_id(ra);
   const auto phys_table_ids = get_physical_table_inputs(ra);
   executor_->setupCaching(phys_inputs, phys_table_ids);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

JoinQualsPerNestingLevel RelAlgExecutor::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
	)

private

Definition at line 4771 of file RelAlgExecutor.cpp.

References ANTI, CHECK, CHECK_LE, RelLeftDeepInnerJoin::getInnerCondition(), RelLeftDeepInnerJoin::getOuterCondition(), INNER, LEFT, anonymous_namespace{RelAlgExecutor.cpp}::left_deep_join_types(), makeJoinQuals(), run_benchmark_import::result, and SEMI.

Referenced by createCompoundWorkUnit(), and createProjectWorkUnit().

                              {
   const auto join_types = left_deep_join_types(join);
   const auto join_condition_quals = makeJoinQuals(
       join->getInnerCondition(), join_types, input_to_nest_level, just_explain);
   MaxRangeTableIndexVisitor rte_idx_visitor;
   JoinQualsPerNestingLevel result(input_descs.size() - 1);
   std::unordered_set<std::shared_ptr<Analyzer::Expr>> visited_quals;
   for (size_t rte_idx = 1; rte_idx < input_descs.size(); ++rte_idx) {
     const auto outer_condition = join->getOuterCondition(rte_idx);
     if (outer_condition) {
       result[rte_idx - 1].quals =
           makeJoinQuals(outer_condition, join_types, input_to_nest_level, just_explain);
       CHECK_LE(rte_idx, join_types.size());
       CHECK(join_types[rte_idx - 1] == JoinType::LEFT);
       result[rte_idx - 1].type = JoinType::LEFT;
       continue;
     }
     for (const auto& qual : join_condition_quals) {
       if (visited_quals.count(qual)) {
         continue;
       }
       const auto qual_rte_idx = rte_idx_visitor.visit(qual.get());
       if (static_cast<size_t>(qual_rte_idx) <= rte_idx) {
         const auto it_ok = visited_quals.emplace(qual);
         CHECK(it_ok.second);
         result[rte_idx - 1].quals.push_back(qual);
       }
     }
     CHECK_LE(rte_idx, join_types.size());
     CHECK(join_types[rte_idx - 1] == JoinType::INNER ||
           join_types[rte_idx - 1] == JoinType::SEMI ||
           join_types[rte_idx - 1] == JoinType::ANTI);
     result[rte_idx - 1].type = join_types[rte_idx - 1];
   }
   return result;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Friends And Related Function Documentation

friend class PendingExecutionClosure

friend

Definition at line 454 of file RelAlgExecutor.h.

Member Data Documentation

std::unique_ptr<TransactionParameters> RelAlgExecutor::dml_transaction_parameters_

private

Definition at line 451 of file RelAlgExecutor.h.

Referenced by executeDelete(), and executeUpdate().

Executor* RelAlgExecutor::executor_

private

Definition at line 439 of file RelAlgExecutor.h.

Referenced by cleanupPostExecution(), computeColRangesCache(), computeStringDictionaryGenerations(), computeTableGenerations(), computeWindow(), createAggregateWorkUnit(), createCompoundWorkUnit(), createFilterWorkUnit(), createProjectWorkUnit(), createTableFunctionWorkUnit(), createUnionWorkUnit(), createWindowFunctionContext(), executeDelete(), executeLogicalValues(), executeModify(), executeRelAlgQueryNoRetry(), executeRelAlgQueryWithFilterPushDown(), executeRelAlgSeq(), executeRelAlgStep(), executeRelAlgSubSeq(), executeSimpleInsert(), executeSort(), executeTableFunction(), executeUpdate(), executeWorkUnit(), getExecutor(), getFilteredCountAll(), getFilterSelectivity(), getNDVEstimation(), getOuterFragmentCount(), getRelAlgTranslator(), handleOutOfMemoryRetry(), initializeParallelismHints(), makeJoinQuals(), prepareLeafExecution(), selectFiltersToBePushedDown(), and setupCaching().

bool RelAlgExecutor::has_step_for_union_

private

Definition at line 448 of file RelAlgExecutor.h.

Referenced by hasStepForUnion(), and setHasStepForUnion().

std::unordered_map<unsigned, AggregatedResult> RelAlgExecutor::leaf_results_

private

Definition at line 446 of file RelAlgExecutor.h.

Referenced by addLeafResult(), executeSort(), executeUpdate(), and executeWorkUnit().

std::unordered_map<unsigned, JoinQualsPerNestingLevel> RelAlgExecutor::left_deep_join_info_

private

Definition at line 444 of file RelAlgExecutor.h.

Referenced by executeRelAlgSeq(), executeRelAlgSubSeq(), and getLeftDeepJoinTreesInfo().

time_t RelAlgExecutor::now_

private

Definition at line 443 of file RelAlgExecutor.h.

Referenced by createAggregateWorkUnit(), createCompoundWorkUnit(), createFilterWorkUnit(), createProjectWorkUnit(), createTableFunctionWorkUnit(), executeRelAlgSeq(), executeRelAlgSubSeq(), getRelAlgTranslator(), and makeJoinQuals().

std::optional<std::function<void()> > RelAlgExecutor::post_execution_callback_

private

Definition at line 452 of file RelAlgExecutor.h.

Referenced by executeDelete(), executePostExecutionCallback(), executeRelAlgQuery(), executeRelAlgQuerySingleStep(), and executeUpdate().

std::unique_ptr<RelAlgDag> RelAlgExecutor::query_dag_

private

Definition at line 440 of file RelAlgExecutor.h.

Referenced by createAggregateWorkUnit(), createCompoundWorkUnit(), createFilterWorkUnit(), createProjectWorkUnit(), executeRelAlgQuery(), executeRelAlgQueryNoRetry(), executeWorkUnit(), getGlobalQueryHint(), getOuterFragmentCount(), getOwnedRelAlgDag(), getParsedQueryHint(), getParsedQueryHints(), getRelAlgDag(), getRootRelAlgNode(), getRootRelAlgNodeShPtr(), getSubqueries(), and prepareForeignTables().

std::shared_ptr<const query_state::QueryState> RelAlgExecutor::query_state_

private

Definition at line 441 of file RelAlgExecutor.h.

Referenced by createAggregateWorkUnit(), createCompoundWorkUnit(), createFilterWorkUnit(), createProjectWorkUnit(), createTableFunctionWorkUnit(), createUnionWorkUnit(), executeRelAlgQueryNoRetry(), getRelAlgTranslator(), and makeJoinQuals().

int64_t RelAlgExecutor::queue_time_ms_

private

Definition at line 447 of file RelAlgExecutor.h.

Referenced by executeRelAlgQuerySingleStep(), and prepareLeafExecution().

SpeculativeTopNBlacklist RelAlgExecutor::speculative_topn_blacklist_

staticprivate

Definition at line 449 of file RelAlgExecutor.h.

Referenced by createSortInputWorkUnit(), and executeSort().

std::vector<std::shared_ptr<Analyzer::Expr> > RelAlgExecutor::target_exprs_owned_

private

Definition at line 445 of file RelAlgExecutor.h.

Referenced by createAggregateWorkUnit(), createCompoundWorkUnit(), createFilterWorkUnit(), createProjectWorkUnit(), createTableFunctionWorkUnit(), createUnionWorkUnit(), executeRelAlgSeq(), executeTableFunction(), executeWorkUnit(), getOuterFragmentCount(), and handleOutOfMemoryRetry().

TemporaryTables RelAlgExecutor::temporary_tables_

private

Definition at line 442 of file RelAlgExecutor.h.

Referenced by addTemporaryTable(), eraseFromTemporaryTables(), executeDelete(), executeProject(), executeRelAlgSeq(), executeRelAlgSubSeq(), executeSort(), executeUpdate(), getOuterFragmentCount(), and handleNop().

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

/home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/RelAlgExecutor.h
/home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/CardinalityEstimator.cpp
/home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/JoinFilterPushDown.cpp
/home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/RelAlgExecutor.cpp

Classes

Public Types

Public Member Functions

Static Public Member Functions

Private Member Functions

Static Private Member Functions

Private Attributes

Static Private Attributes

Friends

Additional Inherited Members

Detailed Description

Member Typedef Documentation

Constructor & Destructor Documentation

Member Function Documentation

Friends And Related Function Documentation

Member Data Documentation