Streaming chunked decompression via a composable vtable method

[joseph-isaacs]

Summary

Adds decompress_chunks, a vtable method that streams an encoding tree's decompressed values through a ChunkSink in cache-resident ~1024-element chunks. Leaf encodings stream blocks straight out of their decompression kernel; wrapper encodings compose by interposing a stack-allocated sink adapter and recursing into their child, so a whole tree decompresses and transforms one L1-resident block at a time instead of materializing a full-length buffer per level.

Two things this buys:

1. Consumers that never materialize fold values directly out of L1 — a sum over FoR(BitPacked) runs 1.29 ms vs 3.36 ms for decompress-then-iterate, matching a hand-written monomorphized loop within noise.
2. Deep encoding stacks canonicalize faster, because each level costs level-wise execution a whole intermediate buffer but costs streaming only one more L1 pass. Marginal cost per added level: 0.93 ms level-wise vs 0.51 ms streaming.

The executor uses it where that second effect pays, decided by measuring the tree rather than by a blanket default.

The vtable API

pub trait VTable {
    /// Can this encoding (recursively) stream without materializing? Default: false.
    fn supports_decompress_chunks(array: ArrayView<'_, Self>) -> bool { false }

    /// Stream decompressed values through `sink` in cache-resident chunks.
    /// Default: reports unsupported — never a silent materializing fallback.
    fn decompress_chunks(
        array: ArrayView<'_, Self>,
        ctx: &mut ExecutionCtx,
        sink: &mut dyn ChunkSink,
    ) -> VortexResult<()>;
}

pub trait ChunkSink {
    fn accept(&mut self, chunk: ChunkMut<'_>, row_range: Range<usize>) -> VortexResult<()>;
}

Design points:

• Two methods, not one. Support must be answerable before any work happens, and it cascades: a wrapper only advertises support when the children it streams from do. So the whole tree is validated up front and ArrayRef::decompress_chunks rejects an unsupported tree without emitting a partial stream.
• No silent fallback. Callers who want one ask by name: ArrayRef::decompress_chunks_or_materialize.
• Dispatch is per chunk, never per value. ChunkMut is a type-erased &mut [T]; sinks recover the typed slice once per chunk and all per-element work stays monomorphized (~0.06 ns/element of dispatch).
• Chunks are mutable producer scratch, which is what lets wrappers transform in place (FoR adds its reference, Patched overwrites patched rows, Filter compacts) with no extra buffer.
• No heap state descending the tree — the sink chain is one stack frame per level.
• Contract: chunks are in-order, contiguous, cover 0..len exactly (debug-checked), may be short; validity is not streamed; primitive dtypes only.

Changes

• vortex-array/src/chunk_iter.rs (new): ChunkSink, ChunkMut, DECOMPRESS_CHUNK_LEN, the ArrayRef entry points, execute_via_chunks, a debug-build coverage checker, and the explicit materializing fallback.
• VTable::{supports_decompress_chunks, decompress_chunks}, wired through DynArrayData.
• execute_until step 2c: canonicalize by streaming when streaming_chain_len(array) >= MIN_STREAMING_CHAIN (4). The chain counts consecutive streaming-capable nodes from the root, following only children that stream, are not already canonical, and preserve row count — that last rule keeps selection encodings such as Filter out of the executor path without a special case. VORTEX_CHUNKED_EXECUTE=0 is a kill switch.
• Implementations: BitPacked (streams FastLanes blocks from its unpack scratch, patches applied per block via a cursor), FoR (fused FoRStrategy kernel when the child is BitPacked with an unsigned reference, otherwise generic in-place add composition), Patched (patches each streamed block in place; lane-transposed layout flattened once), Filter (compacts each block in place against the mask, with a mean-run-length heuristic choosing run-copy vs gather), Constant (re-emits one scratch chunk), Primitive (streams its buffer through one L1 scratch chunk).
• Tests: patches, slicing/offset, fused and generic FoR, Patched-over-Constant, Filter-over-BitPacked across selectivities plus all-true/all-false, non-chunk-aligned lengths, null constants, empty arrays, the error-without-emitting contract, streaming-vs-levelwise equivalence at three stack depths on nullable trees, and the executor depth rule (including that filter push-down beneath a deep stack still streams the levels above it).
• New divan bench encodings/fastlanes/benches/chunked_decompress.rs with hand-written monomorphized baselines to isolate each overhead.

Benchmarks

Divan on a 4-core cloud VM, 4Mi rows unless noted. Methodology note: an earlier revision of this PR reported executor wins that did not reproduce — they were measured while other builds were running. Everything below is from repeated rounds on a quiet machine, and results that were not stable across rounds are called out as such.

Depth sweep — nested FoR over BitPacked, canonicalize:

streaming chain	level-wise	streaming	speedup	level-wise p100	streaming p100
2	2.57 ms	2.58 ms	parity	10.9 ms	3.1 ms
3	3.14 ms	2.74 ms	1.15×	13.6 ms	3.8 ms
5	5.26 ms	4.00 ms	1.32×	17.0 ms	4.6 ms
9	9.07 ms	6.16 ms	1.47×	23.4 ms	6.5 ms

Marginal cost per added level: 0.93 ms level-wise vs 0.51 ms streaming. Tail latency is the other story: p100/median is ~1.2× streaming versus 2.5–5× level-wise, which repeatedly allocates full-length intermediates. This is what MIN_STREAMING_CHAIN = 4 is calibrated against; chain 3 is borderline (1.08–1.15× across rounds) and excluded conservatively.

Streaming consumption — sum without materializing, FoR(ref=1M) over BitPacked(bw=10):

bench	median
hand_fused_sum (monomorphized upper bound)	1.014 ms
chunked_vtable_sum (this API, fused FoR+BitPacked)	1.291 ms
hand_chunked_add_pass_sum (monomorphized, extra in-place pass)	1.411 ms
chunked_vtable_sum_generic_compose (signed ref, generic composition)	1.886 ms
two_pass_sum (decompress then re-read)	3.360 ms

Patched(Constant), 1 patch per 1000 — the 16 MB base buffer is never written: streaming sum 0.955 ms vs 1.953 ms decompress-then-read.

Where streaming does not win, and why. When an encoding's level-wise path already decodes straight into the destination (decode_into for fused FoR/Patched; the in-place compaction kernel for Filter), there is no intermediate to eliminate and streaming only adds a scratch→output copy. Filter(BitPacked) at 65,536 rows (the dominant TPC-H scan tree) measures at parity for materialization and 10–35% slower for consumption except at high selectivity. Hence the cardinality rule in the executor: Filter streams for consumers, never through the executor.

SQL (TPC-H SF-1, DataFusion, Q1 + Q6). Neutral within noise. A trace showed why: streaming-capable trees are ~6% of non-canonical executor invocations there, and the trees that dominate (Filter, decimal_byte_parts) are shallow — chains of 2–3, below the depth where streaming pays. The mechanism is confirmed non-regressing on the real scan path; SQL-level wins need ChunkSink consumers in the aggregate kernels, where the consumption win above would reach query time.

Next step. The remaining structural overhead is destination-blindness: a producer cannot see the consumer's output buffer, so it always decodes into scratch. Letting a sink offer a destination slice per chunk would remove that copy, which should turn the parity/loss cases into wins and lower MIN_STREAMING_CHAIN.

Checks run: cargo nextest run on vortex-array (3439), vortex-fastlanes (327), vortex-file (144), vortex-alp, vortex-btrblocks — all passing; cargo test --doc -p vortex-array; cargo clippy --all-targets on vortex-array/vortex-fastlanes (clean); cargo +nightly fmt --all; TPC-H SF-1 Q1/Q6 A/B via datafusion-bench. Not run: workspace-wide --all-features clippy (CUDA feature stack untouched).

API Changes

New public API in vortex-array: the chunk_iter module (ChunkSink, ChunkMut, DECOMPRESS_CHUNK_LEN, execute_via_chunks), ArrayRef::{supports_decompress_chunks, decompress_chunks, decompress_chunks_or_materialize}, and two defaulted VTable methods. Behavior change: execute_until canonicalizes streaming-capable primitive trees of chain length ≥ 4 via chunk streaming (same logical results, covered by equivalence tests; disable with VORTEX_CHUNKED_EXECUTE=0).

🤖 Generated with Claude Code

https://claude.ai/code/session_01LNgQat1UYMr3pjJrdhPJuh

[codspeed-hq]

Merging this PR will degrade performance by 2.79%

⚠️ Unknown Walltime execution environment detected

Using the Walltime instrument on standard Hosted Runners will lead to inconsistent data.

For the most accurate results, we recommend using CodSpeed Macro Runners: bare-metal machines fine-tuned for performance measurement consistency.

⚠️ Different runtime environments detected

Some benchmarks with significant performance changes were compared across different runtime environments,
which may affect the accuracy of the results.

Open the report in CodSpeed to investigate

⚡ 6 improved benchmarks
❌ 10 regressed benchmarks
✅ 2141 untouched benchmarks
🆕 41 new benchmarks
⏩ 106 skipped benchmarks¹
🗄️ 4 archived benchmarks run²

Warning

Please fix the performance issues or acknowledge them on CodSpeed.

Performance Changes

	Mode	Benchmark	BASE	HEAD	Efficiency
❌	WallTime	mul_u64_nonnull_neon	15.2 µs	19.6 µs	-22.85%
❌	Simulation	decompress[for_bp_u64]	47.7 µs	58.3 µs	-18.28%
❌	WallTime	multiply_shapes_neon[(16384, PerRowPerRow)]	17.3 µs	20.2 µs	-14.75%
❌	Simulation	bitpacked_decompress_u32	55.7 µs	64.9 µs	-14.26%
❌	WallTime	mul_i64_nonnull_neon	17.2 µs	19.9 µs	-13.77%
❌	Simulation	baseline_lt[4, 1024]	82.7 µs	93 µs	-11.06%
❌	Simulation	cold_misaligned[(16, 64)]	345.6 µs	388.6 µs	-11.05%
❌	Simulation	take[all_null_len65536_run4_take2048]	99.7 µs	111.1 µs	-10.2%
❌	Simulation	decompress_rd[f32, (2000, 0.0)]	92.3 µs	102.8 µs	-10.2%
❌	WallTime	mul_i32_nonnull_avx512	7.2 µs	8.1 µs	-10.19%
⚡	Simulation	compact_sliced[(2048, 10)]	250.8 µs	180.7 µs	+38.83%
⚡	Simulation	compact_sliced[(1024, 10)]	149.6 µs	113.8 µs	+31.42%
⚡	WallTime	subtract_shapes_neon[(16384, PerRowNullableConstant)]	12.3 µs	10.9 µs	+12.47%
⚡	WallTime	sub_i64_constant_neon	11.3 µs	10.1 µs	+12.13%
⚡	WallTime	subtract_shapes_neon[(128, PerRowNullableConstant)]	3.9 µs	3.5 µs	+10.39%
⚡	Simulation	compact[(1024, 10)]	279.4 µs	254 µs	+10.01%
🆕	Simulation	chunked_decompress_into_patched_constant	N/A	18.5 ms	N/A
🆕	Simulation	chunked_vtable_decompress_into	N/A	22.9 ms	N/A
🆕	Simulation	chunked_vtable_sum	N/A	8.8 ms	N/A
🆕	Simulation	chunked_vtable_sum_generic_compose	N/A	10.9 ms	N/A
...	...	...	...	...	...

ℹ️ Only the first 20 benchmarks are displayed. Go to the app to view all benchmarks.

Tip

Investigate this regression by commenting @codspeedbot fix this regression on this PR, or directly use the CodSpeed MCP with your agent.

---

_{Comparing claude/decompression-chunked-iteration-u3kcns (377d029) with develop (68e2aee)}

Footnotes

1. 106 benchmarks were skipped, so the baseline results were used instead. If they were deleted from the codebase, click here and archive them to remove them from the performance reports. ↩

2. 4 benchmarks were run, but are now archived. If they were deleted in another branch, consider rebasing to remove them from the report. Instead if they were added back, click here to restore them. ↩