ALP: Add lazy AlpValuesReader for float and double columns

Implements AlpValuesReader (abstract), AlpValuesReaderForFloat, and AlpValuesReaderForDouble. Uses lazy per-vector decoding: initFromPage reads only the header and offset array, vectors are decoded on first access. skip() is O(1) with no decoding.
apache · prtkgaur · Mar 7, 2026 · Mar 7, 2026 · Mar 7, 2026 · Mar 7, 2026
commit 27508d93bf59e23300f78172e753abdd41436ecc
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesReader.java b/parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesReader.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.column.values.alp;
+
+import static org.apache.parquet.column.values.alp.AlpConstants.*;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.ParquetDecodingException;
+
+/**
+ * Abstract base class for ALP values readers with lazy per-vector decoding.
+ *
+ * <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.
+ */
+abstract class AlpValuesReader extends ValuesReader {
+
+  protected int vectorSize;
+  protected int totalCount;
+  protected int numVectors;
+  protected int currentIndex;
+  protected int decodedVectorIndex = -1;
+  protected int[] vectorOffsets;
+  protected byte[] rawData; // all data after header
+
+  @Override
+  public void initFromPage(int valueCount, ByteBufferInputStream stream) throws IOException {
+    int available = (int) stream.available();
+    if (available < HEADER_SIZE) {
+      throw new ParquetDecodingException(
+          "ALP page too small for header: " + available + " bytes");
+    }
+
+    // Read header
+    byte[] headerBytes = new byte[HEADER_SIZE];
+    stream.read(headerBytes);
+    ByteBuffer header = ByteBuffer.wrap(headerBytes).order(ByteOrder.LITTLE_ENDIAN);
+
+    int compressionMode = header.get() & 0xFF;
+    int integerEncoding = header.get() & 0xFF;
+    int logVectorSize = header.get() & 0xFF;
+    totalCount = header.getInt();
+
+    if (compressionMode != COMPRESSION_MODE_ALP) {
+      throw new ParquetDecodingException(
+          "Unsupported ALP compression mode: " + compressionMode);
+    }
+    if (integerEncoding != INTEGER_ENCODING_FOR) {
+      throw new ParquetDecodingException(
+          "Unsupported ALP integer encoding: " + integerEncoding);
+    }
+
+    vectorSize = 1 << logVectorSize;
+    numVectors = (totalCount + vectorSize - 1) / vectorSize;
+    currentIndex = 0;
+    decodedVectorIndex = -1;
+
+    if (numVectors == 0) {
+      vectorOffsets = new int[0];
+      rawData = new byte[0];
+      return;
+    }
+
+    // Read remaining data (offsets + vectors)
+    int remaining = (int) stream.available();
+    rawData = new byte[remaining];
+    stream.read(rawData);
+
+    // Parse offsets from rawData
+    ByteBuffer body = ByteBuffer.wrap(rawData, 0, numVectors * OFFSET_SIZE)
+        .order(ByteOrder.LITTLE_ENDIAN);
+    vectorOffsets = new int[numVectors];
+    for (int i = 0; i < numVectors; i++) {
+      vectorOffsets[i] = body.getInt();
+    }
+  }
+
+  @Override
+  public void skip() {
+    currentIndex++;
+  }
+
+  @Override
+  public void skip(int n) {
+    currentIndex += n;
+  }
+
+  /** Number of elements in the given vector (last vector may be partial). */
+  protected int elementsInVector(int vectorIdx) {
+    if (vectorIdx < totalCount / vectorSize) {
+      return vectorSize;
+    }
+    int rem = totalCount % vectorSize;
+    return (rem == 0) ? vectorSize : rem;
+  }
+}
diff --git a/...t-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesReaderForDouble.java b/...t-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesReaderForDouble.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.column.values.alp;
+
+import org.apache.parquet.io.ParquetDecodingException;
+
+/**
+ * ALP values reader for double columns with lazy per-vector decoding.
+ */
+public class AlpValuesReaderForDouble extends AlpValuesReader {
+
+  private double[] decodedBuffer;
+
+  @Override
+  public double readDouble() {
+    if (currentIndex >= totalCount) {
+      throw new ParquetDecodingException("ALP double reader exhausted at index " + currentIndex);
+    }
+    int vectorIdx = currentIndex / vectorSize;
+    int posInVector = currentIndex % vectorSize;
+    ensureVectorDecoded(vectorIdx);
+    currentIndex++;
+    return decodedBuffer[posInVector];
+  }
+
+  private void ensureVectorDecoded(int vectorIdx) {
+    if (vectorIdx == decodedVectorIndex) {
+      return;
+    }
+    int numElements = elementsInVector(vectorIdx);
+    int dataOffset = vectorOffsets[vectorIdx];
+    AlpCompression.DoubleCompressedVector cv =
+        AlpCompression.DoubleCompressedVector.load(rawData, dataOffset, numElements);
+    decodedBuffer = new double[numElements];
+    AlpCompression.decompressDoubleVector(cv, decodedBuffer);
+    decodedVectorIndex = vectorIdx;
+  }
+}
diff --git a/...et-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesReaderForFloat.java b/...et-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesReaderForFloat.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.column.values.alp;
+
+import org.apache.parquet.io.ParquetDecodingException;
+
+/**
+ * ALP values reader for float columns with lazy per-vector decoding.
+ */
+public class AlpValuesReaderForFloat extends AlpValuesReader {
+
+  private float[] decodedBuffer;
+
+  @Override
+  public float readFloat() {
+    if (currentIndex >= totalCount) {
+      throw new ParquetDecodingException("ALP float reader exhausted at index " + currentIndex);
+    }
+    int vectorIdx = currentIndex / vectorSize;
+    int posInVector = currentIndex % vectorSize;
+    ensureVectorDecoded(vectorIdx);
+    currentIndex++;
+    return decodedBuffer[posInVector];
+  }
+
+  private void ensureVectorDecoded(int vectorIdx) {
+    if (vectorIdx == decodedVectorIndex) {
+      return;
+    }
+    int numElements = elementsInVector(vectorIdx);
+    int dataOffset = vectorOffsets[vectorIdx];
+    AlpCompression.FloatCompressedVector cv =
+        AlpCompression.FloatCompressedVector.load(rawData, dataOffset, numElements);
+    decodedBuffer = new float[numElements];
+    AlpCompression.decompressFloatVector(cv, decodedBuffer);
+    decodedVectorIndex = vectorIdx;
+  }
+}