Add array/object build interface

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

@@ -28,11 +28,6 @@ class SerializedArray extends Variants.SerializedValue implements VariantArray {
   private static final int OFFSET_SIZE_SHIFT = 2;
   private static final int IS_LARGE = 0b10000;
 
-
-  static SerializedArray from(Variant variant) {
-    return from(SerializedMetadata.from(variant.getMetadata()), variant.getValue());
-  }
-
   @VisibleForTesting
   static SerializedArray from(SerializedMetadata metadata, byte[] bytes) {
     return from(metadata, ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN), bytes[0]);

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

@@ -24,7 +24,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
-import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.Pair;
@@ -36,10 +35,6 @@ class SerializedObject extends Variants.SerializedValue implements VariantObject
   private static final int FIELD_ID_SIZE_SHIFT = 4;
   private static final int IS_LARGE = 0b1000000;
 
-  static SerializedObject from(Variant variant) {
-    return from(SerializedMetadata.from(variant.getMetadata()), variant.getValue());
-  }
-
   static SerializedObject from(SerializedMetadata metadata, byte[] bytes) {
     return from(metadata, ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN), bytes[0]);
   }
@@ -116,8 +111,8 @@ private void initOffsetsAndLengths(int numElements) {
     }
   }
 
-  @VisibleForTesting
-  int numElements() {
+  @Override
+  public int numElements() {
     return fieldIds.length;
   }
 

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

@@ -28,10 +28,6 @@ class SerializedPrimitive extends Variants.SerializedValue implements VariantPri
   private static final int PRIMITIVE_TYPE_SHIFT = 2;
   private static final int PRIMITIVE_OFFSET = Variants.HEADER_SIZE;
 
-  static SerializedPrimitive from(Variant variant) {
-    return from(variant.getValue());
-  }
-
   static SerializedPrimitive from(byte[] bytes) {
     return from(ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN), bytes[0]);
   }

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

@@ -26,11 +26,7 @@ class SerializedShortString extends Variants.SerializedValue implements VariantP
   private static final int LENGTH_MASK = 0b11111100;
   private static final int LENGTH_SHIFT = 2;
 
-  static SerializedShortString from(Variant variant) {
-    return from(variant.getValue());
-  }
-
-    static SerializedShortString from(byte[] bytes) {
+  static SerializedShortString from(byte[] bytes) {
     return from(ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN), bytes[0]);
   }
 

core/src/main/java/org/apache/iceberg/variants/Variant.java

@@ -18,34 +18,11 @@
  */
 package org.apache.iceberg.variants;
 
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+/** A variant metadata and value pair. */
+public interface Variant {
+  /** Returns the metadata for all values in the variant. */
+  VariantMetadata metadata();
 
-public final class Variant {
-  private final byte[] value;
-  private final byte[] metadata;
-
-  public Variant(byte[] value, byte[] metadata) {
-    Preconditions.checkArgument(metadata != null && metadata.length >= 1,
-     "Metadata must not be null or empty.");
-    Preconditions.checkArgument(value != null && value.length >= 1,
-        "Value must not be null or empty.");
-
-    Preconditions.checkArgument((metadata[0] & VariantConstants.VERSION_MASK) == VariantConstants.VERSION,
-      "Unsupported metadata version.");
-
-    if (value.length > VariantConstants.SIZE_LIMIT || metadata.length > VariantConstants.SIZE_LIMIT) {
-      throw new VariantSizeLimitException();
-    }
-
-    this.value = value;
-    this.metadata = metadata;
-  }
-
-  public byte[] getMetadata() {
-    return metadata;
-  }
-
-  public byte[] getValue() {
-    return value;
-  }
+  /** Returns the variant value. */
+  VariantValue value();
 }

core/src/main/java/org/apache/iceberg/variants/VariantArray.java

@@ -20,7 +20,9 @@
 
 /** An variant array value. */
 public interface VariantArray extends VariantValue {
-  int numElements();
+  default int numElements() {
+    throw new UnsupportedOperationException();
+  }
 
   /** Returns the {@link VariantValue} at {@code index} in this array. */
   VariantValue get(int index);

core/src/main/java/org/apache/iceberg/variants/VariantArrayBuilder.java

@@ -0,0 +1,120 @@
+/*
+ * 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.math.BigDecimal;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.util.List;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.DateTimeUtil;
+
+public class VariantArrayBuilder extends VariantBuilderBase {
+  private final List<Integer> offsets;
+
+  public VariantArrayBuilder(ByteBufferWrapper buffer, Dictionary dict) {
+    super(buffer, dict);
+    offsets = Lists.newArrayList();
+  }
+
+  public VariantObjectBuilder startObject() {
+    addOffset();
+    return new VariantObjectBuilder(buffer, dict);
+  }
+
+  public VariantArrayBuilder startArray() {
+    addOffset();
+    return new VariantArrayBuilder(buffer, dict);
+  }
+
+  public VariantArrayBuilder writeNull() {
+    addOffset();
+    writeNullInternal();
+    return this;
+  }
+
+  public VariantArrayBuilder writeBoolean(boolean value) {
+    addOffset();
+    writeBooleanInternal(value);
+    return this;
+  }
+
+  public VariantArrayBuilder writeNumeric(long value) {
+    addOffset();
+    writeNumericInternal(value);
+    return this;
+  }
+
+  public VariantArrayBuilder writeDouble(double value) {
+    addOffset();
+    writeDoubleInternal(value);
+    return this;
+  }
+
+  public VariantArrayBuilder writeDecimal(BigDecimal value) {
+    addOffset();
+    writeDecimalInternal(value);
+    return this;
+  }
+
+  public VariantArrayBuilder writeDate(LocalDate value) {
+    addOffset();
+    writeDateInternal(DateTimeUtil.daysFromDate(value));
+    return this;
+  }
+
+  public VariantArrayBuilder writeTimestampTz(OffsetDateTime value) {
+    addOffset();
+    writeTimestampTzInternal(DateTimeUtil.microsFromTimestamptz(value));
+    return this;
+  }
+
+  public VariantArrayBuilder writeTimestampNtz(LocalDateTime value) {
+    addOffset();
+    writeTimestampNtzInternal(DateTimeUtil.microsFromTimestamp(value));
+    return this;
+  }
+
+  public VariantArrayBuilder writeFloat(float value) {
+    addOffset();
+    writeFloatInternal(value);
+    return this;
+  }
+
+  public VariantArrayBuilder writeBinary(byte[] value) {
+    addOffset();
+    writeBinaryInternal(value);
+    return this;
+  }
+
+  public VariantArrayBuilder writeString(String str) {
+    addOffset();
+    writeStringInternal(str);
+    return this;
+  }
+
+  private void addOffset() {
+    offsets.add(buffer.pos - startPos);
+  }
+
+  public void endArray() {
+    super.endArray(startPos, offsets);
+  }
+}