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
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 122cca6f5ab branch-4.2: [fix](cascades) Guard aggregate semi-join
transposition against unsafe expressions #67831 (#70022)
122cca6f5ab is described below
commit 122cca6f5abd0d8788ca77cd3171a49e032e2049
Author: morrySnow <[email protected]>
AuthorDate: Thu Nov 5 10:14:25 2026 +0800
branch-4.2: [fix](cascades) Guard aggregate semi-join transposition against
unsafe expressions #67831 (#70022)
picked from master #67831
Co-authored-by: feiniaofeiafei <[email protected]>
---
.../exploration/TransposeAggSemiJoinProject.java | 2 +-
.../rules/rewrite/TransposeSemiJoinAgg.java | 17 +++
.../rules/rewrite/TransposeSemiJoinAggProject.java | 2 +-
.../TransposeSemiJoinAggExpressionTest.java | 149 +++++++++++++++++++++
.../transposeSemiJoinAggExpression.out | 29 ++++
.../transposeSemiJoinAggExpression.groovy | 90 +++++++++++++
6 files changed, 287 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..9a8477e07ff
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/TransposeSemiJoinAggExpressionTest.java
@@ -0,0 +1,149 @@
+// 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.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+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());
+ Class<?> expectedRoot = kind == ExpressionKind.SAFE
+ ? (withProject ? LogicalProject.class :
LogicalAggregate.class) : LogicalJoin.class;
+ Assertions.assertEquals(expectedRoot,
checker.getPlan().getClass());
+ }
+ }
+}
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]
