Struct mz_compute_client::controller::instance::ActiveInstance

pub(super) struct ActiveInstance<'a, T> {
    compute: &'a mut Instance<T>,
    storage_controller: &'a mut dyn StorageController<Timestamp = T>,
}

A wrapper around Instance with a live storage controller.

Fields

compute: &'a mut Instance<T>

storage_controller: &'a mut dyn StorageController<Timestamp = T>

Implementations

impl<'a, T> ActiveInstance<'a, T>where T: Timestamp + Lattice, ComputeGrpcClient: ComputeClient<T>,

pub fn add_replica( &mut self, id: ReplicaId, config: ReplicaConfig ) -> 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> ) -> Result<(), DataflowCreationError>

Create the described dataflows and initializes state for their output.

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, id: GlobalId, literal_constraints: Option<Vec<Row>>, uuid: Uuid, timestamp: T, finishing: RowSetFinishing, map_filter_project: SafeMfpPlan, target_replica: Option<ReplicaId>, peek_target: PeekTarget ) -> Result<(), PeekError>

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

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

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 update_write_frontiers( &mut self, replica_id: ReplicaId, updates: &BTreeMap<GlobalId, Antichain<T>> )

Accept write frontier updates from the compute layer.

Panics

Panics if any of the updates references an absent collection. Panics if any of the updates regresses an existing write frontier.

fn update_replica_write_frontiers( &mut self, replica_id: ReplicaId, updates: &BTreeMap<GlobalId, Antichain<T>> )

Apply replica write frontier updates.

Panics

Panics if any of the updates references an absent collection. Panics if any of the updates regresses an existing replica write frontier.

fn remove_replica_write_frontiers(&mut self, replica_id: ReplicaId)

Remove frontier tracking state for the given replica.

fn maybe_update_global_write_frontiers( &mut self, updates: &BTreeMap<GlobalId, Antichain<T>> )

Apply global write frontier updates.

Frontier regressions are gracefully ignored.

Panics

Panics if any of the updates references an absent collection.

fn update_read_capabilities( &mut self, updates: BTreeMap<GlobalId, ChangeBatch<T>> )

Applies updates, propagates consequences through other read capabilities, and sends appropriate compaction commands.

fn remove_peek(&mut self, uuid: Uuid)

Removes a registered peek and clean up associated state.

As part of this we:

• 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.

pub fn handle_response( &mut self, response: ComputeResponse<T>, replica_id: ReplicaId ) -> Option<ComputeControllerResponse<T>>

fn handle_frontier_upper( &mut self, id: GlobalId, new_frontier: Antichain<T>, replica_id: ReplicaId )

fn handle_peek_response( &mut self, uuid: Uuid, response: PeekResponse, otel_ctx: OpenTelemetryContext, replica_id: ReplicaId ) -> Option<ComputeControllerResponse<T>>

fn handle_copy_to_response( &mut self, sink_id: GlobalId, response: CopyToResponse, replica_id: ReplicaId ) -> Option<ComputeControllerResponse<T>>

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

fn handle_status_response( &mut self, response: StatusResponse, replica_id: ReplicaId )

impl<'a, T> ActiveInstance<'a, T>where T: TimestampManipulation, ComputeGrpcClient: ComputeClient<T>,

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.

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.

Trait Implementations

impl<'a, T: Debug> Debug for ActiveInstance<'a, T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.