From 2a6dd4ace42fc10de5ce3bc4835d688ef9dbf416 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Mon, 6 Jul 2026 21:46:29 +0200 Subject: [PATCH 1/3] fix: decode fastlanes.rle over F64/F32 (#209) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RleEncodingDecoder threw "fastlanes.rle: unsupported ptype F64" because its ptype switch only had integer arms. The Python Vortex writer (vortex-data 0.69.0) RLE-encodes double columns with long constant runs, so such files could not be read. Per the Rust reference (encodings/fastlanes/src/rle/vtable/mod.rs), the RLE value child is DType::Primitive(dtype.as_ptype(), NonNullable) — i.e. the same ptype as the array, stored raw. F64/F32 therefore serialize identically to the integer arms: a values pool (8-/4-byte elements), a u8/u16 indices child, and per-chunk values_idx_offsets. Nullability is delegated to the indices child (ValidityVTable::validity slices indices and reads its validity), matching how the existing arms already propagate a MaskedArray from a nullable indices child; the new F64/F32 arms reuse that same wrapping. Added LazyRleDoubleArray / LazyRleFloatArray (mirroring LazyRleLongArray's chunk-walk decode) and the F64/F32 decode + empty arms. Buffers stay zero-copy; the value-pool reads use branch-split broadcast handling like the existing arms. Verified on the real seoul-bike-sharing file (F64 weather columns): the CLI export now completes and all 8760 rows of temperature and wind_speed match the parquet oracle exactly. Closes #209 Co-Authored-By: Claude Opus 4.8 --- .../reader/array/LazyRleDoubleArray.java | 75 +++++++ .../reader/array/LazyRleFloatArray.java | 80 ++++++++ .../reader/decode/RleEncodingDecoder.java | 30 +++ .../vortex/reader/array/LazyRleArrayTest.java | 191 ++++++++++++++++++ 4 files changed, 376 insertions(+) create mode 100644 reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyRleDoubleArray.java create mode 100644 reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyRleFloatArray.java diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyRleDoubleArray.java b/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyRleDoubleArray.java new file mode 100644 index 00000000..d5d4894d --- /dev/null +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyRleDoubleArray.java @@ -0,0 +1,75 @@ +package io.github.dfa1.vortex.reader.array; + +import io.github.dfa1.vortex.core.model.DType; + +import java.util.function.DoubleBinaryOperator; +import java.util.function.DoubleConsumer; + +/// Lazy FastLanes-RLE-encoded [DoubleArray]. See [LazyRleLongArray] for semantics. +/// +/// @param dtype logical element type +/// @param length total logical row count +/// @param values concatenated distinct values per chunk +/// @param indices per-row local index table +/// @param valuesIdxOffsets per-chunk values-pool start offsets +/// @param firstOffset absolute origin of the values pool +/// @param valuesLen total values pool length +/// @param numChunks number of FastLanes chunks covered +/// @param offset starting absolute position +@SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared. +public record LazyRleDoubleArray( + DType dtype, long length, double[] values, int[] indices, + long[] valuesIdxOffsets, long firstOffset, long valuesLen, + int numChunks, int offset) + implements DoubleArray { + + @Override + public double getDouble(long i) { + int absRow = (int) (i + offset); + int chunkIdx = absRow >>> RleArrays.FL_LOG2; + int rowInChunk = absRow & RleArrays.FL_MASK; + long valueIdxOffset = valuesIdxOffsets[chunkIdx] - firstOffset; + int numChunkValues = RleArrays.chunkValueCount(chunkIdx, numChunks, valuesIdxOffsets, firstOffset, valuesLen); + if (numChunkValues <= 1) { + return numChunkValues == 1 ? values[(int) valueIdxOffset] : 0.0; + } + int localIdx = indices[chunkIdx * RleArrays.FL_CHUNK_SIZE + rowInChunk]; + if (localIdx >= numChunkValues) { + localIdx = numChunkValues - 1; + } + return values[(int) valueIdxOffset + localIdx]; + } + + @Override + public void forEachDouble(DoubleConsumer c) { + RleArrays.walkChunks(length, offset, numChunks, + (chunkIdx, rowInChunk, end) -> processChunk(chunkIdx, rowInChunk, end, c)); + } + + private void processChunk(int chunkIdx, int rowInChunk, int end, DoubleConsumer c) { + int chunkBase = chunkIdx * RleArrays.FL_CHUNK_SIZE; + long valueIdxOffset = valuesIdxOffsets[chunkIdx] - firstOffset; + int numChunkValues = RleArrays.chunkValueCount(chunkIdx, numChunks, valuesIdxOffsets, firstOffset, valuesLen); + if (numChunkValues <= 1) { + double v = numChunkValues == 1 ? values[(int) valueIdxOffset] : 0.0; + for (int r = rowInChunk; r < end; r++) { + c.accept(v); + } + } else { + for (int r = rowInChunk; r < end; r++) { + int localIdx = indices[chunkBase + r]; + if (localIdx >= numChunkValues) { + localIdx = numChunkValues - 1; + } + c.accept(values[(int) valueIdxOffset + localIdx]); + } + } + } + + @Override + public double fold(double identity, DoubleBinaryOperator op) { + double[] acc = {identity}; + forEachDouble(v -> acc[0] = op.applyAsDouble(acc[0], v)); + return acc[0]; + } +} diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyRleFloatArray.java b/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyRleFloatArray.java new file mode 100644 index 00000000..6b9bf99b --- /dev/null +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyRleFloatArray.java @@ -0,0 +1,80 @@ +package io.github.dfa1.vortex.reader.array; + +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.core.io.VortexFormat; + +import java.lang.foreign.MemorySegment; +import java.lang.foreign.SegmentAllocator; + +/// Lazy FastLanes-RLE-encoded [FloatArray]. See [LazyRleLongArray] for semantics. +/// +/// @param dtype logical element type +/// @param length total logical row count +/// @param values concatenated distinct values per chunk +/// @param indices per-row local index table +/// @param valuesIdxOffsets per-chunk values-pool start offsets +/// @param firstOffset absolute origin of the values pool +/// @param valuesLen total values pool length +/// @param numChunks number of FastLanes chunks covered +/// @param offset starting absolute position +@SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared. +public record LazyRleFloatArray( + DType dtype, long length, float[] values, int[] indices, + long[] valuesIdxOffsets, long firstOffset, long valuesLen, + int numChunks, int offset) + implements FloatArray { + + @Override + public float getFloat(long i) { + int absRow = (int) (i + offset); + int chunkIdx = absRow >>> RleArrays.FL_LOG2; + int rowInChunk = absRow & RleArrays.FL_MASK; + long valueIdxOffset = valuesIdxOffsets[chunkIdx] - firstOffset; + int numChunkValues = RleArrays.chunkValueCount(chunkIdx, numChunks, valuesIdxOffsets, firstOffset, valuesLen); + if (numChunkValues <= 1) { + return numChunkValues == 1 ? values[(int) valueIdxOffset] : 0.0f; + } + int localIdx = indices[chunkIdx * RleArrays.FL_CHUNK_SIZE + rowInChunk]; + if (localIdx >= numChunkValues) { + localIdx = numChunkValues - 1; + } + return values[(int) valueIdxOffset + localIdx]; + } + + /// Bulk-decodes chunk by chunk into a fresh little-endian `f32` segment, + /// with the constant-run fast path (`numChunkValues <= 1`) emitting each + /// value once. See [LazyRleLongArray] for the chunk-walk rationale. + /// + /// @param arena allocator for the output segment + /// @return a little-endian `f32` segment of `length()` decoded elements + @Override + public MemorySegment materialize(SegmentAllocator arena) { + long n = length; + MemorySegment dst = arena.allocate(n * 4L, 4); + RleArrays.walkChunks(length, offset, numChunks, + (chunkIdx, rowInChunk, end) -> processChunk(chunkIdx, rowInChunk, end, dst)); + return dst; + } + + private void processChunk(int chunkIdx, int rowInChunk, int end, MemorySegment dst) { + int chunkBase = chunkIdx * RleArrays.FL_CHUNK_SIZE; + long valueIdxOffset = valuesIdxOffsets[chunkIdx] - firstOffset; + int numChunkValues = RleArrays.chunkValueCount(chunkIdx, numChunks, valuesIdxOffsets, firstOffset, valuesLen); + // dstIdx tracks the logical output row; chunk 0 may start mid-chunk when offset != 0. + long dstIdx = (long) chunkIdx * RleArrays.FL_CHUNK_SIZE + rowInChunk - offset; + if (numChunkValues <= 1) { + float v = numChunkValues == 1 ? values[(int) valueIdxOffset] : 0.0f; + for (int r = rowInChunk; r < end; r++) { + dst.setAtIndex(VortexFormat.LE_FLOAT, dstIdx++, v); + } + } else { + for (int r = rowInChunk; r < end; r++) { + int localIdx = indices[chunkBase + r]; + if (localIdx >= numChunkValues) { + localIdx = numChunkValues - 1; + } + dst.setAtIndex(VortexFormat.LE_FLOAT, dstIdx++, values[(int) valueIdxOffset + localIdx]); + } + } + } +} diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/decode/RleEncodingDecoder.java b/reader/src/main/java/io/github/dfa1/vortex/reader/decode/RleEncodingDecoder.java index c534f030..82ec9aaf 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/decode/RleEncodingDecoder.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/decode/RleEncodingDecoder.java @@ -9,10 +9,14 @@ import io.github.dfa1.vortex.reader.array.Array; import io.github.dfa1.vortex.reader.array.BoolArray; import io.github.dfa1.vortex.reader.array.LazyConstantByteArray; +import io.github.dfa1.vortex.reader.array.LazyConstantDoubleArray; +import io.github.dfa1.vortex.reader.array.LazyConstantFloatArray; import io.github.dfa1.vortex.reader.array.LazyConstantIntArray; import io.github.dfa1.vortex.reader.array.LazyConstantLongArray; import io.github.dfa1.vortex.reader.array.LazyConstantShortArray; import io.github.dfa1.vortex.reader.array.LazyRleByteArray; +import io.github.dfa1.vortex.reader.array.LazyRleDoubleArray; +import io.github.dfa1.vortex.reader.array.LazyRleFloatArray; import io.github.dfa1.vortex.reader.array.LazyRleIntArray; import io.github.dfa1.vortex.reader.array.LazyRleLongArray; import io.github.dfa1.vortex.reader.array.LazyRleShortArray; @@ -96,6 +100,12 @@ public Array decode(DecodeContext ctx) { readBytes(valuesSeg, (int) valuesLen), indices, valuesIdxOffsets, firstOffset, valuesLen, numChunks, offset, ptype == PType.U8); + case F64 -> new LazyRleDoubleArray(ctx.dtype(), rowCount, + readDoubles(valuesSeg, (int) valuesLen), + indices, valuesIdxOffsets, firstOffset, valuesLen, numChunks, offset); + case F32 -> new LazyRleFloatArray(ctx.dtype(), rowCount, + readFloats(valuesSeg, (int) valuesLen), + indices, valuesIdxOffsets, firstOffset, valuesLen, numChunks, offset); default -> throw new VortexException(EncodingId.FASTLANES_RLE, "unsupported ptype " + ptype); }; @@ -114,6 +124,8 @@ private static Array emptyArray(DecodeContext ctx) { case I32, U32 -> new LazyConstantIntArray(dt, 0L, 0); case I16, U16 -> new LazyConstantShortArray(dt, 0L, (short) 0); case I8, U8 -> new LazyConstantByteArray(dt, 0L, (byte) 0); + case F64 -> new LazyConstantDoubleArray(dt, 0L, 0.0); + case F32 -> new LazyConstantFloatArray(dt, 0L, 0.0f); default -> throw new VortexException(EncodingId.FASTLANES_RLE, "unsupported ptype " + ptype); }; } @@ -155,6 +167,24 @@ private static short[] readShorts(MemorySegment buf, int count, PType ptype) { return out; } + private static double[] readDoubles(MemorySegment buf, int count) { + double[] out = new double[count]; + long cap = SegmentBroadcast.capacity(buf, 8); + for (int i = 0; i < count; i++) { + out[i] = buf.get(VortexFormat.LE_DOUBLE, (i % cap) * 8); + } + return out; + } + + private static float[] readFloats(MemorySegment buf, int count) { + float[] out = new float[count]; + long cap = SegmentBroadcast.capacity(buf, 4); + for (int i = 0; i < count; i++) { + out[i] = buf.get(VortexFormat.LE_FLOAT, (i % cap) * 4); + } + return out; + } + private static byte[] readBytes(MemorySegment buf, int count) { byte[] out = new byte[count]; long cap = SegmentBroadcast.capacity(buf, 1); diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/array/LazyRleArrayTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/array/LazyRleArrayTest.java index 94954160..fe0822f9 100644 --- a/reader/src/test/java/io/github/dfa1/vortex/reader/array/LazyRleArrayTest.java +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/array/LazyRleArrayTest.java @@ -21,6 +21,8 @@ class LazyRleArrayTest { private static final DType I8 = DType.I8; private static final DType U16 = DType.U16; private static final DType I16 = DType.I16; + private static final DType F64 = DType.F64; + private static final DType F32 = DType.F32; @Nested class LongDispatch { @@ -328,4 +330,193 @@ void foldStopsWhenLengthSatisfiedBeforeChunksExhausted() { assertThat(sut.fold(0L, Long::sum)).isEqualTo(5L); } } + + /// F64 RLE columns are produced by the Python Vortex writer for double weather + /// columns with long constant runs (issue #209). Fractional values are chosen so a + /// truncating or int-widening decode bug would surface as a wrong assertion. + @Nested + class DoubleDispatch { + + @Test + void singleChunkAllSameValue_constantRunFastPath() { + // Given a single 1024-row chunk with 1 distinct value; the constant-run + // fast path must return it for every row without touching indices. + double[] values = {2.5}; + int[] indices = new int[1024]; + long[] valuesIdxOffsets = {0L}; + var sut = new LazyRleDoubleArray(F64, 1024, values, indices, + valuesIdxOffsets, 0L, 1L, 1, 0); + + // When / Then + assertThat(sut.getDouble(0)).isEqualTo(2.5); + assertThat(sut.getDouble(500)).isEqualTo(2.5); + assertThat(sut.getDouble(1023)).isEqualTo(2.5); + } + + @Test + void singleChunkWithIndices_perRowLookup() { + // Given one chunk with 3 distinct fractional values selected in a 0,1,2 cycle; + // fractional values catch any accidental integer narrowing in the value read. + double[] values = {1.1, 2.2, 3.3}; + int[] indices = new int[1024]; + for (int i = 0; i < 6; i++) { + indices[i] = i % 3; + } + long[] valuesIdxOffsets = {0L}; + var sut = new LazyRleDoubleArray(F64, 6, values, indices, + valuesIdxOffsets, 0L, 3L, 1, 0); + + var result = new ArrayList(); + sut.forEachDouble(result::add); + + assertThat(result).containsExactly(1.1, 2.2, 3.3, 1.1, 2.2, 3.3); + } + + @Test + void multiChunkBoundary_walksAcrossChunks() { + // Given two constant chunks (chunk 0 = -1.5, chunk 1 = 4.25); the negative + // value guards against a sign-loss bug and the boundary crosses a chunk. + double[] values = {-1.5, 4.25}; + int[] indices = new int[2 * 1024]; + long[] valuesIdxOffsets = {0L, 1L}; + var sut = new LazyRleDoubleArray(F64, 1026, values, indices, + valuesIdxOffsets, 0L, 2L, 2, 0); + + // When / Then — rows straddle the 1024-row chunk boundary + assertThat(sut.getDouble(0)).isEqualTo(-1.5); + assertThat(sut.getDouble(1023)).isEqualTo(-1.5); + assertThat(sut.getDouble(1024)).isEqualTo(4.25); + assertThat(sut.getDouble(1025)).isEqualTo(4.25); + } + + @Test + void offsetSkipsLeadingRows() { + // Given chunk 0 = constant 5.75; offset=100 maps logical row 0 to absolute 100, + // matching the slice-with-offset arrays the Python writer emits. + double[] values = {5.75}; + int[] indices = new int[1024]; + long[] valuesIdxOffsets = {0L}; + var sut = new LazyRleDoubleArray(F64, 10, values, indices, + valuesIdxOffsets, 0L, 1L, 1, 100); + + // When / Then + assertThat(sut.getDouble(0)).isEqualTo(5.75); + assertThat(sut.getDouble(9)).isEqualTo(5.75); + } + + @Test + void foldSumsFractionalValues() { + // Given two constant chunks (0.5 and 0.25); a fractional fold sum would be + // wrong if any element were truncated to a long during decode. + double[] values = {0.5, 0.25}; + int[] indices = new int[2 * 1024]; + long[] valuesIdxOffsets = {0L, 1L}; + var sut = new LazyRleDoubleArray(F64, 1026, values, indices, + valuesIdxOffsets, 0L, 2L, 2, 0); + + double result = sut.fold(0.0, Double::sum); + + // 1024 * 0.5 + 2 * 0.25 = 512.0 + 0.5 = 512.5 + assertThat(result).isEqualTo(512.5); + } + + @Test + void indexedClampAndEmptyChunk() { + // Given an indexed chunk whose index[2] overruns the value range: it must + // clamp to the last value (the writer leaves trailing bits 0 for constant runs). + int[] idx = new int[1024]; + idx[0] = 0; + idx[1] = 1; + idx[2] = 9; + var sut = new LazyRleDoubleArray(F64, 3, new double[]{1.0, 2.0, 3.0}, idx, + new long[]{0L}, 0L, 3L, 1, 0); + + // When / Then — out-of-range index clamps to last + assertThat(sut.getDouble(2)).isEqualTo(3.0); + + // empty chunk (0 distinct values) → 0.0 via getDouble and forEach + var empty = new LazyRleDoubleArray(F64, 2, new double[0], new int[1024], + new long[]{0L}, 0L, 0L, 1, 0); + assertThat(empty.getDouble(0)).isZero(); + var zeros = new ArrayList(); + empty.forEachDouble(zeros::add); + assertThat(zeros).containsExactly(0.0, 0.0); + } + } + + /// F32 RLE columns follow the same wire layout as F64 with a 4-byte value child + /// (issue #209); [LazyRleFloatArray] materializes rather than exposing a forEach. + @Nested + class FloatDispatch { + + @Test + void singleChunkWithIndices_perRowLookup() { + // Given one chunk with 3 distinct float values in a 0,1,2 cycle; fractional + // 32-bit values catch a wrong-width read (reading 8 bytes as a double). + float[] values = {1.5f, 2.5f, 3.5f}; + int[] indices = new int[1024]; + for (int i = 0; i < 6; i++) { + indices[i] = i % 3; + } + var sut = new LazyRleFloatArray(F32, 6, values, indices, + new long[]{0L}, 0L, 3L, 1, 0); + + // When / Then + assertThat(sut.getFloat(0)).isEqualTo(1.5f); + assertThat(sut.getFloat(1)).isEqualTo(2.5f); + assertThat(sut.getFloat(5)).isEqualTo(3.5f); + } + + @Test + void constantRunFastPathAndMultiChunk() { + // Given two constant chunks (chunk 0 = -2.25, chunk 1 = 8.75); crosses the + // 1024-row boundary and the negative value guards against sign loss. + var sut = new LazyRleFloatArray(F32, 1026, new float[]{-2.25f, 8.75f}, + new int[2 * 1024], new long[]{0L, 1L}, 0L, 2L, 2, 0); + + // When / Then + assertThat(sut.getFloat(0)).isEqualTo(-2.25f); + assertThat(sut.getFloat(1024)).isEqualTo(8.75f); + } + + @Test + void indexedClampAndEmptyChunk() { + // Given an indexed chunk whose index[2] overruns the value range → clamp to last. + int[] idx = new int[1024]; + idx[0] = 0; + idx[1] = 1; + idx[2] = 9; + var sut = new LazyRleFloatArray(F32, 3, new float[]{1.0f, 2.0f, 3.0f}, idx, + new long[]{0L}, 0L, 3L, 1, 0); + + // When / Then — clamped to last value + assertThat(sut.getFloat(2)).isEqualTo(3.0f); + + // empty chunk (0 distinct values) → 0.0f + var empty = new LazyRleFloatArray(F32, 2, new float[0], new int[1024], + new long[]{0L}, 0L, 0L, 1, 0); + assertThat(empty.getFloat(0)).isZero(); + } + + @Test + void materializeDecodesEveryRow() { + // Given a mixed chunk; materialize must emit each logical row once, honoring + // the constant-run fast path only within a chunk (here indices vary). + float[] values = {10.5f, 20.5f}; + int[] indices = new int[1024]; + indices[0] = 0; + indices[1] = 1; + indices[2] = 1; + var sut = new LazyRleFloatArray(F32, 3, values, indices, + new long[]{0L}, 0L, 2L, 1, 0); + + try (var arena = java.lang.foreign.Arena.ofConfined()) { + var result = sut.materialize(arena); + + assertThat(result.getAtIndex(io.github.dfa1.vortex.core.io.VortexFormat.LE_FLOAT, 0)).isEqualTo(10.5f); + assertThat(result.getAtIndex(io.github.dfa1.vortex.core.io.VortexFormat.LE_FLOAT, 1)).isEqualTo(20.5f); + assertThat(result.getAtIndex(io.github.dfa1.vortex.core.io.VortexFormat.LE_FLOAT, 2)).isEqualTo(20.5f); + } + } + } } From 897a5997aa7991978ca69475de199d3c29e8917b Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Mon, 6 Jul 2026 21:50:06 +0200 Subject: [PATCH 2/3] docs: changelog entry for #209 Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2eca6fac..6368bea8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - CSV export renders unsigned integer columns (U8–U64) with their unsigned values — high-half values previously printed as two's-complement negatives (uci-wine `magnesium` U8 132 exported as -124), silent corruption found by the Raincloud conformance suite. ([#208](https://github.com/dfa1/vortex-java/issues/208)) +- `fastlanes.rle` decodes F64/F32 value pools (`LazyRleDoubleArray`, `LazyRleFloatArray`) — files from the Python bindings RLE-encode double columns with long constant runs, which previously failed to scan with `unsupported ptype F64`. ([#209](https://github.com/dfa1/vortex-java/issues/209)) ## [0.12.0] — 2026-07-04 From 422baeb314143b14b49407db53ce86309adba56e Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Mon, 6 Jul 2026 22:06:09 +0200 Subject: [PATCH 3/3] test: ground-truth RLE F64 interop + review nits (#209) Add RleF64InteropIntegrationTest: the JNI (Rust) writer compresses long constant runs of doubles to fastlanes.rle, so this drives the real decode path (F64/F32 arms, readDoubles/readFloats over a real segment, real valuesIdxOffsets) end to end rather than hand-built lazy records. Each case inspects the file via VortexInspector and assumeTrue it contains fastlanes.rle before comparing values, so it never asserts on an unintended encoding. The JNI compressor does choose RLE for the crafted 512-row constant runs: all three cases (F64, F32, nullable F64) run, none skip. The nullable case decodes without error and checks non-null values exactly; it deliberately does not assert the null mask, since surfacing RLE validity is the separate in-flight fix #210 (matching jniWriter_nullableColumn_decodesWithoutError). Also address review nits on LazyRleArrayTest: replace inline FQNs with imports, add the missing // When / // Then markers, and add a NaN/infinity bit-preservation case for the double arm. Co-Authored-By: Claude Opus 4.8 --- .../RleF64InteropIntegrationTest.java | 259 ++++++++++++++++++ .../vortex/reader/array/LazyRleArrayTest.java | 33 ++- 2 files changed, 288 insertions(+), 4 deletions(-) create mode 100644 integration/src/test/java/io/github/dfa1/vortex/integration/RleF64InteropIntegrationTest.java diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/RleF64InteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/RleF64InteropIntegrationTest.java new file mode 100644 index 00000000..c2e95366 --- /dev/null +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/RleF64InteropIntegrationTest.java @@ -0,0 +1,259 @@ +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.inspect.VortexInspector; +import io.github.dfa1.vortex.reader.ReadRegistry; +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.DoubleArray; +import io.github.dfa1.vortex.reader.array.FloatArray; +import io.github.dfa1.vortex.reader.array.MaskedArray; +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.Float4Vector; +import org.apache.arrow.vector.Float8Vector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.types.FloatingPointPrecision; +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 java.io.IOException; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +/// Ground-truth interop for `fastlanes.rle` over floating-point columns (issue #209). +/// +/// The Rust (JNI) writer compresses long constant runs of doubles to +/// `fastlanes.rle`; before the fix the Java reader threw +/// `fastlanes.rle: unsupported ptype F64`. These tests write such a column with +/// the JNI writer, then confirm via [VortexInspector] that RLE was actually +/// chosen ( `assumeTrue` — a test asserting on whatever encoding the compressor +/// happened to pick would be meaningless), and finally assert the Java reader +/// decodes every value exactly. The nullable case additionally exercises the +/// validity re-wrap the F64 arm shares with the integer arms. +class RleF64InteropIntegrationTest { + + private static final Session SESSION = Session.create(); + private static final BufferAllocator ALLOCATOR = ArrowAllocation.rootAllocator(); + + private static final Schema F64_SCHEMA = new Schema(List.of( + Field.notNullable("v", new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE)) + )); + private static final Schema F64_NULLABLE_SCHEMA = new Schema(List.of( + Field.nullable("v", new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE)) + )); + private static final Schema F32_SCHEMA = new Schema(List.of( + Field.notNullable("v", new ArrowType.FloatingPoint(FloatingPointPrecision.SINGLE)) + )); + + /// Number of rows: long constant runs (512-row blocks of one value) over a + /// handful of distinct values is the canonical RLE-friendly shape and matches + /// the real-world seoul weather columns from #209. + private static final int ROWS = 8_192; + private static final int RUN = 512; + + static { + NativeLoader.loadJni(); + } + + @Test + void jniWriter_javaReader_f64RunLengthEncoded(@TempDir Path tmp) throws IOException { + // Given — 8_192 doubles in 512-row constant runs cycling 5 distinct fractional + // values; fractional values catch any accidental integer-narrowing decode bug. + double[] unique = {-1.5, 2.25, 3.75, -4.125, 5.5}; + double[] vals = new double[ROWS]; + for (int i = 0; i < ROWS; i++) { + vals[i] = unique[(i / RUN) % unique.length]; + } + Path file = tmp.resolve("rle_f64.vtx"); + writeF64(file, vals); + assumeRle(file); + + // When — Java reader decodes the whole column + double[] result = readDoubleColumn(file); + + // Then — every value matches the input exactly + assertThat(result).containsExactly(vals); + } + + @Test + void jniWriter_javaReader_f64NullableDecodesValues(@TempDir Path tmp) throws IOException { + // Given — the same run shape but nullable, with whole 512-row runs set null. + // Exercises the F64 RLE decode path on a nullable schema end-to-end. + double[] unique = {10.5, 20.25, 30.75}; + double[] vals = new double[ROWS]; + boolean[] nulls = new boolean[ROWS]; + for (int i = 0; i < ROWS; i++) { + int block = i / RUN; + nulls[i] = block % 3 == 1; // every third run is entirely null + vals[i] = unique[block % unique.length]; + } + Path file = tmp.resolve("rle_f64_nullable.vtx"); + writeF64Nullable(file, vals, nulls); + assumeRle(file); + + // When — decode the whole column (unwrapping any validity mask) + double[] result = readDoubleColumnUnwrapped(file); + + // Then — decode does not throw, row count is right, and every non-null input + // value is exact. This test deliberately does not assert the null mask: the JNI + // writer folds RLE validity into the indices child, and surfacing it as a + // MaskedArray from the RLE decode path is the subject of the separate in-flight + // fix #210 (the existing jniWriter_nullableColumn_decodesWithoutError follows the + // same convention for the bitpacked case). + assertThat(result).hasSize(ROWS); + for (int i = 0; i < ROWS; i++) { + if (!nulls[i]) { + assertThat(result[i]).as("value@%d", i).isEqualTo(vals[i]); + } + } + } + + @Test + void jniWriter_javaReader_f32RunLengthEncoded(@TempDir Path tmp) throws IOException { + // Given — the same run shape as F64 but 32-bit; guards the 4-byte value read. + // Gated on RLE selection: the compressor may prefer another encoding for F32, + // in which case the test is skipped rather than asserting on the wrong path. + float[] unique = {-1.5f, 2.25f, 3.75f, -4.125f, 5.5f}; + float[] vals = new float[ROWS]; + for (int i = 0; i < ROWS; i++) { + vals[i] = unique[(i / RUN) % unique.length]; + } + Path file = tmp.resolve("rle_f32.vtx"); + writeF32(file, vals); + assumeRle(file); + + // When + float[] result = readFloatColumn(file); + + // Then + assertThat(result).containsExactly(vals); + } + + // ── helpers ─────────────────────────────────────────────────────────────── + + /// Skips the test unless the JNI compressor actually chose `fastlanes.rle`. + private static void assumeRle(Path file) throws IOException { + String report; + try (VortexReader vf = VortexReader.open(file, ReadRegistry.loadAll())) { + report = VortexInspector.inspect(vf); + } + assumeTrue(report.contains("fastlanes.rle"), + () -> "compressor did not choose fastlanes.rle:\n" + report); + } + + private static void writeF64(Path file, double[] vals) throws IOException { + String uri = file.toAbsolutePath().toUri().toString(); + try (VortexWriter writer = VortexWriter.create(SESSION, uri, F64_SCHEMA, new HashMap<>(), ALLOCATOR); + VectorSchemaRoot root = VectorSchemaRoot.create(F64_SCHEMA, ALLOCATOR)) { + Float8Vector vec = (Float8Vector) root.getVector("v"); + vec.allocateNew(vals.length); + for (int i = 0; i < vals.length; i++) { + vec.setSafe(i, vals[i]); + } + root.setRowCount(vals.length); + exportBatch(writer, root); + } + } + + private static void writeF64Nullable(Path file, double[] vals, boolean[] nulls) throws IOException { + String uri = file.toAbsolutePath().toUri().toString(); + try (VortexWriter writer = VortexWriter.create(SESSION, uri, F64_NULLABLE_SCHEMA, new HashMap<>(), ALLOCATOR); + VectorSchemaRoot root = VectorSchemaRoot.create(F64_NULLABLE_SCHEMA, ALLOCATOR)) { + Float8Vector vec = (Float8Vector) root.getVector("v"); + vec.allocateNew(vals.length); + for (int i = 0; i < vals.length; i++) { + if (nulls[i]) { + vec.setNull(i); + } else { + vec.setSafe(i, vals[i]); + } + } + root.setRowCount(vals.length); + exportBatch(writer, root); + } + } + + private static void writeF32(Path file, float[] vals) throws IOException { + String uri = file.toAbsolutePath().toUri().toString(); + try (VortexWriter writer = VortexWriter.create(SESSION, uri, F32_SCHEMA, new HashMap<>(), ALLOCATOR); + VectorSchemaRoot root = VectorSchemaRoot.create(F32_SCHEMA, ALLOCATOR)) { + Float4Vector vec = (Float4Vector) root.getVector("v"); + vec.allocateNew(vals.length); + for (int i = 0; i < vals.length; i++) { + vec.setSafe(i, vals[i]); + } + root.setRowCount(vals.length); + exportBatch(writer, root); + } + } + + private static void exportBatch(VortexWriter writer, VectorSchemaRoot root) throws IOException { + try (ArrowArray arr = ArrowArray.allocateNew(ALLOCATOR); + ArrowSchema schema = ArrowSchema.allocateNew(ALLOCATOR)) { + Data.exportVectorSchemaRoot(ALLOCATOR, root, null, arr, schema); + writer.writeBatch(arr.memoryAddress(), schema.memoryAddress()); + } + } + + private static double[] readDoubleColumn(Path file) throws IOException { + try (var vf = VortexReader.open(file, ReadRegistry.loadAll()); + var iter = vf.scan(ScanOptions.columns("v"))) { + var out = new ArrayList(); + iter.forEachRemaining(c -> { + DoubleArray arr = c.column("v"); + for (long i = 0; i < arr.length(); i++) { + out.add(arr.getDouble(i)); + } + }); + return out.stream().mapToDouble(Double::doubleValue).toArray(); + } + } + + private static double[] readDoubleColumnUnwrapped(Path file) throws IOException { + try (var vf = VortexReader.open(file, ReadRegistry.loadAll()); + var iter = vf.scan(ScanOptions.columns("v"))) { + var out = new ArrayList(); + iter.forEachRemaining(c -> { + Array a = c.column("v"); + DoubleArray arr = (DoubleArray) (a instanceof MaskedArray m ? m.inner() : a); + for (long i = 0; i < arr.length(); i++) { + out.add(arr.getDouble(i)); + } + }); + return out.stream().mapToDouble(Double::doubleValue).toArray(); + } + } + + private static float[] readFloatColumn(Path file) throws IOException { + try (var vf = VortexReader.open(file, ReadRegistry.loadAll()); + var iter = vf.scan(ScanOptions.columns("v"))) { + var out = new ArrayList(); + iter.forEachRemaining(c -> { + FloatArray arr = c.column("v"); + for (long i = 0; i < arr.length(); i++) { + out.add(arr.getFloat(i)); + } + }); + float[] result = new float[out.size()]; + for (int i = 0; i < result.length; i++) { + result[i] = out.get(i); + } + return result; + } + } +} diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/array/LazyRleArrayTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/array/LazyRleArrayTest.java index fe0822f9..483d294a 100644 --- a/reader/src/test/java/io/github/dfa1/vortex/reader/array/LazyRleArrayTest.java +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/array/LazyRleArrayTest.java @@ -1,9 +1,11 @@ package io.github.dfa1.vortex.reader.array; +import io.github.dfa1.vortex.core.io.VortexFormat; import io.github.dfa1.vortex.core.model.DType; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; +import java.lang.foreign.Arena; import java.util.ArrayList; import static org.assertj.core.api.Assertions.assertThat; @@ -420,6 +422,27 @@ void foldSumsFractionalValues() { assertThat(result).isEqualTo(512.5); } + @Test + void preservesNaNAndInfinityBitPatterns() { + // Given a specific quiet-NaN payload plus +/-inf; the decode must copy raw + // IEEE-754 bits verbatim, so bit-exact equality (not value equality, since + // NaN != NaN) is the right assertion. + double nanPayload = Double.longBitsToDouble(0x7FF8_0000_0000_002AL); + double[] values = {nanPayload, Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY}; + int[] indices = new int[1024]; + indices[0] = 0; + indices[1] = 1; + indices[2] = 2; + var sut = new LazyRleDoubleArray(F64, 3, values, indices, + new long[]{0L}, 0L, 3L, 1, 0); + + // When / Then — raw bits round-trip unchanged + assertThat(Double.doubleToRawLongBits(sut.getDouble(0))) + .isEqualTo(0x7FF8_0000_0000_002AL); + assertThat(sut.getDouble(1)).isEqualTo(Double.POSITIVE_INFINITY); + assertThat(sut.getDouble(2)).isEqualTo(Double.NEGATIVE_INFINITY); + } + @Test void indexedClampAndEmptyChunk() { // Given an indexed chunk whose index[2] overruns the value range: it must @@ -510,12 +533,14 @@ void materializeDecodesEveryRow() { var sut = new LazyRleFloatArray(F32, 3, values, indices, new long[]{0L}, 0L, 2L, 1, 0); - try (var arena = java.lang.foreign.Arena.ofConfined()) { + // When + try (var arena = Arena.ofConfined()) { var result = sut.materialize(arena); - assertThat(result.getAtIndex(io.github.dfa1.vortex.core.io.VortexFormat.LE_FLOAT, 0)).isEqualTo(10.5f); - assertThat(result.getAtIndex(io.github.dfa1.vortex.core.io.VortexFormat.LE_FLOAT, 1)).isEqualTo(20.5f); - assertThat(result.getAtIndex(io.github.dfa1.vortex.core.io.VortexFormat.LE_FLOAT, 2)).isEqualTo(20.5f); + // Then — each logical row decodes exactly (row 2 reuses value 1 via its index) + assertThat(result.getAtIndex(VortexFormat.LE_FLOAT, 0)).isEqualTo(10.5f); + assertThat(result.getAtIndex(VortexFormat.LE_FLOAT, 1)).isEqualTo(20.5f); + assertThat(result.getAtIndex(VortexFormat.LE_FLOAT, 2)).isEqualTo(20.5f); } } }