[feature](be) Add range read-ahead for cold queries

[bobhan1]

What problem does this PR solve?

Issue Number: None

Problem Summary:

Doris data pages are usually much smaller than a 1 MiB File Cache block. A cold query that misses File Cache currently expands each page read to block boundaries, which provides a fixed 1 MiB form of read-ahead but also creates large read amplification for sparse row IDs, point queries, TopN, and delayed materialization. At the same time, FileColumnIterator consumes data pages one at a time, so the relatively small scanner batch exposes many serial S3 dependencies even when the query will soon read adjacent pages from the same segment.

This PR adds an opt-in, query-owned range read-ahead path for cloud scans. It plans future pages by compressed bytes across scanner batches, coalesces nearby page intervals into bounded file ranges, reads those ranges concurrently, serves exact page slices to the existing page decoder, and writes back only ranges that the query actually consumes. Complete 1 MiB blocks enter the existing asynchronous File Cache write path directly; partial blocks are completed by a bounded background hole-fill pipeline before entering that same path.

The feature is disabled by default. Admission rejection, cancellation, speculative-read failure, checksum/decode failure, and cache-writeback pressure do not change query correctness: reads fall back to the existing path, while writeback is best effort.

Architecture

flowchart LR
    SCAN["Scanner batch row IDs"] --> ROLE["SegmentIterator<br/>classify eager and lazy columns"]
    ROLE --> WINDOW["ColumnReadAhead<br/>compressed-byte windows"]
    WINDOW --> PLAN["FileRangePlanner<br/>coalesce and optional block completion"]
    PLAN --> ASYNC["AsyncFileRangeReader<br/>bounded concurrent exact reads"]
    ASYNC --> BUFFER["Query-owned range buffers"]
    BUFFER --> PAGE["ReadAheadFileReader<br/>exact page slices"]
    PAGE --> DECODE["Existing PageIO and page decoder"]
    DECODE --> USED{"Was any page in the range consumed?"}
    USED -->|No| DROP["Release buffer"]
    USED -->|Yes| SPLIT["Split at 1 MiB File Cache boundaries"]
    SPLIT -->|Complete block| PHASE1["Existing AsyncCacheWriteManager"]
    SPLIT -->|Partial block| HOLE["PartialBlockWritebackManager"]
    HOLE --> GET["Dedicated S3 GET pool<br/>read block-local holes"]
    GET --> PHASE1

Loading

Read-ahead planning

Columns are divided by data dependency rather than by SQL expression kind:

• An eager column is required before the current filtering or expression stage and can be planned immediately.
• A lazy column depends on an earlier result and uses a smaller speculative window. All currently plannable eager and lazy physical columns are submitted together before eager decoding starts, allowing ranges from different columns to share one admission and scheduling step.
• Array and Map element row IDs are not known until their offset columns have been decoded. Their child pages are therefore planned immediately after the real element-space row IDs become available, using the lazy window. Struct children remain in the parent row-ID space and can be planned with the parent.

Each physical column keeps a high and low watermark measured in compressed page bytes. Pages required by the current batch always enter the plan, even if they exceed the high watermark. Additional pages extend across future scanner batches until the high watermark is reached. Once consumed, discarded, or failed pages reduce the remaining window to the low watermark, the planner refills it. This compensates for scanner batch boundaries without tying I/O depth to row count or data type.

sequenceDiagram
    participant Scanner
    participant Segment as SegmentIterator
    participant Window as ColumnReadAhead
    participant Reader as AsyncFileRangeReader
    participant Decoder as Page decoder

    Scanner->>Segment: Request the next row batch
    Segment->>Window: Refill every currently plannable physical column
    Note over Window: Eager target 8 MiB and refill at 4 MiB<br/>Lazy target 256 KiB and refill at 128 KiB
    Window->>Reader: Submit one segment-level range plan
    par Concurrent exact range reads
        Reader->>Reader: Read range1
    and
        Reader->>Reader: Read range2
    and
        Reader->>Reader: Read range3
    end
    Reader-->>Segment: Publish ready range buffers
    Segment->>Decoder: Decode eager pages
    Decoder-->>Segment: Produce filters and dependent row IDs
    Segment->>Window: Plan newly resolvable dependent child pages
    Segment->>Decoder: Decode selected lazy pages

Loading

flowchart LR
    FULL["Window near its high watermark"] --> USE["Pages are consumed, discarded, or failed"]
    USE --> CHECK{"Remaining compressed bytes<br/>at or below the low watermark?"}
    CHECK -->|No| KEEP["Keep the current plan"]
    CHECK -->|Yes| EXTEND["Append future pages toward<br/>the high watermark"]
    EXTEND --> SUBMIT["Submit newly formed ranges"]
    SUBMIT --> FULL

Loading

Range coalescing and foreground block completion

The coalescing shape follows the same left-to-right idea used by Velox: sort and de-duplicate page intervals, then add the next interval only when all three limits remain satisfied.

1. The gap to the next page is at most 64 KiB.
2. The resulting physical range is at most 2 MiB.
3. Physical bytes divided by the union of requested page bytes is at most 2.0.

After base ranges are formed, the planner evaluates File Cache blocks touched by their boundaries. A block with at least 50% requested-page coverage may be completed in the foreground when the final range still fits the 2 MiB limit. If both boundaries qualify but cannot both fit, the side requiring fewer additional bytes is selected first. This avoids sending a nearly complete block through a second background S3 GET: for object storage, reading a modest number of additional bytes in the existing request is normally cheaper than adding another request and scheduling round.

flowchart LR
    PAGES["Requested page intervals"] --> SORT["Sort and de-duplicate"]
    SORT --> NEXT{"Add the next interval?"}
    NEXT -->|"gap within 64 KiB<br/>range within 2 MiB<br/>amplification within 2.0"| MERGE["Extend current base range"]
    MERGE --> NEXT
    NEXT -->|Any limit exceeded| CLOSE["Close current base range"]
    CLOSE --> COVER{"Boundary block coverage<br/>at least 50 percent?"}
    COVER -->|Yes and final range fits 2 MiB| FILL["Extend to the block boundary"]
    COVER -->|No| EXACT["Keep the base range"]
    FILL --> READ["Submit exact asynchronous range read"]
    EXACT --> READ

Loading

AsyncFileRangeReader performs the final ranges through a NO_WRITE + UNALIGNED request context. CachedRemoteFileReader still reuses already downloaded cache blocks, but a cold miss reads only the exact missing spans and does not reserve or populate File Cache on the query thread.

Concurrency, ownership, and cancellation

AsyncFileRangeReader is a BE-level concurrent file-range executor. It deliberately does not understand pages, coalescing, or writeback policy. A submission atomically reserves all buffers and both query/BE budgets before any range is queued; a rejected submission leaves budgets unchanged and all pages return to the original synchronous path.

The default executor has 64 workers. Fixed safety guards allow at most 128 in-flight ranges and 256 MiB of resident buffers per query, and 1024 ranges and 1 GiB per BE. A range slot is released at terminal I/O state, while its byte reservation follows the range buffer lifetime so decoded pages and writeback cannot escape accounting. Query cancellation cancels queued work and marks running work for cancellation; BE shutdown rejects new submissions, cancels queued work, and waits for running reads before releasing shared resources.

Consumed-range writeback

Writeback metadata and the File Cache invalidation epoch are captured immediately before each asynchronous read submission. A successfully read range becomes eligible only after at least one of its pages is decoded. Purely speculative ranges are released without cache work.

Eligible ranges are split at File Cache block boundaries:

• A fragment covering the complete valid block, including a short physical EOF block, is copied into the existing fixed-block asynchronous cache write path.
• A partial fragment is copied into the BE-level PartialBlockWritebackManager queue.

The existing fixed-block asynchronous cache write path is the only component that creates and persists File Cache blocks. Query range buffers and background partial-block buffers remain separate from its queue and ownership model.

flowchart TD
    READY["Range read is ready"] --> USED{"At least one page was decoded?"}
    USED -->|No| RELEASE["Release query-owned buffer"]
    USED -->|Yes| CUT["Split consumed range at 1 MiB block boundaries"]
    CUT --> COMPLETE{"Fragment covers the complete valid block?"}
    COMPLETE -->|Yes| WRITER["Existing AsyncCacheWriteManager"]
    COMPLETE -->|No| PARTIAL["PartialBlockWritebackManager queue"]
    PARTIAL --> HOLES["Complete block-local holes"]
    HOLES --> WRITER
    WRITER --> CACHE["File Cache block"]

Loading

Background hole fill

The partial-block queue is bounded to 256 MiB per BE. Fragments for the same queued block are merged; a block that is already active is deduplicated. Under pressure, the oldest queued block is evicted. When all pending blocks are already active, the new fragment is rejected. These outcomes affect only cache population.

The scheduler dispatches a block only when the existing cache writer has spare capacity. If capacity is unavailable, the task remains queued and the scheduler waits; S3 GET workers are not occupied while waiting for cache-write capacity. The default dedicated hole-fill pool has 32 workers, so slow object-storage GETs do not block the queue scheduler or the foreground read-ahead executor.

For one partial block, the hole-fill algorithm is:

1. Merge all covered fragments inside that 1 MiB block.
2. Take the complement of the covered union to produce missing intervals.
3. Coalesce adjacent missing intervals with block-local limits: 32 KiB maximum gap, 1 MiB maximum GET, and 2.0 maximum read amplification.
4. Read the resulting ranges through the underlying remote reader without crossing the block boundary.
5. Submit the completed owned buffer to the existing cache writer with spare-capacity-only admission, avoiding another full-block copy.

flowchart TD
    ENQUEUE["Partial block fragment arrives"] --> STATE{"State for the same block"}
    STATE -->|Queued| MERGE["Merge covered intervals into the queued entry"]
    STATE -->|Active| DEDUP["Drop duplicate writeback work"]
    STATE -->|Absent| ADMIT{"Pending-byte budget available?"}
    ADMIT -->|Yes| QUEUE["Append block entry to the queue"]
    ADMIT -->|No and queued entries exist| EVICT["Evict the oldest queued entry"]
    EVICT --> QUEUE
    ADMIT -->|No and every entry is active| REJECT["Reject this cache-only work"]
    MERGE --> WAIT["Wait for existing cache-writer capacity"]
    QUEUE --> WAIT
    WAIT --> ACTIVE["Mark active and dispatch to the dedicated S3 pool"]
    ACTIVE --> COMPLEMENT["Take the complement of covered bytes"]
    COMPLEMENT --> COALESCE["Coalesce block-local holes"]
    COALESCE --> GET["Issue one or more block-local S3 GETs"]
    GET --> HANDOFF["Hand off the completed 1 MiB buffer"]
    HANDOFF --> WRITER["Existing cache writer"]

Loading

For example, if a block already contains [0, 256 KiB), [384 KiB, 416 KiB), and [544 KiB, 1024 KiB), its holes are [256 KiB, 384 KiB) and [416 KiB, 544 KiB). Their 32 KiB separation meets the gap and amplification limits, so they become one S3 GET [256 KiB, 544 KiB). The 32 KiB already present between the holes is overwritten with identical remote data, trading a small byte increase for one fewer object-storage request.

flowchart LR
    C1["Covered<br/>0 to 256 KiB"] --> H1["Hole 1<br/>256 to 384 KiB"]
    H1 --> C2["Covered gap<br/>384 to 416 KiB"]
    C2 --> H2["Hole 2<br/>416 to 544 KiB"]
    H2 --> C3["Covered<br/>544 to 1024 KiB"]
    H1 --> MERGED["Coalesce across the 32 KiB covered gap"]
    H2 --> MERGED
    MERGED --> GET["One S3 GET<br/>256 to 544 KiB"]
    GET --> BLOCK["Completed 1 MiB block"]

Loading

Failure and compatibility boundaries

• The path is enabled only for cloud READER_QUERY segment scans when enable_query_read_ahead=true; the default remains false.
• The legacy Segment File Cache prefetcher is skipped only when the new read-ahead context was created, so the two mechanisms do not issue duplicate reads.
• Generic file readers can use the concurrent range path without File Cache writeback. CachedRemoteFileReader adds the exact no-write read and consumed-range writeback integration.
• Planning/admission errors and asynchronous read failures mark affected pages for fallback. A checksum or decode error evicts the suspect Page Cache entry and retries through the original reader.
• File Cache epoch invalidation, queue pressure, S3 hole-fill failure, and fixed-block writer rejection only drop writeback work.
• Variant-column forwarding is outside this PR's current scope.

Configuration defaults

Configuration	Default	Purpose
enable_query_read_ahead	false	Enable the new cloud query read-ahead path.
read_ahead_io_workers_per_be	64	Concurrent foreground file-range workers per BE.
read_ahead_eager_high_watermark_bytes	8 MiB	Target compressed-page bytes for each eager physical column.
read_ahead_eager_low_watermark_bytes	4 MiB	Refill threshold for an eager column after pages are consumed or discarded.
read_ahead_lazy_high_watermark_bytes	256 KiB	Target compressed-page bytes for each dependency-delayed physical column.
read_ahead_lazy_low_watermark_bytes	128 KiB	Refill threshold for a lazy column.
read_ahead_max_gap_bytes	64 KiB	Maximum gap crossed while coalescing foreground page intervals.
read_ahead_max_range_bytes	2 MiB	Maximum base or block-completed foreground range.
read_ahead_max_read_amplification_ratio	2.0	Maximum foreground coalescing amplification.
read_ahead_block_fill_min_coverage	0.5	Minimum requested-page coverage before completing a block in the foreground.
hole_fill_max_gap_bytes	32 KiB	Maximum covered gap crossed while coalescing block-local holes.
hole_fill_max_range_bytes	1 MiB	Maximum hole-fill GET; reads never cross a File Cache block.
hole_fill_max_read_amplification_ratio	2.0	Maximum block-local hole-fill amplification.
hole_fill_max_pending_bytes_per_be	256 MiB	BE-level pending and active partial-block memory bound.
hole_fill_workers_per_be	32	Dedicated background S3 GET workers.

Implementation structure

The 21 commits are intentionally split by component: generic range coalescing, cache-aware planning, concurrent range execution, exact cache reads, column windows, segment coordination, query ownership, physical/nested-column integration, fixed-block handoff, consumed-range dispatch, block-local hole planning, background writeback, BE lifecycle wiring, production writeback routing, and option-validation ownership. Each core algorithm and concurrency boundary has focused BE unit tests.

Release note

Add an opt-in range read-ahead and File Cache writeback pipeline for cold cloud queries. The feature is disabled by default.

Check List (For Author)

• Test

  • Regression test
  • Unit Test
    • ./run-be-ut.sh -j100 --run '--filter=FileRangeCoalescerTest.*:FileRangePlannerTest.*:AsyncFileRangeReaderTest.*:ColumnReadAheadTest.*:SegmentReadAheadTest.*:SegmentIteratorReadAheadTest.*:RangeWritebackDispatcherTest.*:HoleFillPlannerTest.*:RangeCacheWritebackTest.*:PartialBlockWritebackOptionsTest.*:PartialBlockWritebackManagerTest.*:QueryContextReadAheadTest.*:AsyncCachedRemoteFileReaderTest.*:AsyncCacheWriteManagerTest.*:ColumnReaderTest.*ReadAhead*'
    • Result: 137 tests from 15 suites passed under ASAN.
    • build-support/check-format.sh
    • git diff --check upstream/master...HEAD
  • Manual test
  • No need to test or manual test.

• Behavior changed:

  • No.
  • Yes. When explicitly enabled, eligible cloud query segment reads use bounded asynchronous range read-ahead and consumed-range File Cache writeback. The default behavior is unchanged because the feature gate is off.

• Does this need documentation?

  • No.
  • Yes. The Phase 2 design is complete; a public configuration/documentation PR is pending before this Draft becomes ready for review.

Draft validation still pending

• End-to-end cloud regression with the feature enabled.
• Cold S3 A/B measurements for query latency, GET count, remote bytes, memory high-watermark behavior, and File Cache population.
• Public documentation for the new BE configurations.

Check List (For Reviewer who merge this PR)

• Confirm the release note
• Confirm test cases
• Confirm document
• Add branch pick label

[hello-stephen]

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?