vinooganesh commented on code in PR #3397: URL: https://github.com/apache/parquet-java/pull/3397#discussion_r3942455381
########## parquet-column/src/main/java/org/apache/parquet/column/values/alp/AlpCodec.java: ########## @@ -0,0 +1,302 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.parquet.column.values.alp; + +import static org.apache.parquet.column.values.alp.AlpConstants.*; + +import org.apache.parquet.bytes.BytesUtils; + +/** + * Core ALP (Adaptive Lossless floating-Point) encoding and decoding logic. + * + * <p>ALP works by converting floating-point values to integers using decimal scaling, + * then applying Frame of Reference encoding and bit-packing. + * Values that cannot be losslessly converted are stored as exceptions. + * + * <p>Encoding formula: encoded = fastRound(value * POW10[e] * POW10_NEGATIVE[f]) + * <p>Decoding formula: value = encoded * POW10[f] * POW10_NEGATIVE[e] + * + * <p>The order of operations is critical for IEEE 754 correctness. Both formulas must + * be evaluated as single expressions — storing the intermediate multiplication result + * in a variable before the second multiply changes IEEE 754 rounding and produces extra + * exceptions. Likewise, scaling uses multiply-by-reciprocal (via POW10_NEGATIVE) rather than + * division: this reproduces the exact IEEE 754 rounding of the ALP reference algorithm, so the + * encoded integers — and therefore which values become exceptions and the resulting bytes — are + * identical across implementations. It is about cross-implementation determinism, not any one + * language. + * + * <p>Exception conditions: + * <ul> + * <li>NaN values</li> + * <li>Infinity values</li> + * <li>Negative zero (-0.0)</li> + * <li>Out of integer range</li> + * <li>Round-trip failure (decode(encode(v)) != v)</li> + * </ul> + */ +final class AlpEncoderDecoder { + + private static final double ENCODING_UPPER_LIMIT = 9223372036854774784.0; + private static final double ENCODING_LOWER_LIMIT = -9223372036854774784.0; + private static final float FLOAT_ENCODING_UPPER_LIMIT = 2147483520.0f; + private static final float FLOAT_ENCODING_LOWER_LIMIT = -2147483520.0f; + + private AlpEncoderDecoder() { + // Utility class + } + + /** NaN, Inf, and -0.0 can never be encoded regardless of exponent/factor. */ + static boolean isIntrinsicFloatException(float value) { + if (Float.isNaN(value)) { + return true; + } + if (Float.isInfinite(value)) { + return true; + } + return Float.floatToRawIntBits(value) == FLOAT_NEGATIVE_ZERO_BITS; + } + + /** Full exception check for a given (exponent, factor): intrinsic cases plus round-trip failure. */ + static boolean isFloatException(float value, int exponent, int factor) { + if (isIntrinsicFloatException(value)) { + return true; + } + // Check before rounding: overflow or non-finite after scaling + float scaled = value * FLOAT_POW10[exponent] * FLOAT_POW10_NEGATIVE[factor]; + if (!Float.isFinite(scaled) || scaled > FLOAT_ENCODING_UPPER_LIMIT || scaled < FLOAT_ENCODING_LOWER_LIMIT) { + return true; + } + int encoded = encodeFloat(value, exponent, factor); + float decoded = decodeFloat(encoded, exponent, factor); + return Float.floatToRawIntBits(value) != Float.floatToRawIntBits(decoded); + } + + /** Round float to nearest integer using magic-number trick with sign branching. */ + static int fastRoundFloat(float value) { + if (value >= 0) { + return (int) ((value + MAGIC_FLOAT) - MAGIC_FLOAT); + } else { + return (int) ((value - MAGIC_FLOAT) + MAGIC_FLOAT); + } + } + + static int encodeFloat(float value, int exponent, int factor) { + return fastRoundFloat(value * FLOAT_POW10[exponent] * FLOAT_POW10_NEGATIVE[factor]); + } + + static float decodeFloat(int encoded, int exponent, int factor) { + return encoded * FLOAT_POW10[factor] * FLOAT_POW10_NEGATIVE[exponent]; + } + + /** NaN, Inf, and -0.0 can never be encoded regardless of exponent/factor. */ + static boolean isIntrinsicDoubleException(double value) { + if (Double.isNaN(value)) { + return true; + } + if (Double.isInfinite(value)) { + return true; + } + return Double.doubleToRawLongBits(value) == DOUBLE_NEGATIVE_ZERO_BITS; + } + + /** Full exception check for a given (exponent, factor): intrinsic cases plus round-trip failure. */ + static boolean isDoubleException(double value, int exponent, int factor) { + if (isIntrinsicDoubleException(value)) { + return true; + } + // Check before rounding: overflow or non-finite after scaling + double scaled = value * DOUBLE_POW10[exponent] * DOUBLE_POW10_NEGATIVE[factor]; + if (!Double.isFinite(scaled) || scaled > ENCODING_UPPER_LIMIT || scaled < ENCODING_LOWER_LIMIT) { + return true; + } + long encoded = encodeDouble(value, exponent, factor); + double decoded = decodeDouble(encoded, exponent, factor); + return Double.doubleToRawLongBits(value) != Double.doubleToRawLongBits(decoded); + } + + /** Round double to nearest integer using magic-number trick with sign branching. */ + static long fastRoundDouble(double value) { + if (value >= 0) { + return (long) ((value + MAGIC_DOUBLE) - MAGIC_DOUBLE); + } else { + return (long) ((value - MAGIC_DOUBLE) + MAGIC_DOUBLE); + } + } + + static long encodeDouble(double value, int exponent, int factor) { + return fastRoundDouble(value * DOUBLE_POW10[exponent] * DOUBLE_POW10_NEGATIVE[factor]); + } + + static double decodeDouble(long encoded, int exponent, int factor) { + return encoded * DOUBLE_POW10[factor] * DOUBLE_POW10_NEGATIVE[exponent]; + } + + public static class EncodingParams { + final int exponent; + final int factor; + final int numExceptions; + + EncodingParams(int exponent, int factor, int numExceptions) { + this.exponent = exponent; + this.factor = factor; + this.numExceptions = numExceptions; + } + } + + // Index positions within an (exponent, factor) pair array. + private static final int E = 0; + private static final int F = 1; + + // All valid (exponent, factor) pairs for the full search, precomputed in nested-loop order + // (e ascending, then f ascending) so the tie-break and early-exit behave exactly like an inline + // double loop. Reused across every vector to avoid per-call allocation. + private static final int[][] ALL_VALID_FLOAT_PAIRS = buildAllPairs(FLOAT_MAX_EXPONENT); + private static final int[][] ALL_VALID_DOUBLE_PAIRS = buildAllPairs(DOUBLE_MAX_EXPONENT); + + private static int[][] buildAllPairs(int maxExponent) { + int count = (maxExponent + 1) * (maxExponent + 2) / 2; + int[][] pairs = new int[count][]; + int idx = 0; + for (int e = 0; e <= maxExponent; e++) { + for (int f = 0; f <= e; f++) { + pairs[idx++] = new int[] {e, f}; + } + } + return pairs; + } + + /** Try all (exponent, factor) combos and pick the one with the smallest estimated compressed size. */ + static EncodingParams findBestFloatParams(float[] values, int offset, int length) { + return pickBestFloat(values, offset, length, ALL_VALID_FLOAT_PAIRS); + } + + /** Same as findBestFloatParams but only tries the cached preset combos. */ + static EncodingParams findBestFloatParamsWithPresets(float[] values, int offset, int length, int[][] presets) { + return pickBestFloat(values, offset, length, presets); + } + + /** + * Scores each (exponent, factor) pair by estimated compressed size and returns the best. + * + * <p>Estimated size (in bits) = {@code length * bitWidth + exceptions * (Float.SIZE + Short.SIZE)}, + * where bitWidth is the number of bits needed to represent the signed range (max - min) of + * non-exception encoded values after frame-of-reference subtraction, matching the writer's FOR + * packing. This produces better compression ratios than minimizing exception count alone. Ties in + * size are broken toward the higher exponent, then the higher factor. The first pair (in {@code + * pairs} order) that encodes every value into a single FOR value with zero exceptions wins + * outright. When no pair yields any non-exception values, {@code pairs[0]} is returned. + */ + private static EncodingParams pickBestFloat(float[] values, int offset, int length, int[][] pairs) { + int bestExponent = pairs[0][E]; + int bestFactor = pairs[0][F]; + int bestExceptions = length; + long bestEstimatedSize = Long.MAX_VALUE; + + for (int[] pair : pairs) { + int e = pair[E]; + int f = pair[F]; + int exceptions = 0; + int minEncoded = Integer.MAX_VALUE; + int maxEncoded = Integer.MIN_VALUE; + for (int i = 0; i < length; i++) { + float value = values[offset + i]; + if (isFloatException(value, e, f)) { + exceptions++; + } else { + int encoded = encodeFloat(value, e, f); Review Comment: You were right to ask, and it was worth doing. isFloatException has to encode the value in order to test whether it round trips, so calling it and then encodeFloat really did the work twice. The writer's per-vector loop was worse than the search loop you commented on: it ran the exception check, encoded again, and had a separate placeholder scan that encoded a third time. Fixed in 8e743c60a. There are now tryEncodeFloat and tryEncodeDouble which give back the exception flag and the encoded value together. I went with a small caller-owned EncodeResult holder rather than an Optional, since an Optional would box on every value in what is the hottest loop in the writer. The scaled intermediate gets reused as well, so the two multiplies are also down to one. Encoded bytes are unchanged. tryEncode uses the same expression in the same order as encodeFloat, which matters for cross implementation bit compatibility, and testTryEncodeMatchesSeparateExceptionCheckAndEncode asserts the new path agrees with the old pair on both the flag and the exact encoded bits across every valid (exponent, factor) pair. On whether it was a performance concern, I want to give you the honest answer rather than a flattering one. Measured with AlpEncodingBenchmarks.writeDoubleAndFloatALP over a million rows, before was 197.9 +- 6.6 ms/op and after was 194.2 +- 2.9 ms/op. The error bars overlap, so at the level of writing a whole file this is within noise, because encoding is small next to compression and page assembly. The duplicated work was real and is gone, but I cannot claim a measurable end to end win from it. ########## parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInterOpReadAlp.java: ########## @@ -0,0 +1,1408 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.parquet.hadoop; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.parquet.column.Encoding; +import org.apache.parquet.column.ParquetProperties.WriterVersion; +import org.apache.parquet.column.page.PageReadStore; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroup; +import org.apache.parquet.example.data.simple.convert.GroupRecordConverter; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.io.ColumnIOFactory; +import org.apache.parquet.io.LocalInputFile; +import org.apache.parquet.io.LocalOutputFile; +import org.apache.parquet.io.MessageColumnIO; +import org.apache.parquet.io.RecordReader; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; +import org.apache.parquet.schema.PrimitiveType; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Cross-compatibility test for ALP (Adaptive Lossless floating-Point) encoding. + * + * <p>Reads ALP-encoded parquet files generated by Arrow C++ and verifies that the Java + * implementation decodes them correctly. + * + * <p>Set ALP_TEST_FILE (env var or system property) to a single file, or use the + * ALP_TEST_DATA_DIR property pointing to the alp-test-data/ directory. + * + * @see <a href="https://github.com/apache/arrow/pull/48345";>Arrow C++ ALP PR</a> + * @see <a href="https://github.com/apache/parquet-testing/pull/100";>parquet-testing ALP PR</a> + */ +public class TestInterOpReadAlp { + private static final Logger LOG = LoggerFactory.getLogger(TestInterOpReadAlp.class); + + @TempDir + java.nio.file.Path temp; + + /** Mirrors JUnit 4's {@code TemporaryFolder#newFolder()}: a fresh directory per call. */ + private File newFolder() throws IOException { + return Files.createTempDirectory(temp, "junit").toFile(); + } + + private static final String[] CPP_DOUBLE_FILES = {"alp_spotify1.parquet", "alp_arade.parquet"}; + private static final String[] CPP_FLOAT_FILES = {"alp_float_spotify1.parquet", "alp_float_arade.parquet"}; + + private java.nio.file.Path getTestDataDir() { + String dir = System.getProperty("ALP_TEST_DATA_DIR"); Review Comment: You're right, and I would like to move this over to the standard harness. The blocker is that the ALP fixture files are not in parquet-testing yet. They are proposed in https://github.com/apache/parquet-testing/pull/100, which is still open. Once that merges I will convert this to use InterOpTester.GetInterOpFile with a pinned changeset, the same as TestInterOpReadByteStreamSplit and the other interop tests. Until then the test reads from a local directory and skips when it is not present, which does mean it is effectively a no-op in CI, so I understand it is not carrying its weight right now. If you have any influence over getting parquet-testing #100 looked at, that would help a lot. Cross language read back is the part of this work I am least able to demonstrate on my own, so it matters for the vote more than for this test alone. ########## parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV1ValuesWriterFactory.java: ########## @@ -147,7 +148,13 @@ private ValuesWriter getInt96ValuesWriter(ColumnDescriptor path) { private ValuesWriter getDoubleValuesWriter(ColumnDescriptor path) { final ValuesWriter fallbackWriter; - if (this.parquetProperties.isByteStreamSplitEnabled(path)) { + if (this.parquetProperties.isAlpEnabled(path)) { Review Comment: Thanks for weighing in on this, it is helpful to have a second opinion since I went back and forth on it. I will leave ALP registered in both V1 and V2 then, and I agree that staying consistent with how isByteStreamSplitEnabled is handled is the least surprising option. If the writer version work lands later and there is a reason to pin ALP to a particular version, that seems like a better time to revisit 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]
