[feature](be) Initialize background hole fill

### What problem does this PR solve?

Issue Number: #67611 (tracking)

Related PR: #67292

Problem Summary: Background partial-block writeback needs production defaults and a BE-owned lifetime bounded by the File Cache lifecycle. Add validated hole-fill configuration, create the manager after File Cache initialization, and stop it before File Cache teardown. Configure a 4 GiB BE-wide queued/active buffer budget and a 1.0 hole-read amplification limit. Use hole_fill_workers_per_be for block workers (default 32, range 1 through 128) and hole_fill_remote_read_threads_per_be for the shared multi-range remote-read pool (default 64, positive values). Both settings resize their pools at runtime without rebuilding the manager. Single-range reads execute on block workers outside the remote-read pool limit. Add mutable hole_fill_merge_delay_ms with a 2000 ms default and nonnegative validation. Positive values delay activation from the first admission timestamp; dynamic updates apply to already queued tasks without resetting that timestamp.

### Release note

Support runtime resizing of BE-wide block workers and the shared multi-range remote-read pool. Configure minimum queued-fragment aggregation time through hole_fill_merge_delay_ms; its 2000 ms default keeps queued fragments mergeable before dispatch, and setting it to 0 disables deliberate waiting. Limit queued/active partial-block buffers to 4 GiB per BE and use a 1.0 background read-amplification limit.

### Check List (For Author)

- Test: Unit Test
    - Existing PartialBlockWritebackOptionsTest coverage via run-be-ut.sh -j100 with ASAN/PCH; no new tests for the default-value update.
- Behavior changed: Yes; configure background hole filling, resize its pools at runtime, and adjust queued partial-block aggregation delay.
- Does this need documentation: Yes; configuration and execution contracts are tracked in #67611 and #67292.

be/src/common/config.cpp

@@ -1941,6 +1941,24 @@ DEFINE_Validator(read_ahead_max_read_amplification_ratio,
 DEFINE_mDouble(read_ahead_block_fill_min_coverage, "1.0");
 DEFINE_Validator(read_ahead_block_fill_min_coverage,
                  [](double value) { return std::isfinite(value) && value > 0.0 && value <= 1.0; });
+// Background completion policy and BE-wide resource limits for partial File Cache blocks
+DEFINE_Int64(hole_fill_max_gap_bytes, "32768"); // 32 KiB
+DEFINE_Validator(hole_fill_max_gap_bytes, [](int64_t value) { return value >= 0; });
+DEFINE_Int64(hole_fill_max_range_bytes, "1048576"); // 1 MiB
+DEFINE_Validator(hole_fill_max_range_bytes, [](int64_t value) { return value > 0; });
+DEFINE_Double(hole_fill_max_read_amplification_ratio, "1.0");
+DEFINE_Validator(hole_fill_max_read_amplification_ratio,
+                 [](double value) { return std::isfinite(value) && value >= 1.0; });
+DEFINE_Int64(hole_fill_max_pending_bytes_per_be, "4294967296"); // 4 GiB
+DEFINE_Validator(hole_fill_max_pending_bytes_per_be, [](int64_t value) { return value > 0; });
+DEFINE_mInt32(hole_fill_workers_per_be, "32");
+DEFINE_Validator(hole_fill_workers_per_be, [](int32_t value) { return value > 0 && value <= 128; });
+// With 32 block workers, allow two pooled hole reads per active block on average.
+DEFINE_mInt32(hole_fill_remote_read_threads_per_be, "64");
+DEFINE_Validator(hole_fill_remote_read_threads_per_be, [](int32_t value) { return value > 0; });
+// Minimum aggregation time from first partial-block admission; zero disables the delay.
+DEFINE_mInt32(hole_fill_merge_delay_ms, "2000"); // 2000 ms
+DEFINE_Validator(hole_fill_merge_delay_ms, [](int32_t value) { return value >= 0; });
 // Enable segment file cache block prefetch for compaction
 DEFINE_mBool(enable_compaction_segment_file_cache_prefetch, "false");
 // Number of blocks to prefetch ahead in segment iterator for compaction

be/src/common/config.h

@@ -1912,6 +1912,16 @@ DECLARE_mInt64(read_ahead_max_gap_bytes);
 DECLARE_mInt64(read_ahead_max_range_bytes);
 DECLARE_mDouble(read_ahead_max_read_amplification_ratio);
 DECLARE_mDouble(read_ahead_block_fill_min_coverage);
+// Background completion policy and BE-wide resource limits for partial File Cache blocks
+DECLARE_Int64(hole_fill_max_gap_bytes);
+DECLARE_Int64(hole_fill_max_range_bytes);
+DECLARE_Double(hole_fill_max_read_amplification_ratio);
+DECLARE_Int64(hole_fill_max_pending_bytes_per_be);
+DECLARE_mInt32(hole_fill_workers_per_be);
+// Pool threads for multi-range hole reads; single-range reads run on block workers.
+DECLARE_mInt32(hole_fill_remote_read_threads_per_be);
+// Minimum time queued partial blocks remain mergeable; applies to existing queued tasks.
+DECLARE_mInt32(hole_fill_merge_delay_ms);
 // Enable segment file cache block prefetch for compaction
 DECLARE_mBool(enable_compaction_segment_file_cache_prefetch);
 // Number of blocks to prefetch ahead in segment iterator for compaction

be/src/runtime/exec_env.h

@@ -59,6 +59,7 @@ class FDCache;
 class FileCacheFactory;
 class HdfsMgr;
 class PackedFileManager;
+class PartialBlockWritebackManager;
 } // namespace io
 
 namespace segment_v2 {
@@ -271,6 +272,9 @@ public:
     io::FileRangeReadScheduler* file_range_read_scheduler() {
         return _file_range_read_scheduler.get();
     }
+    io::PartialBlockWritebackManager* partial_block_writeback_manager() {
+        return _partial_block_writeback_manager.get();
+    }
     ThreadPool* peer_race_s3_thread_pool() { return _peer_race_s3_thread_pool.get(); }
 
     void init_file_cache_factory(std::vector<doris::CachePath>& cache_paths);
@@ -515,6 +519,7 @@ private:
     // Threadpool used to prefetch segment file cache blocks
     std::unique_ptr<ThreadPool> _segment_prefetch_thread_pool;
     std::unique_ptr<io::FileRangeReadScheduler> _file_range_read_scheduler;
+    std::unique_ptr<io::PartialBlockWritebackManager> _partial_block_writeback_manager;
     std::unique_ptr<ThreadPool> _peer_race_s3_thread_pool;
 
     FragmentMgr* _fragment_mgr = nullptr;

be/src/runtime/exec_env_init.cpp

@@ -61,6 +61,7 @@
 #include "io/cache/block_file_cache_downloader.h"
 #include "io/cache/block_file_cache_factory.h"
 #include "io/cache/fs_file_cache_storage.h"
+#include "io/cache/partial_block_writeback_manager.h"
 #include "io/fs/file_meta_cache.h"
 #include "io/fs/file_range_read_scheduler.h"
 #include "io/fs/hdfs_file_writer.h"
@@ -327,6 +328,25 @@ Status ExecEnv::_init(const std::vector<StorePath>& store_paths,
     _file_cache_factory = new io::FileCacheFactory();
     std::vector<doris::CachePath> cache_paths;
     init_file_cache_factory(cache_paths);
+    if (config::is_cloud_mode()) {
+        io::PartialBlockWritebackOptions options {
+                .block_size = cast_set<size_t>(config::file_cache_each_block_size),
+                .worker_count = cast_set<size_t>(config::hole_fill_workers_per_be),
+                .remote_read_thread_count = config::hole_fill_remote_read_threads_per_be,
+                .merge_delay_ms = config::hole_fill_merge_delay_ms,
+                .max_pending_bytes = cast_set<size_t>(config::hole_fill_max_pending_bytes_per_be),
+                .hole_fill_coalesce =
+                        {
+                                .max_gap_bytes = cast_set<size_t>(config::hole_fill_max_gap_bytes),
+                                .max_range_bytes =
+                                        cast_set<size_t>(config::hole_fill_max_range_bytes),
+                                .max_read_amplification_ratio =
+                                        config::hole_fill_max_read_amplification_ratio,
+                        },
+        };
+        RETURN_IF_ERROR(io::PartialBlockWritebackManager::create(
+                options, &_partial_block_writeback_manager));
+    }
     doris::io::BeConfDataDirReader::init_be_conf_data_dir(store_paths, spill_store_paths,
                                                           cache_paths);
     _init_runtime_filter_timer_queue();
@@ -893,6 +913,7 @@ void ExecEnv::destroy() {
     SAFE_STOP(_external_scan_context_mgr);
     SAFE_STOP(_fragment_mgr);
     SAFE_SHUTDOWN(_file_range_read_scheduler);
+    SAFE_SHUTDOWN(_partial_block_writeback_manager);
     SAFE_STOP(_runtime_filter_timer_queue);
     // NewLoadStreamMgr should be destoried before storage_engine & after fragment_mgr stopped.
     _load_stream_mgr.reset();
@@ -983,6 +1004,7 @@ void ExecEnv::destroy() {
     SAFE_DELETE(_fragment_mgr);
     SAFE_DELETE(_workload_sched_mgr);
     SAFE_DELETE(_workload_group_manager);
+    _partial_block_writeback_manager.reset(nullptr);
     SAFE_DELETE(_file_cache_factory);
     SAFE_DELETE(_runtime_filter_timer_queue);
     SAFE_DELETE(_dict_factory);
@@ -1058,6 +1080,21 @@ void ExecEnv::destroy() {
 namespace doris::config {
 namespace {
 
+void resize_hole_fill_workers(int32_t old_value, int32_t new_value) {
+    if (old_value == new_value) {
+        return;
+    }
+    auto* manager = ExecEnv::GetInstance()->partial_block_writeback_manager();
+    if (manager == nullptr) {
+        return;
+    }
+    Status status = manager->resize_workers(cast_set<size_t>(new_value));
+    if (!status.ok()) {
+        LOG(WARNING) << "Failed to resize partial block writeback workers from " << old_value
+                     << " to " << new_value << ": " << status;
+    }
+}
+
 void refresh_ms_rpc_rate_limiters() {
     auto* services = ExecEnv::GetInstance()->ms_rpc_rate_limit_services();
     if (services != nullptr) {
@@ -1084,6 +1121,22 @@ void refresh_ms_backpressure_coordinator_params() {
 
 } // namespace
 
+DEFINE_ON_UPDATE(hole_fill_workers_per_be, resize_hole_fill_workers);
+DEFINE_ON_UPDATE(hole_fill_remote_read_threads_per_be, [](int32_t old_value, int32_t new_value) {
+    auto* manager = ExecEnv::GetInstance()->partial_block_writeback_manager();
+    if (manager != nullptr && old_value != new_value) {
+        WARN_IF_ERROR(manager->resize_remote_read_threads(new_value),
+                      "Failed to resize partial block remote-read pool");
+    }
+});
+
+DEFINE_ON_UPDATE(hole_fill_merge_delay_ms, [](int32_t, int32_t new_value) {
+    auto* manager = ExecEnv::GetInstance()->partial_block_writeback_manager();
+    if (manager != nullptr) {
+        manager->set_merge_delay_ms(new_value);
+    }
+});
+
 // Callback to update warmup download rate limiter when config changes is registered
 DEFINE_ON_UPDATE(file_cache_warmup_download_rate_limit_bytes_per_second,
                  [](int64_t old_val, int64_t new_val) {

be/test/io/cache/partial_block_writeback_manager_test.cpp

@@ -33,6 +33,7 @@
 #include <utility>
 #include <vector>
 
+#include "common/config.h"
 #include "cpp/sync_point.h"
 #include "io/cache/block_file_cache_test_common.h"
 #include "io/cache/inflight_write_buffer_index.h"
@@ -1642,5 +1643,47 @@ TEST(PartialBlockWritebackOptionsTest, AcceptsProductionDefaults) {
     EXPECT_TRUE(options.validate().ok());
 }
 
+TEST(PartialBlockWritebackOptionsTest, WorkerConfigIsMutableAndBounded) {
+    const int32_t old_worker_count = config::hole_fill_workers_per_be;
+    const int32_t old_merge_delay = config::hole_fill_merge_delay_ms;
+    Defer restore_worker_count {[&]() {
+        EXPECT_TRUE(config::set_config("hole_fill_workers_per_be", std::to_string(old_worker_count))
+                            .ok());
+        EXPECT_TRUE(config::set_config("hole_fill_merge_delay_ms", std::to_string(old_merge_delay))
+                            .ok());
+    }};
+
+    EXPECT_FALSE(config::set_config("hole_fill_workers_per_be", "0").ok());
+    EXPECT_FALSE(config::set_config("hole_fill_workers_per_be", "129").ok());
+    EXPECT_EQ(config::hole_fill_workers_per_be, old_worker_count);
+    const int32_t new_worker_count = old_worker_count == 1 ? 2 : 1;
+    EXPECT_TRUE(
+            config::set_config("hole_fill_workers_per_be", std::to_string(new_worker_count)).ok());
+    EXPECT_EQ(config::hole_fill_workers_per_be, new_worker_count);
+    EXPECT_FALSE(config::set_config("hole_fill_merge_delay_ms", "-1").ok());
+    EXPECT_EQ(config::hole_fill_merge_delay_ms, old_merge_delay);
+    EXPECT_TRUE(config::set_config("hole_fill_merge_delay_ms", "0").ok());
+    EXPECT_EQ(config::hole_fill_merge_delay_ms, 0);
+    EXPECT_TRUE(config::set_config("hole_fill_merge_delay_ms", "20").ok());
+    EXPECT_EQ(config::hole_fill_merge_delay_ms, 20);
+}
+
+TEST(PartialBlockWritebackOptionsTest, RemoteReadThreadConfigIsMutableAndPositive) {
+    const int32_t old_count = config::hole_fill_remote_read_threads_per_be;
+    Defer restore {[&]() {
+        EXPECT_TRUE(config::set_config("hole_fill_remote_read_threads_per_be",
+                                       std::to_string(old_count))
+                            .ok());
+    }};
+    EXPECT_FALSE(config::set_config("hole_fill_remote_read_threads_per_be", "0").ok());
+    EXPECT_FALSE(config::set_config("hole_fill_remote_read_threads_per_be", "-1").ok());
+    EXPECT_EQ(config::hole_fill_remote_read_threads_per_be, old_count);
+    const int32_t new_count = old_count == 1 ? 2 : 1;
+    EXPECT_TRUE(
+            config::set_config("hole_fill_remote_read_threads_per_be", std::to_string(new_count))
+                    .ok());
+    EXPECT_EQ(config::hole_fill_remote_read_threads_per_be, new_count);
+}
+
 } // namespace
 } // namespace doris::io