This is an automated email from the ASF dual-hosted git repository.
mrhhsg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new d35788599ba [fix](be) Give local runtime filter consumers a private
wrapper when the filter is also merged (#68684)
d35788599ba is described below
commit d35788599ba43835615c9c3bbb9813b551157a94
Author: Jerry Hu <[email protected]>
AuthorDate: Wed Oct 14 14:24:51 2026 +0800
[fix](be) Give local runtime filter consumers a private wrapper when the
filter is also merged (#68684)
### What problem does this PR solve?
Issue Number: None
Problem Summary:
A runtime filter without a remote target can still have two kinds of
consumers at once: plain local-target consumers registered in the local
RF mgr, and local-merge-target consumers registered in the global RF
mgr. `RuntimeFilterProducer::publish()` hands the same
`RuntimeFilterWrapper` to both: the local merger, which used to take
over the first producer's wrapper (`_wrapper = other->_wrapper`) and
keep merging later producers into it, and the local consumer, which
immediately starts probing it.
There is no mutual exclusion between the two. The merger mutates the
wrapper (inserting into the hybrid set, clearing it on IN-overflow,
merging the bloom filter directory) under its own lock while a local
consumer's scan thread may be concurrently reading the very same object.
Depending on timing this can:
- hit `DORIS_CHECK(!_bucket_prune_hashes_started)` once the local
consumer has built its bucket-pruning cache, aborting the process on
ASAN/Debug builds or failing the query on Release builds,
- clear the IN set a local consumer is still probing once the merged set
exceeds `runtime_filter_max_in_num`, silently turning its filter into
"match nothing" and dropping rows,
- race on the hybrid set / bloom filter directory memory itself.
This mix of target kinds is an existing, intentional configuration (see
the comment above `publish()`), not a misconfiguration. Broadcast joins
which share one hash table publish the same wrapper from every instance,
so the merger even merged that wrapper into itself.
Fix: establish the invariant that the merger never writes a producer's
wrapper.
- `RuntimeFilterMerger::merge_from()` takes a deep copy
(`RuntimeFilterWrapper::clone()`) of the first wrapper it receives and
merges all later producers into that private copy. Producers' wrappers
are only read by the merger, so the consumers in local RF mgr can keep
using them, and the shared wrapper of a broadcast join is never
modified.
- `RuntimeFilterWrapper::merge()` used to take over the bloom filter of
the input (`_bloom_filter_func = other->_bloom_filter_func`) when an
IN_OR_BLOOM filter which is still an IN filter meets a bloom filter, and
then inserted its own IN set into it — a write to another producer's
wrapper. It now copies the input bloom filter first.
- `RuntimeFilterWrapper::clone()` (with `MinMaxFuncBase::clone()` and
`BloomFilterFuncBase::clone()`) makes the deep copy. The copy of an
IN_OR_BLOOM filter which is still an IN filter does not copy the unused
bloom filter: a later merge either re-initializes it or takes a copy of
the input one.
Cost: one additional deep copy per merged filter (the first product),
both for the local merge on each BE and for the first filter arriving at
the remote merge node. It is the same work as one more merge.
### Release note
None
### Check List (For Author)
- Test:
- Unit Test:
`RuntimeFilterMergerTest.merge_from_never_writes_producer_wrapper`
(merger owns a private copy; a shared wrapper merged again and a
disabled producer do not touch the producer's wrapper),
`RuntimeFilterWrapperTest.TestMergeInWithBloomCopiesBloomFilter` (IN +
BLOOM merge copies the input bloom filter) and `TestClone`, and 7 cases
in `RuntimeFilterProducerTest` covering mixed local/local-merge targets,
the merge-overflow path, the IN_OR_BLOOM merge which takes a bloom
filter, broadcast join with a shared hash table (including an
early-terminated/disabled instance), an all-disabled mixed case, and the
non-mixed case. They fail on the pre-fix code (either by aborting on the
existing `DORIS_CHECK` or by plain assertion failure). No regression
test was added: the failure is timing-dependent on the publish order
across pipeline instances of the same fragment, which cannot be driven
deterministically from SQL.
- Manual test: No need to test (BE-internal object lifecycle change,
fully covered by the new unit tests)
- Behavior changed: Yes — the local merger (and the merge node) now
merges into a private copy of the first received filter instead of the
producer's own object, so consumers in local RF mgr never observe a
concurrently mutated filter. No code depended on the old (racy,
effectively undefined) behavior.
- Does this need documentation: No
---
be/src/exec/runtime_filter/runtime_filter_merger.h | 22 +-
be/src/exec/runtime_filter/runtime_filter_mgr.cpp | 6 +-
.../runtime_filter/runtime_filter_producer.cpp | 51 ++-
.../exec/runtime_filter/runtime_filter_wrapper.cpp | 39 +-
.../exec/runtime_filter/runtime_filter_wrapper.h | 8 +-
be/src/exprs/bloom_filter_func.h | 34 ++
be/src/exprs/filter_base.h | 1 +
be/src/exprs/minmax_predicate.h | 11 +
.../runtime_filter/runtime_filter_merger_test.cpp | 105 +++++
.../runtime_filter_producer_test.cpp | 496 +++++++++++++++++++++
.../runtime_filter/runtime_filter_wrapper_test.cpp | 313 +++++++++++++
11 files changed, 1072 insertions(+), 14 deletions(-)
diff --git a/be/src/exec/runtime_filter/runtime_filter_merger.h
b/be/src/exec/runtime_filter/runtime_filter_merger.h
index a7bd3605f71..3e0b465b1f8 100644
--- a/be/src/exec/runtime_filter/runtime_filter_merger.h
+++ b/be/src/exec/runtime_filter/runtime_filter_merger.h
@@ -58,7 +58,17 @@ public:
// If input is a disabled predicate, the final result is a disabled
predicate.
// Returns true only for the call that makes the merger ready.
- Status merge_from(const RuntimeFilter* other, bool* ready) {
+ //
+ // `other_wrapper_exclusively_owned` must be true only when the caller
guarantees `other`'s
+ // wrapper has no other reader and will not be read again after this call
(e.g. a producer
+ // whose wrapper is never signaled to a plain local consumer, or an
RPC-only filter built
+ // solely to carry one merge request). The merger then takes the wrapper
over directly
+ // instead of paying for a deep copy. It must stay false whenever
`other`'s wrapper may still
+ // be used by consumers in local RF mgr of the same instance, or is shared
by all producers of
+ // a broadcast join with a shared hash table: the merger must never write
such a wrapper, so
+ // it only merges into its own copy.
+ Status merge_from(const RuntimeFilter* other, bool* ready,
+ bool other_wrapper_exclusively_owned = false) {
std::unique_lock<std::recursive_mutex> l(_rmtx);
_received_producer_num++;
if (_expected_producer_num < _received_producer_num) {
@@ -70,11 +80,13 @@ public:
_rf_state = State::READY;
}
if (_wrapper->get_state() == RuntimeFilterWrapper::State::UNINITED) {
- _wrapper = other->_wrapper;
- return Status::OK();
+ if (other_wrapper_exclusively_owned) {
+ _wrapper = other->_wrapper;
+ return Status::OK();
+ }
+ return other->_wrapper->clone(&_wrapper);
}
- auto st = _wrapper->merge(other->_wrapper.get());
- return st;
+ return _wrapper->merge(other->_wrapper.get(),
other_wrapper_exclusively_owned);
}
// Only raise the expected producer count. RuntimeFilterMgr may compute the
diff --git a/be/src/exec/runtime_filter/runtime_filter_mgr.cpp
b/be/src/exec/runtime_filter/runtime_filter_mgr.cpp
index 29912f5c441..468e8851720 100644
--- a/be/src/exec/runtime_filter/runtime_filter_mgr.cpp
+++ b/be/src/exec/runtime_filter/runtime_filter_mgr.cpp
@@ -589,7 +589,11 @@ Status
RuntimeFilterMergeControllerEntity::merge(std::shared_ptr<QueryContext> q
RETURN_IF_ERROR(tmp_filter->assign(*request, attach_data));
- RETURN_IF_ERROR(cnt_val.merger->merge_from(tmp_filter.get(),
&is_ready));
+ // `tmp_filter` is created solely to carry this one RPC's filter into
the merger and is
+ // discarded right after; nothing else reads its wrapper, so the
merger may take it over
+ // without a copy.
+ RETURN_IF_ERROR(cnt_val.merger->merge_from(tmp_filter.get(), &is_ready,
+
/*other_wrapper_exclusively_owned=*/true));
cnt_val.arrive_id.insert(UniqueId(request->fragment_instance_id()));
}
diff --git a/be/src/exec/runtime_filter/runtime_filter_producer.cpp
b/be/src/exec/runtime_filter/runtime_filter_producer.cpp
index 5a342911627..42d1938b790 100644
--- a/be/src/exec/runtime_filter/runtime_filter_producer.cpp
+++ b/be/src/exec/runtime_filter/runtime_filter_producer.cpp
@@ -26,6 +26,18 @@
#include "util/brpc_closure.h"
namespace doris {
+namespace {
+// Nothing other than the merger can read the wrapper `merge_from` is about to
receive: either no
+// plain local consumer of `state` exists at all
(`_send_to_local_targets(state, this, false)` is
+// then a no-op), or `merger` expects only this one producer, so `merge_from`
becomes ready on
+// this very call and never merges another producer into the wrapper it adopts
here, even though
+// a plain local consumer also reads it.
+bool wrapper_has_no_other_reader(RuntimeState* state, int filter_id,
RuntimeFilterMerger* merger) {
+ return
state->local_runtime_filter_mgr()->get_consume_filters(filter_id).empty() ||
+ merger->get_expected_producer_num() == 1;
+}
+} // namespace
+
Status RuntimeFilterProducer::_send_to_remote_targets(RuntimeState* state,
RuntimeFilter*
merger_filter) {
TNetworkAddress addr;
@@ -44,11 +56,21 @@ Status
RuntimeFilterProducer::_send_to_local_targets(RuntimeState* state, Runtim
return Status::OK();
};
+// Pre-existing complexity debt from the target/broadcast/remote branching
below, already over
+// threshold before this change; the ownership refinement above is factored
out to keep the
+// increase small, but not addressed further here to keep this diff focused.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity)
Status RuntimeFilterProducer::publish(RuntimeState* state, bool
build_hash_table) {
std::unique_lock<std::recursive_mutex> l(_rmtx);
_check_state({State::READY_TO_PUBLISH});
- auto do_merge = [&]() {
+ // `allow_relaxed_ownership` is only meaningful when
`other_wrapper_exclusively_owned` is
+ // false: it lets this call refine that decision using the local-merge
context it fetches
+ // below, instead of the caller guessing it without the context. It must
stay false for the
+ // remote-target call site, whose `true` already holds unconditionally for
a different
+ // reason (see its call below), and for the broadcast case, whose
producers alias one shared
+ // wrapper (see
`RuntimeFilterProducerTest.publish_mixed_targets_shared_wrapper`).
+ auto do_merge = [&](bool other_wrapper_exclusively_owned, bool
allow_relaxed_ownership) {
if (!_need_do_merge(state)) {
// when global consumer not exist, send_to_local_targets will do
nothing, so merge rf is useless
return Status::OK();
@@ -60,8 +82,12 @@ Status RuntimeFilterProducer::publish(RuntimeState* state,
bool build_hash_table
// Filter was removed during a recursive CTE stage reset; this
producer is stale.
return Status::OK();
}
+ if (!other_wrapper_exclusively_owned && allow_relaxed_ownership) {
+ other_wrapper_exclusively_owned = wrapper_has_no_other_reader(
+ state, _wrapper->filter_id(), context->merger.get());
+ }
bool ready = false;
- RETURN_IF_ERROR(context->merger->merge_from(this, &ready));
+ RETURN_IF_ERROR(context->merger->merge_from(this, &ready,
other_wrapper_exclusively_owned));
if (ready) {
if (_has_remote_target) {
RETURN_IF_ERROR(_send_to_remote_targets(state,
context->merger.get()));
@@ -75,13 +101,30 @@ Status RuntimeFilterProducer::publish(RuntimeState* state,
bool build_hash_table
if (!_has_remote_target) {
// A runtime filter may have multiple targets and some of those are
local-merge RF and others are not.
// So for all runtime filters' producers, `publish` should notify all
consumers in global RF mgr which manages local-merge RF and local RF mgr which
manages others.
- RETURN_IF_ERROR(do_merge());
+ // The merger never writes this wrapper while a plain local consumer
may still read it
+ // (see `RuntimeFilterMerger::merge_from`), so by default the
consumers in local RF mgr
+ // can use it right away while the merge of the other producers goes
on. `do_merge` may
+ // still find no such consumer exists (or this is the only producer)
and adopt the
+ // wrapper directly. Broadcast producers alias one shared wrapper
across instances (see
+ // `publish_mixed_targets_shared_wrapper`), so relaxing ownership is
unsound for them;
+ // `build_bf_by_runtime_size` is excluded too, to keep this call's
precondition the same
+ // simple "nothing else can reach this wrapper" rule as the
remote-target call below,
+ // without re-deriving it for a filter whose size-sync step this
branch never exercises.
+ RETURN_IF_ERROR(do_merge(/*other_wrapper_exclusively_owned=*/false,
+
/*allow_relaxed_ownership=*/!_is_broadcast_join &&
+
!_wrapper->build_bf_by_runtime_size()));
RETURN_IF_ERROR(_send_to_local_targets(state, this, false));
} else if (build_hash_table) {
if (_is_broadcast_join) {
RETURN_IF_ERROR(_send_to_remote_targets(state, this));
} else {
- RETURN_IF_ERROR(do_merge());
+ // This path never hands `_wrapper` to a plain local consumer
(that only happens in
+ // the `!_has_remote_target` branch above), and the shared-wrapper
broadcast-join
+ // case is the other arm of this `if`, so no other producer
aliases it either. The
+ // only remaining reference after this call is the one
`_wrapper.reset()` below
+ // drops, so the merger may take `_wrapper` over directly instead
of cloning it.
+ RETURN_IF_ERROR(do_merge(/*other_wrapper_exclusively_owned=*/true,
+ /*allow_relaxed_ownership=*/false));
}
} else {
if (!_is_broadcast_join) {
diff --git a/be/src/exec/runtime_filter/runtime_filter_wrapper.cpp
b/be/src/exec/runtime_filter/runtime_filter_wrapper.cpp
index cd5fc2c04bd..66fa692b65a 100644
--- a/be/src/exec/runtime_filter/runtime_filter_wrapper.cpp
+++ b/be/src/exec/runtime_filter/runtime_filter_wrapper.cpp
@@ -125,7 +125,11 @@ bool RuntimeFilterWrapper::build_bf_by_runtime_size()
const {
return _bloom_filter_func ? _bloom_filter_func->build_bf_by_runtime_size()
: false;
}
-Status RuntimeFilterWrapper::merge(const RuntimeFilterWrapper* other) {
+// Pre-existing size/complexity debt from the per-filter-type switch below,
already over
+// threshold before this change; not addressed here to keep this diff focused.
+//
NOLINTNEXTLINE(readability-function-size,readability-function-cognitive-complexity)
+Status RuntimeFilterWrapper::merge(const RuntimeFilterWrapper* other,
+ bool other_exclusively_owned) {
DORIS_CHECK(!_bucket_prune_hashes_started.load());
if (_state == State::DISABLED) {
return Status::OK();
@@ -189,10 +193,16 @@ Status RuntimeFilterWrapper::merge(const
RuntimeFilterWrapper* other) {
RETURN_IF_ERROR(_bloom_filter_func->init_with_fixed_length(0));
RETURN_IF_ERROR(_change_to_bloom_filter());
}
- } else {
- // case1&case2: use input bf directly and insert hybrid set
data into bf
+ } else if (other_exclusively_owned) {
+ // `other` has no other reader, so its bf can be taken over
directly instead of
+ // cloned before inserting the hybrid set data into it.
_bloom_filter_func = other->_bloom_filter_func;
RETURN_IF_ERROR(_change_to_bloom_filter());
+ } else {
+ // case1&case2: use a copy of input bf and insert hybrid set
data into it. `other`
+ // may still be used by its own consumers, so it must not be
written.
+
RETURN_IF_ERROR(other->_bloom_filter_func->clone(&_bloom_filter_func, true));
+ RETURN_IF_ERROR(_change_to_bloom_filter());
}
} else {
if (other_filter_type == RuntimeFilterType::IN_FILTER) {
@@ -212,6 +222,29 @@ Status RuntimeFilterWrapper::merge(const
RuntimeFilterWrapper* other) {
return Status::OK();
}
+Status RuntimeFilterWrapper::clone(std::shared_ptr<RuntimeFilterWrapper>* res)
const {
+ auto cloned = std::make_shared<RuntimeFilterWrapper>(_column_return_type,
_filter_type,
+ _filter_id,
_state.load(), _max_in_num);
+ cloned->_disable_always_true_logic = _disable_always_true_logic;
+ cloned->_reason.update(_reason.status());
+ if (_hybrid_set) {
+ cloned->_hybrid_set.reset(create_set(_column_return_type,
_hybrid_set->null_aware()));
+ cloned->_hybrid_set->insert(_hybrid_set.get());
+ }
+ if (_minmax_func) {
+ cloned->_minmax_func.reset(_minmax_func->clone());
+ }
+ if (_bloom_filter_func) {
+ // An IN_OR_BLOOM filter which is still an IN filter never reads its
bloom filter: a
+ // later merge which changes the copy to a bloom filter either
re-initializes the bloom
+ // filter or takes a copy of the input one first. So do not copy the
unused bloom filter.
+ RETURN_IF_ERROR(_bloom_filter_func->clone(
+ &cloned->_bloom_filter_func, get_real_type() ==
RuntimeFilterType::BLOOM_FILTER));
+ }
+ *res = std::move(cloned);
+ return Status::OK();
+}
+
Status RuntimeFilterWrapper::_assign(const PInFilter& in_filter, bool
contain_null) {
if (contain_null) {
_hybrid_set->insert((const void*)nullptr);
diff --git a/be/src/exec/runtime_filter/runtime_filter_wrapper.h
b/be/src/exec/runtime_filter/runtime_filter_wrapper.h
index 7fb770e8855..8aa303733d0 100644
--- a/be/src/exec/runtime_filter/runtime_filter_wrapper.h
+++ b/be/src/exec/runtime_filter/runtime_filter_wrapper.h
@@ -57,7 +57,13 @@ public:
Status init(const size_t runtime_size);
Status insert(const ColumnPtr& column, size_t start);
- Status merge(const RuntimeFilterWrapper* wrapper);
+ // `other_exclusively_owned` must be true only when `other` has no other
reader and will not
+ // be read again after this call: a transitional BLOOM directory taken
from it is then moved
+ // in directly instead of cloned. See `RuntimeFilterMerger::merge_from`.
+ Status merge(const RuntimeFilterWrapper* other, bool
other_exclusively_owned = false);
+ // Deep copy of the filter. The result shares no mutable data with this
wrapper, so the
+ // merger can keep merging into the copy while this one is already used by
consumers.
+ Status clone(std::shared_ptr<RuntimeFilterWrapper>* res) const;
template <class T>
Status assign(const T& request, butil::IOBufAsZeroCopyInputStream* data);
diff --git a/be/src/exprs/bloom_filter_func.h b/be/src/exprs/bloom_filter_func.h
index 7e5eaeecb7e..ba4bf8ebb77 100644
--- a/be/src/exprs/bloom_filter_func.h
+++ b/be/src/exprs/bloom_filter_func.h
@@ -132,6 +132,12 @@ public:
_bloom_filter = other_func->_bloom_filter;
}
+ // Deep copy. The returned filter owns a separately allocated bloom
filter. Only the
+ // parameters are copied if `copy_bloom_filter` is false, e.g. the filter
is still used as
+ // an IN filter, so its bloom filter is never read.
+ virtual Status clone(std::shared_ptr<BloomFilterFuncBase>* res,
+ bool copy_bloom_filter) const = 0;
+
virtual void insert_set(std::shared_ptr<HybridSetBase> set) = 0;
virtual void insert_fixed_len(const ColumnPtr& column, size_t start) = 0;
@@ -166,6 +172,26 @@ private:
}
protected:
+ Status _deep_copy_from(const BloomFilterFuncBase& other, bool
copy_bloom_filter) {
+ _bloom_filter_length = other._bloom_filter_length;
+ _runtime_bloom_filter_min_size = other._runtime_bloom_filter_min_size;
+ _runtime_bloom_filter_max_size = other._runtime_bloom_filter_max_size;
+ _build_bf_by_runtime_size = other._build_bf_by_runtime_size;
+ _bloom_filter_size_calculated_by_ndv =
other._bloom_filter_size_calculated_by_ndv;
+ if (!copy_bloom_filter) {
+ return Status::OK();
+ }
+ if (other._bloom_filter == nullptr) {
+ // `other` is not initialized yet, e.g. the producer is disabled
before it is built.
+ return Status::OK();
+ }
+ _bloom_filter_alloced = other._bloom_filter_alloced;
+ _bloom_filter.reset(BloomFilterAdaptor::create(_null_aware));
+ RETURN_IF_ERROR(_bloom_filter->init(_bloom_filter_alloced));
+ _bloom_filter->set_contain_null(other.contain_null());
+ return _bloom_filter->merge(other._bloom_filter.get());
+ }
+
int64_t _bloom_filter_alloced = 0;
std::shared_ptr<BloomFilterAdaptor> _bloom_filter;
int64_t _bloom_filter_length;
@@ -179,6 +205,14 @@ template <PrimitiveType type>
class BloomFilterFunc final : public BloomFilterFuncBase {
public:
BloomFilterFunc(bool null_aware) : BloomFilterFuncBase(null_aware) {}
+
+ Status clone(std::shared_ptr<BloomFilterFuncBase>* res, bool
copy_bloom_filter) const override {
+ auto cloned = std::make_shared<BloomFilterFunc<type>>(_null_aware);
+ RETURN_IF_ERROR(cloned->_deep_copy_from(*this, copy_bloom_filter));
+ *res = std::move(cloned);
+ return Status::OK();
+ }
+
void insert_set(std::shared_ptr<HybridSetBase> set) override {
OpV2::insert_set(*_bloom_filter, set);
_bloom_filter->set_contain_null(set->contain_null());
diff --git a/be/src/exprs/filter_base.h b/be/src/exprs/filter_base.h
index 25778059886..ca808b310ca 100644
--- a/be/src/exprs/filter_base.h
+++ b/be/src/exprs/filter_base.h
@@ -25,6 +25,7 @@ class FilterBase {
public:
FilterBase(bool null_aware) : _null_aware(null_aware) {}
bool contain_null() const { return _null_aware && _contain_null; }
+ bool null_aware() const { return _null_aware; }
void set_contain_null(bool contain_null) { _contain_null |= contain_null; }
diff --git a/be/src/exprs/minmax_predicate.h b/be/src/exprs/minmax_predicate.h
index b0f1b27e7a8..77fe7ec6bc5 100644
--- a/be/src/exprs/minmax_predicate.h
+++ b/be/src/exprs/minmax_predicate.h
@@ -37,6 +37,8 @@ public:
virtual Status assign(void* min_data, void* max_data) = 0;
// merge from other minmax_func
virtual Status merge(MinMaxFuncBase* minmax_func) = 0;
+ // Deep copy. The returned object shares nothing with this one.
+ virtual MinMaxFuncBase* clone() const = 0;
virtual ~MinMaxFuncBase() = default;
virtual void to_pb(PMinMaxFilter* filter) = 0;
@@ -87,6 +89,15 @@ public:
return Status::OK();
}
+ MinMaxFuncBase* clone() const override {
+ auto* cloned = new MinMaxNumFunc(_null_aware);
+ cloned->_max = _max;
+ cloned->_min = _min;
+ cloned->_min_value_set = _min_value_set;
+ cloned->_contain_null = _contain_null;
+ return cloned;
+ }
+
void* get_max() override { return &_max; }
void* get_min() override { return &_min; }
diff --git a/be/test/exec/runtime_filter/runtime_filter_merger_test.cpp
b/be/test/exec/runtime_filter/runtime_filter_merger_test.cpp
index 3d0bb701c8b..5823d39797b 100644
--- a/be/test/exec/runtime_filter/runtime_filter_merger_test.cpp
+++ b/be/test/exec/runtime_filter/runtime_filter_merger_test.cpp
@@ -20,8 +20,11 @@
#include <glog/logging.h>
#include <gtest/gtest.h>
+#include "core/data_type/data_type_number.h"
#include "exec/runtime_filter/runtime_filter_producer.h"
#include "exec/runtime_filter/runtime_filter_test_utils.h"
+#include "exprs/hybrid_set.h"
+#include "testutil/column_helper.h"
namespace doris {
@@ -95,6 +98,108 @@ public:
}
};
+// The merger merges into a private copy of the first wrapper and never writes
a producer's
+// wrapper, which may be shared with consumers in local RF mgr or with other
producers.
+TEST_F(RuntimeFilterMergerTest, merge_from_never_writes_producer_wrapper) {
+ std::shared_ptr<RuntimeFilterMerger> merger;
+ auto desc = TRuntimeFilterDescBuilder().build();
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(RuntimeFilterMerger::create(_query_ctx.get(),
&desc, &merger));
+ merger->increase_expected_producer_num(3);
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_producer_runtime_filter(desc,
&producer));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->init(1));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer->insert(ColumnHelper::create_column<DataTypeInt32>({5}),
0));
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+ auto wrapper = producer->wrapper();
+ auto hashes =
wrapper->get_or_compute_bucket_prune_hashes(std::make_shared<DataTypeInt32>());
+ ASSERT_EQ(hashes->size(), 1);
+
+ bool ready = false;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(merger->merge_from(producer.get(),
&ready));
+ ASSERT_FALSE(ready);
+ ASSERT_NE(merger->_wrapper, wrapper);
+ ASSERT_NE(merger->_wrapper->hybrid_set(), wrapper->hybrid_set());
+ ASSERT_EQ(merger->_wrapper->_state, RuntimeFilterWrapper::State::READY);
+ ASSERT_EQ(merger->_wrapper->hybrid_set()->size(), 1);
+ ASSERT_FALSE(merger->_wrapper->_bucket_prune_hashes_started.load());
+
+ // The same wrapper published by another producer of a broadcast join
changes nothing.
+ std::shared_ptr<RuntimeFilterProducer> producer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_producer_runtime_filter(desc,
&producer2));
+ producer2->set_wrapper(wrapper);
+
producer2->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(merger->merge_from(producer2.get(),
&ready));
+ ASSERT_FALSE(ready);
+ ASSERT_EQ(merger->_wrapper->hybrid_set()->size(), 1);
+
+ // A disabled producer (e.g. of an early terminated instance) disables the
merger's copy only.
+ auto terminated_mgr = std::make_unique<RuntimeFilterMgr>(false);
+ auto terminated_state =
+ RuntimeState::create_unique(TUniqueId(), 0, _query_options,
_query_ctx->query_globals,
+ ExecEnv::GetInstance(),
_query_ctx.get());
+ terminated_state->set_runtime_filter_mgr(terminated_mgr.get());
+ terminated_state->set_desc_tbl(&_tbl);
+ std::shared_ptr<RuntimeFilterProducer> producer3;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ terminated_state->register_producer_runtime_filter(desc,
&producer3));
+
producer3->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::DISABLED,
+ "skip all rf process");
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(merger->merge_from(producer3.get(),
&ready));
+ ASSERT_TRUE(ready);
+ ASSERT_EQ(merger->_wrapper->_state, RuntimeFilterWrapper::State::DISABLED);
+ ASSERT_NE(merger->_wrapper->_reason.status().msg().find("skip all rf
process"),
+ std::string::npos);
+
+ int32_t five = 5;
+ ASSERT_EQ(wrapper->_state, RuntimeFilterWrapper::State::READY);
+ ASSERT_EQ(wrapper->hybrid_set()->size(), 1);
+ ASSERT_TRUE(wrapper->hybrid_set()->find(&five));
+
ASSERT_EQ(wrapper->get_or_compute_bucket_prune_hashes(std::make_shared<DataTypeInt32>()),
+ hashes);
+}
+
+// A caller who guarantees `other`'s wrapper has no other reader (e.g. an
RPC-only filter built
+// solely for this merge, see `RuntimeFilterMergeControllerEntity::merge`) may
let the merger take
+// it over directly instead of paying for a deep copy.
+TEST_F(RuntimeFilterMergerTest, merge_from_exclusively_owned_takes_ownership) {
+ std::shared_ptr<RuntimeFilterMerger> merger;
+ auto desc = TRuntimeFilterDescBuilder().build();
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(RuntimeFilterMerger::create(_query_ctx.get(),
&desc, &merger));
+ merger->increase_expected_producer_num(2);
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_producer_runtime_filter(desc,
&producer));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->init(1));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer->insert(ColumnHelper::create_column<DataTypeInt32>({5}),
0));
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+ auto wrapper = producer->wrapper();
+
+ bool ready = false;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ merger->merge_from(producer.get(), &ready,
/*other_wrapper_exclusively_owned=*/true));
+ ASSERT_FALSE(ready);
+ // No clone: the merger took over the exclusively owned wrapper directly.
+ ASSERT_EQ(merger->_wrapper, wrapper);
+
+ std::shared_ptr<RuntimeFilterProducer> producer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_producer_runtime_filter(desc,
&producer2));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer2->init(1));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer2->insert(ColumnHelper::create_column<DataTypeInt32>({6}),
0));
+
producer2->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ merger->merge_from(producer2.get(), &ready,
/*other_wrapper_exclusively_owned=*/true));
+ ASSERT_TRUE(ready);
+ ASSERT_EQ(merger->_wrapper->hybrid_set()->size(), 2);
+}
+
TEST_F(RuntimeFilterMergerTest, basic) {
test_merge_from(RuntimeFilterWrapper::State::READY,
RuntimeFilterWrapper::State::READY,
RuntimeFilterWrapper::State::READY,
RuntimeFilterWrapper::State::READY);
diff --git a/be/test/exec/runtime_filter/runtime_filter_producer_test.cpp
b/be/test/exec/runtime_filter/runtime_filter_producer_test.cpp
index aa0f93e95ed..42062eec3e4 100644
--- a/be/test/exec/runtime_filter/runtime_filter_producer_test.cpp
+++ b/be/test/exec/runtime_filter/runtime_filter_producer_test.cpp
@@ -20,8 +20,17 @@
#include <glog/logging.h>
#include <gtest/gtest.h>
+#include <algorithm>
+#include <numeric>
+#include <string>
+
+#include "core/data_type/data_type_number.h"
#include "exec/runtime_filter/runtime_filter_consumer.h"
+#include "exec/runtime_filter/runtime_filter_merger.h"
#include "exec/runtime_filter/runtime_filter_test_utils.h"
+#include "exprs/bloom_filter_func.h"
+#include "exprs/hybrid_set.h"
+#include "testutil/column_helper.h"
namespace doris {
@@ -175,6 +184,493 @@ TEST_F(RuntimeFilterProducerTest, set_disable) {
ASSERT_EQ(consumer->_wrapper->_state,
RuntimeFilterWrapper::State::DISABLED);
}
+// A runtime filter whose consumers are partly in local RF mgr and partly in
global RF mgr.
+TEST_F(RuntimeFilterProducerTest,
publish_mixed_targets_local_consumer_gets_private_wrapper) {
+ auto desc = TRuntimeFilterDescBuilder()
+ .set_build_bf_by_runtime_size(false)
+ .set_is_broadcast_join(false)
+ .add_planId_to_target_expr(0)
+ .add_planId_to_target_expr(1)
+ .build();
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_producer_runtime_filter(desc,
&producer));
+ std::shared_ptr<RuntimeFilterProducer> producer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_producer_runtime_filter(desc,
&producer2));
+
+ std::shared_ptr<RuntimeFilterConsumer> local_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_consumer_runtime_filter(desc, false,
0, &local_consumer));
+ std::shared_ptr<RuntimeFilterConsumer> local_consumer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_consumer_runtime_filter(desc, false,
0, &local_consumer2));
+ std::shared_ptr<RuntimeFilterConsumer> merge_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_consumer_runtime_filter(desc, true,
1, &merge_consumer));
+
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->init(1));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer->insert(ColumnHelper::create_column<DataTypeInt32>({5}),
0));
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer2->init(1));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer2->insert(ColumnHelper::create_column<DataTypeInt32>({6}),
0));
+
producer2->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->publish(_runtime_states[0].get(),
true));
+ ASSERT_EQ(local_consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ ASSERT_EQ(merge_consumer->_rf_state,
RuntimeFilterConsumer::State::NOT_READY);
+
+ // The local consumer builds its bucket prune cache before the merger
receives producer2.
+ std::vector<RuntimeFilterExprPtr> exprs;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(local_consumer->acquire_expr(exprs));
+ ASSERT_EQ(exprs.size(), 1);
+ auto hashes =
exprs[0]->get_bucket_prune_hashes(std::make_shared<DataTypeInt32>());
+ ASSERT_EQ(hashes->size(), 1);
+
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer2->publish(_runtime_states[1].get(),
true));
+ ASSERT_EQ(merge_consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ ASSERT_EQ(local_consumer2->_rf_state, RuntimeFilterConsumer::State::READY);
+
+ int32_t five = 5;
+ int32_t six = 6;
+ auto merged_set = merge_consumer->_wrapper->hybrid_set();
+ ASSERT_EQ(merged_set->size(), 2);
+ ASSERT_TRUE(merged_set->find(&five));
+ ASSERT_TRUE(merged_set->find(&six));
+
+ auto local_set = local_consumer->_wrapper->hybrid_set();
+ ASSERT_EQ(local_set->size(), 1);
+ ASSERT_TRUE(local_set->find(&five));
+ ASSERT_FALSE(local_set->find(&six));
+ auto local_set2 = local_consumer2->_wrapper->hybrid_set();
+ ASSERT_EQ(local_set2->size(), 1);
+ ASSERT_TRUE(local_set2->find(&six));
+ ASSERT_FALSE(local_set2->find(&five));
+
+ ASSERT_NE(local_consumer->_wrapper, merge_consumer->_wrapper);
+ ASSERT_NE(local_consumer2->_wrapper, merge_consumer->_wrapper);
+ ASSERT_NE(local_consumer->_wrapper, local_consumer2->_wrapper);
+ ASSERT_NE(local_set, merged_set);
+ ASSERT_NE(local_set2, merged_set);
+ ASSERT_NE(local_consumer->_wrapper->bloom_filter_func(),
+ merge_consumer->_wrapper->bloom_filter_func());
+
+
ASSERT_EQ(exprs[0]->get_bucket_prune_hashes(std::make_shared<DataTypeInt32>()),
hashes);
+ ASSERT_EQ(hashes->size(), 1);
+
+ std::vector<RuntimeFilterExprPtr> merged_exprs;
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(merge_consumer->acquire_expr(merged_exprs));
+ ASSERT_EQ(merged_exprs.size(), 1);
+
ASSERT_EQ(merged_exprs[0]->get_bucket_prune_hashes(std::make_shared<DataTypeInt32>())->size(),
+ 2);
+}
+
+TEST_F(RuntimeFilterProducerTest,
publish_mixed_targets_merge_overflow_keeps_local_filter) {
+ auto desc = TRuntimeFilterDescBuilder()
+ .set_build_bf_by_runtime_size(false)
+ .set_is_broadcast_join(false)
+ .set_type(TRuntimeFilterType::IN)
+ .add_planId_to_target_expr(0)
+ .add_planId_to_target_expr(1)
+ .build();
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_producer_runtime_filter(desc,
&producer));
+ std::shared_ptr<RuntimeFilterProducer> producer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_producer_runtime_filter(desc,
&producer2));
+
+ // Only the first instance has a consumer in local RF mgr.
+ std::shared_ptr<RuntimeFilterConsumer> local_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_consumer_runtime_filter(desc, false,
0, &local_consumer));
+ std::shared_ptr<RuntimeFilterConsumer> merge_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_consumer_runtime_filter(desc, true,
1, &merge_consumer));
+
+ // Each producer is within runtime_filter_max_in_num, but the merged one
is not.
+ std::vector<int32_t> data(600);
+ std::iota(data.begin(), data.end(), 0);
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->init(data.size()));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer->insert(ColumnHelper::create_column<DataTypeInt32>(data),
0));
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+ std::iota(data.begin(), data.end(), 600);
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer2->init(data.size()));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+
producer2->insert(ColumnHelper::create_column<DataTypeInt32>(data), 0));
+
producer2->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->publish(_runtime_states[0].get(),
true));
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer2->publish(_runtime_states[1].get(),
true));
+
+ ASSERT_EQ(merge_consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ ASSERT_EQ(merge_consumer->_wrapper->get_state(),
RuntimeFilterWrapper::State::DISABLED);
+
+ ASSERT_EQ(local_consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ ASSERT_EQ(local_consumer->_wrapper->get_state(),
RuntimeFilterWrapper::State::READY);
+ auto local_set = local_consumer->_wrapper->hybrid_set();
+ ASSERT_EQ(local_set->size(), 600);
+ int32_t value = 0;
+ ASSERT_TRUE(local_set->find(&value));
+ value = 599;
+ ASSERT_TRUE(local_set->find(&value));
+ value = 600;
+ ASSERT_FALSE(local_set->find(&value));
+}
+
+// Broadcast join producers which share one hash table also share one wrapper.
+TEST_F(RuntimeFilterProducerTest, publish_mixed_targets_shared_wrapper) {
+ auto desc = TRuntimeFilterDescBuilder()
+ .set_build_bf_by_runtime_size(false)
+ .set_is_broadcast_join(true)
+ .add_planId_to_target_expr(0)
+ .add_planId_to_target_expr(1)
+ .build();
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_producer_runtime_filter(desc,
&producer));
+ std::shared_ptr<RuntimeFilterProducer> producer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_producer_runtime_filter(desc,
&producer2));
+
+ std::shared_ptr<RuntimeFilterConsumer> local_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_consumer_runtime_filter(desc, false,
0, &local_consumer));
+ std::shared_ptr<RuntimeFilterConsumer> merge_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_consumer_runtime_filter(desc, true,
1, &merge_consumer));
+
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->init(1));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer->insert(ColumnHelper::create_column<DataTypeInt32>({5}),
0));
+ producer2->set_wrapper(producer->wrapper());
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+
producer2->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->publish(_runtime_states[0].get(),
true));
+ std::vector<RuntimeFilterExprPtr> exprs;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(local_consumer->acquire_expr(exprs));
+ ASSERT_EQ(exprs.size(), 1);
+
ASSERT_EQ(exprs[0]->get_bucket_prune_hashes(std::make_shared<DataTypeInt32>())->size(),
1);
+
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer2->publish(_runtime_states[1].get(),
true));
+ ASSERT_EQ(merge_consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ auto merged_set = merge_consumer->_wrapper->hybrid_set();
+ ASSERT_EQ(merged_set->size(), 1);
+ int32_t five = 5;
+ ASSERT_TRUE(merged_set->find(&five));
+ ASSERT_NE(local_consumer->_wrapper, merge_consumer->_wrapper);
+}
+
+// An early terminated instance of a broadcast join publishes its own disabled
wrapper, which must
+// not disable the shared wrapper in place: the local consumers of other
instances still use it.
+TEST_F(RuntimeFilterProducerTest,
publish_mixed_targets_shared_wrapper_with_disabled_instance) {
+ auto desc = TRuntimeFilterDescBuilder()
+ .set_build_bf_by_runtime_size(false)
+ .set_is_broadcast_join(true)
+ .add_planId_to_target_expr(0)
+ .add_planId_to_target_expr(1)
+ .build();
+
+ auto terminated_mgr = std::make_unique<RuntimeFilterMgr>(false);
+ auto terminated_state =
+ RuntimeState::create_unique(TUniqueId(), 0, _query_options,
_query_ctx->query_globals,
+ ExecEnv::GetInstance(),
_query_ctx.get());
+ terminated_state->set_runtime_filter_mgr(terminated_mgr.get());
+ terminated_state->set_desc_tbl(&_tbl);
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_producer_runtime_filter(desc,
&producer));
+ std::shared_ptr<RuntimeFilterProducer> producer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_producer_runtime_filter(desc,
&producer2));
+ std::shared_ptr<RuntimeFilterProducer> terminated_producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ terminated_state->register_producer_runtime_filter(desc,
&terminated_producer));
+
+ std::shared_ptr<RuntimeFilterConsumer> local_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_consumer_runtime_filter(desc, false,
0, &local_consumer));
+ std::shared_ptr<RuntimeFilterConsumer> local_consumer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_consumer_runtime_filter(desc, false,
0, &local_consumer2));
+ std::shared_ptr<RuntimeFilterConsumer> merge_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_consumer_runtime_filter(desc, true,
1, &merge_consumer));
+
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->init(1));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer->insert(ColumnHelper::create_column<DataTypeInt32>({5}),
0));
+ auto shared_wrapper = producer->wrapper();
+ producer2->set_wrapper(shared_wrapper);
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+
producer2->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+ terminated_producer->set_wrapper_state_and_ready_to_publish(
+ RuntimeFilterWrapper::State::DISABLED, "skip all rf process");
+
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->publish(_runtime_states[0].get(),
true));
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(terminated_producer->publish(terminated_state.get(),
true));
+ int32_t five = 5;
+ ASSERT_EQ(shared_wrapper->get_state(), RuntimeFilterWrapper::State::READY);
+ ASSERT_EQ(shared_wrapper->hybrid_set()->size(), 1);
+ ASSERT_TRUE(shared_wrapper->hybrid_set()->find(&five));
+
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer2->publish(_runtime_states[1].get(),
true));
+ ASSERT_EQ(shared_wrapper->get_state(), RuntimeFilterWrapper::State::READY);
+ ASSERT_EQ(shared_wrapper->hybrid_set()->size(), 1);
+ for (const auto& consumer : {local_consumer, local_consumer2}) {
+ ASSERT_EQ(consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ ASSERT_EQ(consumer->_wrapper, shared_wrapper);
+ ASSERT_EQ(consumer->_wrapper->get_state(),
RuntimeFilterWrapper::State::READY);
+ ASSERT_EQ(consumer->_wrapper->hybrid_set()->size(), 1);
+ ASSERT_TRUE(consumer->_wrapper->hybrid_set()->find(&five));
+ }
+ ASSERT_EQ(merge_consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ ASSERT_NE(merge_consumer->_wrapper, shared_wrapper);
+ ASSERT_EQ(merge_consumer->_wrapper->get_state(),
RuntimeFilterWrapper::State::DISABLED);
+ ASSERT_NE(merge_consumer->_wrapper->_reason.status().msg().find("skip all
rf process"),
+ std::string::npos);
+}
+
+TEST_F(RuntimeFilterProducerTest, publish_mixed_targets_disabled) {
+ auto desc = TRuntimeFilterDescBuilder()
+ .set_build_bf_by_runtime_size(false)
+ .set_is_broadcast_join(false)
+ .add_planId_to_target_expr(0)
+ .add_planId_to_target_expr(1)
+ .build();
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_producer_runtime_filter(desc,
&producer));
+ std::shared_ptr<RuntimeFilterProducer> producer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_producer_runtime_filter(desc,
&producer2));
+
+ std::shared_ptr<RuntimeFilterConsumer> local_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_consumer_runtime_filter(desc, false,
0, &local_consumer));
+ std::shared_ptr<RuntimeFilterConsumer> merge_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_consumer_runtime_filter(desc, true,
1, &merge_consumer));
+
+ // Producers are disabled before the filters are built.
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::DISABLED);
+
producer2->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::DISABLED);
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->publish(_runtime_states[0].get(),
true));
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer2->publish(_runtime_states[1].get(),
true));
+
+ ASSERT_EQ(local_consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ ASSERT_EQ(local_consumer->_wrapper->get_state(),
RuntimeFilterWrapper::State::DISABLED);
+ ASSERT_EQ(merge_consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ ASSERT_EQ(merge_consumer->_wrapper->get_state(),
RuntimeFilterWrapper::State::DISABLED);
+ ASSERT_NE(local_consumer->_wrapper, merge_consumer->_wrapper);
+}
+
+// A runtime filter whose only targets are local-merge (global RF mgr)
consumers never hands its
+// wrapper to a plain local consumer of either instance, so once it reaches
the merger nothing
+// else reads it: the merger may adopt it directly instead of cloning it.
+TEST_F(RuntimeFilterProducerTest,
publish_local_merge_only_takes_ownership_without_local_consumer) {
+ auto desc = TRuntimeFilterDescBuilder()
+ .set_build_bf_by_runtime_size(false)
+ .set_is_broadcast_join(false)
+ .add_planId_to_target_expr(0)
+ .build();
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_producer_runtime_filter(desc,
&producer));
+ std::shared_ptr<RuntimeFilterProducer> producer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_producer_runtime_filter(desc,
&producer2));
+
+ std::shared_ptr<RuntimeFilterConsumer> consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_consumer_runtime_filter(desc, true,
0, &consumer));
+
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->init(1));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer->insert(ColumnHelper::create_column<DataTypeInt32>({5}),
0));
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer2->init(1));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer2->insert(ColumnHelper::create_column<DataTypeInt32>({6}),
0));
+
producer2->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+
+ auto wrapper = producer->wrapper();
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->publish(_runtime_states[0].get(),
true));
+ std::shared_ptr<LocalMergeContext> context;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+
_query_ctx->runtime_filter_mgr()->get_local_merge_context(desc.filter_id, 0,
&context));
+ // The merger adopted the first producer's wrapper directly instead of
cloning it: no plain
+ // local consumer of this instance exists to be affected.
+ ASSERT_EQ(context->merger->_wrapper, wrapper);
+ ASSERT_EQ(context->merger->_wrapper->hybrid_set()->size(), 1);
+
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer2->publish(_runtime_states[1].get(),
true));
+ ASSERT_EQ(consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ ASSERT_EQ(consumer->_wrapper, context->merger->_wrapper);
+ ASSERT_EQ(consumer->_wrapper->hybrid_set()->size(), 2);
+ int32_t five = 5;
+ int32_t six = 6;
+ ASSERT_TRUE(wrapper->hybrid_set()->find(&five));
+ ASSERT_TRUE(wrapper->hybrid_set()->find(&six));
+}
+
+// A non-broadcast, non-runtime-sized filter with exactly one expected
producer becomes ready on
+// that producer's only `merge_from` call and is never merged into again, so
the merger may adopt
+// the producer's wrapper directly even though a plain local consumer of the
same instance also
+// reads it.
+TEST_F(RuntimeFilterProducerTest,
publish_single_producer_with_local_consumer_takes_ownership) {
+ auto desc = TRuntimeFilterDescBuilder()
+ .set_build_bf_by_runtime_size(false)
+ .set_is_broadcast_join(false)
+ .add_planId_to_target_expr(0)
+ .add_planId_to_target_expr(1)
+ .build();
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_producer_runtime_filter(desc,
&producer));
+
+ std::shared_ptr<RuntimeFilterConsumer> local_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_consumer_runtime_filter(desc, false,
0, &local_consumer));
+ std::shared_ptr<RuntimeFilterConsumer> merge_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_consumer_runtime_filter(desc, true,
1, &merge_consumer));
+
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->init(1));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer->insert(ColumnHelper::create_column<DataTypeInt32>({5}),
0));
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+
+ auto wrapper = producer->wrapper();
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->publish(_runtime_states[0].get(),
true));
+
+ ASSERT_EQ(local_consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ ASSERT_EQ(local_consumer->_wrapper, wrapper);
+ ASSERT_EQ(merge_consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ // The merger adopted the single producer's wrapper directly: the same
object the plain local
+ // consumer also received, because the merger never merges another
producer into it.
+ ASSERT_EQ(merge_consumer->_wrapper, wrapper);
+ int32_t five = 5;
+ ASSERT_TRUE(wrapper->hybrid_set()->find(&five));
+}
+
+// A non-broadcast filter with only remote targets never hands its wrapper to
a plain local
+// consumer: that only happens in the `!_has_remote_target` branch of
`publish()`. Once such a
+// producer's wrapper reaches the local merger, no other reader remains, so
the merger may take
+// it over directly instead of cloning it.
+TEST_F(RuntimeFilterProducerTest, publish_remote_target_merge_takes_ownership)
{
+ auto desc =
TRuntimeFilterDescBuilder().set_mode(false).set_is_broadcast_join(false).build();
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_producer_runtime_filter(desc,
&producer));
+ std::shared_ptr<RuntimeFilterProducer> producer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_producer_runtime_filter(desc,
&producer2));
+
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->init(1));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer->insert(ColumnHelper::create_column<DataTypeInt32>({5}),
0));
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+
+ auto wrapper = producer->wrapper();
+ // Only the first of the two expected producers has published, so the
merger is not ready
+ // yet and publish() never reaches `_send_to_remote_targets()`.
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->publish(_runtime_states[0].get(),
true));
+ ASSERT_EQ(producer->_wrapper, nullptr);
+
+ std::shared_ptr<LocalMergeContext> context;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+
_query_ctx->runtime_filter_mgr()->get_local_merge_context(desc.filter_id, 0,
&context));
+ // The merger adopted the producer's wrapper directly instead of cloning
it.
+ ASSERT_EQ(context->merger->_wrapper, wrapper);
+ ASSERT_EQ(wrapper->hybrid_set()->size(), 1);
+ int32_t five = 5;
+ ASSERT_TRUE(wrapper->hybrid_set()->find(&five));
+}
+
+// An IN_OR_BLOOM merger which is still an IN filter takes the bloom filter of
a producer which
+// already changed to a bloom filter. It must take a copy, the producer's own
local consumers are
+// probing the original.
+TEST_F(RuntimeFilterProducerTest,
publish_mixed_targets_in_or_bloom_merge_copies_bloom_filter) {
+ auto desc = TRuntimeFilterDescBuilder()
+ .set_build_bf_by_runtime_size(false)
+ .set_is_broadcast_join(false)
+ .add_planId_to_target_expr(0)
+ .add_planId_to_target_expr(1)
+ .build();
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_producer_runtime_filter(desc,
&producer));
+ std::shared_ptr<RuntimeFilterProducer> producer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_producer_runtime_filter(desc,
&producer2));
+
+ // Only the second instance has a consumer in local RF mgr.
+ std::shared_ptr<RuntimeFilterConsumer> local_consumer2;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_consumer_runtime_filter(desc, false,
0, &local_consumer2));
+ std::shared_ptr<RuntimeFilterConsumer> merge_consumer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[1]->register_consumer_runtime_filter(desc, true,
1, &merge_consumer));
+
+ // The first producer stays an IN filter, the second one exceeds
runtime_filter_max_in_num
+ // and changes to a bloom filter.
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->init(1));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer->insert(ColumnHelper::create_column<DataTypeInt32>({5}),
0));
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+ ASSERT_EQ(producer->wrapper()->get_real_type(),
RuntimeFilterType::IN_FILTER);
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer2->init(2000));
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ producer2->insert(ColumnHelper::create_column<DataTypeInt32>({6,
7}), 0));
+
producer2->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+ auto bloom_wrapper = producer2->wrapper();
+ ASSERT_EQ(bloom_wrapper->get_real_type(), RuntimeFilterType::BLOOM_FILTER);
+ char* bloom_data = nullptr;
+ int bloom_len = 0;
+ bloom_wrapper->bloom_filter_func()->get_data(&bloom_data, &bloom_len);
+ ASSERT_GT(bloom_len, 0);
+ const std::string snapshot(bloom_data, bloom_len);
+
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->publish(_runtime_states[0].get(),
true));
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer2->publish(_runtime_states[1].get(),
true));
+
+ ASSERT_EQ(local_consumer2->_rf_state, RuntimeFilterConsumer::State::READY);
+ ASSERT_EQ(local_consumer2->_wrapper, bloom_wrapper);
+ ASSERT_EQ(merge_consumer->_rf_state, RuntimeFilterConsumer::State::READY);
+ ASSERT_EQ(merge_consumer->_wrapper->get_state(),
RuntimeFilterWrapper::State::READY);
+ ASSERT_EQ(merge_consumer->_wrapper->get_real_type(),
RuntimeFilterType::BLOOM_FILTER);
+ ASSERT_NE(merge_consumer->_wrapper->bloom_filter_func(),
bloom_wrapper->bloom_filter_func());
+
+ // The merged filter contains the values of both producers.
+ std::vector<uint8_t> found(3);
+ merge_consumer->_wrapper->bloom_filter_func()->find_fixed_len(
+ ColumnHelper::create_column<DataTypeInt32>({5, 6, 7}),
found.data());
+ ASSERT_TRUE(std::all_of(found.begin(), found.end(), [](uint8_t i) -> bool
{ return i; }));
+
+ // The bloom filter of the second producer is not written by the merge.
+ char* merged_data = nullptr;
+ int merged_len = 0;
+ merge_consumer->_wrapper->bloom_filter_func()->get_data(&merged_data,
&merged_len);
+ ASSERT_NE(merged_data, bloom_data);
+ ASSERT_EQ(bloom_len, merged_len);
+ ASSERT_EQ(snapshot, std::string(bloom_data, bloom_len));
+ ASSERT_NE(snapshot, std::string(merged_data, merged_len));
+}
+
TEST_F(RuntimeFilterProducerTest, publish_release_wrapper) {
auto desc = TRuntimeFilterDescBuilder()
.set_build_bf_by_runtime_size(false)
diff --git a/be/test/exec/runtime_filter/runtime_filter_wrapper_test.cpp
b/be/test/exec/runtime_filter/runtime_filter_wrapper_test.cpp
index 9774ce2f2f1..b4c1978826b 100644
--- a/be/test/exec/runtime_filter/runtime_filter_wrapper_test.cpp
+++ b/be/test/exec/runtime_filter/runtime_filter_wrapper_test.cpp
@@ -22,6 +22,7 @@
#include "core/data_type/data_type_date.h"
#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
#include "exprs/bloom_filter_func.h"
#include "exprs/hybrid_set.h"
#include "exprs/minmax_predicate.h"
@@ -1210,6 +1211,318 @@ TEST_F(RuntimeFilterWrapperTest, TestInOrBloom) {
EXPECT_EQ(wrapper->contain_null(), false);
}
+// Merging a bloom filter into an IN_OR_BLOOM filter which is still an IN
filter takes a copy of
+// the input bloom filter, the input may still be probed by its own consumers.
+TEST_F(RuntimeFilterWrapperTest, TestMergeInWithBloomCopiesBloomFilter) {
+ using DataType = DataTypeInt32;
+ auto make_params = [](int32_t max_in_num) {
+ return RuntimeFilterParams {.filter_id = 0,
+ .filter_type =
RuntimeFilterType::IN_OR_BLOOM_FILTER,
+ .column_return_type =
PrimitiveType::TYPE_INT,
+ .null_aware = false,
+ .max_in_num = max_in_num,
+ .runtime_bloom_filter_min_size = 64,
+ .runtime_bloom_filter_max_size = 128,
+ .bloom_filter_size = 64,
+ .build_bf_by_runtime_size = false,
+ .bloom_filter_size_calculated_by_ndv =
false};
+ };
+ auto in_params = make_params(18);
+ auto wrapper = std::make_shared<RuntimeFilterWrapper>(&in_params);
+ ASSERT_TRUE(wrapper->init(16).ok());
+ ASSERT_EQ(wrapper->get_real_type(), RuntimeFilterType::IN_FILTER);
+ std::vector<int> data_vector(10);
+ std::iota(data_vector.begin(), data_vector.end(), 0);
+
ASSERT_TRUE(wrapper->insert(ColumnHelper::create_column<DataType>(data_vector),
0).ok());
+
+ auto bloom_params = make_params(8);
+ auto bloom_wrapper = std::make_shared<RuntimeFilterWrapper>(&bloom_params);
+ ASSERT_TRUE(bloom_wrapper->init(16).ok());
+ ASSERT_EQ(bloom_wrapper->get_real_type(), RuntimeFilterType::BLOOM_FILTER);
+ std::vector<int> bloom_data_vector(10);
+ std::iota(bloom_data_vector.begin(), bloom_data_vector.end(), 10);
+
ASSERT_TRUE(bloom_wrapper->insert(ColumnHelper::create_column<DataType>(bloom_data_vector),
0)
+ .ok());
+ bloom_wrapper->set_state(RuntimeFilterWrapper::State::READY);
+ char* bloom_data = nullptr;
+ int bloom_len = 0;
+ bloom_wrapper->bloom_filter_func()->get_data(&bloom_data, &bloom_len);
+ ASSERT_GT(bloom_len, 0);
+ const std::string snapshot(bloom_data, bloom_len);
+
+ ASSERT_TRUE(wrapper->merge(bloom_wrapper.get()).ok());
+ EXPECT_EQ(wrapper->get_real_type(), RuntimeFilterType::BLOOM_FILTER);
+ EXPECT_NE(wrapper->bloom_filter_func(),
bloom_wrapper->bloom_filter_func());
+ char* merged_data = nullptr;
+ int merged_len = 0;
+ wrapper->bloom_filter_func()->get_data(&merged_data, &merged_len);
+ EXPECT_NE(merged_data, bloom_data);
+ EXPECT_EQ(merged_len, bloom_len);
+ // The input bloom filter is unchanged, the merged one has the IN values
added.
+ EXPECT_EQ(snapshot, std::string(bloom_data, bloom_len));
+ EXPECT_NE(snapshot, std::string(merged_data, merged_len));
+
+ std::vector<int> final_data_vector(20);
+ std::iota(final_data_vector.begin(), final_data_vector.end(), 0);
+ std::vector<uint8_t> final_res(20);
+ wrapper->bloom_filter_func()->find_fixed_len(
+ ColumnHelper::create_column<DataType>(final_data_vector),
final_res.data());
+ EXPECT_TRUE(
+ std::all_of(final_res.begin(), final_res.end(), [](uint8_t i) ->
bool { return i; }));
+}
+
+// When the caller tells merge() that `other` is exclusively owned (e.g. an
RPC-only filter with
+// no other reader, see `RuntimeFilterMergeControllerEntity::merge`), the
bloom filter is taken
+// over directly instead of cloned.
+TEST_F(RuntimeFilterWrapperTest,
TestMergeInWithBloomExclusivelyOwnedTakesOwnership) {
+ using DataType = DataTypeInt32;
+ auto make_params = [](int32_t max_in_num) {
+ return RuntimeFilterParams {.filter_id = 0,
+ .filter_type =
RuntimeFilterType::IN_OR_BLOOM_FILTER,
+ .column_return_type =
PrimitiveType::TYPE_INT,
+ .null_aware = false,
+ .max_in_num = max_in_num,
+ .runtime_bloom_filter_min_size = 64,
+ .runtime_bloom_filter_max_size = 128,
+ .bloom_filter_size = 64,
+ .build_bf_by_runtime_size = false,
+ .bloom_filter_size_calculated_by_ndv =
false};
+ };
+ auto in_params = make_params(18);
+ auto wrapper = std::make_shared<RuntimeFilterWrapper>(&in_params);
+ ASSERT_TRUE(wrapper->init(16).ok());
+ ASSERT_EQ(wrapper->get_real_type(), RuntimeFilterType::IN_FILTER);
+ std::vector<int> data_vector(10);
+ std::iota(data_vector.begin(), data_vector.end(), 0);
+
ASSERT_TRUE(wrapper->insert(ColumnHelper::create_column<DataType>(data_vector),
0).ok());
+
+ auto bloom_params = make_params(8);
+ auto bloom_wrapper = std::make_shared<RuntimeFilterWrapper>(&bloom_params);
+ ASSERT_TRUE(bloom_wrapper->init(16).ok());
+ ASSERT_EQ(bloom_wrapper->get_real_type(), RuntimeFilterType::BLOOM_FILTER);
+ std::vector<int> bloom_data_vector(10);
+ std::iota(bloom_data_vector.begin(), bloom_data_vector.end(), 10);
+
ASSERT_TRUE(bloom_wrapper->insert(ColumnHelper::create_column<DataType>(bloom_data_vector),
0)
+ .ok());
+ bloom_wrapper->set_state(RuntimeFilterWrapper::State::READY);
+ auto bloom_filter_func_before_merge = bloom_wrapper->bloom_filter_func();
+
+ ASSERT_TRUE(wrapper->merge(bloom_wrapper.get(),
/*other_exclusively_owned=*/true).ok());
+ EXPECT_EQ(wrapper->get_real_type(), RuntimeFilterType::BLOOM_FILTER);
+ // No clone: the merger took over `other`'s bloom filter directly.
+ EXPECT_EQ(wrapper->bloom_filter_func(), bloom_filter_func_before_merge);
+ EXPECT_EQ(wrapper->bloom_filter_func(),
bloom_wrapper->bloom_filter_func());
+
+ std::vector<int> final_data_vector(20);
+ std::iota(final_data_vector.begin(), final_data_vector.end(), 0);
+ std::vector<uint8_t> final_res(20);
+ wrapper->bloom_filter_func()->find_fixed_len(
+ ColumnHelper::create_column<DataType>(final_data_vector),
final_res.data());
+ EXPECT_TRUE(
+ std::all_of(final_res.begin(), final_res.end(), [](uint8_t i) ->
bool { return i; }));
+}
+
+TEST_F(RuntimeFilterWrapperTest, TestClone) {
+ using DataType = DataTypeInt32;
+ auto make_params = [](RuntimeFilterType filter_type, bool null_aware,
int32_t max_in_num,
+ PrimitiveType column_type = PrimitiveType::TYPE_INT)
{
+ return RuntimeFilterParams {.filter_id = 3,
+ .filter_type = filter_type,
+ .column_return_type = column_type,
+ .null_aware = null_aware,
+ .max_in_num = max_in_num,
+ .runtime_bloom_filter_min_size = 64,
+ .runtime_bloom_filter_max_size = 128,
+ .bloom_filter_size = 64,
+ .build_bf_by_runtime_size = false,
+ .bloom_filter_size_calculated_by_ndv =
true};
+ };
+ {
+ // In filter, the copy starts with an empty bucket prune cache.
+ auto params = make_params(RuntimeFilterType::IN_FILTER, true, 8);
+ auto wrapper = std::make_shared<RuntimeFilterWrapper>(¶ms);
+ ASSERT_TRUE(wrapper->init(3).ok());
+ ASSERT_TRUE(wrapper->insert(ColumnHelper::create_column<DataType>({1,
2}), 0).ok());
+ wrapper->hybrid_set()->insert((const void*)nullptr);
+ wrapper->set_state(RuntimeFilterWrapper::State::READY);
+ auto hashes =
wrapper->get_or_compute_bucket_prune_hashes(std::make_shared<DataType>());
+ ASSERT_EQ(hashes->size(), 3);
+
+ std::shared_ptr<RuntimeFilterWrapper> cloned;
+ ASSERT_TRUE(wrapper->clone(&cloned).ok());
+ EXPECT_NE(cloned, wrapper);
+ EXPECT_EQ(cloned->get_state(), RuntimeFilterWrapper::State::READY);
+ EXPECT_EQ(cloned->filter_id(), wrapper->filter_id());
+ EXPECT_EQ(cloned->get_real_type(), RuntimeFilterType::IN_FILTER);
+ EXPECT_EQ(cloned->_max_in_num, wrapper->_max_in_num);
+ EXPECT_NE(cloned->hybrid_set(), wrapper->hybrid_set());
+ EXPECT_EQ(cloned->hybrid_set()->size(), 2);
+ int32_t value = 1;
+ EXPECT_TRUE(cloned->hybrid_set()->find(&value));
+ value = 2;
+ EXPECT_TRUE(cloned->hybrid_set()->find(&value));
+ EXPECT_TRUE(cloned->contain_null());
+ EXPECT_FALSE(cloned->_bucket_prune_hashes_started.load());
+
+ value = 3;
+ cloned->hybrid_set()->insert(&value);
+ EXPECT_EQ(wrapper->hybrid_set()->size(), 2);
+ EXPECT_FALSE(wrapper->hybrid_set()->find(&value));
+ value = 4;
+ wrapper->hybrid_set()->insert(&value);
+ EXPECT_EQ(cloned->hybrid_set()->size(), 3);
+ EXPECT_FALSE(cloned->hybrid_set()->find(&value));
+ }
+ {
+ // MinMax filter
+ auto params = make_params(RuntimeFilterType::MINMAX_FILTER, false, 0);
+ auto wrapper = std::make_shared<RuntimeFilterWrapper>(¶ms);
+ ASSERT_TRUE(wrapper->insert(ColumnHelper::create_column<DataType>({3,
7}), 0).ok());
+ wrapper->set_state(RuntimeFilterWrapper::State::READY);
+
+ std::shared_ptr<RuntimeFilterWrapper> cloned;
+ ASSERT_TRUE(wrapper->clone(&cloned).ok());
+ EXPECT_EQ(cloned->get_state(), RuntimeFilterWrapper::State::READY);
+ EXPECT_NE(cloned->minmax_func(), wrapper->minmax_func());
+ EXPECT_EQ(*(int32_t*)cloned->minmax_func()->get_min(), 3);
+ EXPECT_EQ(*(int32_t*)cloned->minmax_func()->get_max(), 7);
+
+ ASSERT_TRUE(cloned->insert(ColumnHelper::create_column<DataType>({1,
9}), 0).ok());
+ EXPECT_EQ(*(int32_t*)wrapper->minmax_func()->get_min(), 3);
+ EXPECT_EQ(*(int32_t*)wrapper->minmax_func()->get_max(), 7);
+ }
+ {
+ // Bloom filter
+ auto params = make_params(RuntimeFilterType::BLOOM_FILTER, false, 0);
+ auto wrapper = std::make_shared<RuntimeFilterWrapper>(¶ms);
+ ASSERT_TRUE(wrapper->init(80).ok());
+ std::vector<int32_t> data(10);
+ std::iota(data.begin(), data.end(), 0);
+
ASSERT_TRUE(wrapper->insert(ColumnHelper::create_column<DataType>(data),
0).ok());
+ wrapper->set_state(RuntimeFilterWrapper::State::READY);
+
+ std::shared_ptr<RuntimeFilterWrapper> cloned;
+ ASSERT_TRUE(wrapper->clone(&cloned).ok());
+ EXPECT_EQ(cloned->get_state(), RuntimeFilterWrapper::State::READY);
+ EXPECT_NE(cloned->bloom_filter_func(), wrapper->bloom_filter_func());
+ ASSERT_EQ(cloned->bloom_filter_func()->get_size(),
+ wrapper->bloom_filter_func()->get_size());
+ char* origin_data = nullptr;
+ int origin_len = 0;
+ wrapper->bloom_filter_func()->get_data(&origin_data, &origin_len);
+ char* cloned_data = nullptr;
+ int cloned_len = 0;
+ cloned->bloom_filter_func()->get_data(&cloned_data, &cloned_len);
+ ASSERT_EQ(cloned_len, origin_len);
+ EXPECT_NE(cloned_data, origin_data);
+ EXPECT_EQ(memcmp(cloned_data, origin_data, origin_len), 0);
+
+ std::string snapshot(origin_data, origin_len);
+ ASSERT_TRUE(cloned->insert(ColumnHelper::create_column<DataType>({100,
200, 300}), 0).ok());
+ EXPECT_EQ(memcmp(snapshot.data(), origin_data, origin_len), 0);
+ }
+ {
+ // In or bloom filter keeps its real type.
+ auto params = make_params(RuntimeFilterType::IN_OR_BLOOM_FILTER,
false, 2);
+ auto in_wrapper = std::make_shared<RuntimeFilterWrapper>(¶ms);
+ ASSERT_TRUE(in_wrapper->init(1).ok());
+
ASSERT_TRUE(in_wrapper->insert(ColumnHelper::create_column<DataType>({1}),
0).ok());
+ in_wrapper->set_state(RuntimeFilterWrapper::State::READY);
+ // `init()` leaves the bloom filter of an IN filter unallocated, so
allocate it by hand:
+ // the copy must skip the unused bloom filter, not merely find none to
copy.
+
ASSERT_TRUE(in_wrapper->bloom_filter_func()->init_with_fixed_length(1).ok());
+ ASSERT_GT(in_wrapper->bloom_filter_func()->get_size(), 0);
+ std::shared_ptr<RuntimeFilterWrapper> cloned;
+ ASSERT_TRUE(in_wrapper->clone(&cloned).ok());
+ EXPECT_EQ(cloned->get_real_type(), RuntimeFilterType::IN_FILTER);
+ EXPECT_EQ(cloned->hybrid_set()->size(), 1);
+ EXPECT_NE(cloned->bloom_filter_func(),
in_wrapper->bloom_filter_func());
+ // The unused bloom filter of an IN filter is not copied.
+ EXPECT_EQ(cloned->bloom_filter_func()->get_size(), 0);
+
+ auto bloom_wrapper = std::make_shared<RuntimeFilterWrapper>(¶ms);
+ ASSERT_TRUE(bloom_wrapper->init(3).ok());
+ ASSERT_TRUE(
+
bloom_wrapper->insert(ColumnHelper::create_column<DataType>({1, 2, 3}),
0).ok());
+ bloom_wrapper->set_state(RuntimeFilterWrapper::State::READY);
+ ASSERT_TRUE(bloom_wrapper->clone(&cloned).ok());
+ EXPECT_EQ(cloned->get_real_type(), RuntimeFilterType::BLOOM_FILTER);
+ EXPECT_EQ(cloned->hybrid_set(), nullptr);
+ EXPECT_EQ(cloned->bloom_filter_func()->get_size(),
+ bloom_wrapper->bloom_filter_func()->get_size());
+ }
+ {
+ // A disabled filter which is never initialized has no bloom filter
allocated.
+ auto params = make_params(RuntimeFilterType::IN_OR_BLOOM_FILTER,
false, 2);
+ auto wrapper = std::make_shared<RuntimeFilterWrapper>(¶ms);
+ wrapper->set_state(RuntimeFilterWrapper::State::DISABLED, "some
reason");
+ std::shared_ptr<RuntimeFilterWrapper> cloned;
+ ASSERT_TRUE(wrapper->clone(&cloned).ok());
+ EXPECT_EQ(cloned->get_state(), RuntimeFilterWrapper::State::DISABLED);
+ EXPECT_NE(cloned->_reason.status().msg().find("some reason"),
std::string::npos);
+ EXPECT_EQ(cloned->bloom_filter_func()->get_size(), 0);
+ }
+ {
+ // String in filter, the copy owns its strings.
+ auto params =
+ make_params(RuntimeFilterType::IN_FILTER, false, 8,
PrimitiveType::TYPE_STRING);
+ auto wrapper = std::make_shared<RuntimeFilterWrapper>(¶ms);
+ ASSERT_TRUE(wrapper->init(2).ok());
+ const std::string long_value(64, 'x');
+ ASSERT_TRUE(
+
wrapper->insert(ColumnHelper::create_column<DataTypeString>({"a", long_value}),
0)
+ .ok());
+ wrapper->set_state(RuntimeFilterWrapper::State::READY);
+ std::shared_ptr<RuntimeFilterWrapper> cloned;
+ ASSERT_TRUE(wrapper->clone(&cloned).ok());
+ wrapper->hybrid_set()->clear();
+ wrapper.reset();
+ EXPECT_EQ(cloned->hybrid_set()->size(), 2);
+ StringRef value("a");
+ EXPECT_TRUE(cloned->hybrid_set()->find(&value));
+ value = StringRef(long_value);
+ EXPECT_TRUE(cloned->hybrid_set()->find(&value));
+ }
+ {
+ // Null aware bloom filter and minmax filter keep `contain_null`.
+ auto column = ColumnHelper::create_nullable_column<DataType>({1, 2,
0}, {0, 0, 1});
+ for (auto filter_type :
+ {RuntimeFilterType::BLOOM_FILTER,
RuntimeFilterType::MINMAX_FILTER}) {
+ auto params = make_params(filter_type, true, 0);
+ auto wrapper = std::make_shared<RuntimeFilterWrapper>(¶ms);
+ ASSERT_TRUE(wrapper->init(3).ok());
+ ASSERT_TRUE(wrapper->insert(column, 0).ok());
+ wrapper->set_state(RuntimeFilterWrapper::State::READY);
+ ASSERT_TRUE(wrapper->contain_null());
+ std::shared_ptr<RuntimeFilterWrapper> cloned;
+ ASSERT_TRUE(wrapper->clone(&cloned).ok());
+ EXPECT_EQ(cloned->get_real_type(), filter_type);
+ EXPECT_TRUE(cloned->contain_null());
+ }
+ }
+ {
+ // Min filter and max filter
+ for (auto filter_type : {RuntimeFilterType::MIN_FILTER,
RuntimeFilterType::MAX_FILTER}) {
+ auto params = make_params(filter_type, false, 0);
+ auto wrapper = std::make_shared<RuntimeFilterWrapper>(¶ms);
+
ASSERT_TRUE(wrapper->insert(ColumnHelper::create_column<DataType>({3, 7}),
0).ok());
+ wrapper->set_state(RuntimeFilterWrapper::State::READY);
+ std::shared_ptr<RuntimeFilterWrapper> cloned;
+ ASSERT_TRUE(wrapper->clone(&cloned).ok());
+ EXPECT_EQ(cloned->get_real_type(), filter_type);
+ EXPECT_NE(cloned->minmax_func(), wrapper->minmax_func());
+
ASSERT_TRUE(cloned->insert(ColumnHelper::create_column<DataType>({1, 9}),
0).ok());
+ if (filter_type == RuntimeFilterType::MIN_FILTER) {
+ EXPECT_EQ(*(int32_t*)cloned->minmax_func()->get_min(), 1);
+ EXPECT_EQ(*(int32_t*)wrapper->minmax_func()->get_min(), 3);
+ } else {
+ EXPECT_EQ(*(int32_t*)cloned->minmax_func()->get_max(), 9);
+ EXPECT_EQ(*(int32_t*)wrapper->minmax_func()->get_max(), 7);
+ }
+ }
+ }
+}
+
TEST_F(RuntimeFilterWrapperTest, TestErrorPath) {
using DataType = DataTypeInt32;
int32_t filter_id = 0;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
