github-actions[bot] commented on code in PR #65837: URL: https://github.com/apache/doris/pull/65837#discussion_r3765234849
########## 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: **[P2] Reuse exact-set hashes across scan local states** Every pipeline scan local state registers its own consumer, while global/remote/local-merge publication signals those consumers with the same finalized wrapper and HybridSet. This block nevertheless allocates a new column/hash vector and walks the entire set independently in each local state. With the default 40,960-value exact-set ceiling and an automatically selected 64-instance plan, one filter repeats about 2.62 million materializations/CRCs before range work; explicit parallelism can make this worse, and multiple eligible filters multiply it again. This remains after the earlier accumulated-filter fix because that removed repeated old filters only within one local state. Please compute the immutable CRC vector once after READY and reuse it across consumers (keyed by the nullable physical target representation), or otherwise move the one-time work into shared finalized-filter state. -- 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]
