GH-3495: Deprecate LittleEndianDataOutputStream and remove remaining wrapper usages

This is an API cleanup commit with no measurable performance impact;
it removes the last two production usages of LittleEndianDataOutputStream
so the class can be deprecated.

After the previous commit removed LittleEndianDataOutputStream from
PlainValuesWriter, two production usages remained:

- FixedLenByteArrayPlainValuesWriter wrapped its CapacityByteArrayOutputStream
  in a LittleEndianDataOutputStream solely to call Binary.writeTo(out) for the
  fixed-length payload. The fixed-length encoding has no length prefix and the
  wrapper exposed no LE-specific behavior used here -- Binary.writeTo() only
  invokes OutputStream.write(byte[], int, int), which the wrapper passes
  through unchanged. The wrapper has been removed and the writer now writes
  the binary payload directly to the underlying CapacityByteArrayOutputStream.
  The wrapper-specific flush() in getBytes() is also gone (CBOS does not
  buffer).

- DeltaLengthByteArrayValuesWriter had the same pattern: a wrapper used only
  for v.writeTo(out) on the concatenated byte-array payload, with lengths
  written through a separate DeltaBinaryPackingValuesWriterForInteger. The
  wrapper has been removed for the same reasons.

With no remaining production usages, LittleEndianDataOutputStream is marked
@deprecated. The class is retained for binary compatibility (it is part of
the public parquet-common API) and will be removed in a future major release.
The javadoc directs producers of PLAIN-encoded data to write little-endian
values directly into a ByteBuffer with ByteOrder.LITTLE_ENDIAN, which
compiles to a single intrinsic store on little-endian architectures and
avoids the per-call byte decomposition and virtual dispatch performed by
this class.

Benchmarks (BinaryEncodingBenchmark, JMH -wi 5 -i 10 -f 3, 30 samples per
row, 100k values per invocation):

  encodeDeltaLengthByteArray (touched by this commit):
    Param            Before (ops/s)   After (ops/s)   Delta
    LOW   / 10        21,695,420      22,106,165      +1.9%
    LOW   / 100        6,803,834       6,798,992      +0.0%
    LOW   / 1000         865,171         866,820      +0.2%
    HIGH  / 10        19,985,164      20,225,337      +1.2%
    HIGH  / 100        5,677,955       5,600,746      -1.4%
    HIGH  / 1000         695,246         704,673      +1.4%

  encodeDeltaByteArray (uses DeltaLengthByteArrayValuesWriter for suffixes):
    Param            Before (ops/s)   After (ops/s)   Delta
    LOW   / 10        11,423,436      10,854,196      -5.0%
    LOW   / 100        4,810,962       4,821,321      +0.2%
    LOW   / 1000         646,415         667,650      +3.3%
    HIGH  / 10        10,047,817       9,963,667      -0.8%
    HIGH  / 100        4,337,800       4,103,937      -5.4%
    HIGH  / 1000         574,769         580,489      +1.0%

All deltas are within roughly +/- 5% with allocation rates per operation
unchanged within 2%, consistent with measurement noise rather than a
systematic effect on either direction. The motivation for this change is
code health (one fewer wrapper layer in the writer call chain, ability to
deprecate an internal-shaped class), not performance.

All 573 parquet-column tests and 308 parquet-common 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,7 +60,7 @@ 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);
     }
@@ -76,11 +73,6 @@ public long getBufferedSize() {
 
   @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/plain/FixedLenByteArrayPlainValuesWriter.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.io.ParquetEncodingException;
@@ -37,7 +36,6 @@ public class FixedLenByteArrayPlainValuesWriter extends ValuesWriter {
   private static final Logger LOG = LoggerFactory.getLogger(PlainValuesWriter.class);
 
   private CapacityByteArrayOutputStream arrayOut;
-  private LittleEndianDataOutputStream out;
   private int length;
   private ByteBufferAllocator allocator;
 
@@ -46,7 +44,6 @@ public FixedLenByteArrayPlainValuesWriter(
     this.length = length;
     this.allocator = allocator;
     this.arrayOut = new CapacityByteArrayOutputStream(initialSize, pageSize, this.allocator);
-    this.out = new LittleEndianDataOutputStream(arrayOut);
   }
 
   @Override
@@ -56,7 +53,7 @@ public final void writeBytes(Binary v) {
           "Fixed Binary size " + v.length() + " does not match field type length " + length);
     }
     try {
-      v.writeTo(out);
+      v.writeTo(arrayOut);
     } catch (IOException e) {
       throw new ParquetEncodingException("could not write fixed bytes", e);
     }
@@ -69,11 +66,6 @@ public long getBufferedSize() {
 
   @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.from(arrayOut);
   }

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

@@ -24,8 +24,18 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * Based on DataOutputStream but in little endian and without the String/char methods
+ * Based on DataOutputStream but in little endian and without the String/char methods.
+ *
+ * @deprecated As of release following the {@link CapacityByteArrayOutputStream#writeInt(int)}
+ *     and {@link CapacityByteArrayOutputStream#writeLong(long)} additions, this class is no
+ *     longer used by Parquet's own writers. Producers of PLAIN-encoded data should write
+ *     little-endian values directly into a {@link java.nio.ByteBuffer} configured with
+ *     {@link java.nio.ByteOrder#LITTLE_ENDIAN}, which compiles to a single intrinsic store on
+ *     little-endian architectures and avoids the per-call byte decomposition and virtual
+ *     dispatch performed here. This class is retained for binary compatibility and will be
+ *     removed in a future release.
  */
+@Deprecated
 public class LittleEndianDataOutputStream extends OutputStream {
 
   private static final Logger LOG = LoggerFactory.getLogger(LittleEndianDataOutputStream.class);