feat(encoding): cascading compressor with ALP+Bitpacked, FoR+Bitpacked, Dict+Bitpacked

Implements BtrBlocks-style cascading compression (TODO #13):
- CompressorContext, ChildSlot, CascadeStep infrastructure
- Encoding.encodeCascade default + overrides in AlpEncoding, FoR, DictEncoding
- CascadingCompressor: sample-based winner selection + buffer splicing
- WriteOptions.cascading(depth) factory; VortexWriter branches at depth>0
- DictEncoding.decodeLegacyJava updated to decode codes via registry
  (required for bitpacked codes in cascade format)
- VortexVsCsvFileSizeTest: assert Vortex < CSV at depth=2

Result: 100K-row OHLC  Vortex=4.3MB vs CSV=4.6MB (1.08× ratio)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: dfa1

TODO.md

@@ -16,90 +16,14 @@
     - Capture JMH JSON + JFR profile alongside README table; cite hardware (CPU model), JDK build (`java -version`),
       and benchmark commit SHA so numbers don't rot silently.
 
-- [ ] **#13 BtrBlocks-style cascading compressor** (`core`, `writer`)
-    - Reference: https://vortex.dev/blog/btrblocks-compressor
-    - Rust source: `vortex-sampling-compressor` crate in spiraldb/vortex
-    - **Scope (first cut)**: integers + floats only. Strings stay on current first-match.
-    - **Rollout**: opt-in via `WriteOptions.cascading(int allowedCascading)`. Default `allowedCascading=0`
-      preserves today's `findEncoding()` first-match behaviour so existing tests are unaffected.
-    - Exclusion rule: don't apply dict encode to dict encode, FoR to FoR, etc. — encoding ids exclude
-      themselves on recursion via `CompressorContext.excluded`.
-
-    **Sub-tasks**
-
-    - [ ] **a. `CompressorContext` record** (`core/encoding`)
-        - Fields: `int allowedCascading`, `Set<EncodingId> excluded`, `long sampleSeed`,
-          `int minSampleSize` (default 1024), `double sampleFraction` (default 0.01).
-        - Immutable; `withDecrementedDepth()` and `withExcluded(EncodingId)` helpers.
-
-    - [ ] **b. Cascade-aware encode API** (`core/encoding/Encoding.java`)
-        - Add default method:
-          ```java
-          default CascadeStep encodeCascade(DType dtype, Object data, CompressorContext ctx) {
-              EncodeResult r = encode(dtype, data);
-              return CascadeStep.terminal(r);
-          }
-          ```
-        - New record `CascadeStep(EncodeNode partialRoot, List<ByteBuffer> ownedBuffers,
-          List<ChildSlot> openChildren, byte[] statsMin, byte[] statsMax)`.
-        - `ChildSlot(DType childDtype, Object childData, int parentChildIdx)` — slot the
-          compressor recursively encodes, then splices result into `partialRoot.children[parentChildIdx]`
-          while remapping its `bufferIndices` to follow `ownedBuffers`.
-        - Encodings that don't expose intermediates use the default (terminal); they get tried as
-          leaf candidates only.
-
-    - [ ] **c. Refactor cascading-friendly encodings to emit children**
-        - `AlpEncoding`: F64 → `(metadata{expE,expF,patches}, child long[] encodedInts)` +
-          optional patch children. Today's hard-coded `EncodeNode.leaf(VORTEX_PRIMITIVE, 0)` for
-          the I64 child becomes an open `ChildSlot(I64, encodedArr)`.
-        - `DictEncoding`: codes child (currently emitted as `VORTEX_PRIMITIVE` leaf) becomes an
-          open `ChildSlot(unsigned int, codes[])` so cascade can bit-pack it. Values child stays
-          primitive (or recurses too if `allowedCascading>0`).
-        - `FrameOfReferenceEncoding`: **implement encode** (currently decode-only, throws on encode).
-          Produces `(reference, child long[] deltas)` → `ChildSlot(I64, deltas)`. Pure intermediate;
-          requires `allowedCascading>0` because raw FoR without downstream pack is no win.
-        - `DeltaEncoding`: refactor to emit `(bases child, deltas child)` matching Rust's 2-child
-          wire format (currently 1-buffer flat — incompatible with Rust reader, see commit 09685a2).
-          Blocks cross-compat for delta cascades; track as sub-task c.4, can land separately.
-        - `BitpackedEncoding`: stays terminal (no open children). Patch indices/values already handled.
-        - `PrimitiveEncoding`: terminal.
-
-    - [ ] **d. `CascadingCompressor`** (`core/encoding`)
-        - Constructor: `(List<Encoding> encodings, CompressorContext rootCtx)`.
-        - `EncodeResult encode(DType dtype, Object data)` algorithm:
-            1. Build sample: stratified contiguous slices, total ≈ `max(minSampleSize,
-               sampleFraction * n)`. Fixed-seed `Random` for reproducibility.
-            2. Build candidate list: every encoding where `accepts(dtype)` and id ∉ `ctx.excluded`.
-            3. For each candidate: invoke `encodeCascade(dtype, sample, ctx)`. If the step has
-               open children, recurse on each with `ctx.withDecrementedDepth().withExcluded(thisEncoding.id())`.
-               Sum total byte size across all owned buffers.
-            4. Pick winner by smallest total size; require ratio ≥ 1.0 vs raw primitive baseline,
-               else fall through to `PrimitiveEncoding`.
-            5. Re-run winner on **full** data (sample-time stats are estimates only).
-        - Splicing: when a child recursion returns its own `EncodeResult`, append its buffers to
-          the parent's flat list and remap the child's `bufferIndices` by the offset.
-        - Terminal stop: when `ctx.allowedCascading == 0`, only encodings whose default-API
-          `encodeCascade` returns no open children are considered.
-
-    - [ ] **e. `WriteOptions.cascading(int)`** (`writer/WriteOptions.java`)
-        - Add field `int allowedCascading` (default `0`).
-        - `WriteOptions.cascading(int n)` factory mirrors `defaults()`.
-        - `VortexWriter.writeSegment` (writer/VortexWriter.java:212) branches:
-          if `options.allowedCascading() > 0`, use `CascadingCompressor`; else current `findEncoding()`.
-
-    - [ ] **f. Tests** (`core` + `writer` + `integration`)
-        - Unit: `CascadingCompressorTest` — sample stratification, exclusion list propagation,
-          ratio-based selection, depth=0 terminates, depth>0 picks cascade.
-        - Round-trip: each cascade combo (ALP+Bitpacked F64, FoR+Bitpacked I64, Dict+Bitpacked I32,
-          Delta+Bitpacked I64) round-trips bit-exact through `VortexReader`.
-        - Cross-compat: `JavaWritesRustReadsIntegrationTest` runs with `allowedCascading=3`,
-          confirms Rust JNI reader decodes the same OHLC data (gates DeltaEncoding wire-format fix).
-        - `WriteFileSizeIntegrationTest`: assert Java file ≤ JNI file size with cascade on.
-
-    - [ ] **g. Benchmark**
-        - Extend `RustVsJavaWriteBenchmark` (TODO #10b) with cascading-off vs cascading-on variants.
-        - Capture: ratio of compressed bytes (Java cascade / JNI), write throughput, decode
-          throughput on the resulting file.
+- [ ] **#13 Cascading compressor — remaining follow-ups**
+    - [ ] **DeltaEncoding wire-format fix**: refactor to emit `(bases child, deltas child)` matching
+      Rust's 2-child wire format (currently 1-buffer flat — incompatible with Rust reader, see commit 09685a2).
+      Unblocks cross-compat for delta cascades.
+    - [ ] **Cross-compat test**: `JavaWritesRustReadsIntegrationTest` with `allowedCascading=3`
+      confirms Rust JNI reader decodes the same OHLC data (gates DeltaEncoding wire-format fix).
+    - [ ] **Benchmark**: extend `RustVsJavaWriteBenchmark` with cascading-off vs cascading-on variants;
+      capture compressed-bytes ratio (Java cascade / JNI), write throughput, decode throughput.
 
 ## Tooling
 

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

@@ -105,8 +105,21 @@ public EncodeResult encode(DType dtype, Object data) {
 		throw new UnsupportedOperationException("ALP encode not yet supported for " + ptype);
 	}
 
+	@Override
+	public CascadeStep encodeCascade(DType dtype, Object data, CompressorContext ctx) {
+		PType ptype = ((DType.Primitive) dtype).ptype();
+		if (ptype == PType.F64) {
+			return encodeCascadeF64((double[]) data);
+		}
+		return CascadeStep.terminal(encode(dtype, data));
+	}
+
 	// ── F64 encode ────────────────────────────────────────────────────────────
 
+	private record AlpF64Data(int expE, int expF, long[] encodedArr,
+	                          List<Integer> patchIndices, List<Double> patchValues,
+	                          byte[] statsMin, byte[] statsMax) {}
+
 	private static int[] findExponentsF64(double[] values) {
 		int sampleLen = Math.min(SAMPLE_SIZE, values.length);
 		int bestExpE = 0, bestExpF = 0, bestExceptions = sampleLen + 1;
@@ -136,7 +149,7 @@ private static int[] findExponentsF64(double[] values) {
 		return new int[]{bestExpE, bestExpF};
 	}
 
-	private static EncodeResult encodeF64(double[] values, DType dtype) {
+	private static AlpF64Data computeF64(double[] values) {
 		int n = values.length;
 		int[] exps = findExponentsF64(values);
 		int expE = exps[0], expF = exps[1];
@@ -169,45 +182,92 @@ private static EncodeResult encodeF64(double[] values, DType dtype) {
 
 		byte[] statsMin = n > 0 ? scalarF64(min) : null;
 		byte[] statsMax = n > 0 ? scalarF64(max) : null;
+		return new AlpF64Data(expE, expF, encodedArr, patchIndices, patchValues, statsMin, statsMax);
+	}
+
+	private static EncodeResult encodeF64(double[] values, DType dtype) {
+		AlpF64Data d = computeF64(values);
+		int n = values.length;
 
 		MemorySegment encodedBuf = Arena.ofAuto().allocate((long) n * 8, 8);
 		for (int i = 0; i < n; i++) {
-			encodedBuf.setAtIndex(PTypeIO.LE_LONG, i, encodedArr[i]);
+			encodedBuf.setAtIndex(PTypeIO.LE_LONG, i, d.encodedArr()[i]);
 		}
 
 		EncodeNode encodedNode = EncodeNode.leaf(EncodingId.VORTEX_PRIMITIVE, 0);
 
-		if (patchIndices.isEmpty()) {
+		if (d.patchIndices().isEmpty()) {
 			byte[] metaBytes = EncodingProtos.ALPMetadata.newBuilder()
-					.setExpE(expE).setExpF(expF).build().toByteArray();
+					.setExpE(d.expE()).setExpF(d.expF()).build().toByteArray();
 			EncodeNode root = new EncodeNode(EncodingId.VORTEX_ALP,
 					ByteBuffer.wrap(metaBytes), new EncodeNode[]{encodedNode}, new int[0]);
-			return new EncodeResult(root, List.of(encodedBuf), statsMin, statsMax);
+			return new EncodeResult(root, List.of(encodedBuf), d.statsMin(), d.statsMax());
 		}
 
-		int numPatches = patchIndices.size();
+		int numPatches = d.patchIndices().size();
 		MemorySegment idxBuf = Arena.ofAuto().allocate((long) numPatches * 4, 4);
 		MemorySegment valBuf = Arena.ofAuto().allocate((long) numPatches * 8, 8);
 		for (int i = 0; i < numPatches; i++) {
-			idxBuf.setAtIndex(PTypeIO.LE_INT, i, patchIndices.get(i));
-			valBuf.setAtIndex(PTypeIO.LE_DOUBLE, i, patchValues.get(i));
+			idxBuf.setAtIndex(PTypeIO.LE_INT, i, d.patchIndices().get(i));
+			valBuf.setAtIndex(PTypeIO.LE_DOUBLE, i, d.patchValues().get(i));
 		}
 
-		EncodingProtos.PatchesMetadata patches = EncodingProtos.PatchesMetadata.newBuilder()
-				.setLen(numPatches)
-				.setOffset(0)
-				.setIndicesPtype(DTypeProtos.PType.forNumber(PType.U32.ordinal()))
-				.build();
+		EncodingProtos.PatchesMetadata patches = buildPatchesMeta(numPatches);
 		byte[] metaBytes = EncodingProtos.ALPMetadata.newBuilder()
-				.setExpE(expE).setExpF(expF).setPatches(patches).build().toByteArray();
+				.setExpE(d.expE()).setExpF(d.expF()).setPatches(patches).build().toByteArray();
 
 		EncodeNode idxNode = EncodeNode.leaf(EncodingId.VORTEX_PRIMITIVE, 1);
 		EncodeNode valNode = EncodeNode.leaf(EncodingId.VORTEX_PRIMITIVE, 2);
 		EncodeNode root = new EncodeNode(EncodingId.VORTEX_ALP,
 				ByteBuffer.wrap(metaBytes),
 				new EncodeNode[]{encodedNode, idxNode, valNode},
 				new int[0]);
-		return new EncodeResult(root, List.of(encodedBuf, idxBuf, valBuf), statsMin, statsMax);
+		return new EncodeResult(root, List.of(encodedBuf, idxBuf, valBuf), d.statsMin(), d.statsMax());
+	}
+
+	/// Cascade-aware F64 encode: I64 child is an open ChildSlot so the compressor
+	/// can bitpack it. Patch idx/val buffers (if any) stay in ownedBuffers.
+	private static CascadeStep encodeCascadeF64(double[] values) {
+		AlpF64Data d = computeF64(values);
+		int n = values.length;
+
+		if (d.patchIndices().isEmpty()) {
+			byte[] metaBytes = EncodingProtos.ALPMetadata.newBuilder()
+					.setExpE(d.expE()).setExpF(d.expF()).build().toByteArray();
+			// children[0] will be filled by the compressor after recursing on the ChildSlot
+			EncodeNode partialRoot = new EncodeNode(EncodingId.VORTEX_ALP,
+					ByteBuffer.wrap(metaBytes), new EncodeNode[1], new int[0]);
+			ChildSlot slot = new ChildSlot(I64_DTYPE, d.encodedArr(), 0);
+			return new CascadeStep(partialRoot, List.of(), List.of(slot), d.statsMin(), d.statsMax());
+		}
+
+		int numPatches = d.patchIndices().size();
+		MemorySegment idxBuf = Arena.ofAuto().allocate((long) numPatches * 4, 4);
+		MemorySegment valBuf = Arena.ofAuto().allocate((long) numPatches * 8, 8);
+		for (int i = 0; i < numPatches; i++) {
+			idxBuf.setAtIndex(PTypeIO.LE_INT, i, d.patchIndices().get(i));
+			valBuf.setAtIndex(PTypeIO.LE_DOUBLE, i, d.patchValues().get(i));
+		}
+
+		EncodingProtos.PatchesMetadata patches = buildPatchesMeta(numPatches);
+		byte[] metaBytes = EncodingProtos.ALPMetadata.newBuilder()
+				.setExpE(d.expE()).setExpF(d.expF()).setPatches(patches).build().toByteArray();
+
+		// ownedBuffers[0]=idxBuf, [1]=valBuf; child I64 will be appended after (starting at idx 2)
+		EncodeNode idxNode = EncodeNode.leaf(EncodingId.VORTEX_PRIMITIVE, 0);
+		EncodeNode valNode = EncodeNode.leaf(EncodingId.VORTEX_PRIMITIVE, 1);
+		EncodeNode partialRoot = new EncodeNode(EncodingId.VORTEX_ALP,
+				ByteBuffer.wrap(metaBytes), new EncodeNode[]{null, idxNode, valNode}, new int[0]);
+		ChildSlot slot = new ChildSlot(I64_DTYPE, d.encodedArr(), 0);
+		return new CascadeStep(partialRoot, List.of(idxBuf, valBuf), List.of(slot), d.statsMin(), d.statsMax());
+	}
+
+	private static EncodingProtos.PatchesMetadata buildPatchesMeta(int numPatches) {
+		return EncodingProtos.PatchesMetadata.newBuilder()
+				.setLen(numPatches)
+				.setOffset(0)
+				.setIndicesPtype(DTypeProtos.PType.forNumber(PType.U32.ordinal()))
+				.build();
 	}
 
 	private static byte[] scalarF64(double v) {

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

@@ -0,0 +1,38 @@
+package io.github.dfa1.vortex.encoding;
+
+import java.lang.foreign.MemorySegment;
+import java.util.List;
+
+/// One step in cascade-aware encoding: a partially-assembled node tree plus open child slots.
+///
+/// <p>Terminal steps have no open children and carry a fully-resolved {@link EncodeResult}.
+/// Intermediate steps have open children that the {@link CascadingCompressor} recursively fills.
+///
+/// <p>Buffer layout: {@code ownedBuffers} holds buffers belonging to the partial root
+/// (e.g. patch index/value buffers for ALP). Child recursion results are appended after these;
+/// child {@code bufferIndices} are remapped by {@code +ownedBuffers.size()}.
+public record CascadeStep(
+		EncodeNode partialRoot,
+		List<MemorySegment> ownedBuffers,
+		List<ChildSlot> openChildren,
+		byte[] statsMin,
+		byte[] statsMax
+) {
+	/// Convenience: terminal step — no open children, result is final.
+	public static CascadeStep terminal(EncodeResult result) {
+		return new CascadeStep(result.rootNode(), result.buffers(), List.of(), result.statsMin(), result.statsMax());
+	}
+
+	public boolean isTerminal() {
+		return openChildren.isEmpty();
+	}
+
+	/// Total byte size of owned buffers (used for size-based winner selection on samples).
+	public long ownedBytes() {
+		long total = 0;
+		for (MemorySegment seg : ownedBuffers) {
+			total += seg.byteSize();
+		}
+		return total;
+	}
+}