diff --git a/CHANGELOG.md b/CHANGELOG.md index 2493b968e..9cdcb110c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - A nullable timestamp column holding nulls is written as `vortex.datetimeparts` with the validity on its days part, as Rust writes it, instead of `vortex.ext` over a masked i64. ([#627](https://github.com/dfa1/vortex-java/issues/627)) - A string or binary column whose FSST codes pass 2 GiB in one chunk, such as large audio blobs, writes and reads back instead of failing with an integer overflow. ([#629](https://github.com/dfa1/vortex-java/issues/629)) - A string or binary chunk past 2 GiB writes and reads back as `vortex.varbin`, `vortex.varbinview` and compact `vortex.zstd` too, and Rust-written `varbin` columns with `u32` offsets past 2 GiB read instead of failing as out of range. ([#633](https://github.com/dfa1/vortex-java/issues/633)) -- Parquet import reads a `DECIMAL` column as a Vortex decimal with its precision and scale, in all four physical encodings; `INT32`/`INT64` ones lost their scale and the byte-array ones failed. ([#628](https://github.com/dfa1/vortex-java/issues/628)) +- Parquet import reads a `DECIMAL` column as a Vortex decimal with its precision and scale, in all four physical encodings, and names the column of one wider than Vortex's 76 digits; `INT32`/`INT64` ones lost their scale and the byte-array ones failed. ([#628](https://github.com/dfa1/vortex-java/issues/628)) - Parquet import keeps the nulls of a nullable `TIMESTAMP` column instead of failing as if the column were `REQUIRED`. ([#627](https://github.com/dfa1/vortex-java/issues/627)) ## [0.19.0] — 2026-11-01 diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/ParquetImportInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/ParquetImportInteropIntegrationTest.java index 5b38624d4..db52d6ebb 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/ParquetImportInteropIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/ParquetImportInteropIntegrationTest.java @@ -61,6 +61,8 @@ class ParquetImportInteropIntegrationTest { private static final ColumnName TS = ColumnName.of("ts"); /// More than two zones (8192 rows), so the filtered read prunes and the range read slices. private static final int ROWS = 3 * 8192; + /// 10^37: a 12-digit value times it reaches 49 digits, past an i128 (about 1.7e38). + private static final BigInteger WIDE = BigInteger.TEN.pow(37); private static final long START_MICROS = 1_704_067_200_000_000L; // 2024-01-01T00:00:00Z static { @@ -110,7 +112,7 @@ void nullableTimestampWithNulls_jniReadsImportedFile(@TempDir Path tmp) throws E "INT64, 0, 18, 4", "FIXED_LEN_BYTE_ARRAY, 7, 15, 2", "BYTE_ARRAY, 0, 30, 3", - "BYTE_ARRAY, 0, 50, 3", // past decimal128: Rust's i256 values + "BYTE_ARRAY, 0, 50, 3", // past decimal128: values only an i256 holds }) void decimalWithNulls_jniReadsImportedFile(PhysicalType type, int length, int precision, int scale, @TempDir Path tmp) throws Exception { @@ -120,7 +122,16 @@ void decimalWithNulls_jniReadsImportedFile(PhysicalType type, int length, int pr BigInteger[] unscaled = new BigInteger[ROWS]; List expected = new ArrayList<>(ROWS); for (int i = 0; i < ROWS; i++) { - unscaled[i] = isNull.test(i) ? BigInteger.ZERO : BigInteger.valueOf(Math.floorMod(i * 7_919L, 2 * bound) - bound); + // the product wraps 2 * bound many times, so values take both signs + BigInteger value = BigInteger.valueOf(Math.floorMod(i * 7_919_919_919L, 2 * bound) - bound); + if (precision > 38) { + // scaled past what an i128 holds, so Rust reads real i256s + value = value.multiply(WIDE); + } + if (isNull.test(i)) { + value = BigInteger.ZERO; + } + unscaled[i] = value; expected.add(isNull.test(i) ? null : new BigDecimal(unscaled[i], scale)); } Path parquet = decimalParquet(tmp, type, length, precision, scale, unscaled, isNull); diff --git a/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetImporter.java b/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetImporter.java index 2759a69d9..1bff8a5d3 100644 --- a/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetImporter.java +++ b/parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetImporter.java @@ -10,6 +10,7 @@ import dev.hardwood.schema.ColumnSchema; import dev.hardwood.schema.FileSchema; import dev.hardwood.schema.SchemaNode; +import io.github.dfa1.vortex.core.error.VortexException; import io.github.dfa1.vortex.core.model.ColumnName; import io.github.dfa1.vortex.core.model.DType; import io.github.dfa1.vortex.core.model.PType; @@ -32,6 +33,7 @@ import java.util.LinkedHashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; /// Reads a Parquet file and writes a Vortex file. @@ -234,7 +236,14 @@ private static DType mapLeafDType(PhysicalType type, LogicalType logical, boolea // value, big-endian for the byte forms); each maps to Vortex's decimal, as Arrow's // decimal128/256 does in Rust's Arrow import, never to the bare integer that drops the scale if (logical instanceof LogicalType.DecimalType(int precision, int scale)) { - return DType.Decimal.of(precision, scale, nullable); + try { + return DType.Decimal.of(precision, scale, nullable); + } catch (VortexException e) { + // Parquet allows a byte-array DECIMAL wider than Vortex's 76 digits (Rust's Arrow + // has no type for it either) + throw new UnsupportedOperationException("DECIMAL(" + precision + ", " + scale + ") column '" + + name + "' has no Vortex decimal: " + e.getMessage(), e); + } } return switch (type) { case BOOLEAN -> new DType.Bool(nullable); @@ -376,7 +385,7 @@ static void fillRow(RowReader reader, List names, List types, boolean isNull = types.get(c).nullable() && reader.isNull(name); if (types.get(c) instanceof DType.Primitive p && p.ptype() == PType.F16) { // Same carrier as I16, but the value is a 2-byte FIXED_LEN_BYTE_ARRAY, not an INT32 - putHalf(buffers[c], pos, isNull ? null : reader.getBinary(name)); + putHalf(buffers[c], pos, isNull ? null : Objects.requireNonNull(reader.getBinary(name))); continue; } switch (buffers[c]) { @@ -394,9 +403,12 @@ static void fillRow(RowReader reader, List names, List types, case Short[] arr -> arr[pos] = isNull ? null : (short) reader.getInt(name); case byte[] arr -> arr[pos] = (byte) reader.getInt(name); case Byte[] arr -> arr[pos] = isNull ? null : (byte) reader.getInt(name); - case String[] arr -> arr[pos] = isNull ? null : reader.getString(name); - case BigDecimal[] arr -> arr[pos] = isNull ? null : reader.getDecimal(name); - case byte[][] arr -> arr[pos] = isNull ? null : reader.getBinary(name); + case String[] arr -> arr[pos] = isNull ? null : Objects.requireNonNull(reader.getString(name)); + // getString/getBinary/getDecimal return null rather than throwing, so a null + // under a REQUIRED column becomes the NPE the catch below reports with its file + // row (a FLOAT16 one used to become a silent +0.0) + case BigDecimal[] arr -> arr[pos] = isNull ? null : Objects.requireNonNull(reader.getDecimal(name)); + case byte[][] arr -> arr[pos] = isNull ? null : Objects.requireNonNull(reader.getBinary(name)); default -> throw new UnsupportedOperationException( "unexpected buffer type: " + buffers[c].getClass().getSimpleName()); } diff --git a/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetImporterTest.java b/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetImporterTest.java index a64078840..88f9fc1bb 100644 --- a/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetImporterTest.java +++ b/parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetImporterTest.java @@ -375,6 +375,55 @@ void requiredColumnActuallyNull_throwsClearErrorInsteadOfRawNpe() { .hasMessageContaining("malformed.parquet") .hasCauseInstanceOf(NullPointerException.class); } + + @ParameterizedTest(name = "{0}") + @CsvSource({"utf8", "binary", "float16"}) + void requiredObjectColumnActuallyNull_reportsTheFileRow(String kind) { + // Given — getString/getBinary return null instead of throwing; a FLOAT16 null used to + // become a silent +0.0 + DType dtype = switch (kind) { + case "utf8" -> new DType.Utf8(false); + case "binary" -> new DType.Binary(false); + default -> new DType.Primitive(PType.F16, false); + }; + Object buffer = switch (kind) { + case "utf8" -> new String[1]; + case "binary" -> new byte[1][]; + default -> new short[1]; + }; + List names = List.of(ColumnName.of("v")); + ColumnBuilder[] nestedBuilders = new ColumnBuilder[1]; + if (kind.equals("utf8")) { + given(reader.getString("v")).willReturn(null); + } else { + given(reader.getBinary("v")).willReturn(null); + } + + // When / Then + assertThatThrownBy(() -> ParquetImporter.fillRow(reader, names, List.of(dtype), new Object[]{buffer}, + nestedBuilders, 0, "malformed.parquet", 42L)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("REQUIRED") + .hasMessageContaining("row 42"); + } + + @Test + void requiredDecimalActuallyNull_reportsTheFileRow() { + // Given — getDecimal returns null instead of throwing, unlike the primitive getters + List names = List.of(ColumnName.of("amount")); + List types = List.of(DType.Decimal.of(15, 2, false)); + Object[] buffers = {new BigDecimal[1]}; + ColumnBuilder[] nestedBuilders = new ColumnBuilder[1]; + given(reader.getDecimal("amount")).willReturn(null); + + // When / Then — the file's row 70_000, not the row within the chunk + assertThatThrownBy(() -> ParquetImporter.fillRow(reader, names, types, buffers, nestedBuilders, + 0, "malformed.parquet", 70_000L)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("amount") + .hasMessageContaining("REQUIRED") + .hasMessageContaining("row 70000"); + } } @Nested @@ -413,32 +462,107 @@ void decimalColumn_importsAsDecimalWithItsScale(PhysicalType type, int length, i } } + @ParameterizedTest(name = "{0}") + @CsvSource({ + "INT32, 0, 9, 2", + "INT64, 0, 18, 2", + "FIXED_LEN_BYTE_ARRAY, 7, 15, 2", + "BYTE_ARRAY, 0, 30, 2", + }) + void requiredDecimalColumn_importsAsNonNullableDecimal(PhysicalType type, int length, int precision, + int scale, @TempDir Path tmp) throws Exception { + // Given — a REQUIRED DECIMAL column, the non-nullable BigDecimal[] path + List values = List.of(new BigDecimal("17.00"), new BigDecimal("-0.05"), new BigDecimal("0.00")); + Path parquet = writeDecimalParquet(tmp, type, length, precision, scale, RepetitionType.REQUIRED, values); + Path vortex = tmp.resolve("out.vortex"); + + // When + ParquetImporter.importParquet(parquet, vortex); + + // Then + try (VortexReader reader = VortexReader.open(vortex); + ScanIterator iter = reader.scan(ScanOptions.all())) { + DType.Struct schema = (DType.Struct) reader.dtype(); + assertThat(schema.fieldTypes().getFirst()).isEqualTo(DType.Decimal.of(precision, scale, false)); + try (Chunk chunk = iter.next()) { + assertThat(decimals(chunk.column(ColumnName.of("d")), values.size())) + .containsExactlyElementsOf(values); + } + } + } + + @Test + void decimalWiderThanVortexAllows_throwsNamingTheColumn() { + // Given — Parquet allows a BYTE_ARRAY DECIMAL past Vortex's 76 digits + ColumnSchema schema = col("amount", PhysicalType.BYTE_ARRAY, RepetitionType.OPTIONAL, + new LogicalType.DecimalType(100, 2)); + + // When / Then + assertThatThrownBy(() -> ParquetImporter.mapDType(schema)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("amount") + .hasMessageContaining("DECIMAL(100, 2)"); + } + private static Path writeDecimalParquet(Path tmp, PhysicalType type, int length, int precision, int scale) throws Exception { + return writeDecimalParquet(tmp, type, length, precision, scale, RepetitionType.OPTIONAL, VALUES); + } + + private static Path writeDecimalParquet(Path tmp, PhysicalType type, int length, int precision, int scale, + RepetitionType repetition, List values) throws Exception { FileSchema.Builder builder = FileSchema.builder("schema"); LogicalType decimal = new LogicalType.DecimalType(precision, scale); if (length > 0) { - builder.addColumn("d", type, RepetitionType.OPTIONAL, length, decimal); + builder.addColumn("d", type, repetition, length, decimal); } else { - builder.addColumn("d", type, RepetitionType.OPTIONAL, decimal); + builder.addColumn("d", type, repetition, decimal); } - int n = VALUES.size(); + int n = values.size(); boolean[] nulls = new boolean[n]; BigInteger[] unscaled = new BigInteger[n]; for (int i = 0; i < n; i++) { - nulls[i] = VALUES.get(i) == null; - unscaled[i] = nulls[i] ? BigInteger.ZERO : VALUES.get(i).setScale(scale).unscaledValue(); + nulls[i] = values.get(i) == null; + unscaled[i] = nulls[i] ? BigInteger.ZERO : values.get(i).setScale(scale).unscaledValue(); } + // hardwood takes a null mask only for an OPTIONAL column, and its mask-less setters otherwise + boolean[] mask = repetition == RepetitionType.OPTIONAL ? nulls : null; Path parquet = tmp.resolve("decimal.parquet"); try (ParquetFileWriter writer = ParquetFileWriter.create(OutputFile.of(parquet), builder.build())) { writer.columnWriter().writeBatch(batch -> { switch (type) { - case INT32 -> batch.ints(0, Arrays.stream(unscaled).mapToInt(BigInteger::intValueExact).toArray(), nulls); - case INT64 -> batch.longs(0, Arrays.stream(unscaled).mapToLong(BigInteger::longValueExact).toArray(), nulls); - case FIXED_LEN_BYTE_ARRAY -> batch.fixed(0, Arrays.stream(unscaled).map(v -> bigEndian(v, length)) - .toArray(byte[][]::new), nulls); - case BYTE_ARRAY -> batch.bytes(0, Arrays.stream(unscaled).map(BigInteger::toByteArray) - .toArray(byte[][]::new), nulls); + case INT32 -> { + int[] ints = Arrays.stream(unscaled).mapToInt(BigInteger::intValueExact).toArray(); + if (mask == null) { + batch.ints(0, ints); + } else { + batch.ints(0, ints, mask); + } + } + case INT64 -> { + long[] longs = Arrays.stream(unscaled).mapToLong(BigInteger::longValueExact).toArray(); + if (mask == null) { + batch.longs(0, longs); + } else { + batch.longs(0, longs, mask); + } + } + case FIXED_LEN_BYTE_ARRAY -> { + byte[][] fixed = Arrays.stream(unscaled).map(v -> bigEndian(v, length)).toArray(byte[][]::new); + if (mask == null) { + batch.fixed(0, fixed); + } else { + batch.fixed(0, fixed, mask); + } + } + case BYTE_ARRAY -> { + byte[][] bytes = Arrays.stream(unscaled).map(BigInteger::toByteArray).toArray(byte[][]::new); + if (mask == null) { + batch.bytes(0, bytes); + } else { + batch.bytes(0, bytes, mask); + } + } default -> throw new IllegalArgumentException(type.toString()); } });