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 cb580c38d3d branch-4.1: [fix](rbo) Guard partial TopN pushdown through
joins #67796 (#68235)
cb580c38d3d is described below
commit cb580c38d3d5c0b16adf14ab217e0d0a26446333
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sun Sep 20 19:49:10 2026 +0800
branch-4.1: [fix](rbo) Guard partial TopN pushdown through joins #67796
(#68235)
Cherry-picked from #67796
Co-authored-by: morrySnow <[email protected]>
---
.../apache/doris/nereids/properties/DataTrait.java | 3 +-
.../doris/nereids/properties/FuncDepsDG.java | 53 ++++++++
.../rewrite/PushDownTopNDistinctThroughJoin.java | 42 +++++-
.../doris/nereids/properties/FuncDepsDGTest.java | 15 +++
.../PushDownTopNDistinctThroughJoinTest.java | 141 +++++++++++++++++++++
.../push_down_top_n_distinct_through_join.out | 7 +
.../push_down_top_n_distinct_through_join.groovy | 41 ++++++
7 files changed, 294 insertions(+), 8 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DataTrait.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DataTrait.java
index 4b9409f553a..9c3ef231f1d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DataTrait.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DataTrait.java
@@ -73,8 +73,7 @@ public class DataTrait {
}
public boolean isDependent(Set<Slot> dominate, Set<Slot> dependency) {
- return fdDg.findValidFuncDeps(Sets.union(dependency, dominate))
- .isFuncDeps(dominate, dependency);
+ return fdDg.isDependent(dominate, dependency);
}
public boolean isUnique(Slot slot) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/FuncDepsDG.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/FuncDepsDG.java
index 425273fda35..317bdeb34be 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/FuncDepsDG.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/FuncDepsDG.java
@@ -24,6 +24,7 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
@@ -97,6 +98,58 @@ public class FuncDepsDG {
return dgItems.isEmpty();
}
+ /**
+ * Checks whether the determinant closure contains every dependency slot.
+ *
+ * <p>Nodes are indexed by each slot that they are still waiting for. When
a dependency edge adds a slot to
+ * the closure, only nodes waiting for that slot are revisited. Each node
and edge is expanded at most once,
+ * avoiding construction of the graph's full transitive FD relation.</p>
+ */
+ public boolean isDependent(Set<Slot> determinants, Set<Slot> dependencies)
{
+ Set<Slot> closure = new HashSet<>(determinants);
+ if (closure.containsAll(dependencies)) {
+ return true;
+ }
+
+ int[] missingSlotCounts = new int[dgItems.size()];
+ Map<Slot, List<Integer>> waitingNodes = new HashMap<>();
+ ArrayDeque<Integer> readyNodes = new ArrayDeque<>();
+ for (DGItem item : dgItems) {
+ for (Slot slot : item.slots) {
+ if (!closure.contains(slot)) {
+ missingSlotCounts[item.index]++;
+ waitingNodes.computeIfAbsent(slot, key -> new
ArrayList<>()).add(item.index);
+ }
+ }
+ if (missingSlotCounts[item.index] == 0) {
+ readyNodes.add(item.index);
+ }
+ }
+
+ while (!readyNodes.isEmpty()) {
+ DGItem item = dgItems.get(readyNodes.remove());
+ for (int childIndex : item.children) {
+ for (Slot slot : dgItems.get(childIndex).slots) {
+ if (closure.add(slot)) {
+ List<Integer> nodes = waitingNodes.get(slot);
+ if (nodes != null) {
+ for (int nodeIndex : nodes) {
+ missingSlotCounts[nodeIndex]--;
+ if (missingSlotCounts[nodeIndex] == 0) {
+ readyNodes.add(nodeIndex);
+ }
+ }
+ }
+ }
+ }
+ if (closure.containsAll(dependencies)) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
/**
* Finds all functional dependencies that are applicable to a given set of
valid slots.
*/
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownTopNDistinctThroughJoin.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownTopNDistinctThroughJoin.java
index 4b0b8a89dc4..da1eb1eec60 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownTopNDistinctThroughJoin.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownTopNDistinctThroughJoin.java
@@ -17,6 +17,7 @@
package org.apache.doris.nereids.rules.rewrite;
+import org.apache.doris.nereids.properties.DataTrait;
import org.apache.doris.nereids.properties.OrderKey;
import org.apache.doris.nereids.rules.Rule;
import org.apache.doris.nereids.rules.RuleType;
@@ -32,6 +33,7 @@ import org.apache.doris.qe.ConnectContext;
import com.google.common.collect.ImmutableList;
+import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
@@ -104,7 +106,7 @@ public class PushDownTopNDistinctThroughJoin implements
RewriteRuleFactory {
return null;
}
List<OrderKey> pushedOrderKeys =
getPushedOrderKeys(groupBySlots,
- join.left().getOutputSet(), topN.getOrderKeys());
+ join.left(), topN.getOrderKeys());
if (!pushedOrderKeys.isEmpty()) {
LogicalTopN<Plan> left = topN.withLimitOrderKeyAndChild(
topN.getLimit() + topN.getOffset(), 0,
pushedOrderKeys,
@@ -119,7 +121,7 @@ public class PushDownTopNDistinctThroughJoin implements
RewriteRuleFactory {
return null;
}
List<OrderKey> pushedOrderKeys =
getPushedOrderKeys(groupBySlots,
- join.right().getOutputSet(), topN.getOrderKeys());
+ join.right(), topN.getOrderKeys());
if (!pushedOrderKeys.isEmpty()) {
LogicalTopN<Plan> right = topN.withLimitOrderKeyAndChild(
topN.getLimit() + topN.getOffset(), 0,
pushedOrderKeys,
@@ -132,14 +134,14 @@ public class PushDownTopNDistinctThroughJoin implements
RewriteRuleFactory {
Plan leftChild = join.left();
Plan rightChild = join.right();
List<OrderKey> leftPushedOrderKeys =
getPushedOrderKeys(groupBySlots,
- join.left().getOutputSet(), topN.getOrderKeys());
+ join.left(), topN.getOrderKeys());
if (!(join.left() instanceof TopN) &&
!leftPushedOrderKeys.isEmpty()) {
leftChild = topN.withLimitOrderKeyAndChild(
topN.getLimit() + topN.getOffset(), 0,
leftPushedOrderKeys,
PlanUtils.distinct(join.left()));
}
List<OrderKey> rightPushedOrderKeys =
getPushedOrderKeys(groupBySlots,
- join.right().getOutputSet(), topN.getOrderKeys());
+ join.right(), topN.getOrderKeys());
if (!(join.right() instanceof TopN) &&
!rightPushedOrderKeys.isEmpty()) {
rightChild = topN.withLimitOrderKeyAndChild(
topN.getLimit() + topN.getOffset(), 0,
rightPushedOrderKeys,
@@ -160,8 +162,9 @@ public class PushDownTopNDistinctThroughJoin implements
RewriteRuleFactory {
/**
* return pushed order-keys. If top-n distinct cannot be pushed, return
empty list.
*/
- private List<OrderKey> getPushedOrderKeys(Set<Slot> groupBySlots,
Set<Slot> joinChildSlot,
+ private List<OrderKey> getPushedOrderKeys(Set<Slot> groupBySlots, Plan
joinChild,
List<OrderKey> orderKeys) {
+ Set<Slot> joinChildSlot = joinChild.getOutputSet();
// NOTICE: Currently, we have implemented strict restrictions to
ensure that the distinct columns is
// a superset of the output from the corresponding child of the join
operator. In the future, we can relax
// this restriction and only require that there is overlap between
the output of the corresponding child of
@@ -187,6 +190,33 @@ public class PushDownTopNDistinctThroughJoin implements
RewriteRuleFactory {
notFound = true;
}
}
- return pushedOrderKeys.build();
+ List<OrderKey> pushedOrderKeyList = pushedOrderKeys.build();
+ if (pushedOrderKeyList.size() == orderKeys.size()
+ || isOrderKeyPrefixUniqueAfterDistinct(joinChild,
pushedOrderKeyList)) {
+ return pushedOrderKeyList;
+ }
+ return ImmutableList.of();
+ }
+
+ /**
+ * A partial order-key prefix is safe for a hard limit only when it
uniquely orders the rows produced by
+ * {@link PlanUtils#distinct(Plan)}. This is true when a leading part of
the prefix either is already a
+ * non-null unique key, covers every child output, or functionally
determines every remaining child output.
+ */
+ private boolean isOrderKeyPrefixUniqueAfterDistinct(Plan joinChild,
List<OrderKey> orderKeyPrefix) {
+ if (orderKeyPrefix.isEmpty()) {
+ return false;
+ }
+ Set<Slot> childOutput = joinChild.getOutputSet();
+ Set<Slot> prefixSlots = new HashSet<>();
+ for (OrderKey orderKey : orderKeyPrefix) {
+ prefixSlots.add((Slot) orderKey.getExpr());
+ }
+ if (prefixSlots.containsAll(childOutput)) {
+ return true;
+ }
+ DataTrait childTrait = joinChild.getLogicalProperties().getTrait();
+ return childTrait.isUniqueAndNotNull(prefixSlots)
+ || childTrait.isDependent(prefixSlots, childOutput);
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/properties/FuncDepsDGTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/properties/FuncDepsDGTest.java
index d504176e807..3ead4833f17 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/properties/FuncDepsDGTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/properties/FuncDepsDGTest.java
@@ -52,6 +52,21 @@ class FuncDepsDGTest {
Assertions.assertEquals(1, res.size());
}
+ @Test
+ void testDependencyClosure() {
+ FuncDepsDG.Builder dg = new FuncDepsDG.Builder();
+ Slot s1 = new SlotReference("s1", IntegerType.INSTANCE);
+ Slot s2 = new SlotReference("s2", IntegerType.INSTANCE);
+ Slot s3 = new SlotReference("s3", IntegerType.INSTANCE);
+ Slot s4 = new SlotReference("s4", IntegerType.INSTANCE);
+ dg.addDeps(Sets.newHashSet(s1), Sets.newHashSet(s2));
+ dg.addDeps(Sets.newHashSet(s2, s3), Sets.newHashSet(s4));
+
+ FuncDepsDG funcDeps = dg.build();
+ Assertions.assertTrue(funcDeps.isDependent(Sets.newHashSet(s1, s3),
Sets.newHashSet(s2, s4)));
+ Assertions.assertFalse(funcDeps.isDependent(Sets.newHashSet(s1),
Sets.newHashSet(s4)));
+ }
+
@Test
void testCircle() {
FuncDepsDG.Builder dg = new FuncDepsDG.Builder();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PushDownTopNDistinctThroughJoinTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PushDownTopNDistinctThroughJoinTest.java
new file mode 100644
index 00000000000..07b0ae8a679
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PushDownTopNDistinctThroughJoinTest.java
@@ -0,0 +1,141 @@
+// 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.catalog.KeysType;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.common.Pair;
+import org.apache.doris.nereids.trees.expressions.Add;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+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.util.LogicalPlanBuilder;
+import org.apache.doris.nereids.util.MemoPatternMatchSupported;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.nereids.util.PlanConstructor;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class PushDownTopNDistinctThroughJoinTest implements MemoPatternMatchSupported
{
+ private static final PushDownTopNDistinctThroughJoin RULE = new
PushDownTopNDistinctThroughJoin();
+ private static final LogicalOlapScan LEFT_SCAN =
PlanConstructor.newLogicalOlapScan(0, "t1", 0);
+ private static final LogicalOlapScan RIGHT_SCAN =
PlanConstructor.newLogicalOlapScan(1, "t2", 0);
+ private ConnectContext connectContext;
+
+ @BeforeEach
+ void setUp() {
+ connectContext = new ConnectContext();
+ connectContext.setThreadLocalInfo();
+ }
+
+ @AfterEach
+ void tearDown() {
+ ConnectContext.remove();
+ }
+
+ @Test
+ void pushDirectShapeWhenAccumulatedPrefixDeterminesChildOutput() {
+ NamedExpression id = LEFT_SCAN.getOutput().get(0);
+ LogicalPlan left = new LogicalPlanBuilder(LEFT_SCAN)
+ .projectExprs(ImmutableList.of(id,
LEFT_SCAN.getOutput().get(1),
+ new Alias(new Add(id, new IntegerLiteral(1)),
"id_plus_one")))
+ .build();
+ LogicalPlan plan = new LogicalPlanBuilder(left)
+ .join(RIGHT_SCAN, JoinType.LEFT_OUTER_JOIN, Pair.of(0, 0))
+ .distinct(ImmutableList.of(0, 1, 2, 3, 4))
+ .topN(10, 0, ImmutableList.of(0, 1, 3))
+ .build();
+
+ PlanChecker.from(connectContext, plan)
+ .applyTopDown(RULE)
+ .matchesFromRoot(
+ logicalTopN(
+ logicalAggregate(
+ logicalJoin(
+
logicalTopN(logicalAggregate(logicalProject(logicalOlapScan())))
+ .when(topN ->
topN.getLimit() == 10),
+ logicalOlapScan()
+ )
+ )
+ )
+ );
+ }
+
+ @Test
+ void pushAllSlotsProjectShapeForNonNullUniquePrefix() {
+ LogicalOlapScan uniqueScan = newUniqueScan(2, "unique_not_null",
false);
+ LogicalPlan join = new LogicalPlanBuilder(uniqueScan)
+ .join(RIGHT_SCAN, JoinType.LEFT_OUTER_JOIN, Pair.of(0, 0))
+ .build();
+ LogicalPlan plan = new LogicalPlanBuilder(join)
+
.projectExprs(ImmutableList.<NamedExpression>builder().addAll(join.getOutput()).build())
+ .distinct(ImmutableList.of(0, 1, 2, 3))
+ .topN(10, 0, ImmutableList.of(0, 2))
+ .build();
+
+ PlanChecker.from(connectContext, plan)
+ .applyTopDown(RULE)
+ .matchesFromRoot(
+ logicalTopN(
+ logicalProject(
+ logicalAggregate(
+ logicalJoin(
+
logicalTopN(logicalAggregate(logicalOlapScan()))
+ .when(topN ->
topN.getLimit() == 10),
+ logicalOlapScan()
+ )
+ )
+ )
+ )
+ );
+ }
+
+ @Test
+ void rejectNullableUniquePrefix() {
+ LogicalOlapScan nullableUniqueScan = newUniqueScan(3,
"unique_nullable", true);
+ LogicalPlan plan = new LogicalPlanBuilder(nullableUniqueScan)
+ .join(RIGHT_SCAN, JoinType.LEFT_OUTER_JOIN, Pair.of(0, 0))
+ .distinct(ImmutableList.of(0, 1, 2, 3))
+ .topN(10, 0, ImmutableList.of(0, 2))
+ .build();
+
+ PlanChecker.from(connectContext, plan)
+ .applyTopDown(RULE)
+ .matchesFromRoot(
+ logicalTopN(
+ logicalAggregate(
+ logicalJoin(logicalOlapScan(),
logicalOlapScan())
+ )
+ )
+ );
+ }
+
+ private LogicalOlapScan newUniqueScan(long tableId, String tableName,
boolean keyNullable) {
+ OlapTable table = PlanConstructor.newOlapTable(tableId, tableName, 0,
KeysType.UNIQUE_KEYS);
+ table.getFullSchema().get(0).setIsAllowNull(keyNullable);
+ table.getFullSchema().get(1).setIsKey(false);
+ return new LogicalOlapScan(PlanConstructor.getNextRelationId(), table,
ImmutableList.of("db"));
+ }
+}
diff --git
a/regression-test/data/nereids_rules_p0/push_down_top_n/push_down_top_n_distinct_through_join.out
b/regression-test/data/nereids_rules_p0/push_down_top_n/push_down_top_n_distinct_through_join.out
index c5f90edd139..96d3f10ae0d 100644
---
a/regression-test/data/nereids_rules_p0/push_down_top_n/push_down_top_n_distinct_through_join.out
+++
b/regression-test/data/nereids_rules_p0/push_down_top_n/push_down_top_n_distinct_through_join.out
@@ -45,3 +45,10 @@ PhysicalResultSink
6
7
+-- !partial_prefix_asc --
+1 0 10
+
+-- !partial_prefix_desc_offset --
+6 0 60
+7 0 70
+
diff --git
a/regression-test/suites/nereids_rules_p0/push_down_top_n/push_down_top_n_distinct_through_join.groovy
b/regression-test/suites/nereids_rules_p0/push_down_top_n/push_down_top_n_distinct_through_join.groovy
index 3f959c91fdc..e05dd203d86 100644
---
a/regression-test/suites/nereids_rules_p0/push_down_top_n/push_down_top_n_distinct_through_join.groovy
+++
b/regression-test/suites/nereids_rules_p0/push_down_top_n/push_down_top_n_distinct_through_join.groovy
@@ -67,4 +67,45 @@ suite("push_down_top_n_distinct_through_join") {
qt_push_down_topn_through_join_data """
select distinct * from (select t1.id from table_join t1 cross join
table_join t2) t order by id limit 10;
"""
+
+ sql "DROP TABLE IF EXISTS topn_distinct_left"
+ sql "DROP TABLE IF EXISTS topn_distinct_right"
+ sql """
+ CREATE TABLE topn_distinct_left (
+ k INT NOT NULL,
+ id INT NOT NULL
+ ) DUPLICATE KEY(k, id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES("replication_num" = "1")
+ """
+ sql """
+ CREATE TABLE topn_distinct_right (
+ id INT NOT NULL,
+ s INT NOT NULL
+ ) DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES("replication_num" = "1")
+ """
+ sql """
+ INSERT INTO topn_distinct_left VALUES
+ (0, 1), (0, 2), (0, 3), (0, 4), (0, 5), (0, 6), (0, 7), (0, 8)
+ """
+ sql """
+ INSERT INTO topn_distinct_right VALUES
+ (1, 10), (2, 20), (3, 30), (4, 40), (5, 50), (6, 60), (7, 70), (8,
80)
+ """
+
+ order_qt_partial_prefix_asc """
+ SELECT DISTINCT l.id, l.k, r.s
+ FROM topn_distinct_left l LEFT JOIN topn_distinct_right r ON l.id =
r.id
+ ORDER BY l.k ASC, r.s ASC
+ LIMIT 1
+ """
+
+ order_qt_partial_prefix_desc_offset """
+ SELECT DISTINCT l.id, l.k, r.s
+ FROM topn_distinct_left l LEFT JOIN topn_distinct_right r ON l.id =
r.id
+ ORDER BY l.k ASC, r.s DESC
+ LIMIT 2 OFFSET 1
+ """
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
