This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.1 by this push:
new a632e7ae2b4 branch-2.1: [fix](Nereids) handle continuous filter or
project in plan #40176 (#54870)
a632e7ae2b4 is described below
commit a632e7ae2b46b9ea1c19920b4374ceb3186e3b81
Author: yujun <[email protected]>
AuthorDate: Fri Aug 22 10:56:34 2025 +0800
branch-2.1: [fix](Nereids) handle continuous filter or project in plan
#40176 (#54870)
cherry pick from #40176, #54872
---------
Co-authored-by: morrySnow <[email protected]>
---
.../glue/translator/PhysicalPlanTranslator.java | 42 +++++++++++++----
.../test_physical_translator.out | Bin 0 -> 823 bytes
.../test_physical_translator.groovy | 51 +++++++++++++++++++++
3 files changed, 83 insertions(+), 10 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
index 5ee52ca3b7e..f86bbb02c31 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
@@ -1260,6 +1260,12 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
MultiCastDataSink multiCastDataSink = (MultiCastDataSink)
inputFragment.getSink();
DataStreamSink dataStreamSink =
multiCastDataSink.getDataStreamSinks().get(
multiCastDataSink.getDataStreamSinks().size() - 1);
+ if (CollectionUtils.isNotEmpty(dataStreamSink.getConjuncts())
+ ||
CollectionUtils.isNotEmpty(dataStreamSink.getProjections())) {
+ String errMsg = "generate invalid plan \n" +
filter.treeString();
+ LOG.warn(errMsg);
+ throw new AnalysisException(errMsg);
+ }
filter.getConjuncts().stream()
.map(e -> ExpressionTranslator.translate(e, context))
.forEach(dataStreamSink::addConjunct);
@@ -1267,24 +1273,28 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
}
PlanNode planNode = inputFragment.getPlanRoot();
- Plan child = filter.child();
- while (child instanceof PhysicalLimit) {
- child = ((PhysicalLimit<?>) child).child();
- }
- if (planNode instanceof ExchangeNode || planNode instanceof SortNode
|| planNode instanceof UnionNode
- // this means we have filter->limit->project, need a SelectNode
- || child instanceof PhysicalProject) {
- // the three nodes don't support conjuncts, need create a
SelectNode to filter data
+ // the three nodes don't support conjuncts, need create a SelectNode
to filter data
+ if (planNode instanceof ExchangeNode || planNode instanceof SortNode
|| planNode instanceof UnionNode) {
SelectNode selectNode = new SelectNode(context.nextPlanNodeId(),
planNode);
selectNode.setNereidsId(filter.getId());
addConjunctsToPlanNode(filter, selectNode, context);
addPlanRoot(inputFragment, selectNode, filter);
} else {
if (!(filter.child(0) instanceof AbstractPhysicalJoin)) {
+ // already have filter on this node, we should not override
it, so need a new node
+ if (!planNode.getConjuncts().isEmpty()
+ // already have project on this node, filter need
execute after project, so need a new node
+ ||
CollectionUtils.isNotEmpty(planNode.getProjectList())
+ // already have limit on this node, filter need
execute after limit, so need a new node
+ || planNode.hasLimit()) {
+ planNode = new SelectNode(context.nextPlanNodeId(),
planNode);
+ planNode.setNereidsId(filter.getId());
+ addPlanRoot(inputFragment, planNode, filter);
+ }
addConjunctsToPlanNode(filter, planNode, context);
- updateLegacyPlanIdToPhysicalPlan(inputFragment.getPlanRoot(),
filter);
}
}
+ updateLegacyPlanIdToPhysicalPlan(inputFragment.getPlanRoot(), filter);
// in ut, filter.stats may be null
if (filter.getStats() != null) {
inputFragment.getPlanRoot().setCardinalityAfterFilter((long)
filter.getStats().getRowCount());
@@ -1893,8 +1903,15 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
}
PlanFragment inputFragment = project.child(0).accept(this, context);
-
PlanNode inputPlanNode = inputFragment.getPlanRoot();
+ // this means already have project on this node, filter need execute
after project, so need a new node
+ if (CollectionUtils.isNotEmpty(inputPlanNode.getProjectList())) {
+ SelectNode selectNode = new SelectNode(context.nextPlanNodeId(),
inputPlanNode);
+ selectNode.setNereidsId(project.getId());
+ addPlanRoot(inputFragment, selectNode, project);
+ inputPlanNode = selectNode;
+ }
+
List<Expr> projectionExprs = null;
List<Expr> allProjectionExprs = Lists.newArrayList();
List<Slot> slots = null;
@@ -1931,6 +1948,11 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
MultiCastDataSink multiCastDataSink = (MultiCastDataSink)
inputFragment.getSink();
DataStreamSink dataStreamSink =
multiCastDataSink.getDataStreamSinks().get(
multiCastDataSink.getDataStreamSinks().size() - 1);
+ if (CollectionUtils.isNotEmpty(dataStreamSink.getProjections())) {
+ String errMsg = "generate invalid plan \n" +
project.treeString();
+ LOG.warn(errMsg);
+ throw new AnalysisException(errMsg);
+ }
TupleDescriptor projectionTuple = generateTupleDesc(slots, null,
context);
dataStreamSink.setProjections(projectionExprs);
dataStreamSink.setOutputTupleDesc(projectionTuple);
diff --git
a/regression-test/data/nereids_p0/physical_translator/test_physical_translator.out
b/regression-test/data/nereids_p0/physical_translator/test_physical_translator.out
new file mode 100644
index 00000000000..d5166d53ce7
Binary files /dev/null and
b/regression-test/data/nereids_p0/physical_translator/test_physical_translator.out
differ
diff --git
a/regression-test/suites/nereids_p0/physical_translator/test_physical_translator.groovy
b/regression-test/suites/nereids_p0/physical_translator/test_physical_translator.groovy
new file mode 100644
index 00000000000..3e5001a5ea3
--- /dev/null
+++
b/regression-test/suites/nereids_p0/physical_translator/test_physical_translator.groovy
@@ -0,0 +1,51 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_physical_translator") {
+ def tbl = "tbl_test_physical_translator"
+ sql 'SET enable_nereids_planner=true'
+ sql 'SET runtime_filter_mode=OFF'
+ sql 'SET enable_fallback_to_original_planner=false'
+ sql "SET ignore_shape_nodes='PhysicalDistribute'"
+ sql "SET detail_shape_nodes='PhysicalProject'"
+ sql 'SET disable_nereids_rules=PRUNE_EMPTY_PARTITION'
+ sql "drop table if exists ${tbl} force"
+ sql "create table ${tbl} (a int, b int) properties('replication_num' =
'1')"
+ sql "insert into ${tbl} values(1, 10), (200, 300)"
+
+ def sql1 = """
+ SELECT a, x as k1, x as k2 FROM (SELECT a, random(100) as x FROM
${tbl}) t
+ """
+
+ explainAndOrderResult "continue_project", sql1
+ explain {
+ sql sql1
+ contains "VSELECT"
+ }
+
+ def sql2 = """
+ select * from (select a + 2 as x from ${tbl} where b > 10 limit 100)s
where x > 5 and x < 10000
+ """
+
+ explainAndOrderResult "continue_filter", sql2
+ explain {
+ sql sql2
+ contains "VSELECT"
+ }
+
+ sql "drop table if exists ${tbl} force"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
