Add array/object build interface

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

@@ -28,7 +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());
   }

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;
@@ -116,8 +115,8 @@ private void initOffsetsAndLengths(int numElements) {
     }
   }
 
-  @VisibleForTesting
-  int numElements() {
+  @Override
+  public int numElements() {
     return fieldIds.length;
   }
 

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

@@ -30,7 +30,7 @@ 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

@@ -25,15 +25,17 @@ public final class Variant {
   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 != 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.");
+    Preconditions.checkArgument(
+        (metadata[0] & VariantConstants.VERSION_MASK) == VariantConstants.VERSION,
+        "Unsupported metadata version.");
 
-    if (value.length > VariantConstants.SIZE_LIMIT || metadata.length > VariantConstants.SIZE_LIMIT) {
+    if (value.length > VariantConstants.SIZE_LIMIT
+        || metadata.length > VariantConstants.SIZE_LIMIT) {
       throw new VariantSizeLimitException();
     }
 

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 writeNullElement() {
+    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);
+  }
+}