emkornfield commented on code in PR #3397: URL: https://github.com/apache/parquet-java/pull/3397#discussion_r3504172424
########## parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInterOpReadAlp.java: ########## @@ -0,0 +1,1066 @@ +/* + * 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.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assume.assumeTrue; + +import java.io.File; +import java.io.IOException; +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.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +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); + + @Rule + public TemporaryFolder temp = new TemporaryFolder(); + + 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("ALP_TEST_FILE not set or file does not exist, skipping", file != null); + List<Group> rows = readAllRows(file); + assertTrue("Expected at least one row in " + file, rows.size() > 0); + LOG.info("testReadSingleAlpFile: read {} rows from {}", rows.size(), file.getFileName()); + } + + @Test + public void testReadCppDoubleFiles() throws IOException { + java.nio.file.Path dir = getTestDataDir(); + assumeTrue("alp-test-data/ directory not found, skipping", dir != null); + for (String filename : CPP_DOUBLE_FILES) { + java.nio.file.Path file = dir.resolve(filename); + assumeTrue("File not found: " + file, file.toFile().exists()); + List<Group> rows = readAllRows(file); + assertTrue("Expected rows in " + filename, rows.size() > 0); + LOG.info("testReadCppDoubleFiles: {} → {} rows OK", filename, rows.size()); + } + } + + @Test + public void testReadCppFloatFiles() throws IOException { + java.nio.file.Path dir = getTestDataDir(); + assumeTrue("alp-test-data/ directory not found, skipping", dir != null); + for (String filename : CPP_FLOAT_FILES) { + java.nio.file.Path file = dir.resolve(filename); + assumeTrue("File not found: " + file, file.toFile().exists()); + List<Group> rows = readAllRows(file); + assertTrue("Expected rows in " + filename, rows.size() > 0); + 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("alp-test-data/ directory not found, skipping", dir != null); + 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); + assertEquals("Unexpected NaN in " + filename, 0, nanCount); + assertEquals("Unexpected Inf in " + filename, 0, infCount); + } + } + + /** Schema inspection: log encoding types used in each column. */ + @Test + public void testLogSchemaAndEncodings() throws IOException { + java.nio.file.Path dir = getTestDataDir(); + assumeTrue("alp-test-data/ directory not found, skipping", dir != null); + 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 + assertTrue(true); + } + + 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 = + temp.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) + .withAlpEncoding(true) + .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); + assertEquals("Row count mismatch for " + version, DOUBLE_VALUES.length, rows.size()); + for (int i = 0; i < DOUBLE_VALUES.length; i++) { + assertEquals( + "double_col mismatch at row " + i + " for " + version, + DOUBLE_VALUES[i], + rows.get(i).getDouble("double_col", 0), + 0.0); + assertEquals( + "float_col mismatch at row " + i + " for " + version, + FLOAT_VALUES[i], + rows.get(i).getFloat("float_col", 0), + 0.0f); + } + 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 = + temp.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) + .withAlpEncoding(true) + .withAlpVectorSize(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); + assertEquals("Row count mismatch at vectorSize=4096", rowCount, rows.size()); + for (int i = 0; i < rowCount; i++) { + assertEquals( + "double_col mismatch at row " + i, doubles[i], rows.get(i).getDouble("double_col", 0), 0.0); + assertEquals( + "float_col mismatch at row " + i, floats[i], rows.get(i).getFloat("float_col", 0), 0.0f); + } + 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. + * V2 page headers include the encoding value directly; ALP = 10 is supported via the build-time + * patch to the generated Encoding enum in parquet-format-structures. + */ + @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 { + // Check python3 is available + Process check = new ProcessBuilder("python3", "-c", "import pyarrow.parquet") + .redirectErrorStream(true) + .start(); + assumeTrue("python3/pyarrow not available, skipping cross-language test", check.waitFor() == 0); + + for (WriterVersion version : new WriterVersion[] {WriterVersion.PARQUET_1_0, WriterVersion.PARQUET_2_0}) { + MessageType schema = MessageTypeParser.parseMessageType(ALP_SCHEMA); + java.nio.file.Path outPath = + temp.newFolder().toPath().resolve("alp_java_xcompat_" + version.name().toLowerCase() + ".parquet"); + + try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new LocalOutputFile(outPath)) + .withType(schema) + .withCompressionCodec(CompressionCodecName.UNCOMPRESSED) + .withWriterVersion(version) + .withAlpEncoding(true) + .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); + } + } + + // Build a python snippet that reads the file and asserts row count and values + StringBuilder pyScript = new StringBuilder(); + pyScript.append("import pyarrow.parquet as pq, sys\n"); + pyScript.append("t = pq.read_table('").append(outPath.toAbsolutePath()).append("')\n"); + pyScript.append("assert len(t) == ").append(DOUBLE_VALUES.length).append(", f'row count mismatch: {len(t)}'\n"); + pyScript.append("dc = t.column('double_col').to_pylist()\n"); + pyScript.append("fc = t.column('float_col').to_pylist()\n"); + for (int i = 0; i < DOUBLE_VALUES.length; i++) { + pyScript.append("assert abs(dc[").append(i).append("] - ").append(DOUBLE_VALUES[i]) + .append(") < 1e-9, f'double mismatch at ").append(i).append(": {dc[").append(i).append("]}'\n"); + pyScript.append("assert abs(fc[").append(i).append("] - ").append(FLOAT_VALUES[i]) + .append(") < 1e-4, f'float mismatch at ").append(i).append(": {fc[").append(i).append("]}'\n"); + } + pyScript.append("print('OK: " + version.name() + " " + DOUBLE_VALUES.length + " rows verified by pyarrow')\n"); + + Process proc = new ProcessBuilder("python3", "-c", pyScript.toString()) + .redirectErrorStream(true) + .start(); + String output = new String(proc.getInputStream().readAllBytes()); + int exitCode = proc.waitFor(); + LOG.info("pyarrow cross-compat [{}]: {}", version, output.trim()); + // pyarrow may not yet support reading ALP (Arrow C++ PR #48345 in progress). + // Skip rather than fail so the test becomes a passing signal once pyarrow adds ALP support. + assumeTrue( + "pyarrow does not yet support reading ALP encoding (Arrow C++ PR #48345 pending): " + + output, + !output.contains("Unknown encoding type")); + assertEquals( + "pyarrow failed to read Java-written ALP file (" + version + "): " + output, 0, exitCode); + } + } + + // TODO: Uncomment once parquet-testing PR #100 merges + /* + @Test + public void testReadAlpFromParquetTesting() throws IOException { + // InterOpTester will auto-download from parquet-testing + } + */ + + // --------------------------------------------------------------------------- + // Fixture generator: reads the C++-written ALP files from parquet-testing + // PR #100 and re-encodes each as Java ALP at two vector sizes (1024 and 4096). + // Outputs land in ALP_OUTPUT_DIR (default: ${user.dir}/alp-java-generated/) + // and are intended for submission back to parquet-testing PR #100 so the + // C++/Rust/Go readers can verify cross-language compatibility against Java + // writes at both vector sizes. + // --------------------------------------------------------------------------- + + private static final String[] SOURCE_FILES = { + "alp_spotify1.parquet", "alp_arade.parquet", "alp_float_spotify1.parquet", "alp_float_arade.parquet" + }; + private static final int[] GENERATOR_VECTOR_SIZES = {1024, 4096}; + private static final WriterVersion[] GENERATOR_PAGE_VERSIONS = { + WriterVersion.PARQUET_1_0, WriterVersion.PARQUET_2_0 + }; + + /** + * Resolves the explicit output directory from {@code ALP_OUTPUT_DIR} (system property or env var) + * and creates it if missing. Returns {@code null} when the variable is unset — callers should + * {@code assumeTrue} on the result so the test skips cleanly rather than writing stray files + * into the module tree (which would fail the RAT license check on subsequent {@code mvn test} + * runs in the same module). + */ + private java.nio.file.Path getOutputDir() throws IOException { + String dir = System.getProperty("ALP_OUTPUT_DIR"); + if (dir == null) dir = System.getenv("ALP_OUTPUT_DIR"); + if (dir == null) return null; + java.nio.file.Path target = Paths.get(dir); + java.nio.file.Files.createDirectories(target); + return target; + } + + /** + * Read all rows of the given source file and return them as a list of value-only Group rows + * keyed by the source's schema. Used for both the copy and the verification round-trip. + */ + private List<Group> readGroupsWithSchema(java.nio.file.Path source, MessageType[] outSchema) throws IOException { + List<Group> rows = new ArrayList<>(); + try (ParquetFileReader reader = ParquetFileReader.open(new LocalInputFile(source))) { + MessageType schema = reader.getFooter().getFileMetaData().getSchema(); + outSchema[0] = schema; + 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; + } + + /** + * Generates Java-written ALP fixtures from the four PR #100 source datasets across two axes: + * page version (V1, V2) and vectorSize (1024, 4096). Each output is verified bit-exact against + * its source. Produces 16 files total: + * {@code alp_java_{spotify1,arade,float_spotify1,float_arade}_v{1,2}_vs{1024,4096}.parquet}. + * + * <p>Page version is orthogonal to ALP encoding (the page version difference lives in the + * parquet protocol layer, not in the ALP payload), but covering both axes makes the fixture + * set fully symmetric for cross-language compatibility — every reader can verify it handles + * Java-written ALP regardless of how the surrounding pages are framed. + * + * <p>To run: clone https://github.com/prtkgaur/parquet-testing/tree/alpFloatingPointDataset + * and point ALP_TEST_DATA_DIR at its {@code data/} directory. Set ALP_OUTPUT_DIR to choose + * where to write the generated fixtures; if unset, this test skips. + */ + @Test + public void generateAlpFixturesAtMultipleVectorSizes() throws IOException { + java.nio.file.Path sourceDir = getTestDataDir(); + assumeTrue("ALP source dir not found, skipping fixture generator", sourceDir != null); + + java.nio.file.Path outDir = getOutputDir(); + assumeTrue("ALP_OUTPUT_DIR not set, skipping fixture generator", outDir != null); + LOG.info("Generating ALP fixtures to {}", outDir); + + int expectedFiles = + SOURCE_FILES.length * GENERATOR_PAGE_VERSIONS.length * GENERATOR_VECTOR_SIZES.length; + int generated = 0; + for (String sourceFile : SOURCE_FILES) { + java.nio.file.Path source = sourceDir.resolve(sourceFile); + assumeTrue("Source missing: " + source, source.toFile().exists()); + + MessageType[] schemaHolder = new MessageType[1]; + List<Group> sourceRows = readGroupsWithSchema(source, schemaHolder); + MessageType schema = schemaHolder[0]; + + String stem = sourceFile.replace("alp_", "").replace(".parquet", ""); + + for (WriterVersion pageVersion : GENERATOR_PAGE_VERSIONS) { + String pageTag = (pageVersion == WriterVersion.PARQUET_1_0) ? "v1" : "v2"; + for (int vectorSize : GENERATOR_VECTOR_SIZES) { + java.nio.file.Path outPath = + outDir.resolve( + "alp_java_" + stem + "_" + pageTag + "_vs" + vectorSize + ".parquet"); + java.nio.file.Files.deleteIfExists(outPath); + + try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new LocalOutputFile(outPath)) + .withType(schema) + .withCompressionCodec(CompressionCodecName.UNCOMPRESSED) + .withWriterVersion(pageVersion) + .withAlpEncoding(true) + .withAlpVectorSize(vectorSize) + .withDictionaryEncoding(false) + .withConf(new Configuration()) + .build()) { + for (Group row : sourceRows) { + writer.write(row); + } + } + + // Verify: read back and compare against the source row-by-row + List<Group> roundTrip = readGroupsWithSchema(outPath, new MessageType[1]); + assertEquals( + "Row count mismatch for " + outPath.getFileName(), + sourceRows.size(), + roundTrip.size()); + int fieldCount = schema.getFieldCount(); + for (int i = 0; i < sourceRows.size(); i++) { + for (int f = 0; f < fieldCount; f++) { + PrimitiveType.PrimitiveTypeName type = + schema.getType(f).asPrimitiveType().getPrimitiveTypeName(); + String fieldName = schema.getType(f).getName(); + if (type == PrimitiveType.PrimitiveTypeName.DOUBLE) { + double expected = sourceRows.get(i).getDouble(f, 0); + double actual = roundTrip.get(i).getDouble(f, 0); + long eb = Double.doubleToRawLongBits(expected); + long ab = Double.doubleToRawLongBits(actual); + assertEquals( + "double bit mismatch in " + outPath.getFileName() + " row " + i + " field " + fieldName, + eb, + ab); + } else if (type == PrimitiveType.PrimitiveTypeName.FLOAT) { + float expected = sourceRows.get(i).getFloat(f, 0); + float actual = roundTrip.get(i).getFloat(f, 0); + int eb = Float.floatToRawIntBits(expected); + int ab = Float.floatToRawIntBits(actual); + assertEquals( + "float bit mismatch in " + outPath.getFileName() + " row " + i + " field " + fieldName, + eb, + ab); + } + } + } + long bytes = outPath.toFile().length(); + LOG.info( + " wrote {} ({} rows, {} bytes, pageVersion={}, vectorSize={})", + outPath.getFileName(), + sourceRows.size(), + bytes, + pageTag, + vectorSize); + generated++; + } + } + } + assertEquals( + "Expected to generate " + expectedFiles + " fixture files", expectedFiles, generated); + LOG.info("Generated {} ALP fixture files to {}", generated, outDir); + } + + // --------------------------------------------------------------------------- + // Reader-only validation: opens each Java-written fixture and asserts the + // reader path works end-to-end (metadata declares ALP, every row decodes + // without error). Separate from the generator's own round-trip check so it + // surfaces as a distinct signal — useful when the fixtures already exist on + // disk and you want to validate just the read path. + // --------------------------------------------------------------------------- + + @Test + public void readAllFixtureFilesIndependently() throws IOException { + java.nio.file.Path outDir = getOutputDir(); + assumeTrue("ALP_OUTPUT_DIR not set, skipping reader-only test", outDir != null); + File[] files = + outDir.toFile().listFiles((f, n) -> n.startsWith("alp_java_") && n.endsWith(".parquet")); + assumeTrue("No fixture files in " + outDir, files != null && files.length > 0); + java.util.Arrays.sort(files, (a, b) -> a.getName().compareTo(b.getName())); + + int filesRead = 0; + int totalChunks = 0; + int alpChunks = 0; + long totalRows = 0; + for (File pf : files) { + try (ParquetFileReader reader = ParquetFileReader.open(new LocalInputFile(pf.toPath()))) { + ParquetMetadata footer = reader.getFooter(); + // Layer 1: every column chunk declares ALP + for (org.apache.parquet.hadoop.metadata.BlockMetaData block : footer.getBlocks()) { + for (org.apache.parquet.hadoop.metadata.ColumnChunkMetaData col : block.getColumns()) { + totalChunks++; + if (col.getEncodings().contains(Encoding.ALP)) alpChunks++; + } + } + // Layer 2: every row actually decodes (no exceptions thrown, count matches footer) + MessageType schema = footer.getFileMetaData().getSchema(); + MessageColumnIO columnIO = new ColumnIOFactory().getColumnIO(schema); + long rowsDecoded = 0; + 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++) { + Group row = recordReader.read(); + // Touch every present field so any decode-time exception surfaces. + // Optional/null fields have repetitionCount=0 — skip those rather than throw. + for (int f = 0; f < schema.getFieldCount(); f++) { + if (row.getFieldRepetitionCount(f) == 0) continue; + PrimitiveType.PrimitiveTypeName type = + schema.getType(f).asPrimitiveType().getPrimitiveTypeName(); + if (type == PrimitiveType.PrimitiveTypeName.DOUBLE) { + row.getDouble(f, 0); + } else if (type == PrimitiveType.PrimitiveTypeName.FLOAT) { + row.getFloat(f, 0); + } + } + rowsDecoded++; + } + } + totalRows += rowsDecoded; + filesRead++; + } + } + assertEquals("Every column chunk should declare ALP", totalChunks, alpChunks); + LOG.info( + "readAllFixtureFilesIndependently: {} files, {} chunks ({} ALP), {} rows decoded", + filesRead, totalChunks, alpChunks, totalRows); + } + + // --------------------------------------------------------------------------- + // Corner-case fixture per parquet-testing issue #105. Writes a single small + // file with one ALP-encoded column per corner case (no exceptions, 1 exception, + // all exceptions, NaN/Inf/-0.0, nulls, constant/bit_width=0, multi-vector with + // differing exponents) for both f32 and f64, then reads back and verifies + // every value bit-exactly against the expected pattern. + // + // The output file (alp_java_cornercases.parquet) is intended as a candidate + // fixture for parquet-testing PR #100 once Prateek confirms naming and design. + // --------------------------------------------------------------------------- + + private static final int CORNER_ROWS_PER_VECTOR = 1024; + private static final int CORNER_VECTORS = 2; + private static final int CORNER_TOTAL_ROWS = CORNER_ROWS_PER_VECTOR * CORNER_VECTORS; + + /** Expected values per corner-case column. Doubles use NaN to mark a null cell for sentinel use. */ + private static class CornerCaseData { + double[] doubleVals; + float[] floatVals; + boolean[] isNull; // only meaningful for optional columns + boolean isFloat; + boolean isOptional; + String name; + + static CornerCaseData doubles(String name, double[] vals) { + CornerCaseData d = new CornerCaseData(); + d.name = name; + d.doubleVals = vals; + d.isFloat = false; + d.isOptional = false; + return d; + } + + static CornerCaseData floats(String name, float[] vals) { + CornerCaseData d = new CornerCaseData(); + d.name = name; + d.floatVals = vals; + d.isFloat = true; + d.isOptional = false; + return d; + } + + static CornerCaseData optionalDoubles(String name, double[] vals, boolean[] nulls) { + CornerCaseData d = doubles(name, vals); + d.isOptional = true; + d.isNull = nulls; + return d; + } + + static CornerCaseData optionalFloats(String name, float[] vals, boolean[] nulls) { + CornerCaseData d = floats(name, vals); + d.isOptional = true; + d.isNull = nulls; + return d; + } + } + + private List<CornerCaseData> buildCornerCaseColumns() { + List<CornerCaseData> cols = new ArrayList<>(); + int N = CORNER_TOTAL_ROWS; + + // 1. f64_no_exceptions: clean 2-decimal values, both vectors low bit-width + double[] noExc = new double[N]; + for (int i = 0; i < N; i++) noExc[i] = (i % 1000) / 100.0; + cols.add(CornerCaseData.doubles("f64_no_exceptions", noExc)); + + // 2. f64_one_exception_per_vector: clean values with 1 NaN per 1024-row vector + double[] oneExc = new double[N]; + for (int i = 0; i < N; i++) oneExc[i] = (i % 1000) / 100.0; + oneExc[100] = Double.NaN; + oneExc[1100] = Double.NaN; + cols.add(CornerCaseData.doubles("f64_one_exception_per_vector", oneExc)); + + // 3. f64_all_nan: every value NaN — all-exception vector + double[] allNan = new double[N]; + java.util.Arrays.fill(allNan, Double.NaN); + cols.add(CornerCaseData.doubles("f64_all_nan", allNan)); + + // 4. f64_nan_inf_negzero: mix of special values across two vectors + double[] mixed = new double[N]; + for (int i = 0; i < N; i++) mixed[i] = (i % 100) / 10.0; + mixed[0] = Double.NaN; + mixed[1] = Double.POSITIVE_INFINITY; + mixed[2] = Double.NEGATIVE_INFINITY; + mixed[3] = -0.0; + mixed[1024] = Double.NaN; + mixed[1025] = -0.0; + cols.add(CornerCaseData.doubles("f64_nan_inf_negzero", mixed)); + + // 5. f64_constant: every value = 42.0 → max_delta=0 → bit_width=0 + double[] constant = new double[N]; + java.util.Arrays.fill(constant, 42.0); + cols.add(CornerCaseData.doubles("f64_constant", constant)); + + // 6. f64_different_e_f_per_vector: vector 0 uses e=2,f=0 (cents); vector 1 uses e=3,f=0 (mils) + double[] diff = new double[N]; + for (int i = 0; i < CORNER_ROWS_PER_VECTOR; i++) diff[i] = (i % 100) / 100.0; + for (int i = CORNER_ROWS_PER_VECTOR; i < N; i++) diff[i] = ((i - CORNER_ROWS_PER_VECTOR) % 1000) / 1000.0; + cols.add(CornerCaseData.doubles("f64_different_ef_per_vector", diff)); + + // 7. f64_with_nulls: optional column, ~30% nulls scattered + double[] withNulls = new double[N]; + boolean[] nullMask = new boolean[N]; + for (int i = 0; i < N; i++) { + if (i % 3 == 0) { + nullMask[i] = true; + } else { + withNulls[i] = (i % 100) / 100.0; + } + } + cols.add(CornerCaseData.optionalDoubles("f64_with_nulls", withNulls, nullMask)); + + // 8–14: f32 analogues + float[] noExcF = new float[N]; + for (int i = 0; i < N; i++) noExcF[i] = (i % 1000) / 100.0f; + cols.add(CornerCaseData.floats("f32_no_exceptions", noExcF)); + + float[] oneExcF = new float[N]; + for (int i = 0; i < N; i++) oneExcF[i] = (i % 1000) / 100.0f; + oneExcF[100] = Float.NaN; + oneExcF[1100] = Float.NaN; + cols.add(CornerCaseData.floats("f32_one_exception_per_vector", oneExcF)); + + float[] allNanF = new float[N]; + java.util.Arrays.fill(allNanF, Float.NaN); + cols.add(CornerCaseData.floats("f32_all_nan", allNanF)); + + float[] mixedF = new float[N]; + for (int i = 0; i < N; i++) mixedF[i] = (i % 100) / 10.0f; + mixedF[0] = Float.NaN; + mixedF[1] = Float.POSITIVE_INFINITY; + mixedF[2] = Float.NEGATIVE_INFINITY; + mixedF[3] = -0.0f; + mixedF[1024] = Float.NaN; + mixedF[1025] = -0.0f; + cols.add(CornerCaseData.floats("f32_nan_inf_negzero", mixedF)); + + float[] constantF = new float[N]; + java.util.Arrays.fill(constantF, 42.0f); + cols.add(CornerCaseData.floats("f32_constant", constantF)); + + float[] diffF = new float[N]; + for (int i = 0; i < CORNER_ROWS_PER_VECTOR; i++) diffF[i] = (i % 100) / 100.0f; + for (int i = CORNER_ROWS_PER_VECTOR; i < N; i++) diffF[i] = ((i - CORNER_ROWS_PER_VECTOR) % 1000) / 1000.0f; + cols.add(CornerCaseData.floats("f32_different_ef_per_vector", diffF)); + + float[] withNullsF = new float[N]; + boolean[] nullMaskF = new boolean[N]; + for (int i = 0; i < N; i++) { + if (i % 3 == 0) { + nullMaskF[i] = true; + } else { + withNullsF[i] = (i % 100) / 100.0f; + } + } + cols.add(CornerCaseData.optionalFloats("f32_with_nulls", withNullsF, nullMaskF)); + + return cols; + } + + /** + * Dumps the corner-case construction recipe to a CSV truth file alongside the parquet. + * The CSV is built directly from the expected values (not by reading the parquet back), + * which makes it suitable as independent ground truth for cross-language verifiers. + * + * <p>Conventions: + * <ul> + * <li>Header row matches column names. + * <li>Empty field marks a null cell in an optional column. + * <li>Special values use Java's standard toString output: {@code NaN}, {@code Infinity}, + * {@code -Infinity}. These parse correctly via C++ {@code std::stod} / {@code std::stof} + * (per the standard, both are case-insensitive and accept the full word "Infinity"). + * <li>{@code -0.0} round-trips through both Java and C++ string conversion bit-exact. + * </ul> + */ + private void writeCornerCaseCsvTruth(java.nio.file.Path csvPath, List<CornerCaseData> cols) throws IOException { + try (java.io.BufferedWriter w = java.nio.file.Files.newBufferedWriter(csvPath)) { + // Header + for (int i = 0; i < cols.size(); i++) { + if (i > 0) w.write(","); + w.write(cols.get(i).name); + } + w.write("\n"); + // Data + for (int r = 0; r < CORNER_TOTAL_ROWS; r++) { + for (int i = 0; i < cols.size(); i++) { + if (i > 0) w.write(","); + CornerCaseData c = cols.get(i); + if (c.isOptional && c.isNull[r]) { + // empty field = null + } else if (c.isFloat) { + w.write(Float.toString(c.floatVals[r])); + } else { + w.write(Double.toString(c.doubleVals[r])); + } + } + w.write("\n"); + } + } + } + + @Test + public void generateAndVerifyCornerCaseFixture() throws IOException { Review Comment: another corner case is when MIN and MAX INT/LONG values end up getting encoding in FOR, commented on the C++ test but we should ensure we have tests that cover that as well. -- 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]
