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 93c2ada1f. 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 93c2ada1f. 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 93c2ada1f 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 8f90f25fa. 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 8f90f25fa. 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 93c2ada1f 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 93c2ada1f 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. -- 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]
