From 7b71d000a4ff097eb3acbacd4d6be91e0ef92329 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Fri, 21 Aug 2026 12:22:16 +0200 Subject: [PATCH 1/4] compute: serve fast-path peeks on the interactive runtime The interactive runtime holds no local traces, so an index peek there resolves against the sharing registry instead. `PeekScan` and the error walk beneath it become generic over the traces they read, and an interactive peek opens its scan over the registry's `SharedOksHandle` and `SharedErrsHandle`. Both flavours of index peek then spend one budget, report one set of metrics, and reach the peek stash through the one offload driver. A shared peek whose arrangement is not yet published, or whose upper has not sealed the peek timestamp, waits in `pending_work` keyed by a `WorkId` and indexed by the id it waits on. A publication or seal marks that id dirty and wakes the worker, which gives a turn to exactly the items indexed under the ids that changed, so wakeups scale with what changed rather than with total pending work. Past that gate a shared peek is an ordinary index peek: it queues for a turn like any other, and a walk that outruns the activation's fuel or fills a batch bound for the stash leaves for a driver. The sweep still runs on every step, because a persist read and an offloaded walk each wake the worker through a channel of their own rather than through the dirty set, but nothing waiting on a publication or a seal is ever swept. Interactive dataflows build immediately in command arrival order rather than deferring until their dependency is published. An import over an unadopted placeholder produces no data and holds its output frontier at the minimum until a publisher adopts the same slot, so late binding replaces the deferral. The interactive runtime reports only its transient collections' frontiers. It shares the identity of every non-transient collection with maintenance, which owns and reports the real frontiers, and the controller keeps one frontier stream per collection, so reporting the shared ones would race the owner and regress it. For the same reason its logging is forced off: it serves introspection peeks from maintenance's published copies, and its own empty copies would clobber them. The consequence is that nothing the interactive runtime does appears in introspection, tracked as CPU-222. Reconciliation drops `pending_work` and `dep_index`, whose peeks belong to the reconciled-away connection. The standing holds in the registry deliberately survive: one is per collection, carries no dataflow identity, and only rises, so clearing it would drop the arrangement's bound to the minimum until replayed compactions raised it again. Reachable only on a runtime holding the `Interactive` role, which requires the dyncfg that is still off everywhere. Tests are out of line in `compute_state/tests.rs`, per the convention in `src/compute/AGENTS.md`. Co-Authored-By: Claude Opus 5 (1M context) --- src/compute/src/compute_state.rs | 636 +++++++--- src/compute/src/compute_state/error_scan.rs | 49 +- .../src/compute_state/error_scan/tests.rs | 10 +- .../src/compute_state/index_peek_tests.rs | 61 +- src/compute/src/compute_state/index_traces.rs | 125 ++ src/compute/src/compute_state/peek_offload.rs | 41 +- .../src/compute_state/peek_offload/tests.rs | 7 +- src/compute/src/compute_state/peek_scan.rs | 71 +- .../src/compute_state/peek_scan/tests.rs | 9 +- .../src/compute_state/peek_sweep_tests.rs | 4 +- src/compute/src/compute_state/tests.rs | 1069 +++++++++++++++++ src/compute/src/server.rs | 61 + src/compute/src/sharing.rs | 4 - 13 files changed, 1909 insertions(+), 238 deletions(-) create mode 100644 src/compute/src/compute_state/index_traces.rs diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index 5422b6855c8af..f0e48c689ab7b 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_shared_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,47 @@ impl<'a> ActiveComputeState<'a> { } fn handle_allow_compaction(&mut self, id: GlobalId, frontier: Antichain) { + let worker_index = self.timely_worker.index(); + + let interactive = self.compute_state.role == ComputeRuntimeRole::Interactive; + + // The multiplexer broadcasts compaction for the collections its peer publishes, which are the + // ones this runtime may import, so on the interactive runtime a non-transient id is one of + // those. This runtime's own publications are its transient query outputs. + let peer_published = interactive && !id.is_transient(); + if peer_published { + // 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. + self.compute_state + .sharing_registry + .note_standing_hold(id, worker_index, &frontier); + } + + // Whether there is local work is a question about `collections`, NOT about the id: this + // runtime holds empty local copies of the peer's introspection indexes, whose ids are the + // peer's to publish, and the peer renders transient collections of its own (subscribes and + // copy-tos) that this runtime has never seen. Asking the id instead sends a broadcast frontier + // for one of those down the drop path, where `drop_collection` panics on a collection that was + // never installed here. + if interactive && !self.compute_state.collections.contains_key(&id) { + return; + } + + if peer_published { + if !frontier.is_empty() { + // Keeps this runtime's empty local copy of an introspection index in step. + self.compute_state + .traces + .allow_compaction(id, frontier.borrow()); + } + // Never `drop_collection` for one of those. It would also `sharing_registry.remove(&id)` + // and so unpublish an arrangement this runtime does not own. The empty copies live for + // the process lifetime, and the peer drops the real collection on its own stream. + return; + } + if frontier.is_empty() { // Indicates that we may drop `id`, as there are no more valid times to read. self.drop_collection(id); @@ -930,6 +1041,11 @@ impl<'a> ActiveComputeState<'a> { self.compute_state .traces .allow_compaction(id, frontier.borrow()); + // Forward the same frontier to the sharing registry so a cross-runtime publisher of this + // index follows the controller's logical compaction. A no-op unless `id` is published. + self.compute_state + .sharing_registry + .note_allow_compaction(id, worker_index, &frontier); } } @@ -937,9 +1053,18 @@ 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 = if self.compute_state.role == ComputeRuntimeRole::Interactive { + // The interactive runtime maintains no traces of its own. It reads the + // arrangements the maintenance runtime publishes into the sharing registry. + IndexTraces::Shared { + registry: self.compute_state.sharing_registry.clone(), + worker_index: self.timely_worker.index(), + } + } else { + // Acquire a copy of the trace suitable for fulfilling the peek. + IndexTraces::Local(self.compute_state.traces.get(id).unwrap().clone()) + }; + PendingPeek::index(peek, traces) } PeekTarget::Persist { metadata, .. } => { let metadata = metadata.clone(); @@ -965,6 +1090,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 +1115,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 +1170,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 +1214,17 @@ impl<'a> ActiveComputeState<'a> { panic!("dataflow server has already initialized logging"); } + let mut config = config; + // The interactive runtime maintains no introspection indexes of its own: it serves + // introspection peeks from the maintenance runtime's registry-published copies (see + // `logging::publish_logging_index`). Force logging off so its replay stays empty. The + // dataflows still install (empty, per database-issues#4545), so the logging indexes are + // still created and the sanity check below still holds, but they hold no data and, being + // non-maintenance, are never published. + if self.compute_state.role() == ComputeRuntimeRole::Interactive { + config.enable_logging = false; + } + let LoggingTraces { traces, dataflow_index, @@ -1109,6 +1291,14 @@ impl<'a> ActiveComputeState<'a> { pub fn report_frontiers(&mut self) { let mut responses = Vec::new(); + // The interactive runtime installs empty copies of the maintenance runtime's + // logging/introspection indexes (see `initialize_logging`) and shares every non-transient + // collection's identity with the maintenance runtime, which owns and reports the real + // frontiers. Reporting our empty copies' frontiers races the owner's report for the same + // collection id in the controller's single per-collection frontier stream, regressing it. + // Report only the wholly-transient query dataflows this runtime exclusively hosts. + let report_only_transient = self.compute_state.role() == ComputeRuntimeRole::Interactive; + // Maintain a single allocation for `new_frontier` to avoid allocating on every iteration. let mut new_frontier = Antichain::new(); @@ -1119,6 +1309,10 @@ impl<'a> ActiveComputeState<'a> { continue; } + if report_only_transient && !id.is_transient() { + continue; + } + let reported = collection.reported_frontiers(); // Collect the write frontier and check for progress. @@ -1272,6 +1466,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 +1479,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 +1503,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 +1822,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 +2085,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 +2125,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 +2162,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 +2303,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..0314fc9a4282b 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,1041 @@ 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| { + 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, "peek oks", || {}); + PublishArrangement::adopt(&errs, &slot.errs, "peek 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(); + }); + }); +} + +/// 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) = 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, "peek oks", || {}); + PublishArrangement::adopt(&errs, &slot.errs, "peek errs", || {}); + registry_in.notify(id, worker_index); + (oks_input, errs_input) + }); + + // 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 so the registry's +/// `Arc` keeps the snapshot readable after the inputs drop. +fn publish_index_current_worker( + worker: &mut TimelyWorker, + registry: &ArrangementSharingRegistry, + id: GlobalId, + rows: Vec<(Row, Row)>, +) { + let registry_in = registry.clone(); + let (mut oks_input, mut errs_input) = 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, "peek oks", || {}); + PublishArrangement::adopt(&errs, &slot.errs, "peek errs", || {}); + registry_in.notify(id, scope.index()); + (oks_input, errs_input) + }); + + 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, + // and the registry's `Arc` keeps the published chain alive. + drop(oks_input); + drop(errs_input); + for _ in 0..16 { + worker.step(); + } +} + +/// 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. + 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) = 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, "peek oks", || {}); + PublishArrangement::adopt(&errs, &slot.errs, "peek errs", || {}); + registry_in.notify(id, scope.index()); + (oks_input, errs_input) + }); + + 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..33e7586339725 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 diff --git a/src/compute/src/sharing.rs b/src/compute/src/sharing.rs index f156aeb6549f6..ed5335d9b1c36 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; From d2650a97e26a103375592516cead629e7006a9a0 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Mon, 7 Sep 2026 11:13:35 +0200 Subject: [PATCH 2/4] fixup: the interactive runtime installs no logging dataflow The interactive runtime installed empty copies of the logging indexes, with logging forced off, so that `initialize_logging`'s bookkeeping stayed uniform. Every command handler then had to tell those copies apart from the runtime's own collections: `handle_allow_compaction` carried a role check and a transience check to route broadcast compaction around them, and `report_frontiers` skipped non-transient ids so the copies' frontiers would not regress the maintenance runtime's reports. The copies served nothing. Peeks on the logging indexes read the maintenance runtime's publications from the registry, and only the maintenance runtime publishes them. With the interactive runtime installing no logging dataflow, "is this a collection this runtime hosts" is a question `collections` answers, and `handle_allow_compaction` reduces to: an unhosted id is the peer's, so its frontier is the standing hold. The role check survives only as a soft assertion that the maintenance runtime never sees compaction for an unknown collection, which used to panic in `drop_collection`. `report_frontiers` loses its filter. `handle_peek` selects the trace source by matching on the role. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01VDm7opomJLxbNUEP3r9BLk --- src/compute/src/compute_state.rs | 89 +++++++++++--------------------- 1 file changed, 30 insertions(+), 59 deletions(-) diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index f0e48c689ab7b..9d04326f89914 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -853,7 +853,7 @@ impl<'a> ActiveComputeState<'a> { // 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_shared_index`). A not-yet-published dependency + // 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()); @@ -995,42 +995,25 @@ impl<'a> ActiveComputeState<'a> { fn handle_allow_compaction(&mut self, id: GlobalId, frontier: Antichain) { let worker_index = self.timely_worker.index(); - let interactive = self.compute_state.role == ComputeRuntimeRole::Interactive; - - // The multiplexer broadcasts compaction for the collections its peer publishes, which are the - // ones this runtime may import, so on the interactive runtime a non-transient id is one of - // those. This runtime's own publications are its transient query outputs. - let peer_published = interactive && !id.is_transient(); - if peer_published { - // 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. + // 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); - } - - // Whether there is local work is a question about `collections`, NOT about the id: this - // runtime holds empty local copies of the peer's introspection indexes, whose ids are the - // peer's to publish, and the peer renders transient collections of its own (subscribes and - // copy-tos) that this runtime has never seen. Asking the id instead sends a broadcast frontier - // for one of those down the drop path, where `drop_collection` panics on a collection that was - // never installed here. - if interactive && !self.compute_state.collections.contains_key(&id) { - return; - } - - if peer_published { - if !frontier.is_empty() { - // Keeps this runtime's empty local copy of an introspection index in step. - self.compute_state - .traces - .allow_compaction(id, frontier.borrow()); - } - // Never `drop_collection` for one of those. It would also `sharing_registry.remove(&id)` - // and so unpublish an arrangement this runtime does not own. The empty copies live for - // the process lifetime, and the peer drops the real collection on its own stream. return; } @@ -1053,16 +1036,17 @@ impl<'a> ActiveComputeState<'a> { fn handle_peek(&mut self, peek: Peek) { let pending = match &peek.target { PeekTarget::Index { id } => { - let traces = if self.compute_state.role == ComputeRuntimeRole::Interactive { + 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. - IndexTraces::Shared { + ComputeRuntimeRole::Interactive => IndexTraces::Shared { registry: self.compute_state.sharing_registry.clone(), worker_index: self.timely_worker.index(), - } - } else { + }, // Acquire a copy of the trace suitable for fulfilling the peek. - IndexTraces::Local(self.compute_state.traces.get(id).unwrap().clone()) + ComputeRuntimeRole::Maintenance | ComputeRuntimeRole::Solo => { + IndexTraces::Local(self.compute_state.traces.get(id).unwrap().clone()) + } }; PendingPeek::index(peek, traces) } @@ -1214,15 +1198,14 @@ impl<'a> ActiveComputeState<'a> { panic!("dataflow server has already initialized logging"); } - let mut config = config; - // The interactive runtime maintains no introspection indexes of its own: it serves - // introspection peeks from the maintenance runtime's registry-published copies (see - // `logging::publish_logging_index`). Force logging off so its replay stays empty. The - // dataflows still install (empty, per database-issues#4545), so the logging indexes are - // still created and the sanity check below still holds, but they hold no data and, being - // non-maintenance, are never published. + // 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 { - config.enable_logging = false; + return; } let LoggingTraces { @@ -1291,14 +1274,6 @@ impl<'a> ActiveComputeState<'a> { pub fn report_frontiers(&mut self) { let mut responses = Vec::new(); - // The interactive runtime installs empty copies of the maintenance runtime's - // logging/introspection indexes (see `initialize_logging`) and shares every non-transient - // collection's identity with the maintenance runtime, which owns and reports the real - // frontiers. Reporting our empty copies' frontiers races the owner's report for the same - // collection id in the controller's single per-collection frontier stream, regressing it. - // Report only the wholly-transient query dataflows this runtime exclusively hosts. - let report_only_transient = self.compute_state.role() == ComputeRuntimeRole::Interactive; - // Maintain a single allocation for `new_frontier` to avoid allocating on every iteration. let mut new_frontier = Antichain::new(); @@ -1309,10 +1284,6 @@ impl<'a> ActiveComputeState<'a> { continue; } - if report_only_transient && !id.is_transient() { - continue; - } - let reported = collection.reported_frontiers(); // Collect the write frontier and check for progress. From ecd89fd6c78fba585049d310e70c8a85c5f532c2 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Mon, 7 Sep 2026 19:01:15 +0200 Subject: [PATCH 3/4] fixup: compaction reaches a published index through its trace alone `handle_allow_compaction` no longer notes the controller's frontier on the registry: the trace manager's compaction of the index is what the published `since` reports. Tests keep a trace agent alive for as long as they read, since the point closes with the trace. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01VDm7opomJLxbNUEP3r9BLk --- src/compute/src/compute_state.rs | 5 -- src/compute/src/compute_state/tests.rs | 93 ++++++++++++++++---------- 2 files changed, 58 insertions(+), 40 deletions(-) diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index 9d04326f89914..c1aeae24ef33f 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -1024,11 +1024,6 @@ impl<'a> ActiveComputeState<'a> { self.compute_state .traces .allow_compaction(id, frontier.borrow()); - // Forward the same frontier to the sharing registry so a cross-runtime publisher of this - // index follows the controller's logical compaction. A no-op unless `id` is published. - self.compute_state - .sharing_registry - .note_allow_compaction(id, worker_index, &frontier); } } diff --git a/src/compute/src/compute_state/tests.rs b/src/compute/src/compute_state/tests.rs index 0314fc9a4282b..3b64abaf74261 100644 --- a/src/compute/src/compute_state/tests.rs +++ b/src/compute/src/compute_state/tests.rs @@ -281,7 +281,10 @@ fn publish_kv_index_into( ) { let registry_in = registry.clone(); timely::execute_directly(move |worker| { - worker.dataflow::(|scope| { + // 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<_>, @@ -299,8 +302,8 @@ fn publish_kv_index_into( >("test errs"); let slot = registry_in.get_or_create(id, 0, 1); - PublishArrangement::adopt(&oks, &slot.oks, "peek oks", || {}); - PublishArrangement::adopt(&errs, &slot.errs, "peek errs", || {}); + PublishArrangement::adopt(&oks, &slot.oks, || {}); + PublishArrangement::adopt(&errs, &slot.errs, || {}); registry_in.notify(id, 0); for (k, v) in rows { @@ -310,7 +313,10 @@ fn publish_kv_index_into( 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); }); } @@ -483,28 +489,34 @@ fn interactive_shared_peek_defers_until_sealed() { let worker_index = worker.index(); let peers = worker.peers(); - let (mut oks_input, mut errs_input) = worker.dataflow::(move |scope| { - let (oks_input, oks_collection) = scope.new_collection::<(Row, Row), Diff>(); - let oks = oks_collection.mz_arrange::< + 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::< + 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, "peek oks", || {}); - PublishArrangement::adopt(&errs, &slot.errs, "peek errs", || {}); - registry_in.notify(id, worker_index); - (oks_input, errs_input) - }); + 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); @@ -608,16 +620,16 @@ fn interactive_compute_state( } /// 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 so the registry's -/// `Arc` keeps the snapshot readable after the inputs drop. +/// `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) = worker.dataflow::(move |scope| { + 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<_>, @@ -634,10 +646,14 @@ fn publish_index_current_worker( >("test errs"); let slot = registry_in.get_or_create(id, scope.index(), scope.peers()); - PublishArrangement::adopt(&oks, &slot.oks, "peek oks", || {}); - PublishArrangement::adopt(&errs, &slot.errs, "peek errs", || {}); + PublishArrangement::adopt(&oks, &slot.oks, || {}); + PublishArrangement::adopt(&errs, &slot.errs, || {}); registry_in.notify(id, scope.index()); - (oks_input, errs_input) + ( + oks_input, + errs_input, + (oks.trace.clone(), errs.trace.clone()), + ) }); for (k, v) in rows { @@ -650,13 +666,14 @@ fn publish_index_current_worker( for _ in 0..16 { worker.step(); } - // Drop the inputs and drain: the batch seals to the empty upper, readable at any finite ts, - // and the registry's `Arc` keeps the published chain alive. + // 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 @@ -715,7 +732,7 @@ fn interactive_peek_resolves_on_publication_not_on_bare_tick() { 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. - publish_index_current_worker(worker, ®istry, id, kv.clone()); + 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. @@ -782,28 +799,34 @@ fn interactive_peek_resolves_on_seal_via_note_frontier() { // Publish a row at time 0, sealing only to upper {1}. let registry_in = registry.clone(); - let (mut oks_input, mut errs_input) = worker.dataflow::(move |scope| { - let (oks_input, oks_collection) = scope.new_collection::<(Row, Row), Diff>(); - let oks = oks_collection.mz_arrange::< + 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::< + 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, "peek oks", || {}); - PublishArrangement::adopt(&errs, &slot.errs, "peek errs", || {}); - registry_in.notify(id, scope.index()); - (oks_input, errs_input) - }); + 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)); From 528e5bec15a10b9a07a764cf23b58710d48f4900 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Wed, 9 Sep 2026 14:59:47 +0200 Subject: [PATCH 4/4] fixup: the interactive runtime has no logging traces to pad Reconciliation pads every logging index's trace so the controller can read those collections from the minimum time after a reconnect. It takes the ids from the `CreateInstance` config, which the multiplexer delivers to both runtimes, and requires each to name a trace this runtime hosts. The interactive runtime installs no logging dataflow, so it hosts none of them and serves peeks on those ids from the sharing registry instead. The first reconciliation therefore panicked with "logging trace exists" on every interactive worker, taking the process with it. It needs a controller reconnect to reach, which is why the upgrade, restart and zippy suites saw it and the others did not. Skip the padding on the interactive runtime rather than tolerating a missing trace, because a logging trace absent on the maintenance runtime is still a bug and should stay loud. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_015tLhSbZdXrTSK2KwSocT59 --- src/compute/src/server.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/src/compute/src/server.rs b/src/compute/src/server.rs index 33e7586339725..7cf6300de2316 100644 --- a/src/compute/src/server.rs +++ b/src/compute/src/server.rs @@ -920,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