Struct mz_catalog::durable::Transaction

pub struct Transaction<'a> {

    durable_catalog: &'a mut dyn DurableCatalogState,
    databases: TableTransaction<DatabaseKey, DatabaseValue>,
    schemas: TableTransaction<SchemaKey, SchemaValue>,
    items: TableTransaction<ItemKey, ItemValue>,
    comments: TableTransaction<CommentKey, CommentValue>,
    roles: TableTransaction<RoleKey, RoleValue>,
    clusters: TableTransaction<ClusterKey, ClusterValue>,
    cluster_replicas: TableTransaction<ClusterReplicaKey, ClusterReplicaValue>,
    introspection_sources: TableTransaction<ClusterIntrospectionSourceIndexKey, ClusterIntrospectionSourceIndexValue>,
    id_allocator: TableTransaction<IdAllocKey, IdAllocValue>,
    configs: TableTransaction<ConfigKey, ConfigValue>,
    settings: TableTransaction<SettingKey, SettingValue>,
    system_gid_mapping: TableTransaction<GidMappingKey, GidMappingValue>,
    system_configurations: TableTransaction<ServerConfigurationKey, ServerConfigurationValue>,
    default_privileges: TableTransaction<DefaultPrivilegesKey, DefaultPrivilegesValue>,
    source_references: TableTransaction<SourceReferencesKey, SourceReferencesValue>,
    system_privileges: TableTransaction<SystemPrivilegesKey, SystemPrivilegesValue>,
    network_policies: TableTransaction<NetworkPolicyKey, NetworkPolicyValue>,
    storage_collection_metadata: TableTransaction<StorageCollectionMetadataKey, StorageCollectionMetadataValue>,
    unfinalized_shards: TableTransaction<UnfinalizedShardKey, ()>,
    txn_wal_shard: TableTransaction<(), TxnWalShardValue>,
    audit_log_updates: Vec<(AuditLogKey, Diff, u64)>,
    upper: Timestamp,
    op_id: u64,
}

A Transaction batches multiple catalog operations together and commits them atomically. An operation also logically groups multiple catalog updates together.

Fields

durable_catalog: &'a mut dyn DurableCatalogState

databases: TableTransaction<DatabaseKey, DatabaseValue>

schemas: TableTransaction<SchemaKey, SchemaValue>

items: TableTransaction<ItemKey, ItemValue>

comments: TableTransaction<CommentKey, CommentValue>

roles: TableTransaction<RoleKey, RoleValue>

clusters: TableTransaction<ClusterKey, ClusterValue>

cluster_replicas: TableTransaction<ClusterReplicaKey, ClusterReplicaValue>

introspection_sources: TableTransaction<ClusterIntrospectionSourceIndexKey, ClusterIntrospectionSourceIndexValue>

id_allocator: TableTransaction<IdAllocKey, IdAllocValue>

configs: TableTransaction<ConfigKey, ConfigValue>

settings: TableTransaction<SettingKey, SettingValue>

system_gid_mapping: TableTransaction<GidMappingKey, GidMappingValue>

system_configurations: TableTransaction<ServerConfigurationKey, ServerConfigurationValue>

default_privileges: TableTransaction<DefaultPrivilegesKey, DefaultPrivilegesValue>

source_references: TableTransaction<SourceReferencesKey, SourceReferencesValue>

system_privileges: TableTransaction<SystemPrivilegesKey, SystemPrivilegesValue>

network_policies: TableTransaction<NetworkPolicyKey, NetworkPolicyValue>

storage_collection_metadata: TableTransaction<StorageCollectionMetadataKey, StorageCollectionMetadataValue>

unfinalized_shards: TableTransaction<UnfinalizedShardKey, ()>

txn_wal_shard: TableTransaction<(), TxnWalShardValue>

audit_log_updates: Vec<(AuditLogKey, Diff, u64)>

upper: Timestamp

The upper of durable_catalog at the start of the transaction.

op_id: u64

The ID of the current operation of this transaction.

Implementations

impl<'a> Transaction<'a>

pub fn new( durable_catalog: &'a mut dyn DurableCatalogState, _: Snapshot, upper: Timestamp, ) -> Result<Transaction<'_>, CatalogError>

pub fn get_item(&self, id: &CatalogItemId) -> Option<Item>

pub fn get_items(&self) -> impl Iterator<Item = Item>

pub fn insert_audit_log_event(&mut self, event: VersionedEvent)

pub fn insert_audit_log_events( &mut self, events: impl IntoIterator<Item = VersionedEvent>, )

pub fn insert_user_database( &mut self, database_name: &str, owner_id: RoleId, privileges: Vec<MzAclItem>, temporary_oids: &HashSet<u32>, ) -> Result<(DatabaseId, u32), CatalogError>

pub(crate) fn insert_database( &mut self, id: DatabaseId, database_name: &str, owner_id: RoleId, privileges: Vec<MzAclItem>, oid: u32, ) -> Result<u32, CatalogError>

pub fn insert_user_schema( &mut self, database_id: DatabaseId, schema_name: &str, owner_id: RoleId, privileges: Vec<MzAclItem>, temporary_oids: &HashSet<u32>, ) -> Result<(SchemaId, u32), CatalogError>

pub fn insert_system_schema( &mut self, schema_id: u64, schema_name: &str, owner_id: RoleId, privileges: Vec<MzAclItem>, oid: u32, ) -> Result<(), CatalogError>

pub(crate) fn insert_schema( &mut self, schema_id: SchemaId, database_id: Option<DatabaseId>, schema_name: String, owner_id: RoleId, privileges: Vec<MzAclItem>, oid: u32, ) -> Result<(), CatalogError>

pub fn insert_builtin_role( &mut self, id: RoleId, name: String, attributes: RoleAttributes, membership: RoleMembership, vars: RoleVars, oid: u32, ) -> Result<RoleId, CatalogError>

pub fn insert_user_role( &mut self, name: String, attributes: RoleAttributes, membership: RoleMembership, vars: RoleVars, temporary_oids: &HashSet<u32>, ) -> Result<(RoleId, u32), CatalogError>

fn insert_role( &mut self, id: RoleId, name: String, attributes: RoleAttributes, membership: RoleMembership, vars: RoleVars, oid: u32, ) -> Result<(), CatalogError>

pub fn insert_user_cluster( &mut self, cluster_id: ClusterId, cluster_name: &str, introspection_source_indexes: Vec<(&'static BuiltinLog, CatalogItemId, GlobalId)>, owner_id: RoleId, privileges: Vec<MzAclItem>, config: ClusterConfig, temporary_oids: &HashSet<u32>, ) -> Result<(), CatalogError>

Panics if any introspection source id is not a system id

pub fn insert_system_cluster( &mut self, cluster_name: &str, introspection_source_indexes: Vec<(&'static BuiltinLog, CatalogItemId, GlobalId)>, privileges: Vec<MzAclItem>, owner_id: RoleId, config: ClusterConfig, temporary_oids: &HashSet<u32>, ) -> Result<(), CatalogError>

Panics if any introspection source id is not a system id

fn insert_cluster( &mut self, cluster_id: ClusterId, cluster_name: &str, introspection_source_indexes: Vec<(&'static BuiltinLog, CatalogItemId, GlobalId)>, owner_id: RoleId, privileges: Vec<MzAclItem>, config: ClusterConfig, temporary_oids: &HashSet<u32>, ) -> Result<(), CatalogError>

pub fn rename_cluster( &mut self, cluster_id: ClusterId, cluster_name: &str, cluster_to_name: &str, ) -> Result<(), CatalogError>

pub fn rename_cluster_replica( &mut self, replica_id: ReplicaId, replica_name: &QualifiedReplica, replica_to_name: &str, ) -> Result<(), CatalogError>

pub fn insert_cluster_replica( &mut self, cluster_id: ClusterId, replica_name: &str, config: ReplicaConfig, owner_id: RoleId, ) -> Result<ReplicaId, CatalogError>

pub(crate) fn insert_cluster_replica_with_id( &mut self, cluster_id: ClusterId, replica_id: ReplicaId, replica_name: &str, config: ReplicaConfig, owner_id: RoleId, ) -> Result<(), CatalogError>

pub fn insert_user_network_policy( &mut self, name: String, rules: Vec<NetworkPolicyRule>, privileges: Vec<MzAclItem>, owner_id: RoleId, temporary_oids: &HashSet<u32>, ) -> Result<NetworkPolicyId, CatalogError>

pub fn insert_network_policy( &mut self, id: NetworkPolicyId, name: String, rules: Vec<NetworkPolicyRule>, privileges: Vec<MzAclItem>, owner_id: RoleId, oid: u32, ) -> Result<NetworkPolicyId, CatalogError>

pub fn update_introspection_source_index_gids( &mut self, mappings: impl Iterator<Item = (ClusterId, impl Iterator<Item = (String, CatalogItemId, GlobalId, u32)>)>, ) -> Result<(), CatalogError>

Updates persisted information about persisted introspection source indexes.

Panics if provided id is not a system id.

pub fn insert_user_item( &mut self, id: CatalogItemId, global_id: GlobalId, schema_id: SchemaId, item_name: &str, create_sql: String, owner_id: RoleId, privileges: Vec<MzAclItem>, temporary_oids: &HashSet<u32>, versions: BTreeMap<RelationVersion, GlobalId>, ) -> Result<u32, CatalogError>

pub fn insert_item( &mut self, id: CatalogItemId, oid: u32, global_id: GlobalId, schema_id: SchemaId, item_name: &str, create_sql: String, owner_id: RoleId, privileges: Vec<MzAclItem>, extra_versions: BTreeMap<RelationVersion, GlobalId>, ) -> Result<(), CatalogError>

pub fn get_and_increment_id(&mut self, key: String) -> Result<u64, CatalogError>

pub fn get_and_increment_id_by( &mut self, key: String, amount: u64, ) -> Result<Vec<u64>, CatalogError>

pub fn allocate_system_item_ids( &mut self, amount: u64, ) -> Result<Vec<(CatalogItemId, GlobalId)>, CatalogError>

pub fn allocate_introspection_source_index_id( cluster_id: &ClusterId, log_variant: LogVariant, ) -> (CatalogItemId, GlobalId)

Allocates an ID for an introspection source index. These IDs are deterministically derived from the cluster_id and log_variant.

Introspection source indexes are a special edge case of items. They are considered system items, but they are the only system item that can be created by the user at any time. All other system items can only be created by the system during the startup of an upgrade.

Furthermore, all other system item IDs are allocated deterministically in the same order during startup. Therefore, all read-only environmentd processes during an upgrade will allocate the same system IDs to the same items, and due to the way catalog fencing works, only one of them can successfully write the IDs down to the catalog. This removes the need for environmentd processes to coordinate system IDs allocated during read-only mode.

Since introspection IDs can be allocated at any time, read-only instances would either need to coordinate across processes when allocating a new ID or allocate them deterministically. We opted to allocate the IDs deterministically to avoid the overhead of coordination.

Introspection source index IDs are 64 bit integers, with the following format (not to scale):

---

Cluster ID Variant	Cluster ID Inner Value	Log Variant
8-bits	48-bits	8-bits

---

Cluster ID Variant: A unique number indicating the variant of cluster the index belongs to. Cluster ID Inner Value: A per variant unique number indicating the cluster the index belongs to. Log Variant: A unique number indicating the log variant this index is on.

pub fn allocate_user_item_ids( &mut self, amount: u64, ) -> Result<Vec<(CatalogItemId, GlobalId)>, CatalogError>

pub fn allocate_user_replica_id(&mut self) -> Result<ReplicaId, CatalogError>

pub fn allocate_system_replica_id(&mut self) -> Result<ReplicaId, CatalogError>

pub fn allocate_audit_log_id(&mut self) -> Result<u64, CatalogError>

pub fn allocate_storage_usage_ids(&mut self) -> Result<u64, CatalogError>

fn allocate_oids( &mut self, amount: u64, temporary_oids: &HashSet<u32>, ) -> Result<Vec<u32>, CatalogError>

Allocates amount OIDs. OIDs can be recycled if they aren’t currently assigned to any object.

pub fn allocate_oid( &mut self, temporary_oids: &HashSet<u32>, ) -> Result<u32, CatalogError>

Allocates a single OID. OIDs can be recycled if they aren’t currently assigned to any object.

pub(crate) fn insert_id_allocator( &mut self, name: String, next_id: u64, ) -> Result<(), CatalogError>

pub fn remove_database(&mut self, id: &DatabaseId) -> Result<(), CatalogError>

Removes the database id from the transaction.

Returns an error if id is not found.

Runtime is linear with respect to the total number of databases in the catalog. DO NOT call this function in a loop, use Self::remove_databases instead.

pub fn remove_databases( &mut self, databases: &BTreeSet<DatabaseId>, ) -> Result<(), CatalogError>

Removes all databases in databases from the transaction.

Returns an error if any id in databases is not found.

NOTE: On error, there still may be some databases removed from the transaction. It is up to the caller to either abort the transaction or commit.

pub fn remove_schema( &mut self, database_id: &Option<DatabaseId>, schema_id: &SchemaId, ) -> Result<(), CatalogError>

Removes the schema identified by database_id and schema_id from the transaction.

Returns an error if (database_id, schema_id) is not found.

Runtime is linear with respect to the total number of schemas in the catalog. DO NOT call this function in a loop, use Self::remove_schemas instead.

pub fn remove_schemas( &mut self, schemas: &BTreeMap<SchemaId, ResolvedDatabaseSpecifier>, ) -> Result<(), CatalogError>

Removes all schemas in schemas from the transaction.

Returns an error if any id in schemas is not found.

NOTE: On error, there still may be some schemas removed from the transaction. It is up to the caller to either abort the transaction or commit.

pub fn remove_source_references( &mut self, source_id: CatalogItemId, ) -> Result<(), CatalogError>

pub fn remove_role(&mut self, name: &str) -> Result<(), CatalogError>

Removes the role name from the transaction.

Returns an error if name is not found.

Runtime is linear with respect to the total number of roles in the catalog. DO NOT call this function in a loop, use Self::remove_roles instead.

pub fn remove_roles( &mut self, roles: &BTreeSet<RoleId>, ) -> Result<(), CatalogError>

Removes all roles in roles from the transaction.

Returns an error if any id in roles is not found.

NOTE: On error, there still may be some roles removed from the transaction. It is up to the caller to either abort the transaction or commit.

pub fn remove_clusters( &mut self, clusters: &BTreeSet<ClusterId>, ) -> Result<(), CatalogError>

Removes all cluster in clusters from the transaction.

Returns an error if any id in clusters is not found.

NOTE: On error, there still may be some clusters removed from the transaction. It is up to the caller to either abort the transaction or commit.

pub fn remove_cluster_replica( &mut self, id: ReplicaId, ) -> Result<(), CatalogError>

Removes the cluster replica id from the transaction.

Returns an error if id is not found.

Runtime is linear with respect to the total number of cluster replicas in the catalog. DO NOT call this function in a loop, use Self::remove_cluster_replicas instead.

pub fn remove_cluster_replicas( &mut self, replicas: &BTreeSet<ReplicaId>, ) -> Result<(), CatalogError>

Removes all cluster replicas in replicas from the transaction.

Returns an error if any id in replicas is not found.

NOTE: On error, there still may be some cluster replicas removed from the transaction. It is up to the caller to either abort the transaction or commit.

pub fn remove_item(&mut self, id: CatalogItemId) -> Result<(), CatalogError>

Removes item id from the transaction.

Returns an error if id is not found.

Runtime is linear with respect to the total number of items in the catalog. DO NOT call this function in a loop, use Self::remove_items instead.

pub fn remove_items( &mut self, ids: &BTreeSet<CatalogItemId>, ) -> Result<(), CatalogError>

Removes all items in ids from the transaction.

Returns an error if any id in ids is not found.

NOTE: On error, there still may be some items removed from the transaction. It is up to the caller to either abort the transaction or commit.

pub fn remove_system_object_mappings( &mut self, descriptions: BTreeSet<SystemObjectDescription>, ) -> Result<(), CatalogError>

Removes all system object mappings in descriptions from the transaction.

Returns an error if any description in descriptions is not found.

NOTE: On error, there still may be some items removed from the transaction. It is up to the caller to either abort the transaction or commit.

pub fn remove_introspection_source_indexes( &mut self, introspection_source_indexes: BTreeSet<(ClusterId, String)>, ) -> Result<(), CatalogError>

Removes all introspection source indexes in indexes from the transaction.

Returns an error if any index in indexes is not found.

NOTE: On error, there still may be some indexes removed from the transaction. It is up to the caller to either abort the transaction or commit.

pub fn update_item( &mut self, id: CatalogItemId, item: Item, ) -> Result<(), CatalogError>

Updates item id in the transaction to item_name and item.

Returns an error if id is not found.

Runtime is linear with respect to the total number of items in the catalog. DO NOT call this function in a loop, use Self::update_items instead.

pub fn update_items( &mut self, items: BTreeMap<CatalogItemId, Item>, ) -> Result<(), CatalogError>

Updates all items with ids matching the keys of items in the transaction, to the corresponding value in items.

Returns an error if any id in items is not found.

NOTE: On error, there still may be some items updated in the transaction. It is up to the caller to either abort the transaction or commit.

pub fn update_role( &mut self, id: RoleId, role: Role, ) -> Result<(), CatalogError>

Updates role id in the transaction to role.

Returns an error if id is not found.

Runtime is linear with respect to the total number of items in the catalog. DO NOT call this function in a loop, implement and use some Self::update_roles instead. You should model it after Self::update_items.

pub fn update_roles( &mut self, roles: BTreeMap<RoleId, Role>, ) -> Result<(), CatalogError>

Updates all Roles with ids matching the keys of roles in the transaction, to the corresponding value in roles.

Returns an error if any id in roles is not found.

NOTE: On error, there still may be some roles updated in the transaction. It is up to the caller to either abort the transaction or commit.

pub fn update_system_object_mappings( &mut self, mappings: BTreeMap<CatalogItemId, SystemObjectMapping>, ) -> Result<(), CatalogError>

Updates persisted mapping from system objects to global IDs and fingerprints. Each element of mappings should be (old-global-id, new-system-object-mapping).

Panics if provided id is not a system id.

pub fn update_cluster( &mut self, id: ClusterId, cluster: Cluster, ) -> Result<(), CatalogError>

Updates cluster id in the transaction to cluster.

Returns an error if id is not found.

Runtime is linear with respect to the total number of clusters in the catalog. DO NOT call this function in a loop.

pub fn update_cluster_replica( &mut self, replica_id: ReplicaId, replica: ClusterReplica, ) -> Result<(), CatalogError>

Updates cluster replica replica_id in the transaction to replica.

Returns an error if replica_id is not found.

Runtime is linear with respect to the total number of cluster replicas in the catalog. DO NOT call this function in a loop.

pub fn update_database( &mut self, id: DatabaseId, database: Database, ) -> Result<(), CatalogError>

Updates database id in the transaction to database.

Returns an error if id is not found.

Runtime is linear with respect to the total number of databases in the catalog. DO NOT call this function in a loop.

pub fn update_schema( &mut self, schema_id: SchemaId, schema: Schema, ) -> Result<(), CatalogError>

Updates schema schema_id in the transaction to schema.

Returns an error if schema_id is not found.

Runtime is linear with respect to the total number of schemas in the catalog. DO NOT call this function in a loop.

pub fn update_network_policy( &mut self, id: NetworkPolicyId, network_policy: NetworkPolicy, ) -> Result<(), CatalogError>

Updates network_policy_id in the transaction to network policy.

Returns an error if id is not found.

Runtime is linear with respect to the total number of databases in the catalog. DO NOT call this function in a loop.

pub fn remove_network_policies( &mut self, network_policies: &BTreeSet<NetworkPolicyId>, ) -> Result<(), CatalogError>

Removes all network policies in network policies from the transaction.

Returns an error if any id in network policy is not found.

NOTE: On error, there still may be some roles removed from the transaction. It is up to the caller to either abort the transaction or commit.

pub fn set_default_privilege( &mut self, role_id: RoleId, database_id: Option<DatabaseId>, schema_id: Option<SchemaId>, object_type: ObjectType, grantee: RoleId, privileges: Option<AclMode>, ) -> Result<(), CatalogError>

Set persisted default privilege.

DO NOT call this function in a loop, use Self::set_default_privileges instead.

pub fn set_default_privileges( &mut self, default_privileges: Vec<DefaultPrivilege>, ) -> Result<(), CatalogError>

Set persisted default privileges.

pub fn set_system_privilege( &mut self, grantee: RoleId, grantor: RoleId, acl_mode: Option<AclMode>, ) -> Result<(), CatalogError>

Set persisted system privilege.

DO NOT call this function in a loop, use Self::set_system_privileges instead.

pub fn set_system_privileges( &mut self, system_privileges: Vec<MzAclItem>, ) -> Result<(), CatalogError>

Set persisted system privileges.

pub fn set_setting( &mut self, name: String, value: Option<String>, ) -> Result<(), CatalogError>

Set persisted setting.

pub fn set_catalog_content_version( &mut self, version: String, ) -> Result<(), CatalogError>

pub fn insert_introspection_source_indexes( &mut self, introspection_source_indexes: Vec<(ClusterId, String, CatalogItemId, GlobalId)>, temporary_oids: &HashSet<u32>, ) -> Result<(), CatalogError>

Insert persisted introspection source index.

pub fn set_system_object_mappings( &mut self, mappings: Vec<SystemObjectMapping>, ) -> Result<(), CatalogError>

Set persisted system object mappings.

pub fn set_replicas( &mut self, replicas: Vec<ClusterReplica>, ) -> Result<(), CatalogError>

Set persisted replica.

pub fn set_config( &mut self, key: String, value: Option<u64>, ) -> Result<(), CatalogError>

Set persisted configuration.

pub fn get_config(&self, key: String) -> Option<u64>

Get the value of a persisted config.

fn get_setting(&self, name: String) -> Option<String>

Get the value of a persisted setting.

pub fn get_builtin_migration_shard(&self) -> Option<ShardId>

pub fn get_expression_cache_shard(&self) -> Option<ShardId>

pub fn set_enable_0dt_deployment( &mut self, value: bool, ) -> Result<(), CatalogError>

Updates the catalog enable_0dt_deployment “config” value to match the enable_0dt_deployment “system var” value.

These are mirrored so that we can toggle the flag with Launch Darkly, but use it in boot before Launch Darkly is available.

pub fn set_0dt_deployment_max_wait( &mut self, value: Duration, ) -> Result<(), CatalogError>

Updates the catalog with_0dt_deployment_max_wait “config” value to match the with_0dt_deployment_max_wait “system var” value.

These are mirrored so that we can toggle the flag with Launch Darkly, but use it in boot before Launch Darkly is available.

pub fn set_enable_0dt_deployment_panic_after_timeout( &mut self, value: bool, ) -> Result<(), CatalogError>

Updates the catalog 0dt_deployment_panic_after_timeout “config” value to match the 0dt_deployment_panic_after_timeout “system var” value.

These are mirrored so that we can toggle the flag with Launch Darkly, but use it in boot before Launch Darkly is available.

pub fn reset_enable_0dt_deployment(&mut self) -> Result<(), CatalogError>

Removes the catalog enable_0dt_deployment “config” value to match the enable_0dt_deployment “system var” value.

These are mirrored so that we can toggle the flag with LaunchDarkly, but use it in boot before LaunchDarkly is available.

pub fn reset_0dt_deployment_max_wait(&mut self) -> Result<(), CatalogError>

Removes the catalog with_0dt_deployment_max_wait “config” value to match the with_0dt_deployment_max_wait “system var” value.

These are mirrored so that we can toggle the flag with LaunchDarkly, but use it in boot before LaunchDarkly is available.

pub fn reset_enable_0dt_deployment_panic_after_timeout( &mut self, ) -> Result<(), CatalogError>

Removes the catalog enable_0dt_deployment_panic_after_timeout “config” value to match the enable_0dt_deployment_panic_after_timeout “system var” value.

These are mirrored so that we can toggle the flag with LaunchDarkly, but use it in boot before LaunchDarkly is available.

pub fn set_system_config_synced_once(&mut self) -> Result<(), CatalogError>

Updates the catalog system_config_synced “config” value to true.

pub fn update_comment( &mut self, object_id: CommentObjectId, sub_component: Option<usize>, comment: Option<String>, ) -> Result<(), CatalogError>

pub fn drop_comments( &mut self, object_ids: &BTreeSet<CommentObjectId>, ) -> Result<(), CatalogError>

pub fn update_source_references( &mut self, source_id: CatalogItemId, references: Vec<SourceReference>, updated_at: u64, ) -> Result<(), CatalogError>

pub fn upsert_system_config( &mut self, name: &str, value: String, ) -> Result<(), CatalogError>

Upserts persisted system configuration name to value.

pub fn remove_system_config(&mut self, name: &str)

Removes persisted system configuration name.

pub fn clear_system_configs(&mut self)

Removes all persisted system configurations.

pub(crate) fn insert_config( &mut self, key: String, value: u64, ) -> Result<(), CatalogError>

pub fn get_clusters(&self) -> impl Iterator<Item = Cluster>

pub fn get_cluster_replicas(&self) -> impl Iterator<Item = ClusterReplica>

pub fn get_roles(&self) -> impl Iterator<Item = Role>

pub fn get_network_policies(&self) -> impl Iterator<Item = NetworkPolicy>

pub fn get_system_object_mappings( &self, ) -> impl Iterator<Item = SystemObjectMapping>

pub fn get_schemas(&self) -> impl Iterator<Item = Schema>

pub fn get_system_configurations( &self, ) -> impl Iterator<Item = SystemConfiguration>

pub fn get_introspection_source_indexes( &self, cluster_id: ClusterId, ) -> BTreeMap<String, (GlobalId, u32)>

pub fn get_catalog_content_version(&self) -> Option<String>

pub fn get_and_commit_op_updates(&mut self) -> Vec<StateUpdate>

Commit the current operation within the transaction. This does not cause anything to be written durably, but signals to the current transaction that we are moving on to the next operation.

Returns the updates of the committed operation.

fn get_op_updates(&self) -> Vec<StateUpdate>

pub fn is_savepoint(&self) -> bool

fn commit_op(&mut self)

pub fn op_id(&self) -> u64

pub fn upper(&self) -> Timestamp

pub(crate) fn into_parts( self, ) -> (TransactionBatch, &'a mut dyn DurableCatalogState)

pub(crate) async fn commit_internal( self, commit_ts: Timestamp, ) -> Result<(&'a mut dyn DurableCatalogState, Timestamp), CatalogError>

Commits the storage transaction to durable storage. Any error returned outside read-only mode indicates the catalog may be in an indeterminate state and needs to be fully re-read before proceeding. In general, this must be fatal to the calling process. We do not panic/halt inside this function itself so that errors can bubble up during initialization.

The transaction is committed at commit_ts.

Returns what the upper was directly after the transaction committed.

In read-only mode, this will return an error for non-empty transactions indicating that the catalog is not writeable.

pub async fn commit(self, commit_ts: Timestamp) -> Result<(), CatalogError>

Commits the storage transaction to durable storage. Any error returned outside read-only mode indicates the catalog may be in an indeterminate state and needs to be fully re-read before proceeding. In general, this must be fatal to the calling process. We do not panic/halt inside this function itself so that errors can bubble up during initialization.

In read-only mode, this will return an error for non-empty transactions indicating that the catalog is not writeable.

IMPORTANT: It is assumed that the committer of this transaction has already applied all updates from this transaction. Therefore, updates from this transaction will not be returned when calling crate::durable::ReadOnlyDurableCatalogState::sync_to_current_updates or crate::durable::ReadOnlyDurableCatalogState::sync_updates.

An alternative implementation would be for the caller to explicitly consume their updates after committing and only then apply the updates in-memory. While this removes assumptions about the caller in this method, in practice it results in duplicate work on every commit.

Trait Implementations

impl<'a> Debug for Transaction<'a>

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

Formats the value using the given formatter.

impl<'a> PartialEq for Transaction<'a>

fn eq(&self, other: &Self) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl StorageTxn<Timestamp> for Transaction<'_>

fn get_collection_metadata(&self) -> BTreeMap<GlobalId, ShardId>

Retrieve all of the visible storage metadata.

fn insert_collection_metadata( &mut self, metadata: BTreeMap<GlobalId, ShardId>, ) -> Result<(), StorageError<Timestamp>>

Add new storage metadata for a collection.

fn delete_collection_metadata( &mut self, ids: BTreeSet<GlobalId>, ) -> Vec<(GlobalId, ShardId)>

Remove the metadata associated with the identified collections.

fn get_unfinalized_shards(&self) -> BTreeSet<ShardId>

Retrieve all of the shards that are no longer in use by an active collection but are yet to be finalized.

fn insert_unfinalized_shards( &mut self, s: BTreeSet<ShardId>, ) -> Result<(), StorageError<Timestamp>>

Insert the specified values as unfinalized shards.

fn mark_shards_as_finalized(&mut self, shards: BTreeSet<ShardId>)

Mark the specified shards as finalized, deleting them from the unfinalized shard collection.

fn get_txn_wal_shard(&self) -> Option<ShardId>

Get the txn WAL shard for this environment if it exists.

fn write_txn_wal_shard( &mut self, shard: ShardId, ) -> Result<(), StorageError<Timestamp>>

Store the specified shard as the environment’s txn WAL shard.