This is an automated email from the ASF dual-hosted git repository.
vldpyatkov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new c1bfd5eef01 IGNITE-29038 Fix recursive CTE subqueries and multiple
consumers (#13559)
c1bfd5eef01 is described below
commit c1bfd5eef015238126fd6ae052cd75710628cf09
Author: Vladislav Pyatkov <[email protected]>
AuthorDate: Tue Sep 8 21:19:07 2026 +0300
IGNITE-29038 Fix recursive CTE subqueries and multiple consumers (#13559)
---
.../query/calcite/exec/ExecutionContext.java | 10 --
.../query/calcite/exec/LogicalRelImplementor.java | 8 +-
.../calcite/exec/rel/RecursiveTableScanNode.java | 49 +++++++
.../query/calcite/exec/rel/RepeatUnionNode.java | 36 +++++-
.../query/calcite/prepare/IgniteSqlValidator.java | 8 ++
.../query/calcite/prepare/PlanningContext.java | 22 ----
.../calcite/prepare/RecursiveCteValidator.java | 141 +++++++++++++++++++++
.../calcite/rel/IgniteRecursiveTableScan.java | 22 +---
.../query/calcite/rel/IgniteRepeatUnion.java | 27 +---
.../query/calcite/rule/RecursiveCteUtils.java | 11 --
.../rule/RecursiveTableScanConverterRule.java | 3 +-
.../calcite/rule/RepeatUnionConverterRule.java | 2 -
.../integration/RecursiveCteIntegrationTest.java | 104 +++++++++++++++
13 files changed, 347 insertions(+), 96 deletions(-)
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionContext.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionContext.java
index 1d129bd3bb0..41a477ef083 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionContext.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionContext.java
@@ -22,7 +22,6 @@ import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.Comparator;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -49,7 +48,6 @@ import
org.apache.ignite.internal.processors.cache.transactions.TransactionChang
import
org.apache.ignite.internal.processors.query.calcite.exec.exp.ExpressionFactory;
import
org.apache.ignite.internal.processors.query.calcite.exec.exp.ExpressionFactoryImpl;
import
org.apache.ignite.internal.processors.query.calcite.exec.exp.ReflectiveCallNotNullImplementor;
-import
org.apache.ignite.internal.processors.query.calcite.exec.rel.RecursiveCteState;
import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.ExecutionNodeMemoryTracker;
import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.IoTracker;
import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.MemoryTracker;
@@ -141,9 +139,6 @@ public class ExecutionContext<Row> extends
AbstractQueryContext implements DataC
/** Map associates UDF name to instance of class that contains this UDF. */
private final Map<String, Object> udfInstances = new ConcurrentHashMap<>();
- /** Query-local recursive CTE states, keyed by transient table identifier.
*/
- private final Map<String, RecursiveCteState<Row>> recursiveCteStates = new
HashMap<>();
-
/** Session context provider injected into UDF targets. */
private final SessionContextProvider sesCtxProv = new
SessionContextProviderImpl();
@@ -477,11 +472,6 @@ public class ExecutionContext<Row> extends
AbstractQueryContext implements DataC
return ExecutionNodeMemoryTracker.create(qryMemoryTracker,
rowOverhead);
}
- /** Returns the state shared by the repeat union and scans of one
recursive CTE. */
- RecursiveCteState<Row> recursiveCteState(String stateId) {
- return recursiveCteStates.computeIfAbsent(stateId, key -> new
RecursiveCteState<>(this));
- }
-
/** */
public IoTracker ioTracker() {
return ioTracker;
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
index 569fe0b38fd..e2777240471 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
@@ -74,7 +74,7 @@ import
org.apache.ignite.internal.processors.query.calcite.exec.rel.NestedLoopJo
import org.apache.ignite.internal.processors.query.calcite.exec.rel.Node;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.Outbox;
import
org.apache.ignite.internal.processors.query.calcite.exec.rel.ProjectNode;
-import
org.apache.ignite.internal.processors.query.calcite.exec.rel.RecursiveCteState;
+import
org.apache.ignite.internal.processors.query.calcite.exec.rel.RecursiveTableScanNode;
import
org.apache.ignite.internal.processors.query.calcite.exec.rel.RepeatUnionNode;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanNode;
import
org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanStorageNode;
@@ -621,7 +621,7 @@ public class LogicalRelImplementor<Row> implements
IgniteRelVisitor<Node<Row>> {
/** {@inheritDoc} */
@Override public Node<Row> visit(IgniteRecursiveTableScan rel) {
- return new ScanNode<>(ctx, rel.getRowType(),
ctx.recursiveCteState(rel.stateId()).current());
+ return new RecursiveTableScanNode<>(ctx, rel.getRowType());
}
/** {@inheritDoc} */
@@ -646,10 +646,8 @@ public class LogicalRelImplementor<Row> implements
IgniteRelVisitor<Node<Row>> {
/** {@inheritDoc} */
@Override public Node<Row> visit(IgniteRepeatUnion rel) {
- RecursiveCteState<Row> state = ctx.recursiveCteState(rel.stateId());
- RepeatUnionNode<Row> node = new RepeatUnionNode<>(ctx,
rel.getRowType(), state, rel.iterationLimit());
+ RepeatUnionNode<Row> node = new RepeatUnionNode<>(ctx,
rel.getRowType(), rel.iterationLimit());
- state.clear();
node.register(F.asList(visit(rel.getLeft()), visit(rel.getRight())));
return node;
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RecursiveTableScanNode.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RecursiveTableScanNode.java
new file mode 100644
index 00000000000..0bbc221d9de
--- /dev/null
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RecursiveTableScanNode.java
@@ -0,0 +1,49 @@
+/*
+ * 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.
+ */
+
+package org.apache.ignite.internal.processors.query.calcite.exec.rel;
+
+import java.util.Collections;
+import java.util.Iterator;
+import org.apache.calcite.rel.type.RelDataType;
+import
org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+
+import static java.util.Objects.requireNonNull;
+
+/** Scan of the current delta owned by the enclosing recursive union. */
+public class RecursiveTableScanNode<Row> extends ScanNode<Row> {
+ /** Owning recursive union, bound when its sources are registered. */
+ private RepeatUnionNode<Row> repeatUnion;
+
+ /** */
+ public RecursiveTableScanNode(ExecutionContext<Row> ctx, RelDataType
rowType) {
+ super(ctx, rowType, Collections.emptyList());
+ }
+
+ /** Binds this scan to the recursive union that owns its current delta. */
+ void bind(RepeatUnionNode<Row> repeatUnion) {
+ assert this.repeatUnion == null;
+
+ this.repeatUnion = requireNonNull(repeatUnion);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected Iterator<Row> sourceIterator() {
+ return requireNonNull(repeatUnion, "Recursive table scan is not bound
to a repeat union")
+ .current().iterator();
+ }
+}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RepeatUnionNode.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RepeatUnionNode.java
index 33622e6ce57..9586b91a362 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RepeatUnionNode.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RepeatUnionNode.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.query.calcite.exec.rel;
+import java.util.List;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
@@ -55,15 +56,23 @@ public class RepeatUnionNode<Row> extends AbstractNode<Row>
implements Downstrea
public RepeatUnionNode(
ExecutionContext<Row> ctx,
RelDataType rowType,
- RecursiveCteState<Row> state,
int iterationLimit
) {
super(ctx, rowType);
- this.state = state;
+ state = new RecursiveCteState<>(ctx);
this.iterationLimit = iterationLimit;
}
+ /** {@inheritDoc} */
+ @Override public void register(List<Node<Row>> sources) {
+ assert sources.size() == 2;
+
+ bindRecursiveScans(sources.get(RECURSIVE_SOURCE));
+
+ super.register(sources);
+ }
+
/** {@inheritDoc} */
@Override public void request(int rowsCnt) throws Exception {
assert !F.isEmpty(sources()) && sources().size() == 2;
@@ -134,6 +143,11 @@ public class RepeatUnionNode<Row> extends
AbstractNode<Row> implements Downstrea
return this;
}
+ /** Current delta visible to recursive scans owned by this union. */
+ Iterable<Row> current() {
+ return state.current();
+ }
+
/** {@inheritDoc} */
@Override protected void rewindInternal() {
curSrc = SEED_SOURCE;
@@ -155,6 +169,24 @@ public class RepeatUnionNode<Row> extends
AbstractNode<Row> implements Downstrea
return sources().get(curSrc);
}
+ /** Binds recursive scans in this union's recursive term without crossing
nested recursive unions. */
+ private void bindRecursiveScans(Node<Row> node) {
+ if (node instanceof RecursiveTableScanNode) {
+ ((RecursiveTableScanNode<Row>)node).bind(this);
+
+ return;
+ }
+
+ // Nested recursive unions bind their own scans when registering their
sources.
+ if (node instanceof RepeatUnionNode)
+ return;
+
+ if (!F.isEmpty(node.sources())) {
+ for (Node<Row> src : node.sources())
+ bindRecursiveScans(src);
+ }
+ }
+
/** Starts collecting and requests rows from the active input. */
private void requestSource() throws Exception {
if (!writing) {
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java
index d7af46290f4..5f31f4d64a5 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java
@@ -52,6 +52,7 @@ import org.apache.calcite.sql.SqlSelect;
import org.apache.calcite.sql.SqlUpdate;
import org.apache.calcite.sql.SqlUtil;
import org.apache.calcite.sql.SqlWindow;
+import org.apache.calcite.sql.SqlWithItem;
import org.apache.calcite.sql.dialect.CalciteSqlDialect;
import org.apache.calcite.sql.fun.SqlCase;
import org.apache.calcite.sql.parser.SqlParserPos;
@@ -155,6 +156,13 @@ public class IgniteSqlValidator extends SqlValidatorImpl {
super.validateInsert(insert);
}
+ /** {@inheritDoc} */
+ @Override public void validateWithItem(SqlWithItem withItem) {
+ super.validateWithItem(withItem);
+
+ RecursiveCteValidator.validate(this, withItem);
+ }
+
/**
* Validates insert target columns to ensure they do not contain the
version column.
*
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlanningContext.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlanningContext.java
index ffb42003e83..31558c6af82 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlanningContext.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlanningContext.java
@@ -17,18 +17,14 @@
package org.apache.ignite.internal.processors.query.calcite.prepare;
-import java.util.IdentityHashMap;
-import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import org.apache.calcite.plan.Context;
import org.apache.calcite.plan.Contexts;
import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.prepare.CalciteCatalogReader;
import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.schema.SchemaPlus;
-import org.apache.calcite.schema.TransientTable;
import org.apache.calcite.sql.SqlOperatorTable;
import org.apache.calcite.tools.FrameworkConfig;
import org.apache.calcite.tools.RuleSet;
@@ -62,12 +58,6 @@ public final class PlanningContext implements Context {
/** */
private IgnitePlanner planner;
- /** Query-local identifiers of recursive transient tables. */
- private final Map<TransientTable, String> recursiveCteStateIds = new
IdentityHashMap<>();
-
- /** Next query-local recursive CTE identifier. */
- private int nextRecursiveCteStateId;
-
/** */
private final long startTs;
@@ -189,18 +179,6 @@ public final class PlanningContext implements Context {
return planner().cluster();
}
- /** Returns an identifier unique for the given recursive CTE within this
planning context. */
- public synchronized String recursiveCteStateId(RelOptTable table) {
- TransientTable transientTable = table.unwrap(TransientTable.class);
-
- assert transientTable != null;
-
- return recursiveCteStateIds.computeIfAbsent(
- transientTable,
- key -> String.join(".", table.getQualifiedName()) + '#' +
nextRecursiveCteStateId++
- );
- }
-
/** {@inheritDoc} */
@Override public <C> @Nullable C unwrap(Class<C> aCls) {
if (aCls == getClass())
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/RecursiveCteValidator.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/RecursiveCteValidator.java
new file mode 100644
index 00000000000..d809e051d51
--- /dev/null
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/RecursiveCteValidator.java
@@ -0,0 +1,141 @@
+/*
+ * 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.
+ */
+
+package org.apache.ignite.internal.processors.query.calcite.prepare;
+
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.Set;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlWith;
+import org.apache.calcite.sql.SqlWithItem;
+import org.apache.calcite.sql.util.SqlBasicVisitor;
+import org.apache.calcite.sql.validate.SqlValidatorNamespace;
+import org.apache.calcite.sql.validate.SqlWithItemTableRef;
+import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
+import org.apache.ignite.internal.processors.query.IgniteSQLException;
+
+/** Validates restrictions specific to recursive common table expressions. */
+final class RecursiveCteValidator {
+ /** */
+ private RecursiveCteValidator() {
+ // No-op.
+ }
+
+ /** Validates that a recursive CTE does not reference itself from a nested
query. */
+ static void validate(IgniteSqlValidator validator, SqlWithItem withItem) {
+ if (withItem.recursive == null || !withItem.recursive.booleanValue())
+ return;
+
+ Set<SqlNode> topLevelQueries = Collections.newSetFromMap(new
IdentityHashMap<>());
+
+ collectTopLevelQueries(withItem.query, topLevelQueries);
+
+ RecursiveCteReferenceFinder finder =
+ new RecursiveCteReferenceFinder(validator, withItem,
topLevelQueries);
+
+ withItem.query.accept(finder);
+
+ if (finder.nestedSelfReferenceFound) {
+ throw new IgniteSQLException(
+ "Unsupported recursive CTE: self-references inside subqueries
are not supported",
+ IgniteQueryErrorCode.UNSUPPORTED_OPERATION
+ );
+ }
+ }
+
+ /** Collects query nodes that constitute the CTE's top-level query
expression. */
+ private static void collectTopLevelQueries(SqlNode node, Set<SqlNode>
queries) {
+ if (node == null || !node.isA(SqlKind.QUERY))
+ return;
+
+ queries.add(node);
+
+ if (node.isA(SqlKind.SET_QUERY)) {
+ for (SqlNode operand : ((SqlCall)node).getOperandList())
+ collectTopLevelQueries(operand, queries);
+ }
+ else if (node instanceof SqlWith)
+ collectTopLevelQueries(((SqlWith)node).body, queries);
+ else if (node.getKind() == SqlKind.ORDER_BY)
+ collectTopLevelQueries(((SqlCall)node).operand(0), queries);
+ }
+
+ /** Finds references to the recursive CTE located inside nested queries. */
+ private static class RecursiveCteReferenceFinder extends
SqlBasicVisitor<Void> {
+ /** SQL validator. */
+ private final IgniteSqlValidator validator;
+
+ /** Recursive CTE being validated. */
+ private final SqlWithItem withItem;
+
+ /** Nodes that belong to the CTE's top-level query expression. */
+ private final Set<SqlNode> topLevelQueries;
+
+ /** Whether the visitor is inside a nested query. */
+ private boolean insideSubquery;
+
+ /** Whether an unsupported reference was found. */
+ private boolean nestedSelfReferenceFound;
+
+ /** */
+ private RecursiveCteReferenceFinder(
+ IgniteSqlValidator validator,
+ SqlWithItem withItem,
+ Set<SqlNode> topLevelQueries
+ ) {
+ this.validator = validator;
+ this.withItem = withItem;
+ this.topLevelQueries = topLevelQueries;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Void visit(SqlCall call) {
+ if (nestedSelfReferenceFound)
+ return null;
+
+ boolean wasInsideSubquery = insideSubquery;
+
+ insideSubquery |= call.isA(SqlKind.QUERY) &&
!topLevelQueries.contains(call);
+
+ super.visit(call);
+
+ insideSubquery = wasInsideSubquery;
+
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Void visit(SqlIdentifier id) {
+ if (!insideSubquery)
+ return null;
+
+ SqlValidatorNamespace namespace = validator.getNamespace(id);
+ SqlNode resolvedNode = namespace == null ? null :
namespace.resolve().getNode();
+
+ if (resolvedNode instanceof SqlWithItemTableRef
+ && ((SqlWithItemTableRef)resolvedNode).getWithItem() ==
withItem) {
+ nestedSelfReferenceFound = true;
+ }
+
+ return null;
+ }
+ }
+}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRecursiveTableScan.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRecursiveTableScan.java
index d51e1bd53d1..fa12efb5ce1 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRecursiveTableScan.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRecursiveTableScan.java
@@ -26,24 +26,17 @@ import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.RelWriter;
import org.apache.calcite.rel.type.RelDataType;
-import static java.util.Objects.requireNonNull;
-
/** Scan of the current delta of a query-local recursive CTE. */
public class IgniteRecursiveTableScan extends AbstractRelNode implements
IgniteRel {
- /** Query-local recursive state identifier. */
- private final String stateId;
-
/** */
public IgniteRecursiveTableScan(
RelOptCluster cluster,
RelTraitSet traits,
- RelDataType rowType,
- String stateId
+ RelDataType rowType
) {
super(cluster, traits);
this.rowType = rowType;
- this.stateId = stateId;
}
/** Constructor used for deserialization. */
@@ -51,21 +44,15 @@ public class IgniteRecursiveTableScan extends
AbstractRelNode implements IgniteR
this(
input.getCluster(),
input.getTraitSet().replace(IgniteConvention.INSTANCE),
- input.getRowType("rowType"),
- requireNonNull(input.getString("stateId"), "stateId")
+ input.getRowType("rowType")
);
}
- /** Query-local recursive state identifier. */
- public String stateId() {
- return stateId;
- }
-
/** {@inheritDoc} */
@Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
assert inputs.isEmpty();
- return new IgniteRecursiveTableScan(getCluster(), traitSet, rowType,
stateId);
+ return new IgniteRecursiveTableScan(getCluster(), traitSet, rowType);
}
/** {@inheritDoc} */
@@ -77,13 +64,12 @@ public class IgniteRecursiveTableScan extends
AbstractRelNode implements IgniteR
@Override public IgniteRel clone(RelOptCluster cluster, List<IgniteRel>
inputs) {
assert inputs.isEmpty();
- return new IgniteRecursiveTableScan(cluster, getTraitSet(), rowType,
stateId);
+ return new IgniteRecursiveTableScan(cluster, getTraitSet(), rowType);
}
/** {@inheritDoc} */
@Override public RelWriter explainTerms(RelWriter pw) {
return super.explainTerms(pw)
- .item("stateId", stateId)
.item("rowType", rowType);
}
}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRepeatUnion.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRepeatUnion.java
index 0a7db70d139..a5689f2dbe6 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRepeatUnion.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRepeatUnion.java
@@ -22,28 +22,19 @@ import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.RelInput;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelWriter;
import org.apache.calcite.rel.core.RepeatUnion;
-import static java.util.Objects.requireNonNull;
-
/** Coordinator-side iterative UNION ALL for a recursive CTE. */
public class IgniteRepeatUnion extends RepeatUnion implements IgniteRel {
- /** Query-local recursive state identifier. */
- private final String stateId;
-
/** */
public IgniteRepeatUnion(
RelOptCluster cluster,
RelTraitSet traits,
RelNode seed,
RelNode iterative,
- String stateId,
int iterationLimit
) {
super(cluster, traits, seed, iterative, true, iterationLimit, null);
-
- this.stateId = stateId;
}
/** Constructor used for deserialization. */
@@ -53,16 +44,10 @@ public class IgniteRepeatUnion extends RepeatUnion
implements IgniteRel {
input.getTraitSet().replace(IgniteConvention.INSTANCE),
input.getInputs().get(0),
input.getInputs().get(1),
- requireNonNull(input.getString("stateId"), "stateId"),
iterationLimit(input)
);
}
- /** Query-local recursive state identifier. */
- public String stateId() {
- return stateId;
- }
-
/** Maximum number of recursive iterations, or a negative value for no
limit. */
public int iterationLimit() {
return iterationLimit;
@@ -72,7 +57,7 @@ public class IgniteRepeatUnion extends RepeatUnion implements
IgniteRel {
@Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
assert inputs.size() == 2;
- return new IgniteRepeatUnion(getCluster(), traitSet, inputs.get(0),
inputs.get(1), stateId, iterationLimit);
+ return new IgniteRepeatUnion(getCluster(), traitSet, inputs.get(0),
inputs.get(1), iterationLimit);
}
/** {@inheritDoc} */
@@ -84,16 +69,10 @@ public class IgniteRepeatUnion extends RepeatUnion
implements IgniteRel {
@Override public IgniteRel clone(RelOptCluster cluster, List<IgniteRel>
inputs) {
assert inputs.size() == 2;
- return new IgniteRepeatUnion(cluster, getTraitSet(), inputs.get(0),
inputs.get(1), stateId, iterationLimit);
- }
-
- /** {@inheritDoc} */
- @Override public RelWriter explainTerms(RelWriter pw) {
- return super.explainTerms(pw)
- .item("stateId", stateId);
+ return new IgniteRepeatUnion(cluster, getTraitSet(), inputs.get(0),
inputs.get(1), iterationLimit);
}
- /** Reads the optional iteration limit emitted by {@link
RepeatUnion#explainTerms(RelWriter)}. */
+ /** Reads the optional iteration limit from a serialized plan. */
private static int iterationLimit(RelInput input) {
Number iterationLimit = (Number)input.get("iterationLimit");
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveCteUtils.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveCteUtils.java
index 395d85b38d1..ee7b04da2f4 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveCteUtils.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveCteUtils.java
@@ -19,7 +19,6 @@ package
org.apache.ignite.internal.processors.query.calcite.rule;
import java.util.ArrayList;
import java.util.List;
-import org.apache.calcite.plan.RelOptPlanner;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.plan.RelOptUtil;
@@ -36,7 +35,6 @@ import org.apache.calcite.rex.RexSubQuery;
import org.apache.calcite.schema.TransientTable;
import org.apache.calcite.sql.validate.SqlUserDefinedFunction;
import
org.apache.ignite.internal.processors.query.calcite.exec.exp.IgniteScalarFunction;
-import
org.apache.ignite.internal.processors.query.calcite.prepare.PlanningContext;
import
org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
import
org.apache.ignite.internal.processors.query.calcite.trait.RewindabilityTrait;
import org.apache.ignite.internal.processors.query.calcite.trait.TraitUtils;
@@ -55,15 +53,6 @@ final class RecursiveCteUtils {
return table != null && table.unwrap(TransientTable.class) != null;
}
- /** Stable identifier preserved in the serialized physical plan. */
- static String stateId(RelOptPlanner planner, RelOptTable table) {
- PlanningContext ctx =
planner.getContext().unwrap(PlanningContext.class);
-
- assert ctx != null;
-
- return ctx.recursiveCteStateId(table);
- }
-
/** Counts scans of the recursive transient table. */
static int referenceCount(RelNode rel, RelOptTable table) {
rel = original(rel);
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveTableScanConverterRule.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveTableScanConverterRule.java
index 63c0e01401f..003efe72349 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveTableScanConverterRule.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveTableScanConverterRule.java
@@ -56,8 +56,7 @@ public class RecursiveTableScanConverterRule extends
AbstractIgniteConverterRule
return new IgniteRecursiveTableScan(
rel.getCluster(),
traits,
- rel.getRowType(),
- RecursiveCteUtils.stateId(planner, rel.getTable())
+ rel.getRowType()
);
}
}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RepeatUnionConverterRule.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RepeatUnionConverterRule.java
index d53542ea5ac..ef37d92cbb0 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RepeatUnionConverterRule.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RepeatUnionConverterRule.java
@@ -57,7 +57,6 @@ public class RepeatUnionConverterRule extends
AbstractIgniteConverterRule<Logica
if (!rel.all)
throw unsupported("only UNION ALL is supported");
- String stateId = RecursiveCteUtils.stateId(planner, table);
int iterationLimit =
planner.getContext().unwrap(PlanningContext.class).recursiveCteIterationLimit();
RelNode seed = unwrapSpool(rel.getSeedRel(), "seed");
@@ -72,7 +71,6 @@ public class RepeatUnionConverterRule extends
AbstractIgniteConverterRule<Logica
traits,
convert(seed, traits),
convert(iterative, traits),
- stateId,
iterationLimit
);
}
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/RecursiveCteIntegrationTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/RecursiveCteIntegrationTest.java
index 11a333e30fa..be0a8aa3039 100644
---
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/RecursiveCteIntegrationTest.java
+++
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/RecursiveCteIntegrationTest.java
@@ -115,6 +115,88 @@ public class RecursiveCteIntegrationTest extends
AbstractBasicIntegrationTest {
.check();
}
+ /** */
+ @Test
+ public void testRecursiveCteWithMultipleJoinConsumers() {
+ assertQuery("WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "SELECT n + 1 FROM numbers WHERE n < 3" +
+ ") " +
+ "SELECT l.n, r.n " +
+ "FROM numbers l " +
+ "JOIN numbers r ON l.n = r.n")
+ .returns(1, 1)
+ .returns(2, 2)
+ .returns(3, 3)
+ .check();
+ }
+
+ /** */
+ @Test
+ public void testRecursiveCteWithMultipleUnionConsumers() {
+ assertQuery("WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "SELECT n + 1 FROM numbers WHERE n < 3" +
+ ") " +
+ "SELECT n FROM numbers " +
+ "UNION ALL " +
+ "SELECT n FROM numbers")
+ .returns(1)
+ .returns(2)
+ .returns(3)
+ .returns(1)
+ .returns(2)
+ .returns(3)
+ .check();
+ }
+
+ /** */
+ @Test
+ public void testSelfReferenceInScalarSubqueryIsRejected() {
+ assertThrows(
+ "WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "SELECT (SELECT n + 1 FROM numbers) FROM (VALUES (0))" +
+ ") " +
+ "SELECT n FROM numbers FETCH FIRST 3 ROWS ONLY",
+ IgniteSQLException.class,
+ "self-references inside subqueries are not supported"
+ );
+ }
+
+ /** */
+ @Test
+ public void testSelfReferenceInDerivedTableIsRejected() {
+ assertThrows(
+ "WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "SELECT n + 1 FROM (SELECT n FROM numbers) WHERE n < 3" +
+ ") " +
+ "SELECT n FROM numbers",
+ IgniteSQLException.class,
+ "self-references inside subqueries are not supported"
+ );
+ }
+
+ /** */
+ @Test
+ public void testRecursiveCteCanBeReadFromScalarSubquery() {
+ assertQuery(
+ "WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "SELECT n + 1 FROM numbers WHERE n < 3" +
+ ") " +
+ "SELECT (SELECT MAX(n) FROM numbers)"
+ )
+ .returns(3)
+ .check();
+ }
+
/** */
@Test
public void testRecursiveCteWithMultipleRecursiveBranches() {
@@ -178,6 +260,28 @@ public class RecursiveCteIntegrationTest extends
AbstractBasicIntegrationTest {
.check();
}
+ /** */
+ @Test
+ public void testNestedRecursiveCteStatesAreIsolated() {
+ assertQuery("WITH RECURSIVE first_numbers(n) AS (" +
+ "SELECT 10 " +
+ "UNION ALL " +
+ "SELECT n + 1 FROM first_numbers WHERE n < 12" +
+ "), second_numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "SELECT second_numbers.n + 1 " +
+ "FROM second_numbers " +
+ "JOIN first_numbers ON second_numbers.n + 9 = first_numbers.n
" +
+ "WHERE second_numbers.n < 3" +
+ ") " +
+ "SELECT n FROM second_numbers")
+ .returns(1)
+ .returns(2)
+ .returns(3)
+ .check();
+ }
+
/** */
@Test
public void
testIndependentNonDeterministicSubtreeIsEvaluatedForEveryIteration() {
