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 395384e40bd branch-4.2: [fix](fe) Deduplicate recursive CTE fragment
reset requests #67738 (#70038)
395384e40bd is described below
commit 395384e40bd8d18108702af4d031ea68440bbf20
Author: morrySnow <[email protected]>
AuthorDate: Tue Nov 3 17:20:53 2026 +0800
branch-4.2: [fix](fe) Deduplicate recursive CTE fragment reset requests
#67738 (#70038)
picked from master #67738
Co-authored-by: Mryange <[email protected]>
---
.../doris/nereids/rules/rewrite/CTEInline.java | 23 +++++
.../doris/qe/runtime/ThriftPlansBuilder.java | 6 ++
.../doris/nereids/rules/rewrite/CTEInlineTest.java | 39 ++++++++
.../data/recursive_cte/shared_cte_reset_test.out | 8 ++
.../recursive_cte/shared_cte_reset_test.groovy | 108 +++++++++++++++++++++
5 files changed, 184 insertions(+)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/CTEInline.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/CTEInline.java
index 9983c1062da..5ebb4ad0ab6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/CTEInline.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/CTEInline.java
@@ -57,6 +57,9 @@ public class CTEInline extends
DefaultPlanRewriter<LogicalCTEProducer<?>> implem
@Override
public Plan rewriteRoot(Plan plan, JobContext jobContext) {
mustInlineCTEs =
jobContext.getCascadesContext().getStatementContext().getMustInlineCTEs();
+ if (!mustInlineCTEs.isEmpty()) {
+ collectRecursiveCteDependencies(plan);
+ }
Plan root = plan.accept(this, null);
// collect cte id to consumer
@@ -68,6 +71,26 @@ public class CTEInline extends
DefaultPlanRewriter<LogicalCTEProducer<?>> implem
return root;
}
+ private void collectRecursiveCteDependencies(Plan plan) {
+ // Resolve the transitive dependencies before making any
materialization decisions.
+ // Otherwise an outer producer can remain shared by independent
recursive controllers
+ // even when its consumers are inside CTEs that must be inlined.
+ List<LogicalCTEProducer<?>> producers = plan.collectToList(p -> p
instanceof LogicalCTEProducer);
+ boolean changed;
+ do {
+ changed = false;
+ for (LogicalCTEProducer<?> producer : producers) {
+ if (mustInlineCTEs.contains(producer.getCteId())) {
+ List<LogicalCTEConsumer> consumers = producer.child()
+ .collectToList(p -> p instanceof
LogicalCTEConsumer);
+ for (LogicalCTEConsumer consumer : consumers) {
+ changed |= mustInlineCTEs.add(consumer.getCteId());
+ }
+ }
+ }
+ } while (changed);
+ }
+
@Override
public Plan visitLogicalCTEAnchor(LogicalCTEAnchor<? extends Plan, ?
extends Plan> cteAnchor,
LogicalCTEProducer<?> producer) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/ThriftPlansBuilder.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/ThriftPlansBuilder.java
index 5b3d8b24f6d..01401934586 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/ThriftPlansBuilder.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/ThriftPlansBuilder.java
@@ -820,6 +820,7 @@ public class ThriftPlansBuilder {
List<TRecCTETarget> targets = new ArrayList<>();
// reset infos for all instances of child fragments (used to
reset state)
List<TRecCTEResetInfo> fragmentsToReset = new ArrayList<>();
+ Set<String> resetFragmentKeys = new HashSet<>();
// The recursive side is under the right child; collect all
fragments
List<PlanFragment> childFragments = new ArrayList<>();
recursiveCteNode.getChild(1).getChild(0).getFragment().collectAll(PlanFragment.class::isInstance,
@@ -843,6 +844,11 @@ public class ThriftPlansBuilder {
String.format("can't find TNetworkAddress for
fragment %d", childFragmentId));
}
for (TNetworkAddress address : tNetworkAddresses) {
+ String resetFragmentKey = childFragmentId.asInt() + "@"
+ + address.getHostname() + ":" +
address.getPort();
+ if (!resetFragmentKeys.add(resetFragmentKey)) {
+ continue;
+ }
TRecCTEResetInfo tRecCTEResetInfo = new
TRecCTEResetInfo();
tRecCTEResetInfo.setFragmentId(childFragmentId.asInt());
tRecCTEResetInfo.setAddr(address);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/CTEInlineTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/CTEInlineTest.java
index 1acebe8e3f8..232da8467d3 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/CTEInlineTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/CTEInlineTest.java
@@ -34,6 +34,7 @@ import org.apache.doris.utframe.TestWithFeService;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -113,6 +114,44 @@ public class CTEInlineTest extends TestWithFeService
implements MemoPatternMatch
);
}
+ @Test
+ public void inlineTransitiveRecursiveDependencies() {
+ boolean oldEnableCteMaterialize =
connectContext.getSessionVariable().enableCTEMaterialize;
+ int oldCteInlineMode =
connectContext.getSessionVariable().cteInlineMode;
+ int oldInlineCteReferencedThreshold =
connectContext.getSessionVariable().inlineCTEReferencedThreshold;
+ connectContext.getSessionVariable().enableCTEMaterialize = true;
+ connectContext.getSessionVariable().cteInlineMode = 0;
+ connectContext.getSessionVariable().inlineCTEReferencedThreshold = 1;
+ try {
+ for (String input : new String[] {"base", "middle"}) {
+ String sql = "with recursive "
+ + "base as (select 1 as src, 2 as dst union all select
2, 3), "
+ + "middle as (select * from base union all select *
from base), "
+ + "edges as (select * from " + input + " union all
select * from " + input + "), "
+ + "ordinary as (select id from cte_inline_tbl), "
+ + "r1(n) as (select 1 union all "
+ + "select e.dst from r1 r join edges e on r.n =
e.src), "
+ + "r2(n) as (select 2 union all "
+ + "select e.dst from r2 r join edges e on r.n = e.src)
"
+ + "select r1.n, r2.n from r1 join r2 on r1.n = r2.n "
+ + "join ordinary a on a.id = r1.n join ordinary b on
b.id = r2.n";
+ LogicalPlan unboundPlan = new NereidsParser().parseSingle(sql);
+ NereidsPlanner planner = new NereidsPlanner(new
StatementContext(connectContext,
+ new OriginStatement(sql, 0)));
+ planner.planWithLock(unboundPlan, PhysicalProperties.ANY,
+ ExplainCommand.ExplainLevel.REWRITTEN_PLAN);
+ List<LogicalCTEConsumer> consumers = planner.getRewrittenPlan()
+ .collectToList(p -> p instanceof LogicalCTEConsumer);
+ Assertions.assertEquals(2, consumers.size());
+ Assertions.assertTrue(consumers.stream().allMatch(c ->
c.getName().equals("ordinary")));
+ }
+ } finally {
+ connectContext.getSessionVariable().enableCTEMaterialize =
oldEnableCteMaterialize;
+ connectContext.getSessionVariable().cteInlineMode =
oldCteInlineMode;
+ connectContext.getSessionVariable().inlineCTEReferencedThreshold =
oldInlineCteReferencedThreshold;
+ }
+ }
+
@Test
public void refreshCteConsumersAfterNormalizeEliminatesEmptyBranch() {
int oldCteInlineMode =
connectContext.getSessionVariable().cteInlineMode;
diff --git a/regression-test/data/recursive_cte/shared_cte_reset_test.out
b/regression-test/data/recursive_cte/shared_cte_reset_test.out
new file mode 100644
index 00000000000..b62c44c996c
--- /dev/null
+++ b/regression-test/data/recursive_cte/shared_cte_reset_test.out
@@ -0,0 +1,8 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !shared_cte_reset --
+1 MP:a
+2 MP:a
+3 MP:b
+a MP:a
+b MP:b
+
diff --git a/regression-test/suites/recursive_cte/shared_cte_reset_test.groovy
b/regression-test/suites/recursive_cte/shared_cte_reset_test.groovy
new file mode 100644
index 00000000000..05980941140
--- /dev/null
+++ b/regression-test/suites/recursive_cte/shared_cte_reset_test.groovy
@@ -0,0 +1,108 @@
+// 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("shared_cte_reset_test", "rec_cte") {
+ qt_shared_cte_reset """
+ WITH RECURSIVE
+ org_mp_bridge AS (
+ SELECT '1' AS organization_id, 'a' AS marketplace_account_id
+ UNION ALL SELECT '2', 'a'
+ UNION ALL SELECT '3', 'b'
+ ),
+ edges AS (
+ SELECT CONCAT('ORG:', organization_id) AS src,
+ CONCAT('MP:', marketplace_account_id) AS dst
+ FROM org_mp_bridge
+ UNION
+ SELECT CONCAT('MP:', marketplace_account_id),
+ CONCAT('ORG:', organization_id)
+ FROM org_mp_bridge
+ ),
+ nodes AS (
+ SELECT src AS node FROM edges
+ UNION
+ SELECT dst FROM edges
+ ),
+ reach (start_node, node) AS (
+ SELECT node, node FROM nodes
+ UNION
+ SELECT r.start_node, e.dst
+ FROM reach r
+ JOIN edges e ON r.node = e.src
+ ),
+ node_group AS (
+ SELECT node, MIN(start_node) AS reconciliation_group_id
+ FROM reach
+ GROUP BY node
+ ),
+ org_group AS (
+ SELECT DISTINCT b.organization_id, g.reconciliation_group_id
+ FROM org_mp_bridge b
+ JOIN node_group g ON g.node = CONCAT('ORG:', b.organization_id)
+ ),
+ mp_group AS (
+ SELECT DISTINCT b.marketplace_account_id, g.reconciliation_group_id
+ FROM org_mp_bridge b
+ JOIN node_group g ON g.node = CONCAT('MP:',
b.marketplace_account_id)
+ )
+ SELECT organization_id, reconciliation_group_id FROM org_group
+ UNION ALL
+ SELECT marketplace_account_id, reconciliation_group_id FROM mp_group
+ ORDER BY organization_id, reconciliation_group_id
+ """
+
+ // Each recursive controller must own its transitive ordinary CTE
dependencies.
+ // Keep materialization enabled so this also checks that the planner
enforces inlining.
+ sql "set enable_cte_materialize = true"
+ sql "set inline_cte_referenced_threshold = 1"
+ for (def input : ["base", "middle"]) {
+ def query = """
+ WITH RECURSIVE
+ base AS (
+ SELECT 1 AS src, 2 AS dst
+ UNION ALL SELECT 2, 3
+ UNION ALL SELECT 3, 4
+ ),
+ middle AS (
+ SELECT * FROM base
+ UNION ALL SELECT * FROM base
+ ),
+ edges AS (
+ SELECT * FROM ${input}
+ UNION ALL SELECT * FROM ${input}
+ ),
+ r1(n) AS (
+ SELECT 1
+ UNION ALL
+ SELECT e.dst FROM r1 r JOIN edges e ON r.n = e.src
+ ),
+ r2(n) AS (
+ SELECT 2
+ UNION ALL
+ SELECT e.dst FROM r2 r JOIN edges e ON r.n = e.src
+ )
+ SELECT r1.n, r2.n FROM r1 JOIN r2 ON r1.n = r2.n
+ ORDER BY r1.n, r2.n
+ """
+ explain {
+ sql query
+ notContains "MultiCastDataSinks"
+ }
+ // Exercise multiple rounds and final close, which previously raised
NOT_FOUND.
+ sql query
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
