[Enhancement](file-cache) Add fine-grained control for compaction file cache (#60609)

Support selective caching of index files only during compaction in cloud
mode.

Changes:
- Add two mBool configs to control index-only caching for base and
cumulative compaction:
  * enable_base_compaction_output_write_index_only (default: false)
  * enable_cumu_compaction_output_write_index_only (default: false)

- Extend RowsetWriterContext with compaction_output_write_index_only
field to mark whether only index files should be cached

- Modify get_file_writer_options() to accept is_index_file parameter:
* When compaction_output_write_index_only=true and is_index_file=false,
set write_file_cache=false to skip caching data files
  * Index files continue to be cached normally

- Update file writer creation call sites to pass is_index_file
parameter:
  * Index file writers: pass true
  * Segment (data) file writers: pass false

Benefits:
- Reduces cache pressure by avoiding caching large data files during
compaction
- Preserves index file caching for query performance
- Provides separate control for base and cumulative compaction
strategies
- Maintains backward compatibility with default settings

### What problem does this PR solve?

Issue Number: close #xxx

Related PR: #xxx

Problem Summary:

### Release note

None

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [ ] Regression test
    - [x] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [x] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

Author: gavinchou

be/src/cloud/config.cpp

@@ -155,5 +155,8 @@ DEFINE_mBool(enable_cache_read_from_peer, "true");
 // If the value is -1, use the `rehash_tablet_after_be_dead_seconds` setting in the `fe` configuration as the expiration time.
 DEFINE_mInt64(cache_read_from_peer_expired_seconds, "-1");
 
+DEFINE_mBool(enable_file_cache_write_base_compaction_index_only, "false");
+DEFINE_mBool(enable_file_cache_write_cumu_compaction_index_only, "false");
+
 #include "common/compile_check_end.h"
 } // namespace doris::config

be/src/cloud/config.h

@@ -195,5 +195,11 @@ DECLARE_mBool(enable_cache_read_from_peer);
 
 DECLARE_mInt64(cache_read_from_peer_expired_seconds);
 
+// Base compaction output: only write index files to file cache, not data files
+DECLARE_mBool(enable_file_cache_write_base_compaction_index_only);
+
+// Cumulative compaction output: only write index files to file cache, not data files
+DECLARE_mBool(enable_file_cache_write_cumu_compaction_index_only);
+
 #include "common/compile_check_end.h"
 } // namespace doris::config

be/src/olap/compaction.cpp

@@ -88,6 +88,32 @@ using std::vector;
 
 namespace doris {
 using namespace ErrorCode;
+
+// Determine whether to enable index-only file cache mode for compaction output.
+// This function decides if only index files should be written to cache, based on:
+// - write_file_cache: whether file cache is enabled
+// - compaction_type: type of compaction (base or cumulative)
+// - enable_base_index_only: config flag for base compaction
+// - enable_cumu_index_only: config flag for cumulative compaction
+// Returns true if index-only mode should be enabled, false otherwise.
+bool should_enable_compaction_cache_index_only(bool write_file_cache, ReaderType compaction_type,
+                                               bool enable_base_index_only,
+                                               bool enable_cumu_index_only) {
+    if (!write_file_cache) {
+        return false;
+    }
+
+    if (compaction_type == ReaderType::READER_BASE_COMPACTION && enable_base_index_only) {
+        return true;
+    }
+
+    if (compaction_type == ReaderType::READER_CUMULATIVE_COMPACTION && enable_cumu_index_only) {
+        return true;
+    }
+
+    return false;
+}
+
 namespace {
 #include "common/compile_check_begin.h"
 
@@ -1749,6 +1775,13 @@ Status CloudCompactionMixin::construct_output_rowset_writer(RowsetWriterContext&
     ctx.write_file_cache = should_cache_compaction_output();
     ctx.file_cache_ttl_sec = _tablet->ttl_seconds();
     ctx.approximate_bytes_to_write = _input_rowsets_total_size;
+
+    // Set fine-grained control: only write index files to cache if configured
+    ctx.compaction_output_write_index_only = should_enable_compaction_cache_index_only(
+            ctx.write_file_cache, compaction_type(),
+            config::enable_file_cache_write_base_compaction_index_only,
+            config::enable_file_cache_write_cumu_compaction_index_only);
+
     ctx.tablet = _tablet;
     ctx.job_id = _uuid;
 

be/src/olap/rowset/beta_rowset_writer.cpp

@@ -1050,8 +1050,9 @@ Status BaseBetaRowsetWriter::_build_tmp(RowsetSharedPtr& rowset_ptr) {
 }
 
 Status BaseBetaRowsetWriter::_create_file_writer(const std::string& path,
-                                                 io::FileWriterPtr& file_writer) {
-    io::FileWriterOptions opts = _context.get_file_writer_options();
+                                                 io::FileWriterPtr& file_writer,
+                                                 bool is_index_file) {
+    io::FileWriterOptions opts = _context.get_file_writer_options(is_index_file);
     Status st = _context.fs()->create_file(path, &file_writer, &opts);
     if (!st.ok()) {
         LOG(WARNING) << "failed to create writable file. path=" << path << ", err: " << st;
@@ -1069,9 +1070,9 @@ Status BaseBetaRowsetWriter::create_file_writer(uint32_t segment_id, io::FileWri
         std::string prefix =
                 std::string {InvertedIndexDescriptor::get_index_file_path_prefix(segment_path)};
         std::string index_path = InvertedIndexDescriptor::get_index_file_path_v2(prefix);
-        return _create_file_writer(index_path, file_writer);
+        return _create_file_writer(index_path, file_writer, true /* is_index_file */);
     } else if (file_type == FileType::SEGMENT_FILE) {
-        return _create_file_writer(segment_path, file_writer);
+        return _create_file_writer(segment_path, file_writer, false /* is_index_file */);
     }
     return Status::Error<ErrorCode::INTERNAL_ERROR>(
             fmt::format("failed to create file = {}, file type = {}", segment_path, file_type));
@@ -1081,7 +1082,8 @@ Status BaseBetaRowsetWriter::create_index_file_writer(uint32_t segment_id,
                                                       IndexFileWriterPtr* index_file_writer) {
     RETURN_IF_ERROR(RowsetWriter::create_index_file_writer(segment_id, index_file_writer));
     // used for inverted index format v1
-    (*index_file_writer)->set_file_writer_opts(_context.get_file_writer_options());
+    (*index_file_writer)
+            ->set_file_writer_opts(_context.get_file_writer_options(true /* is_index_file */));
     return Status::OK();
 }
 
@@ -1091,7 +1093,7 @@ Status BetaRowsetWriter::create_segment_writer_for_segcompaction(
     std::string path = BetaRowset::local_segment_path_segcompacted(_context.tablet_path,
                                                                    _context.rowset_id, begin, end);
     io::FileWriterPtr file_writer;
-    RETURN_IF_ERROR(_create_file_writer(path, file_writer));
+    RETURN_IF_ERROR(_create_file_writer(path, file_writer, false /* is_index_file */));
 
     IndexFileWriterPtr index_file_writer;
     if (_context.tablet_schema->has_inverted_index() || _context.tablet_schema->has_ann_index()) {
@@ -1100,7 +1102,8 @@ Status BetaRowsetWriter::create_segment_writer_for_segcompaction(
         if (_context.tablet_schema->get_inverted_index_storage_format() !=
             InvertedIndexStorageFormatPB::V1) {
             std::string index_path = InvertedIndexDescriptor::get_index_file_path_v2(prefix);
-            RETURN_IF_ERROR(_create_file_writer(index_path, idx_file_writer));
+            RETURN_IF_ERROR(
+                    _create_file_writer(index_path, idx_file_writer, true /* is_index_file */));
         }
         index_file_writer = std::make_unique<IndexFileWriter>(
                 _context.fs(), prefix, _context.rowset_id.to_string(), _num_segcompacted,

be/src/olap/rowset/beta_rowset_writer.h

@@ -202,7 +202,8 @@ class BaseBetaRowsetWriter : public RowsetWriter {
 protected:
     Status _generate_delete_bitmap(int32_t segment_id);
     virtual Status _build_rowset_meta(RowsetMeta* rowset_meta, bool check_segment_num = false);
-    Status _create_file_writer(const std::string& path, io::FileWriterPtr& file_writer);
+    Status _create_file_writer(const std::string& path, io::FileWriterPtr& file_writer,
+                               bool is_index_file = false);
     virtual Status _close_file_writers();
     virtual Status _check_segment_number_limit(size_t segnum);
     virtual int64_t _num_seg() const;

be/src/olap/rowset/rowset_writer_context.h

@@ -106,6 +106,8 @@ struct RowsetWriterContext {
     bool is_hot_data = false;
     uint64_t file_cache_ttl_sec = 0;
     uint64_t approximate_bytes_to_write = 0;
+    // If true, compaction output only writes index files to file cache, not data files
+    bool compaction_output_write_index_only = false;
     /// end file cache opts
 
     // segcompaction for this RowsetWriter, only enabled when importing data
@@ -234,13 +236,17 @@ struct RowsetWriterContext {
 
     io::FileSystem& fs_ref() const { return *fs(); }
 
-    io::FileWriterOptions get_file_writer_options() {
-        io::FileWriterOptions opts {.write_file_cache = write_file_cache,
-                                    .is_cold_data = is_hot_data,
-                                    .file_cache_expiration_time = file_cache_ttl_sec,
-                                    .approximate_bytes_to_write = approximate_bytes_to_write};
+    io::FileWriterOptions get_file_writer_options(bool is_index_file = false) {
+        bool should_write_cache = write_file_cache;
+        // If configured to only write index files to cache, skip cache for data files
+        if (compaction_output_write_index_only && !is_index_file) {
+            should_write_cache = false;
+        }
 
-        return opts;
+        return io::FileWriterOptions {.write_file_cache = should_write_cache,
+                                      .is_cold_data = is_hot_data,
+                                      .file_cache_expiration_time = file_cache_ttl_sec,
+                                      .approximate_bytes_to_write = approximate_bytes_to_write};
     }
 };