uqa-execution 0.4.6

Volcano physical operators with row-batch pipelines
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Allocate public role OIDs independently from durable role incarnations.

use super::super::roles::locking::ROLE_CATALOG_CLASS_ID;
use super::context::RoleExecutionContext;
use crate::row_locks::{shared_objects::SharedCatalogLock, RelationLockMode};
use uqa_sql::{catalog::roles::RoleDefinition, SQLError};

pub(super) const MEMBERSHIP_CATALOG_CLASS_ID: u32 = 1261;

pub(super) fn reserve_membership_oid(
    context: &RoleExecutionContext<'_>,
    staged: &std::collections::BTreeSet<i64>,
    allocate: impl FnMut() -> Result<i64, SQLError>,
) -> Result<i64, SQLError> {
    let assigned = |oid| {
        let _roles = context.analysis.roles.role_definitions();
        let memberships = context.analysis.roles.role_memberships();
        staged.contains(&oid) || memberships.values().any(|membership| membership.oid == oid)
    };
    crate::catalog::identity::reserve_catalog_oid(
        context.locks,
        MEMBERSHIP_CATALOG_CLASS_ID,
        "role membership",
        |oid| Ok(assigned(oid)),
        allocate,
    )
}

pub(super) fn allocate_oid() -> Result<i64, SQLError> {
    crate::catalog::identity::allocate_catalog_oid("role")
}

pub(super) fn reserve_definition(
    context: &RoleExecutionContext<'_>,
    statement: &uqa_sql::ast::CreateRoleStmt,
) -> Result<RoleDefinition, SQLError> {
    if context
        .analysis
        .roles
        .role_definitions()
        .contains_key(&statement.name)
    {
        return Err(SQLError::Routine {
            sqlstate: "42710".into(),
            message: format!("role \"{}\" already exists", statement.name),
        });
    }
    let name_guard = context.locks.acquire_shared_catalog(
        SharedCatalogLock::Name {
            class_id: ROLE_CATALOG_CLASS_ID,
            name: &statement.name,
        },
        RelationLockMode::AccessExclusive,
    )?;
    context.locks.refresh_shared_catalog()?;
    if context
        .analysis
        .roles
        .role_definitions()
        .contains_key(&statement.name)
    {
        return Err(SQLError::Routine {
            sqlstate: "23505".into(),
            message: "duplicate key value violates unique constraint \"pg_authid_rolname_index\""
                .into(),
        });
    }
    name_guard.retain();
    reserve_oid(context, statement, allocate_oid)
}

pub(super) fn reserve_oid(
    context: &RoleExecutionContext<'_>,
    statement: &uqa_sql::ast::CreateRoleStmt,
    allocate: impl FnMut() -> Result<i64, SQLError>,
) -> Result<RoleDefinition, SQLError> {
    let oid = crate::catalog::identity::reserve_catalog_oid(
        context.locks,
        ROLE_CATALOG_CLASS_ID,
        "role",
        |oid| {
            Ok(context
                .analysis
                .roles
                .role_definitions()
                .values()
                .any(|role| role.oid == oid))
        },
        allocate,
    )?;
    let object_id = crate::catalog::identity::new_nonzero_catalog_identity("role", "identity")
        .map_err(|error| SQLError::Internal(error.to_string()))?;
    Ok(RoleDefinition::from_create(statement, oid, object_id))
}