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 0083a3d285c branch-4.2: [improvement](fe) Prune partition keys from
partition topn sort #63379 (#68171)
0083a3d285c is described below
commit 0083a3d285cf43ec18083854fd74147e2f0d03f8
Author: Calvin Kirs <[email protected]>
AuthorDate: Sun Sep 20 11:26:49 2026 +0800
branch-4.2: [improvement](fe) Prune partition keys from partition topn sort
#63379 (#68171)
https://github.com/apache/doris/pull/63379
---------
Signed-off-by: Aurélien Pupier <[email protected]>
Co-authored-by: Aurélien Pupier <[email protected]>
---
.../properties/ChildOutputPropertyDeriver.java | 9 +++-
.../org/apache/doris/nereids/rules/RuleSet.java | 2 -
...ogicalPartitionTopNToPhysicalPartitionTopN.java | 39 +++------------
.../nereids/rules/rewrite/EliminateOrderByKey.java | 31 +++++++++++-
.../plans/physical/PhysicalPartitionTopN.java | 21 ++++++++
.../translator/PhysicalPlanTranslatorTest.java | 33 +++++++++++++
.../rules/implementation/ImplementationTest.java | 23 +++++++++
.../rules/rewrite/EliminateOrderByKeyTest.java | 57 ++++++++++++++++++++--
.../GeneratePartitionTopnFromWindowTest.java | 51 +++++++++++++++++++
.../shape_check/tpcds_sf100/rf_prune/query44.out | 12 ++---
.../data/shape_check/tpcds_sf100/shape/query44.out | 12 ++---
.../tpcds_sf1000/bs_downgrade_shape/query44.out | 12 ++---
.../shape_check/tpcds_sf1000/shape/query44.out | 12 ++---
.../tpcds_sf1000_constraints/shape/query44.out | 12 ++---
.../explain/test_global_partition_topn_plan.groovy | 55 +++++++++++++++++++++
15 files changed, 312 insertions(+), 69 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
index 8c89d45891b..e60fad74499 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
@@ -435,7 +435,14 @@ public class ChildOutputPropertyDeriver extends
PlanVisitor<PhysicalProperties,
Preconditions.checkState(childDistSpec instanceof
DistributionSpecHash,
"child dist spec is not hash spec");
- return new PhysicalProperties(childDistSpec, new
OrderSpec(partitionTopN.getOrderKeys()));
+ // Declare the delivered order via getOutputOrderKeys() (=
[partitionKeys, orderKeys]), NOT
+ // getOrderKeys() (= orderKeys only). getOrderKeys() is the
executable sort key list sent to BE, from
+ // which order keys that duplicate a partition key are pruned
(they are constant within each partition,
+ // so sorting on them is redundant). The two-phase-global output
order property, however, must remain
+ // the full [partitionKeys, orderKeys] -- the same order this node
declared before that pruning split.
+ // Keeping it full stays in lockstep with the parent window's
required order (RequestPropertyDeriver),
+ // so OrderSpec.satisfy passes and no redundant sort enforcer is
inserted above this PartitionTopN.
+ return new PhysicalProperties(childDistSpec, new
OrderSpec(partitionTopN.getOutputOrderKeys()));
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleSet.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleSet.java
index 908ce436541..0316b30d76f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleSet.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleSet.java
@@ -109,7 +109,6 @@ import
org.apache.doris.nereids.rules.implementation.LogicalWorkTableReferenceTo
import org.apache.doris.nereids.rules.implementation.SplitAggMultiPhase;
import
org.apache.doris.nereids.rules.implementation.SplitAggMultiPhaseWithoutGbyKey;
import org.apache.doris.nereids.rules.implementation.SplitAggWithoutDistinct;
-import org.apache.doris.nereids.rules.rewrite.CreatePartitionTopNFromWindow;
import org.apache.doris.nereids.rules.rewrite.EliminateFilter;
import org.apache.doris.nereids.rules.rewrite.EliminateOuterJoin;
import org.apache.doris.nereids.rules.rewrite.MaxMinFilterPushDown;
@@ -170,7 +169,6 @@ public class RuleSet {
public static final List<RuleFactory> PUSH_DOWN_FILTERS = ImmutableList.of(
new MaxMinFilterPushDown(),
- new CreatePartitionTopNFromWindow(),
new PushDownFilterThroughProject(),
new PushDownFilterThroughSort(),
new PushDownFilterThroughVectorSearchTopN(),
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalPartitionTopNToPhysicalPartitionTopN.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalPartitionTopNToPhysicalPartitionTopN.java
index 7af0ebca511..1e4a22a53f4 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalPartitionTopNToPhysicalPartitionTopN.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalPartitionTopNToPhysicalPartitionTopN.java
@@ -50,15 +50,14 @@ public class LogicalPartitionTopNToPhysicalPartitionTopN
extends OneImplementati
private List<PhysicalPartitionTopN<? extends Plan>>
generatePhysicalPartitionTopn(
LogicalPartitionTopN<? extends Plan> logicalPartitionTopN) {
+ List<OrderKey> orderKeys =
!logicalPartitionTopN.getOrderKeys().isEmpty()
+ ? logicalPartitionTopN.getOrderKeys().stream()
+ .map(OrderExpression::getOrderKey)
+ .collect(ImmutableList.toImmutableList()) :
+ ImmutableList.of();
if (logicalPartitionTopN.getPartitionKeys().isEmpty()
|| !checkTwoPhaseGlobalPartitionTopn(logicalPartitionTopN)) {
// if no partition by keys, use local partition topn combined with
further full sort
- List<OrderKey> orderKeys =
!logicalPartitionTopN.getOrderKeys().isEmpty()
- ? logicalPartitionTopN.getOrderKeys().stream()
- .map(OrderExpression::getOrderKey)
- .collect(ImmutableList.toImmutableList()) :
- ImmutableList.of();
-
PhysicalPartitionTopN<Plan> onePhaseLocalPartitionTopN = new
PhysicalPartitionTopN<>(
logicalPartitionTopN.getFunction(),
logicalPartitionTopN.getPartitionKeys(),
@@ -71,13 +70,10 @@ public class LogicalPartitionTopNToPhysicalPartitionTopN
extends OneImplementati
return ImmutableList.of(onePhaseLocalPartitionTopN);
} else {
- // if partition by keys exist, the order keys will be set as
original partition keys combined with
- // orderby keys, to meet upper window operator's order requirement.
- ImmutableList<OrderKey> fullOrderKeys =
getAllOrderKeys(logicalPartitionTopN);
PhysicalPartitionTopN<Plan> onePhaseGlobalPartitionTopN = new
PhysicalPartitionTopN<>(
logicalPartitionTopN.getFunction(),
logicalPartitionTopN.getPartitionKeys(),
- fullOrderKeys,
+ orderKeys,
logicalPartitionTopN.hasGlobalLimit(),
logicalPartitionTopN.getPartitionLimit(),
PartitionTopnPhase.ONE_PHASE_GLOBAL_PTOPN,
@@ -87,7 +83,7 @@ public class LogicalPartitionTopNToPhysicalPartitionTopN
extends OneImplementati
PhysicalPartitionTopN<Plan> twoPhaseLocalPartitionTopN = new
PhysicalPartitionTopN<>(
logicalPartitionTopN.getFunction(),
logicalPartitionTopN.getPartitionKeys(),
- fullOrderKeys,
+ orderKeys,
logicalPartitionTopN.hasGlobalLimit(),
logicalPartitionTopN.getPartitionLimit(),
PartitionTopnPhase.TWO_PHASE_LOCAL_PTOPN,
@@ -97,7 +93,7 @@ public class LogicalPartitionTopNToPhysicalPartitionTopN
extends OneImplementati
PhysicalPartitionTopN<Plan> twoPhaseGlobalPartitionTopN = new
PhysicalPartitionTopN<>(
logicalPartitionTopN.getFunction(),
logicalPartitionTopN.getPartitionKeys(),
- fullOrderKeys,
+ orderKeys,
logicalPartitionTopN.hasGlobalLimit(),
logicalPartitionTopN.getPartitionLimit(),
PartitionTopnPhase.TWO_PHASE_GLOBAL_PTOPN,
@@ -166,23 +162,4 @@ public class LogicalPartitionTopNToPhysicalPartitionTopN
extends OneImplementati
}
return true;
}
-
- private ImmutableList<OrderKey> getAllOrderKeys(LogicalPartitionTopN<?
extends Plan> logicalPartitionTopN) {
- ImmutableList.Builder<OrderKey> builder = ImmutableList.builder();
-
- if (!logicalPartitionTopN.getPartitionKeys().isEmpty()) {
-
builder.addAll(logicalPartitionTopN.getPartitionKeys().stream().map(partitionKey
-> {
- return new OrderKey(partitionKey, true, false);
- }).collect(ImmutableList.toImmutableList()));
- }
-
- if (!logicalPartitionTopN.getOrderKeys().isEmpty()) {
- builder.addAll(logicalPartitionTopN.getOrderKeys().stream()
- .map(OrderExpression::getOrderKey)
- .collect(ImmutableList.toImmutableList())
- );
- }
-
- return builder.build();
- }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/EliminateOrderByKey.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/EliminateOrderByKey.java
index ae3654b6e0c..762896a2408 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/EliminateOrderByKey.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/EliminateOrderByKey.java
@@ -89,7 +89,17 @@ public class EliminateOrderByKey implements
RewriteRuleFactory {
for (OrderExpression orderExpression : orderExpressions) {
orderKeys.add(orderExpression.getOrderKey());
}
- List<OrderKey> retainExpression = eliminate(dataTrait, orderKeys);
+ // an order key that repeats one of the window's own partition
keys is constant within each
+ // partition, so ordering by it is redundant and can be pruned (in
addition to the data-trait
+ // based elimination below).
+ Set<Expression> partitionKeyConstants =
ImmutableSet.copyOf(windowExpression.getPartitionKeys());
+ List<OrderKey> retainExpression = eliminate(dataTrait, orderKeys,
partitionKeyConstants);
+ // When every order key is redundant the retained list becomes
empty, which is fine: an order key
+ // is only pruned when it is constant within each partition
(duplicate / equal to a partition key /
+ // uniform / functionally constant), so it carried no
intra-partition ordering. Dropping it changes
+ // neither the framed result (for a RANGE frame a constant key
keeps the whole partition as one peer
+ // group, exactly like an empty ORDER BY) nor the partition topn
pushdown, and it lets partition topn
+ // drop the redundant partition key. This matches the existing
behavior introduced with this rule.
if (retainExpression.size() == orderKeys.size()) {
newNamedExpressions.add(expr);
continue;
@@ -117,6 +127,19 @@ public class EliminateOrderByKey implements
RewriteRuleFactory {
}
private static List<OrderKey> eliminate(DataTrait dataTrait,
List<OrderKey> inputOrderKeys) {
+ return eliminate(dataTrait, inputOrderKeys, ImmutableSet.of());
+ }
+
+ /**
+ * Eliminate redundant order keys.
+ *
+ * @param partitionKeyConstants exprs that are constant within the scope
of the order keys (e.g. a window's
+ * partition keys, which are constant inside each partition); an
order key equal to one of them is
+ * redundant. Empty for a plain sort. They are still recorded as
dominants so that order keys
+ * functionally dependent on them are eliminated too.
+ */
+ private static List<OrderKey> eliminate(DataTrait dataTrait,
List<OrderKey> inputOrderKeys,
+ Set<Expression> partitionKeyConstants) {
Set<Slot> validSlots = new HashSet<>();
for (OrderKey inputOrderKey : inputOrderKeys) {
Expression expr = inputOrderKey.getExpr();
@@ -137,6 +160,12 @@ public class EliminateOrderByKey implements
RewriteRuleFactory {
if (orderExprWithEqualSet.contains(expr)) {
continue;
}
+ // eliminate by partition key (constant within each partition for
window order keys)
+ if (partitionKeyConstants.contains(expr)) {
+ orderExprWithEqualSet.add(expr);
+ orderExprWithEqualSet.addAll(dataTrait.calEqualSet((Slot)
expr));
+ continue;
+ }
// eliminate by uniform
if (dataTrait.isUniformAndNotNull((Slot) expr)) {
orderExprWithEqualSet.add(expr);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPartitionTopN.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPartitionTopN.java
index 21eef99135b..3ac2b4bdaba 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPartitionTopN.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPartitionTopN.java
@@ -103,10 +103,31 @@ public class PhysicalPartitionTopN<CHILD_TYPE extends
Plan> extends PhysicalUnar
return partitionKeys;
}
+ /**
+ * Executable sort keys sent to BE (PartitionSortNode.SortInfo): only the
within-partition order keys.
+ * Order keys that duplicate a partition key are already pruned (constant
within a partition, so sorting on
+ * them is redundant). For the order this node declares to its parent, use
{@link #getOutputOrderKeys()}.
+ */
public List<OrderKey> getOrderKeys() {
return orderKeys;
}
+ /**
+ * Output order property declared to the parent (for
ChildOutputPropertyDeriver), NOT executed by BE:
+ * [partitionKeys, orderKeys]. Each partition key is wrapped as an
ascending, nulls-last OrderKey and
+ * prepended to the executable {@link #getOrderKeys()}. This is the full
order the (two-phase-global)
+ * PartitionTopN delivers, kept in lockstep with the parent window's
required order so OrderSpec.satisfy
+ * passes without a redundant sort enforcer, even though BE only sorts by
the pruned getOrderKeys().
+ */
+ public List<OrderKey> getOutputOrderKeys() {
+ return ImmutableList.<OrderKey>builder()
+ .addAll(partitionKeys.stream()
+ .map(partitionKey -> new OrderKey(partitionKey, true,
false))
+ .collect(ImmutableList.toImmutableList()))
+ .addAll(orderKeys)
+ .build();
+ }
+
@Override
public boolean hasGlobalLimit() {
return hasGlobalLimit;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorTest.java
index 981acf431d1..2a24bc5e6b3 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorTest.java
@@ -20,6 +20,7 @@ package org.apache.doris.nereids.glue.translator;
import org.apache.doris.analysis.Expr;
import org.apache.doris.analysis.GroupingInfo;
import org.apache.doris.analysis.SlotRef;
+import org.apache.doris.analysis.SortInfo;
import org.apache.doris.analysis.TupleDescriptor;
import org.apache.doris.analysis.TupleId;
import org.apache.doris.catalog.Column;
@@ -49,6 +50,7 @@ import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.nereids.util.PlanConstructor;
import org.apache.doris.planner.AggregationNode;
import org.apache.doris.planner.OlapScanNode;
+import org.apache.doris.planner.PartitionSortNode;
import org.apache.doris.planner.PlanFragment;
import org.apache.doris.planner.PlanNode;
import org.apache.doris.planner.PlanNodeId;
@@ -242,6 +244,37 @@ public class PhysicalPlanTranslatorTest extends
TestWithFeService {
});
}
+ @Test
+ public void testPartitionTopNPrunesPartitionKeyFromSortInfo() throws
Exception {
+ connectContext.getSessionVariable().setEnablePartitionTopN(true);
+ String sql = "select * from (select a, b, rank() over(partition by a
order by a, b) as rk "
+ + "from test_db.t) t where rk <= 1";
+ Planner planner = getSQLPlanner(sql);
+ Assertions.assertNotNull(planner);
+
+ List<PartitionSortNode> partitionSortNodes = new ArrayList<>();
+ for (PlanFragment fragment : planner.getFragments()) {
+ PlanNode root = fragment.getPlanRoot();
+ if (root != null) {
+ root.collect(PartitionSortNode.class, partitionSortNodes);
+ }
+ }
+ Assertions.assertFalse(partitionSortNodes.isEmpty());
+
+ Field sortInfoField = PartitionSortNode.class.getDeclaredField("info");
+ sortInfoField.setAccessible(true);
+ SortInfo sortInfo = (SortInfo)
sortInfoField.get(partitionSortNodes.get(0));
+ List<Expr> orderingExprs = sortInfo.getOrderingExprs();
+ Assertions.assertEquals(1, orderingExprs.size());
+ // The redundant partition key `a` must be pruned, leaving exactly the
slot `b`. Assert the slot
+ // identity directly: a substring check on the WITH_TABLE-rendered SQL
would false-pass because the
+ // table prefix `test_db` already contains the letter "b".
+ Expr orderingExpr = orderingExprs.get(0);
+ Assertions.assertInstanceOf(SlotRef.class, orderingExpr);
+ Assertions.assertEquals("b", ((SlotRef) orderingExpr).getColumnName());
+ Assertions.assertNotEquals("a", ((SlotRef)
orderingExpr).getColumnName());
+ }
+
private OlapScanNode getFirstOlapScanNode(String sql) throws Exception {
Planner planner = getSQLPlanner(sql);
Assertions.assertNotNull(planner);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/implementation/ImplementationTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/implementation/ImplementationTest.java
index 40b9609e58e..973681a8ed0 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/implementation/ImplementationTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/implementation/ImplementationTest.java
@@ -28,9 +28,11 @@ import
org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.literal.Literal;
import org.apache.doris.nereids.trees.plans.GroupPlan;
import org.apache.doris.nereids.trees.plans.LimitPhase;
+import org.apache.doris.nereids.trees.plans.PartitionTopnPhase;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.RelationId;
+import org.apache.doris.nereids.trees.plans.WindowFuncType;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
import org.apache.doris.nereids.trees.plans.logical.LogicalLimit;
@@ -41,6 +43,7 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.nereids.trees.plans.logical.LogicalSort;
import org.apache.doris.nereids.trees.plans.logical.LogicalTopN;
import org.apache.doris.nereids.trees.plans.physical.PhysicalLimit;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalPartitionTopN;
import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN;
@@ -136,4 +139,24 @@ public class ImplementationTest {
Assertions.assertEquals(limit, physicalLimit.getLimit());
Assertions.assertEquals(offset, physicalLimit.getOffset());
}
+
+ @Test
+ public void physicalPartitionTopNSeparatesSortKeysFromOutputOrderKeys() {
+ SlotReference partitionKey = new SlotReference("col1",
IntegerType.INSTANCE);
+ SlotReference orderExpr = new SlotReference("col2",
IntegerType.INSTANCE);
+ OrderKey orderKey = new OrderKey(orderExpr, true, true);
+ PhysicalPartitionTopN<GroupPlan> partitionTopN = new
PhysicalPartitionTopN<>(
+ WindowFuncType.ROW_NUMBER,
+ ImmutableList.of(partitionKey),
+ ImmutableList.of(orderKey),
+ false,
+ 10,
+ PartitionTopnPhase.TWO_PHASE_GLOBAL_PTOPN,
+ groupPlan.getLogicalProperties(),
+ groupPlan);
+
+ Assertions.assertEquals(ImmutableList.of(orderKey),
partitionTopN.getOrderKeys());
+ Assertions.assertEquals(ImmutableList.of(new OrderKey(partitionKey,
true, false), orderKey),
+ partitionTopN.getOutputOrderKeys());
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/EliminateOrderByKeyTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/EliminateOrderByKeyTest.java
index a54d65d9fcb..ac19d0228bc 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/EliminateOrderByKeyTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/EliminateOrderByKeyTest.java
@@ -176,28 +176,34 @@ public class EliminateOrderByKeyTest extends
TestWithFeService implements MemoPa
@Test
void testWindowDup() {
+ // partition by a order by a,a: order key a repeats the partition key
(constant within partition) and
+ // the second a is a duplicate, so every order key is redundant and
all are pruned to empty. A constant
+ // order key carries no intra-partition ordering, so emptying it is
correctness-neutral.
PlanChecker.from(connectContext)
.analyze("select sum(a) over (partition by a order by a,a)
from eliminate_order_by_constant_t")
.rewrite()
.printlnTree()
.matches(logicalWindow()
.when(window -> ((WindowExpression)
window.getWindowExpressions().get(0).child(0))
- .getOrderKeys().size() == 1));
+ .getOrderKeys().isEmpty()));
}
@Test
void testWindowFd() {
+ // leading order key a repeats the partition key, the rest are
functionally dependent on a; only b remains.
PlanChecker.from(connectContext)
.analyze("select sum(a) over (partition by a order by
a,a+1,abs(a),1-a,b) from eliminate_order_by_constant_t")
.rewrite()
.printlnTree()
.matches(logicalWindow()
.when(window -> ((WindowExpression)
window.getWindowExpressions().get(0).child(0))
- .getOrderKeys().size() == 2));
+ .getOrderKeys().size() == 1));
}
@Test
void testWindowUniform() {
+ // order by b with b=100 makes b uniform, so the only order key is
redundant and is pruned to empty.
+ // A constant order key carries no intra-partition ordering, so
emptying it is correctness-neutral.
PlanChecker.from(connectContext)
.analyze("select sum(a) over (partition by a order by b) from
eliminate_order_by_constant_t where b=100")
.rewrite()
@@ -207,6 +213,49 @@ public class EliminateOrderByKeyTest extends
TestWithFeService implements MemoPa
.getOrderKeys().isEmpty()));
}
+ @Test
+ void testWindowPartitionKey() {
+ // an order key that repeats the window's own partition key is
constant within each partition,
+ // so it is redundant and should be pruned, leaving only the remaining
order key(s).
+ PlanChecker.from(connectContext)
+ .analyze("select sum(a) over (partition by a order by a,b)
from eliminate_order_by_constant_t")
+ .rewrite()
+ .printlnTree()
+ .matches(logicalWindow()
+ .when(window -> ((WindowExpression)
window.getWindowExpressions().get(0).child(0))
+ .getOrderKeys().size() == 1));
+ }
+
+ @Test
+ void testWindowAggPrunedToEmpty() {
+ // A non-ranking aggregate window whose only order key repeats the
partition key (constant within the
+ // partition) is pruned to empty, just like a ranking window. A
constant order key defines no peer
+ // ordering: for a RANGE frame the whole partition is a single peer
group whether the constant key is
+ // present or not, and a ROWS frame is positional, so emptying it does
not change the result.
+ PlanChecker.from(connectContext)
+ .analyze("select sum(b) over (partition by a order by a rows
between unbounded preceding "
+ + "and current row) from
eliminate_order_by_constant_t")
+ .rewrite()
+ .printlnTree()
+ .matches(logicalWindow()
+ .when(window -> ((WindowExpression)
window.getWindowExpressions().get(0).child(0))
+ .getOrderKeys().isEmpty()));
+ }
+
+ @Test
+ void testWindowRowNumberPrunedToEmpty() {
+ // row_number() is a ranking function: it ignores the frame (even
though it is standardized to a
+ // default ROWS frame), so an order key equal to the partition key is
redundant and is pruned to
+ // empty. This keeps the pushed-down partition topn's SortInfo free of
the duplicate partition key.
+ PlanChecker.from(connectContext)
+ .analyze("select row_number() over (partition by a order by a)
from eliminate_order_by_constant_t")
+ .rewrite()
+ .printlnTree()
+ .matches(logicalWindow()
+ .when(window -> ((WindowExpression)
window.getWindowExpressions().get(0).child(0))
+ .getOrderKeys().isEmpty()));
+ }
+
@Test
void testWindowMulti() {
PlanChecker.from(connectContext)
@@ -216,7 +265,7 @@ public class EliminateOrderByKeyTest extends
TestWithFeService implements MemoPa
.printlnTree()
.matches(logicalWindow()
.when(window -> ((WindowExpression)
window.getWindowExpressions().get(0).child(0))
- .getOrderKeys().size() == 2
+ .getOrderKeys().size() == 1
&& ((WindowExpression)
window.getWindowExpressions().get(1).child(0))
.getOrderKeys().size() == 1));
}
@@ -231,7 +280,7 @@ public class EliminateOrderByKeyTest extends
TestWithFeService implements MemoPa
.printlnTree()
.matches(logicalWindow()
.when(window -> ((WindowExpression)
window.getWindowExpressions().get(0).child(0))
- .getOrderKeys().size() == 2
+ .getOrderKeys().size() == 1
&& ((WindowExpression)
window.getWindowExpressions().get(1).child(0))
.getOrderKeys().size() == 1));
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/GeneratePartitionTopnFromWindowTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/GeneratePartitionTopnFromWindowTest.java
index 9c3993d4d2f..427c977c45b 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/GeneratePartitionTopnFromWindowTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/GeneratePartitionTopnFromWindowTest.java
@@ -118,6 +118,57 @@ public class GeneratePartitionTopnFromWindowTest
implements MemoPatternMatchSupp
);
}
+ @Test
+ public void testPartitionTopNPrunesPartitionKeyFromLogicalOrderKeys() {
+ ConnectContext context = MemoTestUtils.createConnectContext();
+ context.getSessionVariable().setEnablePartitionTopN(true);
+ NamedExpression gender = scan.getOutput().get(1).toSlot();
+ NamedExpression age = scan.getOutput().get(3).toSlot();
+
+ List<Expression> partitionKeyList = ImmutableList.of(gender);
+ List<OrderExpression> orderKeyList = ImmutableList.of(
+ new OrderExpression(new OrderKey(gender, true, true)),
+ new OrderExpression(new OrderKey(age, true, true)));
+ List<OrderExpression> expectedOrderKeyList = ImmutableList.of(
+ new OrderExpression(new OrderKey(age, true, true)));
+ WindowFrame windowFrame = new
WindowFrame(WindowFrame.FrameUnitsType.ROWS,
+ WindowFrame.FrameBoundary.newPrecedingBoundary(),
+ WindowFrame.FrameBoundary.newCurrentRowBoundary());
+ WindowExpression window1 = new WindowExpression(new RowNumber(),
partitionKeyList, orderKeyList, windowFrame);
+ Alias windowAlias1 = new Alias(window1, window1.toSql());
+ List<NamedExpression> expressions = Lists.newArrayList(windowAlias1);
+ LogicalWindow<LogicalOlapScan> window = new
LogicalWindow<>(expressions, scan);
+ Expression filterPredicate = new
LessThanEqual(window.getOutput().get(4).toSlot(), Literal.of(100));
+
+ LogicalPlan plan = new LogicalPlanBuilder(window)
+ .filter(filterPredicate)
+ .project(ImmutableList.of(0))
+ .build();
+
+ // EliminateOrderByKey prunes the order key that repeats the partition
key (gender) from the window;
+ // CreatePartitionTopNFromWindow then generates a PartitionTopN that
inherits the pruned order keys.
+ PlanChecker.from(context, plan)
+ .applyTopDown(new EliminateOrderByKey())
+ .applyTopDown(new CreatePartitionTopNFromWindow())
+ .matches(
+ logicalProject(
+ logicalFilter(
+ logicalWindow(
+ logicalPartitionTopN(
+ logicalOlapScan()
+ ).when(logicalPartitionTopN ->
logicalPartitionTopN.getFunction()
+ == WindowFuncType.ROW_NUMBER
+ &&
logicalPartitionTopN.getPartitionKeys().equals(partitionKeyList)
+ &&
logicalPartitionTopN.getOrderKeys().equals(expectedOrderKeyList)
+ &&
!logicalPartitionTopN.hasGlobalLimit()
+ &&
logicalPartitionTopN.getPartitionLimit() == 100)
+ ).when(logicalWindow ->
logicalWindow.getActualWindowExpressions().get(0)
+
.getOrderKeys().equals(expectedOrderKeyList))
+ ).when(filter ->
filter.getConjuncts().equals(ImmutableSet.of(filterPredicate)))
+ )
+ );
+ }
+
@Test
public void testMultipleWindowsWithDifferentOrders() {
ConnectContext context = MemoTestUtils.createConnectContext();
diff --git a/regression-test/data/shape_check/tpcds_sf100/rf_prune/query44.out
b/regression-test/data/shape_check/tpcds_sf100/rf_prune/query44.out
index 0bac032032c..f101442dc5b 100644
--- a/regression-test/data/shape_check/tpcds_sf100/rf_prune/query44.out
+++ b/regression-test/data/shape_check/tpcds_sf100/rf_prune/query44.out
@@ -9,11 +9,11 @@ PhysicalResultSink
------------PhysicalProject
--------------hashJoin[INNER_JOIN broadcast] hashCondition=((asceding.rnk =
descending.rnk)) otherCondition=()
----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i2.i_item_sk
= descending.item_sk)) otherCondition=() build RFs:RF1 item_sk->[i_item_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i1.i_item_sk
= asceding.item_sk)) otherCondition=() build RFs:RF1 item_sk->[i_item_sk]
--------------------PhysicalProject
-----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i2.i_product_name)] apply RFs: RF1
+----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i1.i_product_name)] apply RFs: RF1
--------------------PhysicalProject
-----------------------filter((V21.rnk < 11))
+----------------------filter((V11.rnk < 11))
------------------------PhysicalWindow
--------------------------PhysicalQuickSort[MERGE_SORT]
----------------------------PhysicalDistribute[DistributionSpecGather]
@@ -39,11 +39,11 @@ PhysicalResultSink
------------------------------------------------------filter((store_sales.ss_store_sk
= 146) and ss_addr_sk IS NULL)
--------------------------------------------------------PhysicalOlapScan[store_sales]
----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i1.i_item_sk
= asceding.item_sk)) otherCondition=() build RFs:RF0 item_sk->[i_item_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i2.i_item_sk
= descending.item_sk)) otherCondition=() build RFs:RF0 item_sk->[i_item_sk]
--------------------PhysicalProject
-----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i1.i_product_name)] apply RFs: RF0
+----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i2.i_product_name)] apply RFs: RF0
--------------------PhysicalProject
-----------------------filter((V11.rnk < 11))
+----------------------filter((V21.rnk < 11))
------------------------PhysicalWindow
--------------------------PhysicalQuickSort[MERGE_SORT]
----------------------------PhysicalDistribute[DistributionSpecGather]
diff --git a/regression-test/data/shape_check/tpcds_sf100/shape/query44.out
b/regression-test/data/shape_check/tpcds_sf100/shape/query44.out
index 0bac032032c..f101442dc5b 100644
--- a/regression-test/data/shape_check/tpcds_sf100/shape/query44.out
+++ b/regression-test/data/shape_check/tpcds_sf100/shape/query44.out
@@ -9,11 +9,11 @@ PhysicalResultSink
------------PhysicalProject
--------------hashJoin[INNER_JOIN broadcast] hashCondition=((asceding.rnk =
descending.rnk)) otherCondition=()
----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i2.i_item_sk
= descending.item_sk)) otherCondition=() build RFs:RF1 item_sk->[i_item_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i1.i_item_sk
= asceding.item_sk)) otherCondition=() build RFs:RF1 item_sk->[i_item_sk]
--------------------PhysicalProject
-----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i2.i_product_name)] apply RFs: RF1
+----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i1.i_product_name)] apply RFs: RF1
--------------------PhysicalProject
-----------------------filter((V21.rnk < 11))
+----------------------filter((V11.rnk < 11))
------------------------PhysicalWindow
--------------------------PhysicalQuickSort[MERGE_SORT]
----------------------------PhysicalDistribute[DistributionSpecGather]
@@ -39,11 +39,11 @@ PhysicalResultSink
------------------------------------------------------filter((store_sales.ss_store_sk
= 146) and ss_addr_sk IS NULL)
--------------------------------------------------------PhysicalOlapScan[store_sales]
----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i1.i_item_sk
= asceding.item_sk)) otherCondition=() build RFs:RF0 item_sk->[i_item_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i2.i_item_sk
= descending.item_sk)) otherCondition=() build RFs:RF0 item_sk->[i_item_sk]
--------------------PhysicalProject
-----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i1.i_product_name)] apply RFs: RF0
+----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i2.i_product_name)] apply RFs: RF0
--------------------PhysicalProject
-----------------------filter((V11.rnk < 11))
+----------------------filter((V21.rnk < 11))
------------------------PhysicalWindow
--------------------------PhysicalQuickSort[MERGE_SORT]
----------------------------PhysicalDistribute[DistributionSpecGather]
diff --git
a/regression-test/data/shape_check/tpcds_sf1000/bs_downgrade_shape/query44.out
b/regression-test/data/shape_check/tpcds_sf1000/bs_downgrade_shape/query44.out
index 5d21002157f..ddd91fb608d 100644
---
a/regression-test/data/shape_check/tpcds_sf1000/bs_downgrade_shape/query44.out
+++
b/regression-test/data/shape_check/tpcds_sf1000/bs_downgrade_shape/query44.out
@@ -9,11 +9,11 @@ PhysicalResultSink
------------PhysicalProject
--------------hashJoin[INNER_JOIN broadcast] hashCondition=((asceding.rnk =
descending.rnk)) otherCondition=()
----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i2.i_item_sk
= descending.item_sk)) otherCondition=() build RFs:RF1 item_sk->[i_item_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i1.i_item_sk
= asceding.item_sk)) otherCondition=() build RFs:RF1 item_sk->[i_item_sk]
--------------------PhysicalProject
-----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i2.i_product_name)] apply RFs: RF1
+----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i1.i_product_name)] apply RFs: RF1
--------------------PhysicalProject
-----------------------filter((V21.rnk < 11))
+----------------------filter((V11.rnk < 11))
------------------------PhysicalWindow
--------------------------PhysicalQuickSort[MERGE_SORT]
----------------------------PhysicalDistribute[DistributionSpecGather]
@@ -39,11 +39,11 @@ PhysicalResultSink
------------------------------------------------------filter((store_sales.ss_store_sk
= 4) and ss_hdemo_sk IS NULL)
--------------------------------------------------------PhysicalOlapScan[store_sales]
----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i1.i_item_sk
= asceding.item_sk)) otherCondition=() build RFs:RF0 item_sk->[i_item_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i2.i_item_sk
= descending.item_sk)) otherCondition=() build RFs:RF0 item_sk->[i_item_sk]
--------------------PhysicalProject
-----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i1.i_product_name)] apply RFs: RF0
+----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i2.i_product_name)] apply RFs: RF0
--------------------PhysicalProject
-----------------------filter((V11.rnk < 11))
+----------------------filter((V21.rnk < 11))
------------------------PhysicalWindow
--------------------------PhysicalQuickSort[MERGE_SORT]
----------------------------PhysicalDistribute[DistributionSpecGather]
diff --git a/regression-test/data/shape_check/tpcds_sf1000/shape/query44.out
b/regression-test/data/shape_check/tpcds_sf1000/shape/query44.out
index 5d21002157f..ddd91fb608d 100644
--- a/regression-test/data/shape_check/tpcds_sf1000/shape/query44.out
+++ b/regression-test/data/shape_check/tpcds_sf1000/shape/query44.out
@@ -9,11 +9,11 @@ PhysicalResultSink
------------PhysicalProject
--------------hashJoin[INNER_JOIN broadcast] hashCondition=((asceding.rnk =
descending.rnk)) otherCondition=()
----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i2.i_item_sk
= descending.item_sk)) otherCondition=() build RFs:RF1 item_sk->[i_item_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i1.i_item_sk
= asceding.item_sk)) otherCondition=() build RFs:RF1 item_sk->[i_item_sk]
--------------------PhysicalProject
-----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i2.i_product_name)] apply RFs: RF1
+----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i1.i_product_name)] apply RFs: RF1
--------------------PhysicalProject
-----------------------filter((V21.rnk < 11))
+----------------------filter((V11.rnk < 11))
------------------------PhysicalWindow
--------------------------PhysicalQuickSort[MERGE_SORT]
----------------------------PhysicalDistribute[DistributionSpecGather]
@@ -39,11 +39,11 @@ PhysicalResultSink
------------------------------------------------------filter((store_sales.ss_store_sk
= 4) and ss_hdemo_sk IS NULL)
--------------------------------------------------------PhysicalOlapScan[store_sales]
----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i1.i_item_sk
= asceding.item_sk)) otherCondition=() build RFs:RF0 item_sk->[i_item_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i2.i_item_sk
= descending.item_sk)) otherCondition=() build RFs:RF0 item_sk->[i_item_sk]
--------------------PhysicalProject
-----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i1.i_product_name)] apply RFs: RF0
+----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i2.i_product_name)] apply RFs: RF0
--------------------PhysicalProject
-----------------------filter((V11.rnk < 11))
+----------------------filter((V21.rnk < 11))
------------------------PhysicalWindow
--------------------------PhysicalQuickSort[MERGE_SORT]
----------------------------PhysicalDistribute[DistributionSpecGather]
diff --git
a/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query44.out
b/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query44.out
index f14c3f32e7c..df9bdfd72de 100644
---
a/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query44.out
+++
b/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query44.out
@@ -9,11 +9,11 @@ PhysicalResultSink
------------PhysicalProject
--------------hashJoin[INNER_JOIN broadcast] hashCondition=((asceding.rnk =
descending.rnk)) otherCondition=()
----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i2.i_item_sk
= descending.item_sk)) otherCondition=() build RFs:RF1 item_sk->[i_item_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i1.i_item_sk
= asceding.item_sk)) otherCondition=() build RFs:RF1 item_sk->[i_item_sk]
--------------------PhysicalProject
-----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i2.i_product_name)] apply RFs: RF1
+----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i1.i_product_name)] apply RFs: RF1
--------------------PhysicalProject
-----------------------filter((V21.rnk < 11))
+----------------------filter((V11.rnk < 11))
------------------------PhysicalWindow
--------------------------PhysicalQuickSort[MERGE_SORT]
----------------------------PhysicalDistribute[DistributionSpecGather]
@@ -39,11 +39,11 @@ PhysicalResultSink
------------------------------------------------------filter((store_sales.ss_store_sk
= 4) and ss_hdemo_sk IS NULL)
--------------------------------------------------------PhysicalOlapScan[store_sales]
----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i1.i_item_sk
= asceding.item_sk)) otherCondition=() build RFs:RF0 item_sk->[i_item_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((i2.i_item_sk
= descending.item_sk)) otherCondition=() build RFs:RF0 item_sk->[i_item_sk]
--------------------PhysicalProject
-----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i1.i_product_name)] apply RFs: RF0
+----------------------PhysicalLazyMaterializeOlapScan[item
lazySlots:(i2.i_product_name)] apply RFs: RF0
--------------------PhysicalProject
-----------------------filter((V11.rnk < 11))
+----------------------filter((V21.rnk < 11))
------------------------PhysicalWindow
--------------------------PhysicalQuickSort[MERGE_SORT]
----------------------------PhysicalDistribute[DistributionSpecGather]
diff --git
a/regression-test/suites/nereids_p0/explain/test_global_partition_topn_plan.groovy
b/regression-test/suites/nereids_p0/explain/test_global_partition_topn_plan.groovy
index 3948bb8f288..6eaa71be477 100644
---
a/regression-test/suites/nereids_p0/explain/test_global_partition_topn_plan.groovy
+++
b/regression-test/suites/nereids_p0/explain/test_global_partition_topn_plan.groovy
@@ -74,4 +74,59 @@ suite("test_global_partition_topn_plan") {
contains"PhysicalPartitionTopN"
contains"PhysicalQuickSort"
}
+
+ explain {
+ sql """select c2, c3, rk from (
+ select c2, c3, rank() over (partition by c2 order by c2, c3)
as rk
+ from test_global_partition_topn_plan
+ ) tmp where rk <= 1"""
+ check { String explainStr ->
+ def lines = explainStr.readLines()
+ def partitionTopNBlocks = []
+ lines.eachWithIndex { line, index ->
+ if (line.contains("VPartitionTopN")) {
+ partitionTopNBlocks.add(lines.subList(index,
Math.min(index + 10, lines.size())).join("\n"))
+ }
+ }
+ assertTrue(!partitionTopNBlocks.isEmpty(), explainStr)
+ partitionTopNBlocks.each { block ->
+ assertTrue(block.find("partition by: c2\\[#\\d+\\]") != null,
block)
+ assertTrue(block.find("order by: c3\\[#\\d+\\] ASC") != null,
block)
+ assertTrue(block.find("order by: c2\\[#\\d+\\] ASC") == null,
block)
+ }
+ }
+ }
+
+ sql "SET global_partition_topn_threshold=2"
+ explain {
+ sql """shape plan select rn from (
+ select row_number() over (partition by c2 order by c2, c3) as
rn
+ from test_global_partition_topn_plan
+ ) tmp where rn <= 100"""
+ contains"PhysicalPartitionTopN"
+ notContains"PhysicalQuickSort"
+ }
+
+ explain {
+ sql """select * from (
+ select l.c2 as lc2, r.c2 as rc2,
+ row_number() over (partition by l.c2 order by r.c2) as
rn
+ from test_global_partition_topn_plan l
+ join test_global_partition_topn_plan r on l.c1 = r.c1
+ ) tmp where rn <= 1"""
+ check { String explainStr ->
+ def lines = explainStr.readLines()
+ def partitionTopNBlocks = []
+ lines.eachWithIndex { line, index ->
+ if (line.contains("VPartitionTopN")) {
+ partitionTopNBlocks.add(lines.subList(index,
Math.min(index + 10, lines.size())).join("\n"))
+ }
+ }
+ assertTrue(!partitionTopNBlocks.isEmpty(), explainStr)
+ partitionTopNBlocks.each { block ->
+ assertTrue(block.find("partition by: c2\\[#\\d+\\]") != null,
block)
+ assertTrue(block.find("order by: c2\\[#\\d+\\] ASC") != null,
block)
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
