HappenLee commented on code in PR #65837: URL: https://github.com/apache/doris/pull/65837#discussion_r3766310568
########## 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; Review Comment: Fixed in 05fbc3805d2. IN-filter consumers now retain the shared finalized RuntimeFilterWrapper, and the wrapper computes bucket CRCs with std::call_once. Nullable and non-nullable physical targets have separate immutable shared vectors, so all scan local states reuse the same materialization. ExactSetHashesSharedAcrossConsumers verifies pointer reuse and the nullable cache split. ########## 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 05fbc3805d2. RuntimeFilterBucketPruneClassifier now rejects OlapScanNode.isPointQuery() before marking the target eligible, matching computeTabletInfo(), which intentionally omits bucket metadata for point queries. Eligible scans therefore keep the strict serialization invariant, while point queries skip this optimization. Added testPointQueryRejected. ########## 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: Fixed in 05fbc3805d2. Unit-test expected buckets now use the actual write-routing APIs RawValue::zlib_crc32 and HashUtil::zlib_crc_hash_null instead of the production pruner API. I also added an end-to-end nullable HASH-distribution regression using a null-safe broadcast join: it returns the NULL row (NULL, 90) and asserts RuntimeFilterPrunedBuckets is positive. The suite passed twice against the replaced $run_path/doris build (generation and normal verification). -- 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]
