Skip to content

Commit 3c9ff7a

Browse files
jackwenerjackwener
authored
[feature](Nereids): push down topN through join (#24720)
Push TopN through Join. JoinType just can be left/right outer join or cross join, because data of their one child can't be filtered. new TopN is (original limit + original offset, 0) as limit and offset.
1 parent 47694c5 commit 3c9ff7a

12 files changed

Lines changed: 312 additions & 69 deletions

File tree

‎fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@
7373
import org.apache.doris.nereids.rules.rewrite.InferPredicates;
7474
import org.apache.doris.nereids.rules.rewrite.InferSetOperatorDistinct;
7575
import org.apache.doris.nereids.rules.rewrite.LeadingJoin;
76+
import org.apache.doris.nereids.rules.rewrite.LimitSortToTopN;
7677
import org.apache.doris.nereids.rules.rewrite.MergeFilters;
7778
import org.apache.doris.nereids.rules.rewrite.MergeOneRowRelationIntoUnion;
7879
import org.apache.doris.nereids.rules.rewrite.MergeProjects;
@@ -90,9 +91,9 @@
9091
import org.apache.doris.nereids.rules.rewrite.PushProjectThroughUnion;
9192
import org.apache.doris.nereids.rules.rewrite.PushdownFilterThroughProject;
9293
import org.apache.doris.nereids.rules.rewrite.PushdownLimit;
94+
import org.apache.doris.nereids.rules.rewrite.PushdownTopNThroughJoin;
9395
import org.apache.doris.nereids.rules.rewrite.PushdownTopNThroughWindow;
9496
import org.apache.doris.nereids.rules.rewrite.ReorderJoin;
95-
import org.apache.doris.nereids.rules.rewrite.ReplaceLimitNode;
9697
import org.apache.doris.nereids.rules.rewrite.RewriteCteChildren;
9798
import org.apache.doris.nereids.rules.rewrite.SemiJoinCommute;
9899
import org.apache.doris.nereids.rules.rewrite.SimplifyAggGroupBy;
@@ -275,9 +276,10 @@ public class Rewriter extends AbstractBatchJobExecutor {
275276
// we should refactor like AggregateStrategies, e.g. LimitStrategies,
276277
// generate one PhysicalLimit if current distribution is gather or two
277278
// PhysicalLimits with gather exchange
278-
new ReplaceLimitNode(),
279+
new LimitSortToTopN(),
279280
new SplitLimit(),
280281
new PushdownLimit(),
282+
new PushdownTopNThroughJoin(),
281283
new PushdownTopNThroughWindow(),
282284
new CreatePartitionTopNFromWindow()
283285
)

‎fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -165,8 +165,6 @@ public enum RuleType {
165165
COLUMN_PRUNING(RuleTypeClass.REWRITE),
166166
ELIMINATE_SORT(RuleTypeClass.REWRITE),
167167

168-
PUSHDOWN_TOP_N_THROUGH_PROJECTION_WINDOW(RuleTypeClass.REWRITE),
169-
PUSHDOWN_TOP_N_THROUGH_WINDOW(RuleTypeClass.REWRITE),
170168
PUSHDOWN_MIN_MAX_THROUGH_JOIN(RuleTypeClass.REWRITE),
171169
PUSHDOWN_SUM_THROUGH_JOIN(RuleTypeClass.REWRITE),
172170
PUSHDOWN_COUNT_THROUGH_JOIN(RuleTypeClass.REWRITE),
@@ -248,7 +246,12 @@ public enum RuleType {
248246
PUSH_LIMIT_THROUGH_PROJECT_WINDOW(RuleTypeClass.REWRITE),
249247
PUSH_LIMIT_THROUGH_UNION(RuleTypeClass.REWRITE),
250248
PUSH_LIMIT_THROUGH_WINDOW(RuleTypeClass.REWRITE),
251-
PUSH_LIMIT_INTO_SORT(RuleTypeClass.REWRITE),
249+
LIMIT_SORT_TO_TOP_N(RuleTypeClass.REWRITE),
250+
// topN push down
251+
PUSH_TOP_N_THROUGH_JOIN(RuleTypeClass.REWRITE),
252+
PUSH_TOP_N_THROUGH_PROJECT_JOIN(RuleTypeClass.REWRITE),
253+
PUSH_TOP_N_THROUGH_PROJECT_WINDOW(RuleTypeClass.REWRITE),
254+
PUSH_TOP_N_THROUGH_WINDOW(RuleTypeClass.REWRITE),
252255
// adjust nullable
253256
ADJUST_NULLABLE(RuleTypeClass.REWRITE),
254257
ADJUST_CONJUNCTS_RETURN_TYPE(RuleTypeClass.REWRITE),

‎fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/EliminateLimit.java‎

Lines changed: 24 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -19,18 +19,35 @@
1919

2020
import org.apache.doris.nereids.rules.Rule;
2121
import org.apache.doris.nereids.rules.RuleType;
22+
import org.apache.doris.nereids.trees.UnaryNode;
23+
import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
2224
import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation;
2325

26+
import com.google.common.collect.ImmutableList;
27+
28+
import java.util.List;
29+
2430
/**
2531
* Eliminate limit = 0.
2632
*/
27-
public class EliminateLimit extends OneRewriteRuleFactory {
33+
public class EliminateLimit implements RewriteRuleFactory {
34+
2835
@Override
29-
public Rule build() {
30-
return logicalLimit()
31-
.when(limit -> limit.getLimit() == 0)
32-
.thenApply(ctx -> new LogicalEmptyRelation(ctx.statementContext.getNextRelationId(),
33-
ctx.root.getOutput()))
34-
.toRule(RuleType.ELIMINATE_LIMIT);
36+
public List<Rule> buildRules() {
37+
return ImmutableList.of(
38+
logicalLimit()
39+
.when(limit -> limit.getLimit() == 0)
40+
.thenApply(ctx -> new LogicalEmptyRelation(ctx.statementContext.getNextRelationId(),
41+
ctx.root.getOutput()))
42+
.toRule(RuleType.ELIMINATE_LIMIT),
43+
logicalLimit(logicalOneRowRelation())
44+
.then(limit -> limit.getLimit() > 0 && limit.getOffset() == 0
45+
? limit.child() : new LogicalEmptyRelation(StatementScopeIdGenerator.newRelationId(),
46+
limit.child().getOutput()))
47+
.toRule(RuleType.ELIMINATE_LIMIT_ON_ONE_ROW_RELATION),
48+
logicalLimit(logicalEmptyRelation())
49+
.then(UnaryNode::child)
50+
.toRule(RuleType.ELIMINATE_LIMIT_ON_EMPTY_RELATION)
51+
);
3552
}
3653
}

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/ReplaceLimitNode.java renamed to fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/LimitSortToTopN.java

Lines changed: 4 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,7 @@
1919

2020
import org.apache.doris.nereids.rules.Rule;
2121
import org.apache.doris.nereids.rules.RuleType;
22-
import org.apache.doris.nereids.trees.UnaryNode;
23-
import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
2422
import org.apache.doris.nereids.trees.plans.Plan;
25-
import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation;
2623
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
2724
import org.apache.doris.nereids.trees.plans.logical.LogicalSort;
2825
import org.apache.doris.nereids.trees.plans.logical.LogicalTopN;
@@ -35,7 +32,7 @@
3532
/**
3633
* rule to eliminate limit node by replace to other nodes.
3734
*/
38-
public class ReplaceLimitNode implements RewriteRuleFactory {
35+
public class LimitSortToTopN implements RewriteRuleFactory {
3936
@Override
4037
public List<Rule> buildRules() {
4138
return ImmutableList.of(
@@ -47,8 +44,8 @@ public List<Rule> buildRules() {
4744
limit.getLimit(),
4845
limit.getOffset(),
4946
sort.child(0));
50-
}).toRule(RuleType.PUSH_LIMIT_INTO_SORT),
51-
//limit->proj->sort ==> proj->topN
47+
}).toRule(RuleType.LIMIT_SORT_TO_TOP_N),
48+
// limit -> proj -> sort ==> proj -> topN
5249
logicalLimit(logicalProject(logicalSort()))
5350
.then(limit -> {
5451
LogicalProject<LogicalSort<Plan>> project = limit.child();
@@ -58,15 +55,7 @@ public List<Rule> buildRules() {
5855
limit.getOffset(),
5956
sort.child(0));
6057
return project.withChildren(Lists.newArrayList(topN));
61-
}).toRule(RuleType.PUSH_LIMIT_INTO_SORT),
62-
logicalLimit(logicalOneRowRelation())
63-
.then(limit -> limit.getLimit() > 0 && limit.getOffset() == 0
64-
? limit.child() : new LogicalEmptyRelation(StatementScopeIdGenerator.newRelationId(),
65-
limit.child().getOutput()))
66-
.toRule(RuleType.ELIMINATE_LIMIT_ON_ONE_ROW_RELATION),
67-
logicalLimit(logicalEmptyRelation())
68-
.then(UnaryNode::child)
69-
.toRule(RuleType.ELIMINATE_LIMIT_ON_EMPTY_RELATION)
58+
}).toRule(RuleType.LIMIT_SORT_TO_TOP_N)
7059
);
7160
}
7261
}

‎fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushdownFilterThroughProject.java‎

Lines changed: 19 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -41,27 +41,26 @@ public class PushdownFilterThroughProject implements RewriteRuleFactory {
4141
@Override
4242
public List<Rule> buildRules() {
4343
return ImmutableList.of(
44-
RuleType.PUSHDOWN_FILTER_THROUGH_PROJECT.build(logicalFilter(logicalProject())
45-
.whenNot(filter -> filter.child().getProjects().stream().anyMatch(
46-
expr -> expr.anyMatch(WindowExpression.class::isInstance)))
47-
.then(PushdownFilterThroughProject::pushdownFilterThroughProject)),
48-
// filter(project(limit)) will change to filter(limit(project)) by PushdownProjectThroughLimit,
49-
// then we should change filter(limit(project)) to project(filter(limit))
50-
RuleType.PUSHDOWN_FILTER_THROUGH_PROJECT_UNDER_LIMIT
51-
.build(logicalFilter(logicalLimit(logicalProject()))
52-
.whenNot(filter -> filter.child().child().getProjects().stream()
53-
.anyMatch(expr -> expr
54-
.anyMatch(WindowExpression.class::isInstance)))
55-
.then(filter -> {
56-
LogicalLimit<LogicalProject<Plan>> limit = filter.child();
57-
LogicalProject<Plan> project = limit.child();
44+
logicalFilter(logicalProject())
45+
.whenNot(filter -> filter.child().getProjects().stream().anyMatch(
46+
expr -> expr.anyMatch(WindowExpression.class::isInstance)))
47+
.then(PushdownFilterThroughProject::pushdownFilterThroughProject)
48+
.toRule(RuleType.PUSHDOWN_FILTER_THROUGH_PROJECT),
49+
// filter(project(limit)) will change to filter(limit(project)) by PushdownProjectThroughLimit,
50+
// then we should change filter(limit(project)) to project(filter(limit))
51+
logicalFilter(logicalLimit(logicalProject()))
52+
.whenNot(filter -> filter.child().child().getProjects().stream()
53+
.anyMatch(expr -> expr.anyMatch(WindowExpression.class::isInstance)))
54+
.then(filter -> {
55+
LogicalLimit<LogicalProject<Plan>> limit = filter.child();
56+
LogicalProject<Plan> project = limit.child();
5857

59-
return project.withProjectsAndChild(project.getProjects(),
60-
new LogicalFilter<>(
61-
ExpressionUtils.replace(filter.getConjuncts(),
62-
project.getAliasToProducer()),
63-
limit.withChildren(project.child())));
64-
}))
58+
return project.withProjectsAndChild(project.getProjects(),
59+
new LogicalFilter<>(
60+
ExpressionUtils.replace(filter.getConjuncts(),
61+
project.getAliasToProducer()),
62+
limit.withChildren(project.child())));
63+
}).toRule(RuleType.PUSHDOWN_FILTER_THROUGH_PROJECT_UNDER_LIMIT)
6564
);
6665
}
6766

‎fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushdownProjectThroughLimit.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
2525

2626
/**
27+
* <pre>
2728
* Before:
2829
* project
2930
* │
@@ -42,6 +43,7 @@
4243
* │
4344
* ▼
4445
* plan node
46+
* </pre>
4547
*/
4648
public class PushdownProjectThroughLimit extends OneRewriteRuleFactory {
4749

@@ -50,9 +52,7 @@ public Rule build() {
5052
return logicalProject(logicalLimit()).thenApply(ctx -> {
5153
LogicalProject<LogicalLimit<Plan>> logicalProject = ctx.root;
5254
LogicalLimit<Plan> logicalLimit = logicalProject.child();
53-
return new LogicalLimit<>(logicalLimit.getLimit(), logicalLimit.getOffset(),
54-
logicalLimit.getPhase(), logicalProject.withProjectsAndChild(logicalProject.getProjects(),
55-
logicalLimit.child()));
55+
return logicalLimit.withChildren(logicalProject.withChildren(logicalLimit.child()));
5656
}).toRule(RuleType.PUSHDOWN_PROJECT_THROUGH_LIMIT);
5757
}
5858
}

‎fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushdownTopNThroughJoin.java‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package org.apache.doris.nereids.rules.rewrite;
19+
20+
import org.apache.doris.nereids.properties.OrderKey;
21+
import org.apache.doris.nereids.rules.Rule;
22+
import org.apache.doris.nereids.rules.RuleType;
23+
import org.apache.doris.nereids.trees.expressions.Slot;
24+
import org.apache.doris.nereids.trees.plans.Plan;
25+
import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
26+
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
27+
import org.apache.doris.nereids.trees.plans.logical.LogicalTopN;
28+
import org.apache.doris.nereids.util.Utils;
29+
30+
import com.google.common.collect.ImmutableList;
31+
32+
import java.util.List;
33+
import java.util.Set;
34+
import java.util.stream.Collectors;
35+
36+
/**
37+
* Push down TopN through Outer Join into left child .....
38+
*/
39+
public class PushdownTopNThroughJoin implements RewriteRuleFactory {
40+
41+
@Override
42+
public List<Rule> buildRules() {
43+
return ImmutableList.of(
44+
// topN -> join
45+
logicalTopN(logicalJoin())
46+
// TODO: complex orderby
47+
.when(topN -> topN.getOrderKeys().stream().map(OrderKey::getExpr)
48+
.allMatch(Slot.class::isInstance))
49+
.then(topN -> {
50+
LogicalJoin<Plan, Plan> join = topN.child();
51+
Plan newJoin = pushLimitThroughJoin(topN, join);
52+
if (newJoin == null || topN.child().children().equals(newJoin.children())) {
53+
return null;
54+
}
55+
return topN.withChildren(newJoin);
56+
})
57+
.toRule(RuleType.PUSH_TOP_N_THROUGH_JOIN),
58+
59+
// topN -> project -> join
60+
logicalTopN(logicalProject(logicalJoin()).when(LogicalProject::isAllSlots))
61+
// TODO: complex project
62+
.when(topN -> topN.getOrderKeys().stream().map(OrderKey::getExpr)
63+
.allMatch(Slot.class::isInstance))
64+
.then(topN -> {
65+
LogicalProject<LogicalJoin<Plan, Plan>> project = topN.child();
66+
LogicalJoin<Plan, Plan> join = project.child();
67+
68+
Plan newJoin = pushLimitThroughJoin(topN, join);
69+
if (newJoin == null || join.children().equals(newJoin.children())) {
70+
return null;
71+
}
72+
return topN.withChildren(project.withChildren(newJoin));
73+
}).toRule(RuleType.PUSH_TOP_N_THROUGH_PROJECT_JOIN)
74+
);
75+
}
76+
77+
private Plan pushLimitThroughJoin(LogicalTopN<? extends Plan> topN, LogicalJoin<Plan, Plan> join) {
78+
switch (join.getJoinType()) {
79+
case LEFT_OUTER_JOIN:
80+
Set<Slot> rightOutputSet = join.right().getOutputSet();
81+
if (topN.getOrderKeys().stream().map(OrderKey::getExpr)
82+
.anyMatch(e -> Utils.isIntersecting(rightOutputSet, e.getInputSlots()))) {
83+
return null;
84+
}
85+
return join.withChildren(topN.withChildren(join.left()), join.right());
86+
case RIGHT_OUTER_JOIN:
87+
Set<Slot> leftOutputSet = join.left().getOutputSet();
88+
if (topN.getOrderKeys().stream().map(OrderKey::getExpr)
89+
.anyMatch(e -> Utils.isIntersecting(leftOutputSet, e.getInputSlots()))) {
90+
return null;
91+
}
92+
return join.withChildren(join.left(), topN.withChildren(join.right()));
93+
case CROSS_JOIN:
94+
List<Slot> orderbySlots = topN.getOrderKeys().stream().map(OrderKey::getExpr)
95+
.flatMap(e -> e.getInputSlots().stream()).collect(Collectors.toList());
96+
if (join.left().getOutputSet().containsAll(orderbySlots)) {
97+
return join.withChildren(topN.withChildren(join.left()), join.right());
98+
} else if (join.right().getOutputSet().containsAll(orderbySlots)) {
99+
return join.withChildren(join.left(), topN.withChildren(join.right()));
100+
} else {
101+
return null;
102+
}
103+
default:
104+
// don't push limit.
105+
return null;
106+
}
107+
}
108+
}

‎fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushdownTopNThroughWindow.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ public List<Rule> buildRules() {
5959
return topn;
6060
}
6161
return topn.withChildren(newWindow.get());
62-
}).toRule(RuleType.PUSHDOWN_TOP_N_THROUGH_WINDOW),
62+
}).toRule(RuleType.PUSH_TOP_N_THROUGH_WINDOW),
6363

6464
// topn -> projection -> window
6565
logicalTopN(logicalProject(logicalWindow())).then(topn -> {
@@ -79,7 +79,7 @@ public List<Rule> buildRules() {
7979
return topn;
8080
}
8181
return topn.withChildren(project.withChildren(newWindow.get()));
82-
}).toRule(RuleType.PUSHDOWN_TOP_N_THROUGH_PROJECTION_WINDOW)
82+
}).toRule(RuleType.PUSH_TOP_N_THROUGH_PROJECT_WINDOW)
8383
);
8484
}
8585

‎fe/fe-core/src/main/java/org/apache/doris/nereids/util/ExpressionUtils.java‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -351,10 +351,6 @@ public static List<Expression> mergeArguments(Object... arguments) {
351351
return builder.build();
352352
}
353353

354-
public static boolean isAllLiteral(Expression... children) {
355-
return Arrays.stream(children).allMatch(c -> c instanceof Literal);
356-
}
357-
358354
public static boolean isAllLiteral(List<Expression> children) {
359355
return children.stream().allMatch(c -> c instanceof Literal);
360356
}

0 commit comments

Comments
 (0)