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 72a587f5413 [fix](be) Use file-backed staging for SNII ANN indexes
(#66856)
72a587f5413 is described below
commit 72a587f541321fcb656ed8a3de5c9aad4fdec12d
Author: Jack <[email protected]>
AuthorDate: Mon Aug 24 10:16:50 2026 +0800
[fix](be) Use file-backed staging for SNII ANN indexes (#66856)
### What problem does this PR solve?
Issue Number: None
Related PR: #66052
Problem Summary:
SNII staged every serialized Faiss ANN sub-file in a growing in-memory
vector. A large HNSW or IVF index therefore retained a second complete
copy of the serialized index and could exhaust BE memory during index
construction.
A 64 MiB staging reproduction increased allocated heap by 67,388,976
bytes before this change. Fault-injection reproductions also showed that
final buffered append or fsync failures could escape
`AnnIndexColumnWriter::finish()` as `CLuceneError`; the IVF data path
additionally leaked its raw output owner.
This change reuses the self-cleaning `StagedBlobFile` abstraction for
ANN sub-files and keeps only the fixed CLucene output buffer resident.
It also gives both `ann.faiss` and `ann.ivfdata` explicit RAII
ownership, closes them in a catchable context, and converts staging
failures to `Status`.
The successful Faiss sub-file bytes, SNII container layout, persisted
index format, and query behavior are unchanged. Existing SNII indexes
remain readable after upgrade.
### Release note
Reduce peak BE memory while building SNII ANN indexes by staging
serialized Faiss files on temporary storage instead of retaining a
complete in-memory copy.
---
be/src/storage/index/ann/ann_index_writer.cpp | 13 +-
be/src/storage/index/ann/faiss_ann_index.cpp | 190 ++++++----
be/src/storage/index/index_file_writer.cpp | 54 ++-
be/src/storage/index/index_file_writer.h | 14 +
be/src/storage/index/snii/bkd/staged_blob_file.cpp | 28 +-
be/src/storage/index/snii/bkd/staged_blob_file.h | 10 +-
.../storage/index/snii/snii_bkd_index_writer.cpp | 22 +-
be/src/storage/index/snii/snii_bkd_index_writer.h | 10 +-
.../index/snii/snii_blob_staging_directory.cpp | 111 +++---
.../index/snii/snii_blob_staging_directory.h | 48 ++-
.../index/snii/writer/snii_compound_writer.cpp | 42 ++-
.../index/snii/writer/snii_compound_writer.h | 4 +
be/src/storage/segment/segment_writer.cpp | 39 ++
be/src/storage/segment/segment_writer.h | 6 +
be/src/storage/segment/vertical_segment_writer.cpp | 39 ++
be/src/storage/segment/vertical_segment_writer.h | 6 +
.../storage/index/snii/snii_ann_container_test.cpp | 418 ++++++++++++++++++++-
.../storage/index/snii/snii_bkd_adapter_test.cpp | 45 +++
be/test/storage/index/snii/staged_file_probe.h | 94 +++++
.../snii/writer/snii_compound_writer_test.cpp | 87 +++++
20 files changed, 1081 insertions(+), 199 deletions(-)
diff --git a/be/src/storage/index/ann/ann_index_writer.cpp
b/be/src/storage/index/ann/ann_index_writer.cpp
index d041eb7900d..391b4161557 100644
--- a/be/src/storage/index/ann/ann_index_writer.cpp
+++ b/be/src/storage/index/ann/ann_index_writer.cpp
@@ -171,6 +171,17 @@ Status AnnIndexColumnWriter::_build_and_save(Int64
min_train_rows, Int64 effecti
// full-segment build buffer is released before saving the index.
PODArray<float> empty_buffered_vectors;
_buffered_vectors.swap(empty_buffered_vectors);
- return _vector_index->save(_dir.get());
+ Status status = _vector_index->save(_dir.get());
+ if (!status.ok()) {
+ // A failed save keeps whatever it managed to write, and under SNII
that is
+ // an ANN-sized staging file plus its descriptor. Nothing downstream
+ // unwinds it: the segment flush returns before clear() and
+ // close_inverted_index(), and ADD INDEX holds every producer until the
+ // whole rowset has been closed. Drop BOTH owners of the staging area
here
+ // -- the index file writer's, and this writer's own.
+ _index_file_writer->discard_ann_staging_directory(_index_meta);
+ _dir.reset();
+ }
+ return status;
}
} // namespace doris::segment_v2
diff --git a/be/src/storage/index/ann/faiss_ann_index.cpp
b/be/src/storage/index/ann/faiss_ann_index.cpp
index bf14128dc78..f143b7ec6a5 100644
--- a/be/src/storage/index/ann/faiss_ann_index.cpp
+++ b/be/src/storage/index/ann/faiss_ann_index.cpp
@@ -219,17 +219,41 @@ FaissVectorIndex::~FaissVectorIndex() {
}
}
+namespace {
+
+Status close_index_output(std::unique_ptr<lucene::store::IndexOutput>& output,
+ const char* description) {
+ DCHECK(output != nullptr);
+ try {
+ output->close();
+ } catch (const CLuceneError& e) {
+ output.reset();
+ return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>("Failed
to close {}: {}",
+
description, e.what());
+ }
+ output.reset();
+ return Status::OK();
+}
+
+} // namespace
+
struct FaissIndexWriter : faiss::IOWriter {
public:
- FaissIndexWriter() = default;
- FaissIndexWriter(lucene::store::IndexOutput* output) : _output(output) {}
+ explicit FaissIndexWriter(lucene::store::IndexOutput* output) :
_output(output) {}
~FaissIndexWriter() override {
if (_output != nullptr) {
- _output->close();
- delete _output;
+ try {
+ _output->close();
+ } catch (const CLuceneError& e) {
+ // Serialization already failed, so this is cleanup only. A
+ // destructor must not turn an I/O error into std::terminate.
+ LOG(WARNING) << "Failed to close vector index output during
cleanup: " << e.what();
+ }
}
}
+ Status close() { return close_index_output(_output, "vector index
output"); }
+
size_t operator()(const void* ptr, size_t size, size_t nitems) override {
size_t bytes = size * nitems;
if (bytes > 0) {
@@ -252,7 +276,7 @@ public:
return nitems;
};
- lucene::store::IndexOutput* _output = nullptr;
+ std::unique_ptr<lucene::store::IndexOutput> _output;
};
struct FaissIndexReader : faiss::IOReader {
@@ -953,84 +977,88 @@ doris::Status FaissVectorIndex::range_search(const float*
query_vec, const float
doris::Status FaissVectorIndex::save(lucene::store::Directory* dir) {
auto start_time = std::chrono::high_resolution_clock::now();
- if (_index_type == AnnIndexType::IVF_ON_DISK) {
- // IVF_ON_DISK: write ivf data to a separate file, then write index
metadata.
- //
- // Why do we replace invlists here in save() instead of at build()
time?
- // During build/train/add, IndexIVF needs a writable
ArrayInvertedLists to
- // receive vectors via add_entries(). PreadInvertedLists inherits from
- // ReadOnlyInvertedLists and does not support writes. The original
- // OnDiskInvertedLists does support writes but requires mmap on a real
file,
- // which is unavailable at build time (Directory is only passed to
save()).
- // So the standard faiss pattern is: build in-memory with
ArrayInvertedLists,
- // then convert to on-disk format at serialization time. The
replace_invlists
- // call in Step 2 is purely a serialization format switch (to emit
"ilod"
- // fourcc instead of "ilar"), not a runtime data structure change.
- //
- // Step 1: The in-memory index has ArrayInvertedLists. Write them to
ann.ivfdata
- // by converting to OnDiskInvertedLists format:
- // For each list: [codes: capacity*code_size][ids:
capacity*sizeof(idx_t)]
- auto* ivf = dynamic_cast<faiss::IndexIVF*>(_index.get());
- DCHECK(ivf != nullptr);
- auto* ails = dynamic_cast<faiss::ArrayInvertedLists*>(ivf->invlists);
- DCHECK(ails != nullptr);
-
- const size_t nlist = ails->nlist;
- const size_t code_size = ails->code_size;
-
- // Build OnDiskOneList metadata and write data to ann.ivfdata
- std::vector<faiss::OnDiskOneList> lists(nlist);
- lucene::store::IndexOutput* ivfdata_output =
dir->createOutput(faiss_ivfdata_file_name);
- size_t offset = 0;
- for (size_t i = 0; i < nlist; i++) {
- size_t list_size = ails->list_size(i);
- lists[i].size = list_size;
- lists[i].capacity = list_size;
- lists[i].offset = offset;
-
- if (list_size > 0) {
- // Write codes
- const uint8_t* codes = ails->get_codes(i);
- size_t codes_bytes = list_size * code_size;
- ivfdata_output->writeBytes(codes,
cast_set<Int32>(codes_bytes));
-
- // Write ids
- const faiss::idx_t* ids = ails->get_ids(i);
- size_t ids_bytes = list_size * sizeof(faiss::idx_t);
- ivfdata_output->writeBytes(reinterpret_cast<const
uint8_t*>(ids),
- cast_set<Int32>(ids_bytes));
- }
+ try {
+ if (_index_type == AnnIndexType::IVF_ON_DISK) {
+ // IVF_ON_DISK: write ivf data to a separate file, then write
index metadata.
+ //
+ // Why do we replace invlists here in save() instead of at build()
time?
+ // During build/train/add, IndexIVF needs a writable
ArrayInvertedLists to
+ // receive vectors via add_entries(). PreadInvertedLists inherits
from
+ // ReadOnlyInvertedLists and does not support writes. The original
+ // OnDiskInvertedLists does support writes but requires mmap on a
real file,
+ // which is unavailable at build time (Directory is only passed to
save()).
+ // So the standard faiss pattern is: build in-memory with
ArrayInvertedLists,
+ // then convert to on-disk format at serialization time. The
replace_invlists
+ // call in Step 2 is purely a serialization format switch (to emit
"ilod"
+ // fourcc instead of "ilar"), not a runtime data structure change.
+ //
+ // Step 1: The in-memory index has ArrayInvertedLists. Write them
to ann.ivfdata
+ // by converting to OnDiskInvertedLists format:
+ // For each list: [codes: capacity*code_size][ids:
capacity*sizeof(idx_t)]
+ auto* ivf = dynamic_cast<faiss::IndexIVF*>(_index.get());
+ DCHECK(ivf != nullptr);
+ auto* ails =
dynamic_cast<faiss::ArrayInvertedLists*>(ivf->invlists);
+ DCHECK(ails != nullptr);
+
+ const size_t nlist = ails->nlist;
+ const size_t code_size = ails->code_size;
+
+ // Build OnDiskOneList metadata and write data to ann.ivfdata
+ std::vector<faiss::OnDiskOneList> lists(nlist);
+ auto ivfdata_output = std::unique_ptr<lucene::store::IndexOutput>(
+ dir->createOutput(faiss_ivfdata_file_name));
+ size_t offset = 0;
+ for (size_t i = 0; i < nlist; i++) {
+ size_t list_size = ails->list_size(i);
+ lists[i].size = list_size;
+ lists[i].capacity = list_size;
+ lists[i].offset = offset;
+
+ if (list_size > 0) {
+ // Write codes
+ const uint8_t* codes = ails->get_codes(i);
+ size_t codes_bytes = list_size * code_size;
+ ivfdata_output->writeBytes(codes,
cast_set<Int32>(codes_bytes));
+
+ // Write ids
+ const faiss::idx_t* ids = ails->get_ids(i);
+ size_t ids_bytes = list_size * sizeof(faiss::idx_t);
+ ivfdata_output->writeBytes(reinterpret_cast<const
uint8_t*>(ids),
+ cast_set<Int32>(ids_bytes));
+ }
- offset += list_size * (code_size + sizeof(faiss::idx_t));
+ offset += list_size * (code_size + sizeof(faiss::idx_t));
+ }
+ size_t totsize = offset;
+ RETURN_IF_ERROR(close_index_output(ivfdata_output, "IVF data
output"));
+
+ // Step 2: Replace ArrayInvertedLists with OnDiskInvertedLists so
that
+ // write_index serializes in "ilod" format (metadata only).
+ auto* od = new faiss::OnDiskInvertedLists();
+ od->nlist = nlist;
+ od->code_size = code_size;
+ od->lists = std::move(lists);
+ od->totsize = totsize;
+ od->ptr = nullptr;
+ od->read_only = true;
+ // filename is not used during load (we use separate ivfdata file),
+ // but write it for format completeness.
+ od->filename = faiss_ivfdata_file_name;
+ ivf->replace_invlists(od, true);
+
+ // Step 3: Write index metadata to ann.faiss (includes "ilod"
fourcc)
+ FaissIndexWriter writer(dir->createOutput(faiss_index_fila_name));
+ RETURN_IF_CATCH_EXCEPTION(faiss::write_index(_index.get(),
&writer));
+ RETURN_IF_ERROR(writer.close());
+ } else {
+ // HNSW / IVF: write the full index to ann.faiss
+ FaissIndexWriter writer(dir->createOutput(faiss_index_fila_name));
+ RETURN_IF_CATCH_EXCEPTION(faiss::write_index(_index.get(),
&writer));
+ RETURN_IF_ERROR(writer.close());
}
- size_t totsize = offset;
- ivfdata_output->close();
- delete ivfdata_output;
-
- // Step 2: Replace ArrayInvertedLists with OnDiskInvertedLists so that
- // write_index serializes in "ilod" format (metadata only).
- auto* od = new faiss::OnDiskInvertedLists();
- od->nlist = nlist;
- od->code_size = code_size;
- od->lists = std::move(lists);
- od->totsize = totsize;
- od->ptr = nullptr;
- od->read_only = true;
- // filename is not used during load (we use separate ivfdata file),
- // but write it for format completeness.
- od->filename = faiss_ivfdata_file_name;
- ivf->replace_invlists(od, true);
-
- // Step 3: Write index metadata to ann.faiss (includes "ilod" fourcc)
- lucene::store::IndexOutput* idx_output =
dir->createOutput(faiss_index_fila_name);
- auto writer = std::make_unique<FaissIndexWriter>(idx_output);
- faiss::write_index(_index.get(), writer.get());
-
- } else {
- // HNSW / IVF: write the full index to ann.faiss
- lucene::store::IndexOutput* idx_output =
dir->createOutput(faiss_index_fila_name);
- auto writer = std::make_unique<FaissIndexWriter>(idx_output);
- faiss::write_index(_index.get(), writer.get());
+ } catch (const CLuceneError& e) {
+ return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
+ "Failed to save vector index: {}", e.what());
}
auto end_time = std::chrono::high_resolution_clock::now();
diff --git a/be/src/storage/index/index_file_writer.cpp
b/be/src/storage/index/index_file_writer.cpp
index e00dc047efc..540f01e3e6a 100644
--- a/be/src/storage/index/index_file_writer.cpp
+++ b/be/src/storage/index/index_file_writer.cpp
@@ -134,8 +134,8 @@ Status IndexFileWriter::_insert_directory_into_map(int64_t
index_id,
Result<std::shared_ptr<DorisFSDirectory>> IndexFileWriter::open(const
TabletIndex* index_meta) {
// No index under SNII writes through a CLucene filesystem directory: text
- // postings go through the SPIMI writer, and an ANN index stages into
memory
- // (see open_ann_directory) so that nothing has to be cleaned off disk.
+ // postings go through the SPIMI writer, and an ANN index uses
self-cleaning
+ // per-file staging (see open_ann_directory).
if (_storage_format == InvertedIndexStorageFormatPB::SNII) {
return
ResultError(Status::Error<ErrorCode::INVERTED_INDEX_NOT_SUPPORTED>(
"SNII format does not open CLucene filesystem directories"));
@@ -186,6 +186,37 @@ Result<std::shared_ptr<lucene::store::Directory>>
IndexFileWriter::_open_snii_an
return dir;
}
+void IndexFileWriter::discard_ann_staging_directory(const TabletIndex*
index_meta) {
+ // Only SNII stages an ANN index somewhere disposable. Branching here
rather
+ // than in the caller keeps the format knowledge on the side that owns it,
+ // exactly as open_ann_directory() does.
+ if (_storage_format != InvertedIndexStorageFormatPB::SNII) {
+ return;
+ }
+ DCHECK(index_meta != nullptr);
+ const auto key = std::make_pair(index_meta->index_id(),
index_meta->get_index_suffix());
+ _indices_dirs.erase(key);
+ _snii_blob_dir_metas.erase(key);
+}
+
+void IndexFileWriter::abandon_snii_staging() {
+ if (_storage_format != InvertedIndexStorageFormatPB::SNII) {
+ return;
+ }
+ // Empty each directory before dropping the map. Releasing only this
writer's
+ // reference would free nothing while a producer is still alive holding the
+ // same directory through its own _dir -- which is exactly the state a
segment
+ // that failed before clear() is in. Emptying makes the abort independent
of
+ // who else is still holding on.
+ for (const auto& [key, dir] : _indices_dirs) {
+ if (std::strcmp(dir->getObjectName(),
+ snii_doris::SniiBlobStagingDirectory::getClassName())
== 0) {
+
static_cast<snii_doris::SniiBlobStagingDirectory*>(dir.get())->discard_staged_files();
+ }
+ }
+ _release_snii_blob_directories();
+}
+
Status IndexFileWriter::_seal_snii_blob_directories() {
DORIS_CHECK(_storage_format == InvertedIndexStorageFormatPB::SNII);
for (const auto& [key, dir] : _indices_dirs) {
@@ -202,25 +233,26 @@ Status IndexFileWriter::_seal_snii_blob_directories() {
DORIS_CHECK(std::strcmp(dir->getObjectName(),
snii_doris::SniiBlobStagingDirectory::getClassName()) == 0);
- // Nothing here can throw: the staged bytes are plain buffers, and each
- // source keeps its own alive, so finish() may pull them after this
- // directory is gone.
+ // The sources TAKE the staged files, so they own them alone from here
on:
+ // finish() may pull the bytes after this directory is gone, and it can
+ // unlink each sub-file as soon as it has copied it instead of waiting
for
+ // whoever else happens to still hold the directory.
auto* staging =
static_cast<snii_doris::SniiBlobStagingDirectory*>(dir.get());
// All cold: a faiss index is read at QUERY time, never at container
open,
// so nothing here belongs in the hot area the text metadata groups
share.
RETURN_IF_ERROR(add_snii_blob_index(meta_it->second.get(),
doris::snii::format::LogicalIndexKind::kAnn,
- staging->blob_sources(), {}));
+ staging->take_blob_sources(), {}));
}
return Status::OK();
}
void IndexFileWriter::_release_snii_blob_directories() {
- // Dropping the map is the whole release: a staging directory holds its
bytes
- // in memory and owns no file, so there is nothing on disk to remove and --
- // unlike DorisFSDirectory::deleteDirectory() -- no throwing call to make
from
- // a Status-returning close path. Any buffer a registered blob source still
- // needs stays alive through that source until finish() has pulled it.
+ // Dropping the map is all this has to do: sealing already handed the
staged
+ // files to the blob sources, which unlink them as finish() copies them,
and
+ // an unsealed directory unlinks its own on the way out. So -- unlike
+ // DorisFSDirectory::deleteDirectory() -- there is no throwing cleanup
call to
+ // make from this Status-returning close path.
_indices_dirs.clear();
_snii_blob_dir_metas.clear();
}
diff --git a/be/src/storage/index/index_file_writer.h
b/be/src/storage/index/index_file_writer.h
index f77454f90e6..f1f312a7149 100644
--- a/be/src/storage/index/index_file_writer.h
+++ b/be/src/storage/index/index_file_writer.h
@@ -91,6 +91,20 @@ public:
// need the DorisFSDirectory subclass.
Result<std::shared_ptr<lucene::store::Directory>> open_ann_directory(
const TabletIndex* index_meta);
+ // SNII only: drops the staging directory of one ANN index whose
serialization
+ // failed. Its sub-files unlink themselves once their last owner is gone,
and
+ // the producer releases its own reference alongside this call -- without
that
+ // a failed save keeps an ANN-sized file and its descriptor on the temp
+ // filesystem until this writer is destroyed, which for a rowset build is
not
+ // until every other segment has been written. V1/V2 keep their directory:
its
+ // files ARE the index output, and begin_close() is what removes them.
+ void discard_ann_staging_directory(const TabletIndex* index_meta);
+ // SNII only: drops the staging of EVERY index on this writer because the
+ // segment they belong to is being abandoned. Layered above the per-index
+ // discard, not a duplicate of it: that one fires the instant one ANN
+ // serialization fails, while this one covers a segment that failed AFTER
its
+ // indexes staged successfully, when nothing else will ever seal them.
+ void abandon_snii_staging();
// Write-path facts for one SNII index flush.
struct SniiAddIndexOptions {
// This flush serves a stream/broker load (DataWriteType::TYPE_DIRECT):
diff --git a/be/src/storage/index/snii/bkd/staged_blob_file.cpp
b/be/src/storage/index/snii/bkd/staged_blob_file.cpp
index 7b63e024b1f..b327612cce5 100644
--- a/be/src/storage/index/snii/bkd/staged_blob_file.cpp
+++ b/be/src/storage/index/snii/bkd/staged_blob_file.cpp
@@ -26,6 +26,7 @@
#include "common/check.h"
#include "storage/index/snii/writer/temp_dir.h"
+#include "util/debug_points.h"
namespace doris::snii::bkd {
@@ -60,6 +61,8 @@ StagedBlobFile::~StagedBlobFile() {
Status StagedBlobFile::append(Slice data) {
DORIS_CHECK_GE(fd_, 0);
DORIS_CHECK(!finalized_);
+ DBUG_EXECUTE_IF("StagedBlobFile::append_error",
+ { return Status::IOError("injected blob staging append
failure"); })
const uint8_t* cursor = data.data();
size_t remaining = data.size();
while (remaining > 0) {
@@ -80,13 +83,24 @@ Status StagedBlobFile::append(Slice data) {
Status StagedBlobFile::finalize() {
DORIS_CHECK_GE(fd_, 0);
DORIS_CHECK(!finalized_);
- // The descriptor stays open on purpose: read_at reads through this same
one,
- // so the file survives an unlink and cannot be swapped out from under us.
- // A deferred write error would surface at close(), which happens in
remove()
- // after the container has already sealed -- so force it out here instead.
- if (::fsync(fd_) != 0) {
- return Status::IOError("failed to flush a blob staging file: {}",
std::strerror(errno));
- }
+ // Kept as a seam for the seal-failure plumbing (an error here has to
become a
+ // Status and unwind the staging). Nothing in the seal itself can fail any
+ // more; the reachable staging failure is append().
+ DBUG_EXECUTE_IF("StagedBlobFile::finalize_error",
+ { return Status::IOError("injected blob staging finalize
failure"); })
+ // NO fsync. This file is scratch: SniiCompoundWriter copies every byte of
it
+ // into the real container, IndexFileWriter::begin_close() makes THAT
durable,
+ // and remove() then unlinks this inode -- so its contents can never
recover
+ // anything. Forcing a GiB-scale HNSW/IVF payload to storage here would
buy a
+ // full durable write plus a latency barrier before the second, real write,
+ // once per sub-file, and could fail a build on scratch writeback alone.
+ //
+ // Error reporting does not depend on it either: the descriptor stays open
on
+ // purpose (read_at reads through this same one, so the file survives an
+ // unlink and cannot be swapped out from under us), and a writeback failure
+ // surfaces there as EIO -- which read_at reports, as it does a short
read. A
+ // damaged scratch file therefore still fails the build instead of being
+ // checksummed into the container.
finalized_ = true;
return Status::OK();
}
diff --git a/be/src/storage/index/snii/bkd/staged_blob_file.h
b/be/src/storage/index/snii/bkd/staged_blob_file.h
index 8d7d67fc29a..f9bd73dceeb 100644
--- a/be/src/storage/index/snii/bkd/staged_blob_file.h
+++ b/be/src/storage/index/snii/bkd/staged_blob_file.h
@@ -25,7 +25,9 @@
#include "storage/index/snii/common/slice.h"
#include "storage/index/snii/io/file_writer.h"
-// Build-time staging for one blob sub-file (design 10).
+// Build-time staging for one blob sub-file (design 10). Native BKD and ANN
both
+// use this file because their compound-container producers finish long before
+// the container pulls the payload.
//
// The container is a PULL consumer: SniiCompoundWriter::add_blob_index
registers
// a BlobFileSource and only asks for the bytes at finish(), because placement
@@ -51,8 +53,10 @@ public:
StagedBlobFile(const StagedBlobFile&) = delete;
StagedBlobFile& operator=(const StagedBlobFile&) = delete;
- // io::FileWriter. append() is the producer side; finalize() flushes and
- // switches the file to readable.
+ // io::FileWriter. append() is the producer side; finalize() switches the
file
+ // to readable. It makes NO durability promise -- this is scratch that the
+ // container copies out and then unlinks, so a barrier here would be a
second
+ // full write for nothing (see the .cpp).
Status append(Slice data) override;
Status finalize() override;
uint64_t bytes_written() const override { return bytes_written_; }
diff --git a/be/src/storage/index/snii/snii_bkd_index_writer.cpp
b/be/src/storage/index/snii/snii_bkd_index_writer.cpp
index 11fd8a7e9b8..eaa2e518e84 100644
--- a/be/src/storage/index/snii/snii_bkd_index_writer.cpp
+++ b/be/src/storage/index/snii/snii_bkd_index_writer.cpp
@@ -196,14 +196,17 @@ Status SniiBkdIndexColumnWriter::finish() {
RETURN_IF_ERROR(null_writer.finish(_rid, &null_sink));
// The container pulls at IndexFileWriter::finish_close(), long after this
- // returns, so the staging file has to outlive this object. Ownership moves
- // into the read callback itself.
- _data = std::move(data);
- std::shared_ptr<::doris::snii::bkd::StagedBlobFile> staged = _data;
+ // returns, so the staging file has to outlive this object. Ownership MOVES
+ // into the read callback, which is then its only owner: the per-blob
release
+ // in SniiCompoundWriter::finish() unlinks bkd_data as soon as the bytes
are
+ // in the container, instead of whenever this writer happens to be
destroyed.
+ // Keeping a second reference here would defeat that -- a producer
routinely
+ // outlives the seal (see the header).
+ std::shared_ptr<::doris::snii::bkd::StagedBlobFile> staged =
std::move(data);
::doris::snii::writer::BlobFileSource cold;
cold.name = kDataFileName;
cold.length = staged->bytes_written();
- cold.read_fn = [staged](uint64_t offset, size_t len, uint8_t* out) {
+ cold.read_fn = [staged = std::move(staged)](uint64_t offset, size_t len,
uint8_t* out) {
return staged->read_at(offset, len, out);
};
@@ -219,11 +222,12 @@ Status SniiBkdIndexColumnWriter::finish() {
}
void SniiBkdIndexColumnWriter::close_on_error() {
- // The builder unlinks its own spilled runs; the staging file removes
itself
- // on destruction. Dropping both here means an aborted segment leaves no
temp
- // file behind even if this writer is kept alive for a while.
+ // The builder unlinks its own spilled runs. Dropping it here means an
aborted
+ // segment leaves no temp file behind even if this writer is kept alive
for a
+ // while. There is nothing else to drop: bkd_data only exists between
+ // finish()'s create() and its handover to the read callback, and it
unlinks
+ // itself if finish() fails in between.
_builder.reset();
- _data.reset();
}
} // namespace doris::segment_v2
diff --git a/be/src/storage/index/snii/snii_bkd_index_writer.h
b/be/src/storage/index/snii/snii_bkd_index_writer.h
index 62571b86fe8..537138af795 100644
--- a/be/src/storage/index/snii/snii_bkd_index_writer.h
+++ b/be/src/storage/index/snii/snii_bkd_index_writer.h
@@ -90,10 +90,12 @@ private:
std::vector<uint32_t> _null_docids;
std::unique_ptr<::doris::snii::bkd::BkdBuilder> _builder;
- // Staged bkd_data, kept alive until the container has pulled its bytes at
- // IndexFileWriter::finish_close(). Destroying it earlier would unlink the
- // temp file out from under the pull.
- std::shared_ptr<::doris::snii::bkd::StagedBlobFile> _data;
+ // No staged-file member on purpose. bkd_data has to outlive this writer --
+ // the container pulls it at IndexFileWriter::finish_close() -- but it must
+ // not outlive the pull, and this writer routinely does: IndexBuilder's
SNII
+ // ADD INDEX path holds every producer in _index_column_writers until the
+ // whole rowset has been closed. finish() therefore hands the file to the
+ // read callback and keeps no reference of its own.
};
} // namespace segment_v2
diff --git a/be/src/storage/index/snii/snii_blob_staging_directory.cpp
b/be/src/storage/index/snii/snii_blob_staging_directory.cpp
index c8b39bdb23d..92e7a7eed8a 100644
--- a/be/src/storage/index/snii/snii_blob_staging_directory.cpp
+++ b/be/src/storage/index/snii/snii_blob_staging_directory.cpp
@@ -19,25 +19,22 @@
#include <fmt/format.h>
-#include <cstring>
#include <utility>
#include "common/check.h"
#include "common/status.h"
+#include "storage/index/snii/bkd/staged_blob_file.h"
namespace doris::segment_v2::snii_doris {
-// Appends into one staged buffer. BufferedIndexOutput already batches the
-// writer's byte-at-a-time calls, so flushBuffer sees 64 KiB chunks and the
-// buffer grows geometrically through vector::insert.
-//
-// The buffer is held by shared_ptr because a blob source handed to the
container
-// keeps it alive on its own: an output may be closed and destroyed, and the
-// directory itself dropped, long before finish() pulls the bytes.
+// Appends into one staged file. BufferedIndexOutput batches the writer's
+// byte-at-a-time calls, so flushBuffer issues 64 KiB sequential writes while
the
+// complete ANN payload remains outside the process heap.
class SniiBlobStagingDirectory::StagingIndexOutput final
: public lucene::store::BufferedIndexOutput {
public:
- explicit StagingIndexOutput(std::shared_ptr<Buffer> buffer) :
_buffer(std::move(buffer)) {}
+ explicit StagingIndexOutput(std::shared_ptr<snii::bkd::StagedBlobFile>
file)
+ : _file(std::move(file)) {}
~StagingIndexOutput() override {
// MUST close here, qualified. ~BufferedIndexOutput also calls close()
if
@@ -49,15 +46,25 @@ public:
try {
StagingIndexOutput::close();
} catch (const CLuceneError&) {
- // A destructor may not throw. Nothing here can fail anyway --
- // flushBuffer only appends to a vector -- but the base close() is
- // declared throwing, so the guard has to exist.
+ // A destructor may not throw. The normal success path closes
+ // explicitly so it can report this staging I/O failure.
}
}
- void close() override { BufferedIndexOutput::close(); }
+ void close() override {
+ if (_closed) {
+ return;
+ }
+ BufferedIndexOutput::close();
+ Status status = _file->finalize();
+ if (!status.ok()) {
+ const std::string message = status.to_string();
+ _CLTHROWA(CL_ERR_IO, message.c_str());
+ }
+ _closed = true;
+ }
- int64_t length() const override { return
static_cast<int64_t>(_buffer->size()); }
+ int64_t length() const override { return
static_cast<int64_t>(_file->bytes_written()); }
protected:
void flushBuffer(const uint8_t* b, const int32_t size) override {
@@ -66,11 +73,16 @@ protected:
if (b == nullptr || size <= 0) {
return;
}
- _buffer->insert(_buffer->end(), b, b + size);
+ Status status = _file->append(snii::Slice(b,
static_cast<size_t>(size)));
+ if (!status.ok()) {
+ const std::string message = status.to_string();
+ _CLTHROWA(CL_ERR_IO, message.c_str());
+ }
}
private:
- const std::shared_ptr<Buffer> _buffer;
+ const std::shared_ptr<snii::bkd::StagedBlobFile> _file;
+ bool _closed = false;
};
SniiBlobStagingDirectory::~SniiBlobStagingDirectory() = default;
@@ -83,7 +95,7 @@ const char* SniiBlobStagingDirectory::getObjectName() const {
return getClassName();
}
-const std::shared_ptr<SniiBlobStagingDirectory::Buffer>*
SniiBlobStagingDirectory::find_file(
+const std::shared_ptr<snii::bkd::StagedBlobFile>*
SniiBlobStagingDirectory::find_file(
const char* name) const {
// A null name is a caller bug, not an absent file; reporting "absent"
would
// hide it, and every caller formats `name` into its error message.
@@ -117,13 +129,13 @@ int64_t SniiBlobStagingDirectory::fileModified(const
char* name) const {
}
int64_t SniiBlobStagingDirectory::fileLength(const char* name) const {
- const std::shared_ptr<Buffer>* buffer = find_file(name);
- if (buffer == nullptr) {
+ const std::shared_ptr<snii::bkd::StagedBlobFile>* file = find_file(name);
+ if (file == nullptr) {
const std::string message =
fmt::format("File does not exist in the SNII staging
directory: {}", name);
_CLTHROWA(CL_ERR_IO, message.c_str()); // CLuceneError STRDUPs the
message
}
- return static_cast<int64_t>((*buffer)->size());
+ return static_cast<int64_t>((*file)->bytes_written());
}
bool SniiBlobStagingDirectory::openInput(const char* name,
lucene::store::IndexInput*& ret,
@@ -151,12 +163,18 @@ void SniiBlobStagingDirectory::touchFile(const char*
/*name*/) {
lucene::store::IndexOutput* SniiBlobStagingDirectory::createOutput(const char*
name) {
DORIS_CHECK(name != nullptr);
// Same semantics as a filesystem directory: creating an existing name
- // truncates it. The buffer is replaced rather than cleared, so a blob
source
- // already taken over the old content keeps reading the old content
instead of
- // seeing it mutate underneath.
- auto buffer = std::make_shared<Buffer>();
- _files[name] = buffer;
- return _CLNEW StagingIndexOutput(std::move(buffer));
+ // truncates it. The file is replaced rather than reused, so a blob source
+ // already taken over the old content keeps reading the old content instead
+ // of seeing it mutate underneath.
+ std::unique_ptr<snii::bkd::StagedBlobFile> created;
+ Status status = snii::bkd::StagedBlobFile::create(name, &created);
+ if (!status.ok()) {
+ const std::string message = status.to_string();
+ _CLTHROWA(CL_ERR_IO, message.c_str());
+ }
+ auto file = std::shared_ptr<snii::bkd::StagedBlobFile>(std::move(created));
+ _files[name] = file;
+ return _CLNEW StagingIndexOutput(std::move(file));
}
bool SniiBlobStagingDirectory::doDeleteFile(const char* name) {
@@ -169,8 +187,8 @@ bool SniiBlobStagingDirectory::doDeleteFile(const char*
name) {
void SniiBlobStagingDirectory::close() {
// Deliberately keeps the staged files: close() is what a CLucene writer
calls
- // when it is done producing, and the harvest happens afterwards. The
buffers
- // die with the directory, or with the last blob source over them.
+ // when it is done producing, and the harvest happens afterwards. Each temp
+ // file is unlinked by its final StagedBlobFile owner.
}
std::string SniiBlobStagingDirectory::toString() const {
@@ -178,33 +196,40 @@ std::string SniiBlobStagingDirectory::toString() const {
staged_bytes());
}
-std::vector<snii::writer::BlobFileSource>
SniiBlobStagingDirectory::blob_sources() const {
+std::vector<snii::writer::BlobFileSource>
SniiBlobStagingDirectory::take_blob_sources() {
+ // Moved out before anything is built from it: a source has to be the
file's
+ // only owner, or the per-blob release in SniiCompoundWriter::finish()
frees
+ // nothing and the staging survives until this directory does.
+ std::map<std::string, std::shared_ptr<snii::bkd::StagedBlobFile>> taken;
+ taken.swap(_files);
std::vector<snii::writer::BlobFileSource> sources;
- sources.reserve(_files.size());
+ sources.reserve(taken.size());
// std::map iterates in name order, which is the order the filesystem
harvest
// produced by sorting list(). Two builds of one index therefore lay their
// sub-files out identically in the container.
- for (const auto& [name, buffer] : _files) {
+ for (auto& [name, staged] : taken) {
+ auto file = std::move(staged);
+ const uint64_t length = file->bytes_written();
sources.push_back(snii::writer::BlobFileSource {
.name = name,
- .length = buffer->size(),
- .read_fn = [buffer](uint64_t offset, size_t len, uint8_t* out)
-> Status {
- if (offset > buffer->size() || len > buffer->size() -
offset) {
- return Status::Error<ErrorCode::INTERNAL_ERROR>(
- "SNII staging read [{}, +{}) is outside the
staged {} bytes",
- offset, len, buffer->size());
- }
- std::memcpy(out, buffer->data() + offset, len);
- return Status::OK();
- }});
+ .length = length,
+ .read_fn = [file = std::move(file)](uint64_t offset, size_t
len, uint8_t* out)
+ -> Status { return file->read_at(offset, len, out);
}});
}
return sources;
}
+void SniiBlobStagingDirectory::discard_staged_files() {
+ // Same swap-with-empty as take_blob_sources(): clear() would leave the
map's
+ // nodes, and it is the shared_ptr elements themselves that unlink the
files.
+ std::map<std::string, std::shared_ptr<snii::bkd::StagedBlobFile>> dropped;
+ dropped.swap(_files);
+}
+
uint64_t SniiBlobStagingDirectory::staged_bytes() const {
uint64_t total = 0;
- for (const auto& [name, buffer] : _files) {
- total += buffer->size();
+ for (const auto& [name, file] : _files) {
+ total += file->bytes_written();
}
return total;
}
diff --git a/be/src/storage/index/snii/snii_blob_staging_directory.h
b/be/src/storage/index/snii/snii_blob_staging_directory.h
index a1f4256a3e9..7618272ad82 100644
--- a/be/src/storage/index/snii/snii_blob_staging_directory.h
+++ b/be/src/storage/index/snii/snii_blob_staging_directory.h
@@ -31,22 +31,23 @@
class CLuceneError;
+namespace doris::snii::bkd {
+class StagedBlobFile;
+}
+
namespace doris::segment_v2::snii_doris {
-// Write-only, memory-backed lucene::store::Directory: the staging area an ANN
+// Write-only, file-backed lucene::store::Directory: the staging area an ANN
// index is built into before it is sealed as a blob logical index of a SNII
// container.
//
// WHY IT EXISTS. A SNII container cannot take the faiss bytes as they are
// produced -- blob payloads are streamed by SniiCompoundWriter::finish(),
after
// the text physical sections -- so the bytes must be parked somewhere in
-// between. Borrowing a CLucene filesystem directory for that, as the V1/V2
-// formats do, buys two problems SNII has no use for: the temp directory is
-// removed by exactly one call, so every early return on the close path leaks
it
-// until a BE restart wipes the tmp dir; and that call, deleteDirectory(),
-// throws CLuceneError, which must not cross a Status-returning close. Parking
-// the bytes here removes both -- nothing lands on disk, and there is nothing
to
-// delete.
+// between. A complete in-memory copy is unbounded for HNSW and IVF indexes.
+// Each sub-file therefore uses the same self-cleaning staging file as native
+// BKD. This retains only BufferedIndexOutput's fixed buffer while still
avoiding
+// the throwing, whole-directory cleanup used by the V1/V2 formats.
//
// SCOPE. Only the write side is real. The faiss writer uses createOutput() and
// toString() and nothing else, and a sealed ANN blob is read back through
@@ -74,16 +75,27 @@ public:
static const char* getClassName();
const char* getObjectName() const override;
- // Blob sources over the staged buffers, in name order -- the same order
the
+ // Blob sources over the staged files, in name order -- the same order the
// filesystem harvest produced by sorting list(), so the container lays two
// builds of one index out identically.
//
- // Each source keeps its buffer alive on its own, so the sources stay valid
- // after this directory is destroyed: SniiCompoundWriter::finish() pulls
them
- // long after the ANN writer is gone.
- std::vector<snii::writer::BlobFileSource> blob_sources() const;
-
- // Bytes currently held in memory across every staged sub-file.
+ // TAKES the files. This directory holds nothing afterwards and the
returned
+ // sources are their ONLY owners, which is what lets the container unlink
each
+ // sub-file the moment it has copied its bytes (see the per-blob release in
+ // SniiCompoundWriter::finish()). Leaving a second owner here would defeat
+ // that: an ANN producer routinely outlives the seal -- IndexBuilder's SNII
+ // ADD INDEX path keeps every AnnIndexColumnWriter alive until the whole
+ // rowset has been closed -- so one rowset's staging would pile up at once.
+ std::vector<snii::writer::BlobFileSource> take_blob_sources();
+
+ // Drops every staged file NOW, whoever else still holds this directory.
The
+ // abort path needs that: an ANN producer keeps the directory alive through
+ // its own _dir, so releasing only the index file writer's reference would
+ // free nothing.
+ void discard_staged_files();
+
+ // Logical bytes across every staged sub-file. Zero once
take_blob_sources()
+ // has handed them over.
uint64_t staged_bytes() const;
protected:
@@ -92,11 +104,9 @@ protected:
private:
class StagingIndexOutput;
- using Buffer = std::vector<uint8_t>;
-
- const std::shared_ptr<Buffer>* find_file(const char* name) const;
+ const std::shared_ptr<snii::bkd::StagedBlobFile>* find_file(const char*
name) const;
- std::map<std::string, std::shared_ptr<Buffer>> _files;
+ std::map<std::string, std::shared_ptr<snii::bkd::StagedBlobFile>> _files;
};
} // namespace doris::segment_v2::snii_doris
diff --git a/be/src/storage/index/snii/writer/snii_compound_writer.cpp
b/be/src/storage/index/snii/writer/snii_compound_writer.cpp
index e085071fcfc..c250302f0bf 100644
--- a/be/src/storage/index/snii/writer/snii_compound_writer.cpp
+++ b/be/src/storage/index/snii/writer/snii_compound_writer.cpp
@@ -31,6 +31,7 @@
#include "storage/index/snii/format/metadata_directory.h"
#include "storage/index/snii/format/tail_pointer.h"
#include "storage/index/snii/reader/snii_segment_reader.h"
+#include "util/defer_op.h"
namespace doris::snii::writer {
@@ -45,6 +46,19 @@ Status SniiCompoundWriter::poison(Status status) {
DCHECK(!status.ok());
if (failed_.ok()) {
failed_ = std::move(status);
+ // First transition only. A poisoned compound can never seal, so every
+ // registered blob source is already dead -- and a source is routinely
the
+ // SOLE owner of a staging file holding a whole faiss index or a BKD
leaf
+ // region, because the producer hands it over at registration. Waiting
for
+ // finish() to release them is not enough: after a poison, production
+ // usually never calls it. SegmentWriter::_write_inverted_index()
returns
+ // before close_inverted_index(), and SegmentCreator::flush() keeps the
+ // failed writer, so the files and their descriptors would stay pinned
+ // until this writer is destroyed.
+ //
+ // Safe from inside finish()'s own loops: this only swaps the contents
of
+ // each entry's source vectors, never resizes blobs_.
+ release_all_blob_sources();
}
return failed_;
}
@@ -314,6 +328,13 @@ void
SniiCompoundWriter::release_blob_sources(std::vector<BlobFileSource>* files
files->swap(released);
}
+void SniiCompoundWriter::release_all_blob_sources() {
+ for (PendingBlobIndex& blob : blobs_) {
+ release_blob_sources(&blob.cold_files);
+ release_blob_sources(&blob.hot_files);
+ }
+}
+
SniiIndexInput SniiStreamedIndexSession::attach_encoded_norms(SniiIndexInput
in,
TrackedEncodedNorms* encoded_norms,
uint64_t
reserved_bytes) {
@@ -699,9 +720,9 @@ Status SniiCompoundWriter::write_tail() {
Status SniiCompoundWriter::finish() {
if (out_ == nullptr)
return Status::Error<ErrorCode::INVALID_ARGUMENT, false>("compound:
null file writer");
- if (!failed_.ok()) {
- return failed_;
- }
+ // poison() -- the only writer of failed_ -- already released every blob
+ // source at the transition, so there is nothing left to drop here.
+ if (!failed_.ok()) return failed_;
if (finished_)
return Status::Error<ErrorCode::INTERNAL_ERROR, false>("compound:
finish called twice");
// Crash-safety invariant 6: a begun-but-unfinished streamed session
already
@@ -713,6 +734,7 @@ Status SniiCompoundWriter::finish() {
"compound: finish with an unfinished streamed index session;
the half-fed "
"index must never be sealed away silently");
finished_ = true;
+ Defer release_blobs([this] { release_all_blob_sources(); });
RETURN_IF_ERROR(ensure_bootstrap()); // empty container still gets a header
// Aux sections were written per index at add/finish_streamed time, right
after each
@@ -724,13 +746,13 @@ Status SniiCompoundWriter::finish() {
status = write_blob_files(blob.cold_files, &blob.cold_refs);
if (!status.ok()) return poison(status);
// The sources are dead the instant their bytes are in the container
and
- // their extents are in cold_refs -- and a source can OWN its bytes
(the
- // ANN staging directory hands over shared buffers holding a whole
faiss
- // index), so holding the vector until this writer is destroyed would
pin
- // that memory across every remaining blob and, because a rowset build
- // keeps one writer per segment alive until every segment has been
closed,
- // across every segment of the rowset. Released per blob rather than
after
- // the loop so a multi-blob container never holds two at once.
+ // their extents are in cold_refs -- and a source can own resources
(the
+ // ANN staging directory hands over staged files holding a whole Faiss
+ // index), so holding the vector until this writer is destroyed would
retain
+ // those files across every remaining blob and, because a rowset build
keeps
+ // one writer per segment alive until every segment has been closed,
across
+ // every segment of the rowset. Released per blob rather than after
the loop
+ // so a multi-blob container never retains two at once.
release_blob_sources(&blob.cold_files);
}
status = write_tail();
diff --git a/be/src/storage/index/snii/writer/snii_compound_writer.h
b/be/src/storage/index/snii/writer/snii_compound_writer.h
index e37dbfcf42c..aa2c2d266f4 100644
--- a/be/src/storage/index/snii/writer/snii_compound_writer.h
+++ b/be/src/storage/index/snii/writer/snii_compound_writer.h
@@ -299,6 +299,10 @@ private:
// it serves, so keeping it alive keeps that memory resident for the
writer's
// whole life.
static void release_blob_sources(std::vector<BlobFileSource>* files);
+ // A terminal finish failure can leave both already-visited and pending
blobs.
+ // Drop every callback owner immediately instead of retaining its staging
+ // resource until this compound writer is destroyed.
+ void release_all_blob_sources();
Status write_blob_files(const std::vector<BlobFileSource>& files,
std::vector<format::NamedBlobFileRef>* refs);
// Emits the blob hot-file region and the blob directory entries; see the
.cpp.
diff --git a/be/src/storage/segment/segment_writer.cpp
b/be/src/storage/segment/segment_writer.cpp
index 70de1bcc30d..2fa8233286d 100644
--- a/be/src/storage/segment/segment_writer.cpp
+++ b/be/src/storage/segment/segment_writer.cpp
@@ -499,7 +499,33 @@ Status SegmentWriter::finalize_columns_data() {
return Status::OK();
}
+void SegmentWriter::_abandon_index_staging() {
+ // No clear() here: abandon_snii_staging() empties the staging directories
+ // themselves, so it does not matter whether the column writers -- which
hold
+ // the same directories -- are still alive.
+ if (_index_file_writer != nullptr) {
+ _index_file_writer->abandon_snii_staging();
+ }
+}
+
+// A failure below can land AFTER the ANN and BKD indexes have already been
built
+// into their staging files. The caller then returns before
close_inverted_index(),
+// so the seal that would have consumed them never runs, and the rowset writer
+// keeps this segment's IndexFileWriter -- with its staging files and their
open
+// descriptors -- until the whole load or compaction unwinds. Drop them here.
+//
+// ONLY on failure. On the success path close_inverted_index() is what consumes
+// the staging, so dropping it here would silently seal a container with no ANN
+// or BKD index in it.
Status SegmentWriter::finalize_columns_index(uint64_t* index_size) {
+ Status status = _finalize_columns_index_impl(index_size);
+ if (!status.ok()) {
+ _abandon_index_staging();
+ }
+ return status;
+}
+
+Status SegmentWriter::_finalize_columns_index_impl(uint64_t* index_size) {
uint64_t index_start = _file_writer->bytes_appended();
// Record each index range separately. Vertical compaction writes column
groups as
// data+index pairs, so a single [first index, EOF) range would include
later column data.
@@ -568,8 +594,21 @@ Status SegmentWriter::finalize_footer(uint64_t*
segment_file_size,
return Status::OK();
}
+// Wrapped for the same reason as finalize_columns_index(): a footer or
file-close
+// failure lands after the indexes have staged, and the caller returns before
+// close_inverted_index(). An inner failure has already abandoned the staging,
so
+// the second call is a no-op.
Status SegmentWriter::finalize(uint64_t* segment_file_size, uint64_t*
index_size,
SegmentIndexFileCacheInfo*
index_file_cache_info) {
+ Status status = _finalize_impl(segment_file_size, index_size,
index_file_cache_info);
+ if (!status.ok()) {
+ _abandon_index_staging();
+ }
+ return status;
+}
+
+Status SegmentWriter::_finalize_impl(uint64_t* segment_file_size, uint64_t*
index_size,
+ SegmentIndexFileCacheInfo*
index_file_cache_info) {
MonotonicStopWatch timer;
timer.start();
// check disk capacity
diff --git a/be/src/storage/segment/segment_writer.h
b/be/src/storage/segment/segment_writer.h
index cba1ac4f2f1..f6d5346971f 100644
--- a/be/src/storage/segment/segment_writer.h
+++ b/be/src/storage/segment/segment_writer.h
@@ -129,6 +129,12 @@ public:
uint64_t primary_keys_size() const { return _primary_keys_size; }
private:
+ // Bodies of finalize()/finalize_columns_index(); the public wrappers add
the
+ // abandon-on-failure step. See the .cpp.
+ Status _finalize_impl(uint64_t* segment_file_size, uint64_t* index_size,
+ SegmentIndexFileCacheInfo* index_file_cache_info);
+ Status _finalize_columns_index_impl(uint64_t* index_size);
+ void _abandon_index_staging();
friend class TestSegmentWriter;
DISALLOW_COPY_AND_ASSIGN(SegmentWriter);
Status _create_column_writer(uint32_t cid, const TabletColumn& column,
diff --git a/be/src/storage/segment/vertical_segment_writer.cpp
b/be/src/storage/segment/vertical_segment_writer.cpp
index 63511be63b1..ce79424bceb 100644
--- a/be/src/storage/segment/vertical_segment_writer.cpp
+++ b/be/src/storage/segment/vertical_segment_writer.cpp
@@ -561,7 +561,33 @@ uint64_t
VerticalSegmentWriter::_estimated_remaining_size() {
return size;
}
+void VerticalSegmentWriter::_abandon_index_staging() {
+ // No clear() here: abandon_snii_staging() empties the staging directories
+ // themselves, so it does not matter whether the column writers -- which
hold
+ // the same directories -- are still alive.
+ if (_index_file_writer != nullptr) {
+ _index_file_writer->abandon_snii_staging();
+ }
+}
+
+// A failure below can land AFTER the ANN and BKD indexes have already been
built
+// into their staging files. The caller then returns before
close_inverted_index(),
+// so the seal that would have consumed them never runs, and the rowset writer
+// keeps this segment's IndexFileWriter -- with its staging files and their
open
+// descriptors -- until the whole load or compaction unwinds. Drop them here.
+//
+// ONLY on failure. On the success path close_inverted_index() is what consumes
+// the staging, so dropping it here would silently seal a container with no ANN
+// or BKD index in it.
Status VerticalSegmentWriter::finalize_columns_index(uint64_t* index_size) {
+ Status status = _finalize_columns_index_impl(index_size);
+ if (!status.ok()) {
+ _abandon_index_staging();
+ }
+ return status;
+}
+
+Status VerticalSegmentWriter::_finalize_columns_index_impl(uint64_t*
index_size) {
uint64_t index_start = _file_writer->bytes_appended();
// Record the common index range for cloud index-only file-cache preload.
// This VerticalSegmentWriter path is used when cloud load, compaction, or
schema change flushes
@@ -616,8 +642,21 @@ Status VerticalSegmentWriter::finalize_footer(uint64_t*
segment_file_size,
return Status::OK();
}
+// Wrapped for the same reason as finalize_columns_index(): a footer or
file-close
+// failure lands after the indexes have staged, and the caller returns before
+// close_inverted_index(). An inner failure has already abandoned the staging,
so
+// the second call is a no-op.
Status VerticalSegmentWriter::finalize(uint64_t* segment_file_size, uint64_t*
index_size,
SegmentIndexFileCacheInfo*
index_file_cache_info) {
+ Status status = _finalize_impl(segment_file_size, index_size,
index_file_cache_info);
+ if (!status.ok()) {
+ _abandon_index_staging();
+ }
+ return status;
+}
+
+Status VerticalSegmentWriter::_finalize_impl(uint64_t* segment_file_size,
uint64_t* index_size,
+ SegmentIndexFileCacheInfo*
index_file_cache_info) {
MonotonicStopWatch timer;
timer.start();
// check disk capacity
diff --git a/be/src/storage/segment/vertical_segment_writer.h
b/be/src/storage/segment/vertical_segment_writer.h
index 40a33b6472f..e93829ce36f 100644
--- a/be/src/storage/segment/vertical_segment_writer.h
+++ b/be/src/storage/segment/vertical_segment_writer.h
@@ -135,6 +135,12 @@ public:
}
private:
+ // Bodies of finalize()/finalize_columns_index(); the public wrappers add
the
+ // abandon-on-failure step. See the .cpp.
+ Status _finalize_impl(uint64_t* segment_file_size, uint64_t* index_size,
+ SegmentIndexFileCacheInfo* index_file_cache_info);
+ Status _finalize_columns_index_impl(uint64_t* index_size);
+ void _abandon_index_staging();
void _init_column_meta(ColumnMetaPB* meta, uint32_t column_id, const
TabletColumn& column,
const ColumnWriterOptions& opts);
Status _create_column_writer(uint32_t cid, const TabletColumn& column,
diff --git a/be/test/storage/index/snii/snii_ann_container_test.cpp
b/be/test/storage/index/snii/snii_ann_container_test.cpp
index 2b665c1d4fa..63aa983edf4 100644
--- a/be/test/storage/index/snii/snii_ann_container_test.cpp
+++ b/be/test/storage/index/snii/snii_ann_container_test.cpp
@@ -22,8 +22,9 @@
// what faiss emits. What was missing was the adapter on both ends:
//
// write: the ANN writer had nowhere to write, because SNII opens no CLucene
-// filesystem directory. It now gets a memory-backed staging directory
-// (open_ann_directory) that begin_close() seals into a kAnn blob.
+// filesystem directory. It now gets a bounded, file-backed staging
+// directory (open_ann_directory) that begin_close() seals into a kAnn
+// blob.
// read: _open() refused too. A blob entry records ABSOLUTE container
offsets,
// the same thing a V2 compound entry records, so DorisCompoundReader
is
// reused over the container stream rather than reimplemented.
@@ -42,21 +43,30 @@
#include <CLucene.h>
#include <gen_cpp/olap_file.pb.h>
#include <gtest/gtest.h>
+#ifdef ADDRESS_SANITIZER
+#include <sanitizer/allocator_interface.h>
+#endif
#include <algorithm>
+#include <array>
#include <cstring>
#include <filesystem>
#include <memory>
+#include <roaring/roaring.hh>
+#include <set>
#include <string>
#include <vector>
#include "common/config.h"
#include "common/status.h"
+#include "exec/scan/vector_search_user_params.h"
#include "io/fs/local_file_system.h"
#include "runtime/exec_env.h"
+#include "storage/cache/ann_index_ivf_list_cache.h"
#include "storage/index/ann/ann_index_files.h"
#include "storage/index/ann/ann_index_reader.h"
#include "storage/index/ann/ann_index_writer.h"
+#include "storage/index/ann/ann_search_params.h"
#include "storage/index/index_file_reader.h"
#include "storage/index/index_file_writer.h"
#include "storage/index/inverted/inverted_index_compound_reader.h"
@@ -64,6 +74,8 @@
#include "storage/index/snii/format/metadata_directory.h"
#include "storage/index/snii/io/local_file.h"
#include "storage/index/snii/reader/snii_segment_reader.h"
+#include "storage/index/snii/snii_blob_staging_directory.h"
+#include "storage/index/snii/staged_file_probe.h"
#include "storage/olap_common.h"
#include "storage/options.h"
#include "storage/tablet/tablet_schema.h"
@@ -148,6 +160,16 @@ TabletIndex make_ivf_on_disk_index_meta() {
return meta;
}
+// Faiss opens its sub-files as ann.faiss / ann.ivfdata, so this tag selects
+// exactly the ANN staging and leaves the native BKD staging of other suites
+// alone. Observed on the filesystem rather than through the directory's own
+// bookkeeping: the bookkeeping is what the retention below disagrees with.
+constexpr const char* kAnnStageTag = "ann.";
+
+std::set<std::string> staged_ann_files() {
+ return doris::snii_test::snii_staged_files(kAnnStageTag);
+}
+
TabletIndex make_fake_bkd_index_meta() {
TabletIndexPB pb;
pb.set_index_type(IndexType::INVERTED);
@@ -181,26 +203,54 @@ protected:
// the wrong entry would then also disagree on fileLength.
_synthetic["fake_bkd_data"] = synthetic_bytes(0x11, 4096 + 17);
_synthetic["fake_bkd_index"] = synthetic_bytes(0x77, 1024 + 3);
+ // IVF-on-disk reads its lists through AnnIndexIVFListCache, which
+ // exec_env_init installs unconditionally at BE startup -- so this is
the
+ // production path, not a test convenience. Installed only if nobody
else
+ // in this binary already has one, and torn down only by whoever
installed
+ // it: create_global_cache DCHECKs on a second install, and clobbering
+ // another fixture's cache is how shared-state bugs start.
+ if (AnnIndexIVFListCache::instance() == nullptr) {
+ (void)AnnIndexIVFListCache::create_global_cache(8 * 1024 * 1024);
+ _owns_ivf_list_cache = true;
+ }
}
void TearDown() override {
+ if (_owns_ivf_list_cache) {
+ AnnIndexIVFListCache::destroy_global_cache();
+ _owns_ivf_list_cache = false;
+ }
EXPECT_TRUE(io::global_local_filesystem()->delete_directory(kTestDir).ok());
}
- // Builds one ANN index on `writer` and seals it into the writer's staging
- // area, i.e. everything up to (but not including) begin_close().
- void feed_ann_index(IndexFileWriter* writer, const TabletIndex* meta) {
- AnnIndexColumnWriter ann(writer, meta);
- ASSERT_TRUE(ann.init().ok());
+ // Feeds the standard vector set into an ANN writer the CALLER owns. The
+ // producer therefore outlives the call -- which is the shape both a failed
+ // segment flush and the whole ADD INDEX path leave behind, and the shape
+ // finish_ann_index() below cannot express.
+ static Status feed_and_finish(AnnIndexColumnWriter* ann) {
const std::vector<float> vectors = make_vectors();
std::vector<size_t> offsets(kRows + 1);
for (uint32_t i = 0; i <= kRows; ++i) {
offsets[i] = static_cast<size_t>(i) * kDim;
}
- ASSERT_TRUE(ann.add_array_values(sizeof(float), vectors.data(),
/*null_map=*/nullptr,
- reinterpret_cast<const
uint8_t*>(offsets.data()), kRows)
- .ok());
- ASSERT_TRUE(ann.finish().ok());
+ RETURN_IF_ERROR(ann->add_array_values(sizeof(float), vectors.data(),
/*null_map=*/nullptr,
+ reinterpret_cast<const
uint8_t*>(offsets.data()),
+ kRows));
+ return ann->finish();
+ }
+
+ // NOLINTNEXTLINE(readability-non-const-parameter): AnnIndexColumnWriter's
+ // constructor takes a mutable IndexFileWriter*, so this cannot be const.
+ Status finish_ann_index(IndexFileWriter* writer, const TabletIndex* meta) {
+ AnnIndexColumnWriter ann(writer, meta);
+ RETURN_IF_ERROR(ann.init());
+ return feed_and_finish(&ann);
+ }
+
+ // Builds one ANN index on `writer` and seals it into the writer's staging
+ // area, i.e. everything up to (but not including) begin_close().
+ void feed_ann_index(IndexFileWriter* writer, const TabletIndex* meta) {
+ assert_ok(finish_ann_index(writer, meta));
}
// One SNII container holding an ANN blob and a NON-ANN blob, and the path
@@ -244,6 +294,7 @@ protected:
TabletIndex _meta;
TabletIndex _fake_bkd_meta;
TabletIndex _ivf_on_disk_meta;
+ bool _owns_ivf_list_cache = false;
std::map<std::string, std::vector<uint8_t>> _synthetic;
};
@@ -334,6 +385,64 @@ TEST_F(SniiAnnContainerTest,
FaissLoadsTheIndexBackOutOfTheContainer) {
assert_ok(ann_reader->load_index(&io_ctx));
}
+// IVF-on-disk is the only ANN shape that emits two sub-files, and ann.ivfdata
--
+// the bulk of the index -- is the one whose bytes reach faiss through
+// StagedBlobFile::read_at and DorisCompoundReader rather than through faiss's
own
+// file handling. Nothing else exercises that: the load test above uses _meta
(a
+// single ann.faiss), IvfOnDiskSubFilesAreSealedInAscendingNameOrder checks
only
+// the two names and their order, and the standalone IVF save/load test runs
on a
+// RAMDirectory. A wrong length or a wrong absolute extent on ann.ivfdata would
+// leave every one of those green and surface only when the index is queried.
+//
+// The oracle is make_vectors(), not anything the container reported: a row's
own
+// vector must come back as its own row id at distance zero. nlist is 4 and the
+// default ivf_nprobe is 32, so every list is probed and the answer is exact.
+TEST_F(SniiAnnContainerTest, IvfOnDiskAnswersAQueryThroughTheProductionReader)
{
+ const std::string prefix = std::string(kTestDir) + "/ivf_query_seg";
+ io::FileWriterPtr file_writer;
+ assert_ok(io::global_local_filesystem()->create_file(
+ InvertedIndexDescriptor::get_index_file_path_v2(prefix),
&file_writer));
+ IndexFileWriter writer(io::global_local_filesystem(), prefix,
"snii_ivf_query_rowset",
+ /*seg_id=*/0, InvertedIndexStorageFormatPB::SNII,
std::move(file_writer),
+ /*can_use_ram_dir=*/false,
+ /*tablet_id=*/9912);
+ feed_ann_index(&writer, &_ivf_on_disk_meta);
+ ASSERT_FALSE(testing::Test::HasFatalFailure());
+ assert_ok(writer.begin_close());
+ assert_ok(writer.finish_close());
+
+ auto reader =
std::make_shared<IndexFileReader>(io::global_local_filesystem(), prefix,
+
InvertedIndexStorageFormatPB::SNII);
+ assert_ok(reader->init());
+ auto ann_reader = std::make_shared<AnnIndexReader>(&_ivf_on_disk_meta,
reader);
+ io::IOContext io_ctx;
+ assert_ok(ann_reader->load_index(&io_ctx));
+
+ constexpr uint32_t kProbeRow = 17;
+ const std::vector<float> vectors = make_vectors();
+ const std::vector<float> query(vectors.begin() +
static_cast<int64_t>(kProbeRow) * kDim,
+ vectors.begin() +
static_cast<int64_t>(kProbeRow + 1) * kDim);
+ roaring::Roaring candidates;
+ candidates.addRange(0, kRows);
+
+ AnnTopNParam param {.query_value = query.data(),
+ .query_value_size = kDim,
+ .limit = 4,
+ ._user_params = VectorSearchUserParams {},
+ .roaring = &candidates,
+ .rows_of_segment = kRows,
+ .enable_result_cache = false};
+ AnnIndexStats stats;
+ assert_ok(ann_reader->query(&io_ctx, ¶m, &stats));
+
+ ASSERT_NE(param.row_ids, nullptr);
+ ASSERT_FALSE(param.row_ids->empty());
+ EXPECT_EQ((*param.row_ids)[0], kProbeRow)
+ << "the nearest neighbour of a row's own vector must be that row";
+ ASSERT_NE(param.distance, nullptr);
+ EXPECT_NEAR(param.distance[0], 0.0F, 1e-4F);
+}
+
// A BKD blob has its own reader and no CLucene representation at all. Serving
a
// directory over it would hand a caller bytes no CLucene code can parse, so
the
// read adapter refuses by KIND. Probing an ABSENT id cannot test this: the
@@ -394,6 +503,39 @@ TEST_F(SniiAnnContainerTest,
AnnStagingUnderSniiCreatesNoFilesystemDirectory) {
assert_ok(writer.finish_close());
}
+// ANN serialization can be much larger than the final rowset writer's other
+// resident state. Staging must therefore retain only a fixed-size write
buffer,
+// not a second in-memory copy of the complete faiss output.
+TEST_F(SniiAnnContainerTest, LargeAnnOutputUsesBoundedHeapForStaging) {
+#ifndef ADDRESS_SANITIZER
+ GTEST_SKIP() << "heap sampling requires the ASAN allocator interface";
+#else
+ constexpr size_t kChunkBytes = 64U << 10;
+ constexpr size_t kStagedBytes = 64U << 20;
+ // The ASAN counter is process-wide, so leave ample room for unrelated
+ // background allocations. The old vector staging retained the complete
+ // 64 MiB payload and therefore still exceeds this limit by a wide margin.
+ constexpr size_t kMaximumHeapGrowth = kStagedBytes / 2;
+
+ snii_doris::SniiBlobStagingDirectory staging;
+ std::unique_ptr<lucene::store::IndexOutput>
output(staging.createOutput("ann.faiss"));
+ std::array<uint8_t, kChunkBytes> chunk {};
+
+ const size_t baseline = __sanitizer_get_current_allocated_bytes();
+ size_t peak = baseline;
+ for (size_t written = 0; written < kStagedBytes; written += chunk.size()) {
+ output->writeBytes(chunk.data(), static_cast<int32_t>(chunk.size()));
+ peak = std::max(peak, __sanitizer_get_current_allocated_bytes());
+ }
+ output->close();
+ peak = std::max(peak, __sanitizer_get_current_allocated_bytes());
+
+ EXPECT_EQ(staging.staged_bytes(), kStagedBytes);
+ EXPECT_LT(peak - baseline, kMaximumHeapGrowth)
+ << "ANN staging retained heap proportional to the serialized
index";
+#endif
+}
+
// Scoped debug-point switch: enables the fault injection sites for one test
and
// restores the process-wide config afterwards.
class ScopedDebugPoints {
@@ -412,6 +554,260 @@ private:
const bool _was_enabled;
};
+TEST_F(SniiAnnContainerTest, AnnStagingFinalizeFailureIsReturnedFromFinish) {
+ ScopedDebugPoints debug_points;
+ debug_points.enable("StagedBlobFile::finalize_error");
+
+ const std::string prefix = std::string(kTestDir) + "/finalize_failure_seg";
+ io::FileWriterPtr file_writer;
+ assert_ok(io::global_local_filesystem()->create_file(
+ InvertedIndexDescriptor::get_index_file_path_v2(prefix),
&file_writer));
+ IndexFileWriter writer(io::global_local_filesystem(), prefix,
"snii_ann_finalize_failure",
+ /*seg_id=*/0, InvertedIndexStorageFormatPB::SNII,
std::move(file_writer),
+ /*can_use_ram_dir=*/false,
+ /*tablet_id=*/9904);
+
+ Status finish_status = Status::OK();
+ ASSERT_NO_THROW({ finish_status = finish_ann_index(&writer, &_meta); });
+ EXPECT_FALSE(finish_status.ok());
+ EXPECT_NE(finish_status.to_string().find("injected blob staging finalize
failure"),
+ std::string::npos)
+ << finish_status.to_string();
+}
+
+TEST_F(SniiAnnContainerTest, FinalBufferedAppendFailureIsReturnedFromFinish) {
+ ScopedDebugPoints debug_points;
+ debug_points.enable("StagedBlobFile::append_error");
+
+ const std::string prefix = std::string(kTestDir) + "/append_failure_seg";
+ io::FileWriterPtr file_writer;
+ assert_ok(io::global_local_filesystem()->create_file(
+ InvertedIndexDescriptor::get_index_file_path_v2(prefix),
&file_writer));
+ IndexFileWriter writer(io::global_local_filesystem(), prefix,
"snii_ann_append_failure",
+ /*seg_id=*/0, InvertedIndexStorageFormatPB::SNII,
std::move(file_writer),
+ /*can_use_ram_dir=*/false,
+ /*tablet_id=*/9905);
+
+ Status finish_status = Status::OK();
+ ASSERT_NO_THROW({ finish_status = finish_ann_index(&writer, &_meta); });
+ EXPECT_FALSE(finish_status.ok());
+ EXPECT_NE(finish_status.to_string().find("injected blob staging append
failure"),
+ std::string::npos)
+ << finish_status.to_string();
+}
+
+TEST_F(SniiAnnContainerTest, IvfDataWriteFailuresAreReturnedFromFinish) {
+ struct FaultCase {
+ const char* debug_point;
+ const char* message;
+ const char* suffix;
+ int64_t tablet_id;
+ };
+ const std::array faults {
+ FaultCase {"StagedBlobFile::append_error", "injected blob staging
append failure",
+ "append", 9906},
+ FaultCase {"StagedBlobFile::finalize_error", "injected blob
staging finalize failure",
+ "finalize", 9907},
+ };
+
+ for (const auto& fault : faults) {
+ SCOPED_TRACE(fault.debug_point);
+ ScopedDebugPoints debug_points;
+ debug_points.enable(fault.debug_point);
+
+ const std::string prefix = std::string(kTestDir) + "/ivf_" +
fault.suffix + "_failure_seg";
+ io::FileWriterPtr file_writer;
+ assert_ok(io::global_local_filesystem()->create_file(
+ InvertedIndexDescriptor::get_index_file_path_v2(prefix),
&file_writer));
+ IndexFileWriter writer(io::global_local_filesystem(), prefix,
"snii_ivf_write_failure",
+ /*seg_id=*/0,
InvertedIndexStorageFormatPB::SNII,
+ std::move(file_writer),
+ /*can_use_ram_dir=*/false, fault.tablet_id);
+
+ Status finish_status = Status::OK();
+ ASSERT_NO_THROW({ finish_status = finish_ann_index(&writer,
&_ivf_on_disk_meta); });
+ EXPECT_FALSE(finish_status.ok());
+ EXPECT_NE(finish_status.to_string().find("Failed to close IVF data
output"),
+ std::string::npos)
+ << finish_status.to_string();
+ EXPECT_NE(finish_status.to_string().find(fault.message),
std::string::npos)
+ << finish_status.to_string();
+ }
+}
+
+// A failed faiss serialization must unlink its staging file THERE, not
whenever
+// the writers happen to be destroyed. Both owners are still alive at that
point
+// -- the ANN writer through _dir, the IndexFileWriter through _indices_dirs --
+// and neither caller unwinds them:
VerticalSegmentWriter::finalize_columns_index()
+// returns before clear() and close_inverted_index(), and IndexBuilder's SNII
ADD
+// INDEX path keeps every producer alive until the whole rowset has been
closed.
+// So both writers are held here on purpose; the failure tests above scope
their
+// producer out inside finish_ann_index() and cannot see this.
+TEST_F(SniiAnnContainerTest, AFailedSaveUnlinksItsStagingFile) {
+ struct Case {
+ const char* what;
+ const TabletIndex* meta;
+ int64_t tablet_id;
+ };
+ const std::array cases {
+ Case {.what = "hnsw", .meta = &_meta, .tablet_id = 9908},
+ Case {.what = "ivf_on_disk", .meta = &_ivf_on_disk_meta,
.tablet_id = 9909},
+ };
+
+ for (const auto& one : cases) {
+ SCOPED_TRACE(one.what);
+ const std::set<std::string> before = staged_ann_files();
+ ScopedDebugPoints debug_points;
+ // append(), not the seal: a write is the staging failure production
can
+ // actually hit (ENOSPC on the scratch volume), and the seal makes no
+ // durability call any more.
+ debug_points.enable("StagedBlobFile::append_error");
+
+ const std::string prefix = std::string(kTestDir) + "/failed_save_" +
one.what + "_seg";
+ io::FileWriterPtr file_writer;
+ assert_ok(io::global_local_filesystem()->create_file(
+ InvertedIndexDescriptor::get_index_file_path_v2(prefix),
&file_writer));
+ IndexFileWriter writer(io::global_local_filesystem(), prefix,
"snii_ann_failed_save_rowset",
+ /*seg_id=*/0,
InvertedIndexStorageFormatPB::SNII,
+ std::move(file_writer),
/*can_use_ram_dir=*/false, one.tablet_id);
+
+ AnnIndexColumnWriter ann(&writer, one.meta);
+ assert_ok(ann.init());
+ Status finish_status = Status::OK();
+ ASSERT_NO_THROW({ finish_status = feed_and_finish(&ann); });
+ ASSERT_FALSE(finish_status.ok()) << "the injected fault must fail the
save";
+ // The specific fault, not just any error: a save that failed for some
+ // other reason would not prove anything about the staging file.
+ ASSERT_NE(finish_status.to_string().find("injected blob staging append
failure"),
+ std::string::npos)
+ << finish_status.to_string();
+
+ EXPECT_EQ(staged_ann_files(), before)
+ << "the failed save left its staging file on the temp
filesystem while the ANN "
+ "writer and the IndexFileWriter are both still alive";
+ }
+}
+
+// Sealing hands the staged files to the container, so a producer that outlives
+// the seal must not keep them pinned. IndexBuilder's SNII ADD INDEX path
+// (_handle_single_rowset_snii) is exactly that shape: it builds every segment,
+// closes every IndexFileWriter, and only then clears _index_column_writers. If
+// the handover does not happen, one whole rowset's ANN staging sits on the
temp
+// filesystem at once -- which is precisely what SniiCompoundWriter::finish()
+// releases per blob to avoid.
+TEST_F(SniiAnnContainerTest,
SealingDrainsStagedFilesWhileProducersAreStillAlive) {
+ struct Segment {
+ std::unique_ptr<IndexFileWriter> file_writer;
+ std::unique_ptr<AnnIndexColumnWriter> producer;
+ };
+ constexpr int kSegments = 3;
+
+ const std::set<std::string> before = staged_ann_files();
+ std::vector<Segment> segments;
+ for (int seg_id = 0; seg_id < kSegments; ++seg_id) {
+ const std::string prefix =
+ std::string(kTestDir) + "/held_producer_seg" +
std::to_string(seg_id);
+ io::FileWriterPtr file_writer;
+ assert_ok(io::global_local_filesystem()->create_file(
+ InvertedIndexDescriptor::get_index_file_path_v2(prefix),
&file_writer));
+ Segment segment;
+ segment.file_writer = std::make_unique<IndexFileWriter>(
+ io::global_local_filesystem(), prefix,
"snii_ann_held_producer_rowset", seg_id,
+ InvertedIndexStorageFormatPB::SNII, std::move(file_writer),
+ /*can_use_ram_dir=*/false, /*tablet_id=*/9910);
+ segment.producer =
+
std::make_unique<AnnIndexColumnWriter>(segment.file_writer.get(), &_meta);
+ assert_ok(segment.producer->init());
+ Status finish_status = Status::OK();
+ ASSERT_NO_THROW({ finish_status =
feed_and_finish(segment.producer.get()); });
+ assert_ok(finish_status);
+ segments.push_back(std::move(segment));
+ }
+ // Sanity: their absence below must mean the seal drained them, not that
they
+ // were never staged.
+ ASSERT_EQ(staged_ann_files().size(), before.size() + kSegments);
+
+ for (auto& segment : segments) {
+ assert_ok(segment.file_writer->begin_close());
+ }
+ // Every producer is STILL alive here, exactly as it is in IndexBuilder
when
+ // it runs its begin_close loop.
+ EXPECT_EQ(staged_ann_files(), before)
+ << "sealing left the staged files pinned by the producers";
+
+ for (auto& segment : segments) {
+ assert_ok(segment.file_writer->finish_close());
+ }
+}
+
+// A segment that fails AFTER its ANN indexes have staged must not leave them
+// behind. This is the state such a failure leaves: every index staged fine,
the
+// producers are still alive because clear() has not run, and nobody will ever
+// call begin_close() -- so the seal that normally consumes the staging never
+// happens. Distinct from AFailedSaveUnlinksItsStagingFile, where serialization
+// itself failed and the per-index discard fires immediately.
+TEST_F(SniiAnnContainerTest, AbandoningASegmentDropsEveryStagedAnnFile) {
+ const std::set<std::string> before = staged_ann_files();
+ const std::string prefix = std::string(kTestDir) + "/abandoned_seg";
+ io::FileWriterPtr file_writer;
+ assert_ok(io::global_local_filesystem()->create_file(
+ InvertedIndexDescriptor::get_index_file_path_v2(prefix),
&file_writer));
+ IndexFileWriter writer(io::global_local_filesystem(), prefix,
"snii_abandoned_rowset",
+ /*seg_id=*/0, InvertedIndexStorageFormatPB::SNII,
std::move(file_writer),
+ /*can_use_ram_dir=*/false,
+ /*tablet_id=*/9913);
+
+ std::vector<std::unique_ptr<AnnIndexColumnWriter>> producers;
+ for (const TabletIndex* meta : {&_meta, &_ivf_on_disk_meta}) {
+ auto producer = std::make_unique<AnnIndexColumnWriter>(&writer, meta);
+ assert_ok(producer->init());
+ Status finish_status = Status::OK();
+ ASSERT_NO_THROW({ finish_status = feed_and_finish(producer.get()); });
+ assert_ok(finish_status);
+ producers.push_back(std::move(producer));
+ }
+ ASSERT_EQ(staged_ann_files().size(), before.size() + 3);
+
+ writer.abandon_snii_staging();
+ // The producers are still alive, as they are in a segment writer whose
+ // finalize failed before clear().
+ EXPECT_EQ(staged_ann_files(), before)
+ << "abandoning the segment left its staged ANN files on the temp
filesystem";
+}
+
+// The same handover inside ONE container: _indices_dirs holds every staging
+// directory for the whole of finish(), so the per-blob release in
+// SniiCompoundWriter::finish() can only free a sub-file that the directory has
+// already given up. Two ANN indexes, three sub-files (hnsw writes ann.faiss,
+// ivf_on_disk writes ann.faiss and ann.ivfdata).
+TEST_F(SniiAnnContainerTest, EveryAnnIndexInOneContainerIsDrainedBySealing) {
+ const std::set<std::string> before = staged_ann_files();
+ const std::string prefix = std::string(kTestDir) + "/multi_ann_seg";
+ io::FileWriterPtr file_writer;
+ assert_ok(io::global_local_filesystem()->create_file(
+ InvertedIndexDescriptor::get_index_file_path_v2(prefix),
&file_writer));
+ IndexFileWriter writer(io::global_local_filesystem(), prefix,
"snii_multi_ann_rowset",
+ /*seg_id=*/0, InvertedIndexStorageFormatPB::SNII,
std::move(file_writer),
+ /*can_use_ram_dir=*/false,
+ /*tablet_id=*/9911);
+
+ std::vector<std::unique_ptr<AnnIndexColumnWriter>> producers;
+ for (const TabletIndex* meta : {&_meta, &_ivf_on_disk_meta}) {
+ auto producer = std::make_unique<AnnIndexColumnWriter>(&writer, meta);
+ assert_ok(producer->init());
+ Status finish_status = Status::OK();
+ ASSERT_NO_THROW({ finish_status = feed_and_finish(producer.get()); });
+ assert_ok(finish_status);
+ producers.push_back(std::move(producer));
+ }
+ ASSERT_EQ(staged_ann_files().size(), before.size() + 3);
+
+ assert_ok(writer.begin_close());
+ EXPECT_EQ(staged_ann_files(), before)
+ << "one container's seal did not drain every ANN index it sealed";
+ assert_ok(writer.finish_close());
+}
+
// begin_close() returns Status. Nothing on its SNII path may throw across that
// boundary -- the non-SNII branch of the same function wraps its directory
// teardown in try/catch precisely because DorisFSDirectory::deleteDirectory()
diff --git a/be/test/storage/index/snii/snii_bkd_adapter_test.cpp
b/be/test/storage/index/snii/snii_bkd_adapter_test.cpp
index c6ae01eabc2..3243602176c 100644
--- a/be/test/storage/index/snii/snii_bkd_adapter_test.cpp
+++ b/be/test/storage/index/snii/snii_bkd_adapter_test.cpp
@@ -36,6 +36,7 @@
#include <cstdint>
#include <memory>
#include <optional>
+#include <set>
#include <string>
#include <vector>
@@ -58,6 +59,7 @@
#include "storage/index/snii/reader/snii_segment_reader.h"
#include "storage/index/snii/snii_bkd_index_reader.h"
#include "storage/index/snii/snii_bkd_index_writer.h"
+#include "storage/index/snii/staged_file_probe.h"
#include "storage/key_coder.h"
#include "storage/tablet/tablet_schema.h"
@@ -286,6 +288,49 @@ TEST_F(SniiBkdAdapterTest,
NumericColumnLandsInTheContainerAsAQueryableBkd) {
}
}
+// The bkd_data staging file must be gone once the container has sealed, even
+// though the producer is still alive. It routinely is: IndexBuilder's SNII ADD
+// INDEX path keeps every writer in _index_column_writers until the whole
rowset
+// has been closed, and a close error returns before that map is cleared. Since
+// bkd_data is sized by the point count, a producer that keeps holding it pins
one
+// such file per numeric index per segment on the temp filesystem.
+//
+// write_segment() above already keeps its writer alive across begin_close(),
so
+// this asserts what that flow was silently doing rather than inventing a
shape.
+TEST_F(SniiBkdAdapterTest,
SealingDrainsTheStagedDataWhileTheProducerIsStillAlive) {
+ const std::set<std::string> before =
doris::snii_test::snii_staged_files("bkd_data");
+ const std::vector<Row> rows = sample_rows(3000, 500);
+ const std::string prefix = test_path("held_producer");
+
+ io::FileWriterPtr file_writer;
+ assert_ok(io::global_local_filesystem()->create_file(
+ InvertedIndexDescriptor::get_index_file_path_v2(prefix),
&file_writer));
+ IndexFileWriter index_file_writer(io::global_local_filesystem(), prefix,
"held_rowset",
+ /*seg_id=*/0,
InvertedIndexStorageFormatPB::SNII,
+ std::move(file_writer),
/*can_use_ram_dir=*/true,
+ /*tablet_id=*/302);
+
+ SniiBkdIndexColumnWriter writer(&index_file_writer, &_meta, kFieldType);
+ assert_ok(writer.init());
+ std::vector<int64_t> values;
+ for (const Row& row : rows) {
+ if (!row.is_null) {
+ values.push_back(row.value);
+ }
+ }
+ assert_ok(writer.add_values("c1", values.data(), values.size()));
+ assert_ok(writer.finish());
+ // Sanity: its absence below must mean the seal drained it, not that it was
+ // never staged.
+ ASSERT_EQ(doris::snii_test::snii_staged_files("bkd_data").size(),
before.size() + 1);
+
+ assert_ok(index_file_writer.begin_close());
+ // `writer` is STILL in scope here, exactly as it is in IndexBuilder.
+ EXPECT_EQ(doris::snii_test::snii_staged_files("bkd_data"), before)
+ << "sealing left bkd_data pinned by the producer";
+ assert_ok(index_file_writer.finish_close());
+}
+
// NULL rows are carried by the SNII null-bitmap POD, not by the point set: a
// NULL that leaked in as a point would answer `col > x` for a row that has no
// value at all.
diff --git a/be/test/storage/index/snii/staged_file_probe.h
b/be/test/storage/index/snii/staged_file_probe.h
new file mode 100644
index 00000000000..d822eaba479
--- /dev/null
+++ b/be/test/storage/index/snii/staged_file_probe.h
@@ -0,0 +1,94 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <gtest/gtest.h>
+
+#include <filesystem>
+#include <set>
+#include <string>
+#include <system_error>
+
+#include "storage/index/snii/writer/temp_dir.h"
+
+// Observing SNII build-time staging files from a test.
+//
+// WHY THIS IS NOT A ONE-LINE DIRECTORY SCAN. TmpFileDirs::get_tmp_file_dir()
+// round-robins over its configured paths with an atomic counter, and every
+// resolve_temp_dir() advances it -- so do IndexFileWriter's constructor and
+// every StagedBlobFile::create. The doris_be_test binary is monolithic and
some
+// fixtures install two or three StorePath roots without putting the previous
+// TmpFileDirs back, so a test cannot assume there is only one root, nor that
two
+// scans land on the same one. A before/after count taken through
+// resolve_temp_dir() can therefore compare two different directories and
report
+// "nothing left behind" while a staging file is still linked.
+namespace doris::snii_test {
+
+// Every configured SNII temp root. A round-robin returns to its first value
+// after exactly one cycle, which is what bounds the probe. Deliberately does
not
+// install a single-root TmpFileDirs instead: that would pull the temp root out
+// from under whichever test runs next.
+inline std::set<std::string> snii_temp_roots() {
+ std::set<std::string> roots;
+ const std::string first = snii::writer::resolve_temp_dir();
+ roots.insert(first);
+ for (int i = 0; i < 64; ++i) {
+ const std::string next = snii::writer::resolve_temp_dir();
+ if (next == first) {
+ return roots;
+ }
+ roots.insert(next);
+ }
+ ADD_FAILURE() << "resolve_temp_dir() did not cycle within 64 calls, so the
staged-file "
+ "probe cannot enumerate the temp roots";
+ return roots;
+}
+
+// Full paths of the staging files named for `tag`, across every root.
+// StagedBlobFile::create names them "snii_bkdstage_<tag>_<pid>_<seq>.stage",
so
+// the tag selects one producer's sub-files and leaves other suites' staging
+// alone. Compare the returned SETS, not their sizes: equal counts can still
hide
+// one file leaking while another is created.
+//
+// A filesystem error fails the calling test rather than silently reporting an
+// empty set, which would make "nothing was left behind" unfalsifiable.
+inline std::set<std::string> snii_staged_files(const std::string& tag) {
+ const std::string prefix = "snii_bkdstage_" + tag;
+ std::set<std::string> found;
+ for (const std::string& root : snii_temp_roots()) {
+ std::error_code ec;
+ std::filesystem::directory_iterator it(root, ec);
+ if (ec) {
+ ADD_FAILURE() << "cannot scan SNII temp root " << root << ": " <<
ec.message();
+ continue;
+ }
+ while (it != std::filesystem::directory_iterator()) {
+ if (it->path().filename().string().starts_with(prefix)) {
+ found.insert(it->path().string());
+ }
+ it.increment(ec);
+ if (ec) {
+ ADD_FAILURE() << "cannot walk SNII temp root " << root << ": "
<< ec.message();
+ break;
+ }
+ }
+ }
+ return found;
+}
+
+} // namespace doris::snii_test
diff --git a/be/test/storage/index/snii/writer/snii_compound_writer_test.cpp
b/be/test/storage/index/snii/writer/snii_compound_writer_test.cpp
index 9d073835b86..3711b43dc30 100644
--- a/be/test/storage/index/snii/writer/snii_compound_writer_test.cpp
+++ b/be/test/storage/index/snii/writer/snii_compound_writer_test.cpp
@@ -23,7 +23,9 @@
#include <cstdio>
#include <cstdlib>
#include <cstring>
+#include <filesystem>
#include <functional>
+#include <memory>
#include <string>
#include <utility>
#include <vector>
@@ -32,6 +34,7 @@
#include "common/status.h"
#include "gen_cpp/snii.pb.h"
#include "storage/index/inverted/common_grams/common_grams_segment_metadata.h"
+#include "storage/index/snii/bkd/staged_blob_file.h"
#include "storage/index/snii/common/slice.h"
#include "storage/index/snii/encoding/byte_source.h"
#include "storage/index/snii/encoding/crc32c.h"
@@ -403,6 +406,36 @@ void VerifyAppendFailurePoisonsWriter(size_t
fail_on_append) {
EXPECT_EQ(0U, file.finalize_calls());
}
+void VerifyFinishFailureReleasesBlobSource(size_t fail_on_append, size_t
staged_bytes) {
+ FailOnAppendWriter file(fail_on_append);
+ SniiCompoundWriter writer(&file);
+
+ std::unique_ptr<bkd::StagedBlobFile> created;
+ ASSERT_TRUE(bkd::StagedBlobFile::create("compound_failure",
&created).ok());
+ const std::string path = created->path();
+ std::vector<uint8_t> bytes(staged_bytes, 0x5A);
+ ASSERT_TRUE(created->append(Slice(bytes)).ok());
+ ASSERT_TRUE(created->finalize().ok());
+
+ auto staged = std::shared_ptr<bkd::StagedBlobFile>(std::move(created));
+ std::vector<BlobFileSource> cold_files;
+ cold_files.push_back(BlobFileSource {
+ .name = "ann.faiss",
+ .length = staged->bytes_written(),
+ .read_fn = [staged](uint64_t offset, size_t len, uint8_t* out) ->
Status {
+ return staged->read_at(offset, len, out);
+ }});
+ ASSERT_TRUE(
+ writer.add_blob_index(7, "", LogicalIndexKind::kAnn,
std::move(cold_files), {}).ok());
+ staged.reset();
+ ASSERT_TRUE(std::filesystem::exists(path));
+
+ const Status status = writer.finish();
+ ASSERT_FALSE(status.ok());
+ EXPECT_FALSE(std::filesystem::exists(path))
+ << "a terminal compound failure retained the callback-owned
staging file";
+}
+
// A FileReader decorator that counts how many physical reads (single or
batched)
// touch a given byte window. Used to assert that the BSBF section is NOT read
at
// all on the non-resident (L1) path: with the P1 cold-read fix, open must not
@@ -1026,6 +1059,60 @@ TEST(SniiCompoundWriter,
BsbfAppendFailureReleasesReservationsBeforeReturn) {
VerifyAuxiliaryAppendFailureReleasesReservations(AuxiliarySection::kBsbf);
}
+TEST(SniiCompoundWriter,
FinishFailureReleasesBlobSourcesWhileWriterRemainsAlive) {
+ // Append 1 fails while writing a new container's bootstrap, before the
blob
+ // is read. Append 3 fails on the second blob chunk, after bootstrap and
one
+ // complete chunk have reached the output.
+ VerifyFinishFailureReleasesBlobSource(/*fail_on_append=*/1,
/*staged_bytes=*/1);
+ VerifyFinishFailureReleasesBlobSource(
+ /*fail_on_append=*/3, SniiCompoundWriter::kBlobCopyChunkBytes + 1);
+}
+
+// A poisoned compound can never seal, so every registered blob source is dead
--
+// and after the producer hands its file over, that source is the file's ONLY
+// owner. Production frequently never calls finish() after a poison:
+// SegmentWriter::_write_inverted_index() returns before
close_inverted_index(),
+// and SegmentCreator::flush() keeps the failed writer, so a release that waits
+// for finish() never runs and the staging file and its descriptor stay pinned.
+TEST(SniiCompoundWriter, PoisonReleasesBlobSourcesWithoutWaitingForFinish) {
+ // Append 1 is the bootstrap header; append 2 is the first posting byte
range
+ // of the text index below, which is what poisons the writer.
+ FailOnAppendWriter file(/*fail_on_append=*/2);
+ SniiCompoundWriter writer(&file);
+
+ std::unique_ptr<bkd::StagedBlobFile> created;
+ ASSERT_TRUE(bkd::StagedBlobFile::create("poison_boundary", &created).ok());
+ const std::string path = created->path();
+ const std::vector<uint8_t> bytes(64, 0x31);
+ ASSERT_TRUE(created->append(Slice(bytes)).ok());
+ ASSERT_TRUE(created->finalize().ok());
+
+ auto staged = std::shared_ptr<bkd::StagedBlobFile>(std::move(created));
+ std::vector<BlobFileSource> cold_files;
+ cold_files.push_back(BlobFileSource {
+ .name = "bkd_data",
+ .length = staged->bytes_written(),
+ .read_fn = [staged](uint64_t offset, size_t len, uint8_t* out) ->
Status {
+ return staged->read_at(offset, len, out);
+ }});
+ ASSERT_TRUE(
+ writer.add_blob_index(5, "", LogicalIndexKind::kBkd,
std::move(cold_files), {}).ok());
+ // The registered source is now the only owner, which is what a native-BKD
+ // producer leaves behind once it has handed the file over.
+ staged.reset();
+ ASSERT_TRUE(std::filesystem::exists(path));
+
+ // A later text index fails while streaming its physical sections.
+ const Status poisoned = writer.add_logical_index(MakeIndex(6, "body", 30));
+ ASSERT_FALSE(poisoned.ok());
+ ASSERT_NE(poisoned.to_string().find("injected append failure"),
std::string::npos)
+ << poisoned.to_string();
+
+ // No finish() here on purpose: production does not reach one.
+ EXPECT_FALSE(std::filesystem::exists(path))
+ << "a poisoned compound kept its blob callback, pinning the
staging file: " << path;
+}
+
TEST(SniiCompoundWriter, ReopeningLogicalReaderClearsPreviousCommonGramsState)
{
auto with_common_grams = EmptyIndex(7, "with");
with_common_grams.common_grams_metadata = CompleteCommonGramsMetadata();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
