alex-plekhanov commented on code in PR #13479: URL: https://github.com/apache/ignite/pull/13479#discussion_r3901925453
########## modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveCteConverterRule.java: ########## @@ -0,0 +1,107 @@ +/* + * 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.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 RecursiveCteConverterRule extends AbstractIgniteConverterRule<LogicalRepeatUnion> { + /** Instance. */ + public static final RelOptRule INSTANCE = new RecursiveCteConverterRule(); + + /** */ + private RecursiveCteConverterRule() { + 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); + + RelNode seed = unwrapSpool(rel.getSeedRel(), table, "seed"); + RelNode iterative = unwrapSpool(rel.getIterativeRel(), table, "recursive term"); + + if (RecursiveCteUtils.referenceCount(iterative, table) > 1) + throw unsupported("the recursive term must contain no more than one self-reference"); + + RelOptCluster cluster = rel.getCluster(); + RelTraitSet traits = cluster.traitSetOf(IgniteConvention.INSTANCE).replace(single()); + iterative = RecursiveCteUtils.markStaticInputs(iterative, table); + + return new IgniteRepeatUnion( + cluster, + traits, + convert(seed, traits), + convert(iterative, traits), + stateId, + rel.iterationLimit + ); + } + + /** */ + private static RelNode unwrapSpool(RelNode rel, RelOptTable table, 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 (!RecursiveCteUtils.sameTransientTable(spool.getTable(), table)) { Review Comment: Redundant braces ########## modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/RecursiveCteIntegrationTest.java: ########## @@ -43,18 +96,209 @@ public void testHierarchicalQueryIsNotSupported() { "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"; + + String plan = (String)sql("EXPLAIN PLAN FOR " + qry).get(0).get(0); + + assertTrue(plan, plan.contains("IgniteRepeatUnion")); + assertTrue(plan, plan.contains("IgniteRecursiveTableScan")); + assertFalse(plan, plan.contains("IgniteRecursiveTableSpool")); Review Comment: There is no such node now ########## modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/RecursiveCtePlannerTest.java: ########## @@ -0,0 +1,184 @@ +/* + * 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 java.util.List; +import org.apache.calcite.plan.RelOptUtil; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.Exchange; +import org.apache.calcite.rel.core.Spool; +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.IgniteRel; +import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRepeatUnion; +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); + + IgniteRel plan = physicalPlan( + "WITH RECURSIVE numbers(n) AS (" + + "SELECT 1 " + + "UNION ALL " + + "SELECT n + 1 FROM numbers WHERE n < 3" + + ") " + + "SELECT n FROM numbers", + schema + ); + + assertRecursivePlan(plan); + IgniteRepeatUnion repeatUnion = findFirstNode(plan, byClass(IgniteRepeatUnion.class)); + + assertTrue(planDescription(plan), repeatUnion.getLeft() instanceof IgniteValues); + assertEquals(1, findNodes(plan, byClass(IgniteRecursiveTableScan.class)).size()); + + checkSplitAndSerialization(plan, schema); Review Comment: Let's avoid manual plans check and use assertPlan instead: ``` assertPlan(sql, schema, isInstanceOf(IgniteRepeatUnion.class) .and(hasDistribution(IgniteDistributions.single())) .and(input(0, isInstanceOf(IgniteValues.class))) .and(input(1, hasChildThat(isInstanceOf(IgniteRecursiveTableScan.class))) )); ``` ########## modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/RecursiveCteIntegrationTest.java: ########## @@ -43,18 +96,209 @@ public void testHierarchicalQueryIsNotSupported() { "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"; + + String plan = (String)sql("EXPLAIN PLAN FOR " + qry).get(0).get(0); + + assertTrue(plan, plan.contains("IgniteRepeatUnion")); + assertTrue(plan, plan.contains("IgniteRecursiveTableScan")); + assertFalse(plan, plan.contains("IgniteRecursiveTableSpool")); + } - Throwable err = GridTestUtils.assertThrows( - log, - () -> sql(qry), + /** */ + @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 testRecursiveTermWithMultipleSelfReferencesIsRejected() { + assertThrows( + "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", + IgniteSQLException.class, + "the recursive term must contain no more than one self-reference" + ); + } + + /** */ + @Test + public void testRecursiveCteWithMultipleRecursiveBranchesIsRejected() { + assertThrows( + "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", IgniteSQLException.class, - "Failed to plan query" + "the recursive term must contain no more than one self-reference" ); + } + + /** */ + @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" + ); + } + + /** */ + @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(); + } + + /** */ + @Test + public void testNonDeterministicTableScanIsNotMaterialized() { Review Comment: Result is not checked. Only plan. Maybe move it to the planner test? ########## modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/RecursiveCtePlannerTest.java: ########## @@ -0,0 +1,184 @@ +/* + * 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 java.util.List; +import org.apache.calcite.plan.RelOptUtil; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.Exchange; +import org.apache.calcite.rel.core.Spool; +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.IgniteRel; +import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRepeatUnion; +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); + + IgniteRel plan = physicalPlan( + "WITH RECURSIVE numbers(n) AS (" + + "SELECT 1 " + + "UNION ALL " + + "SELECT n + 1 FROM numbers WHERE n < 3" + + ") " + + "SELECT n FROM numbers", + schema + ); + + assertRecursivePlan(plan); + IgniteRepeatUnion repeatUnion = findFirstNode(plan, byClass(IgniteRepeatUnion.class)); + + assertTrue(planDescription(plan), repeatUnion.getLeft() instanceof IgniteValues); + assertEquals(1, findNodes(plan, byClass(IgniteRecursiveTableScan.class)).size()); + + checkSplitAndSerialization(plan, schema); + } + + /** A replicated source can be read on the coordinator without an exchange. */ + @Test + public void testRecursiveCteWithReplicatedTable() throws Exception { + IgniteSchema schema = hierarchySchema(IgniteDistributions.broadcast(), false); + + IgniteRel plan = physicalPlan(EMPLOYEE_HIERARCHY_QUERY, schema); + + assertRecursivePlan(plan); + assertTrue(planDescription(plan), findNodes(plan, byClass(Exchange.class)).isEmpty()); + + checkSplitAndSerialization(plan, schema); + } + + /** 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); + + IgniteRel plan = physicalPlan(EMPLOYEE_HIERARCHY_QUERY, schema); + + assertRecursivePlan(plan); + assertFalse(planDescription(plan), findNodes(plan, byClass(Exchange.class)).isEmpty()); + + IgniteRepeatUnion repeatUnion = findFirstNode(plan, byClass(IgniteRepeatUnion.class)); + Spool spool = findFirstNode(repeatUnion.getRight(), byClass(Spool.class)); + + assertNotNull(planDescription(plan), spool); + assertFalse(planDescription(plan), findNodes(spool.getInput(), byClass(Exchange.class)).isEmpty()); + + checkSplitAndSerialization(plan, schema); + } + + /** A replicated indexed input can be rewound without materialization. */ + @Test + public void testRecursiveCteWithReplicatedIndexedTable() throws Exception { + IgniteSchema schema = hierarchySchema(IgniteDistributions.broadcast(), true); + + IgniteRel plan = physicalPlan(INDEXED_EMPLOYEE_HIERARCHY_QUERY, schema); + + assertRecursivePlan(plan); + + IgniteRepeatUnion repeatUnion = findFirstNode(plan, byClass(IgniteRepeatUnion.class)); + RelNode iterative = repeatUnion.getRight(); + + assertFalse(planDescription(plan), findNodes(iterative, byClass(IgniteIndexScan.class)).isEmpty()); + assertTrue(planDescription(plan), findNodes(iterative, byClass(Spool.class)).isEmpty()); + + checkSplitAndSerialization(plan, schema); + } + + /** Calcite places multiple non-recursive branches into the seed input of RepeatUnion. */ + @Test + public void testRecursiveCteWithMultipleSeedBranches() throws Exception { + IgniteSchema schema = new IgniteSchema(DEFAULT_SCHEMA); + + IgniteRel plan = physicalPlan( + "WITH RECURSIVE numbers(n) AS (" + + "SELECT 1 " + + "UNION ALL " + + "SELECT 10 " + + "UNION ALL " + + "SELECT n + 1 FROM numbers WHERE n < 3" + + ") " + + "SELECT n FROM numbers", + schema + ); + + assertRecursivePlan(plan); + + IgniteRepeatUnion repeatUnion = findFirstNode(plan, byClass(IgniteRepeatUnion.class)); + IgniteValues seedValues = findFirstNode(repeatUnion.getLeft(), byClass(IgniteValues.class)); + + assertNotNull(planDescription(plan), seedValues); + assertEquals(planDescription(plan), 2, seedValues.getTuples().size()); + + checkSplitAndSerialization(plan, schema); + } + + /** Creates an employee table with the requested distribution and optional manager index. */ + private static IgniteSchema hierarchySchema(IgniteDistribution distribution, boolean withManagerIndex) { Review Comment: Abbreviation should be used for "index" ########## modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveCteConverterRule.java: ########## @@ -0,0 +1,107 @@ +/* + * 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.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 RecursiveCteConverterRule extends AbstractIgniteConverterRule<LogicalRepeatUnion> { Review Comment: RepeatUnionConverterRule? ########## modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RepeatUnionNode.java: ########## @@ -0,0 +1,163 @@ +/* + * 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.query.calcite.exec.ExecutionContext; +import org.apache.ignite.internal.util.typedef.F; + +/** 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; Review Comment: In case of wrong query user can fail the whole cluster. I think there should be at least some configuration (maybe distributed property) with reasonable default limit (and error message with a hint that property can be changed if limit exceeded). ########## modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/RecursiveCtePlannerTest.java: ########## @@ -0,0 +1,184 @@ +/* + * 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 java.util.List; +import org.apache.calcite.plan.RelOptUtil; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.Exchange; +import org.apache.calcite.rel.core.Spool; +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.IgniteRel; +import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRepeatUnion; +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); + + IgniteRel plan = physicalPlan( + "WITH RECURSIVE numbers(n) AS (" + + "SELECT 1 " + + "UNION ALL " + + "SELECT n + 1 FROM numbers WHERE n < 3" + + ") " + + "SELECT n FROM numbers", + schema + ); + + assertRecursivePlan(plan); + IgniteRepeatUnion repeatUnion = findFirstNode(plan, byClass(IgniteRepeatUnion.class)); + + assertTrue(planDescription(plan), repeatUnion.getLeft() instanceof IgniteValues); + assertEquals(1, findNodes(plan, byClass(IgniteRecursiveTableScan.class)).size()); + + checkSplitAndSerialization(plan, schema); + } + + /** A replicated source can be read on the coordinator without an exchange. */ + @Test + public void testRecursiveCteWithReplicatedTable() throws Exception { + IgniteSchema schema = hierarchySchema(IgniteDistributions.broadcast(), false); + + IgniteRel plan = physicalPlan(EMPLOYEE_HIERARCHY_QUERY, schema); + + assertRecursivePlan(plan); + assertTrue(planDescription(plan), findNodes(plan, byClass(Exchange.class)).isEmpty()); + + checkSplitAndSerialization(plan, schema); + } + + /** 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); + + IgniteRel plan = physicalPlan(EMPLOYEE_HIERARCHY_QUERY, schema); + + assertRecursivePlan(plan); + assertFalse(planDescription(plan), findNodes(plan, byClass(Exchange.class)).isEmpty()); + + IgniteRepeatUnion repeatUnion = findFirstNode(plan, byClass(IgniteRepeatUnion.class)); + Spool spool = findFirstNode(repeatUnion.getRight(), byClass(Spool.class)); + + assertNotNull(planDescription(plan), spool); + assertFalse(planDescription(plan), findNodes(spool.getInput(), byClass(Exchange.class)).isEmpty()); + + checkSplitAndSerialization(plan, schema); + } + + /** A replicated indexed input can be rewound without materialization. */ + @Test + public void testRecursiveCteWithReplicatedIndexedTable() throws Exception { + IgniteSchema schema = hierarchySchema(IgniteDistributions.broadcast(), true); + + IgniteRel plan = physicalPlan(INDEXED_EMPLOYEE_HIERARCHY_QUERY, schema); + + assertRecursivePlan(plan); + + IgniteRepeatUnion repeatUnion = findFirstNode(plan, byClass(IgniteRepeatUnion.class)); + RelNode iterative = repeatUnion.getRight(); + + assertFalse(planDescription(plan), findNodes(iterative, byClass(IgniteIndexScan.class)).isEmpty()); + assertTrue(planDescription(plan), findNodes(iterative, byClass(Spool.class)).isEmpty()); + + checkSplitAndSerialization(plan, schema); + } + + /** Calcite places multiple non-recursive branches into the seed input of RepeatUnion. */ + @Test + public void testRecursiveCteWithMultipleSeedBranches() throws Exception { + IgniteSchema schema = new IgniteSchema(DEFAULT_SCHEMA); + + IgniteRel plan = physicalPlan( + "WITH RECURSIVE numbers(n) AS (" + + "SELECT 1 " + Review Comment: Values are merged on union and this is not quite informative test. Lets check that "union" for seed remains and nested into left hand of RepeatUnion, for example, something like: ``` 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(input(0, isInstanceOf(IgniteUnionAll.class) .and(input(0, isTableScan("T"))) .and(input(1, isTableScan("T"))) )) ); ``` ########## modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/RecursiveCteIntegrationTest.java: ########## @@ -43,18 +96,209 @@ public void testHierarchicalQueryIsNotSupported() { "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"; + + String plan = (String)sql("EXPLAIN PLAN FOR " + qry).get(0).get(0); + + assertTrue(plan, plan.contains("IgniteRepeatUnion")); + assertTrue(plan, plan.contains("IgniteRecursiveTableScan")); + assertFalse(plan, plan.contains("IgniteRecursiveTableSpool")); + } - Throwable err = GridTestUtils.assertThrows( - log, - () -> sql(qry), + /** */ + @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 testRecursiveTermWithMultipleSelfReferencesIsRejected() { + assertThrows( + "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", + IgniteSQLException.class, + "the recursive term must contain no more than one self-reference" Review Comment: Do we really need this limitation? What kind of problems are we face without check for single recursive table usage? ########## modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/RecursiveCteIntegrationTest.java: ########## @@ -43,18 +96,209 @@ public void testHierarchicalQueryIsNotSupported() { "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"; + + String plan = (String)sql("EXPLAIN PLAN FOR " + qry).get(0).get(0); + + assertTrue(plan, plan.contains("IgniteRepeatUnion")); + assertTrue(plan, plan.contains("IgniteRecursiveTableScan")); + assertFalse(plan, plan.contains("IgniteRecursiveTableSpool")); + } - Throwable err = GridTestUtils.assertThrows( - log, - () -> sql(qry), + /** */ + @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 testRecursiveTermWithMultipleSelfReferencesIsRejected() { + assertThrows( + "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", + IgniteSQLException.class, + "the recursive term must contain no more than one self-reference" + ); + } + + /** */ + @Test + public void testRecursiveCteWithMultipleRecursiveBranchesIsRejected() { + assertThrows( + "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", IgniteSQLException.class, - "Failed to plan query" + "the recursive term must contain no more than one self-reference" ); + } + + /** */ + @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" + ); + } + + /** */ + @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(); + } + + /** */ + @Test + public void testNonDeterministicTableScanIsNotMaterialized() { + registerRecursiveFunctions(); + + sql("CREATE TABLE recursive_markers (id INT) WITH TEMPLATE=REPLICATED"); + sql("INSERT INTO recursive_markers VALUES (1)"); + + 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 FROM recursive_markers" + + ") v " + + "WHERE n < 4" + + ") " + + "SELECT n, marker FROM numbers ORDER BY n"; + + String plan = (String)sql("EXPLAIN PLAN FOR " + qry).get(0).get(0); + + assertTrue(plan, plan.contains("IgniteTableScan")); + assertTrue(plan, plan.contains("NEXTRECURSIVEVALUE")); + assertFalse(plan, plan.contains("IgniteTableSpool")); + } + + /** */ + @Test + public void testNonDeterministicIndexScanIsNotMaterialized() { Review Comment: Result is not checked. Only plan. Maybe move it to the planner test? ########## modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java: ########## @@ -162,6 +168,9 @@ public class LogicalRelImplementor<Row> implements IgniteRelVisitor<Node<Row>> { /** */ private final ExpressionFactory<Row> expressionFactory; + /** Query-local recursive CTE states, keyed by transient table identifier. */ + private final Map<String, RecursiveCteState<Row>> recursiveStates = new HashMap<>(); Review Comment: Maybe it's worth to store cte state inside execution context, not inside rel implementor? ########## modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveCteUtils.java: ########## @@ -0,0 +1,214 @@ +/* + * 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.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.SingleRel; +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.BaseQueryContext; + +/** 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) { + BaseQueryContext ctx = planner.getContext().unwrap(BaseQueryContext.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; + } + + /** Marks maximal inputs that do not depend on the current delta for coordinator-local rewindable conversion. */ + static RelNode markStaticInputs(RelNode rel, RelOptTable table) { + 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(new StaticInput(input)); Review Comment: Can we convert traits in markStaticInputs instead of creating new rel node and new rule for this relationship node? ########## modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/BaseQueryContext.java: ########## @@ -312,6 +322,18 @@ public int[] partitions() { return null; } + /** Returns an identifier unique for the given recursive CTE within this planning context. */ + public synchronized String recursiveCteStateId(RelOptTable table) { Review Comment: State id only generated on planning phase, looks like planning context is more suitable place for this method. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
