github-actions[bot] commented on code in PR #65837:
URL: https://github.com/apache/doris/pull/65837#discussion_r3776846781
##########
be/src/exec/operator/olap_scan_operator.cpp:
##########
@@ -827,6 +832,8 @@ Status
OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
p._olap_scan_node.is_preaggregation,
read_row_binlog,
resolve_binlog_scan_type(palo_scan_range),
+ palo_scan_range.bucket_seq,
Review Comment:
[P2] Propagate bucket identity through `ParallelScannerBuilder` too. The
ordinary factory forwards `bucket_seq`/`bucket_num` here, but the
default-enabled parallel branch above hands the builder only tablets/read
sources, and `_build_scanner()` leaves the new `OlapScanner::Params` fields at
`(0, 0)`. A late RF updates the pruner under the real bucket count (for example
8), while every split scanner then asks `is_bucket_pruned(0, 0)`, which can
never match; ready-at-open filters still compact before builder construction,
so the current tests mask this production path. Please carry each tablet's
bucket metadata into every row-range/segment split and add a late-arrival test
through the real parallel scanner factory.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/runtime/ThriftPlansBuilder.java:
##########
@@ -117,6 +118,8 @@ public static Map<DistributedPlanWorker,
TPipelineFragmentParamsList> plansToThr
// we should set runtime predicate first, then we can use heap sort
and to thrift
setRuntimePredicateIfNeed(coordinatorContext.scanNodes);
+ setRuntimeFilterBucketPruneParametersIfNeeded(
Review Comment:
[P2] Attach this metadata for the legacy Coordinator path too. With
`enable_nereids_distribute_planner=false`, Nereids still creates the eligible
runtime-filter descriptor, but `EnvFactory` selects `Coordinator`; this helper
is only called from `ThriftPlansBuilder`, and the legacy serializer never calls
`OlapScanNode.setRuntimeFilterBucketPruneParameters()`. It therefore sends
every range without `bucket_seq`/`bucket_num`, so BE records
`_has_rf_bucket_prune_metadata=false` and silently disables this feature even
though `enable_runtime_filter_bucket_prune` is true. Please move the one-time
annotation to a serializer-neutral point (or invoke it before legacy
assignment/serialization) and add coverage with the distributed planner
disabled that proves both fields arrive and the pruning counter is positive.
##########
be/src/exec/scan/scanner_scheduler.cpp:
##########
@@ -192,6 +199,15 @@ void
ScannerScheduler::_scanner_scan(std::shared_ptr<ScannerContext> ctx,
}
}
+ // A filter may become ready while prepare() is doing tablet
setup. Apply it before
+ // open() so a newly pruned OLAP scanner never initializes its
reader or eagerly reads.
+ if (!eos && !scanner->is_open()) {
+ append_late_arrival_runtime_filter();
+ if (scanner->is_pruned_by_runtime_filter()) {
Review Comment:
[P2] Release prepared reader inputs when this check prunes the scanner.
`OlapScanner::prepare()` has already created its `BlockReader`, cloned
contexts, and retained `ReaderParams.rs_splits`/delete metadata; the normal
prompt release is in `_open_impl()` after reader initialization, which this new
EOS path intentionally skips. `OlapScanner::close()` does not clear that state,
and `ScanLocalState::_scanners` owns every delegate until operator close, so
many eliminated tablets can keep their rowset readers pinned for the remainder
of a long scan. Please add an abandon/cleanup path for prepared-but-unopened
scanners and extend the lifecycle test to assert both no reader init and no
retained read-source splits.
##########
be/src/exec/runtime_filter/runtime_filter_bucket_pruner.cpp:
##########
@@ -0,0 +1,148 @@
+// 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.
+
+#include "exec/runtime_filter/runtime_filter_bucket_pruner.h"
+
+#include <gen_cpp/PlanNodes_types.h>
+
+#include <algorithm>
+#include <memory>
+#include <mutex>
+
+#include "exprs/hybrid_set.h"
+#include "exprs/runtime_filter_expr.h"
+#include "exprs/vexpr.h"
+#include "exprs/vexpr_context.h"
+#include "exprs/vslot_ref.h"
+
+namespace doris {
+
+Status RuntimeFilterBucketPruner::prune_by_runtime_filters(
+ const std::vector<std::unique_ptr<TPaloScanRange>>& ranges,
+ const VExprContextSPtrs& conjuncts, const
std::vector<TRuntimeFilterDesc>& rf_descs,
+ int scan_node_id, int max_in_num, int64_t* newly_pruned_count) {
+ *newly_pruned_count = 0;
+ if (ranges.empty()) {
+ return Status::OK();
+ }
+
+ phmap::flat_hash_set<int> eligible_filter_ids;
+ for (const auto& desc : rf_descs) {
+ if (desc.__isset.bucket_pruning_target_ids &&
+ desc.bucket_pruning_target_ids.contains(scan_node_id)) {
+ eligible_filter_ids.insert(desc.filter_id);
+ }
+ }
+ if (eligible_filter_ids.empty()) {
+ return Status::OK();
+ }
+
+ for (const auto& conjunct_ctx : conjuncts) {
+ VExprSPtr root = conjunct_ctx->root();
+ if (!root->is_rf_wrapper()) {
+ continue;
+ }
+ auto* rf_expr = assert_cast<RuntimeFilterExpr*>(root.get());
+ if (!eligible_filter_ids.contains(rf_expr->filter_id())) {
+ continue;
+ }
+
+ VExprSPtr impl = root->get_impl();
+ DORIS_CHECK(impl != nullptr);
+ std::shared_ptr<HybridSetBase> hybrid_set = impl->get_set_func();
+ if (hybrid_set == nullptr) {
+ // IN_OR_BLOOM may become a Bloom filter at runtime. A Bloom filter
+ // cannot be inverted to a safe finite bucket set.
+ continue;
+ }
+ if (hybrid_set->size() > max_in_num) {
+ continue;
+ }
+
+ DORIS_CHECK_EQ(impl->children().size(), 1);
+ VExprSPtr target_expr = impl->children()[0];
+ DORIS_CHECK_EQ(target_expr->node_type(), TExprNodeType::SLOT_REF);
+
+ std::shared_ptr<const std::vector<uint32_t>> hashes =
+ rf_expr->get_bucket_prune_hashes(target_expr->data_type());
+ phmap::flat_hash_map<int32_t, phmap::flat_hash_set<int32_t>>
new_selected_buckets_by_num;
+ for (const auto& range_ptr : ranges) {
+ DORIS_CHECK(range_ptr != nullptr);
+ const auto& range = *range_ptr;
+ DORIS_CHECK(range.__isset.bucket_seq);
+ DORIS_CHECK(range.__isset.bucket_num);
+ DORIS_CHECK_GT(range.bucket_num, 0);
+ DORIS_CHECK_GE(range.bucket_seq, 0);
+ DORIS_CHECK_LT(range.bucket_seq, range.bucket_num);
+
+ auto [selected_it, inserted] =
+ new_selected_buckets_by_num.try_emplace(range.bucket_num);
+ if (inserted) {
+ auto& selected_buckets = selected_it->second;
+ selected_buckets.reserve(
+ std::min(hashes->size(),
static_cast<size_t>(range.bucket_num)));
+ for (uint32_t hash : *hashes) {
+ selected_buckets.insert(
+ static_cast<int32_t>(hash %
static_cast<uint32_t>(range.bucket_num)));
+ }
+ }
+ }
+
+ std::unique_lock lock(_prune_mutex);
+ for (const auto& range_ptr : ranges) {
+ const auto& range = *range_ptr;
+ auto current_it = _selected_buckets_by_num.find(range.bucket_num);
+ bool was_selected = current_it == _selected_buckets_by_num.end() ||
+ current_it->second.contains(range.bucket_seq);
+ if (was_selected &&
+
!new_selected_buckets_by_num.at(range.bucket_num).contains(range.bucket_seq)) {
+ ++*newly_pruned_count;
+ }
+ }
+ for (auto& [bucket_num, new_selected_buckets] :
new_selected_buckets_by_num) {
+ auto current_it = _selected_buckets_by_num.find(bucket_num);
+ if (current_it == _selected_buckets_by_num.end()) {
+ _selected_buckets_by_num.emplace(bucket_num,
std::move(new_selected_buckets));
+ } else {
+ for (auto bucket_it = current_it->second.begin();
+ bucket_it != current_it->second.end();) {
+ if (!new_selected_buckets.contains(*bucket_it)) {
+ bucket_it = current_it->second.erase(bucket_it);
+ } else {
+ ++bucket_it;
+ }
+ }
+ }
+ }
+ _pruned_tablet_count += *newly_pruned_count;
Review Comment:
[P2] Add only the current filter's pruning delta here. `newly_pruned_count`
is initialized once before the outer conjunct loop, so when two eligible
filters eliminate 3 ranges and then 1 more it progresses `3 -> 4`, while this
line retains `3 + 4 = 7` even though only four ranges were removed. Initial
acquisition and one late poll can both deliver multiple ready filters, and this
retained value is exposed directly as `BucketsPrunedByRuntimeFilter`. Please
keep a per-filter delta (or add the aggregate once after the loop) and cover
two filters whose intersection removes an additional bucket.
--
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]
