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 655b46eb3eb [fix](be) Preserve SNII NULL-docid memory accounting
(#66880)
655b46eb3eb is described below
commit 655b46eb3eb3675c4481d5505998505ec836bcdb
Author: Jack <[email protected]>
AuthorDate: Tue Oct 27 18:30:07 2026 +0800
[fix](be) Preserve SNII NULL-docid memory accounting (#66880)
### What problem does this PR solve?
Issue Number: None
Related PR: #66052
Problem Summary: SNII released the ingestion writer NULL-docid memory
charge before flush, then copied the still-live vector into
`LogicalIndexWriter` without a `Reservation`. A one-million-NULL segment
retained a 4 MiB docid vector while the build reporter observed only
about 674 KiB of bitmap scratch, and the handoff briefly held two
complete vectors. This PR carries the existing move-only vector and
`Reservation` through `IndexFileWriter` and `SniiCompoundWriter`,
precharges geometric vector replacements during growth, and releases the
charge only after the NULL bitmap consumes the docids. A real
one-million-NULL compound build verifies reporter current/peak bytes and
terminal zero balance.
---
be/src/storage/index/index_file_writer.cpp | 5 +-
be/src/storage/index/index_file_writer.h | 2 +-
.../index/inverted/snii/snii_index_writer.cpp | 49 +++++----
.../index/inverted/snii/snii_index_writer.h | 11 +-
.../inverted/snii/writer/logical_index_writer.cpp | 3 -
.../inverted/snii/writer/logical_index_writer.h | 3 +-
.../inverted/snii/writer/snii_compound_writer.cpp | 27 ++++-
.../inverted/snii/writer/snii_compound_writer.h | 8 +-
.../snii/compaction/snii_streamed_session_test.cpp | 52 +++++++++
.../index/inverted/snii/snii_writer_test.cpp | 118 +++++++++++++++++++++
.../snii/writer/spimi_streaming_writer_test.cpp | 2 +-
11 files changed, 234 insertions(+), 46 deletions(-)
diff --git a/be/src/storage/index/index_file_writer.cpp
b/be/src/storage/index/index_file_writer.cpp
index 24195f6250f..30b81302f62 100644
--- a/be/src/storage/index/index_file_writer.cpp
+++ b/be/src/storage/index/index_file_writer.cpp
@@ -325,7 +325,7 @@ void IndexFileWriter::_release_snii_blob_directories() {
}
Status IndexFileWriter::add_snii_index(const TabletIndex* index_meta, uint32_t
doc_count,
- std::vector<uint32_t> null_docids,
+ doris::snii::writer::TrackedNullDocids
null_docids,
doris::snii::writer::SpimiTermBuffer*
const term_buffer,
doris::snii::format::IndexConfig
index_config,
SniiAddIndexOptions options,
@@ -348,13 +348,12 @@ Status IndexFileWriter::add_snii_index(const TabletIndex*
index_meta, uint32_t d
input.index_suffix = index_meta->get_index_suffix();
input.config = index_config;
input.doc_count = doc_count;
- input.null_docids = std::move(null_docids);
input.encoded_norms = std::move(options.encoded_norms);
input.gram_scheme = options.gram_scheme;
input.term_source = term_buffer;
input.mem_reporter = mem_reporter;
snii_resolve_index_write_params(options.is_direct_load,
!input.encoded_norms.empty(), &input);
- RETURN_IF_ERROR(_snii_compound_writer->add_logical_index(input));
+ RETURN_IF_ERROR(_snii_compound_writer->add_logical_index(input,
std::move(null_docids)));
++_snii_index_count;
return Status::OK();
}
diff --git a/be/src/storage/index/index_file_writer.h
b/be/src/storage/index/index_file_writer.h
index 51c254f68e9..ef58cd33e7b 100644
--- a/be/src/storage/index/index_file_writer.h
+++ b/be/src/storage/index/index_file_writer.h
@@ -118,7 +118,7 @@ public:
std::vector<uint8_t> encoded_norms;
};
Status add_snii_index(const TabletIndex* index_meta, uint32_t doc_count,
- std::vector<uint32_t> null_docids,
+ doris::snii::writer::TrackedNullDocids null_docids,
doris::snii::writer::SpimiTermBuffer* const
term_buffer,
doris::snii::format::IndexConfig index_config,
SniiAddIndexOptions options,
diff --git a/be/src/storage/index/inverted/snii/snii_index_writer.cpp
b/be/src/storage/index/inverted/snii/snii_index_writer.cpp
index 1e34461aa55..94766c5ade6 100644
--- a/be/src/storage/index/inverted/snii/snii_index_writer.cpp
+++ b/be/src/storage/index/inverted/snii/snii_index_writer.cpp
@@ -161,6 +161,7 @@ Status SniiIndexColumnWriter::init() {
::doris::snii::writer::BuildMemoryPopulation::kRegistered),
spill_threshold,
::doris::snii::writer::MemoryReporter::CapPolicy::kSpillThreshold,
static_cast<uint64_t>(config::snii_postings_workspace_bytes));
+ _null_docids_reservation = _memory_reporter->make_reservation();
_term_buffer = std::make_unique<::doris::snii::writer::SpimiTermBuffer>(
_has_positions, spill_threshold, _memory_reporter.get());
// Register with the process-wide memory limiter so writers can honor
advisory spill requests.
@@ -352,16 +353,24 @@ Status SniiIndexColumnWriter::add_array_values(size_t
field_size, const uint8_t*
return Status::OK();
}
-void SniiIndexColumnWriter::_report_null_docids_capacity(bool release_all) {
- if (_memory_reporter == nullptr) {
- return;
+Status SniiIndexColumnWriter::_reserve_null_docids_for_append(size_t count) {
+ const size_t need = _null_docids.size() + count;
+ // Repeated NULL runs reuse capacity until geometric growth is needed.
+ if (need <= _null_docids.capacity()) {
+ return Status::OK();
}
- const int64_t now =
- release_all ? 0 : static_cast<int64_t>(_null_docids.capacity() *
sizeof(uint32_t));
- if (now != _null_docids_charged_bytes) {
- _memory_reporter->report(now - _null_docids_charged_bytes);
- _null_docids_charged_bytes = now;
+ const size_t target_capacity = std::max(need, _null_docids.capacity() * 2);
+ // Uninitialized writers in unit tests have no reporter.
+ if (_memory_reporter == nullptr) {
+ _null_docids.reserve(target_capacity);
+ return Status::OK();
}
+ ::doris::snii::writer::MemoryReporter::Reservation replacement;
+ RETURN_IF_ERROR(_null_docids_reservation.prepare_replacement(
+ static_cast<uint64_t>(target_capacity) * sizeof(uint32_t),
&replacement));
+ _null_docids.reserve(target_capacity);
+ _null_docids_reservation = std::move(replacement);
+ return Status::OK();
}
// Hand back the token stream for one row. The two ways to get one differ in
ownership AND in
@@ -540,10 +549,7 @@ Status SniiIndexColumnWriter::add_nulls(uint32_t count) {
// the compaction ran 8+x slower than V3 (whose add_nulls is a roaring
// addRange). Doubling on overflow keeps the O(count) amortization AND
makes
// one large run pay at most one reallocation.
- const size_t need = _null_docids.size() + count;
- if (need > _null_docids.capacity()) {
- _null_docids.reserve(std::max(need, _null_docids.capacity() * 2));
- }
+ RETURN_IF_ERROR(_reserve_null_docids_for_append(count));
for (uint32_t i = 0; i < count; ++i) {
_null_docids.push_back(_rid + i);
}
@@ -552,7 +558,6 @@ Status SniiIndexColumnWriter::add_nulls(uint32_t count) {
_encoded_norms.insert(_encoded_norms.end(), count,
::doris::snii::query::encode_norm(0));
_report_encoded_norms_capacity();
}
- _report_null_docids_capacity();
return Status::OK();
}
@@ -564,13 +569,14 @@ Status SniiIndexColumnWriter::add_array_nulls(const
uint8_t* null_map, size_t nu
if (num_rows == 0 || null_map == nullptr) {
return Status::OK();
}
+ const size_t null_count = std::count(null_map, null_map + num_rows,
static_cast<uint8_t>(1));
+ RETURN_IF_ERROR(_reserve_null_docids_for_append(null_count));
const auto first_row = _rid - num_rows;
for (size_t i = 0; i < num_rows; ++i) {
if (null_map[i] == 1) {
_null_docids.push_back(cast_set<uint32_t>(first_row + i));
}
}
- _report_null_docids_capacity();
return Status::OK();
}
@@ -583,10 +589,6 @@ Status SniiIndexColumnWriter::finish() {
if (!status.ok()) {
return Status::InternalError("SNII term buffer error: {}",
status.to_string());
}
- // Ownership of _null_docids hands off to the flush below (transient,
- // flush-scoped); release the accumulation-phase charge so the retained
- // reporter (and the observation tracker behind it) balances to zero.
- _report_null_docids_capacity(/*release_all=*/true);
// A segment smaller than the sample budget reaches here still holding its
rows back. Solve
// on what there is and tokenize them, so a short segment is indexed like
any other -- and
// so `options.gram_scheme` below carries the rate its rows were actually
cut at.
@@ -599,8 +601,10 @@ Status SniiIndexColumnWriter::finish() {
options.encoded_norms = std::move(_encoded_norms);
}
status = _index_file_writer->add_snii_index(
- _index_meta, cast_set<uint32_t>(_rid), std::move(_null_docids),
_term_buffer.get(),
- _config, std::move(options), _memory_reporter.get());
+ _index_meta, cast_set<uint32_t>(_rid),
+
::doris::snii::writer::TrackedNullDocids(std::move(_null_docids_reservation),
+ std::move(_null_docids)),
+ _term_buffer.get(), _config, std::move(options),
_memory_reporter.get());
_report_encoded_norms_capacity(/*release_all=*/true);
// The tokenizer's buffers outlive this call (they die with _analyzer),
but the reporter is
// about to be handed to the index file writer and become a long-lived
process-wide counter,
@@ -622,12 +626,11 @@ Status
SniiIndexColumnWriter::_latch_analysis_failure(Status status) {
void SniiIndexColumnWriter::close_on_error() {
_term_buffer.reset();
- // Balance the observation-tracker mirror before dropping the reporter.
- _report_null_docids_capacity(/*release_all=*/true);
+ std::vector<uint32_t>().swap(_null_docids);
+ _null_docids_reservation.reset();
_report_encoded_norms_capacity(/*release_all=*/true);
_report_gram_buffers_capacity(/*release_all=*/true);
_memory_reporter.reset();
- _null_docids.clear();
std::vector<uint8_t>().swap(_encoded_norms);
}
diff --git a/be/src/storage/index/inverted/snii/snii_index_writer.h
b/be/src/storage/index/inverted/snii/snii_index_writer.h
index bc7b6ee176a..f522d984bee 100644
--- a/be/src/storage/index/inverted/snii/snii_index_writer.h
+++ b/be/src/storage/index/inverted/snii/snii_index_writer.h
@@ -113,9 +113,7 @@ private:
// Mirror of the above, run once the row's tokens are consumed: closes an
owned stream, or
// charges the reused gram tokenizer's settled buffer capacity to the
memory reporter.
void _finish_plain_lane_row(lucene::analysis::TokenStream* token_stream,
bool owned);
- // Mirrors _null_docids' capacity into _memory_reporter (delta-charged);
- // release_all zeroes the charge (finish() handoff / close_on_error()).
- void _report_null_docids_capacity(bool release_all = false);
+ Status _reserve_null_docids_for_append(size_t count);
void _report_encoded_norms_capacity(bool release_all = false);
// Mirrors the reused gram tokenizer's buffer capacity into
_memory_reporter
// (delta-charged), same shape as the two above. A no-op until the reusable
@@ -187,15 +185,10 @@ private:
// analyzed row, and always null off the gram lane.
inverted_index::GramTokenizer* _gram_tokenizer = nullptr;
std::unique_ptr<::doris::snii::writer::MemoryReporter> _memory_reporter;
+ ::doris::snii::writer::MemoryReporter::Reservation
_null_docids_reservation;
std::unique_ptr<::doris::snii::writer::SpimiTermBuffer> _term_buffer;
std::vector<uint32_t> _null_docids;
std::vector<uint8_t> _encoded_norms;
- // Bytes of _null_docids capacity currently mirrored into _memory_reporter
- // (and through it the SNII index-build observation tracker). Re-charged on
- // growth in add_nulls / add_array_nulls, released in finish() /
close_on_error() --
- // without it a large interleaved-null segment accumulates untracked RSS
the
- // Limiter cannot see.
- int64_t _null_docids_charged_bytes = 0;
int64_t _encoded_norms_charged_bytes = 0;
int64_t _gram_buffers_charged_bytes = 0;
Status _failure_status = Status::OK();
diff --git a/be/src/storage/index/inverted/snii/writer/logical_index_writer.cpp
b/be/src/storage/index/inverted/snii/writer/logical_index_writer.cpp
index e9f93783f34..2a540a37fcd 100644
--- a/be/src/storage/index/inverted/snii/writer/logical_index_writer.cpp
+++ b/be/src/storage/index/inverted/snii/writer/logical_index_writer.cpp
@@ -522,9 +522,6 @@ uint32_t high_df_floor_for(const SniiIndexInput& in) {
format::kHighDfDigestDivisor);
}
-LogicalIndexWriter::LogicalIndexWriter(const SniiIndexInput& in)
- : LogicalIndexWriter(in,
TrackedNullDocids(std::vector<uint32_t>(in.null_docids))) {}
-
LogicalIndexWriter::LogicalIndexWriter(const SniiIndexInput& in,
TrackedNullDocids null_docids)
: index_id_(in.index_id),
index_suffix_(in.index_suffix),
diff --git a/be/src/storage/index/inverted/snii/writer/logical_index_writer.h
b/be/src/storage/index/inverted/snii/writer/logical_index_writer.h
index fb13a27b0c7..7a3375f74f1 100644
--- a/be/src/storage/index/inverted/snii/writer/logical_index_writer.h
+++ b/be/src/storage/index/inverted/snii/writer/logical_index_writer.h
@@ -194,7 +194,7 @@ struct FreqStats {
// Builds and holds the section bytes + meta sub-sections for one logical
index.
class LogicalIndexWriter {
public:
- explicit LogicalIndexWriter(const SniiIndexInput& in);
+ LogicalIndexWriter(const SniiIndexInput& in, TrackedNullDocids
null_docids);
// Out-of-line: stream_state_ points at the private nested BlockState,
which
// is incomplete here (unique_ptr needs the complete type at destruction).
~LogicalIndexWriter();
@@ -281,7 +281,6 @@ public:
private:
friend class SniiStreamedIndexSession;
- LogicalIndexWriter(const SniiIndexInput& in, TrackedNullDocids
null_docids);
// One DICT block's directory record. The block's serialized bytes are
appended to
// the in-RAM dict buffer as soon as the block is cut; only this compact
summary
diff --git a/be/src/storage/index/inverted/snii/writer/snii_compound_writer.cpp
b/be/src/storage/index/inverted/snii/writer/snii_compound_writer.cpp
index dca6daa2ad6..11868e3c1ae 100644
--- a/be/src/storage/index/inverted/snii/writer/snii_compound_writer.cpp
+++ b/be/src/storage/index/inverted/snii/writer/snii_compound_writer.cpp
@@ -150,6 +150,31 @@ Status SniiCompoundWriter::inherit(const
reader::SniiRewriteSnapshot& snapshot,
}
Status SniiCompoundWriter::add_logical_index(const SniiIndexInput& in) {
+ MemoryReporter::Reservation null_docids_reservation =
+ in.mem_reporter == nullptr ? MemoryReporter::Reservation()
+ : in.mem_reporter->make_reservation();
+ if (in.mem_reporter != nullptr) {
+ // A hard cap must reject the copy before its allocation becomes
resident.
+ RETURN_IF_ERROR(null_docids_reservation.set_bytes(
+ static_cast<uint64_t>(in.null_docids.size()) *
sizeof(uint32_t)));
+ }
+ std::vector<uint32_t> null_docids(in.null_docids);
+ return _add_logical_index(
+ in, TrackedNullDocids(std::move(null_docids_reservation),
std::move(null_docids)));
+}
+
+Status SniiCompoundWriter::add_logical_index(const SniiIndexInput& in,
+ TrackedNullDocids null_docids) {
+ // Supplying both forms would make the caller's NULL-docid ownership
ambiguous.
+ if (!in.null_docids.empty()) {
+ return Status::Error<ErrorCode::INVALID_ARGUMENT, false>(
+ "compound: tracked NULL docids must not also be present in
input");
+ }
+ return _add_logical_index(in, std::move(null_docids));
+}
+
+Status SniiCompoundWriter::_add_logical_index(const SniiIndexInput& in,
+ TrackedNullDocids null_docids) {
if (out_ == nullptr)
return Status::Error<ErrorCode::INVALID_ARGUMENT, false>("compound:
null file writer");
if (finished_)
@@ -174,7 +199,7 @@ Status SniiCompoundWriter::add_logical_index(const
SniiIndexInput& in) {
}
}
RETURN_IF_ERROR(ensure_bootstrap());
- auto liw = std::make_unique<LogicalIndexWriter>(in);
+ auto liw = std::make_unique<LogicalIndexWriter>(in,
std::move(null_docids));
Placement p;
// The posting region streams DIRECTLY into the container during build()
-- no temp
// round-trip for the bulk -- followed immediately by this index's compact
DICT
diff --git a/be/src/storage/index/inverted/snii/writer/snii_compound_writer.h
b/be/src/storage/index/inverted/snii/writer/snii_compound_writer.h
index 49346ee2b46..fd05f057132 100644
--- a/be/src/storage/index/inverted/snii/writer/snii_compound_writer.h
+++ b/be/src/storage/index/inverted/snii/writer/snii_compound_writer.h
@@ -160,10 +160,11 @@ public:
// holding a partial prefix.
Status inherit(const reader::SniiRewriteSnapshot& snapshot,
io::FileReader* source);
- // Buffers one logical index: builds its section bytes and meta
sub-sections.
- // The actual file writing happens in finish() (single front-to-back pass).
- // The key (index_id, suffix) must not collide with an inherited one.
+ // Builds one logical index synchronously. The convenience overload copies
+ // and charges in.null_docids; ingestion uses the tracked overload to
transfer
+ // the already-charged allocation without a second resident vector.
Status add_logical_index(const SniiIndexInput& in);
+ Status add_logical_index(const SniiIndexInput& in, TrackedNullDocids
null_docids);
// Registers one opaque BLOB logical index (kind must not be kInverted).
// Registration is pure bookkeeping -- NOT A BYTE is written here, so it is
@@ -245,6 +246,7 @@ private:
Status write_index_aux_sections(LogicalIndexWriter& writer, Placement&
placement);
Status write_tail();
Status append(const std::vector<uint8_t>& bytes);
+ Status _add_logical_index(const SniiIndexInput& in, TrackedNullDocids
null_docids);
Status poison(Status status);
// Argument validation for one add_blob_index call; see the .cpp.
static Status validate_blob_registration(format::LogicalIndexKind kind,
diff --git
a/be/test/storage/index/inverted/snii/compaction/snii_streamed_session_test.cpp
b/be/test/storage/index/inverted/snii/compaction/snii_streamed_session_test.cpp
index b332a16e723..3eabe97a226 100644
---
a/be/test/storage/index/inverted/snii/compaction/snii_streamed_session_test.cpp
+++
b/be/test/storage/index/inverted/snii/compaction/snii_streamed_session_test.cpp
@@ -500,6 +500,58 @@ TEST(SniiStreamedWriterSessionTest,
NullDocidsStorageIsTransferredIntoStreamedWr
assert_ok(compound.finish());
}
+TEST(SniiStreamedWriterSessionTest,
OrdinaryLargeNullHandoffTracksRetainedCapacity) {
+ constexpr uint32_t kNullCount = 1U << 20;
+ SniiIndexInput input = empty_input(119, "ordinary_null_handoff",
kNullCount);
+ input.null_docids.resize(kNullCount);
+ std::iota(input.null_docids.begin(), input.null_docids.end(), 0U);
+ const uint64_t retained_null_bytes = input.null_docids.capacity() *
sizeof(uint32_t);
+ const uint64_t bitmap_build_bytes =
format::NullBitmapWriter::build_memory_upper_bound(
+ std::span<const uint32_t>(input.null_docids));
+
+ int64_t mirrored_bytes = 0;
+ int64_t peak_bytes = 0;
+ writer::MemoryReporter reporter([&](int64_t delta) {
+ mirrored_bytes += delta;
+ peak_bytes = std::max(peak_bytes, mirrored_bytes);
+ });
+ input.mem_reporter = &reporter;
+ std::vector<uint32_t> null_docids;
+ null_docids.swap(input.null_docids);
+ auto null_docids_reservation = reporter.make_reservation();
+ assert_ok(null_docids_reservation.set_bytes(retained_null_bytes));
+ writer::TrackedNullDocids
tracked_null_docids(std::move(null_docids_reservation),
+ std::move(null_docids));
+ EXPECT_EQ(reporter.current_bytes(), retained_null_bytes);
+ EXPECT_EQ(mirrored_bytes, retained_null_bytes);
+
+ MemoryFile file;
+ SniiCompoundWriter compound(&file);
+ assert_ok(compound.add_logical_index(input,
std::move(tracked_null_docids)));
+
+ EXPECT_GE(peak_bytes, static_cast<int64_t>(retained_null_bytes +
bitmap_build_bytes));
+ EXPECT_EQ(reporter.current_bytes(), 0);
+ EXPECT_EQ(mirrored_bytes, 0);
+ assert_ok(compound.finish());
+ EXPECT_EQ(reporter.current_bytes(), 0);
+ EXPECT_EQ(mirrored_bytes, 0);
+}
+
+TEST(SniiStreamedWriterSessionTest, OrdinaryNullCopyHonorsReporterCap) {
+ SniiIndexInput input = empty_input(120, "ordinary_null_copy_cap", 1024);
+ input.null_docids.resize(input.doc_count);
+ std::iota(input.null_docids.begin(), input.null_docids.end(), 0U);
+ writer::MemoryReporter reporter(nullptr, input.null_docids.size() *
sizeof(uint32_t) - 1);
+ input.mem_reporter = &reporter;
+ MemoryFile file;
+ SniiCompoundWriter compound(&file);
+ const Status status = compound.add_logical_index(input);
+ EXPECT_TRUE(status.is<ErrorCode::MEM_LIMIT_EXCEEDED>()) << status;
+ EXPECT_EQ(reporter.current_bytes(), 0);
+ EXPECT_EQ(file.bytes_written(), 0);
+ EXPECT_EQ(input.null_docids.size(), input.doc_count);
+}
+
TEST(SniiStreamedWriterSessionTest, NullBitmapFinalizationHonorsReporterCap) {
SniiIndexInput input = empty_input(95, "null_memory_cap", 2);
input.null_docids = {1};
diff --git a/be/test/storage/index/inverted/snii/snii_writer_test.cpp
b/be/test/storage/index/inverted/snii/snii_writer_test.cpp
index 3ec76ae8f8e..2f6208b5b52 100644
--- a/be/test/storage/index/inverted/snii/snii_writer_test.cpp
+++ b/be/test/storage/index/inverted/snii/snii_writer_test.cpp
@@ -31,6 +31,7 @@
#include <cstdint>
#include <map>
#include <optional>
+#include <span>
#include <string>
#include <vector>
@@ -38,8 +39,10 @@
#include "common/exception.h"
#include "common/status.h"
#include "gen_cpp/AgentService_types.h"
+#include "io/fs/local_file_system.h"
#include "runtime/exec_env.h"
#include "runtime/index_policy/index_policy_mgr.h"
+#include "storage/index/index_file_writer.h"
#include "storage/index/inverted/analysis/analyzer/analyzer.h"
#include "storage/index/inverted/analysis/analyzer/custom_analyzer.h"
#include "storage/index/inverted/analysis/gram/gram_extractor.h"
@@ -52,6 +55,7 @@
#include "storage/index/inverted/snii/format/dict_block_directory.h"
#include "storage/index/inverted/snii/format/dict_entry.h"
#include "storage/index/inverted/snii/format/format_constants.h"
+#include "storage/index/inverted/snii/format/null_bitmap.h"
#include "storage/index/inverted/snii/format/phrase_bigram.h"
#include "storage/index/inverted/snii/format/sampled_term_index.h"
#include "storage/index/inverted/snii/query/bm25_scorer.h"
@@ -505,6 +509,120 @@ TEST(SniiWriterNullDocids,
AddNullsGrowsGeometricallyNotQuadratically) {
EXPECT_TRUE(std::ranges::is_sorted(nulls));
}
+namespace {
+constexpr uint32_t kNullHandoffRows = 1U << 20;
+
+void add_null_handoff_rows(doris::segment_v2::SniiIndexColumnWriter* writer,
bool arrays) {
+ ASSERT_OK(writer->add_nulls(1));
+ if (arrays) {
+ std::vector<uint64_t> offsets(kNullHandoffRows + 1, 0);
+ std::vector<uint8_t> null_map(kNullHandoffRows, 1);
+ ASSERT_OK(writer->add_array_values(sizeof(doris::Slice), nullptr,
nullptr, offsets.data(),
+ kNullHandoffRows));
+ ASSERT_OK(writer->add_array_nulls(null_map.data(), kNullHandoffRows));
+ } else {
+ ASSERT_OK(writer->add_nulls(kNullHandoffRows));
+ }
+}
+
+void check_null_handoff_growth(const doris::segment_v2::SniiIndexColumnWriter&
writer,
+ int64_t mirrored_bytes, int64_t peak_bytes) {
+ const auto& nulls = writer.null_docids_for_test();
+ ASSERT_EQ(nulls.size(), kNullHandoffRows + 1);
+ EXPECT_EQ(nulls.front(), 0);
+ EXPECT_EQ(nulls.back(), kNullHandoffRows);
+ const int64_t retained_bytes = nulls.capacity() * sizeof(uint32_t);
+ EXPECT_EQ(writer.memory_reporter_for_test()->current_bytes(),
retained_bytes);
+ EXPECT_EQ(mirrored_bytes, retained_bytes);
+ EXPECT_GE(peak_bytes, retained_bytes +
static_cast<int64_t>(sizeof(uint32_t)));
+}
+
+void check_null_handoff_finish(doris::segment_v2::SniiIndexColumnWriter*
writer,
+ doris::segment_v2::IndexFileWriter*
index_writer,
+ const int64_t* mirrored_bytes, int64_t*
peak_bytes) {
+ auto* reporter = writer->memory_reporter_for_test();
+ const auto& nulls = writer->null_docids_for_test();
+ const int64_t retained_bytes = nulls.capacity() * sizeof(uint32_t);
+ const auto bitmap_bytes =
doris::snii::format::NullBitmapWriter::build_memory_upper_bound(
+ std::span<const uint32_t>(nulls));
+ *peak_bytes = *mirrored_bytes;
+ ASSERT_OK(writer->finish());
+ EXPECT_GE(*peak_bytes, retained_bytes + bitmap_bytes);
+ EXPECT_EQ(reporter->current_bytes(), 0);
+ EXPECT_EQ(*mirrored_bytes, 0);
+ EXPECT_EQ(writer->memory_reporter_for_test(), nullptr);
+ ASSERT_EQ(index_writer->_snii_memory_reporters.size(), 1);
+ EXPECT_EQ(index_writer->_snii_memory_reporters.front().get(), reporter);
+ ASSERT_OK(index_writer->begin_close());
+ ASSERT_OK(index_writer->finish_close());
+}
+} // namespace
+
+TEST(SniiWriterNullDocids,
InitializedScalarAndArrayHandoffPreservesAccounting) {
+ const std::string directory = "./ut_dir/snii_null_docids_handoff";
+ auto fs = doris::io::global_local_filesystem();
+ ASSERT_OK(fs->delete_directory(directory));
+ ASSERT_OK(fs->create_directory(directory));
+ for (const bool arrays : {false, true}) {
+ SCOPED_TRACE(arrays ? "array" : "scalar");
+ int64_t mirrored_bytes = 0;
+ int64_t peak_bytes = 0;
+ {
+ const std::string path = directory + (arrays ? "/array.idx" :
"/scalar.idx");
+ doris::io::FileWriterPtr file;
+ ASSERT_OK(fs->create_file(path, &file));
+ doris::segment_v2::IndexFileWriter index_writer(
+ fs, path, "null_handoff", 0,
doris::InvertedIndexStorageFormatPB::SNII,
+ std::move(file));
+ doris::TabletIndex meta;
+ doris::TabletIndexPB pb;
+ pb.set_index_type(doris::IndexType::INVERTED);
+ pb.set_index_id(119);
+ pb.add_col_unique_id(0);
+ meta.init_from_pb(pb);
+ doris::segment_v2::SniiIndexColumnWriter writer(
+ &index_writer, &meta,
doris::FieldType::OLAP_FIELD_TYPE_VARCHAR);
+ ASSERT_OK(writer.init());
+ auto* reporter = writer.memory_reporter_for_test();
+ mirrored_bytes = reporter->current_bytes();
+ reporter->consume_release_ = [&, original =
reporter->consume_release_](int64_t delta) {
+ original(delta);
+ mirrored_bytes += delta;
+ peak_bytes = std::max(peak_bytes, mirrored_bytes);
+ };
+ add_null_handoff_rows(&writer, arrays);
+ check_null_handoff_growth(writer, mirrored_bytes, peak_bytes);
+ check_null_handoff_finish(&writer, &index_writer, &mirrored_bytes,
&peak_bytes);
+ }
+ EXPECT_EQ(mirrored_bytes, 0);
+ }
+}
+
+TEST(SniiWriterNullDocids, FailedHandoffReleasesNullAllocationBeforeReporter) {
+ doris::TabletIndex meta;
+ doris::TabletIndexPB pb;
+ pb.set_index_type(doris::IndexType::INVERTED);
+ pb.set_index_id(120);
+ pb.add_col_unique_id(0);
+ meta.init_from_pb(pb);
+ doris::segment_v2::IndexFileWriter
index_writer(doris::io::global_local_filesystem(),
+ "missing_null_output",
"null_handoff", 0,
+
doris::InvertedIndexStorageFormatPB::SNII);
+ doris::segment_v2::SniiIndexColumnWriter writer(&index_writer, &meta,
+
doris::FieldType::OLAP_FIELD_TYPE_VARCHAR);
+ ASSERT_OK(writer.init());
+ auto* reporter = writer.memory_reporter_for_test();
+ ASSERT_OK(writer.add_nulls(1U << 20));
+ ASSERT_GT(reporter->current_bytes(), 0);
+ EXPECT_EQ(writer.finish().code(),
doris::ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND);
+ EXPECT_EQ(reporter->current_bytes(), 0);
+ EXPECT_TRUE(index_writer._snii_memory_reporters.empty());
+ EXPECT_TRUE(writer.null_docids_for_test().empty());
+ writer.close_on_error();
+ EXPECT_EQ(writer.memory_reporter_for_test(), nullptr);
+ EXPECT_EQ(writer.null_docids_for_test().capacity(), 0);
+}
+
TEST(SniiWriterFailureLatch, AnalyzerFailureDiscardsStateAndBlocksFinish) {
doris::TabletIndex index_meta;
init_failure_index_meta(&index_meta, 91);
diff --git
a/be/test/storage/index/inverted/snii/writer/spimi_streaming_writer_test.cpp
b/be/test/storage/index/inverted/snii/writer/spimi_streaming_writer_test.cpp
index 15ebca239bd..b588f6877c9 100644
--- a/be/test/storage/index/inverted/snii/writer/spimi_streaming_writer_test.cpp
+++ b/be/test/storage/index/inverted/snii/writer/spimi_streaming_writer_test.cpp
@@ -209,7 +209,7 @@ TEST(SniiSpimiStreamingWriter, StreamingConsumesSource) {
SniiIndexInput in = BaseInput(50);
in.term_source = &buf;
- LogicalIndexWriter writer(in);
+ LogicalIndexWriter writer(in,
TrackedNullDocids(std::move(in.null_docids)));
// build() streams the posting region straight into a FileWriter sink;
this test
// only asserts the source is drained, so a throwaway temp sink suffices.
const std::string post_path = TempPath();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
