HappenLee commented on code in PR #65837: URL: https://github.com/apache/doris/pull/65837#discussion_r3775910897
########## regression-test/suites/query_p0/runtime_filter/rf_bucket_pruning.groovy: ########## @@ -0,0 +1,136 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import org.apache.doris.regression.action.ProfileAction + +suite("rf_bucket_pruning", "nonConcurrent") { + sql "set enable_runtime_filter_prune=false" + sql "set enable_runtime_filter_partition_prune=false" + sql "set enable_runtime_filter_bucket_prune=true" + sql "set runtime_filter_wait_infinitely=true" + sql "set runtime_filter_type='IN'" + sql "set disable_join_reorder=true" + sql "set enable_profile=true" + sql "set profile_level=2" + sql "set parallel_pipeline_task_num=1" Review Comment: Implemented in d5cef5da1e7 and revalidated on 452b5173426. ScannerLateArrivalRfTest.bucket_pruning_after_probe_tasks_start starts four probe/scanner tasks before the exact RF becomes READY, synchronizes publication after every task enters prepare, and verifies one correct result row, zero reads for pruned scanners, and a positive pruning counter. The current head also guarantees latch release and thread joins with RAII. The focused BE run passes 24/24 tests. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTranslator.java: ########## @@ -249,6 +249,11 @@ private void createLegacyRuntimeFilterFromGroup(List<RuntimeFilter> group, RuntimeFilterPartitionPruneClassifier.classify( head.getType(), targetExpr, nereidsTargetExprList.get(i), scanNode); setPartitionPruningMetadata(origFilter, scanNode, classification); + RuntimeFilterBucketPruneClassifier.Classification bucketClassification = + RuntimeFilterBucketPruneClassifier.classify(head.getType(), targetExpr, scanNode); Review Comment: Implemented in d5cef5da1e7. Both translator paths now test enable_runtime_filter_bucket_prune before invoking RuntimeFilterBucketPruneClassifier, so the disabled setting performs no selected-partition classification or discarded catalog work. The translator suite passes 4/4 on the current head. ########## 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: Fixed in 452b5173426. Bucket hashes are generated directly from HybridSet iterator values with RawValue::zlib_crc32, matching write routing without materializing a temporary ColumnString. The retained payload is now DorisVector<uint32_t>, so its buffer uses the allocator-aware tracked storage. StringHashesMatchWriteRoutingWithoutMaterialization includes a 64-KiB value and verifies the exact write-path hashes. ########## 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: Fixed in 452b5173426. OlapScanNode now guards metadata attachment with a plan-scoped synchronized idempotence flag. The first worker populates every shared TScanRange; later worker serializations return without another global range walk. testRuntimeFilterBucketMetadataAttachedOnceAcrossWorkers clears the source map after the first call, then proves a second worker call neither reads it nor loses the attached fields. OlapScanNodeTest passes 8/8. ########## 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: Fixed in 452b5173426. All fallible producer construction, initialization, insertion, and state setup now occurs before any worker starts. A Defer guard always releases filter_published when needed and joins every joinable thread. The normal path explicitly joins workers before the fatal behavior assertions. ScannerLateArrivalRfTest passes 2/2. ########## 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: Fixed in 452b5173426. The duplicate RuntimeFilterBucketPruneRange type and _rf_bucket_prune_ranges vector were removed. RuntimeFilterBucketPruner reads tablet_id, bucket_seq, and bucket_num directly from the owned _scan_ranges TPaloScanRange objects, so existing move/resize compaction also releases pruned metadata. No second per-range retained allocation remains. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTranslator.java: ########## @@ -249,6 +249,11 @@ private void createLegacyRuntimeFilterFromGroup(List<RuntimeFilter> group, RuntimeFilterPartitionPruneClassifier.classify( head.getType(), targetExpr, nereidsTargetExprList.get(i), scanNode); setPartitionPruningMetadata(origFilter, scanNode, classification); + RuntimeFilterBucketPruneClassifier.Classification bucketClassification = Review Comment: Implemented in d5cef5da1e7 and revalidated on 452b5173426. RuntimeFilterTranslatorBucketPruneTest covers grouped same-scan same-target and different-target cases, cast/non-identity suppression, and enabled/disabled serialization. It asserts both planId_to_target_expr and bucket_pruning_target_ids; all 4 translator tests pass on the current head. ########## 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: Fixed in 452b5173426. RuntimeFilterMerger now creates its own fully initialized wrapper and merge_from always copies/merges producer state instead of adopting the first producer wrapper. Bloom-filter first merges also deep-copy their directory. partial_merge_does_not_alias_producer_hash_cache models P0 merge, local P0 cache materialization, then P1 merge; it verifies distinct wrappers, an unchanged one-value P0 cache, and both hashes in the finalized merger cache. -- 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]
