fix(isthmus)!: apply the emit mapping a virtual table carries

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

@@ -789,6 +789,12 @@ public RelNode visit(NamedDdl namedDdl, Context context) {
               namedDdl.getObject()));
     }
 
+    if (namedDdl.getRemap().isPresent()) {
+      throw new UnsupportedOperationException(
+          "Emit mapping on a NamedDdl is not supported: a CreateView produces the view it creates, "
+              + "not columns to select from");
+    }
+
     if (namedDdl.getViewDefinition().isEmpty()) {
       throw new IllegalArgumentException("NamedDdl view definition must be set");
     }
@@ -855,7 +861,9 @@ public RelNode visit(VirtualTableScan virtualTableScan, Context context) {
         }
         tuplesBuilder.add(tupleBuilder.build());
       }
-      return LogicalValues.create(relBuilder.getCluster(), rowType, tuplesBuilder.build());
+      return applyRelCommon(
+          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
@@ -892,8 +900,10 @@ public RelNode visit(VirtualTableScan virtualTableScan, Context context) {
       for (int i = 0; i < rowType.getFieldCount(); i++) {
         topProjectExprs.add(rexBuilder.makeInputRef(union, i));
       }
-      return LogicalProject.create(
-          union, Collections.emptyList(), topProjectExprs, rowType, Collections.emptySet());
+      return applyRelCommon(
+          LogicalProject.create(
+              union, Collections.emptyList(), topProjectExprs, rowType, Collections.emptySet()),
+          virtualTableScan);
     }
   }
 
@@ -1052,6 +1062,12 @@ private RelNode handleCreateTableAs(NamedWrite namedWrite, Context context) {
 
   @Override
   public RelNode visit(NamedWrite write, Context context) {
+    if (write.getRemap().isPresent()) {
+      throw new UnsupportedOperationException(
+          "Emit mapping on a NamedWrite is not supported: a TableModify's row type is a single "
+              + "ROWCOUNT column and a CreateTable produces the table it creates, neither of them "
+              + "columns to select from");
+    }
     RelNode input = write.getInput().accept(this, context);
     final RelOptSchema relOptSchema = requireRelOptSchema();
     final RelOptTable targetTable = relOptSchema.getTableForMember(write.getNames());

isthmus/src/test/java/io/substrait/isthmus/DdlRoundtripTest.java

@@ -163,6 +163,40 @@ void theRootOverACreateStatementIsNamedByTheDeclaredSchema() throws SqlParseExce
     assertEquals(List.of("TOTAL", "DOUBLED"), converted.getNames());
   }
 
+  /**
+   * A write and a DDL relation produce the object they act on rather than columns to select from --
+   * a TableModify's row type is a single ROWCOUNT column, a CreateTable and a CreateView the object
+   * they create -- so an emit mapping over one has nothing to apply to and is refused rather than
+   * dropped.
+   */
+  @Test
+  void anEmitMappingOnAWriteOrADdlIsRefused() {
+    NamedWrite ctas =
+        NamedWrite.builder()
+            .input(computedColumns())
+            .names(List.of("dst1"))
+            .tableSchema(declaredSchema())
+            .operation(AbstractWriteRel.WriteOp.CTAS)
+            .createMode(AbstractWriteRel.CreateMode.REPLACE_IF_EXISTS)
+            .outputMode(AbstractWriteRel.OutputMode.NO_OUTPUT)
+            .remap(Rel.Remap.of(List.of(0)))
+            .build();
+    NamedDdl createView =
+        NamedDdl.builder()
+            .viewDefinition(computedColumns())
+            .names(List.of("dst1"))
+            .tableSchema(declaredSchema())
+            .tableDefaults(ExpressionCreator.struct(false))
+            .operation(AbstractDdlRel.DdlOp.CREATE)
+            .object(AbstractDdlRel.DdlObject.VIEW)
+            .remap(Rel.Remap.of(List.of(0)))
+            .build();
+    SubstraitToCalcite converter = new SubstraitToCalcite(converterProvider, catalogReader);
+
+    assertThrows(UnsupportedOperationException.class, () -> converter.convert(ctas));
+    assertThrows(UnsupportedOperationException.class, () -> converter.convert(createView));
+  }
+
   /** The schema of the object a single DDL statement creates, as Substrait records it. */
   private NamedStruct schemaOf(String sql) throws SqlParseException {
     RelRoot root =

isthmus/src/test/java/io/substrait/isthmus/VirtualTableScanTest.java

@@ -9,6 +9,8 @@
 import com.google.common.collect.ImmutableList;
 import io.substrait.expression.Expression;
 import io.substrait.expression.ExpressionCreator;
+import io.substrait.hint.Hint;
+import io.substrait.relation.Rel;
 import io.substrait.relation.VirtualTableScan;
 import io.substrait.type.NamedStruct;
 import java.io.PrintWriter;
@@ -443,6 +445,62 @@ void nullableStructInsideStructColumnConverts() {
         explain(relNode));
   }
 
+  /**
+   * The emit mapping of a virtual table selects its columns like any other relation's: the scan
+   * produces the whole table and a projection drops what the mapping leaves out. That projection is
+   * what converts back, so such a scan returns as a projection over one -- the same shape every
+   * relation with a mapping comes back as.
+   */
+  @Test
+  void anEmitMappingSelectsTheColumnsItNames() {
+    NamedStruct schema = NamedStruct.of(List.of("col1", "col2"), R.struct(R.I32, R.STRING));
+    VirtualTableScan table =
+        VirtualTableScan.builder()
+            .from(createVirtualTableScan(schema, List.of(sb.i32(2), sb.str("a"))))
+            .remap(Rel.Remap.of(List.of(1)))
+            .build();
+
+    RelNode relNode = substraitToCalcite.convert(table);
+
+    assertEquals(List.of("col2"), relNode.getRowType().getFieldNames());
+    assertEquals(
+        List.of(R.STRING),
+        SubstraitRelVisitor.convert(relNode, extensions).getRecordType().fields());
+  }
+
+  /**
+   * Without a mapping there is no projection to name, and a virtual table's own schema already
+   * names its columns, so the hint is left where it is rather than rebuilding the table around it.
+   */
+  @Test
+  void outputNamesWithoutAMappingAreLeftAlone() {
+    NamedStruct schema = NamedStruct.of(List.of("col1", "col2"), R.struct(R.I32, R.STRING));
+    VirtualTableScan table =
+        VirtualTableScan.builder()
+            .from(createVirtualTableScan(schema, List.of(sb.i32(2), sb.str("a"))))
+            .hint(Hint.builder().addOutputNames("x", "y").build())
+            .build();
+
+    RelNode relNode = substraitToCalcite.convert(table);
+
+    assertInstanceOf(LogicalValues.class, relNode);
+    assertEquals(List.of("col1", "col2"), relNode.getRowType().getFieldNames());
+  }
+
+  /** The names of its hint reach the projection the mapping adds, as they do elsewhere. */
+  @Test
+  void outputNamesReachTheProjectionTheMappingAdds() {
+    NamedStruct schema = NamedStruct.of(List.of("col1", "col2"), R.struct(R.I32, R.STRING));
+    VirtualTableScan table =
+        VirtualTableScan.builder()
+            .from(createVirtualTableScan(schema, List.of(sb.i32(2), sb.str("a"))))
+            .remap(Rel.Remap.of(List.of(1)))
+            .hint(Hint.builder().addOutputNames("label").build())
+            .build();
+
+    assertEquals(List.of("label"), substraitToCalcite.convert(table).getRowType().getFieldNames());
+  }
+
   @SafeVarargs
   private VirtualTableScan createVirtualTableScan(NamedStruct schema, List<Expression>... rows) {
     List<Expression.NestedStruct> structs =