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 | 769 | : ColumnPredicate(column_id, col_name, PrimitiveType::INVALID_TYPE), |
41 | 769 | _mtx(std::make_shared<std::shared_mutex>()) {} |
42 | | SharedPredicate(const ColumnPredicate& other) = delete; |
43 | 769 | ~SharedPredicate() override = default; |
44 | 2.72k | std::string debug_string() const override { |
45 | 2.72k | std::shared_lock<std::shared_mutex> lock(*_mtx); |
46 | 2.72k | fmt::memory_buffer debug_string_buffer; |
47 | 2.72k | fmt::format_to(debug_string_buffer, "SharedPredicate({}, nested={})", |
48 | 2.72k | ColumnPredicate::debug_string(), _nested ? _nested->debug_string() : "null"); |
49 | 2.72k | return fmt::to_string(debug_string_buffer); |
50 | 2.72k | } |
51 | 6.85k | std::shared_ptr<ColumnPredicate> clone(uint32_t column_id) const override { |
52 | | // All scanner thread should share the same SharedPredicate object. |
53 | 6.85k | return std::const_pointer_cast<ColumnPredicate>(shared_from_this()); |
54 | 6.85k | } |
55 | | |
56 | 716 | bool is_runtime_filter() const override { return true; } |
57 | | |
58 | 29.1k | PredicateType type() const override { |
59 | 29.1k | std::shared_lock<std::shared_mutex> lock(*_mtx); |
60 | 29.1k | if (!_nested) { |
61 | | // topn filter is le or ge |
62 | 16.6k | return PredicateType::LE; |
63 | 16.6k | } |
64 | 12.5k | return _nested->type(); |
65 | 29.1k | } |
66 | 2.72k | PrimitiveType primitive_type() const override { |
67 | 2.72k | std::shared_lock<std::shared_mutex> lock(*_mtx); |
68 | 2.72k | if (!_nested) { |
69 | 973 | return PrimitiveType::INVALID_TYPE; |
70 | 973 | } |
71 | 1.75k | return _nested->primitive_type(); |
72 | 2.72k | } |
73 | | |
74 | 3.37k | void set_nested(const std::shared_ptr<ColumnPredicate>& nested) { |
75 | 3.37k | std::unique_lock<std::shared_mutex> lock(*_mtx); |
76 | 3.37k | _nested = nested; |
77 | 3.37k | _version.fetch_add(1, std::memory_order_release); |
78 | 3.37k | } |
79 | | |
80 | 27.3k | 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.80k | Snapshot snapshot() const { |
90 | 1.80k | std::shared_lock<std::shared_mutex> lock(*_mtx); |
91 | 1.80k | return {.version = _version.load(std::memory_order_relaxed), .predicate = _nested}; |
92 | 1.80k | } |
93 | | |
94 | | Status evaluate(const IndexFieldNameAndTypePair& name_with_type, IndexIterator* iterator, |
95 | 12 | uint32_t num_rows, roaring::Roaring* bitmap) const override { |
96 | 12 | std::shared_lock<std::shared_mutex> lock(*_mtx); |
97 | 12 | if (!_nested) { |
98 | 12 | return Status::OK(); |
99 | 12 | } |
100 | 0 | return _nested->evaluate(name_with_type, iterator, num_rows, bitmap); |
101 | 12 | } |
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 | 22.3k | bool evaluate_and(const segment_v2::ZoneMap& zone_map) const override { |
118 | 22.3k | std::shared_lock<std::shared_mutex> lock(*_mtx); |
119 | 22.3k | if (!_nested) { |
120 | 13.8k | return ColumnPredicate::evaluate_and(zone_map); |
121 | 13.8k | } |
122 | 8.42k | return _nested->evaluate_and(zone_map); |
123 | 22.3k | } |
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 | 12.8k | bool can_do_bloom_filter(bool ngram) const override { |
142 | 12.8k | std::shared_lock<std::shared_mutex> lock(*_mtx); |
143 | 12.8k | if (!_nested) { |
144 | 8.73k | return ColumnPredicate::can_do_bloom_filter(ngram); |
145 | 8.73k | } |
146 | 4.15k | return _nested->can_do_bloom_filter(ngram); |
147 | 12.8k | } |
148 | | |
149 | 9.92k | void evaluate_vec(const IColumn& column, uint16_t size, bool* flags) const override { |
150 | 9.92k | std::shared_lock<std::shared_mutex> lock(*_mtx); |
151 | 9.92k | if (!_nested) { |
152 | 9.34M | for (uint16_t i = 0; i < size; ++i) { |
153 | 9.34M | flags[i] = true; |
154 | 9.34M | } |
155 | 5.47k | return; |
156 | 5.47k | } |
157 | 4.45k | _nested->evaluate_vec(column, size, flags); |
158 | 4.45k | } |
159 | | |
160 | 1.43k | void evaluate_and_vec(const IColumn& column, uint16_t size, bool* flags) const override { |
161 | 1.43k | std::shared_lock<std::shared_mutex> lock(*_mtx); |
162 | 1.43k | if (!_nested) { |
163 | 615 | return; |
164 | 615 | } |
165 | 823 | _nested->evaluate_and_vec(column, size, flags); |
166 | 823 | } |
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 | 324 | uint16_t _evaluate_inner(const IColumn& column, uint16_t* sel, uint16_t size) const override { |
199 | 324 | std::shared_lock<std::shared_mutex> lock(*_mtx); |
200 | 324 | if (!_nested) { |
201 | 79 | return size; |
202 | 79 | } |
203 | 245 | return _nested->evaluate(column, sel, size); |
204 | 324 | } |
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 |