perf(reader): untranspose fastlanes.delta straight into the output window

Takes the better half of #343, which fixed #338 independently and in
parallel — I merged #345 for the same issue without checking for an open
PR first, so this reconciles the two rather than discarding one.

From #343:
- scatterChunk writes each untransposed value straight to its output
  index, dropping the chunk-sized `untransposed` buffer #345 staged it
  in and the separate pass that sliced it. The leading chunk of an
  offset-sliced array maps to a negative index and the trailing chunk
  runs past the row count; one `Long.compareUnsigned` covers both, since
  a negative index reads as a huge unsigned value. The stores are a
  permutation scatter and never vectorize regardless, so the compare
  costs nothing the untranspose was not already paying.
- The returned segment is read-only.
- Element-indexed `getAtIndex` instead of hand-computed byte offsets.
- Reader-module tests that build the delta wire form directly, mirroring
  DeltaEncodingEncoder's layout. These cover offset slicing where it
  belongs — the writer is not on the reader's test classpath and never
  emits a non-zero offset, so #345 had to reach into the integration
  module to cover the same shape.

Kept from #345:
- Only chunks overlapping the row window are reconstructed. #343 walked
  every chunk and discarded the out-of-window stores per element, so a
  one-chunk slice of a thousand-chunk column did a thousand chunks of
  work.
- The metadata range guard. Without it a `deltas_len` of Long.MAX_VALUE
  drives the chunk loop ~9e15 times — a hang, which is worse than the
  OutOfMemoryError it replaced.

Closes #343.

Co-Authored-By: Davide Angelocola <davide.angelocola@gmail.com>
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: dfa1

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);
-        }
-    }
 
 }

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