feat(BitpackedCodec): apply patches on decode (TODO #7a step 3)

Per the FastLanes BitPacked spec, BitPackedMetadata carries an optional
patches PatchesMetadata field (tag 3) and the array exposes two child
slots — slot 0 for patch indices (unsigned, ptype from
PatchesMetadata.indices_ptype), slot 1 for patch values (parent dtype).
Until now the Java decoder ignored both: any column the Rust writer
encoded with bitpacked-plus-patches lost its outlier values.

encodings.proto: add `optional PatchesMetadata patches = 3` to
BitPackedMetadata. The field was previously dropped silently because
protobuf treats unknown fields as opaque.

BitpackedCodec.decode: when `meta.hasPatches()`, decode child 0 as the
indices array (its own ptype, rowCount = patches.len), decode child 1
as the values array (parent dtype, rowCount = patches.len), and
overwrite `output[absIdx - offset]` with the value bytes for each
patch. Slot ordering verified against
spiraldb/vortex `encodings/fastlanes/src/bitpacking/array/mod.rs`
(BitPackedSlots: patch_indices=0, patch_values=1).

BitpackedCodecPatchesTest: bit-pack [10,20,30,40,50] via the
production encoder (bitWidth=6), attach synthetic PatchesMetadata
that rewrites indices {1,3} with values {777,999}, and assert the
decoded buffer is [10,777,30,999,50].

TODO.md: move #7a to Done. Trim #7 to pcodec — bitpacked, delta, for,
sparse, alp, dict, fsst, sequence, varbin, constant, and runend all
landed.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: dfa1

TODO.md

@@ -11,6 +11,10 @@
     - Writer embeds min/max per flat segment in `ArrayNode.stats` (Protobuf ScalarValue)
     - Reader peeks stats from segment FlatBuffer tail, skips chunks excluded by `RowFilter`
     - Supports `Gte`, `Lte`, `Eq`, `And` predicates on I8/I16/I32/I64, U8/U16/U32/U64, F32/F64
+- [x] **#7a `fastlanes.bitpacked` — spec-compliant rewrite** (all 5 steps incl. patches)
+    - Protobuf `BitPackedMetadata` (tags 1+2+3); unified FastLanes `unpack` for I8/I16/I32/I64 (signed + unsigned);
+      patches decoded from children slots 0 (indices) + 1 (values), overwritten by absolute index.
+    - Encoder writes spec-compliant protobuf metadata; round-trip + patch decode covered by tests.
 - [x] **#7b `fastlanes.for` decoder** (reference + bitpacked residuals)
 - [x] **#7c `vortex.sparse` decoder** (fill value + patches)
 - [x] **#7d `vortex.alp` decoder** (ALP inverse + patches)
@@ -24,29 +28,8 @@
 ## Open
 
 - [ ] **#7 Additional encodings**
-    - `fastlanes.bitpacked` — integer bit-packing (steps 1/2/4/5 done; step 3 patches still open, see #7a)
-    - `fastlanes.delta` — delta encoding for monotonic sequences
-    - `dict` — dictionary encoding for low-cardinality columns
-    - `pcodec` — float compression
-    - `fsst` — string compression (in-flight, see `FsstCodec.java`)
-
-- [ ] **#7a Fix `fastlanes.bitpacked` — spec-compliant rewrite**
-    - Root cause: current code guesses format by metadata byte size (9 = Java, 2 = JNI). Wrong.
-      The spec always uses protobuf metadata regardless of writer origin.
-    - **Spec** (from `encodings/fastlanes/src/bitpacking/vtable/mod.rs`):
-        - Metadata: protobuf `BitPackedMetadata` — `bit_width u32` (tag 1), `offset u32` (tag 2, 0≤offset<1024),
-          `patches PatchesMetadata` (tag 3, optional)
-        - Buffer size: `ceil((len + offset) / 1024) * 128 * bit_width` bytes
-        - FastLanes block layout: `LANES = 1024 / T` (T = element bit-width); `FL_ORDER = [0,4,2,6,1,5,3,7]`; logical
-          index for `(row, lane)` = `FL_ORDER[row/8]*16 + (row%8)*128 + lane` — same formula for all types
-    - Step 1: delete `decodeJni()` and `decodeJava()`; parse metadata as protobuf (tags 1+2)
-    - Step 2: implement single unified `unpack(buf, bitWidth, offset, T, rowCount) → long[]` using the FastLanes
-      algorithm above
-    - Step 3: handle patches — decode child slots (indices + values), overwrite output at patch indices (**missing**; steps 1,2,4,5 done)
-    - Step 4: align `encode()` to write protobuf metadata (tags 1+2) instead of the 9-byte custom format
-    - Step 5: update `BitpackedCodecTest` for round-trip with spec-compliant metadata
-    - Reference: `spiraldb/vortex` `encodings/fastlanes/src/bitpacking/`, `spiraldb/fastlanes-rs` `src/bitpacking.rs` +
-      `src/macros.rs`
+    - `pcodec` — float compression (only remaining gap; bitpacked, delta, for, sparse, alp, dict, fsst, sequence,
+      varbin, constant, runend all landed)
 
 ## Performance
 

core/src/main/java/io/github/dfa1/vortex/encoding/BitpackedCodec.java

@@ -1,6 +1,7 @@
 package io.github.dfa1.vortex.encoding;
 
 import com.google.protobuf.InvalidProtocolBufferException;
+import dev.vortex.proto.DTypeProtos;
 import dev.vortex.proto.EncodingProtos;
 import dev.vortex.proto.ScalarProtos;
 import io.github.dfa1.vortex.core.Array;
@@ -578,7 +579,61 @@ public Array decode(DecodeContext ctx) {
 		MemorySegment output = ctx.arena().allocate(rowCount * ptype.byteSize());
 		fastlanesUnpackToSeg(packed, bitWidth, offset, typeBits, rowCount, output);
 
+		if (meta.hasPatches()) {
+			applyPatches(ctx, meta.getPatches(), output, ptype.byteSize());
+		}
+
 		return new Array(ctx.dtype(), rowCount,
 				new MemorySegment[]{output.asReadOnly()}, Array.NO_CHILDREN, ArrayStats.empty());
 	}
+
+	// ── Patches ───────────────────────────────────────────────────────────────
+
+	private static Array decodeChildAs(DecodeContext parent, int childIdx, DType dtype, long rowCount) {
+		ArrayNode childNode = parent.node().children()[childIdx];
+		DecodeContext childCtx = new DecodeContext(
+				childNode, dtype, rowCount, parent.segmentBuffers(), parent.registry(), parent.arena());
+		return parent.registry().decode(childCtx);
+	}
+
+	private static long readUnsignedIdx(MemorySegment seg, long i, PType ptype) {
+		return switch (ptype) {
+			case U8 -> Byte.toUnsignedLong(seg.get(ValueLayout.JAVA_BYTE, i));
+			case U16 -> Short.toUnsignedLong(seg.get(LE_SHORT, i * 2));
+			case U32 -> Integer.toUnsignedLong(seg.get(LE_INT, i * 4));
+			case U64 -> seg.get(LE_LONG, i * 8);
+			default -> throw new IllegalStateException(
+					"fastlanes.bitpacked: non-unsigned patch index ptype " + ptype);
+		};
+	}
+
+	private static PType ptypeFromProto(DTypeProtos.PType proto) {
+		return PType.values()[proto.getNumber()];
+	}
+
+	private static void applyPatches(DecodeContext ctx, EncodingProtos.PatchesMetadata pm,
+	                                 MemorySegment out, int elemBytes) {
+		long numPatches = pm.getLen();
+		if (numPatches == 0) {
+			return;
+		}
+		long offset = pm.getOffset();
+		PType idxPtype = ptypeFromProto(pm.getIndicesPtype());
+
+		Array idxArr = decodeChildAs(ctx, 0, new DType.Primitive(idxPtype, false), numPatches);
+		Array valArr = decodeChildAs(ctx, 1, ctx.dtype(), numPatches);
+
+		MemorySegment idxSeg = idxArr.buffer(0);
+		MemorySegment valSeg = valArr.buffer(0);
+
+		long n = ctx.rowCount();
+		for (long i = 0; i < numPatches; i++) {
+			long absIdx = readUnsignedIdx(idxSeg, i, idxPtype) - offset;
+			if (absIdx < 0 || absIdx >= n) {
+				throw new IllegalStateException(
+						"fastlanes.bitpacked: patch index " + absIdx + " out of range [0," + n + ")");
+			}
+			MemorySegment.copy(valSeg, i * elemBytes, out, absIdx * elemBytes, elemBytes);
+		}
+	}
 }

core/src/main/proto/encodings.proto

@@ -44,6 +44,7 @@ message ALPMetadata {
 message BitPackedMetadata {
   uint32 bit_width = 1;
   uint32 offset = 2;
+  optional PatchesMetadata patches = 3;
 }
 
 message SequenceMetadata {

core/src/test/java/io/github/dfa1/vortex/encoding/BitpackedCodecPatchesTest.java

@@ -0,0 +1,88 @@
+package io.github.dfa1.vortex.encoding;
+
+import dev.vortex.proto.DTypeProtos;
+import dev.vortex.proto.EncodingProtos;
+import io.github.dfa1.vortex.core.Array;
+import io.github.dfa1.vortex.core.ArrayStats;
+import io.github.dfa1.vortex.core.DType;
+import io.github.dfa1.vortex.core.PType;
+import org.junit.jupiter.api.Test;
+
+import java.lang.foreign.Arena;
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class BitpackedCodecPatchesTest {
+
+	private static final DType I32_DTYPE = new DType.Primitive(PType.I32, false);
+	private static final ValueLayout.OfInt LE_INT = ValueLayout.JAVA_INT_UNALIGNED.withOrder(ByteOrder.LITTLE_ENDIAN);
+
+	@Test
+	void decode_appliesPatches_overridingBitPackedValues() {
+		// Given — bit-pack [10, 20, 30, 40, 50] via the production encoder (bitWidth = 6),
+		// then attach synthetic patches metadata that rewrites indices [1, 3] with [777, 999].
+		int[] base = {10, 20, 30, 40, 50};
+		BitpackedCodec sut = new BitpackedCodec();
+		EncodeResult packed = sut.encode(I32_DTYPE, base);
+
+		ByteBuffer packedBuf = packed.buffers().get(0);
+		byte[] packedBytes = new byte[packedBuf.remaining()];
+		packedBuf.duplicate().get(packedBytes);
+
+		// Build new BitPackedMetadata that re-uses the packed bytes but advertises patches.
+		EncodingProtos.PatchesMetadata patches = EncodingProtos.PatchesMetadata.newBuilder()
+				.setLen(2)
+				.setOffset(0)
+				.setIndicesPtype(DTypeProtos.PType.U32)
+				.build();
+		byte[] metaBytes = EncodingProtos.BitPackedMetadata.newBuilder()
+				.setBitWidth(6) // matches encoder's choice for max=50
+				.setOffset(0)
+				.setPatches(patches)
+				.build()
+				.toByteArray();
+
+		byte[] idxBuf = new byte[2 * 4];
+		ByteBuffer.wrap(idxBuf).order(ByteOrder.LITTLE_ENDIAN).putInt(1).putInt(3);
+		byte[] valBuf = new byte[2 * 4];
+		ByteBuffer.wrap(valBuf).order(ByteOrder.LITTLE_ENDIAN).putInt(777).putInt(999);
+
+		ArrayNode idxNode = new ArrayNode(CodecId.VORTEX_PRIMITIVE, null,
+				new ArrayNode[0], new int[]{1}, ArrayStats.empty());
+		ArrayNode valNode = new ArrayNode(CodecId.VORTEX_PRIMITIVE, null,
+				new ArrayNode[0], new int[]{2}, ArrayStats.empty());
+		ArrayNode bpNode = new ArrayNode(CodecId.FASTLANES_BITPACKED,
+				ByteBuffer.wrap(metaBytes),
+				new ArrayNode[]{idxNode, valNode},
+				new int[]{0},
+				ArrayStats.empty());
+
+		MemorySegment[] segments = {
+				MemorySegment.ofArray(packedBytes),
+				MemorySegment.ofArray(idxBuf),
+				MemorySegment.ofArray(valBuf)
+		};
+
+		CodecRegistry registry = CodecRegistry.empty();
+		registry.register(new BitpackedCodec());
+		registry.register(new PrimitiveCodec());
+
+		DecodeContext ctx = new DecodeContext(
+				bpNode, I32_DTYPE, base.length, segments, registry, Arena.global());
+
+		// When
+		Array result = sut.decode(ctx);
+
+		// Then
+		assertThat(result.length()).isEqualTo(base.length);
+		assertThat(result.buffer(0).get(LE_INT, 0L)).isEqualTo(10);
+		assertThat(result.buffer(0).get(LE_INT, 4L)).isEqualTo(777);
+		assertThat(result.buffer(0).get(LE_INT, 8L)).isEqualTo(30);
+		assertThat(result.buffer(0).get(LE_INT, 12L)).isEqualTo(999);
+		assertThat(result.buffer(0).get(LE_INT, 16L)).isEqualTo(50);
+	}
+}