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

doc/user/data/metrics.yml

@@ -1113,18 +1113,22 @@ metrics:
   help: Time in seek_fulfillment method including frontier checks and data collection.
   source: src/compute/src/metrics.rs
   visibility: internal
+- name: mz_index_peek_stashed_total
+  help: The number of index peek walks that answered with a handle to the peek response stash. Always a subset of the `offloaded` substrate of `mz_index_peek_walks_total`, because the driver that writes to the stash is the promoted one. How long such a walk spent writing is not separable from the walk itself, since one loop does both, and `mz_index_peek_offload_seconds` bounds it.
+  source: src/compute/src/metrics.rs
+  visibility: internal
 - name: mz_index_peek_total_seconds_bucket
-  help: 'Time one visit to an index peek spent on the timely worker. Observed per visit rather than per peek, so a peek revisited while the frontiers it reads are behind its timestamp contributes an observation per visit. A walk itself takes at most one of those visits: a suspension that is not the walk''s end either promotes it or diverts it to the peek stash. A peek whose walk was offloaded contributes only the inline slice that promoted it, which the inline budget bounds. Its time away from the worker is `mz_index_peek_offload_seconds`. Excluding peeks that use the peek response stash.'
+  help: 'Time one visit to an index peek spent on the timely worker. Observed per visit rather than per peek, so a peek revisited while the frontiers it reads are behind its timestamp contributes an observation per visit. A walk itself takes at most one of those visits: a suspension that is not the walk''s end promotes it. A peek whose walk was offloaded contributes only the inline slice that promoted it, which the inline budget bounds. Its time away from the worker is `mz_index_peek_offload_seconds`. A peek bound for the peek response stash is included, because the slice that promoted it ran here like any other.'
   labels:
   - le
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_total_seconds_count
-  help: 'Time one visit to an index peek spent on the timely worker. Observed per visit rather than per peek, so a peek revisited while the frontiers it reads are behind its timestamp contributes an observation per visit. A walk itself takes at most one of those visits: a suspension that is not the walk''s end either promotes it or diverts it to the peek stash. A peek whose walk was offloaded contributes only the inline slice that promoted it, which the inline budget bounds. Its time away from the worker is `mz_index_peek_offload_seconds`. Excluding peeks that use the peek response stash.'
+  help: 'Time one visit to an index peek spent on the timely worker. Observed per visit rather than per peek, so a peek revisited while the frontiers it reads are behind its timestamp contributes an observation per visit. A walk itself takes at most one of those visits: a suspension that is not the walk''s end promotes it. A peek whose walk was offloaded contributes only the inline slice that promoted it, which the inline budget bounds. Its time away from the worker is `mz_index_peek_offload_seconds`. A peek bound for the peek response stash is included, because the slice that promoted it ran here like any other.'
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_total_seconds_sum
-  help: 'Time one visit to an index peek spent on the timely worker. Observed per visit rather than per peek, so a peek revisited while the frontiers it reads are behind its timestamp contributes an observation per visit. A walk itself takes at most one of those visits: a suspension that is not the walk''s end either promotes it or diverts it to the peek stash. A peek whose walk was offloaded contributes only the inline slice that promoted it, which the inline budget bounds. Its time away from the worker is `mz_index_peek_offload_seconds`. Excluding peeks that use the peek response stash.'
+  help: 'Time one visit to an index peek spent on the timely worker. Observed per visit rather than per peek, so a peek revisited while the frontiers it reads are behind its timestamp contributes an observation per visit. A walk itself takes at most one of those visits: a suspension that is not the walk''s end promotes it. A peek whose walk was offloaded contributes only the inline slice that promoted it, which the inline budget bounds. Its time away from the worker is `mz_index_peek_offload_seconds`. A peek bound for the peek response stash is included, because the slice that promoted it ran here like any other.'
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_walks_total
@@ -3622,25 +3626,6 @@ metrics:
   - version
   source: src/environmentd/src/environmentd/main.rs
   visibility: internal
-- name: mz_stashed_peek_seconds_bucket
-  help: Time spent reading a peek result and stashing it in the peek result stash (aka. persist blob).
-  labels:
-  - le
-  - worker_id
-  source: src/compute/src/metrics.rs
-  visibility: internal
-- name: mz_stashed_peek_seconds_count
-  help: Time spent reading a peek result and stashing it in the peek result stash (aka. persist blob).
-  labels:
-  - worker_id
-  source: src/compute/src/metrics.rs
-  visibility: internal
-- name: mz_stashed_peek_seconds_sum
-  help: Time spent reading a peek result and stashing it in the peek result stash (aka. persist blob).
-  labels:
-  - worker_id
-  source: src/compute/src/metrics.rs
-  visibility: internal
 - name: mz_statement_logging_actual_bytes
   help: The total amount of SQL text that was logged by statement logging.
   source: src/adapter/src/metrics.rs

misc/python/materialize/mzcompose/__init__.py

@@ -776,8 +776,6 @@ def get_default_system_parameters(
     "compute_index_peek_permits",
     "compute_peek_response_stash_read_batch_size_bytes",
     "compute_peek_response_stash_read_memory_budget_bytes",
-    "compute_peek_stash_num_batches",
-    "compute_peek_stash_batch_size",
     "storage_statistics_retention_duration",
     "enable_paused_cluster_readhold_downgrade",
     "kafka_retry_backoff",

misc/python/materialize/parallel_workload/action.py

@@ -3352,8 +3352,6 @@ def __init__(
             "mz_metrics_lgalloc_map_refresh_interval",
             "mz_metrics_lgalloc_refresh_interval",
             "mz_metrics_rusage_refresh_interval",
-            "compute_peek_stash_num_batches",
-            "compute_peek_stash_batch_size",
             "compute_peek_response_stash_batch_max_runs",
             "compute_peek_response_stash_read_batch_size_bytes",
             "compute_peek_response_stash_read_memory_budget_bytes",

src/compute-types/src/dyncfgs.rs

@@ -600,23 +600,6 @@ pub const PEEK_RESPONSE_STASH_READ_MEMORY_BUDGET_BYTES: Config<usize> = Config::
     ParameterScope::Environment,
 );
 
-/// The number of batches to pump from the peek result iterator when stashing peek responses.
-pub const PEEK_STASH_NUM_BATCHES: Config<usize> = Config::new(
-    "compute_peek_stash_num_batches",
-    100,
-    "The number of batches to pump from the peek result iterator (in one iteration through the worker loop) when stashing peek responses.",
-    ParameterScope::Environment,
-);
-
-/// The size of each batch, as number of rows, pumped from the peek result
-/// iterator when stashing peek responses.
-pub const PEEK_STASH_BATCH_SIZE: Config<usize> = Config::new(
-    "compute_peek_stash_batch_size",
-    100000,
-    "The size, as number of rows, of each batch pumped from the peek result iterator (in one iteration through the worker loop) when stashing peek responses.",
-    ParameterScope::Environment,
-);
-
 /// Whether compute should stop peeks that iterate over too many rows.
 pub const ENABLE_PEEK_ROW_ITERATION_LIMIT: Config<bool> = Config::new(
     "enable_compute_peek_row_iteration_limit",
@@ -626,21 +609,29 @@ pub const ENABLE_PEEK_ROW_ITERATION_LIMIT: Config<bool> = Config::new(
 );
 
 /// The maximum number of rows a peek may iterate over on each worker.
+///
+/// The count spans a peek's whole walk of its arrangement, rows written to the peek stash
+/// included, because a peek walks its arrangement once and the count travels with that walk.
 pub const PEEK_ROW_ITERATION_LIMIT: Config<usize> = Config::new(
     "compute_peek_row_iteration_limit",
     1000,
-    "The maximum number of rows a peek may iterate over on each worker when enable_compute_peek_row_iteration_limit is enabled. Does not apply once a peek's results move to the peek stash.",
+    "The maximum number of rows a peek may iterate over on each worker when enable_compute_peek_row_iteration_limit is enabled. The count spans the peek's whole walk, rows written to the peek stash included.",
     ParameterScope::Environment,
 );
 
-/// Whether a fast-path index peek may move its walk off the timely worker.
+/// Whether a fast-path index peek may move its walk off the timely worker for latency.
 ///
-/// Off, a peek walks the arrangement to completion on the worker that owns it, so an expensive
+/// Off, a peek walks the arrangement on the worker that owns it until it answers, so an expensive
 /// peek delays every other message that worker serves. On, a peek that outruns
 /// [`INDEX_PEEK_INLINE_BUDGET`] is promoted and finishes away from the worker.
 ///
-/// The kill switch for the whole mechanism, and off by default so production placement is
-/// unchanged until the path earns trust.
+/// It gates promotion for latency and only that. A peek whose accumulated rows outgrow what it may
+/// answer with inline is promoted whichever way this is set, because the driver that writes to the
+/// peek stash is the promoted one and such a peek has no other route to an answer. Off therefore
+/// means an ordinary peek runs where it used to, not that no peek ever leaves the worker.
+///
+/// The kill switch for the placement, and off by default so production placement is unchanged
+/// until the path earns trust.
 ///
 /// Environment-scoped because it selects between two execution paths whose output-equivalence is
 /// an assumption rather than a guarantee. A peek is broadcast to every replica of its cluster and
@@ -676,8 +667,8 @@ pub const ENABLE_INDEX_PEEK_OFFLOAD: Config<bool> = Config::new(
 /// without discarding the positions they have walked.
 ///
 /// Zero walks one position rather than none. A peek granted no fuel suspends before it has walked
-/// anywhere, and a suspension holding no full batch is a promotion, so zero would promote every
-/// point lookup for a walk that visited nothing.
+/// anywhere, and a suspension is a promotion, so zero would promote every point lookup for a walk
+/// that visited nothing.
 ///
 /// Environment-scoped because the threshold decides which peeks leave the worker. A value that
 /// differs by replica selects a different execution path for the same peek exactly as
@@ -725,6 +716,11 @@ pub const INDEX_PEEK_ACTIVATION_BUDGET: Config<usize> = Config::new(
 /// Counted in the same unit as [`INDEX_PEEK_INLINE_BUDGET`], for the same reasons, and likewise
 /// read through a handle.
 ///
+/// An upper bound rather than a period. A walk whose rows are bound for the peek stash suspends
+/// once its accumulation crosses `peek_response_stash_threshold_bytes`, which at that threshold's
+/// default is by far the smaller trigger, so such a walk yields per batch and the fuel it did not
+/// spend is not carried into the next slice.
+///
 /// Replica-scoped, unlike the switch and the two budgets. It selects no execution path and changes
 /// no result. It only sets how often a walk that is already promoted yields and rechecks
 /// cancellation, which is the timing of the replica process's own execution. Divergence is safe
@@ -849,8 +845,6 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&PEEK_RESPONSE_STASH_BATCH_MAX_RUNS)
         .add(&PEEK_RESPONSE_STASH_READ_BATCH_SIZE_BYTES)
         .add(&PEEK_RESPONSE_STASH_READ_MEMORY_BUDGET_BYTES)
-        .add(&PEEK_STASH_NUM_BATCHES)
-        .add(&PEEK_STASH_BATCH_SIZE)
         .add(&ENABLE_PEEK_ROW_ITERATION_LIMIT)
         .add(&PEEK_ROW_ITERATION_LIMIT)
         .add(&ENABLE_INDEX_PEEK_OFFLOAD)