Coverage Report

Created: 2026-06-28 02:25

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
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/status.h"
29
#include "core/block/column_numbers.h"
30
#include "core/block/column_with_type_and_name.h"
31
#include "core/block/columns_with_type_and_name.h"
32
#include "core/column/column.h"
33
#include "core/column/column_const.h"
34
#include "exec/common/util.hpp"
35
#include "exprs/function_context.h"
36
#include "exprs/lambda_function/lambda_execution_context.h"
37
#include "exprs/vexpr.h"
38
#include "runtime/runtime_state.h"
39
#include "runtime/thread_context.h"
40
#include "storage/olap_common.h"
41
#include "storage/segment/column_reader.h"
42
#include "util/simd/bits.h"
43
44
namespace doris {
45
class RowDescriptor;
46
} // namespace doris
47
48
namespace doris {
49
50
VExprContext::VExprContext(VExprSPtr expr)
51
655k
        : _root(std::move(expr)),
52
655k
          _lambda_execution_context(std::make_unique<LambdaExecutionContext>()) {}
53
54
655k
VExprContext::~VExprContext() {
55
    // In runtime filter, only create expr context to get expr root, will not call
56
    // prepare or open, so that it is not need to call close. And call close may core
57
    // because the function context in expr is not set.
58
655k
    if (!_prepared || !_opened) {
59
141k
        return;
60
141k
    }
61
513k
    try {
62
513k
        close();
63
513k
    } catch (const Exception& e) {
64
0
        LOG(WARNING) << "Exception occurs when expr context deconstruct: " << e.to_string();
65
0
    }
66
513k
}
67
68
57
LambdaExecutionContext& VExprContext::lambda_execution_context() {
69
57
    return *_lambda_execution_context;
70
57
}
71
72
189k
Status VExprContext::execute(Block* block, int* result_column_id) {
73
189k
    Status st;
74
189k
    RETURN_IF_CATCH_EXCEPTION({
75
189k
        st = _root->execute(this, block, result_column_id);
76
189k
        _last_result_column_id = *result_column_id;
77
        // We should first check the status, as some expressions might incorrectly set result_column_id, even if the st is not ok.
78
189k
        if (st.ok() && _last_result_column_id != -1) {
79
189k
            block->get_by_position(*result_column_id).column->sanity_check();
80
189k
            RETURN_IF_ERROR(
81
189k
                    block->get_by_position(*result_column_id).check_type_and_column_match());
82
189k
        }
83
189k
    });
84
189k
    return st;
85
189k
}
86
87
285
Status VExprContext::execute(const Block* block, ColumnPtr& result_column) {
88
285
    Status st;
89
285
    RETURN_IF_CATCH_EXCEPTION(
90
285
            { st = _root->execute_column(this, block, nullptr, block->rows(), result_column); });
91
285
    return st;
92
285
}
93
94
19
Status VExprContext::execute(const Block* block, ColumnWithTypeAndName& result_data) {
95
19
    Status st;
96
19
    ColumnPtr result_column;
97
19
    RETURN_IF_CATCH_EXCEPTION(
98
19
            { st = _root->execute_column(this, block, nullptr, block->rows(), result_column); });
99
19
    RETURN_IF_ERROR(st);
100
19
    result_data.column = result_column;
101
19
    result_data.type = execute_type(block);
102
19
    result_data.name = _root->expr_name();
103
19
    return Status::OK();
104
19
}
105
106
47
DataTypePtr VExprContext::execute_type(const Block* block) {
107
47
    return _root->execute_type(block);
108
47
}
109
110
30
Status VExprContext::execute_const_expr(ColumnWithTypeAndName& result) {
111
30
    Status st;
112
30
    RETURN_IF_CATCH_EXCEPTION(
113
30
            { st = _root->execute_column(this, nullptr, nullptr, 1, result.column); });
114
30
    RETURN_IF_ERROR(st);
115
30
    result.type = _root->execute_type(nullptr);
116
30
    result.name = _root->expr_name();
117
30
    return Status::OK();
118
30
}
119
120
50
[[nodiscard]] const std::string& VExprContext::expr_name() const {
121
50
    return _root->expr_name();
122
50
}
123
124
0
bool VExprContext::is_blockable() const {
125
0
    return _root->is_blockable();
126
0
}
127
128
258k
Status VExprContext::prepare(RuntimeState* state, const RowDescriptor& row_desc) {
129
258k
    _prepared = true;
130
258k
    Status st;
131
258k
    RETURN_IF_CATCH_EXCEPTION({ st = _root->prepare(state, row_desc, this); });
132
258k
    return st;
133
258k
}
134
135
258k
Status VExprContext::open(RuntimeState* state) {
136
258k
    DCHECK(_prepared);
137
258k
    if (_opened) {
138
13
        return Status::OK();
139
13
    }
140
258k
    _opened = true;
141
    // Fragment-local state is only initialized for original contexts. Clones inherit the
142
    // original's fragment state and only need to have thread-local state initialized.
143
258k
    FunctionContext::FunctionStateScope scope =
144
258k
            _is_clone ? FunctionContext::THREAD_LOCAL : FunctionContext::FRAGMENT_LOCAL;
145
258k
    Status st;
146
258k
    RETURN_IF_CATCH_EXCEPTION({ st = _root->open(state, this, scope); });
147
258k
    return st;
148
258k
}
149
150
513k
void VExprContext::close() {
151
    // Sometimes expr context may not have a root, then it need not call close
152
513k
    if (_root == nullptr) {
153
0
        return;
154
0
    }
155
513k
    FunctionContext::FunctionStateScope scope =
156
513k
            _is_clone ? FunctionContext::THREAD_LOCAL : FunctionContext::FRAGMENT_LOCAL;
157
513k
    _root->close(this, scope);
158
513k
}
159
160
254k
Status VExprContext::clone(RuntimeState* state, VExprContextSPtr& new_ctx) {
161
254k
    DCHECK(_prepared) << "expr context not prepared";
162
254k
    DCHECK(_opened);
163
254k
    DCHECK(new_ctx.get() == nullptr);
164
165
254k
    new_ctx = std::make_shared<VExprContext>(_root);
166
254k
    for (auto& _fn_context : _fn_contexts) {
167
43
        new_ctx->_fn_contexts.push_back(_fn_context->clone());
168
43
    }
169
170
254k
    new_ctx->_is_clone = true;
171
254k
    new_ctx->_prepared = true;
172
254k
    new_ctx->_opened = true;
173
    // segment_v2::AnnRangeSearchRuntime should be cloned as well.
174
    // The object of segment_v2::AnnRangeSearchRuntime is not shared by threads.
175
254k
    new_ctx->_ann_range_search_runtime = this->_ann_range_search_runtime;
176
177
254k
    return _root->open(state, new_ctx.get(), FunctionContext::THREAD_LOCAL);
178
254k
}
179
180
0
void VExprContext::clone_fn_contexts(VExprContext* other) {
181
0
    for (auto& _fn_context : _fn_contexts) {
182
0
        other->_fn_contexts.push_back(_fn_context->clone());
183
0
    }
184
0
}
185
186
int VExprContext::register_function_context(RuntimeState* state, const DataTypePtr& return_type,
187
120
                                            const std::vector<DataTypePtr>& arg_types) {
188
120
    _fn_contexts.push_back(FunctionContext::create_context(state, return_type, arg_types));
189
120
    _fn_contexts.back()->set_check_overflow_for_decimal(state->check_overflow_for_decimal());
190
120
    _fn_contexts.back()->set_enable_strict_mode(state->enable_strict_mode());
191
120
    return static_cast<int>(_fn_contexts.size()) - 1;
192
120
}
193
194
18
Status VExprContext::evaluate_inverted_index(uint32_t segment_num_rows) {
195
18
    Status st;
196
18
    RETURN_IF_CATCH_EXCEPTION({ st = _root->evaluate_inverted_index(this, segment_num_rows); });
197
18
    return st;
198
18
}
199
200
18
bool VExprContext::all_expr_inverted_index_evaluated() {
201
18
    return _index_context->has_index_result_for_expr(_root.get());
202
18
}
203
204
0
Status VExprContext::filter_block(VExprContext* vexpr_ctx, Block* block) {
205
0
    if (vexpr_ctx == nullptr || block->rows() == 0) {
206
0
        return Status::OK();
207
0
    }
208
0
    ColumnPtr filter_column;
209
0
    RETURN_IF_ERROR(vexpr_ctx->execute(block, filter_column));
210
0
    size_t filter_column_id = block->columns();
211
0
    block->insert({filter_column, vexpr_ctx->execute_type(block), "filter_column"});
212
0
    vexpr_ctx->_memory_usage = filter_column->allocated_bytes();
213
0
    return Block::filter_block(block, filter_column_id, filter_column_id);
214
0
}
215
216
Status VExprContext::filter_block(const VExprContextSPtrs& expr_contexts, Block* block,
217
1.05k
                                  size_t column_to_keep) {
218
1.05k
    if (expr_contexts.empty() || block->rows() == 0) {
219
1.05k
        return Status::OK();
220
1.05k
    }
221
222
2
    ColumnNumbers columns_to_filter(column_to_keep);
223
2
    std::iota(columns_to_filter.begin(), columns_to_filter.end(), 0);
224
225
2
    return execute_conjuncts_and_filter_block(expr_contexts, block, columns_to_filter,
226
2
                                              static_cast<int>(column_to_keep));
227
1.05k
}
228
229
Status VExprContext::execute_conjuncts(const VExprContextSPtrs& ctxs,
230
                                       const std::vector<IColumn::Filter*>* filters, Block* block,
231
48
                                       IColumn::Filter* result_filter, bool* can_filter_all) {
232
48
    return execute_conjuncts(ctxs, filters, false, block, result_filter, can_filter_all);
233
48
}
234
235
Status VExprContext::execute_filter(const Block* block, uint8_t* __restrict result_filter_data,
236
93
                                    size_t rows, bool accept_null, bool* can_filter_all) {
237
93
    return _root->execute_filter(this, block, result_filter_data, rows, accept_null,
238
93
                                 can_filter_all);
239
93
}
240
241
Status VExprContext::execute_conjuncts(const VExprContextSPtrs& ctxs,
242
                                       const std::vector<IColumn::Filter*>* filters,
243
                                       bool accept_null, const Block* block,
244
69
                                       IColumn::Filter* result_filter, bool* can_filter_all) {
245
69
    size_t rows = block->rows();
246
69
    DCHECK_EQ(result_filter->size(), rows);
247
69
    *can_filter_all = false;
248
69
    auto* __restrict result_filter_data = result_filter->data();
249
93
    for (const auto& ctx : ctxs) {
250
93
        RETURN_IF_ERROR(
251
93
                ctx->execute_filter(block, result_filter_data, rows, accept_null, can_filter_all));
252
93
        if (*can_filter_all) {
253
26
            return Status::OK();
254
26
        }
255
93
    }
256
43
    if (filters != nullptr) {
257
22
        for (auto* filter : *filters) {
258
0
            auto* __restrict filter_data = filter->data();
259
0
            const size_t size = filter->size();
260
0
            for (size_t i = 0; i < size; ++i) {
261
0
                result_filter_data[i] &= filter_data[i];
262
0
            }
263
0
            if (memchr(result_filter_data, 0x1, size) == nullptr) {
264
0
                *can_filter_all = true;
265
0
                return Status::OK();
266
0
            }
267
0
        }
268
22
    }
269
43
    return Status::OK();
270
43
}
271
272
Status VExprContext::execute_conjuncts(const VExprContextSPtrs& conjuncts, const Block* block,
273
6
                                       ColumnUInt8& null_map, IColumn::Filter& filter) {
274
6
    const auto& rows = block->rows();
275
6
    if (rows == 0) {
276
0
        return Status::OK();
277
0
    }
278
6
    if (null_map.size() != rows) {
279
0
        return Status::InternalError("null_map.size()!=rows, null_map.size()={}, rows={}",
280
0
                                     null_map.size(), rows);
281
0
    }
282
283
6
    auto* final_null_map = null_map.get_data().data();
284
6
    auto* final_filter_ptr = filter.data();
285
286
6
    for (const auto& conjunct : conjuncts) {
287
4
        ColumnPtr result_column;
288
4
        RETURN_IF_ERROR(conjunct->execute(block, result_column));
289
4
        auto [filter_column, is_const] = unpack_if_const(result_column);
290
4
        const auto* nullable_column = assert_cast<const ColumnNullable*>(filter_column.get());
291
4
        if (!is_const) {
292
4
            const ColumnPtr& nested_column = nullable_column->get_nested_column_ptr();
293
4
            const IColumn::Filter& result =
294
4
                    assert_cast<const ColumnUInt8&>(*nested_column).get_data();
295
4
            const auto* __restrict filter_data = result.data();
296
4
            const auto* __restrict null_map_data = nullable_column->get_null_map_data().data();
297
4
            DCHECK_EQ(rows, nullable_column->size());
298
299
32
            for (size_t i = 0; i != rows; ++i) {
300
                // null and null    => null
301
                // null and true    => null
302
                // null and false   => false
303
28
                final_null_map[i] = (final_null_map[i] & (null_map_data[i] | filter_data[i])) |
304
28
                                    (null_map_data[i] & (final_null_map[i] | final_filter_ptr[i]));
305
28
                final_filter_ptr[i] = final_filter_ptr[i] & filter_data[i];
306
28
            }
307
4
        } else {
308
0
            bool filter_data = nullable_column->get_bool(0);
309
0
            bool null_map_data = nullable_column->is_null_at(0);
310
0
            for (size_t i = 0; i != rows; ++i) {
311
                // null and null    => null
312
                // null and true    => null
313
                // null and false   => false
314
0
                final_null_map[i] = (final_null_map[i] & (null_map_data | filter_data)) |
315
0
                                    (null_map_data & (final_null_map[i] | final_filter_ptr[i]));
316
0
                final_filter_ptr[i] = final_filter_ptr[i] & filter_data;
317
0
            }
318
0
        }
319
4
    }
320
6
    return Status::OK();
321
6
}
322
323
// TODO Performance Optimization
324
// need exception safety
325
Status VExprContext::execute_conjuncts_and_filter_block(const VExprContextSPtrs& ctxs, Block* block,
326
                                                        std::vector<uint32_t>& columns_to_filter,
327
2
                                                        int column_to_keep) {
328
2
    IColumn::Filter result_filter(block->rows(), 1);
329
2
    bool can_filter_all;
330
331
2
    _reset_memory_usage(ctxs);
332
333
2
    RETURN_IF_ERROR(
334
2
            execute_conjuncts(ctxs, nullptr, false, block, &result_filter, &can_filter_all));
335
336
    // Accumulate the usage of `result_filter` into the first context.
337
2
    if (!ctxs.empty()) {
338
2
        ctxs[0]->_memory_usage += result_filter.allocated_bytes();
339
2
    }
340
2
    if (can_filter_all) {
341
3
        for (auto& col : columns_to_filter) {
342
3
            auto& column = block->get_by_position(col).column;
343
3
            if (column->is_exclusive()) {
344
3
                column->assert_mutable()->clear();
345
3
            } else {
346
0
                column = column->clone_empty();
347
0
            }
348
3
        }
349
1
    } else {
350
1
        try {
351
1
            Block::filter_block_internal(block, columns_to_filter, result_filter);
352
1
        } catch (const Exception& e) {
353
0
            std::string str;
354
0
            for (auto ctx : ctxs) {
355
0
                if (str.length()) {
356
0
                    str += ",";
357
0
                }
358
0
                str += ctx->root()->debug_string();
359
0
            }
360
361
0
            return Status::InternalError(
362
0
                    "filter_block_internal meet exception, exprs=[{}], exception={}", str,
363
0
                    e.what());
364
0
        }
365
1
    }
366
2
    Block::erase_useless_column(block, column_to_keep);
367
2
    return Status::OK();
368
2
}
369
370
Status VExprContext::execute_conjuncts_and_filter_block(const VExprContextSPtrs& ctxs, Block* block,
371
                                                        std::vector<uint32_t>& columns_to_filter,
372
                                                        int column_to_keep,
373
19
                                                        IColumn::Filter& filter) {
374
19
    _reset_memory_usage(ctxs);
375
19
    filter.resize_fill(block->rows(), 1);
376
19
    bool can_filter_all;
377
19
    RETURN_IF_ERROR(execute_conjuncts(ctxs, nullptr, false, block, &filter, &can_filter_all));
378
379
    // Accumulate the usage of `result_filter` into the first context.
380
19
    if (!ctxs.empty()) {
381
19
        ctxs[0]->_memory_usage += filter.allocated_bytes();
382
19
    }
383
19
    if (can_filter_all) {
384
3
        for (auto& col : columns_to_filter) {
385
3
            auto& column = block->get_by_position(col).column;
386
3
            if (column->is_exclusive()) {
387
3
                column->assert_mutable()->clear();
388
3
            } else {
389
0
                column = column->clone_empty();
390
0
            }
391
3
        }
392
18
    } else {
393
18
        RETURN_IF_CATCH_EXCEPTION(Block::filter_block_internal(block, columns_to_filter, filter));
394
18
    }
395
396
19
    Block::erase_useless_column(block, column_to_keep);
397
19
    return Status::OK();
398
19
}
399
400
Status VExprContext::get_output_block_after_execute_exprs(
401
5
        const VExprContextSPtrs& output_vexpr_ctxs, const Block& input_block, Block* output_block) {
402
5
    ColumnsWithTypeAndName result_columns;
403
5
    _reset_memory_usage(output_vexpr_ctxs);
404
405
28
    for (const auto& vexpr_ctx : output_vexpr_ctxs) {
406
28
        ColumnPtr result_column;
407
28
        RETURN_IF_ERROR(vexpr_ctx->execute(&input_block, result_column));
408
409
28
        auto type = vexpr_ctx->execute_type(&input_block);
410
28
        const auto& name = vexpr_ctx->expr_name();
411
412
28
        vexpr_ctx->_memory_usage += result_column->allocated_bytes();
413
28
        result_columns.emplace_back(result_column, type, name);
414
28
    }
415
5
    *output_block = {result_columns};
416
5
    return Status::OK();
417
5
}
418
419
26
void VExprContext::_reset_memory_usage(const VExprContextSPtrs& contexts) {
420
26
    std::for_each(contexts.begin(), contexts.end(),
421
49
                  [](auto&& context) { context->_memory_usage = 0; });
422
26
}
423
424
7
void VExprContext::prepare_ann_range_search(const doris::VectorSearchUserParams& params) {
425
7
    if (_root == nullptr) {
426
0
        return;
427
0
    }
428
429
7
    _root->prepare_ann_range_search(params, _ann_range_search_runtime, _suitable_for_ann_index);
430
7
    VLOG_DEBUG << fmt::format("Prepare ann range search result {}, _suitable_for_ann_index {}",
431
0
                              this->_ann_range_search_runtime.to_string(),
432
0
                              this->_suitable_for_ann_index);
433
7
    return;
434
7
}
435
436
Status VExprContext::evaluate_ann_range_search(
437
        const std::vector<std::unique_ptr<segment_v2::IndexIterator>>& cid_to_index_iterators,
438
        const std::vector<ColumnId>& idx_to_cid,
439
        const std::vector<std::unique_ptr<segment_v2::ColumnIterator>>& column_iterators,
440
        const std::unordered_map<VExprContext*, std::unordered_map<ColumnId, VExpr*>>&
441
                common_expr_to_slotref_map,
442
        size_t rows_of_segment, roaring::Roaring& row_bitmap,
443
        segment_v2::AnnIndexStats& ann_index_stats, bool enable_result_cache,
444
24
        bool* ann_range_search_executed) {
445
24
    if (ann_range_search_executed != nullptr) {
446
23
        *ann_range_search_executed = false;
447
23
    }
448
24
    if (_root == nullptr) {
449
0
        return Status::OK();
450
0
    }
451
452
24
    AnnRangeSearchEvaluationResult evaluation_result;
453
24
    RETURN_IF_ERROR(_root->evaluate_ann_range_search(
454
24
            _ann_range_search_runtime, cid_to_index_iterators, idx_to_cid, column_iterators,
455
24
            rows_of_segment, row_bitmap, ann_index_stats, enable_result_cache, evaluation_result));
456
457
23
    if (!evaluation_result.executed) {
458
19
        return Status::OK();
459
19
    }
460
4
    if (ann_range_search_executed != nullptr) {
461
4
        *ann_range_search_executed = true;
462
4
    }
463
464
4
    DCHECK(_index_context != nullptr);
465
4
    _index_context->set_index_result_for_expr(
466
4
            _root.get(),
467
4
            segment_v2::InvertedIndexResultBitmap(std::make_shared<roaring::Roaring>(row_bitmap),
468
4
                                                  std::make_shared<roaring::Roaring>()));
469
470
4
    if (!evaluation_result.dist_fulfilled) {
471
        // Do not perform index scan in this case.
472
2
        return Status::OK();
473
2
    }
474
475
4
    DCHECK_LT(_ann_range_search_runtime.src_col_idx, idx_to_cid.size());
476
2
    const auto src_col_idx = cast_set<int>(_ann_range_search_runtime.src_col_idx);
477
2
    const auto src_col_key = cast_set<ColumnId>(_ann_range_search_runtime.src_col_idx);
478
2
    auto slot_ref_map_it = common_expr_to_slotref_map.find(this);
479
2
    if (slot_ref_map_it == common_expr_to_slotref_map.end()) {
480
1
        return Status::OK();
481
1
    }
482
1
    auto& slot_ref_map = slot_ref_map_it->second;
483
1
    auto slot_ref_it = slot_ref_map.find(src_col_key);
484
1
    if (slot_ref_it == slot_ref_map.end()) {
485
0
        return Status::OK();
486
0
    }
487
1
    const VExpr* slot_ref_expr_addr = slot_ref_it->second;
488
1
    _index_context->set_true_for_index_status(slot_ref_expr_addr, src_col_idx);
489
490
1
    VLOG_DEBUG << fmt::format(
491
0
            "Evaluate ann range search for expr {}, src_col_idx {}, cid {}, row_bitmap "
492
0
            "cardinality {}",
493
0
            _root->debug_string(), src_col_idx, idx_to_cid[_ann_range_search_runtime.src_col_idx],
494
0
            row_bitmap.cardinality());
495
1
    return Status::OK();
496
1
}
497
498
0
uint64_t VExprContext::get_digest(uint64_t seed) const {
499
0
    return _root->get_digest(seed);
500
0
}
501
502
8
double VExprContext::execute_cost() const {
503
8
    if (_root == nullptr) {
504
        // When there is no expression root, treat the cost as a base value.
505
        // This avoids null dereferences while keeping a deterministic cost.
506
0
        return 0.0;
507
0
    }
508
8
    return _root->execute_cost();
509
8
}
510
511
} // namespace doris