(avro) branch main updated: AVRO-4240: [Java] Size DataFileWriter output buffer to fit entire block frame (#3705)

[dmollitor]

This is an automated email from the ASF dual-hosted git repository.

dmollitor pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/avro.git


The following commit(s) were added to refs/heads/main by this push:
     new ca72772df3 AVRO-4240: [Java] Size DataFileWriter output buffer to fit 
entire block frame (#3705)
ca72772df3 is described below

commit ca72772df3881ab62b67986081a86adee9d088c9
Author: belugabehr <12578579+belugab...@users.noreply.github.com>
AuthorDate: Mon Apr 13 09:41:13 2026 -0400

    AVRO-4240: [Java] Size DataFileWriter output buffer to fit entire block 
frame (#3705)
---
 .../java/org/apache/avro/file/DataFileWriter.java  | 25 +++++++++++++++++++---
 1 file changed, 22 insertions(+), 3 deletions(-)

diff --git 
a/lang/java/avro/src/main/java/org/apache/avro/file/DataFileWriter.java 
b/lang/java/avro/src/main/java/org/apache/avro/file/DataFileWriter.java
index c4e031b75d..466fa28135 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/file/DataFileWriter.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/file/DataFileWriter.java
@@ -246,13 +246,16 @@ public class DataFileWriter<D> implements Closeable, 
Flushable {
     return this;
   }
 
-  private void init(OutputStream outs) throws IOException {
+  private void init(OutputStream outs) {
     this.underlyingStream = outs;
-    this.out = new BufferedFileOutputStream(outs);
+    // Size the output buffer to fit an entire block frame in a single flush:
+    // maxBlockSize() for compressed data + 20 bytes for two varint-encoded 
longs
+    // (up to 10 bytes each) + sync.length for the sync marker
+    this.out = new BufferedFileOutputStream(outs, maxBlockSize() + 20 + 
sync.length);
     EncoderFactory efactory = new EncoderFactory();
     this.vout = efactory.directBinaryEncoder(out, null);
     dout.setSchema(schema);
-    buffer = new NonCopyingByteArrayOutputStream(Math.min((int) (syncInterval 
* 1.25), Integer.MAX_VALUE / 2 - 1));
+    buffer = new NonCopyingByteArrayOutputStream(maxBlockSize());
     this.bufOut = this.initEncoder.apply(buffer);
     if (this.codec == null) {
       this.codec = CodecFactory.nullCodec().createInstance();
@@ -260,6 +263,17 @@ public class DataFileWriter<D> implements Closeable, 
Flushable {
     this.isOpen = true;
   }
 
+  /**
+   * Returns the estimated maximum compressed block size. Blocks are flushed 
when
+   * uncompressed data reaches {@link #syncInterval}, but compression may 
increase
+   * size (e.g. uncompressible data with codec framing overhead), so we allow 
25%
+   * headroom. The result is clamped to avoid integer overflow when used for
+   * buffer allocation.
+   */
+  private int maxBlockSize() {
+    return Math.min((int) (syncInterval * 1.25), Integer.MAX_VALUE / 2 - 1);
+  }
+
   private static final SecureRandom RNG = new SecureRandom();
 
   private static byte[] generateSync() {
@@ -494,6 +508,11 @@ public class DataFileWriter<D> implements Closeable, 
Flushable {
       this.out = new PositionFilter(out);
     }
 
+    public BufferedFileOutputStream(OutputStream out, int bufferSize) {
+      super(null, bufferSize);
+      this.out = new PositionFilter(out);
+    }
+
     public long tell() {
       return position + count;
     }