diff --git a/Cargo.lock b/Cargo.lock index 5ffa6e4970737..ab380d93fdfac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7058,6 +7058,7 @@ dependencies = [ "mz-expr", "mz-metrics", "mz-ore", + "mz-persist", "mz-persist-client", "mz-persist-types", "mz-proto", @@ -7076,6 +7077,7 @@ dependencies = [ "scopeguard", "serde", "smallvec", + "thiserror 2.0.18", "timely", "tokio", "tracing", diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index 318bc3ba145be..79612c7cfb4cf 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -1144,6 +1144,10 @@ 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`. + 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. A peek whose walk was offloaded contributes only the inline slice that offloaded it, and its time away from the worker is `mz_index_peek_offload_seconds`. labels: @@ -3709,25 +3713,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 diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 363cceec80530..420de4d2ba825 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -832,8 +832,6 @@ def get_default_system_parameters( "compute_index_peek_permit_fraction", "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", diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index b3be88dbb9c26..ce0797188507d 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -3428,8 +3428,6 @@ def __init__( "mz_metrics_lgalloc_refresh_interval", "mz_metrics_rusage_refresh_interval", "mz_metrics_usage_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", diff --git a/src/adapter/src/coord/peek.rs b/src/adapter/src/coord/peek.rs index a8d653495210f..e5a78effe5dd9 100644 --- a/src/adapter/src/coord/peek.rs +++ b/src/adapter/src/coord/peek.rs @@ -1111,7 +1111,9 @@ impl crate::coord::Coordinator { // - ProtoBatch is lost in flight // - ProtoBatch is lost because when combining PeekResponse // from workers a cancellation or error "overrides" other - // results, meaning we drop them + // results, meaning we drop them, which includes the + // `max_result_size` rejection the controller raises once it + // has summed the workers' stashed rows // - This task here is not run to completion before it can // delete all batches // diff --git a/src/compute-client/src/protocol/response.rs b/src/compute-client/src/protocol/response.rs index dae4d168af85b..41159fdf732bf 100644 --- a/src/compute-client/src/protocol/response.rs +++ b/src/compute-client/src/protocol/response.rs @@ -218,6 +218,18 @@ impl PeekResponse { Self::Error(_) | Self::Canceled => 0, } } + + /// The size of the rows this response answers with, stashed rows included. + /// + /// This is what `max_result_size` bounds, so a stashed answer has to be measured the way an + /// inline one is. Reading [`PeekResponse::inline_byte_len`] instead would leave a stashed + /// answer bounded per worker only, since the batches a worker writes are invisible to it. + pub fn answer_byte_len(&self) -> usize { + match self { + Self::Rows(_) | Self::Error(_) | Self::Canceled => self.inline_byte_len(), + Self::Stashed(stashed) => stashed.answer_byte_len(), + } + } } /// The error of an unsuccessful peek. @@ -291,7 +303,14 @@ pub struct StashedPeekResponse { /// This does _NOT_ include rows in `inline_rows`. pub num_rows_batches: u64, /// The sum of the encoded sizes of all batches in this response. + /// + /// What the batches cost in blob storage. `max_result_size` bounds + /// [`StashedPeekResponse::stashed_byte_len`] instead. pub encoded_size_bytes: usize, + /// The size of the stashed rows, measured as an inline answer measures its own. + /// + /// Does _NOT_ include `inline_rows`. + pub stashed_byte_len: usize, /// [RelationDesc] for the rows in these stashed batches of results. pub relation_desc: RelationDesc, /// The [ShardId] under which result batches have been stashed. @@ -322,6 +341,13 @@ impl StashedPeekResponse { self.encoded_size_bytes + inline_size } + + /// The size of the rows in this result, measured as an inline answer measures its own. + pub fn answer_byte_len(&self) -> usize { + let inline_size: usize = self.inline_rows.iter().map(|r| r.byte_len()).sum(); + + self.stashed_byte_len.saturating_add(inline_size) + } } /// Various responses that can be communicated after a COPY TO command. diff --git a/src/compute-client/src/service.rs b/src/compute-client/src/service.rs index 70c5cb56126ee..46686b213b5a0 100644 --- a/src/compute-client/src/service.rs +++ b/src/compute-client/src/service.rs @@ -566,12 +566,12 @@ impl PendingSubscribe { struct PendingPeek { /// The responses merged so far. response: PeekResponse, - /// Inline result bytes seen so far, across all shards. + /// Answer bytes seen so far, across all shards, stashed rows included. /// /// Tracked separately from `response` because a worker's rows are dropped as soon as any /// worker reports an error. Without this the aggregate size check would depend on the order /// the responses happen to arrive in. - inline_byte_len: usize, + answer_byte_len: usize, /// The shards that have provided responses. ready_shards: BTreeSet, } @@ -580,7 +580,7 @@ impl PendingPeek { fn new() -> Self { Self { response: PeekResponse::Rows(vec![RowCollection::default()]), - inline_byte_len: 0, + answer_byte_len: 0, ready_shards: BTreeSet::new(), } } @@ -589,20 +589,24 @@ impl PendingPeek { let first = self.ready_shards.insert(shard_id); assert!(first, "duplicate peek response"); - self.inline_byte_len = self - .inline_byte_len - .saturating_add(response.inline_byte_len()); + // Stashed rows count here as inline ones do, so a peek whose workers each stash a share + // under the ceiling is still bounded by what the client receives in total. + self.answer_byte_len = self + .answer_byte_len + .saturating_add(response.answer_byte_len()); let current = mem::replace(&mut self.response, PeekResponse::Canceled); self.response = merge_peek_responses(current, response); // Merging eagerly is what keeps the controller's memory bounded, so the size check has to // happen on every response rather than once at the end. - if self.inline_byte_len > max_result_size.cast_into() { - // NOTE: Tests match on this exact message, so nothing else may produce it. - let error = PeekError::unstructured(format!( - "total result exceeds max size of {}", - ByteSize::b(max_result_size) - )); + if self.answer_byte_len > max_result_size.cast_into() { + // NOTE: an error merged over a stashed response drops the batches that response + // names, and only the reader, which never sees this response, deletes a stashed + // answer's parts. This is the leak DB-50 reports for a cancelled peek, which drops + // the same batches at the same merge. + let error = PeekError::ResultExceedsMaxSize { + max_result_size: max_result_size.cast_into(), + }; let current = mem::replace(&mut self.response, PeekResponse::Canceled); self.response = merge_peek_responses(current, PeekResponse::Error(error)); } @@ -636,6 +640,7 @@ fn merge_peek_responses(resp1: PeekResponse, resp2: PeekResponse) -> PeekRespons let StashedPeekResponse { num_rows_batches: num_rows_batches1, encoded_size_bytes: encoded_size_bytes1, + stashed_byte_len: stashed_byte_len1, relation_desc: relation_desc1, shard_id: shard_id1, batches: mut batches1, @@ -644,6 +649,7 @@ fn merge_peek_responses(resp1: PeekResponse, resp2: PeekResponse) -> PeekRespons let StashedPeekResponse { num_rows_batches: num_rows_batches2, encoded_size_bytes: encoded_size_bytes2, + stashed_byte_len: stashed_byte_len2, relation_desc: relation_desc2, shard_id: shard_id2, batches: mut batches2, @@ -671,6 +677,7 @@ fn merge_peek_responses(resp1: PeekResponse, resp2: PeekResponse) -> PeekRespons Stashed(Box::new(StashedPeekResponse { num_rows_batches: num_rows_batches1 + num_rows_batches2, encoded_size_bytes: encoded_size_bytes1 + encoded_size_bytes2, + stashed_byte_len: stashed_byte_len1.saturating_add(stashed_byte_len2), relation_desc: relation_desc1, shard_id: shard_id1, batches: batches1, diff --git a/src/compute-client/src/service/tests.rs b/src/compute-client/src/service/tests.rs index c9b06dbe246b2..c4e51ba17fd50 100644 --- a/src/compute-client/src/service/tests.rs +++ b/src/compute-client/src/service/tests.rs @@ -11,8 +11,10 @@ use super::*; use std::num::NonZeroUsize; +use std::str::FromStr; -use mz_repr::Row; +use mz_persist_types::ShardId; +use mz_repr::{RelationDesc, Row}; #[mz_ore::test] fn pending_peek_response_precedence() { @@ -60,10 +62,9 @@ fn peek_max_size_wins_over_row_iteration_limit_in_every_order() { [2, 0, 1], [2, 1, 0], ]; - let expected = PeekResponse::Error(PeekError::unstructured(format!( - "total result exceeds max size of {}", - ByteSize::b(max_result_size) - ))); + let expected = PeekResponse::Error(PeekError::ResultExceedsMaxSize { + max_result_size: max_result_size.cast_into(), + }); for permutation in permutations { let mut pending = PendingPeek::new(); @@ -74,3 +75,43 @@ fn peek_max_size_wins_over_row_iteration_limit_in_every_order() { assert_eq!(pending.response, expected, "{permutation:?}"); } } + +/// A stashed answer is bounded by `max_result_size` across the workers that produced it, not +/// per worker. +/// +/// Each worker decides on its own whether its share is large enough to stash, so a peek whose +/// answer exceeds the ceiling can arrive as several stashed responses each well under it. Reading +/// only the rows a response carries inline would leave such an answer bounded at the ceiling times +/// the worker count. +#[mz_ore::test] +fn a_stashed_answer_is_bounded_across_workers() { + let stashed = |stashed_byte_len| { + PeekResponse::Stashed(Box::new(StashedPeekResponse { + num_rows_batches: 1, + encoded_size_bytes: 0, + stashed_byte_len, + relation_desc: RelationDesc::empty(), + shard_id: ShardId::from_str("s00000000-0000-0000-0000-000000000000").expect("valid"), + batches: Vec::new(), + inline_rows: Vec::new(), + })) + }; + let max_result_size = 100; + + // One worker's share, under the ceiling on its own. + let mut pending = PendingPeek::new(); + pending.absorb(0, stashed(60), max_result_size); + assert!( + matches!(pending.response, PeekResponse::Stashed(_)), + "a share under the ceiling answers with its batches" + ); + + // A second share of the same size puts the answer over it. + pending.absorb(1, stashed(60), max_result_size); + assert_eq!( + pending.response, + PeekResponse::Error(PeekError::ResultExceedsMaxSize { + max_result_size: max_result_size.cast_into(), + }), + ); +} diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs index dd9d636075bc8..28f2e494d46ac 100644 --- a/src/compute-types/src/dyncfgs.rs +++ b/src/compute-types/src/dyncfgs.rs @@ -626,23 +626,6 @@ pub const PEEK_RESPONSE_STASH_READ_MEMORY_BUDGET_BYTES: Config = 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 = 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 = 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 = Config::new( "enable_compute_peek_row_iteration_limit", @@ -652,19 +635,24 @@ pub const ENABLE_PEEK_ROW_ITERATION_LIMIT: Config = 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 = 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 to completion on the worker that owns it, delaying every other message that -/// worker serves. On, a peek that outruns [`INDEX_PEEK_INLINE_BUDGET`] finishes away from it. +/// Off, a peek walks on the worker that owns it until it answers, delaying every other message +/// that worker serves. On, a peek that outruns [`INDEX_PEEK_INLINE_BUDGET`] finishes away from it. /// -/// The kill switch for the whole mechanism. +/// This gates latency offload only. A peek whose rows outgrow an inline answer is offloaded either +/// way, because the driver that writes to the peek stash is the offloaded one. Off means an ordinary +/// peek runs where it used to, not that none leaves the worker. pub const ENABLE_INDEX_PEEK_OFFLOAD: Config = Config::new( "enable_compute_index_peek_offload", false, @@ -711,9 +699,12 @@ pub const INDEX_PEEK_ACTIVATION_BUDGET: Config = Config::new( /// /// At a plausible 100ns to 1us per position this bounds cancellation latency to single-digit /// milliseconds. Larger than [`INDEX_PEEK_INLINE_BUDGET`] because an offloaded scan is off the -/// worker's critical path, so its slices answer to cancellation latency rather than to the -/// worker's availability. A check is a few loads and no hand-off, so a finer granularity costs -/// little. +/// worker's critical path, so its slices answer to cancellation latency, not to the worker's +/// availability. A check is a few loads and no hand-off, so a finer granularity costs little. +/// +/// An upper bound, not a period: a walk bound for the peek stash suspends once its accumulation +/// crosses `peek_response_stash_threshold_bytes`, by far the smaller trigger at that threshold's +/// default, and unspent fuel is not carried over. pub const INDEX_PEEK_YIELD_GRANULARITY: Config = Config::new( "compute_index_peek_yield_granularity", 10000, @@ -826,8 +817,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) diff --git a/src/compute/Cargo.toml b/src/compute/Cargo.toml index f81d09c5c947c..393ba13caeeee 100644 --- a/src/compute/Cargo.toml +++ b/src/compute/Cargo.toml @@ -33,6 +33,7 @@ mz-expr = { path = "../expr" } mz-metrics = { path = "../metrics" } mz-ore = { path = "../ore", features = ["async", "process", "tracing", "columnar", "differential-dataflow", "pager", "region"] } mz-proto = { path = "../proto" } +mz-persist = { path = "../persist" } mz-persist-client = { path = "../persist-client" } mz-persist-types = { path = "../persist-types" } mz-repr = { path = "../repr" } @@ -47,6 +48,7 @@ prost.workspace = true scopeguard.workspace = true serde.workspace = true smallvec = { workspace = true, features = ["serde", "union"] } +thiserror.workspace = true timely.workspace = true tokio.workspace = true tracing.workspace = true diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index ad8e8ba9443c7..d439514102c64 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -14,7 +14,6 @@ use std::rc::Rc; use std::sync::Arc; use std::time::{Duration, Instant}; -use bytesize::ByteSize; use differential_dataflow::Hashable; use differential_dataflow::lattice::Lattice; use differential_dataflow::trace::TraceReader; @@ -29,8 +28,7 @@ use mz_compute_client::protocol::response::{ use mz_compute_types::dataflows::DataflowDescription; use mz_compute_types::dyncfgs::{ ENABLE_PEEK_RESPONSE_STASH, ENABLE_PEEK_ROW_ITERATION_LIMIT, - PEEK_RESPONSE_STASH_BATCH_MAX_RUNS, PEEK_RESPONSE_STASH_THRESHOLD_BYTES, - PEEK_ROW_ITERATION_LIMIT, PEEK_STASH_BATCH_SIZE, PEEK_STASH_NUM_BATCHES, + PEEK_RESPONSE_STASH_THRESHOLD_BYTES, PEEK_ROW_ITERATION_LIMIT, }; use mz_compute_types::plan::render_plan::RenderPlan; use mz_dyncfg::{ConfigSet, ConfigValHandle}; @@ -86,7 +84,7 @@ use self::peek_budget::InlineBudget; use self::peek_metrics::IndexPeekMetrics; use self::peek_metrics::PeekWalkMetrics; pub(crate) use self::peek_offload::PeekPermits; -use self::peek_offload::{OffloadConfig, OffloadOutcome, OffloadedPeek}; +use self::peek_offload::{OffloadConfig, OffloadedPeek}; use self::peek_scan::{IndexPeekScan, PeekScan, ScanOutcome, entry_byte_len}; /// Cheap handles on the dyncfgs that bound how many rows a peek may examine. @@ -1224,14 +1222,16 @@ impl<'a> ActiveComputeState<'a> { .finishing .is_streamable(peek.peek.result_desc.arity()); - let peek_stash_enabled = { + // Whether a diverted peek has somewhere to write its rows. A flag here and the location + // itself only where an offload needs it, because this runs for every peek the sweep gives + // a turn, including the point lookups that answer inline and never reach the stash. + let has_stash_location = { let enabled = ENABLE_PEEK_RESPONSE_STASH.get(&self.compute_state.worker_config); - let peek_persist_stash_available = - self.compute_state.peek_stash_persist_location.is_some(); - if !peek_persist_stash_available && enabled { + let located = self.compute_state.peek_stash_persist_location.is_some(); + if !located && enabled { error!("missing peek_stash_persist_location but peek stash is enabled"); } - enabled && peek_persist_stash_available + enabled && located }; let peek_stash_threshold_bytes = @@ -1250,7 +1250,7 @@ impl<'a> ActiveComputeState<'a> { let status = peek.seek_fulfillment( upper, self.compute_state.max_result_size, - peek_stash_enabled && peek_stash_eligible, + peek_stash_eligible && has_stash_location, peek_stash_threshold_bytes, row_iteration_limit, &mut unspent, @@ -1282,11 +1282,26 @@ impl<'a> ActiveComputeState<'a> { let config = OffloadConfig::new(&self.compute_state.worker_config); let walk_metrics = self.compute_state.peek_walk_metrics.clone(); let worker = std::thread::current(); + // Read off the scan rather than decided again, so a walk that can offer a batch + // always has somewhere to write it. + let stash = self + .compute_state + .peek_stash_persist_location + .as_ref() + .filter(|_| scan.stash_eligible()) + .cloned() + .map(|location| { + peek_stash::StashTarget::new( + &peek.peek, + Arc::clone(&self.compute_state.persist_clients), + location, + ) + }); let offloaded = OffloadedPeek::start( peek.peek, - peek.trace_bundle, scan, + stash, &permits, config, walk_metrics, @@ -1297,20 +1312,6 @@ impl<'a> ActiveComputeState<'a> { .pending_peeks .push_back(PendingPeek::Offloaded(offloaded)); } - PeekStatus::UsePeekStash => { - let _span = span!(parent: &peek.span, Level::DEBUG, "process_stash_peek").entered(); - - let location = self - .compute_state - .peek_stash_persist_location - .clone() - .expect("stash location established before diverting"); - let stash_task = self.start_stash_upload(&location, peek.peek, peek.trace_bundle); - - self.compute_state - .pending_peeks - .push_back(PendingPeek::Stash(stash_task)); - } } } @@ -1335,47 +1336,16 @@ impl<'a> ActiveComputeState<'a> { result }), PendingPeek::Offloaded(offloaded) => match offloaded.result.try_recv() { - Ok((outcome, duration)) => { - // Both outcomes are timed, because a walk that hands back is the expensive - // case rather than an aborted one. + Ok((response, duration)) => { + // Covers the writing to the peek stash too, because the walk that produces the + // rows is the one that writes them. self.compute_state .metrics .index_peek_offload_seconds .observe(duration.as_secs_f64()); - match outcome { - OffloadOutcome::Answered(response) => { - trace!(?offloaded.peek, ?duration, "finished offloaded index peek walk"); - Some(response) - } - OffloadOutcome::NeedsStash => { - let _span = - span!(parent: &offloaded.span, Level::DEBUG, "process_stash_peek") - .entered(); - trace!(?offloaded.peek, ?duration, "handing offloaded index peek to the stash"); - - match self.compute_state.peek_stash_persist_location.clone() { - Some(location) => { - let PendingPeek::Offloaded(offloaded) = pending else { - unreachable!("matched as an offloaded peek") - }; - let stash_task = self.start_stash_upload( - &location, - offloaded.peek, - offloaded.trace_bundle, - ); - self.compute_state - .pending_peeks - .push_back(PendingPeek::Stash(stash_task)); - return; - } - None => Some(PeekResponse::Error(PeekError::unstructured( - "peek result is too large to answer inline and this replica \ - has no peek stash location", - ))), - } - } - } + trace!(?offloaded.peek, ?duration, "finished offloaded index peek walk"); + Some(response) } Err(oneshot::error::TryRecvError::Empty) => None, // The task drops its sender without sending only on a cancellation, which removes @@ -1394,23 +1364,6 @@ impl<'a> ActiveComputeState<'a> { ))) } }, - PendingPeek::Stash(stashing_peek) => { - let num_batches = PEEK_STASH_NUM_BATCHES.get(&self.compute_state.worker_config); - let batch_size = PEEK_STASH_BATCH_SIZE.get(&self.compute_state.worker_config); - stashing_peek.pump_rows(num_batches, batch_size); - - if let Ok((response, duration)) = stashing_peek.result.try_recv() { - self.compute_state - .metrics - .stashed_peek_seconds - .observe(duration.as_secs_f64()); - trace!(?stashing_peek.peek, ?duration, "finished stashing peek response in persist"); - - Some(response) - } else { - None - } - } }; if let Some(response) = response { @@ -1422,32 +1375,6 @@ impl<'a> ActiveComputeState<'a> { } } - /// Starts a peek stash upload for `peek`, which walks the ok trace of `trace_bundle` and - /// writes the rows it produces to persist. - /// - /// The walk starts over, so rows a previous walk of the same peek accumulated are produced - /// again rather than carried across. - fn start_stash_upload( - &self, - persist_location: &PersistLocation, - peek: Peek, - trace_bundle: TraceBundle, - ) -> peek_stash::StashingPeek { - // NOTE: The row iteration limit does not follow a peek into the stash. The stash restarts - // the scan and produces in bounded bursts, so a stashed peek may examine any number of - // rows. - let batch_max_runs = - PEEK_RESPONSE_STASH_BATCH_MAX_RUNS.get(&self.compute_state.worker_config); - - peek_stash::StashingPeek::start_upload( - Arc::clone(&self.compute_state.persist_clients), - persist_location, - peek, - trace_bundle, - batch_max_runs, - ) - } - /// Scan the peeks a driver is finishing and the peeks awaiting a turn, and attempt to retire /// each. /// @@ -1639,9 +1566,6 @@ pub enum PendingPeek { Index(IndexPeek), /// A peek against a Persist-backed collection. Persist(PersistPeek), - /// A peek against an index that is being stashed in the peek stash by an - /// async background task. - Stash(peek_stash::StashingPeek), /// A peek against an index whose walk was offloaded from the worker and is running as an async /// task. Offloaded(OffloadedPeek), @@ -1767,7 +1691,6 @@ impl PendingPeek { match self { PendingPeek::Index(p) => &p.span, PendingPeek::Persist(p) => &p.span, - PendingPeek::Stash(p) => &p.span, PendingPeek::Offloaded(p) => &p.span, } } @@ -1776,7 +1699,6 @@ impl PendingPeek { match self { PendingPeek::Index(p) => &p.peek, PendingPeek::Persist(p) => &p.peek, - PendingPeek::Stash(p) => &p.peek, PendingPeek::Offloaded(p) => &p.peek, } } @@ -1908,10 +1830,7 @@ impl PersistPeek { if let Some(row) = eval_result { total_size = total_size.saturating_add(entry_byte_len(&row)); if total_size > max_result_size { - return Err(PeekError::unstructured(format!( - "result exceeds max size of {}", - ByteSize::b(u64::cast_from(max_result_size)) - ))); + return Err(PeekError::ResultExceedsMaxSize { max_result_size }); } result.push((row, count)); limit_remaining = limit_remaining.saturating_sub(count.get()); @@ -2030,55 +1949,35 @@ impl IndexPeek { let outcome = scan.step(row_iteration_limit, fuel); - // A suspension the offload can resume is the one outcome that leaves the walk unfinished, - // so it is the one outcome this driver reports nothing for. Everything else ends the walk - // here, and this driver is the one that accounts for it. - let offloaded = matches!(outcome, ScanOutcome::Suspended) && !scan.batch_ready(); let phases = scan.phases(); - if !offloaded { - metrics.walk.walked_inline(); - metrics.walk.observe_error_phase(&phases); - } - - let rows = match outcome { - ScanOutcome::Finished(Ok(rows)) => rows, - ScanOutcome::Finished(Err(error)) => { - return PeekStatus::Ready(PeekResponse::Error(error)); - } - // A scan that suspends without a batch has work left and rows it may still - // accumulate, so it travels to the offloaded walk with its positions and their cost. - // The offload costs one hand-off rather than a second walk. - ScanOutcome::Suspended if !scan.batch_ready() => return PeekStatus::Offload(scan), - // Diversion is sound only for a scan whose error walk is over. The stash answers the - // peek from a walk of the ok trace alone and never reads the error trace, so a peek - // diverted with its error trace half-read would return rows where it must report an - // error. Only the ok walk accumulates rows, so a scan holding a full batch has read - // the error trace out, and the guard states that rather than assuming it. - ScanOutcome::Suspended => { - if !scan.error_trace_clean() { - soft_panic_or_log!( - "peek on {} suspended before its error trace was read out", - self.peek.target.id() - ); - return PeekStatus::Ready(PeekResponse::Error(PeekError::unstructured( - "peek suspended before its error trace was read out", - ))); - } - // Dropped here rather than with the scan, because discarding it is this driver's - // decision: the stash walks the ok trace again and produces these rows a second - // time. - let _batch = scan.take_batch(); - return PeekStatus::UsePeekStash; + match outcome { + // Both answers end the walk on this worker, so this driver accounts for it either + // way. + ScanOutcome::Finished(result) => { + metrics.walk.walked_inline(); + metrics.walk.observe_error_phase(&phases); + PeekStatus::Ready(match result { + Ok(rows) => { + metrics.walk.observe_ok_phase(&phases); + metrics + .walk + .rows_response(rows, &self.peek.finishing.order_by) + } + // The ok phase goes unreported, because an error can come from either walk + // and its numbers describe a finished ok walk only when rows came out of it. + Err(error) => PeekResponse::Error(error), + }) } - }; - - metrics.walk.observe_ok_phase(&phases); - - PeekStatus::Ready( - metrics - .walk - .rows_response(rows, &self.peek.finishing.order_by), - ) + // The one outcome that leaves the walk unfinished, and so the one this driver + // reports nothing for. + // + // A scan suspends out of fuel or holding a full batch, and this driver can carry on + // with neither: it walks under a budget the slice has spent, and it writes no rows, so + // a batch handed to it here would have to be dropped. Every position the scan walked + // travels with it, and so does their cost, which is what makes offload cost one + // hand-off rather than a second walk. + ScanOutcome::Suspended => PeekStatus::Offload(scan), + } } } @@ -2088,11 +1987,12 @@ enum PeekStatus { /// The frontiers of objects are not yet advanced enough, peek is still /// pending. NotReady, - /// The result size is above the configured threshold and the peek is - /// eligible for using the peek result stash. - UsePeekStash, - /// The walk stopped with work left and nothing to hand over, so it is finished away from the - /// worker. Carries the scan, which resumes from the cursor positions it stopped on. + /// The walk stopped with work left, so it is finished away from the worker. Carries the scan, + /// which resumes from the cursor positions it stopped on. + /// + /// A walk stops either because it spent the fuel this activation granted it or because its + /// accumulated rows grew into a batch bound for the peek stash. Both leave here, because the + /// driver that finishes a walk is also the one that writes to the stash. Offload(IndexPeekScan), /// The peek result is ready. Ready(PeekResponse), diff --git a/src/compute/src/compute_state/index_peek_tests.rs b/src/compute/src/compute_state/index_peek_tests.rs index 835b72a7c8c68..49e5e112ce6dc 100644 --- a/src/compute/src/compute_state/index_peek_tests.rs +++ b/src/compute/src/compute_state/index_peek_tests.rs @@ -193,6 +193,7 @@ impl TestMetrics { BTreeMap::from([ ("walks_inline", metrics.index_peek_walks_inline.get()), ("walks_offloaded", metrics.index_peek_walks_offloaded.get()), + ("walks_stashed", metrics.index_peek_stashed_total.get()), ( "error_scan_seconds", metrics.index_peek_error_scan_seconds.get_sample_count(), @@ -245,6 +246,7 @@ fn expected_observations( BTreeMap::from([ ("walks_inline", walks_inline), ("walks_offloaded", 0), + ("walks_stashed", 0), ("error_scan_seconds", error_scan), ("cursor_setup_seconds", cursor_setup), ("row_iteration_seconds", rows), @@ -262,7 +264,6 @@ fn expected_observations( #[derive(Debug, PartialEq)] enum Answer { NotReady, - UsePeekStash, Offload, Ready(PeekResponse), } @@ -271,7 +272,6 @@ impl From for Answer { fn from(status: PeekStatus) -> Self { match status { PeekStatus::NotReady => Answer::NotReady, - PeekStatus::UsePeekStash => Answer::UsePeekStash, // The scan an offload carries has no comparison of its own. What is comparable // is that the walk left this driver rather than answering here. PeekStatus::Offload(_) => Answer::Offload, @@ -359,11 +359,14 @@ fn an_error_answered_peek_reports_no_phase_timers() { assert_eq!(metrics.observations(), expected_observations(1, 0, 0, 0)); } -/// A peek whose accumulated rows fill a batch is diverted to the stash rather than answered -/// inline. The phases the walk did pass through are reported, and those only a peek answered -/// inline reaches are not. +/// A peek whose accumulated rows fill a batch leaves the worker, however much fuel the slice +/// still had, and the walk it leaves with reports nothing. +/// +/// A batch-ready suspension is an offload because the driver that finishes a walk is also the +/// one that writes to the peek stash. Withholding it here would strand the peek: the scan +/// makes no progress until its batch is taken, and this driver never takes one. #[mz_ore::test] -fn a_scan_that_fills_a_batch_diverts_the_peek_to_the_stash() { +fn a_scan_that_fills_a_batch_leaves_the_worker_with_fuel_to_spare() { let keys: Vec = (0..6).map(ok_row).collect(); let mut subject = index_peek_over( index_peek(trivial_finishing(), None), @@ -373,29 +376,20 @@ fn a_scan_that_fills_a_batch_diverts_the_peek_to_the_stash() { let metrics = TestMetrics::new(); // A threshold of zero bytes is crossed by the first row, so the scan fills a batch well - // before the trace runs out. - let answer = subject.collect_finished_data( - u64::MAX, - true, - 0, - None, - &mut unbounded_fuel(), - &metrics.as_metrics(), - ); + // before the trace runs out and well before unbounded fuel could run out. + let mut fuel = unbounded_fuel(); + let answer = + subject.collect_finished_data(u64::MAX, true, 0, None, &mut fuel, &metrics.as_metrics()); - assert_eq!(Answer::from(answer), Answer::UsePeekStash); - assert_eq!(metrics.observations(), expected_observations(1, 1, 1, 0)); + assert_eq!(Answer::from(answer), Answer::Offload); + assert!(fuel > 0, "the slice stopped for the batch, not for fuel"); + assert_eq!(metrics.observations(), expected_observations(0, 0, 0, 0)); } -/// A peek whose walk both fills a batch and runs out of fuel is diverted to the stash rather -/// than offloaded, and the driver that diverted it accounts for the walk. -/// -/// This is the livelock the placement policy is built around. An offloaded scan holding a full -/// batch has nowhere to write it, so stepping it spends no fuel and advances no cursor, and a -/// driver that resumed it would yield forever. The two causes of a suspension coincide here, -/// which is the case an offload condition written as "the fuel ran out" would get wrong. +/// A peek whose walk both fills a batch and runs out of fuel leaves the worker exactly as one +/// that did only the first, so the two causes of a suspension need no telling apart. #[mz_ore::test] -fn a_batch_ready_suspension_is_diverted_rather_than_offloaded() { +fn a_batch_ready_suspension_out_of_fuel_is_offloaded_too() { let keys: Vec = (0..6).map(ok_row).collect(); let mut subject = index_peek_over( index_peek(trivial_finishing(), None), @@ -411,13 +405,13 @@ fn a_batch_ready_suspension_is_diverted_rather_than_offloaded() { let mut fuel = 1; let answer = subject.collect_finished_data(u64::MAX, true, 0, None, &mut fuel, &metrics.as_metrics()); - assert_eq!(Answer::from(answer), Answer::UsePeekStash); + assert_eq!(Answer::from(answer), Answer::Offload); assert_eq!(fuel, 0, "the slice spent every position it was given"); - assert_eq!(metrics.observations(), expected_observations(1, 1, 1, 0)); + assert_eq!(metrics.observations(), expected_observations(0, 0, 0, 0)); } /// A peek whose walk outruns the fuel it was granted leaves the worker rather than being -/// answered or diverted, and the walk it leaves with reports nothing. +/// answered, and the walk it leaves with reports nothing. /// /// Reporting here as well as in the driver that finishes the walk would count one walk twice, /// on both substrates and in every phase histogram, and the numbers a scan carries are diff --git a/src/compute/src/compute_state/peek_budget.rs b/src/compute/src/compute_state/peek_budget.rs index 872dbace5fa55..169ae2bc800f2 100644 --- a/src/compute/src/compute_state/peek_budget.rs +++ b/src/compute/src/compute_state/peek_budget.rs @@ -33,11 +33,9 @@ impl InlineBudgetConfig { return ActivationBudget::Unbounded; } - // Both floors keep the parameters monotone down to zero rather than wedging there. A - // per-peek budget of zero suspends every scan before it walks a position, and a suspension - // holding no full batch is an offload, so every point lookup would pay for a task and a - // permit to walk nothing. An aggregate of zero passes every peek over on every activation, - // and the activation a passed-over peek asks for finds the same empty budget. + // Zero must not wedge. A per-peek budget of zero suspends every scan before it walks a + // position, and every suspension is an offload, so point lookups would pay for a task and + // a permit to walk nothing. An aggregate of zero passes every peek over forever. ActivationBudget::Bounded { per_peek: self.per_peek.get().max(1), remaining: self.aggregate.get().max(1), @@ -54,10 +52,12 @@ impl InlineBudgetConfig { /// /// Both count cursor positions, the unit the scan charges. enum ActivationBudget { - /// Every peek walks to completion where it started, and nothing is offloaded or passed over. + /// Every peek walks on the worker until it answers or until its rows belong in the peek stash, + /// and nothing is passed over. /// - /// What the kill switch restores. A scan suspends only out of fuel or holding a full batch, - /// so unbounded fuel leaves the batch as the only suspension and the peek takes the stash. + /// What the kill switch restores. A stash-bound peek still offloads, because the driver that + /// writes to the stash is the offloaded one, so this restores where an ordinary peek runs and + /// not a guarantee that none leaves the worker. Unbounded, /// A peek may spend `per_peek` before it is offloaded, and all peeks together may spend /// `remaining` before the rest of this activation's work gets the worker back. diff --git a/src/compute/src/compute_state/peek_budget/tests.rs b/src/compute/src/compute_state/peek_budget/tests.rs index 4981bf649e358..39eb54882b1da 100644 --- a/src/compute/src/compute_state/peek_budget/tests.rs +++ b/src/compute/src/compute_state/peek_budget/tests.rs @@ -59,8 +59,9 @@ fn beginning_an_activation_refills_the_aggregate() { } /// With the offload off, every peek is granted unbounded fuel however much the peeks before it -/// spent, which is what makes the kill switch restore the worker's old behaviour rather than -/// approximate it. +/// spent, which is how the kill switch keeps a peek that answers inline on the worker. It says +/// nothing about a peek whose rows belong in the stash, which suspends on its batch rather +/// than on its fuel and leaves the worker either way. #[mz_ore::test] fn the_kill_switch_grants_every_peek_an_unbounded_slice() { let config = mz_dyncfgs::all_dyncfgs(); diff --git a/src/compute/src/compute_state/peek_metrics.rs b/src/compute/src/compute_state/peek_metrics.rs index d4d190a2d16e3..f5aa7820dc3e3 100644 --- a/src/compute/src/compute_state/peek_metrics.rs +++ b/src/compute/src/compute_state/peek_metrics.rs @@ -35,6 +35,8 @@ pub(super) struct PeekWalkMetrics { walks_inline: IntCounter, /// Counts walks that ended away from the timely worker. walks_offloaded: IntCounter, + /// Counts walks that answered from the peek response stash. + walks_stashed: IntCounter, error_scan_seconds: Histogram, cursor_setup_seconds: Histogram, row_iteration_seconds: Histogram, @@ -54,6 +56,7 @@ impl PeekWalkMetrics { Self { walks_inline: metrics.index_peek_walks_inline.clone(), walks_offloaded: metrics.index_peek_walks_offloaded.clone(), + walks_stashed: metrics.index_peek_stashed_total.clone(), error_scan_seconds: metrics.index_peek_error_scan_seconds.clone(), cursor_setup_seconds: metrics.index_peek_cursor_setup_seconds.clone(), row_iteration_seconds: metrics.index_peek_row_iteration_seconds.clone(), @@ -81,20 +84,29 @@ impl PeekWalkMetrics { /// Counts a walk that the timely worker drove to an outcome. /// - /// A peek diverted to the peek stash counts here, because the walk that decided that ran on - /// the worker. The stash's own walk of the same trace counts on neither substrate. + /// A walk that suspends leaves the worker rather than finishing here, so a peek answered from + /// the peek stash never counts here: the driver that writes to the stash is the offloaded one. pub(super) fn walked_inline(&self) { self.walks_inline.inc(); } /// Counts a walk that an offloaded task drove to an outcome, whatever that outcome is. /// - /// A walk cancelled while queued or while running counts on neither substrate, so the two sum - /// to the walks that ended rather than to the walks that were admitted. + /// A walk cancelled while queued or while running counts on neither substrate, as does one + /// whose task died without an outcome, which the worker answers with an error of its own. The + /// two therefore sum to the walks that reached an outcome rather than to the peeks answered. pub(super) fn walked_offloaded(&self) { self.walks_offloaded.inc(); } + /// Counts a walk that answered with a handle to the peek response stash. + /// + /// Counted alongside [`Self::walked_offloaded`] rather than instead of it, so the two + /// substrates still sum to the walks that ended. + pub(super) fn walked_to_stash(&self) { + self.walks_stashed.inc(); + } + /// Reports the phases that precede the walk over the ok trace. /// /// Reported for every terminal outcome, a hand-off to the peek stash included, because both diff --git a/src/compute/src/compute_state/peek_offload.rs b/src/compute/src/compute_state/peek_offload.rs index d0064fd306fb9..ca15cd98228ec 100644 --- a/src/compute/src/compute_state/peek_offload.rs +++ b/src/compute/src/compute_state/peek_offload.rs @@ -6,17 +6,18 @@ //! Driving an index peek's walk away from the timely worker that owns it. //! //! An offloaded walk steps its scan on the blocking pool, so neither the timely worker nor an -//! async one carries it. It returns to its async task only to answer, and checks for cancellation -//! every `yield_granularity` positions in between. The scan and the permit that admitted it travel -//! together, and every way the walk ends, including a panic, drops the two together. +//! async one carries it. It returns to its async task only to write a batch or to answer, and +//! checks for cancellation every `yield_granularity` positions in between. The scan and the permit +//! that admitted it travel together, and every way the walk ends, a panic included, drops the two +//! together. //! -//! The permit bounds the walks that run, and nothing else. An offloaded walk that has not been -//! admitted queues holding its scan, which retains its accumulated rows and pins the batches its -//! cursors were opened over, so retained memory grows with offloaded walks rather than running -//! ones. +//! The permit bounds the walks that run, not the walks that exist: an unadmitted walk queues +//! holding its scan, which pins the batches its cursors were opened over, so retained memory grows +//! with offloaded walks rather than running ones. //! -//! This driver performs no IO. A walk whose rows outgrow an inline answer hands back rather than -//! writing them, which leaves the writing to the worker. +//! This driver performs the only IO, so the scan stays free of async colouring. A walk +//! whose rows outgrow an inline answer hands over a full batch, the driver writes it to the peek +//! stash, and the walk carries on from where it stopped. use std::sync::{Arc, Mutex}; use std::thread::Thread; @@ -24,19 +25,22 @@ use std::time::{Duration, Instant}; use mz_compute_client::protocol::command::Peek; use mz_compute_client::protocol::response::{PeekError, PeekResponse}; -use mz_compute_types::dyncfgs::{INDEX_PEEK_PERMIT_FRACTION, INDEX_PEEK_YIELD_GRANULARITY}; +use mz_compute_types::dyncfgs::{ + INDEX_PEEK_PERMIT_FRACTION, INDEX_PEEK_YIELD_GRANULARITY, PEEK_RESPONSE_STASH_BATCH_MAX_RUNS, +}; use mz_dyncfg::{ConfigSet, ConfigValHandle}; use mz_expr::ColumnOrder; use mz_ore::cast::CastLossy; use mz_ore::soft_panic_or_log; use mz_ore::task::AbortOnDropHandle; use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot}; -use tracing::debug; +use tracing::{debug, warn}; +use uuid::Uuid; -use crate::arrangement::manager::TraceBundle; use crate::compute_state::PeekRowIterationConfig; use crate::compute_state::peek_metrics::PeekWalkMetrics; -use crate::compute_state::peek_scan::{IndexPeekScan, ScanOutcome}; +use crate::compute_state::peek_scan::{IndexPeekScan, RowBatch, ScanOutcome}; +use crate::compute_state::peek_stash::{StashTarget, StashUpload}; /// The bound on how many offloaded peek walks run at once. /// @@ -101,11 +105,12 @@ impl PeekPermits { /// /// A handle lets a configuration change reach a walk already under way without discarding the /// positions it has visited. The granularity and the row limit are read at every slice boundary, -/// the permit fraction once on the worker, since it sizes the bound the walk queues on. +/// the batch runs where the walk opens its upload, and the permit fraction once on the worker. #[derive(Clone, Debug)] pub(super) struct OffloadConfig { permit_fraction: ConfigValHandle, yield_granularity: ConfigValHandle, + batch_max_runs: ConfigValHandle, row_iteration: PeekRowIterationConfig, } @@ -114,32 +119,20 @@ impl OffloadConfig { Self { permit_fraction: INDEX_PEEK_PERMIT_FRACTION.handle(config), yield_granularity: INDEX_PEEK_YIELD_GRANULARITY.handle(config), + batch_max_runs: PEEK_RESPONSE_STASH_BATCH_MAX_RUNS.handle(config), row_iteration: PeekRowIterationConfig::new(config), } } } -/// What an offloaded walk hands back to the worker that offloaded it. -pub enum OffloadOutcome { - /// The walk answered the peek. - Answered(PeekResponse), - /// The walk accumulated more rows than the peek may answer with inline, and answering it needs - /// them written to the peek stash. This driver performs no IO, so the walk stops here and the - /// worker takes the peek to the stash instead. - NeedsStash, -} - /// An index peek whose walk is running away from the worker that owns it. /// /// Note that `OffloadedPeek` intentionally does not implement or derive `Clone`, as each one is /// meant to be dropped once it has been responded to. pub struct OffloadedPeek { pub(crate) peek: Peek, - /// The traces the walk reads. Retained so the stash can walk the ok trace again from here - /// when the walk hands back, under the compaction hold this bundle carries. - pub(crate) trace_bundle: TraceBundle, - /// The outcome of the walk, eventually. - pub(crate) result: oneshot::Receiver<(OffloadOutcome, Duration)>, + /// The peek's answer, eventually. + pub(crate) result: oneshot::Receiver<(PeekResponse, Duration)>, /// The `tracing::Span` tracking this peek's operation. pub(crate) span: tracing::Span, /// The task driving the walk. Dropping this aborts it. The blocking thread stepping the scan @@ -152,26 +145,21 @@ impl OffloadedPeek { /// Offloads `scan` to a task that finishes the walk away from the worker, waking `worker` /// once the outcome is ready. /// - /// `peek` and `trace_bundle` stay with the worker, because the bundle is what the stash - /// restarts the walk from if the walk hands back. + /// `stash` is where the walk writes rows the peek may not answer with inline. It is `Some` + /// exactly when `scan` was opened stash-eligible, so a scan that offers a batch always has a + /// target. Should it not, the walk fails the peek. /// - /// The scan must not already hold a full batch: its first step would report that it needs the - /// stash, costing a permit and a hand-off for a peek the worker could have diverted itself. - /// The `debug_assert` below catches that in CI, not in an optimized build. + /// The scan may already hold a full batch. This driver takes it, so offloading is how a peek + /// too large to answer inline reaches the stash. pub(super) fn start( peek: Peek, - trace_bundle: TraceBundle, scan: IndexPeekScan, + stash: Option, permits: &PeekPermits, config: OffloadConfig, metrics: PeekWalkMetrics, worker: Thread, ) -> Self { - debug_assert!( - !scan.batch_ready(), - "offloaded a peek scan that already holds a full batch" - ); - let (mut result_tx, result_rx) = oneshot::channel(); let semaphore = permits.resize(config.permit_fraction.get()); @@ -205,8 +193,9 @@ impl OffloadedPeek { _permit: permit, result_tx, }; - let (state, outcome) = Self::walk(state, &config, &metrics, order_by).await; - let Some(outcome) = outcome else { + let (state, response) = + Self::walk(state, peek_uuid, stash, &config, &metrics, order_by).await; + let Some(response) = response else { return; }; let result_tx = state.result_tx; @@ -214,10 +203,18 @@ impl OffloadedPeek { // Past the walk rather than at the permit, so a walk that took a permit and was // then cancelled is not counted, as `walked_offloaded` states. metrics.walked_offloaded(); + if matches!(response, PeekResponse::Stashed(_)) { + metrics.walked_to_stash(); + } - match result_tx.send((outcome, start.elapsed())) { + match result_tx.send((response, start.elapsed())) { Ok(()) => {} - Err((_outcome, elapsed)) => { + // TODO: a dropped stashed response leaves its parts in blob storage. The + // upload's own cleanup cannot reach them, because a finished batch belongs to + // the response rather than to the upload, and rebuilding a deletable batch + // from what the response carries needs a `WriteHandle` this task does not + // hold. A reader-side sweep or persist's own garbage collection covers it. + Err((_response, elapsed)) => { debug!(duration = ?elapsed, "dropping result for cancelled peek {peek_uuid}") } } @@ -231,35 +228,45 @@ impl OffloadedPeek { Self { peek, - trace_bundle, result: result_rx, span: tracing::Span::current(), _abort_handle: task_handle.abort_on_drop(), } } - /// Drives the scan in `state` to an outcome. `None` means the peek was cancelled, which is the - /// one way the walk ends without an outcome. + /// Drives the scan in `state` to the peek's answer, writing what it may not answer with inline + /// to `stash`. `None` means the peek was cancelled, which is the one way the walk ends without + /// an answer. async fn walk( mut state: WalkState, + peek_uuid: Uuid, + stash: Option, config: &OffloadConfig, metrics: &PeekWalkMetrics, order_by: Arc<[ColumnOrder]>, - ) -> (WalkState, Option) { + ) -> (WalkState, Option) { + // Opened by the first batch the scan hands over, so a walk that never crosses the stash + // threshold neither opens a shard nor writes a byte. Whether it is open is also what + // decides how the peek is answered: an upload answers with a handle, and no upload means + // every row the walk produced is still here to answer with. + let mut upload: Option = None; + loop { // Stepped on the blocking pool, because the walk is CPU-bound for its whole length and // would otherwise hold an async worker there. Not `block_in_place`, which parks a core // for the same span. The scan crosses to the pool and back, which it can because it // owns `Arc`-backed batch snapshots and holds no trace handle. - let config = config.clone(); + let walk_config = config.clone(); let (stepped, outcome) = mz_ore::task::spawn_blocking( || "peek_offload::walk", - move || state.step_until_blocked(&config), + move || state.step_until_blocked(&walk_config), ) .await; state = stepped; let scan = &mut state.scan; + // A cancelled walk gives up its upload by dropping it, which deletes what it wrote, + // so this is an ordinary return. let Some(outcome) = outcome else { return (state, None); }; @@ -271,54 +278,81 @@ impl OffloadedPeek { let phases = scan.phases(); metrics.observe_error_phase(&phases); metrics.observe_ok_phase(&phases); - // Onto the blocking pool for the same reason a slice goes there: building - // the answer sorts and copies the whole row set, and a finishing that - // carries an order accumulates the whole result before it can. - let answer_metrics = metrics.clone(); - let order_by = Arc::clone(&order_by); - let response = mz_ore::task::spawn_blocking( - || "peek_offload::answer", - move || answer_metrics.rows_response(rows, &order_by), - ) - .await; - return (state, Some(OffloadOutcome::Answered(response))); + let response = match upload { + // Onto the blocking pool for the same reason the walk runs there: + // building the answer sorts and copies the whole row set, and a + // finishing that carries an order never reaches the stash, so it + // accumulates the whole result before it can. + None => { + let answer_metrics = metrics.clone(); + let order_by = Arc::clone(&order_by); + mz_ore::task::spawn_blocking( + || "peek_offload::answer", + move || answer_metrics.rows_response(rows, &order_by), + ) + .await + } + Some(upload) => stashed_answer(peek_uuid, upload, rows).await, + }; + return (state, Some(response)); } ScanOutcome::Finished(Err(error)) => { metrics.observe_error_phase(&scan.phases()); - return ( - state, - Some(OffloadOutcome::Answered(PeekResponse::Error(error))), - ); + // The peek is answered with the error rather than with the rows written so + // far, so nothing will ever read them, and the upload's drop deletes them. + return (state, Some(PeekResponse::Error(error))); } - // A suspension holding a full batch is not one this walk can resume: the scan - // stops advancing until a driver that writes rows takes the batch, and this one - // cannot. The accumulated rows are dropped, which is sound because the stash walks - // the ok trace again from the trace bundle. - ScanOutcome::Suspended if scan.batch_ready() => { - metrics.observe_error_phase(&scan.phases()); - // The stash answers from the ok trace alone, so a peek diverted with its - // error trace half-read would return rows where it owes an error. Only the ok - // walk accumulates, so a full batch implies the error walk is over, and the - // guard states that rather than assuming it, as the inline driver does. - if !scan.error_trace_clean() { - soft_panic_or_log!( - "peek on {} suspended before its error trace was read out", - scan.target_id() - ); - return ( - state, - Some(OffloadOutcome::Answered(PeekResponse::Error( - PeekError::unstructured( - "peek suspended before its error trace was read out", - ), - ))), - ); + ScanOutcome::Suspended => { + // `step_until_blocked` returns a suspension only with a batch, and a scan that + // has one makes no progress until it is taken. + if let Some(batch) = scan.take_batch() { + let Some(stash) = &stash else { + // Only an eligible scan fills a batch, and eligibility is what gave + // this walk its target, so this is a defect at the offload site. + // Answered as well as logged, because the walk has stopped either way. + soft_panic_or_log!( + "offloaded walk holds a batch and has no stash target" + ); + metrics.observe_error_phase(&scan.phases()); + return ( + state, + Some(PeekResponse::Error(PeekError::unstructured( + "internal error: offloaded peek walk has nowhere to write its rows", + ))), + ); + }; + + let open = match &mut upload { + Some(open) => open, + none => match stash.open(config.batch_max_runs.get()).await { + Ok(opened) => none.insert(opened), + Err(error) => { + warn!(%peek_uuid, %error, "peek stash failed to open a shard"); + metrics.observe_error_phase(&scan.phases()); + return ( + state, + Some(PeekResponse::Error(PeekError::unstructured( + error.to_string(), + ))), + ); + } + }, + }; + + if let Err(error) = open.push(batch).await { + // Persist rejects only a batch handed to it wrongly, so this is a + // defect in the upload rather than a blip. + warn!(%peek_uuid, %error, "peek stash rejected a batch"); + metrics.observe_error_phase(&scan.phases()); + return ( + state, + Some(PeekResponse::Error(PeekError::unstructured( + error.to_string(), + ))), + ); + } } - return (state, Some(OffloadOutcome::NeedsStash)); } - // `step_until_blocked` returns a suspension only with a batch, so this arm is not - // reached, and going back to the pool is the right thing if it ever is. - ScanOutcome::Suspended => {} } } } @@ -335,7 +369,7 @@ struct WalkState { _permit: OwnedSemaphorePermit, /// The sending end of the peek's result channel. Its receiver is dropped by cancellation and /// by nothing else, so a closed channel is the cancellation signal. - result_tx: oneshot::Sender<(OffloadOutcome, Duration)>, + result_tx: oneshot::Sender<(PeekResponse, Duration)>, } impl WalkState { @@ -352,7 +386,7 @@ impl WalkState { } // A granularity of zero would spend no fuel, and a scan stepped with no fuel makes no - // progress, so the walk would spin without ever reaching an outcome. + // progress, so the walk would spin without ever reaching an answer. let mut fuel = config.yield_granularity.get().max(1); let row_iteration_limit = config.row_iteration.current_limit(); @@ -365,5 +399,23 @@ impl WalkState { } } +/// Finishes `upload` with `inline_rows`, the rows the walk still held when it ended, and builds +/// the response that names the stashed batch. +async fn stashed_answer( + peek_uuid: Uuid, + upload: StashUpload, + inline_rows: RowBatch, +) -> PeekResponse { + match upload.finish(inline_rows).await { + Ok(response) => response, + // A defect in the upload rather than a blip, like a rejected push. The parts stay behind, + // see `StashUpload::finish`. + Err(error) => { + warn!(%peek_uuid, %error, "peek stash failed to finish a batch"); + PeekResponse::Error(PeekError::unstructured(error.to_string())) + } + } +} + #[cfg(test)] mod tests; diff --git a/src/compute/src/compute_state/peek_offload/tests.rs b/src/compute/src/compute_state/peek_offload/tests.rs index de84be261c275..b81569ab88529 100644 --- a/src/compute/src/compute_state/peek_offload/tests.rs +++ b/src/compute/src/compute_state/peek_offload/tests.rs @@ -11,17 +11,24 @@ use std::num::NonZeroUsize; +use mz_compute_types::dyncfgs::{ENABLE_PEEK_ROW_ITERATION_LIMIT, PEEK_ROW_ITERATION_LIMIT}; use mz_dyncfg::ConfigUpdates; use mz_expr::RowSetFinishing; use mz_expr::row::RowCollection; +use mz_ore::cast::CastLossy; use mz_ore::metrics::MetricsRegistry; -use mz_repr::Row; +use mz_ore::num::NonNeg; +use mz_persist_client::cache::PersistClientCache; +use mz_persist_types::PersistLocation; +use mz_repr::{IntoRowIterator, Row, RowIterator, RowRef}; +use crate::arrangement::manager::TraceBundle; use crate::compute_state::index_peek_tests::{ cancelling_errors, index_peek, ok_row, rows_answer, trace_bundle, trivial_finishing, wide_ok_rows, }; use crate::compute_state::peek_scan::PeekScan; +use crate::compute_state::peek_stash::tests::{CountedBlob, stashed_rows}; use crate::metrics::{ComputeMetrics, WorkerMetrics}; use crate::server::ComputeRuntimeRole; @@ -59,9 +66,46 @@ fn worker_metrics() -> WorkerMetrics { ComputeMetrics::register_with(&MetricsRegistry::new(), ComputeRuntimeRole::Solo).for_worker(0) } -/// The size the scan accounts a single-column row at. -fn row_size() -> usize { - ok_row(0).byte_len() + size_of::() +/// The size the scan accounts the widest row of `keys` at. +/// +/// Taken as the widest rather than as one row's, because a `UInt64` packs into the fewest bytes +/// that hold its value: a threshold built from the widest row is one that the same number of +/// rows crosses in every batch of a walk, whatever values those rows carry. +/// +/// The rows these tests walk are [`wide_ok_rows`], which carry the `UInt64` the peek's result +/// description declares, and that is the schema the stash writes its batch under. +fn widest_row_size(keys: &[Row]) -> usize { + keys.iter() + .map(crate::compute_state::peek_scan::entry_byte_len) + .max() + .expect("an index a peek walks holds rows") +} + +/// The stash `peek` writes to, over the in-memory location `clients` opens. +fn stash_target(peek: &Peek, clients: &Arc) -> StashTarget { + StashTarget::new(peek, Arc::clone(clients), PersistLocation::new_in_mem()) +} + +/// The batches and the inline rows of a stashed response, as one sorted row set. +/// +/// A peek's answer is both halves together, so a test that compared only one of them would +/// pass on a driver that wrote the rows to the wrong half. +async fn stashed_answer_rows( + clients: &PersistClientCache, + response: PeekResponse, +) -> (Vec, Vec) { + let PeekResponse::Stashed(stashed) = response else { + panic!("a walk that reached the stash answers with a stashed response, not {response:?}") + }; + + let mut inline: Vec = stashed + .inline_rows + .iter() + .flat_map(|rows| rows.clone().into_row_iter().map(RowRef::to_owned)) + .collect(); + inline.sort(); + + (stashed_rows(clients, *stashed).await, inline) } /// Opens a scan of `peek` over `bundle`, the way the peek path opens one. @@ -88,18 +132,42 @@ fn open( /// The configuration an offloaded walk reads, with the yield granularity set to /// `yield_granularity` so a test can choose how many slices a walk is cut into. fn offload_config(yield_granularity: usize) -> OffloadConfig { + offload_config_with(yield_granularity, |_updates| ()) +} + +/// The same, with `configure` applied on top, as `UpdateConfiguration` applies a change. +fn offload_config_with( + yield_granularity: usize, + configure: impl FnOnce(&mut ConfigUpdates), +) -> OffloadConfig { let config = mz_dyncfgs::all_dyncfgs(); let mut updates = ConfigUpdates::default(); updates.add(&INDEX_PEEK_YIELD_GRANULARITY, yield_granularity); + configure(&mut updates); updates.apply(&config); OffloadConfig::new(&config) } +/// The configuration a walk whose blob traffic a test counts reads. +/// +/// A builder over the run limit merges its runs, which writes parts of its own and leaves the +/// parts it merged from behind, so a test counting what a walk wrote raises the limit past the +/// runs the walk produces. +fn counted_blob_config(yield_granularity: usize) -> OffloadConfig { + offload_config_with(yield_granularity, |updates| { + updates.add(&PEEK_RESPONSE_STASH_BATCH_MAX_RUNS, NO_RUN_MERGING); + }) +} + +/// A run limit above the parts any walk here writes, which is at most one per key of the +/// longest trace these tests hold. +const NO_RUN_MERGING: usize = 8_000; + /// A finishing that orders the peek's one column descending, which is the reverse of the order /// the trace holds its keys in. /// /// A peek carrying an order is never eligible for the peek stash, because `is_streamable` -/// requires an empty `order_by`, so an offloaded walk of one answers rather than handing back. +/// requires an empty `order_by`, so an offloaded walk of one answers with its rows. fn descending_finishing() -> RowSetFinishing { let mut finishing = trivial_finishing(); finishing.order_by = vec![ColumnOrder { @@ -125,37 +193,16 @@ fn ordered_rows_answer(rows: impl IntoIterator) -> PeekResponse { PeekResponse::Rows(vec![builder.build()]) } -/// What an offloaded walk handed back, in a form a test can compare whole. +/// Runs the runtime until `offloaded`'s walk answers the peek, and reports the answer. /// -/// Mirrors [`OffloadOutcome`], which carries no comparison of its own because nothing on the -/// peek path compares one. -#[derive(Debug, PartialEq)] -enum HandBack { - Answered(PeekResponse), - NeedsStash, -} - -impl From for HandBack { - fn from(outcome: OffloadOutcome) -> Self { - match outcome { - OffloadOutcome::Answered(response) => HandBack::Answered(response), - OffloadOutcome::NeedsStash => HandBack::NeedsStash, - } - } -} - -/// Runs the runtime until `offloaded`'s walk hands something back, and reports what. -/// -/// Bounded, so a walk that yields without ever reaching an outcome fails here rather than -/// hanging the suite. That is the failure a scan holding a full batch produces, which is the -/// case this driver's batch-ready arm exists to avoid. -async fn hand_back(offloaded: &mut OffloadedPeek) -> HandBack { +/// Bounded, so a walk that never reaches an answer fails here rather than hanging the suite. +async fn answer(offloaded: &mut OffloadedPeek) -> PeekResponse { let Ok(handed_back) = tokio::time::timeout(DRIVE_BOUND, &mut offloaded.result).await else { - panic!("the offloaded walk handed nothing back within {DRIVE_BOUND:?}"); + panic!("the offloaded walk did not answer within {DRIVE_BOUND:?}"); }; - let (outcome, _elapsed) = - handed_back.expect("the offloaded walk ended without handing anything back"); - HandBack::from(outcome) + let (response, _elapsed) = + handed_back.expect("the offloaded walk ended without answering the peek"); + response } /// Runs the runtime until `condition` holds, where `what` names what the test is waiting for. @@ -184,21 +231,17 @@ async fn an_offloaded_walk_finishes_the_answer_the_inline_slice_started() { let mut bundle = trace_bundle(&keys, cancelling_errors(2)); let mut scan = open(&mut bundle, &peek, None); - // Two positions leave the walk suspended with nothing to hand over, which is the state - // the worker offloads and the only one it offloads. + // Two positions leave the walk suspended for want of fuel, with the rest of the trace + // ahead of it. let mut fuel = 2; assert_eq!(scan.step(None, &mut fuel), ScanOutcome::Suspended); - assert!( - !scan.batch_ready(), - "a scan holding a full batch is diverted rather than offloaded" - ); let metrics = worker_metrics(); let permits = PeekPermits::new(1); let mut offloaded = OffloadedPeek::start( peek.clone(), - bundle, scan, + None, &permits, offload_config(*INDEX_PEEK_YIELD_GRANULARITY.default()), PeekWalkMetrics::new(&metrics), @@ -206,8 +249,8 @@ async fn an_offloaded_walk_finishes_the_answer_the_inline_slice_started() { ); assert_eq!( - hand_back(&mut offloaded).await, - HandBack::Answered(rows_answer((0..6).map(ok_row))), + answer(&mut offloaded).await, + rows_answer((0..6).map(ok_row)), ); assert_eq!( metrics.index_peek_walks_offloaded.get(), @@ -240,8 +283,8 @@ async fn an_offloaded_walk_answers_in_the_order_the_peek_asked_for() { let permits = PeekPermits::new(1); let mut offloaded = OffloadedPeek::start( peek.clone(), - bundle, scan, + None, &permits, offload_config(*INDEX_PEEK_YIELD_GRANULARITY.default()), PeekWalkMetrics::new(&metrics), @@ -249,51 +292,121 @@ async fn an_offloaded_walk_answers_in_the_order_the_peek_asked_for() { ); assert_eq!( - hand_back(&mut offloaded).await, - HandBack::Answered(ordered_rows_answer((0..6).rev().map(ok_row))), + answer(&mut offloaded).await, + ordered_rows_answer((0..6).rev().map(ok_row)), ); } -/// An offloaded walk whose accumulated rows grow into a full batch hands the peek back rather -/// than stepping a scan that cannot advance. +/// An offloaded walk whose accumulated rows grow into a full batch writes the batch to the stash +/// and carries on from where it stopped, so one walk produces the whole answer. /// -/// A scan holding a full batch spends no fuel and moves no cursor when stepped, and this -/// driver has nowhere to write the batch, so a driver that treated the suspension as resumable -/// would yield forever without ever reaching an outcome. The bound in [`hand_back`] is what -/// turns that into a failure rather than a hang. +/// The rows the walk was still holding when it ended travel with the response rather than +/// through a batch of their own, and the two halves together are every row the peek owes. The +/// count of stashed rows is asserted as well, because a driver that wrote the tail to both +/// halves would still answer with the right set. +/// +/// What says the trace was walked once is the count of cursor positions the walk reported, not +/// the rows it answered with: a second walk of the same trace produces the same rows, so an +/// answer-only test passes against the very regression it is meant to catch. #[mz_ore::test(tokio::test)] -async fn an_offloaded_walk_that_fills_a_batch_hands_back_rather_than_spinning() { - let keys: Vec = (0..6).map(ok_row).collect(); +async fn an_offloaded_walk_writes_full_batches_to_the_stash_and_walks_on() { + let keys = wide_ok_rows(8); let peek = index_peek(trivial_finishing(), None); let mut bundle = trace_bundle(&keys, cancelling_errors(0)); - // Two rows fit under the threshold and the third crosses it, so the slice below offloads - // a scan with room left and the offloaded walk is the one that fills the batch. - let mut scan = open(&mut bundle, &peek, Some(2 * row_size())); + // Two rows fit under the threshold and the third crosses it, so the walk fills a batch + // twice over and holds the last two rows when the trace runs out. + let mut scan = open(&mut bundle, &peek, Some(2 * widest_row_size(&keys))); let mut fuel = 2; assert_eq!(scan.step(None, &mut fuel), ScanOutcome::Suspended); - assert!( - !scan.batch_ready(), - "the inline slice must offload a scan that still has room" - ); let metrics = worker_metrics(); let permits = PeekPermits::new(1); + let clients = Arc::new(PersistClientCache::new_no_metrics()); let mut offloaded = OffloadedPeek::start( peek.clone(), - bundle, scan, + Some(stash_target(&peek, &clients)), &permits, offload_config(*INDEX_PEEK_YIELD_GRANULARITY.default()), PeekWalkMetrics::new(&metrics), std::thread::current(), ); - assert_eq!(hand_back(&mut offloaded).await, HandBack::NeedsStash); + let response = answer(&mut offloaded).await; + let PeekResponse::Stashed(stashed) = &response else { + panic!("a walk that reached the stash answers with a handle, not {response:?}"); + }; + assert_eq!( + stashed.num_rows_batches, 6, + "the stashed row count covers the batches alone" + ); + + let (batched, inline) = stashed_answer_rows(&clients, response).await; + assert_eq!(batched, keys[..6]); + assert_eq!(inline, keys[6..]); assert_eq!( metrics.index_peek_walks_offloaded.get(), 1, - "a hand-back is a terminal outcome of the offloaded walk" + "one walk produced the whole answer" + ); + assert_eq!( + metrics.index_peek_stashed_total.get(), + 1, + "the walk answered from the stash" + ); + assert_eq!( + metrics.index_peek_row_iteration_rows.get_sample_count(), + 1, + "one walk reports one ok phase, whatever it was cut into" + ); + assert_eq!( + metrics.index_peek_row_iteration_rows.get_sample_sum(), + f64::cast_lossy(keys.len()), + "the walk evaluated each cursor position of the trace once" + ); +} + +/// An offloaded walk stops once the stash holds every row the peek's finishing can use, rather +/// than walking the rest of a trace whose rows no answer could contain. +#[mz_ore::test(tokio::test)] +async fn an_offloaded_walk_stops_where_the_finishing_has_what_it_needs() { + let keys = wide_ok_rows(8); + let mut finishing = trivial_finishing(); + finishing.limit = Some(NonNeg::try_from(2).expect("non-negative")); + let peek = index_peek(finishing, None); + let mut bundle = trace_bundle(&keys, cancelling_errors(0)); + // Crossed by the second row, so the first batch already holds what the limit asks for. + let mut scan = open(&mut bundle, &peek, Some(widest_row_size(&keys))); + + let mut fuel = 1; + assert_eq!(scan.step(None, &mut fuel), ScanOutcome::Suspended); + + let metrics = worker_metrics(); + let permits = PeekPermits::new(1); + let clients = Arc::new(PersistClientCache::new_no_metrics()); + let mut offloaded = OffloadedPeek::start( + peek.clone(), + scan, + Some(stash_target(&peek, &clients)), + &permits, + offload_config(*INDEX_PEEK_YIELD_GRANULARITY.default()), + PeekWalkMetrics::new(&metrics), + std::thread::current(), + ); + + let (batched, inline) = stashed_answer_rows(&clients, answer(&mut offloaded).await).await; + assert_eq!(batched, keys[..2], "the walk wrote what the limit asks for"); + assert_eq!( + inline, + Vec::::new(), + "the batch that satisfied the stash was everything the walk held" + ); + assert_eq!( + metrics.index_peek_row_iteration_seconds.get_sample_count(), + 1, + "a walk that answered the peek reports its ok phase, whether it reached the end of the \ + trace or the end of what the finishing can use" ); } @@ -324,8 +437,8 @@ async fn a_walk_cancelled_while_queued_never_takes_a_permit() { let offloaded = OffloadedPeek::start( peek.clone(), - bundle, scan, + None, &permits, offload_config(*INDEX_PEEK_YIELD_GRANULARITY.default()), PeekWalkMetrics::new(&metrics), @@ -402,8 +515,9 @@ async fn a_walk_cancelled_while_running_reports_no_outcome() { _permit: permit, result_tx, }; - let (_state, outcome) = OffloadedPeek::walk(state, &config, &walk_metrics, order_by).await; - outcome.is_some() + let (_state, response) = + OffloadedPeek::walk(state, Uuid::nil(), None, &config, &walk_metrics, order_by).await; + response.is_some() }); tokio::time::sleep(DRIVE_PAUSE).await; @@ -456,8 +570,8 @@ async fn a_walk_waits_for_a_permit_held_elsewhere() { let mut offloaded = OffloadedPeek::start( peek.clone(), - bundle, scan, + None, &permits, offload_config(*INDEX_PEEK_YIELD_GRANULARITY.default()), PeekWalkMetrics::new(&metrics), @@ -480,8 +594,8 @@ async fn a_walk_waits_for_a_permit_held_elsewhere() { drop(held); assert_eq!( - hand_back(&mut offloaded).await, - HandBack::Answered(rows_answer((0..6).map(ok_row))), + answer(&mut offloaded).await, + rows_answer((0..6).map(ok_row)) ); assert_eq!( metrics.index_peek_permit_wait_seconds.get_sample_count(), @@ -521,8 +635,8 @@ async fn dropping_an_offloaded_peek_cancels_its_walk() { let offloaded = OffloadedPeek::start( peek.clone(), - bundle, scan, + None, &permits, offload_config(1), PeekWalkMetrics::new(&metrics), @@ -577,6 +691,188 @@ fn the_permit_fraction_scales_with_the_worker_count() { ); } +/// A walk that is aborted after its upload has written parts leaves nothing in blob storage. +/// +/// This is the whole of the cleanup a cancellation gets. Cancelling a peek removes the pending +/// entry, which drops the handle to this task and aborts it, and an aborted task is dropped +/// rather than polled again, so nothing the walk would have called runs. What deletes the parts +/// is `Drop for StashUpload` spawning the deletion onto the runtime the upload captured when it +/// opened, and this is the one test that drives an abort into that drop into that spawn. +#[mz_ore::test(tokio::test)] +async fn an_aborted_walk_deletes_the_parts_its_upload_wrote() { + let keys = wide_ok_rows(LONG_WALK_KEYS); + let peek = index_peek(trivial_finishing(), None); + let mut bundle = trace_bundle(&keys, cancelling_errors(0)); + // Crossed by the third row, so the walk opens an upload a few positions in and keeps + // feeding it for the rest of a trace it cannot reach the end of before it is aborted. + let scan = open(&mut bundle, &peek, Some(2 * widest_row_size(&keys))); + + let metrics = worker_metrics(); + let permits = PeekPermits::new(1); + let blob = CountedBlob::new(); + let offloaded = OffloadedPeek::start( + peek.clone(), + scan, + Some(stash_target(&peek, blob.clients())), + &permits, + // One position per slice, so the walk is still far from the end of the trace when the + // first part reaches blob storage. + counted_blob_config(1), + PeekWalkMetrics::new(&metrics), + std::thread::current(), + ); + + blob.wait_until_something_is_written("a walk past the stash threshold") + .await; + + // Dropping the whole entry is what a cancellation does, and it is the abort rather than + // any signal the walk observes. + drop(offloaded); + + blob.wait_until_nothing_is_left("an aborted walk").await; + assert_eq!( + blob.deletes_of_nothing(), + 0, + "the deletes must name the keys the upload wrote" + ); + assert_eq!( + metrics.index_peek_walks_offloaded.get(), + 0, + "an aborted walk reaches no outcome and counts on neither substrate" + ); + assert_eq!( + metrics.index_peek_stashed_total.get(), + 0, + "an aborted walk answers from no stash" + ); +} + +/// A walk that observes its cancellation while feeding an upload gives the upload up, which +/// deletes the parts it had written, and reports no outcome. +/// +/// The branch that sees the cancellation has to give the upload up rather than return past it. +#[mz_ore::test(tokio::test)] +async fn a_walk_cancelled_while_uploading_deletes_what_it_wrote() { + let keys = wide_ok_rows(LONG_WALK_KEYS); + let peek = index_peek(trivial_finishing(), None); + let mut bundle = trace_bundle(&keys, cancelling_errors(0)); + let scan = open(&mut bundle, &peek, Some(2 * widest_row_size(&keys))); + + let metrics = worker_metrics(); + let walk_metrics = PeekWalkMetrics::new(&metrics); + let permits = PeekPermits::new(1); + let semaphore = permits.resize(1.0); + let permit = Arc::clone(&semaphore) + .try_acquire_owned() + .expect("a permit is free"); + + let blob = CountedBlob::new(); + let stash = stash_target(&peek, blob.clients()); + let (result_tx, result_rx) = oneshot::channel(); + let config = counted_blob_config(1); + let order_by: Arc<[ColumnOrder]> = peek.finishing.order_by.as_slice().into(); + let walk = mz_ore::task::spawn(|| "peek_offload_test::walk", async move { + let state = WalkState { + scan, + _permit: permit, + result_tx, + }; + let (_state, response) = OffloadedPeek::walk( + state, + Uuid::nil(), + Some(stash), + &config, + &walk_metrics, + order_by, + ) + .await; + response.is_some() + }); + + blob.wait_until_something_is_written("a walk past the stash threshold") + .await; + assert!( + !walk.is_finished(), + "the walk must still be under way when it is cancelled" + ); + + drop(result_rx); + + wait_until(|| walk.is_finished(), "the cancelled walk stopping").await; + assert_eq!(walk.await, false, "a cancelled walk reports no outcome"); + + blob.wait_until_nothing_is_left("a cancelled walk").await; + assert_eq!( + blob.deletes_of_nothing(), + 0, + "the deletes must name the keys the upload wrote" + ); + assert_eq!( + semaphore.available_permits(), + 1, + "a cancelled walk releases the permit that admitted it" + ); +} + +/// A walk that fails after its upload has written parts answers with the failure and deletes +/// those parts, because nothing will ever read them. +/// +/// A failure reachable only once rows are already in the stash is what this driver created: +/// the walk that produces the rows is now the walk that writes them, so its own error arm has +/// an upload to answer for. +#[mz_ore::test(tokio::test)] +async fn a_walk_that_fails_after_writing_deletes_what_it_wrote() { + let keys = wide_ok_rows(20); + let peek = index_peek(trivial_finishing(), None); + let mut bundle = trace_bundle(&keys, cancelling_errors(0)); + // Crossed by the third row, so two batches reach the stash before the limit below trips. + let scan = open(&mut bundle, &peek, Some(2 * widest_row_size(&keys))); + + // Past the rows two batches hold and short of the trace, so the walk fails with an upload + // open and parts written. + let limit = 8; + + let metrics = worker_metrics(); + let permits = PeekPermits::new(1); + let blob = CountedBlob::new(); + let mut offloaded = OffloadedPeek::start( + peek.clone(), + scan, + Some(stash_target(&peek, blob.clients())), + &permits, + offload_config_with(1, |updates| { + updates.add(&PEEK_RESPONSE_STASH_BATCH_MAX_RUNS, NO_RUN_MERGING); + updates.add(&ENABLE_PEEK_ROW_ITERATION_LIMIT, true); + updates.add(&PEEK_ROW_ITERATION_LIMIT, limit); + }), + PeekWalkMetrics::new(&metrics), + std::thread::current(), + ); + + assert_eq!( + answer(&mut offloaded).await, + PeekResponse::Error(PeekError::RowIterationLimitExceeded { limit }), + "the peek is answered with the failure rather than with the rows already written", + ); + + blob.wait_until_nothing_is_left("a failed walk").await; + assert_eq!( + blob.deletes_of_nothing(), + 0, + "the deletes must name the keys the upload wrote" + ); + assert_eq!( + metrics.index_peek_stashed_total.get(), + 0, + "a walk answered with an error answered from no stash" + ); + assert_eq!( + metrics.index_peek_walks_offloaded.get(), + 1, + "a failure is a terminal outcome, so the driver that reached it counts the walk" + ); +} + #[mz_ore::test] fn permits_resize_around_the_walks_holding_them() { let permits = PeekPermits::new(4); diff --git a/src/compute/src/compute_state/peek_scan.rs b/src/compute/src/compute_state/peek_scan.rs index 8b4d98eae98d6..695cf2cb4a5b9 100644 --- a/src/compute/src/compute_state/peek_scan.rs +++ b/src/compute/src/compute_state/peek_scan.rs @@ -41,7 +41,7 @@ pub(super) type IndexPeekScan = PeekScan< pub(super) type RowBatch = Vec<(Row, NonZeroI64)>; /// The byte size of a row's count, as an answer built from a [`RowBatch`] stores it. -const COUNT_BYTE_SIZE: usize = size_of::(); +pub(super) const COUNT_BYTE_SIZE: usize = size_of::(); /// The byte size of a row's offset into the answer's packed row data. const OFFSET_BYTE_SIZE: usize = size_of::(); @@ -89,8 +89,8 @@ pub(super) enum ScanOutcome { /// The scan retains what it accumulated. A driver that can write rows collects them through /// [`PeekScan::take_batch`], and one that cannot is never handed rows it would have to drop. /// - /// A scan holding a full batch makes no progress when stepped, so a driver that declines a - /// batch has to ask [`PeekScan::batch_ready`] before it steps again or it spins forever. + /// A driver must take every batch it is offered. A scan holding one makes no progress when + /// stepped, so a driver that steps without taking spins forever. Suspended, /// The walk is over. `Ok` carries the rows accumulated since the last batch was taken, which /// together with the batches already taken are the peek's answer. `Err` is the peek's answer @@ -139,7 +139,13 @@ where results: RowBatch, /// The byte size of `results`, as an answer built from them would store them. total_size: usize, - /// The ceiling on what the scan may hold, above which the peek fails. + /// The byte size of the batches already handed to a driver, which `total_size` no longer + /// counts. Together the two are the size of the answer the peek is building. + handed_off_size: usize, + /// The rows the answer holds so far, batches already handed to a driver included, counted in + /// copies because that is what the finishing's limit counts. Thinning takes its drops back off. + answer_rows: u64, + /// The ceiling on the whole answer, above which the peek fails. max_result_size: usize, /// Whether this peek may divert its rows to the peek stash. peek_stash_eligible: bool, @@ -150,7 +156,7 @@ where /// exactly this many rows, just at least those that would have been returned. max_results: Option, /// Orders the rows that thinning keeps. `None` when the finishing imposes no ordering, in - /// which case thinning keeps any `max_results` of them. + /// which case the walk ends at the limit rather than thinning at all. comparator: Option, /// Worker time the error walk spent, summed over the slices it was cut into. pub(super) error_scan_time: Duration, @@ -217,6 +223,8 @@ where ended: None, results: Vec::new(), total_size: 0, + handed_off_size: 0, + answer_rows: 0, max_result_size: usize::cast_from(max_result_size), peek_stash_eligible, peek_stash_threshold_bytes, @@ -283,11 +291,6 @@ where Some(self.take_results()) } - /// The collection this scan reads. - pub(super) fn target_id(&self) -> GlobalId { - self.target_id - } - /// The number of cursor positions the ok walk has evaluated. pub(super) fn rows_processed(&self) -> usize { self.oks.rows_processed() @@ -314,11 +317,19 @@ where matches!(self.error_phase, ErrorPhase::Clean) } + /// Whether this scan may divert rows to the peek stash, and so whether it ever fills a batch. + /// + /// A driver reads this rather than deciding eligibility again, so it cannot end up holding a + /// batch it has nowhere to write. A scan holding an untaken batch makes no progress. + pub(super) fn stash_eligible(&self) -> bool { + self.peek_stash_eligible + } + /// Whether the accumulated rows have grown past what this peek may answer with inline, which /// is when [`PeekScan::take_batch`] hands them over. /// - /// This is how a driver tells a [`ScanOutcome::Suspended`] it can resume from one it cannot: - /// a scan holding a full batch stays where it stands until the batch is taken. + /// A scan whose batch is ready stays where it stands until the batch is taken, so this is also + /// whether stepping the scan again can make progress. pub(super) fn batch_ready(&self) -> bool { self.peek_stash_eligible && self.total_size > self.peek_stash_threshold_bytes } @@ -328,6 +339,9 @@ where /// Every path that hands rows out goes through here, so `total_size` stays an account of /// `results`. fn take_results(&mut self) -> RowBatch { + // Carried rather than dropped, because `max_result_size` bounds the answer the peek + // returns and not the prefix the scan happens to be holding. + self.handed_off_size = self.handed_off_size.saturating_add(self.total_size); self.total_size = 0; mem::take(&mut self.results) } @@ -378,12 +392,29 @@ where } } + /// Whether the answer holds every row the peek's finishing can use. + /// + /// Only ever true without an ordering: an ordered finishing ranks rows against the whole + /// trace, so no prefix of the walk satisfies it. + fn finishing_satisfied(&self) -> bool { + self.comparator.is_none() + && self + .max_results + .is_some_and(|max_results| self.answer_rows >= u64::cast_from(max_results)) + } + /// Advances the walk over the ok trace, accumulating the rows it produces. fn step_ok_phase( &mut self, row_iteration_limit: Option, fuel: &mut usize, ) -> ScanOutcome { + // Ahead of the batch guard, so a scan whose last batch completed the answer ends here + // rather than walking one more row into a batch of its own. + if self.finishing_satisfied() { + return ScanOutcome::Finished(Ok(self.take_results())); + } + // A scan holding a full batch stays where it is until the batch is taken, so the bound on // what one scan retains is the scan's own rather than a rule each driver keeps. Past the // stash threshold the result-size ceiling no longer bounds that growth either. @@ -406,15 +437,19 @@ where self.total_size = self.total_size.saturating_add(entry_byte_len(&row)); let batch_ready = self.batch_ready(); - // Rows bound for the stash are answered by a handle rather than by themselves, so the - // ceiling on an inline answer does not apply to a prefix that has grown past the - // stash threshold. - if !batch_ready && self.total_size > self.max_result_size { + // Measured against everything the answer will contain, the batches already handed off + // included, because a stashed answer reaches the client like any other. Checking only + // the retained prefix would leave a stashed answer unbounded, since the prefix is + // handed away and reset well below the ceiling. + let answer_size = self.handed_off_size.saturating_add(self.total_size); + if answer_size > self.max_result_size { break self.fail(PeekError::ResultExceedsMaxSize { max_result_size: self.max_result_size, }); } + // Positive here: the walk errors on a negative multiplicity rather than yielding it. + self.answer_rows = self.answer_rows.saturating_add(copies.get().unsigned_abs()); self.results.push((row, copies)); // Ahead of thinning, so that a row which both fills a batch and completes a thinned @@ -423,6 +458,10 @@ where break ScanOutcome::Suspended; } + if self.finishing_satisfied() { + break ScanOutcome::Finished(Ok(self.take_results())); + } + if let Some(outcome) = self.thin() { break outcome; } @@ -433,13 +472,16 @@ where outcome } - /// Thins the accumulated rows down towards the ones the peek's finishing needs, once the scan - /// holds many more than that. + /// Thins the accumulated rows down to the ones an ordered finishing ranks first, once the scan + /// holds many more than it needs. /// - /// Returns the outcome that ends the scan if thinning has produced every row the finishing - /// can use. + /// Does nothing without an ordering: such a scan ends at [`PeekScan::finishing_satisfied`] + /// instead of accumulating past its limit. fn thin(&mut self) -> Option { let max_results = self.max_results?; + let Some(comparator) = &self.comparator else { + return None; + }; // We use a threshold twice what we intend, to amortize the work across all of the // insertions. We could tighten this, but it works for the moment. @@ -457,13 +499,6 @@ where return None; } - let Some(comparator) = &self.comparator else { - // Without an ordering, any `max_results` rows answer the peek, so the rows in hand are - // an answer and the rest of the trace does not have to be walked. - self.results.truncate(max_results); - return Some(ScanOutcome::Finished(Ok(self.take_results()))); - }; - // We can sort `results` and then truncate to `max_results`. This has an effect similar to // a priority queue, without its interactive dequeueing properties. // TODO: Had we left these as `Vec` we would avoid the unpacking; we should consider @@ -477,12 +512,17 @@ where self.result_sort_time += sort_start.elapsed(); let dropped = self.results.drain(max_results..); - let dropped_size = dropped - .into_iter() - .fold(0, |acc: usize, (row, _count): (Row, _)| { - acc.saturating_add(entry_byte_len(&row)) - }); + let (dropped_size, dropped_rows) = dropped.into_iter().fold( + (0usize, 0u64), + |(size, rows), (row, count): (Row, NonZeroI64)| { + ( + size.saturating_add(entry_byte_len(&row)), + rows.saturating_add(count.get().unsigned_abs()), + ) + }, + ); self.total_size = self.total_size.saturating_sub(dropped_size); + self.answer_rows = self.answer_rows.saturating_sub(dropped_rows); None } diff --git a/src/compute/src/compute_state/peek_scan/tests.rs b/src/compute/src/compute_state/peek_scan/tests.rs index f8befb83160ad..c244f44a307b0 100644 --- a/src/compute/src/compute_state/peek_scan/tests.rs +++ b/src/compute/src/compute_state/peek_scan/tests.rs @@ -64,11 +64,16 @@ fn expected(values: impl IntoIterator) -> RowBatch { .collect() } -/// A walk over an ok trace holding `keys`. +/// A walk over an ok trace holding `keys`, each once. fn ok_iterator(keys: &[Row]) -> PeekResultIterator { + ok_iterator_with_copies(keys, Diff::ONE) +} + +/// A walk over an ok trace holding `copies` of each of `keys`. +fn ok_iterator_with_copies(keys: &[Row], copies: Diff) -> PeekResultIterator { let updates: Vec<((Row, Row), Timestamp, Diff)> = keys .iter() - .map(|key| ((key.clone(), Row::default()), Timestamp::MIN, Diff::ONE)) + .map(|key| ((key.clone(), Row::default()), Timestamp::MIN, copies)) .collect(); let mut batcher = OrdValBatcher::::new(None, 0); batcher.push_into(updates); @@ -114,6 +119,8 @@ fn scan(error_phase: ErrorPhase, keys: &[Row]) -> PeekScan { ended: None, results: Vec::new(), total_size: 0, + handed_off_size: 0, + answer_rows: 0, max_result_size: usize::MAX, peek_stash_eligible: false, peek_stash_threshold_bytes: usize::MAX, @@ -381,10 +388,10 @@ fn take_batch_yields_nothing_without_the_stash() { ); } -/// A finishing that imposes no ordering is answered by any `max_results` rows, so thinning -/// truncates and the scan stops rather than walking the rest of the trace. +/// A finishing that imposes no ordering is answered by any `max_results` rows, so the scan ends +/// as soon as it holds that many rather than walking the rest of the trace. #[mz_ore::test] -fn unordered_thinning_truncates_and_ends_the_scan() { +fn unordered_thinning_ends_the_scan_at_the_limit() { let keys = rows(0..10); let mut subject = scan(ErrorPhase::Clean, &keys); subject.max_results = Some(2); @@ -396,8 +403,8 @@ fn unordered_thinning_truncates_and_ends_the_scan() { ); assert_eq!( subject.rows_processed(), - 4, - "the scan must stop at the threshold rather than walk the trace out" + 2, + "the scan must stop at the limit rather than walk the trace out" ); assert_eq!( subject.total_size, 0, @@ -405,6 +412,34 @@ fn unordered_thinning_truncates_and_ends_the_scan() { ); } +/// The finishing's limit counts a row as often as the answer holds it, so a trace of few rows at +/// a high multiplicity reaches the limit in fewer cursor positions than it has rows. +#[mz_ore::test] +fn the_limit_counts_copies_rather_than_distinct_rows() { + let keys = rows(0..10); + let mut subject = scan(ErrorPhase::Clean, &keys); + subject.oks = ok_iterator_with_copies(&keys, Diff::from(4)); + subject.max_results = Some(6); + + let mut fuel = usize::MAX; + let ScanOutcome::Finished(Ok(answer)) = subject.step(None, &mut fuel) else { + panic!("a scan that reaches its limit finishes"); + }; + assert_eq!( + answer, + vec![ + (row(0), NonZeroI64::new(4).expect("non-zero")), + (row(1), NonZeroI64::new(4).expect("non-zero")), + ], + "the row that crosses the limit is answered with whole" + ); + assert_eq!( + subject.rows_processed(), + 2, + "eight copies of two rows is past a limit of six, so the walk stops there" + ); +} + /// A finishing that imposes an ordering keeps the rows that order ranks first, thinning down /// to `max_results` each time it holds twice that many. #[mz_ore::test] @@ -453,24 +488,65 @@ fn accumulation_past_the_result_size_ceiling_fails_the_peek() { assert_eq!(subject.rows_processed(), 4); } -/// The result-size ceiling bounds an inline answer, which rows bound for the stash are not. -/// A scan whose batches are taken therefore walks the whole trace with a ceiling below what it -/// produces, rather than failing the peek. +/// The result-size ceiling bounds the whole answer, the batches already handed off included, +/// so a peek bound for the stash fails on it like any other. +/// +/// The ceiling here sits above the stash threshold, so the scan hands off batches and only +/// their sum reaches it. A ceiling compared against the retained prefix alone would never be +/// reached, because a prefix is handed away and reset well below it, and the peek would return +/// an answer of any size at all. /// -/// Driven with unbounded fuel, so every suspension is a full batch, and a scan that grew its -/// prefix past the threshold instead of stopping would fail on the ceiling. +/// Driven with unbounded fuel, so every suspension is a full batch. #[mz_ore::test] -fn a_prefix_bound_for_the_stash_is_not_bound_by_the_result_size_ceiling() { +fn a_stashed_answer_is_bound_by_the_result_size_ceiling_across_its_batches() { let keys = rows(0..8); let mut subject = scan(ErrorPhase::Clean, &keys); - subject.max_result_size = 3 * row_size(); + subject.max_result_size = 5 * row_size(); subject.peek_stash_eligible = true; subject.peek_stash_threshold_bytes = 2 * row_size(); let mut collected = RowBatch::new(); - let mut completed = false; + let mut failure = None; // Bounded so that a regression which restarts the ok walk on every resumption fails here // rather than spinning. + for _ in 0..RESUMPTION_BOUND { + let mut fuel = usize::MAX; + match subject.step(None, &mut fuel) { + ScanOutcome::Suspended => { + collected.extend(subject.take_batch().expect("a full batch")); + } + ScanOutcome::Finished(Ok(rest)) => { + collected.extend(rest); + break; + } + ScanOutcome::Finished(Err(error)) => { + failure = Some(error); + break; + } + } + } + + assert_eq!( + failure, + Some(PeekError::ResultExceedsMaxSize { + max_result_size: 5 * row_size() + }), + "a stashed answer past the ceiling must fail rather than answer, having collected {collected:?}" + ); +} + +/// A peek whose whole answer stays under the ceiling still reaches the stash, so the bound +/// across batches does not cost the stash the results it exists for. +#[mz_ore::test] +fn a_stashed_answer_under_the_ceiling_still_walks_the_whole_trace() { + let keys = rows(0..8); + let mut subject = scan(ErrorPhase::Clean, &keys); + subject.max_result_size = 100 * row_size(); + subject.peek_stash_eligible = true; + subject.peek_stash_threshold_bytes = 2 * row_size(); + + let mut collected = RowBatch::new(); + let mut completed = false; for _ in 0..RESUMPTION_BOUND { let mut fuel = usize::MAX; match subject.step(None, &mut fuel) { diff --git a/src/compute/src/compute_state/peek_stash.rs b/src/compute/src/compute_state/peek_stash.rs index f9ab8ce4b6d05..9891b552a6fad 100644 --- a/src/compute/src/compute_state/peek_stash.rs +++ b/src/compute/src/compute_state/peek_stash.rs @@ -6,138 +6,108 @@ //! For eligible peeks, we send the result back via the peek stash (aka persist //! blob), instead of inline in `ComputeResponse`. -use std::num::{NonZeroI64, NonZeroU64}; +use std::num::{NonZeroU64, NonZeroUsize}; use std::sync::Arc; -use std::time::{Duration, Instant}; use mz_compute_client::protocol::command::Peek; -use mz_compute_client::protocol::response::{PeekError, PeekResponse, StashedPeekResponse}; +use mz_compute_client::protocol::response::{PeekResponse, StashedPeekResponse}; use mz_expr::row::RowCollection; -use mz_ore::cast::CastFrom; -use mz_ore::task::AbortOnDropHandle; +use mz_ore::task::RuntimeExt; +use mz_persist::location::ExternalError; use mz_persist_client::Schemas; +use mz_persist_client::batch::{Added, Batch, BatchBuilder}; use mz_persist_client::cache::PersistClientCache; +use mz_persist_client::error::InvalidUsage; use mz_persist_types::codec_impls::UnitSchema; use mz_persist_types::{PersistLocation, ShardId}; -use mz_repr::{Diff, RelationDesc, Row, Timestamp}; +use mz_repr::{RelationDesc, Timestamp}; use mz_storage_types::sources::SourceData; use timely::progress::Antichain; +use tokio::runtime::Handle; use tokio::sync::oneshot; -use tracing::debug; +use tracing::warn; use uuid::Uuid; -use crate::arrangement::manager::{PaddedTrace, TraceBundle}; -use crate::compute_state::peek_result_iterator; -use crate::compute_state::peek_result_iterator::PeekResultIterator; -use crate::typedefs::RowRowAgent; +use crate::compute_state::peek_scan::{RowBatch, entry_byte_len}; -/// An async task that stashes a peek response in persist and yields a handle to -/// the batch in a [PeekResponse::Stashed]. +/// A failure that leaves an upload unable to answer the peek whose rows it holds. /// -/// Note that `StashingPeek` intentionally does not implement or derive -/// `Clone`, as each `StashingPeek` is meant to be dropped after it's -/// done or no longer needed. -pub struct StashingPeek { - pub peek: Peek, - /// Iterator for the results. The worker thread has to continually pump - /// results from this to the `rows_tx` channel. - peek_iterator: Option>>>, - /// Carries result rows from the worker thread, which owns the walk, to the async upload task. - /// The walk stays on the worker because the upload is what the task is for, so the rows have - /// to cross threads and the upload makes progress only while the worker keeps pumping. - rows_tx: Option, PeekError>>>, - /// The result of the background task, eventually. - pub result: oneshot::Receiver<(PeekResponse, Duration)>, - /// The `tracing::Span` tracking this peek's operation - pub span: tracing::Span, - /// A background task that's responsible for producing the peek results. - /// If we're no longer interested in the results, we abort the task. - _abort_handle: AbortOnDropHandle<()>, +/// The variant names the step that refused, and the driver reports it as the peek's error. +#[derive(Debug, thiserror::Error)] +pub(super) enum StashError { + /// The stash location did not open, so nothing was written. + #[error("peek stash could not open its persist location: {0}")] + OpenLocation(#[source] ExternalError), + /// Persist refused a row the upload handed it. + #[error("peek stash could not write a row: {0}")] + WriteRow(#[source] InvalidUsage), + /// Persist refused to finish the batch, which takes with it every part already written. + #[error("peek stash could not finish its batch: {0}")] + FinishBatch(#[source] InvalidUsage), + /// The task finishing the batch ended without delivering one, which only a runtime that is + /// going away can cause. + #[error("peek stash lost the task finishing its batch")] + LostFinishTask, } -impl StashingPeek { - pub fn start_upload( - persist_clients: Arc, - persist_location: &PersistLocation, - mut peek: Peek, - mut trace_bundle: TraceBundle, - batch_max_runs: usize, - ) -> Self { - let (rows_tx, rows_rx) = tokio::sync::mpsc::channel(10); - let (result_tx, result_rx) = oneshot::channel::<(PeekResponse, Duration)>(); - - let persist_clients = Arc::clone(&persist_clients); - let persist_location = persist_location.clone(); - - let peek_uuid = peek.uuid; - let relation_desc = peek.result_desc.clone(); - - let oks_handle = trace_bundle.oks_mut(); - - let peek_iterator = peek_result_iterator::PeekResultIterator::new( - peek.target.id(), - peek.map_filter_project.clone(), - peek.timestamp, - peek.literal_constraints.as_deref_mut(), - oks_handle, - None, - 0, - ); - - let rows_needed_by_finishing = peek.finishing.num_rows_needed(); - - let task_handle = mz_ore::task::spawn( - || format!("peek_stash::stash_peek_response({peek_uuid})"), - async move { - let start = Instant::now(); - - let result = Self::do_upload( - &persist_clients, - persist_location, - batch_max_runs, - peek.uuid, - relation_desc, - rows_needed_by_finishing, - rows_rx, - ) - .await; - - let result = match result { - Ok(peek_response) => peek_response, - Err(e) => PeekResponse::Error(PeekError::unstructured(e.to_string())), - }; - match result_tx.send((result, start.elapsed())) { - Ok(()) => {} - Err((_result, elapsed)) => { - debug!(duration = ?elapsed, "dropping result for cancelled peek {}", peek_uuid) - } - } - }, - ); +/// A peek's answer on its way to the peek stash, written to persist a batch of rows at a time. +/// +/// The upload owns the IO the stash needs, so a walk that feeds it performs none: a driver that can +/// await pushes the rows the walk produced and finishes the upload, and the walk itself neither +/// opens a client nor writes a byte. That split keeps a walk drivable from a timely worker and from +/// an async task alike. +/// +/// The rows an upload is given are the rows it writes, in the order it is given them. An upload +/// that does not reach a reader deletes what it can, whether it is dropped by a driver that will +/// answer with something else or stopped part-way through finishing. +/// [`StashUpload::abandon`] bounds what that reaches and what it costs. +pub(super) struct StashUpload { + /// The description the stashed response reports, and the schema the batch is written under. + relation_desc: RelationDesc, + /// The shard the batch belongs to, derived from the peek's uuid so that a reader holding the + /// response can find it. + shard_id: ShardId, + /// The parts persist has taken so far. Taken by whichever of [`StashUpload::finish`] and + /// [`StashUpload::abandon`] gets there first, and a `None` says the parts are accounted for + /// and nothing is left to delete. + batch_builder: Option>, + /// The upper the batch is finished at, one step beyond the timestamp every row is written at. + upper: Antichain, + /// Rows written so far, counting a row with a diff of `n` as `n` rows, which is how the + /// finishing counts them. + num_rows: u64, + /// The size of those rows, measured as an inline answer measures its own, so that + /// `max_result_size` bounds a stashed answer and an inline one alike. + stashed_byte_len: usize, + /// Whether persist has taken a part off this builder and into blob storage. Until it has, + /// everything the upload holds is in memory, so an abandoned upload drops its builder instead + /// of paying a write and a delete to reclaim nothing. + wrote_parts: bool, + /// The runtime an abandoned upload's deletion is spawned on, held rather than taken from the + /// ambient context because [`StashUpload::abandon`] runs where there may be none: a `Drop` + /// carries no runtime context of its own, and `Handle::current` panics without one. + runtime: Handle, +} - Self { - peek, - peek_iterator: Some(peek_iterator), - rows_tx: Some(rows_tx), - result: result_rx, - span: tracing::Span::current(), - _abort_handle: task_handle.abort_on_drop(), - } - } +/// The `expect` message where the builder is taken: it is present until `finish` or `abandon` +/// takes it, and neither runs twice. +const BUILDER_TAKEN: &str = "an upload holds its builder until it is finished or abandoned"; - async fn do_upload( +impl StashUpload { + /// Opens an upload for the peek `peek_uuid` identifies. + /// + /// Fails when the stash location does not open, in which case nothing has been written. + pub(super) async fn open( persist_clients: &PersistClientCache, persist_location: PersistLocation, batch_max_runs: usize, peek_uuid: Uuid, relation_desc: RelationDesc, - max_rows: Option, // The number of rows needed by the RowSetFinishing's offset + limit - mut rows_rx: tokio::sync::mpsc::Receiver, PeekError>>, - ) -> Result { + ) -> Result { let client = persist_clients .open(persist_location) .await - .map_err(|e| e.to_string())?; + .map_err(StashError::OpenLocation)?; let shard_id = format!("s{}", peek_uuid); let shard_id = ShardId::try_from(shard_id).expect("can parse"); @@ -157,7 +127,7 @@ impl StashingPeek { // // TODO: We _could_ work around the above by teaching the bare columnar // Row encoder about zero-column rows. - let mut batch_builder = client + let batch_builder = client .batch_builder::( shard_id, write_schemas, @@ -166,93 +136,276 @@ impl StashingPeek { ) .await; - let mut num_rows: u64 = 0; - - 'outer: loop { - let row = rows_rx.recv().await; - match row { - Some(Ok(rows)) => { - for (row, diff) in rows { - num_rows += - u64::from(NonZeroU64::try_from(diff).expect("diff fits into u64")); - let diff: i64 = diff.into(); - - batch_builder - .add(&SourceData(Ok(row)), &(), &Timestamp::default(), &diff) - .await - .expect("invalid usage"); - - if let Some(max_rows) = max_rows { - if num_rows >= u64::cast_from(max_rows) { - // Drop the receiver so the producer's next - // try_reserve() fails, stopping row production. - drop(rows_rx); - break 'outer; - } - } - } - } - Some(Err(err)) => return Ok(PeekResponse::Error(err)), - None => { - break; - } - } + Ok(Self { + relation_desc, + shard_id, + batch_builder: Some(batch_builder), + upper, + num_rows: 0, + stashed_byte_len: 0, + wrote_parts: false, + runtime: Handle::current(), + }) + } + + /// Writes `rows` to the stash. + /// + /// Every row given is written. Stopping where the peek's finishing has all it can use is the + /// scan's to decide, since it is the scan that holds the finishing and produces the rows. + /// + /// Fails where persist rejects the write, which leaves the upload unusable and the rows it + /// holds unanswerable. + pub(super) async fn push(&mut self, rows: RowBatch) -> Result<(), StashError> { + let batch_builder = self.batch_builder.as_mut().expect(BUILDER_TAKEN); + + for (row, diff) in rows { + self.num_rows += u64::from(NonZeroU64::try_from(diff).expect("diff fits into u64")); + self.stashed_byte_len = self.stashed_byte_len.saturating_add(entry_byte_len(&row)); + let diff: i64 = diff.into(); + + let added = batch_builder + .add(&SourceData(Ok(row)), &(), &Timestamp::default(), &diff) + .await + .map_err(StashError::WriteRow)?; + self.wrote_parts |= matches!(added, Added::RecordAndParts); } - let batch = batch_builder.finish(upper).await.expect("invalid usage"); + Ok(()) + } + + /// Finishes the batch and builds the response that names it. + /// + /// `inline_rows` are rows of the same answer that never reached the stash, carried beside the + /// batch instead of paying a write. They sit outside the stashed row count, which describes the + /// batch alone, and outside the stash's ordering, which is sound because a peek reaches the + /// stash only with an empty `order_by`. + /// + /// Fails where persist rejects the batch, which takes the parts with it: persist keeps the + /// builder and hands back no handle to what it holds. Only a batch whose bounds do not admit + /// its own updates is refused, which this upload's fixed lower, upper and timestamp cannot + /// produce. + pub(super) async fn finish( + mut self, + inline_rows: RowBatch, + ) -> Result { + let delivered = self.finish_batch().await?; + + // Built before the batch leaves its guard, so that an unwind here still deletes what was + // written rather than leaving it for nobody. + let inline_rows = inline_rows + .into_iter() + .map(|(row, copies)| { + let copies = NonZeroUsize::try_from(copies).expect("fits into usize"); + (row, copies) + }) + .collect(); + + let batch = delivered.take(); let stashed_response = StashedPeekResponse { - num_rows_batches: u64::cast_from(num_rows), + num_rows_batches: self.num_rows, encoded_size_bytes: batch.encoded_size_bytes(), - relation_desc, - shard_id, + stashed_byte_len: self.stashed_byte_len, + relation_desc: self.relation_desc.clone(), + shard_id: self.shard_id, batches: vec![batch.into_transmittable_batch()], - inline_rows: vec![RowCollection::new(vec![], &[])], + inline_rows: vec![RowCollection::new(inline_rows, &[])], + }; + Ok(PeekResponse::Stashed(Box::new(stashed_response))) + } + + /// Finishes the batch as work of its own, and leaves the upload holding no parts. + /// + /// Flushing the buffered part and the uploads still in flight is the longest await an upload + /// makes, so a cancellation most likely lands there, with the builder already out of the + /// upload. A task the cancellation cannot reach holds it instead, and whoever ends up holding a + /// batch nobody will read deletes it. + async fn finish_batch(&mut self) -> Result { + let batch_builder = self.batch_builder.take().expect(BUILDER_TAKEN); + let upper = self.upper.clone(); + let shard_id = self.shard_id; + let runtime = self.runtime.clone(); + + let (tx, rx) = oneshot::channel(); + let _handle = + self.runtime + .spawn_named(|| format!("peek_stash::finish({shard_id})"), async move { + let delivered = batch_builder + .finish(upper) + .await + .map(|batch| DeliveredBatch::new(batch, runtime, shard_id)) + .map_err(StashError::FinishBatch); + // A send that finds no receiver hands the delivery straight back, and dropping it + // here deletes the batch. An error nobody is left to read is simply dropped. + let _undelivered = tx.send(delivered); + }); + + // The task is detached rather than held, so it outlives this await and the only way the + // channel closes without a delivery is a runtime that is going away. + rx.await.map_err(|_| StashError::LostFinishTask)? + } + + /// Schedules the deletion of whatever parts the upload still holds, and leaves it holding + /// none. + /// + /// Persist hands back a deletable handle only by finishing the batch, so the buffered rows are + /// written out first. That write reaches `persist_blob_target_size`, 128 MiB by default, and + /// nothing bounds how many abandoned uploads carry one at once, because the walk releases its + /// permit before this finishes. + /// + /// The handle does not reach parts a run merge already dropped from shard state. A reader + /// deleting a response it has finished with reaches the same set, so that bounds the builder + /// rather than abandonment. + /// + /// TODO: a builder teardown that surrendered the written parts without flushing the buffered + /// one would make this cost a delete and nothing else. + fn abandon(&mut self) { + let Some(batch_builder) = self.batch_builder.take() else { + return; + }; + + // An upload persist never took a part off holds its rows in memory alone, so its builder + // goes with it. Finishing here would upload the buffered part just to delete it again, + // and a part reaches blob storage only once the buffer passes + // `persist_blob_target_size`, which is far above the stash threshold that opened this + // upload. So this is the case nearly every abandoned upload is in. + if !self.wrote_parts { + drop(batch_builder); + return; + } + + let upper = self.upper.clone(); + let shard_id = self.shard_id; + + // Scheduled, not awaited: the caller that matters most cannot await at all. Cancelling a + // peek aborts the walk driving it, and an aborted task is dropped rather than polled again, + // so the deletion reaches blob storage only as work outside that task. Entering the runtime + // explicitly lets this run from a `Drop`, which has no guaranteed runtime context. + // + // NOTE: this reaches only an upload a live replica gives up on. A replica that dies + // mid-upload, or one whose runtime is shutting down, leaves the parts behind for a + // reader-side sweep or persist's garbage collection. + let _handle = + self.runtime + .spawn_named(|| format!("peek_stash::discard({shard_id})"), async move { + match batch_builder.finish(upper).await { + Ok(batch) => batch.delete().await, + Err(error) => { + warn!(%shard_id, %error, "peek stash cannot delete an abandoned batch") + } + } + }); + } +} + +impl Drop for StashUpload { + /// Deletes the parts of an upload that ends without [`StashUpload::finish`], which is the + /// only cleanup a walk aborted mid-upload gets. + fn drop(&mut self) { + self.abandon(); + } +} + +/// A finished batch on its way to the response that will name it, deleted from blob storage if it +/// never arrives. +/// +/// A batch nobody takes out of a delivery is one no reader will be told how to find, so dropping it +/// deletes it. That covers both ways a delivery goes unclaimed: a send that finds no receiver, and +/// a receiver dropped between the send and the take. Persist's own `Drop for Batch` covers neither, +/// logging the blob keys and leaving them. +struct DeliveredBatch { + /// Taken by [`DeliveredBatch::take`], and a `None` says the batch has an owner that will + /// answer for it. + batch: Option>, + runtime: Handle, + shard_id: ShardId, +} + +impl DeliveredBatch { + fn new( + batch: Batch, + runtime: Handle, + shard_id: ShardId, + ) -> Self { + Self { + batch: Some(batch), + runtime, + shard_id, + } + } + + /// Claims the batch, consuming the delivery. + /// + /// The caller takes over the obligation: a batch dropped without being transmitted or deleted + /// leaves its blobs behind. + fn take(mut self) -> Batch { + self.batch + .take() + .expect("a delivery holds its batch until it is claimed") + } +} + +impl Drop for DeliveredBatch { + fn drop(&mut self) { + let Some(batch) = self.batch.take() else { + return; }; - let result = PeekResponse::Stashed(Box::new(stashed_response)); - Ok(result) + + let shard_id = self.shard_id; + let _handle = self + .runtime + .spawn_named(|| format!("peek_stash::delete({shard_id})"), async move { + batch.delete().await + }); } +} + +/// Where a peek's rows go when they may not be answered with inline, and what opening the upload +/// that writes them takes. +/// +/// A driver holds a target, not an open upload: a walk that never crosses the stash threshold opens +/// no shard and writes no byte, and most walks never do. A driver gets one exactly when its scan was +/// opened stash-eligible, so a walk with no target has a scan that offers no batch. +pub(super) struct StashTarget { + persist_clients: Arc, + persist_location: PersistLocation, + /// The peek's uuid, which the shard the batch belongs to is derived from. + peek_uuid: Uuid, + /// The description the rows are written under, and the one the response reports. + relation_desc: RelationDesc, +} - /// Pumps rows from the [PeekResultIterator] to the async task, via our - /// `rows_tx`. Will pump at most `batch_size` rows in one batch, and at most - /// the given `num_batches` batches. - pub fn pump_rows(&mut self, mut num_batches: usize, batch_size: usize) { - while num_batches > 0 - && let Some(row_iter) = self.peek_iterator.as_mut() - { - // Try to reserve space in the channel before pulling rows from the - // iterator. - let permit = match self - .rows_tx - .as_mut() - .expect("missing rows_tx") - .try_reserve() - { - Ok(permit) => permit, - Err(_) => { - // Channel is full, can't send more rows right now. - break; - } - }; - - let rows: Result, _> = row_iter.take(batch_size).collect(); - match rows { - Ok(rows) if rows.is_empty() => { - // Iterator is exhausted, we're done - drop(permit); - self.peek_iterator.take(); - self.rows_tx.take(); - break; - } - Ok(rows) => { - permit.send(Ok(rows)); - } - Err(e) => { - permit.send(Err(e)); - } - } - - num_batches -= 1; +impl StashTarget { + /// The stash `peek`'s rows go to, at `persist_location`. + pub(super) fn new( + peek: &Peek, + persist_clients: Arc, + persist_location: PersistLocation, + ) -> Self { + Self { + persist_clients, + persist_location, + peek_uuid: peek.uuid, + relation_desc: peek.result_desc.clone(), } } + + /// Opens the upload, whose batch builder holds at most `batch_max_runs` runs. + pub(super) async fn open(&self, batch_max_runs: usize) -> Result { + StashUpload::open( + &self.persist_clients, + self.persist_location.clone(), + batch_max_runs, + self.peek_uuid, + self.relation_desc.clone(), + ) + .await + } } + +/// Tests of the incremental stash upload, over the persist location a replica would write to. +/// +/// [`tests::stashed_rows`] is shared with the drivers that feed an upload, which read a response +/// back the same way. +#[cfg(test)] +pub(crate) mod tests; diff --git a/src/compute/src/compute_state/peek_stash/tests.rs b/src/compute/src/compute_state/peek_stash/tests.rs new file mode 100644 index 0000000000000..b351a7dbe8bd9 --- /dev/null +++ b/src/compute/src/compute_state/peek_stash/tests.rs @@ -0,0 +1,595 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +//! Tests of the upload that writes a peek's answer to the peek response stash. + +use std::num::NonZeroI64; +use std::time::Duration; + +use mz_compute_types::dyncfgs::{ + PEEK_RESPONSE_STASH_BATCH_MAX_RUNS, PEEK_RESPONSE_STASH_READ_MEMORY_BUDGET_BYTES, +}; +use mz_dyncfg::{ConfigUpdates, ConfigVal}; +use mz_ore::cast::CastLossy; +use mz_ore::metrics::MetricsRegistry; +use mz_persist_client::cfg::PersistConfig; +use mz_persist_client::rpc::PubSubClientConnection; +use mz_repr::{Datum, Row, SqlScalarType}; + +use super::*; + +/// How many turns a test gives blob storage to catch up before it declares a deletion lost. +/// +/// Deleting the parts of an abandoned upload is scheduled rather than awaited, and the write +/// that earns the handle to delete runs partly on persist's isolated runtime, so a test can +/// only wait for it. Bounded by turns rather than by a deadline, so a deletion that never +/// happens fails the test rather than hanging the suite. +const BLOB_TURNS: usize = 1_000; + +/// A run limit above what any upload here produces, so that its builder never merges runs. +/// +/// A merge writes parts of its own and leaves the parts it merged from behind, both of which +/// are persist's business rather than the upload's. Raising the limit past the runs a test +/// produces is what makes the parts it counts the parts the upload is answerable for. +const NO_RUN_MERGING: usize = 64; + +/// A persist client cache whose blob traffic a test can count. +/// +/// Nothing an upload writes is inlined into shard state, where no blob counter would see it, +/// which is what makes "the parts an upload wrote" something a test can observe at all. How +/// soon a part gets there is the constructors' difference. The counts are read off the registry +/// because persist keeps its own metric handles private. +pub(crate) struct CountedBlob { + registry: MetricsRegistry, + clients: Arc, +} + +impl CountedBlob { + /// A cache in which every row an upload takes becomes a part in blob storage. + pub(crate) fn new() -> Self { + Self::with_part_size(Some(0)) + } + + /// A cache that leaves persist's part size where production has it, so the rows an upload + /// takes stay buffered in its builder and reach blob storage only if a batch is finished. + pub(crate) fn with_buffered_parts() -> Self { + Self::with_part_size(None) + } + + fn with_part_size(blob_target_size: Option) -> Self { + let cfg = PersistConfig::new_for_tests(); + + let mut updates = ConfigUpdates::default(); + let part_size = blob_target_size + .map(|size| ("persist_blob_target_size", ConfigVal::Usize(size))) + .into_iter(); + for (name, val) in part_size.chain([ + // Without this persist keeps a small part in shard state, where no blob counter + // sees it and no delete has anything to remove. + ( + "persist_inline_writes_single_max_bytes", + ConfigVal::Usize(0), + ), + // A part per row would otherwise hold the builder in its outstanding-write stall, + // so raise the bound past the parts a test produces. + ( + "persist_batch_builder_max_outstanding_parts", + ConfigVal::Usize(8_192), + ), + ]) { + assert!( + cfg.entry(name).is_some(), + "persist no longer has a config named {name}" + ); + updates.add_dynamic(name, val); + } + updates.apply(&cfg); + + let registry = MetricsRegistry::new(); + let clients = Arc::new(PersistClientCache::new(cfg, ®istry, |_, _| { + PubSubClientConnection::noop() + })); + Self { registry, clients } + } + + pub(crate) fn clients(&self) -> &Arc { + &self.clients + } + + /// Blob keys written. + pub(crate) fn written(&self) -> u64 { + self.succeeded("blob_set") + } + + /// Blob keys deleted. + pub(crate) fn deleted(&self) -> u64 { + self.succeeded("blob_delete") + } + + /// Deletes that found nothing to delete, which is how a delete of a key that was never + /// written shows up. + pub(crate) fn deletes_of_nothing(&self) -> u64 { + self.counter("mz_persist_external_blob_delete_noop_count", &[]) + } + + /// Waits until every key written has been deleted, which is what an upload that reaches no + /// reader owes. + pub(crate) async fn wait_until_nothing_is_left(&self, what: &str) { + for _ in 0..BLOB_TURNS { + let written = self.written(); + if written > 0 && self.deleted() == written { + return; + } + tokio::time::sleep(Duration::from_millis(1)).await; + } + panic!( + "{what} left {} of {} blob keys behind after {BLOB_TURNS} turns", + self.written() - self.deleted(), + self.written(), + ); + } + + /// Waits until at least one part has reached blob storage. + pub(crate) async fn wait_until_something_is_written(&self, what: &str) { + for _ in 0..BLOB_TURNS { + if self.written() > 0 { + return; + } + tokio::time::sleep(Duration::from_millis(1)).await; + } + panic!("{what} wrote no blob key within {BLOB_TURNS} turns"); + } + + fn succeeded(&self, op: &str) -> u64 { + self.counter("mz_persist_external_succeeded_count", &[("op", op)]) + } + + /// Sums the counter series named `name` whose labels include all of `labels`, reading zero + /// for a series that has not been incremented yet. + fn counter(&self, name: &str, labels: &[(&str, &str)]) -> u64 { + let Some(family) = self + .registry + .gather() + .into_iter() + .find(|m| m.name() == name) + else { + return 0; + }; + family + .get_metric() + .iter() + .filter(|metric| { + labels.iter().all(|(name, value)| { + metric + .get_label() + .iter() + .any(|label| label.name() == *name && label.value() == *value) + }) + }) + .map(|metric| u64::cast_lossy(metric.get_counter().value())) + .sum() + } +} + +/// The description of the single-column result every peek here asks for. +fn result_desc() -> RelationDesc { + RelationDesc::builder() + .with_column("value", SqlScalarType::UInt64.nullable(false)) + .finish() +} + +/// A batch of `values`, each carrying `diff` copies of itself. +fn batch(values: impl IntoIterator, diff: i64) -> RowBatch { + let diff = NonZeroI64::new(diff).expect("a row carries a non-zero diff"); + values + .into_iter() + .map(|value| (Row::pack_slice(&[Datum::UInt64(value)]), diff)) + .collect() +} + +/// An upload opened against `clients`. +async fn open_upload(clients: &PersistClientCache) -> StashUpload { + open_upload_with_runs(clients, *PEEK_RESPONSE_STASH_BATCH_MAX_RUNS.default()).await +} + +/// An upload whose batch builder holds at most `batch_max_runs` runs. +/// +/// A builder that has more runs than that merges them, which writes parts of its own and +/// leaves the parts it merged from behind. A test counting what an upload wrote raises the +/// limit above the runs it produces, so that the parts it counts are the parts the upload owns. +async fn open_upload_with_runs(clients: &PersistClientCache, batch_max_runs: usize) -> StashUpload { + StashUpload::open( + clients, + PersistLocation::new_in_mem(), + batch_max_runs, + Uuid::new_v4(), + result_desc(), + ) + .await + .expect("the in-memory location opens") +} + +/// Writes `rows` to `upload`, which persist accepts for every batch built here. +async fn push(upload: &mut StashUpload, rows: RowBatch) { + upload.push(rows).await.expect("persist takes the rows") +} + +/// Finishes `upload` beside `inline_rows`, which persist accepts for every batch built here. +async fn finish(upload: StashUpload, inline_rows: RowBatch) -> PeekResponse { + upload + .finish(inline_rows) + .await + .expect("persist finishes the batch") +} + +/// The values a finished upload holds, in ascending order, each repeated as often as the diff +/// it was written with. +/// +/// A stashed response names a persist batch rather than carrying rows, so what the upload wrote +/// is only visible from the batch. Persist consolidates a batch rather than preserving the +/// order it was written in, so the values are sorted here and compared as the multiset they +/// are. +async fn stashed_values(clients: &PersistClientCache, response: PeekResponse) -> Vec { + let PeekResponse::Stashed(stashed) = response else { + panic!("an upload finishes into a stashed response, not {response:?}"); + }; + + stashed_rows(clients, *stashed) + .await + .into_iter() + .map(|row| row.unpack_first().unwrap_uint64()) + .collect() +} + +/// The rows the batches of `stashed` hold, in [`Row`] order, each repeated as often as the diff +/// it was written with. +/// +/// A stashed response names a persist batch rather than carrying rows, so what an upload wrote +/// is only visible from the batch. Read as the coordinator reads one, deletions included, and +/// sorted because persist consolidates a batch rather than preserving the order it was written +/// in. Rows the response carries in `inline_rows` are not here: they never reached a batch. +pub(crate) async fn stashed_rows( + clients: &PersistClientCache, + stashed: StashedPeekResponse, +) -> Vec { + // Opened out of the cache that opened the upload, because two `PersistLocation`s naming the + // same in-memory URI reach the same blob only through one cache. + let mut client = clients + .open(PersistLocation::new_in_mem()) + .await + .expect("the in-memory location opens"); + + let shard_id = stashed.shard_id; + let batches = stashed + .batches + .into_iter() + .map(|batch| client.batch_from_transmittable_batch(&shard_id, batch)) + .collect(); + let read_schemas: Schemas = Schemas { + id: None, + key: Arc::new(stashed.relation_desc), + val: Arc::new(UnitSchema), + }; + let mut cursor = client + .read_batches_consolidated::<_, _, _, i64>( + shard_id, + Antichain::from_elem(Timestamp::default()), + read_schemas, + batches, + |_stats| true, + *PEEK_RESPONSE_STASH_READ_MEMORY_BUDGET_BYTES.default(), + ) + .await + .expect("the batch is readable at the timestamp it was written at"); + + let mut rows = Vec::new(); + while let Some(updates) = cursor.next().await { + for ((key, _val), _time, diff) in updates { + let row = key.0.expect("the peek stash holds no errors"); + let copies = usize::try_from(diff).expect("a stashed row carries a positive diff"); + rows.extend(std::iter::repeat_n(row, copies)); + } + } + rows.sort(); + + // Deleted as the coordinator deletes them once it has read them. A batch dropped without + // this leaves its blob keys behind and says so in a warning. + for batch in cursor.into_lease() { + batch.delete().await; + } + rows +} + +/// An upload with no limit to reach holds every row pushed into it, over as many pushes as the +/// driver made. +#[mz_ore::test(tokio::test)] +async fn an_upload_holds_the_rows_pushed_into_it() { + let clients = PersistClientCache::new_no_metrics(); + let mut upload = open_upload(&clients).await; + + push(&mut upload, batch(0..3, 1)).await; + push(&mut upload, batch(3..5, 1)).await; + + let response = finish(upload, RowBatch::new()).await; + let PeekResponse::Stashed(stashed) = &response else { + panic!("an upload finishes into a stashed response, not {response:?}"); + }; + assert_eq!( + stashed.num_rows_batches, 5, + "the response counts the rows the upload wrote" + ); + assert_eq!( + stashed_values(&clients, response).await, + vec![0, 1, 2, 3, 4] + ); +} + +/// An upload writes every row it is given, whole, however often the answer holds it. Where the +/// finishing's limit lands is the scan's to decide, not the upload's. +#[mz_ore::test(tokio::test)] +async fn an_upload_writes_every_row_it_is_given() { + let clients = PersistClientCache::new_no_metrics(); + let mut upload = open_upload(&clients).await; + + push(&mut upload, batch(0..2, 3)).await; + + assert_eq!( + stashed_values(&clients, finish(upload, RowBatch::new()).await).await, + vec![0, 0, 0, 1, 1, 1], + ); +} + +/// Rows a walk still held when it ended travel with the response rather than through the +/// batch, so a reader sees them beside the stashed rows and the stashed row count counts only +/// what the batch holds. +#[mz_ore::test(tokio::test)] +async fn rows_that_never_reached_the_stash_travel_inline() { + let clients = PersistClientCache::new_no_metrics(); + let mut upload = open_upload(&clients).await; + + push(&mut upload, batch(0..3, 1)).await; + + let response = finish(upload, batch(3..5, 1)).await; + let PeekResponse::Stashed(stashed) = &response else { + panic!("an upload finishes into a stashed response, not {response:?}"); + }; + assert_eq!( + stashed.num_rows_batches, 3, + "the stashed row count describes the batch alone" + ); + let expected: Vec<(Row, NonZeroUsize)> = batch(3..5, 1) + .into_iter() + .map(|(row, copies)| { + ( + row, + NonZeroUsize::try_from(copies).expect("fits into usize"), + ) + }) + .collect(); + assert_eq!( + stashed.inline_rows, + vec![RowCollection::new(expected, &[])], + "the rows the walk still held travel with the response" + ); + assert_eq!( + stashed_values(&clients, response).await, + vec![0, 1, 2], + "the rows carried inline are not written to the batch as well" + ); +} + +/// An upload built by hand, so that a test can hold the bounds [`StashUpload::open`] never +/// produces. +/// +/// Opening fixes the lower, the upper and the timestamp every row is written at, and that +/// combination is one persist always accepts. The rejection arms exist all the same, because +/// the alternative to reporting a rejection is a panic in an offloaded walk, which lands in the +/// dead-task arm and takes the worker down with it in a test build. +async fn upload_with_bounds( + clients: &PersistClientCache, + lower: Timestamp, + upper: Timestamp, +) -> StashUpload { + let client = clients + .open(PersistLocation::new_in_mem()) + .await + .expect("the in-memory location opens"); + let shard_id = ShardId::try_from(format!("s{}", Uuid::new_v4())).expect("can parse"); + let relation_desc = result_desc(); + let write_schemas: Schemas = Schemas { + id: None, + key: Arc::new(relation_desc.clone()), + val: Arc::new(UnitSchema), + }; + let batch_builder = client + .batch_builder::( + shard_id, + write_schemas, + Antichain::from_elem(lower), + Some(NO_RUN_MERGING), + ) + .await; + + StashUpload { + relation_desc, + shard_id, + batch_builder: Some(batch_builder), + upper: Antichain::from_elem(upper), + num_rows: 0, + stashed_byte_len: 0, + wrote_parts: false, + runtime: Handle::current(), + } +} + +/// An upload that is dropped deletes the parts it had already written. +/// +/// Dropping is the whole of the cleanup, both for a driver that will answer the peek with +/// something other than these rows and for a walk aborted mid-upload: an aborted task is +/// dropped rather than polled again, so nothing it would have called runs. +#[mz_ore::test(tokio::test)] +async fn a_dropped_upload_deletes_the_parts_it_wrote() { + let blob = CountedBlob::new(); + let mut upload = open_upload_with_runs(blob.clients(), NO_RUN_MERGING).await; + + push(&mut upload, batch(0..4, 1)).await; + blob.wait_until_something_is_written("an upload holding four rows") + .await; + + drop(upload); + + blob.wait_until_nothing_is_left("a dropped upload").await; + assert_eq!( + blob.deletes_of_nothing(), + 0, + "the deletes must name the keys the upload wrote" + ); +} + +/// An upload persist has taken no part off costs no blob traffic to give up. +/// +/// This is the case nearly every abandoned upload is in, because a part reaches blob storage +/// only once the builder's buffer passes `persist_blob_target_size`, far above the stash +/// threshold that opens an upload at all. Writing the buffered part out just to delete it again +/// would make giving up cost two round trips per abandoned peek. +#[mz_ore::test(tokio::test)] +async fn a_dropped_upload_holding_no_part_writes_none() { + let blob = CountedBlob::with_buffered_parts(); + let mut upload = open_upload_with_runs(blob.clients(), NO_RUN_MERGING).await; + + push(&mut upload, batch(0..4, 1)).await; + assert_eq!(blob.written(), 0, "four rows stay in the builder"); + + drop(upload); + + // Given the same turns an abandonment that does write gets, so a regression that finishes + // the batch here fails rather than racing. + for _ in 0..BLOB_TURNS { + tokio::time::sleep(Duration::from_millis(1)).await; + } + assert_eq!( + blob.written(), + 0, + "giving up an upload holding no part must write none" + ); + assert_eq!(blob.deleted(), 0, "and must delete nothing"); +} + +/// A finished batch that never reaches the response naming it is deleted. +/// +/// This is the window between persist handing the batch over and the response taking it: a +/// cancellation that lands there leaves a batch no reader will ever be told how to find, and +/// persist's own `Drop for Batch` logs the keys and leaves them. +#[mz_ore::test(tokio::test)] +async fn a_finished_batch_that_reaches_no_response_is_deleted() { + let blob = CountedBlob::new(); + let mut upload = open_upload_with_runs(blob.clients(), NO_RUN_MERGING).await; + + push(&mut upload, batch(0..4, 1)).await; + let delivered = upload + .finish_batch() + .await + .expect("persist finishes the batch"); + assert!( + blob.written() > 0, + "a finished batch has written the parts it holds" + ); + assert_eq!( + blob.deleted(), + 0, + "a batch still on its way to a response is not deleted" + ); + + drop(delivered); + + blob.wait_until_nothing_is_left("an unclaimed delivery") + .await; + assert_eq!( + blob.deletes_of_nothing(), + 0, + "the deletes must name the keys the batch wrote" + ); +} + +/// An upload that finishes into a response leaves its parts alone, because the response is +/// what a reader finds them by. +/// +/// The other tests here would all pass against an upload that deleted everything it wrote +/// unconditionally, which is the shape of cleanup that costs every stashed peek its answer. +#[mz_ore::test(tokio::test)] +async fn a_finished_upload_leaves_its_parts_for_the_response() { + let blob = CountedBlob::new(); + let mut upload = open_upload_with_runs(blob.clients(), NO_RUN_MERGING).await; + + push(&mut upload, batch(0..4, 1)).await; + let response = finish(upload, RowBatch::new()).await; + + // Every chance for a deletion that should not happen to happen. + for _ in 0..BLOB_TURNS { + tokio::task::yield_now().await; + } + assert!( + blob.written() > 0, + "a finished upload has written the parts it holds" + ); + assert_eq!( + blob.deleted(), + 0, + "the parts of a finished upload belong to the response that names them" + ); + assert_eq!( + stashed_values(blob.clients(), response).await, + vec![0, 1, 2, 3], + "the response must still name a readable batch" + ); +} + +/// A write persist rejects is reported rather than raised. +/// +/// The unit that fails is the peek, whose driver answers it with the error. A panic here would +/// instead end the task driving an offloaded walk, and the worker reads a dead task as a defect. +#[mz_ore::test(tokio::test)] +async fn a_rejected_write_is_reported_rather_than_raised() { + let clients = PersistClientCache::new_no_metrics(); + // A lower past the timestamp the upload writes its rows at, which is the one thing + // `BatchBuilder::add` refuses. + let mut upload = upload_with_bounds( + &clients, + Timestamp::default().step_forward(), + Timestamp::default().step_forward().step_forward(), + ) + .await; + + let rejection = upload.push(batch(0..2, 1)).await; + + assert!( + matches!(&rejection, Err(StashError::WriteRow(error)) if error.to_string().contains("not beyond batch lower")), + "persist must reject the write and the upload must report it: {rejection:?}", + ); +} + +/// A batch persist rejects is reported rather than raised. +/// +/// A rejected finish leaves the parts behind, which [`StashUpload::finish`] states: persist +/// keeps the builder it was asked to finish and hands back no handle to what it holds. +#[mz_ore::test(tokio::test)] +async fn a_rejected_batch_is_reported_rather_than_raised() { + let clients = PersistClientCache::new_no_metrics(); + // An upper the rows the upload holds are not below, which is the one thing + // `BatchBuilder::finish` refuses. + let mut upload = upload_with_bounds(&clients, Timestamp::default(), Timestamp::default()).await; + + push(&mut upload, batch(0..2, 1)).await; + let rejection = upload.finish(RowBatch::new()).await; + + assert!( + matches!(&rejection, Err(StashError::FinishBatch(error)) if error.to_string().contains("beyond the expected batch upper")), + "persist must reject the batch and the upload must report it: {rejection:?}", + ); +} diff --git a/src/compute/src/compute_state/peek_sweep_tests.rs b/src/compute/src/compute_state/peek_sweep_tests.rs index e6e90648611cd..3ff4c4ef9e124 100644 --- a/src/compute/src/compute_state/peek_sweep_tests.rs +++ b/src/compute/src/compute_state/peek_sweep_tests.rs @@ -11,11 +11,10 @@ use mz_compute_types::dyncfgs::{ ENABLE_INDEX_PEEK_OFFLOAD, INDEX_PEEK_ACTIVATION_BUDGET, INDEX_PEEK_INLINE_BUDGET, - PEEK_RESPONSE_STASH_READ_MEMORY_BUDGET_BYTES, }; use mz_dyncfg::ConfigUpdates; -use mz_persist_client::Schemas; use mz_persist_client::cache::PersistClientCache; +use mz_repr::{IntoRowIterator, RowIterator, RowRef}; use mz_secrets::InMemorySecretsController; use mz_storage_types::connections::ConnectionContext; use timely::WorkerConfig; @@ -47,13 +46,13 @@ const WIDE_INDEX_KEYS: u64 = 2_000; /// on is how many peeks got a turn rather than how far each one walked. const SMALL_INDEX_KEYS: u64 = 6; -/// How many rows a walk accumulates before its batch is full, in the tests that drive a -/// hand-back. +/// How many rows a walk accumulates before its batch is full, in the tests that drive a peek +/// to the stash. /// /// More than the inline slice can accumulate within the production budget and fewer than the /// wide index holds, which is what puts the crossing after the offload rather than before it /// or never. -const HAND_BACK_AT_ROWS: u64 = 1_500; +const DIVERT_AT_ROWS: u64 = 1_500; /// How many activations a test drives before it declares a peek stuck. /// @@ -182,7 +181,6 @@ impl Harness { Some(match peek { PendingPeek::Index(_) => "index", PendingPeek::Persist(_) => "persist", - PendingPeek::Stash(_) => "stash", PendingPeek::Offloaded(_) => "offloaded", }) } @@ -223,6 +221,26 @@ impl Harness { ) } + /// The walks the peek stash answered. + /// + /// A stashed answer and an inline answer carry the same rows, so this is what turns "the + /// stash path engaged" from an inference about the configuration into an assertion. + fn walks_stashed(&self) -> u64 { + self.state.metrics.index_peek_stashed_total.get() + } + + /// The cursor positions the ok walks that ended here reported, and how many walks reported + /// them. + /// + /// A walk reports its positions once, wherever it ended, and the count is cumulative over + /// every slice it was cut into. A peek answered by two walks of the same trace would + /// report twice, or once at twice the positions, and would answer with the same rows + /// either way. + fn ok_walk_positions(&self) -> (u64, f64) { + let rows = &self.state.metrics.index_peek_row_iteration_rows; + (rows.get_sample_count(), rows.get_sample_sum()) + } + /// Runs activations until nothing is pending, and reports the responses they produced. /// /// Bounded, so a peek that never answers fails here rather than hanging the suite. The @@ -521,7 +539,63 @@ async fn the_kill_switch_answers_the_same_scan_inline() { assert_eq!( harness.walks(), (1, 0), - "the kill switch offloads nothing, however far a peek walks" + "with nothing bound for the stash, the kill switch offloads nothing however far a \ + peek walks" + ); +} + +/// With the kill switch off, a peek whose accumulated rows belong in the stash still leaves +/// the worker, and is answered from the stash exactly as it is with the switch on. +/// +/// The switch gates offload for latency, which is the placement it can revert. Offload for +/// stashing is not: the driver that writes to the stash is the offloaded one, so a worker that +/// kept such a peek would hold a scan that makes no progress until its batch is taken and +/// answer the peek never. What the switch guarantees is that an ordinary peek runs where it +/// used to, not that no peek ever leaves the worker. +#[mz_ore::test(tokio::test)] +async fn the_kill_switch_still_takes_a_stash_bound_peek_off_the_worker() { + let keys = wide_ok_rows(WIDE_INDEX_KEYS); + // The threshold is a size rather than a count, because the size of what a scan has + // accumulated is what it compares against. Summed over the rows it is meant to admit, + // rather than multiplied out, because a row's packed width follows the value it holds. + let threshold: usize = keys + .iter() + .take(usize::cast_from(DIVERT_AT_ROWS)) + .map(peek_scan::entry_byte_len) + .sum(); + + // The offload is left where production ships it, and only the stash is turned on. + let mut harness = Harness::new(move |updates| { + updates.add(&ENABLE_PEEK_RESPONSE_STASH, true); + updates.add(&PEEK_RESPONSE_STASH_THRESHOLD_BYTES, threshold); + }); + harness.state.peek_stash_persist_location = Some(PersistLocation::new_in_mem()); + harness.add_pending( + index_peek_with_uuid(PEEK_A, None), + trace_bundle(&keys, cancelling_errors(0)), + ); + + harness.sweep(); + assert_eq!( + harness.pending(PEEK_A), + Some("offloaded"), + "an unbounded slice walks until its rows belong in the stash, and then leaves" + ); + + let mut responses = harness.drain().await; + assert_eq!(responses.len(), 1, "the stashed peek answers once"); + let (uuid, response) = responses.pop().expect("length checked"); + assert_eq!(uuid, PEEK_A); + assert_eq!(stashed_rows(&harness, response).await, keys); + assert_eq!( + harness.walks(), + (0, 1), + "the offloaded driver is the one that writes to the stash, so it ended the walk" + ); + assert_eq!( + harness.walks_stashed(), + 1, + "the kill switch does not take the stash away from a peek that needs it" ); } @@ -850,13 +924,13 @@ async fn a_walk_cancelled_with_its_outcome_in_flight_is_counted() { /// A harness whose peek of the whole wide index is offloaded and whose offloaded walk then /// crosses the stash threshold, swept once so that the peek is already offloaded. /// -/// `location` is the replica's peek stash location, which has to be present here whatever the -/// hand-back is meant to find: a replica without one makes no scan stash-eligible, so its -/// walks never fill a batch and never hand back at all. -fn offloaded_walk_that_hands_back(keys: &[Row], location: PersistLocation) -> Harness { +/// `location` is the replica's peek stash location, which has to be present: a replica without +/// one makes no scan stash-eligible, so its walks never fill a batch and never reach the stash +/// at all. +fn offloaded_walk_that_crosses_the_threshold(keys: &[Row], location: PersistLocation) -> Harness { assert!( - u64::cast_from(*INDEX_PEEK_INLINE_BUDGET.default()) < HAND_BACK_AT_ROWS - && HAND_BACK_AT_ROWS < WIDE_INDEX_KEYS, + u64::cast_from(*INDEX_PEEK_INLINE_BUDGET.default()) < DIVERT_AT_ROWS + && DIVERT_AT_ROWS < WIDE_INDEX_KEYS, "the walk must cross the stash threshold after it is offloaded and before it ends" ); // The threshold is a size rather than a count, because the size of what a scan has @@ -864,7 +938,7 @@ fn offloaded_walk_that_hands_back(keys: &[Row], location: PersistLocation) -> Ha // rather than multiplied out, because a row's packed width follows the value it holds. let threshold: usize = keys .iter() - .take(usize::cast_from(HAND_BACK_AT_ROWS)) + .take(usize::cast_from(DIVERT_AT_ROWS)) .map(peek_scan::entry_byte_len) .sum(); @@ -888,160 +962,129 @@ fn offloaded_walk_that_hands_back(keys: &[Row], location: PersistLocation) -> Ha harness } -/// Runs activations until the offloaded walk of `PEEK_A` has handed back. +/// The rows a stashed peek response holds, read back the way the coordinator reads one, in +/// [`Row`] order. /// -/// Bounded, so a walk that never hands back fails here rather than hanging the suite. The pause -/// between activations is what lets the offloaded walk run. -async fn sweep_until_handed_back(harness: &mut Harness) { - for _ in 0..SWEEP_BOUND { - if harness.pending(PEEK_A) != Some("offloaded") { - return; - } - tokio::time::sleep(SWEEP_POLL).await; - harness.sweep(); - } - panic!("the offloaded walk had not handed back after {SWEEP_BOUND} activations"); -} - -/// The rows a stashed peek response holds, read back out of `location` the way the coordinator -/// reads one, in [`Row`] order. -/// -/// A stashed response names a persist batch rather than carrying the answer, so what the peek -/// owes its caller is only visible from the batch. The batch is ordered as persist consolidates -/// it rather than as the peek would have answered, and the coordinator orders what it reads -/// back, so the rows are sorted here and compared as the set they are. -async fn stashed_rows( - harness: &Harness, - location: &PersistLocation, - response: PeekResponse, -) -> Vec { +/// A stashed response names a persist batch rather than carrying the whole answer, so what the +/// peek owes its caller is the batch plus the rows the response carries inline, and both halves +/// are here. The batch is ordered as persist consolidates it rather than as the peek would have +/// answered, and the coordinator orders what it reads back, so the rows are sorted here and +/// compared as the set they are. +async fn stashed_rows(harness: &Harness, response: PeekResponse) -> Vec { let PeekResponse::Stashed(stashed) = response else { panic!("a peek taken to the stash answers with a stashed response, not {response:?}"); }; - // Opened out of the harness's own cache, because two `PersistLocation`s naming the same - // in-memory URI reach the same blob only through the cache that opened them. - let mut client = harness - .state - .persist_clients - .open(location.clone()) - .await - .expect("the in-memory location opens"); - - let shard_id = stashed.shard_id; - let batches = stashed - .batches - .into_iter() - .map(|batch| client.batch_from_transmittable_batch(&shard_id, batch)) + let mut rows: Vec = stashed + .inline_rows + .iter() + .flat_map(|rows| rows.clone().into_row_iter().map(RowRef::to_owned)) .collect(); - let read_schemas: Schemas = Schemas { - id: None, - key: Arc::new(stashed.relation_desc), - val: Arc::new(UnitSchema), - }; - let mut cursor = client - .read_batches_consolidated::<_, _, _, i64>( - shard_id, - Antichain::from_elem(Timestamp::default()), - read_schemas, - batches, - |_stats| true, - *PEEK_RESPONSE_STASH_READ_MEMORY_BUDGET_BYTES.default(), - ) - .await - .expect("the batches are readable at the timestamp they were written at"); - - let mut rows = Vec::new(); - while let Some(updates) = cursor.next().await { - for ((key, _val), _time, diff) in updates { - assert_eq!(diff, 1, "the index holds each key once"); - rows.push(key.0.expect("the peek stash holds no errors")); - } - } + rows.extend(peek_stash::tests::stashed_rows(&harness.state.persist_clients, *stashed).await); rows.sort(); - - // Deleted as the coordinator deletes them once it has read them. A batch dropped without - // this leaves its blob keys behind and says so in a warning. - for batch in cursor.into_lease() { - batch.delete().await; - } rows } -/// An offloaded walk that hands back with a stash location present takes the peek to the stash, -/// which answers it with the rows the peek would have answered with inline. +/// An offloaded walk whose accumulated rows cross the stash threshold writes them to the stash +/// and answers the peek with the handle, without a second walk of the trace and without the +/// peek ever leaving the driver that offloaded it. /// -/// This is the arm every hand-back in production takes, because a replica's stash location is -/// set once at instance creation and nothing clears it. The peek has to become pending on the -/// stash rather than be answered where the hand-back arrives: the rows are produced by a -/// second walk that the worker pumps over the activations that follow, and answering here -/// would drop them. +/// The rows the walk was still holding when the trace ran out ride along with the handle, so +/// the answer is both halves together and neither half alone. #[mz_ore::test(tokio::test)] -async fn an_offloaded_hand_back_takes_the_peek_to_the_stash() { +async fn an_offloaded_walk_that_crosses_the_threshold_answers_from_the_stash() { // The wide keys carry the `UInt64` the peek's result description declares, which is the // schema the stash writes its batch under. The narrow fixture rows do not. let keys = wide_ok_rows(WIDE_INDEX_KEYS); let location = PersistLocation::new_in_mem(); - let mut harness = offloaded_walk_that_hands_back(&keys, location.clone()); + let mut harness = offloaded_walk_that_crosses_the_threshold(&keys, location); - sweep_until_handed_back(&mut harness).await; + let mut responses = harness.drain().await; + assert_eq!(responses.len(), 1, "the stashed peek answers once"); + let (uuid, response) = responses.pop().expect("length checked"); + assert_eq!(uuid, PEEK_A); + let PeekResponse::Stashed(stashed) = &response else { + panic!("a peek taken to the stash answers with a handle, not {response:?}"); + }; assert_eq!( - harness.pending(PEEK_A), - Some("stash"), - "the hand-back starts a stash upload and leaves the peek waiting on it" + stashed.num_rows_batches, + DIVERT_AT_ROWS + 1, + "the walk wrote every row it had accumulated when it crossed the threshold" ); assert_eq!( - harness.peek_responses(), - vec![], - "a peek handed to the stash is not answered where the hand-back arrives" + inline_rows(stashed), + usize::cast_from(WIDE_INDEX_KEYS - (DIVERT_AT_ROWS + 1)), + "the rows the walk still held when the trace ran out ride along with the handle" + ); + assert_eq!( + stashed_rows(&harness, response).await, + keys, + "the stash and the rows beside it are the answer the peek would have given inline" ); assert_eq!( harness.walks(), (0, 1), - "a hand-back is a terminal outcome of the offloaded walk" + "one offloaded walk produced the whole answer" ); - - let mut responses = harness.drain().await; - assert_eq!(responses.len(), 1, "the stashed peek answers once"); - let (uuid, response) = responses.pop().expect("length checked"); - assert_eq!(uuid, PEEK_A); + assert_eq!(harness.walks_stashed(), 1, "the stash answered the peek"); + // The answer alone says nothing here: a walk that gave up at the threshold and started + // over would produce these very rows. What says the trace was walked once is the count of + // cursor positions the walk that ended reported. assert_eq!( - stashed_rows(&harness, &location, response).await, - keys, - "the stash holds the rows the peek would have answered with inline" + harness.ok_walk_positions(), + (1, f64::cast_lossy(keys.len())), + "one walk evaluated each cursor position of the index once" ); } -/// An offloaded walk that hands back with nowhere to write the rows answers the peek with an -/// error rather than leaving it pending on a walk that has stopped. +/// The rows a stashed response carries beside its batches. +fn inline_rows(stashed: &mz_compute_client::protocol::response::StashedPeekResponse) -> usize { + stashed + .inline_rows + .iter() + .map(|rows| rows.clone().into_row_iter().count()) + .sum() +} + +/// Cancelling an offloaded peek bound for the stash answers it once, as cancelled, and the walk +/// behind it produces no second answer. /// -/// This arm is defensive only. Reaching it takes a replica that loses its stash location -/// between the offload and the hand-back, which nothing does, and the location is cleared -/// here to stand in for that: `handle_create_instance` sets it and nothing clears it, while a -/// replica that never had one makes no scan stash-eligible, so none of its walks fills a batch -/// and hands back in the first place. The arm production takes is -/// [`an_offloaded_hand_back_takes_the_peek_to_the_stash`]'s. +/// Cancellation both removes the pending peek and aborts the walk, and it is the removal that +/// has to be paired with the abort: a cancellation that answered without stopping the walk +/// would answer the same peek again, the second time with a handle to a batch nothing will +/// read. #[mz_ore::test(tokio::test)] -async fn an_offloaded_hand_back_answers_when_there_is_nowhere_to_write() { +async fn a_cancelled_stash_bound_peek_is_answered_once() { let keys = wide_ok_rows(WIDE_INDEX_KEYS); - let mut harness = offloaded_walk_that_hands_back(&keys, PersistLocation::new_in_mem()); + let mut harness = + offloaded_walk_that_crosses_the_threshold(&keys, PersistLocation::new_in_mem()); - harness.state.peek_stash_persist_location = None; + harness.active().handle_cancel_peek(PEEK_A); assert_eq!( - harness.drain().await, - vec![( - PEEK_A, - PeekResponse::Error(PeekError::unstructured( - "peek result is too large to answer inline and this replica has no peek stash \ - location" - )) - )] + harness.peek_responses(), + vec![(PEEK_A, PeekResponse::Canceled)] + ); + assert_eq!(harness.pending(PEEK_A), None); + + for _ in 0..SWEEP_BOUND { + tokio::time::sleep(SWEEP_POLL).await; + harness.sweep(); + } + assert_eq!( + harness.peek_responses(), + vec![], + "a cancelled peek is answered once and never again" + ); + assert_eq!( + harness.walks_stashed(), + 0, + "a cancelled walk answers from no stash" ); assert_eq!( harness.walks(), - (0, 1), - "a hand-back is a terminal outcome of the offloaded walk" + (0, 0), + "a cancelled walk reaches no outcome and counts on neither substrate" ); } diff --git a/src/compute/src/metrics.rs b/src/compute/src/metrics.rs index 5ef662b1d250a..884796a6c23a9 100644 --- a/src/compute/src/metrics.rs +++ b/src/compute/src/metrics.rs @@ -58,7 +58,6 @@ pub struct ComputeMetrics { // look at. timely_step_duration_seconds: HistogramVec, persist_peek_seconds: HistogramVec, - stashed_peek_seconds: HistogramVec, handle_command_duration_seconds: HistogramVec, // Index peek timing phases (per-cluster, no worker label) @@ -73,6 +72,7 @@ pub struct ComputeMetrics { index_peek_frontier_check_seconds: Histogram, index_peek_row_collection_seconds: Histogram, index_peek_walks_total: raw::IntCounterVec, + index_peek_stashed_total: IntCounter, index_peek_permit_queue_depth: UIntGauge, index_peek_permit_wait_seconds: Histogram, index_peek_offload_seconds: Histogram, @@ -206,12 +206,6 @@ impl ComputeMetrics { var_labels: ["worker_id"], buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0), ), role)), - stashed_peek_seconds: registry.register(with_role(metric!( - name: "mz_stashed_peek_seconds", - help: "Time spent reading a peek result and stashing it in the peek result stash (aka. persist blob).", - var_labels: ["worker_id"], - buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0), - ), role)), handle_command_duration_seconds: registry.register(with_role(metric!( name: "mz_cluster_handle_command_duration_seconds", help: "Time spent in handling commands.", @@ -274,6 +268,10 @@ impl ComputeMetrics { help: "The number of index peek walks that reached an outcome, by the substrate they ended on: `inline` on the timely worker, `offloaded` away from it.", var_labels: ["substrate"], ), role)), + index_peek_stashed_total: registry.register(with_role(metric!( + 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`.", + ), role)), index_peek_permit_queue_depth: registry.register(with_role(metric!( name: "mz_index_peek_permit_queue_depth", help: "The number of offloaded index peek walks waiting for a permit to run.", @@ -328,7 +326,6 @@ impl ComputeMetrics { .timely_step_duration_seconds .with_label_values(&[&worker]); let persist_peek_seconds = self.persist_peek_seconds.with_label_values(&[&worker]); - let stashed_peek_seconds = self.stashed_peek_seconds.with_label_values(&[&worker]); let handle_command_duration_seconds = CommandMetrics::build(|typ| { self.handle_command_duration_seconds .with_label_values(&[worker.as_ref(), typ]) @@ -347,6 +344,7 @@ impl ComputeMetrics { let index_peek_walks_offloaded = self .index_peek_walks_total .with_label_values(&["offloaded"]); + let index_peek_stashed_total = self.index_peek_stashed_total.clone(); let index_peek_permit_queue_depth = self.index_peek_permit_queue_depth.clone(); let index_peek_permit_wait_seconds = self.index_peek_permit_wait_seconds.clone(); let index_peek_offload_seconds = self.index_peek_offload_seconds.clone(); @@ -367,7 +365,6 @@ impl ComputeMetrics { arrangement_maintenance_active_info, timely_step_duration_seconds, persist_peek_seconds, - stashed_peek_seconds, handle_command_duration_seconds, index_peek_total_seconds, index_peek_seek_fulfillment_seconds, @@ -381,6 +378,7 @@ impl ComputeMetrics { index_peek_row_collection_seconds, index_peek_walks_inline, index_peek_walks_offloaded, + index_peek_stashed_total, index_peek_permit_queue_depth, index_peek_permit_wait_seconds, index_peek_offload_seconds, @@ -409,8 +407,6 @@ pub struct WorkerMetrics { pub(crate) timely_step_duration_seconds: Histogram, /// Histogram of persist peek durations. pub(crate) persist_peek_seconds: Histogram, - /// Histogram of stashed peek durations. - pub(crate) stashed_peek_seconds: Histogram, /// Histogram of command handling durations. pub(crate) handle_command_duration_seconds: CommandMetrics, /// Histogram of total index peek durations. @@ -441,6 +437,11 @@ pub struct WorkerMetrics { pub(crate) index_peek_walks_inline: IntCounter, /// Counts index peek walks that ran away from the timely worker. pub(crate) index_peek_walks_offloaded: IntCounter, + /// Counts index peek walks that answered from the peek response stash. + /// + /// Resolved when the worker's metrics are built, so it reports zero before the first stashed + /// answer rather than being absent. Whether a peek reached the stash has no other signal. + pub(crate) index_peek_stashed_total: IntCounter, /// How many offloaded index peek walks are waiting for a permit. pub(crate) index_peek_permit_queue_depth: UIntGauge, /// Histogram of how long an offloaded index peek walk waited for its permit. diff --git a/test/sqllogictest/max_result_size.slt b/test/sqllogictest/max_result_size.slt index 671446a1d3bfe..17e76dd8cdded 100644 --- a/test/sqllogictest/max_result_size.slt +++ b/test/sqllogictest/max_result_size.slt @@ -132,15 +132,69 @@ INSERT INTO t1 SELECT * FROM generate_series(1, 10000), repeat('a', 100); statement ok SET cluster TO 'c1'; -# Note: 'total' in the error message here is important because it indicates we failed when -# aggregating the result from multiple workers, as opposed to on any single worker. -query error total result exceeds max size of 1\.0 MiB +# What this pins is the bound across the workers that produced the result, not the one each +# worker applies to its own share: eight workers split 10000 rows of about 110 bytes, so no +# single share comes near the cap and only the total can reject the query. +query error result exceeds max size of 1\.0 MiB SELECT * FROM t1; # For SUBSCRIBE, we need to add a GROUP BY to force an exchange in the dataflow. query error total result exceeds max size of 1\.0 MiB SUBSCRIBE (SELECT * FROM t1 GROUP BY a, b) +# A peek whose answer goes to the peek stash is bounded by `max_result_size` the same way an +# inline one is. Each worker decides on its own whether its share is large enough to stash, so a +# result over the cap arrives as several stashed responses each well under it, and a controller +# that measured only the rows a response carries inline would bound such an answer at the cap +# times the worker count. +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET max_result_size TO '1MB'; +---- +COMPLETE 0 + +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET enable_compute_peek_response_stash TO 'true'; +---- +COMPLETE 0 + +# Low enough that every worker's share of the result below reaches the stash. +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET compute_peek_response_stash_threshold_bytes TO 1024; +---- +COMPLETE 0 + +# Small enough that the per-chunk check the adapter applies on read-back cannot be what +# rejects the result, so what this pins is the bound across the workers that produced it. +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET compute_peek_response_stash_read_batch_size_bytes TO 65536; +---- +COMPLETE 0 + +# The index is what makes this a fast-path index peek, which is the path that writes to the stash. +statement ok +CREATE INDEX t1_idx ON t1 (a); + +query error result exceeds max size of 1\.0 MiB +SELECT * FROM t1; + +statement ok +DROP INDEX t1_idx; + +simple conn=mz_system,user=mz_system +ALTER SYSTEM RESET compute_peek_response_stash_threshold_bytes; +---- +COMPLETE 0 + +simple conn=mz_system,user=mz_system +ALTER SYSTEM RESET compute_peek_response_stash_read_batch_size_bytes; +---- +COMPLETE 0 + +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET enable_compute_peek_response_stash TO 'false'; +---- +COMPLETE 0 + # Regression: `RowSetFinishing` must not reject a result that fits within # `max_result_size`. This folds to a fast-path Constant of 10000 rows with a # `byte_len()` of 1100000, which fits under the 1140000 cap. The finishing path