uqa-execution 0.5.0

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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

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

//! Capability-scoped `pg_namespace` row synthesis.

use crate::catalog::CatalogReadView;
use uqa_sql::{ResultRow, SQLError};

use super::helpers::acl::acl_identifier;
use super::helpers::oids::{current_user_oid, namespace_oid};
use super::helpers::rows::{catalog_array, int_value, row, str_value};

pub fn build_pg_namespace(catalog: &CatalogReadView) -> Result<Vec<ResultRow>, SQLError> {
    catalog
        .all_schema_names()
        .into_iter()
        .map(|schema| {
            let security = catalog.schema_security(&schema);
            let names = catalog.schema_security_names(&schema)?;
            Ok(row([
                ("oid", int_value(namespace_oid(catalog, &schema))),
                ("nspname", str_value(&schema)),
                (
                    "nspowner",
                    int_value(match security {
                        Some(security) => security.role_owner.oid,
                        None => current_user_oid(),
                    }),
                ),
                ("nspacl", schema_acl_catalog_value(names.as_ref())?),
            ]))
        })
        .collect::<Result<Vec<_>, SQLError>>()
}

/// `pg_my_temp_schema()`: the OID of the session's temporary namespace, or 0 before the session's first temporary object created it.
pub fn pg_my_temp_schema_value(catalog: &CatalogReadView) -> uqa_core::Value {
    uqa_core::Value::Int(
        catalog
            .snapshot()
            .temporary_namespace
            .as_ref()
            .map_or(0, |temporary| i64::from(temporary.oids.namespace)),
    )
}

/// `pg_is_other_temp_schema(oid)`: whether the namespace is another session's temporary namespace or TOAST namespace, which `isOtherTempNamespace` decides by the name alone.
pub fn pg_is_other_temp_schema_value(
    catalog: &CatalogReadView,
    arguments: &[uqa_core::Value],
) -> Result<uqa_core::Value, SQLError> {
    let [argument] = arguments else {
        return Err(SQLError::BadArity {
            name: "pg_is_other_temp_schema".into(),
            expected: "1".into(),
            actual: arguments.len(),
        });
    };
    let oid = match argument {
        uqa_core::Value::Null => return Ok(uqa_core::Value::Null),
        uqa_core::Value::Int(oid) => *oid,
        other => {
            return Err(SQLError::TypeMismatch(format!(
                "pg_is_other_temp_schema namespace must be oid, got {other:?}"
            )))
        }
    };
    if catalog
        .snapshot()
        .temporary_namespace
        .as_ref()
        .is_some_and(|temporary| temporary.holds_oid(oid))
    {
        return Ok(uqa_core::Value::Bool(false));
    }
    Ok(uqa_core::Value::Bool(
        catalog.all_schema_names().iter().any(|name| {
            namespace_oid(catalog, name) == oid
                && uqa_sql::catalog::temporary_namespace::is_temporary_schema_name(name)
        }),
    ))
}

fn schema_acl_catalog_value(
    security: Option<&crate::catalog::security::SchemaSecurity>,
) -> Result<uqa_core::Value, SQLError> {
    let Some(security) = security else {
        return Ok(uqa_core::Value::Null);
    };
    let Some(acl) = security.acl.as_ref() else {
        return Ok(uqa_core::Value::Null);
    };
    catalog_array(
        acl.iter()
            .map(|entry| {
                let grantee = entry
                    .role
                    .role_name()
                    .map_or_else(String::new, acl_identifier);
                let grantor =
                    acl_identifier(entry.grantor.as_deref().unwrap_or(&security.role_owner));
                let mut privileges = String::new();
                for (enabled, grant_option, code) in [
                    (entry.privileges.usage, entry.grant_options.usage, 'U'),
                    (entry.privileges.create, entry.grant_options.create, 'C'),
                ] {
                    if enabled {
                        privileges.push(code);
                        if grant_option {
                            privileges.push('*');
                        }
                    }
                }
                str_value(format!("{grantee}={privileges}/{grantor}"))
            })
            .collect(),
        "pg_namespace.nspacl",
    )
}

#[cfg(test)]
mod tests;