diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java index ebb1208dc0..6bc5e89117 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java @@ -65,6 +65,7 @@ import org.apache.parquet.HadoopReadOptions; import org.apache.parquet.ParquetReadOptions; import org.apache.parquet.Preconditions; +import org.apache.parquet.bytes.ByteBufferAllocator; import org.apache.parquet.bytes.ByteBufferInputStream; import org.apache.parquet.bytes.ByteBufferReleaser; import org.apache.parquet.bytes.BytesInput; @@ -1378,8 +1379,30 @@ private void readVectored(List allParts, ChunkListBuilder b totalSize += len; } LOG.debug("Reading {} bytes of data with vectored IO in {} ranges", totalSize, ranges.size()); - // Request a vectored read; - f.readVectored(ranges, options.getAllocator()); + // ChecksumFileSystem may allocate internal checksum buffers and return slices of the data buffers. + // Capture every original allocation so the row group can release the actual allocator-owned buffers. + List allocatedBuffers = new ArrayList<>(); + ByteBufferAllocator capturingAllocator = new ByteBufferAllocator() { + @Override + public ByteBuffer allocate(int size) { + ByteBuffer buffer = options.getAllocator().allocate(size); + allocatedBuffers.add(buffer); + return buffer; + } + + @Override + public void release(ByteBuffer buffer) { + options.getAllocator().release(buffer); + } + + @Override + public boolean isDirect() { + return options.getAllocator().isDirect(); + } + }; + // Request a vectored read. + f.readVectored(ranges, capturingAllocator); + builder.addBuffersToRelease(allocatedBuffers); int k = 0; for (ConsecutivePartList consecutivePart : allParts) { ParquetFileRange currRange = ranges.get(k++); diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileReaderBufferLeak.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileReaderBufferLeak.java index a834237bab..a6178a4b79 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileReaderBufferLeak.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileReaderBufferLeak.java @@ -169,6 +169,29 @@ public void testCloseReleasesCurrentRowGroupBuffers() throws Exception { } } + /** + * Verify that closing a reader after a vectored read releases all buffers allocated by the file system. + */ + @Test + public void testClosingVectoredReaderReleasesAllBuffers() throws Exception { + Path path = writeMultiRowGroupFile(500); + + try (TrackingByteBufferAllocator readAllocator = + TrackingByteBufferAllocator.wrap(new HeapByteBufferAllocator())) { + ParquetReadOptions options = ParquetReadOptions.builder() + .withAllocator(readAllocator) + .withUseHadoopVectoredIo(true) + .build(); + InputFile inputFile = HadoopInputFile.fromPath(path, CONF); + + try (ParquetFileReader reader = new ParquetFileReader(inputFile, options)) { + PageReadStore pages = reader.readNextRowGroup(); + assertNotNull(pages); + assertTrue(pages.getRowCount() > 0); + } + } + } + /** * Verify that readNextFilteredRowGroup() releases buffers of the previous row group * when the filter does not trigger column-index filtering (falls back to readNextRowGroup). diff --git a/pom.xml b/pom.xml index ec53d3d721..f9fba8ac71 100644 --- a/pom.xml +++ b/pom.xml @@ -83,7 +83,7 @@ 3.8.0 shaded.parquet - 3.3.0 + 3.3.6 1.18.0 thrift ${thrift.executable}