You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@calcite.apache.org by za...@apache.org on 2019/05/29 22:56:45 UTC
[calcite] branch master updated: [CALCITE-2812] Add algebraic
operators to allow expressing recursive queries
This is an automated email from the ASF dual-hosted git repository.
zabetak pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/calcite.git
The following commit(s) were added to refs/heads/master by this push:
new 73e6d05 [CALCITE-2812] Add algebraic operators to allow expressing recursive queries
73e6d05 is described below
commit 73e6d05fa65f16485caca80571d1fe4fda5c7468
Author: rubenada <ru...@gmail.com>
AuthorDate: Thu May 30 00:06:47 2019 +0200
[CALCITE-2812] Add algebraic operators to allow expressing recursive queries
1. Add Spool, and RepeatUnion operators with their Logical and Enumerable counterparts as well as appropriate converter rules.
2. Add new interface allowing to express transient tables with a sample implementation.
3. Add new methods in RelBuilder useful for building recursive queries.
4. Update the website with examples for building recursive queries.
5. Add HierarchySchema allowing to express interesting hierarchies through recursive queries.
6. Add unit tests verifying the plan created by the builder and correctness of the new operators.
Close apache/calcite#1020
---
.../adapter/enumerable/EnumerableRepeatUnion.java | 95 +++++++++++++
.../enumerable/EnumerableRepeatUnionRule.java | 56 ++++++++
.../adapter/enumerable/EnumerableRules.java | 6 +
.../adapter/enumerable/EnumerableTableSpool.java | 111 +++++++++++++++
.../enumerable/EnumerableTableSpoolRule.java | 53 +++++++
.../apache/calcite/prepare/CalcitePrepareImpl.java | 2 +
.../org/apache/calcite/rel/core/RelFactories.java | 55 +++++++-
.../org/apache/calcite/rel/core/RepeatUnion.java | 120 ++++++++++++++++
.../java/org/apache/calcite/rel/core/Spool.java | 87 ++++++++++++
.../org/apache/calcite/rel/core/TableSpool.java | 51 +++++++
.../calcite/rel/logical/LogicalRepeatUnion.java | 72 ++++++++++
.../calcite/rel/logical/LogicalTableSpool.java | 66 +++++++++
.../org/apache/calcite/schema/TransientTable.java | 33 +++++
.../calcite/schema/impl/ListTransientTable.java | 150 ++++++++++++++++++++
.../java/org/apache/calcite/tools/RelBuilder.java | 93 ++++++++++++
.../org/apache/calcite/util/BuiltInMethod.java | 4 +
.../java/org/apache/calcite/test/CalciteSuite.java | 5 +-
.../org/apache/calcite/test/HierarchySchema.java | 91 ++++++++++++
.../org/apache/calcite/test/RelBuilderTest.java | 73 ++++++++++
.../EnumerableRepeatUnionHierarchyTest.java | 153 ++++++++++++++++++++
.../test/enumerable/EnumerableRepeatUnionTest.java | 157 +++++++++++++++++++++
.../apache/calcite/linq4j/EnumerableDefaults.java | 148 +++++++++++++++++++
site/_docs/algebra.md | 49 +++++++
23 files changed, 1728 insertions(+), 2 deletions(-)
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRepeatUnion.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRepeatUnion.java
new file mode 100644
index 0000000..1406b3b
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRepeatUnion.java
@@ -0,0 +1,95 @@
+/*
+ * 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.calcite.adapter.enumerable;
+
+import org.apache.calcite.linq4j.function.Experimental;
+import org.apache.calcite.linq4j.tree.BlockBuilder;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.linq4j.tree.Expressions;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.RepeatUnion;
+import org.apache.calcite.util.BuiltInMethod;
+
+import java.util.List;
+
+
+/**
+ * Implementation of {@link RepeatUnion} in
+ * {@link EnumerableConvention enumerable calling convention}.
+ *
+ * <p>NOTE: The current API is experimental and subject to change without notice.</p>
+ */
+@Experimental
+public class EnumerableRepeatUnion extends RepeatUnion implements EnumerableRel {
+
+ /**
+ * Creates an EnumerableRepeatUnion
+ */
+ EnumerableRepeatUnion(
+ RelOptCluster cluster,
+ RelTraitSet traits,
+ RelNode seed,
+ RelNode iterative,
+ boolean all,
+ int maxRep) {
+ super(cluster, traits, seed, iterative, all, maxRep);
+ }
+
+ @Override public EnumerableRepeatUnion copy(RelTraitSet traitSet, List<RelNode> inputs) {
+ assert inputs.size() == 2;
+ return new EnumerableRepeatUnion(getCluster(), traitSet,
+ inputs.get(0), inputs.get(1), all, maxRep);
+ }
+
+ @Override public Result implement(EnumerableRelImplementor implementor, Prefer pref) {
+ // TODO only UNION ALL is supported for the moment
+ if (!all) {
+ throw new UnsupportedOperationException(
+ "Only EnumerableRepeatUnion ALL is supported for the moment");
+ }
+
+ // return repeatUnionAll(<seedExp>, <iterativeExp>, maxRep);
+
+ BlockBuilder builder = new BlockBuilder();
+ RelNode seed = getSeedRel();
+ RelNode iteration = getIterativeRel();
+
+ Result seedResult = implementor.visitChild(this, 0, (EnumerableRel) seed, pref);
+ Result iterationResult = implementor.visitChild(this, 1, (EnumerableRel) iteration, pref);
+
+ Expression seedExp = builder.append("seed", seedResult.block);
+ Expression iterativeExp = builder.append("iteration", iterationResult.block);
+
+ Expression unionExp = Expressions.call(
+ BuiltInMethod.REPEAT_UNION_ALL.method,
+ seedExp,
+ iterativeExp,
+ Expressions.constant(maxRep, int.class));
+ builder.add(unionExp);
+
+ PhysType physType = PhysTypeImpl.of(
+ implementor.getTypeFactory(),
+ getRowType(),
+ pref.prefer(seedResult.format));
+ return implementor.result(physType, builder.toBlock());
+ }
+
+}
+
+// End EnumerableRepeatUnion.java
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRepeatUnionRule.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRepeatUnionRule.java
new file mode 100644
index 0000000..b7e78d4
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRepeatUnionRule.java
@@ -0,0 +1,56 @@
+/*
+ * 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.calcite.adapter.enumerable;
+
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.convert.ConverterRule;
+import org.apache.calcite.rel.logical.LogicalRepeatUnion;
+
+/**
+ * Rule to convert a {@link LogicalRepeatUnion} into an {@link EnumerableRepeatUnion}.
+ */
+public class EnumerableRepeatUnionRule extends ConverterRule {
+
+ EnumerableRepeatUnionRule() {
+ super(
+ LogicalRepeatUnion.class,
+ Convention.NONE,
+ EnumerableConvention.INSTANCE,
+ "EnumerableRepeatUnionRule");
+
+ }
+
+ @Override public RelNode convert(RelNode rel) {
+ LogicalRepeatUnion union = (LogicalRepeatUnion) rel;
+ EnumerableConvention out = EnumerableConvention.INSTANCE;
+ RelTraitSet traitSet = union.getTraitSet().replace(out);
+ RelNode seedRel = union.getSeedRel();
+ RelNode iterativeRel = union.getIterativeRel();
+
+ return new EnumerableRepeatUnion(
+ rel.getCluster(),
+ traitSet,
+ convert(seedRel, seedRel.getTraitSet().replace(out)),
+ convert(iterativeRel, iterativeRel.getTraitSet().replace(out)),
+ union.all,
+ union.maxRep);
+ }
+}
+
+// End EnumerableRepeatUnionRule.java
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRules.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRules.java
index 7871638..c1b6845 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRules.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRules.java
@@ -67,6 +67,12 @@ public class EnumerableRules {
public static final EnumerableUnionRule ENUMERABLE_UNION_RULE =
new EnumerableUnionRule();
+ public static final EnumerableRepeatUnionRule ENUMERABLE_REPEAT_UNION_RULE =
+ new EnumerableRepeatUnionRule();
+
+ public static final EnumerableTableSpoolRule ENUMERABLE_TABLE_SPOOL_RULE =
+ new EnumerableTableSpoolRule();
+
public static final EnumerableIntersectRule ENUMERABLE_INTERSECT_RULE =
new EnumerableIntersectRule();
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableTableSpool.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableTableSpool.java
new file mode 100644
index 0000000..bde5b2a
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableTableSpool.java
@@ -0,0 +1,111 @@
+/*
+ * 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.calcite.adapter.enumerable;
+
+import org.apache.calcite.linq4j.function.Experimental;
+import org.apache.calcite.linq4j.tree.BlockBuilder;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.linq4j.tree.Expressions;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelCollationTraitDef;
+import org.apache.calcite.rel.RelDistributionTraitDef;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Spool;
+import org.apache.calcite.rel.core.TableSpool;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.schema.ModifiableTable;
+import org.apache.calcite.util.BuiltInMethod;
+
+/**
+ * Implementation of {@link TableSpool} in
+ * {@link EnumerableConvention enumerable calling convention}
+ * that writes into a {@link ModifiableTable} (which must exist in the current schema).
+ *
+ * <p>NOTE: The current API is experimental and subject to change without notice.</p>
+ */
+@Experimental
+public class EnumerableTableSpool extends TableSpool implements EnumerableRel {
+
+ private EnumerableTableSpool(RelOptCluster cluster, RelTraitSet traitSet, RelNode input,
+ Type readType, Type writeType, String tableName) {
+ super(cluster, traitSet, input, readType, writeType, tableName);
+ }
+
+ /** Creates an EnumerableTableSpool. */
+ public static EnumerableTableSpool create(RelNode input, Type readType, Type writeType,
+ String tableName) {
+ RelOptCluster cluster = input.getCluster();
+ RelMetadataQuery mq = cluster.getMetadataQuery();
+ RelTraitSet traitSet = cluster.traitSetOf(EnumerableConvention.INSTANCE)
+ .replaceIfs(RelCollationTraitDef.INSTANCE,
+ () -> mq.collations(input))
+ .replaceIf(RelDistributionTraitDef.INSTANCE,
+ () -> mq.distribution(input));
+ return new EnumerableTableSpool(cluster, traitSet, input, readType, writeType, tableName);
+ }
+
+ @Override public Result implement(EnumerableRelImplementor implementor, Prefer pref) {
+ // TODO for the moment only LAZY read & write is supported
+ if (readType != Type.LAZY || writeType != Type.LAZY) {
+ throw new UnsupportedOperationException(
+ "EnumerableTableSpool supports for the moment only LAZY read and LAZY write");
+ }
+
+ // ModifiableTable t = (ModifiableTable) root.getRootSchema().getTable(tableName);
+ // return lazyCollectionSpool(t.getModifiableCollection(), <inputExp>);
+
+ BlockBuilder builder = new BlockBuilder();
+
+ RelNode input = getInput();
+ Result inputResult = implementor.visitChild(this, 0, (EnumerableRel) input, pref);
+
+ Expression tableExp = Expressions.convert_(
+ Expressions.call(
+ Expressions.call(
+ implementor.getRootExpression(),
+ BuiltInMethod.DATA_CONTEXT_GET_ROOT_SCHEMA.method),
+ BuiltInMethod.SCHEMA_GET_TABLE.method,
+ Expressions.constant(tableName, String.class)),
+ ModifiableTable.class);
+ Expression collectionExp = Expressions.call(
+ tableExp,
+ BuiltInMethod.MODIFIABLE_TABLE_GET_MODIFIABLE_COLLECTION.method);
+
+ Expression inputExp = builder.append("input", inputResult.block);
+
+ Expression spoolExp = Expressions.call(
+ BuiltInMethod.LAZY_COLLECTION_SPOOL.method,
+ collectionExp,
+ inputExp);
+ builder.add(spoolExp);
+
+ PhysType physType = PhysTypeImpl.of(
+ implementor.getTypeFactory(),
+ getRowType(),
+ pref.prefer(inputResult.format));
+ return implementor.result(physType, builder.toBlock());
+ }
+
+ @Override protected Spool copy(RelTraitSet traitSet, RelNode input,
+ Type readType, Type writeType) {
+ return new EnumerableTableSpool(input.getCluster(), traitSet, input,
+ readType, writeType, tableName);
+ }
+}
+
+// End EnumerableTableSpool.java
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableTableSpoolRule.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableTableSpoolRule.java
new file mode 100644
index 0000000..255ac47
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableTableSpoolRule.java
@@ -0,0 +1,53 @@
+/*
+ * 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.calcite.adapter.enumerable;
+
+import org.apache.calcite.linq4j.function.Experimental;
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.convert.ConverterRule;
+import org.apache.calcite.rel.logical.LogicalTableSpool;
+
+/**
+ * Rule to convert a {@link LogicalTableSpool} into an {@link EnumerableTableSpool}.
+ *
+ * <p>NOTE: The current API is experimental and subject to change without notice.</p>
+ */
+@Experimental
+public class EnumerableTableSpoolRule extends ConverterRule {
+
+ EnumerableTableSpoolRule() {
+ super(
+ LogicalTableSpool.class,
+ Convention.NONE,
+ EnumerableConvention.INSTANCE,
+ "EnumerableTableSpoolRule");
+
+ }
+
+ @Override public RelNode convert(RelNode rel) {
+ LogicalTableSpool spool = (LogicalTableSpool) rel;
+ return EnumerableTableSpool.create(
+ convert(spool.getInput(),
+ spool.getInput().getTraitSet().replace(EnumerableConvention.INSTANCE)),
+ spool.readType,
+ spool.writeType,
+ spool.getTableName());
+ }
+}
+
+// End EnumerableTableSpoolRule.java
diff --git a/core/src/main/java/org/apache/calcite/prepare/CalcitePrepareImpl.java b/core/src/main/java/org/apache/calcite/prepare/CalcitePrepareImpl.java
index 2c08b76..fed31ac 100644
--- a/core/src/main/java/org/apache/calcite/prepare/CalcitePrepareImpl.java
+++ b/core/src/main/java/org/apache/calcite/prepare/CalcitePrepareImpl.java
@@ -204,6 +204,8 @@ public class CalcitePrepareImpl implements CalcitePrepare {
EnumerableRules.ENUMERABLE_COLLECT_RULE,
EnumerableRules.ENUMERABLE_UNCOLLECT_RULE,
EnumerableRules.ENUMERABLE_UNION_RULE,
+ EnumerableRules.ENUMERABLE_REPEAT_UNION_RULE,
+ EnumerableRules.ENUMERABLE_TABLE_SPOOL_RULE,
EnumerableRules.ENUMERABLE_INTERSECT_RULE,
EnumerableRules.ENUMERABLE_MINUS_RULE,
EnumerableRules.ENUMERABLE_TABLE_MODIFICATION_RULE,
diff --git a/core/src/main/java/org/apache/calcite/rel/core/RelFactories.java b/core/src/main/java/org/apache/calcite/rel/core/RelFactories.java
index 7346a44..86addf0 100644
--- a/core/src/main/java/org/apache/calcite/rel/core/RelFactories.java
+++ b/core/src/main/java/org/apache/calcite/rel/core/RelFactories.java
@@ -16,6 +16,7 @@
*/
package org.apache.calcite.rel.core;
+import org.apache.calcite.linq4j.function.Experimental;
import org.apache.calcite.plan.Contexts;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptTable;
@@ -33,11 +34,13 @@ import org.apache.calcite.rel.logical.LogicalJoin;
import org.apache.calcite.rel.logical.LogicalMatch;
import org.apache.calcite.rel.logical.LogicalMinus;
import org.apache.calcite.rel.logical.LogicalProject;
+import org.apache.calcite.rel.logical.LogicalRepeatUnion;
import org.apache.calcite.rel.logical.LogicalSnapshot;
import org.apache.calcite.rel.logical.LogicalSort;
import org.apache.calcite.rel.logical.LogicalSortExchange;
import org.apache.calcite.rel.logical.LogicalTableFunctionScan;
import org.apache.calcite.rel.logical.LogicalTableScan;
+import org.apache.calcite.rel.logical.LogicalTableSpool;
import org.apache.calcite.rel.logical.LogicalUnion;
import org.apache.calcite.rel.logical.LogicalValues;
import org.apache.calcite.rel.metadata.RelColumnMapping;
@@ -109,6 +112,12 @@ public class RelFactories {
public static final SnapshotFactory DEFAULT_SNAPSHOT_FACTORY =
new SnapshotFactoryImpl();
+ public static final SpoolFactory DEFAULT_SPOOL_FACTORY =
+ new SpoolFactoryImpl();
+
+ public static final RepeatUnionFactory DEFAULT_REPEAT_UNION_FACTORY =
+ new RepeatUnionFactoryImpl();
+
/** A {@link RelBuilderFactory} that creates a {@link RelBuilder} that will
* create logical relational expressions for everything. */
public static final RelBuilderFactory LOGICAL_BUILDER =
@@ -125,7 +134,9 @@ public class RelFactories {
DEFAULT_SET_OP_FACTORY,
DEFAULT_VALUES_FACTORY,
DEFAULT_TABLE_SCAN_FACTORY,
- DEFAULT_SNAPSHOT_FACTORY));
+ DEFAULT_SNAPSHOT_FACTORY,
+ DEFAULT_SPOOL_FACTORY,
+ DEFAULT_REPEAT_UNION_FACTORY));
private RelFactories() {
}
@@ -573,6 +584,48 @@ public class RelFactories {
partitionKeys, orderKeys, interval);
}
}
+
+ /**
+ * Can create a {@link Spool} of
+ * the appropriate type for a rule's calling convention.
+ */
+ @Experimental
+ public interface SpoolFactory {
+ /** Creates a {@link TableSpool}. */
+ RelNode createTableSpool(RelNode input, Spool.Type readType, Spool.Type writeType,
+ String tableName);
+ }
+
+ /**
+ * Implementation of {@link SpoolFactory}
+ * that returns Logical Spools.
+ */
+ private static class SpoolFactoryImpl implements SpoolFactory {
+ public RelNode createTableSpool(RelNode input, Spool.Type readType, Spool.Type writeType,
+ String tableName) {
+ return LogicalTableSpool.create(input, readType, writeType, tableName);
+ }
+ }
+
+ /**
+ * Can create a {@link RepeatUnion} of
+ * the appropriate type for a rule's calling convention.
+ */
+ @Experimental
+ public interface RepeatUnionFactory {
+ /** Creates a {@link RepeatUnion}. */
+ RelNode createRepeatUnion(RelNode seed, RelNode iterative, boolean all, int maxRep);
+ }
+
+ /**
+ * Implementation of {@link RepeatUnion}
+ * that returns a {@link LogicalRepeatUnion}.
+ */
+ private static class RepeatUnionFactoryImpl implements RepeatUnionFactory {
+ public RelNode createRepeatUnion(RelNode seed, RelNode iterative, boolean all, int maxRep) {
+ return LogicalRepeatUnion.create(seed, iterative, all, maxRep);
+ }
+ }
}
// End RelFactories.java
diff --git a/core/src/main/java/org/apache/calcite/rel/core/RepeatUnion.java b/core/src/main/java/org/apache/calcite/rel/core/RepeatUnion.java
new file mode 100644
index 0000000..37e00dd
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/rel/core/RepeatUnion.java
@@ -0,0 +1,120 @@
+/*
+ * 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.calcite.rel.core;
+
+import org.apache.calcite.linq4j.function.Experimental;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.BiRel;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelWriter;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.util.Util;
+
+import com.google.common.collect.Lists;
+
+import java.util.List;
+
+/**
+ * Relational expression that computes a Repeat Union (Recursive Union in SQL terminology).
+ *
+ * This operation is executed as follows:
+ * <ul>
+ * <li>Evaluate the left input (i.e., seed relational expression) once.
+ * For UNION (but not UNION ALL), discard duplicated rows.</li>
+ * <li>Evaluate the right input (i.e., iterative relational expression) over and over until it
+ * produces no more results (or until an optional maximum number of iterations is reached).
+ * For UNION (but not UNION ALL), discard duplicated results.</li>
+ * </ul>
+ *
+ * <p>NOTE: The current API is experimental and subject to change without notice.</p>
+ */
+@Experimental
+public abstract class RepeatUnion extends BiRel {
+
+ /**
+ * Whether duplicates will be considered or not
+ */
+ public final boolean all;
+
+ /**
+ * Maximum number of times to repeat the iterative relational expression,
+ * -1 means no limit, 0 means only seed will be evaluated
+ */
+ public final int maxRep;
+
+
+
+ //~ Constructors -----------------------------------------------------------
+ protected RepeatUnion(
+ RelOptCluster cluster,
+ RelTraitSet traitSet,
+ RelNode seed,
+ RelNode iterative,
+ boolean all,
+ int maxRep) {
+
+ super(cluster, traitSet, seed, iterative);
+ if (maxRep < -1) {
+ throw new IllegalArgumentException("Wrong maxRep value");
+ }
+ this.maxRep = maxRep;
+ this.all = all;
+ }
+
+ @Override public double estimateRowCount(RelMetadataQuery mq) {
+ // TODO implement a more accurate row count?
+ double seedRowCount = mq.getRowCount(getSeedRel());
+ if (maxRep == 0) {
+ return seedRowCount;
+ }
+ return seedRowCount
+ + mq.getRowCount(getIterativeRel()) * (maxRep != -1 ? maxRep : 10);
+ }
+
+ @Override public RelWriter explainTerms(RelWriter pw) {
+ super.explainTerms(pw);
+ if (maxRep != -1) {
+ pw.item("maxRep", maxRep);
+ }
+ return pw.item("all", all);
+ }
+
+ public RelNode getSeedRel() {
+ return left;
+ }
+
+ public RelNode getIterativeRel() {
+ return right;
+ }
+
+ @Override protected RelDataType deriveRowType() {
+ final List<RelDataType> inputRowTypes =
+ Lists.transform(getInputs(), RelNode::getRowType);
+ final RelDataType rowType =
+ getCluster().getTypeFactory().leastRestrictive(inputRowTypes);
+ if (rowType == null) {
+ throw new IllegalArgumentException("Cannot compute compatible row type "
+ + "for arguments: "
+ + Util.sepList(inputRowTypes, ", "));
+ }
+ return rowType;
+ }
+}
+
+// End RepeatUnion.java
diff --git a/core/src/main/java/org/apache/calcite/rel/core/Spool.java b/core/src/main/java/org/apache/calcite/rel/core/Spool.java
new file mode 100644
index 0000000..2287364
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/rel/core/Spool.java
@@ -0,0 +1,87 @@
+/*
+ * 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.calcite.rel.core;
+
+import org.apache.calcite.linq4j.function.Experimental;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelWriter;
+import org.apache.calcite.rel.SingleRel;
+
+import java.util.List;
+
+/**
+ * Relational expression that iterates over its input and, apart from returning its results,
+ * will forward them into other consumers.
+ *
+ * <p>NOTE: The current API is experimental and subject to change without notice.</p>
+ */
+@Experimental
+public abstract class Spool extends SingleRel {
+
+ /**
+ * Enumeration representing spool read / write type.
+ */
+ public enum Type {
+ EAGER,
+ LAZY
+ }
+
+ /**
+ * The way the spool consumes elements from its input.
+ * <ul>
+ * <li>EAGER: the spool will consume the elements from its input at once at the initial request.
+ * </li>
+ * <li>LAZY: the spool will consume the elements from its input one by one by request.</li>
+ * </ul>
+ */
+ public final Type readType;
+ /**
+ * The way the spool forwards elements to consumers.
+ * <ul>
+ * <li>EAGER: the spool will forward each element as soon as it returns it.</li>
+ * <li>LAZY: the spool will forward all elements at once when it is done retuning all of them.
+ * </li>
+ * </ul>
+ */
+ public final Type writeType;
+
+ //~ Constructors -----------------------------------------------------------
+ protected Spool(RelOptCluster cluster, RelTraitSet traitSet, RelNode input,
+ Type readType, Type writeType) {
+ super(cluster, traitSet, input);
+ this.readType = readType;
+ this.writeType = writeType;
+ }
+
+ @Override public final RelNode copy(RelTraitSet traitSet,
+ List<RelNode> inputs) {
+ return copy(traitSet, sole(inputs), readType, writeType);
+ }
+
+ protected abstract Spool copy(RelTraitSet traitSet, RelNode input,
+ Type readType, Type writeType);
+
+ @Override public RelWriter explainTerms(RelWriter pw) {
+ return super.explainTerms(pw)
+ .item("readType", readType)
+ .item("writeType", writeType);
+ }
+}
+
+// End Spool.java
diff --git a/core/src/main/java/org/apache/calcite/rel/core/TableSpool.java b/core/src/main/java/org/apache/calcite/rel/core/TableSpool.java
new file mode 100644
index 0000000..1e46a81
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/rel/core/TableSpool.java
@@ -0,0 +1,51 @@
+/*
+ * 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.calcite.rel.core;
+
+import org.apache.calcite.linq4j.function.Experimental;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelWriter;
+
+/**
+ * Spool that writes into a table.
+ *
+ * <p>NOTE: The current API is experimental and subject to change without notice.</p>
+ */
+@Experimental
+public abstract class TableSpool extends Spool {
+
+ protected final String tableName;
+
+ protected TableSpool(RelOptCluster cluster, RelTraitSet traitSet, RelNode input,
+ Type readType, Type writeType, String tableName) {
+ super(cluster, traitSet, input, readType, writeType);
+ this.tableName = tableName;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ @Override public RelWriter explainTerms(RelWriter pw) {
+ super.explainTerms(pw);
+ return pw.item("tableName", tableName);
+ }
+}
+
+// End TableSpool.java
diff --git a/core/src/main/java/org/apache/calcite/rel/logical/LogicalRepeatUnion.java b/core/src/main/java/org/apache/calcite/rel/logical/LogicalRepeatUnion.java
new file mode 100644
index 0000000..d2dd729
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/rel/logical/LogicalRepeatUnion.java
@@ -0,0 +1,72 @@
+/*
+ * 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.calcite.rel.logical;
+
+import org.apache.calcite.linq4j.function.Experimental;
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.RepeatUnion;
+
+import java.util.List;
+
+/**
+ * Sub-class of {@link org.apache.calcite.rel.core.RepeatUnion}
+ * not targeted at any particular engine or calling convention.
+ *
+ * <p>NOTE: The current API is experimental and subject to change without notice.</p>
+ */
+@Experimental
+public class LogicalRepeatUnion extends RepeatUnion {
+
+ //~ Constructors -----------------------------------------------------------
+ private LogicalRepeatUnion(
+ RelOptCluster cluster,
+ RelTraitSet traitSet,
+ RelNode seed,
+ RelNode iterative,
+ boolean all,
+ int maxRep) {
+ super(cluster, traitSet, seed, iterative, all, maxRep);
+ }
+
+ /** Creates a LogicalRepeatUnion. */
+ public static LogicalRepeatUnion create(RelNode seed, RelNode iterative, boolean all) {
+ return create(seed, iterative, all, -1);
+ }
+
+ /** Creates a LogicalRepeatUnion. */
+ public static LogicalRepeatUnion create(RelNode seed, RelNode iterative, boolean all,
+ int maxRep) {
+ RelOptCluster cluster = seed.getCluster();
+ RelTraitSet traitSet = cluster.traitSetOf(Convention.NONE);
+ return new LogicalRepeatUnion(cluster, traitSet, seed, iterative, all, maxRep);
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ @Override public LogicalRepeatUnion copy(RelTraitSet traitSet, List<RelNode> inputs) {
+ assert traitSet.containsIfApplicable(Convention.NONE);
+ assert inputs.size() == 2;
+ return new LogicalRepeatUnion(getCluster(), traitSet,
+ inputs.get(0), inputs.get(1), all, maxRep);
+ }
+
+}
+
+// End LogicalRepeatUnion.java
diff --git a/core/src/main/java/org/apache/calcite/rel/logical/LogicalTableSpool.java b/core/src/main/java/org/apache/calcite/rel/logical/LogicalTableSpool.java
new file mode 100644
index 0000000..b0a2a14
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/rel/logical/LogicalTableSpool.java
@@ -0,0 +1,66 @@
+/*
+ * 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.calcite.rel.logical;
+
+import org.apache.calcite.linq4j.function.Experimental;
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelCollationTraitDef;
+import org.apache.calcite.rel.RelDistributionTraitDef;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Spool;
+import org.apache.calcite.rel.core.TableSpool;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+
+/**
+ * Sub-class of {@link TableSpool} not targeted at any particular engine or calling convention.
+ *
+ * <p>NOTE: The current API is experimental and subject to change without notice.</p>
+ */
+@Experimental
+public class LogicalTableSpool extends TableSpool {
+
+ //~ Constructors -----------------------------------------------------------
+ public LogicalTableSpool(RelOptCluster cluster, RelTraitSet traitSet, RelNode input,
+ Type readType, Type writeType, String tableName) {
+ super(cluster, traitSet, input, readType, writeType, tableName);
+ }
+
+ /** Creates a LogicalTableSpool. */
+ public static LogicalTableSpool create(RelNode input, Type readType, Type writeType,
+ String tableName) {
+ RelOptCluster cluster = input.getCluster();
+ RelMetadataQuery mq = cluster.getMetadataQuery();
+ RelTraitSet traitSet = cluster.traitSetOf(Convention.NONE)
+ .replaceIfs(RelCollationTraitDef.INSTANCE,
+ () -> mq.collations(input))
+ .replaceIf(RelDistributionTraitDef.INSTANCE,
+ () -> mq.distribution(input));
+ return new LogicalTableSpool(cluster, traitSet, input, readType, writeType, tableName);
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ @Override protected Spool copy(RelTraitSet traitSet, RelNode input,
+ Type readType, Type writeType) {
+ return new LogicalTableSpool(input.getCluster(), traitSet, input,
+ readType, writeType, tableName);
+ }
+}
+
+// End LogicalTableSpool.java
diff --git a/core/src/main/java/org/apache/calcite/schema/TransientTable.java b/core/src/main/java/org/apache/calcite/schema/TransientTable.java
new file mode 100644
index 0000000..b3a3702
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/schema/TransientTable.java
@@ -0,0 +1,33 @@
+/*
+ * 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.calcite.schema;
+
+import org.apache.calcite.linq4j.function.Experimental;
+
+/**
+ * A transient table is a named table that may come into existence implicitly during the
+ * evaluation of a query expression or the execution of a trigger. A transient table is
+ * identified by a query name if it arises during the evaluation of a query expression,
+ * or by a transition table name if it arises during the execution of a trigger.
+ * Such tables exist only for the duration of the executing SQL-statement containing the
+ * query expression or for the duration of the executing trigger.
+ */
+@Experimental
+public interface TransientTable extends Table {
+}
+
+// End TransientTable.java
diff --git a/core/src/main/java/org/apache/calcite/schema/impl/ListTransientTable.java b/core/src/main/java/org/apache/calcite/schema/impl/ListTransientTable.java
new file mode 100644
index 0000000..87dcbd4
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/schema/impl/ListTransientTable.java
@@ -0,0 +1,150 @@
+/*
+ * 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.calcite.schema.impl;
+
+import org.apache.calcite.DataContext;
+import org.apache.calcite.adapter.java.AbstractQueryableTable;
+import org.apache.calcite.linq4j.AbstractEnumerable;
+import org.apache.calcite.linq4j.Enumerable;
+import org.apache.calcite.linq4j.Enumerator;
+import org.apache.calcite.linq4j.Linq4j;
+import org.apache.calcite.linq4j.QueryProvider;
+import org.apache.calcite.linq4j.Queryable;
+import org.apache.calcite.linq4j.function.Experimental;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.prepare.Prepare;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.TableModify;
+import org.apache.calcite.rel.logical.LogicalTableModify;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.schema.ModifiableTable;
+import org.apache.calcite.schema.ScannableTable;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.Schemas;
+import org.apache.calcite.schema.TransientTable;
+
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * {@link TransientTable} backed by a Java list. It will be automatically added to the
+ * current schema when {@link #scan(DataContext)} method gets called.
+ *
+ * <p>NOTE: The current API is experimental and subject to change without notice.</p>
+ */
+@Experimental
+public class ListTransientTable extends AbstractQueryableTable
+ implements TransientTable, ModifiableTable, ScannableTable {
+ private static final Type TYPE = Object[].class;
+ private final List rows = new ArrayList();
+ private final String name;
+ private final RelDataType protoRowType;
+
+ public ListTransientTable(String name, RelDataType rowType) {
+ super(TYPE);
+ this.name = name;
+ this.protoRowType = rowType;
+ }
+
+ @Override public TableModify toModificationRel(
+ RelOptCluster cluster,
+ RelOptTable table,
+ Prepare.CatalogReader catalogReader,
+ RelNode child,
+ TableModify.Operation operation,
+ List<String> updateColumnList,
+ List<RexNode> sourceExpressionList,
+ boolean flattened) {
+ return LogicalTableModify.create(table, catalogReader, child, operation,
+ updateColumnList, sourceExpressionList, flattened);
+ }
+
+ @Override public Collection getModifiableCollection() {
+ return rows;
+ }
+
+ @Override public Enumerable<Object[]> scan(DataContext root) {
+ // add the table into the schema, so that it is accessible by any potential operator
+ root.getRootSchema().add(name, this);
+
+ final AtomicBoolean cancelFlag = DataContext.Variable.CANCEL_FLAG.get(root);
+
+ return new AbstractEnumerable<Object[]>() {
+ public Enumerator<Object[]> enumerator() {
+ return new Enumerator<Object[]>() {
+ private final List list = new ArrayList(rows);
+ private int i = -1;
+
+ // TODO cleaner way to handle non-array objects?
+ @Override public Object[] current() {
+ Object current = list.get(i);
+ return current.getClass().isArray()
+ ? (Object[]) current
+ : new Object[]{current};
+ }
+
+ @Override public boolean moveNext() {
+ if (cancelFlag != null && cancelFlag.get()) {
+ return false;
+ }
+
+ return ++i < list.size();
+ }
+
+ @Override public void reset() {
+ i = -1;
+ }
+
+ @Override public void close() {
+ }
+ };
+ }
+ };
+ }
+
+ public Expression getExpression(SchemaPlus schema, String tableName,
+ Class clazz) {
+ return Schemas.tableExpression(schema, elementType, tableName, clazz);
+ }
+
+ @Override public <T> Queryable<T> asQueryable(QueryProvider queryProvider,
+ SchemaPlus schema, String tableName) {
+ return new AbstractTableQueryable<T>(queryProvider, schema, this, tableName) {
+ public Enumerator<T> enumerator() {
+ //noinspection unchecked
+ return (Enumerator<T>) Linq4j.enumerator(rows);
+ }
+ };
+ }
+
+ @Override public RelDataType getRowType(RelDataTypeFactory typeFactory) {
+ return typeFactory.copyType(protoRowType);
+ }
+
+ @Override public Type getElementType() {
+ return TYPE;
+ }
+}
+
+// End ListTransientTable.java
diff --git a/core/src/main/java/org/apache/calcite/tools/RelBuilder.java b/core/src/main/java/org/apache/calcite/tools/RelBuilder.java
index a33b807..4174736 100644
--- a/core/src/main/java/org/apache/calcite/tools/RelBuilder.java
+++ b/core/src/main/java/org/apache/calcite/tools/RelBuilder.java
@@ -25,6 +25,7 @@ import org.apache.calcite.plan.RelOptPredicateList;
import org.apache.calcite.plan.RelOptSchema;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.prepare.RelOptTableImpl;
import org.apache.calcite.rel.RelCollation;
import org.apache.calcite.rel.RelCollations;
import org.apache.calcite.rel.RelDistribution;
@@ -41,11 +42,14 @@ import org.apache.calcite.rel.core.Match;
import org.apache.calcite.rel.core.Minus;
import org.apache.calcite.rel.core.Project;
import org.apache.calcite.rel.core.RelFactories;
+import org.apache.calcite.rel.core.RepeatUnion;
import org.apache.calcite.rel.core.SemiJoin;
import org.apache.calcite.rel.core.Snapshot;
import org.apache.calcite.rel.core.Sort;
+import org.apache.calcite.rel.core.Spool;
import org.apache.calcite.rel.core.TableFunctionScan;
import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.core.TableSpool;
import org.apache.calcite.rel.core.Union;
import org.apache.calcite.rel.core.Values;
import org.apache.calcite.rel.logical.LogicalFilter;
@@ -68,6 +72,8 @@ import org.apache.calcite.rex.RexSimplify;
import org.apache.calcite.rex.RexUtil;
import org.apache.calcite.runtime.Hook;
import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.TransientTable;
+import org.apache.calcite.schema.impl.ListTransientTable;
import org.apache.calcite.server.CalciteServerStatement;
import org.apache.calcite.sql.SemiJoinType;
import org.apache.calcite.sql.SqlAggFunction;
@@ -150,6 +156,8 @@ public class RelBuilder {
private final RelFactories.TableFunctionScanFactory tableFunctionScanFactory;
private final RelFactories.SnapshotFactory snapshotFactory;
private final RelFactories.MatchFactory matchFactory;
+ private final RelFactories.SpoolFactory spoolFactory;
+ private final RelFactories.RepeatUnionFactory repeatUnionFactory;
private final Deque<Frame> stack = new ArrayDeque<>();
private final boolean simplify;
private final RexSimplify simplifier;
@@ -207,6 +215,12 @@ public class RelBuilder {
this.matchFactory =
Util.first(context.unwrap(RelFactories.MatchFactory.class),
RelFactories.DEFAULT_MATCH_FACTORY);
+ this.spoolFactory =
+ Util.first(context.unwrap(RelFactories.SpoolFactory.class),
+ RelFactories.DEFAULT_SPOOL_FACTORY);
+ this.repeatUnionFactory =
+ Util.first(context.unwrap(RelFactories.RepeatUnionFactory.class),
+ RelFactories.DEFAULT_REPEAT_UNION_FACTORY);
final RexExecutor executor =
Util.first(context.unwrap(RexExecutor.class),
Util.first(cluster.getPlanner().getExecutor(), RexUtil.EXECUTOR));
@@ -1721,6 +1735,85 @@ public class RelBuilder {
return setOp(all, SqlKind.EXCEPT, n);
}
+ /**
+ * Creates a {@link TableScan} on a {@link TransientTable} with the given name, using as type
+ * the top of the stack's type.
+ *
+ * @param tableName table name
+ */
+ @Experimental
+ public RelBuilder transientScan(String tableName) {
+ return this.transientScan(tableName, this.peek().getRowType());
+ }
+
+ /**
+ * Creates a {@link TableScan} on a {@link TransientTable} with the given name and type.
+ *
+ * @param tableName table name
+ * @param rowType row type of the table
+ */
+ @Experimental
+ public RelBuilder transientScan(String tableName, RelDataType rowType) {
+ ListTransientTable transientTable = new ListTransientTable(tableName, rowType);
+ RelOptTable relOptTable = RelOptTableImpl.create(
+ relOptSchema,
+ rowType,
+ transientTable,
+ ImmutableList.of(tableName));
+ RelNode scan = scanFactory.createScan(cluster, relOptTable);
+ push(scan);
+ rename(rowType.getFieldNames());
+ return this;
+ }
+
+ /**
+ * Creates a {@link TableSpool} for the most recent relation expression.
+ *
+ * @param readType spool's read type (as described in {@link Spool.Type})
+ * @param writeType spool's write type (as described in {@link Spool.Type})
+ * @param tableName table name
+ */
+ private RelBuilder tableSpool(Spool.Type readType, Spool.Type writeType, String tableName) {
+ RelNode spool = spoolFactory.createTableSpool(peek(), readType, writeType, tableName);
+ replaceTop(spool);
+ return this;
+ }
+
+ /**
+ * Creates a {@link RepeatUnion} associated to a {@link TransientTable} without a maximum number
+ * of iterations, i.e. repeatUnion(tableName, all, -1).
+ *
+ * @param tableName name of the {@link TransientTable} associated to the {@link RepeatUnion}
+ * @param all whether duplicates will be considered or not
+ */
+ @Experimental
+ public RelBuilder repeatUnion(String tableName, boolean all) {
+ return repeatUnion(tableName, all, -1);
+ }
+
+ /**
+ * Creates a {@link RepeatUnion} associated to a {@link TransientTable} of the two most recent
+ * relational expressions on the stack. Warning: if these relational expressions are not
+ * correctly defined, this operation might lead to an infinite loop.
+ * The generated {@link RepeatUnion} will:
+ * - Evaluate its left term once, propagating the results into the {@link TransientTable}.
+ * - Evaluate its right term (which may contain a {@link TableScan} on the
+ * {@link TransientTable}) over and over until it produces no more results (or until an
+ * optional maximum number of iterations is reached). On each iteration, the results will be
+ * propagated into the {@link TransientTable}, overwriting the results from the previous one.
+ *
+ * @param tableName name of the {@link TransientTable} associated to the {@link RepeatUnion}
+ * @param all whether duplicates will be considered or not
+ * @param maxRep maximum number of iterations, -1 means no limit
+ */
+ @Experimental
+ public RelBuilder repeatUnion(String tableName, boolean all, int maxRep) {
+ RelNode iterative = tableSpool(Spool.Type.LAZY, Spool.Type.LAZY, tableName).build();
+ RelNode seed = tableSpool(Spool.Type.LAZY, Spool.Type.LAZY, tableName).build();
+ RelNode repeatUnion = repeatUnionFactory.createRepeatUnion(seed, iterative, all, maxRep);
+ return push(repeatUnion);
+ }
+
/** Creates a {@link Join}. */
public RelBuilder join(JoinRelType joinType, RexNode condition0,
RexNode... conditions) {
diff --git a/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java b/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java
index 2ef7a35..bde4d83 100644
--- a/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java
+++ b/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java
@@ -189,6 +189,10 @@ public enum BuiltInMethod {
Comparator.class),
UNION(ExtendedEnumerable.class, "union", Enumerable.class),
CONCAT(ExtendedEnumerable.class, "concat", Enumerable.class),
+ REPEAT_UNION_ALL(EnumerableDefaults.class, "repeatUnionAll", Enumerable.class,
+ Enumerable.class, int.class),
+ LAZY_COLLECTION_SPOOL(EnumerableDefaults.class, "lazyCollectionSpool", Collection.class,
+ Enumerable.class),
INTERSECT(ExtendedEnumerable.class, "intersect", Enumerable.class),
EXCEPT(ExtendedEnumerable.class, "except", Enumerable.class),
SKIP(ExtendedEnumerable.class, "skip", int.class),
diff --git a/core/src/test/java/org/apache/calcite/test/CalciteSuite.java b/core/src/test/java/org/apache/calcite/test/CalciteSuite.java
index bced1bf..54bd512 100644
--- a/core/src/test/java/org/apache/calcite/test/CalciteSuite.java
+++ b/core/src/test/java/org/apache/calcite/test/CalciteSuite.java
@@ -59,6 +59,8 @@ import org.apache.calcite.sql.validate.LexCaseSensitiveTest;
import org.apache.calcite.sql.validate.LexEscapeTest;
import org.apache.calcite.sql.validate.SqlValidatorUtilTest;
import org.apache.calcite.test.enumerable.EnumerableCorrelateTest;
+import org.apache.calcite.test.enumerable.EnumerableRepeatUnionHierarchyTest;
+import org.apache.calcite.test.enumerable.EnumerableRepeatUnionTest;
import org.apache.calcite.test.fuzzer.RexProgramFuzzyTest;
import org.apache.calcite.tools.FrameworksTest;
import org.apache.calcite.tools.PlannerTest;
@@ -188,7 +190,8 @@ import org.junit.runners.Suite;
// above 10sec
JdbcFrontJdbcBackTest.class,
-
+ EnumerableRepeatUnionTest.class,
+ EnumerableRepeatUnionHierarchyTest.class,
// above 20sec
JdbcTest.class,
CalciteSqlOperatorTest.class,
diff --git a/core/src/test/java/org/apache/calcite/test/HierarchySchema.java b/core/src/test/java/org/apache/calcite/test/HierarchySchema.java
new file mode 100644
index 0000000..debc578
--- /dev/null
+++ b/core/src/test/java/org/apache/calcite/test/HierarchySchema.java
@@ -0,0 +1,91 @@
+/*
+ * 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.calcite.test;
+
+import java.util.Arrays;
+import java.util.Objects;
+
+/**
+ * A Schema representing a hierarchy of employees.
+ *
+ * <p>The Schema is meant to be used with
+ * {@link org.apache.calcite.adapter.java.ReflectiveSchema} thus all
+ * fields, and methods, should be public.
+ */
+public class HierarchySchema {
+ @Override public String toString() {
+ return "HierarchySchema";
+ }
+
+ public final JdbcTest.Employee[] emps = {
+ new JdbcTest.Employee(1, 10, "Emp1", 10000, 1000),
+ new JdbcTest.Employee(2, 10, "Emp2", 8000, 500),
+ new JdbcTest.Employee(3, 10, "Emp3", 7000, null),
+ new JdbcTest.Employee(4, 10, "Emp4", 8000, 500),
+ new JdbcTest.Employee(5, 10, "Emp5", 7000, null),
+ };
+
+ public final JdbcTest.Department[] depts = {
+ new JdbcTest.Department(
+ 10,
+ "Dept",
+ Arrays.asList(emps[0], emps[1], emps[2], emps[3], emps[4]),
+ new JdbcTest.Location(-122, 38)),
+ };
+
+ // Emp1
+ // / \
+ // Emp2 Emp4
+ // / \
+ // Emp3 Emp5
+ public final Hierarchy[] hierarchies = {
+ new Hierarchy(1, 2),
+ new Hierarchy(2, 3),
+ new Hierarchy(2, 5),
+ new Hierarchy(1, 4),
+ };
+
+ /**
+ * Hierarchy representing manager - subordinate
+ */
+ public static class Hierarchy {
+ public final int managerid;
+ public final int subordinateid;
+
+ public Hierarchy(int managerid, int subordinateid) {
+ this.managerid = managerid;
+ this.subordinateid = subordinateid;
+ }
+
+ @Override public String toString() {
+ return "Hierarchy [managerid: " + managerid + ", subordinateid: " + subordinateid + "]";
+ }
+
+ @Override public boolean equals(Object obj) {
+ return obj == this
+ || obj instanceof Hierarchy
+ && managerid == ((Hierarchy) obj).managerid
+ && subordinateid == ((Hierarchy) obj).subordinateid;
+ }
+
+ @Override public int hashCode() {
+ return Objects.hash(managerid, subordinateid);
+ }
+ }
+}
+
+// End HierarchySchema.java
diff --git a/core/src/test/java/org/apache/calcite/test/RelBuilderTest.java b/core/src/test/java/org/apache/calcite/test/RelBuilderTest.java
index a9f7341..06ea18c 100644
--- a/core/src/test/java/org/apache/calcite/test/RelBuilderTest.java
+++ b/core/src/test/java/org/apache/calcite/test/RelBuilderTest.java
@@ -1285,6 +1285,79 @@ public class RelBuilderTest {
assertThat(root, hasTree(expected));
}
+ @Test public void testRepeatUnion1() {
+ // Generates the sequence 1,2,3,...10 using a repeat union. Equivalent SQL:
+ // WITH RECURSIVE delta(n) AS (
+ // VALUES (1)
+ // UNION ALL
+ // SELECT n+1 FROM delta WHERE n < 10
+ // )
+ // SELECT * FROM delta
+ final RelBuilder builder = RelBuilder.create(config().build());
+ RelNode root =
+ builder.values(new String[] { "i" }, 1)
+ .transientScan("DELTA_TABLE")
+ .filter(
+ builder.call(
+ SqlStdOperatorTable.LESS_THAN,
+ builder.field(0),
+ builder.literal(10)))
+ .project(
+ builder.call(SqlStdOperatorTable.PLUS,
+ builder.field(0),
+ builder.literal(1)))
+ .repeatUnion("DELTA_TABLE", true)
+ .build();
+ final String expected = "LogicalRepeatUnion(all=[true])\n"
+ + " LogicalTableSpool(readType=[LAZY], writeType=[LAZY], tableName=[DELTA_TABLE])\n"
+ + " LogicalValues(tuples=[[{ 1 }]])\n"
+ + " LogicalTableSpool(readType=[LAZY], writeType=[LAZY], tableName=[DELTA_TABLE])\n"
+ + " LogicalProject($f0=[+($0, 1)])\n"
+ + " LogicalFilter(condition=[<($0, 10)])\n"
+ + " LogicalTableScan(table=[[DELTA_TABLE]])\n";
+ assertThat(root, hasTree(expected));
+ }
+
+ @Test public void testRepeatUnion2() {
+ // Generates the factorial function from 0 to 7. Equivalent SQL:
+ // WITH RECURSIVE delta (n, fact) AS (
+ // VALUES (0, 1)
+ // UNION ALL
+ // SELECT n+1, (n+1)*fact FROM delta WHERE n < 7
+ // )
+ // SELECT * FROM delta
+ final RelBuilder builder = RelBuilder.create(config().build());
+ RelNode root =
+ builder.values(new String[] { "n", "fact" }, 0, 1)
+ .transientScan("AUX")
+ .filter(
+ builder.call(
+ SqlStdOperatorTable.LESS_THAN,
+ builder.field("n"),
+ builder.literal(7)))
+ .project(
+ Arrays.asList(
+ builder.call(SqlStdOperatorTable.PLUS,
+ builder.field("n"),
+ builder.literal(1)),
+ builder.call(SqlStdOperatorTable.MULTIPLY,
+ builder.call(SqlStdOperatorTable.PLUS,
+ builder.field("n"),
+ builder.literal(1)),
+ builder.field("fact"))),
+ Arrays.asList("n", "fact"))
+ .repeatUnion("AUX", true)
+ .build();
+ final String expected = "LogicalRepeatUnion(all=[true])\n"
+ + " LogicalTableSpool(readType=[LAZY], writeType=[LAZY], tableName=[AUX])\n"
+ + " LogicalValues(tuples=[[{ 0, 1 }]])\n"
+ + " LogicalTableSpool(readType=[LAZY], writeType=[LAZY], tableName=[AUX])\n"
+ + " LogicalProject(n=[+($0, 1)], fact=[*(+($0, 1), $1)])\n"
+ + " LogicalFilter(condition=[<($0, 7)])\n"
+ + " LogicalTableScan(table=[[AUX]])\n";
+ assertThat(root, hasTree(expected));
+ }
+
@Test public void testIntersect() {
// Equivalent SQL:
// SELECT empno FROM emp
diff --git a/core/src/test/java/org/apache/calcite/test/enumerable/EnumerableRepeatUnionHierarchyTest.java b/core/src/test/java/org/apache/calcite/test/enumerable/EnumerableRepeatUnionHierarchyTest.java
new file mode 100644
index 0000000..4274c2a
--- /dev/null
+++ b/core/src/test/java/org/apache/calcite/test/enumerable/EnumerableRepeatUnionHierarchyTest.java
@@ -0,0 +1,153 @@
+/*
+ * 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.calcite.test.enumerable;
+
+import org.apache.calcite.adapter.enumerable.EnumerableRepeatUnion;
+import org.apache.calcite.adapter.java.ReflectiveSchema;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.schema.Schema;
+import org.apache.calcite.test.CalciteAssert;
+import org.apache.calcite.test.HierarchySchema;
+import org.apache.calcite.tools.RelBuilder;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.Arrays;
+import java.util.function.Function;
+
+/**
+ * Unit tests for
+ * {@link EnumerableRepeatUnion}
+ * <a href="https://issues.apache.org/jira/browse/CALCITE-2812">[CALCITE-2812]
+ * Add algebraic operators to allow expressing recursive queries</a>.
+ */
+@RunWith(Parameterized.class)
+public class EnumerableRepeatUnionHierarchyTest {
+
+ // Tests for the following hierarchy:
+ // Emp1
+ // / \
+ // Emp2 Emp4
+ // / \
+ // Emp3 Emp5
+
+ private static final String EMP1 = "empid=1; name=Emp1";
+ private static final String EMP2 = "empid=2; name=Emp2";
+ private static final String EMP3 = "empid=3; name=Emp3";
+ private static final String EMP4 = "empid=4; name=Emp4";
+ private static final String EMP5 = "empid=5; name=Emp5";
+
+ @Parameterized.Parameters(name = "{index} : hierarchy(startId:{0}, ascendant:{1}, maxDepth:{2})")
+ public static Iterable<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ { 1, true, -1, new String[]{EMP1} },
+ { 2, true, -1, new String[]{EMP2, EMP1} },
+ { 3, true, -1, new String[]{EMP3, EMP2, EMP1} },
+ { 4, true, -1, new String[]{EMP4, EMP1} },
+ { 5, true, -1, new String[]{EMP5, EMP2, EMP1} },
+ { 3, true, 0, new String[]{EMP3} },
+ { 3, true, 1, new String[]{EMP3, EMP2} },
+ { 3, true, 2, new String[]{EMP3, EMP2, EMP1} },
+ { 3, true, 10, new String[]{EMP3, EMP2, EMP1} },
+
+ { 1, false, -1, new String[]{EMP1, EMP2, EMP4, EMP3, EMP5} },
+ { 2, false, -1, new String[]{EMP2, EMP3, EMP5} },
+ { 3, false, -1, new String[]{EMP3} },
+ { 4, false, -1, new String[]{EMP4} },
+ { 1, false, 0, new String[]{EMP1} },
+ { 1, false, 1, new String[]{EMP1, EMP2, EMP4} },
+ { 1, false, 2, new String[]{EMP1, EMP2, EMP4, EMP3, EMP5} },
+ { 1, false, 20, new String[]{EMP1, EMP2, EMP4, EMP3, EMP5} },
+ });
+ }
+
+ private final int startId;
+ private final int maxDepth;
+ private final String fromField;
+ private final String toField;
+ private final String[] expected;
+
+ public EnumerableRepeatUnionHierarchyTest(int startId, boolean ascendant,
+ int maxDepth, String[] expected) {
+ this.startId = startId;
+ this.maxDepth = maxDepth;
+ this.expected = expected;
+
+ if (ascendant) {
+ this.fromField = "subordinateid";
+ this.toField = "managerid";
+ } else {
+ this.fromField = "managerid";
+ this.toField = "subordinateid";
+ }
+ }
+
+ @Test public void testHierarchy() {
+ final Schema schema = new ReflectiveSchema(new HierarchySchema());
+ CalciteAssert.that()
+ .withSchema("s", schema)
+ .query("?")
+ .withRel(hierarchy())
+ .returnsOrdered(expected);
+ }
+
+ private Function<RelBuilder, RelNode> hierarchy() {
+
+ // WITH RECURSIVE delta(empid, name) as (
+ // SELECT empid, name FROM emps WHERE empid = <startId>
+ // UNION ALL
+ // SELECT e.empid, e.name FROM delta d
+ // JOIN hierarchies h ON d.empid = h.<fromField>
+ // JOIN emps e ON h.<toField> = e.empid
+ // )
+ // SELECT empid, name FROM delta
+ return builder -> builder
+ .scan("s", "emps")
+ .filter(
+ builder.equals(
+ builder.field("empid"),
+ builder.literal(startId)))
+ .project(
+ builder.field("emps", "empid"),
+ builder.field("emps", "name"))
+
+ .transientScan("#DELTA#")
+ .scan("s", "hierarchies")
+ .join(
+ JoinRelType.INNER,
+ builder.equals(
+ builder.field(2, "#DELTA#", "empid"),
+ builder.field(2, "hierarchies", fromField)))
+ .scan("s", "emps")
+ .join(
+ JoinRelType.INNER,
+ builder.equals(
+ builder.field(2, "hierarchies", toField),
+ builder.field(2, "emps", "empid")))
+ .project(
+ builder.field("emps", "empid"),
+ builder.field("emps", "name"))
+ .repeatUnion("#DELTA#", true, maxDepth)
+ .build();
+ }
+
+}
+
+// End EnumerableRepeatUnionHierarchyTest.java
diff --git a/core/src/test/java/org/apache/calcite/test/enumerable/EnumerableRepeatUnionTest.java b/core/src/test/java/org/apache/calcite/test/enumerable/EnumerableRepeatUnionTest.java
new file mode 100644
index 0000000..d1fd700
--- /dev/null
+++ b/core/src/test/java/org/apache/calcite/test/enumerable/EnumerableRepeatUnionTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.calcite.test.enumerable;
+
+import org.apache.calcite.adapter.enumerable.EnumerableRepeatUnion;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.test.CalciteAssert;
+
+import org.junit.Test;
+
+import java.util.Arrays;
+
+/**
+ * Unit tests for
+ * {@link EnumerableRepeatUnion}
+ * <a href="https://issues.apache.org/jira/browse/CALCITE-2812">[CALCITE-2812]
+ * Add algebraic operators to allow expressing recursive queries</a>.
+ */
+public class EnumerableRepeatUnionTest {
+
+ @Test public void testGenerateNumbers() {
+ CalciteAssert.that()
+ .query("?")
+ .withRel(
+ // WITH RECURSIVE delta(n) AS (
+ // VALUES (1)
+ // UNION ALL
+ // SELECT n+1 FROM delta WHERE n < 10
+ // )
+ // SELECT * FROM delta
+ builder -> builder
+ .values(new String[] { "i" }, 1)
+ .transientScan("DELTA")
+ .filter(
+ builder.call(
+ SqlStdOperatorTable.LESS_THAN,
+ builder.field(0),
+ builder.literal(10)))
+ .project(
+ builder.call(SqlStdOperatorTable.PLUS,
+ builder.field(0),
+ builder.literal(1)))
+ .repeatUnion("DELTA", true)
+ .build()
+ )
+ .returnsOrdered("i=1", "i=2", "i=3", "i=4", "i=5", "i=6", "i=7", "i=8", "i=9", "i=10");
+ }
+
+ @Test public void testFactorial() {
+ CalciteAssert.that()
+ .query("?")
+ .withRel(
+ // WITH RECURSIVE d(n, fact) AS (
+ // VALUES (0, 1)
+ // UNION ALL
+ // SELECT n+1, (n+1)*fact FROM d WHERE n < 7
+ // )
+ // SELECT * FROM delta
+ builder -> builder
+ .values(new String[] { "n", "fact" }, 0, 1)
+ .transientScan("D")
+ .filter(
+ builder.call(
+ SqlStdOperatorTable.LESS_THAN,
+ builder.field("n"),
+ builder.literal(7)))
+ .project(
+ Arrays.asList(
+ builder.call(SqlStdOperatorTable.PLUS,
+ builder.field("n"),
+ builder.literal(1)),
+ builder.call(SqlStdOperatorTable.MULTIPLY,
+ builder.call(SqlStdOperatorTable.PLUS,
+ builder.field("n"),
+ builder.literal(1)),
+ builder.field("fact"))),
+ Arrays.asList("n", "fact"))
+ .repeatUnion("D", true)
+ .build()
+ )
+ .returnsOrdered("n=0; fact=1",
+ "n=1; fact=1",
+ "n=2; fact=2",
+ "n=3; fact=6",
+ "n=4; fact=24",
+ "n=5; fact=120",
+ "n=6; fact=720",
+ "n=7; fact=5040");
+ }
+
+ @Test public void testGenerateNumbersNestedRecursion() {
+ CalciteAssert.that()
+ .query("?")
+ .withRel(
+ // WITH RECURSIVE t_out(n) AS (
+ // WITH RECURSIVE t_in(n) AS (
+ // VALUES (1)
+ // UNION ALL
+ // SELECT n+1 FROM t_in WHERE n < 9
+ // )
+ // SELECT n FROM t_in
+ // UNION ALL
+ // SELECT n*10 FROM t_out WHERE n < 100
+ // )
+ // SELECT n FROM t_out
+ builder -> builder
+ .values(new String[] { "n" }, 1)
+ .transientScan("T_IN")
+ .filter(
+ builder.call(
+ SqlStdOperatorTable.LESS_THAN,
+ builder.field("n"),
+ builder.literal(9)))
+ .project(
+ builder.call(
+ SqlStdOperatorTable.PLUS,
+ builder.field("n"),
+ builder.literal(1)))
+ .repeatUnion("T_IN", true)
+
+ .transientScan("T_OUT")
+ .filter(
+ builder.call(
+ SqlStdOperatorTable.LESS_THAN,
+ builder.field("n"),
+ builder.literal(100)))
+ .project(
+ builder.call(
+ SqlStdOperatorTable.MULTIPLY,
+ builder.field("n"),
+ builder.literal(10)))
+ .repeatUnion("T_OUT", true)
+ .build()
+ )
+ .returnsOrdered(
+ "n=1", "n=2", "n=3", "n=4", "n=5", "n=6", "n=7", "n=8", "n=9",
+ "n=10", "n=20", "n=30", "n=40", "n=50", "n=60", "n=70", "n=80", "n=90",
+ "n=100", "n=200", "n=300", "n=400", "n=500", "n=600", "n=700", "n=800", "n=900");
+ }
+
+}
+
+// End EnumerableRepeatUnionTest.java
diff --git a/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java b/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java
index 2625bb8..5b4c01f 100644
--- a/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java
+++ b/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java
@@ -3333,6 +3333,154 @@ public abstract class EnumerableDefaults {
public void close() {
}
}
+
+ private static final Object DUMMY = new Object();
+
+ /**
+ * Repeat Union All enumerable: it will evaluate the seed enumerable once, and then
+ * it will start to evaluate the iteration enumerable over and over until either it returns
+ * no results, or an optional maximum numbers of iterations is reached
+ * @param seed seed enumerable
+ * @param iteration iteration enumerable
+ * @param maxRep maximum numbers of repetitions for the iteration enumerable (-1 means no limit)
+ * @param <TSource> record type
+ */
+ @SuppressWarnings("unchecked")
+ public static <TSource> Enumerable<TSource> repeatUnionAll(
+ Enumerable<TSource> seed,
+ Enumerable<TSource> iteration,
+ int maxRep) {
+ assert maxRep >= -1;
+ return new AbstractEnumerable<TSource>() {
+ @Override public Enumerator<TSource> enumerator() {
+ return new Enumerator<TSource>() {
+ private TSource current = (TSource) DUMMY;
+ private boolean seedProcessed = false;
+ private int currentRep = 0;
+ private final Enumerator<TSource> seedEnumerator = seed.enumerator();
+ private Enumerator<TSource> iterativeEnumerator = null;
+
+ @Override public TSource current() {
+ if (this.current == DUMMY) {
+ throw new NoSuchElementException();
+ }
+ return this.current;
+ }
+
+ @Override public boolean moveNext() {
+ // if we are not done with the seed moveNext on it
+ if (!this.seedProcessed) {
+ if (this.seedEnumerator.moveNext()) {
+ this.current = this.seedEnumerator.current();
+ return true;
+ } else {
+ this.seedProcessed = true;
+ }
+ }
+
+ // if we are done with the seed, moveNext on the iterative part
+ while (true) {
+ if (maxRep != -1 && this.currentRep == maxRep) {
+ // max number of iterations reached, we are done
+ this.current = (TSource) DUMMY;
+ return false;
+ }
+
+ if (this.iterativeEnumerator == null) {
+ this.iterativeEnumerator = iteration.enumerator();
+ }
+
+ if (this.iterativeEnumerator.moveNext()) {
+ this.current = this.iterativeEnumerator.current();
+ return true;
+ }
+
+ if (this.current == DUMMY) {
+ // current iteration did not return any value, we are done
+ return false;
+ }
+
+ // current iteration level (which returned some values) is finished, go to next one
+ this.current = (TSource) DUMMY;
+ this.iterativeEnumerator.close();
+ this.iterativeEnumerator = null;
+ this.currentRep++;
+ }
+ }
+
+ @Override public void reset() {
+ this.seedEnumerator.reset();
+ if (this.iterativeEnumerator != null) {
+ this.iterativeEnumerator.close();
+ this.iterativeEnumerator = null;
+ }
+ this.currentRep = 0;
+ }
+
+ @Override public void close() {
+ this.seedEnumerator.close();
+ if (this.iterativeEnumerator != null) {
+ this.iterativeEnumerator.close();
+ this.iterativeEnumerator = null;
+ }
+ }
+ };
+ }
+ };
+ }
+
+ /**
+ * Lazy read and lazy write spool that stores data into a collection
+ */
+ @SuppressWarnings("unchecked")
+ public static <TSource> Enumerable<TSource> lazyCollectionSpool(
+ Collection<TSource> outputCollection,
+ Enumerable<TSource> input) {
+
+ return new AbstractEnumerable<TSource>() {
+ @Override public Enumerator<TSource> enumerator() {
+ return new Enumerator<TSource>() {
+ private TSource current = (TSource) DUMMY;
+ private final Enumerator<TSource> inputEnumerator = input.enumerator();
+ private final Collection<TSource> collection = outputCollection;
+ private final Collection<TSource> tempCollection = new ArrayList<>();
+
+ @Override public TSource current() {
+ if (this.current == DUMMY) {
+ throw new NoSuchElementException();
+ }
+ return this.current;
+ }
+
+ @Override public boolean moveNext() {
+ if (this.inputEnumerator.moveNext()) {
+ this.current = this.inputEnumerator.current();
+ this.tempCollection.add(this.current);
+ return true;
+ }
+ this.flush();
+ return false;
+ }
+
+ private void flush() {
+ this.collection.clear();
+ this.collection.addAll(this.tempCollection);
+ this.tempCollection.clear();
+ }
+
+ @Override public void reset() {
+ this.inputEnumerator.reset();
+ this.collection.clear();
+ this.tempCollection.clear();
+ }
+
+ @Override public void close() {
+ this.inputEnumerator.close();
+ }
+ };
+ }
+ };
+ }
}
// End EnumerableDefaults.java
diff --git a/site/_docs/algebra.md b/site/_docs/algebra.md
index 8802f81..2ada663 100644
--- a/site/_docs/algebra.md
+++ b/site/_docs/algebra.md
@@ -247,6 +247,53 @@ Similarly, to reference "DNAME", internal field #9 (8 + 1),
write `builder.field(2, 1, "DNAME")`, `builder.field(2, "DEPT", "DNAME")`,
or `builder.field(2, 1, 1)`.
+### Recursive Queries
+
+Warning: The current API is experimental and subject to change without notice.
+A SQL recursive query, e.g. this one that generates the sequence 1, 2, 3, ...10:
+
+{% highlight sql %}
+WITH RECURSIVE aux(i) AS (
+ VALUES (1)
+ UNION ALL
+ SELECT i+1 FROM aux WHERE i < 10
+)
+SELECT * FROM aux
+{% endhighlight %}
+
+can be generated using a scan on a TransientTable and a RepeatUnion:
+
+{% highlight java %}
+final RelNode node = builder
+ .values(new String[] { "i" }, 1)
+ .transientScan("aux")
+ .filter(
+ builder.call(
+ SqlStdOperatorTable.LESS_THAN,
+ builder.field(0),
+ builder.literal(10)))
+ .project(
+ builder.call(
+ SqlStdOperatorTable.PLUS,
+ builder.field(0),
+ builder.literal(1)))
+ .repeatUnion("aux", true)
+ .build();
+System.out.println(RelOptUtil.toString(node));
+{% endhighlight %}
+
+which produces:
+
+{% highlight text %}
+LogicalRepeatUnion(all=[true])
+ LogicalTableSpool(readType=[LAZY], writeType=[LAZY], tableName=[aux])
+ LogicalValues(tuples=[[{ 1 }]])
+ LogicalTableSpool(readType=[LAZY], writeType=[LAZY], tableName=[aux])
+ LogicalProject($f0=[+($0, 1)])
+ LogicalFilter(condition=[<($0, 10)])
+ LogicalTableScan(table=[[aux]])
+{% endhighlight %}
+
### API summary
#### Relational operators
@@ -260,6 +307,7 @@ return the `RelBuilder`.
|:------------------- |:-----------
| `scan(tableName)` | Creates a [TableScan]({{ site.apiRoot }}/org/apache/calcite/rel/core/TableScan.html).
| `functionScan(operator, n, expr...)`<br/>`functionScan(operator, n, exprList)` | Creates a [TableFunctionScan]({{ site.apiRoot }}/org/apache/calcite/rel/core/TableFunctionScan.html) of the `n` most recent relational expressions.
+| `transientScan(tableName [, rowType])` | Creates a [TableScan]({{ site.apiRoot }}/org/apache/calcite/rel/core/TableScan.html) on a [TransientTable]]({{ site.apiRoot }}/org/apache/calcite/schema/TransientTable.html) with the given type (if not specified, the most recent relational expression's type will be used).
| `values(fieldNames, value...)`<br/>`values(rowType, tupleList)` | Creates a [Values]({{ site.apiRoot }}/org/apache/calcite/rel/core/Values.html).
| `filter(expr...)`<br/>`filter(exprList)` | Creates a [Filter]({{ site.apiRoot }}/org/apache/calcite/rel/core/Filter.html) over the AND of the given predicates.
| `project(expr...)`<br/>`project(exprList [, fieldNames])` | Creates a [Project]({{ site.apiRoot }}/org/apache/calcite/rel/core/Project.html). To override the default name, wrap expressions using `alias`, or specify the `fieldNames` argument.
@@ -279,6 +327,7 @@ return the `RelBuilder`.
| `union(all [, n])` | Creates a [Union]({{ site.apiRoot }}/org/apache/calcite/rel/core/Union.html) of the `n` (default two) most recent relational expressions.
| `intersect(all [, n])` | Creates an [Intersect]({{ site.apiRoot }}/org/apache/calcite/rel/core/Intersect.html) of the `n` (default two) most recent relational expressions.
| `minus(all)` | Creates a [Minus]({{ site.apiRoot }}/org/apache/calcite/rel/core/Minus.html) of the two most recent relational expressions.
+| `repeatUnion(tableName, all [, n])` | Creates a [RepeatUnion]({{ site.apiRoot }}/org/apache/calcite/rel/core/RepeatUnion.html) associated to a [TransientTable]]({{ site.apiRoot }}/org/apache/calcite/schema/TransientTable.html) of the two most recent relational expressions, with `n` maximum number of iterations (default -1, i.e. no limit).
| `snapshot(period)` | Creates a [Snapshot]({{ site.apiRoot }}/org/apache/calcite/rel/core/Snapshot.html) of the given snapshot period.
| `match(pattern, strictStart,` `strictEnd, patterns, measures,` `after, subsets, allRows,` `partitionKeys, orderKeys,` `interval)` | Creates a [Match]({{ site.apiRoot }}/org/apache/calcite/rel/core/Match.html).