Skip to content

Commit 2a6dd4a

Browse files
dfa1claude
dfa1
and
claude
committed
fix: decode fastlanes.rle over F64/F32 (#209)
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 <noreply@anthropic.com>
1 parent 66d32f3 commit 2a6dd4a

4 files changed

Lines changed: 376 additions & 0 deletions

File tree

‎reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyRleDoubleArray.java‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
package io.github.dfa1.vortex.reader.array;
2+
3+
import io.github.dfa1.vortex.core.model.DType;
4+
5+
import java.util.function.DoubleBinaryOperator;
6+
import java.util.function.DoubleConsumer;
7+
8+
/// Lazy FastLanes-RLE-encoded [DoubleArray]. See [LazyRleLongArray] for semantics.
9+
///
10+
/// @param dtype logical element type
11+
/// @param length total logical row count
12+
/// @param values concatenated distinct values per chunk
13+
/// @param indices per-row local index table
14+
/// @param valuesIdxOffsets per-chunk values-pool start offsets
15+
/// @param firstOffset absolute origin of the values pool
16+
/// @param valuesLen total values pool length
17+
/// @param numChunks number of FastLanes chunks covered
18+
/// @param offset starting absolute position
19+
@SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared.
20+
public record LazyRleDoubleArray(
21+
DType dtype, long length, double[] values, int[] indices,
22+
long[] valuesIdxOffsets, long firstOffset, long valuesLen,
23+
int numChunks, int offset)
24+
implements DoubleArray {
25+
26+
@Override
27+
public double getDouble(long i) {
28+
int absRow = (int) (i + offset);
29+
int chunkIdx = absRow >>> RleArrays.FL_LOG2;
30+
int rowInChunk = absRow & RleArrays.FL_MASK;
31+
long valueIdxOffset = valuesIdxOffsets[chunkIdx] - firstOffset;
32+
int numChunkValues = RleArrays.chunkValueCount(chunkIdx, numChunks, valuesIdxOffsets, firstOffset, valuesLen);
33+
if (numChunkValues <= 1) {
34+
return numChunkValues == 1 ? values[(int) valueIdxOffset] : 0.0;
35+
}
36+
int localIdx = indices[chunkIdx * RleArrays.FL_CHUNK_SIZE + rowInChunk];
37+
if (localIdx >= numChunkValues) {
38+
localIdx = numChunkValues - 1;
39+
}
40+
return values[(int) valueIdxOffset + localIdx];
41+
}
42+
43+
@Override
44+
public void forEachDouble(DoubleConsumer c) {
45+
RleArrays.walkChunks(length, offset, numChunks,
46+
(chunkIdx, rowInChunk, end) -> processChunk(chunkIdx, rowInChunk, end, c));
47+
}
48+
49+
private void processChunk(int chunkIdx, int rowInChunk, int end, DoubleConsumer c) {
50+
int chunkBase = chunkIdx * RleArrays.FL_CHUNK_SIZE;
51+
long valueIdxOffset = valuesIdxOffsets[chunkIdx] - firstOffset;
52+
int numChunkValues = RleArrays.chunkValueCount(chunkIdx, numChunks, valuesIdxOffsets, firstOffset, valuesLen);
53+
if (numChunkValues <= 1) {
54+
double v = numChunkValues == 1 ? values[(int) valueIdxOffset] : 0.0;
55+
for (int r = rowInChunk; r < end; r++) {
56+
c.accept(v);
57+
}
58+
} else {
59+
for (int r = rowInChunk; r < end; r++) {
60+
int localIdx = indices[chunkBase + r];
61+
if (localIdx >= numChunkValues) {
62+
localIdx = numChunkValues - 1;
63+
}
64+
c.accept(values[(int) valueIdxOffset + localIdx]);
65+
}
66+
}
67+
}
68+
69+
@Override
70+
public double fold(double identity, DoubleBinaryOperator op) {
71+
double[] acc = {identity};
72+
forEachDouble(v -> acc[0] = op.applyAsDouble(acc[0], v));
73+
return acc[0];
74+
}
75+
}

‎reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyRleFloatArray.java‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
package io.github.dfa1.vortex.reader.array;
2+
3+
import io.github.dfa1.vortex.core.model.DType;
4+
import io.github.dfa1.vortex.core.io.VortexFormat;
5+
6+
import java.lang.foreign.MemorySegment;
7+
import java.lang.foreign.SegmentAllocator;
8+
9+
/// Lazy FastLanes-RLE-encoded [FloatArray]. See [LazyRleLongArray] for semantics.
10+
///
11+
/// @param dtype logical element type
12+
/// @param length total logical row count
13+
/// @param values concatenated distinct values per chunk
14+
/// @param indices per-row local index table
15+
/// @param valuesIdxOffsets per-chunk values-pool start offsets
16+
/// @param firstOffset absolute origin of the values pool
17+
/// @param valuesLen total values pool length
18+
/// @param numChunks number of FastLanes chunks covered
19+
/// @param offset starting absolute position
20+
@SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared.
21+
public record LazyRleFloatArray(
22+
DType dtype, long length, float[] values, int[] indices,
23+
long[] valuesIdxOffsets, long firstOffset, long valuesLen,
24+
int numChunks, int offset)
25+
implements FloatArray {
26+
27+
@Override
28+
public float getFloat(long i) {
29+
int absRow = (int) (i + offset);
30+
int chunkIdx = absRow >>> RleArrays.FL_LOG2;
31+
int rowInChunk = absRow & RleArrays.FL_MASK;
32+
long valueIdxOffset = valuesIdxOffsets[chunkIdx] - firstOffset;
33+
int numChunkValues = RleArrays.chunkValueCount(chunkIdx, numChunks, valuesIdxOffsets, firstOffset, valuesLen);
34+
if (numChunkValues <= 1) {
35+
return numChunkValues == 1 ? values[(int) valueIdxOffset] : 0.0f;
36+
}
37+
int localIdx = indices[chunkIdx * RleArrays.FL_CHUNK_SIZE + rowInChunk];
38+
if (localIdx >= numChunkValues) {
39+
localIdx = numChunkValues - 1;
40+
}
41+
return values[(int) valueIdxOffset + localIdx];
42+
}
43+
44+
/// Bulk-decodes chunk by chunk into a fresh little-endian `f32` segment,
45+
/// with the constant-run fast path (`numChunkValues <= 1`) emitting each
46+
/// value once. See [LazyRleLongArray] for the chunk-walk rationale.
47+
///
48+
/// @param arena allocator for the output segment
49+
/// @return a little-endian `f32` segment of `length()` decoded elements
50+
@Override
51+
public MemorySegment materialize(SegmentAllocator arena) {
52+
long n = length;
53+
MemorySegment dst = arena.allocate(n * 4L, 4);
54+
RleArrays.walkChunks(length, offset, numChunks,
55+
(chunkIdx, rowInChunk, end) -> processChunk(chunkIdx, rowInChunk, end, dst));
56+
return dst;
57+
}
58+
59+
private void processChunk(int chunkIdx, int rowInChunk, int end, MemorySegment dst) {
60+
int chunkBase = chunkIdx * RleArrays.FL_CHUNK_SIZE;
61+
long valueIdxOffset = valuesIdxOffsets[chunkIdx] - firstOffset;
62+
int numChunkValues = RleArrays.chunkValueCount(chunkIdx, numChunks, valuesIdxOffsets, firstOffset, valuesLen);
63+
// dstIdx tracks the logical output row; chunk 0 may start mid-chunk when offset != 0.
64+
long dstIdx = (long) chunkIdx * RleArrays.FL_CHUNK_SIZE + rowInChunk - offset;
65+
if (numChunkValues <= 1) {
66+
float v = numChunkValues == 1 ? values[(int) valueIdxOffset] : 0.0f;
67+
for (int r = rowInChunk; r < end; r++) {
68+
dst.setAtIndex(VortexFormat.LE_FLOAT, dstIdx++, v);
69+
}
70+
} else {
71+
for (int r = rowInChunk; r < end; r++) {
72+
int localIdx = indices[chunkBase + r];
73+
if (localIdx >= numChunkValues) {
74+
localIdx = numChunkValues - 1;
75+
}
76+
dst.setAtIndex(VortexFormat.LE_FLOAT, dstIdx++, values[(int) valueIdxOffset + localIdx]);
77+
}
78+
}
79+
}
80+
}

‎reader/src/main/java/io/github/dfa1/vortex/reader/decode/RleEncodingDecoder.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,10 +9,14 @@
99
import io.github.dfa1.vortex.reader.array.Array;
1010
import io.github.dfa1.vortex.reader.array.BoolArray;
1111
import io.github.dfa1.vortex.reader.array.LazyConstantByteArray;
12+
import io.github.dfa1.vortex.reader.array.LazyConstantDoubleArray;
13+
import io.github.dfa1.vortex.reader.array.LazyConstantFloatArray;
1214
import io.github.dfa1.vortex.reader.array.LazyConstantIntArray;
1315
import io.github.dfa1.vortex.reader.array.LazyConstantLongArray;
1416
import io.github.dfa1.vortex.reader.array.LazyConstantShortArray;
1517
import io.github.dfa1.vortex.reader.array.LazyRleByteArray;
18+
import io.github.dfa1.vortex.reader.array.LazyRleDoubleArray;
19+
import io.github.dfa1.vortex.reader.array.LazyRleFloatArray;
1620
import io.github.dfa1.vortex.reader.array.LazyRleIntArray;
1721
import io.github.dfa1.vortex.reader.array.LazyRleLongArray;
1822
import io.github.dfa1.vortex.reader.array.LazyRleShortArray;
@@ -96,6 +100,12 @@ public Array decode(DecodeContext ctx) {
96100
readBytes(valuesSeg, (int) valuesLen),
97101
indices, valuesIdxOffsets, firstOffset, valuesLen, numChunks, offset,
98102
ptype == PType.U8);
103+
case F64 -> new LazyRleDoubleArray(ctx.dtype(), rowCount,
104+
readDoubles(valuesSeg, (int) valuesLen),
105+
indices, valuesIdxOffsets, firstOffset, valuesLen, numChunks, offset);
106+
case F32 -> new LazyRleFloatArray(ctx.dtype(), rowCount,
107+
readFloats(valuesSeg, (int) valuesLen),
108+
indices, valuesIdxOffsets, firstOffset, valuesLen, numChunks, offset);
99109
default -> throw new VortexException(EncodingId.FASTLANES_RLE, "unsupported ptype " + ptype);
100110
};
101111

@@ -114,6 +124,8 @@ private static Array emptyArray(DecodeContext ctx) {
114124
case I32, U32 -> new LazyConstantIntArray(dt, 0L, 0);
115125
case I16, U16 -> new LazyConstantShortArray(dt, 0L, (short) 0);
116126
case I8, U8 -> new LazyConstantByteArray(dt, 0L, (byte) 0);
127+
case F64 -> new LazyConstantDoubleArray(dt, 0L, 0.0);
128+
case F32 -> new LazyConstantFloatArray(dt, 0L, 0.0f);
117129
default -> throw new VortexException(EncodingId.FASTLANES_RLE, "unsupported ptype " + ptype);
118130
};
119131
}
@@ -155,6 +167,24 @@ private static short[] readShorts(MemorySegment buf, int count, PType ptype) {
155167
return out;
156168
}
157169

170+
private static double[] readDoubles(MemorySegment buf, int count) {
171+
double[] out = new double[count];
172+
long cap = SegmentBroadcast.capacity(buf, 8);
173+
for (int i = 0; i < count; i++) {
174+
out[i] = buf.get(VortexFormat.LE_DOUBLE, (i % cap) * 8);
175+
}
176+
return out;
177+
}
178+
179+
private static float[] readFloats(MemorySegment buf, int count) {
180+
float[] out = new float[count];
181+
long cap = SegmentBroadcast.capacity(buf, 4);
182+
for (int i = 0; i < count; i++) {
183+
out[i] = buf.get(VortexFormat.LE_FLOAT, (i % cap) * 4);
184+
}
185+
return out;
186+
}
187+
158188
private static byte[] readBytes(MemorySegment buf, int count) {
159189
byte[] out = new byte[count];
160190
long cap = SegmentBroadcast.capacity(buf, 1);

0 commit comments

Comments
 (0)