Only add the type to API. Will add VariantLike stable interface after…

… full implementation. Block Transforms, SortOrder

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

@@ -56,7 +56,7 @@ public class Schema implements Serializable {
   private static final int DEFAULT_SCHEMA_ID = 0;
   private static final int DEFAULT_VALUES_MIN_FORMAT_VERSION = 3;
   private 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;

api/src/main/java/org/apache/iceberg/transforms/Identity.java

@@ -38,6 +38,8 @@ class Identity<T> implements Transform<T, T> {
    */
   @Deprecated
   public static <I> Identity<I> get(Type type) {
+    Preconditions.checkArgument(!type.isVariantType(), "Unsupported type for identity: %s", type);
+
     return new Identity<>(type);
   }
 

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

@@ -26,7 +26,6 @@
 import java.util.Map;
 import java.util.Objects;
 import org.apache.iceberg.StructLike;
-import org.apache.iceberg.VariantLike;
 
 public interface Type extends Serializable {
   enum TypeID {
@@ -47,7 +46,7 @@ enum TypeID {
     STRUCT(StructLike.class),
     LIST(List.class),
     MAP(Map.class),
-    VARIANT(VariantLike.class);
+    VARIANT(Object.class);
 
     private final Class<?> javaClass;
 
@@ -94,6 +93,10 @@ default boolean isListType() {
     return false;
   }
 
+  default boolean isVariantType() {
+    return false;
+  }
+
   default boolean isMapType() {
     return false;
   }

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

@@ -420,6 +420,16 @@ public static VariantType get() {
       return INSTANCE;
     }
 
+    @Override
+    public boolean isPrimitiveType() {
+      return false;
+    }
+
+    @Override
+    public boolean isVariantType() {
+      return true;
+    }
+
     @Override
     public TypeID typeId() {
       return TypeID.VARIANT;

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

@@ -22,14 +22,9 @@
 import static org.apache.iceberg.types.Types.NestedField.required;
 import static org.assertj.core.api.Assertions.assertThat;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.util.Base64;
 import java.util.UUID;
-import org.apache.iceberg.TestHelpers.JsonVariant;
 import org.apache.iceberg.TestHelpers.Row;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -252,70 +247,4 @@ public void testEmptySchema() {
     Schema emptySchema = new Schema();
     assertThat(emptySchema.accessorForField(17)).isNull();
   }
-
-  @Test
-  public void testVariant() throws JsonProcessingException {
-    Base64.Encoder encoder = Base64.getEncoder();
-    boolean expectedTrue = true;
-    boolean expectedFalse = false;
-    int expectedInt = 2147483647;
-    long expectedLong = 2147483648L;
-    float expectedFloat = 1.2345f;
-    double expectedDouble = 1.23456;
-    BigDecimal expectedDecimal = new BigDecimal(123456);
-    String expectedString = "abc";
-    String expectedBytes =
-        new String(encoder.encode(expectedString.getBytes()), StandardCharsets.UTF_8);
-    int nestInt = 10;
-
-    String json =
-        "{\"false\":"
-            + expectedFalse
-            + ", \"true\":"
-            + expectedTrue
-            + ", \"string\": \""
-            + expectedString
-            + "\","
-            + "\"int\":"
-            + expectedInt
-            + ","
-            + "\"long\":"
-            + expectedLong
-            + ", \"float\":"
-            + expectedFloat
-            + ","
-            + "\"double\":"
-            + expectedDouble
-            + ", \"bytes\":\""
-            + expectedBytes
-            + "\", \"decimal\":"
-            + expectedDecimal
-            + ","
-            + "\"nest1\": {\"nest2\":"
-            + nestInt
-            + "}"
-            + "}";
-
-    VariantLike variant = JsonVariant.of(json);
-    assertAccessorReturns(Types.VariantType.get(), variant);
-
-    assertThat(variant.get(new String[] {"true"}).get(Boolean.class)).isEqualTo(expectedTrue);
-    assertThat(variant.get(new String[] {"false"}).get(Boolean.class)).isEqualTo(expectedFalse);
-    assertThat(variant.get(new String[] {"string"}).get(String.class)).isEqualTo(expectedString);
-    assertThat(variant.get(new String[] {"int"}).get(Integer.class)).isEqualTo(expectedInt);
-    assertThat(variant.get(new String[] {"long"}).get(Long.class)).isEqualTo(expectedLong);
-    assertThat(variant.get(new String[] {"float"}).get(Float.class)).isEqualTo(expectedFloat);
-    assertThat(variant.get(new String[] {"double"}).get(Double.class)).isEqualTo(expectedDouble);
-    assertThat(variant.get(new String[] {"decimal"}).get(BigDecimal.class))
-        .isEqualTo(expectedDecimal);
-    assertThat(
-            StandardCharsets.UTF_8
-                .decode(variant.get(new String[] {"bytes"}).get(ByteBuffer.class))
-                .toString())
-        .isEqualTo(expectedString);
-    assertThat(variant.get(new String[] {"nest1", "nest2"}).get(Integer.class)).isEqualTo(nestInt);
-    assertThat(variant.get(new String[] {"nest1", "invalid"})).isNull();
-    assertThat(new ObjectMapper().readTree(variant.toJson()))
-        .isEqualTo(new ObjectMapper().readTree(json));
-  }
 }

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

@@ -25,8 +25,6 @@
 import com.esotericsoftware.kryo.io.Input;
 import com.esotericsoftware.kryo.io.Output;
 import com.esotericsoftware.kryo.serializers.ClosureSerializer;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -36,12 +34,10 @@
 import java.io.OutputStream;
 import java.io.Serializable;
 import java.lang.invoke.SerializedLambda;
-import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Set;
 import java.util.stream.IntStream;
 import org.apache.iceberg.expressions.BoundPredicate;
@@ -406,101 +402,6 @@ public int hashCode() {
     }
   }
 
-  /** A VariantLike implementation for testing accepting JSON input */
-  public static class JsonVariant implements VariantLike {
-    public static JsonVariant of(String json) {
-      return new JsonVariant(json);
-    }
-
-    private final JsonNode node;
-
-    private JsonVariant(String json) {
-      try {
-        ObjectMapper mapper = new ObjectMapper();
-        this.node = mapper.readTree(json);
-      } catch (IOException e) {
-        throw new RuntimeException(e);
-      }
-    }
-
-    private JsonVariant(JsonNode node) {
-      this.node = node;
-    }
-
-    @Override
-    public int size() {
-      return node.size();
-    }
-
-    @Override
-    public <T> T get(Class<T> javaClass) {
-      if (javaClass.equals(Boolean.class)) {
-        return (T) (Boolean) node.asBoolean();
-      } else if (javaClass.equals(Integer.class)) {
-        return (T) (Integer) node.asInt();
-      } else if (javaClass.equals(Long.class)) {
-        return (T) (Long) node.asLong();
-      } else if (javaClass.equals(Float.class)) {
-        return (T) Float.valueOf((float) node.asDouble());
-      } else if (javaClass.equals(Double.class)) {
-        return (T) (Double) (node.asDouble());
-      } else if (CharSequence.class.isAssignableFrom(javaClass)) {
-        return (T) node.asText();
-      } else if (javaClass.equals(ByteBuffer.class)) {
-        try {
-          return (T) ByteBuffer.wrap(node.binaryValue());
-        } catch (IOException e) {
-          throw new RuntimeException(e);
-        }
-      } else if (javaClass.equals(BigDecimal.class)) {
-        return (T) node.decimalValue();
-      }
-
-      throw new IllegalArgumentException("Unsupported type: " + javaClass);
-    }
-
-    @Override
-    public VariantLike get(String[] path) {
-      Preconditions.checkState(
-          path != null && path.length > 0, "path must contain at least one element");
-
-      JsonNode childNode = node;
-      for (String pathElement : path) {
-        childNode = childNode.get(pathElement);
-        if (childNode == null) {
-          return null;
-        }
-      }
-
-      return new JsonVariant(childNode);
-    }
-
-    @Override
-    public String toJson() {
-      return node.toString();
-    }
-
-    @Override
-    public boolean equals(Object other) {
-      if (this == other) {
-        return true;
-      }
-
-      if (other == null || getClass() != other.getClass()) {
-        return false;
-      }
-
-      JsonVariant that = (JsonVariant) other;
-
-      return Objects.equals(node, that.node);
-    }
-
-    @Override
-    public int hashCode() {
-      return Objects.hashCode(node);
-    }
-  }
-
   public static class TestFieldSummary implements ManifestFile.PartitionFieldSummary {
     private final boolean containsNull;
     private final Boolean containsNaN;

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");
+  }
 }

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);