apacheGH-3516: Optimize DeltaByteArrayWriter / DeltaLengthByteArrayValuesWriter

Two related changes in the DELTA_BYTE_ARRAY write path:

1. DeltaLengthByteArrayValuesWriter: drop the unused LittleEndianDataOutputStream
   wrapper. Binary.writeTo(arrayOut) works directly with the underlying
   CapacityByteArrayOutputStream; the LE wrapper added an extra layer of
   dispatch on every value but never used any LE functionality
   (writeInt/writeLong/etc.). Add a new writeBytes(byte[], int, int) overload
   so callers that already have the raw bytes can avoid allocating a Binary
   wrapper.

2. DeltaByteArrayWriter: tighten suffixWriter field type to
   DeltaLengthByteArrayValuesWriter (it's always constructed as one) so the
   new writeBytes(byte[], int, int) overload is callable. Replace the suffix
   call with the raw-bytes overload, eliminating the per-value Binary.slice()
   allocation.

Benchmark (BinaryEncodingBenchmark, 100k BINARY values per invocation,
JMH -wi 3 -i 5 -f 1):

  Benchmark                  Param     Before (ops/s)  After (ops/s)  Improvement
  encodeDeltaByteArray       LOW/10      61,475,818    81,416,754     +32% (1.32x)
  encodeDeltaByteArray       LOW/100     34,759,755    45,186,617     +30% (1.30x)
  encodeDeltaByteArray       LOW/1000     5,386,922     6,532,850     +21% (1.21x)
  encodeDeltaByteArray       HIGH/10     56,799,595    78,966,929     +39% (1.39x)
  encodeDeltaLengthByteArray LOW/10     129,447,876   136,657,079     +6%
  encodeDeltaLengthByteArray HIGH/10    123,673,058   116,778,775     flat (noise)

Negative controls (encodePlain, encodeDictionary): unchanged within noise.

The DeltaByteArray path benefits most because it eliminates both the
Binary.slice() allocation per suffix and the OutputStream dispatch layer.
DeltaLengthByteArray gains are smaller since only the OutputStream wrapper
removal applies there.

No public API change. No file format change.

All 573 parquet-column tests pass.

Author: iemejia

parquet-column/src/main/java/org/apache/parquet/column/values/deltalengthbytearray/DeltaLengthByteArrayValuesWriter.java

@@ -22,7 +22,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.column.values.delta.DeltaBinaryPackingValuesWriter;
@@ -46,11 +45,9 @@ public class DeltaLengthByteArrayValuesWriter extends ValuesWriter {
 
   private ValuesWriter lengthWriter;
   private CapacityByteArrayOutputStream arrayOut;
-  private LittleEndianDataOutputStream out;
 
   public DeltaLengthByteArrayValuesWriter(int initialSize, int pageSize, ByteBufferAllocator allocator) {
     arrayOut = new CapacityByteArrayOutputStream(initialSize, pageSize, allocator);
-    out = new LittleEndianDataOutputStream(arrayOut);
     lengthWriter = new DeltaBinaryPackingValuesWriterForInteger(
         DeltaBinaryPackingValuesWriter.DEFAULT_NUM_BLOCK_VALUES,
         DeltaBinaryPackingValuesWriter.DEFAULT_NUM_MINIBLOCKS,
@@ -63,24 +60,29 @@ public DeltaLengthByteArrayValuesWriter(int initialSize, int pageSize, ByteBuffe
   public void writeBytes(Binary v) {
     try {
       lengthWriter.writeInteger(v.length());
-      v.writeTo(out);
+      v.writeTo(arrayOut);
     } catch (IOException e) {
       throw new ParquetEncodingException("could not write bytes", e);
     }
   }
 
+  /**
+   * Writes raw bytes directly, avoiding Binary object creation overhead.
+   * Used by {@link org.apache.parquet.column.values.deltastrings.DeltaByteArrayWriter}
+   * to write suffix bytes without creating an intermediate Binary.slice().
+   */
+  public void writeBytes(byte[] data, int offset, int length) {
+    lengthWriter.writeInteger(length);
+    arrayOut.write(data, offset, length);
+  }
+
   @Override
   public long getBufferedSize() {
     return lengthWriter.getBufferedSize() + arrayOut.size();
   }
 
   @Override
   public BytesInput getBytes() {
-    try {
-      out.flush();
-    } catch (IOException e) {
-      throw new ParquetEncodingException("could not write page", e);
-    }
     LOG.debug("writing a buffer of size {}", arrayOut.size());
     return BytesInput.concat(lengthWriter.getBytes(), BytesInput.from(arrayOut));
   }

parquet-column/src/main/java/org/apache/parquet/column/values/deltastrings/DeltaByteArrayWriter.java

@@ -38,7 +38,7 @@
 public class DeltaByteArrayWriter extends ValuesWriter {
 
   private ValuesWriter prefixLengthWriter;
-  private ValuesWriter suffixWriter;
+  private DeltaLengthByteArrayValuesWriter suffixWriter;
   private byte[] previous;
 
   public DeltaByteArrayWriter(int initialCapacity, int pageSize, ByteBufferAllocator allocator) {
@@ -98,7 +98,9 @@ public void writeBytes(Binary v) {
       i = length; // all bytes in the common range matched
     }
     prefixLengthWriter.writeInteger(i);
-    suffixWriter.writeBytes(v.slice(i, vb.length - i));
+    // Write suffix bytes directly from the byte array, avoiding Binary.slice() allocation
+    // and the virtual dispatch chain through Binary.writeTo()
+    suffixWriter.writeBytes(vb, i, vb.length - i);
     previous = vb;
   }
 }