ALP: Add pipeline integration, reader buffer reuse, and validation

Writer pipeline integration:
- Use CapacityByteArrayOutputStream for encoded vector storage instead
  of List<byte[]>, integrating with Parquet's memory management
- Use BytesInput.concat() for zero-copy page assembly
- Accept (initialCapacity, pageSize, ByteBufferAllocator) constructor
  params; factory now passes pipeline properties

Reader memory efficiency:
- Allocate decoded buffer once in initFromPage() and reuse across all
  vector decodes, eliminating per-vector float[]/double[] allocations
- Improves decode throughput 5-24% across all datasets

Reader validation:
- Validate logVectorSize bounds (MIN_LOG to MAX_LOG)
- Validate non-negative element count
- Validate skip(n) bounds

Author: sfc-gh-pgaur

parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesReader.java

@@ -33,6 +33,10 @@
  * <p>On {@link #initFromPage}, reads the 7-byte header and offset array but does NOT
  * decode any vectors. Vectors are decoded on demand when values are accessed.
  * {@link #skip()} is O(1) — it just advances the index.
+ *
+ * <p>Reuses the decoded buffer across vectors to reduce allocations.
+ * Validates header fields (compression mode, integer encoding, log vector size bounds,
+ * element count) and skip bounds.
  */
 abstract class AlpValuesReader extends ValuesReader {
 
@@ -42,7 +46,7 @@ abstract class AlpValuesReader extends ValuesReader {
   protected int currentIndex;
   protected int decodedVectorIndex = -1;
   protected int[] vectorOffsets;
-  protected byte[] rawData; // all data after header
+  protected byte[] rawData; // all data after header (offsets + vectors)
 
   @Override
   public void initFromPage(int valueCount, ByteBufferInputStream stream) throws IOException {
@@ -67,6 +71,13 @@ public void initFromPage(int valueCount, ByteBufferInputStream stream) throws IO
     if (integerEncoding != INTEGER_ENCODING_FOR) {
       throw new ParquetDecodingException("Unsupported ALP integer encoding: " + integerEncoding);
     }
+    if (logVectorSize < MIN_LOG_VECTOR_SIZE || logVectorSize > MAX_LOG_VECTOR_SIZE) {
+      throw new ParquetDecodingException("Invalid ALP log vector size: " + logVectorSize
+          + ", must be between " + MIN_LOG_VECTOR_SIZE + " and " + MAX_LOG_VECTOR_SIZE);
+    }
+    if (totalCount < 0) {
+      throw new ParquetDecodingException("Invalid ALP element count: " + totalCount);
+    }
 
     vectorSize = 1 << logVectorSize;
     numVectors = (totalCount + vectorSize - 1) / vectorSize;
@@ -76,6 +87,7 @@ public void initFromPage(int valueCount, ByteBufferInputStream stream) throws IO
     if (numVectors == 0) {
       vectorOffsets = new int[0];
       rawData = new byte[0];
+      allocateDecodedBuffer(vectorSize);
       return;
     }
 
@@ -90,6 +102,8 @@ public void initFromPage(int valueCount, ByteBufferInputStream stream) throws IO
     for (int i = 0; i < numVectors; i++) {
       vectorOffsets[i] = body.getInt();
     }
+
+    allocateDecodedBuffer(vectorSize);
   }
 
   @Override
@@ -99,6 +113,10 @@ public void skip() {
 
   @Override
   public void skip(int n) {
+    if (n < 0 || currentIndex + n > totalCount) {
+      throw new ParquetDecodingException(String.format(
+          "Cannot skip %d elements. Current index: %d, total count: %d", n, currentIndex, totalCount));
+    }
     currentIndex += n;
   }
 
@@ -110,4 +128,7 @@ protected int elementsInVector(int vectorIdx) {
     int rem = totalCount % vectorSize;
     return (rem == 0) ? vectorSize : rem;
   }
+
+  /** Allocate the decoded buffer once; called from initFromPage. */
+  protected abstract void allocateDecodedBuffer(int capacity);
 }

parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesReaderForDouble.java

@@ -22,11 +22,18 @@
 
 /**
  * ALP values reader for double columns with lazy per-vector decoding.
+ *
+ * <p>Reuses the decoded buffer across vectors to reduce allocations.
  */
 public class AlpValuesReaderForDouble extends AlpValuesReader {
 
   private double[] decodedBuffer;
 
+  @Override
+  protected void allocateDecodedBuffer(int capacity) {
+    this.decodedBuffer = new double[capacity];
+  }
+
   @Override
   public double readDouble() {
     if (currentIndex >= totalCount) {
@@ -47,7 +54,6 @@ private void ensureVectorDecoded(int vectorIdx) {
     int dataOffset = vectorOffsets[vectorIdx];
     AlpCompression.DoubleCompressedVector cv =
         AlpCompression.DoubleCompressedVector.load(rawData, dataOffset, numElements);
-    decodedBuffer = new double[numElements];
     AlpCompression.decompressDoubleVector(cv, decodedBuffer);
     decodedVectorIndex = vectorIdx;
   }

parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesReaderForFloat.java

@@ -22,11 +22,18 @@
 
 /**
  * ALP values reader for float columns with lazy per-vector decoding.
+ *
+ * <p>Reuses the decoded buffer across vectors to reduce allocations.
  */
 public class AlpValuesReaderForFloat extends AlpValuesReader {
 
   private float[] decodedBuffer;
 
+  @Override
+  protected void allocateDecodedBuffer(int capacity) {
+    this.decodedBuffer = new float[capacity];
+  }
+
   @Override
   public float readFloat() {
     if (currentIndex >= totalCount) {
@@ -47,7 +54,6 @@ private void ensureVectorDecoded(int vectorIdx) {
     int dataOffset = vectorOffsets[vectorIdx];
     AlpCompression.FloatCompressedVector cv =
         AlpCompression.FloatCompressedVector.load(rawData, dataOffset, numElements);
-    decodedBuffer = new float[numElements];
     AlpCompression.decompressFloatVector(cv, decodedBuffer);
     decodedVectorIndex = vectorIdx;
   }