Struct mz_storage_client::storage_collections::StorageCollectionsImpl

pub struct StorageCollectionsImpl<T: TimelyTimestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation> {

    envd_epoch: NonZeroI64,
    read_only: bool,
    finalizable_shards: Arc<ShardIdSet>,
    finalized_shards: Arc<ShardIdSet>,
    collections: Arc<Mutex<BTreeMap<GlobalId, CollectionState<T>>>>,
    txns_read: TxnsRead<T>,
    config: Arc<Mutex<StorageConfiguration>>,
    initial_txn_upper: Antichain<T>,
    persist_location: PersistLocation,
    persist: Arc<PersistClientCache>,
    cmd_tx: UnboundedSender<BackgroundCmd<T>>,
    holds_tx: UnboundedSender<(GlobalId, ChangeBatch<T>)>,
    _background_task: Arc<AbortOnDropHandle<()>>,
    _finalize_shards_task: Arc<AbortOnDropHandle<()>>,
}

Implementation of StorageCollections that is shallow-cloneable and uses a background task for doing work concurrently, in the background.

Fields

envd_epoch: NonZeroI64

The fencing token for this instance of StorageCollections, and really all of the controllers and Coordinator.

read_only: bool

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

When in read-only mode, we are not allowed to affect changes to external systems, including, for example, acquiring and downgrading critical SinceHandles

finalizable_shards: Arc<ShardIdSet>

The set of ShardIds that we have to finalize. These will have been persisted by the caller of StorageCollections::prepare_state.

finalized_shards: Arc<ShardIdSet>

The set of ShardIds that we have finalized. We keep track of shards here until we are given a chance to let our callers know that these have been finalized, for example via StorageCollections::prepare_state.

collections: Arc<Mutex<BTreeMap<GlobalId, CollectionState<T>>>>

Collections maintained by this StorageCollections.

txns_read: TxnsRead<T>

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

config: Arc<Mutex<StorageConfiguration>>

Storage configuration parameters.

initial_txn_upper: Antichain<T>

The upper of the txn shard as it was when we booted. We forward the upper of created/registered tables to make sure that their uppers are at least not less than the initially known txn upper.

NOTE: This works around a quirk in how the adapter chooses the as_of of existing indexes when bootstrapping, where tables that have an upper that is less than the initially known txn upper can lead to indexes that cannot hydrate in read-only mode.

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

cmd_tx: UnboundedSender<BackgroundCmd<T>>

For sending commands to our internal task.

holds_tx: UnboundedSender<(GlobalId, ChangeBatch<T>)>

For sending updates about read holds to our internal task.

_background_task: Arc<AbortOnDropHandle<()>>

Handles to tasks we own, making sure they’re dropped when we are.

_finalize_shards_task: Arc<AbortOnDropHandle<()>>

Implementations

impl<T> StorageCollectionsImpl<T>

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

pub async fn new( persist_location: PersistLocation, persist_clients: Arc<PersistClientCache>, metrics_registry: &MetricsRegistry, _now: NowFn, txns_metrics: Arc<TxnMetrics>, envd_epoch: NonZeroI64, read_only: bool, connection_context: ConnectionContext, txn: &dyn StorageTxn<T>, ) -> Self

Creates and returns a new StorageCollections.

Note that when creating a new StorageCollections, you must also reconcile it with the previous state using StorageCollections::initialize_state, StorageCollections::prepare_state, and StorageCollections::create_collections.

async fn open_data_handles( &self, id: &GlobalId, shard: ShardId, since: Option<&Antichain<T>>, relation_desc: RelationDesc, persist_client: &PersistClient, ) -> (WriteHandle<SourceData, (), T, Diff>, SinceHandleWrapper<T>)

Opens a WriteHandle and a SinceHandleWrapper, for holding back the since.

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.

async fn open_write_handle( &self, id: &GlobalId, shard: ShardId, relation_desc: RelationDesc, persist_client: &PersistClient, ) -> WriteHandle<SourceData, (), T, Diff>

Opens a write handle for the given shard.

async fn open_critical_handle( &self, id: &GlobalId, shard: ShardId, since: Option<&Antichain<T>>, persist_client: &PersistClient, ) -> SinceHandle<SourceData, (), T, Diff, PersistEpoch>

Opens a critical since handle 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.

async fn open_leased_handle( &self, id: &GlobalId, shard: ShardId, relation_desc: RelationDesc, since: Option<&Antichain<T>>, persist_client: &PersistClient, ) -> ReadHandle<SourceData, (), T, Diff>

Opens a leased ReadHandle, for the purpose of holding back a since, 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.

fn register_handles( &self, id: GlobalId, is_in_txns: bool, since_handle: SinceHandleWrapper<T>, write_handle: WriteHandle<SourceData, (), T, Diff>, )

fn send(&self, cmd: BackgroundCmd<T>)

async fn snapshot_stats_inner( &self, id: GlobalId, as_of: SnapshotStatsAsOf<T>, ) -> Result<SnapshotStats, StorageError<T>>

fn install_collection_dependency_read_holds_inner( &self, self_collections: &mut BTreeMap<GlobalId, CollectionState<T>>, id: GlobalId, ) -> Result<(), StorageError<T>>

If this identified collection has a dependency, install a read hold on it.

This is necessary to ensure that the dependency’s since does not advance beyond its dependents’.

fn determine_collection_dependencies( &self, self_collections: &BTreeMap<GlobalId, CollectionState<T>>, data_source: &DataSource, ) -> Result<Vec<GlobalId>, StorageError<T>>

Determine if this collection has another dependency.

Currently, collections have either 0 or 1 dependencies.

fn install_read_capabilities_inner( &self, self_collections: &mut BTreeMap<GlobalId, CollectionState<T>>, from_id: GlobalId, storage_dependencies: &[GlobalId], read_capability: Antichain<T>, ) -> Result<(), StorageError<T>>

Install read capabilities on the given storage_dependencies.

async fn read_handle_for_snapshot( &self, metadata: &CollectionMetadata, id: GlobalId, ) -> Result<ReadHandle<SourceData, (), T, Diff>, StorageError<T>>

fn set_read_policies_inner( &self, collections: &mut BTreeMap<GlobalId, CollectionState<T>>, policies: Vec<(GlobalId, ReadPolicy<T>)>, )

fn update_read_capabilities_inner( cmd_tx: &UnboundedSender<BackgroundCmd<T>>, collections: &mut BTreeMap<GlobalId, CollectionState<T>>, updates: &mut BTreeMap<GlobalId, ChangeBatch<T>>, )

fn synchronize_finalized_shards(&self, storage_metadata: &StorageMetadata)

Remove any shards that we know are finalized

Trait Implementations

impl<T: Clone + TimelyTimestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation> Clone for StorageCollectionsImpl<T>

fn clone(&self) -> StorageCollectionsImpl<T>

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T: Debug + TimelyTimestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation> Debug for StorageCollectionsImpl<T>

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

Formats the value using the given formatter.

impl<T> StorageCollections for StorageCollectionsImpl<T>

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

type Timestamp = T

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

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

On boot, reconcile this StorageCollections with outside state. We get a StorageTxn where we can record any durable state that we need.

fn update_parameters(&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 active_collection_metadatas(&self) -> Vec<(GlobalId, CollectionMetadata)>

Acquire an iterator over CollectionMetadata for all active collections.

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

Atomically gets and returns the frontiers of all the identified collections.

fn active_collection_frontiers( &self, ) -> Vec<CollectionFrontiers<Self::Timestamp>>

Atomically gets and returns the frontiers of all active collections.

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>>> + Send + '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>>>> + Send + '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 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 prepare_state<'life0, 'life1, 'async_trait>( &'life0 self, txn: &'life1 mut (dyn StorageTxn<Self::Timestamp> + Send), ids_to_add: BTreeSet<GlobalId>, ids_to_drop: BTreeSet<GlobalId>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<T>>> + Send + 'async_trait>>

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

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

fn create_collections_for_bootstrap<'life0, 'life1, 'life2, 'async_trait>( &'life0 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>>> + Send + 'async_trait>>

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

Like Self::create_collections, except used specifically for bootstrap.

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

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

Alters the identified ingestion to use the provided SourceDesc.

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

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

Alters each identified collection to use the correlated GenericSourceConnection.

fn alter_table_desc<'life0, 'async_trait>( &'life0 self, table_id: GlobalId, new_desc: RelationDesc, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + Send + 'async_trait>>

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

Updates the RelationDesc for the specified table.

fn drop_collections_unvalidated( &self, storage_metadata: &StorageMetadata, identifiers: Vec<GlobalId>, )

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

fn set_read_policies( &self, policies: Vec<(GlobalId, ReadPolicy<Self::Timestamp>)>, )

Assigns a read policy to specific identifiers.

fn acquire_read_holds( &self, desired_holds: Vec<GlobalId>, ) -> Result<Vec<ReadHold<Self::Timestamp>>, ReadHoldError>

Acquires and returns the earliest possible read holds for the specified collections.

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

Returns the frontiers of the identified collection.

fn create_collections<'life0, 'life1, 'async_trait>( &'life0 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>>> + Send + 'async_trait>>

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

Create the collections described by the individual CollectionDescriptions.