API, Core: Make variant classes serializable

api/src/main/java/org/apache/iceberg/variants/SerializedArray.java

@@ -18,13 +18,14 @@
  */
 package org.apache.iceberg.variants;
 
+import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.util.ByteBuffers;
 
-class SerializedArray implements VariantArray, SerializedValue {
+class SerializedArray implements VariantArray, SerializedValue, Serializable {
   private static final int HEADER_SIZE = 1;
   private static final int OFFSET_SIZE_MASK = 0b1100;
   private static final int OFFSET_SIZE_SHIFT = 2;
@@ -91,4 +92,22 @@ public ByteBuffer buffer() {
   public String toString() {
     return VariantArray.asString(this);
   }
+
+  private Object writeReplace() {
+    return new SerializationProxy(this);
+  }
+
+  private static class SerializationProxy implements Serializable {
+    private final VariantMetadata metadata;
+    private final byte[] valueBytes;
+
+    private SerializationProxy(SerializedArray array) {
+      this.metadata = array.metadata;
+      this.valueBytes = ByteBuffers.toByteArray(array.buffer());
+    }
+
+    private Object readResolve() {
+      return SerializedArray.from(metadata, valueBytes);
+    }
+  }
 }

api/src/main/java/org/apache/iceberg/variants/SerializedMetadata.java

@@ -18,13 +18,14 @@
  */
 package org.apache.iceberg.variants;
 
+import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.util.ByteBuffers;
 
-class SerializedMetadata implements VariantMetadata, Serialized {
+class SerializedMetadata implements VariantMetadata, Serialized, Serializable {
   private static final int HEADER_SIZE = 1;
   private static final int SUPPORTED_VERSION = 1;
   private static final int VERSION_MASK = 0b1111;
@@ -138,4 +139,20 @@ public int writeTo(ByteBuffer buffer, int offset) {
   public String toString() {
     return VariantMetadata.asString(this);
   }
+
+  private Object writeReplace() {
+    return new SerializationProxy(this);
+  }
+
+  private static class SerializationProxy implements Serializable {
+    private final byte[] metadataBytes;
+
+    private SerializationProxy(SerializedMetadata metadata) {
+      this.metadataBytes = ByteBuffers.toByteArray(metadata.buffer());
+    }
+
+    private Object readResolve() {
+      return SerializedMetadata.from(metadataBytes);
+    }
+  }
 }

api/src/main/java/org/apache/iceberg/variants/SerializedObject.java

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.variants;
 
+import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.util.Iterator;
@@ -29,7 +30,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.ByteBuffers;
 
-class SerializedObject implements VariantObject, SerializedValue {
+class SerializedObject implements VariantObject, SerializedValue, Serializable {
   private static final int HEADER_SIZE = 1;
   private static final int OFFSET_SIZE_MASK = 0b1100;
   private static final int OFFSET_SIZE_SHIFT = 2;
@@ -238,4 +239,22 @@ public boolean equals(Object obj) {
   public String toString() {
     return VariantObject.asString(this);
   }
+
+  private Object writeReplace() {
+    return new SerializationProxy(this);
+  }
+
+  private static class SerializationProxy implements Serializable {
+    private final VariantMetadata metadata;
+    private final byte[] valueBytes;
+
+    private SerializationProxy(SerializedObject object) {
+      this.metadata = object.metadata;
+      this.valueBytes = ByteBuffers.toByteArray(object.buffer());
+    }
+
+    private Object readResolve() {
+      return SerializedObject.from(metadata, valueBytes);
+    }
+  }
 }

api/src/main/java/org/apache/iceberg/variants/SerializedPrimitive.java

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.variants;
 
+import java.io.Serializable;
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.nio.ByteBuffer;
@@ -26,7 +27,7 @@
 import org.apache.iceberg.util.ByteBuffers;
 import org.apache.iceberg.util.UUIDUtil;
 
-class SerializedPrimitive implements VariantPrimitive<Object>, SerializedValue {
+class SerializedPrimitive implements VariantPrimitive<Object>, SerializedValue, Serializable {
   private static final int PRIMITIVE_TYPE_SHIFT = 2;
   private static final int PRIMITIVE_OFFSET = 1;
 
@@ -150,4 +151,20 @@ public boolean equals(Object other) {
   public String toString() {
     return VariantPrimitive.asString(this);
   }
+
+  private Object writeReplace() {
+    return new SerializationProxy(this);
+  }
+
+  private static class SerializationProxy implements Serializable {
+    private final byte[] valueBytes;
+
+    private SerializationProxy(SerializedPrimitive primitive) {
+      this.valueBytes = ByteBuffers.toByteArray(primitive.buffer());
+    }
+
+    private Object readResolve() {
+      return SerializedPrimitive.from(valueBytes);
+    }
+  }
 }

api/src/main/java/org/apache/iceberg/variants/SerializedShortString.java

@@ -18,11 +18,13 @@
  */
 package org.apache.iceberg.variants;
 
+import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.ByteBuffers;
 
-class SerializedShortString implements VariantPrimitive<String>, SerializedValue {
+class SerializedShortString implements VariantPrimitive<String>, SerializedValue, Serializable {
   private static final int HEADER_SIZE = 1;
   private static final int LENGTH_MASK = 0b11111100;
   private static final int LENGTH_SHIFT = 2;
@@ -81,4 +83,20 @@ public boolean equals(Object other) {
   public String toString() {
     return VariantPrimitive.asString(this);
   }
+
+  private Object writeReplace() {
+    return new SerializationProxy(this);
+  }
+
+  private static class SerializationProxy implements Serializable {
+    private final byte[] valueBytes;
+
+    private SerializationProxy(SerializedShortString string) {
+      this.valueBytes = ByteBuffers.toByteArray(string.buffer());
+    }
+
+    private Object readResolve() {
+      return SerializedShortString.from(valueBytes);
+    }
+  }
 }

api/src/main/java/org/apache/iceberg/variants/VariantData.java

@@ -18,9 +18,10 @@
  */
 package org.apache.iceberg.variants;
 
+import java.io.Serializable;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
-class VariantData implements Variant {
+class VariantData implements Variant, Serializable {
   private final VariantMetadata metadata;
   private final VariantValue value;
 

api/src/test/java/org/apache/iceberg/variants/TestSerializedArray.java

@@ -23,9 +23,11 @@
 
 import java.nio.ByteBuffer;
 import java.util.Random;
+import org.apache.iceberg.TestHelpers.RoundTripSerializer;
 import org.apache.iceberg.util.RandomUtil;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 public class TestSerializedArray {
@@ -197,4 +199,13 @@ public void testNegativeArraySize() {
         .isInstanceOf(NegativeArraySizeException.class)
         .hasMessage("-1");
   }
+
+  @ParameterizedTest
+  @MethodSource("org.apache.iceberg.TestHelpers#serializers")
+  public void testSerialization(RoundTripSerializer<SerializedArray> serializer) throws Exception {
+    ByteBuffer buffer = VariantTestUtil.createArray(A, B, C, I34, I1234);
+    SerializedArray array = SerializedArray.from(EMPTY_METADATA, buffer, buffer.get(0));
+
+    VariantTestUtil.assertEqual(array, serializer.apply(array));
+  }
 }

api/src/test/java/org/apache/iceberg/variants/TestSerializedMetadata.java

@@ -24,10 +24,12 @@
 import java.nio.ByteBuffer;
 import java.util.Random;
 import java.util.Set;
+import org.apache.iceberg.TestHelpers.RoundTripSerializer;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.util.RandomUtil;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 public class TestSerializedMetadata {
@@ -251,4 +253,14 @@ public void testLengthTooShort() {
             () -> SerializedMetadata.from(new byte[] {(byte) 0b11010001, 0x00, 0x00, 0x00}))
         .isInstanceOf(IndexOutOfBoundsException.class);
   }
+
+  @ParameterizedTest
+  @MethodSource("org.apache.iceberg.TestHelpers#serializers")
+  public void testSerialization(RoundTripSerializer<SerializedMetadata> serializer)
+      throws Exception {
+    SerializedMetadata metadata =
+        SerializedMetadata.from(VariantTestUtil.createMetadata(Set.of("a", "b", "c"), true));
+
+    VariantTestUtil.assertEqual(metadata, serializer.apply(metadata));
+  }
 }

api/src/test/java/org/apache/iceberg/variants/TestSerializedObject.java

@@ -25,6 +25,7 @@
 import java.util.Random;
 import java.util.Set;
 import java.util.stream.Stream;
+import org.apache.iceberg.TestHelpers.RoundTripSerializer;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -339,4 +340,16 @@ public void testThreeByteFieldIds(boolean sortFieldNames) {
     assertThat(object.get("ZZ").type()).isEqualTo(PhysicalType.INT8);
     assertThat(object.get("ZZ").asPrimitive().get()).isEqualTo((byte) 3);
   }
+
+  @ParameterizedTest
+  @MethodSource("org.apache.iceberg.TestHelpers#serializers")
+  public void testSerialization(RoundTripSerializer<SerializedObject> serializer) throws Exception {
+    Map<String, VariantValue> data = ImmutableMap.of("a", I1, "b", I2, "c", I3);
+    ByteBuffer meta = VariantTestUtil.createMetadata(data.keySet(), true /* sort names */);
+    ByteBuffer value = VariantTestUtil.createObject(meta, data);
+    SerializedObject object =
+        SerializedObject.from(SerializedMetadata.from(meta), value, value.get(0));
+
+    VariantTestUtil.assertEqual(object, serializer.apply(object));
+  }
 }

api/src/test/java/org/apache/iceberg/variants/TestSerializedPrimitives.java

@@ -25,8 +25,11 @@
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.util.UUID;
+import org.apache.iceberg.TestHelpers.RoundTripSerializer;
 import org.apache.iceberg.util.DateTimeUtil;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 
 public class TestSerializedPrimitives {
   @Test
@@ -617,4 +620,25 @@ public void testUnsupportedType() {
   private static byte primitiveHeader(int primitiveType) {
     return (byte) (primitiveType << 2);
   }
+
+  @ParameterizedTest
+  @MethodSource("org.apache.iceberg.TestHelpers#serializers")
+  public void testPrimitiveSerialization(RoundTripSerializer<SerializedPrimitive> serializer)
+      throws Exception {
+    // date value
+    SerializedPrimitive primitive =
+        SerializedPrimitive.from(new byte[] {primitiveHeader(11), (byte) 0xF4, 0x43, 0x00, 0x00});
+
+    VariantTestUtil.assertEqual(primitive, serializer.apply(primitive));
+  }
+
+  @ParameterizedTest
+  @MethodSource("org.apache.iceberg.TestHelpers#serializers")
+  public void testShortStringSerialization(RoundTripSerializer<SerializedShortString> serializer)
+      throws Exception {
+    SerializedShortString string =
+        SerializedShortString.from(new byte[] {0b11101, 'i', 'c', 'e', 'b', 'e', 'r', 'g'});
+
+    VariantTestUtil.assertEqual(string, serializer.apply(string));
+  }
 }

api/src/test/java/org/apache/iceberg/variants/TestVariant.java

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.variants;
+
+import java.util.Map;
+import org.apache.iceberg.TestHelpers.RoundTripSerializer;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class TestVariant {
+  @ParameterizedTest
+  @MethodSource("org.apache.iceberg.TestHelpers#serializers")
+  public void testSerialization(RoundTripSerializer<Variant> serializer) throws Exception {
+    Map<String, VariantValue> data =
+        ImmutableMap.of(
+            "a",
+            SerializedPrimitive.from(new byte[] {0b1100, 1}), // int8 = 1
+            "b",
+            VariantTestUtil.createShortString("iceberg"));
+    Variant variant = VariantTestUtil.variant(data);
+
+    Variant result = serializer.apply(variant);
+
+    VariantTestUtil.assertEqual(variant.metadata(), result.metadata());
+    VariantTestUtil.assertEqual(variant.value(), result.value());
+  }
+}