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 | 19.2M | VExprContext::VExprContext(VExprSPtr expr) : _root(std::move(expr)) {} |
52 | | |
53 | 19.1M | 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 | 19.1M | if (!_prepared || !_opened) { |
58 | 238k | return; |
59 | 238k | } |
60 | 18.9M | try { |
61 | 18.9M | close(); |
62 | 18.9M | } catch (const Exception& e) { |
63 | 0 | LOG(WARNING) << "Exception occurs when expr context deconstruct: " << e.to_string(); |
64 | 0 | } |
65 | 18.9M | } |
66 | | |
67 | 21.3k | LambdaExecutionContext& VExprContext::lambda_execution_context() { |
68 | 21.3k | if (!_lambda_execution_context) { |
69 | 2.82k | _lambda_execution_context = std::make_unique<LambdaExecutionContext>(); |
70 | 2.82k | } |
71 | 21.3k | return *_lambda_execution_context; |
72 | 21.3k | } |
73 | | |
74 | 1.12M | Status VExprContext::execute(Block* block, int* result_column_id) { |
75 | 1.12M | Status st; |
76 | 1.12M | RETURN_IF_CATCH_EXCEPTION({ |
77 | 1.12M | st = _root->execute(this, block, result_column_id); |
78 | 1.12M | _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 | 1.12M | if (st.ok() && _last_result_column_id != -1) { |
81 | 1.12M | block->get_by_position(*result_column_id).column->sanity_check(); |
82 | 1.12M | RETURN_IF_ERROR( |
83 | 1.12M | block->get_by_position(*result_column_id).check_type_and_column_match()); |
84 | 1.12M | } |
85 | 1.12M | }); |
86 | 1.12M | return st; |
87 | 1.12M | } |
88 | | |
89 | 1.98M | Status VExprContext::execute(const Block* block, ColumnPtr& result_column) { |
90 | 1.98M | Status st; |
91 | 1.98M | RETURN_IF_CATCH_EXCEPTION( |
92 | 1.98M | { st = _root->execute_column(this, block, nullptr, block->rows(), result_column); }); |
93 | 1.98M | return st; |
94 | 1.98M | } |
95 | | |
96 | 28.3k | Status VExprContext::execute(const Block* block, ColumnWithTypeAndName& result_data) { |
97 | 28.3k | Status st; |
98 | 28.3k | ColumnPtr result_column; |
99 | 28.3k | RETURN_IF_CATCH_EXCEPTION( |
100 | 28.3k | { st = _root->execute_column(this, block, nullptr, block->rows(), result_column); }); |
101 | 28.3k | RETURN_IF_ERROR(st); |
102 | 28.3k | result_data.column = result_column; |
103 | 28.3k | result_data.type = execute_type(block); |
104 | 28.3k | result_data.name = _root->expr_name(); |
105 | 28.3k | return Status::OK(); |
106 | 28.3k | } |
107 | | |
108 | 824k | DataTypePtr VExprContext::execute_type(const Block* block) { |
109 | 824k | return _root->execute_type(block); |
110 | 824k | } |
111 | | |
112 | 1.03M | Status VExprContext::execute_const_expr(ColumnWithTypeAndName& result) { |
113 | 1.03M | Status st; |
114 | 1.03M | RETURN_IF_CATCH_EXCEPTION( |
115 | 1.03M | { st = _root->execute_column(this, nullptr, nullptr, 1, result.column); }); |
116 | 1.03M | RETURN_IF_ERROR(st); |
117 | 1.03M | result.type = _root->execute_type(nullptr); |
118 | 1.03M | result.name = _root->expr_name(); |
119 | 1.03M | return Status::OK(); |
120 | 1.03M | } |
121 | | |
122 | 796k | [[nodiscard]] const std::string& VExprContext::expr_name() const { |
123 | 796k | return _root->expr_name(); |
124 | 796k | } |
125 | | |
126 | 0 | bool VExprContext::is_blockable() const { |
127 | 0 | return _root->is_blockable(); |
128 | 0 | } |
129 | | |
130 | 5.06M | Status VExprContext::prepare(RuntimeState* state, const RowDescriptor& row_desc) { |
131 | 5.06M | _prepared = true; |
132 | 5.06M | Status st; |
133 | 5.06M | RETURN_IF_CATCH_EXCEPTION({ st = _root->prepare(state, row_desc, this); }); |
134 | 5.07M | return st; |
135 | 5.06M | } |
136 | | |
137 | 5.07M | Status VExprContext::open(RuntimeState* state) { |
138 | 5.07M | DCHECK(_prepared); |
139 | 5.07M | if (_opened) { |
140 | 47 | return Status::OK(); |
141 | 47 | } |
142 | 5.07M | _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 | 5.07M | FunctionContext::FunctionStateScope scope = |
146 | 5.07M | _is_clone ? FunctionContext::THREAD_LOCAL : FunctionContext::FRAGMENT_LOCAL; |
147 | 5.07M | Status st; |
148 | 5.07M | RETURN_IF_CATCH_EXCEPTION({ st = _root->open(state, this, scope); }); |
149 | 5.06M | return st; |
150 | 5.07M | } |
151 | | |
152 | 18.9M | void VExprContext::close() { |
153 | | // Sometimes expr context may not have a root, then it need not call close |
154 | 18.9M | if (_root == nullptr) { |
155 | 0 | return; |
156 | 0 | } |
157 | 18.9M | FunctionContext::FunctionStateScope scope = |
158 | 18.9M | _is_clone ? FunctionContext::THREAD_LOCAL : FunctionContext::FRAGMENT_LOCAL; |
159 | 18.9M | _root->close(this, scope); |
160 | 18.9M | } |
161 | | |
162 | 13.8M | Status VExprContext::clone(RuntimeState* state, VExprContextSPtr& new_ctx) { |
163 | 18.4E | DCHECK(_prepared) << "expr context not prepared"; |
164 | 13.8M | DCHECK(_opened); |
165 | 13.8M | DCHECK(new_ctx.get() == nullptr); |
166 | | |
167 | 13.8M | new_ctx = std::make_shared<VExprContext>(_root); |
168 | 13.8M | for (auto& _fn_context : _fn_contexts) { |
169 | 1.12M | new_ctx->_fn_contexts.push_back(_fn_context->clone()); |
170 | 1.12M | } |
171 | | |
172 | 13.8M | new_ctx->_is_clone = true; |
173 | 13.8M | new_ctx->_prepared = true; |
174 | 13.8M | 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 | 13.8M | new_ctx->_ann_range_search_runtime = this->_ann_range_search_runtime; |
178 | | |
179 | 13.8M | return _root->open(state, new_ctx.get(), FunctionContext::THREAD_LOCAL); |
180 | 13.8M | } |
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 | 609k | const std::vector<DataTypePtr>& arg_types) { |
190 | 609k | _fn_contexts.push_back(FunctionContext::create_context(state, return_type, arg_types)); |
191 | 609k | _fn_contexts.back()->set_check_overflow_for_decimal(state->check_overflow_for_decimal()); |
192 | 609k | _fn_contexts.back()->set_enable_strict_mode(state->enable_strict_mode()); |
193 | 609k | return static_cast<int>(_fn_contexts.size()) - 1; |
194 | 609k | } |
195 | | |
196 | 18.0k | Status VExprContext::evaluate_inverted_index(uint32_t segment_num_rows) { |
197 | 18.0k | Status st; |
198 | 18.0k | RETURN_IF_CATCH_EXCEPTION({ st = _root->evaluate_inverted_index(this, segment_num_rows); }); |
199 | 18.0k | return st; |
200 | 18.0k | } |
201 | | |
202 | | ZoneMapFilterResult VExprContext::evaluate_zonemap_filter(const VExprContextSPtrs& conjuncts, |
203 | 21.2k | const ZoneMapEvalContext& ctx) { |
204 | 23.3k | for (const auto& conjunct : conjuncts) { |
205 | 23.3k | DORIS_CHECK(conjunct != nullptr); |
206 | 23.3k | const auto& root = conjunct->root(); |
207 | 23.3k | DORIS_CHECK(root != nullptr); |
208 | 23.3k | if (!root->can_evaluate_zonemap_filter()) { |
209 | 17.0k | continue; |
210 | 17.0k | } |
211 | 6.30k | if (root->evaluate_zonemap_filter(ctx) == ZoneMapFilterResult::kNoMatch) { |
212 | 487 | return ZoneMapFilterResult::kNoMatch; |
213 | 487 | } |
214 | 6.30k | } |
215 | 20.8k | return ZoneMapFilterResult::kMayMatch; |
216 | 21.2k | } |
217 | | |
218 | | ZoneMapFilterResult VExprContext::evaluate_dictionary_filter(const VExprContextSPtrs& conjuncts, |
219 | 80 | const DictionaryEvalContext& ctx) { |
220 | 80 | for (const auto& conjunct : conjuncts) { |
221 | 80 | DORIS_CHECK(conjunct != nullptr); |
222 | 80 | const auto& root = conjunct->root(); |
223 | 80 | DORIS_CHECK(root != nullptr); |
224 | 80 | if (!root->can_evaluate_dictionary_filter()) { |
225 | 0 | continue; |
226 | 0 | } |
227 | 80 | if (root->evaluate_dictionary_filter(ctx) == ZoneMapFilterResult::kNoMatch) { |
228 | 47 | return ZoneMapFilterResult::kNoMatch; |
229 | 47 | } |
230 | 80 | } |
231 | 33 | return ZoneMapFilterResult::kMayMatch; |
232 | 80 | } |
233 | | |
234 | | ZoneMapFilterResult VExprContext::evaluate_bloom_filter(const VExprContextSPtrs& conjuncts, |
235 | 17 | const BloomFilterEvalContext& ctx) { |
236 | 17 | for (const auto& conjunct : conjuncts) { |
237 | 17 | DORIS_CHECK(conjunct != nullptr); |
238 | 17 | const auto& root = conjunct->root(); |
239 | 17 | DORIS_CHECK(root != nullptr); |
240 | 17 | if (!root->can_evaluate_bloom_filter()) { |
241 | 0 | continue; |
242 | 0 | } |
243 | 17 | if (root->evaluate_bloom_filter(ctx) == ZoneMapFilterResult::kNoMatch) { |
244 | 8 | return ZoneMapFilterResult::kNoMatch; |
245 | 8 | } |
246 | 17 | } |
247 | 9 | return ZoneMapFilterResult::kMayMatch; |
248 | 17 | } |
249 | | |
250 | 17.7k | bool VExprContext::all_expr_inverted_index_evaluated() { |
251 | 17.7k | return _index_context->has_index_result_for_expr(_root.get()); |
252 | 17.7k | } |
253 | | |
254 | 49 | Status VExprContext::filter_block(VExprContext* vexpr_ctx, Block* block) { |
255 | 49 | if (vexpr_ctx == nullptr || block->rows() == 0) { |
256 | 0 | return Status::OK(); |
257 | 0 | } |
258 | 49 | ColumnPtr filter_column; |
259 | 49 | RETURN_IF_ERROR(vexpr_ctx->execute(block, filter_column)); |
260 | 49 | size_t filter_column_id = block->columns(); |
261 | 49 | block->insert({filter_column, vexpr_ctx->execute_type(block), "filter_column"}); |
262 | 49 | vexpr_ctx->_memory_usage = filter_column->allocated_bytes(); |
263 | 49 | return Block::filter_block(block, filter_column_id, filter_column_id); |
264 | 49 | } |
265 | | |
266 | | Status VExprContext::filter_block(const VExprContextSPtrs& expr_contexts, Block* block, |
267 | 1.46M | size_t column_to_keep) { |
268 | 1.46M | if (expr_contexts.empty() || block->rows() == 0) { |
269 | 1.44M | return Status::OK(); |
270 | 1.44M | } |
271 | | |
272 | 17.2k | ColumnNumbers columns_to_filter(column_to_keep); |
273 | 17.2k | std::iota(columns_to_filter.begin(), columns_to_filter.end(), 0); |
274 | | |
275 | 17.2k | return execute_conjuncts_and_filter_block(expr_contexts, block, columns_to_filter, |
276 | 17.2k | static_cast<int>(column_to_keep)); |
277 | 1.46M | } |
278 | | |
279 | | Status VExprContext::execute_conjuncts(const VExprContextSPtrs& ctxs, |
280 | | const std::vector<IColumn::Filter*>* filters, Block* block, |
281 | 2.69k | IColumn::Filter* result_filter, bool* can_filter_all) { |
282 | 2.69k | return execute_conjuncts(ctxs, filters, false, block, result_filter, can_filter_all); |
283 | 2.69k | } |
284 | | |
285 | | Status VExprContext::execute_filter(const Block* block, uint8_t* __restrict result_filter_data, |
286 | 170k | size_t rows, bool accept_null, bool* can_filter_all) { |
287 | 170k | return _root->execute_filter(this, block, result_filter_data, rows, accept_null, |
288 | 170k | can_filter_all); |
289 | 170k | } |
290 | | |
291 | | Status VExprContext::execute_conjuncts(const VExprContextSPtrs& ctxs, |
292 | | const std::vector<IColumn::Filter*>* filters, |
293 | | bool accept_null, const Block* block, |
294 | 364k | IColumn::Filter* result_filter, bool* can_filter_all) { |
295 | 364k | size_t rows = block->rows(); |
296 | 364k | DCHECK_EQ(result_filter->size(), rows); |
297 | 364k | *can_filter_all = false; |
298 | 364k | auto* __restrict result_filter_data = result_filter->data(); |
299 | 364k | for (const auto& ctx : ctxs) { |
300 | 154k | RETURN_IF_ERROR( |
301 | 154k | ctx->execute_filter(block, result_filter_data, rows, accept_null, can_filter_all)); |
302 | 154k | if (*can_filter_all) { |
303 | 30.6k | return Status::OK(); |
304 | 30.6k | } |
305 | 154k | } |
306 | 334k | if (filters != nullptr) { |
307 | 46 | for (auto* filter : *filters) { |
308 | 0 | auto* __restrict filter_data = filter->data(); |
309 | 0 | const size_t size = filter->size(); |
310 | 0 | for (size_t i = 0; i < size; ++i) { |
311 | 0 | result_filter_data[i] &= filter_data[i]; |
312 | 0 | } |
313 | 0 | if (memchr(result_filter_data, 0x1, size) == nullptr) { |
314 | 0 | *can_filter_all = true; |
315 | 0 | return Status::OK(); |
316 | 0 | } |
317 | 0 | } |
318 | 46 | } |
319 | 334k | return Status::OK(); |
320 | 334k | } |
321 | | |
322 | | Status VExprContext::execute_conjuncts(const VExprContextSPtrs& conjuncts, const Block* block, |
323 | 371 | ColumnUInt8& null_map, IColumn::Filter& filter) { |
324 | 371 | const auto& rows = block->rows(); |
325 | 371 | if (rows == 0) { |
326 | 0 | return Status::OK(); |
327 | 0 | } |
328 | 371 | 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 | 371 | auto* final_null_map = null_map.get_data().data(); |
334 | 371 | auto* final_filter_ptr = filter.data(); |
335 | | |
336 | 371 | for (const auto& conjunct : conjuncts) { |
337 | 72 | ColumnPtr result_column; |
338 | 72 | RETURN_IF_ERROR(conjunct->execute(block, result_column)); |
339 | 72 | auto [filter_column, is_const] = unpack_if_const(result_column); |
340 | 72 | const auto* nullable_column = assert_cast<const ColumnNullable*>(filter_column.get()); |
341 | 72 | if (!is_const) { |
342 | 60 | const ColumnPtr& nested_column = nullable_column->get_nested_column_ptr(); |
343 | 60 | const IColumn::Filter& result = |
344 | 60 | assert_cast<const ColumnUInt8&>(*nested_column).get_data(); |
345 | 60 | const auto* __restrict filter_data = result.data(); |
346 | 60 | const auto* __restrict null_map_data = nullable_column->get_null_map_data().data(); |
347 | 60 | DCHECK_EQ(rows, nullable_column->size()); |
348 | | |
349 | 779 | for (size_t i = 0; i != rows; ++i) { |
350 | | // null and null => null |
351 | | // null and true => null |
352 | | // null and false => false |
353 | 719 | final_null_map[i] = (final_null_map[i] & (null_map_data[i] | filter_data[i])) | |
354 | 719 | (null_map_data[i] & (final_null_map[i] | final_filter_ptr[i])); |
355 | 719 | final_filter_ptr[i] = final_filter_ptr[i] & filter_data[i]; |
356 | 719 | } |
357 | 60 | } else { |
358 | 12 | bool filter_data = nullable_column->get_bool(0); |
359 | 12 | bool null_map_data = nullable_column->is_null_at(0); |
360 | 68 | for (size_t i = 0; i != rows; ++i) { |
361 | | // null and null => null |
362 | | // null and true => null |
363 | | // null and false => false |
364 | 56 | final_null_map[i] = (final_null_map[i] & (null_map_data | filter_data)) | |
365 | 56 | (null_map_data & (final_null_map[i] | final_filter_ptr[i])); |
366 | 56 | final_filter_ptr[i] = final_filter_ptr[i] & filter_data; |
367 | 56 | } |
368 | 12 | } |
369 | 72 | } |
370 | 371 | return Status::OK(); |
371 | 371 | } |
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 | 19.2k | int column_to_keep) { |
378 | 19.2k | IColumn::Filter result_filter(block->rows(), 1); |
379 | 19.2k | bool can_filter_all; |
380 | | |
381 | 19.2k | _reset_memory_usage(ctxs); |
382 | | |
383 | 19.2k | RETURN_IF_ERROR( |
384 | 19.2k | execute_conjuncts(ctxs, nullptr, false, block, &result_filter, &can_filter_all)); |
385 | | |
386 | | // Accumulate the usage of `result_filter` into the first context. |
387 | 19.2k | if (!ctxs.empty()) { |
388 | 19.2k | ctxs[0]->_memory_usage += result_filter.allocated_bytes(); |
389 | 19.2k | } |
390 | 19.2k | if (can_filter_all) { |
391 | 31.9k | for (auto& col : columns_to_filter) { |
392 | 31.9k | auto& column = block->get_by_position(col).column; |
393 | 31.9k | if (column->is_exclusive()) { |
394 | 30.1k | column->assert_mutable()->clear(); |
395 | 30.1k | } else { |
396 | 1.80k | column = column->clone_empty(); |
397 | 1.80k | } |
398 | 31.9k | } |
399 | 11.5k | } else { |
400 | 11.5k | try { |
401 | 11.5k | Block::filter_block_internal(block, columns_to_filter, result_filter); |
402 | 11.5k | } 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 | 11.5k | } |
416 | 19.2k | Block::erase_useless_column(block, column_to_keep); |
417 | 19.2k | return Status::OK(); |
418 | 19.2k | } |
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 | 107 | IColumn::Filter& filter) { |
424 | 107 | _reset_memory_usage(ctxs); |
425 | 107 | filter.resize_fill(block->rows(), 1); |
426 | 107 | bool can_filter_all; |
427 | 107 | 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 | 107 | if (!ctxs.empty()) { |
431 | 107 | ctxs[0]->_memory_usage += filter.allocated_bytes(); |
432 | 107 | } |
433 | 107 | if (can_filter_all) { |
434 | 106 | for (auto& col : columns_to_filter) { |
435 | 106 | auto& column = block->get_by_position(col).column; |
436 | 106 | if (column->is_exclusive()) { |
437 | 106 | column->assert_mutable()->clear(); |
438 | 106 | } else { |
439 | 0 | column = column->clone_empty(); |
440 | 0 | } |
441 | 106 | } |
442 | 76 | } else { |
443 | 76 | RETURN_IF_CATCH_EXCEPTION(Block::filter_block_internal(block, columns_to_filter, filter)); |
444 | 76 | } |
445 | | |
446 | 107 | Block::erase_useless_column(block, column_to_keep); |
447 | 107 | return Status::OK(); |
448 | 107 | } |
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 | 178k | bool do_projection) { |
458 | 178k | auto rows = input_block.rows(); |
459 | 178k | ColumnsWithTypeAndName result_columns; |
460 | 178k | _reset_memory_usage(output_vexpr_ctxs); |
461 | | |
462 | 796k | for (const auto& vexpr_ctx : output_vexpr_ctxs) { |
463 | 796k | ColumnPtr result_column; |
464 | 796k | RETURN_IF_ERROR(vexpr_ctx->execute(&input_block, result_column)); |
465 | | |
466 | 796k | auto type = vexpr_ctx->execute_type(&input_block); |
467 | 796k | const auto& name = vexpr_ctx->expr_name(); |
468 | | |
469 | 796k | vexpr_ctx->_memory_usage += result_column->allocated_bytes(); |
470 | 796k | if (do_projection) { |
471 | 10.3k | result_columns.emplace_back(result_column->clone_resized(rows), type, name); |
472 | | |
473 | 785k | } else { |
474 | 785k | result_columns.emplace_back(result_column, type, name); |
475 | 785k | } |
476 | 796k | } |
477 | 178k | *output_block = {result_columns}; |
478 | 178k | return Status::OK(); |
479 | 178k | } |
480 | | |
481 | 198k | void VExprContext::_reset_memory_usage(const VExprContextSPtrs& contexts) { |
482 | 198k | std::for_each(contexts.begin(), contexts.end(), |
483 | 820k | [](auto&& context) { context->_memory_usage = 0; }); |
484 | 198k | } |
485 | | |
486 | 22.7k | void VExprContext::prepare_ann_range_search(const doris::VectorSearchUserParams& params) { |
487 | 22.7k | if (_root == nullptr) { |
488 | 0 | return; |
489 | 0 | } |
490 | | |
491 | 22.7k | _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.7k | return; |
496 | 22.7k | } |
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.7k | bool* ann_range_search_executed) { |
507 | 17.7k | if (ann_range_search_executed != nullptr) { |
508 | 17.7k | *ann_range_search_executed = false; |
509 | 17.7k | } |
510 | 17.7k | if (_root == nullptr) { |
511 | 0 | return Status::OK(); |
512 | 0 | } |
513 | | |
514 | 17.7k | AnnRangeSearchEvaluationResult evaluation_result; |
515 | 17.7k | RETURN_IF_ERROR(_root->evaluate_ann_range_search( |
516 | 17.7k | _ann_range_search_runtime, cid_to_index_iterators, idx_to_cid, column_iterators, |
517 | 17.7k | rows_of_segment, row_bitmap, ann_index_stats, enable_result_cache, evaluation_result)); |
518 | | |
519 | 17.7k | if (!evaluation_result.executed) { |
520 | 17.7k | return Status::OK(); |
521 | 17.7k | } |
522 | 38 | if (ann_range_search_executed != nullptr) { |
523 | 27 | *ann_range_search_executed = true; |
524 | 27 | } |
525 | | |
526 | 38 | DCHECK(_index_context != nullptr); |
527 | 38 | _index_context->set_index_result_for_expr( |
528 | 38 | _root.get(), |
529 | 38 | segment_v2::InvertedIndexResultBitmap(std::make_shared<roaring::Roaring>(row_bitmap), |
530 | 38 | std::make_shared<roaring::Roaring>())); |
531 | | |
532 | 38 | if (!evaluation_result.dist_fulfilled) { |
533 | | // Do not perform index scan in this case. |
534 | 2 | return Status::OK(); |
535 | 2 | } |
536 | | |
537 | 38 | DCHECK_LT(_ann_range_search_runtime.src_col_idx, idx_to_cid.size()); |
538 | 36 | const auto src_col_idx = cast_set<int>(_ann_range_search_runtime.src_col_idx); |
539 | 36 | const auto src_col_key = cast_set<ColumnId>(_ann_range_search_runtime.src_col_idx); |
540 | 36 | auto slot_ref_map_it = common_expr_to_slotref_map.find(this); |
541 | 36 | if (slot_ref_map_it == common_expr_to_slotref_map.end()) { |
542 | 1 | return Status::OK(); |
543 | 1 | } |
544 | 35 | auto& slot_ref_map = slot_ref_map_it->second; |
545 | 35 | auto slot_ref_it = slot_ref_map.find(src_col_key); |
546 | 35 | if (slot_ref_it == slot_ref_map.end()) { |
547 | 0 | return Status::OK(); |
548 | 0 | } |
549 | 35 | const VExpr* slot_ref_expr_addr = slot_ref_it->second; |
550 | 35 | _index_context->set_true_for_index_status(slot_ref_expr_addr, src_col_idx); |
551 | | |
552 | 35 | VLOG_DEBUG << fmt::format( |
553 | 11 | "Evaluate ann range search for expr {}, src_col_idx {}, cid {}, row_bitmap " |
554 | 11 | "cardinality {}", |
555 | 11 | _root->debug_string(), src_col_idx, idx_to_cid[_ann_range_search_runtime.src_col_idx], |
556 | 11 | row_bitmap.cardinality()); |
557 | 35 | return Status::OK(); |
558 | 35 | } |
559 | | |
560 | 286k | uint64_t VExprContext::get_digest(uint64_t seed) const { |
561 | 286k | return _root->get_digest(seed); |
562 | 286k | } |
563 | | |
564 | 658k | double VExprContext::execute_cost() const { |
565 | 658k | 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 | 658k | return _root->execute_cost(); |
571 | 658k | } |
572 | | |
573 | | } // namespace doris |