OmniSciDB  ca0c39ec8f
 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*>& getColumnBufferForWindowFunctionExpressions() const;
146 
147  const std::vector<const int8_t*>& getOrderKeyColumnBuffers() const;
148 
149  const std::vector<SQLTypeInfo>& getOrderKeyColumnBufferTypes() const;
150 
152 
154 
156 
158 
159  size_t* getAggregateTreeDepth() const;
160 
161  size_t getAggregateTreeFanout() const;
162 
163  int64_t* getNullValueStartPos() const;
164 
165  int64_t* getNullValueEndPos() const;
166 
167  // Returns a pointer to the partition start bitmap.
168  const int8_t* partitionStart() const;
169 
170  // Returns a pointer to the partition end bitmap.
171  const int8_t* partitionEnd() const;
172 
173  // Returns the element count in the columns used by the window function.
174  size_t elementCount() const;
175 
176  const int32_t* payload() const;
177 
178  const int32_t* offsets() const;
179 
180  const int32_t* counts() const;
181 
182  size_t partitionCount() const;
183 
184  const bool needsToBuildAggregateTree() const;
185 
186  private:
187  // State for a window aggregate. The count field is only used for average.
188  struct AggregateState {
189  int64_t val;
190  int64_t count;
191  std::vector<void*> outputs;
192  llvm::Value* row_number = nullptr;
193  };
194 
195  static Comparator makeComparator(const Analyzer::ColumnVar* col_var,
196  const int8_t* partition_values,
197  const int32_t* partition_indices,
198  const bool nulls_first);
199 
200  void computePartitionBuffer(const size_t partition_idx,
201  int64_t* output_for_partition_buff,
202  const Analyzer::WindowFunction* window_func);
203 
204  void sortPartition(const size_t partition_idx,
205  int64_t* output_for_partition_buff,
206  bool should_parallelize);
207 
208  void computeNullRangeOfSortedPartition(const SQLTypeInfo& order_col_ti,
209  size_t partition_idx,
210  const int32_t* original_col_idx_buf,
211  const int64_t* ordered_col_idx_buf);
212 
214  size_t partition_idx,
215  size_t partition_size,
216  const int8_t* col_buf,
217  const int32_t* original_rowid_buf,
218  const int64_t* ordered_rowid_buf,
219  const SQLTypeInfo& input_col_ti);
220 
221  void fillPartitionStart();
222 
223  void fillPartitionEnd();
224 
228  // Keeps ownership of order column.
229  std::vector<std::vector<std::shared_ptr<Chunk_NS::Chunk>>> order_columns_owner_;
230  // Order column buffers.
231  std::vector<const int8_t*> order_columns_;
232  std::vector<SQLTypeInfo> order_columns_ti_;
233  // Hash table which contains the partitions specified by the window.
234  std::shared_ptr<HashJoin> partitions_;
235  // The number of elements in the table.
236  size_t elem_count_;
237  // The output of the window function.
238  int8_t* output_;
239  std::shared_ptr<std::vector<int64_t>> sorted_partition_buf_;
240  // Keeps ownership of column referenced in window function expression.
241  std::vector<std::vector<std::shared_ptr<Chunk_NS::Chunk>>>
243  // Column buffers used for window function expression
244  std::vector<const int8_t*> window_func_expr_columns_;
245  // we need to build a segment tree depending on the input column type
246  std::vector<std::shared_ptr<void>> segment_trees_owned_;
253  // Markers for partition start used to reinitialize state for aggregate window
254  // functions.
256  // Markers for partition end used to reinitialize state for aggregate window
257  // functions.
258  int8_t* partition_end_;
259  // State for aggregate function over a window.
262  std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner_;
263 
264  // For use when we do not have partitions_ hash table
265  const int32_t dummy_count_;
266  const int32_t dummy_offset_;
267  // dummy_payload_ is only initialized if there is no partitions_ hash table
268  // TODO(todd): There is no need for index buffer for non-partitioned
269  // window functions, as the row to index mapping is the identity function,
270  // so refactor makeComparator and ilk to allow for this
271  int32_t* dummy_payload_;
272 };
273 
274 // Keeps track of the multiple window functions in a window query.
276  public:
278  std::unique_ptr<WindowFunctionContext> window_function_context,
279  const size_t target_index);
280 
281  // Marks the window function at the given target index as active. This simplifies the
282  // code generation since it's now context sensitive. Each value window function can have
283  // its own iteration order, therefore fetching a column at a given position changes
284  // depending on which window function is active.
286  Executor* executor,
287  const size_t target_index) const;
288 
289  // Resets the active window function, which restores the regular (non-window) codegen
290  // behavior.
291  static void resetWindowFunctionContext(Executor* executor);
292 
293  // Gets the current active window function.
294  static WindowFunctionContext* getActiveWindowFunctionContext(Executor* executor);
295 
296  // Creates the context for a window function execution unit.
297  static WindowProjectNodeContext* create(Executor* executor);
298 
299  // Retrieves the context for the active window function execution unit.
300  static const WindowProjectNodeContext* get(Executor* executor);
301 
302  // Resets the active context.
303  static void reset(Executor* executor);
304 
305  private:
306  // A map from target index to the context associated with the window function at that
307  // target index.
308  std::unordered_map<size_t, std::unique_ptr<WindowFunctionContext>> window_contexts_;
309 };
310 
312 
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_
int64_t * ordered_partition_null_start_pos_
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)
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
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_
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: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:114
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
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_
std::vector< int64_t * > aggregate_tree_for_integer_type_
Definition: WindowContext.h:63
const ExecutorDeviceType device_type_
double ** getAggregationTreesForDoubleTypeWindowExpr() const
int64_t * ordered_partition_null_end_pos_