[opt](cloud) optimize load performance for inverted index when pack small files (#59011)

Related PR: #57770 

Problem Summary:

When merging small files with inverted indexes, the segment close
operation was synchronously waiting for inverted index files to be
uploaded to S3. This blocking behavior significantly impacted the
memtable flush thread performance, causing bottlenecks in the data
loading pipeline.

Solution:

The solution introduces a two-phase close mechanism for inverted index
file writers:

1. **Asynchronous Close Phase**: During segment close, inverted index
files are closed asynchronously and the S3 upload task is submitted
immediately without waiting for completion.

2. **Wait Phase**: When the load channel closes, the system waits for
all pending S3 upload tasks to complete, ensuring data consistency.

Author: liaoxin01

be/src/index-tools/index_tool.cpp

@@ -621,11 +621,16 @@ int main(int argc, char** argv) {
         _CLLDELETE(analyzer);
         _CLLDELETE(char_string_reader);
 
-        auto ret = index_file_writer->close();
+        auto ret = index_file_writer->begin_close();
         if (!ret.ok()) {
             std::cerr << "IndexFileWriter close error:" << ret.msg() << std::endl;
             return -1;
         }
+        ret = index_file_writer->finish_close();
+        if (!ret.ok()) {
+            std::cerr << "IndexFileWriter wait close error:" << ret.msg() << std::endl;
+            return -1;
+        }
     } else if (FLAGS_operation == "show_nested_files_v2") {
         if (FLAGS_idx_file_path == "") {
             std::cout << "no file flag for show " << std::endl;

be/src/olap/rowset/beta_rowset_writer.cpp

@@ -206,16 +206,24 @@ Status InvertedIndexFileCollection::add(int seg_id, IndexFileWriterPtr&& index_w
     return Status::OK();
 }
 
-Status InvertedIndexFileCollection::close() {
+Status InvertedIndexFileCollection::begin_close() {
     std::lock_guard lock(_lock);
     for (auto&& [id, writer] : _inverted_index_file_writers) {
-        RETURN_IF_ERROR(writer->close());
+        RETURN_IF_ERROR(writer->begin_close());
         _total_size += writer->get_index_file_total_size();
     }
 
     return Status::OK();
 }
 
+Status InvertedIndexFileCollection::finish_close() {
+    std::lock_guard lock(_lock);
+    for (auto&& [id, writer] : _inverted_index_file_writers) {
+        RETURN_IF_ERROR(writer->finish_close());
+    }
+    return Status::OK();
+}
+
 Result<std::vector<const InvertedIndexFileInfo*>>
 InvertedIndexFileCollection::inverted_index_file_info(int seg_id_offset) {
     std::lock_guard lock(_lock);
@@ -1097,7 +1105,8 @@ Status BetaRowsetWriter::create_segment_writer_for_segcompaction(
     _segcompaction_worker->get_file_writer().reset(file_writer.release());
     if (auto& idx_file_writer = _segcompaction_worker->get_inverted_index_file_writer();
         idx_file_writer != nullptr) {
-        RETURN_IF_ERROR(idx_file_writer->close());
+        RETURN_IF_ERROR(idx_file_writer->begin_close());
+        RETURN_IF_ERROR(idx_file_writer->finish_close());
     }
     _segcompaction_worker->get_inverted_index_file_writer().reset(index_file_writer.release());
     return Status::OK();

be/src/olap/rowset/beta_rowset_writer.h

@@ -91,7 +91,10 @@ class InvertedIndexFileCollection {
 
     // Close all file writers
     // If the inverted index file writer is not closed, an error will be thrown during destruction
-    Status close();
+    Status begin_close();
+
+    // Wait for all inverted index file writers to be closed
+    Status finish_close();
 
     // Get inverted index file info in segment id order.
     // `seg_id_offset` is the offset of the segment id relative to the subscript of `_inverted_index_file_writers`,
@@ -214,9 +217,11 @@ class BaseBetaRowsetWriter : public RowsetWriter {
     // Some index files are written during normal compaction and some files are written during index compaction.
     // After all index writes are completed, call this method to write the final compound index file.
     Status _close_inverted_index_file_writers() {
-        RETURN_NOT_OK_STATUS_WITH_WARN(_idx_files.close(),
+        RETURN_NOT_OK_STATUS_WITH_WARN(_idx_files.begin_close(),
                                        "failed to close index file when build new rowset");
         this->_total_index_size += _idx_files.get_total_index_size();
+        RETURN_NOT_OK_STATUS_WITH_WARN(_idx_files.finish_close(),
+                                       "failed to wait close index file when build new rowset");
         return Status::OK();
     }
 

be/src/olap/rowset/segment_creator.cpp

@@ -111,7 +111,9 @@ Status SegmentFlusher::_internal_parse_variant_columns(vectorized::Block& block)
 }
 
 Status SegmentFlusher::close() {
-    return _seg_files.close();
+    RETURN_IF_ERROR(_seg_files.close());
+    RETURN_IF_ERROR(_idx_files.finish_close());
+    return Status::OK();
 }
 
 Status SegmentFlusher::_add_rows(std::unique_ptr<segment_v2::SegmentWriter>& segment_writer,

be/src/olap/rowset/segment_v2/index_file_writer.cpp

@@ -184,15 +184,15 @@ Result<std::unique_ptr<IndexSearcherBuilder>> IndexFileWriter::_construct_index_
     return IndexSearcherBuilder::create_index_searcher_builder(reader_type);
 }
 
-Status IndexFileWriter::close() {
+Status IndexFileWriter::begin_close() {
     DCHECK(!_closed) << debug_string();
     _closed = true;
     if (_indices_dirs.empty()) {
         // An empty file must still be created even if there are no indexes to write
         if (dynamic_cast<io::StreamSinkFileWriter*>(_idx_v2_writer.get()) != nullptr ||
             dynamic_cast<io::S3FileWriter*>(_idx_v2_writer.get()) != nullptr ||
             dynamic_cast<io::PackedFileWriter*>(_idx_v2_writer.get()) != nullptr) {
-            return _idx_v2_writer->close();
+            return _idx_v2_writer->close(true);
         }
         return Status::OK();
     }
@@ -224,7 +224,24 @@ Status IndexFileWriter::close() {
                     err.what());
         }
     }
-    LOG_INFO("IndexFileWriter closing, enable_write_index_searcher_cache: {}",
+    return Status::OK();
+}
+
+Status IndexFileWriter::finish_close() {
+    DCHECK(_closed) << debug_string();
+    if (_indices_dirs.empty()) {
+        // An empty file must still be created even if there are no indexes to write
+        if (dynamic_cast<io::StreamSinkFileWriter*>(_idx_v2_writer.get()) != nullptr ||
+            dynamic_cast<io::S3FileWriter*>(_idx_v2_writer.get()) != nullptr ||
+            dynamic_cast<io::PackedFileWriter*>(_idx_v2_writer.get()) != nullptr) {
+            return _idx_v2_writer->close(false);
+        }
+        return Status::OK();
+    }
+    if (_idx_v2_writer != nullptr && _idx_v2_writer->state() != io::FileWriter::State::CLOSED) {
+        RETURN_IF_ERROR(_idx_v2_writer->close(false));
+    }
+    LOG_INFO("IndexFileWriter finish_close, enable_write_index_searcher_cache: {}",
              config::enable_write_index_searcher_cache);
     Status st = Status::OK();
     if (config::enable_write_index_searcher_cache) {

be/src/olap/rowset/segment_v2/index_file_writer.h

@@ -57,7 +57,13 @@ class IndexFileWriter {
     Status delete_index(const TabletIndex* index_meta);
     Status initialize(InvertedIndexDirectoryMap& indices_dirs);
     Status add_into_searcher_cache();
-    Status close();
+    // Begin the close process. This mainly triggers the asynchronous close operation of
+    // _idx_v2_writer by calling close(true), which starts the close process but returns
+    // immediately without waiting for completion.
+    Status begin_close();
+    // Finish the close process. This waits for the close operation to complete by calling
+    // _idx_v2_writer->close(false), which blocks until the close is fully done.
+    Status finish_close();
     const InvertedIndexFileInfo* get_index_file_info() const {
         DCHECK(_closed) << debug_string();
         return &_file_info;

be/src/olap/rowset/segment_v2/inverted_index_fs_directory.cpp

@@ -433,7 +433,7 @@ void DorisFSDirectory::FSIndexOutputV2::close() {
         _index_v2_file_writer = nullptr;
     })
     if (_index_v2_file_writer) {
-        auto ret = _index_v2_file_writer->close();
+        auto ret = _index_v2_file_writer->close(true);
         DBUG_EXECUTE_IF("DorisFSDirectory::FSIndexOutput._set_writer_close_status_error",
                         { ret = Status::Error<INTERNAL_ERROR>("writer close status error"); })
         if (!ret.ok()) {

be/src/olap/rowset/segment_v2/segment_writer.h

@@ -148,7 +148,7 @@ class SegmentWriter {
             *inverted_index_file_size = 0;
             return Status::OK();
         }
-        RETURN_IF_ERROR(_index_file_writer->close());
+        RETURN_IF_ERROR(_index_file_writer->begin_close());
         *inverted_index_file_size = _index_file_writer->get_index_file_total_size();
         return Status::OK();
     }

be/src/olap/rowset/segment_v2/vertical_segment_writer.h

@@ -125,7 +125,7 @@ class VerticalSegmentWriter {
             *inverted_index_file_size = 0;
             return Status::OK();
         }
-        RETURN_IF_ERROR(_index_file_writer->close());
+        RETURN_IF_ERROR(_index_file_writer->begin_close());
         *inverted_index_file_size = _index_file_writer->get_index_file_total_size();
         return Status::OK();
     }

be/src/olap/task/index_builder.cpp

@@ -368,13 +368,20 @@ Status IndexBuilder::handle_single_rowset(RowsetMetaSharedPtr output_rowset_meta
                 _index_file_writers.emplace(seg_ptr->id(), std::move(index_file_writer));
             }
             for (auto&& [seg_id, index_file_writer] : _index_file_writers) {
-                auto st = index_file_writer->close();
+                auto st = index_file_writer->begin_close();
                 if (!st.ok()) {
                     LOG(ERROR) << "close index_file_writer error:" << st;
                     return st;
                 }
                 inverted_index_size += index_file_writer->get_index_file_total_size();
             }
+            for (auto&& [seg_id, index_file_writer] : _index_file_writers) {
+                auto st = index_file_writer->finish_close();
+                if (!st.ok()) {
+                    LOG(ERROR) << "wait close index_file_writer error:" << st;
+                    return st;
+                }
+            }
             _index_file_writers.clear();
             output_rowset_meta->set_data_disk_size(output_rowset_meta->data_disk_size());
             output_rowset_meta->set_total_disk_size(output_rowset_meta->total_disk_size() +
@@ -610,7 +617,7 @@ Status IndexBuilder::handle_single_rowset(RowsetMetaSharedPtr output_rowset_meta
             _olap_data_convertor->reset();
         }
         for (auto&& [seg_id, index_file_writer] : _index_file_writers) {
-            auto st = index_file_writer->close();
+            auto st = index_file_writer->begin_close();
             DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_file_writer_close_error", {
                 st = Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
                         "debug point: handle_single_rowset_file_writer_close_error");
@@ -621,6 +628,13 @@ Status IndexBuilder::handle_single_rowset(RowsetMetaSharedPtr output_rowset_meta
             }
             inverted_index_size += index_file_writer->get_index_file_total_size();
         }
+        for (auto&& [seg_id, index_file_writer] : _index_file_writers) {
+            auto st = index_file_writer->finish_close();
+            if (!st.ok()) {
+                LOG(ERROR) << "wait close index_file_writer error:" << st;
+                return st;
+            }
+        }
         _index_column_writers.clear();
         _index_file_writers.clear();
         output_rowset_meta->set_data_disk_size(output_rowset_meta->data_disk_size());