fix(parquet): import DECIMAL columns as Vortex decimals

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

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<STRUCT<...>>`); `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                              |
 |---------------------------------------------------|-------------------------------------|

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,27 +25,35 @@
 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;
 
 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<BigDecimal> 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<BigDecimal> all = readDecimalsWithJni(imported, ScanOptions.builder(), precision, scale);
+        // a filtered and a row-range read prune and slice, which a full scan never does
+        List<BigDecimal> filtered = readDecimalsWithJni(imported, ScanOptions.builder().filter(Expression.binary(
+                Expression.BinaryOp.GTE, Expression.column("d"),
+                Expression.literalDecimal(threshold, precision, scale))), precision, scale);
+        List<BigDecimal> 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<Instant> 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<Instant> 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<BigDecimal> 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<BigDecimal> 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<Long> readWithJni(Path file, ImmutableScanOptions.Builder options) throws IOException {
         ScanOptions opts = options.projection(Expression.select(new String[]{"ts"}, Expression.root())).build();
         List<Long> out = new ArrayList<>();

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;
     }
 

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<ColumnName> names, List<DType> 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());

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