This is an automated email from the ASF dual-hosted git repository.
morrySnow 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 af525d567d5 [fix](rbo) Guard partial TopN pushdown through joins
(#67796)
af525d567d5 is described below
commit af525d567d5352fb8ee2bead6aab3dbc9b23984c
Author: morrySnow <[email protected]>
AuthorDate: Thu Sep 17 14:24:56 2026 +0800
[fix](rbo) Guard partial TopN pushdown through joins (#67796)
### What problem does this PR solve?
For a `SELECT DISTINCT` over an outer or cross join, a mixed `ORDER BY`
can begin with columns from one join child and continue with columns
from the other child. The TopN pushdown rule used to keep only that
child's leading order-key prefix and apply a hard `LIMIT` before the
join.
If the prefix contains ties, the remaining order keys decide which rows
belong to the final TopN. Applying the limit to an arbitrary subset of
tied child rows can therefore remove the true result before the join and
return a wrong row.
For example, with multiple left rows sharing the same `k`, this shape is
unsafe:
```sql
SELECT DISTINCT l.id, l.k, r.score
FROM left_table l LEFT JOIN right_table r ON l.id = r.id
ORDER BY l.k, r.score
LIMIT 1;
```
### What is changed and how does it work?
The rule now applies a partial order-key prefix only when that prefix is
provably unique after the child-side `DISTINCT`. The proof accepts a
prefix that:
- covers all output columns of the child;
- contains a non-null unique key; or
- functionally determines every remaining child output column.
An order-key sequence that comes entirely from one child remains
eligible for pushdown. A mixed sequence without a uniqueness proof is
kept above the join only.
The change adds focused unit coverage for rejected non-unique prefixes
and accepted complete, full-output, and functionally determining
prefixes. It also adds result regressions for tied prefixes, mixed-side
ordering, descending ordering, and offsets.
---
.../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 a67f06decd7..149dc125d8a 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;
@@ -33,6 +34,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;
@@ -114,7 +116,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(
childLimit, 0, pushedOrderKeys,
@@ -129,7 +131,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(
childLimit, 0, pushedOrderKeys,
@@ -142,14 +144,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(
childLimit, 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(
childLimit, 0, rightPushedOrderKeys,
@@ -170,8 +172,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
@@ -197,6 +200,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 1d7a6c9c516..208fadaa120 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]
