[feature](be) Split consumed ranges for cache writeback

### What problem does this PR solve?

Issue Number: None

Related PR: None

Problem Summary: Query read-ahead produces variable file ranges, while the existing asynchronous File Cache writer accepts complete fixed-size blocks. Add a synchronous ownership-transfer boundary that splits an already consumed range at cache-block boundaries, routes complete blocks and partial fragments to separate consumers, and handles the short physical EOF block as complete. The dispatcher reports generated and accepted fragments without retaining the query range buffer.

### Release note

None

### Check List (For Author)

- Test: Unit Test
    - ./run-be-ut.sh -j100 --run '--filter=RangeWritebackDispatcherTest.*'
- Behavior changed: No
- Does this need documentation: No

be/src/io/cache/range_writeback_dispatcher.cpp

@@ -0,0 +1,79 @@
+// 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 "io/cache/range_writeback_dispatcher.h"
+
+#include <algorithm>
+#include <utility>
+
+#include "common/logging.h"
+
+namespace doris::io {
+
+RangeWritebackDispatcher::RangeWritebackDispatcher(
+        size_t file_size, size_t block_size, FileCacheBlockFragmentConsumer complete_block_consumer,
+        FileCacheBlockFragmentConsumer partial_block_consumer)
+        : _file_size(file_size),
+          _block_size(block_size),
+          _complete_block_consumer(std::move(complete_block_consumer)),
+          _partial_block_consumer(std::move(partial_block_consumer)) {
+    DORIS_CHECK(_file_size > 0);
+    DORIS_CHECK(_block_size > 0);
+    DORIS_CHECK(_complete_block_consumer);
+    DORIS_CHECK(_partial_block_consumer);
+}
+
+RangeWritebackDispatchResult RangeWritebackDispatcher::dispatch(const FileRange& range,
+                                                                Slice data) const {
+    DORIS_CHECK(range.size > 0);
+    DORIS_CHECK(range.offset < _file_size);
+    DORIS_CHECK(range.size <= _file_size - range.offset);
+    DORIS_CHECK(data.data != nullptr);
+    DORIS_CHECK(data.size == range.size);
+
+    RangeWritebackDispatchResult result;
+    size_t file_offset = range.offset;
+    while (file_offset < range.end()) {
+        const size_t block_offset = file_offset / _block_size * _block_size;
+        const size_t block_valid_size = std::min(_block_size, _file_size - block_offset);
+        const size_t fragment_end = std::min(range.end(), block_offset + block_valid_size);
+        const size_t fragment_size = fragment_end - file_offset;
+        FileCacheBlockFragment fragment {
+                .block_offset = block_offset,
+                .block_valid_size = block_valid_size,
+                .fragment_offset = file_offset - block_offset,
+                .data = Slice(data.data + file_offset - range.offset, fragment_size),
+        };
+        if (fragment.complete()) {
+            ++result.complete_block_count;
+            result.complete_block_bytes += fragment.data.size;
+            if (_complete_block_consumer(fragment)) {
+                ++result.submitted_complete_block_count;
+            }
+        } else {
+            ++result.partial_fragment_count;
+            result.partial_fragment_bytes += fragment.data.size;
+            if (_partial_block_consumer(fragment)) {
+                ++result.submitted_partial_fragment_count;
+            }
+        }
+        file_offset = fragment_end;
+    }
+    return result;
+}
+
+} // namespace doris::io

be/src/io/cache/range_writeback_dispatcher.h

@@ -0,0 +1,72 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <functional>
+
+#include "io/fs/file_range_coalescer.h"
+#include "util/slice.h"
+
+namespace doris::io {
+
+/// One contiguous fragment of a File Cache block. `block_valid_size` is shorter than the fixed
+/// block size only at physical EOF. The data slice remains valid only during the consumer call.
+struct FileCacheBlockFragment {
+    size_t block_offset {0};
+    size_t block_valid_size {0};
+    size_t fragment_offset {0};
+    Slice data;
+
+    bool complete() const { return fragment_offset == 0 && data.size == block_valid_size; }
+};
+
+using FileCacheBlockFragmentConsumer = std::function<bool(const FileCacheBlockFragment& fragment)>;
+
+struct RangeWritebackDispatchResult {
+    /// Fragments classified by whether they cover all valid bytes of their cache block.
+    size_t complete_block_count {0};
+    size_t partial_fragment_count {0};
+    size_t complete_block_bytes {0};
+    size_t partial_fragment_bytes {0};
+    /// Fragments whose selected consumer returned true.
+    size_t submitted_complete_block_count {0};
+    size_t submitted_partial_fragment_count {0};
+};
+
+/// Splits an already consumed query range at File Cache block boundaries. Consumers run
+/// synchronously and must copy or otherwise acquire ownership of the fragment before returning.
+class RangeWritebackDispatcher {
+public:
+    /// Both consumers are required and run inline in dispatch().
+    RangeWritebackDispatcher(size_t file_size, size_t block_size,
+                             FileCacheBlockFragmentConsumer complete_block_consumer,
+                             FileCacheBlockFragmentConsumer partial_block_consumer);
+
+    /// `range` must be non-empty and contained by the file, while `data` must cover it exactly.
+    /// The result counts every classified fragment and each consumer's accepted submissions.
+    RangeWritebackDispatchResult dispatch(const FileRange& range, Slice data) const;
+
+private:
+    const size_t _file_size;
+    const size_t _block_size;
+    const FileCacheBlockFragmentConsumer _complete_block_consumer;
+    const FileCacheBlockFragmentConsumer _partial_block_consumer;
+};
+
+} // namespace doris::io

be/test/io/cache/range_writeback_dispatcher_test.cpp

@@ -0,0 +1,132 @@
+// 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 "io/cache/range_writeback_dispatcher.h"
+
+#include <gtest/gtest.h>
+
+#include <string>
+#include <utility>
+#include <vector>
+
+namespace doris::io {
+namespace {
+
+struct OwnedFragment {
+    size_t block_offset {0};
+    size_t block_valid_size {0};
+    size_t fragment_offset {0};
+    std::string data;
+
+    bool operator==(const OwnedFragment&) const = default;
+};
+
+OwnedFragment copy_fragment(const FileCacheBlockFragment& fragment) {
+    return {.block_offset = fragment.block_offset,
+            .block_valid_size = fragment.block_valid_size,
+            .fragment_offset = fragment.fragment_offset,
+            .data = std::string(fragment.data.data, fragment.data.size)};
+}
+
+TEST(RangeWritebackDispatcherTest, SplitsMixedRangeAtBlockBoundaries) {
+    std::vector<OwnedFragment> complete;
+    std::vector<OwnedFragment> partial;
+    RangeWritebackDispatcher dispatcher(
+            4096, 1024,
+            [&](const FileCacheBlockFragment& fragment) {
+                complete.push_back(copy_fragment(fragment));
+                return true;
+            },
+            [&](const FileCacheBlockFragment& fragment) {
+                partial.push_back(copy_fragment(fragment));
+                return false;
+            });
+    const std::string data(2048, 'r');
+
+    const auto result = dispatcher.dispatch({.offset = 256, .size = data.size()}, Slice(data));
+
+    EXPECT_EQ(complete, (std::vector<OwnedFragment> {{.block_offset = 1024,
+                                                      .block_valid_size = 1024,
+                                                      .fragment_offset = 0,
+                                                      .data = std::string(1024, 'r')}}));
+    EXPECT_EQ(partial, (std::vector<OwnedFragment> {{.block_offset = 0,
+                                                     .block_valid_size = 1024,
+                                                     .fragment_offset = 256,
+                                                     .data = std::string(768, 'r')},
+                                                    {.block_offset = 2048,
+                                                     .block_valid_size = 1024,
+                                                     .fragment_offset = 0,
+                                                     .data = std::string(256, 'r')}}));
+    EXPECT_EQ(result.complete_block_count, 1);
+    EXPECT_EQ(result.partial_fragment_count, 2);
+    EXPECT_EQ(result.complete_block_bytes, 1024);
+    EXPECT_EQ(result.partial_fragment_bytes, 1024);
+    EXPECT_EQ(result.submitted_complete_block_count, 1);
+    EXPECT_EQ(result.submitted_partial_fragment_count, 0);
+}
+
+TEST(RangeWritebackDispatcherTest, TreatsPhysicalEofBlockAsComplete) {
+    std::vector<OwnedFragment> complete;
+    std::vector<OwnedFragment> partial;
+    RangeWritebackDispatcher dispatcher(
+            2500, 1024,
+            [&](const FileCacheBlockFragment& fragment) {
+                complete.push_back(copy_fragment(fragment));
+                return true;
+            },
+            [&](const FileCacheBlockFragment& fragment) {
+                partial.push_back(copy_fragment(fragment));
+                return true;
+            });
+    const std::string data(452, 'e');
+
+    const auto result = dispatcher.dispatch({.offset = 2048, .size = data.size()}, Slice(data));
+
+    EXPECT_EQ(complete, (std::vector<OwnedFragment> {{.block_offset = 2048,
+                                                      .block_valid_size = 452,
+                                                      .fragment_offset = 0,
+                                                      .data = data}}));
+    EXPECT_TRUE(partial.empty());
+    EXPECT_EQ(result.complete_block_count, 1);
+    EXPECT_EQ(result.complete_block_bytes, 452);
+    EXPECT_EQ(result.submitted_complete_block_count, 1);
+}
+
+TEST(RangeWritebackDispatcherTest, PreservesSourceOffsetsAcrossCompleteBlocks) {
+    std::vector<OwnedFragment> complete;
+    RangeWritebackDispatcher dispatcher(
+            3072, 1024,
+            [&](const FileCacheBlockFragment& fragment) {
+                complete.push_back(copy_fragment(fragment));
+                return true;
+            },
+            [](const FileCacheBlockFragment&) { return false; });
+    const std::string data =
+            std::string(1024, 'a') + std::string(1024, 'b') + std::string(1024, 'c');
+
+    const auto result = dispatcher.dispatch({.offset = 0, .size = data.size()}, Slice(data));
+
+    ASSERT_EQ(complete.size(), 3);
+    EXPECT_EQ(complete[0].data, std::string(1024, 'a'));
+    EXPECT_EQ(complete[1].data, std::string(1024, 'b'));
+    EXPECT_EQ(complete[2].data, std::string(1024, 'c'));
+    EXPECT_EQ(result.complete_block_count, 3);
+    EXPECT_EQ(result.complete_block_bytes, data.size());
+}
+
+} // namespace
+} // namespace doris::io