vinooganesh commented on code in PR #3397: URL: https://github.com/apache/parquet-java/pull/3397#discussion_r4089197573
########## parquet-column/src/test/java/org/apache/parquet/column/values/alp/AlpAdversarialTest.java: ########## @@ -0,0 +1,398 @@ +/* + * 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; Review Comment: Done in de1e6764e. Every JUnit assertion and assumption helper is gone from the ALP tests: `assertThrows` became `assertThatThrownBy` in `AlpAdversarialTest` and `AlpValuesEndToEndTest`, and `assumeTrue` became AssertJ's `assumeThat` in `TestInterOpReadAlp`. While converting I also pinned the real exception messages rather than the lowercased substring the old assertions used. That turned out to matter: the old `contains("exponent")` check would have passed on a *factor* error too, since that message mentions the exponent as well. `catchClean` became `assertFailsCleanly`, built on `assertThatThrownBy(...).isInstanceOfAny(...)`. I checked it still holds the property you were after in your earlier comment: it fails when nothing is thrown, when the wrong type is thrown, and when an `OutOfMemoryError` escapes, so an allocation bomb still cannot pass as a clean rejection. To convince myself the new message assertions were load-bearing rather than decorative, I broke four validations in `AlpValuesReader` in turn — the factor check, the skip bounds, the non-increasing offset check, and the wording of the allocation-bound message — and confirmed each one is caught by exactly the converted test. On the CI output you pasted: that one is not from this PR. It is `TestInputFormatFileStatusReuse`, which arrived with #3674 and was migrated to JUnit 5 upstream in #3813. Merging master picks up the fix, and parquet-hadoop's banned-imports check passes on the branch now. ########## parquet-column/src/test/java/org/apache/parquet/column/values/factory/DefaultValuesWriterFactoryTest.java: ########## @@ -631,6 +668,50 @@ private void testFloatingPoint_WithByteStreamSplitAndDictionary( PlainValuesWriter.class); } + private void testFloatingPoint_WithAlp( + PrimitiveTypeName typeName, WriterVersion writerVersion, Class<? extends ValuesWriter> alpWriterClass) { + // Cross-column ALP without a dictionary: the ALP writer is produced directly. + doTestValueWriter( + createColumnDescriptor(typeName), + ParquetProperties.builder() + .withWriterVersion(writerVersion) + .withDictionaryEncoding(false) + .withAlp() + .build(), + alpWriterClass); + // Cross-column ALP with a dictionary: the ALP writer is the dictionary's fallback. + doTestValueWriter( + createColumnDescriptor(typeName), + ParquetProperties.builder() + .withWriterVersion(writerVersion) + .withAlp() + .build(), + DictionaryValuesWriter.class, + alpWriterClass); + // Per-column ALP: only colA opts in, colB keeps the plain fallback. + ParquetProperties properties = ParquetProperties.builder() + .withWriterVersion(writerVersion) + .withDictionaryEncoding(false) + .withAlp("colA") + .build(); + doTestValueWriter(createColumnDescriptor(typeName, "colA"), properties, alpWriterClass); + doTestValueWriter(createColumnDescriptor(typeName, "colB"), properties, PlainValuesWriter.class); + } + + private void testAlpTakesPrecedenceOverByteStreamSplit( + PrimitiveTypeName typeName, WriterVersion writerVersion, Class<? extends ValuesWriter> alpWriterClass) { + // When both ALP and byte-stream-split are enabled the factory prefers ALP. Review Comment: You're right, thank you. Removed in de1e6764e, along with both `Alp_TakesPrecedenceOverByteStreamSplit` tests and the shared helper. Since `validateAlp` rejects ALP and BYTE_STREAM_SPLIT on the same column in `newColumnWriteStore`, the factory's ordering can never be reached in production, so the test was asserting behaviour that cannot happen. One thing worth noting: there was no dedicated precedence *code* to remove alongside it. `DefaultV1ValuesWriterFactory` just has a plain `if (isAlpEnabled) / else if (isByteStreamSplitEnabled)` chain, so ALP-first is an artifact of the ordering rather than a feature, and nothing is left dead by the deletion. The rejection itself is covered by `TestParquetProperties.alp_withByteStreamSplitOnTheSameColumn_fails`, which builds the identical configuration the deleted test used. ########## parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInterOpReadAlp.java: ########## @@ -0,0 +1,775 @@ +/* + * 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.LinkedHashMap; +import java.util.List; +import java.util.Map; +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.column.values.alp.AlpConfig; +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.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Cross-compatibility tests for ALP (Adaptive Lossless floating-Point) encoding. + * + * <p>Reads the ALP conformance file from parquet-testing and checks that every ALP column decodes + * bit-identically to its PLAIN counterpart, and round-trips Java-written ALP files. + */ +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 ALP_EXTENDED_FILE = "alp_extended.zstd.parquet"; + private static final String CHANGESET = "09f3cdb"; + private final InterOpTester interop = new InterOpTester(); + + private static final int EXTENDED_ROWS = 9032; + private static final int EXTENDED_ROW_GROUPS = 5; + private static final int EXTENDED_NULLS = 8; + + /** ALP column -> PLAIN column holding the same values. */ + private static final Map<String, String> ALP_TO_PLAIN = new LinkedHashMap<>(); + + static { + ALP_TO_PLAIN.put("float_alp_32", "float_plain"); + ALP_TO_PLAIN.put("float_alp_1024", "float_plain"); + ALP_TO_PLAIN.put("float_alp_4096", "float_plain"); + ALP_TO_PLAIN.put("double_alp_32", "double_plain"); + ALP_TO_PLAIN.put("double_alp_1024", "double_plain"); + ALP_TO_PLAIN.put("double_alp_4096", "double_plain"); + } + + private java.nio.file.Path getExtendedFile() throws IOException { + return Paths.get(interop.GetInterOpFile(ALP_EXTENDED_FILE, CHANGESET).toString()); + } + + /** Read all rows from a parquet file using LocalInputFile (no Hadoop FileSystem). */ + private List<Group> readAllRows(java.nio.file.Path filePath) throws IOException { + List<Group> rows = new ArrayList<>(); + try (ParquetFileReader reader = ParquetFileReader.open(new LocalInputFile(filePath))) { + ParquetMetadata footer = reader.getFooter(); + MessageType schema = footer.getFileMetaData().getSchema(); + MessageColumnIO columnIO = new ColumnIOFactory().getColumnIO(schema); + PageReadStore pages; + while ((pages = reader.readNextRowGroup()) != null) { + long rowCount = pages.getRowCount(); + RecordReader<Group> recordReader = columnIO.getRecordReader(pages, new GroupRecordConverter(schema)); + for (long i = 0; i < rowCount; i++) { + rows.add(recordReader.read()); + } + } + } + return rows; + } + + private static boolean isNull(Group row, String field) { + return row.getFieldRepetitionCount(field) == 0; + } + + /** Raw IEEE 754 bits of a float or double field, widened to a long for exact comparison. */ + private static long rawBits(Group row, String field, boolean isFloat) { + return isFloat + ? Float.floatToRawIntBits(row.getFloat(field, 0)) & 0xFFFFFFFFL + : Double.doubleToRawLongBits(row.getDouble(field, 0)); + } + + @Test + public void testAlpColumnsMatchPlainColumns() throws IOException { + List<Group> rows = readAllRows(getExtendedFile()); + assertThat(rows.size()).as("row count").isEqualTo(EXTENDED_ROWS); + + for (Map.Entry<String, String> entry : ALP_TO_PLAIN.entrySet()) { + String alpCol = entry.getKey(); + String plainCol = entry.getValue(); + boolean isFloat = alpCol.startsWith("float"); + int nulls = 0; + for (int i = 0; i < rows.size(); i++) { + Group row = rows.get(i); + boolean alpNull = isNull(row, alpCol); + assertThat(alpNull).as("nullness of " + alpCol + " at row " + i).isEqualTo(isNull(row, plainCol)); + if (alpNull) { + nulls++; + continue; + } + long alpBits = rawBits(row, alpCol, isFloat); + long plainBits = rawBits(row, plainCol, isFloat); + assertThat(alpBits) + .as(String.format( + "%s row %d: bits 0x%x, %s bits 0x%x", alpCol, i, alpBits, plainCol, plainBits)) + .isEqualTo(plainBits); + } + assertThat(nulls).as(alpCol + " nulls").isEqualTo(EXTENDED_NULLS); + } + } + + @Test + public void testAlpColumnsAreAlpEncoded() throws IOException { + try (ParquetFileReader reader = ParquetFileReader.open(new LocalInputFile(getExtendedFile()))) { + ParquetMetadata footer = reader.getFooter(); + assertThat(footer.getBlocks().size()).as("row groups").isEqualTo(EXTENDED_ROW_GROUPS); + int alpChunks = 0; + for (org.apache.parquet.hadoop.metadata.BlockMetaData block : footer.getBlocks()) { + for (org.apache.parquet.hadoop.metadata.ColumnChunkMetaData chunk : block.getColumns()) { + String path = chunk.getPath().toDotString(); + boolean hasAlp = chunk.getEncodings().contains(Encoding.ALP); + assertThat(hasAlp) + .as("ALP on " + path + ": " + chunk.getEncodings()) + .isEqualTo(ALP_TO_PLAIN.containsKey(path)); + if (hasAlp) { + alpChunks++; + } + } + } + assertThat(alpChunks).as("ALP column chunks").isEqualTo(ALP_TO_PLAIN.size() * EXTENDED_ROW_GROUPS); + } + } + + /** + * Checks exact bit patterns at known rows, so a fault that corrupted the ALP and PLAIN paths the + * same way would still be caught. NaN payloads must survive, since exceptions are stored verbatim. + */ + @Test + public void testSpecialValueBitPatterns() throws IOException { + List<Group> rows = readAllRows(getExtendedFile()); + assertBits(rows, 1024, 0x7FF8000000000000L, 0x7FC00000L, "canonical NaN"); + assertBits(rows, 1500, 0x7FF800DEADBEEF00L, 0x7FC0DEADL, "NaN with payload"); + assertBits(rows, 2047, 0xFFF8000000000001L, 0xFFC00001L, "negative NaN with payload"); + assertBits(rows, 2000, 0x7FF0000000000000L, 0x7F800000L, "+Inf"); + assertBits(rows, 2001, 0xFFF0000000000000L, 0xFF800000L, "-Inf"); + assertBits(rows, 2002, 0x8000000000000000L, 0x80000000L, "-0.0"); + assertBits(rows, 2003, 0x1L, 0x1L, "smallest subnormal"); + assertBits(rows, 2500, 0x400921FB54442D18L, 0x40490FDBL, "pi"); + assertBits(rows, 7168, 0x401F147AE147AE14L, 0x40F8A3D7L, "constant 7.77"); + assertBits(rows, 9000, 0xC3DBC16D674EC800L, 0xDEDE0B6BL, "-8e18"); + assertBits(rows, 9001, 0x43DBC16D674EC800L, 0x5EDE0B6BL, "8e18"); + for (int row : new int[] {8200, 8900}) { + for (String col : ALP_TO_PLAIN.keySet()) { + assertThat(isNull(rows.get(row), col)) + .as(col + " null at row " + row) + .isTrue(); + } + } + } + + private void assertBits(List<Group> rows, int row, long doubleBits, long floatBits, String what) { + for (String col : ALP_TO_PLAIN.keySet()) { + boolean isFloat = col.startsWith("float"); + assertThat(rawBits(rows.get(row), col, isFloat)) + .as(what + " at row " + row + " in " + col) + .isEqualTo(isFloat ? floatBits : doubleBits); + } + } + + private static final String ALP_SCHEMA = + "message alp_interop { " + "required double double_col; " + "required float float_col; " + "}"; + + private static final double[] DOUBLE_VALUES = { + 1.23, 4.56, 7.89, 0.001, 1000.0, -3.14, 2.718281828, 9.99999, 0.123456789, 100.5 + }; + private static final float[] FLOAT_VALUES = { + 1.23f, 4.56f, 7.89f, 0.001f, 1000.0f, -3.14f, 2.718f, 9.999f, 0.1234f, 100.5f + }; + + /** + * Write an ALP-encoded file from Java using the given page version, then read it back and verify + * all double and float values round-trip exactly. + */ + private void writeAndVerifyAlpFile(WriterVersion version) throws IOException { + MessageType schema = MessageTypeParser.parseMessageType(ALP_SCHEMA); + java.nio.file.Path outPath = + newFolder().toPath().resolve("alp_java_" + version.name().toLowerCase() + ".parquet"); + + try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new LocalOutputFile(outPath)) + .withType(schema) + .withCompressionCodec(CompressionCodecName.UNCOMPRESSED) + .withWriterVersion(version) + .withAlp() + .withDictionaryEncoding(false) + .withConf(new Configuration()) + .build()) { + for (int i = 0; i < DOUBLE_VALUES.length; i++) { + SimpleGroup row = new SimpleGroup(schema); + row.add("double_col", DOUBLE_VALUES[i]); + row.add("float_col", FLOAT_VALUES[i]); + writer.write(row); + } + } + + List<Group> rows = readAllRows(outPath); + assertThat(rows.size()).as("Row count mismatch for " + version).isEqualTo(DOUBLE_VALUES.length); + for (int i = 0; i < DOUBLE_VALUES.length; i++) { + assertThat(rows.get(i).getDouble("double_col", 0)) + .as("double_col mismatch at row " + i + " for " + version) + .isEqualTo(DOUBLE_VALUES[i]); + assertThat(rows.get(i).getFloat("float_col", 0)) + .as("float_col mismatch at row " + i + " for " + version) + .isEqualTo(FLOAT_VALUES[i]); + } + LOG.info( + "writeAndVerifyAlpFile [{}]: wrote and read back {} rows from {}", + version, + rows.size(), + outPath.getFileName()); + } + + /** + * Java writes ALP-encoded floats/doubles using V1 (PARQUET_1_0) data pages and reads them back. + * Verifies the Java write path produces a valid file readable by this implementation. + */ + @Test + public void testJavaWriteAlpV1Pages() throws IOException { + writeAndVerifyAlpFile(WriterVersion.PARQUET_1_0); + } + + /** + * Writes >4096 rows with vectorSize=4096 so multiple full vectors are flushed, then verifies + * the file round-trips exactly. The reader pulls log_vector_size from the on-disk header to + * size its unpacking window, so a wrong header byte would surface as decode garbage — + * round-trip equality is sufficient proof that the configured vector size took effect. + */ + @Test + public void testJavaWriteAlpCustomVectorSize() throws IOException { + MessageType schema = MessageTypeParser.parseMessageType(ALP_SCHEMA); + int rowCount = 4500; // crosses one full vector + partial tail at vectorSize=4096 + double[] doubles = new double[rowCount]; + float[] floats = new float[rowCount]; + // 2-decimal sensor-like data — the ALP sweet spot, so few/no exceptions + for (int i = 0; i < rowCount; i++) { + doubles[i] = (i * 13L % 100000) / 100.0; + floats[i] = (float) ((i * 7L % 10000) / 100.0); + } + + java.nio.file.Path outPath = newFolder().toPath().resolve("alp_java_vs4096.parquet"); + + try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new LocalOutputFile(outPath)) + .withType(schema) + .withCompressionCodec(CompressionCodecName.UNCOMPRESSED) + .withWriterVersion(WriterVersion.PARQUET_2_0) + .withAlp(new AlpConfig(4096)) + .withDictionaryEncoding(false) + .withConf(new Configuration()) + .build()) { + for (int i = 0; i < rowCount; i++) { + SimpleGroup row = new SimpleGroup(schema); + row.add("double_col", doubles[i]); + row.add("float_col", floats[i]); + writer.write(row); + } + } + + List<Group> rows = readAllRows(outPath); + assertThat(rows.size()).as("Row count mismatch at vectorSize=4096").isEqualTo(rowCount); + for (int i = 0; i < rowCount; i++) { + assertThat(rows.get(i).getDouble("double_col", 0)) + .as("double_col mismatch at row " + i) + .isEqualTo(doubles[i]); + assertThat(rows.get(i).getFloat("float_col", 0)) + .as("float_col mismatch at row " + i) + .isEqualTo(floats[i]); + } + LOG.info("testJavaWriteAlpCustomVectorSize: {} rows round-tripped at vectorSize=4096", rowCount); + } + + /** + * Java writes ALP-encoded floats/doubles using V2 (PARQUET_2_0) data pages and reads them back. + */ + @Test + public void testJavaWriteAlpV2Pages() throws IOException { + writeAndVerifyAlpFile(WriterVersion.PARQUET_2_0); + } + + /** + * Writes ALP-encoded V1 and V2 parquet files from Java and verifies that pyarrow (Python Arrow) + * can read them back correctly. Skipped if python3/pyarrow is not available in the environment. + * + * <p>This test addresses cross-language compatibility: Java writes, another implementation reads. + */ + @Test + public void testJavaWrittenAlpFilesReadableByPyarrow() throws IOException, InterruptedException { Review Comment: Agreed, removed in b40a5d761. It was hacky: it shelled out to `python3`, built its assertions by string-concatenating a Python script, and then skipped itself if pyarrow could not read ALP, so in practice it never ran anywhere. One thing I would rather say out loud than let pass quietly. The C++ interop test and `testAlpColumnsMatchPlainColumns` both cover the inbound direction, C++ writes and Java reads. The pyarrow test was nominally the outbound check, Java writes and another implementation reads. Deleting it costs no CI coverage, since it never ran, but it does mean there is no automated Java-to-C++ check in the repo now. I verified that direction by hand in July against a local Arrow build with ALP support (apache/arrow#48345): it recognised `Encodings: ALP`, decoded cleanly, and all 2048 rows of the extreme-value columns matched the Java output bit for bit. Happy to wire it up properly once the Arrow side is released, if you think it is worth having. `TestInterOpReadAlp` now runs all 12 of its tests with nothing skipped. ########## parquet-column/src/test/java/org/apache/parquet/column/values/alp/AlpExceptionCountTest.java: ########## @@ -0,0 +1,215 @@ +/* + * 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.Assumptions.assumeTrue; + +import java.io.BufferedReader; +import java.io.File; +import java.io.FileReader; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import org.junit.jupiter.api.Test; + +/** + * Measures exception counts and compression ratios using the real Spotify and Arade datasets from + * the parquet-testing repository (Pratik's alpFloatingPointDataset branch). + * + * <p>These are the same datasets used in the meeting discussion to compare Java vs C++ ALP + * exception counts. Run from the project root with the alp-test-data/ directory present: + * + * <pre> + * ./mvnw test -pl parquet-column -Dtest=AlpExceptionCountTest + * </pre> + * + * <p>To use a different data directory: + * + * <pre> + * ./mvnw test -pl parquet-column -Dtest=AlpExceptionCountTest \ + * -DALP_TEST_DATA_DIR=/path/to/alp-test-data + * </pre> + */ +public class AlpExceptionCountTest { Review Comment: Agreed, and removed in b40a5d761. You are right that it stopped earning its place: it needed an `alp-test-data` directory that is not in the repo, so all three tests skipped everywhere, including locally for anyone who had not fetched the datasets by hand. It was genuinely useful while I was checking exception rates against the numbers in the ALP paper, but that is a one-off exercise rather than something worth keeping in the suite. With this and the pyarrow test gone, parquet-column runs 882 tests with nothing skipped. -- 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]
