update

.baseline/checkstyle/checkstyle.xml

@@ -160,6 +160,9 @@
         <module name="ClassTypeParameterName"> <!-- Java Style Guide: Type variable names -->
             <property name="format" value="(^[A-Z][0-9]?)$|([A-Z][a-zA-Z0-9]*[T]$)"/>
         </module>
+        <module name="InterfaceTypeParameterName"> <!-- Java Style Guide: Type variable names -->
+            <property name="format" value="(^[A-Z][0-9]?)$|([A-Z][a-zA-Z0-9]*[T]$)"/>
+        </module>
         <module name="CovariantEquals"/> <!-- Java Coding Guidelines: Override ``Object#equals`` consistently -->
         <module name="DefaultComesLast"/> <!-- Java Style Guide: The default case is present -->
         <module name="EmptyBlock"> <!-- Java Style Guide: Empty blocks: documented -->

api/src/main/java/org/apache/iceberg/Accessors.java

@@ -232,6 +232,11 @@ public Map<Integer, Accessor<StructLike>> struct(
       return accessors;
     }
 
+    @Override
+    public Map<Integer, Accessor<StructLike>> variant(Types.VariantType variant) {
+      return null;
+    }
+
     @Override
     public Map<Integer, Accessor<StructLike>> field(
         Types.NestedField field, Map<Integer, Accessor<StructLike>> fieldResult) {

api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java

@@ -124,6 +124,11 @@ public Type map(Types.MapType map, Supplier<Type> keyFuture, Supplier<Type> valu
     }
   }
 
+  @Override
+  public Type variant(Types.VariantType variant) {
+    return variant;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     return primitive;

api/src/main/java/org/apache/iceberg/types/AssignIds.java

@@ -92,6 +92,11 @@ public Type map(Types.MapType map, Supplier<Type> keyFuture, Supplier<Type> valu
     }
   }
 
+  @Override
+  public Type variant(Types.VariantType variant) {
+    return variant;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     return primitive;

api/src/main/java/org/apache/iceberg/types/CheckCompatibility.java

@@ -250,6 +250,16 @@ public List<String> map(
     }
   }
 
+  @Override
+  public List<String> variant(Types.VariantType readVariant) {
+    if (currentType.isVariantType()) {
+      return NO_ERRORS;
+    }
+
+    // Currently promotion is not allowed to variant type
+    return ImmutableList.of(String.format(": %s cannot be read as a %s", currentType, readVariant));
+  }
+
   @Override
   public List<String> primitive(Type.PrimitiveType readPrimitive) {
     if (currentType.equals(readPrimitive)) {

api/src/main/java/org/apache/iceberg/types/FindTypeVisitor.java

@@ -77,9 +77,9 @@ public Type map(Types.MapType map, Type keyResult, Type valueResult) {
   }
 
   @Override
-  public Type variant() {
-    if (predicate.test(Types.VariantType.get())) {
-      return Types.VariantType.get();
+  public Type variant(Types.VariantType variant) {
+    if (predicate.test(variant)) {
+      return variant;
     }
 
     return null;

api/src/main/java/org/apache/iceberg/types/GetProjectedIds.java

@@ -47,7 +47,9 @@ public Set<Integer> struct(Types.StructType struct, List<Set<Integer>> fieldResu
 
   @Override
   public Set<Integer> field(Types.NestedField field, Set<Integer> fieldResult) {
-    if ((includeStructIds && field.type().isStructType()) || field.type().isPrimitiveType()) {
+    if ((includeStructIds && field.type().isStructType())
+        || field.type().isPrimitiveType()
+        || field.type().isVariantType()) {
       fieldIds.add(field.fieldId());
     }
     return fieldIds;
@@ -72,4 +74,9 @@ public Set<Integer> map(Types.MapType map, Set<Integer> keyResult, Set<Integer>
     }
     return fieldIds;
   }
+
+  @Override
+  public Set<Integer> variant(Types.VariantType variant) {
+    return null;
+  }
 }

api/src/main/java/org/apache/iceberg/types/IndexById.java

@@ -64,4 +64,9 @@ public Map<Integer, Types.NestedField> map(
     }
     return null;
   }
+
+  @Override
+  public Map<Integer, Types.NestedField> variant(Types.VariantType variant) {
+    return null;
+  }
 }

api/src/main/java/org/apache/iceberg/types/IndexByName.java

@@ -177,7 +177,7 @@ public Map<String, Integer> map(
   }
 
   @Override
-  public Map<String, Integer> variant() {
+  public Map<String, Integer> variant(Types.VariantType variant) {
     return nameToId;
   }
 

api/src/main/java/org/apache/iceberg/types/IndexParents.java

@@ -77,7 +77,7 @@ public Map<Integer, Integer> map(
   }
 
   @Override
-  public Map<Integer, Integer> variant() {
+  public Map<Integer, Integer> variant(Types.VariantType variant) {
     return idToParent;
   }
 

api/src/main/java/org/apache/iceberg/types/PrimitiveHolder.java

@@ -33,6 +33,6 @@ class PrimitiveHolder implements Serializable {
   }
 
   Object readResolve() throws ObjectStreamException {
-    return Types.fromPrimitiveString(typeAsString);
+    return Types.fromTypeName(typeAsString);
   }
 }

api/src/main/java/org/apache/iceberg/types/PruneColumns.java

@@ -159,6 +159,11 @@ public Type map(Types.MapType map, Type ignored, Type valueResult) {
     return null;
   }
 
+  @Override
+  public Type variant(Types.VariantType variant) {
+    return null;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     return null;

api/src/main/java/org/apache/iceberg/types/ReassignDoc.java

@@ -96,6 +96,11 @@ public Type map(Types.MapType map, Supplier<Type> keyTypeFuture, Supplier<Type>
     }
   }
 
+  @Override
+  public Type variant(Types.VariantType variant) {
+    return variant;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     return primitive;

api/src/main/java/org/apache/iceberg/types/ReassignIds.java

@@ -157,6 +157,11 @@ public Type map(Types.MapType map, Supplier<Type> keyTypeFuture, Supplier<Type>
     }
   }
 
+  @Override
+  public Type variant(Types.VariantType variant) {
+    return variant;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     return primitive; // nothing to reassign

api/src/main/java/org/apache/iceberg/types/Type.java

@@ -82,6 +82,10 @@ default Types.MapType asMapType() {
     throw new IllegalArgumentException("Not a map type: " + this);
   }
 
+  default Types.VariantType asVariantType() {
+    throw new IllegalArgumentException("Not a variant type: " + this);
+  }
+
   default boolean isNestedType() {
     return false;
   }
@@ -98,6 +102,10 @@ default boolean isMapType() {
     return false;
   }
 
+  default boolean isVariantType() {
+    return false;
+  }
+
   default NestedType asNestedType() {
     throw new IllegalArgumentException("Not a nested type: " + this);
   }

api/src/main/java/org/apache/iceberg/types/TypeUtil.java

@@ -616,8 +616,16 @@ public T map(Types.MapType map, T keyResult, T valueResult) {
       return null;
     }
 
+    /**
+     * @deprecated will be removed in 2.0.0; use {@link #variant(Types.VariantType)} instead.
+     */
+    @Deprecated
     public T variant() {
-      return null;
+      return variant(Types.VariantType.get());
+    }
+
+    public T variant(Types.VariantType variant) {
+      throw new UnsupportedOperationException("Unsupported type: variant");
     }
 
     public T primitive(Type.PrimitiveType primitive) {
@@ -684,7 +692,7 @@ public static <T> T visit(Type type, SchemaVisitor<T> visitor) {
         return visitor.map(map, keyResult, valueResult);
 
       case VARIANT:
-        return visitor.variant();
+        return visitor.variant(type.asVariantType());
 
       default:
         return visitor.primitive(type.asPrimitiveType());
@@ -712,6 +720,10 @@ public T map(Types.MapType map, Supplier<T> keyResult, Supplier<T> valueResult)
       return null;
     }
 
+    public T variant(Types.VariantType variant) {
+      throw new UnsupportedOperationException("Unsupported type: variant");
+    }
+
     public T primitive(Type.PrimitiveType primitive) {
       return null;
     }
@@ -788,6 +800,9 @@ public static <T> T visit(Type type, CustomOrderSchemaVisitor<T> visitor) {
             new VisitFuture<>(map.keyType(), visitor),
             new VisitFuture<>(map.valueType(), visitor));
 
+      case VARIANT:
+        return visitor.variant(type.asVariantType());
+
       default:
         return visitor.primitive(type.asPrimitiveType());
     }

api/src/main/java/org/apache/iceberg/types/Types.java

@@ -40,8 +40,8 @@ public class Types {
 
   private Types() {}
 
-  private static final ImmutableMap<String, PrimitiveType> TYPES =
-      ImmutableMap.<String, PrimitiveType>builder()
+  private static final ImmutableMap<String, Type> TYPES =
+      ImmutableMap.<String, Type>builder()
           .put(BooleanType.get().toString(), BooleanType.get())
           .put(IntegerType.get().toString(), IntegerType.get())
           .put(LongType.get().toString(), LongType.get())
@@ -57,13 +57,14 @@ private Types() {}
           .put(UUIDType.get().toString(), UUIDType.get())
           .put(BinaryType.get().toString(), BinaryType.get())
           .put(UnknownType.get().toString(), UnknownType.get())
+          .put(VariantType.get().toString(), VariantType.get())
           .buildOrThrow();
 
   private static final Pattern FIXED = Pattern.compile("fixed\\[\\s*(\\d+)\\s*\\]");
   private static final Pattern DECIMAL =
       Pattern.compile("decimal\\(\\s*(\\d+)\\s*,\\s*(\\d+)\\s*\\)");
 
-  public static PrimitiveType fromPrimitiveString(String typeString) {
+  public static Type fromTypeName(String typeString) {
     String lowerTypeString = typeString.toLowerCase(Locale.ROOT);
     if (TYPES.containsKey(lowerTypeString)) {
       return TYPES.get(lowerTypeString);
@@ -82,6 +83,15 @@ public static PrimitiveType fromPrimitiveString(String typeString) {
     throw new IllegalArgumentException("Cannot parse type string to primitive: " + typeString);
   }
 
+  public static PrimitiveType fromPrimitiveString(String typeString) {
+    Type type = fromTypeName(typeString);
+    if (type.isPrimitiveType()) {
+      return type.asPrimitiveType();
+    }
+
+    throw new IllegalArgumentException("Cannot parse type string: variant is not a primitive type");
+  }
+
   public static class BooleanType extends PrimitiveType {
     private static final BooleanType INSTANCE = new BooleanType();
 
@@ -431,6 +441,16 @@ public String toString() {
       return "variant";
     }
 
+    @Override
+    public boolean isVariantType() {
+      return true;
+    }
+
+    @Override
+    public VariantType asVariantType() {
+      return this;
+    }
+
     @Override
     public boolean equals(Object o) {
       if (this == o) {
@@ -692,6 +712,10 @@ private NestedField(
         Literal<?> writeDefault) {
       Preconditions.checkNotNull(name, "Name cannot be null");
       Preconditions.checkNotNull(type, "Type cannot be null");
+      Preconditions.checkArgument(
+          isOptional || !type.equals(UnknownType.get()),
+          "Cannot create required field with unknown type: %s",
+          name);
       this.isOptional = isOptional;
       this.id = id;
       this.name = name;
@@ -706,7 +730,10 @@ private static Literal<?> castDefault(Literal<?> defaultValue, Type type) {
         throw new IllegalArgumentException(
             String.format("Invalid default value for %s: %s (must be null)", type, defaultValue));
       } else if (defaultValue != null) {
-        return defaultValue.to(type);
+        Literal<?> typedDefault = defaultValue.to(type);
+        Preconditions.checkArgument(
+            typedDefault != null, "Cannot cast default value to %s: %s", type, defaultValue);
+        return typedDefault;
       }
 
       return null;

api/src/test/java/org/apache/iceberg/TestPartitionSpecValidation.java

@@ -37,7 +37,7 @@ public class TestPartitionSpecValidation {
           NestedField.required(5, "another_d", Types.TimestampType.withZone()),
           NestedField.required(6, "s", Types.StringType.get()),
           NestedField.required(7, "v", Types.VariantType.get()),
-          NestedField.required(8, "u", Types.UnknownType.get()));
+          NestedField.optional(8, "u", Types.UnknownType.get()));
 
   @Test
   public void testMultipleTimestampPartitions() {