#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()));
   }
 }

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