fix(parquet): import a nullable TIMESTAMP column with nulls

CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to **vortex-java** are documented here.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+- 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
 
 ### Performance

integration/src/test/java/io/github/dfa1/vortex/integration/ParquetImportInteropIntegrationTest.java

@@ -0,0 +1,120 @@
+package io.github.dfa1.vortex.integration;
+
+import dev.vortex.api.DataSource;
+import dev.vortex.api.Expression;
+import dev.vortex.api.ImmutableScanOptions;
+import dev.vortex.api.Partition;
+import dev.vortex.api.Scan;
+import dev.vortex.api.ScanOptions;
+import dev.vortex.api.Session;
+import dev.vortex.arrow.ArrowAllocation;
+import dev.vortex.jni.NativeLoader;
+import io.github.dfa1.vortex.core.model.ColumnName;
+import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.core.model.TimeUnit;
+import io.github.dfa1.vortex.core.model.TimestampDtype;
+import io.github.dfa1.vortex.parquet.ImportOptions;
+import io.github.dfa1.vortex.parquet.ParquetExporter;
+import io.github.dfa1.vortex.parquet.ParquetImporter;
+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.TimeStampVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+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.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.
+class ParquetImportInteropIntegrationTest {
+
+    private static final Session SESSION = Session.create();
+    private static final BufferAllocator ALLOCATOR = ArrowAllocation.rootAllocator();
+    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;
+    private static final long START_MICROS = 1_704_067_200_000_000L; // 2024-01-01T00:00:00Z
+
+    static {
+        NativeLoader.loadJni();
+    }
+
+    /// An OPTIONAL `TIMESTAMP` column with nulls, as in Raincloud's `bi-telco` (#627: the import
+    /// failed as "declared REQUIRED"). Nulls only in the second half and small import chunks, so the
+    /// writer's repartition coalesces a chunk without nulls followed by one with nulls.
+    @Test
+    void nullableTimestampWithNulls_jniReadsImportedFile(@TempDir Path tmp) throws Exception {
+        // Given
+        IntPredicate isNull = i -> i % 7 == 0 && i >= ROWS / 2;
+        List<Long> expected = new ArrayList<>(ROWS);
+        List<Instant> instants = new ArrayList<>(ROWS);
+        for (int i = 0; i < ROWS; i++) {
+            long micros = START_MICROS + i * 90_000_000L;
+            expected.add(isNull.test(i) ? null : micros);
+            instants.add(isNull.test(i) ? null : Instant.EPOCH.plusNanos(micros * 1_000L));
+        }
+        Path parquet = toParquet(tmp, TimestampDtype.of(TimeUnit.Microseconds, null, true), instants);
+        Path imported = tmp.resolve("imported.vortex");
+        long threshold = expected.get(ROWS / 2 + 1);
+
+        // When
+        ParquetImporter.importParquet(parquet, imported, ImportOptions.defaults().withChunkSize(4096));
+        List<Long> all = readWithJni(imported, ScanOptions.builder());
+        // a filtered and a row-range read prune and slice, which a full scan never does
+        List<Long> filtered = readWithJni(imported, ScanOptions.builder().filter(Expression.binary(
+                Expression.BinaryOp.GTE, Expression.column("ts"),
+                Expression.literalTimestamp(threshold, Expression.TimeUnit.MICROSECONDS, null))));
+        List<Long> range = readWithJni(imported, ScanOptions.builder().rowRangeBegin(1000).rowRangeEnd(20_000));
+
+        // Then
+        assertThat(all).isEqualTo(expected);
+        assertThat(filtered).isEqualTo(expected.stream().filter(v -> v != null && v >= threshold).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);
+        try (FileChannel channel = FileChannel.open(vortex, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+             VortexWriter writer = VortexWriter.create(channel, schema, WriteOptions.defaults())) {
+            writer.writeChunk(Map.of(TS, values));
+        }
+        Path parquet = tmp.resolve("source.parquet");
+        ParquetExporter.exportParquet(vortex, parquet);
+        return parquet;
+    }
+
+    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<>();
+        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();
+                    TimeStampVector vec = (TimeStampVector) root.getVector("ts");
+                    for (int i = 0; i < root.getRowCount(); i++) {
+                        out.add(vec.isNull(i) ? null : vec.get(i));
+                    }
+                }
+            }
+        }
+        return out;
+    }
+}

parquet/src/main/java/io/github/dfa1/vortex/parquet/ParquetImporter.java

@@ -14,6 +14,7 @@
 import io.github.dfa1.vortex.core.model.DType;
 import io.github.dfa1.vortex.core.model.PType;
 import io.github.dfa1.vortex.writer.encode.DateTimePartsData;
+import io.github.dfa1.vortex.writer.encode.NullableData;
 import io.github.dfa1.vortex.core.model.TimeUnit;
 import io.github.dfa1.vortex.core.model.TimestampDtype;
 import io.github.dfa1.vortex.writer.VortexWriter;
@@ -335,7 +336,7 @@ private static Object allocateBuffer(DType type, int chunkSize) {
             case DType.Bool _ -> nullable ? new Boolean[chunkSize] : new boolean[chunkSize];
             case DType.Utf8 _ -> new String[chunkSize];
             case DType.Binary _ -> new byte[chunkSize][];
-            case DType.Extension _ -> new long[chunkSize]; // timestamp storage
+            case DType.Extension _ -> nullable ? new Long[chunkSize] : new long[chunkSize]; // timestamp storage
             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];
@@ -424,15 +425,37 @@ private static Map<ColumnName, Object> buildChunk(List<ColumnName> names, List<D
             } else {
                 value = trimBuffer(buffers[c], size);
                 if (types.get(c) instanceof DType.Extension) {
-                    boolean nullable = types.get(c).nullable();
-                    value = new DateTimePartsData((long[]) value, nullable);
+                    value = timestampChunk(value);
                 }
             }
             chunk.put(names.get(c), value);
         }
         return chunk;
     }
 
+    /// Wraps a chunk of timestamp storage for the writer. A nullable column's boxed `Long[]` with
+    /// nulls becomes [NullableData] (a zero placeholder under each invalid slot), as
+    /// `TimestampExtensionEncoder` builds for an `Instant` collection with nulls; without nulls it
+    /// stays pre-decomposed [DateTimePartsData], as a non-nullable column's `long[]` always does.
+    private static Object timestampChunk(Object buffer) {
+        if (buffer instanceof long[] values) {
+            return new DateTimePartsData(values, false);
+        }
+        Long[] boxed = (Long[]) buffer;
+        long[] values = new long[boxed.length];
+        boolean[] validity = new boolean[boxed.length];
+        boolean anyNull = false;
+        for (int i = 0; i < boxed.length; i++) {
+            if (boxed[i] == null) {
+                anyNull = true;
+            } else {
+                values[i] = boxed[i];
+                validity[i] = true;
+            }
+        }
+        return anyNull ? new NullableData(values, validity) : new DateTimePartsData(values, true);
+    }
+
     /// Filters the top-level schema nodes to those named in `names`, in the requested order.
     ///
     /// @param all   every top-level schema node, in file order

parquet/src/test/java/io/github/dfa1/vortex/parquet/ParquetExporterTest.java

@@ -41,6 +41,7 @@
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.time.Instant;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 
@@ -517,5 +518,35 @@ void timestampColumn_roundTripThroughParquetAndBack(@TempDir Path tmp) throws Ex
                 }
             }
         }
+
+        @Test
+        void nullableTimestampColumn_withNulls_roundTripThroughParquetAndBack(@TempDir Path tmp) throws Exception {
+            // Given — an OPTIONAL Parquet TIMESTAMP holding nulls, first row included (#627: the
+            // import read a null into a primitive long[] and failed as "declared REQUIRED")
+            DType.Extension tsDtype = TimestampDtype.of(TimeUnit.Microseconds, null, true);
+            List<Instant> instants = Arrays.asList(
+                    null,
+                    Instant.ofEpochMilli(-1_500L),
+                    null,
+                    Instant.ofEpochMilli(1_733_000_000_000L));
+            Path original = writeVortex(tmp, "original.vortex",
+                    List.of(ColumnName.of("events")), List.of(tsDtype),
+                    Map.of(ColumnName.of("events"), instants));
+            Path parquet = tmp.resolve("out.parquet");
+            Path reimported = tmp.resolve("reimported.vortex");
+
+            // When
+            ParquetExporter.exportParquet(original, parquet);
+            ParquetImporter.importParquet(parquet, reimported);
+
+            // Then
+            try (VortexReader reader = VortexReader.open(reimported);
+                 ScanIterator iter = reader.scan(ScanOptions.all())) {
+                assertThat(iter.hasNext()).isTrue();
+                try (Chunk chunk = iter.next()) {
+                    assertThat(chunk.as(ColumnName.of("events"), Instant.class)).containsExactlyElementsOf(instants);
+                }
+            }
+        }
     }
 }

writer/src/main/java/io/github/dfa1/vortex/writer/Repartitioner.java

@@ -293,9 +293,13 @@ private static Object rawValues(Object data) {
     }
 
     /// Rebuilds `like`'s carrier shape around new values (and validity, for a nullable carrier).
+    /// A validity always wraps the plain values: no encoder takes [DateTimePartsData] inside
+    /// [NullableData], so a parts piece coalesced with a nullable one (#627) becomes nullable
+    /// plain storage, the shape a nullable `Instant` column takes.
     private static Object rewrap(Object like, Object values, boolean[] validity) {
-        Object inner = like instanceof NullableData nd ? nd.values() : like;
-        Object wrapped = inner instanceof DateTimePartsData dtp ? new DateTimePartsData((long[]) values, dtp.nullable()) : values;
-        return validity == null ? wrapped : new NullableData(wrapped, validity);
+        if (validity != null) {
+            return new NullableData(values, validity);
+        }
+        return like instanceof DateTimePartsData dtp ? new DateTimePartsData((long[]) values, dtp.nullable()) : values;
     }
 }

writer/src/test/java/io/github/dfa1/vortex/writer/RepartitionerTest.java

@@ -3,6 +3,9 @@
 import io.github.dfa1.vortex.core.model.DType;
 import io.github.dfa1.vortex.core.model.DTypes;
 import io.github.dfa1.vortex.core.model.PType;
+import io.github.dfa1.vortex.core.model.TimeUnit;
+import io.github.dfa1.vortex.core.model.TimestampDtype;
+import io.github.dfa1.vortex.writer.encode.DateTimePartsData;
 import io.github.dfa1.vortex.writer.encode.NullableData;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -91,6 +94,25 @@ void nullableBatches_keepValidityAlignedWithValues() {
         });
     }
 
+    @Test
+    void dateTimePartsBatchMixedWithNullableBatch_coalescesToNullablePlainStorage() {
+        // Given — a nullable timestamp column whose import sends a batch without nulls as
+        // pre-decomposed DateTimePartsData and one with nulls as NullableData (#627). Wrapping the
+        // parts carrier inside NullableData gave encoders a shape none of them accepts.
+        var sut = new Repartitioner(TimestampDtype.of(TimeUnit.Microseconds, null, true));
+        sut.add(new DateTimePartsData(new long[]{1, 2}, true), 2);
+        sut.add(new NullableData(new long[]{0, 4}, new boolean[]{false, true}), 2);
+
+        // When
+        Object result = sut.finish().getFirst();
+
+        // Then
+        assertThat(result).isInstanceOfSatisfying(NullableData.class, nd -> {
+            assertThat((long[]) nd.values()).containsExactly(1, 2, 0, 4);
+            assertThat(nd.validity()).containsExactly(true, true, false, true);
+        });
+    }
+
     // Each batch's first chunk is the fewest 8192-row blocks reaching 1 MiB by Rust 0.89's exact size
     // (vortex-data/vortex#10445). The inlined, all-valid and all-null cases cut elsewhere under the
     // rule before it, which counted every string byte and a bitmap on every nullable block.