This is an automated email from the ASF dual-hosted git repository.
924060929 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 563d6e3727c [fix](cascades) Guard aggregate semi-join transposition
against unsafe expressions (#67831)
563d6e3727c is described below
commit 563d6e3727c7da2f876f6c652e94e48eb2a2b2d7
Author: feiniaofeiafei <[email protected]>
AuthorDate: Wed Sep 16 10:21:17 2026 +0800
[fix](cascades) Guard aggregate semi-join transposition against unsafe
expressions (#67831)
### What problem does this PR solve?
Problem Summary:
Transposing a semi/anti join and an aggregate changes the rows or the
number of times expressions are evaluated. Moving
`Aggregate(Project(assert_true(...))(SemiJoin(t, s)))` below the join
can evaluate assertions on rows that the join should discard. Moving a
join with a volatile ON predicate below an aggregate can filter
individual input rows instead of whole groups and change aggregate
results.
Reject all four transpose rules when join conditions, aggregate group
keys or output expressions, or the intermediate project contain a
`NoneMovableFunction` or a volatile expression. Share the recursive
checks in `canTranspose` and preserve the existing join-type, mark-join
and grouping-key restrictions.
### Release note
Fix incorrect results or unexpected expression errors when semi/anti
joins and aggregates are reordered around volatile or non-movable
expressions.
---
.../exploration/TransposeAggSemiJoinProject.java | 2 +-
.../rules/rewrite/TransposeSemiJoinAgg.java | 17 +++
.../rules/rewrite/TransposeSemiJoinAggProject.java | 2 +-
.../TransposeSemiJoinAggExpressionTest.java | 145 +++++++++++++++++++++
.../transposeSemiJoinAggExpression.out | 29 +++++
.../transposeSemiJoinAggExpression.groovy | 90 +++++++++++++
6 files changed, 283 insertions(+), 2 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/TransposeAggSemiJoinProject.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/TransposeAggSemiJoinProject.java
index 4f63f13e991..459715d4a92 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/TransposeAggSemiJoinProject.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/TransposeAggSemiJoinProject.java
@@ -40,7 +40,7 @@ public class TransposeAggSemiJoinProject extends
OneExplorationRuleFactory {
.then(agg -> {
LogicalProject<LogicalJoin<GroupPlan, GroupPlan>> project
= agg.child();
LogicalJoin<GroupPlan, GroupPlan> join = project.child();
- if (!TransposeSemiJoinAgg.canTranspose(agg, join)) {
+ if (!TransposeSemiJoinAgg.canTranspose(agg, join,
project)) {
return null;
}
Plan newJoin =
join.withChildren(agg.withChildren(project.withChildren(join.left())),
join.right());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/TransposeSemiJoinAgg.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/TransposeSemiJoinAgg.java
index 4006d614ca8..1347deabac8 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/TransposeSemiJoinAgg.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/TransposeSemiJoinAgg.java
@@ -20,9 +20,11 @@ package org.apache.doris.nereids.rules.rewrite;
import org.apache.doris.nereids.rules.Rule;
import org.apache.doris.nereids.rules.RuleType;
import org.apache.doris.nereids.trees.expressions.Slot;
+import
org.apache.doris.nereids.trees.expressions.functions.NoneMovableFunction;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.qe.ConnectContext;
import java.util.Set;
@@ -51,6 +53,10 @@ public class TransposeSemiJoinAgg extends
OneRewriteRuleFactory {
*/
public static boolean canTranspose(LogicalAggregate<? extends Plan>
aggregate,
LogicalJoin<? extends Plan, ? extends Plan> join) {
+ // Transposition changes the rows or the number of times these
expressions are evaluated.
+ if (containsUnsafeExpression(join) ||
containsUnsafeExpression(aggregate)) {
+ return false;
+ }
Set<Slot> canPushDownSlots =
PushDownFilterThroughAggregation.getCanPushDownSlots(aggregate);
// avoid push down scalar agg.
if (canPushDownSlots.isEmpty()) {
@@ -59,4 +65,15 @@ public class TransposeSemiJoinAgg extends
OneRewriteRuleFactory {
Set<Slot> leftConditionSlot = join.getLeftConditionSlot();
return canPushDownSlots.containsAll(leftConditionSlot);
}
+
+ /** Check the intermediate project as well as the aggregate and join. */
+ public static boolean canTranspose(LogicalAggregate<? extends Plan>
aggregate,
+ LogicalJoin<? extends Plan, ? extends Plan> join, LogicalProject<?
extends Plan> project) {
+ return !containsUnsafeExpression(project) && canTranspose(aggregate,
join);
+ }
+
+ private static boolean containsUnsafeExpression(Plan plan) {
+ return plan.getExpressions().stream().anyMatch(expression ->
expression.containsVolatileExpression()
+ || expression.containsType(NoneMovableFunction.class));
+ }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/TransposeSemiJoinAggProject.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/TransposeSemiJoinAggProject.java
index 17ca8f71395..c0112c59c81 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/TransposeSemiJoinAggProject.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/TransposeSemiJoinAggProject.java
@@ -38,7 +38,7 @@ public class TransposeSemiJoinAggProject extends
OneRewriteRuleFactory {
.then(join -> {
LogicalProject<LogicalAggregate<Plan>> project =
join.left();
LogicalAggregate<Plan> aggregate = project.child();
- if (!TransposeSemiJoinAgg.canTranspose(aggregate, join)) {
+ if (!TransposeSemiJoinAgg.canTranspose(aggregate, join,
project)) {
return null;
}
Plan newPlan =
aggregate.withChildren(join.withChildren(aggregate.child(), join.right()));
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/TransposeSemiJoinAggExpressionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/TransposeSemiJoinAggExpressionTest.java
new file mode 100644
index 00000000000..ebdd6c5eeae
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/TransposeSemiJoinAggExpressionTest.java
@@ -0,0 +1,145 @@
+// 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.rules.rewrite;
+
+import org.apache.doris.nereids.rules.exploration.TransposeAggSemiJoin;
+import org.apache.doris.nereids.rules.exploration.TransposeAggSemiJoinProject;
+import org.apache.doris.nereids.trees.expressions.Add;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.GreaterThan;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.AssertTrue;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Random;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.VarcharLiteral;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.util.LogicalPlanBuilder;
+import org.apache.doris.nereids.util.MemoTestUtils;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.nereids.util.PlanConstructor;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Stream;
+
+class TransposeSemiJoinAggExpressionTest {
+ enum Location {
+ HASH_CONDITION, OTHER_CONDITION, PROJECT, GROUP_KEY, AGG_ARGUMENT
+ }
+
+ enum ExpressionKind {
+ SAFE, VOLATILE, NON_MOVABLE
+ }
+
+ static Stream<Arguments> cases() {
+ List<Arguments> cases = new ArrayList<>();
+ for (JoinType joinType : new JoinType[] {JoinType.LEFT_SEMI_JOIN,
JoinType.LEFT_ANTI_JOIN}) {
+ for (boolean eager : new boolean[] {true, false}) {
+ for (boolean withProject : new boolean[] {true, false}) {
+ for (Location location : Location.values()) {
+ if (location == Location.PROJECT && !withProject) {
+ continue;
+ }
+ for (ExpressionKind kind : ExpressionKind.values()) {
+ cases.add(Arguments.of(joinType, eager,
withProject, location, kind));
+ }
+ }
+ }
+ }
+ }
+ return cases.stream();
+ }
+
+ @ParameterizedTest(name = "{0}, eager={1}, project={2}, {3}, {4}")
+ @MethodSource("cases")
+ void expressionMovement(JoinType joinType, boolean eager, boolean
withProject,
+ Location location, ExpressionKind kind) {
+ LogicalOlapScan left = PlanConstructor.newLogicalOlapScan(0, "t1", 0);
+ LogicalOlapScan right = PlanConstructor.newLogicalOlapScan(1, "t2", 0);
+ Slot key = left.getOutput().get(0);
+ Slot value = left.getOutput().get(1);
+ Expression expression;
+ switch (kind) {
+ case VOLATILE:
+ expression = new Cast(new Random(), IntegerType.INSTANCE);
+ break;
+ case NON_MOVABLE:
+ expression = new Cast(new AssertTrue(new GreaterThan(key, new
IntegerLiteral(0)),
+ new VarcharLiteral("positive key")),
IntegerType.INSTANCE);
+ break;
+ default:
+ expression = new Add(key, new IntegerLiteral(1));
+ }
+ Expression hashLeft = location == Location.HASH_CONDITION ? new
Add(key, expression) : key;
+ List<Expression> hash = ImmutableList.of(new EqualTo(hashLeft,
right.getOutput().get(0)));
+ List<Expression> other = location == Location.OTHER_CONDITION
+ ? ImmutableList.of(new GreaterThan(expression, new
IntegerLiteral(0))) : ImmutableList.of();
+ List<Expression> groupKeys = location == Location.GROUP_KEY
+ ? ImmutableList.of(key, expression) : ImmutableList.of(key);
+ Alias projected = new Alias(expression, "checked");
+ Expression argument = location == Location.AGG_ARGUMENT ? expression :
value;
+ LogicalPlanBuilder builder = new LogicalPlanBuilder(left);
+ if (eager) {
+ builder = builder.join(right, joinType, hash, other);
+ if (withProject) {
+ List<NamedExpression> projects = new
ArrayList<>(left.getOutput());
+ if (location == Location.PROJECT && kind !=
ExpressionKind.SAFE) {
+ projects.add(projected);
+ argument = projected.toSlot();
+ }
+ builder = builder.projectExprs(projects);
+ }
+ }
+ builder = builder.agg(groupKeys, ImmutableList.of(key, new Alias(new
Sum(argument), "sum")));
+ if (!eager) {
+ if (withProject) {
+ List<NamedExpression> projects = new
ArrayList<>(builder.build().getOutput());
+ if (location == Location.PROJECT && kind !=
ExpressionKind.SAFE) {
+ projects.add(projected);
+ }
+ builder = builder.projectExprs(projects);
+ }
+ builder = builder.join(right, joinType, hash, other);
+ }
+ LogicalPlan plan = builder.build();
+ PlanChecker checker =
PlanChecker.from(MemoTestUtils.createConnectContext(), plan);
+ if (eager) {
+ checker.applyExploration(withProject ?
TransposeAggSemiJoinProject.INSTANCE.build()
+ : TransposeAggSemiJoin.INSTANCE.build());
+ Assertions.assertEquals(kind == ExpressionKind.SAFE ? 2 : 1,
checker.getAllPlan().size());
+ } else {
+ checker.applyTopDown(withProject ? new
TransposeSemiJoinAggProject() : new TransposeSemiJoinAgg());
+ Assertions.assertEquals(kind == ExpressionKind.SAFE,
+ !plan.treeString().equals(checker.getPlan().treeString()));
+ }
+ }
+}
diff --git
a/regression-test/data/nereids_rules_p0/transposeJoin/transposeSemiJoinAggExpression.out
b/regression-test/data/nereids_rules_p0/transposeJoin/transposeSemiJoinAggExpression.out
new file mode 100644
index 00000000000..fd103e508f6
--- /dev/null
+++
b/regression-test/data/nereids_rules_p0/transposeJoin/transposeSemiJoinAggExpression.out
@@ -0,0 +1,29 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !non_movable_project --
+1 64
+
+-- !non_movable_argument --
+1 64
+
+-- !volatile_semi_plan --
+PhysicalResultSink
+--hashJoin[LEFT_SEMI_JOIN broadcast] hashCondition=((d.k = s.k))
otherCondition=((random() < 0.5))
+----hashAgg[GLOBAL]
+------PhysicalProject
+--------PhysicalOlapScan[transpose_semi_agg_expression_t]
+----PhysicalOlapScan[transpose_semi_agg_expression_s(s)]
+
+-- !volatile_anti_plan --
+PhysicalResultSink
+--hashJoin[LEFT_ANTI_JOIN broadcast] hashCondition=((d.k = s.k))
otherCondition=((random() < 0.5))
+----hashAgg[GLOBAL]
+------PhysicalProject
+--------PhysicalOlapScan[transpose_semi_agg_expression_t]
+----PhysicalOlapScan[transpose_semi_agg_expression_s(s)]
+
+-- !volatile_semi_condition --
+0
+
+-- !volatile_anti_condition --
+0
+
diff --git
a/regression-test/suites/nereids_rules_p0/transposeJoin/transposeSemiJoinAggExpression.groovy
b/regression-test/suites/nereids_rules_p0/transposeJoin/transposeSemiJoinAggExpression.groovy
new file mode 100644
index 00000000000..e1ccbf8cd61
--- /dev/null
+++
b/regression-test/suites/nereids_rules_p0/transposeJoin/transposeSemiJoinAggExpression.groovy
@@ -0,0 +1,90 @@
+// 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.
+
+suite("transposeSemiJoinAggExpression") {
+ sql "set runtime_filter_mode=OFF"
+ sql "set disable_join_reorder=false"
+ sql "set enable_dphyp_optimizer=false"
+
+ sql "drop table if exists transpose_semi_agg_expression_t"
+ sql """
+ create table transpose_semi_agg_expression_t (k int not null, v int
not null)
+ duplicate key(k) distributed by hash(k) buckets 1
+ properties("replication_num"="1")
+ """
+ sql "drop table if exists transpose_semi_agg_expression_s"
+ sql """
+ create table transpose_semi_agg_expression_s (k int not null)
+ duplicate key(k) distributed by hash(k) buckets 1
+ properties("replication_num"="1")
+ """
+ sql "insert into transpose_semi_agg_expression_t select 1, 1 from
numbers(\"number\"=\"64\")"
+ sql "insert into transpose_semi_agg_expression_t values (2, -1)"
+ sql "insert into transpose_semi_agg_expression_s values (1)"
+ sql "analyze table transpose_semi_agg_expression_t with sync"
+ sql "analyze table transpose_semi_agg_expression_s with sync"
+
+ // The unmatched negative row must be removed before evaluating
assert_true.
+ order_qt_non_movable_project """
+ select k, sum(checked) from (
+ select t.k, cast(assert_true(t.v > 0, 'positive rows only') as
int) checked
+ from transpose_semi_agg_expression_t t
+ left semi join transpose_semi_agg_expression_s s on t.k = s.k
+ ) q group by k
+ """
+ order_qt_non_movable_argument """
+ select t.k, sum(cast(assert_true(t.v > 0, 'positive rows only') as
int))
+ from transpose_semi_agg_expression_t t
+ left semi join transpose_semi_agg_expression_s s on t.k = s.k
+ group by t.k
+ """
+
+ // Inspect the original query directly: the join must stay above
aggregation.
+ qt_volatile_semi_plan """
+ explain shape plan
+ select d.k, d.c from (
+ select k, count(*) c from transpose_semi_agg_expression_t group by
k
+ ) d left semi join transpose_semi_agg_expression_s s
+ on d.k = s.k and random() < 0.5
+ """
+ qt_volatile_anti_plan """
+ explain shape plan
+ select d.k, d.c from (
+ select k, count(*) c from transpose_semi_agg_expression_t group by
k
+ ) d left anti join transpose_semi_agg_expression_s s
+ on d.k = s.k and random() < 0.5
+ """
+
+ // A surviving group must retain all 64 rows, regardless of the random
predicate.
+ // Count only invalid results so the expected output is deterministic.
+ order_qt_volatile_semi_condition """
+ select count(*) from (
+ select d.k, d.c from (
+ select k, count(*) c from transpose_semi_agg_expression_t
group by k
+ ) d left semi join transpose_semi_agg_expression_s s
+ on d.k = s.k and random() < 0.5
+ ) q where k = 1 and c <> 64
+ """
+ order_qt_volatile_anti_condition """
+ select count(*) from (
+ select d.k, d.c from (
+ select k, count(*) c from transpose_semi_agg_expression_t
group by k
+ ) d left anti join transpose_semi_agg_expression_s s
+ on d.k = s.k and random() < 0.5
+ ) q where k = 1 and c <> 64
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
