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 1df6a4efe7e IGNITE-29012 SQL Calcite: Support recursive CTE with UNION
DISTINCT (#13563)
1df6a4efe7e is described below
commit 1df6a4efe7ea19a75070195c3ec254d65576f89b
Author: Vladislav Pyatkov <[email protected]>
AuthorDate: Wed Sep 9 22:11:53 2026 +0300
IGNITE-29012 SQL Calcite: Support recursive CTE with UNION DISTINCT (#13563)
---
.../query/calcite/exec/LogicalRelImplementor.java | 2 +-
.../query/calcite/exec/rel/RecursiveCteState.java | 56 +++++++++++++-----
.../query/calcite/exec/rel/RepeatUnionNode.java | 69 +++++++++++-----------
.../query/calcite/rel/IgniteRepeatUnion.java | 10 ++--
.../calcite/rule/RepeatUnionConverterRule.java | 5 +-
.../integration/RecursiveCteIntegrationTest.java | 55 ++++++++++++-----
.../calcite/planner/RecursiveCtePlannerTest.java | 56 ++++++++++++++++--
7 files changed, 172 insertions(+), 81 deletions(-)
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 e2777240471..7b2172c33ac 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
@@ -646,7 +646,7 @@ public class LogicalRelImplementor<Row> implements
IgniteRelVisitor<Node<Row>> {
/** {@inheritDoc} */
@Override public Node<Row> visit(IgniteRepeatUnion rel) {
- RepeatUnionNode<Row> node = new RepeatUnionNode<>(ctx,
rel.getRowType(), rel.iterationLimit());
+ RepeatUnionNode<Row> node = new RepeatUnionNode<>(ctx,
rel.getRowType(), rel.all, rel.iterationLimit());
node.register(F.asList(visit(rel.getLeft()), visit(rel.getRight())));
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
index 8a865164de8..83bc7d15ef2 100644
---
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
@@ -19,18 +19,32 @@ package
org.apache.ignite.internal.processors.query.calcite.exec.rel;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import
org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler;
+import
org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.GroupKey;
import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.RowTracker;
import org.apache.ignite.internal.util.GridUnsafe;
+import org.jetbrains.annotations.Nullable;
/** Query-local current and next deltas of a recursive CTE. */
public class RecursiveCteState<Row> {
+ /** Rows seen across all iterations, or null for UNION ALL. */
+ private final @Nullable Set<GroupKey<Row>> seen;
+
+ /** Row handler used for SQL grouping keys. */
+ private final RowHandler<Row> hnd;
+
+ /** Memory tracker for keys and their rows retained for duplicate
elimination. */
+ private final @Nullable RowTracker<GroupKey<Row>> seenMemoryTracker;
+
/** 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;
+ private List<Row> next = new ArrayList<>();
/** Memory tracker for rows in the current delta. */
private RowTracker<Row> curMemoryTracker;
@@ -39,33 +53,38 @@ public class RecursiveCteState<Row> {
private RowTracker<Row> nextMemoryTracker;
/** */
- public RecursiveCteState(ExecutionContext<Row> ctx) {
+ public RecursiveCteState(ExecutionContext<Row> ctx, boolean all) {
+ seen = all ? null : new HashSet<>();
+ hnd = ctx.rowHandler();
+ seenMemoryTracker = all ? null :
ctx.createNodeMemoryTracker(MemoryTrackingNode.HASH_MAP_ROW_OVERHEAD);
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;
+ /** Adds a new row to the next delta, returning false for duplicates in
DISTINCT mode. */
+ public boolean add(Row row) {
+ if (seen != null) {
+ GroupKey<Row> rowKey = GroupKey.of(row, hnd);
- next = new ArrayList<>();
- }
+ if (!seen.add(rowKey))
+ return false;
- /** Adds a row to the next delta. */
- public void add(Row row) {
- assert next != null;
+ seenMemoryTracker.onRowAdded(rowKey);
+ }
next.add(row);
- nextMemoryTracker.onRowAdded(row);
+
+ // DISTINCT already accounts for the row in seen; null charges only
the delta's reference overhead.
+ nextMemoryTracker.onRowAdded(seen == null ? row : null);
+
+ return true;
}
- /** Makes the collected delta visible to recursive scans. */
+ /** Publishes the collected delta and prepares an empty buffer for the
next iteration. */
public void commit() {
- assert next != null;
-
curMemoryTracker.reset();
cur = next;
- next = null;
+ next = new ArrayList<>();
RowTracker<Row> tracker = curMemoryTracker;
@@ -86,9 +105,14 @@ public class RecursiveCteState<Row> {
/** Clears all query-local rows. */
public void clear() {
cur = Collections.emptyList();
- next = null;
+ next = new ArrayList<>();
curMemoryTracker.reset();
nextMemoryTracker.reset();
+
+ if (seen != null) {
+ seen.clear();
+ seenMemoryTracker.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
index 9586b91a362..8e66d2ece09 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RepeatUnionNode.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RepeatUnionNode.java
@@ -26,7 +26,7 @@ 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. */
+/** Coordinator-side executor for recursive union. */
public class RepeatUnionNode<Row> extends AbstractNode<Row> implements
Downstream<Row> {
/** Index of the seed input. */
private static final int SEED_SOURCE = 0;
@@ -46,21 +46,22 @@ public class RepeatUnionNode<Row> extends AbstractNode<Row>
implements Downstrea
/** Number of rows still requested by downstream. */
private int waiting;
+ /** Number of rows still requested from the active source. */
+ private int pending;
+
/** 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,
+ boolean all,
int iterationLimit
) {
super(ctx, rowType);
- state = new RecursiveCteState<>(ctx);
+ state = new RecursiveCteState<>(ctx, all);
this.iterationLimit = iterationLimit;
}
@@ -87,27 +88,31 @@ public class RepeatUnionNode<Row> extends AbstractNode<Row>
implements Downstrea
/** {@inheritDoc} */
@Override public void push(Row row) throws Exception {
assert downstream() != null;
- assert waiting > 0;
- assert writing;
+ assert waiting > 0 && pending > 0 :
+ "Received a row without outstanding demand [waiting=" + waiting +
", pending=" + pending + ']';
checkState();
- waiting--;
- state.add(row);
+ pending--;
+
+ if (state.add(row)) {
+ waiting--;
+ downstream().push(row);
+ }
- downstream().push(row);
+ if (pending == 0 && waiting > 0)
+ context().execute(this::requestSource, this::onError);
}
/** {@inheritDoc} */
@Override public void end() throws Exception {
assert downstream() != null;
assert waiting > 0;
- assert writing;
checkState();
+ pending = 0;
state.commit();
- writing = false;
if (state.isEmpty()) {
finish();
@@ -115,25 +120,19 @@ public class RepeatUnionNode<Row> extends
AbstractNode<Row> implements Downstrea
return;
}
- if (curSrc == SEED_SOURCE) {
- if (iterationLimit == 0) {
- throw iterationLimitExceeded();
- }
-
- curSrc = RECURSIVE_SOURCE;
- requestSource();
-
- return;
- }
-
- iteration++;
+ if (curSrc == RECURSIVE_SOURCE)
+ iteration++;
- if (iterationLimit >= 0 && iteration == iterationLimit) {
+ if (iterationLimit >= 0 && iteration == iterationLimit)
throw iterationLimitExceeded();
- }
- source().rewind();
- requestSource();
+ if (curSrc == SEED_SOURCE)
+ curSrc = RECURSIVE_SOURCE;
+ else
+ source().rewind();
+
+ // Let the previous scan leave its push loop before requesting the
next iteration.
+ context().execute(this::requestSource, this::onError);
}
/** {@inheritDoc} */
@@ -152,8 +151,8 @@ public class RepeatUnionNode<Row> extends AbstractNode<Row>
implements Downstrea
@Override protected void rewindInternal() {
curSrc = SEED_SOURCE;
waiting = 0;
+ pending = 0;
iteration = 0;
- writing = false;
state.clear();
}
@@ -187,14 +186,14 @@ public class RepeatUnionNode<Row> extends
AbstractNode<Row> implements Downstrea
}
}
- /** Starts collecting and requests rows from the active input. */
+ /** Requests the remaining downstream demand from the active input. */
private void requestSource() throws Exception {
- if (!writing) {
- state.beginWrite();
- writing = true;
- }
+ checkState();
+
+ assert pending == 0 : "Cannot request more rows while the source
request is pending [waiting=" + waiting +
+ ", pending=" + pending + ']';
- source().request(waiting);
+ source().request(pending = waiting);
}
/** */
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRepeatUnion.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRepeatUnion.java
index a5689f2dbe6..b89123eff71 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRepeatUnion.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRepeatUnion.java
@@ -24,7 +24,7 @@ import org.apache.calcite.rel.RelInput;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.RepeatUnion;
-/** Coordinator-side iterative UNION ALL for a recursive CTE. */
+/** Coordinator-side iterative union for a recursive CTE. */
public class IgniteRepeatUnion extends RepeatUnion implements IgniteRel {
/** */
public IgniteRepeatUnion(
@@ -32,9 +32,10 @@ public class IgniteRepeatUnion extends RepeatUnion
implements IgniteRel {
RelTraitSet traits,
RelNode seed,
RelNode iterative,
+ boolean all,
int iterationLimit
) {
- super(cluster, traits, seed, iterative, true, iterationLimit, null);
+ super(cluster, traits, seed, iterative, all, iterationLimit, null);
}
/** Constructor used for deserialization. */
@@ -44,6 +45,7 @@ public class IgniteRepeatUnion extends RepeatUnion implements
IgniteRel {
input.getTraitSet().replace(IgniteConvention.INSTANCE),
input.getInputs().get(0),
input.getInputs().get(1),
+ input.getBoolean("all", true),
iterationLimit(input)
);
}
@@ -57,7 +59,7 @@ public class IgniteRepeatUnion extends RepeatUnion implements
IgniteRel {
@Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
assert inputs.size() == 2;
- return new IgniteRepeatUnion(getCluster(), traitSet, inputs.get(0),
inputs.get(1), iterationLimit);
+ return new IgniteRepeatUnion(getCluster(), traitSet, inputs.get(0),
inputs.get(1), all, iterationLimit);
}
/** {@inheritDoc} */
@@ -69,7 +71,7 @@ public class IgniteRepeatUnion extends RepeatUnion implements
IgniteRel {
@Override public IgniteRel clone(RelOptCluster cluster, List<IgniteRel>
inputs) {
assert inputs.size() == 2;
- return new IgniteRepeatUnion(cluster, getTraitSet(), inputs.get(0),
inputs.get(1), iterationLimit);
+ return new IgniteRepeatUnion(cluster, getTraitSet(), inputs.get(0),
inputs.get(1), all, iterationLimit);
}
/** Reads the optional iteration limit from a serialized plan. */
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RepeatUnionConverterRule.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RepeatUnionConverterRule.java
index ef37d92cbb0..a19f073c003 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RepeatUnionConverterRule.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RepeatUnionConverterRule.java
@@ -53,10 +53,6 @@ public class RepeatUnionConverterRule extends
AbstractIgniteConverterRule<Logica
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");
-
int iterationLimit =
planner.getContext().unwrap(PlanningContext.class).recursiveCteIterationLimit();
RelNode seed = unwrapSpool(rel.getSeedRel(), "seed");
@@ -71,6 +67,7 @@ public class RepeatUnionConverterRule extends
AbstractIgniteConverterRule<Logica
traits,
convert(seed, traits),
convert(iterative, traits),
+ rel.all,
iterationLimit
);
}
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/RecursiveCteIntegrationTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/RecursiveCteIntegrationTest.java
index 8989bad03de..5352cdc3f40 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
@@ -51,9 +51,12 @@ public class RecursiveCteIntegrationTest extends
AbstractBasicIntegrationTest {
.returns(3)
.check();
- assertThrows("WITH " + keyword + "numbers(n) AS (" +
- "SELECT 1 UNION SELECT n + 1 FROM numbers WHERE n < 3) SELECT
* FROM numbers",
- IgniteSQLException.class, "only UNION ALL is supported");
+ assertQuery("WITH " + keyword + "numbers(n) AS (" +
+ "SELECT 1 UNION SELECT n + 1 FROM numbers WHERE n < 3) SELECT
* FROM numbers")
+ .returns(1)
+ .returns(2)
+ .returns(3)
+ .check();
}
}
@@ -244,19 +247,41 @@ public class RecursiveCteIntegrationTest extends
AbstractBasicIntegrationTest {
.check();
}
- /** */
+ /** Both DISTINCT spellings eliminate duplicates in the seed and across
recursive iterations. */
@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,
- "only UNION ALL is supported"
- );
+ public void testRecursiveCteWithDistinctUnion() {
+ for (String union : new String[] {"UNION", "UNION DISTINCT"}) {
+ assertQuery("WITH RECURSIVE numbers(n) AS (" +
+ "SELECT * FROM (VALUES (1), (1), (2)) " + union + " " +
+ "SELECT MOD(n, 3) + 1 FROM numbers" +
+ ") SELECT n FROM numbers")
+ .returns(1)
+ .returns(2)
+ .returns(3)
+ .check();
+ }
+ }
+
+ /** NULLs compare equal and all columns participate in duplicate
elimination. */
+ @Test
+ public void testRecursiveDistinctNulls() {
+ assertQuery("WITH RECURSIVE numbers(n, label) AS (" +
+ "SELECT * FROM (VALUES (1, CAST(NULL AS VARCHAR)), (1, CAST(NULL
AS VARCHAR)), (1, 'x')) " +
+ "UNION DISTINCT SELECT n, label FROM numbers" +
+ ") SELECT n, label FROM numbers")
+ .returns(1, null)
+ .returns(1, "x")
+ .check();
+ }
+
+ /** Duplicate-only input batches must keep requesting rows until the
source ends. */
+ @Test
+ public void testRecursiveDistinctLargeDuplicateBatch() {
+ assertQuery("WITH RECURSIVE numbers(n) AS (" +
+ "SELECT 1 UNION SELECT n FROM numbers CROSS JOIN
TABLE(SYSTEM_RANGE(1, 10000))" +
+ ") SELECT n FROM numbers")
+ .returns(1)
+ .check();
}
/** */
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
index 769b8233558..66c687d631a 100644
---
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
@@ -26,6 +26,9 @@ import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlWithItem;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.sql.util.SqlBasicVisitor;
+import org.apache.calcite.tools.ValidationException;
+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.exp.IgniteScalarFunction;
import
org.apache.ignite.internal.processors.query.calcite.prepare.IgnitePlanner;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexScan;
@@ -37,6 +40,7 @@ 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.apache.ignite.testframework.GridTestUtils;
import org.junit.Test;
/** Planner tests for recursive common table expressions. */
@@ -96,11 +100,15 @@ public class RecursiveCtePlannerTest extends
AbstractPlannerTest {
assertRecursiveFlags("WITH \"Numbers\"(n) AS (SELECT 1 UNION ALL " +
"SELECT n + 1 FROM \"Numbers\" WHERE n < 3) SELECT * FROM
\"Numbers\"", true);
+ }
- assertRecursiveFlags("WITH numbers(n) AS (SELECT 1 UNION ALL " +
+ /** Subquery self-references are inferred as recursive, but rejected
during validation. */
+ @Test
+ public void testUnsupportedSubqueryRecursiveFlags() throws Exception {
+ assertUnsupportedRecursiveFlags("WITH numbers(n) AS (SELECT 1 UNION
ALL " +
"SELECT n + 1 FROM (SELECT * FROM numbers) x WHERE n < 3) SELECT *
FROM numbers", true);
- assertRecursiveFlags("WITH numbers(n) AS (SELECT 1 UNION ALL " +
+ assertUnsupportedRecursiveFlags("WITH numbers(n) AS (SELECT 1 UNION
ALL " +
"SELECT (SELECT n + 1 FROM numbers WHERE n < 3)) SELECT * FROM
numbers", true);
}
@@ -139,22 +147,36 @@ public class RecursiveCtePlannerTest extends
AbstractPlannerTest {
assertRecursiveFlags("WITH numbers(n) AS (SELECT 1 UNION ALL " +
"SELECT * FROM (WITH numbers(n) AS (SELECT 2 UNION ALL SELECT n +
1 FROM numbers WHERE n < 3) " +
"SELECT * FROM numbers)) SELECT * FROM numbers", false, true);
+ }
- assertRecursiveFlags("WITH numbers(n) AS (SELECT 1 UNION ALL " +
+ /** References to an outer recursive CTE from nested WITH queries are
inferred, but unsupported. */
+ @Test
+ public void testUnsupportedNestedCteFlags() throws Exception {
+ assertUnsupportedRecursiveFlags("WITH numbers(n) AS (SELECT 1 UNION
ALL " +
"SELECT n + 1 FROM (WITH numbers(n) AS (SELECT n FROM numbers
WHERE n < 3) " +
"SELECT * FROM numbers)) SELECT * FROM numbers", true, false);
- assertRecursiveFlags("WITH numbers(n) AS (SELECT 1 UNION ALL " +
+ assertUnsupportedRecursiveFlags("WITH numbers(n) AS (SELECT 1 UNION
ALL " +
"SELECT n + 1 FROM (WITH x AS (SELECT n FROM numbers WHERE n < 3),
numbers(n) AS (SELECT 9) " +
"SELECT * FROM x)) SELECT * FROM numbers", true, false, false);
- assertRecursiveFlags("WITH numbers(n) AS (SELECT 1 UNION ALL " +
+ assertUnsupportedRecursiveFlags("WITH numbers(n) AS (SELECT 1 UNION
ALL " +
"SELECT n + 1 FROM (WITH numbers(n) AS (SELECT n FROM numbers
WHERE n < 2 UNION ALL " +
"SELECT n + 1 FROM numbers WHERE n < 2) SELECT * FROM numbers))
SELECT * FROM numbers", true, true);
}
/** Validates real SQL and checks flags on parsed WITH items in pre-order.
*/
private void assertRecursiveFlags(String sql, boolean... expected) throws
Exception {
+ assertRecursiveFlags(sql, expected, false);
+ }
+
+ /** Checks the inferred flags and the validation error for unsupported
recursive references. */
+ private void assertUnsupportedRecursiveFlags(String sql, boolean...
expected) throws Exception {
+ assertRecursiveFlags(sql, expected, true);
+ }
+
+ /** Checks the validation outcome and flags on parsed WITH items in
pre-order. */
+ private void assertRecursiveFlags(String sql, boolean[] expected, boolean
unsupported) throws Exception {
IgniteSchema schema = createSchema(createTable("NUMBERS",
IgniteDistributions.single(), "N", Integer.class));
try (IgnitePlanner planner = plannerCtx(sql, schema).planner()) {
@@ -171,7 +193,17 @@ public class RecursiveCtePlannerTest extends
AbstractPlannerTest {
}
});
- planner.validate(node);
+ if (unsupported) {
+ ValidationException err =
(ValidationException)GridTestUtils.assertThrows(log,
+ () -> planner.validate(node), ValidationException.class,
+ "Unsupported recursive CTE: self-references inside
subqueries are not supported");
+
+ assertTrue(sql, err.getCause() instanceof IgniteSQLException);
+ assertEquals(sql, IgniteQueryErrorCode.UNSUPPORTED_OPERATION,
+ ((IgniteSQLException)err.getCause()).statusCode());
+ }
+ else
+ planner.validate(node);
assertEquals(sql, expected.length, items.size());
@@ -180,6 +212,18 @@ public class RecursiveCtePlannerTest extends
AbstractPlannerTest {
}
}
+ /** DISTINCT semantics survive conversion and plan serialization. */
+ @Test
+ public void testRecursiveDistinctPlan() throws Exception {
+ for (String union : new String[] {"UNION", "UNION DISTINCT"}) {
+ assertPlan("WITH RECURSIVE numbers(n) AS (SELECT 1 " + union +
+ " SELECT n FROM numbers) SELECT n FROM numbers",
+ new IgniteSchema(DEFAULT_SCHEMA),
isInstanceOf(IgniteRepeatUnion.class)
+ .and(rel -> !rel.all)
+ .and(hasDistribution(IgniteDistributions.single())));
+ }
+ }
+
/** A replicated source can be read on the coordinator without an
exchange. */
@Test
public void testRecursiveCteWithReplicatedTable() throws Exception {
