github-actions[bot] commented on code in PR #66477:
URL: https://github.com/apache/doris/pull/66477#discussion_r4056558740
##########
be/src/exec/runtime_filter/runtime_filter_wrapper.cpp:
##########
@@ -623,13 +624,47 @@ bool RuntimeFilterWrapper::contain_null() const {
}
std::shared_ptr<const std::vector<uint32_t>>
-RuntimeFilterWrapper::get_or_compute_bucket_prune_hashes(const DataTypePtr&
target_type) const {
+RuntimeFilterWrapper::get_or_compute_bucket_prune_hashes(const DataTypePtr&
target_type,
+
TDistributionHashType::type hash_type,
+ uint32_t bucket_num)
const {
DORIS_CHECK(_state.load() == State::READY);
DORIS_CHECK(_hybrid_set != nullptr);
DORIS_CHECK(target_type != nullptr);
+ DORIS_CHECK_GT(bucket_num, 0);
PrimitiveType primitive_type = target_type->get_primitive_type();
DORIS_CHECK_EQ(primitive_type, _column_return_type);
+ if (hash_type == TDistributionHashType::IDENTITY) {
+ std::scoped_lock lock(_identity_bucket_prune_hashes_mutex);
+ if (auto it = _identity_bucket_prune_hashes.find(bucket_num);
+ it != _identity_bucket_prune_hashes.end()) {
+ return it->second;
+ }
+ _bucket_prune_hashes_started.store(true);
Review Comment:
[P1] Do not materialize this cache on a wrapper that can still be merged.
With multiple producers and both local-merge and ordinary local targets, the
first `merge_from()` stores the producer's shared wrapper in the unfinished
merger, then `publish()` signals ordinary local consumers with that same
wrapper. If pruning reaches this store before the next producer, `merge()` hits
`DORIS_CHECK(!_bucket_prune_hashes_started)`; if it overlaps just after that
check, merge mutates `_hybrid_set` while this branch iterates it. Please delay
exposure until the merger is complete or publish an independent immutable
snapshot, and add a mixed-target scheduling test.
##########
be/src/exec/runtime_filter/runtime_filter_wrapper.cpp:
##########
@@ -623,13 +624,47 @@ bool RuntimeFilterWrapper::contain_null() const {
}
std::shared_ptr<const std::vector<uint32_t>>
-RuntimeFilterWrapper::get_or_compute_bucket_prune_hashes(const DataTypePtr&
target_type) const {
+RuntimeFilterWrapper::get_or_compute_bucket_prune_hashes(const DataTypePtr&
target_type,
+
TDistributionHashType::type hash_type,
+ uint32_t bucket_num)
const {
DORIS_CHECK(_state.load() == State::READY);
DORIS_CHECK(_hybrid_set != nullptr);
DORIS_CHECK(target_type != nullptr);
+ DORIS_CHECK_GT(bucket_num, 0);
PrimitiveType primitive_type = target_type->get_primitive_type();
DORIS_CHECK_EQ(primitive_type, _column_return_type);
+ if (hash_type == TDistributionHashType::IDENTITY) {
+ std::scoped_lock lock(_identity_bucket_prune_hashes_mutex);
+ if (auto it = _identity_bucket_prune_hashes.find(bucket_num);
+ it != _identity_bucket_prune_hashes.end()) {
+ return it->second;
+ }
+ _bucket_prune_hashes_started.store(true);
+ auto buckets = std::make_shared<std::vector<uint32_t>>();
+ buckets->reserve(_hybrid_set->size() + (_hybrid_set->contain_null() ?
1 : 0));
+ auto* iter = _hybrid_set->begin();
+ while (iter->has_next()) {
+ const void* value = iter->get_value();
+ DORIS_CHECK(value != nullptr);
+ if (is_string_type(primitive_type) || primitive_type ==
TYPE_VARBINARY) {
+ const auto* string_value = reinterpret_cast<const
StringRef*>(value);
+ buckets->push_back(RawValue::identity_hash(string_value->data,
string_value->size,
+ primitive_type, 0,
bucket_num));
+ } else {
+ buckets->push_back(
+ RawValue::identity_hash(value, 0, primitive_type, 0,
bucket_num));
Review Comment:
[P1] Normalize remote DATE values before IDENTITY bucket hashing. Exact
filters serialize legacy DATE as a string, but `_assign` deserializes both DATE
and DATETIME with `DatelikeTargetType::DATE_TIME`; this branch therefore hashes
`YYYY-MM-DD 00:00:00`, while tablet routing hashes the stored DATE as
`YYYY-MM-DD`. For `2026-01-02` with 3 buckets those encodings select buckets 2
and 1 respectively, so a remote filter can prune the bucket containing the
match. Please parse TYPE_DATE as DATE (or normalize by `primitive_type`) and
cover a serialized-filter round trip with a non-power-of-two bucket count.
##########
be/src/exec/runtime_filter/runtime_filter_wrapper.cpp:
##########
@@ -623,13 +624,47 @@ bool RuntimeFilterWrapper::contain_null() const {
}
std::shared_ptr<const std::vector<uint32_t>>
-RuntimeFilterWrapper::get_or_compute_bucket_prune_hashes(const DataTypePtr&
target_type) const {
+RuntimeFilterWrapper::get_or_compute_bucket_prune_hashes(const DataTypePtr&
target_type,
+
TDistributionHashType::type hash_type,
+ uint32_t bucket_num)
const {
DORIS_CHECK(_state.load() == State::READY);
DORIS_CHECK(_hybrid_set != nullptr);
DORIS_CHECK(target_type != nullptr);
+ DORIS_CHECK_GT(bucket_num, 0);
PrimitiveType primitive_type = target_type->get_primitive_type();
DORIS_CHECK_EQ(primitive_type, _column_return_type);
+ if (hash_type == TDistributionHashType::IDENTITY) {
+ std::scoped_lock lock(_identity_bucket_prune_hashes_mutex);
+ if (auto it = _identity_bucket_prune_hashes.find(bucket_num);
+ it != _identity_bucket_prune_hashes.end()) {
+ return it->second;
+ }
+ _bucket_prune_hashes_started.store(true);
+ auto buckets = std::make_shared<std::vector<uint32_t>>();
+ buckets->reserve(_hybrid_set->size() + (_hybrid_set->contain_null() ?
1 : 0));
+ auto* iter = _hybrid_set->begin();
+ while (iter->has_next()) {
+ const void* value = iter->get_value();
+ DORIS_CHECK(value != nullptr);
+ if (is_string_type(primitive_type) || primitive_type ==
TYPE_VARBINARY) {
+ const auto* string_value = reinterpret_cast<const
StringRef*>(value);
+ buckets->push_back(RawValue::identity_hash(string_value->data,
string_value->size,
+ primitive_type, 0,
bucket_num));
+ } else {
+ buckets->push_back(
+ RawValue::identity_hash(value, 0, primitive_type, 0,
bucket_num));
+ }
+ iter->next();
+ }
+ if (_hybrid_set->contain_null()) {
+ buckets->push_back(RawValue::identity_hash(nullptr, 0,
primitive_type, 0, bucket_num));
+ }
+ _identity_bucket_prune_hashes.emplace(bucket_num, buckets);
Review Comment:
[P2] Bound the per-bucket-count IDENTITY cache. A scan may select partitions
with different bucket counts, and each distinct count retained here stores one
`uint32_t` per exact-set value. At the defaults, bucket counts 1..768 and a
40,960-value filter retain about 120 MiB and execute about 31 million hashes
for this one runtime filter, with multiple filters multiplying both costs; the
map has no cache-level bound or eviction. Please retain only the deduplicated
selected buckets, compute per-count values transiently, or enforce a strict
aggregate/LRU limit, and cover many distinct partition bucket counts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
