GH-3315: Variant binary read does not take length into account (#3333)

Author: jerolba

parquet-variant/src/main/java/org/apache/parquet/variant/VariantUtil.java

@@ -612,7 +612,9 @@ static ByteBuffer getBinary(ByteBuffer value) {
     int start = value.position() + 1 + U32_SIZE;
     int length = readUnsigned(value, value.position() + 1, U32_SIZE);
     checkIndex(start + length - 1, value.limit());
-    return slice(value, start);
+    ByteBuffer result = slice(value, start);
+    result.limit(start + length);
+    return result;
   }
 
   static String getString(ByteBuffer value) {

parquet-variant/src/test/java/org/apache/parquet/variant/TestVariantObjectBuilder.java

@@ -18,6 +18,7 @@
  */
 package org.apache.parquet.variant;
 
+import java.nio.ByteBuffer;
 import org.junit.Assert;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -115,6 +116,41 @@ public void testMixedObjectBuilder() {
     });
   }
 
+  @Test
+  public void testMixedBinaryBuilder() {
+    VariantBuilder b = new VariantBuilder();
+    VariantObjectBuilder objBuilder = b.startObject();
+    objBuilder.appendKey("as_binary");
+    objBuilder.appendBinary(ByteBuffer.wrap(new byte[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9}));
+    objBuilder.appendKey("in_array");
+    VariantArrayBuilder arrBinary = objBuilder.startArray();
+    arrBinary.appendBinary(ByteBuffer.wrap(new byte[] {}));
+    arrBinary.appendBinary(ByteBuffer.wrap(new byte[] {10, 11, 12, 13, 14, 15, 16}));
+    arrBinary.appendBinary(ByteBuffer.wrap(new byte[] {17, 18}));
+    objBuilder.endArray();
+    b.endObject();
+
+    VariantTestUtil.testVariant(b.build(), v -> {
+      VariantTestUtil.checkType(v, VariantUtil.OBJECT, Variant.Type.OBJECT);
+      Assert.assertEquals(2, v.numObjectElements());
+      Assert.assertEquals(
+          ByteBuffer.wrap(new byte[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9}),
+          v.getFieldByKey("as_binary").getBinary());
+      Variant nestedArray = v.getFieldByKey("in_array");
+      VariantTestUtil.checkType(nestedArray, VariantUtil.ARRAY, Variant.Type.ARRAY);
+      Assert.assertEquals(3, nestedArray.numArrayElements());
+      Assert.assertEquals(
+          ByteBuffer.wrap(new byte[] {}),
+          nestedArray.getElementAtIndex(0).getBinary());
+      Assert.assertEquals(
+          ByteBuffer.wrap(new byte[] {10, 11, 12, 13, 14, 15, 16}),
+          nestedArray.getElementAtIndex(1).getBinary());
+      Assert.assertEquals(
+          ByteBuffer.wrap(new byte[] {17, 18}),
+          nestedArray.getElementAtIndex(2).getBinary());
+    });
+  }
+
   private void buildNested(int i, VariantObjectBuilder obj) {
     if (i > 0) {
       obj.appendKey("key" + i);