This is an automated email from the ASF dual-hosted git repository.
Mryange 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 6835dc68619 [fix](fe) Deduplicate recursive CTE fragment reset
requests (#67738)
6835dc68619 is described below
commit 6835dc68619899a742b1d9113abd054b1d541df5
Author: Mryange <[email protected]>
AuthorDate: Mon Sep 14 10:12:03 2026 +0800
[fix](fe) Deduplicate recursive CTE fragment reset requests (#67738)
Recursive CTE queries could fail with `Fragment context ... not found`
during recursive fragment cleanup. The FE collected recursive child
fragments through a shared plan tree and could generate duplicate reset
entries for the same fragment on the same BE. The first
`WAIT_FOR_DESTROY` request removed the BE fragment context, while the
duplicate request then failed to find it. This change deduplicates reset
entries by fragment ID and BE address while preserving entries for
different BEs.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../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 1a239e3122a..64813f1b7b5 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
@@ -822,6 +822,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,
@@ -845,6 +846,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 7ae9aa1e995..cdaa42e1182 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;
@@ -95,6 +96,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]
