diff --git a/src/clusterd/src/lib.rs b/src/clusterd/src/lib.rs index 87e9d932c0b29..54caf42e3eca2 100644 --- a/src/clusterd/src/lib.rs +++ b/src/clusterd/src/lib.rs @@ -22,6 +22,7 @@ use mz_build_info::{BuildInfo, build_info}; use mz_cloud_resources::AwsExternalIdPrefix; use mz_cluster_client::client::TimelyConfig; use mz_compute::server::{ComputeInstanceContext, ComputeRuntimeRole}; +use mz_compute::sharing::ArrangementSharingRegistry; use mz_http_util::DynamicFilterTarget; use mz_orchestrator_tracing::{StaticTracingConfig, TracingCliArgs}; use mz_ore::cli::{self, CliConfig}; @@ -473,11 +474,16 @@ async fn run(args: Args) -> Result<(), anyhow::Error> { ); // Start compute server. + // + // The sharing registry is per process rather than per runtime: a reader on one runtime looks up + // the slot a publisher on another runtime filled, so both must hold the same registry. + let sharing_registry = ArrangementSharingRegistry::new(); let compute_client_builder = mz_compute::server::serve( compute_timely_config, ComputeRuntimeRole::Solo, &metrics_registry, persist_clients, + sharing_registry, txns_ctx, tracing_handle, ComputeInstanceContext { diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index 69da191e92f4f..78f220dda3277 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -75,7 +75,8 @@ use crate::logging::compute::{CollectionLogging, ComputeEvent, PeekEvent}; use crate::logging::initialize::LoggingTraces; use crate::metrics::{CollectionMetrics, WorkerMetrics}; use crate::render::{LinearJoinSpec, StartSignal}; -use crate::server::{ComputeInstanceContext, ResponseSender}; +use crate::server::{ComputeInstanceContext, ComputeRuntimeRole, ResponseSender}; +use crate::sharing::ArrangementSharingRegistry; mod peek_result_iterator; mod peek_stash; @@ -189,6 +190,11 @@ pub struct ComputeState { /// A process-global cache of (blob_uri, consensus_uri) -> PersistClient. /// This is intentionally shared between workers. pub persist_clients: Arc, + /// A per-process registry of published index arrangements. + /// + /// Intentionally shared between all workers of the process, each of which publishes into its own + /// worker-ordinal slot. `Clone` shares the same underlying map. + pub sharing_registry: ArrangementSharingRegistry, /// Context necessary for rendering txn-wal operators. pub txns_ctx: TxnsContext, /// History of commands received by this workers and all its peers. @@ -247,12 +253,20 @@ pub struct ComputeState { /// The storage worker forwards its introspection logs to the compute worker. pub storage_log_reader: Option, + + /// Which of the process's compute runtimes this state belongs to. + /// + /// Only the maintenance runtime runs the non-idempotent process-global initializers. The + /// interactive runtime shares the same process and inherits those globals. + role: ComputeRuntimeRole, } impl ComputeState { /// Construct a new `ComputeState`. pub fn new( + role: ComputeRuntimeRole, persist_clients: Arc, + sharing_registry: ArrangementSharingRegistry, txns_ctx: TxnsContext, metrics: WorkerMetrics, tracing_handle: Arc, @@ -273,6 +287,7 @@ impl ComputeState { peek_stash_persist_location: None, compute_logger: None, persist_clients, + sharing_registry, txns_ctx, command_history, max_result_size: u64::MAX, @@ -288,9 +303,15 @@ impl ComputeState { init_system_time: mz_ore::now::SYSTEM_TIME(), replica_expiration: Antichain::default(), storage_log_reader, + role, } } + /// Which of the process's compute runtimes this state serves. + pub(crate) fn role(&self) -> ComputeRuntimeRole { + self.role + } + /// Return a mutable reference to the identified collection. /// /// Panics if the collection doesn't exist. @@ -955,6 +976,8 @@ impl<'a> ActiveComputeState<'a> { Rc::clone(&self.compute_state.worker_config), self.compute_state.workers_per_process, storage_log_reader, + self.compute_state.role(), + self.compute_state.sharing_registry.clone(), ); let dataflow_index = Rc::new(dataflow_index); diff --git a/src/compute/src/logging/initialize.rs b/src/compute/src/logging/initialize.rs index 245cef517d309..73a57062d7e3d 100644 --- a/src/compute/src/logging/initialize.rs +++ b/src/compute/src/logging/initialize.rs @@ -16,7 +16,7 @@ use differential_dataflow::logging::{DifferentialEvent, DifferentialEventBuilder use mz_compute_client::logging::{LogVariant, LoggingConfig}; use mz_dyncfg::ConfigSet; use mz_ore::metrics::MetricsRegistry; -use mz_repr::{Diff, Timestamp}; +use mz_repr::{Diff, GlobalId, Timestamp}; use mz_storage_operators::persist_source::Subtime; use mz_timely_util::columnar::Column; use mz_timely_util::columnar::builder::ColumnBuilder; @@ -35,7 +35,10 @@ use crate::extensions::arrange::{KeyCollection, MzArrange}; use crate::logging::compute::{ComputeEvent, ComputeEventBuilder}; use crate::logging::{BatchLogger, EventQueue, SharedLoggingState}; use crate::render::errors::DataflowErrorSer; -use crate::typedefs::{ErrBatcher, ErrBuilder}; +use crate::server::ComputeRuntimeRole; +use crate::shared_trace::PublishArrangement; +use crate::sharing::ArrangementSharingRegistry; +use crate::typedefs::{ErrAgent, ErrBatcher, ErrBuilder, RowRowAgent}; /// Initialize logging dataflows. /// @@ -48,6 +51,8 @@ pub fn initialize( worker_config: Rc, workers_per_process: usize, storage_log_reader: Option, + role: ComputeRuntimeRole, + sharing_registry: ArrangementSharingRegistry, ) -> LoggingTraces { let interval_ms = std::cmp::max(1, config.interval.as_millis()); @@ -74,6 +79,8 @@ pub fn initialize( worker_config, workers_per_process, storage_log_reader, + role, + sharing_registry, }; // Depending on whether we should log the creation of the logging dataflows, we register the @@ -114,6 +121,11 @@ struct LoggingContext<'a> { workers_per_process: usize, /// Optional reader for storage timely logging events. storage_log_reader: Option, + /// This runtime's role. Only `Maintenance` publishes its logging indexes into the sharing + /// registry. + role: ComputeRuntimeRole, + /// The per-process registry maintenance publishes its logging indexes into. + sharing_registry: ArrangementSharingRegistry, } pub(crate) struct LoggingTraces { @@ -206,6 +218,20 @@ impl LoggingContext<'_> { let traces = collections .into_iter() .map(|(log, collection)| { + // Publish maintenance's logging index into the sharing registry so the + // interactive runtime serves introspection peeks from it. Gated on the + // Maintenance role inside the helper, so this is a no-op (adds no operators) on + // Interactive and Solo. + if let Some(&id) = self.config.index_logs.get(&log) { + publish_logging_index( + self.role, + &self.sharing_registry, + &scope, + id, + &collection.trace, + &errs, + ); + } let bundle = TraceBundle::new(collection.trace, errs.clone()) .with_drop(collection.token); (log, bundle) @@ -353,3 +379,135 @@ impl ExtractTimestamp for (Timestamp, Subtime) { self.0 } } + +/// Publishes a maintenance logging index's `oks`/`errs` arrangements into the sharing registry so +/// the interactive runtime serves introspection peeks from them. +/// +/// Gated strictly on the `Maintenance` role. Interactive must not publish: it reads maintenance's +/// slot, and its own (empty) copy would clobber it. Solo has no registry peer. The gate is +/// deliberately stricter than `ComputeRuntimeRole::publishes`, which also admits Interactive. +/// +/// The arrangements are re-imported from their trace handles into `scope`. The original arrange +/// streams are consumed inside the per-log construction regions, so only the trace handles survive +/// here, and `Arranged::publish` needs a live arrangement stream on this scope to attach its +/// publisher operator. +fn publish_logging_index( + role: ComputeRuntimeRole, + registry: &ArrangementSharingRegistry, + scope: &timely::dataflow::Scope<'_, Timestamp>, + id: GlobalId, + oks_trace: &RowRowAgent, + errs_trace: &ErrAgent, +) { + if role != ComputeRuntimeRole::Maintenance { + return; + } + + // Re-import the trace handles to obtain live arrangement streams `publish` can attach a + // publisher operator to. The publisher refreshes its published chain from the trace, the + // authoritative source, so the re-import replay only drives the publisher's wakeups. + let oks = oks_trace + .clone() + .import_named(scope.clone(), &format!("PublishLog({id})")); + let errs = errs_trace + .clone() + .import_named(scope.clone(), &format!("PublishLogErr({id})")); + + // Adopt the registry's placeholder for `id` rather than publishing fresh and inserting: whichever + // side, this maintenance publish or an interactive import ahead of it, touches `id` first creates + // the slot, so filling it in place cannot overwrite a placeholder a reader has already imported. + // + // Both halves signal on seal. An introspection read whose result is an error (a division-by-zero + // surfacing in `mz_compute_error_counts_raw_unified`) carries its data on the errs stream, so an + // oks-only signal would leave it stuck. + let worker_index = scope.index(); + let slot = registry.get_or_create_placeholder(id, worker_index, scope.peers()); + let oks_registry = registry.clone(); + PublishArrangement::adopt(&oks, &slot.oks, move || { + oks_registry.note_frontier(id, worker_index) + }); + let errs_registry = registry.clone(); + PublishArrangement::adopt(&errs, &slot.errs, move || { + errs_registry.note_frontier(id, worker_index) + }); + // `get_or_create_placeholder` does not notify on create. + registry.notify(id, worker_index); +} + +#[cfg(test)] +mod tests { + use differential_dataflow::input::Input; + use mz_repr::{Diff, GlobalId, Row, Timestamp}; + use mz_row_spine::{RowRowBatcher, RowRowBuilder}; + use mz_timely_util::columnation::ColumnationChunker; + + use crate::extensions::arrange::{KeyCollection, MzArrange}; + use crate::render::errors::DataflowErrorSer; + use crate::server::ComputeRuntimeRole; + use crate::sharing::ArrangementSharingRegistry; + use crate::typedefs::{ErrBatcher, ErrBuilder, ErrSpine, RowRowSpine}; + + use super::publish_logging_index; + + /// A logging/introspection index is a `RowRow` `oks` arrangement plus an (empty) `errs` + /// arrangement, published into the sharing registry only by the maintenance runtime. Interactive + /// and Solo must not publish: interactive reads maintenance's slot rather than clobbering it with + /// its own empty copy, and Solo has no registry peer. + /// + /// Builds real `RowRow`/`Err` arrangements (the exact types the logging path produces) and drives + /// [`publish_logging_index`] for each role, asserting only maintenance ends up published. + #[mz_ore::test] + fn maintenance_publishes_logging_index_others_do_not() { + for (role, expect_published) in [ + (ComputeRuntimeRole::Maintenance, true), + (ComputeRuntimeRole::Interactive, false), + (ComputeRuntimeRole::Solo, false), + ] { + let id = GlobalId::System(1); + let registry = ArrangementSharingRegistry::new(); + 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 log oks"); + + let (mut errs_input, errs_collection) = + scope.new_collection::(); + let errs = KeyCollection::from(errs_collection).mz_arrange::< + ColumnationChunker<_>, + ErrBatcher<_, _>, + ErrBuilder<_, _>, + ErrSpine<_, _>, + >("test log errs"); + + publish_logging_index( + role, + ®istry_in, + &scope.clone(), + id, + &oks.trace, + &errs.trace, + ); + + oks_input.advance_to(Timestamp::from(1_u64)); + oks_input.flush(); + errs_input.advance_to(Timestamp::from(1_u64)); + errs_input.flush(); + }); + }); + + assert_eq!( + registry.handles(&id, 0).is_some(), + expect_published, + "role {role:?} publication mismatch" + ); + } + } +} diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs index c8607c22246cc..4b88e27201a14 100644 --- a/src/compute/src/render.rs +++ b/src/compute/src/render.rs @@ -168,6 +168,7 @@ use crate::logging::compute::{ use crate::render::columnar::CollectionEdge; use crate::render::context::{ArrangementFlavor, Context}; use crate::render::errors::DataflowErrorSer; +use crate::shared_trace::PublishArrangement; use crate::typedefs::{ErrBatcher, ErrBuilder, ErrSpine, KeyBatcher, MzTimestamp}; use mz_row_spine::{DatumSeq, RowRowBatcher, RowRowBuilder}; @@ -743,6 +744,37 @@ impl<'g> Context<'g, mz_repr::Timestamp> { errs.stream = errs.stream.log_dataflow_errors(logger, idx_id); } + // Publish into the per-process sharing registry when this role publishes (see + // `ComputeRuntimeRole::publishes`). Borrows the arrangements, so it must precede + // moving their traces into the `TraceBundle` below. + if compute_state.role().publishes() { + // Adopt the registry's placeholder for `idx_id` rather than publishing fresh and + // inserting: whichever side touches `idx_id` first creates the slot, so filling + // it in place cannot overwrite a placeholder a reader has already imported. + // + // Both halves signal on seal. A read whose result is an error (a runtime + // division-by-zero, or a `WITH MUTUALLY RECURSIVE ... ERROR AT RECURSION LIMIT` + // that trips its limit) carries its data on the errs stream, whose frontier is + // held back until the error is emitted, so an oks-only signal leaves it stuck, + // and vice versa for a normal result. + let worker_index = self.scope.index(); + let slot = compute_state.sharing_registry.get_or_create_placeholder( + idx_id, + worker_index, + self.scope.peers(), + ); + let oks_registry = compute_state.sharing_registry.clone(); + PublishArrangement::adopt(&oks, &slot.oks, move || { + oks_registry.note_frontier(idx_id, worker_index) + }); + let errs_registry = compute_state.sharing_registry.clone(); + PublishArrangement::adopt(&errs, &slot.errs, move || { + errs_registry.note_frontier(idx_id, worker_index) + }); + // `get_or_create_placeholder` does not notify on create. + compute_state.sharing_registry.notify(idx_id, worker_index); + } + compute_state.traces.set( idx_id, TraceBundle::new(oks.trace, errs.trace).with_drop(needed_tokens), @@ -753,6 +785,20 @@ impl<'g> Context<'g, mz_repr::Timestamp> { // just create another handle to that arrangement. let trace = compute_state.traces.get(&gid).unwrap().clone(); compute_state.traces.set(idx_id, trace); + + // Mirror the trace aliasing in the sharing registry: re-register the arrangement + // already published under `gid` on this worker under `idx_id` as well. This arm + // builds no streams of its own, so it installs no seal signal of its own. + // `reexport` records `idx_id` as an alias of `gid` so `gid`'s publisher wakes reads + // waiting on `idx_id`'s seal. Without that, such a read hangs. + if compute_state.role().publishes() { + compute_state.sharing_registry.reexport( + &gid, + idx_id, + self.scope.index(), + self.scope.peers(), + ); + } } None => { println!("collection available: {:?}", bundle.collection.is_none()); @@ -845,6 +891,29 @@ where errs.stream = errs.stream.log_dataflow_errors(logger, idx_id); } + // Publish into the per-process sharing registry when this role publishes, as in the + // unbucketed export path above, which carries the reasoning. The arrangements were + // re-arranged onto `outer` (the worker scope carrying `mz_repr::Timestamp`), so the + // worker ordinal comes from there. + if compute_state.role().publishes() { + let worker_index = outer.index(); + let slot = compute_state.sharing_registry.get_or_create_placeholder( + idx_id, + worker_index, + outer.peers(), + ); + let oks_registry = compute_state.sharing_registry.clone(); + PublishArrangement::adopt(&oks, &slot.oks, move || { + oks_registry.note_frontier(idx_id, worker_index) + }); + let errs_registry = compute_state.sharing_registry.clone(); + PublishArrangement::adopt(&errs, &slot.errs, move || { + errs_registry.note_frontier(idx_id, worker_index) + }); + // `get_or_create_placeholder` does not notify on create. + compute_state.sharing_registry.notify(idx_id, worker_index); + } + compute_state.traces.set( idx_id, TraceBundle::new(oks.trace, errs.trace).with_drop(needed_tokens), @@ -855,6 +924,17 @@ where // just create another handle to that arrangement. let trace = compute_state.traces.get(&gid).unwrap().clone(); compute_state.traces.set(idx_id, trace); + + // Mirror the trace aliasing in the sharing registry: re-register the arrangement + // already published under `gid` on this worker under `idx_id` as well. + if compute_state.role().publishes() { + compute_state.sharing_registry.reexport( + &gid, + idx_id, + outer.index(), + outer.peers(), + ); + } } None => { println!("collection available: {:?}", bundle.collection.is_none()); diff --git a/src/compute/src/server.rs b/src/compute/src/server.rs index 3fdbd78e0060c..a8edb0c786094 100644 --- a/src/compute/src/server.rs +++ b/src/compute/src/server.rs @@ -43,6 +43,7 @@ use uuid::Uuid; use crate::command_channel; use crate::compute_state::{ActiveComputeState, ComputeState, ReportedFrontier}; use crate::metrics::{ComputeMetrics, WorkerMetrics}; +use crate::sharing::ArrangementSharingRegistry; /// Caller-provided configuration for compute. #[derive(Clone, Debug)] @@ -77,16 +78,6 @@ pub enum ComputeRuntimeRole { Maintenance, /// The interactive runtime of a two-runtime process. Shares the process globals owned by /// maintenance and serves reads. - /// - /// Test-only until the interactive runtime exists to construct it. It is present because the - /// `role` label's entire purpose is that two named roles register into one process registry - /// without colliding, and nothing else can express that: `Solo` registers the same metric names - /// with no `role` label, so prometheus rejects it alongside a named role for differing label - /// dimensions rather than treating it as a second series. Verifying non-collision therefore - /// needs a second *named* role. - /// - /// TODO: drop the `cfg` when the interactive runtime lands and constructs this. - #[cfg(test)] Interactive, } @@ -99,7 +90,6 @@ impl ComputeRuntimeRole { match self { ComputeRuntimeRole::Solo => None, ComputeRuntimeRole::Maintenance => Some("maintenance"), - #[cfg(test)] ComputeRuntimeRole::Interactive => Some("interactive"), } } @@ -109,15 +99,25 @@ impl ComputeRuntimeRole { /// `Solo` and `Maintenance` run them. An interactive runtime shares the same process and /// inherits the globals maintenance installs, so re-running them would either double-apply a /// non-idempotent effect or race maintenance. - /// - /// NOTE: every role a release build can construct owns the globals, so this is constantly true - /// outside tests. The distinction becomes load-bearing when the interactive runtime lands. pub fn owns_process_globals(self) -> bool { matches!( self, ComputeRuntimeRole::Solo | ComputeRuntimeRole::Maintenance ) } + + /// Whether this role publishes its rendered indexes into the sharing registry. + /// + /// `Maintenance` publishes its maintained indexes, which its interactive peer reads exclusively + /// from the registry. `Interactive` publishes its transient query outputs, which the result + /// peeks over them likewise read from the registry and rely on for seal notifications. `Solo` + /// has no registry peer, so it does not publish. + pub fn publishes(self) -> bool { + matches!( + self, + ComputeRuntimeRole::Maintenance | ComputeRuntimeRole::Interactive + ) + } } /// Type alias for the storage timely log reader. @@ -127,8 +127,12 @@ pub(crate) type StorageTimelyLogReader = /// Configures the server with compute-specific metrics. #[derive(Clone)] struct Config { + /// Which of the process's compute runtimes this is. + pub role: ComputeRuntimeRole, /// `persist` client cache. pub persist_clients: Arc, + /// A per-process registry of published index arrangements, shared across all workers. + pub sharing_registry: ArrangementSharingRegistry, /// Context necessary for rendering txn-wal operators. pub txns_ctx: TxnsContext, /// A process-global handle to tracing configuration. @@ -151,6 +155,7 @@ pub async fn serve( role: ComputeRuntimeRole, metrics_registry: &MetricsRegistry, persist_clients: Arc, + sharing_registry: ArrangementSharingRegistry, txns_ctx: TxnsContext, tracing_handle: Arc, context: ComputeInstanceContext, @@ -173,7 +178,9 @@ pub async fn serve( mz_timely_util::pool_config::metrics::register(metrics_registry); let config = Config { + role, persist_clients, + sharing_registry, txns_ctx, tracing_handle, metrics: ComputeMetrics::register_with(metrics_registry, role), @@ -293,6 +300,8 @@ impl ResponseSender { /// Much of this state can be viewed as local variables for the worker thread, /// holding state that persists across function calls. struct Worker<'w> { + /// Which of the process's compute runtimes this worker belongs to. + role: ComputeRuntimeRole, /// The underlying Timely worker. timely_worker: &'w mut TimelyWorker, /// The channel over which commands are received. @@ -305,6 +314,8 @@ struct Worker<'w> { /// A process-global cache of (blob_uri, consensus_uri) -> PersistClient. /// This is intentionally shared between workers persist_clients: Arc, + /// A per-process registry of published index arrangements, shared across all workers. + sharing_registry: ArrangementSharingRegistry, /// Context necessary for rendering txn-wal operators. txns_ctx: TxnsContext, /// A process-global handle to tracing configuration. @@ -355,12 +366,14 @@ impl ClusterSpec for Config { spawn_channel_adapter(client_rx, cmd_tx, resp_rx, worker_id); Worker { + role: self.role, timely_worker, command_rx: CommandReceiver::new(cmd_rx, worker_id), response_tx: ResponseSender::new(resp_tx, worker_id), metrics, context: self.context.clone(), persist_clients: Arc::clone(&self.persist_clients), + sharing_registry: self.sharing_registry.clone(), txns_ctx: self.txns_ctx.clone(), compute_state: None, tracing_handle: Arc::clone(&self.tracing_handle), @@ -497,7 +510,9 @@ impl<'w> Worker<'w> { fn handle_command(&mut self, cmd: ComputeCommand) { if matches!(&cmd, ComputeCommand::CreateInstance(_)) { self.compute_state = Some(ComputeState::new( + self.role, Arc::clone(&self.persist_clients), + self.sharing_registry.clone(), self.txns_ctx.clone(), self.metrics.clone(), Arc::clone(&self.tracing_handle),