fix(isthmus)!: apply the projection a read relation carries

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

@@ -4,6 +4,7 @@
 import io.substrait.expression.Expression;
 import io.substrait.expression.Expression.SortDirection;
 import io.substrait.expression.FunctionArg;
+import io.substrait.expression.MaskExpression;
 import io.substrait.extension.FunctionBindingResolver;
 import io.substrait.extension.ResolvedAggregateBinding;
 import io.substrait.extension.ResolvedArgument;
@@ -204,7 +205,7 @@ public RelNode visit(Filter filter, Context context) throws RuntimeException {
   @Override
   public RelNode visit(NamedScan namedScan, Context context) throws RuntimeException {
     RelNode node = relBuilder.scan(namedScan.getNames()).build();
-    return applyRelCommon(node, namedScan);
+    return applyRelCommon(applyProjection(node, namedScan.getProjection()), namedScan);
   }
 
   @Override
@@ -866,11 +867,6 @@ public RelNode visit(NamedDdl namedDdl, Context context) {
 
   @Override
   public RelNode visit(VirtualTableScan virtualTableScan, Context context) {
-    if (virtualTableScan.getProjection().isPresent()) {
-      throw new UnsupportedOperationException(
-          "Projection on a VirtualTableScan is not supported: its columns would have to be "
-              + "masked before an emit mapping selects from them");
-    }
     // A schema's names are one per field at every level of the struct, in depth-first order, so
     // they have to be handed to the conversion rather than paired with the row type afterwards:
     // with a nested struct anywhere in the schema the two lists do not even have the same length.
@@ -921,7 +917,9 @@ public RelNode visit(VirtualTableScan virtualTableScan, Context context) {
         tuplesBuilder.add(tupleBuilder.build());
       }
       return applyRelCommon(
-          LogicalValues.create(relBuilder.getCluster(), rowType, tuplesBuilder.build()),
+          applyProjection(
+              LogicalValues.create(relBuilder.getCluster(), rowType, tuplesBuilder.build()),
+              virtualTableScan.getProjection()),
           virtualTableScan);
     } else {
       // A row that does not fit a LogicalValues tuple keeps its expressions, in a relation of our
@@ -930,7 +928,10 @@ public RelNode visit(VirtualTableScan virtualTableScan, Context context) {
       // consumer whose planner only knows Calcite's own relations can expand it with
       // VirtualTableExpansionRule.
       return applyRelCommon(
-          VirtualTable.create(relBuilder.getCluster(), rowType, convertedRows), virtualTableScan);
+          applyProjection(
+              VirtualTable.create(relBuilder.getCluster(), rowType, convertedRows),
+              virtualTableScan.getProjection()),
+          virtualTableScan);
     }
   }
 
@@ -1182,6 +1183,51 @@ public RelNode visitFallback(Rel rel, Context context) throws RuntimeException {
             rel, rel.getClass().getCanonicalName(), this.getClass().getCanonicalName()));
   }
 
+  /**
+   * Applies a read relation's projection to the node its scan was converted into.
+   *
+   * <p>A projection masks the columns of the initial schema, and the read produces the ones it
+   * leaves: {@link io.substrait.relation.AbstractReadRel#deriveRecordType()} applies it to that
+   * schema, so an emit mapping's indices, and every field reference a parent relation makes, count
+   * the masked columns. Building the scan from the schema alone leaves the node one column list and
+   * the relation another.
+   *
+   * <p>The columns come out in the order the mask lists them. That is the order {@link
+   * io.substrait.expression.MaskExpressionTypeProjector} builds the record type in, and the node
+   * has to carry the columns the relation says it carries. Spec v0.102.0 describes a mask as
+   * removing columns and asks whether reordering should be supported at all, so this order follows
+   * the record type the model derives rather than a rule the specification settles.
+   *
+   * <p>Only a mask that selects whole columns is converted. A mask can also select inside a column
+   * -- some of a struct's fields, some of a list's elements, some of a map's entries -- and
+   * applying that would mean rebuilding the column's value from the parts the mask keeps, which
+   * this conversion does not do. Such a mask is reported rather than applied to the columns it
+   * selects whole, which would drop the rest of what it says.
+   *
+   * @param relNode the node the read was converted into
+   * @param projection the projection the read carries, if any
+   * @return the node, with the masked columns projected out of it
+   */
+  private RelNode applyProjection(RelNode relNode, Optional<MaskExpression> projection) {
+    if (projection.isEmpty()) {
+      return relNode;
+    }
+    List<MaskExpression.StructItem> items = projection.get().getSelect().getStructItems();
+    RelDataType rowType = relNode.getRowType();
+    List<RexNode> rexList = new ArrayList<>(items.size());
+    for (MaskExpression.StructItem item : items) {
+      if (item.getChild().isPresent()) {
+        throw new UnsupportedOperationException(
+            "A read projection that selects inside a column is not supported: only a mask that "
+                + "selects whole columns is applied, and pruning a struct, a list or a map would "
+                + "have to rebuild the column's value");
+      }
+      rexList.add(
+          new RexInputRef(item.getField(), rowType.getFieldList().get(item.getField()).getType()));
+    }
+    return relBuilder.push(relNode).project(rexList).build();
+  }
+
   /**
    * Applies the parts of a relation's {@code RelCommon} that Calcite can hold: its emit mapping
    * first, and then the alternative output field names of its hint.

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

@@ -0,0 +1,222 @@
+package io.substrait.isthmus;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.substrait.expression.Expression;
+import io.substrait.expression.ImmutableMaskExpression;
+import io.substrait.expression.MaskExpression;
+import io.substrait.hint.Hint;
+import io.substrait.relation.Filter;
+import io.substrait.relation.NamedScan;
+import io.substrait.relation.Rel;
+import io.substrait.relation.VirtualTableScan;
+import io.substrait.type.NamedStruct;
+import java.util.List;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.TableScan;
+import org.junit.jupiter.api.Test;
+
+/**
+ * The projection a read relation carries masks the columns of its initial schema, and the read
+ * produces the ones the mask leaves.
+ */
+class ReadProjectionTest extends PlanTestBase {
+
+  private final NamedScan scan =
+      (NamedScan)
+          sb.namedScan(List.of("t"), List.of("a", "b", "c"), List.of(R.I64, N.STRING, R.FP64));
+
+  /** A mask selecting whole columns, by the index each has in the initial schema. */
+  private static MaskExpression columns(int... fields) {
+    ImmutableMaskExpression.StructSelect.Builder select = MaskExpression.StructSelect.builder();
+    for (int field : fields) {
+      select.addStructItems(MaskExpression.StructItem.of(field));
+    }
+    return MaskExpression.builder().select(select.build()).build();
+  }
+
+  @Test
+  void aProjectionMasksTheColumnsANamedScanReads() {
+    NamedScan masked = NamedScan.builder().from(scan).projection(columns(0, 2)).build();
+
+    RelNode relNode = substraitToCalcite.convert(masked);
+
+    assertEquals(List.of("a", "c"), relNode.getRowType().getFieldNames());
+    assertEquals(
+        masked.getRecordType(),
+        SubstraitRelVisitor.convert(relNode, converterProvider).getRecordType());
+  }
+
+  /**
+   * A mask that lists its columns out of schema order produces them in the order it lists. That is
+   * the order the model derives the record type in, and the node has to carry the columns the
+   * relation says it carries. Spec v0.102.0 describes a mask as removing columns and asks whether
+   * reordering should be supported at all, so this pins what the model already derives rather than
+   * a rule the specification settles.
+   */
+  @Test
+  void theColumnsComeOutInTheOrderTheMaskListsThem() {
+    NamedScan reordered = NamedScan.builder().from(scan).projection(columns(2, 0)).build();
+
+    RelNode relNode = substraitToCalcite.convert(reordered);
+
+    assertEquals(List.of("c", "a"), relNode.getRowType().getFieldNames());
+    assertEquals(
+        reordered.getRecordType(),
+        SubstraitRelVisitor.convert(relNode, converterProvider).getRecordType());
+  }
+
+  /**
+   * An emit mapping selects from the columns the mask leaves, not from the schema it masked: on
+   * this scan index 1 is the second column the mask keeps, which is the schema's third.
+   */
+  @Test
+  void anEmitMappingSelectsFromTheMaskedColumns() {
+    NamedScan masked =
+        NamedScan.builder().from(scan).projection(columns(0, 2)).remap(sb.remap(1)).build();
+
+    RelNode relNode = substraitToCalcite.convert(masked);
+
+    assertEquals(List.of("c"), relNode.getRowType().getFieldNames());
+    assertEquals(
+        masked.getRecordType(),
+        SubstraitRelVisitor.convert(relNode, converterProvider).getRecordType());
+  }
+
+  /**
+   * So does a field reference a parent relation makes against the read: index 1 is the fp64 column
+   * the mask leaves there, and not the string the schema has at that index.
+   */
+  @Test
+  void aParentRelationReferencesTheMaskedColumns() {
+    NamedScan masked = NamedScan.builder().from(scan).projection(columns(0, 2)).build();
+    Filter filter = sb.filter(input -> sb.equal(sb.fieldReference(input, 1), sb.fp64(5)), masked);
+
+    RelNode relNode = substraitToCalcite.convert(filter);
+
+    Rel converted = SubstraitRelVisitor.convert(relNode, converterProvider);
+    assertEquals(filter.getRecordType(), converted.getRecordType());
+    assertEquals(filter.getCondition(), assertInstanceOf(Filter.class, converted).getCondition());
+  }
+
+  /** A mask that selects every column in order leaves the scan the columns it already reads. */
+  @Test
+  void aProjectionSelectingEveryColumnLeavesTheScanAlone() {
+    NamedScan masked = NamedScan.builder().from(scan).projection(columns(0, 1, 2)).build();
+
+    assertInstanceOf(TableScan.class, substraitToCalcite.convert(masked));
+  }
+
+  /**
+   * The masked columns are the ones the relation produces, so they are the ones an output name from
+   * a hint names -- and the projection the mask adds is where those names can go.
+   */
+  @Test
+  void anOutputNameFromAHintNamesAMaskedColumn() {
+    NamedScan masked =
+        NamedScan.builder()
+            .from(scan)
+            .projection(columns(0, 2))
+            .hint(Hint.builder().addOutputNames("x", "y").build())
+            .build();
+
+    assertEquals(
+        List.of("x", "y"), substraitToCalcite.convert(masked).getRowType().getFieldNames());
+  }
+
+  /**
+   * A mask selects by index: a schema's names are not uniquified, so a name can stand for more than
+   * one column.
+   */
+  @Test
+  void aProjectionSelectsByIndexWhereTwoColumnsShareAName() {
+    NamedScan sharedNames =
+        (NamedScan)
+            sb.namedScan(List.of("t"), List.of("c", "c", "d"), List.of(R.I64, R.STRING, R.FP64));
+    NamedScan masked = NamedScan.builder().from(sharedNames).projection(columns(1, 2)).build();
+
+    RelNode relNode = substraitToCalcite.convert(masked);
+
+    assertEquals(
+        masked.getRecordType(),
+        SubstraitRelVisitor.convert(relNode, converterProvider).getRecordType());
+  }
+
+  @Test
+  void aProjectionMasksTheColumnsAVirtualTableReads() {
+    NamedStruct schema =
+        NamedStruct.of(List.of("col1", "col2", "col3"), R.struct(R.I32, R.STRING, R.BOOLEAN));
+    VirtualTableScan table =
+        VirtualTableScan.builder()
+            .initialSchema(schema)
+            .addRows(
+                Expression.NestedStruct.builder()
+                    .addFields(sb.i32(2), sb.str("a"), sb.bool(true))
+                    .build())
+            .projection(columns(1, 2))
+            .build();
+
+    RelNode relNode = substraitToCalcite.convert(table);
+
+    assertEquals(List.of("col2", "col3"), relNode.getRowType().getFieldNames());
+    assertEquals(
+        table.getRecordType(),
+        SubstraitRelVisitor.convert(relNode, converterProvider).getRecordType());
+  }
+
+  /** A row that no {@code LogicalValues} tuple holds is computed, and masked the same way. */
+  @Test
+  void aProjectionMasksTheColumnsAComputedVirtualTableReads() {
+    NamedStruct schema =
+        NamedStruct.of(List.of("col1", "col2", "col3"), R.struct(R.I32, R.FP64, R.STRING));
+    VirtualTableScan table =
+        VirtualTableScan.builder()
+            .initialSchema(schema)
+            .addRows(
+                Expression.NestedStruct.builder()
+                    .addFields(sb.i32(2), sb.add(sb.fp64(4.4), sb.fp64(4.5)), sb.str("a"))
+                    .build())
+            .projection(columns(1, 2))
+            .build();
+
+    RelNode relNode = substraitToCalcite.convert(table);
+
+    assertEquals(List.of("col2", "col3"), relNode.getRowType().getFieldNames());
+    assertEquals(
+        table.getRecordType(),
+        SubstraitRelVisitor.convert(relNode, converterProvider).getRecordType());
+  }
+
+  /**
+   * A mask can select inside a column as well, keeping some of a struct's fields or some of a
+   * list's elements. Calcite reads a column as it stands, so such a mask is reported rather than
+   * applied to the column it selects from.
+   */
+  @Test
+  void aProjectionThatSelectsInsideAColumnIsRefused() {
+    NamedScan structScan =
+        (NamedScan)
+            sb.namedScan(List.of("t"), List.of("s", "x", "y"), List.of(R.struct(R.I64, R.STRING)));
+    MaskExpression insideAColumn =
+        MaskExpression.builder()
+            .select(
+                MaskExpression.StructSelect.builder()
+                    .addStructItems(
+                        MaskExpression.StructItem.of(
+                            0,
+                            MaskExpression.StructSelect.builder()
+                                .addStructItems(MaskExpression.StructItem.of(1))
+                                .build()))
+                    .build())
+            .build();
+    NamedScan masked = NamedScan.builder().from(structScan).projection(insideAColumn).build();
+
+    assertTrue(
+        assertThrows(UnsupportedOperationException.class, () -> substraitToCalcite.convert(masked))
+            .getMessage()
+            .contains("selects inside a column"));
+  }
+}

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

@@ -9,7 +9,6 @@
 import com.google.common.collect.ImmutableList;
 import io.substrait.expression.Expression;
 import io.substrait.expression.ExpressionCreator;
-import io.substrait.expression.MaskExpression;
 import io.substrait.hint.Hint;
 import io.substrait.relation.Rel;
 import io.substrait.relation.VirtualTableScan;
@@ -489,34 +488,6 @@ void outputNamesWithoutAMappingAreLeftAlone() {
     assertEquals(List.of("col1", "col2"), relNode.getRowType().getFieldNames());
   }
 
-  /**
-   * A projection masks a read relation's columns before anything else selects from them -- {@link
-   * io.substrait.relation.AbstractReadRel#deriveRecordType()} applies it to the initial schema --
-   * so an emit mapping's indices count the columns it leaves. Isthmus builds the row type from the
-   * unmasked schema and reads the projection nowhere, so a scan carrying one is refused rather than
-   * converted against the wrong columns.
-   */
-  @Test
-  void aProjectionOnAVirtualTableIsRefused() {
-    NamedStruct schema = NamedStruct.of(List.of("col1", "col2"), R.struct(R.I32, R.STRING));
-    VirtualTableScan table =
-        VirtualTableScan.builder()
-            .from(virtualTable(schema, List.of(sb.i32(2), sb.str("a"))))
-            .projection(
-                MaskExpression.builder()
-                    .select(
-                        MaskExpression.StructSelect.builder()
-                            .addStructItems(MaskExpression.StructItem.of(1))
-                            .build())
-                    .build())
-            .build();
-
-    assertTrue(
-        assertThrows(UnsupportedOperationException.class, () -> substraitToCalcite.convert(table))
-            .getMessage()
-            .contains("Projection on a VirtualTableScan is not supported"));
-  }
-
   /**
    * A virtual table's row type carries the names its schema gives it, which nothing uniquifies, so
    * the mapping has to select its columns by index: resolving a field by name would give the