Repository: apache/iceberg (language: python)

Issue:
Proposal: add Variant type to iceberg
### Feature Request / Improvement

As discussed in the mailing list, and [described in this doc](https://docs.google.com/document/d/1QjhpG_SVNPZh3anFcpicMQx90ebwjL7rmzFYfUP89Iw/edit), I'd like to add Variant type to Iceberg.

We will adopt the Spark Variant spec and move the spec to Parquet - see [discussion](https://lists.apache.org/thread/0k5oj3mn0049fcxoxm3gx3d7r28gw4rj). There are some ongoing discussion on how Parquet will adopt the spec and implementation - seed the [doc]( https://docs.google.com/document/d/1guEzBQjzOEEZvvibeZjNraKmZHWtxQR95O_DvtZU0xw/edit), but from Iceberg side, I would like to start the implementation assuming the spec is available and make the final changes when the spec is finalized.


### Query engine

None

### Willingness to contribute

- [X] I can contribute this improvement/feature independently
- [ ] I would be willing to contribute this improvement/feature with guidance from the Iceberg community
- [ ] I cannot contribute this improvement/feature at this time


Discussion / hints:
Since variant type is a major feature, this issue might be better labeled as `proposal` following https://iceberg.apache.org/contribute/#what-is-an-improvement-proposal

Fix patch:
diff --git a/api/src/main/java/org/apache/iceberg/Schema.java b/api/src/main/java/org/apache/iceberg/Schema.java
index a94e8771875a..bd07e9798e9b 100644
--- a/api/src/main/java/org/apache/iceberg/Schema.java
+++ b/api/src/main/java/org/apache/iceberg/Schema.java
@@ -60,7 +60,7 @@ public class Schema implements Serializable {
 
   @VisibleForTesting
   static final Map<Type.TypeID, Integer> MIN_FORMAT_VERSIONS =
-      ImmutableMap.of(Type.TypeID.TIMESTAMP_NANO, 3);
+      ImmutableMap.of(Type.TypeID.TIMESTAMP_NANO, 3, Type.TypeID.VARIANT, 3);
 
   private final StructType struct;
   private final int schemaId;
diff --git a/api/src/main/java/org/apache/iceberg/expressions/ExpressionUtil.java b/api/src/main/java/org/apache/iceberg/expressions/ExpressionUtil.java
index 82d513ced7dd..9be144b2ce98 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/ExpressionUtil.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/ExpressionUtil.java
@@ -534,7 +534,8 @@ private static String sanitize(Type type, Object value, long now, int today) {
       case DECIMAL:
       case FIXED:
       case BINARY:
-        // for boolean, uuid, decimal, fixed, and binary, match the string result
+      case VARIANT:
+        // for boolean, uuid, decimal, fixed, variant, and binary, match the string result
         return sanitizeSimpleString(value.toString());
     }
     throw new UnsupportedOperationException(
@@ -562,7 +563,7 @@ private static String sanitize(Literal<?> literal, long now, int today) {
     } else if (literal instanceof Literals.DoubleLiteral) {
       return sanitizeNumber(((Literals.DoubleLiteral) literal).value(), "float");
     } else {
-      // for uuid, decimal, fixed, and binary, match the string result
+      // for uuid, decimal, fixed, variant, and binary, match the string result
       return sanitizeSimpleString(literal.value().toString());
     }
   }
diff --git a/api/src/main/java/org/apache/iceberg/transforms/Identity.java b/api/src/main/java/org/apache/iceberg/transforms/Identity.java
index d4e5e532943b..04f0c25e9222 100644
--- a/api/src/main/java/org/apache/iceberg/transforms/Identity.java
+++ b/api/src/main/java/org/apache/iceberg/transforms/Identity.java
@@ -38,6 +38,9 @@ class Identity<T> implements Transform<T, T> {
    */
   @Deprecated
   public static <I> Identity<I> get(Type type) {
+    Preconditions.checkArgument(
+        type.typeId() != Type.TypeID.VARIANT, "Unsupported type for identity: %s", type);
+
     return new Identity<>(type);
   }
 
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 d1dc4adc214a..f0750f337e2e 100644
--- a/api/src/main/java/org/apache/iceberg/types/FindTypeVisitor.java
+++ b/api/src/main/java/org/apache/iceberg/types/FindTypeVisitor.java
@@ -76,6 +76,15 @@ public Type map(Types.MapType map, Type keyResult, Type valueResult) {
     return valueResult;
   }
 
+  @Override
+  public Type variant() {
+    if (predicate.test(Types.VariantType.get())) {
+      return Types.VariantType.get();
+    }
+
+    return null;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     if (predicate.test(primitive)) {
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 9183ea85f467..131434c9a156 100644
--- a/api/src/main/java/org/apache/iceberg/types/IndexByName.java
+++ b/api/src/main/java/org/apache/iceberg/types/IndexByName.java
@@ -176,6 +176,11 @@ public Map<String, Integer> map(
     return nameToId;
   }
 
+  @Override
+  public Map<String, Integer> variant() {
+    return nameToId;
+  }
+
   @Override
   public Map<String, Integer> primitive(Type.PrimitiveType primitive) {
     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 bcd1e1ee900c..952447ed2799 100644
--- a/api/src/main/java/org/apache/iceberg/types/IndexParents.java
+++ b/api/src/main/java/org/apache/iceberg/types/IndexParents.java
@@ -76,6 +76,11 @@ public Map<Integer, Integer> map(
     return idToParent;
   }
 
+  @Override
+  public Map<Integer, Integer> variant() {
+    return idToParent;
+  }
+
   @Override
   public Map<Integer, Integer> primitive(Type.PrimitiveType primitive) {
     return idToParent;
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 571bf9a14e43..30870535521f 100644
--- a/api/src/main/java/org/apache/iceberg/types/Type.java
+++ b/api/src/main/java/org/apache/iceberg/types/Type.java
@@ -45,7 +45,8 @@ enum TypeID {
     DECIMAL(BigDecimal.class),
     STRUCT(StructLike.class),
     LIST(List.class),
-    MAP(Map.class);
+    MAP(Map.class),
+    VARIANT(Object.class);
 
     private final Class<?> javaClass;
 
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 8a9184569aec..7fcf3db3a40d 100644
--- a/api/src/main/java/org/apache/iceberg/types/TypeUtil.java
+++ b/api/src/main/java/org/apache/iceberg/types/TypeUtil.java
@@ -534,6 +534,7 @@ private static int estimateSize(Type type) {
       case FIXED:
         return ((Types.FixedType) type).length();
       case BINARY:
+      case VARIANT:
         return 80;
       case DECIMAL:
         // 12 (header) + (12 + 12 + 4) (BigInteger) + 4 (scale) = 44 bytes
@@ -612,6 +613,10 @@ public T map(Types.MapType map, T keyResult, T valueResult) {
       return null;
     }
 
+    public T variant() {
+      return null;
+    }
+
     public T primitive(Type.PrimitiveType primitive) {
       return null;
     }
@@ -675,6 +680,9 @@ public static <T> T visit(Type type, SchemaVisitor<T> visitor) {
 
         return visitor.map(map, keyResult, valueResult);
 
+      case VARIANT:
+        return visitor.variant();
+
       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 4bb1674f3be5..927b3a5065ad 100644
--- a/api/src/main/java/org/apache/iceberg/types/Types.java
+++ b/api/src/main/java/org/apache/iceberg/types/Types.java
@@ -412,6 +412,41 @@ public String toString() {
     }
   }
 
+  public static class VariantType implements Type {
+    private static final VariantType INSTANCE = new VariantType();
+
+    public static VariantType get() {
+      return INSTANCE;
+    }
+
+    @Override
+    public TypeID typeId() {
+      return TypeID.VARIANT;
+    }
+
+    @Override
+    public String toString() {
+      return "variant";
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      } else if (!(o instanceof VariantType)) {
+        return false;
+      }
+
+      VariantType that = (VariantType) o;
+      return typeId() == that.typeId();
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(VariantType.class, typeId());
+    }
+  }
+
   public static class DecimalType extends PrimitiveType {
     public static DecimalType of(int precision, int scale) {
       return new DecimalType(precision, scale);
diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java
index 3cdc53995dce..0e323bca1c97 100644
--- a/core/src/main/java/org/apache/iceberg/TableMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java
@@ -295,8 +295,9 @@ public String toString() {
         sortOrders != null && !sortOrders.isEmpty(), "Sort orders cannot be null or empty");
     Preconditions.checkArgument(
         formatVersion <= SUPPORTED_TABLE_FORMAT_VERSION,
-        "Unsupported format version: v%s",
-        formatVersion);
+        "Unsupported format version: v%s (supported: v%s)",
+        formatVersion,
+        SUPPORTED_TABLE_FORMAT_VERSION);
     Preconditions.checkArgument(
         formatVersion == 1 || uuid != null, "UUID is required in format v%s", formatVersion);
     Preconditions.checkArgument(


Test patch:
diff --git a/api/src/test/java/org/apache/iceberg/TestPartitionSpecValidation.java b/api/src/test/java/org/apache/iceberg/TestPartitionSpecValidation.java
index 263db427aa16..971f5a9e4510 100644
--- a/api/src/test/java/org/apache/iceberg/TestPartitionSpecValidation.java
+++ b/api/src/test/java/org/apache/iceberg/TestPartitionSpecValidation.java
@@ -21,6 +21,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
+import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.transforms.Transforms;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.NestedField;
@@ -34,7 +35,8 @@ public class TestPartitionSpecValidation {
           NestedField.required(3, "another_ts", Types.TimestampType.withZone()),
           NestedField.required(4, "d", Types.TimestampType.withZone()),
           NestedField.required(5, "another_d", Types.TimestampType.withZone()),
-          NestedField.required(6, "s", Types.StringType.get()));
+          NestedField.required(6, "s", Types.StringType.get()),
+          NestedField.required(7, "v", Types.VariantType.get()));
 
   @Test
   public void testMultipleTimestampPartitions() {
@@ -312,4 +314,15 @@ public void testAddPartitionFieldsWithAndWithoutFieldIds() {
     assertThat(spec.fields().get(2).fieldId()).isEqualTo(1006);
     assertThat(spec.lastAssignedFieldId()).isEqualTo(1006);
   }
+
+  @Test
+  public void testVariantUnsupported() {
+    assertThatThrownBy(
+            () ->
+                PartitionSpec.builderFor(SCHEMA)
+                    .add(7, 1005, "variant_partition1", Transforms.bucket(5))
+                    .build())
+        .isInstanceOf(ValidationException.class)
+        .hasMessage("Cannot partition by non-primitive source field: variant");
+  }
 }
diff --git a/api/src/test/java/org/apache/iceberg/TestSchema.java b/api/src/test/java/org/apache/iceberg/TestSchema.java
index e79adbd09fb7..e9cb387eebb5 100644
--- a/api/src/test/java/org/apache/iceberg/TestSchema.java
+++ b/api/src/test/java/org/apache/iceberg/TestSchema.java
@@ -38,7 +38,10 @@
 public class TestSchema {
 
   private static final List<Type> TEST_TYPES =
-      ImmutableList.of(Types.TimestampNanoType.withoutZone(), Types.TimestampNanoType.withZone());
+      ImmutableList.of(
+          Types.TimestampNanoType.withoutZone(),
+          Types.TimestampNanoType.withZone(),
+          Types.VariantType.get());
 
   private static final Schema INITIAL_DEFAULT_SCHEMA =
       new Schema(
diff --git a/api/src/test/java/org/apache/iceberg/transforms/TestBucketing.java b/api/src/test/java/org/apache/iceberg/transforms/TestBucketing.java
index fc4333d7c6c5..5f0cac2b5e8c 100644
--- a/api/src/test/java/org/apache/iceberg/transforms/TestBucketing.java
+++ b/api/src/test/java/org/apache/iceberg/transforms/TestBucketing.java
@@ -417,6 +417,20 @@ public void testVerifiedIllegalNumBuckets() {
         .hasMessage("Invalid number of buckets: 0 (must be > 0)");
   }
 
+  @Test
+  public void testVariantUnsupported() {
+    assertThatThrownBy(() -> Transforms.bucket(Types.VariantType.get(), 3))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Cannot bucket by type: variant");
+
+    Transform<Object, Integer> bucket = Transforms.bucket(3);
+    assertThatThrownBy(() -> bucket.bind(Types.VariantType.get()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Cannot bucket by type: variant");
+
+    assertThat(bucket.canTransform(Types.VariantType.get())).isFalse();
+  }
+
   private byte[] randomBytes(int length) {
     byte[] bytes = new byte[length];
     testRandom.nextBytes(bytes);
diff --git a/api/src/test/java/org/apache/iceberg/transforms/TestIdentity.java b/api/src/test/java/org/apache/iceberg/transforms/TestIdentity.java
index 93d3281411f3..b5076e08a947 100644
--- a/api/src/test/java/org/apache/iceberg/transforms/TestIdentity.java
+++ b/api/src/test/java/org/apache/iceberg/transforms/TestIdentity.java
@@ -19,6 +19,7 @@
 package org.apache.iceberg.transforms;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
@@ -155,4 +156,21 @@ public void testBigDecimalToHumanString() {
         .as("Should not modify Strings")
         .isEqualTo(decimalString);
   }
+
+  @Test
+  public void testVariantUnsupported() {
+    assertThatThrownBy(() -> Transforms.identity().bind(Types.VariantType.get()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Cannot bind to unsupported type: variant");
+
+    assertThatThrownBy(() -> Transforms.fromString(Types.VariantType.get(), "identity"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Unsupported type for identity: variant");
+
+    assertThatThrownBy(() -> Transforms.identity(Types.VariantType.get()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Unsupported type for identity: variant");
+
+    assertThat(Transforms.identity().canTransform(Types.VariantType.get())).isFalse();
+  }
 }
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 97bb2b1a19e4..af2ebae7e1a8 100644
--- a/api/src/test/java/org/apache/iceberg/types/TestSerializableTypes.java
+++ b/api/src/test/java/org/apache/iceberg/types/TestSerializableTypes.java
@@ -45,7 +45,7 @@ public void testIdentityTypes() throws Exception {
           Types.TimestampNanoType.withZone(),
           Types.StringType.get(),
           Types.UUIDType.get(),
-          Types.BinaryType.get()
+          Types.BinaryType.get(),
         };
 
     for (Type type : identityPrimitives) {
@@ -127,6 +127,15 @@ public void testLists() throws Exception {
     }
   }
 
+  @Test
+  public void testVariant() throws Exception {
+    Types.VariantType variant = Types.VariantType.get();
+    Type copy = TestHelpers.roundTripSerialize(variant);
+    assertThat(copy)
+        .as("Variant serialization should be equal to starting type")
+        .isEqualTo(variant);
+  }
+
   @Test
   public void testSchema() throws Exception {
     Schema schema =
diff --git a/core/src/test/java/org/apache/iceberg/TestSortOrder.java b/core/src/test/java/org/apache/iceberg/TestSortOrder.java
index 6e99af3e260e..3d139543b71c 100644
--- a/core/src/test/java/org/apache/iceberg/TestSortOrder.java
+++ b/core/src/test/java/org/apache/iceberg/TestSortOrder.java
@@ -326,6 +326,22 @@ public void testSortedColumnNames() {
     assertThat(sortedCols).containsExactly("s.id", "data");
   }
 
+  @TestTemplate
+  public void testVariantUnsupported() {
+    Schema v3Schema =
+        new Schema(
+            Types.NestedField.required(3, "id", Types.LongType.get()),
+            Types.NestedField.required(4, "data", Types.StringType.get()),
+            Types.NestedField.required(
+                5,
+                "struct",
+                Types.StructType.of(Types.NestedField.optional(6, "v", Types.VariantType.get()))));
+
+    assertThatThrownBy(() -> SortOrder.builderFor(v3Schema).withOrderId(10).asc("struct.v").build())
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Unsupported type for identity: variant");
+  }
+
   @TestTemplate
   public void testPreservingOrderSortedColumnNames() {
     SortOrder order =
diff --git a/core/src/test/java/org/apache/iceberg/TestTableMetadata.java b/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
index 71254b3abb1b..c9a8eb75a986 100644
--- a/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
+++ b/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
@@ -885,7 +885,8 @@ public void testV2UUIDValidation() {
 
   @Test
   public void testVersionValidation() {
-    int unsupportedVersion = TableMetadata.SUPPORTED_TABLE_FORMAT_VERSION + 1;
+    int supportedVersion = TableMetadata.SUPPORTED_TABLE_FORMAT_VERSION;
+    int unsupportedVersion = supportedVersion + 1;
     assertThatThrownBy(
             () ->
                 new TableMetadata(
@@ -914,7 +915,62 @@ public void testVersionValidation() {
                     ImmutableList.of(),
                     ImmutableList.of()))
         .isInstanceOf(IllegalArgumentException.class)
-        .hasMessage("Unsupported format version: v" + unsupportedVersion);
+        .hasMessage(
+            "Unsupported format version: v%s (supported: v%s)",
+            unsupportedVersion, supportedVersion);
+
+    assertThatThrownBy(
+            () ->
+                TableMetadata.newTableMetadata(
+                    TEST_SCHEMA,
+                    PartitionSpec.unpartitioned(),
+                    SortOrder.unsorted(),
+                    TEST_LOCATION,
+                    ImmutableMap.of(),
+                    unsupportedVersion))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage(
+            "Unsupported format version: v%s (supported: v%s)",
+            unsupportedVersion, supportedVersion);
+
+    // should be allowed in the supported version
+    assertThat(
+            new TableMetadata(
+                null,
+                supportedVersion,
+                UUID.randomUUID().toString(),
+                TEST_LOCATION,
+                SEQ_NO,
+                System.currentTimeMillis(),
+                LAST_ASSIGNED_COLUMN_ID,
+                7,
+                ImmutableList.of(TEST_SCHEMA),
+                SPEC_5.specId(),
+                ImmutableList.of(SPEC_5),
+                SPEC_5.lastAssignedFieldId(),
+                3,
+                ImmutableList.of(SORT_ORDER_3),
+                ImmutableMap.of(),
+                -1L,
+                ImmutableList.of(),
+                null,
+                ImmutableList.of(),
+                ImmutableList.of(),
+                ImmutableMap.of(),
+                ImmutableList.of(),
+                ImmutableList.of(),
+                ImmutableList.of()))
+        .isNotNull();
+
+    assertThat(
+            TableMetadata.newTableMetadata(
+                TEST_SCHEMA,
+                PartitionSpec.unpartitioned(),
+                SortOrder.unsorted(),
+                TEST_LOCATION,
+                ImmutableMap.of(),
+                supportedVersion))
+        .isNotNull();
   }
 
   @Test
@@ -1647,46 +1703,6 @@ public void testConstructV3Metadata() {
         3);
   }
 
-  @Test
-  public void testV3TimestampNanoTypeSupport() {
-    Schema v3Schema =
-        new Schema(
-            Types.NestedField.required(3, "id", Types.LongType.get()),
-            Types.NestedField.required(4, "data", Types.StringType.get()),
-            Types.NestedField.required(
-                5,
-                "struct",
-                Types.StructType.of(
-                    Types.NestedField.optional(
-                        6, "ts_nanos", Types.TimestampNanoType.withZone()))));
-
-    for (int unsupportedFormatVersion : ImmutableList.of(1, 2)) {
-      assertThatThrownBy(
-              () ->
-                  TableMetadata.newTableMetadata(
-                      v3Schema,
-                      PartitionSpec.unpartitioned(),
-                      SortOrder.unsorted(),
-                      TEST_LOCATION,
-                      ImmutableMap.of(),
-                      unsupportedFormatVersion))
-          .isInstanceOf(IllegalStateException.class)
-          .hasMessage(
-              "Invalid schema for v%s:\n"
-                  + "- Invalid type for struct.ts_nanos: timestamptz_ns is not supported until v3",
-              unsupportedFormatVersion);
-    }
-
-    // should be allowed in v3
-    TableMetadata.newTableMetadata(
-        v3Schema,
-        PartitionSpec.unpartitioned(),
-        SortOrder.unsorted(),
-        TEST_LOCATION,
-        ImmutableMap.of(),
-        3);
-  }
-
   @Test
   public void onlyMetadataLocationIsUpdatedWithoutTimestampAndMetadataLogEntry() {
     String uuid = "386b9f01-002b-4d8c-b77f-42c3fd3b7c9b";