feat(core): add typeVariationReference to UserDefined type

core/src/main/java/io/substrait/dsl/SubstraitBuilder.java

@@ -1507,7 +1507,24 @@ public Expression.WindowFunctionInvocation windowFn(
    * @return a new {@link Type.UserDefined}
    */
   public Type.UserDefined userDefinedType(String urn, String typeName) {
-    return Type.UserDefined.builder().urn(urn).name(typeName).nullable(false).build();
+    return userDefinedType(urn, typeName, 0);
+  }
+
+  /**
+   * Creates a user-defined type with the specified URN, type name, and type variation.
+   *
+   * @param urn the URN of the extension containing the type
+   * @param typeName the name of the user-defined type
+   * @param typeVariationReference the type variation reference
+   * @return a new {@link Type.UserDefined}
+   */
+  public Type.UserDefined userDefinedType(String urn, String typeName, int typeVariationReference) {
+    return Type.UserDefined.builder()
+        .urn(urn)
+        .name(typeName)
+        .nullable(false)
+        .typeVariationReference(typeVariationReference)
+        .build();
   }
 
   // Misc

core/src/main/java/io/substrait/type/Type.java

@@ -422,11 +422,11 @@ public <R, E extends Throwable> R accept(final TypeVisitor<R, E> typeVisitor) th
   }
 
   @Value.Immutable
-  abstract class UserDefined implements Type {
+  interface UserDefined extends Type {
 
-    public abstract String urn();
+    String urn();
 
-    public abstract String name();
+    String name();
 
     /**
      * Returns the type parameters for this user-defined type.
@@ -437,16 +437,29 @@ abstract class UserDefined implements Type {
      * @return a list of type parameters, or an empty list if this type is not parameterized
      */
     @Value.Default
-    public java.util.List<Parameter> typeParameters() {
+    default java.util.List<Parameter> typeParameters() {
       return java.util.Collections.emptyList();
     }
 
-    public static ImmutableType.UserDefined.Builder builder() {
+    /**
+     * Returns the type variation reference for this user-defined type.
+     *
+     * <p>Type variations allow different physical representations or semantics for the same logical
+     * type. The reference value maps to an {@code ExtensionTypeVariation} declaration in the plan.
+     *
+     * @return the type variation reference, or {@code 0} if using the default variation
+     */
+    @Value.Default
+    default int typeVariationReference() {
+      return 0;
+    }
+
+    static ImmutableType.UserDefined.Builder builder() {
       return ImmutableType.UserDefined.builder();
     }
 
     @Override
-    public <R, E extends Throwable> R accept(TypeVisitor<R, E> typeVisitor) throws E {
+    default <R, E extends Throwable> R accept(TypeVisitor<R, E> typeVisitor) throws E {
       return typeVisitor.visit(this);
     }
   }

core/src/main/java/io/substrait/type/TypeCreator.java

@@ -112,7 +112,16 @@ public Type.Map map(Type key, Type value) {
   }
 
   public Type userDefined(String urn, String name) {
-    return Type.UserDefined.builder().nullable(nullable).urn(urn).name(name).build();
+    return userDefined(urn, name, 0);
+  }
+
+  public Type userDefined(String urn, String name, int typeVariationReference) {
+    return Type.UserDefined.builder()
+        .nullable(nullable)
+        .urn(urn)
+        .name(name)
+        .typeVariationReference(typeVariationReference)
+        .build();
   }
 
   public static TypeCreator of(boolean nullability) {

core/src/main/java/io/substrait/type/proto/BaseProtoConverter.java

@@ -175,6 +175,7 @@ public final T visit(final Type.Map expr) {
   public final T visit(final Type.UserDefined expr) {
     int ref =
         extensionCollector.getTypeReference(SimpleExtension.TypeAnchor.of(expr.urn(), expr.name()));
-    return typeContainer(expr).userDefined(ref, expr.typeParameters());
+    return typeContainer(expr)
+        .userDefined(ref, expr.typeVariationReference(), expr.typeParameters());
   }
 }

core/src/main/java/io/substrait/type/proto/BaseProtoTypes.java

@@ -131,10 +131,14 @@ public final T struct(T... types) {
 
   public abstract T map(T key, T value);
 
-  public abstract T userDefined(int ref);
+  public T userDefined(int ref, int typeVariationReference) {
+    return userDefined(ref, typeVariationReference, java.util.Collections.emptyList());
+  }
 
   public abstract T userDefined(
-      int ref, java.util.List<io.substrait.type.Type.Parameter> typeParameters);
+      int ref,
+      int typeVariationReference,
+      java.util.List<io.substrait.type.Type.Parameter> typeParameters);
 
   protected abstract T wrap(Object o);
 

core/src/main/java/io/substrait/type/proto/ProtoTypeConverter.java

@@ -107,6 +107,7 @@ public Type from(io.substrait.proto.Type type) {
                   userDefined.getTypeParametersList().stream()
                       .map(this::from)
                       .collect(java.util.stream.Collectors.toList()))
+              .typeVariationReference(userDefined.getTypeVariationReference())
               .build();
         }
       case USER_DEFINED_TYPE_REFERENCE:

core/src/main/java/io/substrait/type/proto/TypeProtoConverter.java

@@ -180,18 +180,15 @@ public Type map(Type key, Type value) {
           Type.Map.newBuilder().setKey(key).setValue(value).setNullability(nullability).build());
     }
 
-    @Override
-    public Type userDefined(int ref) {
-      return wrap(
-          Type.UserDefined.newBuilder().setTypeReference(ref).setNullability(nullability).build());
-    }
-
     @Override
     public Type userDefined(
-        int ref, java.util.List<io.substrait.type.Type.Parameter> typeParameters) {
+        int ref,
+        int typeVariationReference,
+        java.util.List<io.substrait.type.Type.Parameter> typeParameters) {
       return wrap(
           Type.UserDefined.newBuilder()
               .setTypeReference(ref)
+              .setTypeVariationReference(typeVariationReference)
               .setNullability(nullability)
               .addAllTypeParameters(
                   typeParameters.stream()

core/src/test/java/io/substrait/extension/TypeExtensionTest.java

@@ -78,6 +78,24 @@ void roundtripCustomType() {
     io.substrait.proto.Plan protoPlan = planProtoConverter.toProto(plan);
     Plan planReturned = protoPlanConverter.from(protoPlan);
     assertEquals(plan, planReturned);
+
+    // verify default typeVariationReference is 0
+    assertEquals(0, ((Type.UserDefined) customType1).typeVariationReference());
+  }
+
+  @Test
+  void roundtripCustomTypeWithVariationReference() {
+    Type customTypeWithVariation = sb.userDefinedType(URN, "customType1", 42);
+
+    List<String> tableName = Stream.of("example").collect(Collectors.toList());
+    List<String> columnNames = Stream.of("custom_type_column").collect(Collectors.toList());
+    List<Type> types = Stream.of(customTypeWithVariation).collect(Collectors.toList());
+
+    Plan plan = sb.plan(sb.root(sb.namedScan(tableName, columnNames, types)));
+
+    io.substrait.proto.Plan protoPlan = planProtoConverter.toProto(plan);
+    Plan planReturned = protoPlanConverter.from(protoPlan);
+    assertEquals(plan, planReturned);
   }
 
   @Test