This is an automated email from the ASF dual-hosted git repository.
airborne12 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 6f6a8266211 [improvement](be) Decode SNII zstd and raw .prx windows
faster (#69770)
6f6a8266211 is described below
commit 6f6a82662114076eb1bc2ee78efd425994c87bcd
Author: Jack <[email protected]>
AuthorDate: Thu Oct 29 09:10:06 2026 +0800
[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
---
.../index/inverted/snii/encoding/byte_source.h | 6 ++
.../index/inverted/snii/encoding/zstd_codec.cpp | 38 ++++++++++
.../index/inverted/snii/encoding/zstd_codec.h | 21 ++++++
.../index/inverted/snii/format/prx_decode_stats.h | 7 ++
.../storage/index/inverted/snii/format/prx_pod.cpp | 80 +++++++++++++++++-----
.../inverted/snii/reader/snii_index_source.cpp | 2 +-
.../index/inverted/snii/reader/snii_index_source.h | 4 ++
.../inverted/snii/reader/snii_postings_cursor.cpp | 8 ++-
.../inverted/snii/reader/snii_postings_cursor.h | 5 +-
.../inverted/snii/encoding/zstd_codec_test.cpp | 73 ++++++++++++++++++++
.../index/inverted/snii/format/prx_pod_test.cpp | 76 ++++++++++++++++++++
11 files changed, 298 insertions(+), 22 deletions(-)
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<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(); }
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<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
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 <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
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<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);
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<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));
}
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<std::string, Term> _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<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()) {
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<SniiReadWave> _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 <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()));
+}
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 <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());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
