Coverage Report

Created: 2026-10-29 14:40

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/exprs/vtopn_pred.h
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
#pragma once
19
20
#include <gen_cpp/types.pb.h>
21
22
#include <utility>
23
24
#include "core/block/column_numbers.h"
25
#include "core/data_type/data_type.h"
26
#include "exec/common/util.hpp"
27
#include "exprs/function/simple_function_factory.h"
28
#include "exprs/vectorized_fn_call.h"
29
#include "exprs/vexpr.h"
30
#include "exprs/vslot_ref.h"
31
#include "runtime/query_context.h"
32
#include "runtime/runtime_predicate.h"
33
#include "runtime/runtime_state.h"
34
35
namespace doris {
36
37
// only used for dynamic topn filter
38
class VTopNPred : public VExpr {
39
    ENABLE_FACTORY_CREATOR(VTopNPred);
40
41
public:
42
    VTopNPred(const TExprNode& node, int source_node_id, VExprContextSPtr target_ctx)
43
1.95k
            : VExpr(node),
44
1.95k
              _source_node_id(source_node_id),
45
1.95k
              _expr_name(fmt::format("VTopNPred(source_node_id={})", _source_node_id)),
46
1.95k
              _target_ctx(std::move(target_ctx)) {}
47
3.92k
    bool is_topn_filter() const override { return true; }
48
49
1.82k
    static Status create_vtopn_pred(const TExpr& target_expr, int source_node_id, VExprSPtr& expr) {
50
1.82k
        VExprContextSPtr target_ctx;
51
1.82k
        RETURN_IF_ERROR(VExpr::create_expr_tree(target_expr, target_ctx));
52
53
1.82k
        TExprNode node;
54
1.82k
        node.__set_node_type(TExprNodeType::FUNCTION_CALL);
55
1.82k
        node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
56
1.82k
        node.__set_is_nullable(target_ctx->root()->is_nullable());
57
1.82k
        expr = VTopNPred::create_shared(node, source_node_id, target_ctx);
58
59
1.82k
        DCHECK(target_ctx->root() != nullptr);
60
1.82k
        expr->add_child(target_ctx->root());
61
62
1.82k
        return Status::OK();
63
1.82k
    }
64
65
1.67k
    int source_node_id() const { return _source_node_id; }
66
262
    bool is_asc() const { return _predicate->is_asc(); }
67
155
    bool nulls_first() const { return _predicate->nulls_first(); }
68
69
    bool can_evaluate_zonemap_filter() const override;
70
    ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const override;
71
72
48
    Status clone_node(VExprSPtr* cloned_expr) const override {
73
48
        DORIS_CHECK(cloned_expr != nullptr);
74
48
        *cloned_expr = VTopNPred::create_shared(clone_texpr_node(), _source_node_id, nullptr);
75
48
        return Status::OK();
76
48
    }
77
78
1.93k
    Status prepare(RuntimeState* state, const RowDescriptor& desc, VExprContext* context) override {
79
1.93k
        _predicate = &state->get_query_ctx()->get_runtime_predicate(_source_node_id);
80
1.93k
        RETURN_IF_ERROR_OR_PREPARED(VExpr::prepare(state, desc, context));
81
82
1.93k
        ColumnsWithTypeAndName argument_template;
83
1.93k
        argument_template.emplace_back(nullptr, _children[0]->data_type(),
84
1.93k
                                       _children[0]->expr_name());
85
1.93k
        argument_template.emplace_back(nullptr, _children[0]->data_type(), "topn value");
86
87
1.93k
        _function = SimpleFunctionFactory::instance().get_function(
88
1.93k
                _predicate->is_asc() ? "le" : "ge", argument_template, _data_type, {},
89
1.93k
                state->be_exec_version());
90
1.93k
        if (!_function) {
91
0
            return Status::InternalError("get function failed");
92
0
        }
93
1.93k
        return Status::OK();
94
1.93k
    }
95
96
    Status execute_column_impl(VExprContext* context, const Block* block, const Selector* selector,
97
1.59k
                               size_t count, ColumnPtr& result_column) const override {
98
1.59k
        if (!_predicate->has_value()) {
99
1.00k
            result_column = create_always_true_column(count, _data_type->is_nullable());
100
1.00k
            return Status::OK();
101
1.00k
        }
102
103
593
        Block temp_block;
104
105
        // slot
106
593
        ColumnPtr slot_column;
107
593
        RETURN_IF_ERROR(_children[0]->execute_column(context, block, selector, count, slot_column));
108
593
        auto slot_type = _children[0]->execute_type(block);
109
593
        temp_block.insert({slot_column, slot_type, _children[0]->expr_name()});
110
593
        int slot_id = 0;
111
112
        // topn value
113
593
        Field field = _predicate->get_value();
114
593
        auto column_ptr = _children[0]->data_type()->create_column_const(1, field);
115
593
        int topn_value_id = VExpr::insert_param(&temp_block,
116
593
                                                {column_ptr, _children[0]->data_type(), _expr_name},
117
593
                                                std::max(count, column_ptr->size()));
118
119
        // if error(slot_id == -1), will return.
120
593
        ColumnNumbers arguments = {static_cast<uint32_t>(slot_id),
121
593
                                   static_cast<uint32_t>(topn_value_id)};
122
123
593
        uint32_t num_columns_without_result = temp_block.columns();
124
        // prepare a column to save result
125
593
        temp_block.insert({nullptr, _data_type, _expr_name});
126
127
593
        RETURN_IF_ERROR(_function->execute(nullptr, temp_block, arguments,
128
593
                                           num_columns_without_result, temp_block.rows()));
129
593
        result_column = std::move(temp_block.get_by_position(num_columns_without_result).column);
130
593
        result_column = _normalize_filter_result(std::move(result_column));
131
593
        DCHECK_EQ(result_column->size(), count);
132
593
        return Status::OK();
133
593
    }
134
135
    // Returns true only for a direct slot binding whose logical fixed-width type exactly matches
136
    // `data_type`. Eligibility does not depend on whether the dynamic TopN bound has arrived: a
137
    // reader may cache this answer during initialization and observe the bound in a later batch.
138
    bool can_execute_on_raw_fixed_values(const DataTypePtr& data_type,
139
                                         int column_id) const override;
140
141
    // Compares the contiguous non-NULL physical values with one snapshot of the current TopN bound
142
    // and ANDs the result into `matches`. `value_width` must equal the logical Doris value width.
143
    // When no bound is available yet, this is deliberately an all-pass operation.
144
    Status execute_on_raw_fixed_values(const uint8_t* values, size_t num_values, size_t value_width,
145
                                       const DataTypePtr& data_type, int column_id,
146
                                       uint8_t* matches) const override;
147
148
    // Returns true for a directly bound STRING-like or VARBINARY slot. As with the fixed-width
149
    // capability, a not-yet-published bound does not force the scanner onto a materializing path.
150
    bool can_execute_on_raw_binary_values(const DataTypePtr& data_type,
151
                                          int column_id) const override;
152
153
    // Compares decoder-owned immutable byte slices with the current TopN bound and ANDs each
154
    // decision into `matches`; it never copies the slices into a ColumnString/ColumnVarbinary.
155
    Status execute_on_raw_binary_values(const StringRef* values, size_t num_values,
156
                                        const DataTypePtr& data_type, int column_id,
157
                                        uint8_t* matches) const override;
158
159
205
    bool raw_predicate_result_for_null() const override {
160
        // execute_column() is all-pass until publication; afterwards NULLS FIRST keeps NULL while
161
        // NULLS LAST rejects it. Sample this state per fragment just like the mutable bound.
162
205
        return _predicate != nullptr && (!_predicate->has_value() || _predicate->nulls_first());
163
205
    }
164
165
    // Dictionary capability is restricted to a direct slot and NULLS-LAST semantics. It stays
166
    // enabled before the first bound so row-group setup can cache an all-pass bitmap; the scan
167
    // scheduler retains a per-batch residual and defers dictionary-id filtering while NULL still
168
    // matches, then safely resumes it after bound publication.
169
    bool can_evaluate_dictionary_filter() const override;
170
171
    // Evaluates every non-NULL entry in the bound slot dictionary against one current-bound
172
    // snapshot. kNoMatch proves that no entry can pass; kMayMatch includes both a matching entry and
173
    // a not-yet-published bound; kUnsupported reports an absent or incompatible dictionary slot.
174
    ZoneMapFilterResult evaluate_dictionary_filter(const DictionaryEvalContext& ctx) const override;
175
176
2
    const std::string& expr_name() const override { return _expr_name; }
177
178
    // only used in external table (for min-max filter). get `slot > xxx`, not `function(slot) > xxx`.
179
0
    bool get_binary_expr(VExprSPtr& new_root) const {
180
0
        if (!get_child(0)->is_slot_ref()) {
181
            // top rf maybe is `xxx order by abs(column) limit xxx`.
182
0
            return false;
183
0
        }
184
185
0
        if (!_predicate->has_value()) {
186
0
            return false;
187
0
        }
188
189
0
        auto* slot_ref = assert_cast<VSlotRef*>(get_child(0).get());
190
0
        auto slot_data_type = remove_nullable(slot_ref->data_type());
191
0
        {
192
0
            TFunction fn;
193
0
            TFunctionName fn_name;
194
0
            fn_name.__set_db_name("");
195
0
            fn_name.__set_function_name(_predicate->is_asc() ? "le" : "ge");
196
0
            fn.__set_name(fn_name);
197
0
            fn.__set_binary_type(TFunctionBinaryType::BUILTIN);
198
0
            std::vector<TTypeDesc> arg_types;
199
0
            arg_types.push_back(create_type_desc(slot_data_type->get_primitive_type(),
200
0
                                                 slot_data_type->get_precision(),
201
0
                                                 slot_data_type->get_scale()));
202
203
0
            arg_types.push_back(create_type_desc(slot_data_type->get_primitive_type(),
204
0
                                                 slot_data_type->get_precision(),
205
0
                                                 slot_data_type->get_scale()));
206
0
            fn.__set_arg_types(arg_types);
207
0
            fn.__set_ret_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
208
0
            fn.__set_has_var_args(false);
209
210
0
            TExprNode texpr_node;
211
0
            texpr_node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
212
0
            texpr_node.__set_node_type(TExprNodeType::BINARY_PRED);
213
0
            texpr_node.__set_opcode(_predicate->is_asc() ? TExprOpcode::LE : TExprOpcode::GE);
214
0
            texpr_node.__set_fn(fn);
215
0
            texpr_node.__set_num_children(2);
216
0
            texpr_node.__set_is_nullable(is_nullable());
217
0
            new_root = VectorizedFnCall::create_shared(texpr_node);
218
0
        }
219
220
0
        {
221
            // add slot
222
0
            new_root->add_child(children().at(0));
223
0
        }
224
        // add Literal
225
0
        {
226
0
            Field field = _predicate->get_value();
227
0
            TExprNode node = create_texpr_node_from(field, slot_data_type->get_primitive_type(),
228
0
                                                    slot_data_type->get_precision(),
229
0
                                                    slot_data_type->get_scale());
230
0
            new_root->add_child(VLiteral::create_shared(node));
231
0
        }
232
233
        // Since the normal greater than or less than relationship does not consider the relationship of null values, the generated `col >=/<= xxx OR col is null.`
234
0
        if (_predicate->nulls_first()) {
235
0
            VExprSPtr col_is_null_node;
236
0
            {
237
0
                TFunction fn;
238
0
                TFunctionName fn_name;
239
0
                fn_name.__set_db_name("");
240
0
                fn_name.__set_function_name("is_null_pred");
241
0
                fn.__set_name(fn_name);
242
0
                fn.__set_binary_type(TFunctionBinaryType::BUILTIN);
243
0
                std::vector<TTypeDesc> arg_types;
244
0
                arg_types.push_back(create_type_desc(slot_data_type->get_primitive_type(),
245
0
                                                     slot_data_type->get_precision(),
246
0
                                                     slot_data_type->get_scale()));
247
0
                fn.__set_arg_types(arg_types);
248
0
                fn.__set_ret_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
249
0
                fn.__set_has_var_args(false);
250
251
0
                TExprNode texpr_node;
252
0
                texpr_node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
253
0
                texpr_node.__set_node_type(TExprNodeType::FUNCTION_CALL);
254
0
                texpr_node.__set_fn(fn);
255
0
                texpr_node.__set_num_children(1);
256
0
                col_is_null_node = VectorizedFnCall::create_shared(texpr_node);
257
258
                // add slot.
259
0
                col_is_null_node->add_child(children().at(0));
260
0
            }
261
262
0
            VExprSPtr or_node;
263
0
            {
264
0
                TExprNode texpr_node;
265
0
                texpr_node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
266
0
                texpr_node.__set_node_type(TExprNodeType::COMPOUND_PRED);
267
0
                texpr_node.__set_opcode(TExprOpcode::COMPOUND_OR);
268
0
                texpr_node.__set_num_children(2);
269
0
                or_node = VectorizedFnCall::create_shared(texpr_node);
270
0
            }
271
272
0
            or_node->add_child(col_is_null_node);
273
0
            or_node->add_child(new_root);
274
0
            new_root = or_node;
275
0
        }
276
277
0
        return true;
278
0
    }
279
280
private:
281
601
    ColumnPtr _normalize_filter_result(ColumnPtr column) const {
282
601
        const size_t rows = column->size();
283
601
        if (const auto* constant = check_and_get_column<ColumnConst>(*column)) {
284
8
            auto nested = _normalize_filter_result(constant->get_data_column_ptr());
285
8
            return ColumnConst::create(std::move(nested), rows);
286
8
        }
287
288
593
        auto mutable_column = IColumn::mutate(std::move(column));
289
593
        if (auto* nullable = check_and_get_column<ColumnNullable>(*mutable_column)) {
290
591
            auto& values = assert_cast<ColumnUInt8&>(*nullable->get_nested_column_ptr()).get_data();
291
591
            auto& null_map = nullable->get_null_map_data();
292
            // Collapse SQL NULL to the filter decision: NULLS FIRST keeps it, while NULLS LAST
293
            // rejects it. Preserve the nullable wrapper so strict block execution still matches
294
            // the expression's declared Nullable(Boolean) type.
295
1.68M
            for (size_t row = 0; row < values.size(); ++row) {
296
1.68M
                values[row] = null_map[row] ? _predicate->nulls_first() : values[row];
297
1.68M
            }
298
591
            nullable->fill_false_to_nullmap(rows);
299
591
            return mutable_column;
300
591
        }
301
2
        return mutable_column;
302
593
    }
303
304
    int _source_node_id;
305
    std::string _expr_name;
306
    RuntimePredicate* _predicate = nullptr;
307
    FunctionBasePtr _function;
308
    VExprContextSPtr _target_ctx;
309
};
310
311
} // namespace doris