branch-4.0: [fix](file cache) Fix inverted index cache not cleaned on compaction failure #58141 (#58165)

Cherry-picked from #58141

Co-authored-by: zzzxl <yangsiyu@selectdb.com>

Author: github-actions[bot]

be/src/olap/compaction.cpp

@@ -1643,6 +1643,13 @@ Status CloudCompactionMixin::garbage_collection() {
             auto* file_cache = io::FileCacheFactory::instance()->get_by_path(file_key);
             file_cache->remove_if_cached_async(file_key);
         }
+        for (const auto& [_, index_writer] : beta_rowset_writer->index_file_writers()) {
+            for (const auto& file_name : index_writer->get_index_file_names()) {
+                auto file_key = io::BlockFileCache::hash(file_name);
+                auto* file_cache = io::FileCacheFactory::instance()->get_by_path(file_key);
+                file_cache->remove_if_cached_async(file_key);
+            }
+        }
     }
     return Status::OK();
 }

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

@@ -226,6 +226,20 @@ Status IndexFileWriter::close() {
     return Status::OK();
 }
 
+std::vector<std::string> IndexFileWriter::get_index_file_names() const {
+    std::vector<std::string> file_names;
+    if (_storage_format == InvertedIndexStorageFormatPB::V2) {
+        file_names.emplace_back(
+                InvertedIndexDescriptor::get_index_file_name_v2(_rowset_id, _seg_id));
+    } else {
+        for (const auto& [index_info, _] : _indices_dirs) {
+            file_names.emplace_back(InvertedIndexDescriptor::get_index_file_name_v1(
+                    _rowset_id, _seg_id, index_info.first, index_info.second));
+        }
+    }
+    return file_names;
+}
+
 std::string IndexFileWriter::debug_string() const {
     std::stringstream indices_dirs;
     for (const auto& [index, dir] : _indices_dirs) {

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

@@ -69,6 +69,7 @@ class IndexFileWriter {
     const io::FileSystemSPtr& get_fs() const { return _fs; }
     InvertedIndexStorageFormatPB get_storage_format() const { return _storage_format; }
     void set_file_writer_opts(const io::FileWriterOptions& opts) { _opts = opts; }
+    std::vector<std::string> get_index_file_names() const;
     std::string debug_string() const;
 
 private:

be/test/olap/rowset/segment_v2/inverted_index_file_writer_test.cpp

@@ -1658,6 +1658,55 @@ TEST_F(IndexFileWriterTest, DeleteIndexNullMetaTest) {
     ASSERT_TRUE(status.msg().find("Index metadata is null") != std::string::npos);
 }
 
+TEST_F(IndexFileWriterTest, GetIndexFileNamesTest) {
+    // Test V2 format
+    {
+        IndexFileWriter writer(_fs, _index_path_prefix, _rowset_id, _seg_id,
+                               InvertedIndexStorageFormatPB::V2);
+        std::vector<std::string> file_names = writer.get_index_file_names();
+        ASSERT_EQ(file_names.size(), 1);
+        EXPECT_EQ(file_names[0],
+                  InvertedIndexDescriptor::get_index_file_name_v2(_rowset_id, _seg_id));
+    }
+
+    // Test V1 format
+    {
+        IndexFileWriter writer(_fs, _index_path_prefix, _rowset_id, _seg_id,
+                               InvertedIndexStorageFormatPB::V1);
+
+        // Insert some directories
+        int64_t index_id_1 = 1;
+        std::string suffix_1 = "suffix1";
+        EXPECT_TRUE(writer._insert_directory_into_map(index_id_1, suffix_1,
+                                                      std::make_shared<DorisFSDirectory>())
+                            .ok());
+
+        int64_t index_id_2 = 2;
+        std::string suffix_2 = "suffix2";
+        EXPECT_TRUE(writer._insert_directory_into_map(index_id_2, suffix_2,
+                                                      std::make_shared<DorisFSDirectory>())
+                            .ok());
+
+        std::vector<std::string> file_names = writer.get_index_file_names();
+        ASSERT_EQ(file_names.size(), 2);
+
+        std::string expected_name_1 = InvertedIndexDescriptor::get_index_file_name_v1(
+                _rowset_id, _seg_id, index_id_1, suffix_1);
+        std::string expected_name_2 = InvertedIndexDescriptor::get_index_file_name_v1(
+                _rowset_id, _seg_id, index_id_2, suffix_2);
+
+        bool found_1 = false;
+        bool found_2 = false;
+        for (const auto& name : file_names) {
+            if (name == expected_name_1) found_1 = true;
+            if (name == expected_name_2) found_2 = true;
+        }
+
+        EXPECT_TRUE(found_1);
+        EXPECT_TRUE(found_2);
+    }
+}
+
 // Test for add_into_searcher_cache with StreamSinkFileWriter nullptr check
 TEST_F(IndexFileWriterTest, AddIntoSearcherCacheStreamSinkNullTest) {
     IndexFileWriter writer(_fs, _index_path_prefix, _rowset_id, _seg_id,