API: Harden variant binary parsing against malformed input

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

@@ -36,29 +36,50 @@ static SerializedArray from(VariantMetadata metadata, byte[] bytes) {
   }
 
   static SerializedArray from(VariantMetadata metadata, ByteBuffer value, int header) {
+    return from(metadata, value, header, 0);
+  }
+
+  static SerializedArray from(VariantMetadata metadata, ByteBuffer value, int header, int depth) {
     Preconditions.checkArgument(
         value.order() == ByteOrder.LITTLE_ENDIAN, "Unsupported byte order: big endian");
     BasicType basicType = VariantUtil.basicType(header);
     Preconditions.checkArgument(
         basicType == BasicType.ARRAY, "Invalid array, basic type: " + basicType);
-    return new SerializedArray(metadata, value, header);
+    return new SerializedArray(metadata, value, header, depth);
   }
 
   private final VariantMetadata metadata;
   private final ByteBuffer value;
   private final int offsetSize;
   private final int offsetListOffset;
   private final int dataOffset;
+  private final int depth;
   private final VariantValue[] array;
 
-  private SerializedArray(VariantMetadata metadata, ByteBuffer value, int header) {
+  private SerializedArray(VariantMetadata metadata, ByteBuffer value, int header, int depth) {
     this.metadata = metadata;
     this.value = value;
+    this.depth = depth;
     this.offsetSize = 1 + ((header & OFFSET_SIZE_MASK) >> OFFSET_SIZE_SHIFT);
     int numElementsSize = ((header & IS_LARGE) == IS_LARGE) ? 4 : 1;
+    Preconditions.checkArgument(
+        value.remaining() >= HEADER_SIZE + numElementsSize,
+        "Invalid variant array: buffer too small for element count field");
     int numElements = ByteBuffers.readLittleEndianUnsigned(value, HEADER_SIZE, numElementsSize);
+    Preconditions.checkArgument(
+        numElements >= 0, "Invalid variant array: negative element count %s", numElements);
+    Preconditions.checkArgument(
+        numElements <= VariantUtil.MAX_ELEMENTS,
+        "Invalid variant array: element count %s exceeds maximum %s",
+        numElements,
+        VariantUtil.MAX_ELEMENTS);
     this.offsetListOffset = HEADER_SIZE + numElementsSize;
-    this.dataOffset = offsetListOffset + ((1 + numElements) * offsetSize);
+    long offsetTableEnd = (long) offsetListOffset + ((long) numElements + 1L) * offsetSize;
+    Preconditions.checkArgument(
+        offsetTableEnd <= value.remaining(),
+        "Invalid variant array: element count %s exceeds buffer",
+        numElements);
+    this.dataOffset = Math.toIntExact(offsetTableEnd);
     this.array = new VariantValue[numElements];
   }
 
@@ -76,8 +97,16 @@ public VariantValue get(int index) {
       int next =
           ByteBuffers.readLittleEndianUnsigned(
               value, offsetListOffset + (offsetSize * (1 + index)), offsetSize);
+      long dataLen = value.remaining() - (long) dataOffset;
+      Preconditions.checkArgument(
+          offset >= 0 && next >= offset && next <= dataLen,
+          "Invalid variant array: offset range [%s, %s] out of data region [0, %s]",
+          offset,
+          next,
+          dataLen);
       array[index] =
-          VariantValue.from(metadata, VariantUtil.slice(value, dataOffset + offset, next - offset));
+          VariantUtil.fromBuffer(
+              metadata, VariantUtil.slice(value, dataOffset + offset, next - offset), depth + 1);
     }
     return array[index];
   }

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

@@ -59,15 +59,37 @@ static SerializedMetadata from(ByteBuffer metadata) {
   private SerializedMetadata(ByteBuffer metadata, int header) {
     this.isSorted = (header & SORTED_STRINGS) == SORTED_STRINGS;
     this.offsetSize = 1 + ((header & OFFSET_SIZE_MASK) >> OFFSET_SIZE_SHIFT);
+    Preconditions.checkArgument(
+        metadata.remaining() >= HEADER_SIZE + offsetSize,
+        "Invalid variant metadata: buffer too small for dictionary size field");
     int dictSize = ByteBuffers.readLittleEndianUnsigned(metadata, HEADER_SIZE, offsetSize);
-    this.dict = new String[dictSize];
+    Preconditions.checkArgument(
+        dictSize >= 0, "Invalid variant metadata: negative dictionary size %s", dictSize);
+    Preconditions.checkArgument(
+        dictSize <= VariantUtil.MAX_ELEMENTS,
+        "Invalid variant metadata: dictionary size %s exceeds maximum %s",
+        dictSize,
+        VariantUtil.MAX_ELEMENTS);
     this.offsetListOffset = HEADER_SIZE + offsetSize;
-    this.dataOffset = offsetListOffset + ((1 + dictSize) * offsetSize);
-    int endOffset =
-        dataOffset
-            + ByteBuffers.readLittleEndianUnsigned(
-                metadata, offsetListOffset + (offsetSize * dictSize), offsetSize);
-    if (endOffset < metadata.limit()) {
+    long offsetTableEnd = (long) offsetListOffset + ((long) dictSize + 1L) * offsetSize;
+    Preconditions.checkArgument(
+        offsetTableEnd <= metadata.remaining(),
+        "Invalid variant metadata: dictionary size %s exceeds buffer",
+        dictSize);
+    this.dict = new String[dictSize];
+    this.dataOffset = Math.toIntExact(offsetTableEnd);
+    int lastOffset =
+        ByteBuffers.readLittleEndianUnsigned(
+            metadata, offsetListOffset + (offsetSize * dictSize), offsetSize);
+    Preconditions.checkArgument(
+        lastOffset >= 0, "Invalid variant metadata: negative end offset %s", lastOffset);
+    long endOffsetLong = (long) dataOffset + lastOffset;
+    Preconditions.checkArgument(
+        endOffsetLong <= metadata.remaining(),
+        "Invalid variant metadata: end offset %s exceeds buffer",
+        endOffsetLong);
+    int endOffset = (int) endOffsetLong;
+    if (endOffset < metadata.remaining()) {
       this.metadata = VariantUtil.slice(metadata, 0, endOffset);
     } else {
       this.metadata = metadata;
@@ -112,6 +134,12 @@ public String get(int index) {
       int next =
           ByteBuffers.readLittleEndianUnsigned(
               metadata, offsetListOffset + (offsetSize * (1 + index)), offsetSize);
+      Preconditions.checkArgument(
+          offset >= 0 && next >= offset && (long) dataOffset + next <= metadata.remaining(),
+          "Invalid variant metadata: dict entry %s offset range [%s, %s] invalid",
+          index,
+          offset,
+          next);
       dict[index] = VariantUtil.readString(metadata, dataOffset + offset, next - offset);
     }
     return dict[index];

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

@@ -42,12 +42,16 @@ static SerializedObject from(VariantMetadata metadata, byte[] bytes) {
   }
 
   static SerializedObject from(VariantMetadata metadata, ByteBuffer value, int header) {
+    return from(metadata, value, header, 0);
+  }
+
+  static SerializedObject from(VariantMetadata metadata, ByteBuffer value, int header, int depth) {
     Preconditions.checkArgument(
         value.order() == ByteOrder.LITTLE_ENDIAN, "Unsupported byte order: big endian");
     BasicType basicType = VariantUtil.basicType(header);
     Preconditions.checkArgument(
         basicType == BasicType.OBJECT, "Invalid object, basic type: " + basicType);
-    return new SerializedObject(metadata, value, header);
+    return new SerializedObject(metadata, value, header, depth);
   }
 
   private final VariantMetadata metadata;
@@ -60,21 +64,42 @@ static SerializedObject from(VariantMetadata metadata, ByteBuffer value, int hea
   private final int[] offsets;
   private final int[] lengths;
   private final int dataOffset;
+  private final int depth;
   private final VariantValue[] values;
 
-  private SerializedObject(VariantMetadata metadata, ByteBuffer value, int header) {
+  private SerializedObject(VariantMetadata metadata, ByteBuffer value, int header, int depth) {
     this.metadata = metadata;
     this.value = value;
+    this.depth = depth;
     this.offsetSize = 1 + ((header & OFFSET_SIZE_MASK) >> OFFSET_SIZE_SHIFT);
     this.fieldIdSize = 1 + ((header & FIELD_ID_SIZE_MASK) >> FIELD_ID_SIZE_SHIFT);
     int numElementsSize = ((header & IS_LARGE) == IS_LARGE) ? 4 : 1;
+    Preconditions.checkArgument(
+        value.remaining() >= HEADER_SIZE + numElementsSize,
+        "Invalid variant object: buffer too small for element count field");
     int numElements = ByteBuffers.readLittleEndianUnsigned(value, HEADER_SIZE, numElementsSize);
+    Preconditions.checkArgument(
+        numElements >= 0, "Invalid variant object: negative element count %s", numElements);
+    Preconditions.checkArgument(
+        numElements <= VariantUtil.MAX_ELEMENTS,
+        "Invalid variant object: element count %s exceeds maximum %s",
+        numElements,
+        VariantUtil.MAX_ELEMENTS);
     this.fieldIdListOffset = HEADER_SIZE + numElementsSize;
+    long dataStart =
+        (long) fieldIdListOffset
+            + (long) numElements * fieldIdSize
+            + ((long) numElements + 1L) * offsetSize;
+    Preconditions.checkArgument(
+        dataStart <= value.remaining(),
+        "Invalid variant object: element count %s exceeds buffer",
+        numElements);
+    this.offsetListOffset =
+        Math.toIntExact((long) fieldIdListOffset + (long) numElements * fieldIdSize);
     this.fieldIds = new Integer[numElements];
-    this.offsetListOffset = fieldIdListOffset + (numElements * fieldIdSize);
     this.offsets = new int[numElements];
     this.lengths = new int[numElements];
-    this.dataOffset = offsetListOffset + ((1 + numElements) * offsetSize);
+    this.dataOffset = Math.toIntExact(dataStart);
     this.values = new VariantValue[numElements];
 
     if (numElements > 0) {
@@ -96,14 +121,28 @@ private void initOffsetsAndLengths(int numElements) {
     int dataLength =
         ByteBuffers.readLittleEndianUnsigned(
             value, offsetListOffset + (numElements * offsetSize), offsetSize);
+    long dataLen = value.remaining() - (long) dataOffset;
+    Preconditions.checkArgument(
+        dataLength >= 0 && dataLength <= dataLen,
+        "Invalid variant object: data length %s out of data region [0, %s]",
+        dataLength,
+        dataLen);
+    for (int index = 0; index < numElements; index += 1) {
+      Preconditions.checkArgument(
+          offsets[index] >= 0 && offsets[index] <= dataLength,
+          "Invalid variant object: offset %s out of declared data length %s",
+          offsets[index],
+          dataLength);
+    }
     offsetToLength.put(dataLength, 0);
 
     // populate lengths list by sorting offsets
     List<Integer> sortedOffsets =
         offsetToLength.keySet().stream().sorted().collect(Collectors.toList());
-    for (int index = 0; index < numElements; index += 1) {
-      int offset = sortedOffsets.get(index);
-      int length = sortedOffsets.get(index + 1) - offset;
+    // shared offsets are spec-legal (zero-length spans); do not require uniqueness
+    for (int i = 0; i < sortedOffsets.size() - 1; i += 1) {
+      int offset = sortedOffsets.get(i);
+      int length = sortedOffsets.get(i + 1) - offset;
       offsetToLength.put(offset, length);
     }
 
@@ -161,11 +200,19 @@ public String next() {
         };
   }
 
+  // Field id range is validated lazily on first access, not at construction.
   private int id(int index) {
     if (null == fieldIds[index]) {
-      fieldIds[index] =
+      int dictSize = metadata.dictionarySize();
+      int id =
           ByteBuffers.readLittleEndianUnsigned(
               value, fieldIdListOffset + (index * fieldIdSize), fieldIdSize);
+      Preconditions.checkArgument(
+          id >= 0 && id < dictSize,
+          "Invalid variant object: field id %s out of range [0, %s)",
+          id,
+          dictSize);
+      fieldIds[index] = id;
     }
 
     return fieldIds[index];
@@ -182,8 +229,10 @@ public VariantValue get(String name) {
 
     if (null == values[index]) {
       values[index] =
-          VariantValue.from(
-              metadata, VariantUtil.slice(value, dataOffset + offsets[index], lengths[index]));
+          VariantUtil.fromBuffer(
+              metadata,
+              VariantUtil.slice(value, dataOffset + offsets[index], lengths[index]),
+              depth + 1);
     }
 
     return values[index];

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

@@ -51,6 +51,56 @@ static SerializedPrimitive from(ByteBuffer value, int header) {
   private SerializedPrimitive(ByteBuffer value, int header) {
     this.value = value;
     this.type = PhysicalType.from(header >> PRIMITIVE_TYPE_SHIFT);
+    long requiredBytes = PRIMITIVE_OFFSET + payloadSize(type, value);
+    Preconditions.checkArgument(
+        requiredBytes <= value.remaining(),
+        "Invalid variant primitive: %s payload extends past buffer",
+        type);
+  }
+
+  private static long payloadSize(PhysicalType type, ByteBuffer value) {
+    switch (type) {
+      case NULL:
+      case BOOLEAN_TRUE:
+      case BOOLEAN_FALSE:
+        return 0;
+      case INT8:
+        return 1;
+      case INT16:
+        return 2;
+      case INT32:
+      case DATE:
+      case FLOAT:
+        return 4;
+      case INT64:
+      case TIMESTAMPTZ:
+      case TIMESTAMPNTZ:
+      case TIME:
+      case TIMESTAMPTZ_NANOS:
+      case TIMESTAMPNTZ_NANOS:
+      case DOUBLE:
+        return 8;
+      case DECIMAL4:
+        return 5;
+      case DECIMAL8:
+        return 9;
+      case DECIMAL16:
+        return 17;
+      case UUID:
+        return 16;
+      case BINARY:
+      case STRING:
+        Preconditions.checkArgument(
+            PRIMITIVE_OFFSET + 4 <= value.remaining(),
+            "Invalid variant primitive: %s size field extends past buffer",
+            type);
+        int size = ByteBuffers.readLittleEndianInt32(value, PRIMITIVE_OFFSET);
+        Preconditions.checkArgument(
+            size >= 0, "Invalid variant primitive: negative %s size %s", type, size);
+        return 4L + size;
+    }
+
+    throw new UnsupportedOperationException("Unsupported primitive type: " + type);
   }
 
   private Object read() {

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

@@ -47,6 +47,10 @@ static SerializedShortString from(ByteBuffer value, int header) {
   private SerializedShortString(ByteBuffer value, int header) {
     this.value = value;
     this.length = ((header & LENGTH_MASK) >> LENGTH_SHIFT);
+    Preconditions.checkArgument(
+        HEADER_SIZE + length <= value.remaining(),
+        "Invalid variant short string: length %s exceeds buffer",
+        length);
   }
 
   @Override

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

@@ -22,6 +22,8 @@
 import java.nio.ByteOrder;
 import java.nio.charset.StandardCharsets;
 import java.util.function.Function;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.ByteBuffers;
 
 class VariantUtil {
   private static final int BASIC_TYPE_MASK = 0b11;
@@ -30,8 +32,45 @@ class VariantUtil {
   private static final int BASIC_TYPE_OBJECT = 2;
   private static final int BASIC_TYPE_ARRAY = 3;
 
+  /**
+   * Maximum nesting depth in a Variant (permitted depths 0..MAX_VARIANT_DEPTH). Safety limit, not a
+   * spec bound. Matches parquet-java (apache/parquet-java#3562).
+   */
+  static final int MAX_VARIANT_DEPTH = 1000;
+
+  /**
+   * Maximum element count for Variant containers and metadata dictionaries. Safety limit against
+   * buffer-to-heap allocation amplification.
+   */
+  static final int MAX_ELEMENTS = 16_777_216;
+
   private VariantUtil() {}
 
+  /** Parses a variant value; validates input and enforces {@link #MAX_VARIANT_DEPTH}. */
+  static VariantValue fromBuffer(VariantMetadata metadata, ByteBuffer value, int depth) {
+    Preconditions.checkArgument(depth >= 0, "Invalid variant: negative depth %s", depth);
+    Preconditions.checkArgument(
+        depth <= MAX_VARIANT_DEPTH,
+        "Invalid variant: nesting depth %s exceeds maximum %s",
+        depth,
+        MAX_VARIANT_DEPTH);
+    Preconditions.checkArgument(value.remaining() >= 1, "Invalid variant: empty value buffer");
+    int header = ByteBuffers.readByte(value, 0);
+    BasicType basicType = basicType(header);
+    switch (basicType) {
+      case PRIMITIVE:
+        return SerializedPrimitive.from(value, header);
+      case SHORT_STRING:
+        return SerializedShortString.from(value, header);
+      case OBJECT:
+        return SerializedObject.from(metadata, value, header, depth);
+      case ARRAY:
+        return SerializedArray.from(metadata, value, header, depth);
+    }
+
+    throw new UnsupportedOperationException("Unsupported basic type: " + basicType);
+  }
+
   static float readFloat(ByteBuffer buffer, int offset) {
     return buffer.getFloat(buffer.position() + offset);
   }