fix(isthmus)!: round-trip a virtual table whose rows are not all literals

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

@@ -115,6 +115,12 @@ public RelNode visit(Correlate correlate) throws RuntimeException {
 
   @Override
   public RelNode visitOther(RelNode other) throws RuntimeException {
+    // A relation's own expressions can hold a subquery binding outer references, and a subquery's
+    // relation is not an input, so walking inputs never reaches it. Filter and Project scan theirs
+    // before they get here; a virtual table's rows, and a join, calc or sort condition, arrive
+    // here. AbstractRelNode.accept(RexShuttle) returns the relation itself, so the result is the
+    // one already in the tree.
+    other.accept(rexVisitor);
     for (RelNode child : other.getInputs()) {
       reverseAccept(child);
     }

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

@@ -1,5 +1,6 @@
 package io.substrait.isthmus;
 
+import io.substrait.isthmus.calcite.rel.VirtualTable;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.Aggregate;
 import org.apache.calcite.rel.core.Calc;
@@ -63,6 +64,18 @@ public OUTPUT visit(Values values) throws EXCEPTION {
     return visitOther(values);
   }
 
+  /**
+   * Visits a {@link VirtualTable} node, the relation isthmus converts a virtual table whose rows
+   * are not all literals into.
+   *
+   * @param virtualTable the virtual table node
+   * @return the result of visiting this node
+   * @throws EXCEPTION if the visit fails
+   */
+  public OUTPUT visit(VirtualTable virtualTable) throws EXCEPTION {
+    return visitOther(virtualTable);
+  }
+
   /**
    * Visits a {@link Filter} node.
    *
@@ -261,6 +274,8 @@ public final OUTPUT reverseAccept(RelNode node) throws EXCEPTION {
       return this.visit((Aggregate) node);
     } else if (node instanceof TableModify) {
       return this.visit((TableModify) node);
+    } else if (node instanceof VirtualTable) {
+      return this.visit((VirtualTable) node);
     } else {
       return this.visitOther(node);
     }

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

@@ -11,6 +11,7 @@
 import io.substrait.hint.Hint;
 import io.substrait.isthmus.calcite.rel.CreateTable;
 import io.substrait.isthmus.calcite.rel.CreateView;
+import io.substrait.isthmus.calcite.rel.VirtualTable;
 import io.substrait.isthmus.expression.AggregateFunctionConverter;
 import io.substrait.isthmus.expression.ExpressionRexConverter;
 import io.substrait.isthmus.expression.ScalarFunctionConverter;
@@ -77,7 +78,6 @@
 import org.apache.calcite.rel.core.TableModify;
 import org.apache.calcite.rel.logical.LogicalProject;
 import org.apache.calcite.rel.logical.LogicalTableModify;
-import org.apache.calcite.rel.logical.LogicalUnion;
 import org.apache.calcite.rel.logical.LogicalValues;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
@@ -869,45 +869,13 @@ public RelNode visit(VirtualTableScan virtualTableScan, Context context) {
           LogicalValues.create(relBuilder.getCluster(), rowType, tuplesBuilder.build()),
           virtualTableScan);
     } else {
-      // A row that does not fit a LogicalValues tuple is computed instead: we create a
-      // LogicalProject for each row to compute its values, and combine them together using a
-      // LogicalUnion. For example the following:
-      //
-      //   VirtualTable
-      //     (e1, e2)
-      //     (e3, e4)
-      //
-      //  Becomes:
-      //
-      //   LogicalUnion(all=[true])
-      //     LogicalProject(exprs=[e1, e2])
-      //       <Empty Row>
-      //     LogicalProject(exprs=[e3, e4])
-      //       <Empty Row>
-      //
-
-      RelDataType emptyRowType = typeFactory.createStructType(List.of(), List.of());
-      ImmutableList<ImmutableList<RexLiteral>> emptyRowValue = ImmutableList.of(ImmutableList.of());
-
-      List<RelNode> projects = new ArrayList<>();
-      for (final List<RexNode> rexRow : convertedRows) {
-        RelNode values = LogicalValues.create(relBuilder.getCluster(), emptyRowType, emptyRowValue);
-        RelNode project =
-            LogicalProject.create(
-                values, Collections.emptyList(), rexRow, rowType, Collections.emptySet());
-        projects.add(project);
-      }
-      RelNode union = LogicalUnion.create(projects, true);
-
-      // Apply a final LogicalProject on top to capture the field names from the VirtualTable
-      List<RexNode> topProjectExprs = new ArrayList<>();
-      for (int i = 0; i < rowType.getFieldCount(); i++) {
-        topProjectExprs.add(rexBuilder.makeInputRef(union, i));
-      }
-      RelNode topProject =
-          LogicalProject.create(
-              union, Collections.emptyList(), topProjectExprs, rowType, Collections.emptySet());
-      return applyRelCommon(topProject, virtualTableScan, topProject);
+      // A row that does not fit a LogicalValues tuple keeps its expressions, in a relation of our
+      // own: Calcite has none that holds them, and expanding the table into a projection per row
+      // does not come back -- the projection is what converts back, and the table is gone. A
+      // consumer whose planner only knows Calcite's own relations can expand it with
+      // VirtualTableExpansionRule.
+      return applyRelCommon(
+          VirtualTable.create(relBuilder.getCluster(), rowType, convertedRows), virtualTableScan);
     }
   }
 

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

@@ -8,6 +8,7 @@
 import io.substrait.extension.SimpleExtension;
 import io.substrait.isthmus.calcite.rel.CreateTable;
 import io.substrait.isthmus.calcite.rel.CreateView;
+import io.substrait.isthmus.calcite.rel.VirtualTable;
 import io.substrait.isthmus.expression.AggregateFunctionConverter;
 import io.substrait.isthmus.expression.LiteralConverter;
 import io.substrait.isthmus.expression.RexExpressionConverter;
@@ -60,6 +61,7 @@
 import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.rex.RexBuilder;
 import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.sql.SqlKind;
@@ -932,6 +934,53 @@ public Rel handleCreateView(CreateView createView) {
         .build();
   }
 
+  /**
+   * Converts the isthmus {@link VirtualTable}, which is what a virtual table whose rows are not all
+   * literals converts to.
+   *
+   * @param virtualTable Calcite virtual table
+   * @return Substrait virtual table scan
+   */
+  @Override
+  public Rel visit(VirtualTable virtualTable) {
+    // At the row type's field types rather than the values' own, as visit(Values) does: a literal
+    // narrower than its column -- Calcite infers one for a tuple value, and pushes a struct's
+    // nullability down into its fields -- would otherwise disagree with the schema built from the
+    // same row type, and VirtualTableScan rejects the relation on that.
+    List<RelDataTypeField> rowFields = virtualTable.getRowType().getFieldList();
+    LiteralConverter literalConverter = new LiteralConverter(typeConverter);
+    List<Expression.NestedStruct> rows = new ArrayList<>(virtualTable.getRows().size());
+    for (List<RexNode> row : virtualTable.getRows()) {
+      List<Expression> fields = new ArrayList<>(row.size());
+      for (int column = 0; column < row.size(); column++) {
+        RexNode value = row.get(column);
+        RelDataType declaredType = rowFields.get(column).getType();
+        Expression converted =
+            value instanceof RexLiteral
+                ? literalConverter.convert((RexLiteral) value, declaredType)
+                : toExpression(value);
+        // A value that is not a literal is converted from the expressions it is built of and
+        // takes its type from them, which the declared type cannot be put back on: casting at it
+        // would put an expression in the output the input did not have. Refused here rather than
+        // left to VirtualTableScan, whose check compares the two types without promoting either.
+        Type declared = typeConverter.toSubstrait(declaredType);
+        if (!converted.getType().equals(declared)) {
+          throw new UnsupportedOperationException(
+              String.format(
+                  "A virtual table's value %s converts to %s where its column is declared %s: "
+                      + "isthmus cannot convert a value that does not carry its column's type",
+                  value, converted.getType(), declared));
+        }
+        fields.add(converted);
+      }
+      rows.add(ExpressionCreator.nestedStruct(false, fields));
+    }
+    return VirtualTableScan.builder()
+        .initialSchema(typeConverter.toNamedStruct(virtualTable.getRowType()))
+        .addAllRows(rows)
+        .build();
+  }
+
   /**
    * Visits other Calcite nodes (e.g., DDL wrappers).
    *

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

@@ -1,6 +1,7 @@
 package io.substrait.isthmus;
 
 import io.substrait.extension.SimpleExtension;
+import io.substrait.isthmus.calcite.rel.rules.VirtualTableExpansionRule;
 import io.substrait.plan.Plan;
 import io.substrait.plan.Plan.Root;
 import io.substrait.relation.Rel;
@@ -66,6 +67,8 @@ public RelNode substraitRelToCalciteRel(Rel relRoot, Prepare.CatalogReader catal
    *
    * @param plan the Substrait {@link Plan} to convert to SQL, must not be null
    * @param dialect the {@link SqlDialect} to generate the SQL strings for, must not be null
+   *     <p>Any {@link io.substrait.isthmus.calcite.rel.VirtualTable} the conversion produced is
+   *     expanded first: {@link RelToSqlConverter} knows Calcite's own relations only.
    * @return list containing a SQL string for each {@link Plan.Root} in {@code plan}
    */
   public List<String> convert(Plan plan, SqlDialect dialect) {
@@ -75,7 +78,9 @@ public List<String> convert(Plan plan, SqlDialect dialect) {
     for (Root root : plan.getRoots()) {
       result.add(
           relToSql
-              .visitRoot(substraitToCalcite.convert(root).project(true))
+              .visitRoot(
+                  VirtualTableExpansionRule.expandAll(
+                      substraitToCalcite.convert(root).project(true)))
               .asStatement()
               .toSqlString(dialect)
               .getSql());