Coverage Report

Created: 2026-04-14 07:58

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
4.08k
            : VExpr(node),
44
4.08k
              _source_node_id(source_node_id),
45
4.08k
              _expr_name(fmt::format("VTopNPred(source_node_id={})", _source_node_id)),
46
4.08k
              _target_ctx(std::move(target_ctx)) {}
47
8.18k
    bool is_topn_filter() const override { return true; }
48
49
4.10k
    static Status create_vtopn_pred(const TExpr& target_expr, int source_node_id, VExprSPtr& expr) {
50
4.10k
        VExprContextSPtr target_ctx;
51
4.10k
        RETURN_IF_ERROR(VExpr::create_expr_tree(target_expr, target_ctx));
52
53
4.10k
        TExprNode node;
54
4.10k
        node.__set_node_type(TExprNodeType::FUNCTION_CALL);
55
4.10k
        node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
56
4.10k
        node.__set_is_nullable(target_ctx->root()->is_nullable());
57
4.10k
        expr = VTopNPred::create_shared(node, source_node_id, target_ctx);
58
59
4.10k
        DCHECK(target_ctx->root() != nullptr);
60
4.10k
        expr->add_child(target_ctx->root());
61
62
4.10k
        return Status::OK();
63
4.10k
    }
64
65
4.08k
    int source_node_id() const { return _source_node_id; }
66
67
4.07k
    Status prepare(RuntimeState* state, const RowDescriptor& desc, VExprContext* context) override {
68
4.07k
        _predicate = &state->get_query_ctx()->get_runtime_predicate(_source_node_id);
69
4.07k
        RETURN_IF_ERROR_OR_PREPARED(VExpr::prepare(state, desc, context));
70
71
4.07k
        ColumnsWithTypeAndName argument_template;
72
4.07k
        argument_template.emplace_back(nullptr, _children[0]->data_type(),
73
4.07k
                                       _children[0]->expr_name());
74
4.07k
        argument_template.emplace_back(nullptr, _children[0]->data_type(), "topn value");
75
76
4.07k
        _function = SimpleFunctionFactory::instance().get_function(
77
4.07k
                _predicate->is_asc() ? "le" : "ge", argument_template, _data_type, {},
78
4.07k
                state->be_exec_version());
79
4.07k
        if (!_function) {
80
0
            return Status::InternalError("get function failed");
81
0
        }
82
4.07k
        return Status::OK();
83
4.07k
    }
84
85
    Status execute_column(VExprContext* context, const Block* block, Selector* selector,
86
20.4k
                          size_t count, ColumnPtr& result_column) const override {
87
20.4k
        if (!_predicate->has_value()) {
88
16.6k
            result_column = create_always_true_column(count, _data_type->is_nullable());
89
16.6k
            return Status::OK();
90
16.6k
        }
91
92
3.87k
        Block temp_block;
93
94
        // slot
95
3.87k
        ColumnPtr slot_column;
96
3.87k
        RETURN_IF_ERROR(_children[0]->execute_column(context, block, selector, count, slot_column));
97
3.87k
        auto slot_type = _children[0]->execute_type(block);
98
3.87k
        temp_block.insert({slot_column, slot_type, _children[0]->expr_name()});
99
3.87k
        int slot_id = 0;
100
101
        // topn value
102
3.87k
        Field field = _predicate->get_value();
103
3.87k
        auto column_ptr = _children[0]->data_type()->create_column_const(1, field);
104
3.87k
        int topn_value_id = VExpr::insert_param(&temp_block,
105
3.87k
                                                {column_ptr, _children[0]->data_type(), _expr_name},
106
3.87k
                                                std::max(count, column_ptr->size()));
107
108
        // if error(slot_id == -1), will return.
109
3.87k
        ColumnNumbers arguments = {static_cast<uint32_t>(slot_id),
110
3.87k
                                   static_cast<uint32_t>(topn_value_id)};
111
112
3.87k
        uint32_t num_columns_without_result = temp_block.columns();
113
        // prepare a column to save result
114
3.87k
        temp_block.insert({nullptr, _data_type, _expr_name});
115
116
3.87k
        RETURN_IF_ERROR(_function->execute(nullptr, temp_block, arguments,
117
3.87k
                                           num_columns_without_result, temp_block.rows()));
118
3.87k
        result_column = std::move(temp_block.get_by_position(num_columns_without_result).column);
119
3.87k
        if (is_nullable() && _predicate->nulls_first()) {
120
            // null values ​​are always not filtered
121
3.27k
            change_null_to_true(result_column->assume_mutable());
122
3.27k
        }
123
3.87k
        DCHECK_EQ(result_column->size(), count);
124
3.87k
        return Status::OK();
125
3.87k
    }
126
127
0
    const std::string& expr_name() const override { return _expr_name; }
128
129
    // only used in external table (for min-max filter). get `slot > xxx`, not `function(slot) > xxx`.
130
4.00k
    bool get_binary_expr(VExprSPtr& new_root) const {
131
4.00k
        if (!get_child(0)->is_slot_ref()) {
132
            // top rf maybe is `xxx order by abs(column) limit xxx`.
133
0
            return false;
134
0
        }
135
136
4.00k
        if (!_predicate->has_value()) {
137
2.31k
            return false;
138
2.31k
        }
139
140
1.68k
        auto* slot_ref = assert_cast<VSlotRef*>(get_child(0).get());
141
1.68k
        auto slot_data_type = remove_nullable(slot_ref->data_type());
142
1.68k
        {
143
1.68k
            TFunction fn;
144
1.68k
            TFunctionName fn_name;
145
1.68k
            fn_name.__set_db_name("");
146
1.68k
            fn_name.__set_function_name(_predicate->is_asc() ? "le" : "ge");
147
1.68k
            fn.__set_name(fn_name);
148
1.68k
            fn.__set_binary_type(TFunctionBinaryType::BUILTIN);
149
1.68k
            std::vector<TTypeDesc> arg_types;
150
1.68k
            arg_types.push_back(create_type_desc(slot_data_type->get_primitive_type(),
151
1.68k
                                                 slot_data_type->get_precision(),
152
1.68k
                                                 slot_data_type->get_scale()));
153
154
1.68k
            arg_types.push_back(create_type_desc(slot_data_type->get_primitive_type(),
155
1.68k
                                                 slot_data_type->get_precision(),
156
1.68k
                                                 slot_data_type->get_scale()));
157
1.68k
            fn.__set_arg_types(arg_types);
158
1.68k
            fn.__set_ret_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
159
1.68k
            fn.__set_has_var_args(false);
160
161
1.68k
            TExprNode texpr_node;
162
1.68k
            texpr_node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
163
1.68k
            texpr_node.__set_node_type(TExprNodeType::BINARY_PRED);
164
1.68k
            texpr_node.__set_opcode(_predicate->is_asc() ? TExprOpcode::LE : TExprOpcode::GE);
165
1.68k
            texpr_node.__set_fn(fn);
166
1.68k
            texpr_node.__set_num_children(2);
167
1.68k
            texpr_node.__set_is_nullable(is_nullable());
168
1.68k
            new_root = VectorizedFnCall::create_shared(texpr_node);
169
1.68k
        }
170
171
1.68k
        {
172
            // add slot
173
1.68k
            new_root->add_child(children().at(0));
174
1.68k
        }
175
        // add Literal
176
1.68k
        {
177
1.68k
            Field field = _predicate->get_value();
178
1.68k
            TExprNode node = create_texpr_node_from(field, slot_data_type->get_primitive_type(),
179
1.68k
                                                    slot_data_type->get_precision(),
180
1.68k
                                                    slot_data_type->get_scale());
181
1.68k
            new_root->add_child(VLiteral::create_shared(node));
182
1.68k
        }
183
184
        // 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.`
185
1.68k
        if (_predicate->nulls_first()) {
186
1.58k
            VExprSPtr col_is_null_node;
187
1.58k
            {
188
1.58k
                TFunction fn;
189
1.58k
                TFunctionName fn_name;
190
1.58k
                fn_name.__set_db_name("");
191
1.58k
                fn_name.__set_function_name("is_null_pred");
192
1.58k
                fn.__set_name(fn_name);
193
1.58k
                fn.__set_binary_type(TFunctionBinaryType::BUILTIN);
194
1.58k
                std::vector<TTypeDesc> arg_types;
195
1.58k
                arg_types.push_back(create_type_desc(slot_data_type->get_primitive_type(),
196
1.58k
                                                     slot_data_type->get_precision(),
197
1.58k
                                                     slot_data_type->get_scale()));
198
1.58k
                fn.__set_arg_types(arg_types);
199
1.58k
                fn.__set_ret_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
200
1.58k
                fn.__set_has_var_args(false);
201
202
1.58k
                TExprNode texpr_node;
203
1.58k
                texpr_node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
204
1.58k
                texpr_node.__set_node_type(TExprNodeType::FUNCTION_CALL);
205
1.58k
                texpr_node.__set_fn(fn);
206
1.58k
                texpr_node.__set_num_children(1);
207
1.58k
                col_is_null_node = VectorizedFnCall::create_shared(texpr_node);
208
209
                // add slot.
210
1.58k
                col_is_null_node->add_child(children().at(0));
211
1.58k
            }
212
213
1.58k
            VExprSPtr or_node;
214
1.58k
            {
215
1.58k
                TExprNode texpr_node;
216
1.58k
                texpr_node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
217
1.58k
                texpr_node.__set_node_type(TExprNodeType::COMPOUND_PRED);
218
1.58k
                texpr_node.__set_opcode(TExprOpcode::COMPOUND_OR);
219
1.58k
                texpr_node.__set_num_children(2);
220
1.58k
                or_node = VectorizedFnCall::create_shared(texpr_node);
221
1.58k
            }
222
223
1.58k
            or_node->add_child(col_is_null_node);
224
1.58k
            or_node->add_child(new_root);
225
1.58k
            new_root = or_node;
226
1.58k
        }
227
228
1.68k
        return true;
229
4.00k
    }
230
231
private:
232
    int _source_node_id;
233
    std::string _expr_name;
234
    RuntimePredicate* _predicate = nullptr;
235
    FunctionBasePtr _function;
236
    VExprContextSPtr _target_ctx;
237
};
238
239
} // namespace doris