[feature](be) Add query read-ahead runtime ownership

### What problem does this PR solve?

Issue Number: None

Related PR: None

Problem Summary: Concurrent range reads must be shared at BE scope while every scanner of one query shares cancellation and resident-buffer admission. Add a startup-owned FileRangeReadScheduler in Cloud mode, a lazily shared query context, cancellation propagation, and a mutable enable_query_read_ahead switch. Query and BE resident-buffer limits are configurable through read_ahead_max_bytes_per_query and read_ahead_max_bytes_per_be. The scheduler remains available while the switch is disabled so a runtime update can enable read-ahead for newly initialized reads without restarting the BE.

### Release note

Support runtime updates to enable_query_read_ahead and configurable query/BE read-ahead memory limits.

### Check List (For Author)

- Test: Unit Test
    - ./run-be-ut.sh --run --filter='FileRangeReadSchedulerTest.*:QueryContextReadAheadTest.*:SegmentReadAheadTest.*:ColumnReaderTest.*ReadAhead*' -j200
    - ./build.sh --be -j200
- Behavior changed: Yes. Cloud BEs can enable or disable query read-ahead at runtime; new reads observe the updated switch.
- Does this need documentation: Yes. The read-ahead design document lists these settings.

be/src/common/config.cpp

@@ -1908,6 +1908,14 @@ DEFINE_mInt32(segment_file_cache_consume_rowids_batch_size, "8000");
 DEFINE_mBool(enable_query_segment_file_cache_prefetch, "false");
 // Number of blocks to prefetch ahead in segment iterator for query
 DEFINE_mInt32(query_segment_file_cache_prefetch_block_size, "2");
+// Enable exact range read-ahead for query data pages
+DEFINE_mBool(enable_query_read_ahead, "false");
+// Query-level resident read-ahead buffer limit
+DEFINE_Int64(read_ahead_max_bytes_per_query, "268435456"); // 256 MiB
+DEFINE_Validator(read_ahead_max_bytes_per_query, [](int64_t value) { return value > 0; });
+// BE-level resident read-ahead buffer limit
+DEFINE_Int64(read_ahead_max_bytes_per_be, "1073741824"); // 1 GiB
+DEFINE_Validator(read_ahead_max_bytes_per_be, [](int64_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

@@ -1892,6 +1892,12 @@ DECLARE_mInt32(segment_file_cache_consume_rowids_batch_size);
 DECLARE_mBool(enable_query_segment_file_cache_prefetch);
 // Number of blocks to prefetch ahead in segment iterator for query
 DECLARE_mInt32(query_segment_file_cache_prefetch_block_size);
+// Enable exact range read-ahead for query data pages
+DECLARE_mBool(enable_query_read_ahead);
+// Query-level resident read-ahead buffer limit
+DECLARE_Int64(read_ahead_max_bytes_per_query);
+// BE-level resident read-ahead buffer limit
+DECLARE_Int64(read_ahead_max_bytes_per_be);
 // 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

@@ -54,6 +54,7 @@ class ThreadPool;
 struct StorePath;
 struct CachePath;
 namespace io {
+class FileRangeReadScheduler;
 class FDCache;
 class FileCacheFactory;
 class HdfsMgr;
@@ -267,6 +268,9 @@ public:
     ThreadPool* s3_file_system_thread_pool() { return _s3_file_system_thread_pool.get(); }
     ThreadPool* udf_close_workers_pool() { return _udf_close_workers_thread_pool.get(); }
     ThreadPool* segment_prefetch_thread_pool() { return _segment_prefetch_thread_pool.get(); }
+    io::FileRangeReadScheduler* file_range_read_scheduler() {
+        return _file_range_read_scheduler.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);
@@ -510,6 +514,7 @@ private:
     std::unique_ptr<ThreadPool> _udf_close_workers_thread_pool;
     // 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<ThreadPool> _peer_race_s3_thread_pool;
 
     FragmentMgr* _fragment_mgr = nullptr;

be/src/runtime/exec_env_init.cpp

@@ -62,6 +62,7 @@
 #include "io/cache/block_file_cache_factory.h"
 #include "io/cache/fs_file_cache_storage.h"
 #include "io/fs/file_meta_cache.h"
+#include "io/fs/file_range_read_scheduler.h"
 #include "io/fs/hdfs_file_writer.h"
 #include "io/fs/local_file_reader.h"
 #include "load/channel/load_channel_mgr.h"
@@ -308,6 +309,15 @@ Status ExecEnv::_init(const std::vector<StorePath>& store_paths,
                               .build(&_peer_race_s3_thread_pool));
     RETURN_IF_ERROR(init_mem_env());
 
+    if (config::is_cloud_mode()) {
+        io::FileRangeReadSchedulerOptions options {
+                .max_bytes_per_query = cast_set<size_t>(config::read_ahead_max_bytes_per_query),
+                .max_bytes_per_be = cast_set<size_t>(config::read_ahead_max_bytes_per_be),
+        };
+        RETURN_IF_ERROR(io::FileRangeReadScheduler::create(
+                options, _segment_prefetch_thread_pool.get(), &_file_range_read_scheduler));
+    }
+
     // NOTE: runtime query statistics mgr could be visited by query and daemon thread
     // so it should be created before all query begin and deleted after all query and daemon thread stoppped
     _runtime_query_statistics_mgr = new RuntimeQueryStatisticsMgr();
@@ -881,6 +891,7 @@ void ExecEnv::destroy() {
 
     SAFE_STOP(_external_scan_context_mgr);
     SAFE_STOP(_fragment_mgr);
+    SAFE_SHUTDOWN(_file_range_read_scheduler);
     SAFE_STOP(_runtime_filter_timer_queue);
     // NewLoadStreamMgr should be destoried before storage_engine & after fragment_mgr stopped.
     _load_stream_mgr.reset();
@@ -980,6 +991,7 @@ void ExecEnv::destroy() {
     _peer_race_s3_thread_pool.reset(nullptr);
     _send_table_stats_thread_pool.reset(nullptr);
     _buffered_reader_prefetch_thread_pool.reset(nullptr);
+    _file_range_read_scheduler.reset(nullptr);
     _segment_prefetch_thread_pool.reset(nullptr);
     _s3_file_upload_thread_pool.reset(nullptr);
     _send_batch_thread_pool.reset(nullptr);

be/src/runtime/query_context.cpp

@@ -40,6 +40,7 @@
 #include "exprs/function/ai/ai_request_executor.h"
 #include "io/cache/block_file_cache_factory.h"
 #include "io/cache/remote_scan_cache_write_limiter.h"
+#include "io/fs/file_range_read_scheduler.h"
 #include "runtime/exec_env.h"
 #include "runtime/fragment_mgr.h"
 #include "runtime/memory/heap_profiler.h"
@@ -235,6 +236,12 @@ void QueryContext::record_spill_data_dir(SpillDataDir* data_dir) {
 
 QueryContext::~QueryContext() {
     SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(query_mem_tracker());
+    {
+        std::lock_guard lock(_file_range_read_context_mutex);
+        if (_file_range_read_context != nullptr) {
+            _file_range_read_context->cancel();
+        }
+    }
     // query mem tracker consumption is equal to 0, it means that after QueryContext is created,
     // it is found that query already exists in _query_ctx_map, and query mem tracker is not used.
     // query mem tracker consumption is not equal to 0 after use, because there is memory consumed
@@ -310,6 +317,12 @@ void QueryContext::cancel(Status new_status) {
     if (!_exec_status.update(new_status)) {
         return;
     }
+    {
+        std::lock_guard lock(_file_range_read_context_mutex);
+        if (_file_range_read_context != nullptr) {
+            _file_range_read_context->cancel();
+        }
+    }
     // Tasks should be always runnable.
     _execution_dependency->set_always_ready();
     _memory_sufficient_dependency->set_always_ready();
@@ -361,6 +374,22 @@ void QueryContext::cancel(Status new_status) {
     }
 }
 
+std::shared_ptr<io::FileRangeReadContext> QueryContext::get_or_create_file_range_read_context(
+        io::FileRangeReadScheduler* scheduler) {
+    DORIS_CHECK(scheduler != nullptr);
+    std::lock_guard lock(_file_range_read_context_mutex);
+    if (_file_range_read_context == nullptr) {
+        _file_range_read_scheduler = scheduler;
+        _file_range_read_context = scheduler->create_context();
+        if (is_cancelled()) {
+            _file_range_read_context->cancel();
+        }
+    } else {
+        DORIS_CHECK(_file_range_read_scheduler == scheduler);
+    }
+    return _file_range_read_context;
+}
+
 void QueryContext::set_load_error_url(std::string error_url) {
     std::lock_guard<std::mutex> lock(_error_url_lock);
     _load_error_url = error_url;

be/src/runtime/query_context.h

@@ -48,6 +48,8 @@
 namespace doris {
 
 namespace io {
+class FileRangeReadContext;
+class FileRangeReadScheduler;
 class RemoteScanCacheWriteLimiter;
 } // namespace io
 
@@ -262,6 +264,11 @@ public:
         return _remote_scan_cache_write_limiter.get();
     }
 
+    /// Return the query-wide read-ahead context, creating it from `scheduler` on first use. Query
+    /// cancellation propagates to every range registered through the returned context.
+    std::shared_ptr<io::FileRangeReadContext> get_or_create_file_range_read_context(
+            io::FileRangeReadScheduler* scheduler);
+
     // plan node id -> TFileScanRangeParams
     // only for file scan node
     std::map<int, TFileScanRangeParams> file_scan_range_params_map;
@@ -423,6 +430,9 @@ private:
     std::mutex _cte_scan_lock;
     std::shared_ptr<MemShareArbitrator> _mem_arb = nullptr;
     std::unique_ptr<io::RemoteScanCacheWriteLimiter> _remote_scan_cache_write_limiter;
+    std::mutex _file_range_read_context_mutex;
+    io::FileRangeReadScheduler* _file_range_read_scheduler = nullptr;
+    std::shared_ptr<io::FileRangeReadContext> _file_range_read_context;
 
 public:
     // when fragment of pipeline is closed, it will register its profile to this map by using add_fragment_profile

be/test/runtime/query_context_read_ahead_test.cpp

@@ -0,0 +1,81 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <memory>
+
+#include "common/status.h"
+#include "io/fs/file_range_read_scheduler.h"
+#include "runtime/query_context.h"
+#include "testutil/mock/mock_query_context.h"
+#include "util/threadpool.h"
+
+namespace doris {
+namespace {
+
+ThreadPool* read_ahead_test_executor() {
+    static std::unique_ptr<ThreadPool> executor = []() {
+        std::unique_ptr<ThreadPool> result;
+        Status status = ThreadPoolBuilder("QueryReadAheadTest")
+                                .set_min_threads(1)
+                                .set_max_threads(4)
+                                .build(&result);
+        DORIS_CHECK(status.ok());
+        return result;
+    }();
+    return executor.get();
+}
+
+std::unique_ptr<io::FileRangeReadScheduler> create_range_scheduler() {
+    io::FileRangeReadSchedulerOptions options {
+            .max_bytes_per_query = 4096,
+            .max_bytes_per_be = 8192,
+    };
+    std::unique_ptr<io::FileRangeReadScheduler> scheduler;
+    EXPECT_TRUE(io::FileRangeReadScheduler::create(options, read_ahead_test_executor(), &scheduler)
+                        .ok());
+    return scheduler;
+}
+
+} // namespace
+
+TEST(QueryContextReadAheadTest, SharesAndCancelsRangeContext) {
+    auto scheduler = create_range_scheduler();
+    auto query = MockQueryContext::create();
+
+    auto first = query->get_or_create_file_range_read_context(scheduler.get());
+    auto second = query->get_or_create_file_range_read_context(scheduler.get());
+    ASSERT_NE(first, nullptr);
+    EXPECT_EQ(first, second);
+    EXPECT_FALSE(first->cancelled());
+
+    query->cancel(Status::Cancelled("cancel read-ahead test query"));
+    EXPECT_TRUE(first->cancelled());
+}
+
+TEST(QueryContextReadAheadTest, CreatesCancelledContextAfterQueryCancellation) {
+    auto scheduler = create_range_scheduler();
+    auto query = MockQueryContext::create();
+    query->cancel(Status::Cancelled("cancel before creating read-ahead context"));
+
+    auto range_context = query->get_or_create_file_range_read_context(scheduler.get());
+    ASSERT_NE(range_context, nullptr);
+    EXPECT_TRUE(range_context->cancelled());
+}
+
+} // namespace doris