Struct Instance

pub(super) struct Instance<T: ComputeControllerTimestamp> {

    build_info: &'static BuildInfo,
    storage_collections: Arc<dyn StorageCollections<Timestamp = T> + Send + Sync>,
    initialized: bool,
    read_only: bool,
    workload_class: Option<String>,
    replicas: BTreeMap<ReplicaId, ReplicaState<T>>,
    collections: BTreeMap<GlobalId, CollectionState<T>>,
    log_sources: BTreeMap<LogVariant, GlobalId>,
    peeks: BTreeMap<Uuid, PendingPeek<T>>,
    subscribes: BTreeMap<GlobalId, ActiveSubscribe<T>>,
    copy_tos: BTreeSet<GlobalId>,
    history: ComputeCommandHistory<DeleteOnDropGauge<AtomicU64, Vec<String>>, T>,
    command_rx: UnboundedReceiver<Box<dyn FnOnce(&mut Instance<T>) + Send>>,
    response_tx: UnboundedSender<ComputeControllerResponse<T>>,
    introspection_tx: UnboundedSender<(IntrospectionType, Vec<(Row, Diff)>)>,
    metrics: InstanceMetrics,
    dyncfg: Arc<ConfigSet>,
    peek_stash_persist_location: PersistLocation,
    now: NowFn,
    wallclock_lag: WallclockLagFn<T>,
    wallclock_lag_last_recorded: DateTime<Utc>,
    read_hold_tx: ChangeTx<T>,
    replica_tx: InstrumentedUnboundedSender<(ReplicaId, u64, ComputeResponse<T>), DeleteOnDropCounter<AtomicU64, Vec<String>>>,
    replica_rx: InstrumentedUnboundedReceiver<(ReplicaId, u64, ComputeResponse<T>), DeleteOnDropCounter<AtomicU64, Vec<String>>>,
}

The state we keep for a compute instance.

Fields

build_info: &'static BuildInfo

Build info for spawning replicas

storage_collections: Arc<dyn StorageCollections<Timestamp = T> + Send + Sync>

A handle providing access to storage collections.

initialized: bool

Whether instance initialization has been completed.

read_only: bool

Whether or not this instance is in read-only mode.

When in read-only mode, neither the controller nor the instances controlled by it are allowed to affect changes to external systems (largely persist).

workload_class: Option<String>

The workload class of this instance.

This is currently only used to annotate metrics.

replicas: BTreeMap<ReplicaId, ReplicaState<T>>

The replicas of this compute instance.

collections: BTreeMap<GlobalId, CollectionState<T>>

Currently installed compute collections.

New entries are added for all collections exported from dataflows created through Instance::create_dataflow.

Entries are removed by Instance::cleanup_collections. See that method’s documentation about the conditions for removing collection state.

log_sources: BTreeMap<LogVariant, GlobalId>

IDs of log sources maintained by this compute instance.

peeks: BTreeMap<Uuid, PendingPeek<T>>

Currently outstanding peeks.

New entries are added for all peeks initiated through Instance::peek.

The entry for a peek is only removed once all replicas have responded to the peek. This is currently required to ensure all replicas have stopped reading from the peeked collection’s inputs before we allow them to compact. database-issues#4822 tracks changing this so we only have to wait for the first peek response.

subscribes: BTreeMap<GlobalId, ActiveSubscribe<T>>

Currently in-progress subscribes.

New entries are added for all subscribes exported from dataflows created through Instance::create_dataflow.

The entry for a subscribe is removed once at least one replica has reported the subscribe to have advanced to the empty frontier or to have been dropped, implying that no further updates will be emitted for this subscribe.

Note that subscribes are tracked both in collections and subscribes. collections keeps track of the subscribe’s upper and since frontiers and ensures appropriate read holds on the subscribe’s input. subscribes is only used to track which updates have been emitted, to decide if new ones should be emitted or suppressed.

copy_tos: BTreeSet<GlobalId>

Tracks all in-progress COPY TOs.

New entries are added for all s3 oneshot sinks (corresponding to a COPY TO) exported from dataflows created through Instance::create_dataflow.

The entry for a copy to is removed once at least one replica has finished or the exporting collection is dropped.

history: ComputeCommandHistory<DeleteOnDropGauge<AtomicU64, Vec<String>>, T>

The command history, used when introducing new replicas or restarting existing replicas.

command_rx: UnboundedReceiver<Box<dyn FnOnce(&mut Instance<T>) + Send>>

Receiver for commands to be executed.

response_tx: UnboundedSender<ComputeControllerResponse<T>>

Sender for responses to be delivered.

introspection_tx: UnboundedSender<(IntrospectionType, Vec<(Row, Diff)>)>

Sender for introspection updates to be recorded.

metrics: InstanceMetrics

The registry the controller uses to report metrics.

dyncfg: Arc<ConfigSet>

Dynamic system configuration.

peek_stash_persist_location: PersistLocation

The persist location where we can stash large peek results.

now: NowFn

A function that produces the current wallclock time.

wallclock_lag: WallclockLagFn<T>

A function that computes the lag between the given time and wallclock time.

wallclock_lag_last_recorded: DateTime<Utc>

The last time wallclock lag introspection was recorded.

read_hold_tx: ChangeTx<T>

Sender for updates to collection read holds.

Copies of this sender are given to ReadHolds that are created in CollectionState::new.

replica_tx: InstrumentedUnboundedSender<(ReplicaId, u64, ComputeResponse<T>), DeleteOnDropCounter<AtomicU64, Vec<String>>>

A sender for responses from replicas.

replica_rx: InstrumentedUnboundedReceiver<(ReplicaId, u64, ComputeResponse<T>), DeleteOnDropCounter<AtomicU64, Vec<String>>>

A receiver for responses from replicas.

Implementations

impl<T: ComputeControllerTimestamp> Instance<T>

fn collection( &self, id: GlobalId, ) -> Result<&CollectionState<T>, CollectionMissing>

Acquire a handle to the collection state associated with id.

fn collection_mut( &mut self, id: GlobalId, ) -> Result<&mut CollectionState<T>, CollectionMissing>

Acquire a mutable handle to the collection state associated with id.

fn expect_collection(&self, id: GlobalId) -> &CollectionState<T>

Acquire a handle to the collection state associated with id.

Panics

Panics if the identified collection does not exist.

fn expect_collection_mut(&mut self, id: GlobalId) -> &mut CollectionState<T>

Acquire a mutable handle to the collection state associated with id.

Panics

Panics if the identified collection does not exist.

fn collections_iter( &self, ) -> impl Iterator<Item = (GlobalId, &CollectionState<T>)>

fn add_collection( &mut self, id: GlobalId, as_of: Antichain<T>, shared: SharedCollectionState<T>, storage_dependencies: BTreeMap<GlobalId, ReadHold<T>>, compute_dependencies: BTreeMap<GlobalId, ReadHold<T>>, replica_input_read_holds: Vec<ReadHold<T>>, write_only: bool, storage_sink: bool, initial_as_of: Option<Antichain<T>>, refresh_schedule: Option<RefreshSchedule>, )

Add a collection to the instance state.

Panics

Panics if a collection with the same ID exists already.

fn remove_collection(&mut self, id: GlobalId)

fn add_replica_state( &mut self, id: ReplicaId, client: ReplicaClient<T>, config: ReplicaConfig, epoch: u64, )

fn deliver_response(&self, response: ComputeControllerResponse<T>)

Enqueue the given response for delivery to the controller clients.

fn deliver_introspection_updates( &self, type_: IntrospectionType, updates: Vec<(Row, Diff)>, )

Enqueue the given introspection updates for recording.

fn replica_exists(&self, id: ReplicaId) -> bool

Returns whether the identified replica exists.

fn peeks_targeting( &self, replica_id: ReplicaId, ) -> impl Iterator<Item = (Uuid, &PendingPeek<T>)>

Return the IDs of pending peeks targeting the specified replica.

fn subscribes_targeting( &self, replica_id: ReplicaId, ) -> impl Iterator<Item = GlobalId> + '_

Return the IDs of in-progress subscribes targeting the specified replica.

fn update_frontier_introspection(&mut self)

Update introspection with the current collection frontiers.

We could also do this directly in response to frontier changes, but doing it periodically lets us avoid emitting some introspection updates that can be consolidated (e.g. a write frontier updated immediately followed by a read frontier update).

This method is invoked by ComputeController::maintain, which we expect to be called once per second during normal operation.

fn refresh_state_metrics(&self)

Refresh the controller state metrics for this instance.

We could also do state metric updates directly in response to state changes, but that would mean littering the code with metric update calls. Encapsulating state metric maintenance in a single method is less noisy.

This method is invoked by ComputeController::maintain, which we expect to be called once per second during normal operation.

fn refresh_wallclock_lag(&mut self)

Refresh the wallclock lag introspection and metrics with the current lag values.

This method produces wallclock lag metrics of two different shapes:

• Histories: For each replica and each collection, we measure the lag of the write frontier behind the wallclock time every second. Every minute we emit the maximum lag observed over the last minute, together with the current time.
• Histograms: For each collection, we measure the lag of the write frontier behind wallclock time every second. Every minute we emit all lags observed over the last minute, together with the current histogram period.

Histories are emitted to both Mz introspection and Prometheus, histograms only to introspection. We treat lags of unreadable collections (i.e. collections that contain no readable times) as undefined and set them to NULL in introspection and u64::MAX in Prometheus.

This method is invoked by ComputeController::maintain, which we expect to be called once per second during normal operation.

fn maybe_record_wallclock_lag(&mut self)

Produce new wallclock lag introspection updates, provided enough time has passed since the last recording. We emit new introspection updates if the system time has passed into a new multiple of the recording interval (typically 1 minute) since the last refresh. The storage controller uses the same approach, ensuring that both controllers commit their lags at roughly the same time, avoiding confusion caused by inconsistencies.

fn report_dependency_updates(&self, id: GlobalId, diff: Diff)

Report updates (inserts or retractions) to the identified collection’s dependencies.

Panics

Panics if the identified collection does not exist.

pub fn collection_hydrated( &self, collection_id: GlobalId, ) -> Result<bool, CollectionLookupError>

Returns true if the given collection is hydrated on at least one replica.

This also returns true in case this cluster does not have any replicas.

pub fn collections_hydrated_on_replicas( &self, target_replica_ids: Option<Vec<ReplicaId>>, exclude_collections: &BTreeSet<GlobalId>, ) -> Result<bool, HydrationCheckBadTarget>

Returns true if each non-transient, non-excluded collection is hydrated on at least one replica.

This also returns true in case this cluster does not have any replicas.

pub fn collections_hydrated( &self, exclude_collections: &BTreeSet<GlobalId>, ) -> bool

Returns true if all non-transient, non-excluded collections are hydrated on at least one replica.

This also returns true in case this cluster does not have any replicas.

fn cleanup_collections(&mut self)

Clean up collection state that is not needed anymore.

Three conditions need to be true before we can remove state for a collection:

1. A client must have explicitly dropped the collection. If that is not the case, clients can still reasonably assume that the controller knows about the collection and can answer queries about it.
2. There must be no outstanding read capabilities on the collection. As long as someone still holds read capabilities on a collection, we need to keep it around to be able to properly handle downgrading of said capabilities.
3. All replica frontiers for the collection must have advanced to the empty frontier. Advancement to the empty frontiers signals that replicas are done computing the collection and that they won’t send more ComputeResponses for it. As long as we might receive responses for a collection we want to keep it around to be able to validate and handle these responses.

pub fn dump(&self) -> Result<Value, Error>

Returns the state of the Instance formatted as JSON.

The returned value is not guaranteed to be stable and may change at any point in time.

impl<T> Instance<T>

where T: ComputeControllerTimestamp,

fn new( build_info: &'static BuildInfo, storage: Arc<dyn StorageCollections<Timestamp = T> + Send + Sync>, peek_stash_persist_location: PersistLocation, arranged_logs: Vec<(LogVariant, GlobalId, SharedCollectionState<T>)>, metrics: InstanceMetrics, now: NowFn, wallclock_lag: WallclockLagFn<T>, dyncfg: Arc<ConfigSet>, command_rx: UnboundedReceiver<Box<dyn FnOnce(&mut Instance<T>) + Send>>, response_tx: UnboundedSender<ComputeControllerResponse<T>>, read_hold_tx: ChangeTx<T>, introspection_tx: UnboundedSender<(IntrospectionType, Vec<(Row, Diff)>)>, ) -> Self

async fn run(self)

pub fn update_configuration(&mut self, config_params: ComputeParameters)

Update instance configuration.

pub fn initialization_complete(&mut self)

Marks the end of any initialization commands.

Intended to be called by Controller, rather than by other code. Calling this method repeatedly has no effect.

pub fn allow_writes(&mut self)

Allows this instance to affect writes to external systems (persist).

Calling this method repeatedly has no effect.

pub fn shutdown(&mut self)

Shut down this instance.

This method runs various assertions ensuring the instance state is empty. It exists to help us find bugs where the client drops a compute instance that still has replicas or collections installed, and later assumes that said replicas/collections still exists.

Panics

Panics if the compute instance still has active replicas. Panics if the compute instance still has collections installed.

fn send(&mut self, cmd: ComputeCommand<T>)

Sends a command to all replicas of this instance.

pub fn add_replica( &mut self, id: ReplicaId, config: ReplicaConfig, epoch: Option<u64>, ) -> Result<(), ReplicaExists>

Add a new instance replica, by ID.

pub fn remove_replica(&mut self, id: ReplicaId) -> Result<(), ReplicaMissing>

Remove an existing instance replica, by ID.

fn rehydrate_replica(&mut self, id: ReplicaId)

Rehydrate the given instance replica.

Panics

Panics if the specified replica does not exist.

fn rehydrate_failed_replicas(&mut self)

Rehydrate any failed replicas of this instance.

pub fn create_dataflow( &mut self, dataflow: DataflowDescription<Plan<T>, (), T>, import_read_holds: Vec<ReadHold<T>>, subscribe_target_replica: Option<ReplicaId>, shared_collection_state: BTreeMap<GlobalId, SharedCollectionState<T>>, ) -> Result<(), DataflowCreationError>

Creates the described dataflow and initializes state for its output.

This method expects a DataflowDescription with an as_of frontier specified, as well as for each imported collection a read hold in import_read_holds at at least the as_of.

If a subscribe_target_replica is given, any subscribes exported by the dataflow are configured to target that replica, i.e., only subscribe responses sent by that replica are considered.

fn maybe_schedule_collection(&mut self, id: GlobalId)

Schedule the identified collection if all its inputs are available.

Panics

Panics if the identified collection does not exist.

fn schedule_collections(&mut self)

Schedule any unscheduled collections that are ready.

pub fn drop_collections( &mut self, ids: Vec<GlobalId>, ) -> Result<(), CollectionMissing>

Drops the read capability for the given collections and allows their resources to be reclaimed.

pub fn peek( &mut self, peek_target: PeekTarget, literal_constraints: Option<Vec<Row>>, uuid: Uuid, timestamp: T, result_desc: RelationDesc, finishing: RowSetFinishing, map_filter_project: SafeMfpPlan, read_hold: ReadHold<T>, target_replica: Option<ReplicaId>, peek_response_tx: Sender<PeekResponse>, ) -> Result<(), PeekError>

Initiate a peek request for the contents of id at timestamp.

pub fn cancel_peek(&mut self, uuid: Uuid, reason: PeekResponse)

Cancels an existing peek request.

pub fn set_read_policy( &mut self, policies: Vec<(GlobalId, ReadPolicy<T>)>, ) -> Result<(), ReadPolicyError>

Assigns a read policy to specific identifiers.

The policies are assigned in the order presented, and repeated identifiers should conclude with the last policy. Changing a policy will immediately downgrade the read capability if appropriate, but it will not “recover” the read capability if the prior capability is already ahead of it.

Identifiers not present in policies retain their existing read policies.

It is an error to attempt to set a read policy for a collection that is not readable in the context of compute. At this time, only indexes are readable compute collections.

fn maybe_update_global_write_frontier( &mut self, id: GlobalId, new_frontier: Antichain<T>, )

Advance the global write frontier of the given collection.

Frontier regressions are gracefully ignored.

Panics

Panics if the identified collection does not exist.

fn apply_read_hold_change(&mut self, id: GlobalId, update: ChangeBatch<T>)

Apply a collection read hold change.

fn finish_peek(&mut self, uuid: Uuid, response: PeekResponse)

Fulfills a registered peek and cleans up associated state.

As part of this we:

• Send a PeekResponse through the peek’s response channel.
• Emit a CancelPeek command to instruct replicas to stop spending resources on this peek, and to allow the ComputeCommandHistory to reduce away the corresponding Peek command.
• Remove the read hold for this peek, unblocking compaction that might have waited on it.

fn handle_response( &mut self, (replica_id, epoch, response): (ReplicaId, u64, ComputeResponse<T>), )

Handles a response from a replica. Replica IDs are re-used across replica restarts, so we use the replica epoch to drop stale responses.

fn handle_frontiers_response( &mut self, id: GlobalId, frontiers: FrontiersResponse<T>, replica_id: ReplicaId, )

Handle new frontiers, returning any compute response that needs to be sent to the client.

fn handle_peek_response( &mut self, uuid: Uuid, response: PeekResponse, otel_ctx: OpenTelemetryContext, replica_id: ReplicaId, )

fn handle_copy_to_response( &mut self, sink_id: GlobalId, response: CopyToResponse, replica_id: ReplicaId, )

fn handle_subscribe_response( &mut self, subscribe_id: GlobalId, response: SubscribeResponse<T>, replica_id: ReplicaId, )

fn handle_status_response( &self, response: StatusResponse, _replica_id: ReplicaId, )

fn dependency_write_frontiers<'b>( &'b self, collection: &'b CollectionState<T>, ) -> impl Iterator<Item = Antichain<T>> + 'b

Return the write frontiers of the dependencies of the given collection.

fn transitive_storage_dependency_write_frontiers<'b>( &'b self, collection: &'b CollectionState<T>, ) -> impl Iterator<Item = Antichain<T>> + 'b

Return the write frontiers of transitive storage dependencies of the given collection.

fn downgrade_warmup_capabilities(&mut self)

Downgrade the warmup capabilities of collections as much as possible.

The only requirement we have for a collection’s warmup capability is that it is for a time that is available in all of the collection’s inputs. For each input the latest time that is the case for is write_frontier - 1. So the farthest we can downgrade a collection’s warmup capability is the minimum of write_frontier - 1 of all its inputs.

This method expects to be periodically called as part of instance maintenance work. We would like to instead update the warmup capabilities synchronously in response to frontier updates of dependency collections, but that is not generally possible because we don’t learn about frontier updates of storage collections synchronously. We could do synchronous updates for compute dependencies, but we refrain from doing for simplicity.

fn forward_implied_capabilities(&mut self)

Forward the implied capabilities of collections, if possible.

The implied capability of a collection controls (a) which times are still readable (for indexes) and (b) with which as-of the collection gets installed on a new replica. We are usually not allowed to advance an implied capability beyond the frontier that follows from the collection’s read policy applied to its write frontier:

• For sink collections, some external consumer might rely on seeing all distinct times in the input reflected in the output. If we’d forward the implied capability of a sink, we’d risk skipping times in the output across replica restarts.
• For index collections, we might make the index unreadable by advancing its read frontier beyond its write frontier.

There is one case where forwarding an implied capability is fine though: an index installed on a cluster that has no replicas. Such indexes are not readable anyway until a new replica is added, so advancing its read frontier can’t make it unreadable. We can thus advance the implied capability as long as we make sure that when a new replica is added, the expected relationship between write frontier, read policy, and implied capability can be restored immediately (modulo computation time).

Forwarding implied capabilities is not necessary for the correct functioning of the controller but an optimization that is beneficial in two ways:

• It relaxes read holds on inputs to forwarded collections, allowing their compaction.
• It reduces the amount of historical detail new replicas need to process when computing forwarded collections, as forwarding the implied capability also forwards the corresponding dataflow as-of.

pub fn maintain(&mut self)

Process pending maintenance work.

This method is invoked periodically by the global controller. It is a good place to perform maintenance work that arises from various controller state changes and that cannot conveniently be handled synchronously with those state changes.