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 tables(&self) -> Box<dyn CreationRelationNames + '_>;
25    fn views(&self) -> Box<dyn CreationRelationNames + '_>;
26    fn sequences(&self) -> Box<dyn CreationRelationNames + '_>;
27    fn foreign_tables(&self) -> Box<dyn CreationRelationNames + '_>;
28    fn indexes(&self) -> Box<dyn CreationRelationNames + '_>;
29}
30
31pub fn temporary_creation_parts(
32    state: &dyn RelationCandidateState,
33    name: &str,
34) -> Result<(String, String), SQLError> {
35    let (schema, relation) =
36        RelationIdentity::parse_reference(name).map_err(SQLError::Unsupported)?;
37    let temporary_schema = state.temporary_schema_name();
38    if schema
39        .as_deref()
40        .is_some_and(|schema| schema != "pg_temp" && schema != temporary_schema)
41    {
42        return Err(SQLError::Unsupported(
43            "temporary relations cannot specify a schema name".into(),
44        ));
45    }
46    Ok((temporary_schema, relation))
47}
48
49pub fn api_relation_name(
50    state: &dyn RelationCandidateState,
51    catalog: &dyn SchemaPrivilegeCatalog,
52    name: &str,
53) -> Result<String, String> {
54    let (schema, relation) = RelationIdentity::parse_reference(name)?;
55    if let Some(schema) = schema {
56        if !catalog.schemas().contains_key(&schema) {
57            return Err(format!("schema `{schema}` does not exist"));
58        }
59        return Ok(RelationIdentity::new(schema, relation).qualified_name());
60    }
61    let search_path = state.search_path();
62    let schemas = catalog.schemas();
63    let schema = search_path
64        .iter()
65        .find(|schema| {
66            schema.as_str() != "pg_catalog"
67                && schema.as_str() != "information_schema"
68                && schemas.contains_key(schema.as_str())
69        })
70        .cloned()
71        .ok_or_else(|| "no schema has been selected to create in".to_string())?;
72    Ok(RelationIdentity::new(schema, relation).qualified_name())
73}
74
75pub fn sql_creation_schema(
76    state: &dyn RelationCandidateState,
77    privileges: &SchemaPrivilegeInquiry<'_>,
78    schema: Option<&str>,
79    current_user: &str,
80) -> Option<String> {
81    if let Some(schema) = schema {
82        privileges
83            .schema_security_for_privilege(schema)
84            .is_some()
85            .then(|| schema.to_string())
86    } else {
87        let search_path = state.search_path().clone();
88        search_path.into_iter().find(|schema| {
89            privileges.schema_security_for_privilege(schema).is_some()
90                && privileges.schema_has_privilege_for_role(
91                    schema,
92                    current_user,
93                    SchemaAclPrivilege::Usage,
94                )
95        })
96    }
97}
98
99pub fn missing_creation_schema(schema: Option<String>) -> SQLError {
100    SQLError::Routine {
101        sqlstate: "3F000".into(),
102        message: schema.map_or_else(
103            || "no schema has been selected to create in".into(),
104            |schema| format!("schema \"{schema}\" does not exist"),
105        ),
106    }
107}
108
109pub fn ensure_creation_privilege(
110    names: &dyn RoleReferenceNames,
111    privileges: &SchemaPrivilegeInquiry<'_>,
112    canonical_name: &str,
113) -> Result<(), SQLError> {
114    let relation =
115        RelationIdentity::from_legacy_name(canonical_name).map_err(SQLError::Unsupported)?;
116    let current_user = names.current_user_name();
117    privileges.require_schema_privilege(&relation.schema, &current_user, SchemaAclPrivilege::Create)
118}
119
120pub fn resolve_index_table_name(
121    names: &dyn RoleReferenceNames,
122    state: &dyn RelationCandidateState,
123    privileges: &SchemaPrivilegeInquiry<'_>,
124    catalog: &dyn CreationRelationGuards,
125    name: &str,
126) -> Result<Option<String>, SQLError> {
127    let (qualified_schema, _) =
128        RelationIdentity::parse_reference(name).map_err(SQLError::Unsupported)?;
129    if let Some(schema) = qualified_schema.as_deref() {
130        if schema != "pg_temp" && schema != state.temporary_schema_name() {
131            if privileges.schema_security_for_privilege(schema).is_none() {
132                return Err(SQLError::Routine {
133                    sqlstate: "3F000".into(),
134                    message: format!("schema \"{schema}\" does not exist"),
135                });
136            }
137            let current_user = names.current_user_name();
138            privileges.require_schema_privilege(
139                schema,
140                &current_user,
141                SchemaAclPrivilege::Usage,
142            )?;
143        }
144    }
145    let current_user = names.current_user_name();
146    for relation in relation_lookup_candidates(state, name)
147        .map_err(|error| SQLError::Internal(format!("resolve index table `{name}`: {error}")))?
148    {
149        if qualified_schema.is_none()
150            && relation.schema != state.temporary_schema_name()
151            && !privileges.schema_has_privilege_for_role(
152                &relation.schema,
153                &current_user,
154                SchemaAclPrivilege::Usage,
155            )
156        {
157            continue;
158        }
159        if catalog.tables().contains(&relation) {
160            return Ok(Some(relation.qualified_name()));
161        }
162        if catalog.views().contains(&relation)
163            || catalog.sequences().contains(&relation)
164            || catalog.foreign_tables().contains(&relation)
165            || catalog.indexes().contains(&relation)
166        {
167            return Ok(None);
168        }
169    }
170    Ok(None)
171}
172
173#[cfg(test)]
174mod tests;
uqa_sql/catalog/resolution/creation.rs

uqa_sql/catalog/resolution/
creation.rs
