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 6f2a343d282 IGNITE-27822 Support recursive query for Calcite engine
(#13479)
6f2a343d282 is described below
commit 6f2a343d282441dc7360960a6542d4459a297358
Author: Vladislav Pyatkov <[email protected]>
AuthorDate: Fri Sep 4 12:50:53 2026 +0300
IGNITE-27822 Support recursive query for Calcite engine (#13479)
---
.../calcite/DistributedCalciteConfiguration.java | 33 +++
.../query/calcite/exec/ExecutionContext.java | 10 +
.../query/calcite/exec/LogicalRelImplementor.java | 20 ++
.../query/calcite/exec/rel/RecursiveCteState.java | 94 +++++++++
.../query/calcite/exec/rel/RepeatUnionNode.java | 183 ++++++++++++++++
.../calcite/metadata/IgniteMdFragmentMapping.java | 12 ++
.../processors/query/calcite/prepare/Cloner.java | 12 ++
.../query/calcite/prepare/IgniteRelShuttle.java | 12 ++
.../query/calcite/prepare/PlannerPhase.java | 4 +
.../query/calcite/prepare/PlanningContext.java | 41 ++++
.../query/calcite/prepare/PrepareServiceImpl.java | 2 +
.../query/calcite/rel/IgniteHashIndexSpool.java | 19 ++
.../query/calcite/rel/IgniteIndexScan.java | 16 ++
.../calcite/rel/IgniteRecursiveTableScan.java | 89 ++++++++
.../query/calcite/rel/IgniteRelVisitor.java | 10 +
.../query/calcite/rel/IgniteRepeatUnion.java | 102 +++++++++
.../query/calcite/rel/IgniteSortedIndexSpool.java | 34 +++
.../query/calcite/rel/IgniteTableScan.java | 11 +
.../rel/ProjectableFilterableTableScan.java | 30 +++
.../rel/logical/IgniteLogicalIndexScan.java | 10 +
.../rel/logical/IgniteLogicalTableScan.java | 11 +
.../query/calcite/rule/RecursiveCteUtils.java | 218 ++++++++++++++++++++
.../rule/RecursiveTableScanConverterRule.java | 63 ++++++
.../calcite/rule/RepeatUnionConverterRule.java | 102 +++++++++
.../CalciteQueryProcessorPropertiesTest.java | 52 +++++
.../integration/MemoryQuotasIntegrationTest.java | 18 ++
.../integration/RecursiveCteIntegrationTest.java | 208 +++++++++++++++++--
.../calcite/planner/RecursiveCtePlannerTest.java | 229 +++++++++++++++++++++
.../apache/ignite/testsuites/PlannerTestSuite.java | 2 +
.../sql/hierarchy/test_recursive_hierarchy.test | 64 ++++++
.../sql/hierarchy/test_recursive_sequence.test | 62 ++++++
.../sql/hierarchy/test_recursive_subquery.test | 54 +++++
32 files changed, 1809 insertions(+), 18 deletions(-)
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/DistributedCalciteConfiguration.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/DistributedCalciteConfiguration.java
index 26c3a7b14c9..3aee850cd26 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/DistributedCalciteConfiguration.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/DistributedCalciteConfiguration.java
@@ -42,18 +42,28 @@ public class DistributedCalciteConfiguration extends
DistributedSqlConfiguration
/** Plan cache size property name. */
public static final String PLAN_CACHE_SIZE_PROPERTY_NAME =
"sql.calcite.planCacheSize";
+ /** Recursive CTE iteration limit property name. */
+ public static final String RECURSIVE_CTE_ITERATION_LIMIT_PROPERTY_NAME =
+ "sql.calcite.recursiveCteIterationLimit";
+
/** Default value of the disabled rules. */
public static final String[] DFLT_DISABLED_RULES = new String[0];
/** Default value of plan cache size. */
public static final int DFLT_PLAN_CACHE_SIZE = 1024;
+ /** Default recursive CTE iteration limit. */
+ public static final int DFLT_RECURSIVE_CTE_ITERATION_LIMIT = 100;
+
/** Globally disabled rules. */
private volatile DistributedChangeableProperty<String[]> disabledRules;
/** Plan cache size. */
private volatile DistributedChangeableProperty<Integer> planCacheSize;
+ /** Recursive CTE iteration limit. */
+ private volatile DistributedChangeableProperty<Integer>
recursiveCteIterationLimit;
+
/** */
private QueryPlanCache qryPlanCache;
@@ -91,6 +101,13 @@ public class DistributedCalciteConfiguration extends
DistributedSqlConfiguration
return getProperty(planCacheSize, DFLT_PLAN_CACHE_SIZE);
}
+ /**
+ * @return Maximum number of recursive CTE iterations, or a negative value
for no limit.
+ */
+ public int recursiveCteIterationLimit() {
+ return getProperty(recursiveCteIterationLimit,
DFLT_RECURSIVE_CTE_ITERATION_LIMIT);
+ }
+
/** */
private <T extends Serializable> T
getProperty(DistributedChangeableProperty<T> prop, T dflt) {
T res = prop == null ? dflt : prop.get();
@@ -140,6 +157,21 @@ public class DistributedCalciteConfiguration extends
DistributedSqlConfiguration
);
planCacheSize.addListener(planCacheCleaner);
+
+ registerProperty(
+ dispatcher,
+ RECURSIVE_CTE_ITERATION_LIMIT_PROPERTY_NAME,
+ prop -> recursiveCteIterationLimit = prop,
+ () -> new SimpleDistributedProperty<>(
+ RECURSIVE_CTE_ITERATION_LIMIT_PROPERTY_NAME,
+ Integer::parseInt,
+ "Maximum number of recursive CTE iterations before query
execution fails. " +
+ "A negative value disables the limit. NOTE: cleans the
planning cache on change."
+ ),
+ log
+ );
+
+ recursiveCteIterationLimit.addListener(planCacheCleaner);
}
/** {@inheritDoc} */
@@ -148,5 +180,6 @@ public class DistributedCalciteConfiguration extends
DistributedSqlConfiguration
setDefaultValue(disabledRules, DFLT_DISABLED_RULES, log);
setDefaultValue(planCacheSize, DFLT_PLAN_CACHE_SIZE, log);
+ setDefaultValue(recursiveCteIterationLimit,
DFLT_RECURSIVE_CTE_ITERATION_LIMIT, log);
}
}
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 41a477ef083..1d129bd3bb0 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,6 +22,7 @@ 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;
@@ -48,6 +49,7 @@ 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;
@@ -139,6 +141,9 @@ 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();
@@ -472,6 +477,11 @@ 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 ed04f327eb3..569fe0b38fd 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,6 +74,8 @@ 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.RepeatUnionNode;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanNode;
import
org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanStorageNode;
import
org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanTableRowNode;
@@ -101,8 +103,10 @@ import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteMergeJoin;
import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteNestedLoopJoin;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteProject;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReceiver;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRecursiveTableScan;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRelVisitor;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRepeatUnion;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSort;
import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteSortedIndexSpool;
@@ -615,6 +619,11 @@ 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());
+ }
+
/** {@inheritDoc} */
@Override public Node<Row> visit(IgniteValues rel) {
List<RexLiteral> vals = Commons.flat(Commons.cast(rel.getTuples()));
@@ -635,6 +644,17 @@ public class LogicalRelImplementor<Row> implements
IgniteRelVisitor<Node<Row>> {
return node;
}
+ /** {@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());
+
+ state.clear();
+ node.register(F.asList(visit(rel.getLeft()), visit(rel.getRight())));
+
+ return node;
+ }
+
/** {@inheritDoc} */
@Override public Node<Row> visit(IgniteLimit rel) {
long offset = validateAndGetOffset(rel.offset(),
LimitNode.OFFSET_DEFAULT);
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RecursiveCteState.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RecursiveCteState.java
new file mode 100644
index 00000000000..8a865164de8
--- /dev/null
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RecursiveCteState.java
@@ -0,0 +1,94 @@
+/*
+ * 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.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import
org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.RowTracker;
+import org.apache.ignite.internal.util.GridUnsafe;
+
+/** Query-local current and next deltas of a recursive CTE. */
+public class RecursiveCteState<Row> {
+ /** Rows visible to the recursive table scan. */
+ private List<Row> cur = Collections.emptyList();
+
+ /** Rows produced by the active seed or recursive term. */
+ private List<Row> next;
+
+ /** Memory tracker for rows in the current delta. */
+ private RowTracker<Row> curMemoryTracker;
+
+ /** Memory tracker for rows in the next delta. */
+ private RowTracker<Row> nextMemoryTracker;
+
+ /** */
+ public RecursiveCteState(ExecutionContext<Row> ctx) {
+ curMemoryTracker =
ctx.createNodeMemoryTracker(GridUnsafe.OBJ_REF_SIZE);
+ nextMemoryTracker =
ctx.createNodeMemoryTracker(GridUnsafe.OBJ_REF_SIZE);
+ }
+
+ /** Starts collecting the next delta. */
+ public void beginWrite() {
+ assert next == null;
+
+ next = new ArrayList<>();
+ }
+
+ /** Adds a row to the next delta. */
+ public void add(Row row) {
+ assert next != null;
+
+ next.add(row);
+ nextMemoryTracker.onRowAdded(row);
+ }
+
+ /** Makes the collected delta visible to recursive scans. */
+ public void commit() {
+ assert next != null;
+
+ curMemoryTracker.reset();
+ cur = next;
+ next = null;
+
+ RowTracker<Row> tracker = curMemoryTracker;
+
+ curMemoryTracker = nextMemoryTracker;
+ nextMemoryTracker = tracker;
+ }
+
+ /** Current delta. */
+ public Iterable<Row> current() {
+ return () -> cur.iterator();
+ }
+
+ /** Returns whether the current delta is empty. */
+ public boolean isEmpty() {
+ return cur.isEmpty();
+ }
+
+ /** Clears all query-local rows. */
+ public void clear() {
+ cur = Collections.emptyList();
+ next = null;
+
+ curMemoryTracker.reset();
+ nextMemoryTracker.reset();
+ }
+}
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
new file mode 100644
index 00000000000..33622e6ce57
--- /dev/null
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RepeatUnionNode.java
@@ -0,0 +1,183 @@
+/*
+ * 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 org.apache.calcite.rel.type.RelDataType;
+import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
+import org.apache.ignite.internal.processors.query.IgniteSQLException;
+import
org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.internal.util.typedef.F;
+
+import static
org.apache.ignite.internal.processors.query.calcite.DistributedCalciteConfiguration.RECURSIVE_CTE_ITERATION_LIMIT_PROPERTY_NAME;
+
+/** Coordinator-side executor for recursive UNION ALL. */
+public class RepeatUnionNode<Row> extends AbstractNode<Row> implements
Downstream<Row> {
+ /** Index of the seed input. */
+ private static final int SEED_SOURCE = 0;
+
+ /** Index of the recursive-term input. */
+ private static final int RECURSIVE_SOURCE = 1;
+
+ /** Query-local recursive state. */
+ private final RecursiveCteState<Row> state;
+
+ /** Maximum number of recursive iterations, or a negative value for no
limit. */
+ private final int iterationLimit;
+
+ /** Index of the active source. */
+ private int curSrc = SEED_SOURCE;
+
+ /** Number of rows still requested by downstream. */
+ private int waiting;
+
+ /** Number of completed recursive iterations. */
+ private int iteration;
+
+ /** Whether the active input is being collected into the next delta. */
+ private boolean writing;
+
+ /** */
+ public RepeatUnionNode(
+ ExecutionContext<Row> ctx,
+ RelDataType rowType,
+ RecursiveCteState<Row> state,
+ int iterationLimit
+ ) {
+ super(ctx, rowType);
+
+ this.state = state;
+ this.iterationLimit = iterationLimit;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void request(int rowsCnt) throws Exception {
+ assert !F.isEmpty(sources()) && sources().size() == 2;
+ assert rowsCnt > 0 && waiting == 0;
+
+ checkState();
+
+ waiting = rowsCnt;
+ requestSource();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void push(Row row) throws Exception {
+ assert downstream() != null;
+ assert waiting > 0;
+ assert writing;
+
+ checkState();
+
+ waiting--;
+ state.add(row);
+
+ downstream().push(row);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void end() throws Exception {
+ assert downstream() != null;
+ assert waiting > 0;
+ assert writing;
+
+ checkState();
+
+ state.commit();
+ writing = false;
+
+ if (state.isEmpty()) {
+ finish();
+
+ return;
+ }
+
+ if (curSrc == SEED_SOURCE) {
+ if (iterationLimit == 0) {
+ throw iterationLimitExceeded();
+ }
+
+ curSrc = RECURSIVE_SOURCE;
+ requestSource();
+
+ return;
+ }
+
+ iteration++;
+
+ if (iterationLimit >= 0 && iteration == iterationLimit) {
+ throw iterationLimitExceeded();
+ }
+
+ source().rewind();
+ requestSource();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected Downstream<Row> requestDownstream(int idx) {
+ assert idx >= 0 && idx < 2;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void rewindInternal() {
+ curSrc = SEED_SOURCE;
+ waiting = 0;
+ iteration = 0;
+ writing = false;
+ state.clear();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void closeInternal() {
+ state.clear();
+
+ super.closeInternal();
+ }
+
+ /** */
+ private Node<Row> source() {
+ return sources().get(curSrc);
+ }
+
+ /** Starts collecting and requests rows from the active input. */
+ private void requestSource() throws Exception {
+ if (!writing) {
+ state.beginWrite();
+ writing = true;
+ }
+
+ source().request(waiting);
+ }
+
+ /** */
+ private void finish() throws Exception {
+ waiting = -1;
+ state.clear();
+ downstream().end();
+ }
+
+ /** */
+ private IgniteSQLException iterationLimitExceeded() {
+ return new IgniteSQLException(
+ "Recursive CTE iteration limit exceeded [limit=" + iterationLimit +
+ ", property=" + RECURSIVE_CTE_ITERATION_LIMIT_PROPERTY_NAME +
']',
+ IgniteQueryErrorCode.QUERY_CANCELED
+ );
+ }
+}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdFragmentMapping.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdFragmentMapping.java
index 51b0ca886ad..99ad37affff 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdFragmentMapping.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdFragmentMapping.java
@@ -37,6 +37,7 @@ import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexBound;
import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexCount;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexScan;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReceiver;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRecursiveTableScan;
import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableFunctionScan;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteTrimExchange;
@@ -216,6 +217,17 @@ public class IgniteMdFragmentMapping implements
MetadataHandler<FragmentMappingM
rel.getTable().unwrap(IgniteTable.class).colocationGroup(ctx));
}
+ /**
+ * See {@link IgniteMdFragmentMapping#fragmentMapping(RelNode,
RelMetadataQuery, MappingQueryContext)}
+ */
+ public FragmentMapping fragmentMapping(
+ IgniteRecursiveTableScan rel,
+ RelMetadataQuery mq,
+ MappingQueryContext ctx
+ ) {
+ return FragmentMapping.create(ctx.localNodeId());
+ }
+
/**
* See {@link IgniteMdFragmentMapping#fragmentMapping(RelNode,
RelMetadataQuery, MappingQueryContext)}
*/
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Cloner.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Cloner.java
index a60ae8a9cb2..dc6b0e3c8bc 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Cloner.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Cloner.java
@@ -33,8 +33,10 @@ import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteMergeJoin;
import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteNestedLoopJoin;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteProject;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReceiver;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRecursiveTableScan;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRelVisitor;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRepeatUnion;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSort;
import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteSortedIndexSpool;
@@ -179,6 +181,11 @@ public class Cloner implements IgniteRelVisitor<IgniteRel>
{
return rel.clone(cluster, F.asList());
}
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteRecursiveTableScan rel) {
+ return rel.clone(cluster, F.asList());
+ }
+
/** {@inheritDoc} */
@Override public IgniteRel visit(IgniteValues rel) {
return rel.clone(cluster, F.asList());
@@ -189,6 +196,11 @@ public class Cloner implements IgniteRelVisitor<IgniteRel>
{
return rel.clone(cluster, Commons.transform(rel.getInputs(), rel0 ->
visit((IgniteRel)rel0)));
}
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteRepeatUnion rel) {
+ return rel.clone(cluster, F.asList(visit((IgniteRel)rel.getLeft()),
visit((IgniteRel)rel.getRight())));
+ }
+
/** {@inheritDoc} */
@Override public IgniteRel visit(IgniteSort rel) {
return rel.clone(cluster, F.asList(visit((IgniteRel)rel.getInput())));
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteRelShuttle.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteRelShuttle.java
index c922ccba7c8..e5b317a5eaf 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteRelShuttle.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteRelShuttle.java
@@ -32,8 +32,10 @@ import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteMergeJoin;
import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteNestedLoopJoin;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteProject;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReceiver;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRecursiveTableScan;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRelVisitor;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRepeatUnion;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSort;
import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteSortedIndexSpool;
@@ -142,6 +144,11 @@ public class IgniteRelShuttle implements
IgniteRelVisitor<IgniteRel> {
return processNode(rel);
}
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteRepeatUnion rel) {
+ return processNode(rel);
+ }
+
/** {@inheritDoc} */
@Override public IgniteRel visit(IgniteSort rel) {
return processNode(rel);
@@ -172,6 +179,11 @@ public class IgniteRelShuttle implements
IgniteRelVisitor<IgniteRel> {
return processNode(rel);
}
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteRecursiveTableScan rel) {
+ return processNode(rel);
+ }
+
/** {@inheritDoc} */
@Override public IgniteRel visit(IgniteReceiver rel) {
return processNode(rel);
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java
index 71753b173f2..eac30d2d05e 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java
@@ -62,6 +62,8 @@ import
org.apache.ignite.internal.processors.query.calcite.rule.LogicalScanConve
import
org.apache.ignite.internal.processors.query.calcite.rule.MergeJoinConverterRule;
import
org.apache.ignite.internal.processors.query.calcite.rule.NestedLoopJoinConverterRule;
import
org.apache.ignite.internal.processors.query.calcite.rule.ProjectConverterRule;
+import
org.apache.ignite.internal.processors.query.calcite.rule.RecursiveTableScanConverterRule;
+import
org.apache.ignite.internal.processors.query.calcite.rule.RepeatUnionConverterRule;
import
org.apache.ignite.internal.processors.query.calcite.rule.SetOpConverterRule;
import
org.apache.ignite.internal.processors.query.calcite.rule.SortAggregateConverterRule;
import
org.apache.ignite.internal.processors.query.calcite.rule.SortConverterRule;
@@ -311,6 +313,8 @@ public enum PlannerPhase {
//CoreRules.WINDOW_REDUCE_EXPRESSIONS,
ValuesConverterRule.INSTANCE,
+ RepeatUnionConverterRule.INSTANCE,
+ RecursiveTableScanConverterRule.INSTANCE,
LogicalScanConverterRule.INDEX_SCAN,
LogicalScanConverterRule.TABLE_SCAN,
IndexCountRule.INSTANCE,
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 b522f1a0db3..ffb42003e83 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,14 +17,18 @@
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;
@@ -34,6 +38,8 @@ import org.apache.ignite.internal.util.typedef.internal.U;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import static
org.apache.ignite.internal.processors.query.calcite.DistributedCalciteConfiguration.DFLT_RECURSIVE_CTE_ITERATION_LIMIT;
+
/**
* Planning context.
*/
@@ -56,12 +62,21 @@ 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;
/** */
private final long plannerTimeout;
+ /** Maximum number of recursive CTE iterations, or a negative value for no
limit. */
+ private int recursiveCteIterationLimit =
DFLT_RECURSIVE_CTE_ITERATION_LIMIT;
+
/**
* Private constructor, used by a builder.
*/
@@ -115,6 +130,20 @@ public final class PlanningContext implements Context {
return plannerTimeout;
}
+ /**
+ * @return Maximum number of recursive CTE iterations, or a negative value
for no limit.
+ */
+ public int recursiveCteIterationLimit() {
+ return recursiveCteIterationLimit;
+ }
+
+ /**
+ * @param recursiveCteIterationLimit Maximum number of recursive CTE
iterations, or a negative value for no limit.
+ */
+ void recursiveCteIterationLimit(int recursiveCteIterationLimit) {
+ this.recursiveCteIterationLimit = recursiveCteIterationLimit;
+ }
+
/**
* @return Schema.
*/
@@ -160,6 +189,18 @@ 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/PrepareServiceImpl.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PrepareServiceImpl.java
index 52c1bc819b9..a2f2ffdff6a 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PrepareServiceImpl.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PrepareServiceImpl.java
@@ -103,6 +103,8 @@ public class PrepareServiceImpl extends AbstractService
implements PrepareServic
assert distrCfg != null;
+
ctx.recursiveCteIterationLimit(distrCfg.recursiveCteIterationLimit());
+
String[] disbledRules = distrCfg.disabledRules();
if (!F.isEmpty(disbledRules))
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteHashIndexSpool.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteHashIndexSpool.java
index e8fb9c25f24..07d7b7b6456 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteHashIndexSpool.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteHashIndexSpool.java
@@ -29,6 +29,7 @@ import org.apache.calcite.rel.RelWriter;
import org.apache.calcite.rel.core.Spool;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
import org.apache.calcite.util.ImmutableBitSet;
import
org.apache.ignite.internal.processors.query.calcite.metadata.cost.IgniteCost;
import
org.apache.ignite.internal.processors.query.calcite.metadata.cost.IgniteCostFactory;
@@ -92,6 +93,24 @@ public class IgniteHashIndexSpool extends Spool implements
IgniteRel {
return visitor.visit(this);
}
+ /** {@inheritDoc} */
+ @Override public RelNode accept(RexShuttle shuttle) {
+ List<RexNode> newSearchRow = shuttle.apply(searchRow);
+ RexNode newCondition = shuttle.apply(cond);
+
+ if (newSearchRow == searchRow && newCondition == cond)
+ return this;
+
+ return new IgniteHashIndexSpool(
+ getCluster(),
+ getTraitSet(),
+ getInput(),
+ newSearchRow,
+ newCondition,
+ allowNulls
+ );
+ }
+
/** */
@Override public IgniteRel clone(RelOptCluster cluster, List<IgniteRel>
inputs) {
return new IgniteHashIndexSpool(cluster, getTraitSet(), inputs.get(0),
searchRow, cond, allowNulls);
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteIndexScan.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteIndexScan.java
index a7397ffcedb..0d202e154b2 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteIndexScan.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteIndexScan.java
@@ -28,6 +28,8 @@ import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.util.ImmutableBitSet;
import
org.apache.ignite.internal.processors.query.calcite.prepare.bounds.SearchBounds;
+import org.apache.ignite.internal.processors.query.calcite.schema.IgniteIndex;
+import org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
import org.jetbrains.annotations.Nullable;
import static
org.apache.ignite.internal.processors.query.calcite.trait.TraitUtils.changeTraits;
@@ -134,6 +136,20 @@ public class IgniteIndexScan extends AbstractIndexScan
implements SourceAwareIgn
return visitor.visit(this);
}
+ /** {@inheritDoc} */
+ @Override protected IgniteIndexScan copy(
+ RelTraitSet traitSet,
+ @Nullable RelDataType rowType,
+ @Nullable List<RexNode> projects,
+ @Nullable RexNode condition
+ ) {
+ IgniteIndex idx =
getTable().unwrap(IgniteTable.class).getIndex(idxName);
+ List<SearchBounds> newSearchBounds = idx.toSearchBounds(getCluster(),
condition, requiredColumns);
+
+ return new IgniteIndexScan(sourceId, getCluster(), traitSet,
getTable(), idxName, rowType, projects, condition,
+ newSearchBounds, requiredColumns, collation);
+ }
+
/** {@inheritDoc} */
@Override public IgniteRel clone(long sourceId) {
return new IgniteIndexScan(sourceId, getCluster(), getTraitSet(),
getTable(),
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
new file mode 100644
index 00000000000..d51e1bd53d1
--- /dev/null
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRecursiveTableScan.java
@@ -0,0 +1,89 @@
+/*
+ * 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.rel;
+
+import java.util.List;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.AbstractRelNode;
+import org.apache.calcite.rel.RelInput;
+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
+ ) {
+ super(cluster, traits);
+
+ this.rowType = rowType;
+ this.stateId = stateId;
+ }
+
+ /** Constructor used for deserialization. */
+ public IgniteRecursiveTableScan(RelInput input) {
+ this(
+ input.getCluster(),
+ input.getTraitSet().replace(IgniteConvention.INSTANCE),
+ input.getRowType("rowType"),
+ requireNonNull(input.getString("stateId"), "stateId")
+ );
+ }
+
+ /** 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);
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> T accept(IgniteRelVisitor<T> visitor) {
+ return visitor.visit(this);
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel clone(RelOptCluster cluster, List<IgniteRel>
inputs) {
+ assert inputs.isEmpty();
+
+ return new IgniteRecursiveTableScan(cluster, getTraitSet(), rowType,
stateId);
+ }
+
+ /** {@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/IgniteRelVisitor.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRelVisitor.java
index 456b7b46d26..e82b63001a3 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRelVisitor.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRelVisitor.java
@@ -89,6 +89,11 @@ public interface IgniteRelVisitor<T> {
*/
T visit(IgniteTableScan rel);
+ /**
+ * See {@link IgniteRelVisitor#visit(IgniteRel)}
+ */
+ T visit(IgniteRecursiveTableScan rel);
+
/**
* See {@link IgniteRelVisitor#visit(IgniteRel)}
*/
@@ -144,6 +149,11 @@ public interface IgniteRelVisitor<T> {
*/
T visit(IgniteUnionAll rel);
+ /**
+ * See {@link IgniteRelVisitor#visit(IgniteRel)}
+ */
+ T visit(IgniteRepeatUnion rel);
+
/**
* See {@link IgniteRelVisitor#visit(IgniteRel)}
*/
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
new file mode 100644
index 00000000000..0a7db70d139
--- /dev/null
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRepeatUnion.java
@@ -0,0 +1,102 @@
+/*
+ * 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.rel;
+
+import java.util.List;
+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. */
+ public IgniteRepeatUnion(RelInput input) {
+ this(
+ input.getCluster(),
+ 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;
+ }
+
+ /** {@inheritDoc} */
+ @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);
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> T accept(IgniteRelVisitor<T> visitor) {
+ return visitor.visit(this);
+ }
+
+ /** {@inheritDoc} */
+ @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);
+ }
+
+ /** Reads the optional iteration limit emitted by {@link
RepeatUnion#explainTerms(RelWriter)}. */
+ private static int iterationLimit(RelInput input) {
+ Number iterationLimit = (Number)input.get("iterationLimit");
+
+ return iterationLimit == null ? -1 : iterationLimit.intValue();
+ }
+}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteSortedIndexSpool.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteSortedIndexSpool.java
index 33d45ea51a7..654ba228663 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteSortedIndexSpool.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteSortedIndexSpool.java
@@ -30,10 +30,12 @@ import org.apache.calcite.rel.RelWriter;
import org.apache.calcite.rel.core.Spool;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
import
org.apache.ignite.internal.processors.query.calcite.externalize.RelInputEx;
import
org.apache.ignite.internal.processors.query.calcite.metadata.cost.IgniteCost;
import
org.apache.ignite.internal.processors.query.calcite.metadata.cost.IgniteCostFactory;
import
org.apache.ignite.internal.processors.query.calcite.prepare.bounds.SearchBounds;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
/**
* Relational operator that returns the sorted contents of a table
@@ -87,6 +89,38 @@ public class IgniteSortedIndexSpool extends Spool implements
IgniteRel {
return visitor.visit(this);
}
+ /** {@inheritDoc} */
+ @Override public RelNode accept(RexShuttle shuttle) {
+ boolean[] boundsChanged = {false};
+
+ List<SearchBounds> newSearchBounds = searchBounds == null ? null :
Commons.transform(searchBounds, bounds -> {
+ if (bounds == null)
+ return null;
+
+ return bounds.transform(node -> {
+ RexNode newNode = shuttle.apply(node);
+
+ boundsChanged[0] |= newNode != node;
+
+ return newNode;
+ });
+ });
+
+ RexNode newCondition = shuttle.apply(condition);
+
+ if (!boundsChanged[0] && newCondition == condition)
+ return this;
+
+ return new IgniteSortedIndexSpool(
+ getCluster(),
+ getTraitSet(),
+ getInput(),
+ collation,
+ newCondition,
+ newSearchBounds
+ );
+ }
+
/** */
@Override public IgniteRel clone(RelOptCluster cluster, List<IgniteRel>
inputs) {
return new IgniteSortedIndexSpool(cluster, getTraitSet(),
inputs.get(0), collation, condition, searchBounds);
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteTableScan.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteTableScan.java
index c35b415f197..5ca8fdc00a2 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteTableScan.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteTableScan.java
@@ -129,6 +129,17 @@ public class IgniteTableScan extends
ProjectableFilterableTableScan implements S
return visitor.visit(this);
}
+ /** {@inheritDoc} */
+ @Override protected IgniteTableScan copy(
+ RelTraitSet traitSet,
+ @Nullable RelDataType rowType,
+ @Nullable List<RexNode> projects,
+ @Nullable RexNode condition
+ ) {
+ return new IgniteTableScan(sourceId, getCluster(), traitSet,
getTable(), rowType, projects, condition,
+ requiredColumns);
+ }
+
/** {@inheritDoc} */
@Override public IgniteRel clone(long sourceId) {
return new IgniteTableScan(sourceId, getCluster(), getTraitSet(),
getTable(), rowType, projects, condition,
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/ProjectableFilterableTableScan.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/ProjectableFilterableTableScan.java
index bef05a9b5e3..ab648abb1dd 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/ProjectableFilterableTableScan.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/ProjectableFilterableTableScan.java
@@ -121,6 +121,36 @@ public abstract class ProjectableFilterableTableScan
extends TableScan {
return this;
}
+ /** {@inheritDoc} */
+ @Override public RelNode accept(RexShuttle shuttle) {
+ List<RexNode> newProjects = projects == null ? null :
shuttle.apply(projects);
+ RexNode newCondition = shuttle.apply(condition);
+
+ if (newProjects == projects && newCondition == condition)
+ return this;
+
+ RelDataType newRowType = rowType;
+
+ if (newProjects != projects) {
+ newRowType = RexUtil.createStructType(
+ getCluster().getTypeFactory(),
+ newProjects,
+ getRowType().getFieldNames(),
+ null
+ );
+ }
+
+ return copy(getTraitSet(), newRowType, newProjects, newCondition);
+ }
+
+ /** Creates a copy with transformed projects and condition. */
+ protected abstract ProjectableFilterableTableScan copy(
+ RelTraitSet traitSet,
+ @Nullable RelDataType rowType,
+ @Nullable List<RexNode> projects,
+ @Nullable RexNode condition
+ );
+
/** {@inheritDoc} */
@Override public RelWriter explainTerms(RelWriter pw) {
return explainTerms0(super.explainTerms(pw));
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/logical/IgniteLogicalIndexScan.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/logical/IgniteLogicalIndexScan.java
index c0cfa088e6f..80c6f474614 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/logical/IgniteLogicalIndexScan.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/logical/IgniteLogicalIndexScan.java
@@ -85,4 +85,14 @@ public class IgniteLogicalIndexScan extends
AbstractIndexScan {
) {
super(cluster, traits, tbl, idxName, rowType, proj, cond,
searchBounds, requiredCols);
}
+
+ /** {@inheritDoc} */
+ @Override protected IgniteLogicalIndexScan copy(
+ RelTraitSet traitSet,
+ @Nullable RelDataType rowType,
+ @Nullable List<RexNode> projects,
+ @Nullable RexNode condition
+ ) {
+ return create(getCluster(), traitSet, getTable(), idxName, rowType,
projects, condition, requiredColumns);
+ }
}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/logical/IgniteLogicalTableScan.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/logical/IgniteLogicalTableScan.java
index 1cbd7544bf5..016d909853c 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/logical/IgniteLogicalTableScan.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/logical/IgniteLogicalTableScan.java
@@ -76,4 +76,15 @@ public class IgniteLogicalTableScan extends
ProjectableFilterableTableScan {
return new IgniteLogicalTableScan(getCluster(), getTraitSet(),
getTable(), hints,
rowType, projects(), condition(), requiredColumns());
}
+
+ /** {@inheritDoc} */
+ @Override protected IgniteLogicalTableScan copy(
+ RelTraitSet traitSet,
+ @Nullable RelDataType rowType,
+ @Nullable List<RexNode> projects,
+ @Nullable RexNode condition
+ ) {
+ return new IgniteLogicalTableScan(getCluster(), traitSet, getTable(),
getHints(), rowType, projects, condition,
+ requiredColumns);
+ }
}
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
new file mode 100644
index 00000000000..395d85b38d1
--- /dev/null
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveCteUtils.java
@@ -0,0 +1,218 @@
+/*
+ * 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.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;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.plan.volcano.RelSubset;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Aggregate;
+import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+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;
+
+import static
org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions.single;
+
+/** Utilities shared by recursive CTE converter rules. */
+final class RecursiveCteUtils {
+ /** */
+ private RecursiveCteUtils() {
+ // No-op.
+ }
+
+ /** Returns whether the table is Calcite's query-local transient table. */
+ static boolean isTransient(RelOptTable table) {
+ 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);
+
+ int cnt = isRecursiveScan(rel, table) ? 1 : 0;
+
+ for (RelNode input : rel.getInputs())
+ cnt += referenceCount(input, table);
+
+ return cnt;
+ }
+
+ /** Converts maximal inputs that do not depend on the current delta to
coordinator-local rewindable execution. */
+ static RelNode convertStaticInputs(RelNode rel, RelOptTable table,
RelTraitSet traits) {
+ rel = original(rel);
+
+ if (isRecursiveScan(rel, table))
+ return rel;
+
+ List<RelNode> inputs = rel.getInputs();
+
+ if (inputs.isEmpty())
+ return rel;
+
+ List<RelNode> newInputs = new ArrayList<>(inputs.size());
+
+ for (RelNode input : inputs) {
+ if (referenceCount(input, table) == 0 && isInvariant(input))
+ newInputs.add(convertStaticInput(input, traits));
+ else
+ newInputs.add(convertStaticInputs(input, table, traits));
+ }
+
+ return rel.copy(rel.getTraitSet(), newInputs);
+ }
+
+ /** Converts an iteration-independent input to coordinator-local
rewindable execution. */
+ private static RelNode convertStaticInput(RelNode input, RelTraitSet
traits) {
+ IgniteDistribution inputDistribution =
+
(IgniteDistribution)input.getCluster().getMetadataQuery().distribution(original(input));
+
+ if (inputDistribution.satisfies(single()))
+ return RelOptRule.convert(input,
traits.replace(RewindabilityTrait.REWINDABLE));
+
+ RelNode convertedInput = RelOptRule.convert(input, traits);
+
+ return TraitUtils.convertRewindability(
+ input.getCluster().getPlanner(),
+ RewindabilityTrait.REWINDABLE,
+ convertedInput
+ );
+ }
+
+ /** Returns whether the subtree produces the same result on every
recursive iteration. */
+ private static boolean isInvariant(RelNode rel) {
+ rel = original(rel);
+
+ if (!RelOptUtil.getVariablesUsed(rel).isEmpty())
+ return false;
+
+ DeterminismChecker checker = new DeterminismChecker();
+
+ rel.accept(checker);
+
+ if (!checker.deterministic)
+ return false;
+
+ if (rel instanceof Aggregate) {
+ for (AggregateCall call : ((Aggregate)rel).getAggCallList()) {
+ if (!call.getAggregation().isDeterministic())
+ return false;
+ }
+ }
+
+ for (RelNode input : rel.getInputs()) {
+ if (!isInvariant(input))
+ return false;
+ }
+
+ return true;
+ }
+
+ /**
+ * Returns the logical expression represented by a Volcano subset.
Converter rule inputs can be subsets whose own
+ * input list is empty.
+ */
+ static RelNode original(RelNode rel) {
+ while (rel instanceof RelSubset) {
+ RelNode original = ((RelSubset)rel).getOriginal();
+
+ if (original == null)
+ return rel;
+
+ rel = original;
+ }
+
+ return rel;
+ }
+
+ /** Returns whether both optimizer tables represent the same transient
table instance. */
+ static boolean sameTransientTable(RelOptTable first, RelOptTable second) {
+ TransientTable firstTable = first == null ? null :
first.unwrap(TransientTable.class);
+ TransientTable secondTable = second == null ? null :
second.unwrap(TransientTable.class);
+
+ return firstTable != null && firstTable == secondTable;
+ }
+
+ /** */
+ private static boolean isRecursiveScan(RelNode rel, RelOptTable table) {
+ return rel instanceof TableScan
+ && sameTransientTable(((TableScan)rel).getTable(), table);
+ }
+
+ /** Finds non-deterministic expressions in one relational node. */
+ private static class DeterminismChecker extends RexShuttle {
+ /** Whether all visited expressions are deterministic. */
+ private boolean deterministic = true;
+
+ /** {@inheritDoc} */
+ @Override public RexNode visitCall(RexCall call) {
+ if (!call.getOperator().isDeterministic() ||
isNonDeterministicUdf(call)) {
+ deterministic = false;
+
+ return call;
+ }
+
+ return super.visitCall(call);
+ }
+
+ /** {@inheritDoc} */
+ @Override public RexNode visitSubQuery(RexSubQuery subQuery) {
+ if (!isInvariant(subQuery.rel)) {
+ deterministic = false;
+
+ return subQuery;
+ }
+
+ return super.visitSubQuery(subQuery);
+ }
+
+ /** Returns whether the call targets an Ignite UDF declared as
non-deterministic. */
+ private static boolean isNonDeterministicUdf(RexCall call) {
+ if (!(call.getOperator() instanceof SqlUserDefinedFunction))
+ return false;
+
+ Object function =
((SqlUserDefinedFunction)call.getOperator()).getFunction();
+
+ return function instanceof IgniteScalarFunction
+ && !((IgniteScalarFunction)function).isDeterministic();
+ }
+ }
+}
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
new file mode 100644
index 00000000000..63c0e01401f
--- /dev/null
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveTableScanConverterRule.java
@@ -0,0 +1,63 @@
+/*
+ * 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.rule;
+
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.PhysicalNode;
+import org.apache.calcite.rel.logical.LogicalTableScan;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteConvention;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRecursiveTableScan;
+import
org.apache.ignite.internal.processors.query.calcite.trait.RewindabilityTrait;
+import org.jetbrains.annotations.Nullable;
+
+import static
org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions.single;
+
+/** Converts a scan of the recursive transient table to a query-local delta
scan. */
+public class RecursiveTableScanConverterRule extends
AbstractIgniteConverterRule<LogicalTableScan> {
+ /** Instance. */
+ public static final RelOptRule INSTANCE = new
RecursiveTableScanConverterRule();
+
+ /** */
+ private RecursiveTableScanConverterRule() {
+ super(LogicalTableScan.class, "RecursiveTableScanConverterRule");
+ }
+
+ /** {@inheritDoc} */
+ @Override protected @Nullable PhysicalNode convert(
+ RelOptPlanner planner,
+ RelMetadataQuery mq,
+ LogicalTableScan rel
+ ) {
+ if (!RecursiveCteUtils.isTransient(rel.getTable()))
+ return null;
+
+ RelTraitSet traits =
rel.getCluster().traitSetOf(IgniteConvention.INSTANCE)
+ .replace(single())
+ .replace(RewindabilityTrait.REWINDABLE);
+
+ return new IgniteRecursiveTableScan(
+ rel.getCluster(),
+ traits,
+ rel.getRowType(),
+ RecursiveCteUtils.stateId(planner, rel.getTable())
+ );
+ }
+}
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
new file mode 100644
index 00000000000..d53542ea5ac
--- /dev/null
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RepeatUnionConverterRule.java
@@ -0,0 +1,102 @@
+/*
+ * 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.rule;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.PhysicalNode;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Spool;
+import org.apache.calcite.rel.logical.LogicalRepeatUnion;
+import org.apache.calcite.rel.logical.LogicalTableSpool;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
+import org.apache.ignite.internal.processors.query.IgniteSQLException;
+import
org.apache.ignite.internal.processors.query.calcite.prepare.PlanningContext;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteConvention;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRepeatUnion;
+
+import static
org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions.single;
+
+/** Converts Calcite's logical recursive union to coordinator-side execution.
*/
+public class RepeatUnionConverterRule extends
AbstractIgniteConverterRule<LogicalRepeatUnion> {
+ /** Instance. */
+ public static final RelOptRule INSTANCE = new RepeatUnionConverterRule();
+
+ /** */
+ private RepeatUnionConverterRule() {
+ super(LogicalRepeatUnion.class, "RecursiveCteConverterRule");
+ }
+
+ /** {@inheritDoc} */
+ @Override protected PhysicalNode convert(RelOptPlanner planner,
RelMetadataQuery mq, LogicalRepeatUnion rel) {
+ RelOptTable table = rel.getTransientTable();
+
+ if (table == null || !RecursiveCteUtils.isTransient(table))
+ throw unsupported("a transient table is required");
+
+ // TODO: IGNITE-29012 Support recursive CTE with UNION DISTINCT.
+ 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");
+ RelNode iterative = unwrapSpool(rel.getIterativeRel(), "recursive
term");
+
+ RelOptCluster cluster = rel.getCluster();
+ RelTraitSet traits =
cluster.traitSetOf(IgniteConvention.INSTANCE).replace(single());
+ iterative = RecursiveCteUtils.convertStaticInputs(iterative, table,
traits);
+
+ return new IgniteRepeatUnion(
+ cluster,
+ traits,
+ convert(seed, traits),
+ convert(iterative, traits),
+ stateId,
+ iterationLimit
+ );
+ }
+
+ /** */
+ private static RelNode unwrapSpool(RelNode rel, String term) {
+ rel = RecursiveCteUtils.original(rel);
+
+ if (!(rel instanceof LogicalTableSpool))
+ throw unsupported("the " + term + " must use a transient table
spool");
+
+ LogicalTableSpool spool = (LogicalTableSpool)rel;
+
+ if (spool.readType != Spool.Type.LAZY || spool.writeType !=
Spool.Type.LAZY)
+ throw unsupported("only lazy transient table spools are
supported");
+
+ return spool.getInput();
+ }
+
+ /** */
+ private static IgniteSQLException unsupported(String detail) {
+ return new IgniteSQLException(
+ "Unsupported recursive CTE: " + detail,
+ IgniteQueryErrorCode.UNSUPPORTED_OPERATION
+ );
+ }
+}
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorPropertiesTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorPropertiesTest.java
index 71b16cc4536..bd0f3646d74 100644
---
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorPropertiesTest.java
+++
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorPropertiesTest.java
@@ -38,8 +38,11 @@ import org.junit.Test;
import static
org.apache.ignite.internal.processors.query.QueryParserMetricsHolder.QUERY_PARSER_METRIC_GROUP_NAME;
import static
org.apache.ignite.internal.processors.query.calcite.DistributedCalciteConfiguration.DFLT_PLAN_CACHE_SIZE;
+import static
org.apache.ignite.internal.processors.query.calcite.DistributedCalciteConfiguration.DFLT_RECURSIVE_CTE_ITERATION_LIMIT;
+import static
org.apache.ignite.internal.processors.query.calcite.DistributedCalciteConfiguration.RECURSIVE_CTE_ITERATION_LIMIT_PROPERTY_NAME;
import static
org.apache.ignite.internal.processors.query.calcite.QueryChecker.containsIndexScan;
import static
org.apache.ignite.internal.processors.query.calcite.QueryChecker.containsSubPlan;
+import static
org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause;
import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
import static org.hamcrest.CoreMatchers.not;
@@ -54,6 +57,12 @@ public class CalciteQueryProcessorPropertiesTest extends
AbstractBasicIntegratio
@Override protected void afterTest() throws Exception {
changeDistributedProperty(DistributedCalciteConfiguration.DISABLED_RULES_PROPERTY_NAME,
" ",
pVal -> F.compareArrays(pVal, new String[0]) == 0);
+
+ changeDistributedProperty(
+ RECURSIVE_CTE_ITERATION_LIMIT_PROPERTY_NAME,
+ String.valueOf(DFLT_RECURSIVE_CTE_ITERATION_LIMIT),
+ pVal -> pVal.equals(DFLT_RECURSIVE_CTE_ITERATION_LIMIT)
+ );
}
/** */
@@ -212,6 +221,49 @@ public class CalciteQueryProcessorPropertiesTest extends
AbstractBasicIntegratio
checkPlanCacheSize(grid(1), 100);
}
+ /** */
+ @Test
+ public void testRecursiveCteIterationLimit() throws Exception {
+ for (Ignite ig : G.allGrids()) {
+ DistributedChangeableProperty<Integer> prop =
+ distributedProperty(ig,
RECURSIVE_CTE_ITERATION_LIMIT_PROPERTY_NAME);
+
+ assertNotNull(prop);
+ assertEquals(DFLT_RECURSIVE_CTE_ITERATION_LIMIT,
prop.get().intValue());
+ }
+
+ String qry = "WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "SELECT n + 1 FROM numbers WHERE n < 5" +
+ ") " +
+ "SELECT n FROM numbers";
+
+ // Populate the plan cache with the default iteration limit.
+ assertQuery(qry)
+ .returns(1)
+ .returns(2)
+ .returns(3)
+ .returns(4)
+ .returns(5)
+ .check();
+
+ changeDistributedProperty(
+ RECURSIVE_CTE_ITERATION_LIMIT_PROPERTY_NAME,
+ "2",
+ pVal -> pVal.equals(2)
+ );
+
+ // The property listener must invalidate the cached plan and apply the
new limit to the same query.
+ assertThrowsAnyCause(
+ log,
+ () -> sql(qry),
+ IgniteSQLException.class,
+ "Recursive CTE iteration limit exceeded [limit=2, property=" +
+ RECURSIVE_CTE_ITERATION_LIMIT_PROPERTY_NAME + ']'
+ );
+ }
+
/** */
private void checkPlanCacheSize(IgniteEx grid, int expectedCacheSize) {
MetricRegistryImpl mreg =
grid.context().metric().registry(QUERY_PARSER_METRIC_GROUP_NAME);
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/MemoryQuotasIntegrationTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/MemoryQuotasIntegrationTest.java
index 2d905e457d7..ca271177986 100644
---
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/MemoryQuotasIntegrationTest.java
+++
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/MemoryQuotasIntegrationTest.java
@@ -243,6 +243,24 @@ public class MemoryQuotasIntegrationTest extends
AbstractBasicIntegrationTest {
IgniteSQLException.class, "Query quota exceeded");
}
+ /** */
+ @Test
+ public void testRecursiveDeltaIsAccountedForMemoryQuota() {
+ assertThrows(
+ "WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "SELECT n + 1 " +
+ "FROM numbers " +
+ "CROSS JOIN (VALUES (1), (2)) AS fanout(x) " +
+ "WHERE n < 19" +
+ ") " +
+ "SELECT COUNT(*) FROM numbers",
+ IgniteSQLException.class,
+ "Query quota exceeded"
+ );
+ }
+
/** */
@Test
public void testRightMeterializedJoins() {
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 ff77e4f1e0f..11a333e30fa 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
@@ -17,44 +17,216 @@
package org.apache.ignite.internal.processors.query.calcite.integration;
-import org.apache.calcite.plan.RelOptPlanner;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.cache.query.annotations.QuerySqlFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
-import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.internal.processors.query.calcite.QueryChecker;
import org.junit.Test;
/**
* Integration tests for recursive common table expressions.
*/
public class RecursiveCteIntegrationTest extends AbstractBasicIntegrationTest {
+ /** Number of invocations of a non-deterministic function. */
+ private static final AtomicInteger nonDeterministicCallCnt = new
AtomicInteger();
+
+ /** */
+ @Test
+ public void testEmployeeHierarchy() {
+ createEmployeeTable();
+
+ assertQuery("WITH RECURSIVE employee_hierarchy (id, manager_id, name,
depth) AS (" +
+ "SELECT id, manager_id, name, 0 FROM employee WHERE manager_id IS
NULL " +
+ "UNION ALL " +
+ "SELECT e.id, e.manager_id, e.name, h.depth + 1 " +
+ "FROM employee e " +
+ "JOIN employee_hierarchy h ON e.manager_id = h.id" +
+ ") " +
+ "SELECT id, manager_id, name, depth FROM employee_hierarchy ORDER
BY depth, id")
+ .returns(1, null, "CEO", 0)
+ .returns(2, 1, "Manager", 1)
+ .returns(4, 1, "Accountant", 1)
+ .returns(3, 2, "Developer", 2)
+ .check();
+ }
+
/** */
@Test
- public void testHierarchicalQueryIsNotSupported() {
+ public void testRecursionStopsWhenDeltaIsEmpty() {
sql("CREATE TABLE employee (id INT PRIMARY KEY, manager_id INT, name
VARCHAR)");
- sql("INSERT INTO employee VALUES " +
- "(1, NULL, 'CEO'), " +
- "(2, 1, 'Manager'), " +
- "(3, 2, 'Developer'), " +
- "(4, 1, 'Accountant')");
+ sql("INSERT INTO employee VALUES (1, NULL, 'CEO')");
- String qry = "WITH RECURSIVE employee_hierarchy (id, manager_id, name,
depth) AS (" +
+ assertQuery("WITH RECURSIVE employee_hierarchy (id, manager_id, name,
depth) AS (" +
"SELECT id, manager_id, name, 0 FROM employee WHERE manager_id IS
NULL " +
"UNION ALL " +
"SELECT e.id, e.manager_id, e.name, h.depth + 1 " +
"FROM employee e " +
"JOIN employee_hierarchy h ON e.manager_id = h.id" +
") " +
- "SELECT id, manager_id, name, depth FROM employee_hierarchy ORDER
BY depth, id";
+ "SELECT id, manager_id, name, depth FROM employee_hierarchy")
+ .returns(1, null, "CEO", 0)
+ .check();
+ }
+
+ /** */
+ @Test
+ public void testRecursiveTermIsNotExecutedWhenSeedIsEmpty() {
+ sql("CREATE TABLE empty_seed (n INT PRIMARY KEY)");
+
+ assertQuery("WITH RECURSIVE numbers(n) AS (" +
+ "SELECT n FROM empty_seed " +
+ "UNION ALL " +
+ "SELECT v.n FROM numbers RIGHT JOIN (VALUES (42)) v(n) ON TRUE" +
+ ") " +
+ "SELECT n FROM numbers FETCH FIRST 1 ROW ONLY")
+ .resultSize(0)
+ .check();
+ }
+
+ /** */
+ @Test
+ public void testRecursiveTermWithoutSelfReferenceAfterOptimization() {
+ assertQuery("WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "SELECT n + 1 FROM numbers WHERE FALSE" +
+ ") " +
+ "SELECT n FROM numbers")
+ .returns(1)
+ .check();
+ }
+
+ /** */
+ @Test
+ public void testRecursiveTermWithMultipleSelfReferences() {
+ assertQuery("WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "SELECT left_numbers.n + 1 " +
+ "FROM numbers left_numbers " +
+ "JOIN numbers right_numbers ON left_numbers.n =
right_numbers.n " +
+ "WHERE left_numbers.n < 3" +
+ ") " +
+ "SELECT n FROM numbers")
+ .returns(1)
+ .returns(2)
+ .returns(3)
+ .check();
+ }
- Throwable err = GridTestUtils.assertThrows(
- log,
- () -> sql(qry),
+ /** */
+ @Test
+ public void testRecursiveCteWithMultipleRecursiveBranches() {
+ assertQuery("WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "(" +
+ "SELECT n + 1 FROM numbers WHERE n < 3 " +
+ "UNION ALL " +
+ "SELECT n + 10 FROM numbers WHERE n < 3" +
+ ")" +
+ ") " +
+ "SELECT n FROM numbers")
+ .returns(1)
+ .returns(2)
+ .returns(11)
+ .returns(3)
+ .returns(12)
+ .check();
+ }
+
+ /** */
+ @Test
+ public void testRecursiveCteWithDistinctUnionIsRejected() {
+ assertThrows(
+ "WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION " +
+ "SELECT n + 1 FROM numbers WHERE n < 3" +
+ ") " +
+ "SELECT n FROM numbers",
IgniteSQLException.class,
- "Failed to plan query"
+ "only UNION ALL is supported"
);
+ }
+
+ /** */
+ @Test
+ public void testStateIsIsolatedBetweenSameNamedRecursiveCtes() {
+ assertQuery("SELECT /*+ MERGE_JOIN */ l.n, r.n " +
+ "FROM (" +
+ "WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "SELECT n + 1 FROM numbers WHERE n < 3" +
+ ") " +
+ "SELECT n, n + 9 AS join_key FROM numbers" +
+ ") l " +
+ "JOIN (" +
+ "WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 10 " +
+ "UNION ALL " +
+ "SELECT n + 1 FROM numbers WHERE n < 12" +
+ ") " +
+ "SELECT n, n - 9 AS join_key FROM numbers" +
+ ") r ON l.join_key = r.n")
+ .matches(QueryChecker.containsSubPlan("IgniteMergeJoin"))
+ .returns(1, 10)
+ .returns(2, 11)
+ .returns(3, 12)
+ .check();
+ }
+
+ /** */
+ @Test
+ public void
testIndependentNonDeterministicSubtreeIsEvaluatedForEveryIteration() {
+ registerRecursiveFunctions();
+
+ nonDeterministicCallCnt.set(0);
+
+ String qry = "WITH RECURSIVE numbers(n, marker) AS (" +
+ "SELECT 1, 0 " +
+ "UNION ALL " +
+ "SELECT n + 1, v.marker " +
+ "FROM numbers " +
+ "CROSS JOIN (SELECT nextRecursiveValue() AS marker) v " +
+ "WHERE n < 4" +
+ ") " +
+ "SELECT n, marker FROM numbers ORDER BY n";
+
+ assertQuery(qry)
+ .returns(1, 0)
+ .returns(2, 1)
+ .returns(3, 2)
+ .returns(4, 3)
+ .check();
+ }
+
+ /** SQL functions used by recursive CTE tests. */
+ public static class RecursiveFunctions {
+ /** Returns a different value on every invocation. */
+ @QuerySqlFunction(deterministic = false)
+ public static int nextRecursiveValue() {
+ return nonDeterministicCallCnt.incrementAndGet();
+ }
+ }
+
+ /** */
+ private void createEmployeeTable() {
+ sql("CREATE TABLE employee (id INT PRIMARY KEY, manager_id INT, name
VARCHAR)");
+
+ sql("INSERT INTO employee VALUES " +
+ "(1, NULL, 'CEO'), " +
+ "(2, 1, 'Manager'), " +
+ "(3, 2, 'Developer'), " +
+ "(4, 1, 'Accountant')");
+ }
- assertEquals(1, err.getSuppressed().length);
- assertTrue(err.getSuppressed()[0] instanceof
RelOptPlanner.CannotPlanException);
- assertTrue(err.getSuppressed()[0].getMessage().contains(
- "There are not enough rules to produce a node with desired
properties"));
+ /** Registers SQL functions used by recursive CTE tests. */
+ private void registerRecursiveFunctions() {
+ client.getOrCreateCache(new CacheConfiguration<Integer,
Integer>("recursive_functions")
+ .setSqlSchema("PUBLIC")
+ .setSqlFunctionClasses(RecursiveFunctions.class));
}
}
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/RecursiveCtePlannerTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/RecursiveCtePlannerTest.java
new file mode 100644
index 00000000000..658c1431bda
--- /dev/null
+++
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/RecursiveCtePlannerTest.java
@@ -0,0 +1,229 @@
+/*
+ * 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.planner;
+
+import org.apache.calcite.rel.core.Exchange;
+import org.apache.calcite.rel.core.Spool;
+import org.apache.calcite.sql.type.SqlTypeName;
+import
org.apache.ignite.internal.processors.query.calcite.exec.exp.IgniteScalarFunction;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexScan;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRecursiveTableScan;
+import
org.apache.ignite.internal.processors.query.calcite.rel.IgniteRepeatUnion;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteUnionAll;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteValues;
+import org.apache.ignite.internal.processors.query.calcite.schema.IgniteSchema;
+import
org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
+import
org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
+import org.junit.Test;
+
+/** Planner tests for recursive common table expressions. */
+public class RecursiveCtePlannerTest extends AbstractPlannerTest {
+ /** Employee hierarchy query used for distribution and index planning
checks. */
+ private static final String EMPLOYEE_HIERARCHY_QUERY =
+ "WITH RECURSIVE employee_hierarchy (id, manager_id, depth) AS (" +
+ "SELECT id, manager_id, 0 FROM employee WHERE manager_id IS NULL "
+
+ "UNION ALL " +
+ "SELECT e.id, e.manager_id, h.depth + 1 " +
+ "FROM employee e " +
+ "JOIN employee_hierarchy h ON e.manager_id = h.id" +
+ ") " +
+ "SELECT id, manager_id, depth FROM employee_hierarchy";
+
+ /** Employee hierarchy query that requests indexed correlated lookups in
the recursive term. */
+ private static final String INDEXED_EMPLOYEE_HIERARCHY_QUERY =
+ EMPLOYEE_HIERARCHY_QUERY
+ .replace("SELECT e.id", "SELECT /*+ CNL_JOIN */ e.id")
+ .replace("FROM employee e", "FROM employee /*+
FORCE_INDEX(EMPLOYEE_MANAGER_IDX) */ e");
+
+ /** Checks the physical operators used to maintain the recursive delta. */
+ @Test
+ public void testRecursiveDeltaPlan() throws Exception {
+ IgniteSchema schema = new IgniteSchema(DEFAULT_SCHEMA);
+
+ String sql =
+ "WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 " +
+ "UNION ALL " +
+ "SELECT n + 1 FROM numbers WHERE n < 3" +
+ ") " +
+ "SELECT n FROM numbers";
+
+ assertPlan(sql, schema, isInstanceOf(IgniteRepeatUnion.class)
+ .and(hasDistribution(IgniteDistributions.single()))
+ .and(input(0, isInstanceOf(IgniteValues.class)))
+ .and(input(1,
hasChildThat(isInstanceOf(IgniteRecursiveTableScan.class))))
+ );
+ }
+
+ /** A replicated source can be read on the coordinator without an
exchange. */
+ @Test
+ public void testRecursiveCteWithReplicatedTable() throws Exception {
+ IgniteSchema schema = hierarchySchema(IgniteDistributions.broadcast(),
false);
+
+ assertPlan(EMPLOYEE_HIERARCHY_QUERY, schema,
isInstanceOf(IgniteRepeatUnion.class)
+ .and(hasDistribution(IgniteDistributions.single()))
+ .and(hasChildThat(isInstanceOf(Exchange.class)).negate())
+ );
+ }
+
+ /** A partitioned source has to be transferred to the coordinator-side
recursive plan. */
+ @Test
+ public void testRecursiveCteWithPartitionedTable() throws Exception {
+ IgniteDistribution distribution = IgniteDistributions.affinity(0,
"EMPLOYEE", "hash");
+ IgniteSchema schema = hierarchySchema(distribution, false);
+
+ assertPlan(EMPLOYEE_HIERARCHY_QUERY, schema,
isInstanceOf(IgniteRepeatUnion.class)
+ .and(hasDistribution(IgniteDistributions.single()))
+ .and(input(1, hasChildThat(isInstanceOf(Spool.class)
+ .and(hasChildThat(isInstanceOf(Exchange.class))))))
+ );
+ }
+
+ /** A replicated indexed input can be rewound without materialization. */
+ @Test
+ public void testRecursiveCteWithReplicatedIndexedTable() throws Exception {
+ IgniteSchema schema = hierarchySchema(IgniteDistributions.broadcast(),
true);
+
+ assertPlan(INDEXED_EMPLOYEE_HIERARCHY_QUERY, schema,
isInstanceOf(IgniteRepeatUnion.class)
+ .and(hasDistribution(IgniteDistributions.single()))
+ .and(input(1, hasChildThat(isInstanceOf(IgniteIndexScan.class))))
+ .and(input(1, hasChildThat(isInstanceOf(Spool.class)).negate()))
+ );
+ }
+
+ /** A non-deterministic projection in a table scan must be evaluated on
every iteration. */
+ @Test
+ public void testNonDeterministicTableScanIsNotMaterialized() throws
Exception {
+ IgniteSchema schema = recursiveMarkersSchema(false);
+
+ String sql = "WITH RECURSIVE numbers(n, marker) AS (" +
+ "SELECT 1, 0 " +
+ "UNION ALL " +
+ "SELECT n + 1, v.marker " +
+ "FROM numbers " +
+ "CROSS JOIN (" +
+ "SELECT nextRecursiveValue() AS marker FROM recursive_markers"
+
+ ") v " +
+ "WHERE n < 4" +
+ ") " +
+ "SELECT n, marker FROM numbers";
+
+ assertPlan(sql, schema, isInstanceOf(IgniteRepeatUnion.class)
+ .and(input(1, hasChildThat(isInstanceOf(IgniteTableScan.class)
+ .and(scan -> scan.projects() != null)
+ .and(scan ->
scan.projects().toString().contains("NEXTRECURSIVEVALUE")))))
+ .and(input(1, hasChildThat(isInstanceOf(Spool.class)).negate()))
+ );
+ }
+
+ /** A non-deterministic condition in an index scan must be evaluated on
every iteration. */
+ @Test
+ public void testNonDeterministicIndexScanIsNotMaterialized() throws
Exception {
+ IgniteSchema schema = recursiveMarkersSchema(true);
+
+ String sql = "WITH RECURSIVE numbers(n, marker) AS (" +
+ "SELECT 1, 0 " +
+ "UNION ALL " +
+ "SELECT n + 1, " +
+ "(SELECT marker FROM recursive_markers /*+ FORCE_INDEX */ " +
+ "WHERE id = numbers.n AND nextRecursiveValue() > 0) " +
+ "FROM numbers " +
+ "WHERE n < 4" +
+ ") " +
+ "SELECT n, marker FROM numbers";
+
+ assertPlan(sql, schema, isInstanceOf(IgniteRepeatUnion.class)
+ .and(input(1, hasChildThat(isInstanceOf(IgniteIndexScan.class)
+ .and(scan -> scan.condition() != null)
+ .and(scan ->
scan.condition().toString().contains("NEXTRECURSIVEVALUE")))))
+ .and(input(1, hasChildThat(isInstanceOf(Spool.class)).negate()))
+ );
+ }
+
+ /** Calcite places multiple non-recursive branches into the seed input of
RepeatUnion. */
+ @Test
+ public void testRecursiveCteWithMultipleSeedBranches() throws Exception {
+ IgniteSchema schema = createSchema(
+ createTable("T", IgniteDistributions.single(), "ID",
SqlTypeName.INTEGER)
+ );
+
+ String sql =
+ "WITH RECURSIVE numbers(n) AS (" +
+ "SELECT ID FROM T WHERE ID = 1 " +
+ "UNION ALL " +
+ "SELECT ID FROM T WHERE ID = 2 " +
+ "UNION ALL " +
+ "SELECT n + 1 FROM numbers WHERE n < 3" +
+ ") " +
+ "SELECT n FROM numbers";
+
+ assertPlan(sql, schema, isInstanceOf(IgniteRepeatUnion.class)
+ .and(hasDistribution(IgniteDistributions.single()))
+ .and(input(0, isInstanceOf(IgniteUnionAll.class)
+ .and(input(0, isTableScan("T")))
+ .and(input(1, isTableScan("T")))))
+ );
+ }
+
+ /** Creates an employee table with the requested distribution and optional
manager index. */
+ private static IgniteSchema hierarchySchema(IgniteDistribution
distribution, boolean withManagerIdx) {
+ TestTable table = createTable(
+ "EMPLOYEE",
+ distribution,
+ "ID", Integer.class,
+ "MANAGER_ID", Integer.class
+ );
+
+ if (withManagerIdx)
+ table.addIndex("EMPLOYEE_MANAGER_IDX", 1);
+
+ return createSchema(table);
+ }
+
+ /** Creates a replicated table and registers a non-deterministic function
used by scan tests. */
+ private static IgniteSchema recursiveMarkersSchema(boolean withIndex)
throws NoSuchMethodException {
+ TestTable table = createTable(
+ "RECURSIVE_MARKERS",
+ IgniteDistributions.broadcast(),
+ "ID", SqlTypeName.INTEGER,
+ "MARKER", SqlTypeName.INTEGER
+ );
+
+ if (withIndex)
+ table.addIndex("RECURSIVE_MARKERS_ID_IDX", 0);
+
+ IgniteSchema schema = createSchema(table);
+
+ schema.addFunction(
+ "NEXTRECURSIVEVALUE",
+ IgniteScalarFunction.create(
+ RecursiveCtePlannerTest.class.getMethod("nextRecursiveValue"),
+ false
+ )
+ );
+
+ return schema;
+ }
+
+ /** Function used only to build a non-deterministic expression in planner
tests. */
+ public static int nextRecursiveValue() {
+ return 1;
+ }
+
+}
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/testsuites/PlannerTestSuite.java
b/modules/calcite/src/test/java/org/apache/ignite/testsuites/PlannerTestSuite.java
index be2d0283c15..a04f5c1becc 100644
---
a/modules/calcite/src/test/java/org/apache/ignite/testsuites/PlannerTestSuite.java
+++
b/modules/calcite/src/test/java/org/apache/ignite/testsuites/PlannerTestSuite.java
@@ -39,6 +39,7 @@ import
org.apache.ignite.internal.processors.query.calcite.planner.PlanExecution
import
org.apache.ignite.internal.processors.query.calcite.planner.PlanSplitterTest;
import
org.apache.ignite.internal.processors.query.calcite.planner.PlannerTimeoutTest;
import
org.apache.ignite.internal.processors.query.calcite.planner.ProjectFilterScanMergePlannerTest;
+import
org.apache.ignite.internal.processors.query.calcite.planner.RecursiveCtePlannerTest;
import
org.apache.ignite.internal.processors.query.calcite.planner.RexSimplificationPlannerTest;
import
org.apache.ignite.internal.processors.query.calcite.planner.SerializationPlannerTest;
import
org.apache.ignite.internal.processors.query.calcite.planner.SetOpPlannerTest;
@@ -91,6 +92,7 @@ import org.junit.runners.Suite;
InlineIndexScanPlannerTest.class,
UserDefinedViewsPlannerTest.class,
RexSimplificationPlannerTest.class,
+ RecursiveCtePlannerTest.class,
SerializationPlannerTest.class,
UncollectPlannerTest.class,
WindowPlannerTest.class,
diff --git
a/modules/calcite/src/test/sql/hierarchy/test_recursive_hierarchy.test
b/modules/calcite/src/test/sql/hierarchy/test_recursive_hierarchy.test
new file mode 100644
index 00000000000..986e28ec98d
--- /dev/null
+++ b/modules/calcite/src/test/sql/hierarchy/test_recursive_hierarchy.test
@@ -0,0 +1,64 @@
+# name: test/sql/hierarchy/test_recursive_hierarchy.test
+# description: Test recursive traversal of parent-child rows
+# group: [hierarchy]
+
+statement ok
+CREATE TABLE nodes (id INTEGER PRIMARY KEY, parent_id INTEGER);
+
+statement ok
+INSERT INTO nodes VALUES (1, NULL), (2, 1), (3, 1), (4, 2), (5, 2), (6, 3),
(10, NULL), (11, 10);
+
+# Traverse all trees. The seed has two roots and the recursive term joins a
base table.
+query III
+WITH RECURSIVE tree(id, parent_id, depth) AS (
+ SELECT id, parent_id, 0
+ FROM nodes
+ WHERE parent_id IS NULL
+ UNION ALL
+ SELECT n.id, n.parent_id, t.depth + 1
+ FROM nodes n
+ JOIN tree t ON n.parent_id = t.id
+)
+SELECT id, parent_id, depth FROM tree ORDER BY id;
+----
+1 NULL 0
+2 1 1
+3 1 1
+4 2 2
+5 2 2
+6 3 2
+10 NULL 0
+11 10 1
+
+# Start from an ordinary row rather than a root and stop when no next rows
exist.
+query III
+WITH RECURSIVE tree(id, parent_id, depth) AS (
+ SELECT id, parent_id, 0
+ FROM nodes
+ WHERE id = 2
+ UNION ALL
+ SELECT n.id, n.parent_id, t.depth + 1
+ FROM nodes n
+ JOIN tree t ON n.parent_id = t.id
+)
+SELECT id, parent_id, depth FROM tree ORDER BY id;
+----
+2 1 0
+4 2 1
+5 2 1
+
+# An empty seed produces an empty result and does not start an iteration.
+query I
+WITH RECURSIVE tree(id) AS (
+ SELECT id
+ FROM nodes
+ WHERE id < 0
+ UNION ALL
+ SELECT n.id
+ FROM nodes n
+ JOIN tree t ON n.parent_id = t.id
+)
+SELECT id FROM tree;
+----
+
+# End of test.
diff --git
a/modules/calcite/src/test/sql/hierarchy/test_recursive_sequence.test
b/modules/calcite/src/test/sql/hierarchy/test_recursive_sequence.test
new file mode 100644
index 00000000000..6d8e15c91a6
--- /dev/null
+++ b/modules/calcite/src/test/sql/hierarchy/test_recursive_sequence.test
@@ -0,0 +1,62 @@
+# name: test/sql/hierarchy/test_recursive_sequence.test
+# description: Test recursive CTEs used to generate integer sequences
+# group: [hierarchy]
+
+# Fixed recursion bound.
+query I
+WITH RECURSIVE numbers(n) AS (
+ SELECT 1
+ UNION ALL
+ SELECT n + 1
+ FROM numbers
+ WHERE n < 5
+)
+SELECT n FROM numbers ORDER BY n;
+----
+1
+2
+3
+4
+5
+
+statement ok
+CREATE TABLE ranges (id INTEGER PRIMARY KEY, last_value INTEGER);
+
+statement ok
+INSERT INTO ranges VALUES (1, 3), (2, 2);
+
+# The seed contains several rows and every row has its own recursion bound.
+query II
+WITH RECURSIVE numbers(id, n, last_value) AS (
+ SELECT id, 1, last_value
+ FROM ranges
+ UNION ALL
+ SELECT id, n + 1, last_value
+ FROM numbers
+ WHERE n < last_value
+)
+SELECT id, n FROM numbers ORDER BY id, n;
+----
+1 1
+1 2
+1 3
+2 1
+2 2
+
+# A predicate outside the CTE filters rows after the recursion is complete.
+query I
+WITH RECURSIVE numbers(n) AS (
+ SELECT 1
+ UNION ALL
+ SELECT n + 1
+ FROM numbers
+ WHERE n < 6
+)
+SELECT n
+FROM numbers
+WHERE n >= 4
+ORDER BY n;
+----
+4
+5
+6
diff --git
a/modules/calcite/src/test/sql/hierarchy/test_recursive_subquery.test
b/modules/calcite/src/test/sql/hierarchy/test_recursive_subquery.test
new file mode 100644
index 00000000000..092eb389cc2
--- /dev/null
+++ b/modules/calcite/src/test/sql/hierarchy/test_recursive_subquery.test
@@ -0,0 +1,54 @@
+# name: test/sql/hierarchy/test_recursive_subquery.test
+# description: Test recursive CTEs with source CTEs and subqueries
+# group: [hierarchy]
+
+statement ok
+CREATE TABLE groups (id INTEGER PRIMARY KEY, parent_id INTEGER);
+
+statement ok
+INSERT INTO groups VALUES (1, NULL), (2, 1), (3, 1), (4, 2), (5, NULL);
+
+statement ok
+CREATE TABLE items (id INTEGER PRIMARY KEY, group_id INTEGER);
+
+statement ok
+INSERT INTO items VALUES (100, 1), (200, 2), (300, 3), (400, 4), (500, 5);
+
+# A non-recursive CTE supplies rows to the seed of a recursive CTE.
+query II
+WITH RECURSIVE roots(id) AS (
+ SELECT id FROM groups WHERE id = 2
+), descendants(id, depth) AS (
+ SELECT id, 0 FROM roots
+ UNION ALL
+ SELECT g.id, d.depth + 1
+ FROM groups g
+ JOIN descendants d ON g.parent_id = d.id
+)
+SELECT id, depth FROM descendants ORDER BY id;
+----
+2 0
+4 1
+
+# The complete recursive query is used as an EXISTS subquery and correlated
with the outer row.
+query I
+SELECT i.id
+FROM items i
+WHERE EXISTS (
+ WITH RECURSIVE descendants(id) AS (
+ SELECT id FROM groups WHERE id = 1
+ UNION ALL
+ SELECT g.id
+ FROM groups g
+ JOIN descendants d ON g.parent_id = d.id
+ )
+ SELECT 1
+ FROM descendants d
+ WHERE d.id = i.group_id
+)
+ORDER BY i.id;
+----
+100
+200
+300
+400
