Coverage Report

Created: 2026-11-07 17:03

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/predicate/shared_predicate.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 <atomic>
21
#include <cstdint>
22
#include <memory>
23
24
#include "common/factory_creator.h"
25
#include "core/column/column_dictionary.h"
26
#include "storage/index/bloom_filter/bloom_filter.h"
27
#include "storage/index/inverted/inverted_index_reader.h"
28
#include "storage/predicate/column_predicate.h"
29
30
namespace doris {
31
32
// SharedPredicate only used on topn runtime predicate.
33
// Runtime predicate globally share one predicate, to ensure that updates can be real-time.
34
// At the beginning nested predicate may be nullptr, in which case predicate always returns true.
35
class SharedPredicate final : public ColumnPredicate {
36
    ENABLE_FACTORY_CREATOR(SharedPredicate);
37
38
public:
39
    SharedPredicate(uint32_t column_id, std::string col_name)
40
30
            : ColumnPredicate(column_id, col_name, PrimitiveType::INVALID_TYPE),
41
30
              _mtx(std::make_shared<std::shared_mutex>()) {}
42
    SharedPredicate(const ColumnPredicate& other) = delete;
43
30
    ~SharedPredicate() override = default;
44
0
    std::string debug_string() const override {
45
0
        std::shared_lock<std::shared_mutex> lock(*_mtx);
46
0
        fmt::memory_buffer debug_string_buffer;
47
0
        fmt::format_to(debug_string_buffer, "SharedPredicate({}, nested={})",
48
0
                       ColumnPredicate::debug_string(), _nested ? _nested->debug_string() : "null");
49
0
        return fmt::to_string(debug_string_buffer);
50
0
    }
51
12
    std::shared_ptr<ColumnPredicate> clone(uint32_t column_id) const override {
52
        // All scanner thread should share the same SharedPredicate object.
53
12
        return std::const_pointer_cast<ColumnPredicate>(shared_from_this());
54
12
    }
55
56
24
    bool is_runtime_filter() const override { return true; }
57
58
1.56k
    PredicateType type() const override {
59
1.56k
        std::shared_lock<std::shared_mutex> lock(*_mtx);
60
1.56k
        if (!_nested) {
61
            // topn filter is le or ge
62
20
            return PredicateType::LE;
63
20
        }
64
1.54k
        return _nested->type();
65
1.56k
    }
66
0
    PrimitiveType primitive_type() const override {
67
0
        std::shared_lock<std::shared_mutex> lock(*_mtx);
68
0
        if (!_nested) {
69
0
            return PrimitiveType::INVALID_TYPE;
70
0
        }
71
0
        return _nested->primitive_type();
72
0
    }
73
74
1.04k
    void set_nested(const std::shared_ptr<ColumnPredicate>& nested) {
75
1.04k
        std::unique_lock<std::shared_mutex> lock(*_mtx);
76
1.04k
        _nested = nested;
77
1.04k
        _version.fetch_add(1, std::memory_order_release);
78
1.04k
    }
79
80
1.33k
    uint64_t version() const { return _version.load(std::memory_order_acquire); }
81
82
    struct Snapshot {
83
        uint64_t version;
84
        std::shared_ptr<const ColumnPredicate> predicate;
85
    };
86
87
    // Keep the immutable comparison alive after releasing the lock. A concurrent update must
88
    // not pair version 2 with version 1's bound, or hold the producer lock during index I/O.
89
1.01k
    Snapshot snapshot() const {
90
1.01k
        std::shared_lock<std::shared_mutex> lock(*_mtx);
91
1.01k
        return {.version = _version.load(std::memory_order_relaxed), .predicate = _nested};
92
1.01k
    }
93
94
    Status evaluate(const IndexFieldNameAndTypePair& name_with_type, IndexIterator* iterator,
95
0
                    uint32_t num_rows, roaring::Roaring* bitmap) const override {
96
0
        std::shared_lock<std::shared_mutex> lock(*_mtx);
97
0
        if (!_nested) {
98
0
            return Status::OK();
99
0
        }
100
0
        return _nested->evaluate(name_with_type, iterator, num_rows, bitmap);
101
0
    }
102
103
    void evaluate_and(const IColumn& column, const uint16_t* sel, uint16_t size,
104
0
                      bool* flags) const override {
105
0
        std::shared_lock<std::shared_mutex> lock(*_mtx);
106
0
        if (!_nested) {
107
0
            return;
108
0
        }
109
0
        return _nested->evaluate_and(column, sel, size, flags);
110
0
    }
111
112
    void evaluate_or(const IColumn& column, const uint16_t* sel, uint16_t size,
113
0
                     bool* flags) const override {
114
0
        DCHECK(false) << "should not reach here";
115
0
    }
116
117
1.42k
    bool evaluate_and(const segment_v2::ZoneMap& zone_map) const override {
118
1.42k
        std::shared_lock<std::shared_mutex> lock(*_mtx);
119
1.42k
        if (!_nested) {
120
1.29k
            return ColumnPredicate::evaluate_and(zone_map);
121
1.29k
        }
122
132
        return _nested->evaluate_and(zone_map);
123
1.42k
    }
124
125
0
    bool evaluate_del(const segment_v2::ZoneMap& zone_map) const override {
126
0
        std::shared_lock<std::shared_mutex> lock(*_mtx);
127
0
        if (!_nested) {
128
0
            return ColumnPredicate::evaluate_del(zone_map);
129
0
        }
130
0
        return _nested->evaluate_del(zone_map);
131
0
    }
132
133
0
    bool evaluate_and(const BloomFilter* bf) const override {
134
0
        std::shared_lock<std::shared_mutex> lock(*_mtx);
135
0
        if (!_nested) {
136
0
            return ColumnPredicate::evaluate_and(bf);
137
0
        }
138
0
        return _nested->evaluate_and(bf);
139
0
    }
140
141
22
    bool can_do_bloom_filter(bool ngram) const override {
142
22
        std::shared_lock<std::shared_mutex> lock(*_mtx);
143
22
        if (!_nested) {
144
20
            return ColumnPredicate::can_do_bloom_filter(ngram);
145
20
        }
146
2
        return _nested->can_do_bloom_filter(ngram);
147
22
    }
148
149
1.59k
    void evaluate_vec(const IColumn& column, uint16_t size, bool* flags) const override {
150
1.59k
        std::shared_lock<std::shared_mutex> lock(*_mtx);
151
1.59k
        if (!_nested) {
152
8.50k
            for (uint16_t i = 0; i < size; ++i) {
153
8.48k
                flags[i] = true;
154
8.48k
            }
155
20
            return;
156
20
        }
157
1.57k
        _nested->evaluate_vec(column, size, flags);
158
1.57k
    }
159
160
0
    void evaluate_and_vec(const IColumn& column, uint16_t size, bool* flags) const override {
161
0
        std::shared_lock<std::shared_mutex> lock(*_mtx);
162
0
        if (!_nested) {
163
0
            return;
164
0
        }
165
0
        _nested->evaluate_and_vec(column, size, flags);
166
0
    }
167
168
0
    std::string get_search_str() const override {
169
0
        std::shared_lock<std::shared_mutex> lock(*_mtx);
170
0
        if (!_nested) {
171
0
            DCHECK(false) << "should not reach here";
172
0
        }
173
0
        return _nested->get_search_str();
174
0
    }
175
176
0
    bool evaluate_and(ParquetPredicate::ColumnStat* statistic) const override {
177
0
        std::shared_lock<std::shared_mutex> lock(*_mtx);
178
0
        if (!_nested) {
179
            // at the begining _nested will be null, so return true.
180
0
            return true;
181
0
        }
182
0
        return _nested->evaluate_and(statistic);
183
0
    }
184
185
    bool evaluate_and(ParquetPredicate::CachedPageIndexStat* statistic,
186
0
                      RowRanges* row_ranges) const override {
187
0
        std::shared_lock<std::shared_mutex> lock(*_mtx);
188
189
0
        if (!_nested) {
190
            // at the begining _nested will be null, so return true.
191
0
            row_ranges->add(statistic->row_group_range);
192
0
            return true;
193
0
        }
194
0
        return _nested->evaluate_and(statistic, row_ranges);
195
0
    }
196
197
private:
198
0
    uint16_t _evaluate_inner(const IColumn& column, uint16_t* sel, uint16_t size) const override {
199
0
        std::shared_lock<std::shared_mutex> lock(*_mtx);
200
0
        if (!_nested) {
201
0
            return size;
202
0
        }
203
0
        return _nested->evaluate(column, sel, size);
204
0
    }
205
206
    mutable std::shared_ptr<std::shared_mutex> _mtx;
207
    std::shared_ptr<ColumnPredicate> _nested;
208
    std::atomic<uint64_t> _version {0};
209
};
210
211
} //namespace doris