This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new e6e42a76954 branch-4.2: [improvement](be) Decode SNII zstd and raw
.prx windows faster #69770 (#69788)
e6e42a76954 is described below
commit e6e42a76954f5c28bd51915e33b703deb63e269d
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Oct 29 16:30:42 2026 +0800
branch-4.2: [improvement](be) Decode SNII zstd and raw .prx windows faster
#69770 (#69788)
Cherry-picked from #69770
Co-authored-by: Jack <[email protected]>
---
.../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 5c541276dd8..3455b3921b5 100644
--- a/be/src/storage/index/inverted/snii/format/prx_pod.cpp
+++ b/be/src/storage/index/inverted/snii/format/prx_pod.cpp
@@ -964,9 +964,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;
@@ -976,24 +989,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");
@@ -1055,7 +1098,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");
@@ -1070,7 +1113,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,
@@ -1145,7 +1190,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 abb12c1cbf0..d2135be63af 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>
@@ -543,6 +545,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]
