diff --git a/api/src/main/java/org/apache/iceberg/Accessors.java b/api/src/main/java/org/apache/iceberg/Accessors.java
index 08233624f244..0b36730fbb4b 100644
--- a/api/src/main/java/org/apache/iceberg/Accessors.java
+++ b/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) {
diff --git a/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java b/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java
index e58f76a8de56..75055cddc197 100644
--- a/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java
+++ b/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;
diff --git a/api/src/main/java/org/apache/iceberg/types/AssignIds.java b/api/src/main/java/org/apache/iceberg/types/AssignIds.java
index 68588f581adc..b2f72751eb89 100644
--- a/api/src/main/java/org/apache/iceberg/types/AssignIds.java
+++ b/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;
diff --git a/api/src/main/java/org/apache/iceberg/types/CheckCompatibility.java b/api/src/main/java/org/apache/iceberg/types/CheckCompatibility.java
index 502e52c345e5..725f7f42562e 100644
--- a/api/src/main/java/org/apache/iceberg/types/CheckCompatibility.java
+++ b/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)) {
diff --git a/api/src/main/java/org/apache/iceberg/types/FindTypeVisitor.java b/api/src/main/java/org/apache/iceberg/types/FindTypeVisitor.java
index f0750f337e2e..64faebb48243 100644
--- a/api/src/main/java/org/apache/iceberg/types/FindTypeVisitor.java
+++ b/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;
diff --git a/api/src/main/java/org/apache/iceberg/types/GetProjectedIds.java b/api/src/main/java/org/apache/iceberg/types/GetProjectedIds.java
index a8a7de065ece..1ec70b8578bc 100644
--- a/api/src/main/java/org/apache/iceberg/types/GetProjectedIds.java
+++ b/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;
+  }
 }
diff --git a/api/src/main/java/org/apache/iceberg/types/IndexById.java b/api/src/main/java/org/apache/iceberg/types/IndexById.java
index 40280c5ed9dd..a7b96eb381f7 100644
--- a/api/src/main/java/org/apache/iceberg/types/IndexById.java
+++ b/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;
+  }
 }
diff --git a/api/src/main/java/org/apache/iceberg/types/IndexByName.java b/api/src/main/java/org/apache/iceberg/types/IndexByName.java
index 131434c9a156..60258f5c5c3e 100644
--- a/api/src/main/java/org/apache/iceberg/types/IndexByName.java
+++ b/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;
   }
 
diff --git a/api/src/main/java/org/apache/iceberg/types/IndexParents.java b/api/src/main/java/org/apache/iceberg/types/IndexParents.java
index 952447ed2799..6e611d47e912 100644
--- a/api/src/main/java/org/apache/iceberg/types/IndexParents.java
+++ b/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;
   }
 
diff --git a/api/src/main/java/org/apache/iceberg/types/PrimitiveHolder.java b/api/src/main/java/org/apache/iceberg/types/PrimitiveHolder.java
index 42f0da38167d..928a65878d3a 100644
--- a/api/src/main/java/org/apache/iceberg/types/PrimitiveHolder.java
+++ b/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);
   }
 }
diff --git a/api/src/main/java/org/apache/iceberg/types/PruneColumns.java b/api/src/main/java/org/apache/iceberg/types/PruneColumns.java
index daf2e6bbc0ca..56f01cf34bb5 100644
--- a/api/src/main/java/org/apache/iceberg/types/PruneColumns.java
+++ b/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;
diff --git a/api/src/main/java/org/apache/iceberg/types/ReassignDoc.java b/api/src/main/java/org/apache/iceberg/types/ReassignDoc.java
index 9ce04a7bd103..328d81c42885 100644
--- a/api/src/main/java/org/apache/iceberg/types/ReassignDoc.java
+++ b/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;
diff --git a/api/src/main/java/org/apache/iceberg/types/ReassignIds.java b/api/src/main/java/org/apache/iceberg/types/ReassignIds.java
index 565ceee2a901..3d114f093f6b 100644
--- a/api/src/main/java/org/apache/iceberg/types/ReassignIds.java
+++ b/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
diff --git a/api/src/main/java/org/apache/iceberg/types/Type.java b/api/src/main/java/org/apache/iceberg/types/Type.java
index f4c6f22134a5..67e40df9e939 100644
--- a/api/src/main/java/org/apache/iceberg/types/Type.java
+++ b/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);
   }
diff --git a/api/src/main/java/org/apache/iceberg/types/TypeUtil.java b/api/src/main/java/org/apache/iceberg/types/TypeUtil.java
index 39f2898757a6..4892696ab450 100644
--- a/api/src/main/java/org/apache/iceberg/types/TypeUtil.java
+++ b/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());
     }
diff --git a/api/src/main/java/org/apache/iceberg/types/Types.java b/api/src/main/java/org/apache/iceberg/types/Types.java
index 6882f718508b..c1935d6980e9 100644
--- a/api/src/main/java/org/apache/iceberg/types/Types.java
+++ b/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) {
diff --git a/api/src/test/java/org/apache/iceberg/types/TestReadabilityChecks.java b/api/src/test/java/org/apache/iceberg/types/TestReadabilityChecks.java
index 2d02da5346a7..debb9c9dc1d6 100644
--- a/api/src/test/java/org/apache/iceberg/types/TestReadabilityChecks.java
+++ b/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()));
diff --git a/api/src/test/java/org/apache/iceberg/types/TestSerializableTypes.java b/api/src/test/java/org/apache/iceberg/types/TestSerializableTypes.java
index a222e8e66b8e..790f59587c59 100644
--- a/api/src/test/java/org/apache/iceberg/types/TestSerializableTypes.java
+++ b/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 =
diff --git a/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java b/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
index 36384d232af3..b6556d70bd85 100644
--- a/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
+++ b/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
@@ -23,12 +23,18 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
+import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.Types.IntegerType;
 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 TestTypeUtil {
   @Test
@@ -645,4 +651,95 @@ public void testReassignOrRefreshIdsCaseInsensitive() {
                 required(2, "FIELD2", Types.IntegerType.get())));
     assertThat(actualSchema.asStruct()).isEqualTo(expectedSchema.asStruct());
   }
+
+  private static Stream<Arguments> testTypes() {
+    return Stream.of(
+        Arguments.of(Types.UnknownType.get()),
+        Arguments.of(Types.VariantType.get()),
+        Arguments.of(Types.TimestampNanoType.withoutZone()),
+        Arguments.of(Types.TimestampNanoType.withZone()));
+  }
+
+  @ParameterizedTest
+  @MethodSource("testTypes")
+  public void testAssignIdsWithType(Type testType) {
+    Types.StructType sourceType =
+        Types.StructType.of(required(0, "id", IntegerType.get()), required(1, "data", testType));
+    Type expectedType =
+        Types.StructType.of(required(10, "id", IntegerType.get()), required(11, "data", testType));
+
+    Type assignedType = TypeUtil.assignIds(sourceType, oldId -> oldId + 10);
+    assertThat(assignedType).isEqualTo(expectedType);
+  }
+
+  @ParameterizedTest
+  @MethodSource("testTypes")
+  public void testAssignFreshIdsWithType(Type testType) {
+    Schema schema = new Schema(required(0, "id", IntegerType.get()), required(1, "data", testType));
+
+    Schema assignedSchema = TypeUtil.assignFreshIds(schema, new AtomicInteger(10)::incrementAndGet);
+    Schema expectedSchema =
+        new Schema(required(11, "id", IntegerType.get()), required(12, "data", testType));
+    assertThat(assignedSchema.asStruct()).isEqualTo(expectedSchema.asStruct());
+  }
+
+  @ParameterizedTest
+  @MethodSource("testTypes")
+  public void testReassignIdsWithType(Type testType) {
+    Schema schema = new Schema(required(0, "id", IntegerType.get()), required(1, "data", testType));
+    Schema sourceSchema =
+        new Schema(required(1, "id", IntegerType.get()), required(2, "data", testType));
+
+    Schema reassignedSchema = TypeUtil.reassignIds(schema, sourceSchema);
+    assertThat(reassignedSchema.asStruct()).isEqualTo(sourceSchema.asStruct());
+  }
+
+  @ParameterizedTest
+  @MethodSource("testTypes")
+  public void testIndexByIdWithType(Type testType) {
+    Schema schema = new Schema(required(0, "id", IntegerType.get()), required(1, "data", testType));
+
+    Map<Integer, Types.NestedField> indexByIds = TypeUtil.indexById(schema.asStruct());
+    assertThat(indexByIds.get(1).type()).isEqualTo(testType);
+  }
+
+  @ParameterizedTest
+  @MethodSource("testTypes")
+  public void testIndexNameByIdWithType(Type testType) {
+    Schema schema = new Schema(required(0, "id", IntegerType.get()), required(1, "data", testType));
+
+    Map<Integer, String> indexNameByIds = TypeUtil.indexNameById(schema.asStruct());
+    assertThat(indexNameByIds.get(1)).isEqualTo("data");
+  }
+
+  @ParameterizedTest
+  @MethodSource("testTypes")
+  public void testProjectWithType(Type testType) {
+    Schema schema = new Schema(required(0, "id", IntegerType.get()), required(1, "data", testType));
+
+    Schema expectedSchema = new Schema(required(1, "data", testType));
+    Schema projectedSchema = TypeUtil.project(schema, Sets.newHashSet(1));
+    assertThat(projectedSchema.asStruct()).isEqualTo(expectedSchema.asStruct());
+  }
+
+  @ParameterizedTest
+  @MethodSource("testTypes")
+  public void testGetProjectedIdsWithType(Type testType) {
+    Schema schema = new Schema(required(0, "id", IntegerType.get()), required(1, "data", testType));
+
+    Set<Integer> projectedIds = TypeUtil.getProjectedIds(schema);
+    assertThat(Set.of(0, 1)).isEqualTo(projectedIds);
+  }
+
+  @ParameterizedTest
+  @MethodSource("testTypes")
+  public void testReassignDocWithType(Type testType) {
+    Schema schema = new Schema(required(0, "id", IntegerType.get()), required(1, "data", testType));
+    Schema docSourceSchema =
+        new Schema(
+            required(0, "id", IntegerType.get(), "id"), required(1, "data", testType, "data"));
+
+    Schema reassignedSchema = TypeUtil.reassignDoc(schema, docSourceSchema);
+    assertThat(reassignedSchema.asStruct()).isEqualTo(docSourceSchema.asStruct());
+  }
 }
diff --git a/api/src/test/java/org/apache/iceberg/types/TestTypes.java b/api/src/test/java/org/apache/iceberg/types/TestTypes.java
index cbc37291375f..b3381d1ff440 100644
--- a/api/src/test/java/org/apache/iceberg/types/TestTypes.java
+++ b/api/src/test/java/org/apache/iceberg/types/TestTypes.java
@@ -25,6 +25,30 @@
 
 public class TestTypes {
 
+  @Test
+  public void fromTypeName() {
+    assertThat(Types.fromTypeName("boolean")).isSameAs(Types.BooleanType.get());
+    assertThat(Types.fromTypeName("BooLean")).isSameAs(Types.BooleanType.get());
+
+    assertThat(Types.fromTypeName("timestamp")).isSameAs(Types.TimestampType.withoutZone());
+    assertThat(Types.fromTypeName("timestamptz")).isSameAs(Types.TimestampType.withZone());
+    assertThat(Types.fromTypeName("timestamp_ns")).isSameAs(Types.TimestampNanoType.withoutZone());
+    assertThat(Types.fromTypeName("timestamptz_ns")).isSameAs(Types.TimestampNanoType.withZone());
+
+    assertThat(Types.fromTypeName("Fixed[ 3 ]")).isEqualTo(Types.FixedType.ofLength(3));
+
+    assertThat(Types.fromTypeName("Decimal( 2 , 3 )")).isEqualTo(Types.DecimalType.of(2, 3));
+
+    assertThat(Types.fromTypeName("Decimal(2,3)")).isEqualTo(Types.DecimalType.of(2, 3));
+
+    assertThat(Types.fromTypeName("variant")).isSameAs(Types.VariantType.get());
+    assertThat(Types.fromTypeName("Variant")).isSameAs(Types.VariantType.get());
+
+    assertThatExceptionOfType(IllegalArgumentException.class)
+        .isThrownBy(() -> Types.fromTypeName("abcdefghij"))
+        .withMessage("Cannot parse type string to primitive: abcdefghij");
+  }
+
   @Test
   public void fromPrimitiveString() {
     assertThat(Types.fromPrimitiveString("boolean")).isSameAs(Types.BooleanType.get());
@@ -43,6 +67,13 @@ public void fromPrimitiveString() {
 
     assertThat(Types.fromPrimitiveString("Decimal(2,3)")).isEqualTo(Types.DecimalType.of(2, 3));
 
+    assertThatExceptionOfType(IllegalArgumentException.class)
+        .isThrownBy(() -> Types.fromPrimitiveString("variant"))
+        .withMessage("Cannot parse type string: variant is not a primitive type");
+    assertThatExceptionOfType(IllegalArgumentException.class)
+        .isThrownBy(() -> Types.fromPrimitiveString("Variant"))
+        .withMessage("Cannot parse type string: variant is not a primitive type");
+
     assertThatExceptionOfType(IllegalArgumentException.class)
         .isThrownBy(() -> Types.fromPrimitiveString("abcdefghij"))
         .withMessage("Cannot parse type string to primitive: abcdefghij");
diff --git a/core/src/main/java/org/apache/iceberg/SchemaParser.java b/core/src/main/java/org/apache/iceberg/SchemaParser.java
index 27e6ed048712..04655ce3f7d7 100644
--- a/core/src/main/java/org/apache/iceberg/SchemaParser.java
+++ b/core/src/main/java/org/apache/iceberg/SchemaParser.java
@@ -143,8 +143,8 @@ static void toJson(Type.PrimitiveType primitive, JsonGenerator generator) throws
   }
 
   static void toJson(Type type, JsonGenerator generator) throws IOException {
-    if (type.isPrimitiveType()) {
-      toJson(type.asPrimitiveType(), generator);
+    if (type.isPrimitiveType() || type.isVariantType()) {
+      generator.writeString(type.toString());
     } else {
       Type.NestedType nested = type.asNestedType();
       switch (type.typeId()) {
@@ -179,7 +179,7 @@ public static String toJson(Schema schema, boolean pretty) {
 
   private static Type typeFromJson(JsonNode json) {
     if (json.isTextual()) {
-      return Types.fromPrimitiveString(json.asText());
+      return Types.fromTypeName(json.asText());
     } else if (json.isObject()) {
       JsonNode typeObj = json.get(TYPE);
       if (typeObj != null) {
diff --git a/core/src/main/java/org/apache/iceberg/SchemaUpdate.java b/core/src/main/java/org/apache/iceberg/SchemaUpdate.java
index 2b541080ac72..7726c3a785d0 100644
--- a/core/src/main/java/org/apache/iceberg/SchemaUpdate.java
+++ b/core/src/main/java/org/apache/iceberg/SchemaUpdate.java
@@ -722,6 +722,11 @@ public Type map(Types.MapType map, Type kResult, Type valueResult) {
       }
     }
 
+    @Override
+    public Type variant(Types.VariantType variant) {
+      return variant;
+    }
+
     @Override
     public Type primitive(Type.PrimitiveType primitive) {
       return primitive;
diff --git a/core/src/main/java/org/apache/iceberg/mapping/MappingUtil.java b/core/src/main/java/org/apache/iceberg/mapping/MappingUtil.java
index de6ce2ad0425..fc3d920a4069 100644
--- a/core/src/main/java/org/apache/iceberg/mapping/MappingUtil.java
+++ b/core/src/main/java/org/apache/iceberg/mapping/MappingUtil.java
@@ -302,6 +302,11 @@ public MappedFields map(Types.MapType map, MappedFields keyResult, MappedFields
           MappedField.of(map.valueId(), "value", valueResult));
     }
 
+    @Override
+    public MappedFields variant(Types.VariantType variant) {
+      return null; // no mapping because variant has no nested fields with IDs
+    }
+
     @Override
     public MappedFields primitive(Type.PrimitiveType primitive) {
       return null; // no mapping because primitives have no nested fields
diff --git a/core/src/main/java/org/apache/iceberg/schema/SchemaWithPartnerVisitor.java b/core/src/main/java/org/apache/iceberg/schema/SchemaWithPartnerVisitor.java
index 9b2226f5714d..694bfb2f6242 100644
--- a/core/src/main/java/org/apache/iceberg/schema/SchemaWithPartnerVisitor.java
+++ b/core/src/main/java/org/apache/iceberg/schema/SchemaWithPartnerVisitor.java
@@ -107,6 +107,9 @@ public static <P, T> T visit(
 
         return visitor.map(map, partner, keyResult, valueResult);
 
+      case VARIANT:
+        return visitor.variant(type.asVariantType(), partner);
+
       default:
         return visitor.primitive(type.asPrimitiveType(), partner);
     }
@@ -160,6 +163,10 @@ public R map(Types.MapType map, P partner, R keyResult, R valueResult) {
     return null;
   }
 
+  public R variant(Types.VariantType variant, P partner) {
+    throw new UnsupportedOperationException("Unsupported type: variant");
+  }
+
   public R primitive(Type.PrimitiveType primitive, P partner) {
     return null;
   }
diff --git a/core/src/main/java/org/apache/iceberg/schema/UnionByNameVisitor.java b/core/src/main/java/org/apache/iceberg/schema/UnionByNameVisitor.java
index 68172b7062a6..7c4dac9feff1 100644
--- a/core/src/main/java/org/apache/iceberg/schema/UnionByNameVisitor.java
+++ b/core/src/main/java/org/apache/iceberg/schema/UnionByNameVisitor.java
@@ -142,6 +142,11 @@ public Boolean map(
     return false;
   }
 
+  @Override
+  public Boolean variant(Types.VariantType variant, Integer partnerId) {
+    return partnerId == null;
+  }
+
   @Override
   public Boolean primitive(Type.PrimitiveType primitive, Integer partnerId) {
     return partnerId == null;
diff --git a/core/src/main/java/org/apache/iceberg/types/FixupTypes.java b/core/src/main/java/org/apache/iceberg/types/FixupTypes.java
index 23fccddda3d9..1e4c0b597a6a 100644
--- a/core/src/main/java/org/apache/iceberg/types/FixupTypes.java
+++ b/core/src/main/java/org/apache/iceberg/types/FixupTypes.java
@@ -147,6 +147,12 @@ public Type map(Types.MapType map, Supplier<Type> keyTypeFuture, Supplier<Type>
     }
   }
 
+  @Override
+  public Type variant(Types.VariantType variant) {
+    // nothing to fix up
+    return variant;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     if (sourceType.equals(primitive)) {
diff --git a/core/src/test/java/org/apache/iceberg/TestSchemaParser.java b/core/src/test/java/org/apache/iceberg/TestSchemaParser.java
index ebd197a68af0..40db5cfee2cb 100644
--- a/core/src/test/java/org/apache/iceberg/TestSchemaParser.java
+++ b/core/src/test/java/org/apache/iceberg/TestSchemaParser.java
@@ -123,4 +123,14 @@ public void testPrimitiveTypeDefaultValues(Type.PrimitiveType type, Object defau
     assertThat(serialized.findField("col_with_default").initialDefault()).isEqualTo(defaultValue);
     assertThat(serialized.findField("col_with_default").writeDefault()).isEqualTo(defaultValue);
   }
+
+  @Test
+  public void testVariantType() throws IOException {
+    Schema schema =
+        new Schema(
+            Types.NestedField.required(1, "id", Types.IntegerType.get()),
+            Types.NestedField.optional(2, "data", Types.VariantType.get()));
+
+    writeAndValidate(schema);
+  }
 }
diff --git a/core/src/test/java/org/apache/iceberg/TestSchemaUnionByFieldName.java b/core/src/test/java/org/apache/iceberg/TestSchemaUnionByFieldName.java
index 656e72a0c19c..8649ada99bef 100644
--- a/core/src/test/java/org/apache/iceberg/TestSchemaUnionByFieldName.java
+++ b/core/src/test/java/org/apache/iceberg/TestSchemaUnionByFieldName.java
@@ -26,7 +26,7 @@
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.types.Type.PrimitiveType;
+import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.BinaryType;
 import org.apache.iceberg.types.Types.BooleanType;
@@ -42,13 +42,16 @@
 import org.apache.iceberg.types.Types.StringType;
 import org.apache.iceberg.types.Types.StructType;
 import org.apache.iceberg.types.Types.TimeType;
+import org.apache.iceberg.types.Types.TimestampNanoType;
 import org.apache.iceberg.types.Types.TimestampType;
 import org.apache.iceberg.types.Types.UUIDType;
+import org.apache.iceberg.types.Types.UnknownType;
+import org.apache.iceberg.types.Types.VariantType;
 import org.junit.jupiter.api.Test;
 
 public class TestSchemaUnionByFieldName {
 
-  private static List<? extends PrimitiveType> primitiveTypes() {
+  private static List<? extends Type> primitiveTypes() {
     return Lists.newArrayList(
         StringType.get(),
         TimeType.get(),
@@ -63,11 +66,15 @@ private static List<? extends PrimitiveType> primitiveTypes() {
         FixedType.ofLength(10),
         DecimalType.of(10, 2),
         LongType.get(),
-        FloatType.get());
+        FloatType.get(),
+        VariantType.get(),
+        UnknownType.get(),
+        TimestampNanoType.withoutZone(),
+        TimestampNanoType.withZone());
   }
 
   private static NestedField[] primitiveFields(
-      Integer initialValue, List<? extends PrimitiveType> primitiveTypes) {
+      Integer initialValue, List<? extends Type> primitiveTypes) {
     AtomicInteger atomicInteger = new AtomicInteger(initialValue);
     return primitiveTypes.stream()
         .map(
@@ -75,7 +82,7 @@ private static NestedField[] primitiveFields(
                 optional(
                     atomicInteger.incrementAndGet(),
                     type.toString(),
-                    Types.fromPrimitiveString(type.toString())))
+                    Types.fromTypeName(type.toString())))
         .toArray(NestedField[]::new);
   }
 
@@ -88,7 +95,7 @@ public void testAddTopLevelPrimitives() {
 
   @Test
   public void testAddTopLevelListOfPrimitives() {
-    for (PrimitiveType primitiveType : primitiveTypes()) {
+    for (Type primitiveType : primitiveTypes()) {
       Schema newSchema =
           new Schema(optional(1, "aList", Types.ListType.ofOptional(2, primitiveType)));
       Schema applied = new SchemaUpdate(new Schema(), 0).unionByNameWith(newSchema).apply();
@@ -98,7 +105,7 @@ public void testAddTopLevelListOfPrimitives() {
 
   @Test
   public void testAddTopLevelMapOfPrimitives() {
-    for (PrimitiveType primitiveType : primitiveTypes()) {
+    for (Type primitiveType : primitiveTypes()) {
       Schema newSchema =
           new Schema(
               optional(1, "aMap", Types.MapType.ofOptional(2, 3, primitiveType, primitiveType)));
@@ -109,7 +116,7 @@ public void testAddTopLevelMapOfPrimitives() {
 
   @Test
   public void testAddTopLevelStructOfPrimitives() {
-    for (PrimitiveType primitiveType : primitiveTypes()) {
+    for (Type primitiveType : primitiveTypes()) {
       Schema currentSchema =
           new Schema(
               optional(1, "aStruct", Types.StructType.of(optional(2, "primitive", primitiveType))));
@@ -120,7 +127,7 @@ public void testAddTopLevelStructOfPrimitives() {
 
   @Test
   public void testAddNestedPrimitive() {
-    for (PrimitiveType primitiveType : primitiveTypes()) {
+    for (Type primitiveType : primitiveTypes()) {
       Schema currentSchema = new Schema(optional(1, "aStruct", Types.StructType.of()));
       Schema newSchema =
           new Schema(
diff --git a/core/src/test/java/org/apache/iceberg/mapping/TestNameMapping.java b/core/src/test/java/org/apache/iceberg/mapping/TestNameMapping.java
index d30a93d50d49..3ebf4d9242ab 100644
--- a/core/src/test/java/org/apache/iceberg/mapping/TestNameMapping.java
+++ b/core/src/test/java/org/apache/iceberg/mapping/TestNameMapping.java
@@ -289,4 +289,16 @@ public void testMappingFindByName() {
                 "location",
                 MappedFields.of(MappedField.of(11, "latitude"), MappedField.of(12, "longitude"))));
   }
+
+  @Test
+  public void testMappingVariantType() {
+    Schema schema =
+        new Schema(
+            required(1, "id", Types.LongType.get()), required(2, "data", Types.VariantType.get()));
+
+    MappedFields expected = MappedFields.of(MappedField.of(1, "id"), MappedField.of(2, "data"));
+
+    NameMapping mapping = MappingUtil.create(schema);
+    assertThat(mapping.asMappedFields()).isEqualTo(expected);
+  }
 }
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java
index af0fa84f67a1..ad8a4beb55d0 100644
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java
+++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java
@@ -565,6 +565,11 @@ public String map(Types.MapType map, String keyResult, String valueResult) {
       return "map<" + keyResult + ", " + valueResult + ">";
     }
 
+    @Override
+    public String variant(Types.VariantType variant) {
+      return "variant";
+    }
+
     @Override
     public String primitive(Type.PrimitiveType primitive) {
       switch (primitive.typeId()) {
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSpark3Util.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSpark3Util.java
index 6f900ffebb10..e4e66abfefa0 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSpark3Util.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSpark3Util.java
@@ -105,12 +105,13 @@ public void testDescribeSchema() {
                 3,
                 "pairs",
                 Types.MapType.ofOptional(4, 5, Types.StringType.get(), Types.LongType.get())),
-            required(6, "time", Types.TimestampType.withoutZone()));
+            required(6, "time", Types.TimestampType.withoutZone()),
+            required(7, "v", Types.VariantType.get()));
 
     assertThat(Spark3Util.describe(schema))
         .as("Schema description isn't correct.")
         .isEqualTo(
-            "struct<data: list<string> not null,pairs: map<string, bigint>,time: timestamp not null>");
+            "struct<data: list<string> not null,pairs: map<string, bigint>,time: timestamp not null,v: variant not null>");
   }
 
   @Test