[improvement](be) Decode SNII zstd and raw .prx windows faster (#69770)

### What problem does this PR solve?

Issue Number: None

Related PR: #69649, #69769

Problem Summary: A phrase query decodes the positions of the selected
documents of every .prx window it reads. Timed per window on a 10M-row
AgentLogs load, a zstd window cost 18.6 us against 1.4 us for a PFOR
window: 11.7 us to decode the plain payload, 4.8 us to decompress, and
about 2 us more because every window created and freed its own zstd
context (about 96 KB) and a zero-filled output vector. Two commits:

1. **Decode the plain payload without a Status per document.**
`decode_payload_csr_selective` made at least one out-of-line call
returning a Status for every document of the window, selected or not
(`get_varint32_fast`, `skip_varints`, `decode_delta_run`). The loop now
keeps the cursor in a register and handles the usual cases inline: a
single-byte count, walking past the deltas of a document that is not
selected, and a run of single-byte deltas. Anything else -- a multi-byte
value, a run that does not fit, a prefix sum that would overflow -- goes
to the checked decoders from the same cursor, so they decide and report
as before. The output vector grows as before; nothing is reserved up
front.
2. **Reuse one zstd context and buffer per index source.**
`ZstdDecompressor` keeps one context, created at the first zstd frame,
and one output buffer that only grows and is not zero-filled
(decompression writes every byte before it is read). The buffer comes
from the Doris allocator (`DorisUniqueBufferPtr`), so the decoding
task's memory tracker accounts for it while it is held.
`SniiIndexSource` owns one, and the cursors it opens pass it to the .prx
decoder through `PrxDecodeContext`. The cursors of a source decode on
its thread one window at a time, and the decoded payload is consumed
before the next window is read. A source lives for one query on one
segment, so a query holds at most one context and one buffer per segment
it reads, and only if it decodes a zstd window; before, each window
created and released them. Compaction decodes through an allocation
gate, which keeps the old path, as does a cursor given no decompressor.

Hot queries, old against new read path in one process (run-time switch
in an experiment build, 6 paired cycles), on 99M AgentLogs rows written
with the smallest frames, as every segment written before #69649 is
(mostly zstd windows): DS7 (unfiltered phrase) 0.87x, A6 0.89x, A10
0.93x, DS4 (phrase prefix) 0.94x, A9 0.95x, DS1 0.96x; repeating the old
path against itself gave 0.93-1.05. The gain follows the zstd windows a
query reads: the phrase queries of AgentLogs written with #69649 read
PFOR windows and do not change, and on TextBench, whose large windows
stay zstd, the time per window is the decompression itself and the
change is within noise. Every query returns the same result.

### Release note

None

be/src/storage/index/inverted/snii/encoding/byte_source.h

@@ -80,6 +80,12 @@ public:
     Status get_zigzag(int64_t* v);
     Status get_bytes(size_t n, Slice* out);
 
+    // Raw cursor access for a hot loop that decodes many small runs without a Status per value and
+    // hands anything unusual back to the checked decoders above (set_cursor, then the call).
+    const uint8_t* cursor() const { return s_.data() + pos_; }
+    const uint8_t* end() const { return s_.data() + s_.size(); }
+    void set_cursor(const uint8_t* p) { pos_ = static_cast<size_t>(p - s_.data()); }
+
     size_t remaining() const { return s_.size() - pos_; }
     size_t position() const { return pos_; }
     bool eof() const { return pos_ == s_.size(); }

be/src/storage/index/inverted/snii/encoding/zstd_codec.cpp

@@ -49,4 +49,42 @@ Status zstd_decompress(Slice input, size_t expected_uncomp_len, std::vector<uint
     return Status::OK();
 }
 
+ZstdDecompressor::~ZstdDecompressor() {
+    ZSTD_freeDCtx(static_cast<ZSTD_DCtx*>(_context));
+}
+
+Status ZstdDecompressor::decompress(Slice input, size_t expected_uncomp_len, Slice* out) {
+    if (_context == nullptr) {
+        _context = ZSTD_createDCtx();
+        if (_context == nullptr) {
+            return Status::Error<ErrorCode::MEM_ALLOC_FAILED, false>(
+                    "zstd decompress: cannot create a context");
+        }
+    }
+    if (expected_uncomp_len > _capacity) {
+        // Every byte is written by the decompression before it is read, so the buffer is not
+        // zero-filled. The old buffer goes first, so the two are never held together.
+        _buffer.reset();
+        _capacity = 0;
+        _buffer = make_unique_buffer<uint8_t>(expected_uncomp_len);
+        if (_buffer == nullptr) {
+            return Status::Error<ErrorCode::MEM_ALLOC_FAILED, false>(
+                    "zstd decompress: cannot allocate {} bytes", expected_uncomp_len);
+        }
+        _capacity = expected_uncomp_len;
+    }
+    const size_t n = ZSTD_decompressDCtx(static_cast<ZSTD_DCtx*>(_context), _buffer.get(),
+                                         expected_uncomp_len, input.data(), input.size());
+    if (ZSTD_isError(n)) {
+        return Status::Error<ErrorCode::INVERTED_INDEX_FILE_CORRUPTED, false>(
+                std::string("zstd decompress: ") + ZSTD_getErrorName(n));
+    }
+    if (n != expected_uncomp_len) {
+        return Status::Error<ErrorCode::INVERTED_INDEX_FILE_CORRUPTED, false>(
+                "zstd decompressed length mismatch");
+    }
+    *out = Slice(_buffer.get(), expected_uncomp_len);
+    return Status::OK();
+}
+
 } // namespace doris::snii

be/src/storage/index/inverted/snii/encoding/zstd_codec.h

@@ -22,6 +22,7 @@
 #include <vector>
 
 #include "common/status.h"
+#include "core/custom_allocator.h"
 #include "storage/index/inverted/snii/common/slice.h"
 
 namespace doris::snii {
@@ -30,4 +31,24 @@ namespace doris::snii {
 Status zstd_compress(Slice input, int level, std::vector<uint8_t>* out);
 Status zstd_decompress(Slice input, size_t expected_uncomp_len, std::vector<uint8_t>* out);
 
+// Decompresses many small frames in a row -- one per .prx window of a posting -- with one
+// decompression context and one output buffer, instead of a context created and a zeroed vector
+// allocated for every frame. The view returned by decompress() is valid until the next call.
+// Not thread-safe; owned by a single reader. The buffer lasts as long as the decompressor and is as
+// large as the largest frame decoded, so it is charged to the memory tracker of the decoding task.
+class ZstdDecompressor {
+public:
+    ZstdDecompressor() = default;
+    ~ZstdDecompressor();
+    ZstdDecompressor(const ZstdDecompressor&) = delete;
+    ZstdDecompressor& operator=(const ZstdDecompressor&) = delete;
+
+    Status decompress(Slice input, size_t expected_uncomp_len, Slice* out);
+
+private:
+    void* _context = nullptr; // ZSTD_DCtx, created on the first frame
+    DorisUniqueBufferPtr<uint8_t> _buffer;
+    size_t _capacity = 0;
+};
+
 } // namespace doris::snii

be/src/storage/index/inverted/snii/format/prx_decode_stats.h

@@ -23,6 +23,10 @@
 
 #include "common/status.h"
 
+namespace doris::snii {
+class ZstdDecompressor;
+} // namespace doris::snii
+
 namespace doris::snii::format {
 
 // Optional allocation seam for full CSR decode callers that must gate every
@@ -73,6 +77,9 @@ struct PrxDecodeContext {
     PrxDecodeStats* stats = nullptr;
     PrxDecodedShape* shape = nullptr;
     PrxCsrAllocationGate* allocation_gate = nullptr;
+    // Optional: a reader that decodes many windows in a row passes its own decompressor, so the
+    // zstd windows reuse one context and one buffer. An allocation gate takes precedence.
+    ZstdDecompressor* zstd = nullptr;
 };
 
 } // namespace doris::snii::format

be/src/storage/index/inverted/snii/format/prx_pod.cpp

@@ -1057,9 +1057,22 @@ Status decode_payload_csr_selective(Slice plain, std::span<const uint32_t> doc_o
     uint64_t total_pos = 0;
     uint32_t decoded_max_frequency = 0;
     bool decoded_zero_frequency = false;
+    // The loop below runs once per document of the window, selected or not, so it keeps the cursor
+    // in a register and builds no Status on the usual path: a single-byte count, a skipped run of
+    // short varints, a selected run of single-byte deltas. Anything else -- a multi-byte value, a
+    // run that does not fit, a prefix sum that would overflow -- is handed to the checked decoders
+    // with the cursor where it was, so they decide and report exactly as they always did.
+    const uint8_t* const end = src.end();
+    const uint8_t* p = src.cursor();
     for (uint32_t d = 0; d < doc_count; ++d) {
         uint32_t pos_count = 0;
-        RETURN_IF_ERROR(src.get_varint32_fast(&pos_count));
+        if (p < end && *p < 0x80) {
+            pos_count = *p++;
+        } else {
+            src.set_cursor(p);
+            RETURN_IF_ERROR(src.get_varint32(&pos_count));
+            p = src.cursor();
+        }
         decoded_max_frequency = std::max(decoded_max_frequency, pos_count);
         decoded_zero_frequency |= pos_count == 0;
         total_pos += pos_count;
@@ -1069,24 +1082,54 @@ Status decode_payload_csr_selective(Slice plain, std::span<const uint32_t> doc_o
         }
         const bool selected = next_doc < doc_ordinals.size() && doc_ordinals[next_doc] == d;
         if (!selected) {
-            // Skip this doc's position deltas without decoding them -- the CSR
-            // layout is sequential so we must advance past them, but only the
-            // candidate (selected) docs' positions are ever used. With a sparse
-            // candidate set (the common phrase / phrase-prefix case after docid
-            // narrowing) most docs in a window are skipped, so this avoids the
-            // dominant varint-decode cost.
-            RETURN_IF_ERROR(src.skip_varints(pos_count));
+            // The CSR layout is sequential, so a skipped doc's deltas are walked past without being
+            // decoded. With a sparse candidate set most docs of a window are skipped.
+            const uint8_t* q = p;
+            uint32_t skipped = 0;
+            for (; skipped < pos_count && q < end; ++skipped) {
+                while (q < end && (*q & 0x80) != 0) {
+                    ++q;
+                }
+                if (q >= end) {
+                    break;
+                }
+                ++q;
+            }
+            if (skipped != pos_count) {
+                return Status::Error<ErrorCode::INVERTED_INDEX_FILE_CORRUPTED, false>(
+                        "byte_source: varint skip past end");
+            }
+            p = q;
             continue;
         }
-        // Selected doc: decode its `pos_count` ascending position deltas with a
-        // tight inline prefix-sum decoder (single-byte fast path, no per-value
-        // get_varint32/Status call chain). This is the CPU hotspot for narrowed
-        // phrase/phrase-prefix candidate sets, where each selected doc's varint
-        // run dominates after non-selected docs are skipped.
-        RETURN_IF_ERROR(src.decode_delta_run(pos_count, pos_flat));
+        // Selected doc: its `pos_count` ascending deltas become positions. A single-byte delta is
+        // below 0x80, so the running sum cannot overflow unless it is checked.
+        const size_t flat_before = pos_flat->size();
+        const uint8_t* q = p;
+        uint32_t position = 0;
+        uint32_t decoded = 0;
+        if (static_cast<size_t>(end - q) >= pos_count) {
+            for (; decoded < pos_count; ++decoded) {
+                const uint8_t byte = *q;
+                if (byte >= 0x80 || __builtin_add_overflow(position, uint32_t {byte}, &position)) {
+                    break;
+                }
+                pos_flat->push_back(position);
+                ++q;
+            }
+        }
+        if (decoded == pos_count) {
+            p = q;
+        } else {
+            pos_flat->resize(flat_before);
+            src.set_cursor(p);
+            RETURN_IF_ERROR(src.decode_delta_run(pos_count, pos_flat));
+            p = src.cursor();
+        }
         pos_off->push_back(static_cast<uint32_t>(pos_flat->size()));
         ++next_doc;
     }
+    src.set_cursor(p);
     if (!src.eof())
         return Status::Error<ErrorCode::INVERTED_INDEX_FILE_CORRUPTED, false>(
                 "prx: trailing bytes after payload");
@@ -1148,7 +1191,7 @@ Status decode_csr_frame(const PrxFrameView& encoded, std::span<const uint32_t> d
                         bool decode_all_docs, bool all_docs_selected,
                         std::vector<uint32_t>* pos_flat, std::vector<uint32_t>* pos_off,
                         PrxDecodeStats* stats, PrxDecodedShape* shape,
-                        PrxCsrAllocationGate* allocation_gate) {
+                        PrxCsrAllocationGate* allocation_gate, ZstdDecompressor* zstd) {
     if (!decode_all_docs && allocation_gate != nullptr) {
         return Status::Error<ErrorCode::INVALID_ARGUMENT, false>(
                 "prx: selective decode cannot use an allocation gate");
@@ -1163,7 +1206,9 @@ Status decode_csr_frame(const PrxFrameView& encoded, std::span<const uint32_t> d
 
     std::vector<uint8_t> local_decompressed;
     Slice plain = encoded.payload;
-    if (encoded.codec == PrxCodec::kZstd) {
+    if (encoded.codec == PrxCodec::kZstd && allocation_gate == nullptr && zstd != nullptr) {
+        RETURN_IF_ERROR(zstd->decompress(encoded.payload, encoded.uncompressed_length, &plain));
+    } else if (encoded.codec == PrxCodec::kZstd) {
         std::vector<uint8_t>* decompressed = &local_decompressed;
         if (allocation_gate != nullptr) {
             RETURN_IF_ERROR(allocation_gate->reserve_decompression(encoded.uncompressed_length,
@@ -1238,7 +1283,8 @@ Status read_prx_window_csr_impl(ByteSource* source, std::span<const uint32_t> do
     RETURN_IF_ERROR(decode_csr_frame(encoded, doc_ordinals, decode_all_docs, all_docs_selected,
                                      pos_flat, pos_off, stats,
                                      context == nullptr ? nullptr : context->shape,
-                                     context == nullptr ? nullptr : context->allocation_gate));
+                                     context == nullptr ? nullptr : context->allocation_gate,
+                                     context == nullptr ? nullptr : context->zstd));
     if (collect_stats) {
         // Exclude the logical-selection scan from decode timing.
         frame_stats.decode_ns = elapsed_ns(decode_start);

be/src/storage/index/inverted/snii/reader/snii_index_source.cpp

@@ -124,7 +124,7 @@ Status SniiIndexSource::_cursor(Term& term, bool positions, bool scoring, SniiRe
     }
     *out = std::make_unique<SniiPostingsCursor>(_idx, term.hit.entry, term.hit.frq_base,
                                                 term.hit.prx_base, positions, scoring, norms, wave,
-                                                _prx_stats);
+                                                _prx_stats, &_zstd);
     if (auto prelude = term.prelude.lock()) {
         (*out)->set_prelude(std::move(prelude));
     }

be/src/storage/index/inverted/snii/reader/snii_index_source.h

@@ -28,6 +28,7 @@
 
 #include "common/status.h"
 #include "storage/index/inverted/query/spi/index_source.h"
+#include "storage/index/inverted/snii/encoding/zstd_codec.h"
 #include "storage/index/inverted/snii/format/frq_prelude.h"
 #include "storage/index/inverted/snii/format/norms_pod.h"
 #include "storage/index/inverted/snii/reader/dict_block_cache.h"
@@ -89,6 +90,9 @@ private:
 
     const LogicalIndexReader& _idx;
     format::PrxDecodeStats* _prx_stats;
+    // The cursors decode on the source's thread one window at a time, so they share one zstd
+    // context and buffer instead of creating them for every window.
+    ZstdDecompressor _zstd;
     SniiReadWave _wave;
     std::unordered_map<std::string, Term> _terms;
     DictBlockCache _scoring_dict_cache;

be/src/storage/index/inverted/snii/reader/snii_postings_cursor.cpp

@@ -98,7 +98,8 @@ uint64_t SniiReadWave::held_bytes() const {
 SniiPostingsCursor::SniiPostingsCursor(const LogicalIndexReader& idx, format::DictEntry entry,
                                        uint64_t frq_base, uint64_t prx_base, bool positions,
                                        bool scoring, const format::NormsPodReader* norms,
-                                       SniiReadWave* wave, format::PrxDecodeStats* prx_stats)
+                                       SniiReadWave* wave, format::PrxDecodeStats* prx_stats,
+                                       ZstdDecompressor* zstd)
         : _idx(idx),
           _entry(std::move(entry)),
           _frq_base(frq_base),
@@ -108,6 +109,7 @@ SniiPostingsCursor::SniiPostingsCursor(const LogicalIndexReader& idx, format::Di
           _norms(norms),
           _wave(wave),
           _prx_stats(prx_stats),
+          _zstd(zstd),
           _kind(_posting_kind(_entry)) {
     if (_wave == nullptr) {
         _own_wave = std::make_unique<SniiReadWave>(_idx.reader());
@@ -533,7 +535,7 @@ Status SniiPostingsCursor::_ensure_positions(std::span<const uint32_t> ordinals)
     const uint32_t window = _current_window;
     RETURN_IF_ERROR(_ensure_prx(window));
     ByteSource source(_windows[window].prx);
-    format::PrxDecodeContext context {.stats = _prx_stats};
+    format::PrxDecodeContext context {.stats = _prx_stats, .zstd = _zstd};
     if (ordinals.empty() || ordinals.size() == _doc_count) {
         RETURN_IF_ERROR(format::read_prx_window_csr(&source, &_pos_flat, &_pos_off, &context));
     } else {
@@ -795,7 +797,7 @@ Status SniiPostingsCursor::_decode_selected(std::span<const uint32_t> ordinals,
     RETURN_IF_ERROR(_ensure_prx(_current_window));
     ByteSource source(_windows[_current_window].prx);
     format::PrxDecodedShape shape;
-    format::PrxDecodeContext context {.stats = _prx_stats, .shape = &shape};
+    format::PrxDecodeContext context {.stats = _prx_stats, .shape = &shape, .zstd = _zstd};
     RETURN_IF_ERROR(format::read_prx_window_csr_selective(&source, ordinals, &_selected_flat,
                                                           &_selected_off, &context));
     if (!source.eof()) {

be/src/storage/index/inverted/snii/reader/snii_postings_cursor.h

@@ -85,7 +85,8 @@ public:
     SniiPostingsCursor(const LogicalIndexReader& idx, format::DictEntry entry, uint64_t frq_base,
                        uint64_t prx_base, bool positions, bool scoring,
                        const format::NormsPodReader* norms, SniiReadWave* wave = nullptr,
-                       format::PrxDecodeStats* prx_stats = nullptr);
+                       format::PrxDecodeStats* prx_stats = nullptr,
+                       ZstdDecompressor* zstd = nullptr);
     ~SniiPostingsCursor() override;
 
     // A prelude the caller already read; call before the first read.
@@ -187,6 +188,8 @@ private:
     const format::NormsPodReader* _norms;
     SniiReadWave* _wave;
     format::PrxDecodeStats* _prx_stats;
+    // Decompresses the zstd .prx windows when given; each window decompresses on its own otherwise.
+    ZstdDecompressor* _zstd;
     Kind _kind;
     // The wave of a cursor given none.
     std::unique_ptr<SniiReadWave> _own_wave;

be/test/storage/index/inverted/snii/encoding/zstd_codec_test.cpp

@@ -19,11 +19,15 @@
 
 #include <gtest/gtest.h>
 
+#include <algorithm>
 #include <vector>
 
 #include "common/status.h"
+#include "runtime/memory/mem_tracker_limiter.h"
+#include "runtime/thread_context.h"
 
 using namespace doris::snii;
+using doris::AttachTask; // SCOPED_ATTACH_TASK names it unqualified
 
 TEST(SniiZstd, RoundTrip) {
     std::vector<uint8_t> in;
@@ -49,3 +53,72 @@ TEST(SniiZstd, EmptyInput) {
     ASSERT_TRUE(zstd_decompress(Slice(comp), 0, &decomp).ok());
     EXPECT_TRUE(decomp.empty());
 }
+
+namespace {
+std::vector<uint8_t> pattern(size_t n, int seed) {
+    std::vector<uint8_t> v(n);
+    for (size_t i = 0; i < n; ++i) {
+        v[i] = static_cast<uint8_t>((i * 31 + seed) % 13);
+    }
+    return v;
+}
+} // namespace
+
+// One decompressor serves frames that grow and shrink; every output equals its input.
+TEST(SniiZstdDecompressor, ReusedAcrossFramesOfDifferentSizes) {
+    ZstdDecompressor decompressor;
+    for (size_t n : {100UL, 5000UL, 300UL, 0UL, 20000UL, 7UL}) {
+        const std::vector<uint8_t> in = pattern(n, static_cast<int>(n));
+        std::vector<uint8_t> comp;
+        ASSERT_TRUE(zstd_compress(Slice(in), 3, &comp).ok());
+        Slice out;
+        ASSERT_TRUE(decompressor.decompress(Slice(comp), in.size(), &out).ok()) << n;
+        ASSERT_EQ(out.size(), in.size());
+        EXPECT_TRUE(std::equal(in.begin(), in.end(), out.data())) << n;
+    }
+}
+
+// The errors are the ones zstd_decompress reports, and a failed frame does not spoil the next one.
+TEST(SniiZstdDecompressor, ReportsCorruptionAndRecovers) {
+    const std::vector<uint8_t> in = pattern(1000, 3);
+    std::vector<uint8_t> comp;
+    ASSERT_TRUE(zstd_compress(Slice(in), 3, &comp).ok());
+    ZstdDecompressor decompressor;
+    Slice out;
+
+    const doris::Status wrong_length = decompressor.decompress(Slice(comp), in.size() - 1, &out);
+    EXPECT_TRUE(wrong_length.is<doris::ErrorCode::INVERTED_INDEX_FILE_CORRUPTED>())
+            << wrong_length.to_string();
+
+    std::vector<uint8_t> garbage = comp;
+    garbage[garbage.size() / 2] ^= 0xFF;
+    garbage.resize(garbage.size() - 3);
+    const doris::Status corrupt = decompressor.decompress(Slice(garbage), in.size(), &out);
+    EXPECT_TRUE(corrupt.is<doris::ErrorCode::INVERTED_INDEX_FILE_CORRUPTED>())
+            << corrupt.to_string();
+
+    ASSERT_TRUE(decompressor.decompress(Slice(comp), in.size(), &out).ok());
+    ASSERT_EQ(out.size(), in.size());
+    EXPECT_TRUE(std::equal(in.begin(), in.end(), out.data()));
+}
+
+// The buffer a decompressor keeps between frames is as large as the largest frame it decoded, so
+// it is charged to the memory tracker of the task that decodes, until the decompressor goes away.
+TEST(SniiZstdDecompressor, ChargesItsBufferToTheTaskMemTracker) {
+    const std::vector<uint8_t> in = pattern(8 << 20, 5);
+    std::vector<uint8_t> comp;
+    ASSERT_TRUE(zstd_compress(Slice(in), 3, &comp).ok());
+    auto tracker = doris::MemTrackerLimiter::create_shared(doris::MemTrackerLimiter::Type::OTHER,
+                                                           "UT-SniiZstdDecompressor");
+    SCOPED_ATTACH_TASK(tracker);
+    const int64_t before = tracker->consumption();
+    {
+        ZstdDecompressor decompressor;
+        Slice out;
+        ASSERT_TRUE(decompressor.decompress(Slice(comp), in.size(), &out).ok());
+        doris::thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+        EXPECT_GE(tracker->consumption() - before, static_cast<int64_t>(in.size()));
+    }
+    doris::thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+    EXPECT_LT(tracker->consumption() - before, static_cast<int64_t>(in.size()));
+}

be/test/storage/index/inverted/snii/format/prx_pod_test.cpp

@@ -19,9 +19,11 @@
 
 #include <gtest/gtest.h>
 
+#include <algorithm>
 #include <cstdint>
 #include <cstring>
 #include <numeric>
+#include <random>
 #include <string>
 #include <type_traits>
 #include <vector>
@@ -550,6 +552,80 @@ TEST(PrxPodTest, CodecSelectionMatrixMatchesCsrOracleAtLowAndHighTf) {
     }
 }
 
+// The plain-payload selective decoder keeps a fast path for single-byte counts and deltas and hands
+// everything else to the checked decoders. Random windows mix both kinds of value -- multi-byte
+// deltas, docs without positions, positions near the top of the 32-bit range -- and random
+// selections, and every result must equal the positions the window was built from.
+TEST(PrxPodTest, SelectivePlainDecodeMatchesInputForRandomWindows) {
+    std::mt19937 rng(20261028);
+    // A reader-owned decompressor, shared by every window below as a cursor shares it.
+    doris::snii::ZstdDecompressor shared_zstd;
+    for (int round = 0; round < 300; ++round) {
+        const uint32_t doc_count = 1 + rng() % 400;
+        const uint32_t style = rng() % 4;
+        PerDoc docs(doc_count);
+        for (auto& doc : docs) {
+            const uint32_t count = (style == 3 && rng() % 4 == 0) ? 0 : rng() % 6;
+            uint32_t position = rng() % 200;
+            for (uint32_t i = 0; i < count; ++i) {
+                doc.push_back(position);
+                uint32_t step = 1 + rng() % 100; // single-byte delta
+                if (style >= 1 && rng() % 8 == 0) {
+                    step = 128 + rng() % 20000; // two or three bytes
+                }
+                if (style == 2 && rng() % 40 == 0) {
+                    step = 1U << (21 + rng() % 8); // up to five bytes
+                }
+                position = (UINT32_MAX - position < step) ? position : position + step;
+            }
+            // keep the list strictly valid for the builder: ascending, duplicates allowed
+            std::sort(doc.begin(), doc.end());
+        }
+        std::vector<uint32_t> ordinals;
+        const uint32_t mode = rng() % 4;
+        for (uint32_t d = 0; d < doc_count; ++d) {
+            const bool take = mode == 0   ? (rng() % 10 == 0)
+                              : mode == 1 ? (rng() % 2 == 0)
+                              : mode == 2 ? true
+                                          : (d == 0 || d + 1 == doc_count);
+            if (take) {
+                ordinals.push_back(d);
+            }
+        }
+        for (PrxCodec codec : {PrxCodec::kRaw, PrxCodec::kZstd}) {
+            SCOPED_TRACE("round=" + std::to_string(round) + " docs=" + std::to_string(doc_count) +
+                         " style=" + std::to_string(style) + " mode=" + std::to_string(mode) +
+                         " codec=" + std::to_string(static_cast<uint8_t>(codec)));
+            const std::vector<uint8_t> encoded = BuildProfileFrame(docs, codec);
+            std::vector<uint32_t> expected_flat;
+            std::vector<uint32_t> expected_offsets;
+            ExpectedSelectiveCsr(docs, ordinals, &expected_flat, &expected_offsets);
+            std::vector<uint32_t> flat;
+            std::vector<uint32_t> offsets;
+            ByteSource source((Slice(encoded)));
+            ASSERT_TRUE(read_prx_window_csr_selective(&source, ordinals, &flat, &offsets).ok());
+            EXPECT_EQ(flat, expected_flat);
+            EXPECT_EQ(offsets, expected_offsets);
+
+            // The same window through the shared decompressor, selectively and in full.
+            PrxDecodeContext context {.zstd = &shared_zstd};
+            ByteSource reused_source((Slice(encoded)));
+            ASSERT_TRUE(read_prx_window_csr_selective(&reused_source, ordinals, &flat, &offsets,
+                                                      &context)
+                                .ok());
+            EXPECT_EQ(flat, expected_flat);
+            EXPECT_EQ(offsets, expected_offsets);
+            std::vector<uint32_t> all(doc_count);
+            std::iota(all.begin(), all.end(), 0U);
+            ExpectedSelectiveCsr(docs, all, &expected_flat, &expected_offsets);
+            ByteSource full_source((Slice(encoded)));
+            ASSERT_TRUE(read_prx_window_csr(&full_source, &flat, &offsets, &context).ok());
+            EXPECT_EQ(flat, expected_flat);
+            EXPECT_EQ(offsets, expected_offsets);
+        }
+    }
+}
+
 TEST(PrxPodTest, ZstdDeclaredLengthMismatchCommitsNoStats) {
     ByteSink sink;
     ASSERT_TRUE(build_prx_window(PerDoc {{1, 3}, {7}}, 3, &sink).ok());