diff --git a/CHANGELOG.md b/CHANGELOG.md index af9224f26..221c37b29 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Fixed +- 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 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/docs/reference.md b/docs/reference.md index 5037a8fb6..72301d90d 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -492,7 +492,9 @@ A decoder claims its ids via `LayoutDecoder.layoutIds()` (a set — the zoned de Supports flat schemas and nested `LIST`/`STRUCT` columns, recursively composable (e.g. `LIST>`); `MAP` and `VARIANT` are not yet supported. Un-annotated `BYTE_ARRAY` maps -to `DType.Binary`. +to `DType.Binary`. A top-level `DECIMAL` column, in any of its four physical encodings (`INT32`, +`INT64`, `FIXED_LEN_BYTE_ARRAY`, `BYTE_ARRAY`), maps to `DType.Decimal` with the annotation's +precision and scale; a `DECIMAL` nested in a `LIST`/`STRUCT` is not yet supported. | Method | Notes | |---------------------------------------------------|-------------------------------------| 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 0fe3d82dc..5b38624d4 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 @@ -1,5 +1,11 @@ package io.github.dfa1.vortex.integration; +import dev.hardwood.OutputFile; +import dev.hardwood.metadata.LogicalType; +import dev.hardwood.metadata.PhysicalType; +import dev.hardwood.metadata.RepetitionType; +import dev.hardwood.schema.FileSchema; +import dev.hardwood.writer.ParquetFileWriter; import dev.vortex.api.DataSource; import dev.vortex.api.Expression; import dev.vortex.api.ImmutableScanOptions; @@ -19,18 +25,25 @@ import io.github.dfa1.vortex.writer.VortexWriter; import io.github.dfa1.vortex.writer.WriteOptions; import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.vector.FieldVector; import org.apache.arrow.vector.TimeStampVector; import org.apache.arrow.vector.VectorSchemaRoot; import org.apache.arrow.vector.ipc.ArrowReader; +import org.apache.arrow.vector.types.pojo.ArrowType; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import java.io.IOException; +import java.math.BigDecimal; +import java.math.BigInteger; import java.nio.channels.FileChannel; import java.nio.file.Path; import java.nio.file.StandardOpenOption; import java.time.Instant; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.Map; import java.util.function.IntPredicate; @@ -38,8 +51,9 @@ import static org.assertj.core.api.Assertions.assertThat; /// A Parquet file imported by [ParquetImporter] reads in vortex-jni with the values and nulls the -/// Parquet file holds. The Parquet input is written by [ParquetExporter] from a Java-written Vortex -/// file, so each case controls the exact Parquet column shape the Raincloud corpus tripped on. +/// Parquet file holds. Each case writes its own Parquet input ([ParquetExporter] from a Java-written +/// Vortex file, or hardwood's writer directly), so it controls the exact Parquet column shape the +/// Raincloud corpus tripped on. class ParquetImportInteropIntegrationTest { private static final Session SESSION = Session.create(); @@ -86,6 +100,50 @@ void nullableTimestampWithNulls_jniReadsImportedFile(@TempDir Path tmp) throws E assertThat(range).isEqualTo(expected.subList(1000, 20_000)); } + /// An OPTIONAL `DECIMAL` column in each Parquet physical encoding imports as a Vortex decimal + /// that vortex-jni reads back as the same values with the column's precision and scale (#628: + /// INT32/INT64 dropped the scale, FIXED_LEN_BYTE_ARRAY as in TPC-H lineitem and BYTE_ARRAY threw). + @ParameterizedTest(name = "{0} DECIMAL({2},{3})") + @CsvSource({ + // physical type, fixed length, precision, scale + "INT32, 0, 9, 2", + "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 + }) + void decimalWithNulls_jniReadsImportedFile(PhysicalType type, int length, int precision, int scale, + @TempDir Path tmp) throws Exception { + // Given + IntPredicate isNull = i -> i % 7 == 0 && i >= ROWS / 2; + long bound = precision < 10 ? 99_999_999L : 999_999_999_999L; + 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); + expected.add(isNull.test(i) ? null : new BigDecimal(unscaled[i], scale)); + } + Path parquet = decimalParquet(tmp, type, length, precision, scale, unscaled, isNull); + Path imported = tmp.resolve("imported.vortex"); + BigInteger threshold = unscaled[ROWS / 2 + 1]; + + // When + ParquetImporter.importParquet(parquet, imported, ImportOptions.defaults().withChunkSize(4096)); + List all = readDecimalsWithJni(imported, ScanOptions.builder(), precision, scale); + // a filtered and a row-range read prune and slice, which a full scan never does + List filtered = readDecimalsWithJni(imported, ScanOptions.builder().filter(Expression.binary( + Expression.BinaryOp.GTE, Expression.column("d"), + Expression.literalDecimal(threshold, precision, scale))), precision, scale); + List range = readDecimalsWithJni(imported, ScanOptions.builder().rowRangeBegin(1000).rowRangeEnd(20_000), + precision, scale); + + // Then + BigDecimal min = new BigDecimal(threshold, scale); + assertThat(all).isEqualTo(expected); + assertThat(filtered).isEqualTo(expected.stream().filter(v -> v != null && v.compareTo(min) >= 0).toList()); + assertThat(range).isEqualTo(expected.subList(1000, 20_000)); + } + private static Path toParquet(Path tmp, DType.Extension dtype, List values) throws IOException { Path vortex = tmp.resolve("source.vortex"); DType.Struct schema = new DType.Struct(List.of(TS), List.of(dtype), false); @@ -98,6 +156,71 @@ private static Path toParquet(Path tmp, DType.Extension dtype, List val return parquet; } + private static Path decimalParquet(Path tmp, PhysicalType type, int length, int precision, int scale, + BigInteger[] unscaled, IntPredicate isNull) throws IOException { + FileSchema.Builder builder = FileSchema.builder("schema"); + LogicalType decimal = new LogicalType.DecimalType(precision, scale); + if (length > 0) { + builder.addColumn("d", type, RepetitionType.OPTIONAL, length, decimal); + } else { + builder.addColumn("d", type, RepetitionType.OPTIONAL, decimal); + } + boolean[] nulls = new boolean[unscaled.length]; + for (int i = 0; i < nulls.length; i++) { + nulls[i] = isNull.test(i); + } + 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); + default -> throw new IllegalArgumentException(type.toString()); + } + }); + } + return parquet; + } + + /// `value` as `length` big-endian two's-complement bytes, sign-extended, as Parquet stores a + /// fixed-length DECIMAL. + private static byte[] bigEndian(BigInteger value, int length) { + byte[] minimal = value.toByteArray(); + byte[] out = new byte[length]; + Arrays.fill(out, 0, length - minimal.length, (byte) (value.signum() < 0 ? 0xFF : 0)); + System.arraycopy(minimal, 0, out, length - minimal.length, minimal.length); + return out; + } + + /// Reads column `d`, checking every batch carries Arrow's decimal type at `precision`/`scale`. + private static List readDecimalsWithJni(Path file, ImmutableScanOptions.Builder options, + int precision, int scale) throws IOException { + ScanOptions opts = options.projection(Expression.select(new String[]{"d"}, Expression.root())).build(); + List out = new ArrayList<>(); + DataSource ds = DataSource.open(SESSION, file.toAbsolutePath().toUri().toString()); + Scan scan = ds.scan(opts); + while (scan.hasNext()) { + Partition partition = scan.next(); + try (ArrowReader reader = partition.scanArrow(ALLOCATOR)) { + while (reader.loadNextBatch()) { + VectorSchemaRoot root = reader.getVectorSchemaRoot(); + FieldVector vec = root.getVector("d"); + ArrowType.Decimal arrowType = (ArrowType.Decimal) vec.getField().getType(); + assertThat(arrowType.getPrecision()).isEqualTo(precision); + assertThat(arrowType.getScale()).isEqualTo(scale); + for (int i = 0; i < root.getRowCount(); i++) { + out.add(vec.isNull(i) ? null : (BigDecimal) vec.getObject(i)); + } + } + } + } + return out; + } + private static List readWithJni(Path file, ImmutableScanOptions.Builder options) throws IOException { ScanOptions opts = options.projection(Expression.select(new String[]{"ts"}, Expression.root())).build(); List out = new ArrayList<>(); diff --git a/parquet/src/main/java/io/github/dfa1/vortex/parquet/ColumnBuilder.java b/parquet/src/main/java/io/github/dfa1/vortex/parquet/ColumnBuilder.java index 0a53a11c7..fa8f21ee9 100644 --- a/parquet/src/main/java/io/github/dfa1/vortex/parquet/ColumnBuilder.java +++ b/parquet/src/main/java/io/github/dfa1/vortex/parquet/ColumnBuilder.java @@ -163,6 +163,11 @@ final class LeafColumnBuilder implements ColumnBuilder { throw new UnsupportedOperationException( "nested Extension (timestamp) columns are not supported: " + dtype); } + // a nested DECIMAL used to map to its bare integer and be truncated through Number; until + // the builder carries BigDecimal elements it fails at the import's start instead + if (dtype instanceof DType.Decimal) { + throw new UnsupportedOperationException("nested DECIMAL columns are not supported: " + dtype); + } this.dtype = dtype; } 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 3bb894726..2759a69d9 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 @@ -20,6 +20,7 @@ import io.github.dfa1.vortex.writer.VortexWriter; import java.io.IOException; +import java.math.BigDecimal; import java.net.URI; import java.nio.channels.FileChannel; import java.nio.file.Path; @@ -229,6 +230,12 @@ static DType mapDType(ColumnSchema col) { } private static DType mapLeafDType(PhysicalType type, LogicalType logical, boolean nullable, String name) { + // DECIMAL annotates any of INT32, INT64, FIXED_LEN_BYTE_ARRAY or BYTE_ARRAY (the unscaled + // 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); + } return switch (type) { case BOOLEAN -> new DType.Bool(nullable); case INT32 -> mapInt32(logical, nullable); @@ -337,6 +344,7 @@ private static Object allocateBuffer(DType type, int chunkSize) { case DType.Utf8 _ -> new String[chunkSize]; case DType.Binary _ -> new byte[chunkSize][]; case DType.Extension _ -> nullable ? new Long[chunkSize] : new long[chunkSize]; // timestamp storage + case DType.Decimal _ -> new BigDecimal[chunkSize]; case DType.Primitive p -> switch (p.ptype()) { case I8, U8 -> nullable ? new Byte[chunkSize] : new byte[chunkSize]; case I16, U16 -> nullable ? new Short[chunkSize] : new short[chunkSize]; @@ -387,6 +395,7 @@ static void fillRow(RowReader reader, List names, List types, 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); default -> throw new UnsupportedOperationException( "unexpected buffer type: " + buffers[c].getClass().getSimpleName()); @@ -497,6 +506,7 @@ private static Object trimBuffer(Object buffer, int size) { case byte[] arr -> size == arr.length ? arr : Arrays.copyOf(arr, size); case Byte[] arr -> size == arr.length ? arr : Arrays.copyOf(arr, size); case String[] arr -> size == arr.length ? arr : Arrays.copyOf(arr, size); + case BigDecimal[] arr -> size == arr.length ? arr : Arrays.copyOf(arr, size); case byte[][] arr -> size == arr.length ? arr : Arrays.copyOf(arr, size); default -> throw new UnsupportedOperationException( "unexpected buffer type: " + buffer.getClass().getSimpleName()); diff --git a/parquet/src/test/java/io/github/dfa1/vortex/parquet/ColumnBuilderTest.java b/parquet/src/test/java/io/github/dfa1/vortex/parquet/ColumnBuilderTest.java index 99518f6f6..22d3ef5da 100644 --- a/parquet/src/test/java/io/github/dfa1/vortex/parquet/ColumnBuilderTest.java +++ b/parquet/src/test/java/io/github/dfa1/vortex/parquet/ColumnBuilderTest.java @@ -27,6 +27,17 @@ class ColumnBuilderTest { @Nested class Leaf { + @Test + void nestedDecimal_throwsAtConstruction() { + // Given — a DECIMAL field inside a STRUCT/LIST (#628 maps it to DType.Decimal) + DType.Decimal decimal = DType.Decimal.of(15, 2, true); + + // When / Then + assertThatThrownBy(() -> ColumnBuilder.forType(decimal)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("DECIMAL"); + } + @Test void nullablePrimitive_wrapsInNullableData_withDefaultsAtNullPositions() { // Given — Raincloud-style nullable int leaf; Hardwood decodes INT32 as Integer 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 2701df860..a64078840 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 @@ -7,6 +7,9 @@ import dev.hardwood.metadata.PhysicalType; import dev.hardwood.metadata.RepetitionType; import dev.hardwood.reader.RowReader; +import dev.hardwood.OutputFile; +import dev.hardwood.schema.FileSchema; +import dev.hardwood.writer.ParquetFileWriter; import dev.hardwood.schema.ColumnSchema; import dev.hardwood.schema.SchemaNode; import io.github.dfa1.vortex.core.model.DType; @@ -16,6 +19,8 @@ import io.github.dfa1.vortex.reader.ScanIterator; import io.github.dfa1.vortex.reader.ScanOptions; import io.github.dfa1.vortex.reader.VortexReader; +import io.github.dfa1.vortex.reader.array.Array; +import io.github.dfa1.vortex.reader.array.DecimalArray; import io.github.dfa1.vortex.reader.array.LongArray; import io.github.dfa1.vortex.reader.array.MaskedArray; import io.github.dfa1.vortex.reader.array.VarBinArray; @@ -28,7 +33,11 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import java.math.BigDecimal; +import java.math.BigInteger; import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import static org.assertj.core.api.Assertions.assertThat; @@ -156,10 +165,10 @@ void byteArray_withoutAnnotation_mapsToBinary() { @Test void byteArray_unsupportedAnnotation_stillThrows() { - // Given — a BYTE_ARRAY annotation that is neither string-like nor absent (e.g. - // DECIMAL) must not be silently downgraded to raw Binary, losing its semantic + // Given — a BYTE_ARRAY annotation that is neither string-like, DECIMAL nor absent (e.g. + // BSON) must not be silently downgraded to raw Binary, losing its semantic ColumnSchema schema = col("amount", PhysicalType.BYTE_ARRAY, RepetitionType.REQUIRED, - new LogicalType.DecimalType(10, 2)); + new LogicalType.BsonType()); // When / Then assertThatThrownBy(() -> ParquetImporter.mapDType(schema)) @@ -368,6 +377,98 @@ void requiredColumnActuallyNull_throwsClearErrorInsteadOfRawNpe() { } } + @Nested + class Decimals { + + private static final List VALUES = Arrays.asList( + new BigDecimal("17.00"), null, new BigDecimal("-0.05"), new BigDecimal("1234567.89"), null); + + @ParameterizedTest(name = "{0}") + @CsvSource({ + // physical type, fixed length, precision, scale + "INT32, 0, 9, 2", + "INT64, 0, 18, 2", + "FIXED_LEN_BYTE_ARRAY, 7, 15, 2", // TPC-H lineitem as parquet-cpp writes it (#628) + "BYTE_ARRAY, 0, 30, 2", + }) + void decimalColumn_importsAsDecimalWithItsScale(PhysicalType type, int length, int precision, int scale, + @TempDir Path tmp) throws Exception { + // Given — an OPTIONAL DECIMAL column holding nulls, a negative and a fractional value + Path parquet = writeDecimalParquet(tmp, type, length, precision, scale); + 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, true)); + try (Chunk chunk = iter.next()) { + assertThat(decimals(chunk.column(ColumnName.of("d")), VALUES.size())) + .containsExactlyElementsOf(VALUES); + } + } + } + + private static Path writeDecimalParquet(Path tmp, PhysicalType type, int length, int precision, int scale) + 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); + } else { + builder.addColumn("d", type, RepetitionType.OPTIONAL, decimal); + } + 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(); + } + 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); + default -> throw new IllegalArgumentException(type.toString()); + } + }); + } + return parquet; + } + + /// `value` as `length` big-endian two's-complement bytes, sign-extended, as Parquet stores a + /// fixed-length DECIMAL. + private static byte[] bigEndian(BigInteger value, int length) { + byte[] minimal = value.toByteArray(); + byte[] out = new byte[length]; + Arrays.fill(out, 0, length - minimal.length, (byte) (value.signum() < 0 ? 0xFF : 0)); + System.arraycopy(minimal, 0, out, length - minimal.length, minimal.length); + return out; + } + + private static List decimals(Array column, int n) { + List out = new ArrayList<>(n); + for (int i = 0; i < n; i++) { + if (column instanceof MaskedArray masked) { + out.add(masked.isValid(i) ? ((DecimalArray) masked.inner()).getDecimal(i) : null); + } else { + out.add(((DecimalArray) column).getDecimal(i)); + } + } + return out; + } + } + @Nested class Import {