github-actions[bot] commented on code in PR #65837:
URL: https://github.com/apache/doris/pull/65837#discussion_r3777457520
##########
be/src/exec/scan/scanner_scheduler.cpp:
##########
@@ -176,14 +176,21 @@ void
ScannerScheduler::_scanner_scan(std::shared_ptr<ScannerContext> ctx,
Status status = Status::OK();
bool eos = false;
+ auto append_late_arrival_runtime_filter = [&] {
+ Status rf_status = scanner->try_append_late_arrival_runtime_filter();
+ if (!rf_status.ok()) {
+ LOG(WARNING) << "Failed to append late arrival runtime filter: "
+ << rf_status.to_string();
+ }
+ };
ASSIGN_STATUS_IF_CATCH_EXCEPTION(
RuntimeState* state = ctx->state(); DCHECK(nullptr != state);
// scanner->open may alloc plenty amount of memory(read blocks of
data),
// so better to also check low memory and clear free blocks here.
if (ctx->low_memory_mode()) { ctx->clear_free_blocks(); }
- if (scanner->check_partition_pruned()) { eos = true; }
+ if (scanner->is_pruned_by_runtime_filter()) { eos = true; }
Review Comment:
[P2] Release OLAP inputs when this first check prunes a never-prepared
scanner. `OlapScanner` already moved `Params.read_source` into `ReaderParams`
in its constructor, and bounded scanner concurrency leaves other delegates
pending while an active scanner can publish a late filter. When such a pending
delegate is first scheduled, this sets EOS before `prepare()`, so the later
cleanup block is skipped; the current hook cannot be used because it asserts
`_has_prepared`, and `close()` does not clear `ReaderParams`. The delegate then
retains cloned rowset readers, delete metadata, and tablet ownership until the
whole scan local state closes. Please make abandonment clear constructor-owned
inputs whether or not prepare ran, call it on this path too, and add a
bounded-concurrency test where pruning is published before another scanner's
first schedule.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/RuntimeFilterBucketPruneClassifier.java:
##########
@@ -0,0 +1,121 @@
+// 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.
+
+package org.apache.doris.nereids.glue.translator;
+
+import org.apache.doris.analysis.Expr;
+import org.apache.doris.analysis.SlotRef;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.DistributionInfo;
+import org.apache.doris.catalog.HashDistributionInfo;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.planner.OlapScanNode;
+import org.apache.doris.planner.PlanNode;
+import org.apache.doris.thrift.TRuntimeFilterType;
+
+/** Classifies direct single-column HASH targets for BE-side runtime-filter
bucket pruning. */
+final class RuntimeFilterBucketPruneClassifier {
+ private RuntimeFilterBucketPruneClassifier() {
+ }
+
+ static Classification classify(TRuntimeFilterType filterType, Expr
targetExpr, PlanNode scanNode) {
+ if (filterType != TRuntimeFilterType.IN && filterType !=
TRuntimeFilterType.IN_OR_BLOOM) {
+ return Classification.unsupported("runtime filter is not IN or
IN_OR_BLOOM");
+ }
+ if (!(scanNode instanceof OlapScanNode)) {
+ return Classification.unsupported("target scan is not an
OlapScanNode");
+ }
+ if (!(targetExpr instanceof SlotRef)) {
+ return Classification.unsupported("target expression is not a
direct SlotRef");
+ }
+
+ Column targetColumn = ((SlotRef) targetExpr).getColumn();
+ if (targetColumn == null) {
+ return Classification.unsupported("target SlotRef has no column");
+ }
+
+ OlapScanNode olapScanNode = (OlapScanNode) scanNode;
+ if (olapScanNode.isPointQuery()) {
+ return Classification.unsupported("target scan is a point query");
+ }
+ OlapTable table = olapScanNode.getOlapTable();
+ if (table == null || olapScanNode.getSelectedPartitionIds().isEmpty())
{
+ return Classification.unsupported("target scan has no selected
partitions");
+ }
+
+ Column distributionColumn = null;
+ for (Long partitionId : olapScanNode.getSelectedPartitionIds()) {
+ Partition partition = table.getPartition(partitionId);
+ if (partition == null) {
+ return Classification.unsupported("selected partition does not
exist");
+ }
+ DistributionInfo distributionInfo =
partition.getDistributionInfo();
+ if (!(distributionInfo instanceof HashDistributionInfo)) {
+ return Classification.unsupported("distribution type is not
HASH");
+ }
+ HashDistributionInfo hashDistributionInfo = (HashDistributionInfo)
distributionInfo;
+ if (hashDistributionInfo.getDistributionColumns().size() != 1) {
+ return Classification.unsupported("HASH distribution is not
single-column");
+ }
+ Column currentDistributionColumn =
hashDistributionInfo.getDistributionColumns().get(0);
+ if (!sameColumn(targetColumn, currentDistributionColumn)) {
+ return Classification.unsupported("target SlotRef is not the
HASH distribution column");
+ }
+ if (distributionColumn != null && !sameColumn(distributionColumn,
currentDistributionColumn)) {
+ return Classification.unsupported("selected partitions use
different distribution columns");
+ }
+ distributionColumn = currentDistributionColumn;
+ }
+ return Classification.supported();
+ }
+
+ private static boolean sameColumn(Column targetColumn, Column
distributionColumn) {
+ if (targetColumn == distributionColumn) {
+ return true;
+ }
+ return targetColumn.tryGetBaseColumnName()
Review Comment:
[P1] Reject computed MV columns even when their alias and type match the
base distribution column. A reduced failing plan is:
```text
HashJoin(mv.k = dim.k)
OlapScan(selected sync-MV, mv.k := abs(base.k))
```
`getOutputByIndex()` makes the target slot from the selected-index `Column`;
for a non-`SlotRef` `defineExpr`, `tryGetBaseColumnName()` falls back to the MV
alias. Historical duplicate-name MV metadata is retained for compatibility, so
a computed `k INT` can pass this comparison against base HASH column `k INT`.
BE then hashes the RF value `abs(k)` to choose buckets, although the rollup
tablet was placed by `hash(base.k)`; for `base.k=-1` and RF value `1`, this can
prune the matching row. Please require a direct base-column identity (for
example a direct `SlotRef` definition) and reject complex MV definitions, with
a historical alias-collision test.
##########
be/src/exec/scan/parallel_scanner_builder.h:
##########
@@ -54,7 +55,17 @@ class ParallelScannerBuilder {
_is_preaggregation(is_preaggregation),
_tablets(tablets.cbegin(), tablets.cend()),
_key_ranges(key_ranges.cbegin(), key_ranges.cend()),
- _read_sources(read_sources) {}
+ _read_sources(read_sources) {
+ DORIS_CHECK_EQ(_tablets.size(), scan_ranges.size());
+ for (size_t i = 0; i < _tablets.size(); ++i) {
Review Comment:
[P2] Avoid constructing this node-based identity map for every parallel
scan. `ParallelScannerBuilder` is used for no-join, composite-HASH,
session-disabled, and old-FE scans too, but its constructor still inserts one
`unordered_map` node per tablet (usually just storing unused `0,0`) while the
original ranges, copied tablets, and read sources are live. At the supported
20,000-partition x 768-bucket envelope, a one-BE/one-instance plan can create
15.36 million nodes plus the bucket array, adding hundreds of MiB of untracked
peak memory before any scanner is returned. This is distinct from the removed
local-state metadata vector: it is a later, higher-overhead allocation
introduced by the parallel-identity fix. Please carry the pair in an existing
compact per-tablet work item (or another allocator-aware aligned structure),
skip identity storage when metadata is absent, and cover a high-cardinality
ineligible parallel build.
--
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]
