GH-3471: Fix ByteBuffer handling in VariantUtil and VariantBuilder

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

@@ -374,7 +374,7 @@ public void appendBinary(ByteBuffer binary) {
     writePos += 1;
     VariantUtil.writeLong(writeBuffer, writePos, binarySize, VariantUtil.U32_SIZE);
     writePos += VariantUtil.U32_SIZE;
-    ByteBuffer.wrap(writeBuffer, writePos, binarySize).put(binary);
+    ByteBuffer.wrap(writeBuffer, writePos, binarySize).put(binary.duplicate());
     writePos += binarySize;
   }
 

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

@@ -843,7 +843,7 @@ static HashMap<String, Integer> getMetadataMap(ByteBuffer metadata) {
       } else {
         // ByteBuffer does not have an array, so we need to use the `get` method to read the bytes.
         byte[] metadataArray = new byte[nextOffset - offset];
-        slice(metadata, stringStart + offset).get(metadataArray);
+        slice(metadata, pos + stringStart + offset).get(metadataArray);
         result.put(new String(metadataArray), id);
       }
       offset = nextOffset;

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

@@ -341,4 +341,39 @@ public void testInvalidObject() {
       Assert.assertEquals("Cannot read ARRAY value as OBJECT", e.getMessage());
     }
   }
+
+  @Test
+  public void testMetadataWithNonZeroPositionReadOnly() {
+    // Build a variant with object fields to populate the metadata dictionary
+    VariantBuilder vb = new VariantBuilder();
+    VariantObjectBuilder obj = vb.startObject();
+    obj.appendKey("name");
+    obj.appendString("Alice");
+    obj.appendKey("age");
+    obj.appendInt(30);
+    vb.endObject();
+    Variant variant = vb.build();
+
+    // Get the raw metadata bytes
+    ByteBuffer metaBuf = variant.getMetadataBuffer();
+    byte[] metaBytes = new byte[metaBuf.remaining()];
+    metaBuf.duplicate().get(metaBytes);
+
+    // Embed in a larger buffer with a non-zero position, then make read-only
+    // to force the else-branch in getMetadataMap.
+    byte[] padded = new byte[10 + metaBytes.length];
+    System.arraycopy(metaBytes, 0, padded, 10, metaBytes.length);
+    ByteBuffer offsetMetadata = ByteBuffer.wrap(padded);
+    offsetMetadata.position(10);
+    offsetMetadata.limit(10 + metaBytes.length);
+    offsetMetadata = offsetMetadata.asReadOnlyBuffer();
+
+    // ImmutableMetadata calls getMetadataMap, which had the bug.
+    // getMetadataMap builds a key->id dictionary from the metadata buffer.
+    // With a non-zero position and read-only buffer, the else-branch is taken,
+    // which previously used the wrong offset.
+    ImmutableMetadata immutableMetadata = new ImmutableMetadata(offsetMetadata);
+    Assert.assertEquals(0, immutableMetadata.getOrInsert("name"));
+    Assert.assertEquals(1, immutableMetadata.getOrInsert("age"));
+  }
 }

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

@@ -387,6 +387,17 @@ public void testBinaryBuilder() {
     }
   }
 
+  @Test
+  public void testBinaryBuilderDoesNotMutateCallerBuffer() {
+    ByteBuffer buf = ByteBuffer.wrap(new byte[] {0, 1, 2, 3});
+    int positionBefore = buf.position();
+    int remainingBefore = buf.remaining();
+    VariantBuilder vb = new VariantBuilder();
+    vb.appendBinary(buf);
+    Assert.assertEquals(positionBefore, buf.position());
+    Assert.assertEquals(remainingBefore, buf.remaining());
+  }
+
   @Test
   public void testStringBuilder() {
     IntStream.range(VariantUtil.MAX_SHORT_STR_SIZE - 3, VariantUtil.MAX_SHORT_STR_SIZE + 3)