compute: make the peek stash a state transition of one scan

[antiguru]

Deletes the second walk a stashed peek used to cost. A peek whose accumulated rows crossed the stash threshold abandoned its walk, and a separate path re-read the trace bundle from the beginning to stream rows into persist. The promoted driver now writes each full batch as the walk produces it, so a peek walks its arrangement exactly once whatever its answer turns out to be. This is the layer that makes the one-path claim in the design document true.

The upload becomes a handle the driver feeds. StashUpload keeps everything about the persist interaction, the shard derived from the peek uuid, the schemas, the batch builder, the max_rows early exit, and takes rows through push instead of through a channel the worker pumps. Complete carries whatever the scan still holds, and the driver assembles the answer around it: a driver that never opened an upload answers with a row collection, and one that did finishes and answers with the stashed handle, whose inline_rows carry the rows that never reached the stash. The hand-back the layer below needed goes away, because the task now has somewhere to write.

PendingPeek ends with fewer index-peek states than it had before this work started, which is the trade the design says the effort is making. PendingPeek::Stash, StashingPeek and its worker-driven pump, start_stash_upload, and PEEK_STASH_NUM_BATCHES are all gone. PEEK_STASH_BATCH_SIZE goes too, against the plan: it counted rows, the upload cuts a batch on the byte threshold the scan already tracks, and nothing read it. A live tunable that does nothing is worse than an absent one, because an operator who reaches for it gets silence rather than an error.

Two consequences worth stating plainly rather than discovering later.

The kill switch no longer means no peek ever leaves the worker. With UsePeekStash gone, a large streamable peek would otherwise have no route to the stash at all, which is a functional regression rather than a placement change. The switch therefore gates budget-based promotion only: a scan that suspends because its prefix is batch-ready is handed to a task whichever way the switch is set, because that promotion is for stashing rather than for latency. With the switch off, ordinary peeks behave exactly as they do today.

The row iteration limit now follows a peek into the peek stash. #38158 stopped at the stash because a stashed peek restarted its scan and the restart charged the same rows twice. Deleting the restart is what this change does, so the count simply continues because the scan does. The dyncfg description said otherwise and no longer does.

Cancellation deletes what it wrote. A cancelled peek used to leave the parts already written in blob storage, since impl Drop for Batch only logs the dangling keys and the reader-side delete runs after a successful read that a cancelled peek never reaches. Cancellation aborts the promoted task rather than signalling it, so an await placed after the cancellation check would never run, and the obligation lives on the upload itself: dropping one spawns a Batch::delete() onto a runtime handle captured when it opened. A guard covers the window inside finish, where persist has taken the builder and no upload holds the parts any more.

Reclaiming those blobs must not cost more than it saves. A builder whose part write was in flight when its walk was aborted holds a write persist has already marked as waited on, and finishing it panics rather than returning; this replica aborts the process on any uncaught panic, so reclaiming one query's blob storage could take the replica with it. The panic is caught and the shard logged. Failing to reclaim is an outcome this path already tolerates for a replica that dies mid-upload, which makes a leak the right answer there and an abort the wrong one. A Linear issue tracks the persist-side fix that would remove the need for the catch.

mz_index_peek_stashed_total counts the walks the stash answered. It is incremented beside the offloaded substrate counter, so it is a strict subset by construction, and every inline-driver test asserts it stays at zero. mz_stashed_peek_seconds goes: its only observer was the deleted walk, and a registered histogram that never observes reports a flat zero, which reads worse in a graph than an absent series. The duration is deliberately not reconstructed, because one loop now walks and writes, so the write is not separable from the walk.

What an upload can delete is not always everything it wrote. Persist merges runs once a builder holds more than peek_response_stash_batch_max_runs of them, and a merge writes a fresh output and drops the inputs it read without entering them into shard state. Parts flush at persist_blob_target_size, so an upload small enough never to merge deletes all of what it wrote and a large one deletes the output of its last merge. A reader deleting a response it has finished with reaches exactly the same parts, so this is a property of the builder rather than of abandonment, and it is unchanged by this PR.

Nothing bounds the total size of a stashed answer, which is also true today. max_result_size bounds the prefix a single scan retains between batches and never the sum across them. The only ceiling is the reader's own budget.

Known gaps, none of which this PR closes:

• No sqllogictest or testdrive exercises a real stashed fast-path peek through the adapter reader. inline_rows was always empty from a worker on the old path and is now non-empty up to the threshold, and the adapter merges it across workers, so that merge has no coverage in CI.
• Nothing bounds how many abandoned uploads hold a part at once. Each can hold up to persist_blob_target_size, and the abandon task outlives the walk's permit.
• The caught persist panic is verified by construction rather than by a test, because the panic handler that turns it into an abort is not installed under cargo test.

🤖 Opened by Claude Code on behalf of @antiguru