This is an automated email from the ASF dual-hosted git repository. yiguolei pushed a commit to branch branch-4.2 in repository https://gitbox.apache.org/repos/asf/doris.git
commit b872a1b25fc344e62618222f93684513ca9385bd Author: minghong <[email protected]> AuthorDate: Tue Sep 15 18:44:18 2026 +0800 branch-4.1 [fix](runtime filter) do not push runtime filter below repeat when target slot is not in all grouping sets (#67992) Port the repeat protection of RuntimeFilterPushDownVisitor#visitPhysicalRepeat from #62383. Repeat erases a slot to NULL for the grouping sets which do not contain it, so pushing a runtime filter below the repeat would filter out rows which are still needed to compute the other grouping sets. Only push through repeat when the target slot appears in all grouping sets. branch-4.1 keeps two push down visitors, RuntimeFilterPushDownVisitor for join runtime filters and runtimefilterv2/PushDownVisitor for set operation runtime filters, so both get the same protection. Tests: RuntimeFilterTest testRuntimeFilterBlockByGroupingSetsPartialColumn testRuntimeFilterPushThroughGroupingSetsCommonColumn testSetOperationRuntimeFilterBlockByGroupingSetsPartialColumn testSetOperationRuntimeFilterPushThroughGroupingSetsCommonColumn ### What problem does this PR solve? Issue Number: close #xxx Related PR: #xxx Problem Summary: ### Release note None ### Check List (For Author) - Test <!-- At least one of them must be included. --> - [ ] Regression test - [ ] Unit Test - [ ] Manual test (add detailed scripts or steps below) - [ ] No need to test or manual test. Explain why: - [ ] This is a refactor/code format and no logic has been changed. - [ ] Previous test can cover this change. - [ ] No code files have been changed. - [ ] Other reason <!-- Add your reason? --> - Behavior changed: - [ ] No. - [ ] Yes. <!-- Explain the behavior change --> - Does this need documentation? - [ ] No. - [ ] Yes. <!-- Add document PR link here. eg: https://github.com/apache/doris-website/pull/1214 --> ### Check List (For Reviewer who merge this PR) - [ ] Confirm the release note - [ ] Confirm test cases - [ ] Confirm document - [ ] Add branch pick label <!-- Add branch pick label that this PR should merge into --> --- .../post/RuntimeFilterPushDownVisitor.java | 17 +++++ .../post/runtimefilterv2/PushDownVisitor.java | 17 +++++ .../nereids/postprocess/RuntimeFilterTest.java | 74 ++++++++++++++++++++++ 3 files changed, 108 insertions(+) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPushDownVisitor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPushDownVisitor.java index 0e3648df252..032894cf860 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPushDownVisitor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPushDownVisitor.java @@ -39,6 +39,7 @@ import org.apache.doris.nereids.trees.plans.physical.PhysicalLazyMaterializeOlap import org.apache.doris.nereids.trees.plans.physical.PhysicalNestedLoopJoin; import org.apache.doris.nereids.trees.plans.physical.PhysicalProject; import org.apache.doris.nereids.trees.plans.physical.PhysicalRelation; +import org.apache.doris.nereids.trees.plans.physical.PhysicalRepeat; import org.apache.doris.nereids.trees.plans.physical.PhysicalSchemaScan; import org.apache.doris.nereids.trees.plans.physical.PhysicalSetOperation; import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN; @@ -469,6 +470,22 @@ public class RuntimeFilterPushDownVisitor extends PlanVisitor<Boolean, PushDownC return pushedDown; } + @Override + public Boolean visitPhysicalRepeat(PhysicalRepeat<? extends Plan> repeat, PushDownContext ctx) { + if (!repeat.getOutputSet().containsAll(ctx.probeExpr.getInputSlots())) { + return false; + } + // Only push through Repeat if the probe slot appears in ALL grouping sets. + // A slot absent from a grouping set is erased to NULL for the rows of that group, + // so filtering on it before the repeat would incorrectly discard rows which are + // still needed to compute the other grouping sets. + Set<Expression> commonGroupingSetExpressions = repeat.getCommonGroupingSetExpressions(); + if (!commonGroupingSetExpressions.containsAll(ctx.probeExpr.getInputSlots())) { + return false; + } + return repeat.child().accept(this, ctx); + } + @Override public Boolean visitPhysicalTopN(PhysicalTopN<? extends Plan> topN, PushDownContext ctx) { return false; diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/runtimefilterv2/PushDownVisitor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/runtimefilterv2/PushDownVisitor.java index 2dbc6598718..29dbf51fc91 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/runtimefilterv2/PushDownVisitor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/runtimefilterv2/PushDownVisitor.java @@ -28,6 +28,7 @@ import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin; import org.apache.doris.nereids.trees.plans.physical.PhysicalNestedLoopJoin; import org.apache.doris.nereids.trees.plans.physical.PhysicalProject; import org.apache.doris.nereids.trees.plans.physical.PhysicalRelation; +import org.apache.doris.nereids.trees.plans.physical.PhysicalRepeat; import org.apache.doris.nereids.trees.plans.physical.PhysicalSetOperation; import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN; import org.apache.doris.nereids.trees.plans.physical.PhysicalWindow; @@ -171,6 +172,22 @@ public class PushDownVisitor extends PlanVisitor<Boolean, PushDownContext> { return pushed; } + @Override + public Boolean visitPhysicalRepeat(PhysicalRepeat<? extends Plan> repeat, PushDownContext ctx) { + if (!repeat.getOutputSet().containsAll(ctx.getTargetExpression().getInputSlots())) { + return false; + } + // Only push through Repeat if the target slot appears in ALL grouping sets. + // A slot absent from a grouping set is erased to NULL for the rows of that group, + // so filtering on it before the repeat would incorrectly discard rows which are + // still needed to compute the other grouping sets. + Set<Expression> commonGroupingSetExpressions = repeat.getCommonGroupingSetExpressions(); + if (!commonGroupingSetExpressions.containsAll(ctx.getTargetExpression().getInputSlots())) { + return false; + } + return repeat.child().accept(this, ctx); + } + @Override public Boolean visitPhysicalTopN(PhysicalTopN<? extends Plan> topN, PushDownContext ctx) { return false; diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java index 4122ceaabbe..bd8a8af9bc3 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java @@ -29,6 +29,7 @@ import org.apache.doris.nereids.parser.NereidsParser; import org.apache.doris.nereids.processor.post.PlanPostProcessors; import org.apache.doris.nereids.processor.post.RuntimeFilterContext; import org.apache.doris.nereids.processor.post.RuntimeFilterGenerator; +import org.apache.doris.nereids.processor.post.runtimefilterv2.RuntimeFilterV2; import org.apache.doris.nereids.properties.PhysicalProperties; import org.apache.doris.nereids.trees.expressions.Add; import org.apache.doris.nereids.trees.expressions.Alias; @@ -70,7 +71,9 @@ import org.junit.jupiter.api.Test; import org.mockito.Mockito; import java.util.ArrayList; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; @@ -361,6 +364,25 @@ public class RuntimeFilterTest extends SSBTestBase { return Optional.of(filters); } + /** + * Collect the runtime filters generated by the set operation push down process (runtime filter v2). + */ + private List<RuntimeFilterV2> getSetOperationRuntimeFilters(String sql) { + PlanChecker checker = PlanChecker.from(connectContext) + .analyze(sql) + .rewrite() + .optimize(); + PhysicalPlan plan = new PlanPostProcessors(checker.getCascadesContext()).process(checker.getBestPlanTree()); + Map<RuntimeFilterId, RuntimeFilterV2> filters = new LinkedHashMap<>(); + for (AbstractPhysicalPlan node : plan.<AbstractPhysicalPlan>collectToList( + p -> p instanceof AbstractPhysicalPlan)) { + for (RuntimeFilterV2 filter : node.getRuntimeFiltersV2()) { + filters.put(filter.getId(), filter); + } + } + return ImmutableList.copyOf(filters.values()); + } + private void checkRuntimeFilterExprs(List<RuntimeFilter> filters, List<Pair<String, String>> colNames) { Assertions.assertEquals(filters.size(), colNames.size()); for (RuntimeFilter filter : filters) { @@ -401,6 +423,58 @@ public class RuntimeFilterTest extends SSBTestBase { Assertions.assertEquals(0, filters.size()); } + @Test + public void testRuntimeFilterBlockByGroupingSetsPartialColumn() { + // RF on lo_custkey should be blocked because lo_custkey is NOT in all grouping sets. + // grouping sets ((lo_partkey), (lo_custkey, lo_partkey)) — the first set lacks lo_custkey, + // so lo_custkey is erased to NULL for the rows of that group. + // The subquery must be on the LEFT (probe) side so that the RF is pushed through the Repeat. + String sql = "SELECT lo_custkey FROM (" + + " SELECT lo_custkey, lo_partkey FROM lineorder" + + " GROUP BY GROUPING SETS ((lo_partkey), (lo_custkey, lo_partkey))" + + ") t INNER JOIN customer ON t.lo_custkey = c_custkey"; + List<RuntimeFilter> filters = getRuntimeFilters(sql).get(); + Assertions.assertEquals(0, filters.size(), + "RF should be blocked when probe slot is not in all grouping sets"); + } + + @Test + public void testRuntimeFilterPushThroughGroupingSetsCommonColumn() { + // RF on lo_partkey should push through because lo_partkey IS in all grouping sets. + String sql = "SELECT lo_partkey FROM (" + + " SELECT lo_custkey, lo_partkey FROM lineorder" + + " GROUP BY GROUPING SETS ((lo_partkey), (lo_custkey, lo_partkey))" + + ") t INNER JOIN part ON t.lo_partkey = p_partkey"; + List<RuntimeFilter> filters = getRuntimeFilters(sql).get(); + Assertions.assertEquals(1, filters.size(), + "RF should push through when probe slot is in all grouping sets"); + checkRuntimeFilterExprs(filters, ImmutableList.of(Pair.of("p_partkey", "lo_partkey"))); + } + + @Test + public void testSetOperationRuntimeFilterBlockByGroupingSetsPartialColumn() { + // SetOp RF should also be blocked when the target slot is not in all grouping sets. + String sql = "SELECT c_custkey FROM customer INTERSECT SELECT lo_custkey FROM (" + + " SELECT lo_custkey, lo_partkey FROM lineorder" + + " GROUP BY GROUPING SETS ((lo_partkey), (lo_custkey, lo_partkey))" + + ") t"; + List<RuntimeFilterV2> filters = getSetOperationRuntimeFilters(sql); + Assertions.assertEquals(0, filters.size(), + "SetOp RF should be blocked when the target slot is not in all grouping sets"); + } + + @Test + public void testSetOperationRuntimeFilterPushThroughGroupingSetsCommonColumn() { + // SetOp RF should still push through the Repeat for a slot that is common to all grouping sets. + String sql = "SELECT p_partkey FROM part INTERSECT SELECT lo_partkey FROM (" + + " SELECT lo_custkey, lo_partkey FROM lineorder" + + " GROUP BY GROUPING SETS ((lo_partkey), (lo_custkey, lo_partkey))" + + ") t"; + List<RuntimeFilterV2> filters = getSetOperationRuntimeFilters(sql); + Assertions.assertEquals(1, filters.size(), + "SetOp RF should push through when the target slot is in all grouping sets"); + } + @Test public void testNotGenerateRfOnDanglingSlot() { String sql = "select lo_custkey from lineorder union all select c_custkey from customer union all select p_partkey from part;"; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
