Struct mz_adapter::catalog::storage::Transaction

pub struct Transaction<'a> {

    stash: &'a mut Stash,
    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>,
    timestamps: TableTransaction<TimestampKey, TimestampValue>,
    system_gid_mapping: TableTransaction<GidMappingKey, GidMappingValue>,
    system_configurations: TableTransaction<ServerConfigurationKey, ServerConfigurationValue>,
    default_privileges: TableTransaction<DefaultPrivilegesKey, DefaultPrivilegesValue>,
    system_privileges: TableTransaction<SystemPrivilegesKey, SystemPrivilegesValue>,
    audit_log_updates: Vec<(AuditLogKey, (), i64)>,
    storage_usage_updates: Vec<(StorageUsageKey, (), i64)>,
}

A Transaction batches multiple Connection operations together and commits them atomically.

Fields

stash: &'a mut Stash

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>

timestamps: TableTransaction<TimestampKey, TimestampValue>

system_gid_mapping: TableTransaction<GidMappingKey, GidMappingValue>

system_configurations: TableTransaction<ServerConfigurationKey, ServerConfigurationValue>

default_privileges: TableTransaction<DefaultPrivilegesKey, DefaultPrivilegesValue>

system_privileges: TableTransaction<SystemPrivilegesKey, SystemPrivilegesValue>

audit_log_updates: Vec<(AuditLogKey, (), i64)>

storage_usage_updates: Vec<(StorageUsageKey, (), i64)>

Implementations

impl<'a> Transaction<'a>

pub(crate) fn loaded_items(&self) -> Vec<Item>

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

pub(crate) fn insert_storage_usage_event( &mut self, metric: VersionedStorageUsage )

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

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

pub(crate) fn insert_user_role( &mut self, name: String, attributes: RoleAttributes, membership: RoleMembership ) -> Result<RoleId, Error>

pub(crate) fn insert_user_cluster( &mut self, cluster_id: ClusterId, cluster_name: &str, linked_object_id: Option<GlobalId>, introspection_source_indexes: Vec<(BuiltinLog, &GlobalId)>, owner_id: RoleId, privileges: Vec<MzAclItem>, config: ClusterConfig ) -> Result<(), Error>

Panics if any introspection source id is not a system id

fn insert_system_cluster( &mut self, cluster_id: ClusterId, cluster_name: &str, introspection_source_indexes: Vec<(BuiltinLog, &GlobalId)>, privileges: Vec<MzAclItem>, config: ClusterConfig ) -> Result<(), Error>

Panics if any introspection source id is not a system id

fn insert_cluster( &mut self, cluster_id: ClusterId, cluster_name: &str, linked_object_id: Option<GlobalId>, introspection_source_indexes: Vec<(BuiltinLog, &GlobalId)>, owner_id: RoleId, privileges: Vec<MzAclItem>, config: ClusterConfig ) -> Result<(), Error>

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

pub(crate) fn check_migration_has_run( &mut self, name: String ) -> Result<bool, Error>

pub(crate) fn mark_migration_has_run( &mut self, name: String ) -> Result<(), Error>

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

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

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

Updates persisted information about persisted introspection source indexes.

Panics if provided id is not a system id.

pub(crate) fn insert_item( &mut self, id: GlobalId, schema_id: SchemaId, item_name: &str, create_sql: String, owner_id: RoleId, privileges: Vec<MzAclItem> ) -> Result<(), Error>

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

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

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

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

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

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

pub(crate) fn remove_item(&mut self, id: GlobalId) -> Result<(), Error>

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 stash. DO NOT call this function in a loop, use Self::remove_items instead.

pub(crate) fn remove_items( &mut self, ids: BTreeSet<GlobalId> ) -> Result<(), Error>

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 called to either abort the transaction or commit.

pub(crate) fn update_item( &mut self, id: GlobalId, item: Item ) -> Result<(), Error>

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 stash. DO NOT call this function in a loop, use Self::update_items instead.

pub(crate) fn update_items( &mut self, items: BTreeMap<GlobalId, Item> ) -> Result<(), Error>

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 called to either abort the transaction or commit.

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

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 stash. 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(crate) fn update_system_object_mappings( &mut self, mappings: BTreeMap<GlobalId, SystemObjectMapping> ) -> Result<(), Error>

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(crate) fn update_cluster( &mut self, id: ClusterId, cluster: Cluster ) -> Result<(), Error>

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 stash. DO NOT call this function in a loop.

pub(crate) fn update_cluster_replica( &mut self, cluster_id: ClusterId, replica_id: ReplicaId, replica: ClusterReplica ) -> Result<(), Error>

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 stash. DO NOT call this function in a loop.

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

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 stash. DO NOT call this function in a loop.

pub(crate) fn update_schema( &mut self, database_id: Option<DatabaseId>, schema_id: SchemaId, schema: Schema ) -> Result<(), Error>

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 stash. DO NOT call this function in a loop.

pub(crate) 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<(), Error>

Set persisted default privilege.

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

Set persisted system privilege.

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

pub(crate) fn drop_comments( &mut self, object_id: CommentObjectId ) -> Result<Vec<(CommentObjectId, Option<usize>, String)>, Error>

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

Upserts persisted system configuration name to value.

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

Removes persisted system configuration name.

pub(crate) fn clear_system_configs(&mut self)

Removes all persisted system configurations.

pub(crate) fn remove_timestamp(&mut self, timeline: Timeline)

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

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

async fn commit_inner(self) -> Result<(), Error>