This is an automated email from the ASF dual-hosted git repository.
Fokko pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/parquet-java.git
The following commit(s) were added to refs/heads/master by this push:
new 8e30c4cee GH-3530: Eagerly release column buffers (#3571)
8e30c4cee is described below
commit 8e30c4cee3c7e85a8cf2133697f13138509b05b7
Author: Ismaël Mejía <[email protected]>
AuthorDate: Tue Aug 11 22:36:56 2026 +0200
GH-3530: Eagerly release column buffers (#3571)
* GH-3530: Eagerly release column buffers during row group flush
Release each column's compressed page buffers immediately after writing
to disk in flushToFileWriter(), rather than holding all buffers until the
entire flush completes. This is correct resource management that makes
buffers GC-eligible sooner, though benchmarking with a PeakTrackingAllocator
confirms it does not reduce peak memory: the peak is reached during the
write phase (as pages are compressed), not during flush.
Changes:
- Call pageWriter.close() after each column in flushToFileWriter()
- Add writeAllToAndRelease() to ConcatenatingByteBufferCollector for
progressive slab-by-slab memory release during write
- Make close() idempotent (safe to call after eager release or double-close)
- Add RowGroupFlushBenchmark (20-column wide schema, PeakTrackingAllocator)
and BlackHoleOutputFile for measuring flush performance and peak memory
- Add tests for eager release, double-close safety, and output equivalence
* Address review feedback: try-with-resources and progressive release in
writeAllTo
- Merge writeAllToAndRelease into writeAllTo so progressive buffer
release is used in production (via ParquetFileWriter.writeColumnChunk).
- Add comment explaining why the WritableByteChannel is intentionally
not closed (closing it would close the caller's OutputStream).
- Use try-with-resources in flushToFileWriter for idiomatic cleanup.
- Capture buf.size() before write in writeToFileWriter since writeAllTo
now releases buffers progressively.
---
.../parquet/benchmarks/BlackHoleOutputFile.java | 76 ++++++++
.../parquet/benchmarks/RowGroupFlushBenchmark.java | 191 +++++++++++++++++++++
.../bytes/ConcatenatingByteBufferCollector.java | 29 +++-
.../TestConcatenatingByteBufferCollector.java | 68 ++++++++
.../parquet/hadoop/ColumnChunkPageWriteStore.java | 12 +-
5 files changed, 371 insertions(+), 5 deletions(-)
diff --git
a/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/BlackHoleOutputFile.java
b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/BlackHoleOutputFile.java
new file mode 100644
index 000000000..690ddc2bb
--- /dev/null
+++
b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/BlackHoleOutputFile.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.benchmarks;
+
+import java.io.IOException;
+import org.apache.parquet.io.OutputFile;
+import org.apache.parquet.io.PositionOutputStream;
+
+/**
+ * A no-op {@link OutputFile} that discards all written data.
+ * Useful for isolating CPU/encoding cost from filesystem I/O in write
benchmarks.
+ */
+public final class BlackHoleOutputFile implements OutputFile {
+
+ public static final BlackHoleOutputFile INSTANCE = new BlackHoleOutputFile();
+
+ private BlackHoleOutputFile() {}
+
+ @Override
+ public boolean supportsBlockSize() {
+ return false;
+ }
+
+ @Override
+ public long defaultBlockSize() {
+ return -1L;
+ }
+
+ @Override
+ public PositionOutputStream createOrOverwrite(long blockSizeHint) {
+ return create(blockSizeHint);
+ }
+
+ @Override
+ public PositionOutputStream create(long blockSizeHint) {
+ return new PositionOutputStream() {
+ private long pos;
+
+ @Override
+ public long getPos() throws IOException {
+ return pos;
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ ++pos;
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ pos += len;
+ }
+ };
+ }
+
+ @Override
+ public String getPath() {
+ return "/dev/null";
+ }
+}
diff --git
a/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/RowGroupFlushBenchmark.java
b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/RowGroupFlushBenchmark.java
new file mode 100644
index 000000000..9bc5cab0a
--- /dev/null
+++
b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/RowGroupFlushBenchmark.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.benchmarks;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.parquet.bytes.ByteBufferAllocator;
+import org.apache.parquet.bytes.HeapByteBufferAllocator;
+import org.apache.parquet.column.ParquetProperties.WriterVersion;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetFileWriter;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Types;
+import org.openjdk.jmh.annotations.AuxCounters;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+/**
+ * Benchmark measuring row group flush performance and peak buffer memory.
+ *
+ * <p>Uses a wide schema (20 BINARY columns, 200 bytes each) to produce
+ * substantial per-column page buffers. A {@link PeakTrackingAllocator}
+ * wraps the heap allocator to precisely track the peak bytes outstanding
+ * across all parquet-managed ByteBuffers (independent of JVM GC behavior).
+ *
+ * <p>The key metric is {@code peakAllocatorMB}: with the interleaved flush
+ * optimization, each column's pages are finalized, written, and released
+ * before the next column is processed, so peak buffer memory is roughly
+ * 1/N of the total row group size (N = number of columns).
+ *
+ * <p>Writes to {@link BlackHoleOutputFile} to isolate flush cost from
+ * filesystem I/O.
+ */
+@BenchmarkMode({Mode.AverageTime})
+@Fork(
+ value = 1,
+ jvmArgs = {"-Xms512m", "-Xmx1g"})
+@Warmup(iterations = 3)
+@Measurement(iterations = 5)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@State(Scope.Thread)
+public class RowGroupFlushBenchmark {
+
+ private static final int COLUMN_COUNT = 20;
+ private static final int BINARY_VALUE_LENGTH = 200;
+ private static final int ROW_COUNT = 100_000;
+
+ /** Row group sizes: 8MB and 64MB. */
+ @Param({"8388608", "67108864"})
+ public int rowGroupSize;
+
+ /** Wide schema: 20 required BINARY columns. */
+ private static final MessageType WIDE_SCHEMA;
+
+ static {
+ Types.MessageTypeBuilder builder = Types.buildMessage();
+ for (int c = 0; c < COLUMN_COUNT; c++) {
+ builder.required(PrimitiveTypeName.BINARY).named("col_" + c);
+ }
+ WIDE_SCHEMA = builder.named("wide_record");
+ }
+
+ /** Pre-generated column values (one unique value per column). */
+ private Binary[] columnValues;
+
+ @Setup(Level.Trial)
+ public void setup() {
+ Random random = new Random(42);
+ columnValues = new Binary[COLUMN_COUNT];
+ for (int c = 0; c < COLUMN_COUNT; c++) {
+ byte[] value = new byte[BINARY_VALUE_LENGTH];
+ random.nextBytes(value);
+ columnValues[c] = Binary.fromConstantByteArray(value);
+ }
+ }
+
+ /**
+ * Auxiliary counters reported alongside timing. JMH collects these after
+ * each iteration.
+ */
+ @AuxCounters(AuxCounters.Type.EVENTS)
+ @State(Scope.Thread)
+ public static class MemoryCounters {
+ /** Peak bytes outstanding in the parquet ByteBufferAllocator. */
+ public long peakAllocatorBytes;
+
+ /** Convenience: peak in MB (peakAllocatorBytes / 1048576). */
+ public double peakAllocatorMB;
+
+ @Setup(Level.Iteration)
+ public void reset() {
+ peakAllocatorBytes = 0;
+ peakAllocatorMB = 0;
+ }
+ }
+
+ /**
+ * ByteBufferAllocator wrapper that tracks current and peak allocated bytes.
+ * Thread-safe (uses AtomicLong) although the write path is single-threaded.
+ */
+ static class PeakTrackingAllocator implements ByteBufferAllocator {
+ private final ByteBufferAllocator delegate = new HeapByteBufferAllocator();
+ private final AtomicLong currentBytes = new AtomicLong();
+ private final AtomicLong peakBytes = new AtomicLong();
+
+ @Override
+ public ByteBuffer allocate(int size) {
+ ByteBuffer buf = delegate.allocate(size);
+ long current = currentBytes.addAndGet(buf.capacity());
+ peakBytes.accumulateAndGet(current, Math::max);
+ return buf;
+ }
+
+ @Override
+ public void release(ByteBuffer buf) {
+ currentBytes.addAndGet(-buf.capacity());
+ delegate.release(buf);
+ }
+
+ @Override
+ public boolean isDirect() {
+ return delegate.isDirect();
+ }
+
+ long getPeakBytes() {
+ return peakBytes.get();
+ }
+ }
+
+ @Benchmark
+ public void writeWithFlush(MemoryCounters counters) throws IOException {
+ PeakTrackingAllocator allocator = new PeakTrackingAllocator();
+ SimpleGroupFactory factory = new SimpleGroupFactory(WIDE_SCHEMA);
+
+ try (ParquetWriter<Group> writer =
ExampleParquetWriter.builder(BlackHoleOutputFile.INSTANCE)
+ .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
+ .withType(WIDE_SCHEMA)
+ .withCompressionCodec(CompressionCodecName.UNCOMPRESSED)
+ .withWriterVersion(WriterVersion.PARQUET_1_0)
+ .withRowGroupSize(rowGroupSize)
+ .withDictionaryEncoding(false)
+ .withAllocator(allocator)
+ .build()) {
+ for (int i = 0; i < ROW_COUNT; i++) {
+ Group group = factory.newGroup();
+ for (int c = 0; c < COLUMN_COUNT; c++) {
+ group.append("col_" + c, columnValues[c]);
+ }
+ writer.write(group);
+ }
+ }
+
+ counters.peakAllocatorBytes = allocator.getPeakBytes();
+ counters.peakAllocatorMB = allocator.getPeakBytes() / (1024.0 * 1024.0);
+ }
+}
diff --git
a/parquet-common/src/main/java/org/apache/parquet/bytes/ConcatenatingByteBufferCollector.java
b/parquet-common/src/main/java/org/apache/parquet/bytes/ConcatenatingByteBufferCollector.java
index 7a616e9b9..982a8e370 100644
---
a/parquet-common/src/main/java/org/apache/parquet/bytes/ConcatenatingByteBufferCollector.java
+++
b/parquet-common/src/main/java/org/apache/parquet/bytes/ConcatenatingByteBufferCollector.java
@@ -26,6 +26,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
/**
@@ -64,18 +65,42 @@ public class ConcatenatingByteBufferCollector extends
BytesInput implements Auto
@Override
public void close() {
+ if (slabs.isEmpty()) {
+ return;
+ }
for (ByteBuffer slab : slabs) {
allocator.release(slab);
}
slabs.clear();
+ size = 0;
}
+ /**
+ * Writes all collected slabs to the given output stream, releasing each
slab's
+ * {@link ByteBuffer} back to the allocator immediately after it has been
written.
+ * This progressively frees memory during the write rather than holding all
slabs
+ * until {@link #close()} is called.
+ *
+ * <p>After this method returns, the collector is empty and {@link #size()}
returns 0.
+ * Calling {@link #close()} afterwards is safe but has no additional effect.
+ *
+ * @param out the output stream to write to
+ * @throws IOException if an I/O error occurs
+ */
@Override
public void writeAllTo(OutputStream out) throws IOException {
+ // The channel is intentionally not closed: closing it would close the
underlying
+ // OutputStream which is owned by the caller. The channel is a stateless
wrapper
+ // that holds no independent resources.
WritableByteChannel channel = Channels.newChannel(out);
- for (ByteBuffer buffer : slabs) {
- channel.write(buffer.duplicate());
+ Iterator<ByteBuffer> it = slabs.iterator();
+ while (it.hasNext()) {
+ ByteBuffer slab = it.next();
+ channel.write(slab.duplicate());
+ allocator.release(slab);
+ it.remove();
}
+ size = 0;
}
@Override
diff --git
a/parquet-common/src/test/java/org/apache/parquet/bytes/TestConcatenatingByteBufferCollector.java
b/parquet-common/src/test/java/org/apache/parquet/bytes/TestConcatenatingByteBufferCollector.java
index a64bd9d4b..236a39c9e 100644
---
a/parquet-common/src/test/java/org/apache/parquet/bytes/TestConcatenatingByteBufferCollector.java
+++
b/parquet-common/src/test/java/org/apache/parquet/bytes/TestConcatenatingByteBufferCollector.java
@@ -109,4 +109,72 @@ public class TestConcatenatingByteBufferCollector {
}
return cbaos;
}
+
+ @Test
+ public void testWriteAllToReleasesProgressively() throws IOException {
+ byte[] result;
+ ConcatenatingByteBufferCollector collector = new
ConcatenatingByteBufferCollector(allocator);
+ collector.collect(BytesInput.from(bytes("Hello")));
+ collector.collect(BytesInput.from(bytes(" ")));
+ collector.collect(BytesInput.from(bytes("World")));
+
+ Assert.assertEquals(11, collector.size());
+
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ collector.writeAllTo(baos);
+ result = baos.toByteArray();
+
+ // After writeAllTo, the collector should be empty (buffers released
progressively)
+ Assert.assertEquals(0, collector.size());
+
+ // Verify the data was written correctly
+ Assert.assertEquals("Hello World", new String(result,
StandardCharsets.UTF_8));
+
+ // close() after writeAllTo is a safe no-op
+ collector.close();
+ }
+
+ @Test
+ public void testDoubleCloseIsSafe() throws IOException {
+ ConcatenatingByteBufferCollector collector = new
ConcatenatingByteBufferCollector(allocator);
+ collector.collect(BytesInput.from(bytes("test data")));
+
+ Assert.assertEquals(9, collector.size());
+
+ // First close releases the buffers
+ collector.close();
+ Assert.assertEquals(0, collector.size());
+
+ // Second close should be a no-op and not throw
+ collector.close();
+ }
+
+ @Test
+ public void testCloseOnEmpty() {
+ // Close on an empty collector should not throw
+ ConcatenatingByteBufferCollector collector = new
ConcatenatingByteBufferCollector(allocator);
+ collector.close();
+ collector.close(); // double close on empty
+ }
+
+ @Test
+ public void testWriteAllToProducesCorrectOutputWithMultipleTypes() throws
IOException {
+ // Verify that writeAllTo produces correct output with mixed BytesInput
types
+ byte[] result;
+
+ ConcatenatingByteBufferCollector collector = new
ConcatenatingByteBufferCollector(allocator);
+ collector.collect(BytesInput.fromInt(42));
+ collector.collect(BytesInput.from(bytes("parquet")));
+ collector.collect(BytesInput.fromInt(99));
+
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ collector.writeAllTo(baos);
+ result = baos.toByteArray();
+
+ // Verify size: 4 (int) + 7 (string) + 4 (int) = 15 bytes
+ Assert.assertEquals(15, result.length);
+
+ // Already released by writeAllTo, close is a no-op
+ collector.close();
+ }
}
diff --git
a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java
b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java
index 69c28b2cd..9f5bc48f8 100644
---
a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java
+++
b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java
@@ -431,6 +431,7 @@ public class ColumnChunkPageWriteStore implements
PageWriteStore, BloomFilterWri
}
public void writeToFileWriter(ParquetFileWriter writer) throws IOException
{
+ long bytesWritten = buf.size();
if (null == headerBlockEncryptor) {
writer.writeColumnChunk(
path,
@@ -475,7 +476,7 @@ public class ColumnChunkPageWriteStore implements
PageWriteStore, BloomFilterWri
if (LOG.isDebugEnabled()) {
LOG.debug(String.format(
"written %,dB for %s: %,d values, %,dB raw, %,dB comp, %d
pages, encodings: %s",
- buf.size(),
+ bytesWritten,
path,
totalValueCount,
uncompressedLength,
@@ -811,8 +812,13 @@ public class ColumnChunkPageWriteStore implements
PageWriteStore, BloomFilterWri
public void flushToFileWriter(ParquetFileWriter writer) throws IOException {
for (ColumnDescriptor path : schema.getColumns()) {
- ColumnChunkPageWriter pageWriter = writers.get(path);
- pageWriter.writeToFileWriter(writer);
+ // Eagerly release this column's page buffers now that they've been
+ // written to the file writer. This reduces peak memory during flush
+ // from the entire compressed row group down to roughly one column's
+ // worth of compressed pages at a time.
+ try (ColumnChunkPageWriter pageWriter = writers.get(path)) {
+ pageWriter.writeToFileWriter(writer);
+ }
}
}
}
