From 318f91ed7b42426205bcc57d0a0e181eb38f3a61 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sun, 1 Nov 2026 18:48:36 +0100 Subject: [PATCH] fix(parquet): import a nullable TIMESTAMP column with nulls ParquetImporter buffered every timestamp column in a primitive long[], so a null cell in an OPTIONAL TIMESTAMP threw a NullPointerException, reported as "declared REQUIRED". A nullable timestamp column now buffers in a Long[] and a chunk with nulls goes to the writer as NullableData. That exposed a writer bug: Repartitioner, coalescing a DateTimePartsData piece with a NullableData one, wrapped the parts carrier inside NullableData, a shape no encoder accepts. A validity now always wraps the plain storage. Found by the Raincloud corpus: bi-telco and bi-trainsuk2 now import, and their CSV exports are byte-identical to the vortex-jni-written files'. Closes #627 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 5 + .../ParquetImportInteropIntegrationTest.java | 120 ++++++++++++++++++ .../dfa1/vortex/parquet/ParquetImporter.java | 29 ++++- .../vortex/parquet/ParquetExporterTest.java | 31 +++++ .../dfa1/vortex/writer/Repartitioner.java | 10 +- .../dfa1/vortex/writer/RepartitionerTest.java | 22 ++++ 6 files changed, 211 insertions(+), 6 deletions(-) create mode 100644 integration/src/test/java/io/github/dfa1/vortex/integration/ParquetImportInteropIntegrationTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 6cae2faf9..af9224f26 100644 --- a/CHANGELOG.md +++ b/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 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 new file mode 100644 index 000000000..0fe3d82dc --- /dev/null +++ b/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 expected = new ArrayList<>(ROWS); + List 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 all = readWithJni(imported, ScanOptions.builder()); + // a filtered and a row-range read prune and slice, which a full scan never does + List filtered = readWithJni(imported, ScanOptions.builder().filter(Expression.binary( + Expression.BinaryOp.GTE, Expression.column("ts"), + Expression.literalTimestamp(threshold, Expression.TimeUnit.MICROSECONDS, null)))); + List 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 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 readWithJni(Path file, ImmutableScanOptions.Builder options) throws IOException { + ScanOptions opts = options.projection(Expression.select(new String[]{"ts"}, 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(); + 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; + } +} 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 cea5347a1..3bb894726 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 @@ -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,8 +425,7 @@ private static Map buildChunk(List names, List buildChunk(List names, List 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); + } + } + } } } diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/Repartitioner.java b/writer/src/main/java/io/github/dfa1/vortex/writer/Repartitioner.java index 2a229f214..3fd30d371 100644 --- a/writer/src/main/java/io/github/dfa1/vortex/writer/Repartitioner.java +++ b/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; } } diff --git a/writer/src/test/java/io/github/dfa1/vortex/writer/RepartitionerTest.java b/writer/src/test/java/io/github/dfa1/vortex/writer/RepartitionerTest.java index 5196c04e7..1f85d3397 100644 --- a/writer/src/test/java/io/github/dfa1/vortex/writer/RepartitionerTest.java +++ b/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.