HappenLee commented on code in PR #65837:
URL: https://github.com/apache/doris/pull/65837#discussion_r3763858788
##########
fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java:
##########
@@ -1016,9 +1017,13 @@ private void computeTabletInfo() throws UserException {
scanTabletIds.addAll(allTabletIds);
}
- if (!isPointQuery()) {
- for (int i = 0; i < allTabletIds.size(); i++) {
- tabletId2BucketSeq.put(allTabletIds.get(i), i);
+ for (int i = 0; i < allTabletIds.size(); i++) {
Review Comment:
Fixed in b716b37c76e. Restored the point-query exclusion, so short-circuit
lookups build no all-tablet bucket metadata. For ordinary scans, bucket
sequence and bucket count are now packed into one Map<Long, Long> instead of
two maps.
##########
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) {
Review Comment:
Fixed in b716b37c76e. Each update snapshots only the newly appended
runtime-filter conjuncts while holding _conjuncts_lock, releases the lock, and
passes that slice to partition and bucket pruning. Initial ready filters use
the same sliced path, and partition-pruner publication is now safe across
concurrent updates. This reduces staggered processing from F(F+1)/2 filter
passes to F.
##########
be/src/exec/scan/olap_scanner.cpp:
##########
@@ -815,6 +815,15 @@ bool OlapScanner::check_partition_pruned() const {
return
_local_state->is_partition_pruned(_tablet_reader_params.tablet->partition_id());
}
+bool OlapScanner::check_bucket_pruned() const {
+ if (!_local_state) {
+ return false;
+ }
+ auto* olap_local_state = assert_cast<OlapScanLocalState*>(_local_state);
+ return olap_local_state->_is_tablet_pruned_by_runtime_filter(
Review Comment:
Fixed in b716b37c76e. The scheduler now uses the single scanner API
is_pruned_by_runtime_filter() for partition and bucket pruning.
OlapScanLocalState checks the immutable _rf_bucket_prune_ranges eligibility
before calling is_tablet_pruned(), so ineligible and old-FE scans never acquire
the bucket-pruner mutex.
--
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]
