Struct Controller

pub struct Controller<T: Timestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation> {

    pub(crate) build_info: &'static BuildInfo,
    pub(crate) now: NowFn,
    pub(crate) envd_epoch: NonZeroI64,
    pub(crate) read_only: bool,
    pub(crate) collections: BTreeMap<GlobalId, CollectionState<T>>,
    pub(crate) dropped_objects: BTreeMap<GlobalId, BTreeSet<ReplicaId>>,
    pub(crate) persist_table_worker: PersistTableWriteWorker<T>,
    pub(crate) txns_read: TxnsRead<T>,
    pub(crate) txns_metrics: Arc<Metrics>,
    pub(crate) stashed_response: Option<(Option<ReplicaId>, StorageResponse<T>)>,
    pub(crate) pending_table_handle_drops_tx: UnboundedSender<GlobalId>,
    pub(crate) pending_table_handle_drops_rx: UnboundedReceiver<GlobalId>,
    pub(crate) pending_oneshot_ingestions: BTreeMap<Uuid, PendingOneshotIngestion>,
    pub(crate) collection_manager: CollectionManager<T>,
    pub(crate) introspection_ids: BTreeMap<IntrospectionType, GlobalId>,
    pub(crate) introspection_tokens: Arc<Mutex<BTreeMap<GlobalId, Box<dyn Any + Send + Sync>>>>,
    pub(crate) source_statistics: Arc<Mutex<SourceStatistics>>,
    pub(crate) sink_statistics: Arc<Mutex<BTreeMap<GlobalId, StatsState<SinkStatisticsUpdate>>>>,
    pub(crate) statistics_interval_sender: Sender<Duration>,
    pub(crate) instances: BTreeMap<StorageInstanceId, Instance<T>>,
    pub(crate) initialized: bool,
    pub(crate) config: StorageConfiguration,
    pub(crate) persist_location: PersistLocation,
    pub(crate) persist: Arc<PersistClientCache>,
    pub(crate) metrics: StorageControllerMetrics,
    pub(crate) recorded_frontiers: BTreeMap<GlobalId, (Antichain<T>, Antichain<T>)>,
    pub(crate) recorded_replica_frontiers: BTreeMap<(GlobalId, ReplicaId), Antichain<T>>,
    pub(crate) wallclock_lag: WallclockLagFn<T>,
    pub(crate) wallclock_lag_history_last_refresh: Instant,
    pub(crate) wallclock_lag_histogram_last_refresh: Instant,
    pub(crate) storage_collections: Arc<dyn StorageCollections<Timestamp = T> + Send + Sync>,
    pub(crate) migrated_storage_collections: BTreeSet<GlobalId>,
    pub(crate) maintenance_ticker: Interval,
    pub(crate) maintenance_scheduled: bool,
    pub(crate) instance_response_tx: UnboundedSender<(Option<ReplicaId>, StorageResponse<T>)>,
    pub(crate) instance_response_rx: UnboundedReceiver<(Option<ReplicaId>, StorageResponse<T>)>,
    pub(crate) persist_warm_task: Option<AbortOnDropHandle<Box<dyn Debug + Send>>>,
}

A storage controller for a storage instance.

Fields

build_info: &'static BuildInfo

The build information for this process.

now: NowFn

A function that returns the current time.

envd_epoch: NonZeroI64

The fencing token for this instance of the controller.

read_only: bool

Whether or not this controller is in read-only mode.

When in read-only mode, neither this controller nor the instances controlled by it are allowed to affect changes to external systems (largely persist).

collections: BTreeMap<GlobalId, CollectionState<T>>

Collections maintained by the storage controller.

This collection only grows, although individual collections may be rendered unusable. This is to prevent the re-binding of identifiers to other descriptions.

dropped_objects: BTreeMap<GlobalId, BTreeSet<ReplicaId>>

Map from IDs of objects that have been dropped to replicas we are still expecting DroppedId messages from. This is cleared out once all replicas have responded.

We use this only to catch problems in the protocol between controller and replicas, for example we can differentiate between late messages for objects that have already been dropped and unexpected (read erroneous) messages from the replica.

persist_table_worker: PersistTableWriteWorker<T>

Write handle for table shards.

txns_read: TxnsRead<T>

A shared TxnsCache running in a task and communicated with over a channel.

txns_metrics: Arc<Metrics>

stashed_response: Option<(Option<ReplicaId>, StorageResponse<T>)>

pending_table_handle_drops_tx: UnboundedSender<GlobalId>

Channel for sending table handle drops.

pending_table_handle_drops_rx: UnboundedReceiver<GlobalId>

Channel for receiving table handle drops.

pending_oneshot_ingestions: BTreeMap<Uuid, PendingOneshotIngestion>

Closures that can be used to send responses from oneshot ingestions.

collection_manager: CollectionManager<T>

Interface for managed collections

introspection_ids: BTreeMap<IntrospectionType, GlobalId>

Tracks which collection is responsible for which IntrospectionType.

introspection_tokens: Arc<Mutex<BTreeMap<GlobalId, Box<dyn Any + Send + Sync>>>>

Tokens for tasks that drive updating introspection collections. Dropping this will make sure that any tasks (or other resources) will stop when needed.

source_statistics: Arc<Mutex<SourceStatistics>>

Consolidated metrics updates to periodically write. We do not eagerly initialize this, and its contents are entirely driven by StorageResponse::StatisticsUpdates’s, as well as webhook statistics.

sink_statistics: Arc<Mutex<BTreeMap<GlobalId, StatsState<SinkStatisticsUpdate>>>>

Consolidated metrics updates to periodically write. We do not eagerly initialize this, and its contents are entirely driven by StorageResponse::StatisticsUpdates’s.

statistics_interval_sender: Sender<Duration>

A way to update the statistics interval in the statistics tasks.

instances: BTreeMap<StorageInstanceId, Instance<T>>

Clients for all known storage instances.

initialized: bool

Set to true once initialization_complete has been called.

config: StorageConfiguration

Storage configuration to apply to newly provisioned instances, and use during purification.

persist_location: PersistLocation

The persist location where all storage collections are being written to

persist: Arc<PersistClientCache>

A persist client used to write to storage collections

metrics: StorageControllerMetrics

Metrics of the Storage controller

recorded_frontiers: BTreeMap<GlobalId, (Antichain<T>, Antichain<T>)>

(read, write) frontiers that have been recorded in the Frontiers collection, kept to be able to retract old rows.

recorded_replica_frontiers: BTreeMap<(GlobalId, ReplicaId), Antichain<T>>

Write frontiers that have been recorded in the ReplicaFrontiers collection, kept to be able to retract old rows.

wallclock_lag: WallclockLagFn<T>

A function that computes the lag between the given time and wallclock time.

wallclock_lag_history_last_refresh: Instant

The last time WallclockLagHistory introspection was refreshed.

wallclock_lag_histogram_last_refresh: Instant

The last time WallclockLagHistogram introspection was refreshed.

storage_collections: Arc<dyn StorageCollections<Timestamp = T> + Send + Sync>

Handle to a StorageCollections.

migrated_storage_collections: BTreeSet<GlobalId>

Migrated storage collections that can be written even in read only mode.

maintenance_ticker: Interval

Ticker for scheduling periodic maintenance work.

maintenance_scheduled: bool

Whether maintenance work was scheduled.

instance_response_tx: UnboundedSender<(Option<ReplicaId>, StorageResponse<T>)>

Shared transmit channel for replicas to send responses.

instance_response_rx: UnboundedReceiver<(Option<ReplicaId>, StorageResponse<T>)>

Receive end for replica responses.

persist_warm_task: Option<AbortOnDropHandle<Box<dyn Debug + Send>>>

Background task run at startup to warm persist state.

Implementations

impl<T> Controller<T>

where T: Timestamp + Lattice + TotalOrder + Codec64 + From<EpochMillis> + TimestampManipulation + Into<Datum<'static>>, 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, wallclock_lag: WallclockLagFn<T>, txns_metrics: Arc<TxnMetrics>, envd_epoch: NonZeroI64, read_only: bool, metrics_registry: &MetricsRegistry, controller_metrics: ControllerMetrics, connection_context: ConnectionContext, txn: &dyn StorageTxn<T>, storage_collections: Arc<dyn StorageCollections<Timestamp = T> + 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.

pub(crate) fn set_hold_policies( &mut self, policies: Vec<(GlobalId, ReadPolicy<T>)>, )

pub(crate) fn update_write_frontiers( &mut self, updates: &[(GlobalId, Antichain<T>)], )

pub(crate) fn update_hold_capabilities( &mut self, updates: &mut BTreeMap<GlobalId, ChangeBatch<T>>, )

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

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

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

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

pub(crate) async fn open_data_handles( &self, id: &GlobalId, shard: ShardId, relation_desc: RelationDesc, persist_client: &PersistClient, ) -> WriteHandle<SourceData, (), T, StorageDiff>

Opens a write and critical since handles for the given shard.

since is an optional since that the read handle will be forwarded to if it is less than its current since.

This will halt! the process if we cannot successfully acquire a critical handle with our current epoch.

pub(crate) fn register_introspection_collection( &mut self, id: GlobalId, introspection_type: IntrospectionType, write_handle: WriteHandle<SourceData, (), T, StorageDiff>, persist_client: PersistClient, ) -> Result<(), StorageError<T>>

Registers the given introspection collection and does any preparatory work that we have to do before we start writing to it. This preparatory work will include partial truncation or other cleanup schemes, depending on introspection type.

pub(crate) fn reconcile_dangling_statistics(&self)

Remove statistics for sources/sinks that were dropped but still have statistics rows hanging around.

pub(crate) fn append_shard_mappings<I>(&self, global_ids: I, diff: Diff)

where I: Iterator<Item = GlobalId>,

Appends a new global ID, shard ID pair to the appropriate collection. Use a diff of 1 to append a new entry; -1 to retract an existing entry.

Panics

• If self.collections does not have an entry for global_id.
• If IntrospectionType::ShardMapping’s GlobalId is not registered as a managed collection.
• If diff is any value other than 1 or -1.

pub(crate) fn determine_collection_dependencies( &self, self_id: GlobalId, data_source: &DataSource<T>, ) -> Result<Vec<GlobalId>, StorageError<T>>

Determines and returns this collection’s dependencies, if any.

pub(crate) async fn read_handle_for_snapshot( &self, id: GlobalId, ) -> Result<ReadHandle<SourceData, (), T, StorageDiff>, StorageError<T>>

pub(crate) fn record_status_updates(&mut self, updates: Vec<StatusUpdate>)

Handles writing of status updates for sources/sinks to the appropriate status relation

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

pub(crate) fn run_ingestion( &mut self, id: GlobalId, ) -> Result<(), StorageError<T>>

Runs the identified ingestion using the current definition of the ingestion in-memory.

pub(crate) fn run_export(&mut self, id: GlobalId) -> Result<(), StorageError<T>>

Runs the identified export using the current definition of the export that we have in memory.

pub(crate) fn update_frontier_introspection(&mut self)

Update introspection with the current frontiers of storage objects.

This method is invoked by Controller::maintain, which we expect to be called once per second during normal operation.

pub(crate) fn refresh_wallclock_lag(&mut self)

Refresh the wallclock lag introspection and metrics with the current lag values.

We measure the lag of write frontiers behind the wallclock time every second and track the maximum over 60 measurements (i.e., one minute). Every minute, we emit a new lag event to the WallclockLagHistory introspection with the current maximum.

This method is invoked by ComputeController::maintain, which we expect to be called once per second during normal operation.

pub(crate) fn maintain(&mut self)

Run periodic tasks.

This method is invoked roughly once per second during normal operation. It is a good place for tasks that need to run periodically, such as state cleanup or updating of metrics.

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<T> StorageController for Controller<T>

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

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<Self::Timestamp>, collections: Vec<(GlobalId, CollectionDescription<Self::Timestamp>)>, migrated_storage_collections: &'life2 BTreeSet<GlobalId>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + '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<Self::Timestamp>>> + '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<Self::Timestamp>>> + '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<Self::Timestamp>>

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_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_hydrated( &self, collection_id: GlobalId, ) -> Result<bool, StorageError<Self::Timestamp>>

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

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

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

Returns the IDs of all active ingestions for the given storage instance.

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

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

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 cancel_oneshot_ingestion( &mut self, ingestion_id: Uuid, ) -> Result<(), StorageError<Self::Timestamp>>

Cancel a oneshot ingestion.

fn alter_export<'life0, 'async_trait>( &'life0 mut self, id: GlobalId, new_description: ExportDescription<Self::Timestamp>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + '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: 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, 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: Self::Timestamp, advance_to: Self::Timestamp, commands: Vec<(GlobalId, Vec<TableData>)>, ) -> 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 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<T>>, 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 real_time_recent_timestamp<'life0, 'async_trait>( &'life0 self, timestamp_objects: BTreeSet<GlobalId>, timeout: Duration, ) -> Pin<Box<dyn Future<Output = Result<BoxFuture<Result<Self::Timestamp, StorageError<Self::Timestamp>>>, StorageError<Self::Timestamp>>> + 'async_trait>>

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

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 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

Create the sources described in the individual RunIngestionCommand commands.