GH-3549: Leverage JDK11 readNBytes

parquet-common/src/main/java/org/apache/parquet/bytes/BytesInput.java

@@ -19,7 +19,6 @@
 package org.apache.parquet.bytes;
 
 import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
@@ -367,7 +366,8 @@ private StreamBytesInput(InputStream in, int byteCount) {
     @Override
     public void writeAllTo(OutputStream out) throws IOException {
       LOG.debug("write All {} bytes", byteCount);
-      // TODO: more efficient
+      // Cannot transfer in chunks because some InputStreams (e.g., NonBlockedDecompressorStream)
+      // require a full-size buffer to decompress all data in a single read() call.
       out.write(this.toByteArray());
     }
 
@@ -395,8 +395,16 @@ void writeInto(ByteBuffer buffer) {
 
     public byte[] toByteArray() throws IOException {
       LOG.debug("read all {} bytes", byteCount);
+      // Use the 3-arg readNBytes to read directly into a byteCount-sized buffer.
+      // The 1-arg readNBytes(int) internally uses small (8KB) buffers which breaks
+      // block decompressors that require a full-size output buffer (e.g., NonBlockedDecompressorStream
+      // resets the decompressor after finished(), losing remaining data on partial reads).
       byte[] buf = new byte[byteCount];
-      new DataInputStream(in).readFully(buf);
+      int n = in.readNBytes(buf, 0, byteCount);
+      if (n != byteCount) {
+        throw new EOFException(
+            "Reached the end of stream with " + (byteCount - n) + " bytes left to read");
+      }
       return buf;
     }