Core: add variant type support

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

@@ -39,8 +39,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())
@@ -56,13 +56,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);
@@ -81,6 +82,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();
 
@@ -430,6 +440,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) {

api/src/test/java/org/apache/iceberg/types/TestReadabilityChecks.java

@@ -22,10 +22,15 @@
 import static org.apache.iceberg.types.Types.NestedField.required;
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.util.Arrays;
 import java.util.List;
+import java.util.stream.Stream;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.types.Type.PrimitiveType;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 
 public class TestReadabilityChecks {
   private static final Type.PrimitiveType[] PRIMITIVES =
@@ -112,6 +117,40 @@ private void testDisallowPrimitiveToStruct(PrimitiveType from, Schema fromSchema
         .contains("cannot be read as a struct");
   }
 
+  @Test
+  public void testVariantToVariant() {
+    Schema fromSchema = new Schema(required(1, "from_field", Types.VariantType.get()));
+    List<String> errors =
+        CheckCompatibility.writeCompatibilityErrors(
+            new Schema(required(1, "to_field", Types.VariantType.get())), fromSchema);
+    assertThat(errors).as("Should produce 0 error messages").isEmpty();
+  }
+
+  private static Stream<Arguments> incompatibleTypesToVariant() {
+    return Stream.of(
+            Stream.of(
+                Arguments.of(Types.StructType.of(required(1, "from", Types.IntegerType.get()))),
+                Arguments.of(
+                    Types.MapType.ofRequired(
+                        1, 2, Types.StringType.get(), Types.IntegerType.get())),
+                Arguments.of(Types.ListType.ofRequired(1, Types.StringType.get()))),
+            Arrays.stream(PRIMITIVES).map(type -> Arguments.of(type)))
+        .flatMap(s -> s);
+  }
+
+  @ParameterizedTest
+  @MethodSource("incompatibleTypesToVariant")
+  public void testIncompatibleTypesToVariant(Type from) {
+    Schema fromSchema = new Schema(required(3, "from_field", from));
+    List<String> errors =
+        CheckCompatibility.writeCompatibilityErrors(
+            new Schema(required(3, "to_field", Types.VariantType.get())), fromSchema);
+    assertThat(errors).hasSize(1);
+    assertThat(errors.get(0))
+        .as("Should complain that other type to variant is not allowed")
+        .contains("cannot be read as a variant");
+  }
+
   @Test
   public void testRequiredSchemaField() {
     Schema write = new Schema(optional(1, "from_field", Types.IntegerType.get()));

api/src/test/java/org/apache/iceberg/types/TestSerializableTypes.java

@@ -46,6 +46,7 @@ public void testIdentityTypes() throws Exception {
           Types.StringType.get(),
           Types.UUIDType.get(),
           Types.BinaryType.get(),
+          Types.UnknownType.get()
         };
 
     for (Type type : identityPrimitives) {
@@ -136,15 +137,6 @@ public void testVariant() throws Exception {
         .isEqualTo(variant);
   }
 
-  @Test
-  public void testUnknown() throws Exception {
-    Types.UnknownType unknown = Types.UnknownType.get();
-    Type copy = TestHelpers.roundTripSerialize(unknown);
-    assertThat(copy)
-        .as("Unknown serialization should be equal to starting type")
-        .isEqualTo(unknown);
-  }
-
   @Test
   public void testSchema() throws Exception {
     Schema schema =