be/src/storage/predicate/shared_predicate.h

Source
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#pragma once

#include <cstdint>
#include <memory>

#include "common/factory_creator.h"
#include "core/column/column_dictionary.h"
#include "storage/index/bloom_filter/bloom_filter.h"
#include "storage/index/inverted/inverted_index_reader.h"
#include "storage/predicate/column_predicate.h"

namespace doris {

// SharedPredicate only used on topn runtime predicate.
// Runtime predicate globally share one predicate, to ensure that updates can be real-time.
// At the beginning nested predicate may be nullptr, in which case predicate always returns true.
class SharedPredicate final : public ColumnPredicate {
    ENABLE_FACTORY_CREATOR(SharedPredicate);

public:
    SharedPredicate(uint32_t column_id, std::string col_name)
            : ColumnPredicate(column_id, col_name, PrimitiveType::INVALID_TYPE),
              _mtx(std::make_shared<std::shared_mutex>()) {}
    SharedPredicate(const ColumnPredicate& other) = delete;
    SharedPredicate(const SharedPredicate& other, uint32_t column_id)
            : ColumnPredicate(other, column_id),
              _mtx(std::make_shared<std::shared_mutex>()),
              _nested(assert_cast<const SharedPredicate&>(other)._nested
                              ? other._nested->clone(column_id)
                              : nullptr) {}
    ~SharedPredicate() override = default;
    std::string debug_string() const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        fmt::memory_buffer debug_string_buffer;
        fmt::format_to(debug_string_buffer, "SharedPredicate({}, nested={})",
                       ColumnPredicate::debug_string(), _nested ? _nested->debug_string() : "null");
        return fmt::to_string(debug_string_buffer);
    }
    std::shared_ptr<ColumnPredicate> clone(uint32_t column_id) const override {
        // All scanner thread should share the same SharedPredicate object.
        return std::const_pointer_cast<ColumnPredicate>(shared_from_this());
    }

    PredicateType type() const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            // topn filter is le or ge
            return PredicateType::LE;
        }
        return _nested->type();
    }
    PrimitiveType primitive_type() const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            return PrimitiveType::INVALID_TYPE;
        }
        return _nested->primitive_type();
    }

    void set_nested(const std::shared_ptr<ColumnPredicate>& nested) {
        std::unique_lock<std::shared_mutex> lock(*_mtx);
        _nested = nested;
    }

    Status evaluate(const IndexFieldNameAndTypePair& name_with_type, IndexIterator* iterator,
                    uint32_t num_rows, roaring::Roaring* bitmap) const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            return Status::OK();
        }
        return _nested->evaluate(name_with_type, iterator, num_rows, bitmap);
    }

    void evaluate_and(const IColumn& column, const uint16_t* sel, uint16_t size,
                      bool* flags) const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            return;
        }
        return _nested->evaluate_and(column, sel, size, flags);
    }

    void evaluate_or(const IColumn& column, const uint16_t* sel, uint16_t size,
                     bool* flags) const override {
        DCHECK(false) << "should not reach here";
    }

    bool evaluate_and(const segment_v2::ZoneMap& zone_map) const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            return ColumnPredicate::evaluate_and(zone_map);
        }
        return _nested->evaluate_and(zone_map);
    }

    bool evaluate_del(const segment_v2::ZoneMap& zone_map) const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            return ColumnPredicate::evaluate_del(zone_map);
        }
        return _nested->evaluate_del(zone_map);
    }

    bool evaluate_and(const BloomFilter* bf) const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            return ColumnPredicate::evaluate_and(bf);
        }
        return _nested->evaluate_and(bf);
    }

    bool can_do_bloom_filter(bool ngram) const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            return ColumnPredicate::can_do_bloom_filter(ngram);
        }
        return _nested->can_do_bloom_filter(ngram);
    }

    void evaluate_vec(const IColumn& column, uint16_t size, bool* flags) const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            for (uint16_t i = 0; i < size; ++i) {
                flags[i] = true;
            }
            return;
        }
        _nested->evaluate_vec(column, size, flags);
    }

    void evaluate_and_vec(const IColumn& column, uint16_t size, bool* flags) const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            return;
        }
        _nested->evaluate_and_vec(column, size, flags);
    }

    std::string get_search_str() const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            DCHECK(false) << "should not reach here";
        }
        return _nested->get_search_str();
    }

    bool evaluate_and(ParquetPredicate::ColumnStat* statistic) const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            // at the begining _nested will be null, so return true.
            return true;
        }
        return _nested->evaluate_and(statistic);
    }

    bool evaluate_and(ParquetPredicate::CachedPageIndexStat* statistic,
                      RowRanges* row_ranges) const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);

        if (!_nested) {
            // at the begining _nested will be null, so return true.
            row_ranges->add(statistic->row_group_range);
            return true;
        }
        return _nested->evaluate_and(statistic, row_ranges);
    }

private:
    uint16_t _evaluate_inner(const IColumn& column, uint16_t* sel, uint16_t size) const override {
        std::shared_lock<std::shared_mutex> lock(*_mtx);
        if (!_nested) {
            return size;
        }
        return _nested->evaluate(column, sel, size);
    }

    mutable std::shared_ptr<std::shared_mutex> _mtx;
    std::shared_ptr<ColumnPredicate> _nested;
};

} //namespace doris

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

Coverage Report

Created: 2026-03-15 01:14