API: Add Variant data type (apache#11324)

aihuaxu · web-flow · commit 821aec32bb9b · 2024-11-15T12:52:23.000-06:00
diff --git 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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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/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
@@ -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
@@ -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
@@ -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
@@ -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/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(
diff --git 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

Original file line number	Diff line number	Diff line change
`@@ -534,7 +534,8 @@ private static String sanitize(Type type, Object value, long now, int today) {`
`534`	`534`	`case DECIMAL:`
`535`	`535`	`case FIXED:`
`536`	`536`	`case BINARY:`
`537`		`- // for boolean, uuid, decimal, fixed, and binary, match the string result`
	`537`	`+ case VARIANT:`
	`538`	`+ // for boolean, uuid, decimal, fixed, variant, and binary, match the string result`
`538`	`539`	`return sanitizeSimpleString(value.toString());`
`539`	`540`	`}`
`540`	`541`	`throw new UnsupportedOperationException(`
`@@ -562,7 +563,7 @@ private static String sanitize(Literal<?> literal, long now, int today) {`
`562`	`563`	`} else if (literal instanceof Literals.DoubleLiteral) {`
`563`	`564`	`return sanitizeNumber(((Literals.DoubleLiteral) literal).value(), "float");`
`564`	`565`	`} else {`
`565`		`- // for uuid, decimal, fixed, and binary, match the string result`
	`566`	`+ // for uuid, decimal, fixed, variant, and binary, match the string result`
`566`	`567`	`return sanitizeSimpleString(literal.value().toString());`
`567`	`568`	`}`
`568`	`569`	`}`
Original file line number	Diff line number	Diff line change
`@@ -38,6 +38,9 @@ class Identity<T> implements Transform<T, T> {`
`38`	`38`	`*/`
`39`	`39`	`@Deprecated`
`40`	`40`	`public static <I> Identity<I> get(Type type) {`
	`41`	`+ Preconditions.checkArgument(`
	`42`	`+ type.typeId() != Type.TypeID.VARIANT, "Unsupported type for identity: %s", type);`
	`43`	`+`
`41`	`44`	`return new Identity<>(type);`
`42`	`45`	`}`
`43`	`46`