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 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/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..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; @@ -21,6 +23,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 +332,216 @@ 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 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 + // 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); + + // When + try (var arena = Arena.ofConfined()) { + var result = sut.materialize(arena); + + // 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); + } + } + } }