github-actions[bot] commented on code in PR #65837:
URL: https://github.com/apache/doris/pull/65837#discussion_r3775463395
##########
be/src/exec/runtime_filter/runtime_filter_wrapper.cpp:
##########
@@ -615,6 +617,46 @@ bool RuntimeFilterWrapper::contain_null() const {
return false;
}
+std::shared_ptr<const std::vector<uint32_t>>
+RuntimeFilterWrapper::get_or_compute_bucket_prune_hashes(const DataTypePtr&
target_type) const {
+ DORIS_CHECK(_state.load() == State::READY);
+ DORIS_CHECK(_hybrid_set != nullptr);
+ DORIS_CHECK(target_type != nullptr);
+ PrimitiveType primitive_type = target_type->get_primitive_type();
+ DORIS_CHECK_EQ(primitive_type, _column_return_type);
+
+ std::call_once(_bucket_prune_hashes_once, [&] {
+ MutableColumnPtr column = target_type->create_column();
+ auto* iter = _hybrid_set->begin();
+ while (iter->has_next()) {
Review Comment:
[P1] Do not cache hashes on a wrapper that an unfinished merger can still
mutate. `RuntimeFilterMerger::merge_from()` aliases the first producer's
wrapper, while `RuntimeFilterProducer::publish()` deliberately calls
`do_merge()` and then signals ordinary local consumers when one RF has both
local-merge and non-merge targets. With producers P0/P1, a local consumer can
run this `call_once` on W0=`{k0}` after P0 is adopted but before P1 merges
`{k1}` into the same `HybridSet`. The merger later publishes logical `{k0,k1}`,
yet every merged consumer reuses the frozen `{hash(k0)}` vector and may prune
the bucket containing `k1` (and concurrent iteration/insertion also races).
Please keep merger storage independent from producer wrappers, or cache only an
immutable finalized snapshot/generation, and add a two-producer mixed-routing
test that materializes the local cache before the second merge.
##########
be/src/exec/runtime_filter/runtime_filter_wrapper.cpp:
##########
@@ -615,6 +617,46 @@ bool RuntimeFilterWrapper::contain_null() const {
return false;
}
+std::shared_ptr<const std::vector<uint32_t>>
+RuntimeFilterWrapper::get_or_compute_bucket_prune_hashes(const DataTypePtr&
target_type) const {
+ DORIS_CHECK(_state.load() == State::READY);
+ DORIS_CHECK(_hybrid_set != nullptr);
+ DORIS_CHECK(target_type != nullptr);
+ PrimitiveType primitive_type = target_type->get_primitive_type();
+ DORIS_CHECK_EQ(primitive_type, _column_return_type);
+
+ std::call_once(_bucket_prune_hashes_once, [&] {
+ MutableColumnPtr column = target_type->create_column();
+ auto* iter = _hybrid_set->begin();
+ while (iter->has_next()) {
+ const void* value = iter->get_value();
Review Comment:
[P2] Avoid duplicating the whole exact string set to build this cache.
Eligibility is bounded only by `runtime_filter_max_in_num` (40,960 by default),
not by payload bytes, and `StringSet` already owns every VARCHAR/STRING value.
This loop copies all bytes again into a temporary `ColumnString` before
hashing; 10,000 64-KiB values add roughly 625 MiB during scan open (or while
`_conjuncts_lock` is held for a late filter), so a default-on optimization can
fail an otherwise viable query. Please hash each iterator value directly with
the write-routing CRC routine, or reserve/account a byte-bounded
materialization and conservatively skip it on failure. The retained hash vector
should also use query-tracked allocator-aware storage.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java:
##########
@@ -1420,6 +1429,40 @@ void
setPartitionBoundariesForRuntimeFilter(TOlapScanNode olapScanNode) {
}
}
+ private boolean hasRfDrivingBucketPruning() {
+ PlanNodeId myId = this.getId();
+ for (RuntimeFilter rf : runtimeFilters) {
+ if (rf.canPruneBucketsFor(myId)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private void setRuntimeFilterBucketPruneParameters() {
+ for (TScanRangeLocations locations : scanRangeLocations) {
+ TPaloScanRange scanRange =
locations.getScanRange().getPaloScanRange();
+ Long bucketInfo = tabletId2BucketInfo.get(scanRange.getTabletId());
+ Preconditions.checkState(bucketInfo != null &&
decodeBucketNum(bucketInfo) > 0,
Review Comment:
[P2] Attach this metadata once instead of re-walking all ranges per backend
worker. Nereids memoizes `fragment.toThrift()` per worker in
`ThriftPlansBuilder.fragmentToThriftIfAbsent()`, so every distinct worker
re-enters this method and scans the global `scanRangeLocations` list. Yet
`ScanWorkerSelector.buildScanReplicaParams()` aliases the same `TScanRange`
objects into worker assignments, meaning the first pass already sets every
range completely; later passes repeat identical map lookups, checks, decodes,
and setters. This makes the new planning work O(workers x tablets)—for example,
100 workers and 100,000 ranges cause ten million iterations. Please
snapshot/attach the paired fields once before per-worker serialization (or
guard it with plan-scoped idempotence), and add multi-worker coverage that
proves one global walk supplies every assignment.
##########
be/test/exec/scan/scanner_late_arrival_rf_test.cpp:
##########
@@ -135,6 +194,145 @@ TEST_F(ScannerLateArrivalRfTest,
applied_rf_num_advances_after_late_arrival) {
ASSERT_TRUE(scanner->_conjuncts.empty());
}
+TEST_F(ScannerLateArrivalRfTest, bucket_pruning_after_probe_tasks_start) {
+ constexpr int scan_node_id = 0;
+ constexpr int bucket_num = 4;
+ constexpr int filter_value = 7;
+
+ auto desc =
TRuntimeFilterDescBuilder().add_planId_to_target_expr(scan_node_id).build();
+ desc.__set_bucket_pruning_target_ids({scan_node_id});
+
+ ObjectPool pool;
+ DescriptorTblBuilder desc_builder(&pool);
+ desc_builder.declare_tuple() << TupleDescBuilder::SlotType
{std::make_shared<DataTypeInt32>(),
+ "dist_col"};
+ DescriptorTbl* desc_tbl = desc_builder.build();
+ ASSERT_NE(desc_tbl, nullptr);
+
+ TOlapScanNode olap_scan_node;
+ olap_scan_node.__set_tuple_id(0);
+ olap_scan_node.__set_keyType(TKeysType::DUP_KEYS);
+ olap_scan_node.__set_key_column_name({"dist_col"});
+ olap_scan_node.__set_key_column_type({TPrimitiveType::INT});
+
+ TPlanNode plan_node;
+ plan_node.__set_node_id(scan_node_id);
+ plan_node.__set_node_type(TPlanNodeType::OLAP_SCAN_NODE);
+ plan_node.__set_num_children(0);
+ plan_node.__set_limit(-1);
+ plan_node.__set_row_tuples({0});
+ plan_node.__set_runtime_filters({desc});
+ plan_node.__set_olap_scan_node(olap_scan_node);
+
+ auto op = std::make_shared<OlapScanOperatorX>(&pool, plan_node, 0,
*desc_tbl, bucket_num,
+ TQueryCacheParam {});
+ auto* state = _runtime_states[0].get();
+ state->set_desc_tbl(desc_tbl);
+ TQueryOptions query_options =
+ TQueryOptionsBuilder().set_runtime_filter_max_in_num(1024).build();
+ query_options.__set_enable_runtime_filter_bucket_prune(true);
+ state->set_query_options(query_options);
+
+ auto local_state = OlapScanLocalState::create_shared(state, op.get());
+ std::vector<std::shared_ptr<Dependency>> rf_dependencies;
+ ASSERT_TRUE(local_state->_helper.init(state, true, 0, 0, rf_dependencies,
"").ok());
+ ASSERT_TRUE(
+ local_state->_helper
+ .acquire_runtime_filter(state, local_state->_conjuncts,
op->row_descriptor())
+ .ok());
+ ASSERT_TRUE(local_state->_conjuncts.empty());
+ auto task_exec_ctx = std::make_shared<TaskExecutionContext>();
+ state->set_task_execution_context(task_exec_ctx);
+ for (int bucket_seq = 0; bucket_seq < bucket_num; ++bucket_seq) {
+ local_state->_rf_bucket_prune_ranges.push_back({100 + bucket_seq,
bucket_seq, bucket_num});
+ }
+ RuntimeProfile scan_profile("late bucket scan");
+ local_state->_buckets_pruned_by_rf_counter =
+ ADD_COUNTER(&scan_profile, "BucketsPrunedByRuntimeFilter",
TUnit::UNIT);
+ local_state->_scan_timer = ADD_TIMER(&scan_profile, "ScannerGetBlockTime");
+ local_state->_scan_cpu_timer = ADD_TIMER(&scan_profile, "ScannerCpuTime");
+ local_state->_filter_timer = ADD_TIMER(&scan_profile, "ScannerFilterTime");
+ local_state->_rows_read_counter = ADD_COUNTER(&scan_profile, "RowsRead",
TUnit::UNIT);
+
+ uint32_t hash = RawValue::zlib_crc32(&filter_value, sizeof(filter_value),
TYPE_INT, 0);
+ int selected_bucket = static_cast<int>(hash % bucket_num);
+ std::latch prepare_started(bucket_num);
+ std::latch filter_published(1);
+ std::list<std::shared_ptr<ScannerDelegate>> scanner_delegates;
+ std::vector<std::shared_ptr<LateBucketScanner>> scanners;
+ for (int bucket_seq = 0; bucket_seq < bucket_num; ++bucket_seq) {
+ auto scanner = std::make_shared<LateBucketScanner>(
+ state, local_state.get(), 100 + bucket_seq, bucket_seq ==
selected_bucket,
+ &scan_profile, &prepare_started, &filter_published);
+ ASSERT_TRUE(scanner->init(state, {}).ok());
+ scanners.push_back(scanner);
+ ScannerSPtr scanner_base = scanner;
+
scanner_delegates.push_back(std::make_shared<ScannerDelegate>(scanner_base));
+ }
+
+ auto dependency = Dependency::create_shared(0, 0, "late bucket scan
dependency");
+ std::atomic<int64_t> shared_limit {-1};
+ auto scanner_context = ScannerContext::create_shared(
+ state, local_state.get(), desc_tbl->get_tuple_descriptor(0),
nullptr, scanner_delegates,
+ -1, dependency, &shared_limit, nullptr, nullptr, 0, false,
bucket_num);
+ scanner_context->_newly_create_free_blocks_num =
+ ADD_COUNTER(&scan_profile, "NewlyCreatedFreeBlocks", TUnit::UNIT);
+ scanner_context->_scanner_memory_used_counter =
+ ADD_COUNTER(&scan_profile, "ScannerMemoryUsed", TUnit::BYTES);
+ scanner_context->_max_bytes_in_queue = 10 * 1024 * 1024;
+ std::vector<std::shared_ptr<ScanTask>> tasks;
+ for (const auto& scanner_delegate : scanner_delegates) {
+ auto task = std::make_shared<ScanTask>(scanner_delegate);
+ task->set_state(ScanTask::State::IN_FLIGHT);
+ tasks.push_back(std::move(task));
+ }
+ scanner_context->_in_flight_tasks_num = bucket_num;
+
+ std::vector<std::thread> probe_threads;
+ for (const auto& task : tasks) {
+ probe_threads.emplace_back([scanner_context, task] {
+ ScannerScheduler::_scanner_scan(scanner_context, task);
+ });
+ }
+
+ // Every task has passed the scheduler's pre-prepare pruning check while
the RF is not ready.
+ prepare_started.wait();
+ ASSERT_EQ(local_state->_rf_bucket_pruner.pruned_tablet_count(), 0);
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ ASSERT_TRUE(RuntimeFilterProducer::create(_query_ctx.get(), &desc,
&producer).ok());
Review Comment:
[P2] Make worker cleanup unconditional before using fatal assertions here.
After these joinable threads start, each can block on `filter_published`, but
that latch is released only at line 310 and the joins happen later. If this
assertion—or producer setup at lines 303-307—fails, GTest returns from the body
and destruction of `probe_threads` calls `std::terminate` (with workers still
blocked), masking the regression and aborting the whole BE test binary. Set up
fallible producer state before launching, and use an RAII guard that always
releases the latch and joins every thread; keep behavior assertions after
cleanup.
##########
be/src/exec/operator/olap_scan_operator.h:
##########
@@ -134,7 +136,11 @@ class OlapScanLocalState final : public
ScanLocalState<OlapScanLocalState> {
Status _build_key_ranges_and_filters();
+ bool _is_tablet_pruned_by_runtime_filter(int64_t partition_id, int64_t
tablet_id) const;
+
std::vector<std::unique_ptr<TPaloScanRange>> _scan_ranges;
+ std::vector<RuntimeFilterBucketPruneRange> _rf_bucket_prune_ranges;
+ RuntimeFilterBucketPruner _rf_bucket_pruner;
std::vector<SyncRowsetStats> _sync_statistics;
MonotonicStopWatch _sync_cloud_tablets_watcher;
Review Comment:
[P2] Avoid retaining a second untracked tuple for every scan range.
`set_scan_ranges()` already owns a full `TPaloScanRange` in `_scan_ranges`,
including these same three fields, but each executing non-cache-hit eligible
local state also appends this 16-byte struct to an ordinary `std::vector`. A
one-BE/one-instance scan at current limits (20,000 partitions x 768 buckets)
can therefore add about 234 MiB before capacity overhead; ordinary parallel
plans divide the count but retain the same linear aggregate duplication.
`_init_scanners()` then compacts `_scan_ranges` after ready RF pruning but
never releases the corresponding entries here. Please read the bucket fields
from the owned scan ranges or compact the metadata with them. If separate
storage is necessary, reserve/account it with an allocator-aware container and
release pruned entries; add high-range-count coverage for retained memory.
--
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]
