Parquet: Fix the limit handling in ParquetValueReaders#BytesReader (a…

…pache#3646)
ZhuoRoger · Dec 7, 2021 · 2511de8 · 2511de8
1 parent a0ca990
commit 2511de8
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 1 deletion.
diff --git a/data/src/test/java/org/apache/iceberg/data/parquet/TestGenericData.java b/data/src/test/java/org/apache/iceberg/data/parquet/TestGenericData.java
@@ -26,9 +26,11 @@
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.data.DataTest;
 import org.apache.iceberg.data.DataTestHelpers;
+import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.RandomGenericData;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.parquet.Parquet;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -60,5 +62,20 @@ protected void writeAndValidate(Schema schema) throws IOException {
     for (int i = 0; i < expected.size(); i += 1) {
       DataTestHelpers.assertEquals(schema.asStruct(), expected.get(i), rows.get(i));
     }
+
+    // test reuseContainers
+    try (CloseableIterable<Record> reader = Parquet.read(Files.localInput(testFile))
+            .project(schema)
+            .reuseContainers()
+            .createReaderFunc(fileSchema -> GenericParquetReaders.buildReader(schema, fileSchema))
+            .build()) {
+      CloseableIterator it = reader.iterator();
+      int idx = 0;
+      while (it.hasNext()) {
+        GenericRecord actualRecord = (GenericRecord) it.next();
+        DataTestHelpers.assertEquals(schema.asStruct(), expected.get(idx), actualRecord);
+        idx++;
+      }
+    }
   }
 }
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java
@@ -330,7 +330,7 @@ public ByteBuffer read(ByteBuffer reuse) {
       if (reuse != null && reuse.hasArray() && reuse.capacity() >= data.remaining()) {
         data.get(reuse.array(), reuse.arrayOffset(), data.remaining());
         reuse.position(0);
-        reuse.limit(data.remaining());
+        reuse.limit(binary.length());
         return reuse;
       } else {
         byte[] array = new byte[data.remaining()];