morrySnow commented on code in PR #65837: URL: https://github.com/apache/doris/pull/65837#discussion_r3765484230
########## be/src/exec/runtime_filter/runtime_filter_bucket_pruner.cpp: ########## @@ -0,0 +1,167 @@ +// 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 "core/column/column.h" +#include "core/data_type/data_type.h" +#include "core/data_type/primitive_type.h" +#include "core/string_ref.h" +#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 { + +static void materialize_hashes(const VExprSPtr& target_expr, HybridSetBase* hybrid_set, + std::vector<uint32_t>* hashes) { + DORIS_CHECK(target_expr != nullptr); + DORIS_CHECK(hybrid_set != nullptr); + + const DataTypePtr& data_type = target_expr->data_type(); + MutableColumnPtr column = data_type->create_column(); + PrimitiveType primitive_type = data_type->get_primitive_type(); + auto* iter = hybrid_set->begin(); + while (iter->has_next()) { + const void* value = iter->get_value(); + DORIS_CHECK(value != nullptr); + if (is_string_type(primitive_type)) { + const auto* string_value = reinterpret_cast<const StringRef*>(value); + column->insert_data(string_value->data, string_value->size); + } else { + // ColumnVector::insert_data ignores length for fixed-length values. + column->insert_data(reinterpret_cast<const char*>(value), 0); + } + iter->next(); + } + if (hybrid_set->contain_null() && data_type->is_nullable()) { + // contain_null() is true only for a null-aware filter. Keep the bucket that owns + // NULL probe rows by hashing NULL with the same nullable CRC semantics as partitioning. + column->insert_default(); + } + + hashes->assign(column->size(), 0); + if (!hashes->empty()) { + column->update_crcs_with_value(hashes->data(), primitive_type, + static_cast<uint32_t>(column->size())); + } +} + +Status RuntimeFilterBucketPruner::prune_by_runtime_filters( + const std::vector<RuntimeFilterBucketPruneRange>& 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(); + } + + phmap::flat_hash_set<int64_t> newly_pruned; + for (const auto& conjunct_ctx : conjuncts) { + VExprSPtr root = conjunct_ctx->root(); + if (!root->is_rf_wrapper()) { + continue; + } + auto* wrapper = assert_cast<RuntimeFilterExpr*>(root.get()); + if (!eligible_filter_ids.contains(wrapper->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); Review Comment: These `DORIS_CHECK`/`DORIS_CHECK_EQ` assertions turn an invariant that is only guaranteed by FE-side classification code (`RuntimeFilterBucketPruneClassifier` + `castTargetToSourceTypeIfNeeded` in `RuntimeFilterTranslator`) into a BE crash. If any future change produces an IN/IN_OR_BLOOM RF whose probe expr is not a bare `SLOT_REF` (e.g., a new translation path that wraps the slot in a cast), every BE executing the query crashes. The rest of this function already degrades gracefully (`hybrid_set == nullptr` -> continue, size limit -> continue); using `continue` (or skipping the conjunct) for unexpected impl shapes would make the feature robust to FE/BE skew without losing correctness. Given this is a correctness-preserving optimization, crashing the BE is a disproportionate failure mode. ########## 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: `setRuntimeFilterBucketPruneParameters()` hard-fails the whole query plan via `Preconditions.checkState` when bucket metadata is missing for any scan range, even though bucket pruning is purely an optimization. The metadata gap is real for point-query scans: `computeTabletInfo()` only populates `tabletId2BucketInfo` when `!isPointQuery()` (line ~1025), while `RuntimeFilterBucketPruneClassifier.classify()` never gates on `isPointQuery()`. Today short-circuit point queries are join-free so they can't be RF targets and the check is unreachable, but the invariant is split across two unlinked places in the FE (classification eligibility vs. metadata population). If any future change classifies a point-query scan or a scan whose `computeTabletInfo()` ran under a different `isPointQuery()` value, a valid query turns into a plan failure instead of simply skipping the optimization. Consider gating classification on `!isPointQuery()` or skipping ranges without metadata (BE already handles m issing bucket fields gracefully via `__isset` checks). ########## be/src/exec/operator/olap_scan_operator.cpp: ########## @@ -1109,10 +1113,43 @@ void OlapScanLocalState::set_scan_ranges(RuntimeState* state, for (auto& scan_range : scan_ranges) { DCHECK(scan_range.scan_range.__isset.palo_scan_range); _scan_ranges.emplace_back(new TPaloScanRange(scan_range.scan_range.palo_scan_range)); + const auto& palo_scan_range = scan_range.scan_range.palo_scan_range; + if (palo_scan_range.__isset.bucket_seq || palo_scan_range.__isset.bucket_num) { + DORIS_CHECK(palo_scan_range.__isset.bucket_seq); + DORIS_CHECK(palo_scan_range.__isset.bucket_num); + _rf_bucket_prune_ranges.emplace_back(palo_scan_range.tablet_id, + palo_scan_range.bucket_seq, + palo_scan_range.bucket_num); + } COUNTER_UPDATE(_tablet_counter, 1); } } +Status OlapScanLocalState::_on_runtime_filter_update(const VExprContextSPtrs& new_conjuncts) { + RETURN_IF_ERROR(Base::_on_runtime_filter_update(new_conjuncts)); + if (!state()->query_options().enable_runtime_filter_bucket_prune || + _rf_bucket_prune_ranges.empty()) { + return Status::OK(); + } + + int64_t newly_pruned = 0; + RETURN_IF_ERROR(_rf_bucket_pruner.prune_by_runtime_filters( Review Comment: `prune_by_runtime_filters` is invoked with `_parent->runtime_filter_descs()` and `new_conjuncts`, but the bucket metadata in `_rf_bucket_prune_ranges` was captured in `set_scan_ranges()` from the thrift scan ranges. If the FE ever marks a scan eligible (`bucket_pruning_target_ids`) without sending `bucket_seq`/`bucket_num` on its ranges (or BE/FE versions skew), the DORIS_CHECKs in `set_scan_ranges` crash on scan init. The code already guards with `__isset` checks before recording ranges; consider returning `Status::OK()` (conservative fallback, matching the feature's stated fallback policy) rather than asserting, since the FE-side metadata is optional by thrift definition. ########## be/test/exec/runtime_filter/runtime_filter_bucket_pruner_test.cpp: ########## @@ -0,0 +1,240 @@ +// 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 <gtest/gtest.h> + +#include <cstdint> +#include <memory> +#include <set> +#include <utility> +#include <vector> + +#include "core/column/column_vector.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "exec/runtime_filter/runtime_filter_definitions.h" +#include "exprs/create_predicate_function.h" +#include "exprs/runtime_filter_expr.h" +#include "exprs/vdirect_in_predicate.h" +#include "exprs/vexpr_context.h" +#include "exprs/vslot_ref.h" + +namespace doris { + +class RuntimeFilterBucketPrunerTest : public testing::Test { +protected: + static constexpr int SCAN_NODE_ID = 10; + + VExprContextSPtr make_in_conjunct(int filter_id, const std::vector<int32_t>& values) { + std::shared_ptr<HybridSetBase> set(create_set(TYPE_INT, false)); + for (const int32_t value : values) { + set->insert(&value); + } + + TExprNode node; + node.__set_type(create_type_desc(TYPE_BOOLEAN)); + node.__set_node_type(TExprNodeType::IN_PRED); + node.in_predicate.__set_is_not_in(false); + node.__set_opcode(TExprOpcode::FILTER_IN); + node.__set_is_nullable(false); + auto impl = VDirectInPredicate::create_shared(node, std::move(set), true); + impl->add_child(VSlotRef::create_shared(/*slot_id=*/1, /*column_id=*/0, + /*column_uniq_id=*/1, + std::make_shared<DataTypeInt32>(), "dist_col")); + auto wrapper = RuntimeFilterExpr::create_shared(node, impl, 0, false, filter_id); + return std::make_shared<VExprContext>(wrapper); + } + + VExprContextSPtr make_non_exact_conjunct(int filter_id) { + TExprNode node; + node.__set_type(create_type_desc(TYPE_BOOLEAN)); + node.__set_node_type(TExprNodeType::BLOOM_PRED); + node.__set_opcode(TExprOpcode::RT_FILTER); + node.__set_is_nullable(false); + auto impl = VDirectInPredicate::create_shared(node, nullptr, true); + impl->add_child(VSlotRef::create_shared(/*slot_id=*/1, /*column_id=*/0, + /*column_uniq_id=*/1, + std::make_shared<DataTypeInt32>(), "dist_col")); + auto wrapper = RuntimeFilterExpr::create_shared(node, impl, 0, false, filter_id); + return std::make_shared<VExprContext>(wrapper); + } + + VExprContextSPtr make_null_aware_in_conjunct(int filter_id) { + std::shared_ptr<HybridSetBase> set(create_set(TYPE_INT, true)); + set->insert(static_cast<const void*>(nullptr)); + + TExprNode node; + node.__set_type(create_type_desc(TYPE_BOOLEAN)); + node.__set_node_type(TExprNodeType::NULL_AWARE_IN_PRED); + node.in_predicate.__set_is_not_in(false); + node.__set_opcode(TExprOpcode::FILTER_IN); + node.__set_is_nullable(false); + auto impl = VDirectInPredicate::create_shared(node, std::move(set), true); + impl->add_child(VSlotRef::create_shared( + /*slot_id=*/1, /*column_id=*/0, /*column_uniq_id=*/1, + std::make_shared<DataTypeNullable>(std::make_shared<DataTypeInt32>()), "dist_col")); + auto wrapper = RuntimeFilterExpr::create_shared(node, impl, 0, false, filter_id); + return std::make_shared<VExprContext>(wrapper); + } + + TRuntimeFilterDesc bucket_prune_desc(int filter_id) { + TRuntimeFilterDesc desc; + desc.__set_filter_id(filter_id); + desc.__set_bucket_pruning_target_ids({SCAN_NODE_ID}); + return desc; + } + + std::vector<RuntimeFilterBucketPruneRange> four_bucket_ranges() { + std::vector<RuntimeFilterBucketPruneRange> ranges; + for (int32_t bucket_seq = 0; bucket_seq < 4; ++bucket_seq) { + ranges.push_back({100 + bucket_seq, bucket_seq, 4}); + } + return ranges; + } + + int32_t bucket_for_value(int32_t value, int32_t bucket_num) { Review Comment: Test-quality note: the tests derive the expected bucket (`bucket_for_value`, `bucket_for_null`) using the exact same `update_crcs_with_value` API the production pruner uses, so they are self-consistent and cannot detect a divergence between the pruner's hash and the actual write-path distribution hash (`RawValue::zlib_crc32` / `HashUtil::zlib_crc_hash_null` used in `VOlapTablePartitionParam::find_tablets`). Such a divergence would silently lose rows in production while CI stays green. Suggest pinning at least one known value->bucket expectation computed via the write-path hash (e.g., `RawValue::zlib_crc32` with seed 0), plus an end-to-end case with a nullable distribution column and a null-aware IN filter to pin the NULL bucket semantics. ########## be/src/exec/runtime_filter/runtime_filter_bucket_pruner.cpp: ########## @@ -0,0 +1,167 @@ +// 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 "core/column/column.h" +#include "core/data_type/data_type.h" +#include "core/data_type/primitive_type.h" +#include "core/string_ref.h" +#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 { + +static void materialize_hashes(const VExprSPtr& target_expr, HybridSetBase* hybrid_set, + std::vector<uint32_t>* hashes) { + DORIS_CHECK(target_expr != nullptr); + DORIS_CHECK(hybrid_set != nullptr); + + const DataTypePtr& data_type = target_expr->data_type(); + MutableColumnPtr column = data_type->create_column(); + PrimitiveType primitive_type = data_type->get_primitive_type(); + auto* iter = hybrid_set->begin(); + while (iter->has_next()) { + const void* value = iter->get_value(); + DORIS_CHECK(value != nullptr); + if (is_string_type(primitive_type)) { + const auto* string_value = reinterpret_cast<const StringRef*>(value); + column->insert_data(string_value->data, string_value->size); + } else { + // ColumnVector::insert_data ignores length for fixed-length values. + column->insert_data(reinterpret_cast<const char*>(value), 0); + } + iter->next(); + } + if (hybrid_set->contain_null() && data_type->is_nullable()) { + // contain_null() is true only for a null-aware filter. Keep the bucket that owns + // NULL probe rows by hashing NULL with the same nullable CRC semantics as partitioning. + column->insert_default(); + } + + hashes->assign(column->size(), 0); + if (!hashes->empty()) { + column->update_crcs_with_value(hashes->data(), primitive_type, + static_cast<uint32_t>(column->size())); + } +} + +Status RuntimeFilterBucketPruner::prune_by_runtime_filters( + const std::vector<RuntimeFilterBucketPruneRange>& 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(); + } + + phmap::flat_hash_set<int64_t> newly_pruned; + for (const auto& conjunct_ctx : conjuncts) { + VExprSPtr root = conjunct_ctx->root(); + if (!root->is_rf_wrapper()) { + continue; + } + auto* wrapper = assert_cast<RuntimeFilterExpr*>(root.get()); + if (!eligible_filter_ids.contains(wrapper->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::vector<uint32_t> hashes; + materialize_hashes(target_expr, hybrid_set.get(), &hashes); + phmap::flat_hash_map<int32_t, phmap::flat_hash_set<int32_t>> selected_buckets_by_num; + for (const auto& range : ranges) { Review Comment: Perf note: this loop re-iterates *all* ranges on every update and re-hashes the entire IN set (`materialize_hashes` is O(set size), called per update). `newly_pruned` is call-local, so tablets already in `_pruned_tablet_ids` from earlier updates are re-examined on each late-arrival RF; and `OlapScanLocalState::_on_runtime_filter_update()` is invoked while holding `_conjuncts_lock`, so scanner threads calling `clone_conjunct_ctxs()` are blocked for the duration of the hashing. With `max_in_num` up to thousands of values and many tablets, and one update per late filter, this can add noticeable latency to late-RF processing. Suggest skipping ranges already in `_pruned_tablet_ids` and caching the per-(filter, bucket_num) bucket computation. -- 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]
