diff --git a/CHANGELOG.md b/CHANGELOG.md index bdef29f6a..adc7ae727 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 +- 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 column whose encoded bytes pass 2 GiB in one chunk, such as large audio blobs under FSST, writes and reads back instead of failing with an integer overflow. ([#629](https://github.com/dfa1/vortex-java/issues/629)) - 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)) diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/NullableTimestampInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/NullableTimestampInteropIntegrationTest.java new file mode 100644 index 000000000..e11bcf73a --- /dev/null +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/NullableTimestampInteropIntegrationTest.java @@ -0,0 +1,187 @@ +package io.github.dfa1.vortex.integration; + +import dev.vortex.api.Session; +import dev.vortex.api.VortexWriter; +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.WriteOptions; +import org.apache.arrow.c.ArrowArray; +import org.apache.arrow.c.ArrowSchema; +import org.apache.arrow.c.Data; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.vector.TimeStampMicroVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.Schema; +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.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +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.Random; +import java.util.function.IntPredicate; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; + +/// A nullable timestamp column is written as vortex-jni writes it: `vortex.datetimeparts`, with the +/// validity on its days child as Rust's `split_temporal` keeps it, and no validity at all when every +/// row is valid. The Java side used to decline datetimeparts for any chunk holding a null, writing +/// `vortex.ext` over a masked i64, so the column's encoding changed with where its nulls fell. +/// +/// Only the data column's root encoding and where its validity sits are compared: the cascade under +/// the days child and the zone-stats table still differ from Rust with or without nulls (#634). +class NullableTimestampInteropIntegrationTest { + + private static final Session SESSION = Session.create(); + private static final BufferAllocator ALLOCATOR = ArrowAllocation.rootAllocator(); + private static final ColumnName TS = ColumnName.of("ts"); + private static final DType.Extension TS_DTYPE = TimestampDtype.of(TimeUnit.Microseconds, null, true); + private static final int ROWS = 4 * 8192; + private static final long START_MICROS = 1_704_067_200_000_000L; // 2024-01-01T00:00:00Z + private static final long[] JITTER_MICROS = new Random(634).longs(ROWS, 0, 60_000_000L).toArray(); + + static { + NativeLoader.loadJni(); + } + + static Stream shapes() { + return Stream.of( + // nullable but every row valid: written without a validity, as Rust's AllValid + Arguments.of("no nulls", (IntPredicate) _ -> false), + Arguments.of("every 7th null", (IntPredicate) i -> i % 7 == 0), + Arguments.of("one null", (IntPredicate) i -> i == 5), + Arguments.of("nulls in the second half", (IntPredicate) i -> i >= ROWS / 2 && i % 3 == 0)); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("shapes") + void javaWritesTheEncodingTreeJniWrites(String name, IntPredicate isNull, @TempDir Path tmp) throws Exception { + // Given + Path jniFile = tmp.resolve("jni.vortex"); + Path javaFile = tmp.resolve("java.vortex"); + writeWithJni(jniFile, isNull); + List expected = EncodingTrees.of(jniFile); + + // When + writeWithJava(javaFile, isNull); + + // Then + assertThat(dataColumn(expected).getFirst()).contains("vortex.datetimeparts"); + assertDataColumnMatches(EncodingTrees.of(javaFile), expected); + } + + /// The Parquet import path hands the writer batches with and without nulls, which its + /// repartition coalesces into one chunk; that chunk is datetimeparts too. + @Test + void parquetImport_writesTheEncodingTreeJniWrites(@TempDir Path tmp) throws Exception { + // Given + IntPredicate isNull = i -> i >= ROWS / 2 && i % 3 == 0; + Path jniFile = tmp.resolve("jni.vortex"); + writeWithJni(jniFile, isNull); + Path source = tmp.resolve("source.vortex"); + writeWithJava(source, isNull); + Path parquet = tmp.resolve("source.parquet"); + ParquetExporter.exportParquet(source, parquet); + Path imported = tmp.resolve("imported.vortex"); + + // When + ParquetImporter.importParquet(parquet, imported, ImportOptions.defaults().withChunkSize(4096)); + + // Then + assertDataColumnMatches(EncodingTrees.of(imported), EncodingTrees.of(jniFile)); + } + + /// Same root encoding for the data column, and a validity (`vortex.bool`) in its days child + /// exactly when Rust writes one. + private static void assertDataColumnMatches(List actual, List expected) { + List actualColumn = dataColumn(actual); + List expectedColumn = dataColumn(expected); + assertThat(actualColumn.getFirst()).isEqualTo(expectedColumn.getFirst()); + assertThat(firstChild(actualColumn).stream().anyMatch(l -> l.contains("vortex.bool"))) + .isEqualTo(firstChild(expectedColumn).stream().anyMatch(l -> l.contains("vortex.bool"))); + } + + /// The data column's encoding subtree: the lines under the zoned layout's first flat, up to the + /// stats table's flat. + private static List dataColumn(List tree) { + int start = tree.indexOf(" vortex.flat") + 1; + int end = tree.subList(start, tree.size()).indexOf(" vortex.flat"); + return tree.subList(start, end < 0 ? tree.size() : start + end); + } + + /// The first child (days) subtree of a data column subtree. + private static List firstChild(List column) { + int depth = indent(column.getFirst()) + 2; + int end = 2; + while (end < column.size() && indent(column.get(end)) > depth) { + end++; + } + return column.subList(1, end); + } + + private static int indent(String line) { + return line.length() - line.stripLeading().length(); + } + + /// One row about every 90 s with a jitter of up to a minute: not an arithmetic sequence, which + /// Rust would write as `vortex.sequence` instead of splitting it (#634). + private static long micros(int i) { + return START_MICROS + i * 90_000_000L + JITTER_MICROS[i]; + } + + private static void writeWithJni(Path path, IntPredicate isNull) throws Exception { + Schema schema = new Schema(List.of(Field.nullable("ts", + new ArrowType.Timestamp(org.apache.arrow.vector.types.TimeUnit.MICROSECOND, null)))); + try (VortexWriter writer = VortexWriter.builder(SESSION, path.toAbsolutePath().toUri().toString(), schema, + ALLOCATOR).build(); + VectorSchemaRoot root = VectorSchemaRoot.create(schema, ALLOCATOR)) { + TimeStampMicroVector vector = (TimeStampMicroVector) root.getVector("ts"); + vector.setInitialCapacity(ROWS); + vector.allocateNew(); + for (int i = 0; i < ROWS; i++) { + if (isNull.test(i)) { + vector.setNull(i); + } else { + vector.setSafe(i, micros(i)); + } + } + vector.setValueCount(ROWS); + root.setRowCount(ROWS); + try (ArrowArray array = ArrowArray.allocateNew(ALLOCATOR); + ArrowSchema arrowSchema = ArrowSchema.allocateNew(ALLOCATOR)) { + Data.exportVectorSchemaRoot(ALLOCATOR, root, null, array, arrowSchema); + writer.writeBatch(array.memoryAddress(), arrowSchema.memoryAddress()); + } + } + } + + private static void writeWithJava(Path path, IntPredicate isNull) throws IOException { + List values = new ArrayList<>(ROWS); + for (int i = 0; i < ROWS; i++) { + values.add(isNull.test(i) ? null : Instant.EPOCH.plusNanos(micros(i) * 1_000L)); + } + DType.Struct schema = new DType.Struct(List.of(TS), List.of(TS_DTYPE), false); + try (FileChannel channel = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + var writer = io.github.dfa1.vortex.writer.VortexWriter.create(channel, schema, WriteOptions.defaults())) { + writer.writeChunk(Map.of(TS, values)); + } + } +} diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/TimestampFilterInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/TimestampFilterInteropIntegrationTest.java index 26abcf0b3..4a29d7f08 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/TimestampFilterInteropIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/TimestampFilterInteropIntegrationTest.java @@ -93,9 +93,8 @@ static void writeFiles() throws IOException { } } } - // As the Parquet importer writes a timestamp column: a nullable column handed over as - // DateTimePartsData, the only input the writer stores as datetimeparts (which has no null - // rows on this path; the unit tests cover nulls). + // As the Parquet importer writes a timestamp column without nulls: a nullable column + // handed over as DateTimePartsData, which the writer stores as datetimeparts. partsFile = write("timestamp-parts.vtx", WriteOptions.defaults(), ids, b -> new DateTimePartsData(ts[b], true)); // The same values as i64 storage under the extension: the default cascade would pick @@ -154,12 +153,21 @@ static Stream cases() { void timestampColumnsAreStoredAsPartsAndAsFlatLongs() throws IOException { // Given the files written above // When - try (VortexReader parts = VortexReader.open(partsFile); - VortexReader flat = VortexReader.open(flatFile)) { - // Then the filters run over both storages the fast path reads - assertThat(parts.footer().arraySpecs()).contains(EncodingId.VORTEX_DATETIMEPARTS); - assertThat(flat.footer().arraySpecs()).doesNotContain(EncodingId.VORTEX_DATETIMEPARTS); - } + String partsData = tsDataEncoding(partsFile); + String flatData = tsDataEncoding(flatFile); + + // Then the filters run over both storages the fast path reads. Only the data column is + // checked: the zone-stats table of a nullable timestamp is datetimeparts in both files. + assertThat(partsData).isEqualTo(EncodingId.VORTEX_DATETIMEPARTS.id()); + assertThat(flatData).isEqualTo(EncodingId.VORTEX_EXT.id()); + } + + /// The root encoding of the ts column's data: the array under the first flat of its zoned layout. + private static String tsDataEncoding(Path file) throws IOException { + List tree = EncodingTrees.of(file); + int tsZoned = tree.lastIndexOf(" vortex.zoned"); + assertThat(tsZoned).as("the ts column's zoned layout in %s", tree).isPositive(); + return tree.get(tree.subList(tsZoned, tree.size()).indexOf(" vortex.flat") + tsZoned + 1).strip(); } @ParameterizedTest(name = "{0}: {1}") diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.java b/writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.java index 0f6aec60b..9e232afb9 100644 --- a/writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.java +++ b/writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.java @@ -357,11 +357,11 @@ private static WriteRegistry buildRegistry(List encoders) { private static List buildCascadeCodecs(WriteOptions options) { List codecs = new ArrayList<>(); // FbsExtension-dtype dispatch order matters: findPrimitiveEncoding picks the first - // accepting codec. DateTimePartsEncoding goes first because it consumes - // pre-decomposed DateTimePartsData (Parquet importer path); when the data is - // raw primitive storage (JDBC's long[] via TimestampExtension.encodeAll) it - // returns notApplicable and spliceResult excludes it, falling back to - // ExtEncoding which cascades the storage child through FoR/Bitpacked/RLE/ALP. + // accepting codec. DateTimePartsEncoding goes first: it splits any vortex.timestamp + // storage (DateTimePartsData, long[], or NullableData over long[] with its validity on + // the days child); for other extensions, Days units or an all-null chunk it returns + // notApplicable and spliceResult excludes it, falling back to ExtEncoding, which + // cascades the storage child through FoR/Bitpacked/RLE/ALP. // FixedSizeListEncoding handles UUID-style fixed-size byte storage downstream // of ExtEncoding. codecs.add(new DateTimePartsEncodingEncoder()); diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/CascadingCompressor.java b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/CascadingCompressor.java index 000e04797..6b25ef835 100644 --- a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/CascadingCompressor.java +++ b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/CascadingCompressor.java @@ -261,8 +261,7 @@ private EncodeResult encodeWithCtx(DType dtype, Object data, EncodeContext ctx) // splice through it so its cascaded children (e.g. datetimeparts → days/seconds/ // subseconds) are recursively compressed rather than stored as raw primitives. Honor the // excluded set so spliceResult's notApplicable retry can rotate to the next accepting - // encoding (e.g. DateTimePartsEncoding → ExtEncoding when the input is raw storage rather - // than DateTimePartsData). + // encoding (e.g. DateTimePartsEncoding → ExtEncoding for a date, or an all-null chunk). if (!(dtype instanceof DType.Primitive)) { return spliceResult(findPrimitiveEncoding(dtype, ctx.excluded()), dtype, data, ctx); } diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/DateTimePartsEncodingEncoder.java b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/DateTimePartsEncodingEncoder.java index 6c6af71a4..79e819483 100644 --- a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/DateTimePartsEncodingEncoder.java +++ b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/DateTimePartsEncodingEncoder.java @@ -1,5 +1,6 @@ package io.github.dfa1.vortex.writer.encode; +import io.github.dfa1.vortex.core.compute.PrimitiveArrays; import io.github.dfa1.vortex.core.model.DType; import io.github.dfa1.vortex.core.model.DTypes; import io.github.dfa1.vortex.core.model.PType; @@ -43,20 +44,20 @@ public boolean accepts(DType dtype) { @Override public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) { DType.Extension ext = (DType.Extension) dtype; - DateTimePartsData d = asParts(dtype, data); - if (d == null) { + Input in = asParts(dtype, data); + if (in == null) { throw new VortexException(EncodingId.VORTEX_DATETIMEPARTS, - "expected timestamp storage, got " + (data == null ? "null" : data.getClass().getSimpleName())); + "expected non-Days vortex.timestamp storage with a valid row, got " + + (data == null ? "null" : data.getClass().getSimpleName())); } - Parts parts = split(d.timestamps(), unitOf(ext)); - long[] days = parts.days(); + Parts parts = split(in.timestamps(), unitOf(ext)); long[] seconds = parts.seconds(); long[] subseconds = parts.subseconds(); - DType daysDtype = d.nullable() ? DTypes.I64.asNullable() : DTypes.I64; - EncodingEncoder primEnc = ctx.lookupEncoder(EncodingId.VORTEX_PRIMITIVE); - EncodeResult daysResult = primEnc.encode(daysDtype, days, ctx); + EncodeResult daysResult = in.validity() == null + ? primEnc.encode(in.daysDtype(), parts.days(), ctx) + : new MaskedEncodingEncoder().encode(in.daysDtype(), in.days(parts), ctx); EncodeResult secondsResult = primEnc.encode(DTypes.I64, seconds, ctx); EncodeResult subsecondsResult = primEnc.encode(DTypes.I64, subseconds, ctx); @@ -82,20 +83,19 @@ public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) { // Bounds come from the undivided timestamps, in storage space: the split children are // days/seconds/subseconds and none of them orders the column. Without this, choosing // datetimeparts over vortex.ext would trade the column's zone map for its size. - MinMax stats = PrimitiveEncodingEncoder.minMaxStats(PType.I64, d.timestamps()); + MinMax stats = in.stats(); return new EncodeResult(root, List.copyOf(allBuffers), PrimitiveEncodingEncoder.minOf(stats), PrimitiveEncodingEncoder.maxOf(stats)); } @Override public CascadeStep encodeCascade(DType dtype, Object data, EncodeContext encodeCtx) { - DateTimePartsData d = asParts(dtype, data); - if (d == null) { + Input in = asParts(dtype, data); + if (in == null) { return CascadeStep.notApplicable(); } DType.Extension ext = (DType.Extension) dtype; - Parts parts = split(d.timestamps(), unitOf(ext)); - long[] days = parts.days(); + Parts parts = split(in.timestamps(), unitOf(ext)); long[] seconds = parts.seconds(); long[] subseconds = parts.subseconds(); @@ -107,13 +107,12 @@ public CascadeStep encodeCascade(DType dtype, Object data, EncodeContext encodeC new EncodeNode[3], new int[0]); - DType daysDtype = d.nullable() ? DTypes.I64.asNullable() : DTypes.I64; List children = List.of( - new ChildSlot(daysDtype, days, 0, SELF), + new ChildSlot(in.daysDtype(), in.days(parts), 0, SELF), new ChildSlot(DTypes.I64, seconds, 1, SELF), new ChildSlot(DTypes.I64, subseconds, 2, SELF)); - MinMax stats = PrimitiveEncodingEncoder.minMaxStats(PType.I64, d.timestamps()); + MinMax stats = in.stats(); return new CascadeStep(partialRoot, List.of(), children, PrimitiveEncodingEncoder.minOf(stats), PrimitiveEncodingEncoder.maxOf(stats), true); } @@ -129,23 +128,46 @@ public CascadeStep encodeCascade(DType dtype, Object data, EncodeContext encodeC /// Restricted to `vortex.timestamp`: a date is already whole days, a time-of-day has no day /// component worth splitting out, and [TimeUnit#Days] cannot be subdivided. /// - /// A nullable column arrives as [NullableData] and is declined here, leaving it on the - /// existing `vortex.ext` path rather than silently dropping its validity. + /// A nullable column with nulls arrives as [NullableData] over `long[]` storage. Its validity + /// goes on the days child, as Rust's `split_temporal` keeps it there (seconds and subseconds + /// stay non-nullable); an all-valid validity is dropped. An all-null chunk is declined: Rust + /// checks for all-null before the extension's schemes, so the cascade writes it as the null + /// constant through `vortex.ext`. The days child takes the column's nullability, as Rust's + /// does, whichever shape the storage arrives in. /// /// @param dtype the column dtype /// @param data the values being encoded /// @return the timestamps to split, or `null` when this encoder does not apply - private static DateTimePartsData asParts(DType dtype, Object data) { + private static Input asParts(DType dtype, Object data) { if (data instanceof DateTimePartsData parts) { - return parts; + return new Input(parts.timestamps(), dtype.nullable(), null); } - if (!(data instanceof long[] storage) || !(dtype instanceof DType.Extension ext) + if (!(dtype instanceof DType.Extension ext) || ext.extensionId() != ExtensionId.VORTEX_TIMESTAMP || !(ext.metadata() instanceof TimestampMetadata(TimeUnit unit, var _)) || unit == TimeUnit.Days) { return null; } - return new DateTimePartsData(storage, false); + return switch (data) { + case long[] storage -> new Input(storage, ext.nullable(), null); + case NullableData(long[] storage, boolean[] validity) -> nullableInput(storage, validity); + default -> null; + }; + } + + /// An all-valid validity is dropped, as Rust's `AllValid` writes no bitmap; an all-null one + /// declines the chunk. + private static Input nullableInput(long[] storage, boolean[] validity) { + int valid = 0; + for (boolean v : validity) { + if (v) { + valid++; + } + } + if (valid == 0) { + return null; + } + return new Input(storage, true, valid == validity.length ? null : validity); } private static TimeUnit unitOf(DType.Extension ext) { @@ -188,6 +210,27 @@ private static Parts split(long[] timestamps, TimeUnit unit) { return new Parts(days, seconds, subseconds); } + /// The timestamps to split, whether the column is nullable, and the validity when the chunk + /// holds nulls (`null` when every row is valid). + @SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared. + private record Input(long[] timestamps, boolean nullable, boolean[] validity) { + + DType daysDtype() { + return nullable ? DTypes.I64.asNullable() : DTypes.I64; + } + + /// The days child: the plain part, or the part under this chunk's validity. + Object days(Parts parts) { + return validity == null ? parts.days() : new NullableData(parts.days(), validity); + } + + /// Bounds over the valid rows only: an invalid slot's placeholder is not a value. + MinMax stats() { + Object values = validity == null ? timestamps : PrimitiveArrays.compact(PType.I64, timestamps, validity); + return PrimitiveEncodingEncoder.minMaxStats(PType.I64, values); + } + } + /// One timestamp column split into its three `vortex.datetimeparts` children. @SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared. private record Parts(long[] days, long[] seconds, long[] subseconds) { diff --git a/writer/src/test/java/io/github/dfa1/vortex/writer/encode/DateTimePartsEncodingEncoderTest.java b/writer/src/test/java/io/github/dfa1/vortex/writer/encode/DateTimePartsEncodingEncoderTest.java index 57e6192ef..06b369e09 100644 --- a/writer/src/test/java/io/github/dfa1/vortex/writer/encode/DateTimePartsEncodingEncoderTest.java +++ b/writer/src/test/java/io/github/dfa1/vortex/writer/encode/DateTimePartsEncodingEncoderTest.java @@ -12,6 +12,8 @@ import io.github.dfa1.vortex.reader.decode.TestRegistry; import io.github.dfa1.vortex.core.model.TimeUnit; import io.github.dfa1.vortex.core.proto.ProtoDateTimePartsMetadata; +import io.github.dfa1.vortex.core.proto.ProtoScalarValue; +import io.github.dfa1.vortex.core.model.TimestampMetadata; import io.github.dfa1.vortex.reader.decode.DateTimePartsEncodingDecoder; import io.github.dfa1.vortex.reader.decode.PrimitiveEncodingDecoder; import org.junit.jupiter.api.Nested; @@ -36,7 +38,7 @@ class DateTimePartsEncodingEncoderTest { private static DType timestampDType(TimeUnit unit) { return new DType.Extension(ExtensionId.VORTEX_TIMESTAMP, - DTypes.I64, new io.github.dfa1.vortex.core.model.TimestampMetadata(unit, null), false); + DTypes.I64, new TimestampMetadata(unit, null), false); } private static ArrayNode toArrayNode(EncodeNode node) { @@ -98,6 +100,102 @@ void encode_missingMetadata_throws() { } } + /// A nullable timestamp's validity goes on the days child, as Rust's `split_temporal` keeps it. + @Nested + class Nullable { + + private static final DType NULLABLE_MS = new DType.Extension(ExtensionId.VORTEX_TIMESTAMP, + DTypes.I64.asNullable(), new TimestampMetadata(TimeUnit.Milliseconds, null), + true); + + @Test + void encodeCascade_withNulls_putsValidityOnDaysChildOnly() { + // Given + boolean[] validity = {true, false, true}; + NullableData data = new NullableData(new long[]{86_400_000L, 0L, 2 * 86_400_000L}, validity); + + // When + CascadeStep result = ENCODER.encodeCascade(NULLABLE_MS, data, EncodeTestHelper.testCtx()); + + // Then + assertThat(result.applicable()).isTrue(); + assertThat(result.openChildren().get(0).childDtype()).isEqualTo(DTypes.I64.asNullable()); + assertThat(result.openChildren().get(0).childData()).isInstanceOfSatisfying(NullableData.class, days -> { + assertThat((long[]) days.values()).containsExactly(1L, 0L, 2L); + assertThat(days.validity()).containsExactly(validity); + }); + assertThat(result.openChildren().get(1).childData()).isInstanceOf(long[].class); + assertThat(result.openChildren().get(2).childData()).isInstanceOf(long[].class); + } + + @Test + void encodeCascade_withNulls_takesBoundsOverValidRowsOnly() { + // Given — the invalid slot's 0 placeholder is below every valid value + NullableData data = new NullableData(new long[]{5_000L, 0L, 9_000L}, new boolean[]{true, false, true}); + + // When + CascadeStep result = ENCODER.encodeCascade(NULLABLE_MS, data, EncodeTestHelper.testCtx()); + + // Then + assertThat(result.statsMin()).isEqualTo(ProtoScalarValue.ofInt64Value(5_000L)); + assertThat(result.statsMax()).isEqualTo(ProtoScalarValue.ofInt64Value(9_000L)); + } + + @Test + void encodeCascade_allValid_writesNoValidity() { + // Given — Rust's AllValid carries no bitmap + NullableData data = new NullableData(new long[]{1L, 2L}, new boolean[]{true, true}); + + // When + CascadeStep result = ENCODER.encodeCascade(NULLABLE_MS, data, EncodeTestHelper.testCtx()); + + // Then + assertThat(result.openChildren().get(0).childDtype()).isEqualTo(DTypes.I64.asNullable()); + assertThat(result.openChildren().get(0).childData()).isInstanceOf(long[].class); + } + + @Test + void encodeCascade_allNull_isNotApplicable() { + // Given — vortex.ext writes an all-null chunk as the null constant + NullableData data = new NullableData(new long[]{0L, 0L}, new boolean[]{false, false}); + + // When + CascadeStep result = ENCODER.encodeCascade(NULLABLE_MS, data, EncodeTestHelper.testCtx()); + + // Then + assertThat(result.applicable()).isFalse(); + } + + @Test + void encode_withNulls_keepsValidityUnderDaysChild() { + // Given + NullableData data = new NullableData(new long[]{86_400_000L, 0L}, new boolean[]{true, false}); + + // When + EncodeResult result = ENCODER.encode(NULLABLE_MS, data, EncodeTestHelper.testCtx()); + + // Then — a bool validity under days, none under seconds or subseconds + EncodeNode[] children = result.rootNode().children(); + assertThat(result.rootNode().encodingId()).isEqualTo(EncodingId.VORTEX_DATETIMEPARTS); + assertThat(children).hasSize(3); + assertThat(contains(children[0], EncodingId.VORTEX_BOOL)).isTrue(); + assertThat(contains(children[1], EncodingId.VORTEX_BOOL)).isFalse(); + assertThat(contains(children[2], EncodingId.VORTEX_BOOL)).isFalse(); + } + + private static boolean contains(EncodeNode node, EncodingId id) { + if (node.encodingId() == id) { + return true; + } + for (EncodeNode child : node.children()) { + if (contains(child, id)) { + return true; + } + } + return false; + } + } + @Nested class Decode {