uqa_sql/catalog/resolution/
creation.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Creation namespaces and index-target lookup over the caller's actual metadata guards.
8
9use super::candidates::{relation_lookup_candidates, RelationCandidateState};
10use crate::catalog::{
11    roles::RoleReferenceNames,
12    security::{
13        schema::SchemaAclPrivilege,
14        schema_inquiry::{SchemaPrivilegeCatalog, SchemaPrivilegeInquiry},
15    },
16};
17use crate::SQLError;
18use uqa_core::RelationIdentity;
19
20pub trait CreationRelationNames {
21    fn contains(&self, relation: &RelationIdentity) -> bool;
22}
23pub trait CreationRelationGuards {
24    fn named_type_exists(&self, identity: &RelationIdentity) -> bool;
25    fn tables(&self) -> Box<dyn CreationRelationNames + '_>;
26    fn views(&self) -> Box<dyn CreationRelationNames + '_>;
27    fn sequences(&self) -> Box<dyn CreationRelationNames + '_>;
28    fn foreign_tables(&self) -> Box<dyn CreationRelationNames + '_>;
29    fn indexes(&self) -> Box<dyn CreationRelationNames + '_>;
30}
31
32/// Domains and row types share the type namespace; sequences and indexes do not define row types.
33pub fn type_name_in_use(catalog: &dyn CreationRelationGuards, identity: &RelationIdentity) -> bool {
34    catalog.named_type_exists(identity)
35        || catalog.tables().contains(identity)
36        || catalog.views().contains(identity)
37        || catalog.foreign_tables().contains(identity)
38}
39
40pub fn ensure_type_name_available(
41    catalog: &dyn CreationRelationGuards,
42    identity: &RelationIdentity,
43) -> Result<(), SQLError> {
44    if type_name_in_use(catalog, identity) {
45        return Err(SQLError::Routine {
46            sqlstate: "42710".into(),
47            message: format!("type \"{}\" already exists", identity.name),
48        });
49    }
50    Ok(())
51}
52
53/// Every relation kind shares the same namespace, independently of query visibility.
54pub fn relation_name_in_use(
55    catalog: &dyn CreationRelationGuards,
56    relation: &RelationIdentity,
57) -> bool {
58    catalog.tables().contains(relation)
59        || catalog.views().contains(relation)
60        || catalog.sequences().contains(relation)
61        || catalog.foreign_tables().contains(relation)
62        || catalog.indexes().contains(relation)
63}
64
65pub fn temporary_creation_parts(
66    state: &dyn RelationCandidateState,
67    name: &str,
68) -> Result<(String, String), SQLError> {
69    let (schema, relation) =
70        RelationIdentity::parse_reference(name).map_err(SQLError::Unsupported)?;
71    let temporary_schema = state.temporary_schema_name();
72    if schema
73        .as_deref()
74        .is_some_and(|schema| schema != "pg_temp" && schema != temporary_schema)
75    {
76        return Err(SQLError::Unsupported(
77            "temporary relations cannot specify a schema name".into(),
78        ));
79    }
80    Ok((temporary_schema, relation))
81}
82
83pub fn api_relation_name(
84    state: &dyn RelationCandidateState,
85    catalog: &dyn SchemaPrivilegeCatalog,
86    name: &str,
87) -> Result<String, String> {
88    let (schema, relation) = RelationIdentity::parse_reference(name)?;
89    if let Some(schema) = schema {
90        if !catalog.schemas().contains_key(&schema) {
91            return Err(format!("schema `{schema}` does not exist"));
92        }
93        return Ok(RelationIdentity::new(schema, relation).qualified_name());
94    }
95    let search_path = state.search_path();
96    let schemas = catalog.schemas();
97    let schema = search_path
98        .iter()
99        .find(|schema| {
100            schema.as_str() != "pg_catalog"
101                && schema.as_str() != "information_schema"
102                && schemas.contains_key(schema.as_str())
103        })
104        .cloned()
105        .ok_or_else(|| "no schema has been selected to create in".to_string())?;
106    Ok(RelationIdentity::new(schema, relation).qualified_name())
107}
108
109pub fn sql_creation_schema(
110    state: &dyn RelationCandidateState,
111    privileges: &SchemaPrivilegeInquiry<'_>,
112    schema: Option<&str>,
113    current_user: &(impl crate::catalog::roles::identity::RoleSubject + ?Sized),
114) -> Option<String> {
115    if let Some(schema) = schema {
116        privileges
117            .schema_security_for_privilege(schema)
118            .is_some()
119            .then(|| schema.to_string())
120    } else {
121        let search_path = state.search_path().clone();
122        search_path.into_iter().find(|schema| {
123            privileges.schema_security_for_privilege(schema).is_some()
124                && privileges.schema_has_privilege_for_role(
125                    schema,
126                    current_user,
127                    SchemaAclPrivilege::Usage,
128                )
129        })
130    }
131}
132
133pub fn missing_creation_schema(schema: Option<String>) -> SQLError {
134    SQLError::Routine {
135        sqlstate: "3F000".into(),
136        message: schema.map_or_else(
137            || "no schema has been selected to create in".into(),
138            |schema| format!("schema \"{schema}\" does not exist"),
139        ),
140    }
141}
142
143pub fn ensure_creation_privilege(
144    names: &dyn RoleReferenceNames,
145    privileges: &SchemaPrivilegeInquiry<'_>,
146    canonical_name: &str,
147) -> Result<(), SQLError> {
148    let relation =
149        RelationIdentity::from_legacy_name(canonical_name).map_err(SQLError::Unsupported)?;
150    let current_user = names.current_role();
151    privileges.require_schema_privilege(&relation.schema, &current_user, SchemaAclPrivilege::Create)
152}
153
154pub fn resolve_index_table_name(
155    names: &dyn RoleReferenceNames,
156    state: &dyn RelationCandidateState,
157    privileges: &SchemaPrivilegeInquiry<'_>,
158    catalog: &dyn CreationRelationGuards,
159    name: &str,
160) -> Result<Option<String>, SQLError> {
161    let (qualified_schema, _) =
162        RelationIdentity::parse_reference(name).map_err(SQLError::Unsupported)?;
163    if let Some(schema) = qualified_schema.as_deref() {
164        if schema != "pg_temp" && schema != state.temporary_schema_name() {
165            if privileges.schema_security_for_privilege(schema).is_none() {
166                return Err(SQLError::Routine {
167                    sqlstate: "3F000".into(),
168                    message: format!("schema \"{schema}\" does not exist"),
169                });
170            }
171            let current_user = names.current_role();
172            privileges.require_schema_privilege(
173                schema,
174                &current_user,
175                SchemaAclPrivilege::Usage,
176            )?;
177        }
178    }
179    let current_user = names.current_role();
180    for relation in relation_lookup_candidates(state, name)
181        .map_err(|error| SQLError::Internal(format!("resolve index table `{name}`: {error}")))?
182    {
183        if qualified_schema.is_none()
184            && relation.schema != state.temporary_schema_name()
185            && !privileges.schema_has_privilege_for_role(
186                &relation.schema,
187                &current_user,
188                SchemaAclPrivilege::Usage,
189            )
190        {
191            continue;
192        }
193        if catalog.tables().contains(&relation) {
194            return Ok(Some(relation.qualified_name()));
195        }
196        if catalog.views().contains(&relation)
197            || catalog.sequences().contains(&relation)
198            || catalog.foreign_tables().contains(&relation)
199            || catalog.indexes().contains(&relation)
200        {
201            return Ok(None);
202        }
203    }
204    Ok(None)
205}
206
207#[cfg(test)]
208mod tests;
uqa_sql/catalog/resolution/creation.rs

uqa_sql/catalog/resolution/
creation.rs
