wgtmac commented on code in PR #3397:
URL: https://github.com/apache/parquet-java/pull/3397#discussion_r3900437837
##########
parquet-format-structures/pom.xml:
##########
@@ -66,6 +66,35 @@
</execution>
</executions>
</plugin>
+ <!--
Review Comment:
We should probably remove these lines by either releasing a new version of
parquet-format, or after merging
https://github.com/apache/parquet-java/pull/3709
##########
parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesReader.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * 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>Reads ALP-encoded values from the interleaved page layout:
+ * <pre>
+ * ┌─────────┬──────────────────────┬──────────────┬──────────────┬─────┐
+ * │ Header │ Offset Array │ Vector 0 │ Vector 1 │ ... │
+ * │ 7 bytes │ 4B × numVectors │ (interleaved)│ (interleaved)│ │
+ * └─────────┴──────────────────────┴──────────────┴──────────────┴─────┘
+ * </pre>
+ *
+ * <p>Each vector is decoded lazily on first access. Skipping values does not
+ * trigger decoding of intermediate vectors.
+ */
+abstract class AlpValuesReader extends ValuesReader {
+
+ protected int vectorSize;
+ protected int totalCount;
+ protected int numVectors;
+ protected int pageValueIndex;
+ protected int currentVectorNumber;
+
+ protected int[] vectorOffsets;
+ protected ByteBuffer vectorsData;
+ protected int offsetArraySize;
+
+ // Scratch buffer for exception positions within a vector; shared by both
readers (int[] in each).
+ protected int[] excPositionsBuffer;
+
+ AlpValuesReader() {
+ this.pageValueIndex = 0;
+ this.totalCount = 0;
+ this.currentVectorNumber = -1;
+ }
+
+ @Override
+ public void initFromPage(int valuesCount, ByteBufferInputStream stream)
+ throws ParquetDecodingException, IOException {
+ ByteBuffer headerBuf =
stream.slice(ALP_HEADER_SIZE).order(ByteOrder.LITTLE_ENDIAN);
+ int compressionMode = headerBuf.get() & 0xFF;
+ int integerEncoding = headerBuf.get() & 0xFF;
+ int logVectorSize = headerBuf.get() & 0xFF;
+ int numElements = headerBuf.getInt();
+
+ if (compressionMode != ALP_COMPRESSION_MODE) {
+ throw new ParquetDecodingException("Unsupported ALP compression mode: "
+ compressionMode);
+ }
+ if (integerEncoding != ALP_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 (numElements < 0) {
+ throw new ParquetDecodingException("Invalid ALP element count: " +
numElements);
+ }
+ // ALP's num_elements is the count of non-null values that went through
encoding;
+ // valuesCount is the page row count, which is larger when the column has
nulls.
+ // The two are equal only for required (non-null) columns.
+ if (numElements > valuesCount) {
+ throw new ParquetDecodingException(
+ "ALP header element count " + numElements + " exceeds page
valuesCount " + valuesCount);
+ }
+
+ this.vectorSize = 1 << logVectorSize;
+ this.totalCount = numElements;
+ this.numVectors = (numElements + vectorSize - 1) / vectorSize;
Review Comment:
This int arithmetic can overflow for a forged numElements. The result can be
negative or trigger a huge allocation.
##########
parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesWriter.java:
##########
@@ -0,0 +1,647 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.parquet.bytes.ByteBufferAllocator;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.BytesUtils;
+import org.apache.parquet.bytes.CapacityByteArrayOutputStream;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.values.ValuesWriter;
+import org.apache.parquet.column.values.bitpacking.BytePacker;
+import org.apache.parquet.column.values.bitpacking.BytePackerForLong;
+import org.apache.parquet.column.values.bitpacking.Packer;
+
+/**
+ * ALP (Adaptive Lossless floating-Point) values writer.
+ *
+ * <p>ALP encoding converts floating-point values to integers using decimal
scaling,
+ * then applies Frame of Reference encoding and bit-packing.
+ * Values that cannot be losslessly converted are stored as exceptions.
+ *
+ * <p>Writing is incremental: values are buffered in a fixed-size vector
buffer,
+ * and each full vector is encoded and flushed to the output stream
immediately.
+ * On {@link #getBytes()}, any remaining partial vector is flushed, and the
+ * final page bytes are assembled.
+ *
+ * <p>Interleaved Page Layout:
+ * <pre>
+ * ┌─────────┬──────────────────────┬──────────────┬──────────────┬─────┐
+ * │ Header │ Offset Array │ Vector 0 │ Vector 1 │ ... │
+ * │ 7 bytes │ 4B × numVectors │ (interleaved)│ (interleaved)│ │
+ * └─────────┴──────────────────────┴──────────────┴──────────────┴─────┘
+ * </pre>
+ *
+ * <p>Each vector contains interleaved:
+ * AlpInfo(4B) + ForInfo(5B/9B) + PackedValues + ExceptionPositions +
ExceptionValues
+ */
+public abstract class AlpValuesWriter extends ValuesWriter {
+
+ protected final int initialCapacity;
+ protected final int pageSize;
+ protected final ByteBufferAllocator allocator;
+ protected final int vectorSize;
+ protected final int logVectorSize;
+
+ AlpValuesWriter(int initialCapacity, int pageSize, ByteBufferAllocator
allocator, int vectorSize) {
+ AlpConstants.validateVectorSize(vectorSize);
+ this.initialCapacity = initialCapacity;
+ this.pageSize = pageSize;
+ this.allocator = allocator;
+ this.vectorSize = vectorSize;
+ this.logVectorSize = Integer.numberOfTrailingZeros(vectorSize);
+ }
+
+ @Override
+ public Encoding getEncoding() {
+ return Encoding.ALP;
+ }
+
+ /** Float writer. Buffers one vector at a time, encodes and flushes when
full. */
+ public static class FloatAlpValuesWriter extends AlpValuesWriter {
+ private final float[] vectorBuffer;
+ private int bufferCount;
+ private int totalCount;
+ private CapacityByteArrayOutputStream encodedVectors;
+ private final List<Integer> vectorByteSizes;
+
+ // Preset caching: collect evenly-spaced sample vectors across the
rowgroup,
+ // then build presets using estimated compressed size (matching C++
AlpSampler).
+ private int vectorsProcessed;
+ private int[][] cachedPresets;
+ // Winning (exponent, factor) pairs from sampled vectors, tallied later
into the preset cache.
+ private final List<int[]> sampledParams;
+ private final int rowgroupSampleJump;
+
+ // Reusable per-vector buffers
+ private final int[] encodedBuffer;
+ private final short[] excPosBuffer;
+ private final float[] excValBuffer;
+ private final byte[] metadataBuf;
+ private final byte[] packBuf;
+ private final int[] packPadBuf;
+
+ public FloatAlpValuesWriter(int initialCapacity, int pageSize,
ByteBufferAllocator allocator) {
+ this(initialCapacity, pageSize, allocator, DEFAULT_VECTOR_SIZE);
+ }
+
+ public FloatAlpValuesWriter(int initialCapacity, int pageSize,
ByteBufferAllocator allocator, int vectorSize) {
+ super(initialCapacity, pageSize, allocator, vectorSize);
+ this.vectorBuffer = new float[vectorSize];
+ this.bufferCount = 0;
+ this.totalCount = 0;
+ this.encodedVectors = new CapacityByteArrayOutputStream(initialCapacity,
pageSize, allocator);
+ this.vectorByteSizes = new ArrayList<>();
+ this.vectorsProcessed = 0;
+ this.cachedPresets = null;
+ this.sampledParams = new ArrayList<>();
+ // Space samples evenly: one sample every jump vectors across the
rowgroup.
+ // Math.max(1, ...) guards against very small rowgroups or large vector
sizes.
+ this.rowgroupSampleJump =
Review Comment:
With the default 20K-row page limit, this cadence yields only about two
samples per page, while eight are required. The preset cache never builds on
normal pages.
##########
parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesReader.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * 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>Reads ALP-encoded values from the interleaved page layout:
+ * <pre>
+ * ┌─────────┬──────────────────────┬──────────────┬──────────────┬─────┐
+ * │ Header │ Offset Array │ Vector 0 │ Vector 1 │ ... │
+ * │ 7 bytes │ 4B × numVectors │ (interleaved)│ (interleaved)│ │
+ * └─────────┴──────────────────────┴──────────────┴──────────────┴─────┘
+ * </pre>
+ *
+ * <p>Each vector is decoded lazily on first access. Skipping values does not
+ * trigger decoding of intermediate vectors.
+ */
+abstract class AlpValuesReader extends ValuesReader {
+
+ protected int vectorSize;
+ protected int totalCount;
+ protected int numVectors;
+ protected int pageValueIndex;
+ protected int currentVectorNumber;
+
+ protected int[] vectorOffsets;
+ protected ByteBuffer vectorsData;
+ protected int offsetArraySize;
+
+ // Scratch buffer for exception positions within a vector; shared by both
readers (int[] in each).
+ protected int[] excPositionsBuffer;
+
+ AlpValuesReader() {
+ this.pageValueIndex = 0;
+ this.totalCount = 0;
+ this.currentVectorNumber = -1;
+ }
+
+ @Override
+ public void initFromPage(int valuesCount, ByteBufferInputStream stream)
+ throws ParquetDecodingException, IOException {
+ ByteBuffer headerBuf =
stream.slice(ALP_HEADER_SIZE).order(ByteOrder.LITTLE_ENDIAN);
+ int compressionMode = headerBuf.get() & 0xFF;
+ int integerEncoding = headerBuf.get() & 0xFF;
+ int logVectorSize = headerBuf.get() & 0xFF;
+ int numElements = headerBuf.getInt();
+
+ if (compressionMode != ALP_COMPRESSION_MODE) {
+ throw new ParquetDecodingException("Unsupported ALP compression mode: "
+ compressionMode);
+ }
+ if (integerEncoding != ALP_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 (numElements < 0) {
+ throw new ParquetDecodingException("Invalid ALP element count: " +
numElements);
+ }
+ // ALP's num_elements is the count of non-null values that went through
encoding;
+ // valuesCount is the page row count, which is larger when the column has
nulls.
+ // The two are equal only for required (non-null) columns.
+ if (numElements > valuesCount) {
+ throw new ParquetDecodingException(
+ "ALP header element count " + numElements + " exceeds page
valuesCount " + valuesCount);
+ }
+
+ this.vectorSize = 1 << logVectorSize;
+ this.totalCount = numElements;
+ this.numVectors = (numElements + vectorSize - 1) / vectorSize;
+ this.pageValueIndex = 0;
+ this.currentVectorNumber = -1;
+
+ this.offsetArraySize = numVectors * Integer.BYTES;
+ ByteBuffer offsetBuf =
stream.slice(offsetArraySize).order(ByteOrder.LITTLE_ENDIAN);
+ this.vectorOffsets = new int[numVectors];
+ for (int v = 0; v < numVectors; v++) {
Review Comment:
These offsets are trusted without validating the first entry, ordering, or
bounds. Decode also ignores the next offset, so malformed bytes can silently
produce the wrong vector.
##########
parquet-column/src/test/java/org/apache/parquet/column/values/alp/AlpAdversarialTest.java:
##########
@@ -0,0 +1,393 @@
+/*
+ * 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.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.DirectByteBufferAllocator;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Adversarial tests for ALP readers: feed malformed page bytes and assert the
reader
+ * fails cleanly rather than crashing, producing silent garbage, or hanging.
+ *
+ * <p>"Fails cleanly" means raising a meaningful exception — preferably
+ * {@link ParquetDecodingException}, but at minimum a typed exception (not a
JVM-level
+ * crash, infinite loop, or wrong answer). The tests cover both:
+ * <ul>
+ * <li>Already-validated cases — the reader explicitly rejects these with a
+ * ParquetDecodingException carrying an explanatory message. These tests
pin
+ * the validation behavior in place.
+ * <li>Currently-unvalidated cases (truncation, corrupted offsets) — the
reader
+ * relies on the underlying ByteBuffer to surface
IndexOutOfBoundsException or
+ * BufferUnderflowException. These tests assert that some Throwable is
raised
+ * so the failure mode stays "loud" even if the explicit message is
missing.
+ * </ul>
+ */
+public class AlpAdversarialTest {
+
+ //
---------------------------------------------------------------------------
+ // Helpers: build a known-good encoded page, then mutate copies of it
+ //
---------------------------------------------------------------------------
+
+ /** Build a valid ALP-encoded double page with N clean values. */
+ private static byte[] validDoublePage(int valueCount, int vectorSize) throws
Exception {
+ AlpValuesWriter.DoubleAlpValuesWriter writer = null;
+ try {
+ int cap = Math.max(512, valueCount * 16);
+ writer = new AlpValuesWriter.DoubleAlpValuesWriter(cap, cap, new
DirectByteBufferAllocator(), vectorSize);
+ // 2-decimal values — the ALP sweet spot, ensures no exceptions
+ for (int i = 0; i < valueCount; i++) {
+ writer.writeDouble((i % 1000) / 100.0);
+ }
+ BytesInput bi = writer.getBytes();
+ ByteBuffer bb = bi.toByteBuffer();
+ byte[] out = new byte[bb.remaining()];
+ bb.duplicate().get(out);
+ return out;
+ } finally {
+ if (writer != null) {
+ writer.reset();
+ writer.close();
+ }
+ }
+ }
+
+ /** Build a valid ALP-encoded float page with N clean values. */
+ private static byte[] validFloatPage(int valueCount, int vectorSize) throws
Exception {
+ AlpValuesWriter.FloatAlpValuesWriter writer = null;
+ try {
+ int cap = Math.max(256, valueCount * 8);
+ writer = new AlpValuesWriter.FloatAlpValuesWriter(cap, cap, new
DirectByteBufferAllocator(), vectorSize);
+ for (int i = 0; i < valueCount; i++) {
+ writer.writeFloat((i % 1000) / 100.0f);
+ }
+ BytesInput bi = writer.getBytes();
+ ByteBuffer bb = bi.toByteBuffer();
+ byte[] out = new byte[bb.remaining()];
+ bb.duplicate().get(out);
+ return out;
+ } finally {
+ if (writer != null) {
+ writer.reset();
+ writer.close();
+ }
+ }
+ }
+
+ /** Sanity baseline: the known-good page actually decodes cleanly. */
+ @Test
+ public void sanityBaselineDecodesClean() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+ reader.initFromPage(32, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ for (int i = 0; i < 32; i++) reader.readDouble();
+ }
+
+ //
---------------------------------------------------------------------------
+ // Header-level validation (already-validated paths)
+ //
---------------------------------------------------------------------------
+
+ @Test
+ public void rejectsBadCompressionMode() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ page[0] = (byte) 0x99; // mode is at byte 0
+ ParquetDecodingException ex = assertThrows(ParquetDecodingException.class,
() -> {
+ new AlpValuesReaderForDouble().initFromPage(32,
ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ });
+ assertThat(ex.getMessage().toLowerCase().contains("compression"))
+ .as(ex.getMessage())
+ .isTrue();
+ }
+
+ @Test
+ public void rejectsBadIntegerEncoding() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ page[1] = (byte) 0x99; // integer_encoding is at byte 1
+ ParquetDecodingException ex = assertThrows(ParquetDecodingException.class,
() -> {
+ new AlpValuesReaderForDouble().initFromPage(32,
ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ });
+ assertThat(ex.getMessage().toLowerCase().contains("integer encoding"))
+ .as(ex.getMessage())
+ .isTrue();
+ }
+
+ @Test
+ public void rejectsLogVectorSizeTooLarge() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ page[2] = (byte) 99; // log_vector_size at byte 2
+ ParquetDecodingException ex = assertThrows(ParquetDecodingException.class,
() -> {
+ new AlpValuesReaderForDouble().initFromPage(32,
ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ });
+ assertThat(ex.getMessage().toLowerCase().contains("vector size"))
+ .as(ex.getMessage())
+ .isTrue();
+ }
+
+ @Test
+ public void rejectsLogVectorSizeTooSmall() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ page[2] = (byte) 2; // below MIN_LOG_VECTOR_SIZE=3
+ assertThrows(ParquetDecodingException.class, () -> {
+ new AlpValuesReaderForDouble().initFromPage(32,
ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ });
+ }
+
+ @Test
+ public void rejectsNegativeNumElements() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ // num_elements is int32 LE at bytes 3..6 — write -1
+ page[3] = (byte) 0xFF;
+ page[4] = (byte) 0xFF;
+ page[5] = (byte) 0xFF;
+ page[6] = (byte) 0xFF;
+ ParquetDecodingException ex = assertThrows(ParquetDecodingException.class,
() -> {
+ new AlpValuesReaderForDouble().initFromPage(32,
ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ });
+ assertThat(ex.getMessage().toLowerCase().contains("element count"))
+ .as(ex.getMessage())
+ .isTrue();
+ }
+
+ @Test
+ public void rejectsNumElementsGreaterThanValuesCount() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ // num_elements stays 32; pass valuesCount=10 (smaller than encoded count)
+ ParquetDecodingException ex = assertThrows(ParquetDecodingException.class,
() -> {
+ new AlpValuesReaderForDouble().initFromPage(10,
ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ });
+ assertThat(ex.getMessage().toLowerCase().contains("exceeds"))
+ .as(ex.getMessage())
+ .isTrue();
+ }
+
+ //
---------------------------------------------------------------------------
+ // Vector-level validation (already-validated paths, surface lazily on
decode)
+ //
---------------------------------------------------------------------------
+
+ /** Helper: find the byte position where the first vector's metadata starts.
*/
+ private static int firstVectorOffset(byte[] page) {
+ // header (7) + first 4 bytes of offset array = the offset value itself
+ int firstVectorOff =
+ ByteBuffer.wrap(page, 7, 4).order(ByteOrder.LITTLE_ENDIAN).getInt();
+ // offsets are measured from the start of the compression body (after the
7B header)
+ return 7 + firstVectorOff;
+ }
+
+ @Test
+ public void rejectsExponentTooHighDouble() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ int v0 = firstVectorOffset(page);
+ page[v0] = (byte) 99; // exponent byte
+ AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+ reader.initFromPage(32, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ ParquetDecodingException ex = assertThrows(ParquetDecodingException.class,
reader::readDouble);
+ assertThat(ex.getMessage().toLowerCase().contains("exponent"))
+ .as(ex.getMessage())
+ .isTrue();
+ }
+
+ @Test
+ public void rejectsExponentTooHighFloat() throws Exception {
+ byte[] page = validFloatPage(32, 16);
+ int v0 = firstVectorOffset(page);
+ page[v0] = (byte) 99;
+ AlpValuesReaderForFloat reader = new AlpValuesReaderForFloat();
+ reader.initFromPage(32, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ ParquetDecodingException ex = assertThrows(ParquetDecodingException.class,
reader::readFloat);
+ assertThat(ex.getMessage().toLowerCase().contains("exponent"))
+ .as(ex.getMessage())
+ .isTrue();
+ }
+
+ @Test
+ public void rejectsFactorGreaterThanExponent() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ int v0 = firstVectorOffset(page);
+ page[v0] = (byte) 2; // exponent
+ page[v0 + 1] = (byte) 5; // factor > exponent
+ AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+ reader.initFromPage(32, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ ParquetDecodingException ex = assertThrows(ParquetDecodingException.class,
reader::readDouble);
+ assertThat(ex.getMessage().toLowerCase().contains("factor"))
+ .as(ex.getMessage())
+ .isTrue();
+ }
+
+ @Test
+ public void rejectsTooManyExceptions() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ int v0 = firstVectorOffset(page);
+ // num_exceptions at v0+2, uint16 LE — set to 9999, way more than
vectorLen=16
+ page[v0 + 2] = (byte) (9999 & 0xFF);
+ page[v0 + 3] = (byte) ((9999 >>> 8) & 0xFF);
+ AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+ reader.initFromPage(32, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ ParquetDecodingException ex = assertThrows(ParquetDecodingException.class,
reader::readDouble);
+ assertThat(ex.getMessage().toLowerCase().contains("numexceptions"))
+ .as(ex.getMessage())
+ .isTrue();
+ }
+
+ @Test
+ public void rejectsBitWidthTooLargeDouble() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ int v0 = firstVectorOffset(page);
+ // Layout: ALP_INFO(4) + frameOfReference(8) then bitWidth byte at v0+12.
99 > 64.
+ page[v0 + 12] = (byte) 99;
+ AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+ reader.initFromPage(32, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ ParquetDecodingException ex = assertThrows(ParquetDecodingException.class,
reader::readDouble);
+ assertThat(ex.getMessage().toLowerCase().contains("bitwidth"))
+ .as(ex.getMessage())
+ .isTrue();
+ }
+
+ @Test
+ public void rejectsBitWidthTooLargeFloat() throws Exception {
+ byte[] page = validFloatPage(32, 16);
+ int v0 = firstVectorOffset(page);
+ // Layout: ALP_INFO(4) + frameOfReference(4) then bitWidth byte at v0+8.
99 > 32.
+ page[v0 + 8] = (byte) 99;
+ AlpValuesReaderForFloat reader = new AlpValuesReaderForFloat();
+ reader.initFromPage(32, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ ParquetDecodingException ex = assertThrows(ParquetDecodingException.class,
reader::readFloat);
+ assertThat(ex.getMessage().toLowerCase().contains("bitwidth"))
+ .as(ex.getMessage())
+ .isTrue();
+ }
+
+ //
---------------------------------------------------------------------------
+ // Currently-unvalidated paths: truncation and corrupted offsets
+ // These currently fail with low-level Throwables (BufferUnderflowException,
+ // IndexOutOfBoundsException). The tests assert any Throwable is raised so we
+ // notice if a regression silently swallows the corruption.
+ //
---------------------------------------------------------------------------
+
+ /** Page with only the 7-byte header — nothing else. */
+ @Test
+ public void rejectsHeaderOnlyPage() {
+ byte[] tiny = new byte[] {0x00, 0x00, 0x0A, 0x20, 0x00, 0x00, 0x00}; // 32
elements, log_vec=10
+ Throwable t = catchAny(() -> {
+ new AlpValuesReaderForDouble().initFromPage(32,
ByteBufferInputStream.wrap(ByteBuffer.wrap(tiny)));
+ });
+ assertThat(t).as("header-only page must raise").isNotNull();
+ }
+
+ @Test
+ public void rejectsPageTruncatedMidOffsetArray() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ // num_vectors = ceil(32/16) = 2, so offset array is 8 bytes. Truncate to
chop the 2nd offset.
+ byte[] truncated = new byte[7 + 4]; // header + first offset only
+ System.arraycopy(page, 0, truncated, 0, truncated.length);
+ Throwable t = catchAny(() -> {
+ new AlpValuesReaderForDouble().initFromPage(32,
ByteBufferInputStream.wrap(ByteBuffer.wrap(truncated)));
+ });
+ assertThat(t).as("truncated offset array must raise").isNotNull();
+ }
+
+ @Test
+ public void rejectsPageTruncatedMidVectorData() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ // chop the last 20 bytes — guaranteed to land in the middle of the second
vector
+ byte[] truncated = new byte[page.length - 20];
+ System.arraycopy(page, 0, truncated, 0, truncated.length);
+ AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+ // initFromPage should still succeed (truncation is inside the vectors
section,
+ // which initFromPage just slices without parsing). The failure surfaces
on decode.
+ reader.initFromPage(32,
ByteBufferInputStream.wrap(ByteBuffer.wrap(truncated)));
+ Throwable t = catchAny(() -> {
+ for (int i = 0; i < 32; i++) reader.readDouble();
+ });
+ assertThat(t).as("truncated vector data must raise on read").isNotNull();
+ }
+
+ @Test
+ public void rejectsCorruptedOffsetPointingPastEnd() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ // Offset array starts at byte 7. Overwrite the first offset (uint32 LE)
with a huge value.
+ page[7] = (byte) 0xFF;
+ page[8] = (byte) 0xFF;
+ page[9] = (byte) 0xFF;
+ page[10] = (byte) 0x7F;
+ AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+ reader.initFromPage(32, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ Throwable t = catchAny(() -> reader.readDouble());
+ assertThat(t).as("corrupted offset must raise on decode").isNotNull();
+ }
+
+ //
---------------------------------------------------------------------------
+ // skip() / read() bounds
+ //
---------------------------------------------------------------------------
+
+ @Test
+ public void rejectsSkipPastEnd() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+ reader.initFromPage(32, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ assertThrows(ParquetDecodingException.class, () -> reader.skip(33));
+ }
+
+ @Test
+ public void rejectsNegativeSkip() throws Exception {
+ byte[] page = validDoublePage(32, 16);
+ AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+ reader.initFromPage(32, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ assertThrows(ParquetDecodingException.class, () -> reader.skip(-1));
+ }
+
+ @Test
+ public void rejectsReadPastEnd() throws Exception {
+ byte[] page = validDoublePage(8, 8);
+ AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+ reader.initFromPage(8, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
+ for (int i = 0; i < 8; i++) reader.readDouble();
+ ParquetDecodingException ex = assertThrows(ParquetDecodingException.class,
reader::readDouble);
+ assertThat(ex.getMessage().toLowerCase().contains("exhausted"))
+ .as(ex.getMessage())
+ .isTrue();
+ }
+
+ //
---------------------------------------------------------------------------
+ // Utility
+ //
---------------------------------------------------------------------------
+
+ @FunctionalInterface
+ private interface ThrowingRunnable {
+ void run() throws Throwable;
+ }
+
+ /** Catch any Throwable (including low-level RuntimeExceptions / Errors). */
+ private static Throwable catchAny(ThrowingRunnable r) {
Review Comment:
Catching Throwable lets OOM and assertion failures count as success. This
test can pass without proving clean rejection.
##########
parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpConstants.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.Preconditions;
+
+/**
+ * Constants for the ALP (Adaptive Lossless floating-Point) encoding.
+ *
+ * <p>ALP encoding converts floating-point values to integers using decimal
scaling,
+ * then applies Frame of Reference encoding and bit-packing.
+ * Values that cannot be losslessly converted are stored as exceptions.
+ *
+ * <p>Based on the paper: "ALP: Adaptive Lossless floating-Point Compression"
(SIGMOD 2024)
+ *
+ * @see <a href="https://dl.acm.org/doi/10.1145/3626717";>ALP Paper</a>
+ */
+public final class AlpConstants {
Review Comment:
Does it need to be public or just default visibility?
##########
parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpEncoderDecoder.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * 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 org.apache.parquet.bytes.BytesUtils;
+
+/**
+ * Core ALP (Adaptive Lossless floating-Point) encoding and decoding logic.
+ *
+ * <p>ALP works by converting floating-point values to integers using decimal
scaling,
+ * then applying Frame of Reference encoding and bit-packing.
+ * Values that cannot be losslessly converted are stored as exceptions.
+ *
+ * <p>Encoding formula: encoded = fastRound(value * POW10[e] *
POW10_NEGATIVE[f])
+ * <p>Decoding formula: value = encoded * POW10[f] * POW10_NEGATIVE[e]
+ *
+ * <p>The order of operations is critical for IEEE 754 correctness. Both
formulas must
+ * be evaluated as single expressions — storing the intermediate
multiplication result
+ * in a variable before the second multiply changes IEEE 754 rounding and
produces extra
+ * exceptions. Likewise, scaling uses multiply-by-reciprocal (via
POW10_NEGATIVE) rather than
+ * division: this reproduces the exact IEEE 754 rounding of the ALP reference
algorithm, so the
+ * encoded integers — and therefore which values become exceptions and the
resulting bytes — are
+ * identical across implementations. It is about cross-implementation
determinism, not any one
+ * language.
+ *
+ * <p>Exception conditions:
+ * <ul>
+ * <li>NaN values</li>
+ * <li>Infinity values</li>
+ * <li>Negative zero (-0.0)</li>
+ * <li>Out of integer range</li>
+ * <li>Round-trip failure (decode(encode(v)) != v)</li>
+ * </ul>
+ */
+final class AlpEncoderDecoder {
+
+ private static final double ENCODING_UPPER_LIMIT = 9223372036854774784.0;
+ private static final double ENCODING_LOWER_LIMIT = -9223372036854774784.0;
+ private static final float FLOAT_ENCODING_UPPER_LIMIT = 2147483520.0f;
+ private static final float FLOAT_ENCODING_LOWER_LIMIT = -2147483520.0f;
+
+ private AlpEncoderDecoder() {
+ // Utility class
+ }
+
+ /** NaN, Inf, and -0.0 can never be encoded regardless of exponent/factor. */
+ static boolean isIntrinsicFloatException(float value) {
+ if (Float.isNaN(value)) {
+ return true;
+ }
+ if (Float.isInfinite(value)) {
+ return true;
+ }
+ return Float.floatToRawIntBits(value) == FLOAT_NEGATIVE_ZERO_BITS;
+ }
+
+ /** Full exception check for a given (exponent, factor): intrinsic cases
plus round-trip failure. */
+ static boolean isFloatException(float value, int exponent, int factor) {
+ if (isIntrinsicFloatException(value)) {
+ return true;
+ }
+ // Check before rounding: overflow or non-finite after scaling
+ float scaled = value * FLOAT_POW10[exponent] *
FLOAT_POW10_NEGATIVE[factor];
+ if (!Float.isFinite(scaled) || scaled > FLOAT_ENCODING_UPPER_LIMIT ||
scaled < FLOAT_ENCODING_LOWER_LIMIT) {
+ return true;
+ }
+ int encoded = encodeFloat(value, exponent, factor);
+ float decoded = decodeFloat(encoded, exponent, factor);
+ return Float.floatToRawIntBits(value) != Float.floatToRawIntBits(decoded);
+ }
+
+ /** Round float to nearest integer using magic-number trick with sign
branching. */
+ static int fastRoundFloat(float value) {
+ if (value >= 0) {
+ return (int) ((value + MAGIC_FLOAT) - MAGIC_FLOAT);
+ } else {
+ return (int) ((value - MAGIC_FLOAT) + MAGIC_FLOAT);
+ }
+ }
+
+ static int encodeFloat(float value, int exponent, int factor) {
+ return fastRoundFloat(value * FLOAT_POW10[exponent] *
FLOAT_POW10_NEGATIVE[factor]);
+ }
+
+ static float decodeFloat(int encoded, int exponent, int factor) {
+ return encoded * FLOAT_POW10[factor] * FLOAT_POW10_NEGATIVE[exponent];
+ }
+
+ /** NaN, Inf, and -0.0 can never be encoded regardless of exponent/factor. */
+ static boolean isIntrinsicDoubleException(double value) {
+ if (Double.isNaN(value)) {
+ return true;
+ }
+ if (Double.isInfinite(value)) {
+ return true;
+ }
+ return Double.doubleToRawLongBits(value) == DOUBLE_NEGATIVE_ZERO_BITS;
+ }
+
+ /** Full exception check for a given (exponent, factor): intrinsic cases
plus round-trip failure. */
+ static boolean isDoubleException(double value, int exponent, int factor) {
+ if (isIntrinsicDoubleException(value)) {
+ return true;
+ }
+ // Check before rounding: overflow or non-finite after scaling
+ double scaled = value * DOUBLE_POW10[exponent] *
DOUBLE_POW10_NEGATIVE[factor];
+ if (!Double.isFinite(scaled) || scaled > ENCODING_UPPER_LIMIT || scaled <
ENCODING_LOWER_LIMIT) {
+ return true;
+ }
+ long encoded = encodeDouble(value, exponent, factor);
+ double decoded = decodeDouble(encoded, exponent, factor);
+ return Double.doubleToRawLongBits(value) !=
Double.doubleToRawLongBits(decoded);
+ }
+
+ /** Round double to nearest integer using magic-number trick with sign
branching. */
+ static long fastRoundDouble(double value) {
+ if (value >= 0) {
+ return (long) ((value + MAGIC_DOUBLE) - MAGIC_DOUBLE);
+ } else {
+ return (long) ((value - MAGIC_DOUBLE) + MAGIC_DOUBLE);
+ }
+ }
+
+ static long encodeDouble(double value, int exponent, int factor) {
+ return fastRoundDouble(value * DOUBLE_POW10[exponent] *
DOUBLE_POW10_NEGATIVE[factor]);
+ }
+
+ static double decodeDouble(long encoded, int exponent, int factor) {
+ return encoded * DOUBLE_POW10[factor] * DOUBLE_POW10_NEGATIVE[exponent];
+ }
+
+ public static class EncodingParams {
+ final int exponent;
+ final int factor;
+ final int numExceptions;
+
+ EncodingParams(int exponent, int factor, int numExceptions) {
+ this.exponent = exponent;
+ this.factor = factor;
+ this.numExceptions = numExceptions;
+ }
+ }
+
+ // Index positions within an (exponent, factor) pair array.
+ private static final int E = 0;
+ private static final int F = 1;
+
+ // All valid (exponent, factor) pairs for the full search, precomputed in
nested-loop order
+ // (e ascending, then f ascending) so the tie-break and early-exit behave
exactly like an inline
+ // double loop. Reused across every vector to avoid per-call allocation.
+ private static final int[][] ALL_VALID_FLOAT_PAIRS =
buildAllPairs(FLOAT_MAX_EXPONENT);
+ private static final int[][] ALL_VALID_DOUBLE_PAIRS =
buildAllPairs(DOUBLE_MAX_EXPONENT);
+
+ private static int[][] buildAllPairs(int maxExponent) {
+ int count = (maxExponent + 1) * (maxExponent + 2) / 2;
+ int[][] pairs = new int[count][];
+ int idx = 0;
+ for (int e = 0; e <= maxExponent; e++) {
+ for (int f = 0; f <= e; f++) {
+ pairs[idx++] = new int[] {e, f};
+ }
+ }
+ return pairs;
+ }
+
+ /** Try all (exponent, factor) combos and pick the one with the smallest
estimated compressed size. */
+ static EncodingParams findBestFloatParams(float[] values, int offset, int
length) {
+ return pickBestFloat(values, offset, length, ALL_VALID_FLOAT_PAIRS);
+ }
+
+ /** Same as findBestFloatParams but only tries the cached preset combos. */
+ static EncodingParams findBestFloatParamsWithPresets(float[] values, int
offset, int length, int[][] presets) {
+ return pickBestFloat(values, offset, length, presets);
+ }
+
+ /**
+ * Scores each (exponent, factor) pair by estimated compressed size and
returns the best.
+ *
+ * <p>Estimated size (in bits) = {@code length * bitWidth + exceptions *
(Float.SIZE + Short.SIZE)},
+ * where bitWidth is the number of bits needed to represent the signed range
(max - min) of
+ * non-exception encoded values after frame-of-reference subtraction,
matching the writer's FOR
+ * packing. This produces better compression ratios than minimizing
exception count alone. Ties in
+ * size are broken toward the higher exponent, then the higher factor. The
first pair (in {@code
+ * pairs} order) that encodes every value into a single FOR value with zero
exceptions wins
+ * outright. When no pair yields any non-exception values, {@code pairs[0]}
is returned.
+ */
+ private static EncodingParams pickBestFloat(float[] values, int offset, int
length, int[][] pairs) {
+ int bestExponent = pairs[0][E];
+ int bestFactor = pairs[0][F];
+ int bestExceptions = length;
+ long bestEstimatedSize = Long.MAX_VALUE;
+
+ for (int[] pair : pairs) {
+ int e = pair[E];
+ int f = pair[F];
+ int exceptions = 0;
+ int minEncoded = Integer.MAX_VALUE;
+ int maxEncoded = Integer.MIN_VALUE;
+ for (int i = 0; i < length; i++) {
+ float value = values[offset + i];
+ if (isFloatException(value, e, f)) {
+ exceptions++;
+ } else {
+ int encoded = encodeFloat(value, e, f);
Review Comment:
I'm not sure if this is a performance concern. But `isFloatException(value,
e, f)` above has already called `encodeFloat(value, e, f)` internally. Should
we consider a method like `tryEncodeFloat` to return a two-state value
(Optional?): either good encoded value or indicate it is an exception?
##########
parquet-column/src/test/java/org/apache/parquet/column/values/alp/AlpValuesEndToEndTest.java:
##########
@@ -0,0 +1,2208 @@
+/*
+ * 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.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.DirectByteBufferAllocator;
+import org.apache.parquet.column.values.bitpacking.BytePacker;
+import org.apache.parquet.column.values.bitpacking.Packer;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+/**
+ * End-to-end tests for ALP encoding and decoding pipeline.
+ */
+public class AlpValuesEndToEndTest {
+
+ private static final int DEFAULT_VECTOR_SIZE =
AlpConstants.DEFAULT_VECTOR_SIZE;
+
+ // ========== Helper Methods ==========
+
+ /**
+ * Round-trip with STRICT raw-bit comparison for every value, including NaN
payloads (the regular
+ * roundTrip helpers only assert isNaN, which would hide a
NaN-payload-preservation bug). ALP must
+ * be losslessly bit-exact for every possible IEEE-754 bit pattern.
+ */
+ private void roundTripDoubleStrict(double[] values) throws Exception {
+ AlpValuesWriter.DoubleAlpValuesWriter writer = null;
+ try {
+ int capacity = Math.max(512, values.length * 16);
+ writer = new AlpValuesWriter.DoubleAlpValuesWriter(
+ capacity, capacity, new DirectByteBufferAllocator(),
DEFAULT_VECTOR_SIZE);
+ for (double v : values) {
+ writer.writeDouble(v);
+ }
+ BytesInput input = writer.getBytes();
+ AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+ reader.initFromPage(values.length,
ByteBufferInputStream.wrap(input.toByteBuffer()));
+ for (int i = 0; i < values.length; i++) {
+ long exp = Double.doubleToRawLongBits(values[i]);
+ long act = Double.doubleToRawLongBits(reader.readDouble());
+ assertThat(act)
+ .as("Raw-bit mismatch at index " + i + " expectedBits=0x" +
Long.toHexString(exp)
+ + " actualBits=0x" + Long.toHexString(act))
+ .isEqualTo(exp);
+ }
+ } finally {
+ if (writer != null) {
+ writer.reset();
+ writer.close();
+ }
+ }
+ }
+
+ private void roundTripFloatStrict(float[] values) throws Exception {
+ AlpValuesWriter.FloatAlpValuesWriter writer = null;
+ try {
+ int capacity = Math.max(256, values.length * 8);
+ writer = new AlpValuesWriter.FloatAlpValuesWriter(
+ capacity, capacity, new DirectByteBufferAllocator(),
DEFAULT_VECTOR_SIZE);
+ for (float v : values) {
+ writer.writeFloat(v);
+ }
+ BytesInput input = writer.getBytes();
+ AlpValuesReaderForFloat reader = new AlpValuesReaderForFloat();
+ reader.initFromPage(values.length,
ByteBufferInputStream.wrap(input.toByteBuffer()));
+ for (int i = 0; i < values.length; i++) {
+ int exp = Float.floatToRawIntBits(values[i]);
+ int act = Float.floatToRawIntBits(reader.readFloat());
+ assertThat(act)
+ .as("Raw-bit mismatch at index " + i + " expectedBits=0x" +
Integer.toHexString(exp)
+ + " actualBits=0x" + Integer.toHexString(act))
+ .isEqualTo(exp);
+ }
+ } finally {
+ if (writer != null) {
+ writer.reset();
+ writer.close();
+ }
+ }
+ }
+
+ // ========== Full-bit-space fuzz (lossless invariant) ==========
+
+ @Test
+ public void testDoubleFullBitSpaceFuzz() throws Exception {
+ // Sweep the entire double bit space: random raw longs -> double. Covers
all subnormals, every
+ // NaN payload, +/-0, +/-Inf, extreme exponents, chaotic mixed magnitudes
within a vector.
+ Random rng = new Random(0x9E3779B97F4A7C15L);
+ for (int v = 0; v < 300; v++) { // ~300k values
+ double[] values = new double[1024];
+ for (int i = 0; i < values.length; i++) {
+ values[i] = Double.longBitsToDouble(rng.nextLong());
+ }
+ roundTripDoubleStrict(values);
+ }
+ }
+
+ @Test
+ public void testFloatFullBitSpaceFuzz() throws Exception {
+ Random rng = new Random(0x9E3779B9L);
+ for (int v = 0; v < 300; v++) {
+ float[] values = new float[1024];
+ for (int i = 0; i < values.length; i++) {
+ values[i] = Float.intBitsToFloat(rng.nextInt());
+ }
+ roundTripFloatStrict(values);
+ }
+ }
+
+ @Test
+ public void testDoubleMixedFuzz() throws Exception {
+ // Mix ALP-friendly decimal-ish values with random raw bits, so FOR
encoding AND the exception
+ // path are both exercised on chaotic data within the same vector.
+ Random rng = new Random(0xD1CE5EEDL);
+ for (int v = 0; v < 300; v++) {
+ double[] values = new double[1024];
+ for (int i = 0; i < values.length; i++) {
+ if (rng.nextInt(4) == 0) {
+ values[i] = Double.longBitsToDouble(rng.nextLong()); // ~25% chaotic
(mostly exceptions)
+ } else {
+ values[i] = Math.round(rng.nextDouble() * 1_000_000.0) / 100.0; //
2-decimal, ALP-friendly
+ }
+ }
+ roundTripDoubleStrict(values);
+ }
+ }
+
+ @Test
+ public void testDoubleDistributionShiftWithinRowGroup() throws Exception {
+ // First vectors are clean 2-decimal data (the sampler builds its preset
(e,f) cache from these),
+ // then later vectors switch to high-precision / very different magnitude
that the cached presets
+ // fit poorly. Once presets are cached the writer only tries those, so ALP
must still stay lossless
+ // (the exception path catches every preset mismatch).
+ Random rng = new Random(7);
+ int totalVectors = 40; // well past SAMPLER_SAMPLE_VECTORS_PER_ROWGROUP so
presets are in use
+ double[] values = new double[totalVectors * 1024];
+ for (int i = 0; i < values.length; i++) {
+ int vec = i / 1024;
+ if (vec < 12) {
+ values[i] = Math.round(rng.nextDouble() * 10000.0) / 100.0; // clean
cents
+ } else {
+ values[i] = rng.nextDouble() * 1e12 + rng.nextDouble(); //
high-precision, big magnitude
+ }
+ }
+ roundTripDoubleStrict(values);
+ }
+
+ // ========== Reader robustness against malformed input ==========
+
+ private void readAllDoubles(byte[] bytes, int valueCount) throws Exception {
+ AlpValuesReaderForDouble reader = new AlpValuesReaderForDouble();
+ reader.initFromPage(valueCount,
ByteBufferInputStream.wrap(ByteBuffer.wrap(bytes)));
+ for (int i = 0; i < valueCount; i++) {
+ reader.readDouble();
+ }
+ }
+
+ @Test
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ public void testReaderRejectsCorruptInputCleanly() throws Exception {
+ // Build a valid ALP double page, then feed corrupted variants. The reader
must fail cleanly
+ // (a catchable exception) and never hang, OOM, or read out of bounds
silently.
+ double[] values = new double[2048];
+ for (int i = 0; i < values.length; i++) {
+ values[i] = (i % 100) / 100.0;
+ }
+ AlpValuesWriter.DoubleAlpValuesWriter writer = new
AlpValuesWriter.DoubleAlpValuesWriter(
+ 65536, 65536, new DirectByteBufferAllocator(), DEFAULT_VECTOR_SIZE);
+ for (double v : values) {
+ writer.writeDouble(v);
+ }
+ byte[] valid = writer.getBytes().toByteArray();
+ writer.reset();
+ writer.close();
+
+ // Sanity: the valid page reads fine.
+ readAllDoubles(valid, values.length);
+
+ // Corruptions that must each fail cleanly (never crash/hang/OOB):
+ java.util.List<byte[]> corrupt = new java.util.ArrayList<>();
+ corrupt.add(java.util.Arrays.copyOf(valid, valid.length / 2)); //
truncated to half
+ corrupt.add(java.util.Arrays.copyOf(valid, 3)); // truncated to a stub
+ corrupt.add(new byte[0]); // empty
+ for (int pos : new int[] {0, 1, 2, 5, 7, 11, 20, valid.length - 1}) {
+ byte[] c = valid.clone();
+ c[pos] = (byte) ~c[pos]; // flip a header/body byte
+ corrupt.add(c);
+ }
+ for (byte[] c : corrupt) {
+ try {
+ readAllDoubles(c, values.length);
+ // Some single-byte flips may still decode to (wrong but in-bounds)
values without throwing;
+ // that is acceptable here. What matters is no crash/hang/OOB, which
we reached this line.
+ } catch (OutOfMemoryError oom) {
+ fail("Malformed input caused an OutOfMemoryError (allocation bomb) - a
corrupt size/count "
+ + "must not drive an unbounded allocation");
+ } catch (Throwable t) {
+ // Any ordinary catchable exception (EOFException,
ParquetDecodingException, IndexOutOfBounds,
+ // BufferUnderflow, NegativeArraySize, ...) is a clean failure: no JVM
crash, no OOB, no hang.
+ }
+ }
+
+ // Claiming far more elements than the data supports must be rejected
without an OOM allocation.
+ try {
+ readAllDoubles(valid, Integer.MAX_VALUE / 2);
Review Comment:
This changes valuesCount, not the ALP header num_elements. It does not
exercise the allocation or overflow path.
##########
parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesWriter.java:
##########
@@ -0,0 +1,647 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.parquet.bytes.ByteBufferAllocator;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.BytesUtils;
+import org.apache.parquet.bytes.CapacityByteArrayOutputStream;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.values.ValuesWriter;
+import org.apache.parquet.column.values.bitpacking.BytePacker;
+import org.apache.parquet.column.values.bitpacking.BytePackerForLong;
+import org.apache.parquet.column.values.bitpacking.Packer;
+
+/**
+ * ALP (Adaptive Lossless floating-Point) values writer.
+ *
+ * <p>ALP encoding converts floating-point values to integers using decimal
scaling,
+ * then applies Frame of Reference encoding and bit-packing.
+ * Values that cannot be losslessly converted are stored as exceptions.
+ *
+ * <p>Writing is incremental: values are buffered in a fixed-size vector
buffer,
+ * and each full vector is encoded and flushed to the output stream
immediately.
+ * On {@link #getBytes()}, any remaining partial vector is flushed, and the
+ * final page bytes are assembled.
+ *
+ * <p>Interleaved Page Layout:
+ * <pre>
+ * ┌─────────┬──────────────────────┬──────────────┬──────────────┬─────┐
+ * │ Header │ Offset Array │ Vector 0 │ Vector 1 │ ... │
+ * │ 7 bytes │ 4B × numVectors │ (interleaved)│ (interleaved)│ │
+ * └─────────┴──────────────────────┴──────────────┴──────────────┴─────┘
+ * </pre>
+ *
+ * <p>Each vector contains interleaved:
+ * AlpInfo(4B) + ForInfo(5B/9B) + PackedValues + ExceptionPositions +
ExceptionValues
+ */
+public abstract class AlpValuesWriter extends ValuesWriter {
+
+ protected final int initialCapacity;
+ protected final int pageSize;
+ protected final ByteBufferAllocator allocator;
+ protected final int vectorSize;
+ protected final int logVectorSize;
+
+ AlpValuesWriter(int initialCapacity, int pageSize, ByteBufferAllocator
allocator, int vectorSize) {
+ AlpConstants.validateVectorSize(vectorSize);
+ this.initialCapacity = initialCapacity;
+ this.pageSize = pageSize;
+ this.allocator = allocator;
+ this.vectorSize = vectorSize;
+ this.logVectorSize = Integer.numberOfTrailingZeros(vectorSize);
+ }
+
+ @Override
+ public Encoding getEncoding() {
+ return Encoding.ALP;
+ }
+
+ /** Float writer. Buffers one vector at a time, encodes and flushes when
full. */
+ public static class FloatAlpValuesWriter extends AlpValuesWriter {
+ private final float[] vectorBuffer;
+ private int bufferCount;
+ private int totalCount;
+ private CapacityByteArrayOutputStream encodedVectors;
+ private final List<Integer> vectorByteSizes;
+
+ // Preset caching: collect evenly-spaced sample vectors across the
rowgroup,
+ // then build presets using estimated compressed size (matching C++
AlpSampler).
+ private int vectorsProcessed;
+ private int[][] cachedPresets;
+ // Winning (exponent, factor) pairs from sampled vectors, tallied later
into the preset cache.
+ private final List<int[]> sampledParams;
+ private final int rowgroupSampleJump;
+
+ // Reusable per-vector buffers
+ private final int[] encodedBuffer;
+ private final short[] excPosBuffer;
+ private final float[] excValBuffer;
+ private final byte[] metadataBuf;
+ private final byte[] packBuf;
+ private final int[] packPadBuf;
+
+ public FloatAlpValuesWriter(int initialCapacity, int pageSize,
ByteBufferAllocator allocator) {
+ this(initialCapacity, pageSize, allocator, DEFAULT_VECTOR_SIZE);
+ }
+
+ public FloatAlpValuesWriter(int initialCapacity, int pageSize,
ByteBufferAllocator allocator, int vectorSize) {
+ super(initialCapacity, pageSize, allocator, vectorSize);
+ this.vectorBuffer = new float[vectorSize];
+ this.bufferCount = 0;
+ this.totalCount = 0;
+ this.encodedVectors = new CapacityByteArrayOutputStream(initialCapacity,
pageSize, allocator);
+ this.vectorByteSizes = new ArrayList<>();
+ this.vectorsProcessed = 0;
+ this.cachedPresets = null;
+ this.sampledParams = new ArrayList<>();
+ // Space samples evenly: one sample every jump vectors across the
rowgroup.
+ // Math.max(1, ...) guards against very small rowgroups or large vector
sizes.
+ this.rowgroupSampleJump =
+ Math.max(1, SAMPLER_ROWGROUP_SIZE /
SAMPLER_SAMPLE_VECTORS_PER_ROWGROUP / vectorSize);
+ // Pre-allocate reusable buffers
+ this.encodedBuffer = new int[vectorSize];
+ this.excPosBuffer = new short[vectorSize];
+ this.excValBuffer = new float[vectorSize];
+ this.metadataBuf = new byte[Math.max(ALP_INFO_SIZE,
FLOAT_FOR_INFO_SIZE)];
+ this.packBuf = new byte[Integer.SIZE]; // max bit width for int
+ this.packPadBuf = new int[PACK_GROUP_SIZE];
+ }
+
+ @Override
+ public void writeFloat(float v) {
+ vectorBuffer[bufferCount++] = v;
+ totalCount++;
+ if (bufferCount == vectorSize) {
+ encodeAndFlushVector(bufferCount);
+ bufferCount = 0;
+ }
+ }
+
+ private void encodeAndFlushVector(int vectorLen) {
+ // Sampling phase first (full search + collect evenly-spaced samples,
then build the preset
+ // cache once enough are gathered); after the cache is built, later
vectors take the else branch.
+ AlpEncoderDecoder.EncodingParams params;
+ if (cachedPresets == null) {
+ params = AlpEncoderDecoder.findBestFloatParams(vectorBuffer, 0,
vectorLen);
+ // Collect one sample every rowgroupSampleJump vectors so that samples
are
+ // evenly distributed across the rowgroup (matching C++ AlpSampler
spacing).
+ if (vectorsProcessed % rowgroupSampleJump == 0
+ && sampledParams.size() < SAMPLER_SAMPLE_VECTORS_PER_ROWGROUP) {
+ sampledParams.add(new int[] {params.exponent, params.factor});
+ }
+ if (sampledParams.size() >= SAMPLER_SAMPLE_VECTORS_PER_ROWGROUP) {
+ buildPresetCache();
+ }
+ } else {
+ params =
AlpEncoderDecoder.findBestFloatParamsWithPresets(vectorBuffer, 0, vectorLen,
cachedPresets);
Review Comment:
`vectorsProcessed` resets for every page. With the default page size, we
only collect about two samples, but need eight to build the cache. So
`cachedPresets` is never used in the normal path.
##########
parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV1ValuesWriterFactory.java:
##########
@@ -147,7 +148,13 @@ private ValuesWriter getInt96ValuesWriter(ColumnDescriptor
path) {
private ValuesWriter getDoubleValuesWriter(ColumnDescriptor path) {
final ValuesWriter fallbackWriter;
- if (this.parquetProperties.isByteStreamSplitEnabled(path)) {
+ if (this.parquetProperties.isAlpEnabled(path)) {
Review Comment:
+1 on `keep ALP registered in both V1 and V2` to avoid confusion and it is
aligned with the change of `isByteStreamSplitEnabled`.
##########
parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesReader.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * 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>Reads ALP-encoded values from the interleaved page layout:
+ * <pre>
+ * ┌─────────┬──────────────────────┬──────────────┬──────────────┬─────┐
+ * │ Header │ Offset Array │ Vector 0 │ Vector 1 │ ... │
+ * │ 7 bytes │ 4B × numVectors │ (interleaved)│ (interleaved)│ │
+ * └─────────┴──────────────────────┴──────────────┴──────────────┴─────┘
+ * </pre>
+ *
+ * <p>Each vector is decoded lazily on first access. Skipping values does not
+ * trigger decoding of intermediate vectors.
+ */
+abstract class AlpValuesReader extends ValuesReader {
+
+ protected int vectorSize;
+ protected int totalCount;
+ protected int numVectors;
+ protected int pageValueIndex;
+ protected int currentVectorNumber;
+
+ protected int[] vectorOffsets;
+ protected ByteBuffer vectorsData;
+ protected int offsetArraySize;
+
+ // Scratch buffer for exception positions within a vector; shared by both
readers (int[] in each).
+ protected int[] excPositionsBuffer;
+
+ AlpValuesReader() {
+ this.pageValueIndex = 0;
+ this.totalCount = 0;
+ this.currentVectorNumber = -1;
+ }
+
+ @Override
+ public void initFromPage(int valuesCount, ByteBufferInputStream stream)
+ throws ParquetDecodingException, IOException {
+ ByteBuffer headerBuf =
stream.slice(ALP_HEADER_SIZE).order(ByteOrder.LITTLE_ENDIAN);
+ int compressionMode = headerBuf.get() & 0xFF;
+ int integerEncoding = headerBuf.get() & 0xFF;
+ int logVectorSize = headerBuf.get() & 0xFF;
+ int numElements = headerBuf.getInt();
+
+ if (compressionMode != ALP_COMPRESSION_MODE) {
+ throw new ParquetDecodingException("Unsupported ALP compression mode: "
+ compressionMode);
+ }
+ if (integerEncoding != ALP_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 (numElements < 0) {
+ throw new ParquetDecodingException("Invalid ALP element count: " +
numElements);
+ }
+ // ALP's num_elements is the count of non-null values that went through
encoding;
+ // valuesCount is the page row count, which is larger when the column has
nulls.
+ // The two are equal only for required (non-null) columns.
+ if (numElements > valuesCount) {
+ throw new ParquetDecodingException(
+ "ALP header element count " + numElements + " exceeds page
valuesCount " + valuesCount);
+ }
+
+ this.vectorSize = 1 << logVectorSize;
+ this.totalCount = numElements;
+ this.numVectors = (numElements + vectorSize - 1) / vectorSize;
+ this.pageValueIndex = 0;
+ this.currentVectorNumber = -1;
+
+ this.offsetArraySize = numVectors * Integer.BYTES;
+ ByteBuffer offsetBuf =
stream.slice(offsetArraySize).order(ByteOrder.LITTLE_ENDIAN);
+ this.vectorOffsets = new int[numVectors];
+ for (int v = 0; v < numVectors; v++) {
+ vectorOffsets[v] = offsetBuf.getInt();
+ }
+
+ // Slice remaining bytes into a 0-based view so decodeVector can use
+ // absolute get methods (vectorsData.get(pos)) directly.
+ int remainingBytes = (int) stream.available();
+ ByteBuffer rawSlice = stream.slice(remainingBytes);
+ this.vectorsData = rawSlice.slice().order(ByteOrder.LITTLE_ENDIAN);
+
+ allocateDecodedBuffer(vectorSize);
+ this.excPositionsBuffer = new int[vectorSize];
+ }
+
+ protected int getVectorLength(int vectorNumber) {
+ if (vectorNumber < numVectors - 1) {
+ return vectorSize;
+ }
+ // Last vector may be partial
+ int lastVectorLen = totalCount % vectorSize;
+ return lastVectorLen == 0 ? vectorSize : lastVectorLen;
+ }
+
+ // Offsets in the page are relative to the compression body (after header),
+ // but vectorsData starts after the offset array, so adjust.
+ protected int getVectorDataPosition(int vectorNumber) {
+ return vectorOffsets[vectorNumber] - offsetArraySize;
+ }
+
+ @Override
+ public void skip() {
+ skip(1);
+ }
+
+ @Override
+ public void skip(int n) {
+ if (n < 0 || pageValueIndex + n > totalCount) {
Review Comment:
This addition can overflow. A large skip can pass the check and make
pageValueIndex negative.
##########
parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpConstants.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.Preconditions;
+
+/**
+ * Constants for the ALP (Adaptive Lossless floating-Point) encoding.
+ *
+ * <p>ALP encoding converts floating-point values to integers using decimal
scaling,
+ * then applies Frame of Reference encoding and bit-packing.
+ * Values that cannot be losslessly converted are stored as exceptions.
+ *
+ * <p>Based on the paper: "ALP: Adaptive Lossless floating-Point Compression"
(SIGMOD 2024)
+ *
+ * @see <a href="https://dl.acm.org/doi/10.1145/3626717";>ALP Paper</a>
+ */
+public final class AlpConstants {
+
+ private AlpConstants() {
+ // Utility class
+ }
+
+ // Page header fields
+ public static final int ALP_COMPRESSION_MODE = 0;
+ public static final int ALP_INTEGER_ENCODING_FOR = 0;
+ public static final int ALP_HEADER_SIZE = 7;
+
+ public static final int DEFAULT_VECTOR_SIZE = 1024;
+ public static final int DEFAULT_VECTOR_SIZE_LOG = 10;
+
+ // BytePacker packs/unpacks 8 values at a time (pack8Values/unpack8Values).
+ static final int PACK_GROUP_SIZE = 8;
+
+ // Capped at 15 (vectorSize=32768) because num_exceptions is uint16,
+ // so vectorSize must not exceed 65535 to avoid overflow when all values are
exceptions.
+ static final int MAX_LOG_VECTOR_SIZE = 15;
+ static final int MIN_LOG_VECTOR_SIZE = 3;
+
+ static final int FLOAT_MAX_EXPONENT = 10;
+ static final int DOUBLE_MAX_EXPONENT = 18;
+
+ // Sampler constants matching C++ AlpConstants.
+ // Sample SAMPLER_SAMPLE_VECTORS_PER_ROWGROUP vectors evenly distributed
across a rowgroup
+ // of SAMPLER_ROWGROUP_SIZE values, then lock in top MAX_PRESET_COMBINATIONS
combos.
+ static final int SAMPLER_ROWGROUP_SIZE = 122_880;
Review Comment:
Unless shared by both encoder and decoder, variables like these are better
placed in the file that directly uses them.
##########
parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesWriter.java:
##########
@@ -0,0 +1,647 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.parquet.bytes.ByteBufferAllocator;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.BytesUtils;
+import org.apache.parquet.bytes.CapacityByteArrayOutputStream;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.values.ValuesWriter;
+import org.apache.parquet.column.values.bitpacking.BytePacker;
+import org.apache.parquet.column.values.bitpacking.BytePackerForLong;
+import org.apache.parquet.column.values.bitpacking.Packer;
+
+/**
+ * ALP (Adaptive Lossless floating-Point) values writer.
+ *
+ * <p>ALP encoding converts floating-point values to integers using decimal
scaling,
+ * then applies Frame of Reference encoding and bit-packing.
+ * Values that cannot be losslessly converted are stored as exceptions.
+ *
+ * <p>Writing is incremental: values are buffered in a fixed-size vector
buffer,
+ * and each full vector is encoded and flushed to the output stream
immediately.
+ * On {@link #getBytes()}, any remaining partial vector is flushed, and the
+ * final page bytes are assembled.
+ *
+ * <p>Interleaved Page Layout:
+ * <pre>
+ * ┌─────────┬──────────────────────┬──────────────┬──────────────┬─────┐
+ * │ Header │ Offset Array │ Vector 0 │ Vector 1 │ ... │
+ * │ 7 bytes │ 4B × numVectors │ (interleaved)│ (interleaved)│ │
Review Comment:
This line is not aligned on my editor.
##########
parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpEncoderDecoder.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * 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.*;
Review Comment:
Please do not use wildcard import. Same to other files.
##########
parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpEncoderDecoder.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * 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 org.apache.parquet.bytes.BytesUtils;
+
+/**
+ * Core ALP (Adaptive Lossless floating-Point) encoding and decoding logic.
+ *
+ * <p>ALP works by converting floating-point values to integers using decimal
scaling,
+ * then applying Frame of Reference encoding and bit-packing.
+ * Values that cannot be losslessly converted are stored as exceptions.
+ *
+ * <p>Encoding formula: encoded = fastRound(value * POW10[e] *
POW10_NEGATIVE[f])
+ * <p>Decoding formula: value = encoded * POW10[f] * POW10_NEGATIVE[e]
+ *
+ * <p>The order of operations is critical for IEEE 754 correctness. Both
formulas must
+ * be evaluated as single expressions — storing the intermediate
multiplication result
+ * in a variable before the second multiply changes IEEE 754 rounding and
produces extra
+ * exceptions. Likewise, scaling uses multiply-by-reciprocal (via
POW10_NEGATIVE) rather than
+ * division: this reproduces the exact IEEE 754 rounding of the ALP reference
algorithm, so the
+ * encoded integers — and therefore which values become exceptions and the
resulting bytes — are
+ * identical across implementations. It is about cross-implementation
determinism, not any one
+ * language.
+ *
+ * <p>Exception conditions:
+ * <ul>
+ * <li>NaN values</li>
+ * <li>Infinity values</li>
+ * <li>Negative zero (-0.0)</li>
+ * <li>Out of integer range</li>
+ * <li>Round-trip failure (decode(encode(v)) != v)</li>
+ * </ul>
+ */
+final class AlpEncoderDecoder {
Review Comment:
nit: rename it to `AlpCodec` or `AlpUtil`
##########
parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpValuesWriter.java:
##########
@@ -0,0 +1,647 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.parquet.bytes.ByteBufferAllocator;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.BytesUtils;
+import org.apache.parquet.bytes.CapacityByteArrayOutputStream;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.values.ValuesWriter;
+import org.apache.parquet.column.values.bitpacking.BytePacker;
+import org.apache.parquet.column.values.bitpacking.BytePackerForLong;
+import org.apache.parquet.column.values.bitpacking.Packer;
+
+/**
+ * ALP (Adaptive Lossless floating-Point) values writer.
+ *
+ * <p>ALP encoding converts floating-point values to integers using decimal
scaling,
+ * then applies Frame of Reference encoding and bit-packing.
+ * Values that cannot be losslessly converted are stored as exceptions.
+ *
+ * <p>Writing is incremental: values are buffered in a fixed-size vector
buffer,
+ * and each full vector is encoded and flushed to the output stream
immediately.
+ * On {@link #getBytes()}, any remaining partial vector is flushed, and the
+ * final page bytes are assembled.
+ *
+ * <p>Interleaved Page Layout:
+ * <pre>
+ * ┌─────────┬──────────────────────┬──────────────┬──────────────┬─────┐
+ * │ Header │ Offset Array │ Vector 0 │ Vector 1 │ ... │
+ * │ 7 bytes │ 4B × numVectors │ (interleaved)│ (interleaved)│ │
+ * └─────────┴──────────────────────┴──────────────┴──────────────┴─────┘
+ * </pre>
+ *
+ * <p>Each vector contains interleaved:
+ * AlpInfo(4B) + ForInfo(5B/9B) + PackedValues + ExceptionPositions +
ExceptionValues
+ */
+public abstract class AlpValuesWriter extends ValuesWriter {
+
+ protected final int initialCapacity;
+ protected final int pageSize;
+ protected final ByteBufferAllocator allocator;
+ protected final int vectorSize;
+ protected final int logVectorSize;
+
+ AlpValuesWriter(int initialCapacity, int pageSize, ByteBufferAllocator
allocator, int vectorSize) {
+ AlpConstants.validateVectorSize(vectorSize);
+ this.initialCapacity = initialCapacity;
+ this.pageSize = pageSize;
+ this.allocator = allocator;
+ this.vectorSize = vectorSize;
+ this.logVectorSize = Integer.numberOfTrailingZeros(vectorSize);
+ }
+
+ @Override
+ public Encoding getEncoding() {
+ return Encoding.ALP;
+ }
+
+ /** Float writer. Buffers one vector at a time, encodes and flushes when
full. */
+ public static class FloatAlpValuesWriter extends AlpValuesWriter {
Review Comment:
Why the writer implementations are nested static class but the readers are
in the separate files? Should we make them consistent?
##########
parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java:
##########
@@ -585,6 +637,74 @@ public Builder withExtendedByteStreamSplitEncoding(boolean
enable) {
return this;
}
+ /**
+ * Enable or disable ALP encoding for FLOAT and DOUBLE columns.
+ *
+ * @param enable whether ALP encoding should be enabled
+ * @return this builder for method chaining.
+ */
+ public Builder withAlpEncoding(boolean enable) {
+ this.alpEnabled.withDefaultValue(enable);
+ return this;
+ }
+
+ /**
+ * Enable or disable ALP encoding for the specified column.
+ *
+ * @param columnPath the path of the column (dot-string)
+ * @param enable whether ALP encoding should be enabled
+ * @return this builder for method chaining.
+ */
+ public Builder withAlpEncoding(String columnPath, boolean enable) {
+ this.alpEnabled.withValue(columnPath, enable);
+ return this;
+ }
+
+ /**
+ * Set the ALP vector size (number of values per encoded vector) for FLOAT
and DOUBLE columns.
+ * Must be a power of 2 in the range supported by {@link AlpConstants}.
+ *
+ * @param vectorSize the vector size
+ * @return this builder for method chaining.
+ */
+ public Builder withAlpVectorSize(int vectorSize) {
Review Comment:
Why not directly use `withAlp(AlpConfig)`? It looks a little bit
over-complicated with the current API, especially `buildAlp()`.
##########
parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInterOpReadAlp.java:
##########
@@ -0,0 +1,1408 @@
+/*
+ * 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.hadoop;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.ParquetProperties.WriterVersion;
+import org.apache.parquet.column.page.PageReadStore;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroup;
+import org.apache.parquet.example.data.simple.convert.GroupRecordConverter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.io.ColumnIOFactory;
+import org.apache.parquet.io.LocalInputFile;
+import org.apache.parquet.io.LocalOutputFile;
+import org.apache.parquet.io.MessageColumnIO;
+import org.apache.parquet.io.RecordReader;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.apache.parquet.schema.PrimitiveType;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Cross-compatibility test for ALP (Adaptive Lossless floating-Point)
encoding.
+ *
+ * <p>Reads ALP-encoded parquet files generated by Arrow C++ and verifies that
the Java
+ * implementation decodes them correctly.
+ *
+ * <p>Set ALP_TEST_FILE (env var or system property) to a single file, or use
the
+ * ALP_TEST_DATA_DIR property pointing to the alp-test-data/ directory.
+ *
+ * @see <a href="https://github.com/apache/arrow/pull/48345";>Arrow C++ ALP
PR</a>
+ * @see <a
href="https://github.com/apache/parquet-testing/pull/100";>parquet-testing ALP
PR</a>
+ */
+public class TestInterOpReadAlp {
+ private static final Logger LOG =
LoggerFactory.getLogger(TestInterOpReadAlp.class);
+
+ @TempDir
+ java.nio.file.Path temp;
+
+ /** Mirrors JUnit 4's {@code TemporaryFolder#newFolder()}: a fresh directory
per call. */
+ private File newFolder() throws IOException {
+ return Files.createTempDirectory(temp, "junit").toFile();
+ }
+
+ private static final String[] CPP_DOUBLE_FILES = {"alp_spotify1.parquet",
"alp_arade.parquet"};
+ private static final String[] CPP_FLOAT_FILES =
{"alp_float_spotify1.parquet", "alp_float_arade.parquet"};
+
+ private java.nio.file.Path getTestDataDir() {
+ String dir = System.getProperty("ALP_TEST_DATA_DIR");
Review Comment:
This does not follow other interop tests in the repo to download it from
parquet-testing repo.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
