GH-3495: Optimize PlainValuesWriter with direct ByteBuffer slab writes

PlainValuesWriter previously wrote values through a two-layer abstraction:
PlainValuesWriter -> LittleEndianDataOutputStream -> CapacityByteArrayOutputStream.
Each writeInt() decomposed the int into 4 bytes in a temp writeBuffer[8]
array, then dispatched through the OutputStream chain. Since
CapacityByteArrayOutputStream already uses ByteBuffer slabs internally, we
can write directly to the slab with putInt()/putLong() using LITTLE_ENDIAN
byte order -- a single JVM intrinsic on x86/ARM -- eliminating the byte
decomposition, temp array, and virtual dispatch.

Changes:
- CapacityByteArrayOutputStream: set ByteOrder.LITTLE_ENDIAN on newly
  allocated slabs in addSlab(); add writeInt(int) and writeLong(long)
  methods that use currentSlab.putInt(v) / currentSlab.putLong(v) directly.
- PlainValuesWriter: remove the LittleEndianDataOutputStream field; route
  writeInteger/writeLong/writeFloat/writeDouble/writeBytes through the
  underlying CapacityByteArrayOutputStream directly. writeFloat and
  writeDouble use Float.floatToIntBits / Double.doubleToLongBits + the new
  writeInt/writeLong methods. getBytes() no longer needs to flush a
  buffering layer; close() no longer closes the defunct stream.

Benchmark (IntEncodingBenchmark.encodePlain, 100k INT32 values per
invocation, JMH -wi 3 -i 5 -f 1):

  Pattern           Before (ops/s)   After (ops/s)   Improvement
  SEQUENTIAL          26,817,451      52,953,193     +97.5% (2.0x)
  RANDOM              28,517,312      37,774,036     +32.5%
  LOW_CARDINALITY     28,705,158      52,819,678     +84.0%
  HIGH_CARDINALITY    28,595,519      37,862,571     +32.4%

The same code path also benefits writeLong, writeFloat, writeDouble, and
the length prefix written by writeBytes(Binary).

Author: iemejia

parquet-column/src/main/java/org/apache/parquet/column/values/plain/PlainValuesWriter.java

@@ -23,7 +23,6 @@
 import org.apache.parquet.bytes.ByteBufferAllocator;
 import org.apache.parquet.bytes.BytesInput;
 import org.apache.parquet.bytes.CapacityByteArrayOutputStream;
-import org.apache.parquet.bytes.LittleEndianDataOutputStream;
 import org.apache.parquet.column.Encoding;
 import org.apache.parquet.column.values.ValuesWriter;
 import org.apache.parquet.io.ParquetEncodingException;
@@ -41,66 +40,44 @@ public class PlainValuesWriter extends ValuesWriter {
   public static final Charset CHARSET = Charset.forName("UTF-8");
 
   private CapacityByteArrayOutputStream arrayOut;
-  private LittleEndianDataOutputStream out;
 
   public PlainValuesWriter(int initialSize, int pageSize, ByteBufferAllocator allocator) {
     arrayOut = new CapacityByteArrayOutputStream(initialSize, pageSize, allocator);
-    out = new LittleEndianDataOutputStream(arrayOut);
   }
 
   @Override
   public final void writeBytes(Binary v) {
     try {
-      out.writeInt(v.length());
-      v.writeTo(out);
+      arrayOut.writeInt(v.length());
+      v.writeTo(arrayOut);
     } catch (IOException e) {
       throw new ParquetEncodingException("could not write bytes", e);
     }
   }
 
   @Override
   public final void writeInteger(int v) {
-    try {
-      out.writeInt(v);
-    } catch (IOException e) {
-      throw new ParquetEncodingException("could not write int", e);
-    }
+    arrayOut.writeInt(v);
   }
 
   @Override
   public final void writeLong(long v) {
-    try {
-      out.writeLong(v);
-    } catch (IOException e) {
-      throw new ParquetEncodingException("could not write long", e);
-    }
+    arrayOut.writeLong(v);
   }
 
   @Override
   public final void writeFloat(float v) {
-    try {
-      out.writeFloat(v);
-    } catch (IOException e) {
-      throw new ParquetEncodingException("could not write float", e);
-    }
+    arrayOut.writeInt(Float.floatToIntBits(v));
   }
 
   @Override
   public final void writeDouble(double v) {
-    try {
-      out.writeDouble(v);
-    } catch (IOException e) {
-      throw new ParquetEncodingException("could not write double", e);
-    }
+    arrayOut.writeLong(Double.doubleToLongBits(v));
   }
 
   @Override
   public void writeByte(int value) {
-    try {
-      out.write(value);
-    } catch (IOException e) {
-      throw new ParquetEncodingException("could not write byte", e);
-    }
+    arrayOut.write(value);
   }
 
   @Override
@@ -110,11 +87,6 @@ public long getBufferedSize() {
 
   @Override
   public BytesInput getBytes() {
-    try {
-      out.flush();
-    } catch (IOException e) {
-      throw new ParquetEncodingException("could not write page", e);
-    }
     if (LOG.isDebugEnabled()) LOG.debug("writing a buffer of size {}", arrayOut.size());
     return BytesInput.from(arrayOut);
   }
@@ -127,7 +99,6 @@ public void reset() {
   @Override
   public void close() {
     arrayOut.close();
-    out.close();
   }
 
   @Override

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

@@ -27,6 +27,7 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.ArrayList;
 import java.util.List;
 import org.apache.parquet.OutputStreamCloseException;
@@ -194,6 +195,7 @@ private void addSlab(int minimumSize) {
     LOG.debug("used {} slabs, adding new slab of size {}", slabs.size(), nextSlabSize);
 
     this.currentSlab = allocator.allocate(nextSlabSize);
+    this.currentSlab.order(ByteOrder.LITTLE_ENDIAN);
     this.slabs.add(currentSlab);
     this.bytesAllocated = Math.addExact(this.bytesAllocated, nextSlabSize);
   }
@@ -225,6 +227,34 @@ public void write(byte b[], int off, int len) {
     bytesUsed = Math.addExact(bytesUsed, len);
   }
 
+  /**
+   * Writes an int in little-endian byte order directly to the underlying slab,
+   * bypassing intermediate byte array decomposition. Slabs are set to
+   * {@link ByteOrder#LITTLE_ENDIAN} order so {@code putInt} produces the correct encoding.
+   *
+   * @param v the int value to write
+   */
+  public void writeInt(int v) {
+    if (currentSlab.remaining() < 4) {
+      addSlab(4);
+    }
+    currentSlab.putInt(v);
+    bytesUsed = Math.addExact(bytesUsed, 4);
+  }
+
+  /**
+   * Writes a long in little-endian byte order directly to the underlying slab.
+   *
+   * @param v the long value to write
+   */
+  public void writeLong(long v) {
+    if (currentSlab.remaining() < 8) {
+      addSlab(8);
+    }
+    currentSlab.putLong(v);
+    bytesUsed = Math.addExact(bytesUsed, 8);
+  }
+
   private void writeToOutput(OutputStream out, ByteBuffer buf, int len) throws IOException {
     if (buf.hasArray()) {
       out.write(buf.array(), buf.arrayOffset(), len);