[fix](be) Release SNII writer reservations on auxiliary failure (#66855)

### What problem does this PR solve?

Issue Number: N/A

Related PR: #66052

Problem Summary:

SNII adopted a `LogicalIndexWriter` into `SniiCompoundWriter::indexes_`
before appending its norms, null bitmap, and block-split bloom-filter
sections. An injected append failure at those three boundaries
reproduced retained `MemoryReporter` charges of 96, 86, and 60 bytes
respectively after `add_logical_index()` returned. The ordinary build
caller transfers reporter ownership only after that call succeeds, so
failure teardown could leave the compound writer holding reservations
that refer to an already-destroyed reporter.

This change keeps the logical writer and its placement local while
writing all auxiliary sections, then adopts both into the compound
writer only after every append succeeds. The same ownership rule is
applied to the streamed path. Poisoning behavior and the successful
append order, offsets, file layout, and bytes are unchanged, so this
does not change the SNII storage format and does not require rebuilding
existing indexes.

Author: airborne12

be/src/storage/index/snii/writer/snii_compound_writer.cpp

@@ -173,11 +173,12 @@ Status SniiCompoundWriter::add_logical_index(const SniiIndexInput& in) {
     status = liw->stream_dict_region_into(out_);
     if (!status.ok()) return poison(status);
     p.dict_len = out_->bytes_written() - p.dict_off;
+    status = write_index_aux_sections(*liw, p);
+    if (!status.ok()) {
+        return poison(status);
+    }
     indexes_.push_back(std::move(liw));
     placements_.push_back(p);
-    // liw has been moved from; write_index_aux_sections works off indexes_.back().
-    status = write_index_aux_sections(indexes_.size() - 1);
-    if (!status.ok()) return poison(status);
     return Status::OK();
 }
 
@@ -479,18 +480,15 @@ Status SniiCompoundWriter::finish_streamed_index(SniiStreamedIndexSession* sessi
     status = session->writer_->stream_dict_region_into(out_);
     if (!status.ok()) return poison(status);
     p.dict_len = out_->bytes_written() - p.dict_off;
-    // The index joins the container (indexes_/placements_) here, but session->finished_
-    // is not set until write_index_aux_sections below also succeeds. A failure ANYWHERE
-    // in this function -- finish_streamed()/stream_dict_region_into() above, or
-    // write_index_aux_sections below -- calls poison(), which sets failed_ before
-    // returning. finish() checks "if (!failed_.ok()) return failed_;" ahead of its
-    // has_active_session() gate, so a poisoned writer fails loudly on its own; it can
-    // never fall through to sealing a tail that silently omits an index whose posting
-    // bytes are already in the file.
+    // The index joins the container only after every section succeeds. A failure
+    // anywhere in this function poisons the compound writer, so finish() cannot seal
+    // a tail that omits posting bytes already written to the file.
+    status = write_index_aux_sections(*session->writer_, p);
+    if (!status.ok()) {
+        return poison(status);
+    }
     indexes_.push_back(std::move(session->writer_));
     placements_.push_back(p);
-    status = write_index_aux_sections(indexes_.size() - 1);
-    if (!status.ok()) return poison(status);
     session->finished_ = true;
     return Status::OK();
 }
@@ -506,29 +504,25 @@ Status SniiCompoundWriter::write_bootstrap() {
 // Writes one index's norms / null bitmap / bsbf directly after its [posting][dict] pair.
 // Bytes are released as soon as they are on disk rather than being held until finish(),
 // which also lowers import peak memory -- a content column's bsbf runs to MBs.
-Status SniiCompoundWriter::write_index_aux_sections(size_t index) {
-    DORIS_CHECK_LT(index, indexes_.size());
-    DORIS_CHECK_LT(index, placements_.size());
-    LogicalIndexWriter& w = *indexes_[index];
-    Placement& p = placements_[index];
-
-    if (w.has_norms() && !w.norms_bytes().empty()) {
-        p.norms_off = out_->bytes_written();
-        RETURN_IF_ERROR(append(w.norms_bytes()));
-        p.norms_len = out_->bytes_written() - p.norms_off;
-        w.release_norms_bytes();
-    }
-    if (w.has_null_bitmap()) {
-        p.null_off = out_->bytes_written();
-        RETURN_IF_ERROR(append(w.null_bitmap_bytes()));
-        p.null_len = out_->bytes_written() - p.null_off;
-        w.release_null_bitmap_bytes();
-    }
-    if (w.has_bsbf()) {
-        p.bsbf_off = out_->bytes_written();
-        RETURN_IF_ERROR(append(w.bsbf_bytes()));
-        p.bsbf_len = out_->bytes_written() - p.bsbf_off;
-        w.release_bsbf_bytes();
+Status SniiCompoundWriter::write_index_aux_sections(LogicalIndexWriter& writer,
+                                                    Placement& placement) {
+    if (writer.has_norms() && !writer.norms_bytes().empty()) {
+        placement.norms_off = out_->bytes_written();
+        RETURN_IF_ERROR(append(writer.norms_bytes()));
+        placement.norms_len = out_->bytes_written() - placement.norms_off;
+        writer.release_norms_bytes();
+    }
+    if (writer.has_null_bitmap()) {
+        placement.null_off = out_->bytes_written();
+        RETURN_IF_ERROR(append(writer.null_bitmap_bytes()));
+        placement.null_len = out_->bytes_written() - placement.null_off;
+        writer.release_null_bitmap_bytes();
+    }
+    if (writer.has_bsbf()) {
+        placement.bsbf_off = out_->bytes_written();
+        RETURN_IF_ERROR(append(writer.bsbf_bytes()));
+        placement.bsbf_len = out_->bytes_written() - placement.bsbf_off;
+        writer.release_bsbf_bytes();
     }
     return Status::OK();
 }

be/src/storage/index/snii/writer/snii_compound_writer.h

@@ -279,12 +279,11 @@ class SniiCompoundWriter {
 
     Status ensure_bootstrap();
     Status write_bootstrap();
-    // Writes indexes_[index]'s norms/null-bitmap/bsbf immediately after its
-    // [posting][dict] pair and fills placements_[index]. Keeping one index's sections
+    // Writes one index's norms/null-bitmap/bsbf immediately after its
+    // [posting][dict] pair and fills its placement. Keeping one index's sections
     // contiguous is what makes a single-index cold query touch one cache block instead
     // of three; the previous layout grouped these by section type across all indexes.
-    // Must be called after indexes_/placements_ have been pushed for this index.
-    Status write_index_aux_sections(size_t index);
+    Status write_index_aux_sections(LogicalIndexWriter& writer, Placement& placement);
     Status write_tail();
     Status append(const std::vector<uint8_t>& bytes);
     Status poison(Status status);

be/test/storage/index/snii/writer/snii_compound_writer_test.cpp

@@ -47,6 +47,7 @@
 #include "storage/index/snii/format/frq_prelude.h"
 #include "storage/index/snii/format/metadata_blob.h"
 #include "storage/index/snii/format/metadata_directory.h"
+#include "storage/index/snii/format/null_bitmap.h"
 #include "storage/index/snii/format/prx_pod.h"
 #include "storage/index/snii/format/sampled_term_index.h"
 #include "storage/index/snii/format/tail_pointer.h"
@@ -336,6 +337,55 @@ class FailOnAppendWriter final : public io::FileWriter {
     std::vector<uint8_t> bytes_;
 };
 
+enum class AuxiliarySection { kNorms, kNullBitmap, kBsbf };
+
+class FailOnAuxiliarySectionWriter final : public io::FileWriter {
+public:
+    explicit FailOnAuxiliarySectionWriter(AuxiliarySection target) : target_(target) {}
+
+    Status append(Slice data) override {
+        if (is_target(data)) {
+            ++target_append_calls_;
+            return Status::Error<doris::ErrorCode::IO_ERROR, false>(
+                    "injected auxiliary section append failure");
+        }
+        bytes_.insert(bytes_.end(), data.data(), data.data() + data.size());
+        return Status::OK();
+    }
+
+    Status finalize() override { return Status::OK(); }
+
+    uint64_t bytes_written() const override { return bytes_.size(); }
+    size_t target_append_calls() const { return target_append_calls_; }
+
+private:
+    bool is_target(Slice data) const {
+        if (target_ == AuxiliarySection::kBsbf) {
+            if (data.size() < kBsbfHeaderSize) {
+                return false;
+            }
+            BsbfHeader header;
+            return BsbfHeader::parse(data.subslice(0, kBsbfHeaderSize), bytes_.size(), &header)
+                           .ok() &&
+                   data.size() == kBsbfHeaderSize + header.num_bytes;
+        }
+
+        ByteSource source(data);
+        FramedSection section;
+        if (!SectionFramer::read(source, &section).ok() || source.remaining() != 0) {
+            return false;
+        }
+        const uint8_t target_type = target_ == AuxiliarySection::kNorms
+                                            ? static_cast<uint8_t>(SectionType::kNormsPod)
+                                            : kNullBitmapSectionType;
+        return section.type == target_type;
+    }
+
+    AuxiliarySection target_;
+    size_t target_append_calls_ = 0;
+    std::vector<uint8_t> bytes_;
+};
+
 void VerifyAppendFailurePoisonsWriter(size_t fail_on_append) {
     FailOnAppendWriter file(fail_on_append);
     SniiCompoundWriter writer(&file);
@@ -487,6 +537,37 @@ SniiIndexInput MakeIndex(uint64_t index_id, const std::string& suffix, uint32_t
     return in;
 }
 
+SniiIndexInput MakeIndexWithAllAuxiliarySections(MemoryReporter* reporter) {
+    SniiIndexInput in;
+    in.index_id = 7;
+    in.index_suffix = "body";
+    in.config = IndexConfig::kDocsPositionsScoring;
+    in.doc_count = 3;
+    in.null_docids = {2};
+    in.encoded_norms = {1, 2, 3};
+    in.terms.push_back(MakeTerm("apple", {0, 1}, true));
+    in.mem_reporter = reporter;
+    return in;
+}
+
+void VerifyAuxiliaryAppendFailureReleasesReservations(AuxiliarySection target) {
+    FailOnAuxiliarySectionWriter file(target);
+    auto reporter = std::make_unique<MemoryReporter>();
+    auto compound = std::make_unique<SniiCompoundWriter>(&file);
+    const SniiIndexInput input = MakeIndexWithAllAuxiliarySections(reporter.get());
+
+    const Status status = compound->add_logical_index(input);
+    ASSERT_FALSE(status.ok());
+    ASSERT_EQ(1U, file.target_append_calls());
+    ASSERT_EQ(0, reporter->current_bytes())
+            << "failed logical writer retained reservations after add_logical_index returned";
+
+    // The caller only transfers reporter ownership after add_logical_index succeeds.
+    // Exercise the real failed-call teardown order under ASAN: the reporter dies first.
+    reporter.reset();
+    compound.reset();
+}
+
 // Locate a term through the full reader walk and return its DictEntry.
 Status LocateEntry(const std::vector<uint8_t>& file, const SampledTermIndexReader& sti,
                    const DictBlockDirectoryReader& dbd, const std::string& term, bool* found,
@@ -933,6 +1014,18 @@ TEST(SniiCompoundWriter, AppendFailurePoisonsWriterBeforeAnyValidFooter) {
     }
 }
 
+TEST(SniiCompoundWriter, NormsAppendFailureReleasesReservationsBeforeReturn) {
+    VerifyAuxiliaryAppendFailureReleasesReservations(AuxiliarySection::kNorms);
+}
+
+TEST(SniiCompoundWriter, NullBitmapAppendFailureReleasesReservationsBeforeReturn) {
+    VerifyAuxiliaryAppendFailureReleasesReservations(AuxiliarySection::kNullBitmap);
+}
+
+TEST(SniiCompoundWriter, BsbfAppendFailureReleasesReservationsBeforeReturn) {
+    VerifyAuxiliaryAppendFailureReleasesReservations(AuxiliarySection::kBsbf);
+}
+
 TEST(SniiCompoundWriter, ReopeningLogicalReaderClearsPreviousCommonGramsState) {
     auto with_common_grams = EmptyIndex(7, "with");
     with_common_grams.common_grams_metadata = CompleteCommonGramsMetadata();