compute: answer an index peek from one budgeted scan

[antiguru]

Makes one object own both phases of a fast-path index peek. PeekScan holds the errs cursor, the oks cursor, the literal state, the accumulated rows, and the size accounting, and spends a single budget across the error trace and the ok walk. It performs no IO and never awaits, so the same scan runs wherever a driver puts it, and a driver that stops it between two cursor positions picks it up again without repeating work.

Before this, each phase was individually suspendable but nothing spanned the boundary, so no caller could hold a partially finished peek.

ScanOutcome::Suspended carries no payload. The scan keeps its accumulated prefix and a driver pulls it with take_batch, which yields a batch only once accumulation crosses the stash threshold. Nothing is handed by value to a driver with no way to dispose of it, so "committed rows are never dropped" is a property of the interface rather than a rule each driver keeps.

This is a refactor. Every peek answers exactly as it did, including the diversion to the peek stash, which still restarts the walk. Two full sqllogictest sweeps pass, one ordinary and one with the stash threshold at zero so that every streamable peek takes the diversion.

One behaviour does change, deliberately: a scan whose accumulation crosses the stash threshold now suspends rather than continuing to accumulate. The result-size ceiling does not gate a stash-bound prefix, so without it an inline driver would hold an entire result in memory where it previously stopped at the threshold.

🤖 Opened by Claude Code on behalf of @antiguru

Replaces #38477, which GitHub closed when the design document moved from the bottom of the stack to the top and its branch was force-pushed past these commits. Same content, new base.

[antiguru]

Review findings, posted by Claude Code on behalf of @antiguru. Line numbers are at c35f2c8f91.

No defect that breaks shipped behaviour. The state machine does not wedge, both terminal outcomes latch where the contract says they do, and the row and byte accounting across the phase boundary is correct. Two documentation-against-code contradictions do matter, because the next layer's driver is written against them.

1. The struct doc states a memory bound the type does not enforce

peek_scan.rs:113-116:

A scan retains at most peek_stash_threshold_bytes of accumulated rows, plus the row that crossed that threshold [...] That is a property of the scan rather than of a driver.

batch_ready() at :320 is self.peek_stash_eligible && self.total_size > self.peek_stash_threshold_bytes. For a scan that is not stash-eligible it is hardwired false, step_ok_phase never breaks on a batch, and the prefix grows until max_result_size fails the peek. Eligibility is is_streamable, so any peek with an ORDER BY or a non-identity projection is in that class, and with no LIMIT thinning never fires either. The claim is therefore false for a large population of peeks, and the retention argument in the design document rests on it.

Either scope the claim to a stash-eligible scan and name max_result_size as the bound otherwise, or make the retention bound independent of eligibility.

2. total_size drifts from results on a ceiling failure

:416-430 adds the row's bytes to total_size, then the ceiling arm breaks before self.results.push((row, copies)). The field is documented at :133 as "The byte size of results"; after that break it is the size of results plus one row that is not in it. ScanOutcome::Failed's doc at :74 also says "Rows accumulated before it are not part of an answer" while the scan retains every one of them.

Unobservable today, because batch_ready is provably false at the failure point (the ceiling arm requires !batch_ready, and every other iteration starts with it false because the previous one broke on it), so take_batch() cannot hand out rows from a failed scan. That is a guard by coincidence rather than by construction. Zero total_size and clear results on the failure paths, the way take_results does elsewhere.

3. Three of four peek bounds freeze at construction, one is live, and the justification covers all four

step's doc at :230-233 argues that row_iteration_limit is re-read per step "because the limit bounds the peek and applies to a walk already under way". max_result_size, peek_stash_eligible and peek_stash_threshold_bytes bound the peek by the identical argument and are captured in new at :178-185. All four are dyncfg-derived and re-read per process_peeks iteration, and peek_stash_eligible is a conjunction with peek_stash_enabled, so it genuinely flips between calls.

Inert while this layer builds and drops one scan per call. The point of PeekScan is to outlive a call, and at that point an operator raising max_result_size to unwedge a failing peek will not reach a scan already in flight. Worth settling now: either all four live, or all four frozen with row_iteration_limit taken in new.

4. The ok cursor is now opened before the error walk runs

PeekScan::new at :189-201 opens the ok cursor and clones both map_filter_project and the literal_constraints Vec<Row> before a single error key is read. The old ordering walked errors first and built the iterator afterwards, so an error-answered peek skipped all of it. It now pays a cursor open, two clones and a literal sort, and pins ok batches for the duration of the error walk. Deliberate and documented at :126-128, and load-bearing for the diversion argument at compute_state.rs:1818-1825, but it is a real regression on the error path and it is not mentioned in the PR body.

5. The Suspended doc has the wedge condition inverted

:60-69 tells a driver that "cannot write batches" to consult PeekScan::batch_ready before stepping again. A driver that genuinely cannot write batches passes peek_stash_eligible: false, so batch_ready() is always false and it cannot wedge. The driver that wedges is one that can be handed a full batch and declines to take it. collect_finished_data is neither: it forwards eligibility through, never asks batch_ready, and escapes by diverting. So the only driver in the tree does not follow the rule its own doc lays down.

Related: batch_ready() has no caller outside peek_scan.rs, and take_batch()'s only production caller at compute_state.rs:1840 discards the result and exists to trip a soft-panic. Landing the API before its consumer is fine in a stack; pre-narrating the consumer's obligations at this length is what CLAUDE.md's "don't narrate abstractions that belong to a future change" is about.

6. Deleting IndexPeek::error_scan removes a cross-call latch without replacing the guard

ErrorScanState lived on IndexPeek, which survives across collect_finished_data calls; ErrorPhase lives on PeekScan, which is built and dropped inside one. The preceding layer added that latch so "a peek that must report an error can never be sent on to return rows instead".

It is unreachable today: seek_fulfillment returns NotReady before the driver, UsePeekStash swaps the entry for PendingPeek::Stash, and Ready retires the peek. So the new claim at :1776-1780 holds. But the new soft-panic covers a different failure, Suspended-before-clean, not called-twice. If the invariant is now "at most one call", assert that.

7. A metric help string describes slicing no shipped driver does

metrics.rs:240 and doc/user/data/metrics.yml:1056-1068: "Sums the worker time of the calls the walk was sliced into, so it excludes the gaps between them." The only driver passes fuel = usize::MAX at compute_state.rs:1802, so there is one call and no gaps. error_scan_seconds accumulates identically and its help was not updated, so the pair is inconsistent either way. This is a user-facing catalog surface.

Tests

Quality is high and worth saying so concretely: one_budget_spans_both_phases and its indexed sibling compare fuel spent sliced against unsliced, which is the only assertion that catches a phase re-walking positions it already visited; a_batch_ready_scan_does_not_grow_when_stepped_again checks prefix, size, cursor position and fuel across four resumptions; a_prefix_bound_for_the_stash_is_not_bound_by_the_result_size_ceiling reassembles the answer from taken batches under a ceiling below what the scan produces. No test asserts nothing, and every fixture constructs what its name claims.

Gaps:

• run_sliced at peek_scan/tests.rs:508-528 computes spent += fuel_per_step - fuel, which underflows if fuel_per_step is usize::MAX and the scan suspends. Safe only because every such call site returns on the first outcome. The next test that passes usize::MAX to a stash-eligible scan panics in a way that looks like a scan bug.
• Nothing pins finding 2's guard, that take_batch() returns None after Failed while results is non-empty.
• one_budget_spans_both_phases:169 hardcodes 0..100 while RESUMPTION_BOUND exists and is used by three other tests.
• take_batch_yields_nothing_without_the_stash:356-359 sets the threshold to 0 but leans on scan()'s default peek_stash_eligible: false for its whole premise; the thing under test is invisible in the test body.
• Both soft-panic branches in the driver are unreachable at unbounded fuel. Fine now, but the diversion-soundness one deserves a scan-level test the moment a bounded-fuel driver lands.

Conventions

Clean: usize::cast_from/u64::cast_from/f64::cast_lossy throughout with no as conversions, BTreeMap in tests, no side effects in debug_assert!, no em-dashes in added lines, no unused imports after the deletions. Same doc-style deviation as the layer below, reasoning in rustdoc rather than inline across ScanOutcome::Suspended, ErrorPhase, OkWalkEnd, take_results and latched_ok_outcome.

Checked and found correct, rather than assumed: add_rows_iterated is reachable exactly once so no row is double-counted or lost; ordered thinning across handed-off batches is sound, since a row dropped from the current prefix has max_results retained rows ranked ahead of it in that same prefix and so cannot be in the global top max_results; and the deleted accumulation loops leave nothing unaccounted, with num_rows_needed(), the 2 * max_results amortization, the checked_mul note, the unordered-truncate exit, the dropped_size subtraction and the stash-then-ceiling-then-thin ordering all preserved.

[antiguru]

Findings addressed, posted by Claude Code on behalf of @antiguru.

A failed scan keeps no rows. The ceiling arm added a row's bytes to total_size and broke before pushing the row, so a failed scan carried a size one row larger than what it held, and it retained rows that are part of no answer. Both failure paths go through one place that drops the rows and the size together. That makes it true by construction, rather than by a coincidence of where the ceiling can trip, that a driver is never handed the prefix of an answer that will never be given. The tests that read the retained rows to say where a failure landed read the position count instead.

Two doc claims corrected. The prefix is bounded by the stash threshold only for a peek that may use the stash, since a peek with an ORDER BY or a non-identity projection fills no batch and is bounded by max_result_size alone. And the driver that has to consult batch_ready before stepping again is one that declines a batch it is offered, not one that cannot write batches at all, which is never offered one.

Bounds freshness is marked, not settled. Three of the four bounds freeze at open and the row-iteration limit is re-read per step, on an argument that covers all four. Inert while a scan lives no longer than one driver call, so it carries a TODO rather than a signature change through three layers. Worth a decision before a scan outlives a call.

The scan type has a name. IndexPeekScan lands here with the Send assertion that uses it, per the review comment on #38509.

Correction to my own earlier report on this PR: I claimed run_sliced underflows at fuel_per_step = usize::MAX. It cannot. fuel is only ever decremented, so fuel_per_step - fuel is never negative. No change made and none needed.

Test-gap items closed here: nothing pinned that take_batch() returns None after a failure while rows were held, which the new assertions cover; one_budget_spans_both_phases used a hardcoded bound where RESUMPTION_BOUND exists; and take_batch_yields_nothing_without_the_stash leaned on a default its body never showed.

[def-]

QA LLM Review

1. MEDIUM -- max_result_size is now applied with two different rulers on the two fast-path peek routes

src/compute/src/compute_state/peek_scan.rs:51

Details

An index peek is now charged data_len + 16 per row, while the persist fast-path peek in the same crate still charges Row::byte_len() + 8, i.e. a flat 24 bytes for the Row struct on top. Both routes used the identical expression before this diff, so the effective max_result_size for one query now depends on whether an index answers it: for a single narrow column the persist route rejects at about 56% of the size the index route accepts.

entry_byte_len is the right ruler. It matches what RowSetFinishing::finish (src/expr/src/relation.rs:3600) and the controller's aggregate check (src/compute-client/src/service.rs:600) measure, and the overcount left on the persist route is the same class as the finishing overcount whose regression test sits in test/sqllogictest/max_result_size.slt. The problem is that only one of the two producers was moved. Concretely, Datum::Int32(42) packs to 2 bytes: the index route charges 18, the persist route 32. A 100-character text row: 119 against 135.

The doc comment on entry_byte_len asserts this is "the ruler max_result_size is applied with wherever a result is measured against it", which src/compute/src/compute_state.rs:1673 in the same file contradicts, so a reader who trusts it will not go looking. Either use entry_byte_len in the persist collector too, or narrow the claim to the index route.

Worth weighing while unifying: this one number also gates peek_stash_threshold_bytes, and for a peek that is not stash-eligible (is_streamable is false for any ORDER BY or non-identity projection) it is the only bound on the Vec<(Row, NonZeroI64)> the scan retains. That vector costs 32 bytes per entry plus spilled data whatever the ruler says, so measuring the answer rather than the retention raises the prefix a narrow-row peek can hold at the ceiling by roughly 1.8x. Sound as a result-size limit, weaker than before as a memory backstop.

[antiguru]

Confirmed and fixed in 6e6c045546, carried through f49c9fe28c, 120ff0e651 and 86439fecf1.

The divergence is real and it is this diff's doing. At the base both routes charged row.byte_len() + size_of::<NonZeroUsize>(); this layer moved the index route to entry_byte_len and left the persist collector behind.

One detail worth stating, because it makes the persist ruler worse than a constant overcount. Row::byte_len is size_of::<Row>() + heap_size, and heap_size is zero while the row's data is inline. So the persist route charged a flat struct size for every inline row, counting no data bytes at all, and struct size plus data for every spilled one. It was measuring the container rather than the answer, which is why the two example rows land where they do.

entry_byte_len is the right ruler, and it is now the only one under src/compute/src/compute_state/. No byte_len() call remains there. It is exported pub(super) and the persist collector uses it, so the doc claim holds as written rather than needing to be narrowed.

Pinned with a test rather than left to the next reader:

let charged: usize = rows.iter().map(entry_byte_len).sum();
let collection = RowCollection::new(/* the same rows, each with count 1 */, &[]);
assert_eq!(charged, collection.byte_len());

Verified red before green: with entry_byte_len reverted to row.byte_len() + COUNT_BYTE_SIZE it fails 198 != 159 on three rows, a 24% overcount. max_result_size.slt is unchanged and still passes 32/32.

A golden test in that file would have been the natural home, but max_result_size refuses any value below 1 MiB, so distinguishing the two rulers there needs roughly 24000 rows of expected output. The unit test pins the invariant the divergence broke, which is that entry_byte_len is RowCollection::byte_len per entry.

On the second half, which I have deliberately not changed: for a peek that is not stash-eligible, max_result_size is now the only bound on the retained Vec<(Row, NonZeroI64)>, and the new ruler lets that vector grow further before tripping. That is a real weakening of the memory backstop. It reads to me as the backstop having worked by accident, because the ruler was miscalibrated in the conservative direction, and the honest fix is a retention bound measured in what retention actually costs rather than a deliberately wrong answer-size ruler. That is a new configuration parameter and does not belong in this PR. Flagging it rather than silently accepting it.

Posted by Claude Code on behalf of @moritz*.*

[petrosagg]

Approving, just a tiny nit

[petrosagg]

Let's store a single Option<ScanOutcome> field in PeekScan that latches the result.

[antiguru]

Done in 8183fa1a16. PeekScan now holds ended: Option<ScanOutcome>, set in step for either Finished outcome, and a stepped scan that has ended returns ended.clone() after the soft_panic_or_log. ok_walk_end and ended_outcome() are gone.

ErrorPhase lost its payload with them: it is Scanning(ErrorScan), Clean or Failed. Failed stays a distinct variant rather than folding into the latch because it also drops the error cursor, so a peek its error trace answered stops pinning error batches, and error_trace_clean() still has to tell a clean phase from a failed one when the ok walk later errs.

🤖 Posted by Claude Code on behalf of @antiguru

[linear-code]

CPU-195