Struct mz_dataflow_types::client::controller::storage::Controller

pub struct Controller<T: Timestamp + Lattice + Codec64 + Unpin> {
    state: StorageControllerState<T>,
    persist_location: PersistLocation,
    persist_client: PersistClient,
    orchestrator: Arc<dyn NamespacedOrchestrator>,
    storaged_image: String,
}

A storage controller for a storage instance.

Fields

state: StorageControllerState<T>

persist_location: PersistLocation

The persist location where all storage collections are being written to

persist_client: PersistClient

A persist client used to write to storage collections

orchestrator: Arc<dyn NamespacedOrchestrator>

An orchestrator to start and stop storage processes.

storaged_image: String

The storaged image to use when starting new storage processes.

Implementations

impl<T> Controller<T> where
    T: Timestamp + Lattice + TotalOrder + TryInto<i64> + TryFrom<i64> + Codec64 + Unpin,
    <T as TryInto<i64>>::Error: Debug,
    <T as TryFrom<i64>>::Error: Debug,
    StorageCommand<T>: RustType<ProtoStorageCommand>,
    StorageResponse<T>: RustType<ProtoStorageResponse>, 

pub async fn new(
    postgres_url: String,
    persist_location: PersistLocation,
    persist_clients: Arc<Mutex<PersistClientCache>>,
    orchestrator: Arc<dyn NamespacedOrchestrator>,
    storaged_image: String
) -> Self

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

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

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

Trait Implementations

impl<T: Debug + Timestamp + Lattice + Codec64 + Unpin> 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 + TryInto<i64> + TryFrom<i64> + Codec64 + Unpin,
    <T as TryInto<i64>>::Error: Debug,
    <T as TryFrom<i64>>::Error: Debug,
    StorageCommand<T>: RustType<ProtoStorageCommand>,
    StorageResponse<T>: RustType<ProtoStorageResponse>, 

fn linearize_sources<'life0, 'async_trait>(
    &'life0 mut self,
    _peek_id: Uuid,
    _source_ids: Vec<GlobalId>
) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'async_trait>> where
    'life0: 'async_trait,
    Self: 'async_trait, 

“Linearize” the listed sources.

If these sources are valid and “linearizable”, then the response will respond with timestamps that are guaranteed to be up-to-date with the max offset found at the time of the command issuance.

Note: “linearizable” in this context may not represent true linearizability in all cases.

type Timestamp = T

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

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

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

Create the sources described in the individual CreateSourceCommand commands.

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

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

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

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>> + Send + '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<T>)>
) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + Send + '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, ChangeBatch<T>)]
) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + Send + 'async_trait>> where
    'life0: 'async_trait,
    'life1: 'async_trait,
    Self: 'async_trait, 

Accept write frontier updates from the compute layer.

fn update_read_capabilities<'life0, 'life1, 'async_trait>(
    &'life0 mut self,
    updates: &'life1 mut BTreeMap<GlobalId, ChangeBatch<T>>
) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + Send + 'async_trait>> where
    'life0: 'async_trait,
    'life1: 'async_trait,
    Self: 'async_trait, 

Applies updates and sends any appropriate compaction command.

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