feat(query): adaptively distribute lazy row fetch

[dantengsky]

I hereby agree to the terms of the CLA available at: https://docs.databend.com/dev/policies/cla/

Summary

Problem

Lazy materialization first scans the columns needed by filtering and sorting, keeps _row_id, and reads the deferred columns only after ORDER BY ... LIMIT has selected the final rows.

In a distributed top-N plan, scan workers send their partial results through a Merge exchange. The coordinator produces the final top-N row IDs and currently performs the entire RowFetch itself. This is efficient when the selected rows are concentrated in a few storage blocks, but the coordinator becomes a remote-I/O bottleneck when those rows are spread across many storage blocks. Always distributing RowFetch would remove that bottleneck but add unnecessary shuffle overhead when the result has a low block count.

This PR keeps RowFetch local when the selected rows touch few blocks and distributes it when they touch many blocks.

Scope

Adaptive routing is available only when the physical plan meets all three conditions:

1. Distributed top-N. Multiple executors produce partial top-N results, and the coordinator merges them before RowFetch. Without adaptive routing, the coordinator would fetch all deferred columns for the final result.
2. Deferred columns from one table. All RowFetch row IDs share one block-prefix domain, so routing by block prefix can send each block to one destination. Plans that fetch deferred columns from multiple tables keep the existing local RowFetch path.
3. A large enough limit. LIMIT > max_threads * 4. This planner heuristic avoids adding an exchange when the result is unlikely to contain enough blocks to benefit; it is not a correctness condition.

These are planner-time guards. For eligible plans, the runtime still counts the distinct blocks in the final top-N result and chooses between local and distributed RowFetch. If any guard fails, the physical plan remains unchanged.

For an eligible plan, the planner adds a RowFetch-specific exchange before RowFetch and a second Merge exchange after RowFetch:

scan workers: scan narrow columns + partial sort
                         |
                         v
coordinator: first Merge + final top-N
                         |
                         v
              adaptive RowFetch exchange
                  /       |       \
                 v        v        v
        destination nodes: fetch assigned blocks
                  \       |       /
                         v
coordinator: second Merge + restore final order

Runtime decision

The coordinator first coalesces all top-N input batches and counts distinct block prefixes in _row_id. Coalescing makes the decision against the complete result; deciding per batch could classify every small batch as local even when the combined result covers many blocks.

The routing rule is:

local_block_threshold = max(max_threads * 8, 128)

distinct blocks <= local_block_threshold  -> local RowFetch
distinct blocks >  local_block_threshold  -> distributed RowFetch

Block count is used instead of row count because remote reads and metadata lookup are organized by block. The threshold allows a bounded number of local I/O waves before a cluster shuffle becomes worthwhile.

Node behavior

Coordinator before RowFetch

• Consumes the first Merge exchange and produces the final top-N row IDs.
• Coalesces the result batches and makes one routing decision.
• In local mode, routes every row ID back to itself.
• In distributed mode, hashes row IDs by block prefix across all RowFetch destinations, including itself.

RowFetch destination nodes

• In local mode, workers receive no rows and perform no RowFetch work.
• In distributed mode, each worker and the coordinator receive only their assigned block prefixes.
• Rows from the same block always go to the same destination. The router does not split or rebalance a block away from its hash destination, preserving block affinity and avoiding duplicate block reads.
• Each active destination prefetches metadata for its assigned blocks in batches and reads the deferred columns locally.
• Completed rows are sent to the coordinator through the second Merge exchange.

Coordinator after RowFetch

• Collects completed rows from all active destinations.
• Restores the final ordering when the original plan contains ORDER BY.
• Returns the same top-N result as the local path.

Fragment scheduling

The fragment that consumes the first Merge exchange must run only on the coordinator. However, the following node-to-node RowFetch exchange needs the same fragment ID registered on every executor so that all destinations can construct their exchange channels. The scheduler therefore sends the real merge-dependent plan to the coordinator and an empty source stub to the other workers. The downstream fragment containing RowFetch runs on every destination node.

Observability

Profiles, metrics, and logs expose the selected mode, input batches, rows, distinct blocks, destination distribution, and block-affinity statistics.

Performance

Manual benchmark on a three-node object-storage deployment with disk cache disabled used two top-N workloads. Both returned 10K rows:

• 99 result blocks: lazy RowFetch read about 0.95 GiB from remote storage.
• 792 result blocks: lazy RowFetch read about 8.02 GiB from remote storage.

Full-column scan (no RowFetch) reads all projected columns during TableScan and does not execute the code changed by this PR; it is included as a control. Lazy RowFetch reads deferred columns after the limit and exercises the adaptive path.

Result block count	Read path	Before runs	After runs	Before median	After median	Change
99	Full-column scan (no RowFetch)	unavailable	unavailable	-	-	-
99	Lazy RowFetch	17.55 / 13.04 / 13.57s	14.26 / 14.00 / 14.07 / 13.88s	13.57s	14.04s	+3.4%
792	Full-column scan (no RowFetch)	9.06 / 7.14 / 7.20s	8.96 / 7.75 / 7.39s	7.20s	7.75s	+7.6%
792	Lazy RowFetch	19.34 / 19.50 / 19.32s	10.00 / 9.37 / 9.29s	19.34s	9.37s	-51.6% (2.06x)

Before is rc7 build 4065bd334b; After is rc7 build b5f73c7e51, which contains the same change set applied to main by this PR. The 99-block full-column scan could not be compared because both builds attempted to spill in a write-restricted test environment. The 99-block local RowFetch path changed by +3.4%, the 792-block full-column control changed by +7.6%, and the target 792-block distributed RowFetch path improved by 51.6%.

Tests

• Unit Test
• Logic Test
• Benchmark Test
• No Test - Explain why

Unit tests cover batch coalescing, the local decision, block-affinity routing, and skewed block distributions. Optimizer integration tests cover plan selection and merge-dependent fragment scheduling. Cluster sqllogic coverage checks that row IDs and fetched payloads remain paired after shuffle and merge.

Type of change

• Bug Fix (non-breaking change which fixes an issue)
• New Feature (non-breaking change which adds functionality)
• Breaking Change (fix or feature that could cause existing functionality not to work as expected)
• Documentation Update
• Refactoring
• Performance Improvement
• Other (please describe):

---

This change is 

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 2ffdf2e0e9

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[github-actions]

🤖 CI Job Analysis

Workflow: 29643421496

📊 Summary

• Total Jobs: 87
• Failed Jobs: 1
• Retryable: 0
• Code Issues: 1

❌ NO RETRY NEEDED

All failures appear to be code/test issues requiring manual fixes.

🔍 Job Details

• ❌ linux / test_private_tasks: Not retryable (Code/Test)

---

🤖 About

Automated analysis using job annotations to distinguish infrastructure issues (auto-retried) from code/test issues (manual fixes needed).