fix(isthmus)!: keep a struct's declared field nullability in Calcite

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

@@ -71,8 +71,13 @@ public CalciteSchema toSchema(@NonNull final Rel rel) {
       // Create the table if it is not present
       CalciteSchema.TableEntry table = schema.getTable(tableName, CASE_SENSITIVE);
       if (table == null) {
+        // A relation's row type says what its columns are, not whether a value is there, and
+        // Calcite builds one NOT NULL wherever it derives one -- so a schema struct contributes
+        // its fields and not its own nullability.
         RelDataType rowType =
-            typeConverter.toCalcite(typeFactory, namedStruct.struct(), namedStruct.names());
+            typeFactory.createTypeWithNullability(
+                typeConverter.toCalcite(typeFactory, namedStruct.struct(), namedStruct.names()),
+                false);
         schema.add(tableName, new SubstraitTable(tableName, rowType));
       }
     }

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

@@ -945,12 +945,12 @@ public RelNode visit(VirtualTableScan virtualTableScan, Context context) {
    *
    * <p>Names are what a row, list or map value is rebuilt for: a row's field types are checked
    * against the schema when the {@link VirtualTableScan} is built, so a value that is one of those
-   * agrees with its declared type on everything a Substrait type says. It can still be stamped
-   * nullable where it was not -- Calcite makes a struct's fields nullable along with the struct,
-   * and a value that cannot be null stands in a column that can. A value that cannot take the
-   * declared type at all -- a call returning a struct, say, whose names are not the declared ones
-   * -- is reported here: {@link LogicalProject} and {@link LogicalValues} check the row type they
-   * are handed with an {@code assert}, which says nothing at all unless assertions are on.
+   * agrees with its declared type on everything a Substrait type says. Its nullability can still
+   * differ, since Calcite derives that from the expression: a value that cannot be null stands in a
+   * column that can. A value that cannot take the declared type at all -- a call returning a
+   * struct, say, whose names are not the declared ones -- is reported here: {@link LogicalProject}
+   * and {@link LogicalValues} check the row type they are handed with an {@code assert}, which says
+   * nothing at all unless assertions are on.
    *
    * @param value the converted row value
    * @param declaredType the type its column is declared at
@@ -971,10 +971,11 @@ private RexNode valueAsDeclared(RexNode value, RelDataType declaredType) {
   /**
    * Whether a converted value can stand at the type it is declared at.
    *
-   * <p>A value that cannot be null stands in a column that can: Calcite makes a struct's fields
-   * nullable along with the struct, so the fields of a nullable struct are declared that way
-   * whatever the values in them are. The other direction is a mismatch, and so is any difference
-   * beyond nullability -- a field name included, which is the one this rename is about.
+   * <p>A value that cannot be null stands in a column that can: Calcite derives a value's
+   * nullability from its expression, and a value that is never null is valid wherever null is
+   * allowed. The other direction is a mismatch -- a scalar subquery, say, which Calcite types
+   * nullable whatever its Substrait type says -- and so is any difference beyond nullability, a
+   * field name included, which is the one this rename is about.
    */
   private boolean fitsDeclared(RelDataType valueType, RelDataType declaredType) {
     return SqlTypeUtil.equalSansNullability(typeFactory, valueType, declaredType)
@@ -1286,15 +1287,50 @@ protected RelNode applyOutputNames(RelNode relNode, Rel rel, RelNode... inputs)
    */
   private boolean describesColumnsOf(List<Type> fields, RelDataType rowType) {
     for (int field = 0; field < fields.size(); field++) {
-      RelDataType declared = typeConverter.toCalcite(typeFactory, fields.get(field));
-      if (!SqlTypeUtil.equalSansNullability(
-          declared, rowType.getFieldList().get(field).getType())) {
+      if (!sameSansNamesAndNullability(
+          typeConverter.toCalcite(typeFactory, fields.get(field)),
+          rowType.getFieldList().get(field).getType())) {
         return false;
       }
     }
     return true;
   }
 
+  /**
+   * Returns whether two types are the same, ignoring nullability and field names at every depth.
+   *
+   * <p>A column's nullability, and that of anything inside it, is Calcite's to derive from the
+   * expression producing it. The names inside a column are not what a hint restates, and a declared
+   * type converts without a name list, so the names inside it are placeholders. Calcite's own
+   * {@link SqlTypeUtil#equalAsStructSansNullability} ignores both only on the outermost struct.
+   */
+  private static boolean sameSansNamesAndNullability(RelDataType declared, RelDataType actual) {
+    if (declared.isStruct() || actual.isStruct()) {
+      if (!declared.isStruct()
+          || !actual.isStruct()
+          || declared.getFieldCount() != actual.getFieldCount()) {
+        return false;
+      }
+      for (int field = 0; field < declared.getFieldCount(); field++) {
+        if (!sameSansNamesAndNullability(
+            declared.getFieldList().get(field).getType(),
+            actual.getFieldList().get(field).getType())) {
+          return false;
+        }
+      }
+      return true;
+    }
+    if (declared.getKeyType() != null && actual.getKeyType() != null) {
+      return sameSansNamesAndNullability(declared.getKeyType(), actual.getKeyType())
+          && sameSansNamesAndNullability(declared.getValueType(), actual.getValueType());
+    }
+    if (declared.getComponentType() != null && actual.getComponentType() != null) {
+      return declared.getSqlTypeName() == actual.getSqlTypeName()
+          && sameSansNamesAndNullability(declared.getComponentType(), actual.getComponentType());
+    }
+    return SqlTypeUtil.equalSansNullability(declared, actual);
+  }
+
   /**
    * Applies an optional field remap to the given node.
    *

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

@@ -982,9 +982,9 @@ public Rel handleCreateView(CreateView createView) {
   @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.
+    // narrower than its column -- Calcite infers one for a tuple value -- 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());

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

@@ -535,7 +535,10 @@ private RelDataType t(boolean nullable, SqlTypeName typeName, Integer... props)
     }
 
     private RelDataType n(Type substraitType, RelDataType type) {
-      return typeFactory.createTypeWithNullability(type, n(substraitType));
+      // enforceTypeWithNullability, not createTypeWithNullability: the latter
+      // widens a record's fields along with the record, which loses the
+      // nullability the Substrait type declares for them.
+      return typeFactory.enforceTypeWithNullability(type, n(substraitType));
     }
   }
 }

isthmus/src/main/java/io/substrait/isthmus/expression/CallConverters.java

@@ -154,8 +154,8 @@ else if (operand instanceof Expression.StructLiteral
    * that: it builds a deliberately non-nullable ROW, keeping the user-defined type's own
    * nullability in the REINTERPRET target type, and so still arrives here as a StructLiteral.
    *
-   * <p>Each literal's nullability is set to match its field type's nullability. Note that Calcite
-   * makes every field of a nullable record type nullable, so a nullable ROW widens its fields.
+   * <p>Each literal's nullability is set to match its field type's nullability, which the ROW's
+   * type records field by field, whatever the ROW's own nullability is.
    */
   public static final SimpleCallConverter ROW =
       (call, visitor) -> {

isthmus/src/main/java/io/substrait/isthmus/expression/LiteralConverter.java

@@ -196,8 +196,9 @@ public Expression.Literal convert(RexLiteral literal) {
    * <p>This overload is useful when the target type comes from a containing schema rather than the
    * literal itself. Calcite may infer a narrower type for a value in a LogicalValues tuple than for
    * the corresponding row field. Nullability is taken from {@code resultType}, so callers that need
-   * a nullability other than the literal's own should widen the Calcite type with {@link
-   * org.apache.calcite.rel.type.RelDataTypeFactory#createTypeWithNullability} before calling.
+   * a nullability other than the literal's own should set it on the Calcite type with {@link
+   * org.apache.calcite.rel.type.RelDataTypeFactory#enforceTypeWithNullability} before calling:
+   * {@code createTypeWithNullability} would make a struct's fields nullable along with it.
    *
    * @param literal the RexLiteral to convert
    * @param resultType the Calcite type required by the containing schema

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

@@ -1,6 +1,7 @@
 package io.substrait.isthmus;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -320,6 +321,44 @@ void nestedStruct() {
             "topStruct1", "inner1", "inner2", "topStruct2", "inner3", "inner4", "topVarChar"));
   }
 
+  /**
+   * A struct declares its own nullability and its fields' separately, and a trip through Calcite
+   * keeps both: {@code createTypeWithNullability} widens a record's fields along with the record,
+   * which dropped what the schema said about them.
+   */
+  @Test
+  void aNullableStructKeepsWhatItsFieldsDeclare() {
+    Type.Struct substrait =
+        TypeCreator.NULLABLE.struct(TypeCreator.REQUIRED.I32, TypeCreator.NULLABLE.FP64);
+
+    RelDataType calcite = TypeConverter.DEFAULT.toCalcite(type, substrait, List.of("a", "b"));
+
+    assertTrue(calcite.isNullable());
+    assertFalse(calcite.getFieldList().get(0).getType().isNullable());
+    assertTrue(calcite.getFieldList().get(1).getType().isNullable());
+    assertEquals(substrait, TypeConverter.DEFAULT.toSubstrait(calcite));
+  }
+
+  /**
+   * The widening this replaces was recursive, so a struct below the outermost one is pinned too: a
+   * nullable struct holding a required struct keeps the inner struct's own nullability as well as
+   * its fields'.
+   */
+  @Test
+  void aNestedStructKeepsWhatItsFieldsDeclare() {
+    Type.Struct substrait =
+        TypeCreator.NULLABLE.struct(
+            TypeCreator.REQUIRED.struct(TypeCreator.REQUIRED.I32, TypeCreator.NULLABLE.FP64));
+
+    RelDataType calcite =
+        TypeConverter.DEFAULT.toCalcite(type, substrait, List.of("inner", "a", "b"));
+
+    assertEquals(
+        "RecordType(RecordType(INTEGER NOT NULL a, DOUBLE b) NOT NULL inner)",
+        calcite.getFullTypeString());
+    assertEquals(substrait, TypeConverter.DEFAULT.toSubstrait(calcite));
+  }
+
   @ParameterizedTest
   @ValueSource(booleans = {true, false})
   void userDefinedType(boolean nullable) {

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

@@ -273,10 +273,11 @@ void nullableNestedStructWithLiteralsTest() {
 
   @Test
   void nullableStructLiteralTest() {
-    // The same nullability, on a value that is a StructLiteral to begin with. Its fields are
-    // nullable because Calcite makes every field of a nullable record type nullable.
+    // The same nullability, on a value that is a StructLiteral to begin with. Each field keeps its
+    // own nullability, whatever the struct's is.
     Expression.StructLiteral structLiteral =
-        ExpressionCreator.struct(true, ExpressionCreator.i32(true, 7));
+        ExpressionCreator.struct(
+            true, ExpressionCreator.i32(true, 7), ExpressionCreator.i32(false, 8));
 
     Project project =
         Project.builder().expressions(List.of(structLiteral)).input(emptyTable).build();

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

@@ -93,6 +93,31 @@ private Project structColumnProject(Rel structScan, Optional<Hint> hint) {
     return builder.build();
   }
 
+  @Test
+  void namesAStructColumnComputedByAnIfThen() {
+    // Calcite types a CASE over a nullable struct with nullable fields, where the relation declares
+    // the struct's first field required. That field's nullability is the expression's to derive,
+    // like the column's own, so it does not keep the names off the column.
+    Type.Struct inner = TypeCreator.NULLABLE.struct(R.I64, N.STRING);
+    Rel structScan = sb.namedScan(List.of("t"), List.of("s", "x", "y", "k"), List.of(inner, R.I64));
+    Rel project =
+        Project.builder()
+            .input(structScan)
+            .remap(Rel.Remap.offset(2, 1))
+            .addExpressions(
+                sb.ifThen(
+                    List.of(
+                        sb.ifClause(
+                            sb.equal(sb.fieldReference(structScan, 1), sb.i64(1)),
+                            sb.fieldReference(structScan, 0))),
+                    sb.fieldReference(structScan, 0)))
+            .hint(Hint.builder().addOutputNames("renamed", "first", "second").build())
+            .build();
+
+    assertEquals(
+        List.of("renamed"), substraitToCalcite.convert(project).getRowType().getFieldNames());
+  }
+
   @Test
   void keepsCalciteNamesWithoutAHint() {
     RelNode node = substraitToCalcite.convert(projectWithHint(Optional.empty()));

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

@@ -10,12 +10,16 @@
 import io.substrait.expression.ImmutableFieldReference;
 import io.substrait.extension.DefaultExtensionCatalog;
 import io.substrait.relation.AbstractWriteRel;
+import io.substrait.relation.Join;
+import io.substrait.relation.NamedScan;
 import io.substrait.relation.NamedUpdate;
 import io.substrait.relation.Rel;
+import io.substrait.type.NamedStruct;
 import io.substrait.type.TypeCreator;
 import java.util.Arrays;
 import java.util.List;
 import org.apache.calcite.jdbc.CalciteSchema;
+import org.apache.calcite.rel.RelNode;
 import org.junit.jupiter.api.Test;
 
 class SchemaCollectorTest extends PlanTestBase {
@@ -122,9 +126,9 @@ void testUpdate() {
 
     CalciteSchema calciteSchema = schemaCollector.toSchema(rel);
     CalciteSchema schema1 = calciteSchema.getSubSchema("schema1", false);
-    hasTable(schema1, "table1", "RecordType(BOOLEAN col1)");
+    hasTable(schema1, "table1", "RecordType(BOOLEAN NOT NULL col1) NOT NULL");
 
-    hasTable(schema1, "table2", "RecordType(BOOLEAN col1)");
+    hasTable(schema1, "table2", "RecordType(BOOLEAN NOT NULL col1) NOT NULL");
   }
 
   @Test
@@ -146,6 +150,26 @@ void canHandleMultipleSchemas() {
     hasTable(level2b, "t2", "RecordType(INTEGER col2) NOT NULL");
   }
 
+  /**
+   * A nullable schema struct still gives a NOT NULL row type, so Calcite widens the columns on the
+   * null-generating side of a LEFT join as it would for any other table.
+   */
+  @Test
+  void aNullableSchemaStructStillWidensUnderAnOuterJoin() {
+    Rel left = sb.namedScan(List.of("l"), List.of("a"), List.of(R.I32));
+    Rel right =
+        NamedScan.builder()
+            .names(List.of("r"))
+            .initialSchema(NamedStruct.of(List.of("b"), N.struct(R.I32)))
+            .build();
+    Rel join = sb.join(input -> sb.bool(true), Join.JoinType.LEFT, left, right);
+
+    RelNode relNode = substraitToCalcite.convert(join);
+    assertEquals(
+        "RecordType(INTEGER NOT NULL a, INTEGER b) NOT NULL",
+        relNode.getRowType().getFullTypeString());
+  }
+
   @Test
   void canHandleDuplicateNamedScans() {
     Rel table = sb.namedScan(List.of("table"), List.of("col1"), List.of(N.BOOLEAN));