compute: thin peek results by partitioning, not sorting - #38041
Closed
aljoscha wants to merge 2 commits into
Closed
Conversation
aljoscha
force-pushed
the
aljoscha/peek-02-thinning
branch
from
August 4, 2026 12:14
6fc9408 to
659e036
Compare
aljoscha
force-pushed
the
aljoscha/peek-02-thinning
branch
from
August 4, 2026 18:04
659e036 to
b6fa502
Compare
Serving a ready index peek walked the whole arrangement in one go. For a large arrangement that pins the worker thread for the full duration of the scan, so dataflows aren't scheduled, commands aren't handled, and the peek can't even observe its own cancellation. A peek now scans in bounded slices. `PeekScan` owns the cursor, the rows collected so far, and the size accounting, and its `step` spends one budget before handing the worker back. The cursor owns the batches it reads rather than borrowing them from the trace, so a scan is self-contained and parking one between activations is safe. `PeekResultIterator::step` charges fuel per cursor position rather than per row returned. Counting rows would let a selective `map_filter_project` over a large arrangement run arbitrarily long without ever reaching a yield point. Fuel bounds how often we get to yield, not the length of any one slice, which the docs now say. Budgets nest. Each peek gets its own turn (`peek_yielding`), bounded by what all peeks together may spend in one activation (`peek_yielding_total`). Peeks that don't get a turn are served first on the next activation, so a long peek can't starve the ones behind it. A budget reports no allowance once spent, on the time bound as much as the work bound. Since a nested allowance is the minimum of the two, checking the deadline only when asked whether to yield would let every peek reached after the shared deadline still take a full turn, and the shared time bound would not cap the activation at all. A slice always advances the cursor at least once, even on a spent budget. A yielded peek keeps the worker from parking, so a slice that does no work at all would be a livelock rather than a slow peek. Making that a property of the loop means a budget of `work:0` degrades to a slow peek instead of hanging one. A yielded peek is work the worker owes itself, so `run_client` doesn't park while any peek has work left. That is also why `handle_peek` no longer serves the peek inline: `process_peeks` runs later in the same iteration, so latency is unchanged, and routing everything through there means a burst of peeks shares one budget instead of each getting its own. Peek timings now accumulate across activations and are reported once the peek is done. Reporting per activation would turn one slow peek into a string of fast ones. The cost is that a peek cancelled mid-scan reports nothing, which the help texts now say. That leaves nothing measuring how long peeks hold the worker per activation, which is the quantity the budgets bound, so `mz_peek_processing_seconds` times one pass over the pending peeks. `YieldSpec` moves out of the linear join into `crate::yielding` so both callers share one policy type and config format. The new `peek-count` script command in the clusterd test driver peeks an index directly and reports only the row count. `count` tallies through an ephemeral reduce dataflow and then peeks that dataflow's single-row output, so it cannot exercise a scan over many rows. It also takes optional literal constraints, which reach the other cursor path, the one that seeks from one literal to the next rather than stepping.
When a peek's finishing bounds how many rows it can need, the scan keeps twice that many and periodically drops the excess. It did so by sorting the whole buffer and truncating, which costs a log factor per row for an order that is thrown away: the result is ordered once at the end, when the `RowCollection` is built. Partitioning gives the same retained rows in O(n) comparisons. Each comparison decodes both rows, so this is the dominant cost of a peek with an `ORDER BY` and a small `LIMIT` over a large arrangement. The partition is not stable, so when rows tie across the cut it is unspecified which survives. That is unobservable: a tie under this comparator means the rows are byte-identical, so whichever way the cut lands the retained run agrees with any other choice on its first `limit + offset` rows, and that prefix is all the finishing reads. `mz_index_peek_result_sort_seconds` becomes `mz_index_peek_result_thinning_seconds`, since it no longer times a sort. The one remaining sort, when the `RowCollection` is built, is timed by `mz_index_peek_row_collection_seconds`.
aljoscha
force-pushed
the
aljoscha/peek-02-thinning
branch
from
August 4, 2026 18:33
b6fa502 to
d450e69
Compare
Contributor
Author
|
@antiguru we might still want to also take that list fix from here, about using partitioning and not sorting? |
Member
|
Taken, as MaterializeInc/materialize#38634, on top of the peek stack. Your reasoning carried over as written, since the argument is what makes the change safe rather than the two-line edit. It holds against what the layers below it added, which I checked rather than assumed. The unordered early stop and the peek stash both require an absent ordering, and thinning requires one, so neither ever consumes a partitioned prefix. Two things beyond the port. Two tests asserted an exact order over a thinned result and were passing on the partition's incidental output; they compare multisets now, since thinning no longer orders what it keeps. And the phase's fields become Posted by Claude Code on behalf of @moritz*.* |
antiguru
added a commit
that referenced
this pull request
Sep 3, 2026
Thinning only needs to know which rows fall outside the first `max_results`, not the order among those that stay. That order is established once anyway, when the answer is collected, so sorting here costs a log factor per row for an ordering that is then thrown away. Over a walk of N rows, thinning runs about N/max_results times on twice `max_results` rows each, which turns `O(N log max_results)` into `O(N)`. It is the walk these peeks spend their time in: a finishing that carries an ordering is never streamable, so such a peek never reaches the peek stash and accumulates until the walk ends. Partitioning is unstable, so when rows tie across the cut it is unspecified which of them survives, and entries carry counts, so the retained multiset does depend on that choice. What the client reads does not, for three reasons together. A tie here means the rows are byte-identical, because the tiebreaker compares the whole encoded row. Exactly `max_results` entries stay, each with a count of at least one, so they expand to at least `max_results` rows. And `max_results` is `limit + offset`, which the finishing reads as `offset..offset + limit` of the merged answer, whose prefix depends only on the prefixes of the runs it merges. The second and third are why a peek result must not be consumed without applying the finishing's limit. Thinning no longer orders what it keeps, so the two tests that asserted an exact order over a thinned result compare multisets instead. They passed on the partition's incidental output, which is not a property the implementation promises. The phase's fields become `thinning_time` and `rows_thinned`, since neither measures a sort any more. The two metrics keep their names, which are already released, and their help text now describes thinning rather than sorting. The reasoning here is Aljoscha's, from #38041, which is closed.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
Part 2 of 2 in a stack that makes index peeks stop monopolizing the compute
worker. Part 1 is #38040. A replica crash fix in the same code went first, in
#38039, and has landed.
When a peek's finishing bounds how many rows it can need, the scan keeps twice
that many and periodically drops the excess. It did so by sorting the whole
buffer and truncating. That costs a log factor per row for an order that is
thrown away: the result is ordered once at the end, when the
RowCollectionisbuilt. Each comparison decodes both rows, so for a peek with an
ORDER BYanda small
LIMITover a large arrangement this is the dominant cost.Part of CPU-195.
Description
select_nth_unstable_bygives the same retained rows in O(n) comparisonsinstead of O(n log k).
The partition is not stable, so when rows tie across the cut it is unspecified
which of them survives. That is unobservable, and the argument is worth stating
because it is the only thing making this a safe swap:
RowComparator::compare_rows(l, r, || l.cmp(r))means theorder_bycolumns compare equal and the tiebreaker compares equal, and thetiebreaker is a full comparison of the encoded rows. So tied rows are
byte-identical.
limit + offsetrows, whichever way the cut lands.RowSetFinishing::finishreads exactlyoffset..offset + limitof the mergedresult, and
merge_sorted's output prefix depends only on the runs' prefixes.So the client sees the same rows either way.
mz_index_peek_result_sort_secondsis renamed tomz_index_peek_result_thinning_seconds, since it no longer times a sort. Theone remaining sort, when the
RowCollectionis built, is timed bymz_index_peek_row_collection_seconds.Not changed on purpose. The unconditional sort in
RowCollection::new,which fires even when
order_byis empty, stays. It is load-bearing fordeterminism, not for correctness: per-worker peek responses are absorbed in
arrival order by a randomized
StreamMap, so sorting each run by encoded bytesplus a byte-order k-way merge in envd is what makes client-visible row order
reproducible independent of worker count and arrival order. Roughly 1100
multi-row, no-
ORDER BY,nosortsqllogictest goldens ride on it, andtest/sqllogictest/range.slt:359pins the length-first-then-bytes encodingorder exactly.
Verification
Covered by the existing suite: any change to which rows survive thinning, or to
their order, shows up in the sqllogictest goldens described above, since they
compare output positionally.