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

be/src/olap/compaction.cpp

@@ -1525,6 +1525,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->inverted_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/inverted_index_file_writer.cpp

@@ -672,4 +672,19 @@ void InvertedIndexFileWriter::copy_files_data(lucene::store::IndexOutput* output
         copyFile(meta.filename.c_str(), meta.directory, output, buffer, buffer_length);
     }
 }
+
+std::vector<std::string> InvertedIndexFileWriter::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;
+}
+
 } // namespace doris::segment_v2

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

@@ -89,6 +89,7 @@ class InvertedIndexFileWriter {
     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 {
         std::stringstream indices_dirs;