apacheGH-3493: Optimize PlainValuesReader with direct ByteBuffer reads

Replace the LittleEndianDataInputStream wrapper with direct ByteBuffer
access using LITTLE_ENDIAN byte order in PlainValuesReader. Each
read{Integer,Long,Float,Double}() previously dispatched through 4
in.read() calls per value and assembled the result with manual bit
shifts; it now compiles to a single ByteBuffer get*() JVM intrinsic.

In initFromPage, the page data is obtained as a single contiguous
ByteBuffer via ByteBufferInputStream.slice(available). The
ByteBufferInputStream.slice() method handles both single-buffer
(zero-copy view) and multi-buffer (copy into contiguous buffer) cases
transparently. In practice page data is almost always a single
contiguous buffer.

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

  Pattern           Before (ops/s)   After (ops/s)   Speedup
  SEQUENTIAL        427,630,411   5,397,298,681     12.6x
  RANDOM            431,052,072   5,437,926,758     12.6x
  LOW_CARDINALITY   423,443,685   5,477,810,011     12.9x
  HIGH_CARDINALITY  426,405,891   5,485,493,740     12.9x

The improvement is consistent regardless of data distribution because
the bottleneck was entirely in the dispatch overhead. All four numeric
plain reader types (int, long, float, double) benefit equally.

All 573 parquet-column tests pass.

Author: iemejia

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

@@ -19,120 +19,92 @@
 package org.apache.parquet.column.values.plain;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import org.apache.parquet.bytes.ByteBufferInputStream;
 import org.apache.parquet.bytes.LittleEndianDataInputStream;
 import org.apache.parquet.column.values.ValuesReader;
-import org.apache.parquet.io.ParquetDecodingException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * Plain encoding for float, double, int, long
+ * Plain encoding for float, double, int, long.
+ *
+ * <p>Reads directly from a {@link ByteBuffer} with {@link ByteOrder#LITTLE_ENDIAN} byte order,
+ * bypassing the {@link LittleEndianDataInputStream} wrapper to avoid per-value virtual dispatch
+ * overhead. The underlying page data is obtained as a single contiguous {@link ByteBuffer} via
+ * {@link ByteBufferInputStream#slice(int)}.
  */
 public abstract class PlainValuesReader extends ValuesReader {
   private static final Logger LOG = LoggerFactory.getLogger(PlainValuesReader.class);
 
-  protected LittleEndianDataInputStream in;
+  ByteBuffer buffer;
 
   @Override
   public void initFromPage(int valueCount, ByteBufferInputStream stream) throws IOException {
     LOG.debug("init from page at offset {} for length {}", stream.position(), stream.available());
-    this.in = new LittleEndianDataInputStream(stream.remainingStream());
+    int available = stream.available();
+    if (available > 0) {
+      this.buffer = stream.slice(available).order(ByteOrder.LITTLE_ENDIAN);
+    } else {
+      this.buffer = ByteBuffer.allocate(0).order(ByteOrder.LITTLE_ENDIAN);
+    }
   }
 
   @Override
   public void skip() {
     skip(1);
   }
 
-  void skipBytesFully(int n) throws IOException {
-    int skipped = 0;
-    while (skipped < n) {
-      skipped += in.skipBytes(n - skipped);
-    }
-  }
-
   public static class DoublePlainValuesReader extends PlainValuesReader {
 
     @Override
     public void skip(int n) {
-      try {
-        skipBytesFully(n * 8);
-      } catch (IOException e) {
-        throw new ParquetDecodingException("could not skip " + n + " double values", e);
-      }
+      buffer.position(buffer.position() + n * 8);
     }
 
     @Override
     public double readDouble() {
-      try {
-        return in.readDouble();
-      } catch (IOException e) {
-        throw new ParquetDecodingException("could not read double", e);
-      }
+      return buffer.getDouble();
     }
   }
 
   public static class FloatPlainValuesReader extends PlainValuesReader {
 
     @Override
     public void skip(int n) {
-      try {
-        skipBytesFully(n * 4);
-      } catch (IOException e) {
-        throw new ParquetDecodingException("could not skip " + n + " floats", e);
-      }
+      buffer.position(buffer.position() + n * 4);
     }
 
     @Override
     public float readFloat() {
-      try {
-        return in.readFloat();
-      } catch (IOException e) {
-        throw new ParquetDecodingException("could not read float", e);
-      }
+      return buffer.getFloat();
     }
   }
 
   public static class IntegerPlainValuesReader extends PlainValuesReader {
 
     @Override
     public void skip(int n) {
-      try {
-        in.skipBytes(n * 4);
-      } catch (IOException e) {
-        throw new ParquetDecodingException("could not skip " + n + " ints", e);
-      }
+      buffer.position(buffer.position() + n * 4);
     }
 
     @Override
     public int readInteger() {
-      try {
-        return in.readInt();
-      } catch (IOException e) {
-        throw new ParquetDecodingException("could not read int", e);
-      }
+      return buffer.getInt();
     }
   }
 
   public static class LongPlainValuesReader extends PlainValuesReader {
 
     @Override
     public void skip(int n) {
-      try {
-        in.skipBytes(n * 8);
-      } catch (IOException e) {
-        throw new ParquetDecodingException("could not skip " + n + " longs", e);
-      }
+      buffer.position(buffer.position() + n * 8);
     }
 
     @Override
     public long readLong() {
-      try {
-        return in.readLong();
-      } catch (IOException e) {
-        throw new ParquetDecodingException("could not read long", e);
-      }
+      return buffer.getLong();
     }
   }
 }

pom.xml

@@ -594,6 +594,8 @@
               <exclude>org.apache.parquet.internal.column.columnindex.IndexIterator</exclude>
               <!-- Removal of a protected method in a class that's not supposed to be subclassed by third-party code -->
               <exclude>org.apache.parquet.column.values.bytestreamsplit.ByteStreamSplitValuesReader#gatherElementDataFromStreams(byte[])</exclude>
+              <!-- Removal of a protected internal field that should not have been part of the public API -->
+              <exclude>org.apache.parquet.column.values.plain.PlainValuesReader#in</exclude>
               <!-- Due to the removal of deprecated methods -->
               <exclude>org.apache.parquet.arrow.schema.SchemaMapping$TypeMappingVisitor#visit(org.apache.parquet.arrow.schema.SchemaMapping$MapTypeMapping)</exclude>
               <!-- Intentional removal of deprecated Pig support from parquet-thrift -->