diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index 5422b6855c8af..c1aeae24ef33f 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -15,7 +15,7 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use differential_dataflow::Hashable; -use differential_dataflow::lattice::Lattice; +use differential_dataflow::lattice::{Lattice, antichain_join}; use differential_dataflow::trace::TraceReader; use mz_compute_client::logging::LoggingConfig; use mz_compute_client::protocol::command::{ @@ -64,13 +64,15 @@ use tokio::sync::{oneshot, watch}; use tracing::{Level, debug, error, info, span, trace, warn}; use uuid::Uuid; -use crate::arrangement::manager::{TraceBundle, TraceManager}; +use crate::arrangement::manager::TraceManager; +use crate::compute_state::error_scan::PeekErrsTrace; +use crate::compute_state::index_traces::{IndexTraces, PeekErrs, PeekOks}; use crate::compute_state::peek_budget::InlineBudget; use crate::compute_state::peek_metrics::{IndexPeekMetrics, PeekWalkMetrics}; pub(crate) use crate::compute_state::peek_offload::PeekPermits; use crate::compute_state::peek_offload::{OffloadConfig, OffloadedPeek}; use crate::compute_state::peek_scan::{ - IndexPeekScan, PeekScan, ScanOutcome, StashBounds, entry_byte_len, rows_response, + IndexPeekScan, PeekOksTrace, PeekScan, ScanOutcome, StashBounds, entry_byte_len, rows_response, }; use crate::logging; use crate::logging::compute::{CollectionLogging, ComputeEvent, PeekEvent}; @@ -81,6 +83,7 @@ use crate::server::{ComputeInstanceContext, ComputeRuntimeRole, ResponseSender}; use crate::sharing::ArrangementSharingRegistry; mod error_scan; +mod index_traces; mod peek_budget; mod peek_metrics; mod peek_offload; @@ -208,6 +211,13 @@ pub struct ComputeState { /// These are polled on every sweep and draw no budget, because the work they are waiting on is /// not running on the worker. pub pending_peeks: VecDeque, + /// Shared-index peeks waiting on a publication or a seal, keyed by the index they wait on. + /// + /// Empty on the maintenance runtime, which reads the traces it maintains. A peek here waits on + /// an event rather than on a turn, so no sweep polls it: it is re-examined only when its key is + /// marked dirty in the sharing registry and the worker is woken. Only peeks wait this way. An + /// interactive dataflow builds immediately and binds its imports through registry placeholders. + pub pending_work: BTreeMap>, /// The persist location where we can stash large peek results. pub peek_stash_persist_location: Option, /// The logger, from Timely's logging framework, if logs are enabled. @@ -345,6 +355,7 @@ impl ComputeState { copy_to_response_buffer: Default::default(), queued_peeks: Default::default(), pending_peeks: Default::default(), + pending_work: Default::default(), peek_stash_persist_location: None, compute_logger: None, persist_clients, @@ -405,6 +416,14 @@ impl ComputeState { probe } + /// Parks `peek` in `pending_work` until its target index is published or seals. + fn enqueue_shared_peek(&mut self, peek: IndexPeek) { + self.pending_work + .entry(peek.peek.target.id()) + .or_default() + .push(peek); + } + /// Apply the current `worker_config` to the compute state. fn apply_worker_config(&mut self) { use mz_compute_types::dyncfgs::*; @@ -413,41 +432,45 @@ impl ComputeState { self.linear_join_spec = LinearJoinSpec::from_config(config); - if ENABLE_LGALLOC.get(config) { - if let Some(path) = &self.context.scratch_directory { - let clear_bytes = LGALLOC_SLOW_CLEAR_BYTES.get(config); - let eager_return = ENABLE_LGALLOC_EAGER_RECLAMATION.get(config); - let file_growth_dampener = LGALLOC_FILE_GROWTH_DAMPENER.get(config); - let interval = LGALLOC_BACKGROUND_INTERVAL.get(config); - let local_buffer_bytes = LGALLOC_LOCAL_BUFFER_BYTES.get(config); - info!( - ?path, - backgrund_interval=?interval, - clear_bytes, - eager_return, - file_growth_dampener, - local_buffer_bytes, - "enabling lgalloc" - ); - let background_worker_config = lgalloc::BackgroundWorkerConfig { - interval, - clear_bytes, - }; - lgalloc::lgalloc_set_config( - lgalloc::LgAlloc::new() - .enable() - .with_path(path.clone()) - .with_background_config(background_worker_config) - .eager_return(eager_return) - .file_growth_dampener(file_growth_dampener) - .local_buffer_bytes(local_buffer_bytes), - ); + // lgalloc is process-global. Only the maintenance runtime configures it; the interactive + // runtime shares the same process and inherits maintenance's configuration. + if self.role.owns_process_globals() { + if ENABLE_LGALLOC.get(config) { + if let Some(path) = &self.context.scratch_directory { + let clear_bytes = LGALLOC_SLOW_CLEAR_BYTES.get(config); + let eager_return = ENABLE_LGALLOC_EAGER_RECLAMATION.get(config); + let file_growth_dampener = LGALLOC_FILE_GROWTH_DAMPENER.get(config); + let interval = LGALLOC_BACKGROUND_INTERVAL.get(config); + let local_buffer_bytes = LGALLOC_LOCAL_BUFFER_BYTES.get(config); + info!( + ?path, + backgrund_interval=?interval, + clear_bytes, + eager_return, + file_growth_dampener, + local_buffer_bytes, + "enabling lgalloc" + ); + let background_worker_config = lgalloc::BackgroundWorkerConfig { + interval, + clear_bytes, + }; + lgalloc::lgalloc_set_config( + lgalloc::LgAlloc::new() + .enable() + .with_path(path.clone()) + .with_background_config(background_worker_config) + .eager_return(eager_return) + .file_growth_dampener(file_growth_dampener) + .local_buffer_bytes(local_buffer_bytes), + ); + } else { + debug!("not enabling lgalloc, scratch directory not specified"); + } } else { - debug!("not enabling lgalloc, scratch directory not specified"); + info!("disabling lgalloc"); + lgalloc::lgalloc_set_config(lgalloc::LgAlloc::new().disable()); } - } else { - info!("disabling lgalloc"); - lgalloc::lgalloc_set_config(lgalloc::LgAlloc::new().disable()); } // Pager backend selection follows scratch-directory availability: @@ -463,12 +486,16 @@ impl ComputeState { mz_ore::pager::set_backend(mz_ore::pager::Backend::Swap); } - crate::memory_limiter::apply_limiter_config(config); + // The memory limiter and the columnation lgalloc region flag are process-global. Only + // maintenance configures them; the interactive runtime inherits maintenance's settings. + if self.role.owns_process_globals() { + crate::memory_limiter::apply_limiter_config(config); - mz_ore::region::ENABLE_LGALLOC_REGION.store( - ENABLE_COLUMNATION_LGALLOC.get(config), - std::sync::atomic::Ordering::Relaxed, - ); + mz_ore::region::ENABLE_LGALLOC_REGION.store( + ENABLE_COLUMNATION_LGALLOC.get(config), + std::sync::atomic::Ordering::Relaxed, + ); + } // NB: arrangement dictionary compression is deliberately NOT applied here. Unlike the // settings above, it is captured once at replica creation (see `handle_create_instance` @@ -608,14 +635,18 @@ impl ComputeState { // every server iteration. self.server_maintenance_interval = COMPUTE_SERVER_MAINTENANCE_INTERVAL.get(config); - let overflowing_behavior = ORE_OVERFLOWING_BEHAVIOR.get(config); - match overflowing_behavior.parse() { - Ok(behavior) => mz_ore::overflowing::set_behavior(behavior), - Err(err) => { - error!( - err, - overflowing_behavior, "Invalid value for ore_overflowing_behavior" - ); + // `set_behavior` mutates a process-global. Only maintenance applies it; the interactive + // runtime inherits the behavior maintenance installs. + if self.role.owns_process_globals() { + let overflowing_behavior = ORE_OVERFLOWING_BEHAVIOR.get(config); + match overflowing_behavior.parse() { + Ok(behavior) => mz_ore::overflowing::set_behavior(behavior), + Err(err) => { + error!( + err, + overflowing_behavior, "Invalid value for ore_overflowing_behavior" + ); + } } } } @@ -736,12 +767,15 @@ impl<'a> ActiveComputeState<'a> { // Apply dictionary compression exactly once, here at instance creation, from the value the // controller captured when the replica was created. We deliberately do NOT re-apply it on // `handle_update_configuration`, so flipping the flag does not retroactively change this - // replica's arrangements. `DICTIONARY_COMPRESSION` is process-global and a replica process - // hosts a single instance, so this single store covers all of the replica's arrangements. - mz_row_spine::DICTIONARY_COMPRESSION.store( - config.arrangement_dictionary_compression, - std::sync::atomic::Ordering::Relaxed, - ); + // replica's arrangements. `DICTIONARY_COMPRESSION` is process-global. Only the maintenance + // runtime stores it; the interactive runtime shares the process and inherits the value, and + // both runtimes host a single instance, so this single store covers all arrangements. + if self.compute_state.role.owns_process_globals() { + mz_row_spine::DICTIONARY_COMPRESSION.store( + config.arrangement_dictionary_compression, + std::sync::atomic::Ordering::Relaxed, + ); + } if let Some(offset) = config.expiration_offset { self.compute_state.apply_expiration_offset(offset); @@ -791,6 +825,37 @@ impl<'a> ActiveComputeState<'a> { &mut self, dataflow: DataflowDescription, ) { + // Tripwire for a Multiplexer routing bug. The Multiplexer routes on `is_peek_dataflow`, and + // this runtime's read path assumes nothing else arrives. Both sides now ask the same + // question of the same description, so a violation means the two runtimes disagree about a + // single dataflow, and it would be rendered against a runtime that cannot serve it. + // + // `soft_assert_or_log!` rather than `debug_assert!`, which compiles out under + // `[profile.optimized]` and `[profile.release]`. Those are the profiles mzcompose and + // `bin/environmentd` build, so a debug assertion here would be absent from the suites that + // exercise two-runtime most broadly. This panics where soft assertions are on and logs an + // error everywhere else, so a routing bug is visible in production without taking the + // replica down for a dataflow that may still render correctly. + mz_ore::soft_assert_or_log!( + self.compute_state.role() != ComputeRuntimeRole::Interactive + || dataflow.is_peek_dataflow(), + "interactive runtime received a dataflow that is not a peek dataflow: \ + export_ids={:?} transient={} single_time={} has_subscribes={} has_copy_tos={}", + dataflow.export_ids().collect::>(), + dataflow.is_transient(), + dataflow.is_single_time(), + dataflow.subscribe_ids().next().is_some(), + dataflow.copy_to_ids().next().is_some(), + ); + + // Every dataflow builds immediately, in command arrival order. Timely allocates per-worker + // channel ids in construction order, so deferring a build until its dependencies publish + // would diverge that order across workers, latently unsound under a multi-worker interactive + // runtime. On the interactive runtime a query dataflow imports its maintenance-index inputs + // from the sharing registry, binding each through a registry placeholder that a maintenance + // publisher adopts later (see `render::import_published_index`). A not-yet-published dependency + // therefore yields an empty import held at the minimum frontier, so the build is always + // possible without waiting. let dataflow_index = Rc::new(self.timely_worker.next_dataflow_index()); let as_of = dataflow.as_of.clone().unwrap(); @@ -912,6 +977,11 @@ impl<'a> ActiveComputeState<'a> { // dataflow can export multiple collections and they all share one suspension token, so the // computation of a dataflow will only start once all its exported collections have been // scheduled. + // + // Every dataflow builds immediately on `CreateDataflow`, inserting its + // `suspended_collections` entry before the `Schedule` that follows in arrival order can + // reach us. A `Schedule` with no entry is therefore a stray or duplicate command, a silent + // no-op. let suspension_token = self.compute_state.suspended_collections.remove(&id); drop(suspension_token); @@ -923,6 +993,30 @@ impl<'a> ActiveComputeState<'a> { } fn handle_allow_compaction(&mut self, id: GlobalId, frontier: Antichain) { + let worker_index = self.timely_worker.index(); + + // A collection this runtime does not host is one its peer publishes and this runtime may + // import. The multiplexer broadcasts `AllowCompaction` for those, and the frontier becomes + // the standing hold: this runtime's own position in the command stream, which the peer's + // publisher bounds its compaction by. An importing dataflow of ours whose `CreateDataflow` + // is still queued here has registered no reader hold yet, so nothing else keeps the + // arrangement at or below the `as_of` it is about to read at. + // + // Hosting is a question about `collections`, not about the id. The peer also renders + // transient collections of its own (subscribes and copy-tos) this runtime has never seen, + // and neither of those may reach `drop_collection`, which would panic on the untracked + // collection or unpublish an arrangement this runtime does not own. + if !self.compute_state.collections.contains_key(&id) { + mz_ore::soft_assert_or_log!( + self.compute_state.role == ComputeRuntimeRole::Interactive, + "compaction for a collection this runtime does not host: {id}" + ); + self.compute_state + .sharing_registry + .note_standing_hold(id, worker_index, &frontier); + return; + } + if frontier.is_empty() { // Indicates that we may drop `id`, as there are no more valid times to read. self.drop_collection(id); @@ -937,9 +1031,19 @@ impl<'a> ActiveComputeState<'a> { fn handle_peek(&mut self, peek: Peek) { let pending = match &peek.target { PeekTarget::Index { id } => { - // Acquire a copy of the trace suitable for fulfilling the peek. - let trace_bundle = self.compute_state.traces.get(id).unwrap().clone(); - PendingPeek::index(peek, trace_bundle) + let traces = match self.compute_state.role { + // The interactive runtime maintains no traces of its own. It reads the + // arrangements the maintenance runtime publishes into the sharing registry. + ComputeRuntimeRole::Interactive => IndexTraces::Shared { + registry: self.compute_state.sharing_registry.clone(), + worker_index: self.timely_worker.index(), + }, + // Acquire a copy of the trace suitable for fulfilling the peek. + ComputeRuntimeRole::Maintenance | ComputeRuntimeRole::Solo => { + IndexTraces::Local(self.compute_state.traces.get(id).unwrap().clone()) + } + }; + PendingPeek::index(peek, traces) } PeekTarget::Persist { metadata, .. } => { let metadata = metadata.clone(); @@ -965,6 +1069,23 @@ impl<'a> ActiveComputeState<'a> { } } + /// Gives a turn to each shared peek waiting on one of the dirtied ids. + /// + /// The interactive worker loop calls this on wake with the dirty set the sharing registry + /// drained. A peek the turn does not resolve returns to [`ComputeState::pending_work`]. + pub(crate) fn resolve_dirty(&mut self, dirty: BTreeSet) { + let mut upper = Antichain::new(); + for id in dirty { + // Taken out first, because serving a peek may park it again under the same key. + let Some(peeks) = self.compute_state.pending_work.remove(&id) else { + continue; + }; + for peek in peeks { + self.serve_index_peek(&mut upper, peek); + } + } + } + fn handle_cancel_peek(&mut self, uuid: Uuid) { let queued = &mut self.compute_state.queued_peeks; if let Some(index) = queued.iter().position(|peek| peek.peek.uuid == uuid) { @@ -973,6 +1094,30 @@ impl<'a> ActiveComputeState<'a> { return; } + // A shared peek waits keyed by its target, so finding it by uuid means a scan. A cancel is + // an event rather than a per-step poll, so the scan is affordable. + let found = self + .compute_state + .pending_work + .iter() + .find_map(|(id, peeks)| { + let index = peeks.iter().position(|peek| peek.peek.uuid == uuid)?; + Some((*id, index)) + }); + if let Some((id, index)) = found { + let peeks = self + .compute_state + .pending_work + .get_mut(&id) + .expect("found above"); + let peek = peeks.remove(index); + if peeks.is_empty() { + self.compute_state.pending_work.remove(&id); + } + self.send_peek_response(PendingPeek::Index(peek), PeekResponse::Canceled); + return; + } + let pending = &mut self.compute_state.pending_peeks; let Some(index) = pending.iter().position(|peek| peek.peek().uuid == uuid) else { return; @@ -1004,6 +1149,11 @@ impl<'a> ActiveComputeState<'a> { // If this collection is an index, remove its trace. self.compute_state.traces.remove(&id); + // Drop any published arrangement for this index from the sharing registry. Done + // unconditionally rather than gated on the role's publish decision, so a slot published by + // a publishing role is always reclaimed. The call is a no-op when nothing was published for + // `id`. + self.compute_state.sharing_registry.remove(&id); // If the collection is unscheduled, remove it from the list of waiting collections. self.compute_state.suspended_collections.remove(&id); @@ -1043,6 +1193,16 @@ impl<'a> ActiveComputeState<'a> { panic!("dataflow server has already initialized logging"); } + // The interactive runtime keeps no introspection state of its own. Its peeks on the logging + // indexes read the maintenance runtime's publications from the registry (see + // `logging::publish_logging_index`), so installing the logging dataflow here would only + // create collections under ids this runtime does not own, which every command handler + // would then have to tell apart from its own. Without a `compute_logger` this runtime's + // own events are not logged. TODO(CPU-222): log them through the maintenance runtime. + if self.compute_state.role() == ComputeRuntimeRole::Interactive { + return; + } + let LoggingTraces { traces, dataflow_index, @@ -1272,6 +1432,7 @@ impl<'a> ActiveComputeState<'a> { batch_bytes: PEEK_RESPONSE_STASH_BATCH_BYTES.get(&self.compute_state.worker_config), }; + let max_result_size = self.compute_state.max_result_size; let metrics = IndexPeekMetrics { seek_fulfillment_seconds: &self .compute_state @@ -1284,7 +1445,7 @@ impl<'a> ActiveComputeState<'a> { let mut unspent = fuel; let status = peek.seek_fulfillment( upper, - self.compute_state.max_result_size, + max_result_size, stash, row_iteration_limit, &mut unspent, @@ -1308,47 +1469,51 @@ impl<'a> ActiveComputeState<'a> { span!(parent: &peek.span, Level::DEBUG, "process_peek_response").entered(); self.send_peek_response(PendingPeek::Index(peek), response); } - PeekStatus::NotReady => self.compute_state.queued_peeks.push_back(peek), + PeekStatus::NotReady => match peek.traces { + // Waits for a turn: the sweep re-examines it on the next activation. + IndexTraces::Local(_) => self.compute_state.queued_peeks.push_back(peek), + // Waits for a publication or a seal: only a dirty mark for its target re-examines + // it, so it leaves the sweep's queue. + IndexTraces::Shared { .. } => self.compute_state.enqueue_shared_peek(peek), + }, PeekStatus::Offload(scan) => { let _span = span!(parent: &peek.span, Level::DEBUG, "offload_index_peek").entered(); - - let permits = Arc::clone(&self.compute_state.peek_permits); - 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, - scan, - stash, - permits, - config, - walk_metrics, - worker, - ); - - self.compute_state - .pending_peeks - .push_back(PendingPeek::Offloaded(offloaded)); + self.offload_index_peek(peek.peek, scan); } } } + /// Hands `scan` to a driver that finishes the walk away from the worker, and records `peek` as + /// one a driver has taken over. + fn offload_index_peek(&mut self, peek: Peek, scan: IndexPeekScan) { + let permits = Arc::clone(&self.compute_state.peek_permits); + 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, + Arc::clone(&self.compute_state.persist_clients), + location, + ) + }); + + let offloaded = + OffloadedPeek::start(peek, scan, stash, permits, config, walk_metrics, worker); + + self.compute_state + .pending_peeks + .push_back(PendingPeek::Offloaded(offloaded)); + } + /// Asks the driver that has taken `pending` over for its outcome, and sends the response when /// one is ready. fn poll_pending_peek(&mut self, mut pending: PendingPeek) { @@ -1623,25 +1788,32 @@ impl PendingPeek { }) } - fn index(peek: Peek, mut trace_bundle: TraceBundle) -> Self { - let empty_frontier = Antichain::new(); - let timestamp_frontier = Antichain::from_elem(peek.timestamp); - trace_bundle - .oks_mut() - .set_logical_compaction(timestamp_frontier.borrow()); - trace_bundle - .errs_mut() - .set_logical_compaction(timestamp_frontier.borrow()); - trace_bundle - .oks_mut() - .set_physical_compaction(empty_frontier.borrow()); - trace_bundle - .errs_mut() - .set_physical_compaction(empty_frontier.borrow()); + /// Builds an index peek over `traces`. + /// + /// Local traces are pinned at the peek's timestamp for the peek's life. Shared ones are not: + /// the compaction frontier a shared peek needs is the one the importing runtime's standing + /// hold in the registry keeps, and its handles are resolved when it runs. + fn index(peek: Peek, mut traces: IndexTraces) -> Self { + if let IndexTraces::Local(trace_bundle) = &mut traces { + let empty_frontier = Antichain::new(); + let timestamp_frontier = Antichain::from_elem(peek.timestamp); + trace_bundle + .oks_mut() + .set_logical_compaction(timestamp_frontier.borrow()); + trace_bundle + .errs_mut() + .set_logical_compaction(timestamp_frontier.borrow()); + trace_bundle + .oks_mut() + .set_physical_compaction(empty_frontier.borrow()); + trace_bundle + .errs_mut() + .set_physical_compaction(empty_frontier.borrow()); + } PendingPeek::Index(IndexPeek { peek, - trace_bundle, + traces, span: tracing::Span::current(), }) } @@ -1879,12 +2051,18 @@ impl PersistPeek { } } -/// An in-progress index-backed peek, and data to eventually fulfill it. +/// An in-progress index-backed peek, and the traces that eventually fulfill it. +/// +/// A peek over shared traces whose arrangement is not yet published, or whose upper has not sealed +/// the peek timestamp, waits in [`ComputeState::pending_work`] until a publication or a seal marks +/// its target id dirty. +/// +/// Note that this intentionally does not implement or derive `Clone`, as each pending peek is +/// meant to be dropped after it has been responded to. pub struct IndexPeek { peek: Peek, - /// The data from which the trace derives. - trace_bundle: TraceBundle, - /// The `tracing::Span` tracking this peek's operation + traces: IndexTraces, + /// The `tracing::Span` tracking this peek's operation. span: tracing::Span, } @@ -1913,34 +2091,35 @@ impl IndexPeek { row_iteration_limit: Option, fuel: &mut usize, metrics: &IndexPeekMetrics<'_>, - ) -> PeekStatus { + ) -> PeekStatus { let method_start = Instant::now(); - self.trace_bundle.oks_mut().read_upper(upper); - if upper.less_equal(&self.peek.timestamp) { - return PeekStatus::NotReady; - } - self.trace_bundle.errs_mut().read_upper(upper); - if upper.less_equal(&self.peek.timestamp) { + // An unpublished shared arrangement parks the peek. The publication that fills its slot + // marks the target dirty, which is what brings the peek back. + let Some((mut oks, mut errs)) = self.traces.resolve(self.peek.target.id()) else { return PeekStatus::NotReady; - } + }; - let read_frontier = self.trace_bundle.compaction_frontier(); - if !read_frontier.less_equal(&self.peek.timestamp) { - let error = format!( - "Arrangement compaction frontier ({:?}) is beyond the time of the attempted read ({})", - read_frontier.elements(), - self.peek.timestamp, - ); - return PeekStatus::Ready(PeekResponse::Error(PeekError::unstructured(error))); + match gate_peek(&self.peek, &mut oks, &mut errs, upper) { + PeekGate::NotReady => return PeekStatus::NotReady, + PeekGate::Compacted(response) => return PeekStatus::Ready(response), + PeekGate::Open => {} } metrics .frontier_check_seconds .observe(method_start.elapsed().as_secs_f64()); - let result = - self.collect_finished_data(max_result_size, stash, row_iteration_limit, fuel, metrics); + let result = Self::walk_traces( + &self.peek, + oks, + errs, + max_result_size, + stash, + row_iteration_limit, + fuel, + metrics, + ); metrics .seek_fulfillment_seconds @@ -1949,61 +2128,138 @@ impl IndexPeek { result } - /// Answers the peek by scanning the traces that fulfil it, for as long as `fuel` allows. + /// Answers `peek` by scanning `oks` and `errs`, for as long as `fuel` allows. /// /// One call opens one scan and either answers from it or hands it on, so nothing survives the /// call. A scan that runs out of fuel with work left leaves with the [`PeekStatus::Offload`] /// that reports it, so the positions it walked are not walked again. - fn collect_finished_data( - &mut self, + fn walk_traces( + peek: &Peek, + mut oks: PeekOks, + mut errs: PeekErrs, max_result_size: u64, stash: StashBounds, row_iteration_limit: Option, fuel: &mut usize, metrics: &IndexPeekMetrics<'_>, - ) -> PeekStatus { - let peek = &self.peek; - let (oks, errs) = self.trace_bundle.oks_errs_mut(); - let mut scan = PeekScan::new(peek, errs, oks, max_result_size, stash); - - let outcome = scan.step(row_iteration_limit, fuel); - - let phases = scan.phases(); - 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); - let start = Instant::now(); - let response = rows_response(rows, &self.peek.finishing.order_by); - metrics.walk.observe_row_collection(start.elapsed()); - response - } - // 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), - }) - } - // 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), + ) -> PeekStatus { + let scan = PeekScan::new(peek, &mut errs, &mut oks, max_result_size, stash); + walk_scan(scan, peek, row_iteration_limit, fuel, metrics) + } +} + +/// Whether a peek's traces admit a read at its timestamp. +enum PeekGate { + /// Both frontiers admit the read. + Open, + /// No `upper` has passed the peek's timestamp, so the answer could still change. + NotReady, + /// Compaction has passed the peek's timestamp. Unlike `NotReady` this never resolves, so it is + /// the peek's answer. + Compacted(PeekResponse), +} + +/// Decides whether `peek` may read `oks` and `errs`, leaving the traces' meet in `upper`. +/// +/// To produce output at `peek.timestamp` we must be certain it is no longer changing. A trace +/// guarantees every future change is at or beyond an element of its `upper`, so an `upper` at or +/// below the timestamp still admits updates. Compaction is the opposite bound: a `since` beyond the +/// timestamp has already destroyed distinctions the read needs. +/// +/// Generic over both traces so a peek served from this runtime's `TraceBundle` and one served from +/// the sharing registry's published handles are gated by the same code rather than by two copies +/// that can drift. +fn gate_peek( + peek: &Peek, + oks: &mut Tr, + errs: &mut ETr, + upper: &mut Antichain, +) -> PeekGate +where + Tr: PeekOksTrace, + ETr: PeekErrsTrace, +{ + oks.read_upper(upper); + if upper.less_equal(&peek.timestamp) { + return PeekGate::NotReady; + } + errs.read_upper(upper); + if upper.less_equal(&peek.timestamp) { + return PeekGate::NotReady; + } + + // The meet of the two traces' logical compaction frontiers, which is what + // `TraceBundle::compaction_frontier` computes over a maintained pair. + let read_frontier = antichain_join( + &oks.get_logical_compaction(), + &errs.get_logical_compaction(), + ); + if !read_frontier.less_equal(&peek.timestamp) { + let error = format!( + "Arrangement compaction frontier ({:?}) is beyond the time of the attempted read ({})", + read_frontier.elements(), + peek.timestamp, + ); + return PeekGate::Compacted(PeekResponse::Error(PeekError::unstructured(error))); + } + + PeekGate::Open +} + +/// Steps a freshly opened `scan` for as long as `fuel` allows, and either answers `peek` from it +/// or hands the scan on to be finished elsewhere. +/// +/// The inline driver, which is what this is, reports a walk it finished and nothing about one it +/// hands on: the driver that finishes an offloaded walk reports every phase of it. +fn walk_scan( + mut scan: PeekScan, + peek: &Peek, + row_iteration_limit: Option, + fuel: &mut usize, + metrics: &IndexPeekMetrics<'_>, +) -> PeekStatus> +where + Tr: PeekOksTrace, + ETr: PeekErrsTrace, +{ + let outcome = scan.step(row_iteration_limit, fuel); + + let phases = scan.phases(); + 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); + let start = Instant::now(); + let response = rows_response(rows, &peek.finishing.order_by); + metrics.walk.observe_row_collection(start.elapsed()); + response + } + // 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), + }) } + // 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), } } /// For keeping track of the state of pending or ready peeks, and managing /// control flow. -enum PeekStatus { +/// +/// `S` is the scan an unfinished walk leaves behind, which follows the traces the walk opened. +enum PeekStatus { /// The frontiers of objects are not yet advanced enough, peek is still /// pending. NotReady, @@ -2013,7 +2269,7 @@ enum PeekStatus { /// 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), + Offload(S), /// The peek result is ready. Ready(PeekResponse), } diff --git a/src/compute/src/compute_state/error_scan.rs b/src/compute/src/compute_state/error_scan.rs index ce93100bb617e..9efe289860887 100644 --- a/src/compute/src/compute_state/error_scan.rs +++ b/src/compute/src/compute_state/error_scan.rs @@ -8,7 +8,7 @@ use std::time::{Duration, Instant}; -use differential_dataflow::trace::{Cursor, TraceReader}; +use differential_dataflow::trace::{Cursor, Navigable, TraceReader}; use mz_compute_client::protocol::response::PeekError; use mz_repr::{Diff, GlobalId, Timestamp}; use timely::order::PartialOrder; @@ -16,19 +16,52 @@ use tracing::error; use crate::arrangement::manager::PaddedTrace; use crate::compute_state::{PeekRowIterationTracker, peek_result_iterator}; +use crate::render::errors::DataflowErrorSer; use crate::typedefs::ErrAgent; /// The error trace of an index, as /// [`TraceBundle::errs_mut`](crate::arrangement::manager::TraceBundle::errs_mut) hands it out. pub(super) type ErrsHandle = PaddedTrace>; +/// A trace an index peek's error walk can read. +/// +/// The bound is spelled once here, so the walk and everything that carries it name the shape +/// rather than restate it. +pub(super) trait PeekErrsTrace: + TraceReader< + Time = Timestamp, + Batch: Navigable< + Cursor: for<'a> Cursor< + Key<'a> = &'a DataflowErrorSer, + TimeGat<'a>: PartialOrder, + DiffGat<'a> = &'a Diff, + >, + >, + > +{ +} + +impl PeekErrsTrace for Tr where + Tr: TraceReader< + Time = Timestamp, + Batch: Navigable< + Cursor: for<'a> Cursor< + Key<'a> = &'a DataflowErrorSer, + TimeGat<'a>: PartialOrder, + DiffGat<'a> = &'a Diff, + >, + >, + > +{ +} + /// A walk over an index peek's error trace, suspendable between cursor positions. /// /// Holds nothing of the ok trace or of the rows a peek returns. A peek reaches those only once /// this walk reports [`ErrorScanStep::Finished`] with an `Ok`. -pub(super) struct ErrorScan { - cursor: peek_result_iterator::TraceCursor, - storage: peek_result_iterator::TraceStorage, +pub(super) struct ErrorScan { + cursor: peek_result_iterator::TraceCursor, + storage: peek_result_iterator::TraceStorage, /// The limit spans this walk and the ok scan after it, so the count accrued here is handed /// on with [`ErrorScanStep::Finished`]. row_iteration_tracker: PeekRowIterationTracker, @@ -48,12 +81,12 @@ pub(super) enum ErrorScanStep { OutOfFuel, } -impl ErrorScan { +impl ErrorScan { /// Opens a walk over `errs`. /// /// The walk starts without a row-iteration limit. The limit in effect is the caller's to /// supply through [`ErrorScan::set_row_iteration_limit`] before each step. - pub(super) fn new(errs: &mut ErrsHandle) -> Self { + pub(super) fn new(errs: &mut Tr) -> Self { let scan_start = Instant::now(); let (cursor, storage) = errs.cursor(); let mut scan = Self::from_cursor(cursor, storage); @@ -63,8 +96,8 @@ impl ErrorScan { /// Opens a walk over an already-opened cursor. pub(super) fn from_cursor( - cursor: peek_result_iterator::TraceCursor, - storage: peek_result_iterator::TraceStorage, + cursor: peek_result_iterator::TraceCursor, + storage: peek_result_iterator::TraceStorage, ) -> Self { Self { cursor, diff --git a/src/compute/src/compute_state/error_scan/tests.rs b/src/compute/src/compute_state/error_scan/tests.rs index 509e96440b670..9392e8b0eca80 100644 --- a/src/compute/src/compute_state/error_scan/tests.rs +++ b/src/compute/src/compute_state/error_scan/tests.rs @@ -51,7 +51,10 @@ pub(crate) fn error_batch( /// Builds a walk over a single-batch error trace holding `updates`, bounded by /// `row_iteration_limit`. -pub(crate) fn error_scan(updates: ErrorUpdates, row_iteration_limit: Option) -> ErrorScan { +pub(crate) fn error_scan( + updates: ErrorUpdates, + row_iteration_limit: Option, +) -> ErrorScan { let storage = vec![error_batch(updates)]; let cursor = CursorList::new(vec![storage[0].cursor()], &storage); let mut scan = ErrorScan::from_cursor(cursor, storage); @@ -79,7 +82,10 @@ pub(crate) fn holding(error: &DataflowErrorSer) -> ErrorUpdates { /// Runs `scan` to an answer in slices of `fuel_per_step` units, and returns that answer, the /// fuel the walk spent, and the number of calls it took. -fn run_sliced(scan: &mut ErrorScan, fuel_per_step: usize) -> (ErrorScanStep, usize, usize) { +fn run_sliced( + scan: &mut ErrorScan, + fuel_per_step: usize, +) -> (ErrorScanStep, usize, usize) { let mut consumed = 0; // Bounded so that a walk which restarts from the first key on each resumption fails the // test instead of hanging it. diff --git a/src/compute/src/compute_state/index_peek_tests.rs b/src/compute/src/compute_state/index_peek_tests.rs index d52ce58ea0a54..9c6653d8aedd1 100644 --- a/src/compute/src/compute_state/index_peek_tests.rs +++ b/src/compute/src/compute_state/index_peek_tests.rs @@ -27,6 +27,7 @@ use super::error_scan::tests::{ ErrorUpdates, PEEK_TIMESTAMP, cancelling, error, error_batch, holding, }; use super::*; +use crate::arrangement::manager::TraceBundle; /// The collection the peeks in these tests read. pub(crate) const TARGET_ID: GlobalId = GlobalId::User(1); @@ -198,10 +199,10 @@ impl TestMetrics { } } - /// How often each metric that `collect_finished_data` can observe into was observed. + /// How often each metric that a walk can observe into was observed. /// /// The two histograms the enclosing `seek_fulfillment` owns are left out, because the - /// tests that read this call `collect_finished_data` directly. + /// tests that read this call [`collect`] directly. fn observations(&self) -> BTreeMap<&'static str, u64> { let metrics = &self.metrics; BTreeMap::from([ @@ -282,8 +283,8 @@ enum Answer { Ready(PeekResponse), } -impl From for Answer { - fn from(status: PeekStatus) -> Self { +impl From> for Answer { + fn from(status: PeekStatus) -> Self { match status { PeekStatus::NotReady => Answer::NotReady, // The scan an offload carries has no comparison of its own. What is comparable @@ -295,10 +296,35 @@ impl From for Answer { } /// An index peek of `peek` over an index holding `keys` and `errors`. +/// Walks `subject` without the frontier gate, so the observations are the walk's alone. +fn collect( + subject: &mut IndexPeek, + max_result_size: u64, + stash: StashBounds, + row_iteration_limit: Option, + fuel: &mut usize, + metrics: &IndexPeekMetrics<'_>, +) -> PeekStatus { + let (oks, errs) = subject + .traces + .resolve(subject.peek.target.id()) + .expect("local traces resolve"); + IndexPeek::walk_traces( + &subject.peek, + oks, + errs, + max_result_size, + stash, + row_iteration_limit, + fuel, + metrics, + ) +} + fn index_peek_over(peek: Peek, keys: &[Row], errors: ErrorUpdates) -> IndexPeek { IndexPeek { peek, - trace_bundle: trace_bundle(keys, errors), + traces: IndexTraces::Local(trace_bundle(keys, errors)), span: tracing::Span::none(), } } @@ -330,7 +356,8 @@ fn a_completed_scan_answers_with_rows_and_reports_every_phase() { ); let metrics = TestMetrics::new(); - let answer = subject.collect_finished_data( + let answer = collect( + &mut subject, u64::MAX, NO_STASH, None, @@ -356,7 +383,8 @@ fn an_error_answered_peek_reports_no_phase_timers() { let mut subject = index_peek_over(index_peek(trivial_finishing(), None), &keys, errors); let metrics = TestMetrics::new(); - let answer = subject.collect_finished_data( + let answer = collect( + &mut subject, u64::MAX, NO_STASH, None, @@ -390,7 +418,8 @@ fn a_scan_that_fills_a_batch_leaves_the_worker_with_fuel_to_spare() { // A threshold of zero bytes is crossed by the first row, so the scan fills a batch well // before the trace runs out and well before unbounded fuel could run out. let mut fuel = unbounded_fuel(); - let answer = subject.collect_finished_data( + let answer = collect( + &mut subject, u64::MAX, STASH_EVERYTHING, None, @@ -420,7 +449,8 @@ fn a_batch_ready_suspension_out_of_fuel_is_offloaded_too() { // suspends holding a full batch and out of fuel, with both causes of a suspension in force // at once. let mut fuel = 1; - let answer = subject.collect_finished_data( + let answer = collect( + &mut subject, u64::MAX, STASH_EVERYTHING, None, @@ -451,8 +481,14 @@ fn a_scan_that_outruns_its_fuel_leaves_the_worker_reporting_nothing() { // An empty error trace is walked out within a position or two, so this fuel is spent // inside the ok walk with most of the six keys still ahead of it. let mut fuel = 2; - let answer = - subject.collect_finished_data(u64::MAX, NO_STASH, None, &mut fuel, &metrics.as_metrics()); + let answer = collect( + &mut subject, + u64::MAX, + NO_STASH, + None, + &mut fuel, + &metrics.as_metrics(), + ); assert_eq!(Answer::from(answer), Answer::Offload); assert_eq!( @@ -482,7 +518,8 @@ fn an_ok_phase_failure_reports_the_phases_the_walk_reached() { // A ceiling of one byte is crossed by the first row the ok walk produces, so the peek // fails inside that walk rather than in the error walk before it. let max_result_size = 1; - let answer = subject.collect_finished_data( + let answer = collect( + &mut subject, max_result_size, NO_STASH, None, diff --git a/src/compute/src/compute_state/index_traces.rs b/src/compute/src/compute_state/index_traces.rs new file mode 100644 index 0000000000000..929e7dbb08c79 --- /dev/null +++ b/src/compute/src/compute_state/index_traces.rs @@ -0,0 +1,125 @@ +// 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. + +//! The traces an index peek reads, from an arrangement this runtime maintains or one the sharing +//! registry publishes. + +use differential_dataflow::trace::TraceReader; +use mz_repr::{Diff, GlobalId, Timestamp}; +use timely::progress::frontier::AntichainRef; + +use crate::arrangement::manager::{PaddedTrace, TraceBundle}; +use crate::compute_state::error_scan::ErrsHandle; +use crate::shared_trace::{SharedErrsHandle, SharedOksHandle}; +use crate::sharing::ArrangementSharingRegistry; +use crate::typedefs::{ErrSpine, RowRowAgent, RowRowSpine}; + +/// Where an index peek finds the traces that answer it. +pub(super) enum IndexTraces { + /// Traces this runtime maintains, pinned for the peek's life. + Local(TraceBundle), + /// An arrangement the sharing registry publishes, resolved on every attempt. A parked peek + /// holds nothing of the arrangement, so an unpublished slot registers no hold at the minimum. + Shared { + registry: ArrangementSharingRegistry, + worker_index: usize, + }, +} + +impl IndexTraces { + /// Handles on the traces of `id` for one attempt, or `None` while a shared index is + /// unpublished. + /// + /// Both variants hand out owned handles so the scan can carry them off the worker. A local + /// handle is a clone of the pinned one, which registers a hold the pinned one already keeps. + pub(super) fn resolve(&mut self, id: GlobalId) -> Option<(PeekOks, PeekErrs)> { + match self { + IndexTraces::Local(bundle) => { + let (oks, errs) = bundle.oks_errs_mut(); + Some((PeekOks::Local(oks.clone()), PeekErrs::Local(errs.clone()))) + } + IndexTraces::Shared { + registry, + worker_index, + } => registry + .handles(&id, *worker_index) + .map(|(oks, errs)| (PeekOks::Shared(oks), PeekErrs::Shared(errs))), + } + } +} + +/// The ok trace an index peek reads. +pub(super) enum PeekOks { + Local(PaddedTrace>), + Shared(SharedOksHandle), +} + +/// The error trace an index peek reads. +pub(super) enum PeekErrs { + Local(ErrsHandle), + Shared(SharedErrsHandle), +} + +/// Both variants read the same batch type, so the enum is a `TraceReader` by delegation. +macro_rules! delegate_trace_reader { + ($ty:ident, $spine:ty) => { + impl TraceReader for $ty { + type Time = Timestamp; + type Batch = <$spine as TraceReader>::Batch; + + fn set_logical_compaction(&mut self, frontier: AntichainRef) { + match self { + $ty::Local(trace) => trace.set_logical_compaction(frontier), + $ty::Shared(trace) => trace.set_logical_compaction(frontier), + } + } + + fn get_logical_compaction(&mut self) -> AntichainRef<'_, Timestamp> { + match self { + $ty::Local(trace) => trace.get_logical_compaction(), + $ty::Shared(trace) => trace.get_logical_compaction(), + } + } + + fn set_physical_compaction(&mut self, frontier: AntichainRef) { + match self { + $ty::Local(trace) => trace.set_physical_compaction(frontier), + $ty::Shared(trace) => trace.set_physical_compaction(frontier), + } + } + + fn get_physical_compaction(&mut self) -> AntichainRef<'_, Timestamp> { + match self { + $ty::Local(trace) => trace.get_physical_compaction(), + $ty::Shared(trace) => trace.get_physical_compaction(), + } + } + + fn map_batches(&self, f: F) { + match self { + $ty::Local(trace) => trace.map_batches(f), + $ty::Shared(trace) => trace.map_batches(f), + } + } + + fn batches_through( + &mut self, + upper: AntichainRef, + ) -> Option> { + match self { + $ty::Local(trace) => trace.batches_through(upper), + $ty::Shared(trace) => trace.batches_through(upper), + } + } + } + }; +} + +delegate_trace_reader!(PeekOks, RowRowSpine); +delegate_trace_reader!(PeekErrs, ErrSpine); diff --git a/src/compute/src/compute_state/peek_offload.rs b/src/compute/src/compute_state/peek_offload.rs index 6c739a097be4b..a90757be5e77d 100644 --- a/src/compute/src/compute_state/peek_offload.rs +++ b/src/compute/src/compute_state/peek_offload.rs @@ -38,8 +38,11 @@ use tracing::{debug, warn}; use uuid::Uuid; use crate::compute_state::PeekRowIterationConfig; +use crate::compute_state::error_scan::PeekErrsTrace; use crate::compute_state::peek_metrics::PeekWalkMetrics; -use crate::compute_state::peek_scan::{IndexPeekScan, RowBatch, ScanOutcome, rows_response}; +use crate::compute_state::peek_scan::{ + PeekOksTrace, PeekScan, RowBatch, ScanOutcome, rows_response, +}; use crate::compute_state::peek_stash::{StashTarget, StashUpload}; /// The bound on how many offloaded peek walks run at once. @@ -221,15 +224,20 @@ impl OffloadedPeek { /// /// 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( + pub(super) fn start( peek: Peek, - scan: IndexPeekScan, + scan: PeekScan, stash: Option, permits: Arc, config: OffloadConfig, metrics: PeekWalkMetrics, worker: Thread, - ) -> Self { + ) -> Self + where + Tr: PeekOksTrace, + ETr: PeekErrsTrace, + PeekScan: Send + 'static, + { let (mut result_tx, result_rx) = oneshot::channel(); permits.resize(config.permit_fraction.get()); @@ -305,14 +313,19 @@ impl OffloadedPeek { /// 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, + async fn walk( + mut state: WalkState, peek_uuid: Uuid, stash: Option, config: &OffloadConfig, metrics: &PeekWalkMetrics, order_by: Arc<[ColumnOrder]>, - ) -> (WalkState, Option) { + ) -> (WalkState, Option) + where + Tr: PeekOksTrace, + ETr: PeekErrsTrace, + PeekScan: Send + 'static, + { // 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 @@ -436,15 +449,23 @@ impl OffloadedPeek { /// on a blocking thread that an abort cannot interrupt, and the permit accounts for that thread /// until the scan leaves it. Fields drop in declaration order, so the scan and its batches go /// before the permit that accounts for them. -struct WalkState { - scan: IndexPeekScan, +struct WalkState +where + Tr: PeekOksTrace, + ETr: PeekErrsTrace, +{ + scan: PeekScan, _permit: WalkPermit, /// 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<(PeekResponse, Duration)>, } -impl WalkState { +impl WalkState +where + Tr: PeekOksTrace, + ETr: PeekErrsTrace, +{ /// Steps the scan until it ends, offers a batch, or the peek is cancelled, whichever comes /// first. `None` is the cancellation. /// diff --git a/src/compute/src/compute_state/peek_offload/tests.rs b/src/compute/src/compute_state/peek_offload/tests.rs index 86b59a97d8ea7..026c3afd61e3b 100644 --- a/src/compute/src/compute_state/peek_offload/tests.rs +++ b/src/compute/src/compute_state/peek_offload/tests.rs @@ -27,7 +27,8 @@ 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, StashBounds}; +use crate::compute_state::index_traces::{PeekErrs, PeekOks}; +use crate::compute_state::peek_scan::{IndexPeekScan, PeekScan, StashBounds}; use crate::compute_state::peek_stash::tests::{CountedBlob, stashed_rows}; use crate::metrics::{ComputeMetrics, WorkerMetrics}; use crate::server::ComputeRuntimeRole; @@ -121,8 +122,8 @@ fn open( let (oks, errs) = bundle.oks_errs_mut(); PeekScan::new( peek, - errs, - oks, + &mut PeekErrs::Local(errs.clone()), + &mut PeekOks::Local(oks.clone()), u64::MAX, StashBounds { eligible: stash_threshold_bytes.is_some(), diff --git a/src/compute/src/compute_state/peek_scan.rs b/src/compute/src/compute_state/peek_scan.rs index 8a2bd9d28ff22..a7e9afa48371f 100644 --- a/src/compute/src/compute_state/peek_scan.rs +++ b/src/compute/src/compute_state/peek_scan.rs @@ -14,7 +14,6 @@ use std::mem; use std::num::{NonZeroI64, NonZeroUsize}; use std::time::{Duration, Instant}; -use differential_dataflow::trace::cursor::BatchCursor; use differential_dataflow::trace::implementations::BatchContainer; use differential_dataflow::trace::{Cursor, Navigable, TraceReader}; use mz_compute_client::protocol::command::Peek; @@ -27,13 +26,48 @@ use mz_repr::fixed_length::ExtendDatums; use mz_repr::{Diff, GlobalId, Row, Timestamp}; use timely::order::PartialOrder; -use crate::compute_state::error_scan::{ErrorScan, ErrorScanStep, ErrsHandle}; +use crate::compute_state::error_scan::{ErrorScan, ErrorScanStep, PeekErrsTrace}; +use crate::compute_state::index_traces::{PeekErrs, PeekOks}; use crate::compute_state::peek_result_iterator::{PeekResultIterator, Step}; -/// The scan an index peek builds, over the ok trace of the arrangement that answers it. -pub(super) type IndexPeekScan = PeekScan< - crate::arrangement::manager::PaddedTrace>, ->; +/// A trace an index peek's ok walk can read. +/// +/// The bound is spelled once here, so the walk and everything that carries it name the shape +/// rather than restate it. +pub(super) trait PeekOksTrace: + TraceReader< + Time = Timestamp, + Batch: Navigable< + Cursor: for<'a> Cursor< + Key<'a>: ExtendDatums + Eq, + KeyContainer: BatchContainer, + Val<'a>: ExtendDatums, + TimeGat<'a>: PartialOrder, + DiffGat<'a> = &'a Diff, + >, + >, + > +{ +} + +impl PeekOksTrace for Tr where + Tr: TraceReader< + Time = Timestamp, + Batch: Navigable< + Cursor: for<'a> Cursor< + Key<'a>: ExtendDatums + Eq, + KeyContainer: BatchContainer, + Val<'a>: ExtendDatums, + TimeGat<'a>: PartialOrder, + DiffGat<'a> = &'a Diff, + >, + >, + > +{ +} + +/// The scan an index peek builds. +pub(super) type IndexPeekScan = PeekScan; /// Rows a scan hands to its driver, in the order the scan produced them. /// @@ -128,9 +162,9 @@ pub(super) enum ScanOutcome { /// The state of a [`PeekScan`]'s walk over its error trace. /// /// Both ended states drop the walk, so a peek pins error batches only while it reads them. -enum ErrorPhase { +enum ErrorPhase { /// The walk is under way, and resumes from the cursor position it stopped on. - Scanning(ErrorScan), + Scanning(ErrorScan), /// The error trace holds no error at the peek's timestamp, which is the only way to the ok /// trace. The rows the walk examined have been handed to the ok walk. Clean, @@ -146,15 +180,16 @@ enum ErrorPhase { /// A stash-eligible scan retains at most the threshold before its first batch and the batch size /// after, plus the row that crossed either. A scan that cannot use the stash fills no batch, and /// `max_result_size` alone bounds its prefix. -pub(super) struct PeekScan +pub(super) struct PeekScan where - Tr: TraceReader, + Tr: PeekOksTrace, + ETr: PeekErrsTrace, { /// The time at which the error trace is read. peek_timestamp: Timestamp, /// The collection the peek reads, for logging. target_id: GlobalId, - error_phase: ErrorPhase, + error_phase: ErrorPhase, /// The walk over the ok trace, reached only once the error walk reports the error trace /// clean. Its cursor is opened with the scan, and nothing advances it before then. oks: PeekResultIterator, @@ -195,16 +230,10 @@ where pub(super) rows_thinned: usize, } -impl PeekScan +impl PeekScan where - Tr: TraceReader, - for<'a> BatchCursor: Cursor< - Key<'a>: ExtendDatums + Eq, - KeyContainer: BatchContainer, - Val<'a>: ExtendDatums, - TimeGat<'a>: PartialOrder, - DiffGat<'a> = &'a Diff, - >, + Tr: PeekOksTrace, + ETr: PeekErrsTrace, { /// Opens a scan of `peek` over the traces that answer it. /// @@ -213,7 +242,7 @@ where /// the caller's to supply to each [`PeekScan::step`]. pub(super) fn new( peek: &Peek, - errs_handle: &mut ErrsHandle, + errs_handle: &mut ETr, oks_handle: &mut Tr, max_result_size: u64, stash: StashBounds, diff --git a/src/compute/src/compute_state/peek_scan/tests.rs b/src/compute/src/compute_state/peek_scan/tests.rs index 7858c62847ca5..459d198999ab5 100644 --- a/src/compute/src/compute_state/peek_scan/tests.rs +++ b/src/compute/src/compute_state/peek_scan/tests.rs @@ -21,6 +21,7 @@ use timely::container::PushInto; use timely::progress::Antichain; use crate::arrangement::manager::{PaddedTrace, TraceBundle}; +use crate::compute_state::error_scan::ErrsHandle; use crate::compute_state::error_scan::tests::PEEK_TIMESTAMP; use crate::compute_state::index_peek_tests::{ answering_errors, cancelling_errors, index_peek, ok_row as row, trace_bundle, trivial_finishing, @@ -116,7 +117,7 @@ fn ok_iterator_with_copies(keys: &[Row], copies: Diff) -> PeekResultIterator ErrorScan { +fn clean_error_scan(keys: usize) -> ErrorScan { crate::compute_state::error_scan::tests::error_scan(cancelling_errors(keys), None) } @@ -125,7 +126,7 @@ fn clean_error_scan(keys: usize) -> ErrorScan { /// Mirrors what [`PeekScan::new`] builds. Tests use this rather than `new` to hold a second /// cursor layout under test, and to start from an [`ErrorPhase`] that a fresh scan cannot be /// in. -fn scan(error_phase: ErrorPhase, keys: &[Row]) -> PeekScan { +fn scan(error_phase: ErrorPhase, keys: &[Row]) -> PeekScan { PeekScan { peek_timestamp: PEEK_TIMESTAMP, target_id: GlobalId::User(1), @@ -613,7 +614,7 @@ fn open( max_result_size: u64, peek_stash_eligible: bool, peek_stash_threshold_bytes: usize, -) -> PeekScan { +) -> PeekScan { let (oks, errs) = bundle.oks_errs_mut(); PeekScan::new( peek, @@ -635,7 +636,7 @@ fn open( /// this reports is comparable across runs that cross the stash threshold and runs that do /// not. fn run_sliced( - subject: &mut PeekScan, + subject: &mut PeekScan, fuel_per_step: usize, row_iteration_limit: Option, ) -> (ScanOutcome, RowBatch, usize) { diff --git a/src/compute/src/compute_state/peek_sweep_tests.rs b/src/compute/src/compute_state/peek_sweep_tests.rs index 810179ce6268d..78ced1854c706 100644 --- a/src/compute/src/compute_state/peek_sweep_tests.rs +++ b/src/compute/src/compute_state/peek_sweep_tests.rs @@ -29,6 +29,7 @@ use super::index_peek_tests::{ TARGET_ID, cancelling_errors, index_peek_with_uuid, rows_answer, trace_bundle, wide_ok_rows, }; use super::*; +use crate::arrangement::manager::TraceBundle; /// The per-peek budget the budget-arming tests configure. Distinct from both the unbounded /// grant and the parameter's default, so that an assertion on it says the configured value was @@ -157,7 +158,8 @@ impl Harness { /// Queues `peek` over `bundle`, as a peek whose frontiers were not yet ready is queued. fn add_pending(&mut self, peek: Peek, bundle: TraceBundle) { - let PendingPeek::Index(pending) = PendingPeek::index(peek, bundle) else { + let PendingPeek::Index(pending) = PendingPeek::index(peek, IndexTraces::Local(bundle)) + else { unreachable!("built as an index peek") }; let uuid = pending.peek.uuid; diff --git a/src/compute/src/compute_state/tests.rs b/src/compute/src/compute_state/tests.rs index 73508ae6a5e8b..3b64abaf74261 100644 --- a/src/compute/src/compute_state/tests.rs +++ b/src/compute/src/compute_state/tests.rs @@ -11,7 +11,38 @@ use mz_dyncfg::ConfigUpdates; +use std::rc::Rc; + +use differential_dataflow::input::Input; +use differential_dataflow::operators::arrange::TraceAgent; +use differential_dataflow::trace::{Builder, Description, Trace}; +use mz_compute_types::dataflows::{BuildDesc, IndexDesc}; +use mz_compute_types::plan::LirRelationExpr; +use mz_expr::{ + AggregateExpr, AggregateFunc, MapFilterProject, MirRelationExpr, MirScalarExpr, + OptimizedMirRelationExpr, RowSetFinishing, +}; +use mz_repr::optimize::OptimizerFeatures; +use mz_repr::{Datum, Diff, RelationDesc, ReprRelationType, SqlScalarType}; +use mz_row_spine::{RowRowBatcher, RowRowBuilder}; +use mz_timely_util::columnation::{ColumnationChunker, ColumnationStack}; +use timely::container::PushInto; +use timely::dataflow::operators::generic::OperatorInfo; +use timely::progress::Timestamp as _; +use uuid::Uuid; + +use mz_persist_client::cache::PersistClientCache; +use mz_secrets::{InMemorySecretsController, SecretsController}; +use mz_storage_types::connections::ConnectionContext; +use mz_txn_wal::operator::TxnsContext; + use super::*; +use crate::arrangement::manager::PaddedTrace; +use crate::arrangement::manager::TraceBundle; +use crate::extensions::arrange::{KeyCollection, MzArrange}; +use crate::render::errors::DataflowErrorSer; +use crate::shared_trace::PublishArrangement; +use crate::typedefs::{ErrAgent, ErrBatcher, ErrBuilder, ErrSpine, RowRowAgent, RowRowSpine}; #[mz_ore::test] fn row_iteration_limit_observes_updates_and_disabled_rows() { @@ -46,3 +77,1064 @@ fn row_iteration_limit_observes_updates_and_disabled_rows() { Err(PeekError::RowIterationLimitExceeded { limit: 5 }) ); } + +fn row(x: i64) -> Row { + Row::pack_slice(&[Datum::Int64(x)]) +} + +/// Builds a one-batch `[0, upper)` oks trace with `rows`, wrapped exactly like a real +/// index's `TraceBundle.oks` (a `PaddedTrace>`), but constructed directly +/// (bypassing rendering a dataflow) for test purposes. +/// +/// The batch is inserted through the `TraceWriter` (not `Trace::insert` on the bare spine +/// directly), because the writer tracks its own idea of the trace's current upper and +/// asserts new batches are contiguous with it; inserting straight into the spine before +/// wrapping desyncs that bookkeeping, and the writer's `Drop` (which seals the trace to the +/// empty frontier) then panics. Closing the trace this way is fine for a test snapshot: an +/// empty (fully closed) upper is readable at any finite peek timestamp. +fn oks_trace_with_rows( + upper: Timestamp, + rows: Vec<((Row, Row), Timestamp, Diff)>, +) -> PaddedTrace> { + let spine: RowRowSpine = + Trace::new(OperatorInfo::new(0, 0, Rc::from(vec![0])), None, None); + let (agent, mut writer) = + TraceAgent::new(spine, OperatorInfo::new(1, 0, Rc::from(vec![0])), None); + + let description = Description::new( + Antichain::from_elem(Timestamp::minimum()), + Antichain::from_elem(upper), + Antichain::from_elem(Timestamp::minimum()), + ); + let mut chunk = ColumnationStack::default(); + for row in rows { + chunk.push_into(row); + } + let batch = RowRowBuilder::::seal(&mut vec![chunk], description); + writer.insert(batch, Some(Timestamp::minimum())); + + agent.into() +} + +/// Builds a one-batch `[0, upper)` errs trace with no errors, wrapped like a real index's +/// `TraceBundle.errs`. +fn errs_trace_empty(upper: Timestamp) -> PaddedTrace> { + let spine: ErrSpine = + Trace::new(OperatorInfo::new(2, 0, Rc::from(vec![0])), None, None); + let (agent, mut writer) = + TraceAgent::new(spine, OperatorInfo::new(3, 0, Rc::from(vec![0])), None); + + let description = Description::new( + Antichain::from_elem(Timestamp::minimum()), + Antichain::from_elem(upper), + Antichain::from_elem(Timestamp::minimum()), + ); + let chunk = ColumnationStack::default(); + let batch = ErrBuilder::::seal(&mut vec![chunk], description); + writer.insert(batch, Some(Timestamp::minimum())); + + agent.into() +} + +/// A peek that may not use the stash, so its whole answer is built inline. +const NO_STASH: StashBounds = StashBounds { + eligible: false, + threshold_bytes: usize::MAX, + batch_bytes: 0, +}; + +/// A peek whose every row is bound for the stash. +const STASH_EVERYTHING: StashBounds = StashBounds { + eligible: true, + threshold_bytes: 0, + batch_bytes: 0, +}; + +/// The metrics an index peek walk observes into, over a registry the test owns. +struct TestMetrics { + metrics: WorkerMetrics, + walk: PeekWalkMetrics, +} + +impl TestMetrics { + fn new() -> Self { + let metrics = crate::metrics::ComputeMetrics::register_with( + &MetricsRegistry::new(), + ComputeRuntimeRole::Maintenance, + ) + .for_worker(0); + let walk = PeekWalkMetrics::new(&metrics); + Self { metrics, walk } + } + + fn as_metrics(&self) -> IndexPeekMetrics<'_> { + IndexPeekMetrics { + seek_fulfillment_seconds: &self.metrics.index_peek_seek_fulfillment_seconds, + frontier_check_seconds: &self.metrics.index_peek_frontier_check_seconds, + walk: &self.walk, + } + } +} + +fn make_peek(timestamp: Timestamp) -> Peek { + let result_desc = RelationDesc::builder() + .with_column("k", SqlScalarType::Int64.nullable(false)) + .with_column("v", SqlScalarType::Int64.nullable(false)) + .finish(); + Peek { + target: PeekTarget::Index { + id: GlobalId::User(1), + }, + result_desc, + literal_constraints: None, + uuid: Uuid::new_v4(), + timestamp, + finishing: RowSetFinishing::trivial(2), + map_filter_project: MapFilterProject::new(2) + .into_plan() + .expect("identity MFP plans") + .into_nontemporal() + .expect("identity MFP has no temporal filters"), + otel_ctx: OpenTelemetryContext::empty(), + } +} + +/// The traces of an index holding `kv` at `peek_ts`, sealed to `upper`, as a maintained index +/// hands them to a peek. +fn kv_trace_bundle(upper: Timestamp, peek_ts: Timestamp, kv: &[(Row, Row)]) -> TraceBundle { + let rows = kv + .iter() + .cloned() + .map(|(k, v)| ((k, v), peek_ts, Diff::ONE)) + .collect(); + TraceBundle::new(oks_trace_with_rows(upper, rows), errs_trace_empty(upper)) +} + +/// Walks `peek` over `bundle` with more fuel than the walk can spend, so the outcome reports +/// where the walk itself ended rather than where the budget cut it off. +fn walk_local( + peek: Peek, + bundle: TraceBundle, + stash: StashBounds, + metrics: &TestMetrics, +) -> PeekStatus { + let mut local = IndexPeek { + peek, + traces: IndexTraces::Local(bundle), + span: tracing::Span::none(), + }; + let mut upper = Antichain::new(); + let mut fuel = usize::MAX; + local.seek_fulfillment( + &mut upper, + u64::MAX, + stash, + None, + &mut fuel, + &metrics.as_metrics(), + ) +} + +/// Walks `peek` over the arrangement `registry` publishes, as the interactive runtime does, with +/// the same fuel [`walk_local`] grants. +fn walk_shared( + registry: &ArrangementSharingRegistry, + worker_index: usize, + peek: Peek, + stash: StashBounds, + metrics: &TestMetrics, +) -> PeekStatus { + let mut shared = IndexPeek { + peek, + traces: IndexTraces::Shared { + registry: registry.clone(), + worker_index, + }, + span: tracing::Span::none(), + }; + let mut upper = Antichain::new(); + let mut fuel = usize::MAX; + shared.seek_fulfillment( + &mut upper, + u64::MAX, + stash, + None, + &mut fuel, + &metrics.as_metrics(), + ) +} + +/// Publishes `rows` (at time 0, sealed to 1) as a real index arrangement into a fresh registry +/// under `id` on worker 0 of 1, mirroring how a maintained index publishes on the maintenance +/// runtime. +fn publish_kv_index(id: GlobalId, rows: Vec<(Row, Row)>) -> ArrangementSharingRegistry { + let registry = ArrangementSharingRegistry::new(); + publish_kv_index_into(®istry, id, rows); + registry +} + +/// Like [`publish_kv_index`], but publishes into an existing `registry`. +fn publish_kv_index_into( + registry: &ArrangementSharingRegistry, + id: GlobalId, + rows: Vec<(Row, Row)>, +) { + let registry_in = registry.clone(); + timely::execute_directly(move |worker| { + // The trace lives as long as an agent does, and the point closes when it drops, so the + // agents must outlive the stepping that seals the batches. Production keeps them in the + // trace manager. `execute_directly` steps only after this closure returns, so step here. + let keep = worker.dataflow::(|scope| { + let (mut oks_input, oks_collection) = scope.new_collection::<(Row, Row), Diff>(); + let oks = oks_collection.mz_arrange::< + ColumnationChunker<_>, + RowRowBatcher<_, _>, + RowRowBuilder<_, _>, + RowRowSpine<_, _>, + >("test oks"); + let (mut errs_input, errs_collection) = + scope.new_collection::(); + let errs = KeyCollection::from(errs_collection).mz_arrange::< + ColumnationChunker<_>, + ErrBatcher<_, _>, + ErrBuilder<_, _>, + ErrSpine<_, _>, + >("test errs"); + + let slot = registry_in.get_or_create(id, 0, 1); + PublishArrangement::adopt(&oks, &slot.oks, || {}); + PublishArrangement::adopt(&errs, &slot.errs, || {}); + registry_in.notify(id, 0); + + for (k, v) in rows { + oks_input.update((k, v), Diff::ONE); + } + oks_input.advance_to(Timestamp::from(1_u64)); + oks_input.flush(); + errs_input.advance_to(Timestamp::from(1_u64)); + errs_input.flush(); + (oks.trace.clone(), errs.trace.clone()) + }); + while worker.step() {} + drop(keep); + }); +} + +/// The interactive inline walk over the sharing registry returns the same `PeekResponse` as the +/// maintenance runtime's local trace walk over the same rows. +#[mz_ore::test] +#[cfg_attr(miri, ignore)] // differential-dataflow's Columnation isn't miri-clean +fn interactive_shared_peek_matches_local_path() { + let metrics = TestMetrics::new(); + + let kv = vec![(row(1), row(10)), (row(2), row(20)), (row(3), row(30))]; + let peek_ts = Timestamp::new(0); + let trace_upper = Timestamp::new(1); + + // The maintenance runtime's walk over an equivalent, locally built trace bundle. + let bundle = kv_trace_bundle(trace_upper, peek_ts, &kv); + let local_response = match walk_local(make_peek(peek_ts), bundle, NO_STASH, &metrics) { + PeekStatus::Ready(response) => response, + _ => panic!("a walk with fuel to spare must answer"), + }; + + // The interactive runtime's walk: publish the same rows and read them off the registry. + let shared_registry = publish_kv_index(GlobalId::User(1), kv.clone()); + let shared_response = + match walk_shared(&shared_registry, 0, make_peek(peek_ts), NO_STASH, &metrics) { + PeekStatus::Ready(response) => response, + _ => panic!("a walk with fuel to spare must answer"), + }; + + assert_eq!( + local_response, shared_response, + "shared-registry peek must return the local path's rows" + ); +} + +/// The interactive walk defers an over-threshold result to the peek stash, exactly as the +/// maintenance walk does, rather than returning it inline. +/// +/// An interactive walk that could not reach the stash would answer inline, and a result over +/// `max_result_size` would then fail with "result exceeds max size" on a query that streams fine +/// through the stash on the maintenance runtime. Every peek routes to interactive while the +/// feature is on, and no other test approaches the limit, so nothing else would catch it. +#[mz_ore::test] +#[cfg_attr(miri, ignore)] // differential-dataflow's Columnation isn't miri-clean +fn interactive_shared_peek_defers_over_threshold_result_to_the_stash() { + let metrics = TestMetrics::new(); + + let kv = vec![(row(1), row(10)), (row(2), row(20)), (row(3), row(30))]; + let peek_ts = Timestamp::new(0); + let trace_upper = Timestamp::new(1); + + // The walk stops with a batch to hand over rather than answering inline, which is how a + // result too large for an inline answer reaches the stash: the driver that finishes the walk + // writes the batch. + let shared_registry = publish_kv_index(GlobalId::User(1), kv.clone()); + let shared_scan = match walk_shared( + &shared_registry, + 0, + make_peek(peek_ts), + STASH_EVERYTHING, + &metrics, + ) { + PeekStatus::Offload(scan) => scan, + other => panic!( + "an over-threshold interactive walk must stop with a batch, got {}", + status_name(&other) + ), + }; + assert!( + shared_scan.stash_eligible() && shared_scan.batch_ready(), + "the suspended interactive walk must hold a batch bound for the stash" + ); + + // The maintenance walk over the same rows makes the same call, which is the property that + // matters: routing a peek to interactive must not change whether it stashes. + let bundle = kv_trace_bundle(trace_upper, peek_ts, &kv); + let local_scan = match walk_local(make_peek(peek_ts), bundle, STASH_EVERYTHING, &metrics) { + PeekStatus::Offload(scan) => scan, + other => panic!( + "an over-threshold maintenance walk must stop with a batch, got {}", + status_name(&other) + ), + }; + assert!( + local_scan.stash_eligible() && local_scan.batch_ready(), + "the suspended maintenance walk must hold a batch bound for the stash" + ); +} + +/// Names a [`PeekStatus`] for an assertion message, which the scan it may carry cannot render. +fn status_name(status: &PeekStatus) -> &'static str { + match status { + PeekStatus::NotReady => "NotReady", + PeekStatus::Offload(_) => "Offload", + PeekStatus::Ready(_) => "Ready", + } +} + +/// A local index peek whose timestamp has been compacted past returns a compaction-frontier +/// error. The interactive inline path mirrors this exact gate over the registry handles, so +/// this asserts the error string the shared path reproduces. +#[mz_ore::test] +#[cfg_attr(miri, ignore)] +fn seek_fulfillment_compacted_past_errors() { + let metrics = TestMetrics::new(); + + // A peek at time 1, against a trace that has compacted its logical frontier to time 5: the + // read is beyond the trace's compaction frontier. + let peek_timestamp = Timestamp::new(1); + let trace_upper = Timestamp::new(10); + + let mut bundle = kv_trace_bundle(trace_upper, peek_timestamp, &[]); + let compacted = Antichain::from_elem(Timestamp::new(5)); + bundle.oks_mut().set_logical_compaction(compacted.borrow()); + bundle.errs_mut().set_logical_compaction(compacted.borrow()); + + let response = match walk_local(make_peek(peek_timestamp), bundle, NO_STASH, &metrics) { + PeekStatus::Ready(response) => response, + _ => panic!("a compacted-past read must resolve directly"), + }; + assert!( + matches!(&response, PeekResponse::Error(PeekError::Unstructured(msg)) if msg.contains("compaction frontier")), + "expected a compaction-frontier error, got {response:?}", + ); +} + +/// A peek for an index that is not yet published defers via `NotReady` (rather than blocking or +/// erroring), and resolves with the correct rows once the maintenance runtime publishes and the +/// pending-peek retry runs again. +#[mz_ore::test] +#[cfg_attr(miri, ignore)] +fn interactive_shared_peek_defers_until_published() { + let metrics = TestMetrics::new(); + let id = GlobalId::User(1); + let kv = vec![(row(1), row(10)), (row(2), row(20))]; + let registry = ArrangementSharingRegistry::new(); + + let peek = make_peek(Timestamp::new(0)); + assert!( + matches!( + walk_shared(®istry, 0, peek.clone(), NO_STASH, &metrics), + PeekStatus::NotReady, + ), + "an unpublished index must defer", + ); + + // Publishing lets the turn a dirty mark grants resolve the peek. + publish_kv_index_into(®istry, id, kv.clone()); + assert!( + matches!( + walk_shared(®istry, 0, peek, NO_STASH, &metrics), + PeekStatus::Ready(PeekResponse::Rows(_)), + ), + "a published index must resolve", + ); +} + +/// A peek at a timestamp the arrangement's upper has not yet sealed defers via `NotReady`, then +/// resolves once the upper advances past the peek timestamp. Uses a live worker so the +/// published trace carries a finite (non-empty) upper, which `execute_directly`'s +/// run-to-completion sealing cannot stage. +#[mz_ore::test] +#[cfg_attr(miri, ignore)] +fn interactive_shared_peek_defers_until_sealed() { + let id = GlobalId::User(1); + timely::execute_directly(move |worker| { + let metrics = TestMetrics::new(); + let registry = ArrangementSharingRegistry::new(); + let registry_in = registry.clone(); + let worker_index = worker.index(); + let peers = worker.peers(); + + let (mut oks_input, mut errs_input, _keep) = + worker.dataflow::(move |scope| { + let (oks_input, oks_collection) = scope.new_collection::<(Row, Row), Diff>(); + let oks = oks_collection.mz_arrange::< + ColumnationChunker<_>, + RowRowBatcher<_, _>, + RowRowBuilder<_, _>, + RowRowSpine<_, _>, + >("test oks"); + let (errs_input, errs_collection) = + scope.new_collection::(); + let errs = KeyCollection::from(errs_collection).mz_arrange::< + ColumnationChunker<_>, + ErrBatcher<_, _>, + ErrBuilder<_, _>, + ErrSpine<_, _>, + >("test errs"); + + let slot = registry_in.get_or_create(id, worker_index, peers); + PublishArrangement::adopt(&oks, &slot.oks, || {}); + PublishArrangement::adopt(&errs, &slot.errs, || {}); + registry_in.notify(id, worker_index); + ( + oks_input, + errs_input, + (oks.trace.clone(), errs.trace.clone()), + ) + }); + + // A row at time 0, batch sealed so the trace's upper is {1}. + oks_input.update((row(1), row(10)), Diff::ONE); + oks_input.advance_to(Timestamp::from(1_u64)); + oks_input.flush(); + errs_input.advance_to(Timestamp::from(1_u64)); + errs_input.flush(); + for _ in 0..16 { + worker.step(); + } + + // upper {1} does not seal a peek at time 1: defer. + assert!( + matches!( + walk_shared( + ®istry, + worker_index, + make_peek(Timestamp::new(1)), + NO_STASH, + &metrics, + ), + PeekStatus::NotReady, + ), + "an unsealed peek must defer", + ); + + // Advance the upper past the peek timestamp; the retry now resolves. + oks_input.advance_to(Timestamp::from(2_u64)); + oks_input.flush(); + errs_input.advance_to(Timestamp::from(2_u64)); + errs_input.flush(); + for _ in 0..16 { + worker.step(); + } + + assert!( + matches!( + walk_shared( + ®istry, + worker_index, + make_peek(Timestamp::new(1)), + NO_STASH, + &metrics, + ), + PeekStatus::Ready(PeekResponse::Rows(_)), + ), + "a sealed peek must resolve", + ); + + // Keep the publisher inputs alive until here so the publication stayed open. + let _keep = (&oks_input, &errs_input); + }); +} + +fn test_compute_instance_context() -> ComputeInstanceContext { + ComputeInstanceContext { + scratch_directory: None, + worker_core_affinity: false, + connection_context: ConnectionContext::for_tests(InMemorySecretsController::new().reader()), + } +} + +/// Builds a persist client cache inside a Tokio runtime context, which its pubsub task needs. +/// Returns the runtime too so the caller keeps it alive for the cache's lifetime. The cache is +/// an `Arc` (so `Send`) and can move into a timely worker closure, unlike the `Rc`-holding +/// `ComputeState`, which must be built on the worker thread. +fn test_persist_clients() -> (tokio::runtime::Runtime, Arc) { + let runtime = tokio::runtime::Runtime::new().expect("tokio runtime"); + let clients = { + let _guard = runtime.enter(); + Arc::new(PersistClientCache::new_no_metrics()) + }; + (runtime, clients) +} + +/// Builds an interactive-runtime `ComputeState` over `registry`, with a fresh, isolated metrics +/// registry. Enough to drive `handle_peek`/`resolve_dirty` in a test `ActiveComputeState`. +fn interactive_compute_state( + persist_clients: Arc, + registry: ArrangementSharingRegistry, +) -> ComputeState { + let metrics_registry = MetricsRegistry::new(); + let metrics = crate::metrics::ComputeMetrics::register_with( + &metrics_registry, + ComputeRuntimeRole::Interactive, + ) + .for_worker(0); + ComputeState::new( + ComputeRuntimeRole::Interactive, + persist_clients, + registry, + TxnsContext::default(), + metrics, + Arc::new(TracingHandle::disabled()), + test_compute_instance_context(), + metrics_registry, + 1, + Arc::new(PeekPermits::new(1)), + None, + ) +} + +/// Publishes `rows` as a `RowRow` index under `id` on the CURRENT worker (no nested +/// `execute_directly`), sealing the batch and draining to the empty upper. Returns the trace +/// agents, which the caller keeps for as long as it reads: the point closes with the trace. +fn publish_index_current_worker( + worker: &mut TimelyWorker, + registry: &ArrangementSharingRegistry, + id: GlobalId, + rows: Vec<(Row, Row)>, +) -> (RowRowAgent, ErrAgent) { + let registry_in = registry.clone(); + let (mut oks_input, mut errs_input, keep) = worker.dataflow::(move |scope| { + let (oks_input, oks_collection) = scope.new_collection::<(Row, Row), Diff>(); + let oks = oks_collection.mz_arrange::< + ColumnationChunker<_>, + RowRowBatcher<_, _>, + RowRowBuilder<_, _>, + RowRowSpine<_, _>, + >("test oks"); + let (errs_input, errs_collection) = scope.new_collection::(); + let errs = KeyCollection::from(errs_collection).mz_arrange::< + ColumnationChunker<_>, + ErrBatcher<_, _>, + ErrBuilder<_, _>, + ErrSpine<_, _>, + >("test errs"); + + let slot = registry_in.get_or_create(id, scope.index(), scope.peers()); + PublishArrangement::adopt(&oks, &slot.oks, || {}); + PublishArrangement::adopt(&errs, &slot.errs, || {}); + registry_in.notify(id, scope.index()); + ( + oks_input, + errs_input, + (oks.trace.clone(), errs.trace.clone()), + ) + }); + + for (k, v) in rows { + oks_input.update((k, v), Diff::ONE); + } + oks_input.advance_to(Timestamp::from(1_u64)); + oks_input.flush(); + errs_input.advance_to(Timestamp::from(1_u64)); + errs_input.flush(); + for _ in 0..16 { + worker.step(); + } + // Drop the inputs and drain: the batch seals to the empty upper, readable at any finite ts. + // The returned agents keep the trace, and with it the publication, alive. + drop(oks_input); + drop(errs_input); + for _ in 0..16 { + worker.step(); + } + keep +} + +/// A peek issued before its index is published enqueues in `pending_work` (never the maintenance +/// `pending_peeks` poll path) and is served only when the target id is presented as dirty to +/// `resolve_dirty`. A re-examination with an empty dirty set, even after the data is published +/// and ready, serves nothing: this is the no-polling property. +#[mz_ore::test] +#[cfg_attr(miri, ignore)] +fn interactive_peek_resolves_on_publication_not_on_bare_tick() { + let id = GlobalId::User(1); + let kv = vec![(row(1), row(10)), (row(2), row(20))]; + // The persist cache spawns a task that needs a Tokio reactor; build it (and keep the + // runtime alive) before entering the timely worker thread. + let (_rt, persist_clients) = test_persist_clients(); + + timely::execute_directly(move |worker| { + let registry = ArrangementSharingRegistry::new(); + // Part A: register this interactive worker's waker, as startup does. + registry.register_waker(0, std::thread::current()); + + let mut compute_state = interactive_compute_state(persist_clients, registry.clone()); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let mut response_tx = ResponseSender::for_test(tx); + + // A peek issued before publication enqueues, does not respond, and does not touch the + // poll path. + { + let mut active = ActiveComputeState { + timely_worker: &mut *worker, + compute_state: &mut compute_state, + response_tx: &mut response_tx, + }; + active.handle_peek(make_peek(Timestamp::new(0))); + assert_eq!( + active + .compute_state + .pending_work + .values() + .map(Vec::len) + .sum::(), + 1, + "an unpublished peek must enqueue in pending_work" + ); + assert!( + active.compute_state.pending_peeks.is_empty(), + "the interactive peek must not use the pending_peeks poll path" + ); + assert!( + active.compute_state.pending_work.contains_key(&id), + "the peek must be indexed under its target id" + ); + + // No-polling: with no dirtied id, re-examination serves nothing. + active.resolve_dirty(BTreeSet::new()); + } + assert!(rx.try_recv().is_err(), "no response before publication"); + + // Publish the index from this same worker. `insert` marks the id dirty for worker 0. + let _keep = publish_index_current_worker(worker, ®istry, id, kv.clone()); + + // No-polling: the data is now published and ready, yet a re-examination with an empty + // dirty set must NOT serve the peek. Only a dirtied id triggers work. + { + let mut active = ActiveComputeState { + timely_worker: &mut *worker, + compute_state: &mut compute_state, + response_tx: &mut response_tx, + }; + active.resolve_dirty(BTreeSet::new()); + } + assert!( + rx.try_recv().is_err(), + "a bare tick (empty dirty set) must not resolve pending work" + ); + + // The genuine wake: drain the dirty inbox (the id, marked by `insert`) and resolve. + let dirty = registry.take_dirty(0); + assert_eq!( + dirty, + BTreeSet::from([id]), + "publication must have marked the id dirty" + ); + { + let mut active = ActiveComputeState { + timely_worker: &mut *worker, + compute_state: &mut compute_state, + response_tx: &mut response_tx, + }; + active.resolve_dirty(dirty); + assert!( + active.compute_state.pending_work.is_empty(), + "a served peek is removed from the store" + ); + } + let response = match rx.try_recv() { + Ok((ComputeResponse::PeekResponse(_, response, _), _)) => response, + other => panic!("expected a peek response, got {other:?}"), + }; + assert!( + matches!(response, PeekResponse::Rows(_)), + "the served peek must carry rows, got {response:?}" + ); + }); +} + +/// A published-but-not-sealed peek stays enqueued and is served only after a frontier advance +/// drives `note_frontier` (the seal signal `export_index` wires). A re-examination after the +/// seal but with no dirty mark serves nothing (no-polling). +#[mz_ore::test] +#[cfg_attr(miri, ignore)] +fn interactive_peek_resolves_on_seal_via_note_frontier() { + let id = GlobalId::User(1); + let (_rt, persist_clients) = test_persist_clients(); + + timely::execute_directly(move |worker| { + let registry = ArrangementSharingRegistry::new(); + let worker_index = worker.index(); + registry.register_waker(worker_index, std::thread::current()); + + let mut compute_state = interactive_compute_state(persist_clients, registry.clone()); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let mut response_tx = ResponseSender::for_test(tx); + + // Publish a row at time 0, sealing only to upper {1}. + let registry_in = registry.clone(); + let (mut oks_input, mut errs_input, _keep) = + worker.dataflow::(move |scope| { + let (oks_input, oks_collection) = scope.new_collection::<(Row, Row), Diff>(); + let oks = oks_collection.mz_arrange::< + ColumnationChunker<_>, + RowRowBatcher<_, _>, + RowRowBuilder<_, _>, + RowRowSpine<_, _>, + >("test oks"); + let (errs_input, errs_collection) = + scope.new_collection::(); + let errs = KeyCollection::from(errs_collection).mz_arrange::< + ColumnationChunker<_>, + ErrBatcher<_, _>, + ErrBuilder<_, _>, + ErrSpine<_, _>, + >("test errs"); + + let slot = registry_in.get_or_create(id, scope.index(), scope.peers()); + PublishArrangement::adopt(&oks, &slot.oks, || {}); + PublishArrangement::adopt(&errs, &slot.errs, || {}); + registry_in.notify(id, scope.index()); + ( + oks_input, + errs_input, + (oks.trace.clone(), errs.trace.clone()), + ) + }); + + oks_input.update((row(1), row(10)), Diff::ONE); + oks_input.advance_to(Timestamp::from(1_u64)); + oks_input.flush(); + errs_input.advance_to(Timestamp::from(1_u64)); + errs_input.flush(); + for _ in 0..16 { + worker.step(); + } + // Drain the publication's dirty mark so the seal signal is observed in isolation. + let _ = registry.take_dirty(worker_index); + + // A peek at ts 1: published but not sealed (upper {1}). Enqueues. + { + let mut active = ActiveComputeState { + timely_worker: &mut *worker, + compute_state: &mut compute_state, + response_tx: &mut response_tx, + }; + active.handle_peek(make_peek(Timestamp::new(1))); + assert_eq!( + active + .compute_state + .pending_work + .values() + .map(Vec::len) + .sum::(), + 1, + "an unsealed peek must enqueue" + ); + } + assert!(rx.try_recv().is_err(), "an unsealed peek does not respond"); + + // Advance the upper past the peek ts and step, so the shared trace seals ts 1. + oks_input.advance_to(Timestamp::from(2_u64)); + oks_input.flush(); + errs_input.advance_to(Timestamp::from(2_u64)); + errs_input.flush(); + for _ in 0..16 { + worker.step(); + } + + // No-polling: the seal alone does not re-examine the peek until the id is dirtied. + { + let mut active = ActiveComputeState { + timely_worker: &mut *worker, + compute_state: &mut compute_state, + response_tx: &mut response_tx, + }; + active.resolve_dirty(BTreeSet::new()); + } + assert!( + rx.try_recv().is_err(), + "a seal with no dirty mark must not resolve the peek" + ); + + // The seal signal: `export_index`'s frontier hook calls `note_frontier`. Drive it. + registry.notify(id, worker_index); + let dirty = registry.take_dirty(worker_index); + assert_eq!(dirty, BTreeSet::from([id])); + { + let mut active = ActiveComputeState { + timely_worker: &mut *worker, + compute_state: &mut compute_state, + response_tx: &mut response_tx, + }; + active.resolve_dirty(dirty); + assert!( + active.compute_state.pending_work.is_empty(), + "a sealed peek is served and removed" + ); + } + let response = match rx.try_recv() { + Ok((ComputeResponse::PeekResponse(_, response, _), _)) => response, + other => panic!("expected a peek response, got {other:?}"), + }; + assert!( + matches!(response, PeekResponse::Rows(_)), + "the sealed peek must carry rows, got {response:?}" + ); + + // Keep the publisher inputs alive until here so the publication stayed open. + let _keep = (&oks_input, &errs_input); + }); +} + +/// A `(k, v)` `ReprRelationType` of two non-null `int64` columns, matching the rows +/// [`publish_index_current_worker`] publishes. +fn two_int64_type() -> ReprRelationType { + let desc = RelationDesc::builder() + .with_column("k", SqlScalarType::Int64.nullable(false)) + .with_column("v", SqlScalarType::Int64.nullable(false)) + .finish(); + ReprRelationType::from(desc.typ()) +} + +/// Converts a lowered index-only dataflow into the `` shape the +/// compute protocol ships, mirroring `compute-client`'s `Instance::create_dataflow`. The test +/// dataflows import only shared indexes (no storage sources) and export no sinks, so the augment +/// step is trivial. +fn to_render_dataflow( + lowered: DataflowDescription, +) -> DataflowDescription { + assert!( + lowered.source_imports.is_empty(), + "index-only test dataflow imports no storage sources" + ); + let objects_to_build = lowered + .objects_to_build + .into_iter() + .map(|o| BuildDesc { + id: o.id, + plan: RenderPlan::try_from(o.plan).expect("render plan conversion"), + }) + .collect(); + DataflowDescription { + source_imports: BTreeMap::new(), + objects_to_build, + index_imports: lowered.index_imports, + index_exports: lowered.index_exports, + sink_exports: BTreeMap::new(), + as_of: lowered.as_of, + until: lowered.until, + initial_storage_as_of: lowered.initial_storage_as_of, + refresh_schedule: lowered.refresh_schedule, + debug_name: lowered.debug_name, + time_dependence: lowered.time_dependence, + } +} + +/// A real query dataflow that imports the maintenance index `index_id` (arranging `on_id` by +/// `[0]`) and exports `out_index_id` = `count(*)` over it. Built by lowering hand-written MIR, +/// exactly as the controller would ship it. No optimization is needed: a reduce lowers +/// faithfully. +fn reduce_count_dataflow( + index_id: GlobalId, + on_id: GlobalId, + reduce_id: GlobalId, + out_index_id: GlobalId, + as_of: Timestamp, +) -> DataflowDescription { + let on_type = two_int64_type(); + let mut mir = DataflowDescription::::new("test-reduce".into()); + mir.import_index( + index_id, + IndexDesc { + on_id, + key: vec![MirScalarExpr::column(0)], + }, + on_type.clone(), + false, + ); + let count = AggregateExpr { + func: AggregateFunc::Count, + expr: MirScalarExpr::literal_true(), + distinct: false, + }; + let reduce = MirRelationExpr::Reduce { + input: Box::new(MirRelationExpr::global_get(on_id, on_type)), + group_key: vec![], + aggregates: vec![count], + monotonic: false, + expected_group_size: None, + }; + let reduce_type = reduce.typ(); + mir.insert_plan( + reduce_id, + OptimizedMirRelationExpr::declare_optimized(reduce), + ); + mir.set_as_of(Antichain::from_elem(as_of)); + mir.export_index( + out_index_id, + IndexDesc { + on_id: reduce_id, + key: vec![MirScalarExpr::column(0)], + }, + reduce_type, + ); + let lowered = LirRelationExpr::finalize_dataflow(mir, &OptimizerFeatures::default(), None) + .expect("lowering the reduce dataflow"); + to_render_dataflow(lowered) +} + +/// A peek over a single-column `int64` result, for reading a `count(*)` query output. +fn make_count_peek(id: GlobalId, timestamp: Timestamp) -> Peek { + let result_desc = RelationDesc::builder() + .with_column("count", SqlScalarType::Int64.nullable(false)) + .finish(); + Peek { + target: PeekTarget::Index { id }, + result_desc, + literal_constraints: None, + uuid: Uuid::new_v4(), + timestamp, + finishing: RowSetFinishing::trivial(1), + map_filter_project: MapFilterProject::new(1) + .into_plan() + .expect("identity MFP plans") + .into_nontemporal() + .expect("identity MFP has no temporal filters"), + otel_ctx: OpenTelemetryContext::empty(), + } +} + +/// An interactive query dataflow that imports a not-yet-published maintenance index is built +/// IMMEDIATELY in arrival order. The import binds through a registry placeholder rather than +/// deferring, so the output collection appears in `collections` right away and nothing lands in +/// `pending_work`. With the placeholder unadopted, the import produces no data and the output +/// frontier holds at the minimum, so a result peek at the as_of stays pending. +#[mz_ore::test] +#[cfg_attr(miri, ignore)] +fn interactive_build_is_immediate() { + let index_id = GlobalId::User(1); + let on_id = GlobalId::User(2); + let reduce_id = GlobalId::User(3); + // Transient with a non-empty `until`, matching the bounded-read contract the Multiplexer + // enforces for anything it routes to the interactive runtime (see the debug_assert in + // `handle_create_dataflow`). + let out_index_id = GlobalId::Transient(4); + let (_rt, persist_clients) = test_persist_clients(); + + timely::execute_directly(move |worker| { + let registry = ArrangementSharingRegistry::new(); + registry.register_waker(0, std::thread::current()); + let mut compute_state = interactive_compute_state(persist_clients, registry.clone()); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let mut response_tx = ResponseSender::for_test(tx); + + let as_of = Timestamp::new(0); + let mut dataflow = reduce_count_dataflow(index_id, on_id, reduce_id, out_index_id, as_of); + dataflow.until = Antichain::from_elem(as_of.step_forward()); + + // The build is NOT deferred, even though the imported index is unpublished: the import + // binds through a placeholder that a maintenance publisher adopts later. + { + let mut active = ActiveComputeState { + timely_worker: &mut *worker, + compute_state: &mut compute_state, + response_tx: &mut response_tx, + }; + active.handle_create_dataflow(dataflow); + assert!( + active.compute_state.pending_work.is_empty(), + "an immediately-built dataflow must not sit in pending_work" + ); + assert!( + active.compute_state.collections.contains_key(&out_index_id), + "the query output collection is built immediately" + ); + // Binding the import through get-or-create created a placeholder slot for the + // unpublished dependency, so its handles now exist. + assert!( + active + .compute_state + .sharing_registry + .handles(&index_id, 0) + .is_some(), + "the interactive import created a placeholder slot for its dependency" + ); + + // Start the (suspended) dataflow, as a `Schedule` command would. + active.handle_schedule(out_index_id); + } + + // Step so the reduce runs over the empty, unadopted placeholder input. Its output frontier + // is held at the minimum, so it never seals past the peek time. + for _ in 0..64 { + worker.step(); + } + + // A result peek at the as_of cannot resolve while the output frontier is held at the + // minimum: it stays pending rather than returning wrong (empty) rows. + { + let mut active = ActiveComputeState { + timely_worker: &mut *worker, + compute_state: &mut compute_state, + response_tx: &mut response_tx, + }; + active.handle_peek(make_count_peek(out_index_id, as_of)); + assert_eq!( + active + .compute_state + .pending_work + .values() + .map(Vec::len) + .sum::(), + 1, + "the result peek stays pending while the output frontier is held at the minimum" + ); + } + assert!( + rx.try_recv().is_err(), + "no result is produced while the placeholder input is unadopted" + ); + + // Tear down the built dataflow so the worker can shut down. Its import over the never + // adopted placeholder holds a frontier at the minimum forever, so without dropping it the + // dataflow never completes and `execute_directly` would wedge on teardown. + { + let mut active = ActiveComputeState { + timely_worker: &mut *worker, + compute_state: &mut compute_state, + response_tx: &mut response_tx, + }; + active.handle_allow_compaction(out_index_id, Antichain::new()); + } + for _ in 0..16 { + worker.step(); + } + }); +} diff --git a/src/compute/src/server.rs b/src/compute/src/server.rs index 3c96b9c0e8598..7cf6300de2316 100644 --- a/src/compute/src/server.rs +++ b/src/compute/src/server.rs @@ -295,6 +295,17 @@ impl ResponseSender { self.nonce = Some(nonce); } + /// Builds a `ResponseSender` with the nonce pre-initialized, for tests that drive an + /// `ActiveComputeState` outside the full `serve` protocol. + #[cfg(test)] + pub(crate) fn for_test(inner: mpsc::UnboundedSender<(ComputeResponse, Uuid)>) -> Self { + Self { + inner, + worker_id: 0, + nonce: Some(Uuid::nil()), + } + } + /// Send a compute response. pub fn send(&self, response: ComputeResponse) -> Result<(), SendError> { let nonce = self.nonce.expect("nonce must be initialized"); @@ -526,7 +537,28 @@ impl<'w> Worker<'w> { self.handle_pending_commands()?; + let role = self.role; + let worker_index = self.timely_worker.index(); if let Some(mut compute_state) = self.activate_compute() { + if role == ComputeRuntimeRole::Interactive { + // Give a turn to the shared-index peeks whose dependency was marked dirty by a + // publication or a seal since the last drain. An empty dirty set means the + // worker woke for a command or its maintenance tick, and no peek waiting on an + // event is touched, so this costs what changed rather than what is pending. + let dirty = compute_state + .compute_state + .sharing_registry + .take_dirty(worker_index); + if !dirty.is_empty() { + compute_state.resolve_dirty(dirty); + } + } + // The sweep is what serves a peek awaiting a turn and what retires one a driver + // has taken over. Neither of those waits on a dirty mark, on either runtime: a + // persist read and an offloaded walk both wake the worker through a channel of + // their own. A shared peek reaches the sweep only once it has passed the gates + // that admit the read, so a peek waiting on a publication or a seal costs it + // nothing. compute_state.process_peeks(); compute_state.process_subscribes(); compute_state.process_copy_tos(); @@ -556,6 +588,16 @@ impl<'w> Worker<'w> { Arc::clone(&self.peek_permits), self.storage_log_reader.take(), )); + + // The interactive runtime resolves deferred peeks on notification from the sharing + // registry. Register this worker's thread so a publication or seal in the registry can + // unpark it to re-examine pending work. Registered from the worker's own thread, which + // is what makes `current()` the right handle. Only the interactive runtime defers work + // this way; the maintenance runtime keeps its poll. + if self.role == ComputeRuntimeRole::Interactive { + self.sharing_registry + .register_waker(self.timely_worker.index(), std::thread::current()); + } } self.activate_compute().unwrap().handle_compute_command(cmd); } @@ -633,6 +675,12 @@ impl<'w> Worker<'w> { compute_state.command_history.discard_peeks(); compute_state.command_history.reduce(); + // NOTE: the standing holds in the sharing registry are deliberately NOT cleared here. + // One is per collection and carries no dataflow identity, so it cannot go stale across a + // reconnection, and it only ever rises. Clearing it would drop the hold back to the + // minimum time until the replayed compactions raised it again. See + // `doc/developer/design/20260720_two_runtime_compute/design.md`. + // At this point, we need to sort out which of the *certainly installed* dataflows are // suitable replacements for the requested dataflows. A dataflow is "certainly installed" // as of a frontier if its compaction allows it to go no further. We ignore peeks for this @@ -807,6 +855,19 @@ impl<'w> Worker<'w> { } } + // And the shared peeks waiting on a publication or a seal, which belong to the + // reconciled-away client connection just as much. They wait outside both queues above, + // so the sweep never sees them. + for peek in std::mem::take(&mut compute_state.pending_work) + .into_values() + .flatten() + { + // Log dropping the peek request, reusing `PendingPeek`'s log event. + if let Some(logger) = compute_state.compute_logger.as_mut() { + logger.log(&PendingPeek::Index(peek).as_log_event(false)); + } + } + for (&id, collection) in compute_state.collections.iter_mut() { // Adjust reported frontiers: // * For dataflows we continue to use, reset to ensure we report something not @@ -859,7 +920,14 @@ impl<'w> Worker<'w> { // these collections when it reconnects to a replica. // // TODO(database-issues#8152): Consider resolving this with controller-side reconciliation instead. - if let Some(config) = old_instance_config { + // + // The interactive runtime installs no logging dataflow and so hosts none of these + // traces (see `ActiveComputeState::initialize_logging`). It receives the same + // `CreateInstance` config as its peer, naming every logging index, and serves peeks on + // those ids from the sharing registry rather than from a local trace. There is nothing + // of its own to pad. + let hosts_logging = compute_state.role() != ComputeRuntimeRole::Interactive; + if let Some(config) = old_instance_config.filter(|_| hosts_logging) { for id in config.logging.index_logs.values() { let trace = compute_state .traces diff --git a/src/compute/src/sharing.rs b/src/compute/src/sharing.rs index c802cb3d0e7eb..ed6a4f484077f 100644 --- a/src/compute/src/sharing.rs +++ b/src/compute/src/sharing.rs @@ -19,10 +19,6 @@ //! is shared. Worker `i` publishes into slot `i`; a reader on worker `i` of another runtime looks up //! slot `i`, which is sound only because both sides shard keys by the same `key.hashed() % peers`. -// TODO(CPU-215): drop once `crate::render` and `crate::compute_state` call this registry. Only the -// registry's constructor is reachable yet, so the rest reads as dead. -#![allow(dead_code)] - use std::collections::{BTreeMap, BTreeSet}; use std::sync::{Arc, Mutex, MutexGuard}; use std::thread::Thread;