Struct Controller 

pub struct Controller { /* private fields */ }

A storage controller for a storage instance.

Implementations

impl Controller

where Self: StorageController,

pub async fn new( build_info: &'static BuildInfo, persist_location: PersistLocation, persist_clients: Arc<PersistClientCache>, now: NowFn, wallclock_lag: WallclockLagFn<Timestamp>, txns_metrics: Arc<TxnMetrics>, read_only: bool, metrics_registry: &MetricsRegistry, controller_metrics: ControllerMetrics, connection_context: ConnectionContext, txn: &dyn StorageTxn, storage_collections: Arc<dyn StorageCollections + Send + Sync>, ) -> 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 Debug for Controller

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

Formats the value using the given formatter.

impl StorageController for Controller

fn config(&self) -> &StorageConfiguration

Get the current configuration

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

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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. “Execute” the collection. What that means is contingent on the type of collection. so consult the code for more details.

fn create_oneshot_ingestion<'life0, 'async_trait>( &'life0 mut self, ingestion_id: Uuid, collection_id: GlobalId, instance_id: StorageInstanceId, request: OneshotIngestionRequest, result_tx: OneshotResultCallback<ProtoBatch>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + 'async_trait>>

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

Create a oneshot ingestion.

fn alter_export_connections<'life0, 'async_trait>( &'life0 mut self, exports: BTreeMap<GlobalId, StorageSinkConnection>, ) -> 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_sinks( &mut self, storage_metadata: &StorageMetadata, identifiers: Vec<GlobalId>, ) -> Result<(), StorageError>

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

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

Returns the CollectionMetadata of the collection identified by id.

fn collection_hydrated( &self, collection_id: GlobalId, ) -> Result<bool, StorageError>

Returns true iff the given collection/ingestion has been hydrated.

fn collections_hydrated_on_replicas( &self, target_replica_ids: Option<Vec<ReplicaId>>, target_cluster_id: &StorageInstanceId, exclude_collections: &BTreeSet<GlobalId>, ) -> Result<bool, StorageError>

Returns true if each non-transient, non-excluded collection is hydrated on at least one of the provided replicas.

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

Returns the since/upper frontiers of the identified collection.

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

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 active_ingestion_exports( &self, instance_id: StorageInstanceId, ) -> Box<dyn Iterator<Item = &GlobalId> + '_>

Returns the IDs of ingestion exports running on the given instance. This includes the ingestion itself, if any, and running source tables (aka. subsources).

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

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, workload_class: Option<String>, )

Creates a storage instance with the specified ID.

fn drop_instance(&mut self, id: StorageInstanceId)

Drops the storage instance with the given ID.

fn update_instance_workload_class( &mut self, id: StorageInstanceId, workload_class: Option<String>, )

Updates a storage instance’s workload class.

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 evolve_nullability_for_bootstrap<'life0, 'life1, 'async_trait>( &'life0 mut self, storage_metadata: &'life1 StorageMetadata, collections: Vec<(GlobalId, RelationDesc)>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + 'async_trait>>

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

Across versions of Materialize the nullability of columns for some objects can change based on updates to our optimizer.

fn check_alter_ingestion_source_desc( &mut self, ingestion_id: GlobalId, source_desc: &SourceDesc, ) -> Result<(), StorageError>

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

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

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

Alters each identified ingestion to use the correlated SourceDesc.

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

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

Alters each identified collection to use the correlated GenericSourceConnection.

fn alter_ingestion_export_data_configs<'life0, 'async_trait>( &'life0 mut self, source_exports: BTreeMap<GlobalId, SourceExportDataConfig>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + 'async_trait>>

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

Alters the data config for the specified source exports of the specified ingestions.

fn alter_table_desc<'life0, 'async_trait>( &'life0 mut self, existing_collection: GlobalId, new_collection: GlobalId, new_desc: RelationDesc, expected_version: RelationVersion, register_ts: Timestamp, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + 'async_trait>>

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

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

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

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

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

fn cancel_oneshot_ingestion( &mut self, ingestion_id: Uuid, ) -> Result<(), StorageError>

Cancel a oneshot ingestion.

fn alter_export<'life0, 'async_trait>( &'life0 mut self, id: GlobalId, new_description: ExportDescription, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError>> + 'async_trait>>

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

Alter the sink identified by the given id to match the provided ExportDescription.

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

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>

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>

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

fn drop_sinks_unvalidated( &mut self, storage_metadata: &StorageMetadata, sinks_to_drop: Vec<GlobalId>, )

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

fn append_table( &mut self, write_ts: Timestamp, advance_to: Timestamp, commands: Vec<(GlobalId, Vec<TableData>)>, ) -> 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 channel that can be used to monotonically append to the specified GlobalId.

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

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

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( &mut self, storage_metadata: &StorageMetadata, ) -> Result<Option<Response>, Error>

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 append_introspection_updates( &mut self, type_: IntrospectionType, updates: Vec<(Row, Diff)>, )

Records append-only updates for the given introspection type.

fn append_status_introspection_updates( &mut self, type_: IntrospectionType, updates: Vec<StatusUpdate>, )

Records append-only status updates for the given introspection type.

fn update_introspection_collection( &mut self, type_: IntrospectionType, op: StorageWriteOp, )

Updates the desired state of the given introspection type.

fn append_only_introspection_tx( &self, type_: IntrospectionType, ) -> UnboundedSender<(Vec<AppendOnlyUpdate>, Sender<Result<(), StorageError>>)>

Returns a sender for updates to the specified append-only introspection collection.

fn differential_introspection_tx( &self, type_: IntrospectionType, ) -> UnboundedSender<(StorageWriteOp, Sender<Result<(), StorageError>>)>

Returns a sender for updates to the specified differential introspection collection.

fn real_time_recent_timestamp<'life0, 'async_trait>( &'life0 self, timestamp_objects: BTreeSet<GlobalId>, timeout: Duration, ) -> Pin<Box<dyn Future<Output = Result<BoxFuture<Result<Timestamp, StorageError>>, StorageError>> + 'async_trait>>

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

fn dump(&self) -> Result<Value, Error>

Returns the state of the StorageController formatted as JSON.

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

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

Create the sources described in the individual RunIngestionCommand commands.