fix(isthmus): use schema types for virtual table literals

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

@@ -163,19 +163,19 @@ public Rel visit(org.apache.calcite.rel.core.Values values) {
     NamedStruct type = typeConverter.toNamedStruct(values.getRowType());
 
     LiteralConverter literalConverter = new LiteralConverter(typeConverter);
-    List<Type> schemaFieldTypes = type.struct().fields();
     List<Expression.NestedStruct> structs =
         values.getTuples().stream()
             .map(
                 list -> {
-                  // Use schema nullability since Calcite infers non-nullable for all non-null
-                  // values
+                  // Calcite may infer a narrower type for a tuple literal than for its row field.
+                  // Virtual table rows must use the complete schema type.
                   List<Expression> fields =
                       IntStream.range(0, list.size())
                           .mapToObj(
                               i ->
                                   literalConverter.convert(
-                                      list.get(i), schemaFieldTypes.get(i).nullable()))
+                                      list.get(i),
+                                      values.getRowType().getFieldList().get(i).getType()))
                           .collect(Collectors.toUnmodifiableList());
                   return ExpressionCreator.nestedStruct(false, fields);
                 })

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

@@ -21,6 +21,7 @@
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.util.DateString;
@@ -91,9 +92,7 @@ private static BigDecimal bd(RexLiteral literal) {
    * @throws UnsupportedOperationException if the literal type/value cannot be handled
    */
   public Expression.Literal convert(RexLiteral literal) {
-    // convert type first to guarantee we can handle the value.
-    final Type type = typeConverter.toSubstrait(literal.getType());
-    return convert(literal, type.nullable());
+    return convert(literal, literal.getType());
   }
 
   /**
@@ -108,14 +107,34 @@ public Expression.Literal convert(RexLiteral literal) {
    * @return the converted Substrait Literal
    */
   public Expression.Literal convert(RexLiteral literal, boolean nullable) {
+    return convert(literal, literal.getType(), nullable);
+  }
+
+  /**
+   * Converts a RexLiteral to a Substrait Literal using the specified result type.
+   *
+   * <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.
+   *
+   * @param literal the RexLiteral to convert
+   * @param resultType the Calcite type required by the containing schema
+   * @return the converted Substrait Literal
+   */
+  public Expression.Literal convert(RexLiteral literal, RelDataType resultType) {
+    Type type = typeConverter.toSubstrait(resultType);
+    return convert(literal, resultType, type.nullable());
+  }
+
+  private Expression.Literal convert(RexLiteral literal, RelDataType resultType, boolean nullable) {
     if (literal.isNull()) {
-      final Type type = typeConverter.toSubstrait(literal.getType());
+      final Type type = typeConverter.toSubstrait(resultType);
       final Type typeWithNullability =
           nullable ? TypeCreator.asNullable(type) : TypeCreator.asNotNullable(type);
       return ExpressionCreator.typedNull(typeWithNullability);
     }
 
-    switch (literal.getType().getSqlTypeName()) {
+    switch (resultType.getSqlTypeName()) {
       case TINYINT:
         return ExpressionCreator.i8(nullable, i(literal).intValue());
       case SMALLINT:
@@ -145,23 +164,23 @@ public Expression.Literal convert(RexLiteral literal, boolean nullable) {
         {
           BigDecimal bd = bd(literal);
           return ExpressionCreator.decimal(
-              nullable, bd, literal.getType().getPrecision(), literal.getType().getScale());
+              nullable, bd, resultType.getPrecision(), resultType.getScale());
         }
       case VARCHAR:
         {
-          if (literal.getType().getPrecision() == RelDataType.PRECISION_NOT_SPECIFIED) {
+          if (resultType.getPrecision() == RelDataType.PRECISION_NOT_SPECIFIED) {
             return ExpressionCreator.string(nullable, s(literal));
           }
 
-          return ExpressionCreator.varChar(nullable, s(literal), literal.getType().getPrecision());
+          return ExpressionCreator.varChar(nullable, s(literal), resultType.getPrecision());
         }
       case BINARY:
         return ExpressionCreator.fixedBinary(
             nullable,
             ByteString.copyFrom(
                 padRightIfNeeded(
                     literal.getValueAs(org.apache.calcite.avatica.util.ByteString.class),
-                    literal.getType().getPrecision())));
+                    resultType.getPrecision())));
       case VARBINARY:
         return ExpressionCreator.binary(
             nullable, ByteString.copyFrom(literal.getValueAs(byte[].class)));
@@ -246,21 +265,29 @@ public Expression.Literal convert(RexLiteral literal, boolean nullable) {
         {
           List<RexLiteral> literals = (List<RexLiteral>) literal.getValue();
           return ExpressionCreator.struct(
-              nullable, literals.stream().map(this::convert).collect(Collectors.toList()));
+              nullable,
+              IntStream.range(0, literals.size())
+                  .mapToObj(
+                      i -> convert(literals.get(i), resultType.getFieldList().get(i).getType()))
+                  .collect(Collectors.toList()));
         }
 
       case ARRAY:
         {
           List<RexLiteral> literals = (List<RexLiteral>) literal.getValue();
+          RelDataType componentType = Objects.requireNonNull(resultType.getComponentType());
           return ExpressionCreator.list(
-              nullable, literals.stream().map(this::convert).collect(Collectors.toList()));
+              nullable,
+              literals.stream()
+                  .map(nestedLiteral -> convert(nestedLiteral, componentType))
+                  .collect(Collectors.toList()));
         }
 
       default:
         throw new UnsupportedOperationException(
             String.format(
                 "Unable to convert the value of %s of type %s to a literal.",
-                literal, literal.getType().getSqlTypeName()));
+                literal, resultType.getSqlTypeName()));
     }
   }
 

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

@@ -2,22 +2,29 @@
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 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 com.google.common.collect.ImmutableList;
 import io.substrait.expression.Expression;
 import io.substrait.expression.ExpressionCreator;
 import io.substrait.relation.VirtualTableScan;
 import io.substrait.type.NamedStruct;
 import java.io.PrintWriter;
 import java.io.StringWriter;
+import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.stream.Collectors;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.RelWriter;
 import org.apache.calcite.rel.externalize.RelWriterImpl;
+import org.apache.calcite.rel.logical.LogicalValues;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.sql.SqlExplainLevel;
+import org.apache.calcite.sql.type.SqlTypeName;
 import org.junit.jupiter.api.Test;
 
 class VirtualTableScanTest extends PlanTestBase {
@@ -111,6 +118,26 @@ void mixedNullabilityRoundTrip() {
     assertFullRoundTrip(virtualTableScan);
   }
 
+  @Test
+  void valuesLiteralUsesSchemaType() {
+    RelDataType rowType = typeFactory.builder().add("col1", SqlTypeName.INTEGER).build();
+    RexLiteral literal =
+        builder
+            .getRexBuilder()
+            .makeExactLiteral(BigDecimal.ONE, typeFactory.createSqlType(SqlTypeName.TINYINT));
+    LogicalValues values =
+        LogicalValues.create(
+            builder.getCluster(), rowType, ImmutableList.of(ImmutableList.of(literal)));
+
+    VirtualTableScan converted =
+        assertInstanceOf(
+            VirtualTableScan.class, SubstraitRelVisitor.convert(values, converterProvider));
+    assertEquals(R.I32, converted.getInitialSchema().struct().fields().get(0));
+    Expression.I32Literal convertedLiteral =
+        assertInstanceOf(Expression.I32Literal.class, converted.getRows().get(0).fields().get(0));
+    assertEquals(1, convertedLiteral.value());
+  }
+
   @SafeVarargs
   private VirtualTableScan createVirtualTableScan(NamedStruct schema, List<Expression>... rows) {
     List<Expression.NestedStruct> structs =