#include "Execute.h"
#include "AggregateUtils.h"
#include "CodeGenerator.h"
#include "ColumnFetcher.h"
#include "Descriptors/QueryCompilationDescriptor.h"
#include "Descriptors/QueryFragmentDescriptor.h"
#include "DynamicWatchdog.h"
#include "EquiJoinCondition.h"
#include "ErrorHandling.h"
#include "ExpressionRewrite.h"
#include "ExternalCacheInvalidators.h"
#include "GpuMemUtils.h"
#include "InPlaceSort.h"
#include "JoinHashTable/BaselineJoinHashTable.h"
#include "JoinHashTable/OverlapsJoinHashTable.h"
#include "JsonAccessors.h"
#include "OutputBufferInitialization.h"
#include "QueryEngine/QueryDispatchQueue.h"
#include "QueryRewrite.h"
#include "QueryTemplateGenerator.h"
#include "ResultSetReductionJIT.h"
#include "RuntimeFunctions.h"
#include "SpeculativeTopN.h"
#include "TableFunctions/TableFunctionCompilationContext.h"
#include "TableFunctions/TableFunctionExecutionContext.h"
#include "CudaMgr/CudaMgr.h"
#include "DataMgr/BufferMgr/BufferMgr.h"
#include "Parser/ParserNode.h"
#include "Shared/SystemParameters.h"
#include "Shared/TypedDataAccessors.h"
#include "Shared/checked_alloc.h"
#include "Shared/measure.h"
#include "Shared/misc.h"
#include "Shared/scope.h"
#include "Shared/shard_key.h"
#include "Shared/threadpool.h"
#include "AggregatedColRange.h"
#include "StringDictionaryGenerations.h"
#include <llvm/Transforms/Utils/BasicBlockUtils.h>
#include <boost/filesystem/operations.hpp>
#include <boost/filesystem/path.hpp>
#include <chrono>
#include <ctime>
#include <future>
#include <iostream>
#include <memory>
#include <numeric>
#include <set>
#include <thread>
#include "ArrayOps.cpp"
#include "DateAdd.cpp"
#include "StringFunctions.cpp"

Classes
class	anonymous_namespace{Execute.cpp}::OutVecOwner

Namespaces
	anonymous_namespace{Execute.cpp}

Macros
#define	EXECUTE_INCLUDE

Functions
	cpu_code_cache_ (code_cache_size)

	gpu_code_cache_ (code_cache_size)

	block_size_x_ (block_size_x)

	grid_size_x_ (grid_size_x)

	max_gpu_slab_size_ (max_gpu_slab_size)

	debug_dir_ (debug_dir)

	debug_file_ (debug_file)

	executor_id_ (executor_id)

	catalog_ (nullptr)

	temporary_tables_ (nullptr)

	input_table_info_cache_ (this)

ResultSetPtr	anonymous_namespace{Execute.cpp}::get_merged_result (std::vector< std::pair< ResultSetPtr, std::vector< size_t >>> &results_per_device)

ReductionCode	anonymous_namespace{Execute.cpp}::get_reduction_code (std::vector< std::pair< ResultSetPtr, std::vector< size_t >>> &results_per_device, int64_t *compilation_queue_time)

std::unordered_set< int >	get_available_gpus (const Catalog_Namespace::Catalog &cat)

size_t	get_context_count (const ExecutorDeviceType device_type, const size_t cpu_count, const size_t gpu_count)

size_t	anonymous_namespace{Execute.cpp}::compute_buffer_entry_guess (const std::vector< InputTableInfo > &query_infos)

std::string	anonymous_namespace{Execute.cpp}::get_table_name (const InputDescriptor &input_desc, const Catalog_Namespace::Catalog &cat)

size_t	anonymous_namespace{Execute.cpp}::getDeviceBasedScanLimit (const ExecutorDeviceType device_type, const int device_count)

void	anonymous_namespace{Execute.cpp}::checkWorkUnitWatchdog (const RelAlgExecutionUnit &ra_exe_unit, const std::vector< InputTableInfo > &table_infos, const Catalog_Namespace::Catalog &cat, const ExecutorDeviceType device_type, const int device_count)

bool	is_trivial_loop_join (const std::vector< InputTableInfo > &query_infos, const RelAlgExecutionUnit &ra_exe_unit)

template<typename T >
std::vector< std::string >	anonymous_namespace{Execute.cpp}::expr_container_to_string (const T &expr_container)

template<>
std::vector< std::string >	anonymous_namespace{Execute.cpp}::expr_container_to_string (const std::list< Analyzer::OrderEntry > &expr_container)

std::string	anonymous_namespace{Execute.cpp}::join_type_to_string (const JoinType type)

std::string	anonymous_namespace{Execute.cpp}::sort_algorithm_to_string (const SortAlgorithm algorithm)

std::string	ra_exec_unit_desc_for_caching (const RelAlgExecutionUnit &ra_exe_unit)

std::ostream &	operator<< (std::ostream &os, const RelAlgExecutionUnit &ra_exe_unit)

RelAlgExecutionUnit	anonymous_namespace{Execute.cpp}::replace_scan_limit (const RelAlgExecutionUnit &ra_exe_unit_in, const size_t new_scan_limit)

int64_t	anonymous_namespace{Execute.cpp}::inline_null_val (const SQLTypeInfo &ti, const bool float_argument_input)

void	anonymous_namespace{Execute.cpp}::fill_entries_for_empty_input (std::vector< TargetInfo > &target_infos, std::vector< int64_t > &entry, const std::vector< Analyzer::Expr * > &target_exprs, const QueryMemoryDescriptor &query_mem_desc)

ResultSetPtr	anonymous_namespace{Execute.cpp}::build_row_for_empty_input (const std::vector< Analyzer::Expr * > &target_exprs_in, const QueryMemoryDescriptor &query_mem_desc, const ExecutorDeviceType device_type)

size_t	anonymous_namespace{Execute.cpp}::permute_storage_columnar (const ResultSetStorage input_storage, const QueryMemoryDescriptor &input_query_mem_desc, const ResultSetStorage output_storage, size_t output_row_index, const QueryMemoryDescriptor &output_query_mem_desc, const std::vector< uint32_t > &top_permutation)

size_t	anonymous_namespace{Execute.cpp}::permute_storage_row_wise (const ResultSetStorage input_storage, const ResultSetStorage output_storage, size_t output_row_index, const QueryMemoryDescriptor &output_query_mem_desc, const std::vector< uint32_t > &top_permutation)

bool	anonymous_namespace{Execute.cpp}::has_lazy_fetched_columns (const std::vector< ColumnLazyFetchInfo > &fetched_cols)

const ColumnDescriptor *	anonymous_namespace{Execute.cpp}::try_get_column_descriptor (const InputColDescriptor *col_desc, const Catalog_Namespace::Catalog &cat)

std::map< size_t, std::vector < uint64_t > >	get_table_id_to_frag_offsets (const std::vector< InputDescriptor > &input_descs, const std::map< int, const TableFragments * > &all_tables_fragments)

std::ostream &	operator<< (std::ostream &os, FetchResult const &fetch_result)

bool	anonymous_namespace{Execute.cpp}::check_rows_less_than_needed (const ResultSetPtr &results, const size_t scan_limit)

std::tuple< bool, int64_t, int64_t >	anonymous_namespace{Execute.cpp}::get_hpt_overflow_underflow_safe_scaled_values (const int64_t chunk_min, const int64_t chunk_max, const SQLTypeInfo &lhs_type, const SQLTypeInfo &rhs_type)

Variables
bool g_enable_watchdog	false

bool	g_enable_dynamic_watchdog {false}

bool	g_use_tbb_pool {false}

bool	g_enable_filter_function {true}

unsigned	g_dynamic_watchdog_time_limit {10000}

bool	g_allow_cpu_retry {true}

bool	g_null_div_by_zero {false}

unsigned	g_trivial_loop_join_threshold {1000}

bool	g_from_table_reordering {true}

bool	g_inner_join_fragment_skipping {true}

bool	g_enable_smem_group_by

std::unique_ptr< llvm::Module >	udf_gpu_module

std::unique_ptr< llvm::Module >	udf_cpu_module

bool	g_enable_filter_push_down {false}

float	g_filter_push_down_low_frac {-1.0f}

float	g_filter_push_down_high_frac {-1.0f}

size_t	g_filter_push_down_passing_row_ubound {0}

bool	g_enable_columnar_output {false}

bool	g_enable_overlaps_hashjoin {false}

bool	g_enable_hashjoin_many_to_many {false}

size_t	g_overlaps_max_table_size_bytes {1024 * 1024 * 1024}

bool	g_strip_join_covered_quals {false}

size_t	g_constrained_by_in_threshold {10}

size_t	g_big_group_threshold {20000}

bool	g_enable_window_functions {true}

bool	g_enable_table_functions {false}

size_t	g_max_memory_allocation_size {2000000000}

size_t	g_min_memory_allocation_size

bool	g_enable_bump_allocator {false}

double	g_bump_allocator_step_reduction {0.75}

bool	g_enable_direct_columnarization {true}

bool	g_enable_experimental_string_functions

bool	g_enable_runtime_query_interrupt {false}

bool	g_use_estimator_result_cache {true}

unsigned	g_pending_query_interrupt_freq {1000}

double	g_running_query_interrupt_freq {0.5}

size_t	g_gpu_smem_threshold

bool	g_enable_smem_grouped_non_count_agg

bool	g_enable_smem_non_grouped_agg

bool	g_is_test_env {false}

size_t	g_approx_quantile_buffer {1000}

size_t	g_approx_quantile_centroids {300}

bool	g_cache_string_hash

Macro Definition Documentation

#define EXECUTE_INCLUDE

Definition at line 3241 of file Execute.cpp.

Function Documentation

block_size_x_ ( block_size_x )

Referenced by Executor::blockSize().

Here is the caller graph for this function:

catalog_ ( nullptr )

Referenced by Executor::addDeletedColumn(), Executor::blockSize(), Executor::collectAllDeviceResults(), Executor::collectAllDeviceShardedTopResults(), Executor::computeColRangesCache(), Executor::computeStringDictionaryGenerations(), Fragmenter_Namespace::InsertOrderFragmenter::conditionallyInstantiateFileMgrWithParams(), Executor::createKernels(), Fragmenter_Namespace::InsertOrderFragmenter::deleteFragments(), Executor::deviceCount(), Executor::deviceCycles(), Executor::executeWorkUnitImpl(), Executor::getCatalog(), Executor::getColLazyFetchInfo(), Executor::getColumnDescriptor(), Executor::getNumBytesForFetchedRow(), Executor::getPhysicalColumnDescriptor(), ResultSet::getStringDictionaryPayloadCopy(), Executor::gridSize(), Fragmenter_Namespace::InsertOrderFragmenter::insertData(), Fragmenter_Namespace::InsertOrderFragmenter::insertDataImpl(), Executor::isArchMaxwell(), Executor::isCPUOnly(), Executor::numBlocksPerMP(), ResultSet::radixSortOnGpu(), Executor::reduceMultiDeviceResults(), Executor::reduceMultiDeviceResultSets(), Fragmenter_Namespace::InsertOrderFragmenter::replicateData(), Executor::resultsUnion(), Executor::setCatalog(), Executor::setupCaching(), Executor::skipFragment(), Fragmenter_Namespace::InsertOrderFragmenter::updateChunkStats(), Executor::warpSize(), StorageIOFacility< EXECUTOR_TRAITS, FRAGMENT_UPDATER >::yieldDeleteCallback(), and StorageIOFacility< EXECUTOR_TRAITS, FRAGMENT_UPDATER >::yieldUpdateCallback().

Here is the caller graph for this function:

cpu_code_cache_ ( code_cache_size )

Referenced by Executor::optimizeAndCodegenCPU().

Here is the caller graph for this function:

debug_dir_ ( debug_dir )

debug_file_ ( debug_file )

executor_id_ ( executor_id )

Referenced by Executor::attachExecutorToQuerySession(), Executor::clearQuerySessionStatus(), Executor::executeUpdate(), Executor::executeWorkUnit(), Executor::removeFromQuerySessionList(), and Executor::updateQuerySessionStatus().

Here is the caller graph for this function:

std::unordered_set<int> get_available_gpus ( const Catalog_Namespace::Catalog & cat )

Definition at line 1030 of file Execute.cpp.

References CHECK_GT, Data_Namespace::DataMgr::getCudaMgr(), Catalog_Namespace::Catalog::getDataMgr(), CudaMgr_Namespace::CudaMgr::getDeviceCount(), and Data_Namespace::DataMgr::gpusPresent().

Referenced by Executor::executeWorkUnitImpl().

                                                                               {
   std::unordered_set<int> available_gpus;
   if (cat.getDataMgr().gpusPresent()) {
     int gpu_count = cat.getDataMgr().getCudaMgr()->getDeviceCount();
     CHECK_GT(gpu_count, 0);
     for (int gpu_id = 0; gpu_id < gpu_count; ++gpu_id) {
       available_gpus.insert(gpu_id);
     }
   }
   return available_gpus;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

size_t get_context_count	(	const ExecutorDeviceType	device_type,
		const size_t	cpu_count,
		const size_t	gpu_count
	)

Definition at line 1042 of file Execute.cpp.

References GPU.

Referenced by Executor::executeWorkUnitImpl().

                                                  {
   return device_type == ExecutorDeviceType::GPU ? gpu_count
                                                 : static_cast<size_t>(cpu_count);
 }

Here is the caller graph for this function:

std::map<size_t, std::vector<uint64_t> > get_table_id_to_frag_offsets	(	const std::vector< InputDescriptor > &	input_descs,
		const std::map< int, const TableFragments * > &	all_tables_fragments
	)

Definition at line 2257 of file Execute.cpp.

References CHECK.

Referenced by Executor::getRowCountAndOffsetForAllFrags().

                                                                     {
   std::map<size_t, std::vector<uint64_t>> tab_id_to_frag_offsets;
   for (auto& desc : input_descs) {
     const auto fragments_it = all_tables_fragments.find(desc.getTableId());
     CHECK(fragments_it != all_tables_fragments.end());
     const auto& fragments = *fragments_it->second;
     std::vector<uint64_t> frag_offsets(fragments.size(), 0);
     for (size_t i = 0, off = 0; i < fragments.size(); ++i) {
       frag_offsets[i] = off;
       off += fragments[i].getNumTuples();
     }
     tab_id_to_frag_offsets.insert(std::make_pair(desc.getTableId(), frag_offsets));
   }
   return tab_id_to_frag_offsets;
 }

Here is the caller graph for this function:

gpu_code_cache_ ( code_cache_size )

grid_size_x_ ( grid_size_x )

Referenced by Executor::gridSize(), and Executor::numBlocksPerMP().

Here is the caller graph for this function:

input_table_info_cache_ ( this )

Definition at line 154 of file Execute.cpp.

Referenced by Executor::clearMetaInfoCache(), and Executor::getTableInfo().

154 {}

Here is the caller graph for this function:

bool is_trivial_loop_join	(	const std::vector< InputTableInfo > &	query_infos,
		const RelAlgExecutionUnit &	ra_exe_unit
	)

Definition at line 1150 of file Execute.cpp.

References CHECK, g_trivial_loop_join_threshold, and RelAlgExecutionUnit::input_descs.

Referenced by anonymous_namespace{IRCodegen.cpp}::check_if_loop_join_is_allowed().

                                                                   {
   if (ra_exe_unit.input_descs.size() < 2) {
     return false;
   }
 
   // We only support loop join at the end of folded joins
   // where ra_exe_unit.input_descs.size() > 2 for now.
   const auto inner_table_id = ra_exe_unit.input_descs.back().getTableId();
 
   std::optional<size_t> inner_table_idx;
   for (size_t i = 0; i < query_infos.size(); ++i) {
     if (query_infos[i].table_id == inner_table_id) {
       inner_table_idx = i;
       break;
     }
   }
   CHECK(inner_table_idx);
   return query_infos[*inner_table_idx].info.getNumTuples() <=
          g_trivial_loop_join_threshold;
 }

Here is the caller graph for this function:

max_gpu_slab_size_ ( max_gpu_slab_size )

Referenced by Executor::maxGpuSlabSize().

Here is the caller graph for this function:

std::ostream& operator<<	(	std::ostream &	os,
		const RelAlgExecutionUnit &	ra_exe_unit
	)

Definition at line 1276 of file Execute.cpp.

                                                                                {
   os << "\n\tTable/Col/Levels: ";
   for (const auto& input_col_desc : ra_exe_unit.input_col_descs) {
     const auto& scan_desc = input_col_desc->getScanDesc();
     os << "(" << scan_desc.getTableId() << ", " << input_col_desc->getColId() << ", "
        << scan_desc.getNestLevel() << ") ";
   }
   if (!ra_exe_unit.simple_quals.empty()) {
     os << "\n\tSimple Quals: "
        << boost::algorithm::join(expr_container_to_string(ra_exe_unit.simple_quals),
                                  ", ");
   }
   if (!ra_exe_unit.quals.empty()) {
     os << "\n\tQuals: "
        << boost::algorithm::join(expr_container_to_string(ra_exe_unit.quals), ", ");
   }
   if (!ra_exe_unit.join_quals.empty()) {
     os << "\n\tJoin Quals: ";
     for (size_t i = 0; i < ra_exe_unit.join_quals.size(); i++) {
       const auto& join_condition = ra_exe_unit.join_quals[i];
       os << "\t\t" << std::to_string(i) << " "
          << join_type_to_string(join_condition.type);
       os << boost::algorithm::join(expr_container_to_string(join_condition.quals), ", ");
     }
   }
   if (!ra_exe_unit.groupby_exprs.empty()) {
     os << "\n\tGroup By: "
        << boost::algorithm::join(expr_container_to_string(ra_exe_unit.groupby_exprs),
                                  ", ");
   }
   os << "\n\tProjected targets: "
      << boost::algorithm::join(expr_container_to_string(ra_exe_unit.target_exprs), ", ");
   os << "\n\tHas Estimator: " << ::toString(ra_exe_unit.estimator == nullptr);
   os << "\n\tSort Info: ";
   const auto& sort_info = ra_exe_unit.sort_info;
   os << "\n\t  Order Entries: "
      << boost::algorithm::join(expr_container_to_string(sort_info.order_entries), ", ");
   os << "\n\t  Algorithm: " << sort_algorithm_to_string(sort_info.algorithm);
   os << "\n\t  Limit: " << std::to_string(sort_info.limit);
   os << "\n\t  Offset: " << std::to_string(sort_info.offset);
   os << "\n\tScan Limit: " << std::to_string(ra_exe_unit.scan_limit);
   os << "\n\tBump Allocator: " << ::toString(ra_exe_unit.use_bump_allocator);
   if (ra_exe_unit.union_all) {
     os << "\n\tUnion: " << std::string(*ra_exe_unit.union_all ? "UNION ALL" : "UNION");
   }
   return os;
 }

Here is the call graph for this function:

std::ostream& operator<<	(	std::ostream &	os,
		FetchResult const &	fetch_result
	)

Definition at line 2344 of file Execute.cpp.

References FetchResult::col_buffers, FetchResult::frag_offsets, FetchResult::num_rows, and shared::printContainer().

                                                                         {
   return os << "col_buffers" << shared::printContainer(fetch_result.col_buffers)
             << " num_rows" << shared::printContainer(fetch_result.num_rows)
             << " frag_offsets" << shared::printContainer(fetch_result.frag_offsets);
 }

Here is the call graph for this function:

std::string ra_exec_unit_desc_for_caching ( const RelAlgExecutionUnit & ra_exe_unit )

Definition at line 1225 of file Execute.cpp.

References RelAlgExecutionUnit::estimator, RelAlgExecutionUnit::groupby_exprs, RelAlgExecutionUnit::input_col_descs, RelAlgExecutionUnit::join_quals, anonymous_namespace{Execute.cpp}::join_type_to_string(), RelAlgExecutionUnit::quals, RelAlgExecutionUnit::scan_limit, RelAlgExecutionUnit::simple_quals, RelAlgExecutionUnit::target_exprs, to_string(), and toString().

Referenced by RelAlgExecutor::executeWorkUnit().

                                                                                 {
   // todo(yoonmin): replace a cache key as a DAG representation of a query plan
   // instead of ra_exec_unit description if possible
   std::ostringstream os;
   for (const auto& input_col_desc : ra_exe_unit.input_col_descs) {
     const auto& scan_desc = input_col_desc->getScanDesc();
     os << scan_desc.getTableId() << "," << input_col_desc->getColId() << ","
        << scan_desc.getNestLevel();
   }
   if (!ra_exe_unit.simple_quals.empty()) {
     for (const auto& qual : ra_exe_unit.simple_quals) {
       if (qual) {
         os << qual->toString() << ",";
       }
     }
   }
   if (!ra_exe_unit.quals.empty()) {
     for (const auto& qual : ra_exe_unit.quals) {
       if (qual) {
         os << qual->toString() << ",";
       }
     }
   }
   if (!ra_exe_unit.join_quals.empty()) {
     for (size_t i = 0; i < ra_exe_unit.join_quals.size(); i++) {
       const auto& join_condition = ra_exe_unit.join_quals[i];
       os << std::to_string(i) << join_type_to_string(join_condition.type);
       for (const auto& qual : join_condition.quals) {
         if (qual) {
           os << qual->toString() << ",";
         }
       }
     }
   }
   if (!ra_exe_unit.groupby_exprs.empty()) {
     for (const auto& qual : ra_exe_unit.groupby_exprs) {
       if (qual) {
         os << qual->toString() << ",";
       }
     }
   }
   for (const auto& expr : ra_exe_unit.target_exprs) {
     if (expr) {
       os << expr->toString() << ",";
     }
   }
   os << ::toString(ra_exe_unit.estimator == nullptr);
   os << std::to_string(ra_exe_unit.scan_limit);
   return os.str();
 }