From 6fc940871a7a0c79e771986d9164139306303854 Mon Sep 17 00:00:00 2001 From: Aljoscha Krettek Date: Tue, 4 Aug 2026 08:41:14 +0000 Subject: [PATCH] compute: thin peek results by partitioning, not sorting 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`. --- doc/user/data/metrics.yml | 12 ++++---- src/compute/src/compute_state.rs | 7 +++-- src/compute/src/compute_state/peek_scan.rs | 34 ++++++++++++++++++---- src/compute/src/metrics.rs | 16 +++++----- 4 files changed, 47 insertions(+), 22 deletions(-) diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index f1055d71d6b75..c3bc557d5321c 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -1032,18 +1032,18 @@ metrics: help: Time checking trace frontiers. source: src/compute/src/metrics.rs visibility: internal -- name: mz_index_peek_result_sort_seconds_bucket - help: Time sorting intermediate peek results down to the rows the finishing can need, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing. +- name: mz_index_peek_result_thinning_seconds_bucket + help: Time thinning intermediate peek results down to the rows the finishing can need, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing. labels: - le source: src/compute/src/metrics.rs visibility: internal -- name: mz_index_peek_result_sort_seconds_count - help: Time sorting intermediate peek results down to the rows the finishing can need, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing. +- name: mz_index_peek_result_thinning_seconds_count + help: Time thinning intermediate peek results down to the rows the finishing can need, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing. source: src/compute/src/metrics.rs visibility: internal -- name: mz_index_peek_result_sort_seconds_sum - help: Time sorting intermediate peek results down to the rows the finishing can need, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing. +- name: mz_index_peek_result_thinning_seconds_sum + help: Time thinning intermediate peek results down to the rows the finishing can need, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing. source: src/compute/src/metrics.rs visibility: internal - name: mz_index_peek_row_collection_seconds_bucket diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index abab4271cc58f..91af88260a0b3 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -1077,7 +1077,10 @@ impl<'a> ActiveComputeState<'a> { .compute_state .metrics .index_peek_row_iteration_seconds, - result_sort_seconds: &self.compute_state.metrics.index_peek_result_sort_seconds, + result_thinning_seconds: &self + .compute_state + .metrics + .index_peek_result_thinning_seconds, row_collection_seconds: &self .compute_state .metrics @@ -1679,7 +1682,7 @@ pub(crate) struct IndexPeekMetrics<'a> { pub error_scan_seconds: &'a prometheus::Histogram, pub cursor_setup_seconds: &'a prometheus::Histogram, pub row_iteration_seconds: &'a prometheus::Histogram, - pub result_sort_seconds: &'a prometheus::Histogram, + pub result_thinning_seconds: &'a prometheus::Histogram, pub row_collection_seconds: &'a prometheus::Histogram, } diff --git a/src/compute/src/compute_state/peek_scan.rs b/src/compute/src/compute_state/peek_scan.rs index 477c333da29ea..0107902726ade 100644 --- a/src/compute/src/compute_state/peek_scan.rs +++ b/src/compute/src/compute_state/peek_scan.rs @@ -160,7 +160,7 @@ impl PeekScan { .row_iteration_seconds .observe(self.row_iteration_time.as_secs_f64()); metrics - .result_sort_seconds + .result_thinning_seconds .observe(self.thinning_time.as_secs_f64()); match stop { @@ -262,17 +262,39 @@ impl PeekScan { return Some(Stop::Complete); } - // Sorting and truncating has an effect similar to a priority queue, - // without its interactive dequeueing properties. + // Partition rather than sort: we only need to know which rows fall + // outside the first `max_results`, not the order among those that stay. + // The final ordering is established once, when the results are + // collected. Sorting here would cost a log factor per row for an order + // we then throw away. + // + // Partitioning is not stable, so when rows tie across the cut it is + // unspecified which of them survives, and since entries carry a count + // the retained multiset genuinely differs between choices. What the + // client sees does not, for three reasons together: + // + // - A tie under this comparator means the rows are byte-identical, + // because the tiebreaker compares the whole encoded row. + // - We keep exactly `max_results` entries, each with a count of at + // least one, so the retained run expands to at least `max_results` + // rows and its first `max_results` are the same either way. + // - `max_results` is `limit + offset`, and the finishing reads exactly + // `offset..offset + limit` of the merged result, whose prefix + // depends only on the runs' prefixes. + // + // NOTE: The second and third points are why a peek result must not be + // consumed without applying the finishing's limit. + // // TODO: Had we left these as `Vec` we would avoid the unpacking. // We should consider doing that, although it will require a re-pivot of // the code to branch on this inner test (as we prefer not to maintain // `Vec` in the other case). let sort_start = Instant::now(); let comparator = &self.comparator; - self.results.sort_by(|left, right| { - comparator.compare_rows(&left.0, &right.0, || left.0.cmp(&right.0)) - }); + self.results + .select_nth_unstable_by(max_results, |left, right| { + comparator.compare_rows(&left.0, &right.0, || left.0.cmp(&right.0)) + }); self.thinning_time += sort_start.elapsed(); let dropped_size = self diff --git a/src/compute/src/metrics.rs b/src/compute/src/metrics.rs index bb17a6e5df2b1..14fa1b97b44f1 100644 --- a/src/compute/src/metrics.rs +++ b/src/compute/src/metrics.rs @@ -66,7 +66,7 @@ pub struct ComputeMetrics { index_peek_error_scan_seconds: Histogram, index_peek_cursor_setup_seconds: Histogram, index_peek_row_iteration_seconds: Histogram, - index_peek_result_sort_seconds: Histogram, + index_peek_result_thinning_seconds: Histogram, index_peek_frontier_check_seconds: Histogram, index_peek_row_collection_seconds: Histogram, @@ -209,9 +209,9 @@ impl ComputeMetrics { help: "Time iterating rows and evaluating MFP, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.", buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0), )), - index_peek_result_sort_seconds: registry.register(metric!( - name: "mz_index_peek_result_sort_seconds", - help: "Time sorting intermediate peek results down to the rows the finishing can need, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.", + index_peek_result_thinning_seconds: registry.register(metric!( + name: "mz_index_peek_result_thinning_seconds", + help: "Time thinning intermediate peek results down to the rows the finishing can need, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.", buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0), )), index_peek_frontier_check_seconds: registry.register(metric!( @@ -275,7 +275,7 @@ impl ComputeMetrics { let index_peek_error_scan_seconds = self.index_peek_error_scan_seconds.clone(); let index_peek_cursor_setup_seconds = self.index_peek_cursor_setup_seconds.clone(); let index_peek_row_iteration_seconds = self.index_peek_row_iteration_seconds.clone(); - let index_peek_result_sort_seconds = self.index_peek_result_sort_seconds.clone(); + let index_peek_result_thinning_seconds = self.index_peek_result_thinning_seconds.clone(); let index_peek_frontier_check_seconds = self.index_peek_frontier_check_seconds.clone(); let index_peek_row_collection_seconds = self.index_peek_row_collection_seconds.clone(); let replica_expiration_timestamp_seconds = self @@ -303,7 +303,7 @@ impl ComputeMetrics { index_peek_error_scan_seconds, index_peek_cursor_setup_seconds, index_peek_row_iteration_seconds, - index_peek_result_sort_seconds, + index_peek_result_thinning_seconds, index_peek_frontier_check_seconds, index_peek_row_collection_seconds, replica_expiration_timestamp_seconds, @@ -347,8 +347,8 @@ pub struct WorkerMetrics { pub(crate) index_peek_cursor_setup_seconds: Histogram, /// Histogram of index peek row iteration durations. pub(crate) index_peek_row_iteration_seconds: Histogram, - /// Histogram of index peek result sort durations. - pub(crate) index_peek_result_sort_seconds: Histogram, + /// Histogram of index peek result thinning durations. + pub(crate) index_peek_result_thinning_seconds: Histogram, /// Histogram of index peek frontier check durations. pub(crate) index_peek_frontier_check_seconds: Histogram, /// Histogram of index peek row collection construction durations.