diff --git a/TODO.md b/TODO.md index 411ce1a1..e1e81530 100644 --- a/TODO.md +++ b/TODO.md @@ -19,10 +19,6 @@ Out of scope for the #287 rewrite (which, together with the follow-up hot-path p `vortex-jni` gap with a scalar, branch-free algorithm — encode at parity, decode ~1.16x faster on `JavaVsJniFsstBenchmark` — see [ADR-0022](adr/0022-fsst-module-extraction.md)): -- [ ] **True per-row lazy/random-access decompression** exploiting FSST's headline random-access - property — today `FsstEncodingDecoder.decode()` eagerly materializes the whole column up front - regardless of what is queried. Connects to [ADR-0010](adr/0010-lazy-decode.md) (Lazy decode) but - is a separate initiative. - [ ] **OptFSST** (2026 arXiv follow-up: DP-based training instead of greedy, ~4x slower training for 7–17% better compression) — a documented future option, not adopted, since it moves off the classic greedy-FSST speed/compression tradeoff this rewrite targets (matching what `vortex-jni` diff --git a/adr/0026-fsst-per-row-lazy-decode.md b/adr/0026-fsst-per-row-lazy-decode.md new file mode 100644 index 00000000..22980f6a --- /dev/null +++ b/adr/0026-fsst-per-row-lazy-decode.md @@ -0,0 +1,160 @@ +# ADR 0026: FSST per-row lazy decode + +- **Status:** Accepted +- **Date:** 2026-09-13 +- **Deciders:** project maintainer +- **Supersedes:** — +- **Superseded by:** — +- **Related:** [ADR 0010 — Lazy decode for 1:1 transform encodings](0010-lazy-decode.md), + [ADR 0022 — Extract FSST into a standalone module](0022-fsst-module-extraction.md) + +## Context + +ADR 0010 grouped `Fsst` with `Bitpacked`, `Pco`, and `Zstd` as encodings that "must remain eager": +*"their output shape differs from their input (compact compressed bytes → wider element array), so +element-at-i requires unpacking a window."* That reasoning holds for the other three — a bitpacked +value, a Pco block, and a Zstd frame all require decoding a run of neighboring elements to recover +one. It does not hold for FSST. + +FSST's wire shape already carries two children that make row `i` independent of every other row: +an uncompressed-lengths child (`n` elements, one decoded byte count per row) and a codes-offsets +child (`n + 1` elements, `codesOffsets[i] .. codesOffsets[i + 1]` is row `i`'s compressed code +range). Nothing about decoding row 5 touches row 4's or row 6's bytes. `FsstEncodingDecoder.decode()` +nonetheless decompressed the entire column into one flat buffer up front, regardless of which rows +(if any) a caller went on to read — the exact waste ADR 0010's motivating table catalogs for +`WHERE` filters, projections that drop the column, and `LIMIT`/`take` slices. + +Two observations sharpen this: + +1. **Lengths need no decompression at all.** The uncompressed-lengths child already states each + row's decoded byte count directly. `JavaVsJniFsstBenchmark.javaFsstDecode` — which scans the + whole column and sums `forEachByteLength` — was paying the full decompression cost for a query + the wire format could answer without running the FSST algorithm even once. +2. **A single row decodes in isolation.** `Decompressor.decompress(MemorySegment, long, long, + MemorySegment, long)` already takes an arbitrary `[start, end)` code range; the existing eager + path called it in row-batches purely to avoid an OSR mega-loop over the whole chunk (see the + removed `ROWS_PER_DECODE_BATCH` comment), not because decoding required more than one row's + context. + +## Decision + +`FsstEncodingDecoder.decode()` no longer decompresses. It builds a `Decompressor` from the wire +symbol table (unchanged — at most 255 symbols, negligible cost) and returns a +`LazyFsstVarBinArray` holding that decompressor plus the still-compressed code stream and the two +per-row children: + +```java +return new LazyFsstVarBinArray(ctx.dtype(), n, decompressor, compressedBytes, + uncompLensSeg, uncompLenPType, codesOffsetsSeg, codesOffPType); +``` + +`LazyFsstVarBinArray implements VarBinArray` (`reader.array`, the same package as every other +`VarBin*Array`): + +- **`getByteLength(i)`** reads the uncompressed-lengths child directly — no decompression. +- **`forEachByteLength(c)`** walks the same child in a branch-split, per-ptype loop (CLAUDE.md + hot-loop rule) — still no decompression, and no per-row cross-validation (matching + `VarBinOffsetArray.forEachByteLength`'s existing "bulk walk trusts the data" convention). +- **`getBytes(i)` / `getString(i)`** decompress *only* row `i`'s code range, on that call, into a + freshly sized scratch array. +- **`bytesSegment()`** returns the `MemorySegment.NULL` sentinel and **`segmentIfPresent()`** + returns empty — the established convention for every other non-contiguous `VarBinArray` + (`VarBinChunkedArray`, `VarBinRunEndArray`, `VarBinSparseArray`, `VarBinConstantArray`). A caller + that needs the flat bytes-plus-offsets shape gets it from `VarBinArray.toOffsetMode`, which walks + every row through `getBytes` — i.e., decodes the whole column exactly once, only when actually + asked to. + +### Validation moves from decode-time to access-time + +The old eager path validated the whole column up front: a prefix-sum pass over the uncompressed +lengths, a first/last code-offset bounds check, and a post-decode "did the batch produce exactly +the claimed byte count" comparison. All of that was `O(rows)` or `O(bytes)` work — exactly the cost +this ADR removes from `decode()`. + +Per-row validation instead happens inside the accessors, mirroring +`VarBinArrays.checkedLength`'s existing rationale ("offsets arrive from an untrusted file and are +deliberately not scanned at decode time"): + +- **`getByteLength(i)`** cross-checks the claimed length against the maximum a decode of that row's + code range could produce (FSST's own bound: at most 8 output bytes per compressed byte — see + `Decompressor`'s "unconditional 8-byte store" javadoc). This is `O(1)` — no decompression — yet it + still rejects a claimed length like "1.5 GB from a 1-byte code range" without ever running the + decompressor. +- **`getBytes(i)`** sizes its scratch buffer from that same bound (never from the untrusted claimed + length), decompresses, and only then compares the actual decoded byte count against the claim — + a strictly *more* precise version of the old aggregate check, now scoped to the one row that was + actually read. + +Only one guard remains eager: `n >= Integer.MAX_VALUE` is still rejected in `decode()`, since it +bounds the codes-offsets child's own `n + 1`-element decode (`ctx.decodeChildSegment`), which +`decode()` always performs regardless of laziness. + +## Consequences + +### Positive + +- **Filter/projection/take pushdown works for FSST columns for free.** A column requested but never + read costs nothing; a filter that rejects 99% of rows only ever decompresses the 1% it keeps. +- **`forEachByteLength`-shaped aggregations (cardinality checks, length histograms, `LIMIT` + planning) cost zero decompression.** `JavaVsJniFsstBenchmark.javaFsstDecode` goes from "decompress + everything, then sum lengths already known" to "read the lengths child directly." +- **Per-row validation is strictly more precise than the old aggregate check**, since a per-row + mismatch is caught at the row it occurs in rather than only detectable in aggregate. +- **ADR 0010's blanket "decompression-style encodings stay eager" rule gets a documented exception** + for the one member of that group whose rows are independently addressable in the wire format. + +### Negative + +- **A full-column scan that reads every row's bytes now issues one `Decompressor.decompress` call + per row instead of one call per 256-row batch.** The `ROWS_PER_DECODE_BATCH` batching existed to + keep the decode loop out of OSR compilation, not to amortize per-call overhead across rows; this + ADR does not re-measure whether per-row calls reintroduce that OSR cost at full-scan scale — no + benchmark in this codebase exercises full-column `getString`/`getBytes` for FSST today (the only + FSST throughput benchmark, `javaFsstDecode`, uses `forEachByteLength`, which never decompresses in + either the old or new design). A future benchmark that does full-column FSST string materialization + should be added before relying on this path's throughput. +- **Two heap allocations per accessed row** (`getBytes`'s scratch array plus the trimmed copy) where + the old path allocated once for the whole batch. Acceptable for the row-at-a-time access pattern + this ADR targets; would need revisiting if profiling shows it dominates a hot path. + +### Risks to manage + +- **A corrupted length that happens to fit the code-range bound still passes `getByteLength`.** + E.g. a code range that could produce up to 8 bytes but a claimed length of 5 when the true decode + produces 2 — `getByteLength` cannot detect this (it has no way to decode without paying the cost + it exists to avoid); only `getBytes`/`getString` catch it, by comparing the actual decode output. + Callers that call `getByteLength` alone (e.g. `forEachByteLength`-style aggregations) do not get + this stronger guarantee — consistent with `VarBinOffsetArray`'s existing lenient bulk-walk + behavior, but worth remembering when reasoning about what a length-only scan has actually verified. + +## Alternatives considered + +### A — Keep eager decode, skip only when `forEachByteLength` is the sole call site + +Special-case `forEachByteLength` to read the lengths child directly while leaving `getBytes`/ +`getString` behind an eager `decode()`. + +Rejected: still pays full decompression for every filtered-out or unprojected row whenever any row +is read as a string — the dominant win (filter/take pushdown) requires per-row laziness, not just a +length-only fast path. + +### B — Batch decode lazily in windows (e.g. re-run the old 256-row batching, triggered on first +access to any row in the window) + +Would preserve the batching that avoided OSR compilation while still deferring work for untouched +windows. + +Rejected as unnecessary complexity for this change: the per-row `Decompressor.decompress` call is +already a short, non-mega loop by construction (one row's codes), so there is no OSR concern to +re-solve. If full-column-scan benchmarking (see Consequences → Negative) later shows per-row call +overhead dominates, revisit with real numbers rather than pre-optimizing here. + +## References + +- `Decompressor.decompress(MemorySegment, long, long, MemorySegment, long)` — the per-range decode + primitive this ADR calls once per row instead of once per batch. +- `LazyFsstVarBinArray` (`reader.array`) — the lazy implementation. +- `VarBinArrays.checkedLength` — the "don't scan offsets at decode time" convention this ADR extends + to FSST's length/offset children. +- [ADR 0010](0010-lazy-decode.md) — the original lazy-decode framework; this ADR narrows its + "decompression encodings stay eager" exclusion to exclude FSST specifically. diff --git a/adr/ADR.md b/adr/ADR.md index 577af09f..49e5e1c4 100644 --- a/adr/ADR.md +++ b/adr/ADR.md @@ -40,3 +40,4 @@ the decision shipped in (blank = not yet shipped). | 0023 | Adopt the Vortex editions model as a client-side write/read policy | Accepted | | | 0024 | JPMS adoption for core/reader/writer | Proposed | | | 0025 | Centralize zone-map MIN/MAX stats computation in the writer | Accepted | | +| 0026 | FSST per-row lazy decode | Accepted | | diff --git a/docs/compatibility.md b/docs/compatibility.md index 3e3f8b82..56a0cc00 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -152,7 +152,7 @@ decoder falls into one of three shapes: | `vortex.sequence` | Lazy | Lazy | `LazySequenceXxxArray`; `base + i * multiplier` per access, no buffer, ADR 0015 | | `vortex.struct` | Zero-copy | Zero-copy | `StructArray` wraps fields | | `vortex.chunked` | Lazy | Lazy | `ChunkedXxxArray` (primitive/Bool) + `VarBinChunkedArray` (Utf8/Binary), ADR 0012 | -| `vortex.fsst` | Materialized | Materialized | symbol-table decompression | +| `vortex.fsst` | Lazy | Lazy | `LazyFsstVarBinArray` — per-row code range is independent, so `getBytes(i)` decompresses only row `i`; `getByteLength`/`forEachByteLength` read the uncompressed-lengths child directly, no decompression at all, ADR 0026 | | `vortex.list` | Lazy | Lazy | `ListArray` wraps elements + offsets children; shape inherits from child | | `vortex.listview` | Lazy | Lazy | `ListViewArray` wraps elements + offsets + sizes children; a validity child yields a `MaskedArray` over it | | `vortex.map` | Lazy | Lazy | `MapArray` wraps the entries child (a `ListViewArray`, or a `MaskedArray` over one when the map is nullable) | @@ -171,10 +171,11 @@ decoder falls into one of three shapes: | `vortex.variant` | Lazy | Lazy | container wraps constant/chunked core (inner-typed) + optional shredded child | | `vortex.onpair` | n/a | n/a | not ported | -Decompression-style encodings (Bitpacked / Pco / Zstd / Fsst / Delta) stay Materialized by design -— element-at-`i` requires decoding a window, so they must allocate output (ADR 0010 §"Decompression -encodings stay eager"). Their output can itself be wrapped in a 1:1 lazy transform (e.g. ALP over -Bitpacked produces `LazyAlp(MaterializedXxx)`). +Decompression-style encodings (Bitpacked / Pco / Zstd / Delta) stay Materialized by design — +element-at-`i` requires decoding a window, so they must allocate output (ADR 0010). Their output +can itself be wrapped in a 1:1 lazy transform (e.g. ALP over Bitpacked produces +`LazyAlp(MaterializedXxx)`). Fsst is the one exception: its per-row code range is independent of +every other row, so it stays Lazy instead (ADR 0026). ### Unknown encodings diff --git a/docs/reference.md b/docs/reference.md index 15399e10..f72a53ce 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -379,7 +379,11 @@ The `vortex-fsst` module is the standalone FSST (Fast Static Symbol Table) strin algorithm, usable independently of Vortex. It depends only on the JDK (`java.lang.foreign`), never on `core`/`reader`/`writer`; `writer`/`reader` depend on it. The `vortex.fsst` encoding adapter is one caller — the module itself knows nothing of the Vortex wire format. See -[ADR 0022](../adr/0022-fsst-module-extraction.md). +[ADR 0022](../adr/0022-fsst-module-extraction.md). On the read side, `FsstEncodingDecoder` never +decompresses eagerly: it returns a `LazyFsstVarBinArray` that decompresses row `i`'s code range only +when that row is actually read, since FSST's per-row code range is independent of every other row +(unlike `Bitpacked`/`Pco`/`Zstd`, which do need a decoded window) — see +[ADR 0026](../adr/0026-fsst-per-row-lazy-decode.md). Compress: train a `Compressor` over a corpus, then compress rows against its table. Decompress: build a `Decompressor` (from the trained `Compressor`, or from raw table arrays). Hot-path methods diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyFsstVarBinArray.java b/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyFsstVarBinArray.java new file mode 100644 index 00000000..56094fd4 --- /dev/null +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyFsstVarBinArray.java @@ -0,0 +1,252 @@ +package io.github.dfa1.vortex.reader.array; + +import io.github.dfa1.vortex.core.error.VortexException; +import io.github.dfa1.vortex.core.io.VortexFormat; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.core.model.EncodingId; +import io.github.dfa1.vortex.core.model.PType; +import io.github.dfa1.vortex.fsst.Decompressor; +import io.github.dfa1.vortex.reader.decode.SegmentBroadcast; + +import java.lang.foreign.MemorySegment; +import java.lang.foreign.ValueLayout; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Objects; +import java.util.Optional; +import java.util.function.IntConsumer; + +/// Lazy [VarBinArray] backed by an undecoded `vortex.fsst` column. +/// +/// FSST's per-row code range (`codesOffsets[i] .. codesOffsets[i+1]`) is independent of every +/// other row, so — unlike the window-based encodings (`Bitpacked`, `Pco`, `Zstd`) ADR 0010 +/// originally grouped it with — a single row decodes without touching its neighbors. This class +/// exploits that: [#getBytes(long)] decompresses only the requested row's code range on each +/// call, and [#getByteLength(long)] and [#forEachByteLength(IntConsumer)] read the wire's own +/// per-row uncompressed-length child directly, never invoking the decompressor at all. See +/// ADR 0026. +/// +/// [#bytesSegment()] is the [MemorySegment#NULL] sentinel and [#segmentIfPresent()] is empty — no +/// single contiguous buffer holds the expanded rows, the same convention [VarBinChunkedArray], +/// [VarBinRunEndArray], [VarBinSparseArray], and [VarBinConstantArray] use. A consumer that needs +/// the flat bytes-plus-offsets shape gets it via +/// [VarBinArray#toOffsetMode(VarBinArray, java.lang.foreign.SegmentAllocator)], which walks every +/// row through [#getBytes(long)] — decoding the whole column exactly once, on demand. +/// +/// Per-row bounds are validated on access, never at construction: an untrusted length or code +/// range surfaces as a [VortexException] the first time the offending row is read, mirroring +/// [VarBinArrays#checkedLength(MemorySegment, long, long)]'s "offsets are not scanned at decode +/// time" convention. A row's claimed uncompressed length is cross-checked against the maximum a +/// symbol-table decode of its code range could ever produce (8 bytes per compressed byte, the FSST +/// paper's own bound — see [Decompressor#decompress(MemorySegment, long, long, MemorySegment, +/// long)]) rather than trusted outright, so a corrupted or adversarial length can never drive an +/// oversized allocation. +/// +/// @param dtype logical type (Utf8 or Binary) +/// @param length number of logical elements (rows) +/// @param decompressor decoder bound to this column's symbol table +/// @param compressedBytes the FSST code stream (buffer 2 of the `vortex.fsst` node) +/// @param uncompressedLengths per-row decoded byte count child (length = `length`, broadcastable) +/// @param uncompressedLengthsPType physical type of `uncompressedLengths` +/// @param codesOffsets per-row code range child (length = `length + 1`, broadcastable) +/// @param codesOffsetsPType physical type of `codesOffsets` +public record LazyFsstVarBinArray( + DType dtype, long length, Decompressor decompressor, + MemorySegment compressedBytes, + MemorySegment uncompressedLengths, PType uncompressedLengthsPType, + MemorySegment codesOffsets, PType codesOffsetsPType) + implements VarBinArray { + + /// No single contiguous segment backs the lazily decoded rows. + /// + /// @return the [MemorySegment#NULL] sentinel + @Override + public MemorySegment bytesSegment() { + return MemorySegment.NULL; + } + + /// No single contiguous segment backs the lazily decoded rows. + /// + /// @return always empty + @Override + public Optional segmentIfPresent() { + return Optional.empty(); + } + + @Override + public byte[] getBytes(long i) { + Objects.checkIndex(i, length); + CodeRange range = codeRange(i); + long maxLen = range.maxDecodedLength(); + if (maxLen > Integer.MAX_VALUE - 7) { + throw new VortexException(EncodingId.VORTEX_FSST, "decoded length too large: row " + i + + " code range implies up to " + maxLen + " bytes"); + } + // 7 bytes of trailing slack for the decompressor's unconditional 8-byte-store trick + // (see Decompressor); sliced off below so the caller sees an exactly-sized array. + byte[] scratch = new byte[(int) maxLen + 7]; + long decodedLen; + try { + decodedLen = decompressor.decompress(compressedBytes, range.start(), range.end(), + MemorySegment.ofArray(scratch), 0); + } catch (IndexOutOfBoundsException e) { + // Two adversarial shapes land here: a trailing escape code with no literal byte after + // it (reads one byte past the row's own code range, potentially past compressedBytes + // entirely) and a non-escape code naming a symbol index past the trained table's size + // (ArrayIndexOutOfBoundsException, a subtype of IndexOutOfBoundsException, indexing + // Decompressor's packedSymbols/lengths arrays). Both must surface as VortexException, + // never a raw JDK exception (the reader parses untrusted binary input). + throw new VortexException(EncodingId.VORTEX_FSST, + "row " + i + " code range [" + range.start() + ", " + range.end() + + ") decodes past its bounds or references an unknown symbol code", e); + } + long claimedLen = uncompressedLength(i); + if (decodedLen != claimedLen) { + throw new VortexException(EncodingId.VORTEX_FSST, "row " + i + " decoded " + decodedLen + + " bytes but uncompressed lengths claim " + claimedLen); + } + return Arrays.copyOf(scratch, (int) decodedLen); + } + + @Override + public String getString(long i) { + return new String(getBytes(i), StandardCharsets.UTF_8); + } + + @Override + public int getByteLength(long i) { + Objects.checkIndex(i, length); + long maxLen = codeRange(i).maxDecodedLength(); + long claimedLen = uncompressedLength(i); + if (claimedLen < 0 || claimedLen > maxLen) { + throw new VortexException(EncodingId.VORTEX_FSST, "decoded length too large: row " + i + + " claims " + claimedLen + " bytes but code range implies at most " + maxLen); + } + return (int) claimedLen; + } + + /// Sums per-row lengths straight from the uncompressed-lengths child — no decompression, no + /// per-row bounds cross-check (matching [VarBinOffsetArray#forEachByteLength(IntConsumer)]'s + /// "bulk walk trusts the data, typed accessors validate it" convention). Branch-split per ptype + /// and on the broadcast case so the per-row body stays a uniform, fixed-stride read + /// (CLAUDE.md hot-loop rule). + /// + /// @param c consumer called once per row with the claimed byte length at that index + @Override + public void forEachByteLength(IntConsumer c) { + long n = length; + long cap = SegmentBroadcast.capacity(uncompressedLengths, uncompressedLengthsPType.byteSize()); + if (cap == 0) { + if (n == 0) { + return; + } + throw new VortexException(EncodingId.VORTEX_FSST, "empty uncompressed-lengths child"); + } + if (cap >= n) { + forEachClaimedLength(c, n); + return; + } + for (long i = 0; i < n; i++) { + c.accept((int) readAt(uncompressedLengths, (i % cap) * uncompressedLengthsPType.byteSize(), + uncompressedLengthsPType)); + } + } + + /// Zero-copy truncation: rows are resolved on read, so trailing rows past the new length + /// simply go unvisited. + /// + /// @param rows number of leading rows to keep + /// @return a length-`rows` view over the same underlying segments + @Override + public VarBinArray limited(long rows) { + if (rows >= length) { + return this; + } + return new LazyFsstVarBinArray(dtype, rows, decompressor, compressedBytes, + uncompressedLengths, uncompressedLengthsPType, codesOffsets, codesOffsetsPType); + } + + /// Fast path of [#forEachByteLength(IntConsumer)] when the uncompressed-lengths child holds at + /// least `n` physical elements: reads at a constant stride per ptype, no per-row modulo. + private void forEachClaimedLength(IntConsumer c, long n) { + switch (uncompressedLengthsPType) { + case U8 -> { + for (long i = 0; i < n; i++) { + c.accept(Byte.toUnsignedInt(uncompressedLengths.get(ValueLayout.JAVA_BYTE, i))); + } + } + case U16 -> { + for (long i = 0; i < n; i++) { + c.accept(Short.toUnsignedInt(uncompressedLengths.get(VortexFormat.LE_SHORT, i * 2))); + } + } + case U32 -> { + for (long i = 0; i < n; i++) { + c.accept((int) Integer.toUnsignedLong( + uncompressedLengths.getAtIndex(VortexFormat.LE_INT, i))); + } + } + case I32 -> { + for (long i = 0; i < n; i++) { + c.accept(uncompressedLengths.getAtIndex(VortexFormat.LE_INT, i)); + } + } + case I64, U64 -> { + for (long i = 0; i < n; i++) { + c.accept((int) uncompressedLengths.getAtIndex(VortexFormat.LE_LONG, i)); + } + } + default -> throw new VortexException(EncodingId.VORTEX_FSST, + "unsupported ptype " + uncompressedLengthsPType); + } + } + + /// Row `i`'s validated compressed code range, plus the FSST-paper bound (8 bytes per + /// compressed byte — [Decompressor]'s "unconditional 8-byte store" trick) on how many bytes + /// decoding it could ever produce. + /// + /// @param start start offset into [#compressedBytes()], inclusive + /// @param end end offset into [#compressedBytes()], exclusive + private record CodeRange(long start, long end) { + long maxDecodedLength() { + return (end - start) * 8; + } + } + + /// Reads and validates row `i`'s code range against [#compressedBytes()]'s actual size. + private CodeRange codeRange(long i) { + long start = codeOffset(i); + long end = codeOffset(i + 1); + if (start < 0 || end < start || end > compressedBytes.byteSize()) { + throw new VortexException(EncodingId.VORTEX_FSST, "invalid code offsets [" + start + + ", " + end + ") of " + compressedBytes.byteSize() + " at row " + i); + } + return new CodeRange(start, end); + } + + /// Reads codes-offsets element `idx`, broadcasting if the physical child is shorter than + /// `length + 1` (the [SegmentBroadcast] convention shared with every other lazy accessor in + /// this codebase). + private long codeOffset(long idx) { + long byteOffset = SegmentBroadcast.elementOffset(codesOffsets, idx, codesOffsetsPType.byteSize()); + return readAt(codesOffsets, byteOffset, codesOffsetsPType); + } + + /// Reads the claimed uncompressed length of row `i`, broadcasting if the physical child is + /// shorter than `length`. + private long uncompressedLength(long i) { + long byteOffset = SegmentBroadcast.elementOffset(uncompressedLengths, i, uncompressedLengthsPType.byteSize()); + return readAt(uncompressedLengths, byteOffset, uncompressedLengthsPType); + } + + private static long readAt(MemorySegment seg, long byteOffset, PType ptype) { + return switch (ptype) { + case U8 -> Byte.toUnsignedLong(seg.get(ValueLayout.JAVA_BYTE, byteOffset)); + case U16 -> Short.toUnsignedLong(seg.get(VortexFormat.LE_SHORT, byteOffset)); + case U32 -> Integer.toUnsignedLong(seg.get(VortexFormat.LE_INT, byteOffset)); + case I32 -> seg.get(VortexFormat.LE_INT, byteOffset); + case I64, U64 -> seg.get(VortexFormat.LE_LONG, byteOffset); + default -> throw new VortexException(EncodingId.VORTEX_FSST, "unsupported ptype " + ptype); + }; + } +} diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/decode/FsstEncodingDecoder.java b/reader/src/main/java/io/github/dfa1/vortex/reader/decode/FsstEncodingDecoder.java index a61c1254..e7c56680 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/decode/FsstEncodingDecoder.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/decode/FsstEncodingDecoder.java @@ -4,7 +4,7 @@ import io.github.dfa1.vortex.core.model.PType; import io.github.dfa1.vortex.core.error.VortexException; import io.github.dfa1.vortex.reader.array.Array; -import io.github.dfa1.vortex.reader.array.VarBinOffsetArray; +import io.github.dfa1.vortex.reader.array.LazyFsstVarBinArray; import io.github.dfa1.vortex.core.model.EncodingId; import io.github.dfa1.vortex.core.io.VortexFormat; import io.github.dfa1.vortex.core.proto.ProtoFSSTMetadata; @@ -19,17 +19,12 @@ /// /// This class is a thin wire adapter over the standalone `vortex-fsst` module (issue #287): after /// parsing the `vortex.fsst` wire buffers (symbol table, per-row uncompressed lengths, code offsets, -/// [ProtoFSSTMetadata]), it hands the symbol table and each row's code range to a [Decompressor], -/// which runs the FSST paper's Algorithm 1 decode. +/// [ProtoFSSTMetadata]), it builds a [Decompressor] bound to the symbol table and hands it — along +/// with the still-compressed code stream and the per-row length/offset children — to a +/// [LazyFsstVarBinArray], which decompresses each row's code range only when that row is actually +/// read (ADR 0026). No decompression happens here. public final class FsstEncodingDecoder implements EncodingDecoder { - /// Rows decoded per [Decompressor#decompress] call on the fast path. Batching (rather than one - /// whole-chunk call) keeps each call's loop short enough that execution stays in the cleanly - /// compiled method entry instead of an OSR-compiled mega-loop — a single 50k-row call measured - /// ~1.8x slower per byte than short calls on the same data — while still amortizing the per-row - /// offset read down to one read per batch. - private static final int ROWS_PER_DECODE_BATCH = 256; - @Override public EncodingId encodingId() { return EncodingId.VORTEX_FSST; @@ -69,10 +64,11 @@ public Array decode(DecodeContext ctx) { MemorySegment symbolLensBuf = ctx.buffer(1); MemorySegment compressedBytes = ctx.buffer(2); + // These children carry one length/offset per row (or n+1 for offsets) — proportional to + // row count, not to the compressed string bytes — so decoding them here is unavoidable + // metadata cost, not the eager decompression this class exists to avoid. MemorySegment uncompLensSeg = ctx.decodeChildSegment(0, new DType.Primitive(uncompLenPType, false), n); MemorySegment codesOffsetsSeg = ctx.decodeChildSegment(1, new DType.Primitive(codesOffPType, false), n + 1); - long uncompLensCap = SegmentBroadcast.capacity(uncompLensSeg, uncompLenPType.byteSize()); - long codesOffCap = SegmentBroadcast.capacity(codesOffsetsSeg, codesOffPType.byteSize()); // Read the wire symbol table into parallel code-indexed arrays once per chunk (there are at // most 255 symbols), then hand them to the decompressor. symbolsBuf carries one LSB-first @@ -88,152 +84,17 @@ public Array decode(DecodeContext ctx) { } Decompressor decompressor = Decompressor.of(packedSymbols, symbolLengths); + // Bounds the codes-offsets child's `n + 1` element count decoded above; not, unlike the old + // eager path, about sizing a flat I32 offsets buffer (there is none anymore). if (n >= Integer.MAX_VALUE) { throw new VortexException(EncodingId.VORTEX_FSST, "row count too large: " + n); } - // The decoded row boundaries are fully determined by the uncompressed-lengths child, so the - // output offsets are its prefix sums — computed in one per-ptype loop (no switch, no - // modulo in the body; hot-loop rule) that also yields the total for sizing the output. - MemorySegment outOffsets = ctx.arena().allocate((n + 1) * 4L, 4); - outOffsets.setAtIndex(VortexFormat.LE_INT, 0, 0); - long totalUncompressed = writeDecodedOffsets(uncompLensSeg, outOffsets, n, uncompLenPType, uncompLensCap); - - // The output offsets are I32 prefix sums, so a decoded chunk past 2 GB would silently wrap - // (and adversarial I32/I64-typed lengths can make the running sum go negative). Reject both - // before sizing/allocating outBytes, so a hostile length child can never drive a wrapped or - // enormous allocation. - if (totalUncompressed > Integer.MAX_VALUE || totalUncompressed < 0) { - throw new VortexException(EncodingId.VORTEX_FSST, - "decoded length too large: " + totalUncompressed); - } - - // Allocate 7 bytes of slack past the true logical length: the decompressor's unconditional - // 8-byte-store trick writes a full 8 bytes for the final symbol even when it contributes as - // few as 1 real byte. The slack is sliced off before the buffer is exposed, so callers still - // see an exactly-sized buffer. - MemorySegment outBytes = ctx.arena().allocate(totalUncompressed + 7); - - if (codesOffCap == 0) { - // No physical offsets at all: only valid for a zero-row chunk, where there is nothing - // to decode. - if (n > 0) { - throw new VortexException(EncodingId.VORTEX_FSST, "empty codes-offsets child"); - } - } else if (codesOffCap == 1) { - // Constant-broadcast offsets child: all n + 1 logical offsets share one physical value, - // so every row's code range is the empty [off, off) and the column decodes to nothing. - // The batched fast path below must NOT run here — it reads offsets at row indices (256, - // 512, ...), which would run off this 1-element segment. A non-zero claimed length means - // the lengths child disagrees with the (empty) code ranges, i.e. a malformed file. - if (totalUncompressed != 0) { - throw new VortexException(EncodingId.VORTEX_FSST, "constant code offsets imply an " - + "empty column but uncompressed lengths claim " + totalUncompressed + " bytes"); - } - } else if (codesOffCap >= n + 1) { - // Fast path. Row i's code range is [offsets[i], offsets[i+1]) out of ONE shared offsets - // array, so consecutive rows are contiguous by construction and the code stream decodes - // in row batches — only every ROWS_PER_DECODE_BATCH-th offset is read (no per-row - // offset reads, no per-row switch/modulo). - long firstOffset = readUnsigned(codesOffsetsSeg, 0, codesOffPType); - long lastOffset = readUnsigned(codesOffsetsSeg, Math.min(n, codesOffCap - 1), codesOffPType); - if (firstOffset > lastOffset || lastOffset > compressedBytes.byteSize()) { - throw new VortexException(EncodingId.VORTEX_FSST, "invalid code offsets: [" - + firstOffset + ", " + lastOffset + ") of " + compressedBytes.byteSize()); - } - long outPos = 0L; - long batchStart = firstOffset; - for (long i = 0; i < n; i += ROWS_PER_DECODE_BATCH) { - long batchEndRow = Math.min(i + ROWS_PER_DECODE_BATCH, n); - long batchEnd = batchEndRow == n - ? lastOffset - : readUnsigned(codesOffsetsSeg, batchEndRow, codesOffPType); - outPos = decompressor.decompress(compressedBytes, batchStart, batchEnd, outBytes, outPos); - batchStart = batchEnd; - } - if (outPos != totalUncompressed) { - throw new VortexException(EncodingId.VORTEX_FSST, "decoded " + outPos - + " bytes but uncompressed lengths claim " + totalUncompressed); - } - } else { - // Defensive broadcast path (1 < physical offsets < n + 1): ranges wrap around the - // physical elements, so decode row by row and record the offsets the decode actually - // produced, overwriting the claimed prefix sums. - long outPos = 0L; - for (long i = 0; i < n; i++) { - long cStart = readUnsigned(codesOffsetsSeg, i % codesOffCap, codesOffPType); - long cEnd = readUnsigned(codesOffsetsSeg, (i + 1) % codesOffCap, codesOffPType); - outPos = decompressor.decompress(compressedBytes, cStart, cEnd, outBytes, outPos); - outOffsets.setAtIndex(VortexFormat.LE_INT, i + 1, (int) outPos); - } - } - - return new VarBinOffsetArray(ctx.dtype(), n, - outBytes.asSlice(0, totalUncompressed).asReadOnly(), outOffsets.asReadOnly(), PType.I32); - } - - /// Writes the prefix sums of the `count` unsigned per-row lengths in `seg` into `outOffsets` - /// (I32 slots `1 .. count`; slot 0 is the caller's) and returns the total. Branch-split per - /// ptype and on the broadcast case so the per-element loop bodies carry no switch and no - /// modulo (hot-loop rule). - private static long writeDecodedOffsets(MemorySegment seg, MemorySegment outOffsets, long count, - PType ptype, long cap) { - if (cap == 0 && count > 0) { - throw new VortexException(EncodingId.VORTEX_FSST, "empty uncompressed-lengths child"); - } - long sum = 0L; - if (cap >= count) { - switch (ptype) { - case U8 -> { - for (long i = 0; i < count; i++) { - sum += Byte.toUnsignedLong(seg.get(ValueLayout.JAVA_BYTE, i)); - outOffsets.setAtIndex(VortexFormat.LE_INT, i + 1, (int) sum); - } - } - case U16 -> { - for (long i = 0; i < count; i++) { - sum += Short.toUnsignedLong(seg.get(VortexFormat.LE_SHORT, i * 2)); - outOffsets.setAtIndex(VortexFormat.LE_INT, i + 1, (int) sum); - } - } - case U32 -> { - for (long i = 0; i < count; i++) { - sum += Integer.toUnsignedLong(seg.getAtIndex(VortexFormat.LE_INT, i)); - outOffsets.setAtIndex(VortexFormat.LE_INT, i + 1, (int) sum); - } - } - case I32 -> { - for (long i = 0; i < count; i++) { - sum += seg.getAtIndex(VortexFormat.LE_INT, i); - outOffsets.setAtIndex(VortexFormat.LE_INT, i + 1, (int) sum); - } - } - case I64, U64 -> { - for (long i = 0; i < count; i++) { - sum += seg.getAtIndex(VortexFormat.LE_LONG, i); - outOffsets.setAtIndex(VortexFormat.LE_INT, i + 1, (int) sum); - } - } - default -> throw new VortexException(EncodingId.VORTEX_FSST, "unsupported ptype " + ptype); - } - return sum; - } - // Broadcast slow path (constant-encoded child): only ever a handful of physical elements. - for (long i = 0; i < count; i++) { - sum += readUnsigned(seg, i % cap, ptype); - outOffsets.setAtIndex(VortexFormat.LE_INT, i + 1, (int) sum); - } - return sum; - } - - private static long readUnsigned(MemorySegment seg, long idx, PType ptype) { - return switch (ptype) { - case U8 -> Byte.toUnsignedLong(seg.get(ValueLayout.JAVA_BYTE, idx)); - case U16 -> Short.toUnsignedLong(seg.get(VortexFormat.LE_SHORT, idx * 2)); - case U32 -> Integer.toUnsignedLong(seg.getAtIndex(VortexFormat.LE_INT, idx)); - case I32 -> seg.getAtIndex(VortexFormat.LE_INT, idx); - case I64, U64 -> seg.getAtIndex(VortexFormat.LE_LONG, idx); - default -> throw new VortexException(EncodingId.VORTEX_FSST, "unsupported ptype " + ptype); - }; + // No decompression happens here: every per-row length, code range, and bounds check is + // deferred to LazyFsstVarBinArray's accessors, which run only for rows a caller actually + // reads (ADR 0026) — mirroring VarBinArrays' "offsets are not scanned at decode time" + // convention for the other VarBinArray implementations. + return new LazyFsstVarBinArray(ctx.dtype(), n, decompressor, compressedBytes, + uncompLensSeg, uncompLenPType, codesOffsetsSeg, codesOffPType); } } diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/decode/FsstEncodingDecoderTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/decode/FsstEncodingDecoderTest.java index 0a183201..a3f0f3bb 100644 --- a/reader/src/test/java/io/github/dfa1/vortex/reader/decode/FsstEncodingDecoderTest.java +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/decode/FsstEncodingDecoderTest.java @@ -16,6 +16,8 @@ import java.lang.foreign.MemorySegment; import java.lang.foreign.ValueLayout; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; @@ -73,10 +75,11 @@ void decodesTinyNode_provingHarnessIsSound() { @Test void constantCodeOffsetsManyRows_decodesEmptyColumn() { // Given — a column of all-empty rows: every compressed offset collapses to one physical - // value (a constant-encoded offsets child, capacity 1), while the row count exceeds - // ROWS_PER_DECODE_BATCH (256). This is the regression guard for the batched fast path, - // which read offsets at row indices 256, 512, ... — off the end of the 1-element offsets - // segment — until the capacity-1 case was split out to decode nothing instead. + // value (a constant-encoded offsets child, capacity 1), across enough rows (300) that a + // batched decode loop would have walked past a 1-element offsets segment — the + // regression this fixture used to guard when decode() batch-decoded eagerly. The lazy + // per-row decoder has no batch loop to run off the end of, but the broadcast offsets + // must still resolve correctly for every row. int rows = 300; long[] symbols = {}; byte[] symbolLengths = {}; @@ -93,6 +96,269 @@ void constantCodeOffsetsManyRows_decodesEmptyColumn() { assertThat(result.getString(0)).isEmpty(); assertThat(result.getString(rows - 1)).isEmpty(); } + + @Test + void bytesSegment_isNullSentinel_noContiguousBufferBacksLazyRows() { + // Given — no single flat buffer holds the lazily decoded rows (same convention as + // VarBinChunkedArray / VarBinRunEndArray / VarBinConstantArray). + long[] symbols = {packSymbol("ab")}; + byte[] symbolLengths = {2}; + byte[] compressed = {0x00}; + long[] uncompLengths = {2}; + long[] codeOffsets = {0, 1}; + + // When + VarBinArray result = decodeFsst(1, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // Then + assertThat(result.bytesSegment()).isSameAs(MemorySegment.NULL); + assertThat(result.segmentIfPresent()).isEmpty(); + } + + @Test + void limited_returnsShorterZeroCopyView() { + // Given + long[] symbols = {packSymbol("ab")}; + byte[] symbolLengths = {2}; + byte[] compressed = {0x00, 0x00}; + long[] uncompLengths = {2, 2}; + long[] codeOffsets = {0, 1, 2}; + VarBinArray sut = decodeFsst(2, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // When + VarBinArray result = sut.limited(1); + + // Then + assertThat(result.length()).isEqualTo(1); + assertThat(result.getString(0)).isEqualTo("ab"); + } + } + + @Nested + class Laziness { + + @Test + void decode_doesNotDecompressUntilRowIsAccessed() { + // Given — row 0 is well-formed ("ab"); row 1's code range points past the compressed + // buffer entirely, which would blow up any eager whole-column decode. decode() itself + // must not touch row 1's code range at all. + long[] symbols = {packSymbol("ab")}; + byte[] symbolLengths = {2}; + byte[] compressed = {0x00}; // only 1 code byte physically present + long[] uncompLengths = {2, 99}; + long[] codeOffsets = {0, 1, 50}; // row 1 = [1, 50) — far past the buffer + + // When + VarBinArray result = decodeFsst(2, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // Then — decode() succeeded (no eager validation of row 1), row 0 reads correctly, and + // only reading row 1 surfaces the malformed offsets. + assertThat(result.getString(0)).isEqualTo("ab"); + assertThatExceptionOfType(VortexException.class) + .isThrownBy(() -> result.getString(1)) + .withMessageContaining("invalid code offsets"); + } + + @Test + void forEachByteLength_neverTouchesCodeOffsetsOrCompressedBytes() { + // Given — the codes-offsets child and compressed buffer are both nonsense (a range past + // the buffer's end), but forEachByteLength only ever reads the uncompressed-lengths + // child, so it must return the claimed lengths without tripping over either. + long[] symbols = {}; + byte[] symbolLengths = {}; + byte[] compressed = {0x00}; + long[] uncompLengths = {2, 3}; + long[] codeOffsets = {0, 50, 100}; // wildly out of range for `compressed` + VarBinArray sut = decodeFsst(2, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // When + List lengths = new ArrayList<>(); + sut.forEachByteLength(lengths::add); + + // Then + assertThat(lengths).containsExactly(2, 3); + } + } + + @Nested + class AdversarialEdgeCases { + + @Test + void zeroRows_decodesToNothing() { + // Given — n == 0: no row is ever legal to read, and forEachByteLength must not try to + // divide by a broadcast capacity of 0 just because the length child happens to be empty. + long[] symbols = {}; + byte[] symbolLengths = {}; + byte[] compressed = {}; + long[] uncompLengths = {}; + long[] codeOffsets = {}; + + // When + VarBinArray result = decodeFsst(0, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + List lengths = new ArrayList<>(); + result.forEachByteLength(lengths::add); + + // Then + assertThat(result.length()).isZero(); + assertThat(lengths).isEmpty(); + } + + @Test + void singleEmptyRowAmidNonEmptyRows_decodesCorrectly() { + // Given — a genuinely empty row (start == end) sitting between two non-empty rows in the + // physical (non-broadcast) fast path — distinct from the existing all-empty-broadcast + // fixture, which never exercises a per-row zero-length range. + long[] symbols = {packSymbol("ab")}; + byte[] symbolLengths = {2}; + byte[] compressed = {0x00, 0x00}; + long[] uncompLengths = {2, 0, 2}; + long[] codeOffsets = {0, 1, 1, 2}; // row1 = [1, 1): empty + VarBinArray result = decodeFsst(3, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // When / Then + assertThat(result.getString(0)).isEqualTo("ab"); + assertThat(result.getString(1)).isEmpty(); + assertThat(result.getString(2)).isEqualTo("ab"); + } + + @Test + void laterRowIsIndependentlyAccessible_whenAnEarlierRowIsStructurallyInvalid() { + // Given — row 0's own offsets are descending (structurally invalid on its own), but row + // 1's range is well-formed. True random access means reading row 1 must not require + // validating — or even touching — row 0 first. + long[] symbols = {packSymbol("ab")}; + byte[] symbolLengths = {2}; + byte[] compressed = {0x00, 0x00}; + long[] uncompLengths = {99, 2}; // row 0's claim is irrelevant; never read + long[] codeOffsets = {3, 1, 2}; // row 0 = [3, 1): invalid; row 1 = [1, 2): valid + VarBinArray result = decodeFsst(2, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // When / Then — row 1 alone, never touching row 0. + assertThat(result.getString(1)).isEqualTo("ab"); + } + + @Test + void getBytes_repeatedAccessOfSameRowIsIdempotent() { + // Given — nothing about decoding row i should mutate shared state that a later re-read + // of the same row would observe. + long[] symbols = {packSymbol("ab")}; + byte[] symbolLengths = {2}; + byte[] compressed = {0x00}; + long[] uncompLengths = {2}; + long[] codeOffsets = {0, 1}; + VarBinArray result = decodeFsst(1, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // When + String first = result.getString(0); + String second = result.getString(0); + + // Then + assertThat(first).isEqualTo("ab"); + assertThat(second).isEqualTo("ab"); + } + + @Test + void getByteLength_claimedLengthExactlyAtCodeRangeBound_passes() { + // Given — a maximum-length-8 FSST symbol decoded from a single compressed byte: the + // claimed length (8) exactly equals the code range's bound (1 code * 8 bytes/code). The + // boundary must be inclusive (<=), not exclusive. + long[] symbols = {packSymbol("abcdefgh")}; + byte[] symbolLengths = {8}; + byte[] compressed = {0x00}; + long[] uncompLengths = {8}; + long[] codeOffsets = {0, 1}; + VarBinArray result = decodeFsst(1, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // When / Then + assertThat(result.getByteLength(0)).isEqualTo(8); + assertThat(result.getString(0)).isEqualTo("abcdefgh"); + } + + @Test + void getByteLength_negativeCodeStart_throws() { + // Given — an adversarial signed code-offset that is itself negative (not merely + // descending relative to its pair). + long[] symbols = {packSymbol("ab")}; + byte[] symbolLengths = {2}; + byte[] compressed = {0x00, 0x00}; + long[] uncompLengths = {2}; + long[] codeOffsets = {-1, 5}; + VarBinArray result = decodeFsst(1, symbols, symbolLengths, compressed, + PType.I32, uncompLengths, PType.I32, codeOffsets); + + // When / Then + assertThatExceptionOfType(VortexException.class) + .isThrownBy(() -> result.getByteLength(0)) + .withMessageContaining("invalid code offsets"); + } + + @Test + void getBytes_codeRangeImpliesLengthBeyondIntMax_throwsWithoutAllocating() { + // Given — a single row whose code range is exactly large enough that the FSST 8-bytes- + // per-compressed-byte bound crosses Integer.MAX_VALUE - 7. The guard must reject this + // BEFORE sizing a scratch array, or the (int) cast would silently wrap into a negative + // size instead of a clean VortexException. + long codeLen = 256L * 1024 * 1024; // 256 MiB of code bytes => 2 GiB max output + long[] symbols = {}; + byte[] symbolLengths = {}; + long[] uncompLengths = {0}; // irrelevant: the bound check fires first + long[] codeOffsets = {0, codeLen}; + VarBinArray result = decodeFsstWithHugeCompressedBuffer(codeLen, symbols, symbolLengths, + PType.U8, uncompLengths, PType.I64, codeOffsets); + + // When / Then + assertThatExceptionOfType(VortexException.class) + .isThrownBy(() -> result.getBytes(0)) + .withMessageContaining("decoded length too large"); + } + + @Test + void getBytes_truncatedEscapeAtBufferEnd_throwsVortexExceptionNotRaw() { + // Given — a code range whose last code is the escape marker (0xFF) with no following + // literal byte, and the buffer ends exactly there. Decompressor reads one byte past the + // declared range to fetch the (nonexistent) literal; that read falls off the segment + // entirely and must surface as a VortexException, not a raw IndexOutOfBoundsException + // (the reader parses untrusted binary input; CLAUDE.md security contract). + long[] symbols = {}; + byte[] symbolLengths = {}; + byte[] compressed = {(byte) 0xFF}; // ESCAPE, no literal follows + long[] uncompLengths = {1}; + long[] codeOffsets = {0, 1}; + VarBinArray result = decodeFsst(1, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // When / Then + assertThatExceptionOfType(VortexException.class) + .isThrownBy(() -> result.getBytes(0)); + } + + @Test + void getBytes_codeReferencesUnknownSymbol_throwsVortexExceptionNotRaw() { + // Given — a code byte that names a symbol index past the trained table's size (the table + // has 1 symbol, code 0; this row's code is 5). Decompressor indexes packedSymbols[5] on + // an array of length 1, which must surface as a VortexException, not a raw + // ArrayIndexOutOfBoundsException. + long[] symbols = {packSymbol("ab")}; + byte[] symbolLengths = {2}; + byte[] compressed = {0x05}; // code 5: no such symbol was trained + long[] uncompLengths = {2}; + long[] codeOffsets = {0, 1}; + VarBinArray result = decodeFsst(1, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // When / Then + assertThatExceptionOfType(VortexException.class) + .isThrownBy(() -> result.getBytes(0)); + } } @Nested @@ -100,8 +366,9 @@ class Guards { @Test void rowCountAtIntMax_throws() { - // Given — n == Integer.MAX_VALUE overflows the I32 offset array sizing; the guard rejects - // it before any allocation. Empty buffers/children suffice because the check fires first. + // Given — n == Integer.MAX_VALUE would overflow the codes-offsets child's `n + 1` + // element count; the guard rejects it before that child is even decoded. Empty + // buffers/children suffice because the check fires first. long[] symbols = {}; byte[] symbolLengths = {}; byte[] compressed = {}; @@ -109,45 +376,51 @@ void rowCountAtIntMax_throws() { long[] uncompLengths = {0}; long[] codeOffsets = {0}; - // When / Then + // When / Then — decode() itself throws; this is the one guard that still fires eagerly. assertThatExceptionOfType(VortexException.class) .isThrownBy(() -> decodeFsst(Integer.MAX_VALUE, symbols, symbolLengths, compressed, PType.U8, uncompLengths, PType.U8, codeOffsets)) .withMessageContaining("row count too large"); } + // The remaining guards below all validate untrusted per-row content (lengths, code + // offsets), which — like every other VarBinArray's offsets (see VarBinArrays.checkedLength) + // — is deliberately not scanned at decode() time. decode() always succeeds; the malformed + // input surfaces the first time the offending row is actually read, which is what proves + // these rows are not being decoded up front. See the Laziness nested class above. + @Test void emptyUncompressedLengthsChild_throws() { - // Given — n > 0 but the uncompressed-lengths child has zero physical elements. There is - // no per-row length to read, so the prefix-sum loop cannot proceed. + // Given — n > 0 but the uncompressed-lengths child has zero physical elements. long[] symbols = {packSymbol("ab")}; byte[] symbolLengths = {2}; byte[] compressed = {0x00}; long[] uncompLengths = {}; // empty length child with n == 1 long[] codeOffsets = {0, 1}; + VarBinArray result = decodeFsst(1, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); - // When / Then + // When / Then — decode() succeeded; reading row 0's length is where this surfaces. assertThatExceptionOfType(VortexException.class) - .isThrownBy(() -> decodeFsst(1, symbols, symbolLengths, compressed, - PType.U8, uncompLengths, PType.U8, codeOffsets)) - .withMessageContaining("empty uncompressed-lengths child"); + .isThrownBy(() -> result.getByteLength(0)) + .withMessageContaining("empty"); } @Test void emptyCodesOffsetsChild_throws() { - // Given — n > 0 but the code-offsets child is empty. Without offsets there is no code - // range for any row, so the fast path's codesOffCap == 0 branch must reject it. + // Given — n > 0 but the code-offsets child is empty: no code range exists for any row. long[] symbols = {packSymbol("ab")}; byte[] symbolLengths = {2}; byte[] compressed = {0x00}; long[] uncompLengths = {2}; long[] codeOffsets = {}; // empty offsets child with n == 1 + VarBinArray result = decodeFsst(1, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); // When / Then assertThatExceptionOfType(VortexException.class) - .isThrownBy(() -> decodeFsst(1, symbols, symbolLengths, compressed, - PType.U8, uncompLengths, PType.U8, codeOffsets)) - .withMessageContaining("empty codes-offsets child"); + .isThrownBy(() -> result.getByteLength(0)) + .withMessageContaining("empty"); } @Test @@ -159,11 +432,12 @@ void codeOffsetsDescending_throws() { byte[] compressed = {0x00, 0x00}; long[] uncompLengths = {2}; long[] codeOffsets = {2, 0}; // first (2) > last (0) + VarBinArray result = decodeFsst(1, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); // When / Then assertThatExceptionOfType(VortexException.class) - .isThrownBy(() -> decodeFsst(1, symbols, symbolLengths, compressed, - PType.U8, uncompLengths, PType.U8, codeOffsets)) + .isThrownBy(() -> result.getByteLength(0)) .withMessageContaining("invalid code offsets"); } @@ -176,11 +450,12 @@ void codeOffsetsPastCompressedBuffer_throws() { byte[] compressed = {0x00}; // only 1 code byte physically present long[] uncompLengths = {2}; long[] codeOffsets = {0, 5}; // claims 5 code bytes, buffer has 1 + VarBinArray result = decodeFsst(1, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); // When / Then assertThatExceptionOfType(VortexException.class) - .isThrownBy(() -> decodeFsst(1, symbols, symbolLengths, compressed, - PType.U8, uncompLengths, PType.U8, codeOffsets)) + .isThrownBy(() -> result.getByteLength(0)) .withMessageContaining("invalid code offsets"); } @@ -188,17 +463,19 @@ void codeOffsetsPastCompressedBuffer_throws() { void decodedLengthMismatchesClaim_throws() { // Given — a valid symbol table plus a code stream that decodes to FEWER bytes than the // uncompressed-lengths child claims: the stream is a single "ab" symbol (2 bytes) but the - // length child claims 5. The fast-path post-decode check must catch the disagreement. + // length child claims 5 (within the code range's 8-byte max, so getByteLength alone would + // not catch this — only an actual decode-and-compare does). long[] symbols = {packSymbol("ab")}; byte[] symbolLengths = {2}; byte[] compressed = {0x00}; // decodes to exactly "ab" (2 bytes) long[] uncompLengths = {5}; // but the child claims 5 bytes long[] codeOffsets = {0, 1}; + VarBinArray result = decodeFsst(1, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); // When / Then assertThatExceptionOfType(VortexException.class) - .isThrownBy(() -> decodeFsst(1, symbols, symbolLengths, compressed, - PType.U8, uncompLengths, PType.U8, codeOffsets)) + .isThrownBy(() -> result.getString(0)) .withMessageContaining("uncompressed lengths claim"); } @@ -206,57 +483,57 @@ void decodedLengthMismatchesClaim_throws() { void constantCodeOffsetsButLengthsClaimBytes_throws() { // Given — a capacity-1 (constant) offsets child means every row's code range is empty, // so the column must decode to zero bytes; but the uncompressed-lengths child claims - // non-zero bytes per row. The two disagree, so the file is malformed and must be - // rejected rather than silently returning zero-filled output. + // non-zero bytes per row. The two disagree, so the file is malformed; getByteLength's + // code-range cross-check catches it without needing to decode anything. int rows = 300; long[] symbols = {packSymbol("ab")}; byte[] symbolLengths = {2}; byte[] compressed = {0x00}; long[] uncompLengths = {5}; // broadcast: claims 5 bytes per row long[] codeOffsets = {0}; // capacity 1: all ranges empty + VarBinArray result = decodeFsst(rows, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); // When / Then assertThatExceptionOfType(VortexException.class) - .isThrownBy(() -> decodeFsst(rows, symbols, symbolLengths, compressed, - PType.U8, uncompLengths, PType.U8, codeOffsets)) - .withMessageContaining("constant code offsets imply an empty column"); + .isThrownBy(() -> result.getByteLength(0)) + .withMessageContaining("decoded length too large"); } @Test void totalUncompressedOverflowsIntMax_throws() { - // Given — an I64 length child whose few huge values sum past Integer.MAX_VALUE. The - // output offsets are I32 prefix sums, so this would silently wrap; the step-1 guard must - // reject it. Placed BEFORE the outBytes allocation, so the test must fail without ever - // attempting a multi-gigabyte allocation. Two rows of ~1.5 GB each overflow I32. + // Given — an I64 length child whose per-row value is individually implausible for its + // 1-code range (max 8 bytes), even though each row's raw value alone fits comfortably + // under Integer.MAX_VALUE. getByteLength's code-range cross-check catches it per row. long huge = 1_500_000_000L; long[] symbols = {packSymbol("ab")}; byte[] symbolLengths = {2}; byte[] compressed = {0x00, 0x00}; - long[] uncompLengths = {huge, huge}; // sum == 3e9 > Integer.MAX_VALUE + long[] uncompLengths = {huge, huge}; long[] codeOffsets = {0, 1, 2}; + VarBinArray result = decodeFsst(2, symbols, symbolLengths, compressed, + PType.I64, uncompLengths, PType.U8, codeOffsets); // When / Then assertThatExceptionOfType(VortexException.class) - .isThrownBy(() -> decodeFsst(2, symbols, symbolLengths, compressed, - PType.I64, uncompLengths, PType.U8, codeOffsets)) + .isThrownBy(() -> result.getByteLength(0)) .withMessageContaining("decoded length too large"); } @Test void negativeTotalUncompressed_throws() { - // Given — an adversarial I32 length whose value is negative; the running prefix sum goes - // negative, which the step-1 guard must reject (a negative size would otherwise flow into - // the allocation and throw a raw exception instead of a sanitized VortexException). + // Given — an adversarial I32 length whose value is negative. long[] symbols = {packSymbol("ab")}; byte[] symbolLengths = {2}; byte[] compressed = {0x00}; - long[] uncompLengths = {-1}; // I32 -1 => sum == -1 + long[] uncompLengths = {-1}; // I32 -1 long[] codeOffsets = {0, 1}; + VarBinArray result = decodeFsst(1, symbols, symbolLengths, compressed, + PType.I32, uncompLengths, PType.U8, codeOffsets); // When / Then assertThatExceptionOfType(VortexException.class) - .isThrownBy(() -> decodeFsst(1, symbols, symbolLengths, compressed, - PType.I32, uncompLengths, PType.U8, codeOffsets)) + .isThrownBy(() -> result.getByteLength(0)) .withMessageContaining("decoded length too large"); } } @@ -271,9 +548,27 @@ void negativeTotalUncompressed_throws() { private static VarBinArray decodeFsst(long rowCount, long[] symbols, byte[] symbolLengths, byte[] compressed, PType uncompLenPType, long[] uncompLengths, PType codesOffPType, long[] codeOffsets) { + return decodeFsst(rowCount, symbols, symbolLengths, bytes(compressed), + uncompLenPType, uncompLengths, codesOffPType, codeOffsets); + } + + /// Variant of [#decodeFsst] for the code-range-overflow guard: a real 256 MiB+ compressed + /// buffer is unwieldy as a `byte[]` literal, and its content is irrelevant — only its + /// `byteSize()` needs to bound the (huge) code range under test. Allocated zero-filled, so it + /// never round-trips through the decompressor before the guard rejects it. + private static VarBinArray decodeFsstWithHugeCompressedBuffer(long compressedByteSize, + long[] symbols, byte[] symbolLengths, PType uncompLenPType, long[] uncompLengths, + PType codesOffPType, long[] codeOffsets) { + MemorySegment compressedSeg = Arena.ofAuto().allocate(compressedByteSize); + return decodeFsst(1, symbols, symbolLengths, compressedSeg, + uncompLenPType, uncompLengths, codesOffPType, codeOffsets); + } + + private static VarBinArray decodeFsst(long rowCount, long[] symbols, byte[] symbolLengths, + MemorySegment compressedSeg, PType uncompLenPType, long[] uncompLengths, + PType codesOffPType, long[] codeOffsets) { MemorySegment symbolsSeg = leLongs(symbols); MemorySegment symbolLensSeg = bytes(symbolLengths); - MemorySegment compressedSeg = bytes(compressed); MemorySegment uncompLensSeg = typedSegment(uncompLenPType, uncompLengths); MemorySegment codeOffsetsSeg = typedSegment(codesOffPType, codeOffsets); MemorySegment[] segs = {symbolsSeg, symbolLensSeg, compressedSeg, uncompLensSeg, codeOffsetsSeg};