vortex-datafusion v2 scan should preserve ordered splits and enforce fetch across splits

[gatesn]

VortexDataSource::open consumes DataSourceScan::partitions() with try_flatten_unordered even when ScanRequest.ordered is true. It also forwards fetch as ScanRequest.limit without enforcing the limit across partition output.

A DataSource should be able to expose independent partitions in a deterministic order. When ordering is requested, v2 should emit their results in partition order. Otherwise, it should retain concurrent consumption. V2 should enforce fetch across all returned batches, slice the final batch if needed, and stop polling once the limit is met.

Please clarify the ScanRequest.limit contract so sources do not need their own cross-partition coordinator. Add a regression test with a delayed first partition that checks ordered output and a global fetch = 1.

[gatesn]

@claude take a stab at this one

[vortex-claude]

Claude finished @gatesn's task in 10m 54s —— View job • claude/issue-10056-20260925-1405 • Create PR ➔

---

Task: v2 scan ordered splits + global fetch enforcement

• Gather context (read source.rs, ScanRequest, scan partition APIs)
• Understand the ScanRequest.limit contract
• Preserve ordered splits when ordered is true; keep concurrency otherwise
• Enforce fetch across all returned batches (slice final batch, stop polling once met)
• Clarify ScanRequest.limit contract in docs
• Add regression test: delayed first partition, ordered output, global fetch=1
• Commit and push

Summary

Implemented in vortex-datafusion/src/v2/source.rs (VortexDataSource::open) and documented in vortex-scan/src/lib.rs.

Ordered vs. concurrent split consumption

open() now branches on ordered:

• ordered == true → try_flatten() (sequential), so split output is emitted in partition order.
• ordered == false → try_flatten_unordered(num_partitions * 2) (unchanged concurrent behavior).

Global fetch enforcement

The combined batch stream is wrapped in a stream::unfold that tracks emitted rows, slices the batch that crosses the limit (RecordBatch::slice), and — via a done flag — returns None without polling upstream again once the limit is met, so partition streams stop being pulled.

ScanRequest.limit contract clarification

Documented that limit is a per-partition upper bound, not a global one. The cross-partition consumer (here, VortexDataSource) owns enforcing the limit globally, so individual sources don't need their own cross-partition coordinator while still being able to prune I/O per partition.

Regression test

ordered_scan_enforces_global_fetch builds a mock 2-partition DataSource where partition 0 is delayed 100ms and partition 1 is immediate. With ordered = true and limit = 1, it asserts exactly one row is returned and its value comes from the first partition (0), not the faster second partition (100). Under the old unordered path this would have returned 100, so the test guards the fix.

Checks

⚠️ I could not run cargo fmt/clippy/nextest — cargo and git config invocations require interactive approval that isn't available in this environment, so verification was done by manual review only. Please run:

cargo +nightly-2026-09-10 fmt -p vortex-datafusion
cargo clippy -p vortex-datafusion --all-targets --all-features -- -D warnings
cargo nextest run -p vortex-datafusion

• Branch: claude/issue-10056-20260925-1405