OmniSciDB  cde582ebc3
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
WindowContext.h
Go to the documentation of this file.
1 /*
2  * Copyright 2022 HEAVY.AI, Inc.
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  * http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16 
17 #pragma once
18 
19 #include "Analyzer/Analyzer.h"
20 #include "DataMgr/Chunk/Chunk.h"
23 #include "Shared/sqltypes.h"
24 
25 #include <functional>
26 #include <unordered_map>
27 
28 // Returns true for value window functions, false otherwise.
30  switch (kind) {
35  return true;
36  }
37  default: {
38  return false;
39  }
40  }
41 }
42 
43 // Returns true for aggregate window functions, false otherwise.
45  switch (kind) {
52  return true;
53  }
54  default: {
55  return false;
56  }
57  }
58 }
59 
60 class Executor;
61 
63  std::vector<int64_t*> aggregate_tree_for_integer_type_;
64  std::vector<double*> aggregate_tree_for_double_type_;
65  std::vector<SumAndCountPair<int64_t>*> derived_aggregate_tree_for_integer_type_;
66  std::vector<SumAndCountPair<double>*> derived_aggregate_tree_for_double_type_;
67 };
68 
69 // Per-window function context which encapsulates the logic for computing the various
70 // window function kinds and keeps ownership of buffers which contain the results. For
71 // rank functions, the code generated for the projection simply reads the values and
72 // writes them to the result set. For value and aggregate functions, only the iteration
73 // order is written to the buffer, the rest is handled by generating code in a similar way
74 // we do for non-window queries.
76  public:
77  // non-partitioned version
79  const size_t elem_count,
80  const ExecutorDeviceType device_type,
81  std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner);
82 
83  // partitioned version
85  const Analyzer::WindowFunction* window_func,
86  QueryPlanHash cache_key,
87  const std::shared_ptr<HashJoin>& partitions,
88  const size_t elem_count,
89  const ExecutorDeviceType device_type,
90  std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
91  size_t aggregation_tree_fan_out = g_window_function_aggregation_tree_fanout);
92 
94 
96 
98 
99  // Adds the order column buffer to the context and keeps ownership of it.
100  void addOrderColumn(const int8_t* column,
101  const SQLTypeInfo& ti,
102  const std::vector<std::shared_ptr<Chunk_NS::Chunk>>& chunks_owner);
103 
105 
107  const int8_t* column,
108  const std::vector<std::shared_ptr<Chunk_NS::Chunk>>& chunks_owner);
109 
110  enum class WindowComparatorResult { LT, EQ, GT };
111  using Comparator =
112  std::function<WindowFunctionContext::WindowComparatorResult(const int64_t lhs,
113  const int64_t rhs)>;
114 
115  std::vector<Comparator> createComparator(size_t partition_idx);
116 
117  // Computes the window function result to be used during the actual projection query.
118  void compute(
119  std::unordered_map<QueryPlanHash, size_t>& sorted_partition_key_ref_count_map,
120  std::unordered_map<QueryPlanHash, std::shared_ptr<std::vector<int64_t>>>&
121  sorted_partition_cache);
122 
123  // Returns a pointer to the window function associated with this context.
125 
126  // Returns a pointer to the output buffer of the window function result.
127  const int8_t* output() const;
128 
129  // Returns a pointer to the sorted row index buffer
130  const int64_t* sortedPartition() const;
131 
132  // Returns a pointer to the value field of the aggregation state.
133  const int64_t* aggregateState() const;
134 
135  // Returns a pointer to the count field of the aggregation state.
136  const int64_t* aggregateStateCount() const;
137 
138  // Returns a handle to the pending outputs for the aggregate window function.
139  int64_t aggregateStatePendingOutputs() const;
140 
141  const int64_t* partitionStartOffset() const;
142 
143  const int64_t* partitionNumCountBuf() const;
144 
145  const std::vector<const int8_t*>& getOrderKeyColumnBuffers() const;
146 
147  const std::vector<SQLTypeInfo>& getOrderKeyColumnBufferTypes() const;
148 
150 
152 
154 
156 
157  size_t* getAggregateTreeDepth() const;
158 
159  size_t getAggregateTreeFanout() const;
160 
161  int64_t* getNullValueStartPos() const;
162 
163  int64_t* getNullValueEndPos() const;
164 
165  // Returns a pointer to the partition start bitmap.
166  const int8_t* partitionStart() const;
167 
168  // Returns a pointer to the partition end bitmap.
169  const int8_t* partitionEnd() const;
170 
171  // Returns the element count in the columns used by the window function.
172  size_t elementCount() const;
173 
174  const int32_t* payload() const;
175 
176  const int32_t* offsets() const;
177 
178  const int32_t* counts() const;
179 
180  size_t partitionCount() const;
181 
182  const bool needsToBuildAggregateTree() const;
183 
184  private:
185  // State for a window aggregate. The count field is only used for average.
186  struct AggregateState {
187  int64_t val;
188  int64_t count;
189  std::vector<void*> outputs;
190  llvm::Value* row_number = nullptr;
191  };
192 
193  static Comparator makeComparator(const Analyzer::ColumnVar* col_var,
194  const int8_t* partition_values,
195  const int32_t* partition_indices,
196  const bool nulls_first);
197 
198  void computePartitionBuffer(const size_t partition_idx,
199  int64_t* output_for_partition_buff,
200  const Analyzer::WindowFunction* window_func);
201 
202  void sortPartition(const size_t partition_idx,
203  int64_t* output_for_partition_buff,
204  bool should_parallelize);
205 
207  size_t partition_idx,
208  const int64_t* orederd_col_idx_buf);
209 
211  size_t partition_idx,
212  size_t partition_size,
213  const int8_t* col_buf,
214  const int32_t* original_rowid_buf,
215  const int64_t* ordered_rowid_buf,
216  SQLTypeInfo input_col_ti);
217 
218  void fillPartitionStart();
219 
220  void fillPartitionEnd();
221 
225  // Keeps ownership of order column.
226  std::vector<std::vector<std::shared_ptr<Chunk_NS::Chunk>>> order_columns_owner_;
227  // Order column buffers.
228  std::vector<const int8_t*> order_columns_;
229  std::vector<SQLTypeInfo> order_columns_ti_;
230  // Hash table which contains the partitions specified by the window.
231  std::shared_ptr<HashJoin> partitions_;
232  // The number of elements in the table.
233  size_t elem_count_;
234  // The output of the window function.
235  int8_t* output_;
236  std::shared_ptr<std::vector<int64_t>> sorted_partition_buf_;
237  // Keeps ownership of column referenced in window function expression.
238  std::vector<std::vector<std::shared_ptr<Chunk_NS::Chunk>>>
240  // Column buffers used for window function expression
241  std::vector<const int8_t*> window_func_expr_columns_;
242  // we need to build a segment tree depending on the input column type
243  std::vector<std::shared_ptr<void>> segment_trees_owned_;
250  // Markers for partition start used to reinitialize state for aggregate window
251  // functions.
253  // Markers for partition end used to reinitialize state for aggregate window
254  // functions.
255  int8_t* partition_end_;
256  // State for aggregate function over a window.
259  std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner_;
260 
261  // For use when we do not have partitions_ hash table
262  const int32_t dummy_count_;
263  const int32_t dummy_offset_;
264  // dummy_payload_ is only initialized if there is no partitions_ hash table
265  // TODO(todd): There is no need for index buffer for non-partitioned
266  // window functions, as the row to index mapping is the identity function,
267  // so refactor makeComparator and ilk to allow for this
268  int32_t* dummy_payload_;
269 };
270 
271 // Keeps track of the multiple window functions in a window query.
273  public:
275  std::unique_ptr<WindowFunctionContext> window_function_context,
276  const size_t target_index);
277 
278  // Marks the window function at the given target index as active. This simplifies the
279  // code generation since it's now context sensitive. Each value window function can have
280  // its own iteration order, therefore fetching a column at a given position changes
281  // depending on which window function is active.
283  Executor* executor,
284  const size_t target_index) const;
285 
286  // Resets the active window function, which restores the regular (non-window) codegen
287  // behavior.
288  static void resetWindowFunctionContext(Executor* executor);
289 
290  // Gets the current active window function.
291  static WindowFunctionContext* getActiveWindowFunctionContext(Executor* executor);
292 
293  // Creates the context for a window function execution unit.
294  static WindowProjectNodeContext* create(Executor* executor);
295 
296  // Retrieves the context for the active window function execution unit.
297  static const WindowProjectNodeContext* get(Executor* executor);
298 
299  // Resets the active context.
300  static void reset(Executor* executor);
301 
302  private:
303  // A map from target index to the context associated with the window function at that
304  // target index.
305  std::unordered_map<size_t, std::unique_ptr<WindowFunctionContext>> window_contexts_;
306 };
307 
309 
size_t getAggregateTreeFanout() const
Defines data structures for the semantic analysis phase of query processing.
std::vector< SumAndCountPair< double > * > derived_aggregate_tree_for_double_type_
Definition: WindowContext.h:66
void addOrderColumn(const int8_t *column, const SQLTypeInfo &ti, const std::vector< std::shared_ptr< Chunk_NS::Chunk >> &chunks_owner)
std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner_
IndexPair computeNullRangeOfSortedPartition(SQLTypeInfo order_col_ti, size_t partition_idx, const int64_t *orederd_col_idx_buf)
const int32_t dummy_count_
std::vector< double * > aggregate_tree_for_double_type_
Definition: WindowContext.h:64
ExecutorDeviceType
int64_t * getNullValueEndPos() const
const int32_t dummy_offset_
const int8_t * partitionStart() const
const std::vector< SQLTypeInfo > & getOrderKeyColumnBufferTypes() const
void setSortedPartitionCacheKey(QueryPlanHash cache_key)
static WindowProjectNodeContext * create(Executor *executor)
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)
size_t elementCount() const
const int8_t * output() const
const Analyzer::WindowFunction * window_func_
const int32_t * counts() const
Constants for Builtin SQL Types supported by HEAVY.AI.
const int32_t * offsets() const
static WindowFunctionContext * getActiveWindowFunctionContext(Executor *executor)
const bool needsToBuildAggregateTree() const
size_t * getAggregateTreeDepth() const
int64_t ** getAggregationTreesForIntegerTypeWindowExpr() const
int64_t * getNullValueStartPos() const
SumAndCountPair< double > ** getDerivedAggregationTreesForDoubleTypeWindowExpr() const
const int64_t * partitionStartOffset() const
std::vector< std::shared_ptr< void > > segment_trees_owned_
std::shared_ptr< std::vector< int64_t > > sorted_partition_buf_
size_t partitionCount() const
AggregateState aggregate_state_
bool window_function_is_value(const SqlWindowFunctionKind kind)
Definition: WindowContext.h:29
size_t g_window_function_aggregation_tree_fanout
const int64_t * aggregateStateCount() const
std::vector< Comparator > createComparator(size_t partition_idx)
QueryPlanHash sorted_partition_cache_key_
const int8_t * partitionEnd() const
std::vector< std::vector< std::shared_ptr< Chunk_NS::Chunk > > > window_func_expr_columns_owner_
void addColumnBufferForWindowFunctionExpression(const int8_t *column, const std::vector< std::shared_ptr< Chunk_NS::Chunk >> &chunks_owner)
static void reset(Executor *executor)
const WindowFunctionContext * activateWindowFunctionContext(Executor *executor, const size_t target_index) const
static Comparator makeComparator(const Analyzer::ColumnVar *col_var, const int8_t *partition_values, const int32_t *partition_indices, const bool nulls_first)
QueryPlanHash partition_cache_key_
int64_t aggregateStatePendingOutputs() const
const int64_t * aggregateState() const
std::function< bool(const PermutationIdx, const PermutationIdx)> Comparator
Definition: ResultSet.h:155
WindowFunctionContext & operator=(const WindowFunctionContext &)=delete
std::pair< int64_t, int64_t > IndexPair
void buildAggregationTreeForPartition(SqlWindowFunctionKind agg_type, size_t partition_idx, size_t partition_size, const int8_t *col_buf, const int32_t *original_rowid_buf, const int64_t *ordered_rowid_buf, SQLTypeInfo input_col_ti)
SumAndCountPair< int64_t > ** getDerivedAggregationTreesForIntegerTypeWindowExpr() const
Executor(const ExecutorId id, Data_Namespace::DataMgr *data_mgr, const size_t block_size_x, const size_t grid_size_x, const size_t max_gpu_slab_size, const std::string &debug_dir, const std::string &debug_file)
Definition: Execute.cpp:244
bool window_function_is_aggregate(const SqlWindowFunctionKind kind)
Definition: WindowContext.h:44
void addWindowFunctionContext(std::unique_ptr< WindowFunctionContext > window_function_context, const size_t target_index)
std::vector< const int8_t * > window_func_expr_columns_
AggregateTreeForWindowFraming aggregate_trees_
void sortPartition(const size_t partition_idx, int64_t *output_for_partition_buff, bool should_parallelize)
WindowFunctionContext(const Analyzer::WindowFunction *window_func, const size_t elem_count, const ExecutorDeviceType device_type, std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner)
const std::vector< const int8_t * > & getOrderKeyColumnBuffers() const
std::shared_ptr< HashJoin > partitions_
size_t QueryPlanHash
const int64_t * partitionNumCountBuf() const
bool window_function_requires_peer_handling(const Analyzer::WindowFunction *window_func)
SqlWindowFunctionKind
Definition: sqldefs.h:110
std::unordered_map< size_t, std::unique_ptr< WindowFunctionContext > > window_contexts_
void computePartitionBuffer(const size_t partition_idx, int64_t *output_for_partition_buff, const Analyzer::WindowFunction *window_func)
int64_t * partition_start_offset_
std::vector< const int8_t * > order_columns_
const int64_t * sortedPartition() const
static void resetWindowFunctionContext(Executor *executor)
std::vector< SQLTypeInfo > order_columns_ti_
std::vector< SumAndCountPair< int64_t > * > derived_aggregate_tree_for_integer_type_
Definition: WindowContext.h:65
const Analyzer::WindowFunction * getWindowFunction() const
int64_t * aggregate_tree_null_end_pos_
const int32_t * payload() const
std::function< WindowFunctionContext::WindowComparatorResult(const int64_t lhs, const int64_t rhs)> Comparator
size_t * aggregate_trees_depth_
std::vector< std::vector< std::shared_ptr< Chunk_NS::Chunk > > > order_columns_owner_
int64_t * aggregate_tree_null_start_pos_
std::vector< int64_t * > aggregate_tree_for_integer_type_
Definition: WindowContext.h:63
const ExecutorDeviceType device_type_
double ** getAggregationTreesForDoubleTypeWindowExpr() const