RussellSpitzer commented on code in PR #3397: URL: https://github.com/apache/parquet-java/pull/3397#discussion_r3973409372
########## parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInterOpReadAlp.java: ########## @@ -0,0 +1,1406 @@ +/* + * 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.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.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"); + if (dir == null) dir = System.getenv("ALP_TEST_DATA_DIR"); + if (dir != null && new File(dir).isDirectory()) return Paths.get(dir); + // Default: alp-test-data/ relative to project root (two levels up from target/test-classes) + java.nio.file.Path candidate = Paths.get(System.getProperty("user.dir")).resolve("alp-test-data"); + return candidate.toFile().isDirectory() ? candidate : null; + } + + private java.nio.file.Path getSingleTestFile() { + String filePath = System.getProperty("ALP_TEST_FILE"); + if (filePath == null) filePath = System.getenv("ALP_TEST_FILE"); + if (filePath != null && new File(filePath).exists()) return Paths.get(filePath); + return null; + } + + /** 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; + } + + @Test + public void testReadSingleAlpFile() throws IOException { + java.nio.file.Path file = getSingleTestFile(); + assumeTrue(file != null, "ALP_TEST_FILE not set or file does not exist, skipping"); + List<Group> rows = readAllRows(file); + assertThat(rows.size() > 0).as("Expected at least one row in " + file).isTrue(); + LOG.info("testReadSingleAlpFile: read {} rows from {}", rows.size(), file.getFileName()); + } + + @Test + public void testReadCppDoubleFiles() throws IOException { + java.nio.file.Path dir = getTestDataDir(); + assumeTrue(dir != null, "alp-test-data/ directory not found, skipping"); + for (String filename : CPP_DOUBLE_FILES) { + java.nio.file.Path file = dir.resolve(filename); + assumeTrue(file.toFile().exists(), "File not found: " + file); + List<Group> rows = readAllRows(file); + assertThat(rows.size() > 0).as("Expected rows in " + filename).isTrue(); + LOG.info("testReadCppDoubleFiles: {} → {} rows OK", filename, rows.size()); + } + } + + @Test + public void testReadCppFloatFiles() throws IOException { + java.nio.file.Path dir = getTestDataDir(); + assumeTrue(dir != null, "alp-test-data/ directory not found, skipping"); + for (String filename : CPP_FLOAT_FILES) { + java.nio.file.Path file = dir.resolve(filename); + assumeTrue(file.toFile().exists(), "File not found: " + file); + List<Group> rows = readAllRows(file); + assertThat(rows.size() > 0).as("Expected rows in " + filename).isTrue(); + LOG.info("testReadCppFloatFiles: {} → {} rows OK", filename, rows.size()); + } + } + + /** Verify no NaN/Inf corruption in float/double columns across all test files. */ + @Test + public void testNoCorruptionInCppFiles() throws IOException { + java.nio.file.Path dir = getTestDataDir(); + assumeTrue(dir != null, "alp-test-data/ directory not found, skipping"); + String[] allFiles = { + "alp_spotify1.parquet", "alp_arade.parquet", "alp_float_spotify1.parquet", "alp_float_arade.parquet" + }; + for (String filename : allFiles) { + java.nio.file.Path file = dir.resolve(filename); + if (!file.toFile().exists()) continue; + int nanCount = 0; + int infCount = 0; + int totalValues = 0; + try (ParquetFileReader reader = ParquetFileReader.open(new LocalInputFile(file))) { + MessageType schema = reader.getFooter().getFileMetaData().getSchema(); + MessageColumnIO columnIO = new ColumnIOFactory().getColumnIO(schema); + PageReadStore pages; + while ((pages = reader.readNextRowGroup()) != null) { + RecordReader<Group> recordReader = + columnIO.getRecordReader(pages, new GroupRecordConverter(schema)); + for (long i = 0; i < pages.getRowCount(); i++) { + Group row = recordReader.read(); + for (int f = 0; f < schema.getFieldCount(); f++) { + PrimitiveType.PrimitiveTypeName type = + schema.getType(f).asPrimitiveType().getPrimitiveTypeName(); + try { + if (type == PrimitiveType.PrimitiveTypeName.FLOAT) { + float v = row.getFloat(f, 0); + totalValues++; + if (Float.isNaN(v)) nanCount++; + if (Float.isInfinite(v)) infCount++; + } else if (type == PrimitiveType.PrimitiveTypeName.DOUBLE) { + double v = row.getDouble(f, 0); + totalValues++; + if (Double.isNaN(v)) nanCount++; + if (Double.isInfinite(v)) infCount++; + } + } catch (Exception ignored) { + } + } + } + } + } + LOG.info("{}: {} values, {} NaN, {} Inf", filename, totalValues, nanCount, infCount); + assertThat(nanCount).as("Unexpected NaN in " + filename).isEqualTo(0); + assertThat(infCount).as("Unexpected Inf in " + filename).isEqualTo(0); + } + } + + /** Schema inspection: log encoding types used in each column. */ + @Test + public void testLogSchemaAndEncodings() throws IOException { + java.nio.file.Path dir = getTestDataDir(); + assumeTrue(dir != null, "alp-test-data/ directory not found, skipping"); + String[] allFiles = { + "alp_spotify1.parquet", "alp_arade.parquet", "alp_float_spotify1.parquet", "alp_float_arade.parquet" + }; + for (String filename : allFiles) { + java.nio.file.Path file = dir.resolve(filename); + if (!file.toFile().exists()) continue; + try (ParquetFileReader reader = ParquetFileReader.open(new LocalInputFile(file))) { + ParquetMetadata footer = reader.getFooter(); + MessageType schema = footer.getFileMetaData().getSchema(); + LOG.info("=== {} ===", filename); + LOG.info("Schema: {}", schema); + if (!footer.getBlocks().isEmpty()) { + for (org.apache.parquet.hadoop.metadata.ColumnChunkMetaData col : + footer.getBlocks().get(0).getColumns()) { + LOG.info( + " column={} encodings={} compression={}", + col.getPath(), + col.getEncodings(), + col.getCodec()); + } + } + } catch (Exception e) { + LOG.warn("Could not read metadata for {}: {}", filename, e.getMessage()); + } + } + // This test always passes — it's for inspection + assertThat(true).isTrue(); + } + + 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. Review Comment: Java doc here is not in sync. One of the reasons I'm always pushing on not including details in javadocs because they changed very fast :) -- 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]
