HappenLee commented on code in PR #65837: URL: https://github.com/apache/doris/pull/65837#discussion_r3766594002
########## 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: Updated in 5e54eef7510. The finalized wrapper now owns one std::call_once cache shared by nullable and non-nullable targets. It hashes normal values once and appends HashUtil::zlib_crc_hash_null(0) whenever the exact set contains NULL. Nullable pruning remains exact; a non-nullable target may conservatively retain one extra bucket but cannot prune matching rows. ExactSetHashesSharedAcrossConsumers now verifies both target nullabilities receive the same vector, and NonNullableTargetConservativelyKeepsNullBucket pins the intended conservative behavior. -- 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]
