Struct mz_storage_client::controller::Controller

pub struct Controller<T: Timestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation> { /* private fields */ }

A storage controller for a storage instance.

Implementations

impl<T> Controller<T>where T: Timestamp + Lattice + TotalOrder + Codec64 + From<EpochMillis> + TimestampManipulation, StorageCommand<T>: RustType<ProtoStorageCommand>, StorageResponse<T>: RustType<ProtoStorageResponse>, Self: StorageController<Timestamp = T>,

pub async fn new( build_info: &'static BuildInfo, postgres_url: String, persist_location: PersistLocation, persist_clients: Arc<PersistClientCache>, now: NowFn, postgres_factory: &StashFactory, envd_epoch: NonZeroI64, metrics_registry: MetricsRegistry ) -> Self

Create a new storage controller from a client it should wrap.

Note that when creating a new storage controller, you must also reconcile it with the previous state.

Trait Implementations

impl<T: Debug + Timestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation> Debug for Controller<T>

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

Formats the value using the given formatter.

impl<T> StorageController for Controller<T>where T: Timestamp + Lattice + TotalOrder + Codec64 + From<EpochMillis> + TimestampManipulation + Into<Timestamp>, StorageCommand<T>: RustType<ProtoStorageCommand>, StorageResponse<T>: RustType<ProtoStorageResponse>, MetadataExportFetcher: MetadataExport<T>, DurableExportMetadata<T>: RustType<DurableExportMetadata>,

fn drop_sinks(&mut self, identifiers: Vec<GlobalId>) -> Result<(), StorageError>

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

type Timestamp = T

fn initialization_complete(&mut self)

Marks the end of any initialization commands.

fn update_configuration(&mut self, config_params: StorageParameters)

Update storage configuration.

fn collection( &self, id: GlobalId ) -> Result<&CollectionState<Self::Timestamp>, StorageError>

Acquire an immutable reference to the collection state, should it exist.

fn collection_mut( &mut self, id: GlobalId ) -> Result<&mut CollectionState<Self::Timestamp>, StorageError>

Acquire a mutable reference to the collection state, should it exist.

fn collections( &self ) -> Box<dyn Iterator<Item = (&GlobalId, &CollectionState<Self::Timestamp>)> + '_>

Acquire an iterator over all collection states.

fn create_instance( &mut self, id: StorageInstanceId, variable_length_row_encoding: bool )

Creates a storage instance with the specified ID.

fn drop_instance(&mut self, id: StorageInstanceId)

Drops the storage instance with the given ID.

fn connect_replica( &mut self, instance_id: StorageInstanceId, replica_id: ReplicaId, location: ClusterReplicaLocation )

Connects the storage instance to the specified replica.

fn drop_replica( &mut self, instance_id: StorageInstanceId, _replica_id: ReplicaId )

Disconnects the storage instance from the specified replica.

fn migrate_collections<'life0, 'async_trait>( &'life0 mut self, _collections: Vec<(GlobalId, CollectionDescription<Self::Timestamp>)> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Migrate any storage controller state from previous versions to this version’s expectations.

fn create_collections<'life0, 'async_trait>( &'life0 mut self, collections: Vec<(GlobalId, CollectionDescription<Self::Timestamp>)> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Create the sources described in the individual RunIngestionCommand commands.

fn check_alter_collection( &mut self, id: GlobalId, ingestion: IngestionDescription ) -> Result<(), StorageError>

Check that the collection associated with id can be altered to represent the given ingestion.

fn alter_collection<'life0, 'async_trait>( &'life0 mut self, id: GlobalId, ingestion: IngestionDescription ) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Alter the identified collection to use the described ingestion.

fn export( &self, id: GlobalId ) -> Result<&ExportState<Self::Timestamp>, StorageError>

Acquire an immutable reference to the export state, should it exist.

fn export_mut( &mut self, id: GlobalId ) -> Result<&mut ExportState<Self::Timestamp>, StorageError>

Acquire a mutable reference to the export state, should it exist.

fn prepare_export( &mut self, id: GlobalId, from_id: GlobalId ) -> Result<CreateExportToken<T>, StorageError>

Notify the storage controller to prepare for an export to be created

fn cancel_prepare_export(&mut self, _: CreateExportToken<T>)

Cancel the pending export

fn create_exports<'life0, 'async_trait>( &'life0 mut self, exports: Vec<(CreateExportToken<Self::Timestamp>, ExportDescription<Self::Timestamp>)> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Create the sinks described by the ExportDescription.

fn drop_sources( &mut self, identifiers: Vec<GlobalId> ) -> Result<(), StorageError>

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

fn drop_sources_unvalidated(&mut self, identifiers: Vec<GlobalId>)

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

fn drop_sinks_unvalidated(&mut self, identifiers: Vec<GlobalId>)

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

fn append_table( &mut self, commands: Vec<(GlobalId, Vec<Update<Self::Timestamp>>, Self::Timestamp)> ) -> Result<Receiver<Result<(), StorageError>>, StorageError>

Append updates into the local input named id and advance its upper to upper.

fn monotonic_appender( &self, id: GlobalId ) -> Result<MonotonicAppender, StorageError>

Returns a MonotonicAppender which is a oneshot-esque struct that can be used to monotonically append to the specified GlobalId.

fn snapshot<'life0, 'async_trait>( &'life0 self, id: GlobalId, as_of: Self::Timestamp ) -> Pin<Box<dyn Future<Output = Result<Vec<(Row, Diff)>, StorageError>> + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Returns the snapshot of the contents of the local input named id at as_of.

fn snapshot_stats<'life0, 'async_trait>( &'life0 self, id: GlobalId, as_of: Antichain<Self::Timestamp> ) -> Pin<Box<dyn Future<Output = Result<SnapshotStats<Self::Timestamp>, StorageError>> + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Returns aggregate statistics about the contents of the local input named id at as_of.

fn set_read_policy( &mut self, policies: Vec<(GlobalId, ReadPolicy<Self::Timestamp>)> )

Assigns a read policy to specific identifiers.

fn update_write_frontiers( &mut self, updates: &[(GlobalId, Antichain<Self::Timestamp>)] )

Ingests write frontier updates for collections that this controller maintains and potentially generates updates to read capabilities, which are passed on to StorageController::update_read_capabilities.

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

Applies updates and sends any appropriate compaction command.

fn ready<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = ()> + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Waits until the controller is ready to process a response.

fn process<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Processes the work queued by StorageController::ready.

fn reconcile_state<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = ()> + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Signal to the controller that the adapter has populated all of its initial state and the controller can reconcile (i.e. drop) any unclaimed resources.

fn inspect_persist_state<'life0, 'async_trait>( &'life0 self, id: GlobalId ) -> Pin<Box<dyn Future<Output = Result<Value, Error>> + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Exposes the internal state of the data shard for debugging and QA.

fn record_frontiers<'life0, 'async_trait>( &'life0 mut self, external_frontiers: BTreeMap<(GlobalId, ReplicaId), Antichain<Self::Timestamp>> ) -> Pin<Box<dyn Future<Output = ()> + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Records the current frontiers of all known storage objects.