mirror of
https://github.com/apache/doris.git
synced 2026-11-04 07:31:17 +00:00
[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.
This commit is contained in:
11 files changed
+234
-46
No files matched your search
@@ -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();
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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) {
|
||||
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 size_t target_capacity = std::max(need, _null_docids.capacity() * 2);
|
||||
// Uninitialized writers in unit tests have no reporter.
|
||||
if (_memory_reporter == nullptr) {
|
||||
return;
|
||||
}
|
||||
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;
|
||||
_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);
|
||||
}
|
||||
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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};
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in new issue
Block a user