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,
    StorageCommand<T>: RustType<ProtoStorageCommand>,
    StorageResponse<T>: RustType<ProtoStorageResponse>,
    MetadataExportFetcher: MetadataExport<T>,
    DurableExportMetadata<T>: Data,

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)

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,
    id: StorageInstanceId,
    location: ClusterReplicaLocation
)

Connects the storage instance to 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 CreateSourceCommand commands.

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(
    &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 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 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.