This is an automated email from the ASF dual-hosted git repository. englefly pushed a commit to branch rf-tablet-prune in repository https://gitbox.apache.org/repos/asf/doris.git
commit b1813caa8ede19407bdc7ef4fcfb8564b8516997 Author: englefly <[email protected]> AuthorDate: Wed Jul 22 10:03:01 2026 +0800 (fe-part) prune tablets by in-runtime-filter --- .../RuntimeFilterTabletPruneClassifier.java | 136 ++++++++++++++++ .../glue/translator/RuntimeFilterTranslator.java | 43 +++-- .../org/apache/doris/planner/OlapScanNode.java | 65 ++++++++ .../org/apache/doris/planner/RuntimeFilter.java | 39 +++++ .../java/org/apache/doris/qe/SessionVariable.java | 14 ++ .../RuntimeFilterTabletPruneClassifierTest.java | 178 +++++++++++++++++++++ gensrc/thrift/PlanNodes.thrift | 17 ++ 7 files changed, 476 insertions(+), 16 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTabletPruneClassifier.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTabletPruneClassifier.java new file mode 100644 index 00000000000..72f3dd67999 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTabletPruneClassifier.java @@ -0,0 +1,136 @@ +// 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.planner.OlapScanNode; +import org.apache.doris.planner.PlanNode; +import org.apache.doris.thrift.TRuntimeFilterType; + +import java.util.List; + +/** + * Classifies whether one runtime-filter target can safely drive BE-side tablet pruning. + * + * <p>BE prunes a tablet when none of the filter values hashes into the tablet's bucket + * index (hash(value) % bucket_num). This requires the filter values to be enumerable + * and hash-compatible with the raw distribution column values, so this gate is much + * stricter than the partition-pruning one: + * + * <ul> + * <li>Only IN filters qualify: BLOOM filters are bitmaps (values cannot be + * enumerated) and MIN_MAX filters only carry a range.</li> + * <li>The target must be a direct SlotRef on the table's single hash distribution + * column. A cast-wrapped or computed target would be hashed by BE on a value + * whose type/encoding differs from what the load-time distribution hash used.</li> + * <li>Auto-bucket tables are excluded: their bucket count is adjusted dynamically + * and the value seen at planning time may be stale at execution time.</li> + * </ul> + * + * <p>Note this classifier cannot check per-partition bucket counts (partitions are + * selected after RF translation). OlapScanNode.toThrift re-validates that all selected + * partitions share one bucket count before serializing tablet bucket info. + */ +final class RuntimeFilterTabletPruneClassifier { + private RuntimeFilterTabletPruneClassifier() { + } + + static Classification classify(TRuntimeFilterType filterType, Expr targetExpr, PlanNode scanNode) { + if (filterType != TRuntimeFilterType.IN) { + return Classification.unsupported("only IN runtime filter can drive tablet pruning"); + } + if (!(scanNode instanceof OlapScanNode)) { + return Classification.unsupported("target scan is not an OlapScanNode"); + } + OlapScanNode olapScanNode = (OlapScanNode) scanNode; + OlapTable table = olapScanNode.getOlapTable(); + if (table == null) { + return Classification.unsupported("target scan has no OlapTable"); + } + + DistributionInfo distributionInfo = table.getDefaultDistributionInfo(); + if (!(distributionInfo instanceof HashDistributionInfo)) { + return Classification.unsupported("table is not hash distributed"); + } + HashDistributionInfo hashDistribution = (HashDistributionInfo) distributionInfo; + if (hashDistribution.getAutoBucket()) { + return Classification.unsupported("auto bucket table has unstable bucket num"); + } + List<Column> distributionColumns = hashDistribution.getDistributionColumns(); + if (distributionColumns.size() != 1) { + return Classification.unsupported("only single-column distribution key is supported"); + } + + if (!(targetExpr instanceof SlotRef)) { + return Classification.unsupported("target is not a direct SlotRef on the distribution column"); + } + SlotRef slotRef = (SlotRef) targetExpr; + if (slotRef.getColumn() == null) { + return Classification.unsupported("target SlotRef has no resolved column"); + } + if (!sameColumn(slotRef.getColumn(), distributionColumns.get(0))) { + return Classification.unsupported("target SlotRef is not the distribution column"); + } + return Classification.supported(); + } + + private static boolean sameColumn(Column targetColumn, Column distributionColumn) { + if (targetColumn == distributionColumn) { + return true; + } + int targetUniqueId = targetColumn.getUniqueId(); + int distributionUniqueId = distributionColumn.getUniqueId(); + if (targetUniqueId != Column.COLUMN_UNIQUE_ID_INIT_VALUE + && distributionUniqueId != Column.COLUMN_UNIQUE_ID_INIT_VALUE + && targetUniqueId == distributionUniqueId) { + return true; + } + return targetColumn.equals(distributionColumn); + } + + static final class Classification { + private final boolean canPruneTablets; + private final String unsupportedReason; + + private Classification(boolean canPruneTablets, String unsupportedReason) { + this.canPruneTablets = canPruneTablets; + this.unsupportedReason = unsupportedReason; + } + + static Classification supported() { + return new Classification(true, ""); + } + + static Classification unsupported(String reason) { + return new Classification(false, reason); + } + + boolean canPruneTablets() { + return canPruneTablets; + } + + String getUnsupportedReason() { + return unsupportedReason; + } + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTranslator.java index 286330d79ea..fa6c05c77b3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTranslator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTranslator.java @@ -244,14 +244,8 @@ public class RuntimeFilterTranslator { || targetAdjustedAfterNonIdentityList.get(i)) { continue; } - RuntimeFilterPartitionPruneClassifier.Classification classification = - RuntimeFilterPartitionPruneClassifier.classify( - head.getType(), targetExpr, nereidsTargetExprList.get(i), scanNode); - if (classification.canPrunePartitions()) { - origFilter.markTargetCanPrunePartitions(scanNode.getId()); - } - origFilter.setTargetPartitionMonotonicity( - scanNode.getId(), classification.getPartitionMonotonicity()); + classifyAndMarkPruningTargets(head.getType(), targetExpr, + nereidsTargetExprList.get(i), scanNode, origFilter); } origFilter.setBloomFilterSizeCalculatedByNdv(head.isBloomFilterSizeCalculatedByNdv()); setWaitTimeMs(origFilter, head.isNonBlocking(), isLocalTarget); @@ -351,14 +345,8 @@ public class RuntimeFilterTranslator { if (targetAdjustedAfterNonIdentityList.get(i)) { continue; } - RuntimeFilterPartitionPruneClassifier.Classification classification = - RuntimeFilterPartitionPruneClassifier.classify( - filter.getType(), targetExpr, filter.getTargetExpressions().get(i), scanNode); - if (classification.canPrunePartitions()) { - origFilter.markTargetCanPrunePartitions(scanNode.getId()); - } - origFilter.setTargetPartitionMonotonicity( - scanNode.getId(), classification.getPartitionMonotonicity()); + classifyAndMarkPruningTargets(filter.getType(), targetExpr, + filter.getTargetExpressions().get(i), scanNode, origFilter); } origFilter.setBloomFilterSizeCalculatedByNdv(filter.isBloomFilterSizeCalculatedByNdv()); setWaitTimeMs(origFilter, filter.isNonBlocking(), isLocalTarget); @@ -388,6 +376,29 @@ public class RuntimeFilterTranslator { return origFilter; } + /** + * Run the partition-pruning and tablet-pruning classifiers on one final legacy + * target expression and record the results on the legacy runtime filter. + * Shared by the grouped-filter and single-filter translation paths above. + */ + private void classifyAndMarkPruningTargets(TRuntimeFilterType filterType, Expr targetExpr, + Expression nereidsTargetExpr, ScanNode scanNode, + org.apache.doris.planner.RuntimeFilter origFilter) { + RuntimeFilterPartitionPruneClassifier.Classification classification = + RuntimeFilterPartitionPruneClassifier.classify( + filterType, targetExpr, nereidsTargetExpr, scanNode); + if (classification.canPrunePartitions()) { + origFilter.markTargetCanPrunePartitions(scanNode.getId()); + } + origFilter.setTargetPartitionMonotonicity( + scanNode.getId(), classification.getPartitionMonotonicity()); + RuntimeFilterTabletPruneClassifier.Classification tabletClassification = + RuntimeFilterTabletPruneClassifier.classify(filterType, targetExpr, scanNode); + if (tabletClassification.canPruneTablets()) { + origFilter.markTargetCanPruneTablets(scanNode.getId()); + } + } + private void setWaitTimeMs(org.apache.doris.planner.RuntimeFilter filter, boolean isNonBlocking, boolean isLocalTarget) { if (isNonBlocking) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java index 2535961a76f..abd8ca6e826 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java @@ -1358,6 +1358,18 @@ public class OlapScanNode extends ScanNode { setPartitionBoundaries(msg.olap_scan_node); } + // Populate tablet -> bucket index mapping for BE-side runtime filter tablet + // pruning. Only serialize when this scan node actually has at least one IN + // runtime filter whose target can drive tablet pruning according to the + // FE-side classifier, so we don't bloat thrift for tables with many tablets + // but no usable RF target. + // Gated by session variable `enable_runtime_filter_tablet_prune`. + if (rfPruneCtx != null + && rfPruneCtx.getSessionVariable().isEnableRuntimeFilterTabletPrune() + && hasRfDrivingTabletPruning()) { + setTabletBucketInfos(msg.olap_scan_node); + } + super.toThrift(msg); } @@ -1374,6 +1386,59 @@ public class OlapScanNode extends ScanNode { return false; } + private boolean hasRfDrivingTabletPruning() { + // tabletId2BucketSeq is only filled by computeTabletInfo on the non-point-query + // path; point queries (lazyEvaluateRangeLocations) clear it and are excluded here. + if (tabletId2BucketSeq.isEmpty()) { + return false; + } + PlanNodeId myId = this.getId(); + for (RuntimeFilter rf : runtimeFilters) { + if (rf.canPruneTabletsFor(myId)) { + return true; + } + } + return false; + } + + private void setTabletBucketInfos(TOlapScanNode olapScanNode) { + int bucketNum = getUniformSelectedPartitionBucketNum(); + if (bucketNum <= 0) { + return; + } + olapScanNode.setTabletIdToBucketIndex(Maps.newHashMap(tabletId2BucketSeq)); + olapScanNode.setDistributionBucketNum(bucketNum); + } + + // Bucket counts are per-partition (a partition can be added with its own + // DISTRIBUTED BY ... BUCKETS clause, and dynamic partition tables may roll out + // partitions with different bucket counts). tabletId2BucketSeq stores each + // tablet's index within its own partition's bucket list, so those indexes are + // only comparable when all selected partitions share one bucket count. Returns + // the shared bucket count, or -1 when bucket counts differ (tablet pruning must + // then be skipped: hashing RF values with a wrong modulus would wrongly skip + // tablets and lose data). + private int getUniformSelectedPartitionBucketNum() { + int bucketNum = -1; + for (Long partitionId : selectedPartitionIds) { + Partition partition = olapTable.getPartition(partitionId); + if (partition == null) { + return -1; + } + DistributionInfo distributionInfo = partition.getDistributionInfo(); + if (!(distributionInfo instanceof HashDistributionInfo)) { + return -1; + } + int partitionBucketNum = distributionInfo.getBucketNum(); + if (bucketNum == -1) { + bucketNum = partitionBucketNum; + } else if (bucketNum != partitionBucketNum) { + return -1; + } + } + return bucketNum; + } + private void setPartitionBoundaries(TOlapScanNode olapScanNode) { PartitionInfo partitionInfo = olapTable.getPartitionInfo(); PartitionType partType = partitionInfo.getType(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/RuntimeFilter.java b/fe/fe-core/src/main/java/org/apache/doris/planner/RuntimeFilter.java index 77bb4ffcdc1..1930b7a94b2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/RuntimeFilter.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/RuntimeFilter.java @@ -147,6 +147,12 @@ public final class RuntimeFilter { = new HashMap<>(); private final Set<PlanNodeId> partitionPruningTargetScanIds = new HashSet<>(); + // Target scan nodes whose tablets this filter can prune at runtime on BE. + // Filled by RuntimeFilterTabletPruneClassifier at translation time; only IN + // filters whose target is a direct SlotRef on the scan's single-column hash + // distribution key qualify (BE must enumerate the filter values to hash them). + private final Set<PlanNodeId> tabletPruningTargetScanIds = new HashSet<>(); + /** * Internal representation of a runtime filter target. */ @@ -391,6 +397,21 @@ public final class RuntimeFilter { } } + // Target scan nodes whose tablets this filter can prune at runtime. + // Populated upstream by RuntimeFilterTabletPruneClassifier; BE combines + // this with TOlapScanNode.tablet_id_to_bucket_index/distribution_bucket_num + // to skip tablets no filter value hashes into. + // Gated by session variable `enable_runtime_filter_tablet_prune`. + boolean enableRfTabletPrune = rfPruneCtx != null + && rfPruneCtx.getSessionVariable().isEnableRuntimeFilterTabletPrune(); + if (enableRfTabletPrune && !tabletPruningTargetScanIds.isEmpty()) { + Set<Integer> planIds = new HashSet<>(); + for (PlanNodeId id : tabletPruningTargetScanIds) { + planIds.add(id.asInt()); + } + tFilter.setPlanIdToCanPruneTablets(planIds); + } + return tFilter; } @@ -422,6 +443,24 @@ public final class RuntimeFilter { return partitionPruningTargetScanIds.contains(scanNodeId); } + /** + * Record that a target can drive tablet pruning and needs the scan's + * tablet -> bucket index mapping serialized. + */ + public void markTargetCanPruneTablets(PlanNodeId scanNodeId) { + tabletPruningTargetScanIds.add(scanNodeId); + } + + /** + * Returns true iff this RF can drive tablet pruning for the given target + * scan node. Used by OlapScanNode.toThrift to decide whether it is worth + * serializing tablet_id_to_bucket_index to BE. The single source of truth + * for that decision lives in RuntimeFilterTabletPruneClassifier. + */ + public boolean canPruneTabletsFor(PlanNodeId scanNodeId) { + return tabletPruningTargetScanIds.contains(scanNodeId); + } + public boolean hasTargets() { return !targets.isEmpty(); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index df3fcbac6e0..d642f807ebd 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -456,6 +456,9 @@ public class SessionVariable implements Serializable, Writable { public static final String ENABLE_RUNTIME_FILTER_PARTITION_PRUNE = "enable_runtime_filter_partition_prune"; + public static final String ENABLE_RUNTIME_FILTER_TABLET_PRUNE = + "enable_runtime_filter_tablet_prune"; + public static final String ENABLE_PRUNE_NESTED_COLUMN = "enable_prune_nested_column"; static final String SESSION_CONTEXT = "session_context"; @@ -2195,6 +2198,9 @@ public class SessionVariable implements Serializable, Writable { @VarAttrDef.VarAttr(name = ENABLE_RUNTIME_FILTER_PARTITION_PRUNE, needForward = true, fuzzy = true) public boolean enableRuntimeFilterPartitionPrune = true; + @VarAttrDef.VarAttr(name = ENABLE_RUNTIME_FILTER_TABLET_PRUNE, needForward = true) + public boolean enableRuntimeFilterTabletPrune = false; + /** * The client can pass some special information by setting this session variable in the format: "k1:v1;k2:v2". * For example, trace_id can be passed to trace the query request sent by the user. @@ -5485,6 +5491,14 @@ public class SessionVariable implements Serializable, Writable { this.enableRuntimeFilterPartitionPrune = enableRuntimeFilterPartitionPrune; } + public boolean isEnableRuntimeFilterTabletPrune() { + return enableRuntimeFilterTabletPrune; + } + + public void setEnableRuntimeFilterTabletPrune(boolean enableRuntimeFilterTabletPrune) { + this.enableRuntimeFilterTabletPrune = enableRuntimeFilterTabletPrune; + } + public void setFragmentTransmissionCompressionCodec(String codec) { this.fragmentTransmissionCompressionCodec = codec; } diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTabletPruneClassifierTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTabletPruneClassifierTest.java new file mode 100644 index 00000000000..f1b1a10f2d1 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTabletPruneClassifierTest.java @@ -0,0 +1,178 @@ +// 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.CastExpr; +import org.apache.doris.analysis.Expr; +import org.apache.doris.analysis.SlotDescriptor; +import org.apache.doris.analysis.SlotId; +import org.apache.doris.analysis.SlotRef; +import org.apache.doris.analysis.TupleId; +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.HashDistributionInfo; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.PrimitiveType; +import org.apache.doris.catalog.RandomDistributionInfo; +import org.apache.doris.catalog.Type; +import org.apache.doris.nereids.glue.translator.RuntimeFilterTabletPruneClassifier.Classification; +import org.apache.doris.planner.OlapScanNode; +import org.apache.doris.planner.PlanNode; +import org.apache.doris.planner.PlanNodeId; +import org.apache.doris.thrift.TRuntimeFilterType; + +import com.google.common.collect.ImmutableList; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +class RuntimeFilterTabletPruneClassifierTest { + private static final Column DIST_COLUMN = new Column("dist_col", PrimitiveType.INT); + + @Test + void testInFilterOnDistributionColumnSupported() { + Classification classification = classify(TRuntimeFilterType.IN, distSlotRef(), + hashScanNode(1, false)); + + Assertions.assertTrue(classification.canPruneTablets()); + Assertions.assertTrue(classification.getUnsupportedReason().isEmpty()); + } + + @Test + void testBloomFilterUnsupported() { + Classification classification = classify(TRuntimeFilterType.BLOOM, distSlotRef(), + hashScanNode(1, false)); + + assertUnsupported(classification, "IN"); + } + + @Test + void testInOrBloomFilterUnsupported() { + // IN_OR_BLOOM may degrade to a bitmap bloom filter at runtime, whose values + // cannot be enumerated for hashing. + Classification classification = classify(TRuntimeFilterType.IN_OR_BLOOM, distSlotRef(), + hashScanNode(1, false)); + + assertUnsupported(classification, "IN"); + } + + @Test + void testMinMaxFilterUnsupported() { + Classification classification = classify(TRuntimeFilterType.MIN_MAX, distSlotRef(), + hashScanNode(1, false)); + + assertUnsupported(classification, "IN"); + } + + @Test + void testNonOlapScanNodeUnsupported() { + PlanNode scanNode = Mockito.mock(PlanNode.class); + + Classification classification = RuntimeFilterTabletPruneClassifier.classify( + TRuntimeFilterType.IN, distSlotRef(), scanNode); + + assertUnsupported(classification, "OlapScanNode"); + } + + @Test + void testRandomDistributionUnsupported() { + OlapTable table = Mockito.mock(OlapTable.class); + Mockito.when(table.getDefaultDistributionInfo()) + .thenReturn(new RandomDistributionInfo(4)); + OlapScanNode scanNode = Mockito.mock(OlapScanNode.class); + Mockito.when(scanNode.getOlapTable()).thenReturn(table); + + Classification classification = RuntimeFilterTabletPruneClassifier.classify( + TRuntimeFilterType.IN, distSlotRef(), scanNode); + + assertUnsupported(classification, "hash"); + } + + @Test + void testAutoBucketUnsupported() { + Classification classification = classify(TRuntimeFilterType.IN, distSlotRef(), + hashScanNode(1, true)); + + assertUnsupported(classification, "auto bucket"); + } + + @Test + void testMultiColumnDistributionKeyUnsupported() { + Classification classification = classify(TRuntimeFilterType.IN, distSlotRef(), + hashScanNode(2, false)); + + assertUnsupported(classification, "single-column"); + } + + @Test + void testNonDistributionColumnUnsupported() { + Column otherColumn = new Column("other_col", PrimitiveType.INT); + Classification classification = classify(TRuntimeFilterType.IN, slotRefOf(otherColumn), + hashScanNode(1, false)); + + assertUnsupported(classification, "distribution column"); + } + + @Test + void testCastWrappedTargetUnsupported() { + // A casted target would be hashed by BE on a value whose type/encoding + // differs from the raw distribution column values used at load time. + CastExpr castTarget = new CastExpr(Type.BIGINT, distSlotRef(), true); + Classification classification = classify(TRuntimeFilterType.IN, castTarget, + hashScanNode(1, false)); + + assertUnsupported(classification, "SlotRef"); + } + + private static SlotRef distSlotRef() { + return slotRefOf(DIST_COLUMN); + } + + private static SlotRef slotRefOf(Column column) { + SlotDescriptor slotDescriptor = new SlotDescriptor(new SlotId(1), new TupleId(1)); + slotDescriptor.setColumn(column); + slotDescriptor.setType(column.getType()); + return new SlotRef(slotDescriptor); + } + + private static OlapScanNode hashScanNode(int distributionColumnCount, boolean autoBucket) { + ImmutableList.Builder<Column> columns = ImmutableList.builder(); + for (int i = 0; i < distributionColumnCount; i++) { + columns.add(i == 0 ? DIST_COLUMN : new Column("dist_col_" + i, PrimitiveType.INT)); + } + HashDistributionInfo distributionInfo = + new HashDistributionInfo(4, autoBucket, columns.build()); + OlapTable table = Mockito.mock(OlapTable.class); + Mockito.when(table.getDefaultDistributionInfo()).thenReturn(distributionInfo); + OlapScanNode scanNode = Mockito.mock(OlapScanNode.class); + Mockito.when(scanNode.getOlapTable()).thenReturn(table); + Mockito.when(scanNode.getId()).thenReturn(new PlanNodeId(7)); + return scanNode; + } + + private static Classification classify( + TRuntimeFilterType filterType, Expr targetExpr, PlanNode scanNode) { + return RuntimeFilterTabletPruneClassifier.classify(filterType, targetExpr, scanNode); + } + + private static void assertUnsupported(Classification classification, String reasonKeyword) { + Assertions.assertFalse(classification.canPruneTablets()); + Assertions.assertTrue(classification.getUnsupportedReason().contains(reasonKeyword), + "reason should mention '" + reasonKeyword + "' but was: " + + classification.getUnsupportedReason()); + } +} diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift index dddc2d17c32..6d24b011690 100644 --- a/gensrc/thrift/PlanNodes.thrift +++ b/gensrc/thrift/PlanNodes.thrift @@ -1022,6 +1022,16 @@ struct TOlapScanNode { 27: optional list<TPartitionBoundary> partition_boundaries // Slot ids of extra storage key columns used only to align the scan tuple with storage schema. 28: optional set<i32> extra_key_column_slot_ids + // Tablet -> hash bucket index (0 .. distribution_bucket_num-1) for BE-side runtime + // filter tablet pruning. Only set when enable_runtime_filter_tablet_prune=true, at + // least one eligible IN RF targets this scan's distribution column, and ALL selected + // partitions share the same bucket count (bucket counts are per-partition and may + // differ, e.g. partitions added with their own DISTRIBUTED BY ... BUCKETS clause). + 29: optional map<Types.TTabletId, i32> tablet_id_to_bucket_index + // Hash bucket count shared by all selected partitions. BE computes + // hash(rf_value) % distribution_bucket_num to match a tablet's bucket index. + // Only set together with tablet_id_to_bucket_index. + 30: optional i32 distribution_bucket_num } struct TEqJoinCondition { @@ -1629,6 +1639,13 @@ struct TRuntimeFilterDesc { // slice and must be merged before being applied. Computed truthfully by FE after local // exchange planning; replaces inferring this from the target scan's is_serial_operator. 21: optional bool force_local_merge; + + // Target scan node IDs for which this RF can drive BE-side tablet (hash bucket) + // pruning. A scan node appears here iff this is an IN filter whose target is a + // direct SlotRef on the scan's single-column hash distribution key. BE uses this + // together with TOlapScanNode.tablet_id_to_bucket_index/distribution_bucket_num + // to skip tablets whose bucket index no RF value hashes into. + 22: optional set<Types.TPlanNodeId> planId_to_can_prune_tablets; } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
