alex-plekhanov commented on code in PR #13479: URL: https://github.com/apache/ignite/pull/13479#discussion_r3844910469
########## modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RecursiveCteState.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.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> current = Collections.emptyList(); Review Comment: Abbreviation should be used for `current` ########## modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/RecursiveCteIntegrationTest.java: ########## @@ -43,18 +99,149 @@ 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); - Throwable err = GridTestUtils.assertThrows( - log, - () -> sql(qry), + assertTrue(plan, plan.contains("IgniteRepeatUnion")); + assertTrue(plan, plan.contains("IgniteRecursiveTableSpool")); + } + + /** */ + @Test + public void testRecursiveDeltaIsAccountedForMemoryQuota() { Review Comment: Let's move quota test to `MemoryQuotasIntegrationTest` ########## 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.BiRel; +import org.apache.calcite.rel.RelInput; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.RelWriter; + +import static java.util.Objects.requireNonNull; + +/** Coordinator-side iterative UNION ALL for a recursive CTE. */ +public class IgniteRepeatUnion extends BiRel implements IgniteRel { Review Comment: Why not `extends RepeatUnion`? ########## modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/RecursiveCteState.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.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> current = 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> currentMemoryTracker; Review Comment: Abbreviation should be used for `current` ########## modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RecursiveCteUtils.java: ########## @@ -0,0 +1,195 @@ +/* + * 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.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.BaseQueryContext; +import org.apache.ignite.internal.processors.query.calcite.rel.logical.IgniteLogicalRecursiveStaticSpool; + +/** 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; + } + + /** Materializes maximal iteration subtrees that do not depend on the current delta. */ + static RelNode materializeStaticInputs(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 IgniteLogicalRecursiveStaticSpool(input)); + else + newInputs.add(materializeStaticInputs(input, table)); + } + + return rel.copy(rel.getTraitSet(), newInputs); + } + + /** 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); Review Comment: Looks like ProjectableFilterableTableScan, IgniteSortedIndexSpool, IgniteHashIndexSpool are not implement `accept` method as required (not sure if this can affect this ticket). ########## 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: As far as I understand there is no way to now to set iterationLimit to value other than -1. Maybe we should limit it in execution node? -- 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]
