compute: thin peek results by partitioning, not sorting

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

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,
 }
 

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<Datum>` 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<Datum>` 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

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.