feat(core): support max length, scale, precision in Substrait dialect

Add max_length (FIXED_BINARY, VARCHAR, FIXED_CHAR) and max_scale
(DECIMAL, alongside the existing max_precision) to the dialect
supported_types model, mirroring the schema additions from
substrait-io/substrait#1030.

Closes #808

Author: nielspardon

core/src/main/java/io/substrait/dialect/SupportedType.java

@@ -36,12 +36,30 @@ public abstract class SupportedType {
   public abstract Optional<SystemTypeMetadata> systemMetadata();
 
   /**
-   * The maximum precision supported for the type, if constrained.
+   * The maximum precision supported for the type, if constrained. Applies to the
+   * subsecond-precision temporal types ({@code PRECISION_TIME}, {@code PRECISION_TIMESTAMP}, {@code
+   * PRECISION_TIMESTAMP_TZ}, {@code INTERVAL_COMPOUND}, {@code INTERVAL_DAY}) and, together with
+   * {@link #maxScale()}, to {@code DECIMAL}.
    *
    * @return the optional maximum precision
    */
   public abstract Optional<Integer> maxPrecision();
 
+  /**
+   * The maximum scale supported for a {@code DECIMAL} type, if constrained.
+   *
+   * @return the optional maximum scale
+   */
+  public abstract Optional<Integer> maxScale();
+
+  /**
+   * The maximum length supported for a variable- or fixed-length type ({@code FIXED_BINARY}, {@code
+   * VARCHAR}, {@code FIXED_CHAR}), if constrained.
+   *
+   * @return the optional maximum length
+   */
+  public abstract Optional<Integer> maxLength();
+
   /**
    * Dependency (alias) where a {@code USER_DEFINED} type is declared.
    *
@@ -66,6 +84,8 @@ public boolean isBare() {
         && !metadata().isPresent()
         && !systemMetadata().isPresent()
         && !maxPrecision().isPresent()
+        && !maxScale().isPresent()
+        && !maxLength().isPresent()
         && !source().isPresent()
         && !name().isPresent();
   }

core/src/main/java/io/substrait/dialect/SupportedTypeDeserializer.java

@@ -35,6 +35,12 @@ public SupportedType deserialize(JsonParser p, DeserializationContext ctxt) thro
     if (node.hasNonNull("max_precision")) {
       builder.maxPrecision(node.get("max_precision").asInt());
     }
+    if (node.hasNonNull("max_scale")) {
+      builder.maxScale(node.get("max_scale").asInt());
+    }
+    if (node.hasNonNull("max_length")) {
+      builder.maxLength(node.get("max_length").asInt());
+    }
     if (node.hasNonNull("source")) {
       builder.source(node.get("source").asText());
     }

core/src/main/java/io/substrait/dialect/SupportedTypeSerializer.java

@@ -30,6 +30,12 @@ public void serialize(SupportedType value, JsonGenerator gen, SerializerProvider
     if (value.maxPrecision().isPresent()) {
       gen.writeNumberField("max_precision", value.maxPrecision().get());
     }
+    if (value.maxScale().isPresent()) {
+      gen.writeNumberField("max_scale", value.maxScale().get());
+    }
+    if (value.maxLength().isPresent()) {
+      gen.writeNumberField("max_length", value.maxLength().get());
+    }
     if (value.systemMetadata().isPresent()) {
       provider.defaultSerializeField("system_metadata", value.systemMetadata().get(), gen);
     }

core/src/test/java/io/substrait/dialect/SupportedTypeParameterizedRoundTripTest.java

@@ -0,0 +1,69 @@
+package io.substrait.dialect;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.networknt.schema.Error;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies that the parameterized-type constraints {@code max_length} (for {@code FIXED_BINARY},
+ * {@code VARCHAR}, {@code FIXED_CHAR}) and {@code max_precision}/{@code max_scale} (for {@code
+ * DECIMAL}) serialize as configuration objects, validate against the published {@code
+ * dialect_schema.yaml}, and survive a POJO → YAML → POJO round trip.
+ */
+class SupportedTypeParameterizedRoundTripTest {
+
+  private static Dialect dialectWith(SupportedType type) {
+    return Dialect.builder().addSupportedTypes(type).build();
+  }
+
+  private static void assertRoundTrips(SupportedType type) {
+    assertFalse(type.isBare(), "configured type must not be bare");
+    Dialect original = dialectWith(type);
+    String yaml = Dialect.toYaml(original);
+
+    List<Error> errors = SchemaValidator.validate(yaml);
+    assertTrue(errors.isEmpty(), () -> "Generated dialect failed schema validation: " + errors);
+
+    assertEquals(original, Dialect.load(yaml));
+  }
+
+  @Test
+  void fixedBinaryMaxLength() {
+    SupportedType type =
+        SupportedType.builder().type(TypeKind.FIXED_BINARY).maxLength(1024).build();
+
+    assertTrue(Dialect.toYaml(dialectWith(type)).contains("max_length: 1024"));
+    assertRoundTrips(type);
+  }
+
+  @Test
+  void varcharMaxLength() {
+    SupportedType type = SupportedType.builder().type(TypeKind.VARCHAR).maxLength(255).build();
+
+    assertTrue(Dialect.toYaml(dialectWith(type)).contains("max_length: 255"));
+    assertRoundTrips(type);
+  }
+
+  @Test
+  void fixedCharMaxLength() {
+    SupportedType type = SupportedType.builder().type(TypeKind.FIXED_CHAR).maxLength(64).build();
+
+    assertTrue(Dialect.toYaml(dialectWith(type)).contains("max_length: 64"));
+    assertRoundTrips(type);
+  }
+
+  @Test
+  void decimalMaxPrecisionAndScale() {
+    SupportedType type =
+        SupportedType.builder().type(TypeKind.DECIMAL).maxPrecision(38).maxScale(10).build();
+
+    String yaml = Dialect.toYaml(dialectWith(type));
+    assertTrue(yaml.contains("max_precision: 38"), yaml);
+    assertTrue(yaml.contains("max_scale: 10"), yaml);
+    assertRoundTrips(type);
+  }
+}