OmniSciDB  b28c0d5765
 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  default:
37  return false;
38  }
39 }
40 
41 // Returns true for aggregate window functions, false otherwise.
43  switch (kind) {
51  return true;
52  default:
53  return false;
54  }
55 }
56 
58  switch (kind) {
60  return true;
61  default:
62  return false;
63  }
64 }
65 
66 class Executor;
67 
69  std::vector<int64_t*> aggregate_tree_for_integer_type_;
70  std::vector<double*> aggregate_tree_for_double_type_;
71  std::vector<SumAndCountPair<int64_t>*> derived_aggregate_tree_for_integer_type_;
72  std::vector<SumAndCountPair<double>*> derived_aggregate_tree_for_double_type_;
73 };
74 
78  llvm::Value* current_row_pos_lv;
79  llvm::Value* current_col_value_lv;
81  llvm::Value* int64_t_zero_val_lv;
83  llvm::Value* order_key_buf_ptr_lv;
84  std::string order_type_col_name;
87  llvm::Value* null_start_pos_lv;
88  llvm::Value* null_end_pos_lv;
89 };
90 
96 };
97 
98 // Per-window function context which encapsulates the logic for computing the various
99 // window function kinds and keeps ownership of buffers which contain the results. For
100 // rank functions, the code generated for the projection simply reads the values and
101 // writes them to the result set. For value and aggregate functions, only the iteration
102 // order is written to the buffer, the rest is handled by generating code in a similar way
103 // we do for non-window queries.
105  public:
106  // non-partitioned version
108  const size_t elem_count,
109  const ExecutorDeviceType device_type,
110  std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner);
111 
112  // partitioned version
114  const Analyzer::WindowFunction* window_func,
115  QueryPlanHash cache_key,
116  const std::shared_ptr<HashJoin>& partitions,
117  const size_t elem_count,
118  const ExecutorDeviceType device_type,
119  std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
120  size_t aggregation_tree_fan_out = g_window_function_aggregation_tree_fanout);
121 
123 
125 
127 
128  // Adds the order column buffer to the context and keeps ownership of it.
129  void addOrderColumn(const int8_t* column,
130  const SQLTypeInfo& ti,
131  const std::vector<std::shared_ptr<Chunk_NS::Chunk>>& chunks_owner);
132 
134 
136  const int8_t* column,
137  const std::vector<std::shared_ptr<Chunk_NS::Chunk>>& chunks_owner);
138 
139  enum class WindowComparatorResult { LT, EQ, GT };
140  using Comparator =
141  std::function<WindowFunctionContext::WindowComparatorResult(const int64_t lhs,
142  const int64_t rhs)>;
143 
144  std::vector<Comparator> createComparator(size_t partition_idx);
145 
146  // Computes the window function result to be used during the actual projection query.
147  void compute(
148  std::unordered_map<QueryPlanHash, size_t>& sorted_partition_key_ref_count_map,
149  std::unordered_map<QueryPlanHash, std::shared_ptr<std::vector<int64_t>>>&
150  sorted_partition_cache);
151 
152  // Returns a pointer to the window function associated with this context.
154 
155  // Returns a pointer to the output buffer of the window function result.
156  const int8_t* output() const;
157 
158  // Returns a pointer to the sorted row index buffer
159  const int64_t* sortedPartition() const;
160 
161  // Returns a pointer to the value field of the aggregation state.
162  const int64_t* aggregateState() const;
163 
164  // Returns a pointer to the count field of the aggregation state.
165  const int64_t* aggregateStateCount() const;
166 
167  // Returns a handle to the pending outputs for the aggregate window function.
168  int64_t aggregateStatePendingOutputs() const;
169 
170  const int64_t* partitionStartOffset() const;
171 
172  const int64_t* partitionNumCountBuf() const;
173 
174  const std::vector<const int8_t*>& getColumnBufferForWindowFunctionExpressions() const;
175 
176  const std::vector<const int8_t*>& getOrderKeyColumnBuffers() const;
177 
178  const std::vector<SQLTypeInfo>& getOrderKeyColumnBufferTypes() const;
179 
181 
183 
185 
187 
188  size_t* getAggregateTreeDepth() const;
189 
190  size_t getAggregateTreeFanout() const;
191 
192  int64_t* getNullValueStartPos() const;
193 
194  int64_t* getNullValueEndPos() const;
195 
196  // Returns a pointer to the partition start bitmap.
197  const int8_t* partitionStart() const;
198 
199  // Returns a pointer to the partition end bitmap.
200  const int8_t* partitionEnd() const;
201 
202  // Returns the element count in the columns used by the window function.
203  size_t elementCount() const;
204 
205  const int32_t* payload() const;
206 
207  const int32_t* offsets() const;
208 
209  const int32_t* counts() const;
210 
211  size_t partitionCount() const;
212 
213  const bool needsToBuildAggregateTree() const;
214 
215  private:
216  // State for a window aggregate. The count field is only used for average.
217  struct AggregateState {
218  int64_t val;
219  int64_t count;
220  std::vector<void*> outputs;
221  llvm::Value* row_number = nullptr;
222  };
223 
224  static Comparator makeComparator(const Analyzer::ColumnVar* col_var,
225  const int8_t* partition_values,
226  const int32_t* partition_indices,
227  const bool nulls_first);
228 
229  void computePartitionBuffer(const size_t partition_idx,
230  int64_t* output_for_partition_buff,
231  const Analyzer::WindowFunction* window_func);
232 
233  void sortPartition(const size_t partition_idx,
234  int64_t* output_for_partition_buff,
235  bool should_parallelize);
236 
237  void computeNullRangeOfSortedPartition(const SQLTypeInfo& order_col_ti,
238  size_t partition_idx,
239  const int32_t* original_col_idx_buf,
240  const int64_t* ordered_col_idx_buf);
241 
243  size_t partition_idx,
244  size_t partition_size,
245  const int8_t* col_buf,
246  const int32_t* original_rowid_buf,
247  const int64_t* ordered_rowid_buf,
248  const SQLTypeInfo& input_col_ti);
249 
250  void fillPartitionStart();
251 
252  void fillPartitionEnd();
253 
257  // Keeps ownership of order column.
258  std::vector<std::vector<std::shared_ptr<Chunk_NS::Chunk>>> order_columns_owner_;
259  // Order column buffers.
260  std::vector<const int8_t*> order_columns_;
261  std::vector<SQLTypeInfo> order_columns_ti_;
262  // Hash table which contains the partitions specified by the window.
263  std::shared_ptr<HashJoin> partitions_;
264  // The number of elements in the table.
265  size_t elem_count_;
266  // The output of the window function.
267  int8_t* output_;
268  std::shared_ptr<std::vector<int64_t>> sorted_partition_buf_;
269  // Keeps ownership of column referenced in window function expression.
270  std::vector<std::vector<std::shared_ptr<Chunk_NS::Chunk>>>
272  // Column buffers used for window function expression
273  std::vector<const int8_t*> window_func_expr_columns_;
274  // we need to build a segment tree depending on the input column type
275  std::vector<std::shared_ptr<void>> segment_trees_owned_;
282  // Markers for partition start used to reinitialize state for aggregate window
283  // functions.
285  // Markers for partition end used to reinitialize state for aggregate window
286  // functions.
287  int8_t* partition_end_;
288  // State for aggregate function over a window.
291  std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner_;
292 
293  // For use when we do not have partitions_ hash table
294  const int32_t dummy_count_;
295  const int32_t dummy_offset_;
296  // dummy_payload_ is only initialized if there is no partitions_ hash table
297  // TODO(todd): There is no need for index buffer for non-partitioned
298  // window functions, as the row to index mapping is the identity function,
299  // so refactor makeComparator and ilk to allow for this
300  int32_t* dummy_payload_;
301 };
302 
303 // Keeps track of the multiple window functions in a window query.
305  public:
307  std::unique_ptr<WindowFunctionContext> window_function_context,
308  const size_t target_index);
309 
310  // Marks the window function at the given target index as active. This simplifies the
311  // code generation since it's now context sensitive. Each value window function can have
312  // its own iteration order, therefore fetching a column at a given position changes
313  // depending on which window function is active.
315  Executor* executor,
316  const size_t target_index) const;
317 
318  // Resets the active window function, which restores the regular (non-window) codegen
319  // behavior.
320  static void resetWindowFunctionContext(Executor* executor);
321 
322  // Gets the current active window function.
323  static WindowFunctionContext* getActiveWindowFunctionContext(Executor* executor);
324 
325  // Creates the context for a window function execution unit.
326  static WindowProjectNodeContext* create(Executor* executor);
327 
328  // Retrieves the context for the active window function execution unit.
329  static const WindowProjectNodeContext* get(Executor* executor);
330 
331  // Resets the active context.
332  static void reset(Executor* executor);
333 
334  private:
335  // A map from target index to the context associated with the window function at that
336  // target index.
337  std::unordered_map<size_t, std::unique_ptr<WindowFunctionContext>> window_contexts_;
338 };
339 
341 
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:72
void addOrderColumn(const int8_t *column, const SQLTypeInfo &ti, const std::vector< std::shared_ptr< Chunk_NS::Chunk >> &chunks_owner)
llvm::Value * num_elem_current_partition_lv
Definition: WindowContext.h:82
std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner_
int64_t * ordered_partition_null_start_pos_
llvm::Value * current_col_value_lv
Definition: WindowContext.h:79
const int32_t dummy_count_
std::vector< double * > aggregate_tree_for_double_type_
Definition: WindowContext.h:70
ExecutorDeviceType
int64_t * getNullValueEndPos() const
bool window_function_conditional_aggregate(const SqlWindowFunctionKind kind)
Definition: WindowContext.h:57
llvm::Value * current_partition_start_offset_lv
Definition: WindowContext.h:92
const int32_t dummy_offset_
llvm::Value * current_row_pos_lv
Definition: WindowContext.h:78
llvm::Value * target_partition_rowid_ptr_lv
Definition: WindowContext.h:85
const int8_t * partitionStart() const
llvm::Value * frame_end_bound_expr_lv
Definition: WindowContext.h:77
llvm::Value * num_elem_current_partition_lv
Definition: WindowContext.h:93
const std::vector< SQLTypeInfo > & getOrderKeyColumnBufferTypes() const
void setSortedPartitionCacheKey(QueryPlanHash cache_key)
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)
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
llvm::Value * target_partition_sorted_rowid_ptr_lv
Definition: WindowContext.h:95
llvm::Value * target_partition_rowid_ptr_lv
Definition: WindowContext.h:94
static WindowFunctionContext * getActiveWindowFunctionContext(Executor *executor)
const bool needsToBuildAggregateTree() const
size_t * getAggregateTreeDepth() const
int64_t ** getAggregationTreesForIntegerTypeWindowExpr() const
const std::vector< const int8_t * > & getColumnBufferForWindowFunctionExpressions() 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_
llvm::Value * null_start_pos_lv
Definition: WindowContext.h:87
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)
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, const SQLTypeInfo &input_col_ti)
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
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:245
llvm::Value * current_partition_start_offset_lv
Definition: WindowContext.h:80
bool window_function_is_aggregate(const SqlWindowFunctionKind kind)
Definition: WindowContext.h:42
void addWindowFunctionContext(std::unique_ptr< WindowFunctionContext > window_function_context, const size_t target_index)
llvm::Value * target_partition_sorted_rowid_ptr_lv
Definition: WindowContext.h:86
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
llvm::Value * null_end_pos_lv
Definition: WindowContext.h:88
const int64_t * partitionNumCountBuf() const
bool window_function_requires_peer_handling(const Analyzer::WindowFunction *window_func)
SqlWindowFunctionKind
Definition: sqldefs.h:119
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)
llvm::Value * order_key_buf_ptr_lv
Definition: WindowContext.h:83
int64_t * partition_start_offset_
llvm::Value * int64_t_zero_val_lv
Definition: WindowContext.h:81
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:71
const Analyzer::WindowFunction * getWindowFunction() const
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_
llvm::Value * frame_start_bound_expr_lv
Definition: WindowContext.h:76
std::vector< int64_t * > aggregate_tree_for_integer_type_
Definition: WindowContext.h:69
const ExecutorDeviceType device_type_
std::string order_type_col_name
Definition: WindowContext.h:84
double ** getAggregationTreesForDoubleTypeWindowExpr() const
int64_t * ordered_partition_null_end_pos_