This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new d9570d1ea9f branch-4.1 [fix](runtime filter) do not push runtime
filter below repeat when target slot is not in all grouping sets (#67992)
d9570d1ea9f is described below
commit d9570d1ea9f7744c08df6a74a188c8619aa37ac2
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]
