vinooganesh commented on code in PR #3397: URL: https://github.com/apache/parquet-java/pull/3397#discussion_r3942046622
########## 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: Thanks for catching this, it's a good find. You're right that numElements is only bounded by valuesCount, and since a forged file controls that value it can go all the way up to Integer.MAX_VALUE. Two separate things went wrong from there. The `+ vectorSize - 1` overflowed into a negative count, and even when it didn't overflow, a large count drove a big allocation before anything had been validated. While fixing it I found the allocation is genuinely reachable rather than theoretical. MultiBufferInputStream.slice() calls ByteBuffer.allocate(length) before it checks for EOF, so a small forged page really can ask for hundreds of megabytes. This is fixed in 338fb99c6. numVectors is computed in long now and checked against the bytes actually present before anything gets allocated. Since every vector needs a 4 byte offset entry plus at least its ALP and FOR headers, that gives a firm ceiling on how many vectors the remaining bytes could possibly describe, and anything above it throws a ParquetDecodingException that says how far short the page is. ########## 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: You're right on all three counts, thank you. Fixed in 338fb99c6. initFromPage now validates the whole offset array before any vector is decoded. The first offset has to equal the offset array size, offsets have to increase by at least one vector's worth of fixed headers, and each one has to start early enough to leave room for those headers before the body ends. The observation about decode ignoring the next offset was the more important half of this, since that's the case that quietly returns wrong values instead of failing. decodeVector now takes each vector's end position from the following offset, or from the end of the body for the last vector, and rejects a packed body or exception block that runs past it. ########## 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: Good catch, thank you. Fixed in 338fb99c6 by comparing against the remaining count with `n > totalCount - pageValueIndex` rather than adding to pageValueIndex, so a large n can no longer overflow past the check and leave the reader at a negative index. I added testSkipRejectsOverflowingCount to cover it. ########## 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: Thank you for working through the arithmetic on this one, it turned out to be the most valuable comment in the review. I checked it and the numbers come out exactly as you described. At the default 20k row page limit with a 1024 vector size a page holds roughly 19 vectors, and rowgroupSampleJump works out to 15, so each page contributes about two samples. Because reset() cleared sampledParams and vectorsProcessed at every page boundary, the count never got near SAMPLER_SAMPLE_VECTORS_PER_ROWGROUP, buildPresetCache() never ran, and findBest*ParamsWithPresets was effectively dead code. Every vector was paying for a full parameter search. Fixed in eb0e5941d. The sampler is meant to operate over a rowgroup rather than a page, so the sampling state now survives reset() and samples accumulate across pages the way the design intended. I added testPresetCacheBuildsAcrossPageBoundaries, which fails without the change. One consequence I want to flag, since it isn't obvious. Now that the preset path is genuinely reachable, encoded output can shift slightly, because the search narrows to the top MAX_PRESET_COMBINATIONS pairs instead of trying everything. That is the intended ALP sampling behavior and it matches the C++ implementation. Round tripping is still exact, since the chosen pair is recorded in each vector's header and exceptions cover anything it cannot represent. ########## 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: This is the same underlying problem as your comment on the sampling cadence, and it's fixed in eb0e5941d. I've written up the details in that thread rather than repeating them here. Thanks for flagging it from both angles, it made the problem much easier to pin down. ########## 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: You're right, and thank you for reading the tests as carefully as the production code. That test only inflated the page's valuesCount, so the `numElements <= valuesCount` check passed trivially and the allocation path was never touched. I replaced it in 338fb99c6 with testForgedElementCountIsRejectedWithoutHugeAllocation, which overwrites the num_elements field in the header directly and passes Integer.MAX_VALUE as valuesCount so that the allocation bound is what has to reject it. It covers both the counts that are simply too large for the page and the two that overflow the vector count arithmetic. ########## 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: Agreed, and thanks for pointing it out. Catching Throwable meant the test could not really fail, which defeated the point of it. In 338fb99c6 I replaced catchAny with catchClean, which catches Exception and asserts that the type is one of the expected clean failures while letting Error propagate. An OutOfMemoryError from an allocation bomb or an AssertionError from an inner assertion will now fail the test rather than quietly counting as a successful rejection. I applied the same change to the `catch (Throwable)` in the corruption loop in AlpValuesEndToEndTest. ########## 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: You're right that this was more complicated than it needed to be, thanks for pushing on it. The builder was keeping the enabled flag and the vector size as two separate ColumnProperty scaffolds, merging them into an AlpConfig in buildAlp(), and then splitting them apart again in the copy constructor. Nothing actually needed that round trip. Fixed in 99ae547c5. There is now a single ColumnProperty<AlpConfig>, with withAlp(AlpConfig) and withAlp(columnPath, AlpConfig) as the primary API. I kept withAlpEncoding and withAlpVectorSize as thin read-modify-write conveniences on top, since they match the ergonomics of withByteStreamSplitEncoding and it means no existing caller had to change, including ParquetWriter.Builder. buildAlp and the copy constructor split are both gone. I also moved the vector size validation into the AlpConfig constructor, which felt like the right home for it since an invalid size should be rejected however the config was built. That had the nice side effect of removing the last reference to AlpConstants from ParquetProperties, which made your visibility comment straightforward to act on. One thing I want to flag for you. To let the convenience setters modify an existing config rather than replace it, I added getDefaultValue and getValue to ColumnProperty.Builder. That class is package private so it is not public API, but it is a shared file rather than something ALP specific, so please say if you would rather I found another way around it. ########## 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: Good question, and the answer is that it did not need to be public. I checked every reference and ParquetProperties was the only thing outside the package touching it, for DEFAULT_VECTOR_SIZE and validateVectorSize. Once the AlpConfig change above removed both of those, nothing external was left. The class and all of its members are package private now, in 78c1d182a. It seemed worth doing properly before a release rather than after, since it keeps the ALP internals out of the public API surface permanently. ########## 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: Agreed, thank you. I went through each constant and sorted them by who actually reads them, in 78c1d182a. The sampler constants (SAMPLER_ROWGROUP_SIZE, SAMPLER_SAMPLE_VECTORS_PER_ROWGROUP, MAX_PRESET_COMBINATIONS) are only used by the writer, so they moved to AlpValuesWriter. The rounding magic numbers and the powers of ten tables are only used by the codec, so they moved to AlpCodec. What is left in AlpConstants is genuinely shared and all of it describes the wire format: the header and metadata sizes, the mode and encoding markers, the vector size bounds, and the per type exponent limits. It reads much better as a wire format definition than it did as a general bucket. ########## 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: Happy to take this. Renamed to AlpCodec in 78c1d182a, along with its test. I went with AlpCodec over AlpUtil since it really is a codec rather than a collection of helpers. ########## 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: Understood, and sorry for the noise. All five files now use explicit static imports, in 78c1d182a. I did this last on purpose, after the constants had been moved to their new homes, so that the import lists reflect where things ended up rather than needing a second pass. ########## 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: Thanks for spotting that. The cause is that the cell contains `4B × numVectors`, and the HTML entity is seven characters in the source but renders as a single glyph, so the box could never line up in both your editor and the rendered javadoc at the same time. Fixed in 78c1d182a by using the literal character and padding the cell to the right width, so every row is 73 characters in both views. The same diagram had been copied into AlpValuesReader with the same problem, so I fixed it there too. ########## parquet-format-structures/pom.xml: ########## @@ -66,6 +66,35 @@ </execution> </executions> </plugin> + <!-- Review Comment: Your timing on this was good. #3709 merged a couple of days after you left the comment, so parquet.thrift is now inlined in the repo and I was able to take the approach you suggested. In 229e0d99b, ALP = 10 is declared directly in the Encoding enum, and the perl script and the exec-maven-plugin execution that ran it are both gone. I verified the generated Encoding.java still carries ALP(10) after BYTE_STREAM_SPLIT(9) with no patch step involved. There is one wrinkle I want your opinion on. dev/update-parquet-thrift.sh overwrites parquet.thrift wholesale from upstream, so the next time anyone runs it the ALP entry will disappear silently. It also rewrites the parquet-format.version sidecar, so a note there would not survive either. For now I have put a clearly marked notice in the enum itself saying it is a local addition that needs re-applying until ALP is accepted into parquet-format, on the grounds that at least the loss shows up in a diff. If you would prefer this handled a different way, such as a guard in the update script, I am glad to change it. -- 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]
