Struct mz_catalog::durable::persist::PersistHandle

pub(crate) struct PersistHandle<T: TryIntoStateUpdateKind, U: ApplyUpdate<T>> {
    pub(crate) mode: Mode,
    since_handle: SinceHandle<SourceData, (), Timestamp, Diff, i64>,
    write_handle: WriteHandle<SourceData, (), Timestamp, Diff>,
    listen: Listen<SourceData, (), Timestamp, Diff>,
    persist_client: PersistClient,
    shard_id: ShardId,
    pub(crate) snapshot: Vec<(T, Timestamp, Diff)>,
    update_applier: U,
    pub(crate) upper: Timestamp,
    fenceable_token: FenceableToken,
    catalog_content_version: Version,
    metrics: Arc<Metrics>,
}

A handle for interacting with the persist catalog shard.

The catalog shard is used in multiple different contexts, for example pre-open and post-open, but for all contexts the majority of the durable catalog’s behavior is identical. This struct implements those behaviors that are identical while allowing the user to specify the different behaviors via generic parameters.

The behavior of the durable catalog can be different along one of two axes. The first is the format of each individual update, i.e. raw binary, the current protobuf version, previous protobuf versions, etc. The second axis is what to do with each individual update, for example before opening we cache all config updates but don’t cache them after opening. These behaviors are customizable via the T: TryIntoStateUpdateKind and U: ApplyUpdate<T> generic parameters respectively.

Fields

mode: Mode

The Mode that this catalog was opened in.

since_handle: SinceHandle<SourceData, (), Timestamp, Diff, i64>

Since handle to control compaction.

write_handle: WriteHandle<SourceData, (), Timestamp, Diff>

Write handle to persist.

listen: Listen<SourceData, (), Timestamp, Diff>

Listener to catalog changes.

persist_client: PersistClient

Handle for connecting to persist.

shard_id: ShardId

Catalog shard ID.

snapshot: Vec<(T, Timestamp, Diff)>

Cache of the most recent catalog snapshot.

We use a tuple instead of StateUpdate to make consolidation easier.

update_applier: U

Applies custom processing, filtering, and fencing for each individual update.

upper: Timestamp

The current upper of the persist shard.

fenceable_token: FenceableToken

The fence token of the catalog, if one exists.

catalog_content_version: Version

The semantic version of the current binary.

metrics: Arc<Metrics>

Metrics for the persist catalog.

Implementations

impl<T: TryIntoStateUpdateKind, U: ApplyUpdate<T>> PersistHandle<T, U>

async fn increment_catalog_upgrade_shard_version(&self, organization_id: Uuid)

Increment the version in the catalog upgrade shard to the code’s current version.

async fn current_upper(&mut self) -> Timestamp

Fetch the current upper of the catalog state.

pub(crate) async fn compare_and_append<S: IntoStateUpdateKindJson>( &mut self, updates: Vec<(S, Diff)>, ) -> Result<Timestamp, CompareAndAppendError>

Appends updates iff the current global upper of the catalog is self.upper.

Returns the next upper used to commit the transaction.

async fn snapshot_unconsolidated(&mut self) -> Vec<StateUpdate<StateUpdateKind>>

Generates an iterator of StateUpdate that contain all unconsolidated updates to the catalog state up to, and including, as_of.

pub(crate) async fn sync_to_current_upper(&mut self) -> Result<(), FenceError>

Listen and apply all updates that are currently in persist.

Returns an error if this instance has been fenced out.

pub(crate) async fn sync( &mut self, target_upper: Timestamp, ) -> Result<(), FenceError>

Listen and apply all updates up to target_upper.

Returns an error if this instance has been fenced out.

async fn sync_inner( &mut self, target_upper: Timestamp, ) -> Result<(), FenceError>

pub(crate) fn apply_updates( &mut self, updates: impl IntoIterator<Item = StateUpdate<T>>, ) -> Result<(), FenceError>

pub(crate) fn consolidate(&mut self)

async fn with_trace<R>( &mut self, f: impl FnOnce(&Vec<(T, Timestamp, Diff)>) -> Result<R, CatalogError>, ) -> Result<R, CatalogError>

Execute and return the results of f on the current catalog trace.

Will return an error if the catalog has been fenced out.

async fn read_handle(&self) -> ReadHandle<SourceData, (), Timestamp, Diff>

Open a read handle to the catalog.

async fn expire(self: Box<Self>)

Politely releases all external resources that can only be released in an async context.

impl<U: ApplyUpdate<StateUpdateKind>> PersistHandle<StateUpdateKind, U>

async fn with_snapshot<T>( &mut self, f: impl FnOnce(Snapshot) -> Result<T, CatalogError>, ) -> Result<T, CatalogError>

Execute and return the results of f on the current catalog snapshot.

Will return an error if the catalog has been fenced out.

async fn persist_snapshot( &mut self, ) -> impl Iterator<Item = StateUpdate> + DoubleEndedIterator

Generates an iterator of StateUpdate that contain all updates to the catalog state.

The output is fetched directly from persist instead of the in-memory cache.

The output is consolidated and sorted by timestamp in ascending order.

impl PersistHandle<StateUpdateKindJson, UnopenedCatalogStateInner>

pub(crate) async fn new( persist_client: PersistClient, organization_id: Uuid, version: Version, deploy_generation: Option<u64>, metrics: Arc<Metrics>, ) -> Result<PersistHandle<StateUpdateKindJson, UnopenedCatalogStateInner>, DurableCatalogError>

Create a new UnopenedPersistCatalogState to the catalog state associated with organization_id.

All usages of the persist catalog must go through this function. That includes the catalog-debug tool, the adapter’s catalog, etc.

async fn open_inner( self, mode: Mode, initial_ts: EpochMillis, bootstrap_args: &BootstrapArgs, ) -> Result<Box<dyn DurableCatalogState>, CatalogError>

fn is_initialized_inner(&self) -> bool

Reports if the catalog state has been initialized.

NOTE: This is the answer as of the last call to PersistHandle::sync or PersistHandle::sync_to_current_upper, not necessarily what is currently in persist.

async fn get_current_config( &mut self, key: &str, ) -> Result<Option<u64>, DurableCatalogError>

Get the current value of config key.

Some configs need to be read before the catalog is opened for bootstrapping.

pub(crate) async fn get_user_version( &mut self, ) -> Result<Option<u64>, DurableCatalogError>

Get the user version of this instance.

The user version is used to determine if a migration is needed.

async fn get_current_setting( &mut self, name: &str, ) -> Result<Option<String>, DurableCatalogError>

Get the current value of setting name.

Some settings need to be read before the catalog is opened for bootstrapping.

async fn get_catalog_content_version( &mut self, ) -> Result<Option<Version>, DurableCatalogError>

Get the catalog content version.

The catalog content version is the semantic version of the most recent binary that wrote to the catalog.

impl PersistHandle<StateUpdateKindJson, UnopenedCatalogStateInner>

pub(crate) async fn debug_edit<T: Collection>( &mut self, key: T::Key, value: T::Value, ) -> Result<Option<T::Value>, CatalogError>

where T::Key: PartialEq + Eq + Debug + Clone, T::Value: Debug + Clone,

Manually update value of key in collection T to value.

pub(crate) async fn debug_delete<T: Collection>( &mut self, key: T::Key, ) -> Result<(), CatalogError>

where T::Key: PartialEq + Eq + Debug + Clone, T::Value: Debug,

Manually delete key from collection T.

async fn current_snapshot( &mut self, ) -> Result<impl IntoIterator<Item = StateUpdate> + '_, CatalogError>

Generates a Vec<StateUpdate> that contain all updates to the catalog state.

The output is consolidated and sorted by timestamp in ascending order and the current upper.

Trait Implementations

impl<T: Debug + TryIntoStateUpdateKind, U: Debug + ApplyUpdate<T>> Debug for PersistHandle<T, U>

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

Formats the value using the given formatter.

impl DurableCatalogState for PersistHandle<StateUpdateKind, CatalogStateInner>

fn is_read_only(&self) -> bool

Returns true if the catalog is opened in read only mode, false otherwise.

fn is_savepoint(&self) -> bool

Returns true if the catalog is opened is savepoint mode, false otherwise.

fn transaction<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Transaction<'_>, CatalogError>> + Send + 'async_trait>>

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

Creates a new durable catalog state transaction.

fn commit_transaction<'life0, 'async_trait>( &'life0 mut self, txn_batch: TransactionBatch, ) -> Pin<Box<dyn Future<Output = Result<Timestamp, CatalogError>> + Send + 'async_trait>>

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

Commits a durable catalog state transaction.

fn confirm_leadership<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<(), CatalogError>> + Send + 'async_trait>>

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

Confirms that this catalog is connected as the current leader.

fn allocate_id<'life0, 'life1, 'async_trait>( &'life0 mut self, id_type: &'life1 str, amount: u64, ) -> Pin<Box<dyn Future<Output = Result<Vec<u64>, CatalogError>> + Send + 'async_trait>>

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

Allocates and returns amount IDs of id_type.

fn allocate_system_ids<'life0, 'async_trait>( &'life0 mut self, amount: u64, ) -> Pin<Box<dyn Future<Output = Result<Vec<(CatalogItemId, GlobalId)>, CatalogError>> + Send + 'async_trait>>

where Self: Send + 'async_trait, 'life0: 'async_trait,

Allocates and returns amount system CatalogItemIds.

fn allocate_user_id<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<(CatalogItemId, GlobalId), CatalogError>> + Send + 'async_trait>>

where Self: Send + 'async_trait, 'life0: 'async_trait,

Allocates and returns both a user CatalogItemId and GlobalId.

fn allocate_user_cluster_id<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<ClusterId, CatalogError>> + Send + 'async_trait>>

where Self: Send + 'async_trait, 'life0: 'async_trait,

Allocates and returns a user ClusterId.

impl OpenableDurableCatalogState for PersistHandle<StateUpdateKindJson, UnopenedCatalogStateInner>

fn open_savepoint<'life0, 'async_trait>( self: Box<Self>, initial_ts: EpochMillis, bootstrap_args: &'life0 BootstrapArgs, ) -> Pin<Box<dyn Future<Output = Result<Box<dyn DurableCatalogState>, CatalogError>> + Send + 'async_trait>>

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

Opens the catalog in a mode that accepts and buffers all writes, but never durably commits them. This is used to check and see if opening the catalog would be successful, without making any durable changes.

fn open_read_only<'life0, 'async_trait>( self: Box<Self>, bootstrap_args: &'life0 BootstrapArgs, ) -> Pin<Box<dyn Future<Output = Result<Box<dyn DurableCatalogState>, CatalogError>> + Send + 'async_trait>>

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

Opens the catalog in read only mode. All mutating methods will return an error.

fn open<'life0, 'async_trait>( self: Box<Self>, initial_ts: EpochMillis, bootstrap_args: &'life0 BootstrapArgs, ) -> Pin<Box<dyn Future<Output = Result<Box<dyn DurableCatalogState>, CatalogError>> + Send + 'async_trait>>

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

Opens the catalog in a writeable mode. Optionally initializes the catalog, if it has not been initialized, and perform any migrations needed.

fn open_debug<'async_trait>( self: Box<Self>, ) -> Pin<Box<dyn Future<Output = Result<DebugCatalogState, CatalogError>> + Send + 'async_trait>>

where Self: 'async_trait,

Opens the catalog for manual editing of the underlying data. This is helpful for fixing a corrupt catalog.

fn is_initialized<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<bool, CatalogError>> + Send + 'async_trait>>

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

Reports if the catalog state has been initialized.

fn epoch<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Epoch, CatalogError>> + Send + 'async_trait>>

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

Returns the epoch of the current durable catalog state. The epoch acts as a fencing token to prevent split brain issues across two DurableCatalogStates. When a new DurableCatalogState opens the catalog, it will increment the epoch by one (or initialize it to some value if there’s no existing epoch) and store the value in memory. It’s guaranteed that no two DurableCatalogStates will return the same value for their epoch.

fn get_deployment_generation<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<u64, CatalogError>> + Send + 'async_trait>>

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

Get the most recent deployment generation written to the catalog. Not necessarily the deploy generation of this instance.

fn get_enable_0dt_deployment<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Option<bool>, CatalogError>> + Send + 'async_trait>>

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

Get the enable_0dt_deployment config value of this instance.

fn get_0dt_deployment_max_wait<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Option<Duration>, CatalogError>> + Send + 'async_trait>>

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

Get the with_0dt_deployment_max_wait config value of this instance.

fn get_enable_0dt_deployment_panic_after_timeout<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Option<bool>, CatalogError>> + Send + 'async_trait>>

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

Get the enable_0dt_deployment_panic_after_timeout config value of this instance.

fn has_system_config_synced_once<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<bool, DurableCatalogError>> + Send + 'async_trait>>

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

Reports if the remote configuration was synchronized at least once.

fn trace_unconsolidated<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Trace, CatalogError>> + Send + 'async_trait>>

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

Generate an unconsolidated Trace of catalog contents.

fn trace_consolidated<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Trace, CatalogError>> + Send + 'async_trait>>

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

Generate a consolidated Trace of catalog contents.

fn expire<'async_trait>( self: Box<Self>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait,

Politely releases all external resources that can only be released in an async context.

impl ReadOnlyDurableCatalogState for PersistHandle<StateUpdateKind, CatalogStateInner>

fn epoch(&self) -> Epoch

Returns the epoch of the current durable catalog state. The epoch acts as a fencing token to prevent split brain issues across two DurableCatalogStates. When a new DurableCatalogState opens the catalog, it will increment the epoch by one (or initialize it to some value if there’s no existing epoch) and store the value in memory. It’s guaranteed that no two DurableCatalogStates will return the same value for their epoch.

fn expire<'async_trait>( self: Box<Self>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait,

Politely releases all external resources that can only be released in an async context.

fn get_audit_logs<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Vec<VersionedEvent>, CatalogError>> + Send + 'async_trait>>

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

Get all audit log events.

fn get_next_id<'life0, 'life1, 'async_trait>( &'life0 mut self, id_type: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<u64, CatalogError>> + Send + 'async_trait>>

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

Get the next ID of id_type, without allocating it.

fn get_deployment_generation<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<u64, CatalogError>> + Send + 'async_trait>>

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

Get the deployment generation of this instance.

fn snapshot<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Snapshot, CatalogError>> + Send + 'async_trait>>

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

Get a snapshot of the catalog.

fn sync_to_current_updates<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Vec<StateUpdate>, CatalogError>> + Send + 'async_trait>>

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

Listen and return all updates that are currently in the catalog.

fn sync_updates<'life0, 'async_trait>( &'life0 mut self, target_upper: Timestamp, ) -> Pin<Box<dyn Future<Output = Result<Vec<StateUpdate>, CatalogError>> + Send + 'async_trait>>

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

Listen and return all updates in the catalog up to target_upper.

fn get_next_user_item_id<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<u64, CatalogError>> + Send + 'async_trait>>

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

Get the next user ID without allocating it.

fn get_next_system_item_id<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<u64, CatalogError>> + Send + 'async_trait>>

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

Get the next system ID without allocating it.

fn get_next_system_replica_id<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<u64, CatalogError>> + Send + 'async_trait>>

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

Get the next system replica id without allocating it.

fn get_next_user_replica_id<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<u64, CatalogError>> + Send + 'async_trait>>

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

Get the next user replica id without allocating it.