diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/decode/DeltaEncodingDecoder.java b/reader/src/main/java/io/github/dfa1/vortex/reader/decode/DeltaEncodingDecoder.java index 91f3437d..3f34ccf6 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/decode/DeltaEncodingDecoder.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/decode/DeltaEncodingDecoder.java @@ -83,11 +83,13 @@ public Array decode(DecodeContext ctx) { long basesCap = SegmentBroadcast.capacity(basesSeg, elemBytes); long deltasCap = SegmentBroadcast.capacity(deltasSeg, elemBytes); + // The only row-scaled allocation: the output itself, off-heap and at the column's own + // width. Everything below is fixed-size scratch — one chunk's worth, cache-resident, + // reused across chunks. MemorySegment out = ctx.arena().allocate(rowCount * elemBytes); long[] chunkBases = new long[lanes]; long[] chunkDeltas = new long[FastLanes.CHUNK]; long[] chunkUndelta = new long[FastLanes.CHUNK]; - long[] untransposed = new long[FastLanes.CHUNK]; // Each chunk carries its own lane bases, so chunks are independent and only those // overlapping the requested window are reconstructed — reading the tail of a long @@ -99,16 +101,65 @@ public Array decode(DecodeContext ctx) { readElements(basesSeg, ptype, basesCap, chunk * lanes, lanes, chunkBases); readElements(deltasSeg, ptype, deltasCap, chunk * FastLanes.CHUNK, FastLanes.CHUNK, chunkDeltas); undeltaChunk(chunkDeltas, chunkBases, lanes, typeBits, mask, chunkUndelta); - for (int i = 0; i < FastLanes.CHUNK; i++) { - untransposed[FastLanes.transposeIndex(i)] = chunkUndelta[i]; + scatterChunk(out, ptype, chunkUndelta, chunk * FastLanes.CHUNK - offset, rowCount); + } + return array(ctx, ptype, rowCount, out.asReadOnly()); + } + + /// Untransposes one chunk straight into the output window. + /// + /// The value at in-chunk position `i` belongs at logical index + /// `base + FastLanes#transposeIndex(i)`, so untransposing and window-shifting happen in the + /// same store — no second chunk-sized buffer, and no separate pass to slice it. + /// + /// `base` is negative for the leading chunk of an offset-sliced array, and the trailing + /// chunk can run past `rowCount`. One unsigned comparison covers both: a negative index + /// reads as a huge unsigned value and fails the same test as an overrun. The stores are a + /// permutation scatter, so they never vectorize regardless, and the compare costs nothing + /// the untranspose was not already paying. The ptype switch is hoisted out of the loop so + /// each body stays uniform (CLAUDE.md hot-loop rule). + /// + /// @param out output segment of `rowCount` elements + /// @param ptype output element type + /// @param values one chunk of reconstructed values, in transposed order + /// @param base output index the chunk's logical position 0 maps to; may be negative + /// @param rowCount number of rows in the output window + private static void scatterChunk(MemorySegment out, PType ptype, long[] values, long base, long rowCount) { + switch (ptype) { + case I8, U8 -> { + for (int i = 0; i < FastLanes.CHUNK; i++) { + long at = base + FastLanes.transposeIndex(i); + if (Long.compareUnsigned(at, rowCount) < 0) { + out.set(ValueLayout.JAVA_BYTE, at, (byte) values[i]); + } + } } - // The window clips only the first and last chunk; the ones between copy whole. - long chunkStart = chunk * FastLanes.CHUNK; - int from = (int) Math.max(0, offset - chunkStart); - int to = (int) Math.min(FastLanes.CHUNK, offset + rowCount - chunkStart); - writeElements(out, ptype, chunkStart + from - offset, untransposed, from, to - from); + case I16, U16 -> { + for (int i = 0; i < FastLanes.CHUNK; i++) { + long at = base + FastLanes.transposeIndex(i); + if (Long.compareUnsigned(at, rowCount) < 0) { + out.setAtIndex(VortexFormat.LE_SHORT, at, (short) values[i]); + } + } + } + case I32, U32 -> { + for (int i = 0; i < FastLanes.CHUNK; i++) { + long at = base + FastLanes.transposeIndex(i); + if (Long.compareUnsigned(at, rowCount) < 0) { + out.setAtIndex(VortexFormat.LE_INT, at, (int) values[i]); + } + } + } + case I64, U64 -> { + for (int i = 0; i < FastLanes.CHUNK; i++) { + long at = base + FastLanes.transposeIndex(i); + if (Long.compareUnsigned(at, rowCount) < 0) { + out.setAtIndex(VortexFormat.LE_LONG, at, values[i]); + } + } + } + default -> throw new VortexException(EncodingId.FASTLANES_DELTA, "unsupported ptype: " + ptype); } - return array(ctx, ptype, rowCount, out); } /// Wraps a decoded segment in the `Materialized*Array` matching `ptype`. @@ -159,7 +210,7 @@ private static void undeltaChunk(long[] deltas, long[] bases, int lanes, int typ private static void readElements(MemorySegment buf, PType ptype, long cap, long firstIdx, int count, long[] out) { if (firstIdx + count <= cap) { - readContiguous(buf, ptype, firstIdx * ptype.byteSize(), count, out); + readContiguous(buf, ptype, firstIdx, count, out); return; } if (cap == 0) { @@ -169,41 +220,41 @@ private static void readElements(MemorySegment buf, PType ptype, long cap, long readBroadcast(buf, ptype, cap, firstIdx, count, out); } - private static void readContiguous(MemorySegment buf, PType ptype, long base, int count, long[] out) { + private static void readContiguous(MemorySegment buf, PType ptype, long from, int count, long[] out) { switch (ptype) { case I8 -> { for (int i = 0; i < count; i++) { - out[i] = buf.get(ValueLayout.JAVA_BYTE, base + i); + out[i] = buf.get(ValueLayout.JAVA_BYTE, from + i); } } case U8 -> { for (int i = 0; i < count; i++) { - out[i] = Byte.toUnsignedLong(buf.get(ValueLayout.JAVA_BYTE, base + i)); + out[i] = Byte.toUnsignedLong(buf.get(ValueLayout.JAVA_BYTE, from + i)); } } case I16 -> { for (int i = 0; i < count; i++) { - out[i] = buf.get(VortexFormat.LE_SHORT, base + i * 2L); + out[i] = buf.getAtIndex(VortexFormat.LE_SHORT, from + i); } } case U16 -> { for (int i = 0; i < count; i++) { - out[i] = Short.toUnsignedLong(buf.get(VortexFormat.LE_SHORT, base + i * 2L)); + out[i] = Short.toUnsignedLong(buf.getAtIndex(VortexFormat.LE_SHORT, from + i)); } } case I32 -> { for (int i = 0; i < count; i++) { - out[i] = buf.get(VortexFormat.LE_INT, base + i * 4L); + out[i] = buf.getAtIndex(VortexFormat.LE_INT, from + i); } } case U32 -> { for (int i = 0; i < count; i++) { - out[i] = Integer.toUnsignedLong(buf.get(VortexFormat.LE_INT, base + i * 4L)); + out[i] = Integer.toUnsignedLong(buf.getAtIndex(VortexFormat.LE_INT, from + i)); } } case I64, U64 -> { for (int i = 0; i < count; i++) { - out[i] = buf.get(VortexFormat.LE_LONG, base + i * 8L); + out[i] = buf.getAtIndex(VortexFormat.LE_LONG, from + i); } } default -> throw new VortexException(EncodingId.FASTLANES_DELTA, "unsupported ptype: " + ptype); @@ -233,41 +284,5 @@ private static long readOne(MemorySegment buf, PType ptype, long off) { }; } - /// Writes `src[from, from + count)` into `out` at element index `dstIdx`, narrowed to - /// `ptype`'s width. The ptype switch is hoisted out of the loop so each body stays uniform. - /// - /// @param out destination segment, at `ptype`'s width - /// @param ptype element type - /// @param dstIdx element index in `out` to write the first value at - /// @param src reconstructed values, widened to `long` - /// @param from first index in `src` to write - /// @param count number of elements to write - /// @throws VortexException if `ptype` is not an integer ptype - private static void writeElements(MemorySegment out, PType ptype, long dstIdx, long[] src, - int from, int count) { - switch (ptype) { - case I8, U8 -> { - for (int i = 0; i < count; i++) { - out.set(ValueLayout.JAVA_BYTE, dstIdx + i, (byte) src[from + i]); - } - } - case I16, U16 -> { - for (int i = 0; i < count; i++) { - out.setAtIndex(VortexFormat.LE_SHORT, dstIdx + i, (short) src[from + i]); - } - } - case I32, U32 -> { - for (int i = 0; i < count; i++) { - out.setAtIndex(VortexFormat.LE_INT, dstIdx + i, (int) src[from + i]); - } - } - case I64, U64 -> { - for (int i = 0; i < count; i++) { - out.setAtIndex(VortexFormat.LE_LONG, dstIdx + i, src[from + i]); - } - } - default -> throw new VortexException(EncodingId.FASTLANES_DELTA, "unsupported ptype: " + ptype); - } - } } diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/decode/DeltaEncodingDecoderTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/decode/DeltaEncodingDecoderTest.java index bfb0de9c..bcea81a7 100644 --- a/reader/src/test/java/io/github/dfa1/vortex/reader/decode/DeltaEncodingDecoderTest.java +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/decode/DeltaEncodingDecoderTest.java @@ -3,17 +3,23 @@ import io.github.dfa1.vortex.core.error.VortexException; import io.github.dfa1.vortex.core.model.DType; import io.github.dfa1.vortex.core.model.PType; +import io.github.dfa1.vortex.core.compute.FastLanes; +import io.github.dfa1.vortex.core.compute.PrimitiveArrays; import io.github.dfa1.vortex.core.model.EncodingId; import io.github.dfa1.vortex.core.io.VortexFormat; import io.github.dfa1.vortex.core.testing.TestSegments; import io.github.dfa1.vortex.core.proto.ProtoDeltaMetadata; import io.github.dfa1.vortex.reader.ReadRegistry; import io.github.dfa1.vortex.reader.array.Array; +import io.github.dfa1.vortex.reader.array.ByteArray; +import io.github.dfa1.vortex.reader.array.IntArray; import io.github.dfa1.vortex.reader.array.LongArray; +import io.github.dfa1.vortex.reader.array.ShortArray; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.ValueSource; import java.lang.foreign.Arena; import java.lang.foreign.MemorySegment; @@ -51,6 +57,171 @@ void decode_nullMetadata_returnsEmptyArray(PType ptype) { assertThat(result.length()).isZero(); } + /// Round-trips a known sequence through the wire form, for every integer width. The values + /// step by a per-lane amount so the prefix sum is non-trivial and a lane mix-up shows up. + /// + /// The previous decoder staged this through four row-scaled heap `long[]`s (bases, deltas, + /// a full-length `decoded`, and a `result` slice of it), widening every value to 8 bytes + /// whatever the column's width; values now go straight into one arena segment at the + /// column's own width (#338). The reconstruction is what must not change. + @ParameterizedTest + @EnumSource(value = PType.class, names = {"I8", "I16", "I32", "I64", "U8", "U16", "U32", "U64"}) + void decode_roundTripsASingleChunk(PType ptype) { + // Given — 1024 values, small enough to survive I8's 8-bit width + long[] values = new long[FL_CHUNK_SIZE]; + for (int i = 0; i < values.length; i++) { + values[i] = (i * 3) & 0x3F; + } + + // When + LongArray result = decodeDelta(ptype, values, 0, values.length); + + // Then + assertValues(result, values, 0, values.length); + } + + /// Multi-chunk: the per-chunk scratch is reused across iterations, so a chunk boundary is + /// where a stale-scratch or wrong-base bug would surface. Two chunks plus a partial third. + @Test + void decode_roundTripsAcrossChunkBoundaries() { + // Given + long[] values = new long[FL_CHUNK_SIZE * 2 + 100]; + for (int i = 0; i < values.length; i++) { + values[i] = i * 7L; + } + + // When + LongArray result = decodeDelta(PType.I64, values, 0, values.length); + + // Then + assertValues(result, values, 0, values.length); + } + + /// A non-zero `offset` slices the decoded values, and nothing covered it before. It is the + /// sharp edge of writing chunks straight into the output: the leading chunk now maps to a + /// negative output index and the trailing chunk runs past the row count, both of which the + /// scatter has to drop rather than write out of bounds. The encoder always emits offset 0, + /// so this shape only arrives from a sliced array written elsewhere. + @ParameterizedTest + @ValueSource(ints = {1, 7, 1023, 1024, 1025, 2000}) + void decode_offsetSlicesTheWindow(int offset) { + // Given + long[] values = new long[FL_CHUNK_SIZE * 3]; + for (int i = 0; i < values.length; i++) { + values[i] = i * 11L; + } + int rowCount = 500; + + // When + LongArray result = decodeDelta(PType.I64, values, offset, rowCount); + + // Then — rows are values[offset .. offset + rowCount) + assertValues(result, values, offset, rowCount); + } + + /// The window may stop short of the chunk it lands in, so the trailing chunk is only + /// partially written. Rows past `rowCount` must not be stored at all. + @Test + void decode_rowCountShorterThanTheDecodedLength() { + // Given + long[] values = new long[FL_CHUNK_SIZE * 2]; + for (int i = 0; i < values.length; i++) { + values[i] = i * 5L; + } + + // When + LongArray result = decodeDelta(PType.I64, values, 0, 3); + + // Then + assertThat(result.length()).isEqualTo(3L); + assertValues(result, values, 0, 3); + } + + private static void assertValues(LongArray actual, long[] expected, int offset, int count) { + assertThat(actual.length()).isEqualTo((long) count); + for (int i = 0; i < count; i++) { + assertThat(actual.getLong(i)).as("row %d", i).isEqualTo(expected[offset + i]); + } + } + + /// Decodes `values` through the `fastlanes.delta` wire form, mirroring + /// `DeltaEncodingEncoder`'s transpose-then-per-lane-delta layout. Built here rather than + /// called: the writer module is not on the reader's test classpath, and the encoder never + /// emits a non-zero `offset`, which is precisely the case worth covering. + private static LongArray decodeDelta(PType ptype, long[] values, int offset, int rowCount) { + int lanes = FastLanes.lanes(ptype); + int typeBits = ptype.bits(); + long mask = FastLanes.lowMask(typeBits); + int numChunks = (values.length + FastLanes.CHUNK - 1) / FastLanes.CHUNK; + long paddedLen = (long) numChunks * FastLanes.CHUNK; + + long[] basesAll = new long[numChunks * lanes]; + long[] deltasAll = new long[(int) paddedLen]; + long[] transposed = new long[FastLanes.CHUNK]; + + for (int chunk = 0; chunk < numChunks; chunk++) { + long[] chunkBuf = new long[FastLanes.CHUNK]; + int start = chunk * FastLanes.CHUNK; + int end = Math.min(start + FastLanes.CHUNK, values.length); + for (int i = start; i < end; i++) { + chunkBuf[i - start] = values[i] & mask; + } + for (int i = 0; i < FastLanes.CHUNK; i++) { + transposed[i] = chunkBuf[FastLanes.transposeIndex(i)]; + } + System.arraycopy(transposed, 0, basesAll, chunk * lanes, lanes); + for (int lane = 0; lane < lanes; lane++) { + long prev = basesAll[chunk * lanes + lane] & mask; + for (int row = 0; row < typeBits; row++) { + int idx = FastLanes.iterateIndex(row, lane); + long next = transposed[idx] & mask; + deltasAll[chunk * FastLanes.CHUNK + idx] = (next - prev) & mask; + prev = next; + } + } + } + + MemorySegment meta = MemorySegment.ofArray(new ProtoDeltaMetadata(paddedLen, offset).encode()); + ArrayNode bases = new ArrayNode(EncodingId.VORTEX_PRIMITIVE, null, new ArrayNode[0], new int[]{0}); + ArrayNode deltas = new ArrayNode(EncodingId.VORTEX_PRIMITIVE, null, new ArrayNode[0], new int[]{1}); + ArrayNode node = new ArrayNode(EncodingId.FASTLANES_DELTA, meta, new ArrayNode[]{bases, deltas}, new int[0]); + + MemorySegment[] segs = {toSegment(basesAll, ptype), toSegment(deltasAll, ptype)}; + DecodeContext ctx = new DecodeContext(node, new DType.Primitive(ptype, false), rowCount, segs, + REGISTRY, Arena.ofAuto()); + Array decoded = SUT.decode(ctx); + return new WidenedLongView(decoded); + } + + private static MemorySegment toSegment(long[] longs, PType ptype) { + return PrimitiveArrays.fromLongs(longs, ptype, Arena.ofAuto()); + } + + /// Reads any narrow decoded array as `long` so one assertion helper covers every width. + private record WidenedLongView(Array inner) implements LongArray { + + @Override + public DType dtype() { + return inner.dtype(); + } + + @Override + public long length() { + return inner.length(); + } + + @Override + public long getLong(long i) { + return switch (inner) { + case ByteArray ba -> ba.getInt(i); + case ShortArray sa -> sa.getInt(i); + case IntArray ia -> ia.getInt(i); + case LongArray la -> la.getLong(i); + default -> throw new IllegalStateException("unexpected array type " + inner.getClass()); + }; + } + } + @Test void decode_constantChildren_broadcastsAcrossChunk() { // Given a single delta chunk (1024 rows) whose bases and deltas children each hold