diff --git a/be/src/storage/index/inverted/snii/encoding/byte_source.h b/be/src/storage/index/inverted/snii/encoding/byte_source.h index c6d4c9f83a2..86d0568f10d 100644 --- a/be/src/storage/index/inverted/snii/encoding/byte_source.h +++ b/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(p - s_.data()); } + size_t remaining() const { return s_.size() - pos_; } size_t position() const { return pos_; } bool eof() const { return pos_ == s_.size(); } diff --git a/be/src/storage/index/inverted/snii/encoding/zstd_codec.cpp b/be/src/storage/index/inverted/snii/encoding/zstd_codec.cpp index 4139f49f957..ffc3461103c 100644 --- a/be/src/storage/index/inverted/snii/encoding/zstd_codec.cpp +++ b/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(_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( + "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(expected_uncomp_len); + if (_buffer == nullptr) { + return Status::Error( + "zstd decompress: cannot allocate {} bytes", expected_uncomp_len); + } + _capacity = expected_uncomp_len; + } + const size_t n = ZSTD_decompressDCtx(static_cast(_context), _buffer.get(), + expected_uncomp_len, input.data(), input.size()); + if (ZSTD_isError(n)) { + return Status::Error( + std::string("zstd decompress: ") + ZSTD_getErrorName(n)); + } + if (n != expected_uncomp_len) { + return Status::Error( + "zstd decompressed length mismatch"); + } + *out = Slice(_buffer.get(), expected_uncomp_len); + return Status::OK(); +} + } // namespace doris::snii diff --git a/be/src/storage/index/inverted/snii/encoding/zstd_codec.h b/be/src/storage/index/inverted/snii/encoding/zstd_codec.h index c5ebf4206f8..c62056dbd4d 100644 --- a/be/src/storage/index/inverted/snii/encoding/zstd_codec.h +++ b/be/src/storage/index/inverted/snii/encoding/zstd_codec.h @@ -22,6 +22,7 @@ #include #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* out); Status zstd_decompress(Slice input, size_t expected_uncomp_len, std::vector* 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 _buffer; + size_t _capacity = 0; +}; + } // namespace doris::snii diff --git a/be/src/storage/index/inverted/snii/format/prx_decode_stats.h b/be/src/storage/index/inverted/snii/format/prx_decode_stats.h index eda0af5cc0c..18fe996f8de 100644 --- a/be/src/storage/index/inverted/snii/format/prx_decode_stats.h +++ b/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 diff --git a/be/src/storage/index/inverted/snii/format/prx_pod.cpp b/be/src/storage/index/inverted/snii/format/prx_pod.cpp index 8500e23e543..4fe4de76998 100644 --- a/be/src/storage/index/inverted/snii/format/prx_pod.cpp +++ b/be/src/storage/index/inverted/snii/format/prx_pod.cpp @@ -1057,9 +1057,22 @@ Status decode_payload_csr_selective(Slice plain, std::span 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 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( + "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(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(pos_flat->size())); ++next_doc; } + src.set_cursor(p); if (!src.eof()) return Status::Error( "prx: trailing bytes after payload"); @@ -1148,7 +1191,7 @@ Status decode_csr_frame(const PrxFrameView& encoded, std::span d bool decode_all_docs, bool all_docs_selected, std::vector* pos_flat, std::vector* pos_off, PrxDecodeStats* stats, PrxDecodedShape* shape, - PrxCsrAllocationGate* allocation_gate) { + PrxCsrAllocationGate* allocation_gate, ZstdDecompressor* zstd) { if (!decode_all_docs && allocation_gate != nullptr) { return Status::Error( "prx: selective decode cannot use an allocation gate"); @@ -1163,7 +1206,9 @@ Status decode_csr_frame(const PrxFrameView& encoded, std::span d std::vector 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* 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 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); diff --git a/be/src/storage/index/inverted/snii/reader/snii_index_source.cpp b/be/src/storage/index/inverted/snii/reader/snii_index_source.cpp index c3131c987e2..51df3f0e2c6 100644 --- a/be/src/storage/index/inverted/snii/reader/snii_index_source.cpp +++ b/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(_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)); } diff --git a/be/src/storage/index/inverted/snii/reader/snii_index_source.h b/be/src/storage/index/inverted/snii/reader/snii_index_source.h index ef46af713f1..b8bd6306569 100644 --- a/be/src/storage/index/inverted/snii/reader/snii_index_source.h +++ b/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 _terms; DictBlockCache _scoring_dict_cache; diff --git a/be/src/storage/index/inverted/snii/reader/snii_postings_cursor.cpp b/be/src/storage/index/inverted/snii/reader/snii_postings_cursor.cpp index 4fd33f6a4f2..9ac4a57eff9 100644 --- a/be/src/storage/index/inverted/snii/reader/snii_postings_cursor.cpp +++ b/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(_idx.reader()); @@ -533,7 +535,7 @@ Status SniiPostingsCursor::_ensure_positions(std::span 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 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()) { diff --git a/be/src/storage/index/inverted/snii/reader/snii_postings_cursor.h b/be/src/storage/index/inverted/snii/reader/snii_postings_cursor.h index 64c6d653e92..e10a51211e4 100644 --- a/be/src/storage/index/inverted/snii/reader/snii_postings_cursor.h +++ b/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 _own_wave; diff --git a/be/test/storage/index/inverted/snii/encoding/zstd_codec_test.cpp b/be/test/storage/index/inverted/snii/encoding/zstd_codec_test.cpp index 73f00f1c128..d4b325456e5 100644 --- a/be/test/storage/index/inverted/snii/encoding/zstd_codec_test.cpp +++ b/be/test/storage/index/inverted/snii/encoding/zstd_codec_test.cpp @@ -19,11 +19,15 @@ #include +#include #include #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 in; @@ -49,3 +53,72 @@ TEST(SniiZstd, EmptyInput) { ASSERT_TRUE(zstd_decompress(Slice(comp), 0, &decomp).ok()); EXPECT_TRUE(decomp.empty()); } + +namespace { +std::vector pattern(size_t n, int seed) { + std::vector v(n); + for (size_t i = 0; i < n; ++i) { + v[i] = static_cast((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 in = pattern(n, static_cast(n)); + std::vector 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 in = pattern(1000, 3); + std::vector 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()) + << wrong_length.to_string(); + + std::vector 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()) + << 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 in = pattern(8 << 20, 5); + std::vector 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(in.size())); + } + doris::thread_context()->thread_mem_tracker_mgr->flush_untracked_mem(); + EXPECT_LT(tracker->consumption() - before, static_cast(in.size())); +} diff --git a/be/test/storage/index/inverted/snii/format/prx_pod_test.cpp b/be/test/storage/index/inverted/snii/format/prx_pod_test.cpp index 56b3aa1f869..a3a3a46d54a 100644 --- a/be/test/storage/index/inverted/snii/format/prx_pod_test.cpp +++ b/be/test/storage/index/inverted/snii/format/prx_pod_test.cpp @@ -19,9 +19,11 @@ #include +#include #include #include #include +#include #include #include #include @@ -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 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(codec))); + const std::vector encoded = BuildProfileFrame(docs, codec); + std::vector expected_flat; + std::vector expected_offsets; + ExpectedSelectiveCsr(docs, ordinals, &expected_flat, &expected_offsets); + std::vector flat; + std::vector 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 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());