wgtmac commented on code in PR #3397: URL: https://github.com/apache/parquet-java/pull/3397#discussion_r4078572049
########## 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: Should we remove this file as well? I think it is useful to locally verify the ALP PoC but now it does not bring too much value since it cannot run in CI and others are not easy to run it as well. ########## 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: Do we really need this test? It looks hacky to me. I think the C++ ALP implementation has already added an interop test by reading the file from parquet-testing repo. We can remove this now. -- 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]
