fix(isthmus): implement ddl operatoins sas relnode

Author: ZorinAnton

isthmus/src/main/java/io/substrait/isthmus/CalciteOperationToSubstrait.java

@@ -1,4 +1,4 @@
-package io.substrait.isthmus.operation;
+package io.substrait.isthmus;
 
 import io.substrait.relation.Aggregate;
 import io.substrait.relation.ConsistentPartitionWindow;
@@ -43,7 +43,7 @@ public class SqlKindFromRel
   @Override
   public SqlKind visit(Aggregate aggregate, EmptyVisitationContext context)
       throws RuntimeException {
-    // Aggregation is a core part of a query (DQL).
+
     return QUERY_KIND;
   }
 
@@ -56,25 +56,21 @@ public SqlKind visit(EmptyScan emptyScan, EmptyVisitationContext context)
 
   @Override
   public SqlKind visit(Fetch fetch, EmptyVisitationContext context) throws RuntimeException {
-    // Fetch corresponds to LIMIT/OFFSET, a part of a query.
     return QUERY_KIND;
   }
 
   @Override
   public SqlKind visit(Filter filter, EmptyVisitationContext context) throws RuntimeException {
-    // Filter corresponds to a WHERE clause, a part of a query.
     return QUERY_KIND;
   }
 
   @Override
   public SqlKind visit(Join join, EmptyVisitationContext context) throws RuntimeException {
-    // A logical join operation.
     return SqlKind.JOIN;
   }
 
   @Override
   public SqlKind visit(Set set, EmptyVisitationContext context) throws RuntimeException {
-    // Maps Substrait's Set operations to their SQL equivalents.
     switch (set.getSetOp()) {
       case UNION_ALL:
       case UNION_DISTINCT:
@@ -96,38 +92,32 @@ public SqlKind visit(Set set, EmptyVisitationContext context) throws RuntimeExce
   @Override
   public SqlKind visit(NamedScan namedScan, EmptyVisitationContext context)
       throws RuntimeException {
-    // A scan from a named table is the start of a query.
     return QUERY_KIND;
   }
 
   @Override
   public SqlKind visit(LocalFiles localFiles, EmptyVisitationContext context)
       throws RuntimeException {
-    // A scan from local files is a type of query input.
     return QUERY_KIND;
   }
 
   @Override
   public SqlKind visit(Project project, EmptyVisitationContext context) throws RuntimeException {
-    // Project corresponds to the SELECT clause of a query.
     return QUERY_KIND;
   }
 
   @Override
   public SqlKind visit(Expand expand, EmptyVisitationContext context) throws RuntimeException {
-    // Expand is a relational operator used in queries (e.g., for grouping sets).
     return QUERY_KIND;
   }
 
   @Override
   public SqlKind visit(Sort sort, EmptyVisitationContext context) throws RuntimeException {
-    // A sort operation directly maps to ORDER BY.
     return SqlKind.ORDER_BY;
   }
 
   @Override
   public SqlKind visit(Cross cross, EmptyVisitationContext context) throws RuntimeException {
-    // A cross join is a type of join.
     return SqlKind.JOIN;
   }
 
@@ -141,62 +131,53 @@ public SqlKind visit(VirtualTableScan virtualTableScan, EmptyVisitationContext c
   @Override
   public SqlKind visit(ExtensionLeaf extensionLeaf, EmptyVisitationContext context)
       throws RuntimeException {
-    // Unknown extension node.
     return SqlKind.OTHER;
   }
 
   @Override
   public SqlKind visit(ExtensionSingle extensionSingle, EmptyVisitationContext context)
       throws RuntimeException {
-    // Unknown extension node.
     return SqlKind.OTHER;
   }
 
   @Override
   public SqlKind visit(ExtensionMulti extensionMulti, EmptyVisitationContext context)
       throws RuntimeException {
-    // Unknown extension node.
     return SqlKind.OTHER;
   }
 
   @Override
   public SqlKind visit(ExtensionTable extensionTable, EmptyVisitationContext context)
       throws RuntimeException {
-    // Unknown extension node.
     return SqlKind.OTHER;
   }
 
   @Override
   public SqlKind visit(HashJoin hashJoin, EmptyVisitationContext context) throws RuntimeException {
-    // A physical hash join is a type of join.
     return SqlKind.JOIN;
   }
 
   @Override
   public SqlKind visit(MergeJoin mergeJoin, EmptyVisitationContext context)
       throws RuntimeException {
-    // A physical merge join is a type of join.
     return SqlKind.JOIN;
   }
 
   @Override
   public SqlKind visit(NestedLoopJoin nestedLoopJoin, EmptyVisitationContext context)
       throws RuntimeException {
-    // A physical nested loop join is a type of join.
     return SqlKind.JOIN;
   }
 
   @Override
   public SqlKind visit(
       ConsistentPartitionWindow consistentPartitionWindow, EmptyVisitationContext context)
       throws RuntimeException {
-    // This relation represents a window function operation.
     return SqlKind.OVER;
   }
 
   @Override
   public SqlKind visit(NamedWrite write, EmptyVisitationContext context) throws RuntimeException {
-    // DML and CTAS write operations.
     switch (write.getOperation()) {
       case INSERT:
         return SqlKind.INSERT;
@@ -214,13 +195,11 @@ public SqlKind visit(NamedWrite write, EmptyVisitationContext context) throws Ru
   @Override
   public SqlKind visit(ExtensionWrite write, EmptyVisitationContext context)
       throws RuntimeException {
-    // Custom DML/DDL operations are best mapped to OTHER or OTHER_DDL.
     return SqlKind.OTHER_DDL;
   }
 
   @Override
   public SqlKind visit(NamedDdl ddl, EmptyVisitationContext context) throws RuntimeException {
-    // DDL operations, determined by combining the operation and the object type.
     switch (ddl.getOperation()) {
       case CREATE:
       case CREATE_OR_REPLACE:
@@ -251,14 +230,11 @@ public SqlKind visit(NamedDdl ddl, EmptyVisitationContext context) throws Runtim
 
   @Override
   public SqlKind visit(ExtensionDdl ddl, EmptyVisitationContext context) throws RuntimeException {
-    // A custom DDL operation.
     return SqlKind.OTHER_DDL;
   }
 
   @Override
   public SqlKind visit(NamedUpdate update, EmptyVisitationContext context) throws RuntimeException {
-    // An update operation is a specific DML type. Note that modern Substrait
-    // producers should use NamedWrite with WriteOp.UPDATE instead.
     return SqlKind.UPDATE;
   }
 }

…it/isthmus/operation/SqlKindFromRel.java …io/substrait/isthmus/SqlKindFromRel.javaisthmus/src/main/java/io/substrait/isthmus/operation/SqlKindFromRel.java renamed to isthmus/src/main/java/io/substrait/isthmus/SqlKindFromRel.java isthmus/src/main/java/io/substrait/isthmus/operation/SqlKindFromRel.java renamed to isthmus/src/main/java/io/substrait/isthmus/SqlKindFromRel.java

@@ -1,12 +1,10 @@
 package io.substrait.isthmus;
 
-import io.substrait.isthmus.operation.CalciteOperation;
 import io.substrait.isthmus.sql.SubstraitSqlToCalcite;
 import io.substrait.plan.ImmutablePlan.Builder;
 import io.substrait.plan.Plan;
 import io.substrait.plan.Plan.Version;
 import io.substrait.plan.PlanProtoConverter;
-import java.util.List;
 import org.apache.calcite.prepare.Prepare;
 import org.apache.calcite.sql.parser.SqlParseException;
 
@@ -55,13 +53,9 @@ public Plan convert(String sqlStatements, Prepare.CatalogReader catalogReader)
     builder.version(Version.builder().from(Version.DEFAULT_VERSION).producer("isthmus").build());
 
     // TODO: consider case in which one sql passes conversion while others don't
-    List<CalciteOperation> calciteOperations =
-        SubstraitSqlToCalcite.convertQueries(sqlStatements, catalogReader);
-    CalciteOperationToSubstrait calciteOperationToSubstrait =
-        new CalciteOperationToSubstrait(EXTENSION_COLLECTION, featureBoard);
-    calciteOperations.stream()
-        .map(operation -> operation.accept(calciteOperationToSubstrait))
-        .forEach(builder::addRoots);
+    SubstraitSqlToCalcite.convertQueries(sqlStatements, catalogReader).stream()
+        .map(root -> SubstraitRelVisitor.convert(root, EXTENSION_COLLECTION, featureBoard))
+        .forEach(root -> builder.addRoots(root));
 
     return builder.build();
   }

isthmus/src/main/java/io/substrait/isthmus/SqlToSubstrait.java

@@ -10,6 +10,8 @@
 import io.substrait.expression.Expression.SortDirection;
 import io.substrait.expression.FunctionArg;
 import io.substrait.extension.SimpleExtension;
+import io.substrait.isthmus.calcite.rel.CreateTable;
+import io.substrait.isthmus.calcite.rel.CreateView;
 import io.substrait.isthmus.expression.AggregateFunctionConverter;
 import io.substrait.isthmus.expression.ExpressionRexConverter;
 import io.substrait.isthmus.expression.ScalarFunctionConverter;
@@ -24,6 +26,7 @@
 import io.substrait.relation.Join;
 import io.substrait.relation.Join.JoinType;
 import io.substrait.relation.LocalFiles;
+import io.substrait.relation.NamedDdl;
 import io.substrait.relation.NamedScan;
 import io.substrait.relation.NamedUpdate;
 import io.substrait.relation.NamedWrite;
@@ -53,6 +56,7 @@
 import org.apache.calcite.rel.RelCollations;
 import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelRoot;
 import org.apache.calcite.rel.core.AggregateCall;
 import org.apache.calcite.rel.core.CorrelationId;
 import org.apache.calcite.rel.core.JoinRelType;
@@ -68,6 +72,7 @@
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.rex.RexSlot;
 import org.apache.calcite.sql.SqlAggFunction;
+import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.calcite.tools.Frameworks;
@@ -547,6 +552,17 @@ public RelNode visit(NamedUpdate update, Context context) {
         false);
   }
 
+  @Override
+  public RelNode visit(NamedDdl namedDdl, Context context) {
+    if (namedDdl.getViewDefinition().isEmpty()) {
+      throw new IllegalArgumentException("no view definition found");
+    }
+    Rel viewDefinition = namedDdl.getViewDefinition().get();
+    RelNode relNode = viewDefinition.accept(this, context);
+    RelRoot relRoot = RelRoot.of(relNode, SqlKind.SELECT);
+    return new CreateView(namedDdl.getNames(), relRoot);
+  }
+
   @Override
   public RelNode visit(VirtualTableScan virtualTableScan, Context context) {
 
@@ -584,6 +600,13 @@ public RelNode visit(VirtualTableScan virtualTableScan, Context context) {
         relBuilder.getCluster(), rowTypeWithNames, ImmutableList.copyOf(tuples));
   }
 
+  private RelNode handleCreateTableAs(NamedWrite namedWrite, Context context) {
+    Rel input = namedWrite.getInput();
+    RelNode relNode = input.accept(this, context);
+    RelRoot relRoot = RelRoot.of(relNode, SqlKind.SELECT);
+    return new CreateTable(namedWrite.getNames(), relRoot);
+  }
+
   @Override
   public RelNode visit(NamedWrite write, Context context) {
     RelNode input = write.getInput().accept(this, context);
@@ -599,6 +622,8 @@ public RelNode visit(NamedWrite write, Context context) {
       case DELETE:
         operation = TableModify.Operation.DELETE;
         break;
+      case CTAS:
+        return handleCreateTableAs(write, context);
       default:
         throw new UnsupportedOperationException(
             "Write operation '"