feat(isthmus): convert nested struct and map expressions to and from Calcite

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

@@ -127,14 +127,18 @@ else if (operand instanceof Expression.StructLiteral
           };
 
   /**
-   * Converts Calcite ROW constructors into Substrait {@link Expression.StructLiteral}s.
+   * Converts Calcite ROW constructors into Substrait {@link Expression.StructLiteral}s, or into
+   * {@link Expression.NestedStruct}s when the fields are not all literals.
    *
-   * <p>ROW values are always concrete (never null themselves) - if a value is actually null, use
-   * NullLiteral instead of StructLiteral. Therefore, the resulting StructLiteral always has
-   * nullable=false. The ROW's type may be nullable (for regular structs) or non-nullable (for UDT
-   * struct encoding), but the value itself is always concrete.
+   * <p>Either way the struct takes its nullability from the ROW's type, which is where Calcite
+   * keeps it. On a Substrait literal, {@code nullable} marks the literal's type as nullable rather
+   * than the value as null - a value that is actually null is a NullLiteral - so a nullable ROW of
+   * literals does not have to give up being a StructLiteral. The UDT struct encoding depends on
+   * 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.
+   * <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.
    */
   public static final SimpleCallConverter ROW =
       (call, visitor) -> {
@@ -145,10 +149,12 @@ else if (operand instanceof Expression.StructLiteral
         List<Expression> operands =
             call.getOperands().stream().map(visitor).collect(Collectors.toList());
         if (!operands.stream().allMatch(expr -> expr instanceof Expression.Literal)) {
-          throw new IllegalArgumentException("ROW operands must be literals.");
+          return Expression.NestedStruct.builder()
+              .nullable(call.getType().isNullable())
+              .fields(operands)
+              .build();
         }
 
-        // ROW types are never nullable (struct literals are always concrete values).
         // Field nullability comes from individual field types, so match literal nullability
         // to field type nullability.
         List<RelDataTypeField> fieldTypes = call.getType().getFieldList();
@@ -162,9 +168,7 @@ else if (operand instanceof Expression.StructLiteral
                     })
                 .collect(Collectors.toList());
 
-        // Struct literals are always concrete values (never null).
-        // For UDT struct literals, struct-level nullability is in the REINTERPRET target type.
-        return ExpressionCreator.struct(false, literals);
+        return ExpressionCreator.struct(call.getType().isNullable(), literals);
       };
 
   /**

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

@@ -437,7 +437,19 @@ public RexNode visit(Expression.StructLiteral expr, Context context) throws Runt
   public RexNode visit(Expression.ListLiteral expr, Context context) throws RuntimeException {
     List<RexNode> args =
         expr.values().stream().map(l -> l.accept(this, context)).collect(Collectors.toList());
-    return rexBuilder.makeCall(SqlStdOperatorTable.ARRAY_VALUE_CONSTRUCTOR, args);
+    // expr.getType() carries the nullability of the list itself, which Calcite would otherwise
+    // infer as non-nullable from the elements
+    RelDataType listType = typeConverter.toCalcite(typeFactory, expr.getType());
+    return rexBuilder.makeCall(listType, SqlStdOperatorTable.ARRAY_VALUE_CONSTRUCTOR, args);
+  }
+
+  @Override
+  public RexNode visit(Expression.NestedStruct expr, Context context) {
+    List<RexNode> fieldNodes =
+        expr.fields().stream().map(f -> f.accept(this, context)).collect(Collectors.toList());
+    // expr.getType() carries the nullability of the NestedStruct itself
+    RelDataType structType = typeConverter.toCalcite(typeFactory, expr.getType());
+    return rexBuilder.makeCall(structType, SqlStdOperatorTable.ROW, fieldNodes);
   }
 
   @Override
@@ -475,7 +487,25 @@ public RexNode visit(Expression.MapLiteral expr, Context context) throws Runtime
                         entry.getKey().accept(this, context),
                         entry.getValue().accept(this, context)))
             .collect(Collectors.toList());
-    return rexBuilder.makeCall(SqlStdOperatorTable.MAP_VALUE_CONSTRUCTOR, args);
+    // expr.getType() carries the nullability of the map itself, which Calcite would otherwise
+    // infer as non-nullable from the keys and values
+    RelDataType mapType = typeConverter.toCalcite(typeFactory, expr.getType());
+    return rexBuilder.makeCall(mapType, SqlStdOperatorTable.MAP_VALUE_CONSTRUCTOR, args);
+  }
+
+  @Override
+  public RexNode visit(Expression.NestedMap expr, Context context) {
+    List<RexNode> args =
+        expr.values().entrySet().stream()
+            .flatMap(
+                entry ->
+                    Stream.of(
+                        entry.getKey().accept(this, context),
+                        entry.getValue().accept(this, context)))
+            .collect(Collectors.toList());
+    // expr.getType() carries the nullability of the NestedMap itself
+    RelDataType mapType = typeConverter.toCalcite(typeFactory, expr.getType());
+    return rexBuilder.makeCall(mapType, SqlStdOperatorTable.MAP_VALUE_CONSTRUCTOR, args);
   }
 
   @Override

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

@@ -3,21 +3,25 @@
 import io.substrait.expression.Expression;
 import io.substrait.expression.ExpressionCreator;
 import io.substrait.isthmus.CallConverter;
-import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 import org.apache.calcite.rex.RexCall;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.fun.SqlMapValueConstructor;
 
 /**
- * Converts Calcite {@link SqlMapValueConstructor} calls into Substrait map literals.
+ * Converts Calcite {@link SqlMapValueConstructor} calls into Substrait map expressions.
  *
- * <p>Expects an even-numbered operand list (key/value pairs) and produces an {@link Expression} map
- * literal via {@link ExpressionCreator}.
+ * <p>Expects an even-numbered operand list (key/value pairs) and produces an {@link
+ * Expression.MapLiteral} when every key and value is a literal, and an {@link Expression.NestedMap}
+ * otherwise. Either way the map takes its nullability from the call's type, which is where Calcite
+ * keeps it: on a Substrait literal, {@code nullable} marks the literal's type as nullable rather
+ * than the value as null, so a nullable map of literals is still a MapLiteral.
  */
 public class SqlMapValueConstructorCallConverter implements CallConverter {
 
@@ -26,38 +30,53 @@ public SqlMapValueConstructorCallConverter() {}
 
   /**
    * Attempts to convert a Calcite {@link RexCall} representing a {@link SqlMapValueConstructor}
-   * into a Substrait map literal.
+   * into a Substrait map expression.
    *
    * @param call The Calcite call to convert.
    * @param topLevelConverter Function for converting {@link RexNode} operands to Substrait {@link
    *     Expression}s.
    * @return An {@link Optional} containing the converted {@link Expression} if the operator is a
    *     {@link SqlMapValueConstructor}; otherwise {@link Optional#empty()}.
-   * @throws ClassCastException if operands converted by {@code topLevelConverter} are not {@link
-   *     Expression.Literal} instances.
-   * @throws AssertionError if the number of operands is not even (expecting key/value pairs).
+   * @throws IllegalArgumentException if the number of operands is not even (expecting key/value
+   *     pairs).
    */
   @Override
   public Optional<Expression> convert(
       RexCall call, Function<RexNode, Expression> topLevelConverter) {
     SqlOperator operator = call.getOperator();
     if (operator instanceof SqlMapValueConstructor) {
-      return toMapLiteral(call, topLevelConverter);
+      return toMap(call, topLevelConverter);
     }
     return Optional.empty();
   }
 
-  private Optional<Expression> toMapLiteral(
+  private Optional<Expression> toMap(
       RexCall call, Function<RexNode, Expression> topLevelConverter) {
-    List<Expression.Literal> literals =
-        call.operands.stream()
-            .map(t -> ((Expression.Literal) topLevelConverter.apply(t)))
-            .collect(java.util.stream.Collectors.toList());
-    Map<Expression.Literal, Expression.Literal> items = new HashMap<>();
-    assert literals.size() % 2 == 0;
-    for (int i = 0; i < literals.size(); i += 2) {
-      items.put(literals.get(i), literals.get(i + 1));
+    if (call.operands.size() % 2 != 0) {
+      throw new IllegalArgumentException(
+          String.format(
+              "A map value constructor takes key/value pairs, so it must have an even number of"
+                  + " operands, but it has %d.",
+              call.operands.size()));
     }
-    return Optional.of(ExpressionCreator.map(false, items));
+
+    List<Expression> expressions =
+        call.operands.stream().map(topLevelConverter).collect(Collectors.toList());
+
+    // The maps below are LinkedHashMaps so that the pairs keep the order they were written in.
+    if (expressions.stream().allMatch(e -> e instanceof Expression.Literal)) {
+      Map<Expression.Literal, Expression.Literal> literals = new LinkedHashMap<>();
+      for (int i = 0; i < expressions.size(); i += 2) {
+        literals.put(
+            (Expression.Literal) expressions.get(i), (Expression.Literal) expressions.get(i + 1));
+      }
+      return Optional.of(ExpressionCreator.map(call.getType().isNullable(), literals));
+    }
+
+    Map<Expression, Expression> values = new LinkedHashMap<>();
+    for (int i = 0; i < expressions.size(); i += 2) {
+      values.put(expressions.get(i), expressions.get(i + 1));
+    }
+    return Optional.of(ExpressionCreator.nestedMap(call.getType().isNullable(), values));
   }
 }