Struct mz_storage::controller::Controller

pub struct Controller<T: Timestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation> {
    state: StorageControllerState<T>,
    hosts: StorageHosts<T>,
    internal_response_queue: UnboundedReceiver<StorageResponse<T>>,
    persist_location: PersistLocation,
    persist: Arc<Mutex<PersistClientCache>>,
}

A storage controller for a storage instance.

Fields

state: StorageControllerState<T>

hosts: StorageHosts<T>

Storage host provisioning and storage object assignment.

internal_response_queue: UnboundedReceiver<StorageResponse<T>>

Mechanism for returning frontier advancement for tables.

persist_location: PersistLocation

The persist location where all storage collections are being written to

persist: Arc<Mutex<PersistClientCache>>

A persist client used to write to storage collections

Implementations

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

pub async fn new(
    build_info: &'static BuildInfo,
    postgres_url: String,
    persist_location: PersistLocation,
    persist_clients: Arc<Mutex<PersistClientCache>>,
    orchestrator: Arc<dyn NamespacedOrchestrator>,
    storaged_image: String,
    now: NowFn
) -> Self

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

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>,

fn validate_collection_ids(
    &self,
    ids: impl Iterator<Item = GlobalId>
) -> Result<(), StorageError>

Validate that a collection exists for all identifiers, and error if any do not.

fn validate_export_ids(
    &self,
    ids: impl Iterator<Item = GlobalId>
) -> Result<(), StorageError>

Validate that a collection exists for all identifiers, and error if any do not.

fn generate_new_capability_for_collection<F>(
    &mut self,
    id: GlobalId,
    f: F
) -> Result<ChangeBatch<<Self as StorageController>::Timestamp>, StorageError>where
    F: FnOnce(&mut CollectionState<<Self as StorageController>::Timestamp>),

Trait Implementations

impl<T> CollectionManagement 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 truncate_managed_collection<'life0, 'async_trait>(
    &'life0 mut self,
    id: GlobalId
) -> Pin<Box<dyn Future<Output = ()> + 'async_trait>>where
    'life0: 'async_trait,
    Self: 'async_trait,

Effectively truncates the data_shard associated with global_id effective as of the system time.

Panics

• If id does not belong to a collection or is not registered as a managed collection.

fn append_to_managed_collection<'life0, 'async_trait>(
    &'life0 mut self,
    id: GlobalId,
    updates: Vec<(Row, Diff)>
) -> Pin<Box<dyn Future<Output = ()> + 'async_trait>>where
    'life0: 'async_trait,
    Self: 'async_trait,

Append updates to the data_shard associated with global_id effective as of the system time.

Panics

• If id is not registered as a managed collection.

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

Append updates to the data_shard correlated with global_id effective as of the system time.

Panics

• If IntrospectionType::ShardMapping is not correlated with a GlobalId.
• If IntrospectionType::ShardMapping’s GlobalId is not registered as a managed collection.

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

Tracks the mapping of GlobalId to data shards in the collection at self.state.shard_collection_global_id.

However, data is written iff we know of the GlobalId of the IntrospectionType::ShardMapping collection; in other cases, data is dropped on the floor. In these cases, the data is later written by Self::initialize_shard_mapping.

Panics

• If self.state.collections does not have an entry for global_id.
• If IntrospectionType::ShardMapping’s GlobalId is not registered as a managed collection.

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<'life0, 'async_trait>(
    &'life0 mut self,
    identifiers: Vec<GlobalId>
) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + 'async_trait>>where
    'life0: 'async_trait,
    Self: 'async_trait,

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

Create the sources described in the individual CreateSourceCommand commands.

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

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<'life0, 'async_trait>(
    &'life0 mut self,
    id: GlobalId,
    from_id: GlobalId
) -> Pin<Box<dyn Future<Output = Result<CreateExportToken, StorageError>> + 'async_trait>>where
    'life0: 'async_trait,
    Self: 'async_trait,

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

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

Cancel the pending export

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

Create the sinks described by the ExportDescription.

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

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

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

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

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

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 mut self,
    id: GlobalId,
    as_of: Self::Timestamp
) -> Pin<Box<dyn Future<Output = Result<Vec<(Row, Diff)>, StorageError>> + 'async_trait>>where
    'life0: 'async_trait,
    Self: 'async_trait,

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

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

Assigns a read policy to specific identifiers.

fn update_write_frontiers<'life0, 'life1, 'async_trait>(
    &'life0 mut self,
    updates: &'life1 [(GlobalId, Antichain<Self::Timestamp>)]
) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + 'async_trait>>where
    'life0: 'async_trait,
    'life1: 'async_trait,
    Self: 'async_trait,

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<'life0, 'life1, 'async_trait>(
    &'life0 mut self,
    updates: &'life1 mut BTreeMap<GlobalId, ChangeBatch<Self::Timestamp>>
) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + 'async_trait>>where
    'life0: 'async_trait,
    'life1: 'async_trait,
    Self: 'async_trait,

Applies updates and sends any appropriate compaction command.

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

Processes the work queued by StorageController::ready.