github-actions[bot] commented on code in PR #68684:
URL: https://github.com/apache/doris/pull/68684#discussion_r4203575169
##########
be/src/exec/runtime_filter/runtime_filter_wrapper.cpp:
##########
@@ -190,8 +190,9 @@ Status RuntimeFilterWrapper::merge(const
RuntimeFilterWrapper* other) {
RETURN_IF_ERROR(_change_to_bloom_filter());
}
} else {
- // case1&case2: use input bf directly and insert hybrid set
data into bf
- _bloom_filter_func = other->_bloom_filter_func;
+ // 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));
Review Comment:
[P2] Avoid copying a later RPC-only Bloom filter during IN-to-Bloom merge.
With a remote `IN_OR_BLOOM` filter, `enable_sync_runtime_filter_size=false`
lets local build sizes straddle `runtime_filter_max_in_num`, so an IN first
arrival can meet a Bloom arrival here. The global receiver's `tmp_filter` owns
that Bloom directory and has no readers, yet this clone allocates and copies
the full directory under `GlobalMergeContext::mtx` before adding the IN values.
Adopting the first RPC wrapper does not remove this later copy; provide an
ownership-aware path for this RPC input while retaining the copy for local
producer wrappers with consumers.
##########
be/src/exec/runtime_filter/runtime_filter_merger.h:
##########
@@ -70,11 +70,13 @@ class RuntimeFilterMerger : public RuntimeFilter {
_rf_state = State::READY;
}
if (_wrapper->get_state() == RuntimeFilterWrapper::State::UNINITED) {
- _wrapper = other->_wrapper;
- return Status::OK();
+ // The merger owns a private copy of the first wrapper. A
producer's wrapper may
+ // still be used by the consumers in local RF mgr of the same
instance (and is shared
+ // by all producers of a broadcast join with a shared hash table),
so the merger must
+ // never write a producer's wrapper: it only merges the later ones
into its own copy.
+ return other->_wrapper->clone(&_wrapper);
Review Comment:
[P2] Avoid copying the first RPC-only filter at the global merge node.
`RuntimeFilterMergeControllerEntity::merge()` builds a fresh `tmp_filter` from
this RPC and destroys it after `merge_from()`. This unconditional clone
allocates, zeroes, and OR-copies its Bloom directory while holding
`GlobalMergeContext::mtx`; the configured default maximum is 64 MiB per filter,
so concurrent merges add avoidable memory pressure and RPC latency. Keep the
producer-side copy for local readers, but let this uniquely owned RPC input
transfer its wrapper to the merger.
--
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]
