[feature][dingo-calcite] Support window function and cte

dingo-calcite/build.gradle

@@ -40,6 +40,8 @@ dependencies {
     implementation group: 'io.dingodb.expr', name: 'dingo-expr-coding', version: 'dingo-expr'.v()
     implementation group: 'io.dingodb.expr', name: 'dingo-expr-common', version: 'dingo-expr'.v()
     implementation group: 'io.dropwizard.metrics', name: 'metrics-core', version: 'metrics-core'.v()
+    implementation group: 'org.codehaus.janino', name:'janino', version: '3.1.8'
+    implementation group: 'org.codehaus.janino', name:'commons-compiler', version: '3.1.8'
 
     api group: 'io.dingodb.expr', name: 'dingo-expr-parser', version: 'dingo-expr'.v()
     api group: 'io.dingodb.expr', name: 'dingo-expr-rel', version: 'dingo-expr'.v()

dingo-calcite/src/main/codegen/templates/Parser.jj

@@ -3815,14 +3815,15 @@ SqlNodeList WithList() :
 {
     final Span s;
     final List<SqlWithItem> list = new ArrayList<SqlWithItem>();
+    boolean recursive = false;
 }
 {
-    <WITH> { s = span(); }
-    AddWithItem(list) ( <COMMA> AddWithItem(list) )*
+    <WITH> [ <RECURSIVE> { recursive = true; } ] { s = span(); }
+    AddWithItem(list, SqlLiteral.createBoolean(recursive, getPos())) ( <COMMA> AddWithItem(list, SqlLiteral.createBoolean(recursive, getPos())) )*
     { return new SqlNodeList(list, s.end(this)); }
 }
 
-void AddWithItem(List<SqlWithItem> list) :
+void AddWithItem(List<SqlWithItem> list, SqlLiteral recursive) :
 {
     final SqlIdentifier id;
     final SqlNodeList columnList;
@@ -3833,7 +3834,7 @@ void AddWithItem(List<SqlWithItem> list) :
     ( columnList = ParenthesizedSimpleIdentifierList() | { columnList = null; } )
     <AS>
     definition = ParenthesizedExpression(ExprContext.ACCEPT_QUERY)
-    { list.add(new SqlWithItem(id.getParserPosition(), id, columnList, definition)); }
+    { list.add(new SqlWithItem(id.getParserPosition(), id, columnList, definition, recursive)); }
 }
 
 /**

dingo-calcite/src/main/java/io/dingodb/calcite/DingoParser.java

@@ -414,6 +414,11 @@ public RelNode optimize(RelNode relNode) {
         List<RelOptRule> rules = DingoRules.rules();
         ImmutableList.Builder<RelOptRule> builder = ImmutableList.builder();
         builder.addAll(rules);
+
+        builder.addAll(DingoRules.ABSTRACT_RELATIONAL_RULES);
+        builder.addAll(DingoRules.ABSTRACT_RULES);
+        builder.addAll(DingoRules.BASE_RULES);
+
         if (!context.getConfig().topDownOpt()) {
             // This is needed for `IterativeRuleDriver`.
             builder.add(AbstractConverter.ExpandConversionRule.INSTANCE);

dingo-calcite/src/main/java/io/dingodb/calcite/meta/DingoCostModelV1.java

@@ -61,7 +61,9 @@ public static synchronized DingoCostModelV1 getCostModel() {
 
     public static double getScanAvgRowSize(LogicalDingoTableScan tableScan) {
         DingoTable dingoTable = tableScan.getTable().unwrap(DingoTable.class);
-        assert dingoTable != null;
+        if (dingoTable == null) {
+            return 0;
+        }
         String schemaName = dingoTable.getNames().get(1);
         //List<Column> selectionCdList = getSelectionCdList(tableScan, dingoTable);
         return getAvgRowSize(

dingo-calcite/src/main/java/io/dingodb/calcite/rel/DingoRel.java

@@ -20,6 +20,7 @@
 import io.dingodb.calcite.traits.DingoRelStreamingDef;
 import io.dingodb.calcite.visitor.DingoRelVisitor;
 import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.plan.volcano.RelSubset;
 import org.apache.calcite.rel.PhysicalNode;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.util.Pair;
@@ -31,6 +32,10 @@
 
 public interface DingoRel extends PhysicalNode {
     static DingoRel dingo(RelNode rel) {
+        if (rel instanceof RelSubset) {
+            RelSubset relSubset = (RelSubset) rel;
+            return (DingoRel) relSubset.getBest();
+        }
         return (DingoRel) rel;
     }
 

dingo-calcite/src/main/java/io/dingodb/calcite/rel/DingoRepeatUnion.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2021 DataCanvas
+ *
+ * Licensed 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 io.dingodb.calcite.rel;
+
+import io.dingodb.calcite.visitor.DingoRelVisitor;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.RepeatUnion;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.List;
+
+public class DingoRepeatUnion extends RepeatUnion implements DingoRel {
+    public DingoRepeatUnion(
+        RelOptCluster cluster, RelTraitSet traitSet, RelNode seed, RelNode iterative, boolean all,
+        int iterationLimit, @Nullable RelOptTable transientTable) {
+        super(cluster, traitSet, seed, iterative, all, iterationLimit, transientTable);
+    }
+
+    @Override
+    public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+        return new DingoRepeatUnion(getCluster(), traitSet,
+            getSeedRel(), getIterativeRel(), all, iterationLimit, transientTable);
+    }
+
+    @Override
+    public <T> T accept(@NonNull DingoRelVisitor<T> visitor) {
+        return visitor.visit(this);
+    }
+}

dingo-calcite/src/main/java/io/dingodb/calcite/rel/DingoTableSpool.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2021 DataCanvas
+ *
+ * Licensed 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 io.dingodb.calcite.rel;
+
+import io.dingodb.calcite.traits.DingoConvention;
+import io.dingodb.calcite.visitor.DingoRelVisitor;
+import io.dingodb.calcite.visitor.function.DingoTableSpoolVisitFun;
+import lombok.Getter;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptTable;
+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.impl.ListTransientTable;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+public class DingoTableSpool extends TableSpool implements DingoRel {
+    @Getter
+    private ListTransientTable listTransientTable;
+
+    public DingoTableSpool(
+        RelOptCluster cluster, RelTraitSet traitSet, RelNode input, Type readType,
+        Type writeType, RelOptTable table, ListTransientTable listTransientTable
+    ) {
+        super(cluster, traitSet, input, readType, writeType, table);
+        this.listTransientTable = listTransientTable;
+    }
+
+    public static DingoTableSpool create(RelNode input, Type readType,
+                                              Type writeType, RelOptTable table) {
+        RelOptCluster cluster = input.getCluster();
+        RelMetadataQuery mq = cluster.getMetadataQuery();
+        RelTraitSet traitSet = cluster.traitSetOf(DingoConvention.INSTANCE)
+            .replaceIfs(RelCollationTraitDef.INSTANCE,
+                () -> mq.collations(input))
+            .replaceIf(RelDistributionTraitDef.INSTANCE,
+                () -> mq.distribution(input));
+        ListTransientTable tmp = table.unwrap(ListTransientTable.class);
+        return new DingoTableSpool(cluster, traitSet, input, readType, writeType, table, tmp);
+    }
+
+    @Override
+    public <T> T accept(@NonNull DingoRelVisitor<T> visitor) {
+        return visitor.visit(this);
+    }
+
+    @Override
+    protected Spool copy(RelTraitSet relTraitSet, RelNode relNode, Type type, Type type1) {
+        return null;
+    }
+}

dingo-calcite/src/main/java/io/dingodb/calcite/rel/DingoTransientTableScan.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2021 DataCanvas
+ *
+ * Licensed 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 io.dingodb.calcite.rel;
+
+import io.dingodb.calcite.visitor.DingoRelVisitor;
+import io.dingodb.common.type.TupleMapping;
+import lombok.Getter;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.hint.RelHint;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.schema.impl.ListTransientTable;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.List;
+
+public class DingoTransientTableScan extends LogicalDingoTableScan implements DingoRel {
+    @Getter
+    ListTransientTable transientTable;
+
+    public DingoTransientTableScan(
+        RelOptCluster cluster, RelTraitSet traitSet, List<RelHint> hints, RelOptTable table,
+        @Nullable RexNode filter, @Nullable TupleMapping selection, ListTransientTable transientTable
+    ) {
+        super(cluster, traitSet, hints, table, filter, selection);
+        this.transientTable = transientTable;
+    }
+
+    @Override
+    public <T> T accept(@NonNull DingoRelVisitor<T> visitor) {
+        return visitor.visit(this);
+    }
+}

dingo-calcite/src/main/java/io/dingodb/calcite/rel/DingoWindow.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2021 DataCanvas
+ *
+ * Licensed 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 io.dingodb.calcite.rel;
+
+import io.dingodb.calcite.visitor.DingoRelVisitor;
+import org.apache.calcite.adapter.enumerable.PhysType;
+import org.apache.calcite.adapter.enumerable.RexToLixTranslator;
+import org.apache.calcite.linq4j.tree.BlockBuilder;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Window;
+import org.apache.calcite.rel.hint.RelHint;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLiteral;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+import java.lang.reflect.Type;
+import java.util.List;
+
+public class DingoWindow extends Window implements DingoRel {
+
+    public DingoWindow(
+        RelOptCluster cluster, RelTraitSet traitSet, List<RelHint> hints, RelNode input,
+        List<RexLiteral> constants, RelDataType rowType, List<Group> groups
+    ) {
+        super(cluster, traitSet, hints, input, constants, rowType, groups);
+    }
+
+    @Override
+    public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+        return new DingoWindow(this.getCluster(), traitSet, this.hints, sole(inputs), constants, rowType, groups);
+    }
+
+    public static class WindowRelInputGetter
+        implements RexToLixTranslator.InputGetter {
+        private final Expression row;
+        private final PhysType rowPhysType;
+        private final int actualInputFieldCount;
+        private final List<Expression> constants;
+
+        public WindowRelInputGetter(Expression row,
+                                     PhysType rowPhysType, int actualInputFieldCount,
+                                     List<Expression> constants) {
+            this.row = row;
+            this.rowPhysType = rowPhysType;
+            this.actualInputFieldCount = actualInputFieldCount;
+            this.constants = constants;
+        }
+
+        @Override public Expression field(BlockBuilder list, int index, Type storageType) {
+            if (index < actualInputFieldCount) {
+                Expression current = list.append("current", row);
+                return rowPhysType.fieldReference(current, index, storageType);
+            }
+            return constants.get(index - actualInputFieldCount);
+        }
+    }
+
+
+    @Override
+    public <T> T accept(@NonNull DingoRelVisitor<T> visitor) {
+        return visitor.visit(this);
+    }
+}

dingo-calcite/src/main/java/io/dingodb/calcite/rel/LogicalDingoTableScan.java

@@ -124,11 +124,10 @@ public LogicalDingoTableScan(
         this.groupSets = groupSets;
         this.pushDown = pushDown;
         DingoTable dingoTable = table.unwrap(DingoTable.class);
-        assert dingoTable != null;
         this.realSelection = selection;
         this.forDml = forDml;
         // If the columns of the table contain hide and delete, the data shows that they need to be deleted
-        if (selection != null) {
+        if (selection != null && dingoTable != null) {
             int fieldCount = dingoTable.getTable().getColumns().size();
             if (forDml) {
                 int[] mappingTmp = selection.getMappings();
@@ -149,7 +148,7 @@ public LogicalDingoTableScan(
                 }
                 this.selection = TupleMapping.of(mappingList);
             }
-        } else {
+        } else if (dingoTable != null) {
             List<Integer> mapping = dingoTable.getTable()
                 .getColumns()
                 .stream()

dingo-calcite/src/main/java/io/dingodb/calcite/rel/dingo/DingoStreamingConverter.java

@@ -57,8 +57,9 @@ public double estimateRowCount(@NonNull RelMetadataQuery mq) {
         Set<DingoRelPartition> partitions = getStreaming().getPartitions();
         Set<DingoRelPartition> inputPartitions = inputStreaming.getPartitions();
         this.rowCount = rowCount;
-        assert partitions != null && inputPartitions != null;
-        if (partitions.size() > inputPartitions.size()) {
+        if (partitions == null || inputPartitions == null) {
+            return rowCount;
+        } else if (partitions.size() > inputPartitions.size()) {
             for (int i = 0; i < partitions.size() - inputPartitions.size(); ++i) {
                 rowCount /= 1.000001d;
             }