GH-3505: Optimize ByteStreamSplitValuesReader page transposition

ByteStreamSplitValuesReader.decodeData eagerly transposes an entire page
from stream-split layout (elementSizeInBytes streams of valuesCount bytes
each) back to interleaved layout (valuesCount elements of elementSizeInBytes
bytes each). The current loop performs one ByteBuffer.get(int) per byte,
which incurs per-call bounds checks and virtual dispatch through
HeapByteBuffer/DirectByteBuffer for every single byte of the page. For a
100k-value FLOAT page that is 400k get(int) calls; for DOUBLE/LONG it is
800k.

This change rewrites decodeData in three steps:

1. Drop down to a byte[] view of the encoded buffer. When encoded.hasArray()
   is true (the typical case) use the backing array directly with the
   correct base offset; otherwise copy once with a single get(byte[]) call.
   This eliminates the per-byte ByteBuffer.get(int) bounds check and
   virtual dispatch.

2. Specialize loops for the common element sizes (4 and 8). Hoist all
   stream * valuesCount offsets out of the inner loop into local ints
   (s0..s3 for floats/ints, s0..s7 for doubles/longs), and write each
   output slot exactly once in a single sequential pass. Reads come from
   elementSizeInBytes concurrent sequential streams which modern hardware
   prefetchers handle well.

3. Generic fallback for arbitrary element sizes (FIXED_LEN_BYTE_ARRAY of
   any width) keeps the existing behaviour.

Benchmark (new ByteStreamSplitDecodingBenchmark, 100k values per
invocation, JDK 18, JMH -wi 5 -i 10 -f 3, 30 samples per row):

  Type     Before (ops/s)   After (ops/s)    Improvement
  Float       47,798,981     162,294,904    +240% (3.40x)
  Double      26,320,043      66,002,524    +151% (2.51x)
  Int         47,072,832     162,177,747    +245% (3.45x)
  Long        26,795,544      65,999,343    +146% (2.46x)

Decoded output is byte-identical to before; per-op heap allocation is
unchanged (the only allocation is the per-page decode buffer plus the
boxing of returned primitives by the benchmark).

All 573 parquet-column tests pass; 51 BSS-specific tests pass.

Author: iemejia

parquet-column/src/main/java/org/apache/parquet/column/values/bytestreamsplit/ByteStreamSplitValuesReader.java

@@ -49,17 +49,87 @@ protected int nextElementByteOffset() {
     return offset;
   }
 
-  // Decode an entire data page
+  /**
+   * Advances the stream position by {@code count} elements and returns the byte offset
+   * of the first element. Used by batch read methods in subclasses.
+   */
+  protected int advanceByteOffset(int count) {
+    if (indexInStream + count > valuesCount) {
+      throw new ParquetDecodingException("Byte-stream data was already exhausted.");
+    }
+    int offset = indexInStream * elementSizeInBytes;
+    indexInStream += count;
+    return offset;
+  }
+
+  // Decode an entire data page by transposing from stream-split layout to interleaved layout.
+  // The encoded data has elementSizeInBytes separate streams of valuesCount bytes each.
+  // The decoded data has valuesCount elements of elementSizeInBytes bytes each.
   private byte[] decodeData(ByteBuffer encoded, int valuesCount) {
-    assert encoded.limit() == valuesCount * elementSizeInBytes;
-    byte[] decoded = new byte[encoded.limit()];
-    int destByteIndex = 0;
-    for (int srcValueIndex = 0; srcValueIndex < valuesCount; ++srcValueIndex) {
-      for (int stream = 0; stream < elementSizeInBytes; ++stream, ++destByteIndex) {
-        decoded[destByteIndex] = encoded.get(srcValueIndex + stream * valuesCount);
+    int totalBytes = valuesCount * elementSizeInBytes;
+    assert encoded.remaining() >= totalBytes;
+
+    // Bulk access: use the backing array directly if available, otherwise copy once.
+    // This eliminates per-byte ByteBuffer.get(index) bounds checking in the hot loop.
+    byte[] src;
+    int srcBase;
+    if (encoded.hasArray()) {
+      src = encoded.array();
+      srcBase = encoded.arrayOffset() + encoded.position();
+    } else {
+      src = new byte[totalBytes];
+      encoded.get(src);
+      srcBase = 0;
+    }
+
+    byte[] decoded = new byte[totalBytes];
+
+    // Specialized single-pass loops for common element sizes.
+    // Single-pass writes each output location exactly once (sequentially), which is
+    // cache-friendlier than multi-pass for large pages where the output array exceeds L1.
+    // The reads come from elementSizeInBytes concurrent sequential streams, which modern
+    // hardware prefetchers handle well (typically 8-16 tracked streams per core).
+    if (elementSizeInBytes == 4) {
+      int s0 = srcBase,
+          s1 = srcBase + valuesCount,
+          s2 = srcBase + 2 * valuesCount,
+          s3 = srcBase + 3 * valuesCount;
+      for (int i = 0; i < valuesCount; ++i) {
+        int di = i * 4;
+        decoded[di] = src[s0 + i];
+        decoded[di + 1] = src[s1 + i];
+        decoded[di + 2] = src[s2 + i];
+        decoded[di + 3] = src[s3 + i];
+      }
+    } else if (elementSizeInBytes == 8) {
+      int s0 = srcBase,
+          s1 = srcBase + valuesCount,
+          s2 = srcBase + 2 * valuesCount,
+          s3 = srcBase + 3 * valuesCount,
+          s4 = srcBase + 4 * valuesCount,
+          s5 = srcBase + 5 * valuesCount,
+          s6 = srcBase + 6 * valuesCount,
+          s7 = srcBase + 7 * valuesCount;
+      for (int i = 0; i < valuesCount; ++i) {
+        int di = i * 8;
+        decoded[di] = src[s0 + i];
+        decoded[di + 1] = src[s1 + i];
+        decoded[di + 2] = src[s2 + i];
+        decoded[di + 3] = src[s3 + i];
+        decoded[di + 4] = src[s4 + i];
+        decoded[di + 5] = src[s5 + i];
+        decoded[di + 6] = src[s6 + i];
+        decoded[di + 7] = src[s7 + i];
+      }
+    } else {
+      // Generic fallback for arbitrary element sizes (e.g. FIXED_LEN_BYTE_ARRAY)
+      for (int stream = 0; stream < elementSizeInBytes; ++stream) {
+        int srcOffset = srcBase + stream * valuesCount;
+        for (int i = 0; i < valuesCount; ++i) {
+          decoded[i * elementSizeInBytes + stream] = src[srcOffset + i];
+        }
       }
     }
-    assert destByteIndex == decoded.length;
     return decoded;
   }