be/src/exprs/vexpr_context.cpp
Line | Count | Source |
1 | | // Licensed to the Apache Software Foundation (ASF) under one |
2 | | // or more contributor license agreements. See the NOTICE file |
3 | | // distributed with this work for additional information |
4 | | // regarding copyright ownership. The ASF licenses this file |
5 | | // to you under the Apache License, Version 2.0 (the |
6 | | // "License"); you may not use this file except in compliance |
7 | | // with the License. You may obtain a copy of the License at |
8 | | // |
9 | | // http://www.apache.org/licenses/LICENSE-2.0 |
10 | | // |
11 | | // Unless required by applicable law or agreed to in writing, |
12 | | // software distributed under the License is distributed on an |
13 | | // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
14 | | // KIND, either express or implied. See the License for the |
15 | | // specific language governing permissions and limitations |
16 | | // under the License. |
17 | | |
18 | | #include "exprs/vexpr_context.h" |
19 | | |
20 | | #include <algorithm> |
21 | | #include <cstdint> |
22 | | #include <memory> |
23 | | #include <string> |
24 | | #include <utility> |
25 | | |
26 | | #include "common/compiler_util.h" // IWYU pragma: keep |
27 | | #include "common/exception.h" |
28 | | #include "common/logging.h" |
29 | | #include "common/status.h" |
30 | | #include "core/block/column_numbers.h" |
31 | | #include "core/block/column_with_type_and_name.h" |
32 | | #include "core/block/columns_with_type_and_name.h" |
33 | | #include "core/column/column.h" |
34 | | #include "core/column/column_const.h" |
35 | | #include "exec/common/util.hpp" |
36 | | #include "exprs/function_context.h" |
37 | | #include "exprs/lambda_function/lambda_execution_context.h" |
38 | | #include "exprs/vexpr.h" |
39 | | #include "runtime/runtime_state.h" |
40 | | #include "runtime/thread_context.h" |
41 | | #include "storage/olap_common.h" |
42 | | #include "storage/segment/column_reader.h" |
43 | | #include "util/simd/bits.h" |
44 | | |
45 | | namespace doris { |
46 | | class RowDescriptor; |
47 | | } // namespace doris |
48 | | |
49 | | namespace doris { |
50 | | |
51 | 29.5M | VExprContext::VExprContext(VExprSPtr expr) : _root(std::move(expr)) {} |
52 | | |
53 | 29.4M | VExprContext::~VExprContext() { |
54 | | // In runtime filter, only create expr context to get expr root, will not call |
55 | | // prepare or open, so that it is not need to call close. And call close may core |
56 | | // because the function context in expr is not set. |
57 | 29.4M | if (!_prepared || !_opened) { |
58 | 615k | return; |
59 | 615k | } |
60 | 28.8M | try { |
61 | 28.8M | close(); |
62 | 28.8M | } catch (const Exception& e) { |
63 | 0 | LOG(WARNING) << "Exception occurs when expr context deconstruct: " << e.to_string(); |
64 | 0 | } |
65 | 28.8M | } |
66 | | |
67 | 21.4k | LambdaExecutionContext& VExprContext::lambda_execution_context() { |
68 | 21.4k | if (!_lambda_execution_context) { |
69 | 2.82k | _lambda_execution_context = std::make_unique<LambdaExecutionContext>(); |
70 | 2.82k | } |
71 | 21.4k | return *_lambda_execution_context; |
72 | 21.4k | } |
73 | | |
74 | 2.31M | Status VExprContext::execute(Block* block, int* result_column_id) { |
75 | 2.31M | Status st; |
76 | 2.31M | RETURN_IF_CATCH_EXCEPTION({ |
77 | 2.31M | st = _root->execute(this, block, result_column_id); |
78 | 2.31M | _last_result_column_id = *result_column_id; |
79 | | // We should first check the status, as some expressions might incorrectly set result_column_id, even if the st is not ok. |
80 | 2.31M | if (st.ok() && _last_result_column_id != -1) { |
81 | 2.31M | block->get_by_position(*result_column_id).column->sanity_check(); |
82 | 2.31M | RETURN_IF_ERROR( |
83 | 2.31M | block->get_by_position(*result_column_id).check_type_and_column_match()); |
84 | 2.31M | } |
85 | 2.31M | }); |
86 | 2.31M | return st; |
87 | 2.31M | } |
88 | | |
89 | 2.78M | Status VExprContext::execute(const Block* block, ColumnPtr& result_column) { |
90 | 2.78M | Status st; |
91 | 2.78M | RETURN_IF_CATCH_EXCEPTION( |
92 | 2.78M | { st = _root->execute_column(this, block, nullptr, block->rows(), result_column); }); |
93 | 2.78M | return st; |
94 | 2.78M | } |
95 | | |
96 | 32.1k | Status VExprContext::execute(const Block* block, ColumnWithTypeAndName& result_data) { |
97 | 32.1k | Status st; |
98 | 32.1k | ColumnPtr result_column; |
99 | 32.1k | RETURN_IF_CATCH_EXCEPTION( |
100 | 32.1k | { st = _root->execute_column(this, block, nullptr, block->rows(), result_column); }); |
101 | 32.1k | RETURN_IF_ERROR(st); |
102 | 32.1k | result_data.column = result_column; |
103 | 32.1k | result_data.type = execute_type(block); |
104 | 32.1k | result_data.name = _root->expr_name(); |
105 | 32.1k | return Status::OK(); |
106 | 32.1k | } |
107 | | |
108 | 1.23M | DataTypePtr VExprContext::execute_type(const Block* block) { |
109 | 1.23M | return _root->execute_type(block); |
110 | 1.23M | } |
111 | | |
112 | 1.12M | Status VExprContext::execute_const_expr(ColumnWithTypeAndName& result) { |
113 | 1.12M | Status st; |
114 | 1.12M | RETURN_IF_CATCH_EXCEPTION( |
115 | 1.12M | { st = _root->execute_column(this, nullptr, nullptr, 1, result.column); }); |
116 | 1.12M | RETURN_IF_ERROR(st); |
117 | 1.12M | result.type = _root->execute_type(nullptr); |
118 | 1.12M | result.name = _root->expr_name(); |
119 | 1.12M | return Status::OK(); |
120 | 1.12M | } |
121 | | |
122 | 1.20M | [[nodiscard]] const std::string& VExprContext::expr_name() const { |
123 | 1.20M | return _root->expr_name(); |
124 | 1.20M | } |
125 | | |
126 | 0 | bool VExprContext::is_blockable() const { |
127 | 0 | return _root->is_blockable(); |
128 | 0 | } |
129 | | |
130 | 7.67M | Status VExprContext::prepare(RuntimeState* state, const RowDescriptor& row_desc) { |
131 | 7.67M | _prepared = true; |
132 | 7.67M | Status st; |
133 | 7.67M | RETURN_IF_CATCH_EXCEPTION({ st = _root->prepare(state, row_desc, this); }); |
134 | 7.67M | return st; |
135 | 7.67M | } |
136 | | |
137 | 7.68M | Status VExprContext::open(RuntimeState* state) { |
138 | 7.68M | DCHECK(_prepared); |
139 | 7.68M | if (_opened) { |
140 | 3.45k | return Status::OK(); |
141 | 3.45k | } |
142 | 7.68M | _opened = true; |
143 | | // Fragment-local state is only initialized for original contexts. Clones inherit the |
144 | | // original's fragment state and only need to have thread-local state initialized. |
145 | 7.68M | FunctionContext::FunctionStateScope scope = |
146 | 7.68M | _is_clone ? FunctionContext::THREAD_LOCAL : FunctionContext::FRAGMENT_LOCAL; |
147 | 7.68M | Status st; |
148 | 7.68M | RETURN_IF_CATCH_EXCEPTION({ st = _root->open(state, this, scope); }); |
149 | 7.67M | return st; |
150 | 7.68M | } |
151 | | |
152 | 28.8M | void VExprContext::close() { |
153 | | // Sometimes expr context may not have a root, then it need not call close |
154 | 28.8M | if (_root == nullptr) { |
155 | 0 | return; |
156 | 0 | } |
157 | 28.8M | FunctionContext::FunctionStateScope scope = |
158 | 28.8M | _is_clone ? FunctionContext::THREAD_LOCAL : FunctionContext::FRAGMENT_LOCAL; |
159 | 28.8M | _root->close(this, scope); |
160 | 28.8M | } |
161 | | |
162 | 21.0M | Status VExprContext::clone(RuntimeState* state, VExprContextSPtr& new_ctx) { |
163 | 18.4E | DCHECK(_prepared) << "expr context not prepared"; |
164 | 21.0M | DCHECK(_opened); |
165 | 21.0M | DCHECK(new_ctx.get() == nullptr); |
166 | | |
167 | 21.0M | new_ctx = std::make_shared<VExprContext>(_root); |
168 | 21.0M | for (auto& _fn_context : _fn_contexts) { |
169 | 1.58M | new_ctx->_fn_contexts.push_back(_fn_context->clone()); |
170 | 1.58M | } |
171 | | |
172 | 21.0M | new_ctx->_is_clone = true; |
173 | 21.0M | new_ctx->_prepared = true; |
174 | 21.0M | new_ctx->_opened = true; |
175 | | // segment_v2::AnnRangeSearchRuntime should be cloned as well. |
176 | | // The object of segment_v2::AnnRangeSearchRuntime is not shared by threads. |
177 | 21.0M | new_ctx->_ann_range_search_runtime = this->_ann_range_search_runtime; |
178 | | |
179 | 21.0M | return _root->open(state, new_ctx.get(), FunctionContext::THREAD_LOCAL); |
180 | 21.0M | } |
181 | | |
182 | 0 | void VExprContext::clone_fn_contexts(VExprContext* other) { |
183 | 0 | for (auto& _fn_context : _fn_contexts) { |
184 | 0 | other->_fn_contexts.push_back(_fn_context->clone()); |
185 | 0 | } |
186 | 0 | } |
187 | | |
188 | | int VExprContext::register_function_context(RuntimeState* state, const DataTypePtr& return_type, |
189 | 963k | const std::vector<DataTypePtr>& arg_types) { |
190 | 963k | _fn_contexts.push_back(FunctionContext::create_context(state, return_type, arg_types)); |
191 | 963k | _fn_contexts.back()->set_check_overflow_for_decimal(state->check_overflow_for_decimal()); |
192 | 963k | _fn_contexts.back()->set_enable_strict_mode(state->enable_strict_mode()); |
193 | 963k | return static_cast<int>(_fn_contexts.size()) - 1; |
194 | 963k | } |
195 | | |
196 | 18.2k | Status VExprContext::evaluate_inverted_index(uint32_t segment_num_rows) { |
197 | 18.2k | Status st; |
198 | 18.2k | RETURN_IF_CATCH_EXCEPTION({ st = _root->evaluate_inverted_index(this, segment_num_rows); }); |
199 | 18.2k | return st; |
200 | 18.2k | } |
201 | | |
202 | | ZoneMapFilterResult VExprContext::evaluate_zonemap_filter(const VExprContextSPtrs& conjuncts, |
203 | 48.0k | const ZoneMapEvalContext& ctx) { |
204 | 64.2k | for (const auto& conjunct : conjuncts) { |
205 | 64.2k | DORIS_CHECK(conjunct != nullptr); |
206 | 64.2k | const auto& root = conjunct->root(); |
207 | 64.2k | DORIS_CHECK(root != nullptr); |
208 | 64.2k | if (!root->can_evaluate_zonemap_filter()) { |
209 | 19.2k | continue; |
210 | 19.2k | } |
211 | 45.0k | if (root->evaluate_zonemap_filter(ctx) == ZoneMapFilterResult::kNoMatch) { |
212 | 13.8k | return ZoneMapFilterResult::kNoMatch; |
213 | 13.8k | } |
214 | 45.0k | } |
215 | 34.1k | return ZoneMapFilterResult::kMayMatch; |
216 | 48.0k | } |
217 | | |
218 | | ZoneMapFilterResult VExprContext::evaluate_dictionary_filter(const VExprContextSPtrs& conjuncts, |
219 | 41.3k | const DictionaryEvalContext& ctx) { |
220 | 41.3k | for (const auto& conjunct : conjuncts) { |
221 | 41.3k | DORIS_CHECK(conjunct != nullptr); |
222 | 41.3k | const auto& root = conjunct->root(); |
223 | 41.3k | DORIS_CHECK(root != nullptr); |
224 | 41.3k | if (!root->can_evaluate_dictionary_filter()) { |
225 | 0 | continue; |
226 | 0 | } |
227 | 41.3k | if (root->evaluate_dictionary_filter(ctx) == ZoneMapFilterResult::kNoMatch) { |
228 | 41.0k | return ZoneMapFilterResult::kNoMatch; |
229 | 41.0k | } |
230 | 41.3k | } |
231 | 299 | return ZoneMapFilterResult::kMayMatch; |
232 | 41.3k | } |
233 | | |
234 | | ZoneMapFilterResult VExprContext::evaluate_bloom_filter(const VExprContextSPtrs& conjuncts, |
235 | 27 | const BloomFilterEvalContext& ctx) { |
236 | 27 | for (const auto& conjunct : conjuncts) { |
237 | 27 | DORIS_CHECK(conjunct != nullptr); |
238 | 27 | const auto& root = conjunct->root(); |
239 | 27 | DORIS_CHECK(root != nullptr); |
240 | 27 | if (!root->can_evaluate_bloom_filter()) { |
241 | 0 | continue; |
242 | 0 | } |
243 | 27 | if (root->evaluate_bloom_filter(ctx) == ZoneMapFilterResult::kNoMatch) { |
244 | 8 | return ZoneMapFilterResult::kNoMatch; |
245 | 8 | } |
246 | 27 | } |
247 | 19 | return ZoneMapFilterResult::kMayMatch; |
248 | 27 | } |
249 | | |
250 | 17.8k | bool VExprContext::all_expr_inverted_index_evaluated() { |
251 | 17.8k | return _index_context->has_index_result_for_expr(_root.get()); |
252 | 17.8k | } |
253 | | |
254 | 50 | Status VExprContext::filter_block(VExprContext* vexpr_ctx, Block* block) { |
255 | 50 | if (vexpr_ctx == nullptr || block->rows() == 0) { |
256 | 0 | return Status::OK(); |
257 | 0 | } |
258 | 50 | ColumnPtr filter_column; |
259 | 50 | RETURN_IF_ERROR(vexpr_ctx->execute(block, filter_column)); |
260 | 50 | size_t filter_column_id = block->columns(); |
261 | 50 | block->insert({filter_column, vexpr_ctx->execute_type(block), "filter_column"}); |
262 | 50 | vexpr_ctx->_memory_usage = filter_column->allocated_bytes(); |
263 | 50 | return Block::filter_block(block, filter_column_id, filter_column_id); |
264 | 50 | } |
265 | | |
266 | | Status VExprContext::filter_block(const VExprContextSPtrs& expr_contexts, Block* block, |
267 | 1.90M | size_t column_to_keep) { |
268 | 1.90M | if (expr_contexts.empty() || block->rows() == 0) { |
269 | 1.80M | return Status::OK(); |
270 | 1.80M | } |
271 | | |
272 | 104k | ColumnNumbers columns_to_filter(column_to_keep); |
273 | 104k | std::iota(columns_to_filter.begin(), columns_to_filter.end(), 0); |
274 | | |
275 | 104k | return execute_conjuncts_and_filter_block(expr_contexts, block, columns_to_filter, |
276 | 104k | static_cast<int>(column_to_keep)); |
277 | 1.90M | } |
278 | | |
279 | | Status VExprContext::execute_conjuncts(const VExprContextSPtrs& ctxs, |
280 | | const std::vector<IColumn::Filter*>* filters, Block* block, |
281 | 8.29k | IColumn::Filter* result_filter, bool* can_filter_all) { |
282 | 8.29k | return execute_conjuncts(ctxs, filters, false, block, result_filter, can_filter_all); |
283 | 8.29k | } |
284 | | |
285 | | Status VExprContext::execute_filter(const Block* block, uint8_t* __restrict result_filter_data, |
286 | 431k | size_t rows, bool accept_null, bool* can_filter_all) { |
287 | 431k | return _root->execute_filter(this, block, result_filter_data, rows, accept_null, |
288 | 431k | can_filter_all); |
289 | 431k | } |
290 | | |
291 | | Status VExprContext::execute_conjuncts(const VExprContextSPtrs& ctxs, |
292 | | const std::vector<IColumn::Filter*>* filters, |
293 | | bool accept_null, const Block* block, |
294 | 327k | IColumn::Filter* result_filter, bool* can_filter_all) { |
295 | 327k | size_t rows = block->rows(); |
296 | 327k | DCHECK_EQ(result_filter->size(), rows); |
297 | 327k | *can_filter_all = false; |
298 | 327k | auto* __restrict result_filter_data = result_filter->data(); |
299 | 327k | for (const auto& ctx : ctxs) { |
300 | 259k | RETURN_IF_ERROR( |
301 | 259k | ctx->execute_filter(block, result_filter_data, rows, accept_null, can_filter_all)); |
302 | 259k | if (*can_filter_all) { |
303 | 31.0k | return Status::OK(); |
304 | 31.0k | } |
305 | 259k | } |
306 | 296k | if (filters != nullptr) { |
307 | 1.07k | for (auto* filter : *filters) { |
308 | 16 | auto* __restrict filter_data = filter->data(); |
309 | 16 | const size_t size = filter->size(); |
310 | 76 | for (size_t i = 0; i < size; ++i) { |
311 | 60 | result_filter_data[i] &= filter_data[i]; |
312 | 60 | } |
313 | 16 | if (memchr(result_filter_data, 0x1, size) == nullptr) { |
314 | 0 | *can_filter_all = true; |
315 | 0 | return Status::OK(); |
316 | 0 | } |
317 | 16 | } |
318 | 1.07k | } |
319 | 296k | return Status::OK(); |
320 | 296k | } |
321 | | |
322 | | Status VExprContext::execute_conjuncts(const VExprContextSPtrs& conjuncts, const Block* block, |
323 | 391 | ColumnUInt8& null_map, IColumn::Filter& filter) { |
324 | 391 | const auto& rows = block->rows(); |
325 | 391 | if (rows == 0) { |
326 | 0 | return Status::OK(); |
327 | 0 | } |
328 | 391 | if (null_map.size() != rows) { |
329 | 0 | return Status::InternalError("null_map.size()!=rows, null_map.size()={}, rows={}", |
330 | 0 | null_map.size(), rows); |
331 | 0 | } |
332 | | |
333 | 391 | auto* final_null_map = null_map.get_data().data(); |
334 | 391 | auto* final_filter_ptr = filter.data(); |
335 | | |
336 | 391 | for (const auto& conjunct : conjuncts) { |
337 | 84 | ColumnPtr result_column; |
338 | 84 | RETURN_IF_ERROR(conjunct->execute(block, result_column)); |
339 | 84 | auto [filter_column, is_const] = unpack_if_const(result_column); |
340 | 84 | const auto* nullable_column = assert_cast<const ColumnNullable*>(filter_column.get()); |
341 | 84 | if (!is_const) { |
342 | 68 | const ColumnPtr& nested_column = nullable_column->get_nested_column_ptr(); |
343 | 68 | const IColumn::Filter& result = |
344 | 68 | assert_cast<const ColumnUInt8&>(*nested_column).get_data(); |
345 | 68 | const auto* __restrict filter_data = result.data(); |
346 | 68 | const auto* __restrict null_map_data = nullable_column->get_null_map_data().data(); |
347 | 68 | DCHECK_EQ(rows, nullable_column->size()); |
348 | | |
349 | 769 | for (size_t i = 0; i != rows; ++i) { |
350 | | // null and null => null |
351 | | // null and true => null |
352 | | // null and false => false |
353 | 701 | final_null_map[i] = (final_null_map[i] & (null_map_data[i] | filter_data[i])) | |
354 | 701 | (null_map_data[i] & (final_null_map[i] | final_filter_ptr[i])); |
355 | 701 | final_filter_ptr[i] = final_filter_ptr[i] & filter_data[i]; |
356 | 701 | } |
357 | 68 | } else { |
358 | 16 | bool filter_data = nullable_column->get_bool(0); |
359 | 16 | bool null_map_data = nullable_column->is_null_at(0); |
360 | 90 | for (size_t i = 0; i != rows; ++i) { |
361 | | // null and null => null |
362 | | // null and true => null |
363 | | // null and false => false |
364 | 74 | final_null_map[i] = (final_null_map[i] & (null_map_data | filter_data)) | |
365 | 74 | (null_map_data & (final_null_map[i] | final_filter_ptr[i])); |
366 | 74 | final_filter_ptr[i] = final_filter_ptr[i] & filter_data; |
367 | 74 | } |
368 | 16 | } |
369 | 84 | } |
370 | 391 | return Status::OK(); |
371 | 391 | } |
372 | | |
373 | | // TODO Performance Optimization |
374 | | // need exception safety |
375 | | Status VExprContext::execute_conjuncts_and_filter_block(const VExprContextSPtrs& ctxs, Block* block, |
376 | | std::vector<uint32_t>& columns_to_filter, |
377 | 106k | int column_to_keep) { |
378 | 106k | IColumn::Filter result_filter(block->rows(), 1); |
379 | 106k | bool can_filter_all; |
380 | | |
381 | 106k | _reset_memory_usage(ctxs); |
382 | | |
383 | 106k | RETURN_IF_ERROR( |
384 | 106k | execute_conjuncts(ctxs, nullptr, false, block, &result_filter, &can_filter_all)); |
385 | | |
386 | | // Accumulate the usage of `result_filter` into the first context. |
387 | 106k | if (!ctxs.empty()) { |
388 | 106k | ctxs[0]->_memory_usage += result_filter.allocated_bytes(); |
389 | 106k | } |
390 | 106k | if (can_filter_all) { |
391 | 55.2k | for (auto& col : columns_to_filter) { |
392 | 55.2k | auto& column = block->get_by_position(col).column; |
393 | 55.2k | if (column->is_exclusive()) { |
394 | 47.7k | column->assert_mutable()->clear(); |
395 | 47.7k | } else { |
396 | 7.45k | column = column->clone_empty(); |
397 | 7.45k | } |
398 | 55.2k | } |
399 | 92.1k | } else { |
400 | 92.1k | try { |
401 | 92.1k | Block::filter_block_internal(block, columns_to_filter, result_filter); |
402 | 92.1k | } catch (const Exception& e) { |
403 | 0 | std::string str; |
404 | 0 | for (auto ctx : ctxs) { |
405 | 0 | if (str.length()) { |
406 | 0 | str += ","; |
407 | 0 | } |
408 | 0 | str += ctx->root()->debug_string(); |
409 | 0 | } |
410 | |
|
411 | 0 | return Status::InternalError( |
412 | 0 | "filter_block_internal meet exception, exprs=[{}], exception={}", str, |
413 | 0 | e.what()); |
414 | 0 | } |
415 | 92.1k | } |
416 | 106k | Block::erase_useless_column(block, column_to_keep); |
417 | 106k | return Status::OK(); |
418 | 106k | } |
419 | | |
420 | | Status VExprContext::execute_conjuncts_and_filter_block(const VExprContextSPtrs& ctxs, Block* block, |
421 | | std::vector<uint32_t>& columns_to_filter, |
422 | | int column_to_keep, |
423 | 1.12k | IColumn::Filter& filter) { |
424 | 1.12k | _reset_memory_usage(ctxs); |
425 | 1.12k | filter.resize_fill(block->rows(), 1); |
426 | 1.12k | bool can_filter_all; |
427 | 1.12k | RETURN_IF_ERROR(execute_conjuncts(ctxs, nullptr, false, block, &filter, &can_filter_all)); |
428 | | |
429 | | // Accumulate the usage of `result_filter` into the first context. |
430 | 1.12k | if (!ctxs.empty()) { |
431 | 982 | ctxs[0]->_memory_usage += filter.allocated_bytes(); |
432 | 982 | } |
433 | 1.12k | if (can_filter_all) { |
434 | 330 | for (auto& col : columns_to_filter) { |
435 | 330 | auto& column = block->get_by_position(col).column; |
436 | 330 | if (column->is_exclusive()) { |
437 | 330 | column->assert_mutable()->clear(); |
438 | 330 | } else { |
439 | 0 | column = column->clone_empty(); |
440 | 0 | } |
441 | 330 | } |
442 | 867 | } else { |
443 | 867 | RETURN_IF_CATCH_EXCEPTION(Block::filter_block_internal(block, columns_to_filter, filter)); |
444 | 867 | } |
445 | | |
446 | 1.12k | Block::erase_useless_column(block, column_to_keep); |
447 | 1.12k | return Status::OK(); |
448 | 1.12k | } |
449 | | |
450 | | // do_projection: for some query(e.g. in MultiCastDataStreamerSourceOperator::get_block()), |
451 | | // output_vexpr_ctxs will output the same column more than once, and if the output_block |
452 | | // is mem-reused later, it will trigger DCHECK_EQ(d.column->use_count(), 1) failure when |
453 | | // doing Block::clear_column_data, set do_projection to true to copy the column data to |
454 | | // avoid this problem. |
455 | | Status VExprContext::get_output_block_after_execute_exprs( |
456 | | const VExprContextSPtrs& output_vexpr_ctxs, const Block& input_block, Block* output_block, |
457 | 251k | bool do_projection) { |
458 | 251k | auto rows = input_block.rows(); |
459 | 251k | ColumnsWithTypeAndName result_columns; |
460 | 251k | _reset_memory_usage(output_vexpr_ctxs); |
461 | | |
462 | 1.18M | for (const auto& vexpr_ctx : output_vexpr_ctxs) { |
463 | 1.18M | ColumnPtr result_column; |
464 | 1.18M | RETURN_IF_ERROR(vexpr_ctx->execute(&input_block, result_column)); |
465 | | |
466 | 1.18M | auto type = vexpr_ctx->execute_type(&input_block); |
467 | 1.18M | const auto& name = vexpr_ctx->expr_name(); |
468 | | |
469 | 1.18M | vexpr_ctx->_memory_usage += result_column->allocated_bytes(); |
470 | 1.18M | if (do_projection) { |
471 | 35.7k | result_columns.emplace_back(result_column->clone_resized(rows), type, name); |
472 | | |
473 | 1.15M | } else { |
474 | 1.15M | result_columns.emplace_back(result_column, type, name); |
475 | 1.15M | } |
476 | 1.18M | } |
477 | 251k | *output_block = {result_columns}; |
478 | 251k | return Status::OK(); |
479 | 251k | } |
480 | | |
481 | 358k | void VExprContext::_reset_memory_usage(const VExprContextSPtrs& contexts) { |
482 | 358k | std::for_each(contexts.begin(), contexts.end(), |
483 | 1.37M | [](auto&& context) { context->_memory_usage = 0; }); |
484 | 358k | } |
485 | | |
486 | 22.8k | void VExprContext::prepare_ann_range_search(const doris::VectorSearchUserParams& params) { |
487 | 22.8k | if (_root == nullptr) { |
488 | 0 | return; |
489 | 0 | } |
490 | | |
491 | 22.8k | _root->prepare_ann_range_search(params, _ann_range_search_runtime, _suitable_for_ann_index); |
492 | 18.4E | VLOG_DEBUG << fmt::format("Prepare ann range search result {}, _suitable_for_ann_index {}", |
493 | 18.4E | this->_ann_range_search_runtime.to_string(), |
494 | 18.4E | this->_suitable_for_ann_index); |
495 | 22.8k | return; |
496 | 22.8k | } |
497 | | |
498 | | Status VExprContext::evaluate_ann_range_search( |
499 | | const std::vector<std::unique_ptr<segment_v2::IndexIterator>>& cid_to_index_iterators, |
500 | | const std::vector<ColumnId>& idx_to_cid, |
501 | | const std::vector<std::unique_ptr<segment_v2::ColumnIterator>>& column_iterators, |
502 | | const std::unordered_map<VExprContext*, std::unordered_map<ColumnId, VExpr*>>& |
503 | | common_expr_to_slotref_map, |
504 | | size_t rows_of_segment, roaring::Roaring& row_bitmap, |
505 | | segment_v2::AnnIndexStats& ann_index_stats, bool enable_result_cache, |
506 | 17.9k | bool* ann_range_search_executed) { |
507 | 17.9k | if (ann_range_search_executed != nullptr) { |
508 | 17.9k | *ann_range_search_executed = false; |
509 | 17.9k | } |
510 | 17.9k | if (_root == nullptr) { |
511 | 0 | return Status::OK(); |
512 | 0 | } |
513 | | |
514 | 17.9k | AnnRangeSearchEvaluationResult evaluation_result; |
515 | 17.9k | RETURN_IF_ERROR(_root->evaluate_ann_range_search( |
516 | 17.9k | _ann_range_search_runtime, cid_to_index_iterators, idx_to_cid, column_iterators, |
517 | 17.9k | rows_of_segment, row_bitmap, ann_index_stats, enable_result_cache, evaluation_result)); |
518 | | |
519 | 17.9k | if (!evaluation_result.executed) { |
520 | 17.9k | return Status::OK(); |
521 | 17.9k | } |
522 | 49 | if (ann_range_search_executed != nullptr) { |
523 | 27 | *ann_range_search_executed = true; |
524 | 27 | } |
525 | | |
526 | 49 | DCHECK(_index_context != nullptr); |
527 | 49 | _index_context->set_index_result_for_expr( |
528 | 49 | _root.get(), |
529 | 49 | segment_v2::InvertedIndexResultBitmap(std::make_shared<roaring::Roaring>(row_bitmap), |
530 | 49 | std::make_shared<roaring::Roaring>())); |
531 | | |
532 | 49 | if (!evaluation_result.dist_fulfilled) { |
533 | | // Do not perform index scan in this case. |
534 | 2 | return Status::OK(); |
535 | 2 | } |
536 | | |
537 | 49 | DCHECK_LT(_ann_range_search_runtime.src_col_idx, idx_to_cid.size()); |
538 | 47 | const auto src_col_idx = cast_set<int>(_ann_range_search_runtime.src_col_idx); |
539 | 47 | const auto src_col_key = cast_set<ColumnId>(_ann_range_search_runtime.src_col_idx); |
540 | 47 | auto slot_ref_map_it = common_expr_to_slotref_map.find(this); |
541 | 47 | if (slot_ref_map_it == common_expr_to_slotref_map.end()) { |
542 | 1 | return Status::OK(); |
543 | 1 | } |
544 | 46 | auto& slot_ref_map = slot_ref_map_it->second; |
545 | 46 | auto slot_ref_it = slot_ref_map.find(src_col_key); |
546 | 46 | if (slot_ref_it == slot_ref_map.end()) { |
547 | 0 | return Status::OK(); |
548 | 0 | } |
549 | 46 | const VExpr* slot_ref_expr_addr = slot_ref_it->second; |
550 | 46 | _index_context->set_true_for_index_status(slot_ref_expr_addr, src_col_idx); |
551 | | |
552 | 46 | VLOG_DEBUG << fmt::format( |
553 | 22 | "Evaluate ann range search for expr {}, src_col_idx {}, cid {}, row_bitmap " |
554 | 22 | "cardinality {}", |
555 | 22 | _root->debug_string(), src_col_idx, idx_to_cid[_ann_range_search_runtime.src_col_idx], |
556 | 22 | row_bitmap.cardinality()); |
557 | 46 | return Status::OK(); |
558 | 46 | } |
559 | | |
560 | 567k | uint64_t VExprContext::get_digest(uint64_t seed) const { |
561 | 567k | return _root->get_digest(seed); |
562 | 567k | } |
563 | | |
564 | 1.21M | double VExprContext::execute_cost() const { |
565 | 1.21M | if (_root == nullptr) { |
566 | | // When there is no expression root, treat the cost as a base value. |
567 | | // This avoids null dereferences while keeping a deterministic cost. |
568 | 0 | return 0.0; |
569 | 0 | } |
570 | 1.21M | return _root->execute_cost(); |
571 | 1.21M | } |
572 | | |
573 | | } // namespace doris |