#include <WindowContext.h>

Collaboration diagram for WindowFunctionContext:

Classes
struct	AggregateState

Public Types
enum	WindowComparatorResult { WindowComparatorResult::LT, WindowComparatorResult::EQ, WindowComparatorResult::GT }

using	Comparator = std::function< WindowFunctionContext::WindowComparatorResult(const int64_t lhs, const int64_t rhs)>

Public Member Functions
	WindowFunctionContext (const Analyzer::WindowFunction *window_func, const size_t elem_count, const ExecutorDeviceType device_type, std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner)

	WindowFunctionContext (const Analyzer::WindowFunction *window_func, QueryPlanHash cache_key, const std::shared_ptr< HashJoin > &partitions, const size_t elem_count, const ExecutorDeviceType device_type, std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner, size_t aggregation_tree_fan_out=g_window_function_aggregation_tree_fanout)

	WindowFunctionContext (const WindowFunctionContext &)=delete

WindowFunctionContext &	operator= (const WindowFunctionContext &)=delete

	~WindowFunctionContext ()

void	addOrderColumn (const int8_t *column, const SQLTypeInfo &ti, const std::vector< std::shared_ptr< Chunk_NS::Chunk >> &chunks_owner)

void	setSortedPartitionCacheKey (QueryPlanHash cache_key)

void	addColumnBufferForWindowFunctionExpression (const int8_t *column, const std::vector< std::shared_ptr< Chunk_NS::Chunk >> &chunks_owner)

std::vector< Comparator >	createComparator (size_t partition_idx)

void	compute (std::unordered_map< QueryPlanHash, size_t > &sorted_partition_key_ref_count_map, std::unordered_map< QueryPlanHash, std::shared_ptr< std::vector< int64_t >>> &sorted_partition_cache, std::unordered_map< QueryPlanHash, AggregateTreeForWindowFraming > &aggregate_tree_map)

const Analyzer::WindowFunction *	getWindowFunction () const

const int8_t *	output () const

const int64_t *	sortedPartition () const

const int64_t *	aggregateState () const

const int64_t *	aggregateStateCount () const

int64_t	aggregateStatePendingOutputs () const

const int64_t *	partitionStartOffset () const

const int64_t *	partitionNumCountBuf () const

const std::vector< const int8_t * > &	getColumnBufferForWindowFunctionExpressions () const

const std::vector< const int8_t * > &	getOrderKeyColumnBuffers () const

const std::vector< SQLTypeInfo > &	getOrderKeyColumnBufferTypes () const

int64_t **	getAggregationTreesForIntegerTypeWindowExpr () const

double **	getAggregationTreesForDoubleTypeWindowExpr () const

SumAndCountPair< int64_t > **	getDerivedAggregationTreesForIntegerTypeWindowExpr () const

SumAndCountPair< double > **	getDerivedAggregationTreesForDoubleTypeWindowExpr () const

size_t *	getAggregateTreeDepth () const

size_t	getAggregateTreeFanout () const

int64_t *	getNullValueStartPos () const

int64_t *	getNullValueEndPos () const

const int8_t *	partitionStart () const

const int8_t *	partitionEnd () const

size_t	elementCount () const

const int32_t *	payload () const

const int32_t *	offsets () const

const int32_t *	counts () const

size_t	partitionCount () const

const bool	needsToBuildAggregateTree () const

Static Public Attributes
static const int	NUM_EXECUTION_DEVICES = 1

Private Member Functions
void	computePartitionBuffer (const size_t partition_idx, int64_t output_for_partition_buff, const Analyzer::WindowFunction window_func)

void	sortPartition (const size_t partition_idx, int64_t *output_for_partition_buff, bool should_parallelize)

void	computeNullRangeOfSortedPartition (const SQLTypeInfo &order_col_ti, size_t partition_idx, const int32_t original_col_idx_buf, const int64_t ordered_col_idx_buf)

void	buildAggregationTreeForPartition (SqlWindowFunctionKind agg_type, size_t partition_idx, size_t partition_size, const int32_t original_rowid_buf, const int64_t ordered_rowid_buf, const SQLTypeInfo &input_col_ti)

void	fillPartitionStart ()

void	fillPartitionEnd ()

void	resizeStorageForWindowFraming (bool const for_reuse=false)

const QueryPlanHash	computeAggregateTreeCacheKey () const

Static Private Member Functions
static Comparator	makeComparator (const Analyzer::ColumnVar col_var, const int8_t partition_values, const int32_t *partition_indices, const bool asc_ordering, const bool nulls_first)

Private Attributes
const Analyzer::WindowFunction *	window_func_

QueryPlanHash	partition_cache_key_

QueryPlanHash	sorted_partition_cache_key_

std::vector< std::vector < std::shared_ptr < Chunk_NS::Chunk > > >	order_columns_owner_

std::vector< const int8_t * >	order_columns_

std::vector< SQLTypeInfo >	order_columns_ti_

std::shared_ptr< HashJoin >	partitions_

size_t	elem_count_

int8_t *	output_

std::shared_ptr< std::vector < int64_t > >	sorted_partition_buf_

std::vector< std::vector < std::shared_ptr < Chunk_NS::Chunk > > >	window_func_expr_columns_owner_

std::vector< const int8_t * >	window_func_expr_columns_

std::vector< std::shared_ptr < void > >	segment_trees_owned_

AggregateTreeForWindowFraming	aggregate_trees_

size_t	aggregate_trees_fan_out_

size_t *	aggregate_trees_depth_

int64_t *	ordered_partition_null_start_pos_

int64_t *	ordered_partition_null_end_pos_

int64_t *	partition_start_offset_

int8_t *	partition_start_

int8_t *	partition_end_

AggregateState	aggregate_state_

const ExecutorDeviceType	device_type_

std::shared_ptr < RowSetMemoryOwner >	row_set_mem_owner_

const int32_t	dummy_count_

const int32_t	dummy_offset_

int32_t *	dummy_payload_

Detailed Description

Definition at line 136 of file WindowContext.h.

Member Typedef Documentation

using WindowFunctionContext::Comparator = std::function<WindowFunctionContext::WindowComparatorResult(const int64_t lhs, const int64_t rhs)>

Definition at line 181 of file WindowContext.h.

Member Enumeration Documentation

enum WindowFunctionContext::WindowComparatorResult

strong

Enumerator
LT
EQ
GT

Definition at line 178 of file WindowContext.h.

178 { LT, EQ, GT };

StatsRequestPredicateOp::LT

StatsRequestPredicateOp::GT

Constructor & Destructor Documentation

WindowFunctionContext::WindowFunctionContext	(	const Analyzer::WindowFunction *	window_func,
		const size_t	elem_count,
		const ExecutorDeviceType	device_type,
		std::shared_ptr< RowSetMemoryOwner >	row_set_mem_owner
	)

Definition at line 50 of file WindowContext.cpp.

References aggregate_trees_depth_, CHECK_LE, checked_calloc(), checked_malloc(), dummy_payload_, elem_count_, Analyzer::WindowFunction::getKind(), Analyzer::WindowFunction::hasFraming(), gpu_enabled::iota(), Analyzer::WindowFunction::isMissingValueFillingFunction(), NTH_VALUE, ordered_partition_null_end_pos_, ordered_partition_null_start_pos_, partition_start_offset_, and window_func_.

     : window_func_(window_func)
     , partition_cache_key_(EMPTY_HASHED_PLAN_DAG_KEY)
     , sorted_partition_cache_key_(EMPTY_HASHED_PLAN_DAG_KEY)
     , partitions_(nullptr)
     , elem_count_(elem_count)
     , output_(nullptr)
     , sorted_partition_buf_(nullptr)
     , aggregate_trees_fan_out_(g_window_function_aggregation_tree_fanout)
     , aggregate_trees_depth_(nullptr)
     , ordered_partition_null_start_pos_(nullptr)
     , ordered_partition_null_end_pos_(nullptr)
     , partition_start_offset_(nullptr)
     , partition_start_(nullptr)
     , partition_end_(nullptr)
     , device_type_(device_type)
     , row_set_mem_owner_(row_set_mem_owner)
     , dummy_count_(elem_count)
     , dummy_offset_(0)
     , dummy_payload_(nullptr) {
   CHECK_LE(elem_count_, static_cast<size_t>(std::numeric_limits<int32_t>::max()));
   dummy_payload_ =
       reinterpret_cast<int32_t*>(checked_malloc(elem_count_ * sizeof(int32_t)));
   std::iota(dummy_payload_, dummy_payload_ + elem_count_, int32_t(0));
   if (window_func_->hasFraming() || window_func_->isMissingValueFillingFunction() ||
       window_func_->getKind() == SqlWindowFunctionKind::NTH_VALUE) {
     // in this case, we consider all rows of the row belong to the same and only
     // existing partition
     partition_start_offset_ =
         reinterpret_cast<int64_t*>(checked_calloc(2, sizeof(int64_t)));
     partition_start_offset_[1] = elem_count_;
     aggregate_trees_depth_ = reinterpret_cast<size_t*>(checked_calloc(1, sizeof(size_t)));
     ordered_partition_null_start_pos_ =
         reinterpret_cast<int64_t*>(checked_calloc(1, sizeof(int64_t)));
     ordered_partition_null_end_pos_ =
         reinterpret_cast<int64_t*>(checked_calloc(1, sizeof(int64_t)));
   }
 }

Here is the call graph for this function:

WindowFunctionContext::WindowFunctionContext	(	const Analyzer::WindowFunction *	window_func,
		QueryPlanHash	cache_key,
		const std::shared_ptr< HashJoin > &	partitions,
		const size_t	elem_count,
		const ExecutorDeviceType	device_type,
		std::shared_ptr< RowSetMemoryOwner >	row_set_mem_owner,
		size_t	aggregation_tree_fan_out = `g_window_function_aggregation_tree_fanout`
	)

Definition at line 94 of file WindowContext.cpp.

References aggregate_trees_depth_, CHECK, checked_calloc(), counts(), Analyzer::WindowFunction::hasFraming(), Analyzer::WindowFunction::isMissingValueFillingFunction(), ordered_partition_null_end_pos_, ordered_partition_null_start_pos_, gpu_enabled::partial_sum(), partition_start_offset_, partitionCount(), partitions_, and window_func_.

     : window_func_(window_func)
     , partition_cache_key_(partition_cache_key)
     , sorted_partition_cache_key_(EMPTY_HASHED_PLAN_DAG_KEY)
     , partitions_(partitions)
     , elem_count_(elem_count)
     , output_(nullptr)
     , sorted_partition_buf_(nullptr)
     , aggregate_trees_fan_out_(aggregation_tree_fan_out)
     , aggregate_trees_depth_(nullptr)
     , ordered_partition_null_start_pos_(nullptr)
     , ordered_partition_null_end_pos_(nullptr)
     , partition_start_offset_(nullptr)
     , partition_start_(nullptr)
     , partition_end_(nullptr)
     , device_type_(device_type)
     , row_set_mem_owner_(row_set_mem_owner)
     , dummy_count_(elem_count)
     , dummy_offset_(0)
     , dummy_payload_(nullptr) {
   CHECK(partitions_);  // This version should have hash table
   size_t partition_count = partitionCount();
   partition_start_offset_ =
       reinterpret_cast<int64_t*>(checked_calloc(partition_count + 1, sizeof(int64_t)));
   if (window_func_->hasFraming() || window_func_->isMissingValueFillingFunction()) {
     aggregate_trees_depth_ =
         reinterpret_cast<size_t*>(checked_calloc(partition_count, sizeof(size_t)));
     ordered_partition_null_start_pos_ =
         reinterpret_cast<int64_t*>(checked_calloc(partition_count, sizeof(int64_t)));
     ordered_partition_null_end_pos_ =
         reinterpret_cast<int64_t*>(checked_calloc(partition_count, sizeof(int64_t)));
   }
   // the first partition starts at zero position
   std::partial_sum(counts(), counts() + partition_count, partition_start_offset_ + 1);
 }

Here is the call graph for this function:

WindowFunctionContext::WindowFunctionContext ( const WindowFunctionContext & )

delete

WindowFunctionContext::~WindowFunctionContext ( )

Definition at line 137 of file WindowContext.cpp.

References aggregate_trees_depth_, dummy_payload_, ordered_partition_null_end_pos_, ordered_partition_null_start_pos_, partition_end_, partition_start_, and partition_start_offset_.

                                               {
   free(partition_start_);
   free(partition_end_);
   if (dummy_payload_) {
     free(dummy_payload_);
   }
   if (partition_start_offset_) {
     free(partition_start_offset_);
   }
   if (aggregate_trees_depth_) {
     free(aggregate_trees_depth_);
   }
   if (ordered_partition_null_start_pos_) {
     free(ordered_partition_null_start_pos_);
   }
   if (ordered_partition_null_end_pos_) {
     free(ordered_partition_null_end_pos_);
   }
 }

Member Function Documentation

void WindowFunctionContext::addColumnBufferForWindowFunctionExpression	(	const int8_t *	column,
		const std::vector< std::shared_ptr< Chunk_NS::Chunk >> &	chunks_owner
	)

Definition at line 166 of file WindowContext.cpp.

References window_func_expr_columns_, and window_func_expr_columns_owner_.

                                                                  {
   window_func_expr_columns_owner_.push_back(chunks_owner);
   window_func_expr_columns_.push_back(column);
 };

void WindowFunctionContext::addOrderColumn	(	const int8_t *	column,
		const SQLTypeInfo &	ti,
		const std::vector< std::shared_ptr< Chunk_NS::Chunk >> &	chunks_owner
	)

Definition at line 157 of file WindowContext.cpp.

References order_columns_, order_columns_owner_, and order_columns_ti_.

                                                                  {
   order_columns_owner_.push_back(chunks_owner);
   order_columns_.push_back(column);
   order_columns_ti_.push_back(ti);
 }

const int64_t * WindowFunctionContext::aggregateState ( ) const

Definition at line 993 of file WindowContext.cpp.

References aggregate_state_, CHECK, Analyzer::WindowFunction::getKind(), WindowFunctionContext::AggregateState::val, window_func_, and window_function_is_aggregate().

                                                            {
   CHECK(window_function_is_aggregate(window_func_->getKind()));
   return &aggregate_state_.val;
 }

Here is the call graph for this function:

const int64_t * WindowFunctionContext::aggregateStateCount ( ) const

Definition at line 998 of file WindowContext.cpp.

References aggregate_state_, CHECK, WindowFunctionContext::AggregateState::count, Analyzer::WindowFunction::getKind(), window_func_, and window_function_is_aggregate().

                                                                 {
   CHECK(window_function_is_aggregate(window_func_->getKind()));
   return &aggregate_state_.count;
 }

Here is the call graph for this function:

int64_t WindowFunctionContext::aggregateStatePendingOutputs ( ) const

Definition at line 1013 of file WindowContext.cpp.

References aggregate_state_, CHECK, Analyzer::WindowFunction::getKind(), WindowFunctionContext::AggregateState::outputs, window_func_, and window_function_is_aggregate().

                                                                   {
   CHECK(window_function_is_aggregate(window_func_->getKind()));
   return reinterpret_cast<int64_t>(&aggregate_state_.outputs);
 }

Here is the call graph for this function:

void WindowFunctionContext::buildAggregationTreeForPartition	(	SqlWindowFunctionKind	agg_type,
		size_t	partition_idx,
		size_t	partition_size,
		const int32_t *	original_rowid_buf,
		const int64_t *	ordered_rowid_buf,
		const SQLTypeInfo &	input_col_ti
	)

private

Definition at line 1473 of file WindowContext.cpp.

References AggregateTreeForWindowFraming::aggregate_tree_for_double_type_, AggregateTreeForWindowFraming::aggregate_tree_for_integer_type_, aggregate_trees_, AggregateTreeForWindowFraming::aggregate_trees_depth_, aggregate_trees_depth_, aggregate_trees_fan_out_, anonymous_namespace{WindowContext.cpp}::allow_framing_on_time_or_date(), AVG, CHECK, COUNT, decimal_to_int_type(), AggregateTreeForWindowFraming::derived_aggregate_tree_for_double_type_, AggregateTreeForWindowFraming::derived_aggregate_tree_for_integer_type_, get_int_type_by_size(), SQLTypeInfo::get_size(), SQLTypeInfo::get_type(), Analyzer::WindowFunction::getKind(), SQLTypeInfo::is_boolean(), SQLTypeInfo::is_decimal(), SQLTypeInfo::is_fp(), SQLTypeInfo::is_integer(), SQLTypeInfo::is_number(), SQLTypeInfo::is_time_or_date(), kBIGINT, kBOOLEAN, kDECIMAL, kDOUBLE, kFLOAT, kINT, kNUMERIC, kSMALLINT, kTINYINT, MAX, MIN, offsets(), ordered_partition_null_end_pos_, ordered_partition_null_start_pos_, segment_trees_owned_, toString(), run_benchmark_import::type, UNREACHABLE, window_func_, and window_func_expr_columns_.

Referenced by compute().

                                      {
   if (!(input_col_ti.is_number() || input_col_ti.is_boolean() ||
         input_col_ti.is_time_or_date())) {
     throw QueryNotSupported("Window aggregate function over frame on a column type " +
                             ::toString(input_col_ti.get_type()) + " is not supported.");
   }
   if (input_col_ti.is_time_or_date() &&
       !allow_framing_on_time_or_date(window_func_->getKind()) &&
       !(agg_type == SqlWindowFunctionKind::MIN ||
         agg_type == SqlWindowFunctionKind::MAX ||
         agg_type == SqlWindowFunctionKind::COUNT)) {
     throw QueryNotSupported(
         "Aggregation over a window frame for a column type " +
         ::toString(input_col_ti.get_type()) +
         " must use one of the following window aggregate function: MIN / MAX / COUNT");
   }
   const auto type = input_col_ti.is_decimal() ? decimal_to_int_type(input_col_ti)
                     : input_col_ti.is_time_or_date()
                         ? get_int_type_by_size(input_col_ti.get_size())
                         : input_col_ti.get_type();
   if (partition_size > 0) {
     IndexPair order_col_null_range{ordered_partition_null_start_pos_[partition_idx],
                                    ordered_partition_null_end_pos_[partition_idx]};
     const int64_t* ordered_rowid_buf_for_partition =
         ordered_rowid_buf + offsets()[partition_idx];
     switch (type) {
       case kBOOLEAN:
       case kTINYINT: {
         const auto segment_tree = std::make_shared<SegmentTree<int8_t, int64_t>>(
             window_func_expr_columns_,
             input_col_ti,
             original_rowid_buf,
             ordered_rowid_buf_for_partition,
             partition_size,
             agg_type,
             aggregate_trees_fan_out_);
         aggregate_trees_depth_[partition_idx] =
             segment_tree ? segment_tree->getLeafDepth() : 0;
         if (agg_type == SqlWindowFunctionKind::AVG) {
           aggregate_trees_.derived_aggregate_tree_for_integer_type_[partition_idx] =
               segment_tree ? segment_tree->getDerivedAggregatedValues() : nullptr;
         } else {
           aggregate_trees_.aggregate_tree_for_integer_type_[partition_idx] =
               segment_tree ? segment_tree->getAggregatedValues() : nullptr;
         }
         segment_trees_owned_[partition_idx] = std::move(segment_tree);
         break;
       }
       case kSMALLINT: {
         const auto segment_tree = std::make_shared<SegmentTree<int16_t, int64_t>>(
             window_func_expr_columns_,
             input_col_ti,
             original_rowid_buf,
             ordered_rowid_buf_for_partition,
             partition_size,
             agg_type,
             aggregate_trees_fan_out_);
         aggregate_trees_depth_[partition_idx] =
             segment_tree ? segment_tree->getLeafDepth() : 0;
         if (agg_type == SqlWindowFunctionKind::AVG) {
           aggregate_trees_.derived_aggregate_tree_for_integer_type_[partition_idx] =
               segment_tree ? segment_tree->getDerivedAggregatedValues() : nullptr;
         } else {
           aggregate_trees_.aggregate_tree_for_integer_type_[partition_idx] =
               segment_tree ? segment_tree->getAggregatedValues() : nullptr;
         }
         segment_trees_owned_[partition_idx] = std::move(segment_tree);
         break;
       }
       case kINT: {
         const auto segment_tree = std::make_shared<SegmentTree<int32_t, int64_t>>(
             window_func_expr_columns_,
             input_col_ti,
             original_rowid_buf,
             ordered_rowid_buf_for_partition,
             partition_size,
             agg_type,
             aggregate_trees_fan_out_);
         aggregate_trees_depth_[partition_idx] =
             segment_tree ? segment_tree->getLeafDepth() : 0;
         if (agg_type == SqlWindowFunctionKind::AVG) {
           aggregate_trees_.derived_aggregate_tree_for_integer_type_[partition_idx] =
               segment_tree ? segment_tree->getDerivedAggregatedValues() : nullptr;
         } else {
           aggregate_trees_.aggregate_tree_for_integer_type_[partition_idx] =
               segment_tree ? segment_tree->getAggregatedValues() : nullptr;
         }
         segment_trees_owned_[partition_idx] = std::move(segment_tree);
         break;
       }
       case kDECIMAL:
       case kNUMERIC:
       case kBIGINT: {
         const auto segment_tree = std::make_shared<SegmentTree<int64_t, int64_t>>(
             window_func_expr_columns_,
             input_col_ti,
             original_rowid_buf,
             ordered_rowid_buf_for_partition,
             partition_size,
             agg_type,
             aggregate_trees_fan_out_);
         aggregate_trees_depth_[partition_idx] =
             segment_tree ? segment_tree->getLeafDepth() : 0;
         if (agg_type == SqlWindowFunctionKind::AVG) {
           aggregate_trees_.derived_aggregate_tree_for_integer_type_[partition_idx] =
               segment_tree ? segment_tree->getDerivedAggregatedValues() : nullptr;
         } else {
           aggregate_trees_.aggregate_tree_for_integer_type_[partition_idx] =
               segment_tree ? segment_tree->getAggregatedValues() : nullptr;
         }
         segment_trees_owned_[partition_idx] = std::move(segment_tree);
         break;
       }
       case kFLOAT: {
         const auto segment_tree =
             std::make_shared<SegmentTree<float, double>>(window_func_expr_columns_,
                                                          input_col_ti,
                                                          original_rowid_buf,
                                                          ordered_rowid_buf_for_partition,
                                                          partition_size,
                                                          agg_type,
                                                          aggregate_trees_fan_out_);
         aggregate_trees_depth_[partition_idx] =
             segment_tree ? segment_tree->getLeafDepth() : 0;
         if (agg_type == SqlWindowFunctionKind::AVG) {
           aggregate_trees_.derived_aggregate_tree_for_double_type_[partition_idx] =
               segment_tree ? segment_tree->getDerivedAggregatedValues() : nullptr;
         } else {
           aggregate_trees_.aggregate_tree_for_double_type_[partition_idx] =
               segment_tree ? segment_tree->getAggregatedValues() : nullptr;
         }
         segment_trees_owned_[partition_idx] = std::move(segment_tree);
         break;
       }
       case kDOUBLE: {
         const auto segment_tree =
             std::make_shared<SegmentTree<double, double>>(window_func_expr_columns_,
                                                           input_col_ti,
                                                           original_rowid_buf,
                                                           ordered_rowid_buf_for_partition,
                                                           partition_size,
                                                           agg_type,
                                                           aggregate_trees_fan_out_);
         aggregate_trees_depth_[partition_idx] =
             segment_tree ? segment_tree->getLeafDepth() : 0;
         if (agg_type == SqlWindowFunctionKind::AVG) {
           aggregate_trees_.derived_aggregate_tree_for_double_type_[partition_idx] =
               segment_tree ? segment_tree->getDerivedAggregatedValues() : nullptr;
         } else {
           aggregate_trees_.aggregate_tree_for_double_type_[partition_idx] =
               segment_tree ? segment_tree->getAggregatedValues() : nullptr;
         }
         segment_trees_owned_[partition_idx] = std::move(segment_tree);
         break;
       }
       default:
         UNREACHABLE();
     }
   } else {
     // handling a case of an empty partition
     aggregate_trees_depth_[partition_idx] = 0;
     if (input_col_ti.is_integer() || input_col_ti.is_decimal() ||
         input_col_ti.is_boolean() || input_col_ti.is_time_or_date()) {
       if (agg_type == SqlWindowFunctionKind::AVG) {
         aggregate_trees_.derived_aggregate_tree_for_integer_type_[partition_idx] =
             nullptr;
       } else {
         aggregate_trees_.aggregate_tree_for_integer_type_[partition_idx] = nullptr;
       }
     } else {
       CHECK(input_col_ti.is_fp());
       if (agg_type == SqlWindowFunctionKind::AVG) {
         aggregate_trees_.derived_aggregate_tree_for_double_type_[partition_idx] = nullptr;
       } else {
         aggregate_trees_.aggregate_tree_for_double_type_[partition_idx] = nullptr;
       }
     }
   }
   aggregate_trees_.aggregate_trees_depth_ = aggregate_trees_depth_;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void WindowFunctionContext::compute	(	std::unordered_map< QueryPlanHash, size_t > &	sorted_partition_key_ref_count_map,
		std::unordered_map< QueryPlanHash, std::shared_ptr< std::vector< int64_t >>> &	sorted_partition_cache,
		std::unordered_map< QueryPlanHash, AggregateTreeForWindowFraming > &	aggregate_tree_map
	)

Definition at line 549 of file WindowContext.cpp.

References aggregate_trees_, AggregateTreeForWindowFraming::aggregate_trees_depth_, aggregate_trees_depth_, buildAggregationTreeForPartition(), CHECK, computeAggregateTreeCacheKey(), computeNullRangeOfSortedPartition(), computePartitionBuffer(), counts(), DEBUG_TIMER, elem_count_, fillPartitionEnd(), fillPartitionStart(), g_enable_parallel_window_partition_compute, g_parallel_window_partition_compute_threshold, Analyzer::WindowFunction::getArgs(), Analyzer::WindowFunction::getKind(), Analyzer::WindowFunction::getOrderKeys(), Analyzer::WindowFunction::hasFraming(), Analyzer::WindowFunction::isMissingValueFillingFunction(), needsToBuildAggregateTree(), offsets(), output_, threading_serial::parallel_for(), partitionCount(), payload(), resizeStorageForWindowFraming(), row_set_mem_owner_, sorted_partition_buf_, sorted_partition_cache_key_, sortPartition(), logger::thread_local_ids(), toString(), VLOG, window_func_, anonymous_namespace{WindowContext.cpp}::window_function_buffer_element_size(), window_function_is_aggregate(), and window_function_requires_peer_handling().

                                                                                  {
   auto timer = DEBUG_TIMER(__func__);
   CHECK(!output_);
   if (elem_count_ == 0) {
     return;
   }
   size_t output_buf_sz =
       elem_count_ * window_function_buffer_element_size(window_func_->getKind());
   output_ = static_cast<int8_t*>(row_set_mem_owner_->allocate(output_buf_sz,
                                                               /*thread_idx=*/0));
   bool const is_agg_func = window_function_is_aggregate(window_func_->getKind());
   bool const need_window_partition_buf =
       window_func_->hasFraming() || window_func_->isMissingValueFillingFunction();
   if (is_agg_func || need_window_partition_buf) {
     fillPartitionStart();
     if (window_function_requires_peer_handling(window_func_) ||
         need_window_partition_buf) {
       fillPartitionEnd();
     }
   }
   std::unique_ptr<int64_t[]> scratchpad;
   int64_t* intermediate_output_buffer;
   if (is_agg_func || need_window_partition_buf) {
     intermediate_output_buffer = reinterpret_cast<int64_t*>(output_);
   } else {
     output_buf_sz = sizeof(int64_t) * elem_count_;
     scratchpad.reset(new int64_t[elem_count_]);
     intermediate_output_buffer = scratchpad.get();
   }
   const bool should_parallelize{g_enable_parallel_window_partition_compute &&
                                 elem_count_ >=
                                     g_parallel_window_partition_compute_threshold};
 
   auto cached_sorted_partition_it =
       sorted_partition_cache.find(sorted_partition_cache_key_);
   if (cached_sorted_partition_it != sorted_partition_cache.end()) {
     auto& sorted_partition = cached_sorted_partition_it->second;
     VLOG(1) << "Reuse cached sorted partition to compute window function context (key: "
             << sorted_partition_cache_key_
             << ", ordering condition: " << ::toString(window_func_->getOrderKeys())
             << ")";
     DEBUG_TIMER("Window Function Cached Sorted Partition Copy");
     std::memcpy(intermediate_output_buffer, sorted_partition->data(), output_buf_sz);
     if (need_window_partition_buf) {
       sorted_partition_buf_ = sorted_partition;
     }
   } else {
     // ordering partitions if necessary
     const auto sort_partitions = [&](const size_t start, const size_t end) {
       for (size_t partition_idx = start; partition_idx < end; ++partition_idx) {
         sortPartition(partition_idx,
                       intermediate_output_buffer + offsets()[partition_idx],
                       should_parallelize);
       }
     };
 
     if (should_parallelize) {
       auto sorted_partition_copy_timer =
           DEBUG_TIMER("Window Function Partition Sorting Parallelized");
       tbb::parallel_for(tbb::blocked_range<int64_t>(0, partitionCount()),
                         [&, parent_thread_local_ids = logger::thread_local_ids()](
                             const tbb::blocked_range<int64_t>& r) {
                           logger::LocalIdsScopeGuard lisg =
                               parent_thread_local_ids.setNewThreadId();
                           sort_partitions(r.begin(), r.end());
                         });
     } else {
       auto sorted_partition_copy_timer =
           DEBUG_TIMER("Window Function  Partition Sorting Non-Parallelized");
       sort_partitions(0, partitionCount());
     }
     auto sorted_partition_ref_cnt_it =
         sorted_partition_key_ref_count_map.find(sorted_partition_cache_key_);
     bool can_access_sorted_partition =
         sorted_partition_ref_cnt_it != sorted_partition_key_ref_count_map.end() &&
         sorted_partition_ref_cnt_it->second > 1;
     if (can_access_sorted_partition || need_window_partition_buf) {
       // keep the sorted partition only if it will be reused from other window function
       // context of this query
       sorted_partition_buf_ = std::make_shared<std::vector<int64_t>>(elem_count_);
       DEBUG_TIMER("Window Function Sorted Partition Copy For Caching");
       std::memcpy(
           sorted_partition_buf_->data(), intermediate_output_buffer, output_buf_sz);
       auto it = sorted_partition_cache.emplace(sorted_partition_cache_key_,
                                                sorted_partition_buf_);
       if (it.second) {
         VLOG(1) << "Put sorted partition to cache (key: " << sorted_partition_cache_key_
                 << ", ordering condition: " << ::toString(window_func_->getOrderKeys())
                 << ")";
       }
     }
   }
 
   if (need_window_partition_buf) {
     const auto compute_ordered_partition_null_range = [=](const size_t start,
                                                           const size_t end) {
       for (size_t partition_idx = start; partition_idx < end; ++partition_idx) {
         computeNullRangeOfSortedPartition(
             window_func_->getOrderKeys().front()->get_type_info(),
             partition_idx,
             payload() + offsets()[partition_idx],
             intermediate_output_buffer + offsets()[partition_idx]);
       }
     };
     auto partition_count = partitionCount();
 
     if (should_parallelize) {
       auto partition_compuation_timer =
           DEBUG_TIMER("Window Function Ordered-Partition Null-Range Compute");
       tbb::parallel_for(tbb::blocked_range<int64_t>(0, partitionCount()),
                         [&, parent_thread_local_ids = logger::thread_local_ids()](
                             const tbb::blocked_range<int64_t>& r) {
                           logger::LocalIdsScopeGuard lisg =
                               parent_thread_local_ids.setNewThreadId();
                           compute_ordered_partition_null_range(r.begin(), r.end());
                         });
     } else {
       auto partition_compuation_timer = DEBUG_TIMER(
           "Window Function Non-Parallelized Ordered-Partition Null-Range Compute");
       compute_ordered_partition_null_range(0, partitionCount());
     }
     auto const cache_key = computeAggregateTreeCacheKey();
     auto const c_it = aggregate_tree_map.find(cache_key);
     if (c_it != aggregate_tree_map.cend()) {
       VLOG(1) << "Reuse aggregate tree for window function framing";
       resizeStorageForWindowFraming(true);
       aggregate_trees_ = c_it->second;
       memcpy(aggregate_trees_depth_,
              aggregate_trees_.aggregate_trees_depth_,
              sizeof(size_t) * partition_count);
     } else {
       if (needsToBuildAggregateTree()) {
         const auto build_aggregation_tree_for_partitions = [=](const size_t start,
                                                                const size_t end) {
           for (size_t partition_idx = start; partition_idx < end; ++partition_idx) {
             // build a segment tree for the partition
             // todo (yoonmin) : support generic window function expression
             // i.e., when window_func_expr_columns_.size() > 1
             SQLTypeInfo const input_col_ti =
                 window_func_->getArgs().front()->get_type_info();
             const auto partition_size = counts()[partition_idx];
             buildAggregationTreeForPartition(window_func_->getKind(),
                                              partition_idx,
                                              partition_size,
                                              payload() + offsets()[partition_idx],
                                              intermediate_output_buffer,
                                              input_col_ti);
           }
         };
         resizeStorageForWindowFraming();
         if (should_parallelize) {
           auto partition_compuation_timer = DEBUG_TIMER(
               "Window Function Parallelized Segment Tree Construction for Partitions");
           tbb::parallel_for(tbb::blocked_range<int64_t>(0, partitionCount()),
                             [=, parent_thread_local_ids = logger::thread_local_ids()](
                                 const tbb::blocked_range<int64_t>& r) {
                               logger::LocalIdsScopeGuard lisg =
                                   parent_thread_local_ids.setNewThreadId();
                               build_aggregation_tree_for_partitions(r.begin(), r.end());
                             });
         } else {
           auto partition_compuation_timer = DEBUG_TIMER(
               "Window Function Non-Parallelized Segment Tree Construction for "
               "Partitions");
           build_aggregation_tree_for_partitions(0, partition_count);
         }
       }
       CHECK(aggregate_tree_map.emplace(cache_key, aggregate_trees_).second);
       VLOG(2) << "Put aggregate tree for the window framing";
     }
   }
 
   const auto compute_partitions = [=](const size_t start, const size_t end) {
     for (size_t partition_idx = start; partition_idx < end; ++partition_idx) {
       computePartitionBuffer(partition_idx,
                              intermediate_output_buffer + offsets()[partition_idx],
                              window_func_);
     }
   };
 
   if (should_parallelize) {
     auto partition_compuation_timer = DEBUG_TIMER("Window Function Partition Compute");
     tbb::parallel_for(tbb::blocked_range<int64_t>(0, partitionCount()),
                       [&, parent_thread_local_ids = logger::thread_local_ids()](
                           const tbb::blocked_range<int64_t>& r) {
                         logger::LocalIdsScopeGuard lisg =
                             parent_thread_local_ids.setNewThreadId();
                         compute_partitions(r.begin(), r.end());
                       });
   } else {
     auto partition_compuation_timer =
         DEBUG_TIMER("Window Function Non-Parallelized Partition Compute");
     compute_partitions(0, partitionCount());
   }
 
   if (is_agg_func || need_window_partition_buf) {
     // If window function is aggregate we were able to write to the final output buffer
     // directly in computePartition and we are done.
     return;
   }
 
   auto output_i64 = reinterpret_cast<int64_t*>(output_);
   const auto payload_copy = [=](const size_t start, const size_t end) {
     for (size_t i = start; i < end; ++i) {
       output_i64[payload()[i]] = intermediate_output_buffer[i];
     }
   };
   if (should_parallelize) {
     auto payload_copy_timer =
         DEBUG_TIMER("Window Function Non-Aggregate Payload Copy Parallelized");
     tbb::parallel_for(tbb::blocked_range<int64_t>(0, elem_count_),
                       [&, parent_thread_local_ids = logger::thread_local_ids()](
                           const tbb::blocked_range<int64_t>& r) {
                         logger::LocalIdsScopeGuard lisg =
                             parent_thread_local_ids.setNewThreadId();
                         payload_copy(r.begin(), r.end());
                       });
   } else {
     auto payload_copy_timer =
         DEBUG_TIMER("Window Function Non-Aggregate Payload Copy Non-Parallelized");
     payload_copy(0, elem_count_);
   }
 }

Here is the call graph for this function:

QueryPlanHash const WindowFunctionContext::computeAggregateTreeCacheKey ( ) const

private

Definition at line 1814 of file WindowContext.cpp.

References Analyzer::WindowFunction::getArgs(), Analyzer::WindowFunction::getCollation(), Analyzer::WindowFunction::getKind(), Analyzer::WindowFunction::getOrderKeys(), Analyzer::WindowFunction::getPartitionKeys(), hash_value(), toString(), and window_func_.

Referenced by compute().

                                                                               {
   // aggregate tree is constructed per window aggregate function kind, input expression,
   // partition key(s) and ordering key
   //  this means when two window definitions have the same condition listed above but
   //  differ in frame bound declaration,
   // they can share the same aggregate tree
   auto cache_key = boost::hash_value(::toString(window_func_->getKind()));
   boost::hash_combine(cache_key, ::toString(window_func_->getArgs()));
   boost::hash_combine(cache_key, ::toString(window_func_->getPartitionKeys()));
   boost::hash_combine(cache_key, ::toString(window_func_->getOrderKeys()));
   for (auto& order_entry : window_func_->getCollation()) {
     boost::hash_combine(cache_key, order_entry.toString());
   }
   return cache_key;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void WindowFunctionContext::computeNullRangeOfSortedPartition	(	const SQLTypeInfo &	order_col_ti,
		size_t	partition_idx,
		const int32_t *	original_col_idx_buf,
		const int64_t *	ordered_col_idx_buf
	)

private

Definition at line 849 of file WindowContext.cpp.

References counts(), logger::FATAL, SQLTypeInfo::get_size(), SQLTypeInfo::get_type(), SQLTypeInfo::is_boolean(), SQLTypeInfo::is_decimal(), SQLTypeInfo::is_fp(), SQLTypeInfo::is_integer(), SQLTypeInfo::is_time_or_date(), kDOUBLE, kFLOAT, LOG, null_val_bit_pattern(), order_columns_, ordered_partition_null_end_pos_, and ordered_partition_null_start_pos_.

Referenced by compute().

                                         {
   IndexPair null_range;
   const auto partition_size = counts()[partition_idx];
   if (partition_size > 0) {
     if (order_col_ti.is_integer() || order_col_ti.is_decimal() ||
         order_col_ti.is_time_or_date() || order_col_ti.is_boolean()) {
       FindNullRange const null_range_info{
           original_col_idx_buf, ordered_col_idx_buf, partition_size};
       switch (order_col_ti.get_size()) {
         case 8:
           null_range =
               null_range_info.find_null_range_int<int64_t>(order_columns_.front());
           break;
         case 4:
           null_range =
               null_range_info.find_null_range_int<int32_t>(order_columns_.front());
           break;
         case 2:
           null_range =
               null_range_info.find_null_range_int<int16_t>(order_columns_.front());
           break;
         case 1:
           null_range =
               null_range_info.find_null_range_int<int8_t>(order_columns_.front());
           break;
         default:
           LOG(FATAL) << "Invalid type size: " << order_col_ti.get_size();
       }
     } else if (order_col_ti.is_fp()) {
       const auto null_bit_pattern =
           null_val_bit_pattern(order_col_ti, order_col_ti.get_type() == kFLOAT);
       FindNullRange const null_range_info{
           original_col_idx_buf, ordered_col_idx_buf, partition_size, null_bit_pattern};
       switch (order_col_ti.get_type()) {
         case kFLOAT:
           null_range = null_range_info.find_null_range_fp<float>(order_columns_.front());
           break;
         case kDOUBLE:
           null_range = null_range_info.find_null_range_fp<double>(order_columns_.front());
           break;
         default:
           LOG(FATAL) << "Invalid float type";
       }
     } else {
       LOG(FATAL) << "Invalid column type for window aggregation over the frame";
     }
   }
   ordered_partition_null_start_pos_[partition_idx] = null_range.first;
   ordered_partition_null_end_pos_[partition_idx] = null_range.second + 1;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void WindowFunctionContext::computePartitionBuffer	(	const size_t	partition_idx,
		int64_t *	output_for_partition_buff,
		const Analyzer::WindowFunction *	window_func
	)

private

Definition at line 1312 of file WindowContext.cpp.

Referenced by compute().

                                                {
   const size_t partition_size{static_cast<size_t>(counts()[partition_idx])};
   if (partition_size == 0) {
     return;
   }
   const auto offset = offsets()[partition_idx];
   auto partition_comparator = createComparator(partition_idx);
   const auto col_tuple_comparator = [&partition_comparator](const int64_t lhs,
                                                             const int64_t rhs) {
     for (const auto& comparator : partition_comparator) {
       const auto comparator_result = comparator(lhs, rhs);
       switch (comparator_result) {
         case WindowFunctionContext::WindowComparatorResult::LT:
           return true;
         case WindowFunctionContext::WindowComparatorResult::GT:
           return false;
         default:
           // WindowComparatorResult::EQ: continue to next comparator
           continue;
       }
     }
     // If here WindowFunctionContext::WindowComparatorResult::KEQ for all keys
     // return false as sort algo must enforce weak ordering
     return false;
   };
   switch (window_func->getKind()) {
     case SqlWindowFunctionKind::ROW_NUMBER: {
       const auto row_numbers =
           index_to_row_number(output_for_partition_buff, partition_size);
       std::copy(row_numbers.begin(), row_numbers.end(), output_for_partition_buff);
       break;
     }
     case SqlWindowFunctionKind::RANK: {
       const auto rank =
           index_to_rank(output_for_partition_buff, partition_size, col_tuple_comparator);
       std::copy(rank.begin(), rank.end(), output_for_partition_buff);
       break;
     }
     case SqlWindowFunctionKind::DENSE_RANK: {
       const auto dense_rank = index_to_dense_rank(
           output_for_partition_buff, partition_size, col_tuple_comparator);
       std::copy(dense_rank.begin(), dense_rank.end(), output_for_partition_buff);
       break;
     }
     case SqlWindowFunctionKind::PERCENT_RANK: {
       const auto percent_rank = index_to_percent_rank(
           output_for_partition_buff, partition_size, col_tuple_comparator);
       std::copy(percent_rank.begin(),
                 percent_rank.end(),
                 reinterpret_cast<double*>(may_alias_ptr(output_for_partition_buff)));
       break;
     }
     case SqlWindowFunctionKind::CUME_DIST: {
       const auto cume_dist = index_to_cume_dist(
           output_for_partition_buff, partition_size, col_tuple_comparator);
       std::copy(cume_dist.begin(),
                 cume_dist.end(),
                 reinterpret_cast<double*>(may_alias_ptr(output_for_partition_buff)));
       break;
     }
     case SqlWindowFunctionKind::NTILE: {
       const auto& args = window_func->getArgs();
       CHECK_EQ(args.size(), size_t(1));
       const auto n = get_int_constant_from_expr(args.front().get());
       const auto ntile = index_to_ntile(output_for_partition_buff, partition_size, n);
       std::copy(ntile.begin(), ntile.end(), output_for_partition_buff);
       break;
     }
     case SqlWindowFunctionKind::LAG:
     case SqlWindowFunctionKind::LEAD: {
       const auto lag_or_lead = get_lag_or_lead_argument(window_func);
       const auto partition_row_offsets = payload() + offset;
       apply_lag_to_partition(
           lag_or_lead, partition_row_offsets, output_for_partition_buff, partition_size);
       break;
     }
     case SqlWindowFunctionKind::FIRST_VALUE:
     case SqlWindowFunctionKind::LAST_VALUE: {
       const auto target_idx =
           get_target_idx_for_first_or_last_value_func(window_func, partition_size);
       const auto partition_row_offsets = payload() + offset;
       apply_nth_value_to_partition(
           partition_row_offsets, output_for_partition_buff, partition_size, target_idx);
       break;
     }
     case SqlWindowFunctionKind::NTH_VALUE: {
       auto const n_value_ptr =
           dynamic_cast<Analyzer::Constant*>(window_func_->getArgs()[1].get());
       CHECK(n_value_ptr);
       auto const n_value = static_cast<size_t>(n_value_ptr->get_constval().intval);
       const auto partition_row_offsets = payload() + offset;
       if (n_value < partition_size) {
         apply_nth_value_to_partition(
             partition_row_offsets, output_for_partition_buff, partition_size, n_value);
       } else {
         // when NTH_VALUE of the current row is NULL, we keep the NULL value in the
         // current row's output storage in the query output buffer, so we assign the
         // original index of the current row to the corresponding slot in
         // `output_for_partition_buff`
         apply_original_index_to_partition(
             partition_row_offsets, output_for_partition_buff, partition_size);
       }
       break;
     }
     case SqlWindowFunctionKind::AVG:
     case SqlWindowFunctionKind::MIN:
     case SqlWindowFunctionKind::MAX:
     case SqlWindowFunctionKind::SUM:
     case SqlWindowFunctionKind::COUNT:
     case SqlWindowFunctionKind::CONDITIONAL_CHANGE_EVENT:
     case SqlWindowFunctionKind::LEAD_IN_FRAME:
     case SqlWindowFunctionKind::LAG_IN_FRAME:
     case SqlWindowFunctionKind::NTH_VALUE_IN_FRAME:
     case SqlWindowFunctionKind::FIRST_VALUE_IN_FRAME:
     case SqlWindowFunctionKind::LAST_VALUE_IN_FRAME:
     case SqlWindowFunctionKind::COUNT_IF:
     case SqlWindowFunctionKind::SUM_IF:
     case SqlWindowFunctionKind::FORWARD_FILL:
     case SqlWindowFunctionKind::BACKWARD_FILL: {
       const auto partition_row_offsets = payload() + offset;
       if (window_function_requires_peer_handling(window_func)) {
         index_to_partition_end(partitionEnd(),
                                offset,
                                output_for_partition_buff,
                                partition_size,
                                col_tuple_comparator);
       }
       apply_permutation_to_partition(
           output_for_partition_buff, partition_row_offsets, partition_size);
       break;
     }
     default: {
       std::ostringstream oss;
       oss << "Window function not supported yet: " << window_func_->getKind();
       throw std::runtime_error(oss.str());
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

const int32_t * WindowFunctionContext::counts ( ) const

Definition at line 1792 of file WindowContext.cpp.

References device_type_, dummy_count_, and partitions_.

Referenced by CodeGenerator::codegenFixedLengthColVarInWindow(), Executor::codegenLoadPartitionBuffers(), compute(), computeNullRangeOfSortedPartition(), computePartitionBuffer(), fillPartitionEnd(), fillPartitionStart(), partitionCount(), sortPartition(), and WindowFunctionContext().

                                                    {
   if (partitions_) {
     return reinterpret_cast<const int32_t*>(
         partitions_->getJoinHashBuffer(device_type_, 0) + partitions_->countBufferOff());
   }
   return &dummy_count_;
 }

Here is the caller graph for this function:

std::vector< WindowFunctionContext::Comparator > WindowFunctionContext::createComparator ( size_t partition_idx )

Definition at line 904 of file WindowContext.cpp.

References CHECK, CHECK_EQ, Analyzer::WindowFunction::getCollation(), Analyzer::WindowFunction::getOrderKeys(), makeComparator(), offsets(), order_columns_, payload(), and window_func_.

Referenced by computePartitionBuffer(), and sortPartition().

                           {
   // create tuple comparator
   std::vector<WindowFunctionContext::Comparator> partition_comparator;
   const auto& order_keys = window_func_->getOrderKeys();
   const auto& collation = window_func_->getCollation();
   CHECK_EQ(order_keys.size(), collation.size());
   for (size_t order_column_idx = 0; order_column_idx < order_columns_.size();
        ++order_column_idx) {
     auto order_column_buffer = order_columns_[order_column_idx];
     const auto order_col =
         dynamic_cast<const Analyzer::ColumnVar*>(order_keys[order_column_idx].get());
     CHECK(order_col);
     const auto& order_col_collation = collation[order_column_idx];
     auto comparator = makeComparator(order_col,
                                      order_column_buffer,
                                      payload() + offsets()[partition_idx],
                                      !order_col_collation.is_desc,
                                      order_col_collation.nulls_first);
     if (order_col_collation.is_desc) {
       comparator = [comparator](const int64_t lhs, const int64_t rhs) {
         return comparator(rhs, lhs);
       };
     }
     partition_comparator.push_back(comparator);
   }
   return partition_comparator;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

size_t WindowFunctionContext::elementCount ( ) const

Definition at line 1026 of file WindowContext.cpp.

References elem_count_.

Referenced by Executor::codegenCurrentPartitionIndex().

                                                  {
   return elem_count_;
 }

Here is the caller graph for this function:

void WindowFunctionContext::fillPartitionEnd ( )

private

Definition at line 1730 of file WindowContext.cpp.

References agg_count_distinct_bitmap(), Bitmap, checked_calloc(), counts(), CPU, elem_count_, gpu_enabled::partial_sum(), partition_end_, partition_start_offset_, partitionCount(), and partitions_.

Referenced by compute().

                                              {
   CountDistinctDescriptor partition_start_bitmap{CountDistinctImplType::Bitmap,
                                                  0,
                                                  0,
                                                  static_cast<int64_t>(elem_count_),
                                                  false,
                                                  ExecutorDeviceType::CPU,
                                                  1};
   auto bitmap_sz = partition_start_bitmap.bitmapPaddedSizeBytes();
   if (partitions_) {
     bitmap_sz += partitions_->isBitwiseEq() ? 1 : 0;
   }
   partition_end_ = static_cast<int8_t*>(checked_calloc(bitmap_sz, 1));
   auto partition_end_handle = reinterpret_cast<int64_t>(partition_end_);
   int64_t partition_count = partitionCount();
   if (partition_start_offset_) {
     // if we have `partition_start_offset_`, we can reuse it for this logic
     // but note that it has partition_count + 1 elements where the first element is zero
     // which means the first partition's start offset is zero
     // and rest of them can represent values required for this logic
     for (int64_t i = 0; i < partition_count - 1; ++i) {
       if (partition_start_offset_[i + 1] == 0) {
         continue;
       }
       agg_count_distinct_bitmap(
           &partition_end_handle, partition_start_offset_[i + 1] - 1, 0, 0);
     }
     if (elem_count_) {
       agg_count_distinct_bitmap(&partition_end_handle, elem_count_ - 1, 0, 0);
     }
   } else {
     std::vector<size_t> partition_offsets(partition_count);
     std::partial_sum(counts(), counts() + partition_count, partition_offsets.begin());
     for (int64_t i = 0; i < partition_count - 1; ++i) {
       if (partition_offsets[i] == 0) {
         continue;
       }
       agg_count_distinct_bitmap(&partition_end_handle, partition_offsets[i] - 1, 0, 0);
     }
     if (elem_count_) {
       agg_count_distinct_bitmap(&partition_end_handle, elem_count_ - 1, 0, 0);
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void WindowFunctionContext::fillPartitionStart ( )

private

Definition at line 1696 of file WindowContext.cpp.

References agg_count_distinct_bitmap(), Bitmap, checked_calloc(), counts(), CPU, elem_count_, gpu_enabled::partial_sum(), partition_start_, partition_start_offset_, partitionCount(), and partitions_.

Referenced by compute().

                                                {
   CountDistinctDescriptor partition_start_bitmap{CountDistinctImplType::Bitmap,
                                                  0,
                                                  0,
                                                  static_cast<int64_t>(elem_count_),
                                                  false,
                                                  ExecutorDeviceType::CPU,
                                                  1};
   auto bitmap_sz = partition_start_bitmap.bitmapPaddedSizeBytes();
   if (partitions_) {
     bitmap_sz += partitions_->isBitwiseEq() ? 1 : 0;
   }
   partition_start_ = static_cast<int8_t*>(checked_calloc(bitmap_sz, 1));
   int64_t partition_count = partitionCount();
   auto partition_start_handle = reinterpret_cast<int64_t>(partition_start_);
   agg_count_distinct_bitmap(&partition_start_handle, 0, 0, 0);
   if (partition_start_offset_) {
     // if we have `partition_start_offset_`, we can reuse it for this logic
     // but note that it has partition_count + 1 elements where the first element is zero
     // which means the first partition's start offset is zero
     // and rest of them can represent values required for this logic
     for (int64_t i = 0; i < partition_count - 1; ++i) {
       agg_count_distinct_bitmap(
           &partition_start_handle, partition_start_offset_[i + 1], 0, 0);
     }
   } else {
     std::vector<size_t> partition_offsets(partition_count);
     std::partial_sum(counts(), counts() + partition_count, partition_offsets.begin());
     for (int64_t i = 0; i < partition_count - 1; ++i) {
       agg_count_distinct_bitmap(&partition_start_handle, partition_offsets[i], 0, 0);
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

size_t * WindowFunctionContext::getAggregateTreeDepth ( ) const

Definition at line 1680 of file WindowContext.cpp.

References aggregate_trees_depth_.

                                                            {
   return aggregate_trees_depth_;
 }

size_t WindowFunctionContext::getAggregateTreeFanout ( ) const

Definition at line 1684 of file WindowContext.cpp.

References aggregate_trees_fan_out_.

                                                            {
   return aggregate_trees_fan_out_;
 }

double ** WindowFunctionContext::getAggregationTreesForDoubleTypeWindowExpr ( ) const

Definition at line 1664 of file WindowContext.cpp.

References AggregateTreeForWindowFraming::aggregate_tree_for_double_type_, and aggregate_trees_.

                                                                                  {
   return const_cast<double**>(aggregate_trees_.aggregate_tree_for_double_type_.data());
 }

int64_t ** WindowFunctionContext::getAggregationTreesForIntegerTypeWindowExpr ( ) const

Definition at line 1660 of file WindowContext.cpp.

References AggregateTreeForWindowFraming::aggregate_tree_for_integer_type_, and aggregate_trees_.

                                                                                    {
   return const_cast<int64_t**>(aggregate_trees_.aggregate_tree_for_integer_type_.data());
 }

const std::vector< const int8_t * > & WindowFunctionContext::getColumnBufferForWindowFunctionExpressions ( ) const

Definition at line 174 of file WindowContext.cpp.

References window_func_expr_columns_.

                                                                          {
   return window_func_expr_columns_;
 }

SumAndCountPair< double > ** WindowFunctionContext::getDerivedAggregationTreesForDoubleTypeWindowExpr ( ) const

Definition at line 1675 of file WindowContext.cpp.

References aggregate_trees_, and AggregateTreeForWindowFraming::derived_aggregate_tree_for_double_type_.

                                                                                {
   return const_cast<SumAndCountPair<double>**>(
       aggregate_trees_.derived_aggregate_tree_for_double_type_.data());
 }

SumAndCountPair< int64_t > ** WindowFunctionContext::getDerivedAggregationTreesForIntegerTypeWindowExpr ( ) const

Definition at line 1669 of file WindowContext.cpp.

References aggregate_trees_, and AggregateTreeForWindowFraming::derived_aggregate_tree_for_integer_type_.

                                                                                 {
   return const_cast<SumAndCountPair<int64_t>**>(
       aggregate_trees_.derived_aggregate_tree_for_integer_type_.data());
 }

int64_t * WindowFunctionContext::getNullValueEndPos ( ) const

Definition at line 1692 of file WindowContext.cpp.

References ordered_partition_null_end_pos_.

Referenced by Executor::codegenFrameNullRange().

                                                          {
   return ordered_partition_null_end_pos_;
 }

Here is the caller graph for this function:

int64_t * WindowFunctionContext::getNullValueStartPos ( ) const

Definition at line 1688 of file WindowContext.cpp.

References ordered_partition_null_start_pos_.

Referenced by Executor::codegenFrameNullRange().

                                                            {
   return ordered_partition_null_start_pos_;
 }

Here is the caller graph for this function:

const std::vector< const int8_t * > & WindowFunctionContext::getOrderKeyColumnBuffers ( ) const

Definition at line 178 of file WindowContext.cpp.

References order_columns_.

Referenced by Executor::codegenLoadOrderKeyBufPtr(), and Executor::codegenWindowFrameBounds().

           {
   return order_columns_;
 }

Here is the caller graph for this function:

const std::vector< SQLTypeInfo > & WindowFunctionContext::getOrderKeyColumnBufferTypes ( ) const

Definition at line 183 of file WindowContext.cpp.

References order_columns_ti_.

Referenced by CodeGenerator::codegenFixedLengthColVar(), and Executor::codegenLoadOrderKeyBufPtr().

           {
   return order_columns_ti_;
 }

Here is the caller graph for this function:

const Analyzer::WindowFunction * WindowFunctionContext::getWindowFunction ( ) const

Definition at line 980 of file WindowContext.cpp.

References window_func_.

Referenced by Executor::codegenCurrentPartitionIndex(), CodeGenerator::codegenFixedLengthColVar(), CodeGenerator::codegenFixedLengthColVarInWindow(), Executor::codegenLoadCurrentValueFromColBuf(), Executor::codegenLoadOrderKeyBufPtr(), Executor::codegenWindowFrameBounds(), Executor::codegenWindowFunction(), Executor::getFirstOrderColTypeInfo(), and Executor::getOrderKeyTypeName().

                                                                              {
   return window_func_;
 }

Here is the caller graph for this function:

WindowFunctionContext::Comparator WindowFunctionContext::makeComparator	(	const Analyzer::ColumnVar *	col_var,
		const int8_t *	partition_values,
		const int32_t *	partition_indices,
		const bool	asc_ordering,
		const bool	nulls_first
	)

staticprivate

Definition at line 1174 of file WindowContext.cpp.

References logger::FATAL, Analyzer::Expr::get_type_info(), kDOUBLE, kFLOAT, and LOG.

Referenced by createComparator().

                             {
   const auto& ti = col_var->get_type_info();
   if (ti.is_integer() || ti.is_decimal() || ti.is_time() || ti.is_boolean()) {
     switch (ti.get_size()) {
       case 8: {
         return [order_column_buffer, nulls_first, partition_indices, asc_ordering, &ti](
                    const int64_t lhs, const int64_t rhs) {
           return asc_ordering ? integer_comparator_asc<int64_t>(order_column_buffer,
                                                                 ti,
                                                                 partition_indices,
                                                                 lhs,
                                                                 rhs,
                                                                 asc_ordering,
                                                                 nulls_first)
                               : integer_comparator_desc<int64_t>(order_column_buffer,
                                                                  ti,
                                                                  partition_indices,
                                                                  lhs,
                                                                  rhs,
                                                                  asc_ordering,
                                                                  nulls_first);
         };
       }
       case 4: {
         return [order_column_buffer, nulls_first, partition_indices, asc_ordering, &ti](
                    const int64_t lhs, const int64_t rhs) {
           return asc_ordering ? integer_comparator_asc<int32_t>(order_column_buffer,
                                                                 ti,
                                                                 partition_indices,
                                                                 lhs,
                                                                 rhs,
                                                                 asc_ordering,
                                                                 nulls_first)
                               : integer_comparator_desc<int32_t>(order_column_buffer,
                                                                  ti,
                                                                  partition_indices,
                                                                  lhs,
                                                                  rhs,
                                                                  asc_ordering,
                                                                  nulls_first);
         };
       }
       case 2: {
         return [order_column_buffer, nulls_first, partition_indices, asc_ordering, &ti](
                    const int64_t lhs, const int64_t rhs) {
           return asc_ordering ? integer_comparator_asc<int16_t>(order_column_buffer,
                                                                 ti,
                                                                 partition_indices,
                                                                 lhs,
                                                                 rhs,
                                                                 asc_ordering,
                                                                 nulls_first)
                               : integer_comparator_desc<int16_t>(order_column_buffer,
                                                                  ti,
                                                                  partition_indices,
                                                                  lhs,
                                                                  rhs,
                                                                  asc_ordering,
                                                                  nulls_first);
         };
       }
       case 1: {
         return [order_column_buffer, nulls_first, partition_indices, asc_ordering, &ti](
                    const int64_t lhs, const int64_t rhs) {
           return asc_ordering ? integer_comparator_asc<int8_t>(order_column_buffer,
                                                                ti,
                                                                partition_indices,
                                                                lhs,
                                                                rhs,
                                                                asc_ordering,
                                                                nulls_first)
                               : integer_comparator_desc<int8_t>(order_column_buffer,
                                                                 ti,
                                                                 partition_indices,
                                                                 lhs,
                                                                 rhs,
                                                                 asc_ordering,
                                                                 nulls_first);
         };
       }
       default: {
         LOG(FATAL) << "Invalid type size: " << ti.get_size();
       }
     }
   }
   if (ti.is_fp()) {
     switch (ti.get_type()) {
       case kFLOAT: {
         return [order_column_buffer, nulls_first, partition_indices, asc_ordering, &ti](
                    const int64_t lhs, const int64_t rhs) {
           return asc_ordering ? fp_comparator_asc<float, int32_t>(order_column_buffer,
                                                                   ti,
                                                                   partition_indices,
                                                                   lhs,
                                                                   rhs,
                                                                   asc_ordering,
                                                                   nulls_first)
                               : fp_comparator_desc<float, int32_t>(order_column_buffer,
                                                                    ti,
                                                                    partition_indices,
                                                                    lhs,
                                                                    rhs,
                                                                    asc_ordering,
                                                                    nulls_first);
         };
       }
       case kDOUBLE: {
         return [order_column_buffer, nulls_first, partition_indices, asc_ordering, &ti](
                    const int64_t lhs, const int64_t rhs) {
           return asc_ordering ? fp_comparator_asc<double, int64_t>(order_column_buffer,
                                                                    ti,
                                                                    partition_indices,
                                                                    lhs,
                                                                    rhs,
                                                                    asc_ordering,
                                                                    nulls_first)
                               : fp_comparator_desc<double, int64_t>(order_column_buffer,
                                                                     ti,
                                                                     partition_indices,
                                                                     lhs,
                                                                     rhs,
                                                                     asc_ordering,
                                                                     nulls_first);
         };
       }
       default: {
         LOG(FATAL) << "Invalid float type";
       }
     }
   }
   throw std::runtime_error("Type not supported yet");
 }

Here is the call graph for this function:

Here is the caller graph for this function:

const bool WindowFunctionContext::needsToBuildAggregateTree ( ) const

Definition at line 1809 of file WindowContext.cpp.

References elem_count_, Analyzer::WindowFunction::hasAggregateTreeRequiredWindowFunc(), Analyzer::WindowFunction::hasFraming(), and window_func_.

Referenced by compute().

                                                                   {
   return window_func_->hasFraming() &&
          window_func_->hasAggregateTreeRequiredWindowFunc() && elem_count_ > 0;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

const int32_t * WindowFunctionContext::offsets ( ) const

Definition at line 1784 of file WindowContext.cpp.

References device_type_, dummy_offset_, and partitions_.

Referenced by buildAggregationTreeForPartition(), compute(), computePartitionBuffer(), createComparator(), and partitionCount().

                                                     {
   if (partitions_) {
     return reinterpret_cast<const int32_t*>(
         partitions_->getJoinHashBuffer(device_type_, 0) + partitions_->offsetBufferOff());
   }
   return &dummy_offset_;
 }

Here is the caller graph for this function:

WindowFunctionContext& WindowFunctionContext::operator= ( const WindowFunctionContext & )

delete

const int8_t * WindowFunctionContext::output ( ) const

Definition at line 984 of file WindowContext.cpp.

References output_.

Referenced by CodeGenerator::codegenWindowPosition().

                                                   {
   return output_;
 }

Here is the caller graph for this function:

size_t WindowFunctionContext::partitionCount ( ) const

Definition at line 1800 of file WindowContext.cpp.

References CHECK_GE, counts(), offsets(), and partitions_.

Referenced by Executor::codegenCurrentPartitionIndex(), compute(), fillPartitionEnd(), fillPartitionStart(), resizeStorageForWindowFraming(), and WindowFunctionContext().

                                                    {
   if (partitions_) {
     const auto partition_count = counts() - offsets();
     CHECK_GE(partition_count, 0);
     return partition_count;
   }
   return 1;  // non-partitioned window function
 }

Here is the call graph for this function:

Here is the caller graph for this function:

const int8_t * WindowFunctionContext::partitionEnd ( ) const

Definition at line 1022 of file WindowContext.cpp.

References partition_end_.

Referenced by computePartitionBuffer().

                                                         {
   return partition_end_;
 }

Here is the caller graph for this function:

const int64_t * WindowFunctionContext::partitionNumCountBuf ( ) const

Definition at line 1008 of file WindowContext.cpp.

References CHECK, and partition_start_offset_.

Referenced by Executor::codegenCurrentPartitionIndex().

                                                                  {
   CHECK(partition_start_offset_);
   return partition_start_offset_ + 1;
 }

Here is the caller graph for this function:

const int8_t * WindowFunctionContext::partitionStart ( ) const

Definition at line 1018 of file WindowContext.cpp.

References partition_start_.

                                                           {
   return partition_start_;
 }

const int64_t * WindowFunctionContext::partitionStartOffset ( ) const

Definition at line 1003 of file WindowContext.cpp.

References CHECK, and partition_start_offset_.

Referenced by Executor::codegenLoadPartitionBuffers().

                                                                  {
   CHECK(partition_start_offset_);
   return partition_start_offset_;
 }

Here is the caller graph for this function:

const int32_t * WindowFunctionContext::payload ( ) const

Definition at line 1775 of file WindowContext.cpp.

References device_type_, dummy_payload_, and partitions_.

Referenced by Executor::codegenCurrentPartitionIndex(), Executor::codegenLoadPartitionBuffers(), compute(), computePartitionBuffer(), and createComparator().

                                                     {
   if (partitions_) {
     return reinterpret_cast<const int32_t*>(
         partitions_->getJoinHashBuffer(device_type_, 0) +
         partitions_->payloadBufferOff());
   }
   return dummy_payload_;  // non-partitioned window function
 }

Here is the caller graph for this function:

void WindowFunctionContext::resizeStorageForWindowFraming ( bool const for_reuse = false )

private

Definition at line 1454 of file WindowContext.cpp.

References aggregate_trees_, partitionCount(), AggregateTreeForWindowFraming::resizeStorageForWindowFraming(), and segment_trees_owned_.

Referenced by compute().

                                                                         {
   auto const partition_count = partitionCount();
   aggregate_trees_.resizeStorageForWindowFraming(partition_count);
   if (!for_reuse) {
     segment_trees_owned_.resize(partition_count);
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void WindowFunctionContext::setSortedPartitionCacheKey ( QueryPlanHash cache_key )

Definition at line 188 of file WindowContext.cpp.

References sorted_partition_cache_key_.

                                                                               {
   sorted_partition_cache_key_ = cache_key;
 }

const int64_t * WindowFunctionContext::sortedPartition ( ) const

Definition at line 988 of file WindowContext.cpp.

References CHECK, and sorted_partition_buf_.

Referenced by Executor::codegenLoadPartitionBuffers().

                                                             {
   CHECK(sorted_partition_buf_);
   return sorted_partition_buf_->data();
 }

Here is the caller graph for this function:

void WindowFunctionContext::sortPartition	(	const size_t	partition_idx,
		int64_t *	output_for_partition_buff,
		bool	should_parallelize
	)

private

Definition at line 933 of file WindowContext.cpp.

References counts(), createComparator(), GT, gpu_enabled::iota(), LT, and gpu_enabled::sort().

Referenced by compute().

                                                                    {
   const size_t partition_size{static_cast<size_t>(counts()[partition_idx])};
   if (partition_size == 0) {
     return;
   }
   std::iota(
       output_for_partition_buff, output_for_partition_buff + partition_size, int64_t(0));
   auto partition_comparator = createComparator(partition_idx);
   if (!partition_comparator.empty()) {
     const auto col_tuple_comparator = [&partition_comparator](const int64_t lhs,
                                                               const int64_t rhs) {
       for (const auto& comparator : partition_comparator) {
         const auto comparator_result = comparator(lhs, rhs);
         switch (comparator_result) {
           case WindowFunctionContext::WindowComparatorResult::LT:
             return true;
           case WindowFunctionContext::WindowComparatorResult::GT:
             return false;
           default:
             // WindowComparatorResult::EQ: continue to next comparator
             continue;
         }
       }
       // If here WindowFunctionContext::WindowComparatorResult::KEQ for all keys
       // return false as sort algo must enforce weak ordering
       return false;
     };
     if (should_parallelize) {
 #ifdef HAVE_TBB
       tbb::parallel_sort(output_for_partition_buff,
                          output_for_partition_buff + partition_size,
                          col_tuple_comparator);
 #else
       thrust::sort(output_for_partition_buff,
                    output_for_partition_buff + partition_size,
                    col_tuple_comparator);
 #endif
     } else {
       std::sort(output_for_partition_buff,
                 output_for_partition_buff + partition_size,
                 col_tuple_comparator);
     }
   }
 }