Struct mz_catalog::durable::Transaction

pub struct Transaction<'a> { /* private fields */ }

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

Implementations

impl<'a> Transaction<'a>

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

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_storage_usage_event(&mut self, metric: VersionedStorageUsage)

pub fn insert_storage_usage_events( &mut self, metrics: impl IntoIterator<Item = VersionedStorageUsage> )

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

pub fn insert_user_schema( &mut self, database_id: DatabaseId, schema_name: &str, owner_id: RoleId, privileges: Vec<MzAclItem> ) -> Result<(SchemaId, u32), 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 ) -> Result<(RoleId, u32), CatalogError>

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

Panics if any introspection source id is not a system id

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

Panics if any introspection source id is not a system id

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_id: ReplicaId, replica_name: &str, config: ReplicaConfig, owner_id: RoleId ) -> Result<(), CatalogError>

pub fn update_introspection_source_index_gids( &mut self, mappings: impl Iterator<Item = (ClusterId, impl Iterator<Item = (String, 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: GlobalId, schema_id: SchemaId, item_name: &str, create_sql: String, owner_id: RoleId, privileges: Vec<MzAclItem> ) -> Result<u32, CatalogError>

pub fn insert_item( &mut self, id: GlobalId, oid: u32, schema_id: SchemaId, item_name: &str, create_sql: String, owner_id: RoleId, privileges: Vec<MzAclItem> ) -> 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<GlobalId>, CatalogError>

pub fn allocate_user_item_ids( &mut self, amount: u64 ) -> Result<Vec<GlobalId>, 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_oid(&mut self) -> Result<u32, CatalogError>

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

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

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

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

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

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

pub fn remove_item(&mut self, id: GlobalId) -> 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<GlobalId> ) -> 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: GlobalId, 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<GlobalId, 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<GlobalId, 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 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_catalog_content_version( &mut self, version: String ) -> Result<(), CatalogError>

pub fn insert_introspection_source_indexes( &mut self, introspection_source_indexes: Vec<(ClusterId, String, GlobalId)> ) -> Result<Vec<IntrospectionSourceIndex>, 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(&mut self, key: String) -> Option<u64>

Get the value of a persisted config.

pub fn set_persist_txn_tables( &mut self, value: PersistTxnTablesImpl ) -> Result<(), CatalogError>

Updates the catalog persist_txn_tables “config” value to match the persist_txn_tables “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_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_id: CommentObjectId ) -> Result<Vec<(CommentObjectId, Option<usize>, String)>, 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 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_system_object_mappings( &self ) -> impl Iterator<Item = SystemObjectMapping>

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

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

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

pub fn commit_op(&mut self)

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.

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

Commits the storage transaction to durable storage. Any error returned 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.

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

This method tests for self and other values to be equal, and is used by ==.

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

This method 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, String>

Retrieve all of the visible storage metadata.

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

Add new storage metadata for a collection.

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

Remove the metadata associated with the identified collections.

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

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<String> ) -> Result<(), StorageError<Timestamp>>

Insert the specified values as unfinalized shards.

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

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

fn get_persist_txn_shard(&self) -> Option<String>

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

fn write_persist_txn_shard( &mut self, shard: String ) -> Result<(), StorageError<Timestamp>>

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