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(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	0	_nested(other._nested ? other._nested->clone(column_id) : nullptr) {}
46	0	~SharedPredicate() override = default;
47	0	std::string debug_string() const override {
48	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
49	0	fmt::memory_buffer debug_string_buffer;
50	0	fmt::format_to(debug_string_buffer, "SharedPredicate({}, nested={})",
51	0	ColumnPredicate::debug_string(), _nested ? _nested->debug_string() : "null");
52	0	return fmt::to_string(debug_string_buffer);
53	0	}
54	0	std::shared_ptr<ColumnPredicate> clone(uint32_t column_id) const override {
55		// All scanner thread should share the same SharedPredicate object.
56	0	return std::const_pointer_cast<ColumnPredicate>(shared_from_this());
57	0	}
58
59	0	PredicateType type() const override {
60	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
61	0	if (!_nested) {
62		// topn filter is le or ge
63	0	return PredicateType::LE;
64	0	}
65	0	return _nested->type();
66	0	}
67	0	PrimitiveType primitive_type() const override {
68	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
69	0	if (!_nested) {
70	0	return PrimitiveType::INVALID_TYPE;
71	0	}
72	0	return _nested->primitive_type();
73	0	}
74
75	0	void set_nested(const std::shared_ptr<ColumnPredicate>& nested) {
76	0	std::unique_lock<std::shared_mutex> lock(*_mtx);
77	0	_nested = nested;
78	0	}
79
80		Status evaluate(const IndexFieldNameAndTypePair& name_with_type, IndexIterator* iterator,
81	0	uint32_t num_rows, roaring::Roaring* bitmap) const override {
82	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
83	0	if (!_nested) {
84	0	return Status::OK();
85	0	}
86	0	return _nested->evaluate(name_with_type, iterator, num_rows, bitmap);
87	0	}
88
89		void evaluate_and(const IColumn& column, const uint16_t* sel, uint16_t size,
90	0	bool* flags) const override {
91	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
92	0	if (!_nested) {
93	0	return;
94	0	}
95	0	return _nested->evaluate_and(column, sel, size, flags);
96	0	}
97
98		void evaluate_or(const IColumn& column, const uint16_t* sel, uint16_t size,
99	0	bool* flags) const override {
100	0	DCHECK(false) << "should not reach here";
101	0	}
102
103	0	bool evaluate_and(const segment_v2::ZoneMap& zone_map) const override {
104	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
105	0	if (!_nested) {
106	0	return ColumnPredicate::evaluate_and(zone_map);
107	0	}
108	0	return _nested->evaluate_and(zone_map);
109	0	}
110
111	0	bool evaluate_del(const segment_v2::ZoneMap& zone_map) const override {
112	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
113	0	if (!_nested) {
114	0	return ColumnPredicate::evaluate_del(zone_map);
115	0	}
116	0	return _nested->evaluate_del(zone_map);
117	0	}
118
119	0	bool evaluate_and(const BloomFilter* bf) const override {
120	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
121	0	if (!_nested) {
122	0	return ColumnPredicate::evaluate_and(bf);
123	0	}
124	0	return _nested->evaluate_and(bf);
125	0	}
126
127	0	bool can_do_bloom_filter(bool ngram) const override {
128	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
129	0	if (!_nested) {
130	0	return ColumnPredicate::can_do_bloom_filter(ngram);
131	0	}
132	0	return _nested->can_do_bloom_filter(ngram);
133	0	}
134
135	0	void evaluate_vec(const IColumn& column, uint16_t size, bool* flags) const override {
136	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
137	0	if (!_nested) {
138	0	for (uint16_t i = 0; i < size; ++i) {
139	0	flags[i] = true;
140	0	}
141	0	return;
142	0	}
143	0	_nested->evaluate_vec(column, size, flags);
144	0	}
145
146	0	void evaluate_and_vec(const IColumn& column, uint16_t size, bool* flags) const override {
147	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
148	0	if (!_nested) {
149	0	return;
150	0	}
151	0	_nested->evaluate_and_vec(column, size, flags);
152	0	}
153
154	0	std::string get_search_str() const override {
155	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
156	0	if (!_nested) {
157	0	DCHECK(false) << "should not reach here";
158	0	}
159	0	return _nested->get_search_str();
160	0	}
161
162	0	bool evaluate_and(ParquetPredicate::ColumnStat* statistic) const override {
163	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
164	0	if (!_nested) {
165		// at the begining _nested will be null, so return true.
166	0	return true;
167	0	}
168	0	return _nested->evaluate_and(statistic);
169	0	}
170
171		bool evaluate_and(ParquetPredicate::CachedPageIndexStat* statistic,
172	0	RowRanges* row_ranges) const override {
173	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
174
175	0	if (!_nested) {
176		// at the begining _nested will be null, so return true.
177	0	row_ranges->add(statistic->row_group_range);
178	0	return true;
179	0	}
180	0	return _nested->evaluate_and(statistic, row_ranges);
181	0	}
182
183		private:
184	0	uint16_t _evaluate_inner(const IColumn& column, uint16_t* sel, uint16_t size) const override {
185	0	std::shared_lock<std::shared_mutex> lock(*_mtx);
186	0	if (!_nested) {
187	0	return size;
188	0	}
189	0	return _nested->evaluate(column, sel, size);
190	0	}
191
192		mutable std::shared_ptr<std::shared_mutex> _mtx;
193		std::shared_ptr<ColumnPredicate> _nested;
194		};
195
196		} //namespace doris

Coverage Report

Created: 2026-05-08 23:56