[refact](inverted index) refact compound idx writer (#59219)

airborne12 · web-flow · commit 26750bea547f · 2025-12-22T18:53:27.000+08:00
### What problem does this PR solve?
This PR refactors the compound index writer for the inverted index V2
format to optimize performance, particularly for cloud storage systems
like S3. The main improvement is bypassing unnecessary directory
operations by directly creating the output stream using an existing file
writer.

Key changes:

Added a static factory method FSIndexOutputV2::create() for direct
instantiation without requiring a Directory object
Refactored IndexStorageFormatV2::create_output_stream() to return only
an IndexOutput instead of a Directory-IndexOutput pair
Removed the cloud mode check in DorisFSDirectoryFactory::getDirectory()
to simplify code (directory creation is safe for all file systems)
diff --git a/be/src/olap/rowset/segment_v2/index_storage_format_v2.cpp b/be/src/olap/rowset/segment_v2/index_storage_format_v2.cpp
@@ -39,7 +39,6 @@ IndexStorageFormatV2::IndexStorageFormatV2(IndexFileWriter* index_file_writer)
         : IndexStorageFormat(index_file_writer) {}
 
 Status IndexStorageFormatV2::write() {
-    std::unique_ptr<lucene::store::Directory, DirectoryDeleter> out_dir = nullptr;
     std::unique_ptr<lucene::store::IndexOutput> compound_file_output = nullptr;
     ErrorContext error_context;
     try {
@@ -48,11 +47,13 @@ Status IndexStorageFormatV2::write() {
         // Prepare file metadata
         auto file_metadata = prepare_file_metadata(current_offset);
 
-        // Create output stream
-        auto result = create_output_stream();
-        out_dir = std::move(result.first);
-        compound_file_output = std::move(result.second);
-        VLOG_DEBUG << fmt::format("Output compound index file to streams: {}", out_dir->toString());
+        // Create output stream directly without directory operations.
+        // This is important for cloud storage (like S3) where directory operations are not
+        // supported or unnecessary.
+        compound_file_output = create_output_stream();
+        auto index_path = InvertedIndexDescriptor::get_index_file_path_v2(
+                _index_file_writer->_index_path_prefix);
+        VLOG_DEBUG << fmt::format("Output compound index file to: {}", index_path);
 
         // Write version and number of indices
         write_version_and_indices_count(compound_file_output.get());
@@ -75,10 +76,7 @@ Status IndexStorageFormatV2::write() {
         error_context.err_msg.append(err.what());
         LOG(ERROR) << error_context.err_msg;
     }
-    FINALLY({
-        FINALLY_CLOSE(compound_file_output);
-        FINALLY_CLOSE(out_dir);
-    })
+    FINALLY({ FINALLY_CLOSE(compound_file_output); })
 
     return Status::OK();
 }
@@ -177,21 +175,16 @@ std::vector<FileMetadata> IndexStorageFormatV2::prepare_file_metadata(int64_t& c
     return file_metadata;
 }
 
-std::pair<std::unique_ptr<lucene::store::Directory, DirectoryDeleter>,
-          std::unique_ptr<lucene::store::IndexOutput>>
-IndexStorageFormatV2::create_output_stream() {
-    io::Path index_path {InvertedIndexDescriptor::get_index_file_path_v2(
-            _index_file_writer->_index_path_prefix)};
-
-    auto* out_dir = DorisFSDirectoryFactory::getDirectory(_index_file_writer->_fs,
-                                                          index_path.parent_path().c_str());
-    out_dir->set_file_writer_opts(_index_file_writer->_opts);
-    std::unique_ptr<lucene::store::Directory, DirectoryDeleter> out_dir_ptr(out_dir);
-
+std::unique_ptr<lucene::store::IndexOutput> IndexStorageFormatV2::create_output_stream() {
+    // For V2 format, we create the output stream directly using the file writer,
+    // bypassing the directory layer entirely. This optimization is especially important
+    // for cloud storage (like S3) where:
+    // 1. Directory operations (exists, create_directory) are unnecessary overhead
+    // 2. S3 doesn't have a real directory concept - directories are just key prefixes
+    // 3. The file writer is already created and ready to use
     DCHECK(_index_file_writer->_idx_v2_writer != nullptr)
             << "inverted index file writer v2 is nullptr";
-    auto compound_file_output = out_dir->createOutputV2(_index_file_writer->_idx_v2_writer.get());
-    return {std::move(out_dir_ptr), std::move(compound_file_output)};
+    return DorisFSDirectory::FSIndexOutputV2::create(_index_file_writer->_idx_v2_writer.get());
 }
 
 void IndexStorageFormatV2::write_version_and_indices_count(lucene::store::IndexOutput* output) {
diff --git a/be/src/olap/rowset/segment_v2/index_storage_format_v2.h b/be/src/olap/rowset/segment_v2/index_storage_format_v2.h
@@ -43,9 +43,10 @@ class IndexStorageFormatV2 : public IndexStorageFormat {
 private:
     int64_t header_length();
     std::vector<FileMetadata> prepare_file_metadata(int64_t& current_offset);
-    virtual std::pair<std::unique_ptr<lucene::store::Directory, DirectoryDeleter>,
-                      std::unique_ptr<lucene::store::IndexOutput>>
-    create_output_stream();
+    // Creates the output stream for writing the compound file.
+    // For V2 format, we directly create FSIndexOutputV2 using the file writer,
+    // avoiding unnecessary directory operations (important for cloud storage like S3).
+    virtual std::unique_ptr<lucene::store::IndexOutput> create_output_stream();
     void write_version_and_indices_count(lucene::store::IndexOutput* output);
     virtual void write_index_headers_and_metadata(lucene::store::IndexOutput* output,
                                                   const std::vector<FileMetadata>& file_metadata);
diff --git a/be/src/olap/rowset/segment_v2/inverted_index_fs_directory.cpp b/be/src/olap/rowset/segment_v2/inverted_index_fs_directory.cpp
@@ -454,6 +454,26 @@ int64_t DorisFSDirectory::FSIndexOutputV2::length() const {
     return _index_v2_file_writer->bytes_appended();
 }
 
+std::unique_ptr<lucene::store::IndexOutput> DorisFSDirectory::FSIndexOutputV2::create(
+        io::FileWriter* file_writer) {
+    auto ret = std::make_unique<FSIndexOutputV2>();
+    ErrorContext error_context;
+    try {
+        ret->init(file_writer);
+    } catch (CLuceneError& err) {
+        error_context.eptr = std::current_exception();
+        error_context.err_msg.append("FSIndexOutputV2::create init error: ");
+        error_context.err_msg.append(err.what());
+        LOG(ERROR) << error_context.err_msg;
+    }
+    FINALLY_EXCEPTION({
+        if (error_context.eptr) {
+            FINALLY_CLOSE(ret);
+        }
+    })
+    return ret;
+}
+
 DorisFSDirectory::DorisFSDirectory() {
     filemode = 0644;
     this->lockFactory = nullptr;
@@ -905,28 +925,25 @@ DorisFSDirectory* DorisFSDirectoryFactory::getDirectory(const io::FileSystemSPtr
     if (config::inverted_index_ram_dir_enable && can_use_ram_dir) {
         dir = _CLNEW DorisRAMFSDirectory();
     } else {
-        // cloud mode does not need to create directory
-        if (!config::is_cloud_mode()) {
-            bool exists = false;
-            auto st = _fs->exists(file, &exists);
-            DBUG_EXECUTE_IF("DorisFSDirectoryFactory::getDirectory_exists_status_is_not_ok", {
-                st = Status::Error<ErrorCode::INTERNAL_ERROR>(
-                        "debug point: "
-                        "DorisFSDirectoryFactory::getDirectory_exists_status_is_not_ok");
-            })
-            LOG_AND_THROW_IF_ERROR(st, "Get directory exists IO error");
-            if (!exists) {
-                st = _fs->create_directory(file);
-                DBUG_EXECUTE_IF(
-                        "DorisFSDirectoryFactory::getDirectory_create_directory_status_is_not_ok", {
-                            st = Status::Error<ErrorCode::INTERNAL_ERROR>(
-                                    "debug point: "
-                                    "DorisFSDirectoryFactory::getDirectory_create_directory_status_"
-                                    "is_"
-                                    "not_ok");
-                        })
-                LOG_AND_THROW_IF_ERROR(st, "Get directory create directory IO error");
-            }
+        bool exists = false;
+        auto st = _fs->exists(file, &exists);
+        DBUG_EXECUTE_IF("DorisFSDirectoryFactory::getDirectory_exists_status_is_not_ok", {
+            st = Status::Error<ErrorCode::INTERNAL_ERROR>(
+                    "debug point: "
+                    "DorisFSDirectoryFactory::getDirectory_exists_status_is_not_ok");
+        })
+        LOG_AND_THROW_IF_ERROR(st, "Get directory exists IO error");
+        if (!exists) {
+            st = _fs->create_directory(file);
+            DBUG_EXECUTE_IF(
+                    "DorisFSDirectoryFactory::getDirectory_create_directory_status_is_not_ok", {
+                        st = Status::Error<ErrorCode::INTERNAL_ERROR>(
+                                "debug point: "
+                                "DorisFSDirectoryFactory::getDirectory_create_directory_status_"
+                                "is_"
+                                "not_ok");
+                    })
+            LOG_AND_THROW_IF_ERROR(st, "Get directory create directory IO error");
         }
         dir = _CLNEW DorisFSDirectory();
     }
diff --git a/be/src/olap/rowset/segment_v2/inverted_index_fs_directory.h b/be/src/olap/rowset/segment_v2/inverted_index_fs_directory.h
@@ -246,6 +246,11 @@ class DorisFSDirectory::FSIndexOutputV2 : public lucene::store::BufferedIndexOut
     ~FSIndexOutputV2() override;
     void close() override;
     int64_t length() const override;
+
+    // Static factory method to create FSIndexOutputV2 directly without Directory object.
+    // This is useful for compound file creation where we already have a FileWriter
+    // and don't need directory operations (especially for cloud storage like S3).
+    static std::unique_ptr<lucene::store::IndexOutput> create(io::FileWriter* file_writer);
 };
 
 /**
diff --git a/be/test/olap/rowset/segment_v2/inverted_index_file_writer_test.cpp b/be/test/olap/rowset/segment_v2/inverted_index_file_writer_test.cpp
@@ -692,9 +692,8 @@ class IndexStorageFormatV2MockCreateOutputStream : public IndexStorageFormatV2 {
     IndexStorageFormatV2MockCreateOutputStream(IndexFileWriter* index_file_writer)
             : IndexStorageFormatV2(index_file_writer) {}
 
-    MOCK_METHOD((std::pair<std::unique_ptr<lucene::store::Directory, DirectoryDeleter>,
-                           std::unique_ptr<lucene::store::IndexOutput>>),
-                create_output_stream, (), (override));
+    MOCK_METHOD((std::unique_ptr<lucene::store::IndexOutput>), create_output_stream, (),
+                (override));
 };
 
 class IndexFileWriterMockCreateOutputStreamV1 : public IndexFileWriter {
@@ -808,12 +807,9 @@ TEST_F(IndexFileWriterTest, WriteV2OutputTest) {
     EXPECT_CALL(
             *(IndexStorageFormatV2MockCreateOutputStream*)writer_mock._index_storage_format.get(),
             create_output_stream())
-            .WillOnce(::testing::Invoke(
-                    [&]() -> std::pair<std::unique_ptr<lucene::store::Directory, DirectoryDeleter>,
-                                       std::unique_ptr<lucene::store::IndexOutput>> {
-                        return std::make_pair(std::move(out_dir_ptr),
-                                              std::move(compound_file_output));
-                    }));
+            .WillOnce(::testing::Invoke([&]() -> std::unique_ptr<lucene::store::IndexOutput> {
+                return std::move(compound_file_output);
+            }));
 
     int64_t index_id = 1;
     std::string index_suffix = "suffix1";
@@ -871,12 +867,9 @@ TEST_F(IndexFileWriterTest, WriteV2OutputCloseErrorTest) {
     EXPECT_CALL(
             *(IndexStorageFormatV2MockCreateOutputStream*)writer_mock._index_storage_format.get(),
             create_output_stream())
-            .WillOnce(::testing::Invoke(
-                    [&]() -> std::pair<std::unique_ptr<lucene::store::Directory, DirectoryDeleter>,
-                                       std::unique_ptr<lucene::store::IndexOutput>> {
-                        return std::make_pair(std::move(out_dir_ptr),
-                                              std::move(compound_file_output));
-                    }));
+            .WillOnce(::testing::Invoke([&]() -> std::unique_ptr<lucene::store::IndexOutput> {
+                return std::move(compound_file_output);
+            }));
 
     int64_t index_id = 1;
     std::string index_suffix = "suffix1";
