Struct mz_storage_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, persist_location: PersistLocation, persist_clients: Arc<PersistClientCache>, now: NowFn, envd_epoch: NonZeroI64, metrics_registry: MetricsRegistry, persist_txn_tables: PersistTxnTablesImpl, connection_context: ConnectionContext, txn: &dyn StorageTxn<T> ) -> 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.

Panics

If this function is called before prepare_initialization.

Trait Implementations

impl<T> Debug for Controller<T>

where T: Debug + Timestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation,

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

Formats the value using the given formatter.

impl<'a, T> StorageController for Controller<T>

where T: Timestamp + Lattice + TotalOrder + Codec64 + From<EpochMillis> + TimestampManipulation + Into<Datum<'a>> + Display, StorageCommand<T>: RustType<ProtoStorageCommand>, StorageResponse<T>: RustType<ProtoStorageResponse>,

fn config(&self) -> &StorageConfiguration

Get the current configuration

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

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Create and “execute” the described collection.

“Execute” is in scare quotes because what executing a collection means varies widely based on the type of collection you’re creating.

The general process creating a collection undergoes is:

1. Enrich the description we get from the user with the metadata only the storage controller’s metadata. This is mostly a matter of separating concerns.
2. Generate write and read persist handles for the collection.
3. Store the collection’s metadata in the appropriate field.
4. “Execte” the collection. What that means is contingent on the type of collection. so consult the code for more details.

fn alter_export_connections<'life0, 'async_trait>( &'life0 mut self, exports: BTreeMap<GlobalId, StorageSinkConnection> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Create the sinks described by the ExportDescription.

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

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

fn init_txns<'life0, 'async_trait>( &'life0 mut self, init_ts: T ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

With the CRDB based timestamp oracle, there is no longer write timestamp fencing. As in, when a new Coordinator, B, starts up, there is nothing that prevents an old Coordinator, A, from getting a new write timestamp that is higher than B’s boot timestamp. Below is the implications for all persist transaction scenarios, on means a Coordinator turning the persist txn flag on, off means a Coordinator turning the persist txn flag off.

The following series of events is a concern:

1. A writes at t_0, s.t. t_0 > B’s boot timestamp.
2. B writes at t_1, s.t. t_1 > t_0.
3. A writes at t_2, s.t. t_2 > t_1.
4. etc.

• off -> off: If B`` manages to append t_1before A appendst_0then thet_0append will panic and we won't acknowledge the write to the user (or similarlyt_2andt_1`). Before persist-txn, appends are not atomic, so we might get a partial append. This is fine because we only support single table transactions.
• on -> on: The txn-shard is meant to correctly handle two writers so this should be fine. Note it’s possible that we have two Coordinators interleaving write transactions without the leadership check described below, but that should be fine.
• off -> on: If A gets a write timestamp higher than B’s boot timestamp, then A can write directly to a data shard after it’s been registered with a txn-shard, breaking the invariant that no data shard is written to directly while it’s registered to a transaction shard. To mitigate this, we must do a leadership check AFTER getting the write timestamp. In order for B to register a data shard in the txn shard, it must first become the leader then second get a register timestamp. So if A gets a write timestamp higher than B’s register timestamp, it will fail the leadership check before attempting the append.
• on -> off: If A tries to write to the txn-shard at a timestamp higher than B’s boot timestamp, it will fail because the shards have been forgotten. So everything should be ok.

In general, all transitions make the following steps:

1. Get write timestamp, ts.
2. Apply all transactions to all data shards up to ts.
3. Register/forget all data shards. So if we crash at any point in these steps, for example after only applying some transactions, then the next Coordinator can pick up where we left off and finish whatever needs finishing.

H/t jkosh44 for the above notes from the discussion in which we hashed this all out.

type Timestamp = T

fn initialization_complete(&mut self)

Marks the end of any initialization commands.

fn update_parameters(&mut self, config_params: StorageParameters)

Update storage configuration with new parameters.

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

Returns the CollectionMetadata of the collection identified by id.

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

Returns the since/upper frontiers of the identified collection.

fn collections_frontiers( &self, ids: Vec<GlobalId> ) -> Result<Vec<(GlobalId, Antichain<T>, Antichain<T>)>, StorageError<Self::Timestamp>>

Returns the since/upper frontiers of the identified collections.

fn active_collection_metadatas(&self) -> Vec<(GlobalId, CollectionMetadata)>

Acquire an iterator over CollectionMetadata for all active collections.

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

Checks whether a collection exists under the given GlobalId. Returns an error if the collection does not exist.

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, 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 check_alter_ingestion_source_desc( &mut self, ingestion_id: GlobalId, source_desc: &SourceDesc ) -> Result<(), StorageError<Self::Timestamp>>

Check that the ingestion associated with id can use the provided SourceDesc.

fn alter_ingestion_source_desc<'life0, 'async_trait>( &'life0 mut self, ingestion_id: GlobalId, source_desc: SourceDesc ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Alters the identified collection to use the provided SourceDesc.

fn alter_ingestion_connections<'life0, 'async_trait>( &'life0 mut self, source_connections: BTreeMap<GlobalId, GenericSourceConnection<InlinedConnection>> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Alters each identified collection to use the correlated GenericSourceConnection.

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

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

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

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

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

where Self: 'async_trait, 'life0: 'async_trait,

Create the sinks described by the ExportDescription.

fn drop_tables( &mut self, identifiers: Vec<GlobalId>, ts: Self::Timestamp ) -> Result<(), StorageError<Self::Timestamp>>

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

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

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

fn drop_sources_unvalidated( &mut self, storage_metadata: &StorageMetadata, ids: Vec<GlobalId> ) -> Result<(), StorageError<Self::Timestamp>>

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, write_ts: Self::Timestamp, advance_to: Self::Timestamp, commands: Vec<(GlobalId, Vec<TimestamplessUpdate>)> ) -> Result<Receiver<Result<(), StorageError<Self::Timestamp>>>, StorageError<Self::Timestamp>>

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

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

Returns a MonotonicAppender which is a channel that can be used to monotonically append to the specified GlobalId.

fn webhook_statistics( &self, id: GlobalId ) -> Result<Arc<WebhookStatistics>, StorageError<Self::Timestamp>>

Returns a shared WebhookStatistics which can be used to report user-facing statistics for this given webhhook, specified by the GlobalId.

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

where Self::Timestamp: Timestamp + Lattice + Codec64, 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, StorageError<Self::Timestamp>>> + '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 snapshot_parts_stats<'life0, 'async_trait>( &'life0 self, id: GlobalId, as_of: Antichain<Self::Timestamp> ) -> Pin<Box<dyn Future<Output = BoxFuture<'static, Result<SnapshotPartsStats, StorageError<Self::Timestamp>>>> + '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 acquire_read_holds( &mut self, desired_holds: Vec<GlobalId> ) -> Result<Vec<ReadHold<Self::Timestamp>>, ReadHoldError>

Acquires and returns the desired read holds, advancing them to the since frontier when necessary.

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, 'life1, 'async_trait>( &'life0 mut self, storage_metadata: &'life1 StorageMetadata ) -> Pin<Box<dyn Future<Output = Result<Option<Response<T>>, Error>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Processes the work queued by StorageController::ready.

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, (Antichain<Self::Timestamp>, Antichain<Self::Timestamp>)> ) -> Pin<Box<dyn Future<Output = ()> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Records the current read and write frontiers of all known storage objects.

fn record_replica_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 per-replica write frontiers of all known storage objects.

fn record_introspection_updates<'life0, 'async_trait>( &'life0 mut self, type_: IntrospectionType, updates: Vec<(Row, Diff)> ) -> Pin<Box<dyn Future<Output = ()> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Records updates for the given introspection type.

fn initialize_state<'life0, 'life1, 'async_trait>( &'life0 mut self, txn: &'life1 mut dyn StorageTxn<T>, init_ids: BTreeSet<GlobalId>, drop_ids: BTreeSet<GlobalId> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<T>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

On boot, seed the controller’s metadata/state.

fn prepare_state<'life0, 'life1, 'async_trait>( &'life0 mut self, txn: &'life1 mut dyn StorageTxn<T>, ids_to_add: BTreeSet<GlobalId>, ids_to_drop: BTreeSet<GlobalId> ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<T>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Update the implementor of StorageTxn with the appropriate metadata given the IDs to add and drop.

fn acquire_read_hold( &mut self, id: GlobalId ) -> Result<ReadHold<Self::Timestamp>, ReadHoldError>

Acquires and returns the earliest legal read hold.

fn acquire_read_hold_at_time( &mut self, id: GlobalId, desired_hold: Antichain<Self::Timestamp> ) -> Result<ReadHold<Self::Timestamp>, ReadHoldError>

Acquires and returns a read hold at desired_time. Returns ReadHoldError::SinceViolation when that is not possible.