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::ast::RelationPersistence;
11use crate::catalog::temporary_namespace::{is_temporary_schema_name, temporary_toast_schema_name};
12use crate::catalog::{
13    roles::RoleReferenceNames,
14    security::{
15        schema::SchemaAclPrivilege,
16        schema_inquiry::{SchemaPrivilegeCatalog, SchemaPrivilegeInquiry},
17    },
18};
19use crate::SQLError;
20use uqa_core::RelationIdentity;
21
22pub trait CreationRelationNames {
23    fn contains(&self, relation: &RelationIdentity) -> bool;
24}
25pub trait CreationRelationGuards {
26    fn named_type_exists(&self, identity: &RelationIdentity) -> bool;
27    fn tables(&self) -> Box<dyn CreationRelationNames + '_>;
28    fn views(&self) -> Box<dyn CreationRelationNames + '_>;
29    fn sequences(&self) -> Box<dyn CreationRelationNames + '_>;
30    fn foreign_tables(&self) -> Box<dyn CreationRelationNames + '_>;
31    fn indexes(&self) -> Box<dyn CreationRelationNames + '_>;
32    /// The relations of standalone composite types, which occupy the relation namespace as well as the type namespace.
33    fn composite_types(&self) -> Box<dyn CreationRelationNames + '_>;
34}
35
36/// Domains and row types share the type namespace; sequences and indexes do not define row types.
37pub fn type_name_in_use(catalog: &dyn CreationRelationGuards, identity: &RelationIdentity) -> bool {
38    catalog.named_type_exists(identity)
39        || catalog.tables().contains(identity)
40        || catalog.views().contains(identity)
41        || catalog.foreign_tables().contains(identity)
42        || catalog.composite_types().contains(identity)
43}
44
45pub fn ensure_type_name_available(
46    catalog: &dyn CreationRelationGuards,
47    identity: &RelationIdentity,
48) -> Result<(), SQLError> {
49    if type_name_in_use(catalog, identity) {
50        return Err(SQLError::Routine {
51            sqlstate: "42710".into(),
52            message: format!("type \"{}\" already exists", identity.name),
53        });
54    }
55    Ok(())
56}
57
58/// Every relation kind shares the same namespace, independently of query visibility.
59pub fn relation_name_in_use(
60    catalog: &dyn CreationRelationGuards,
61    relation: &RelationIdentity,
62) -> bool {
63    catalog.tables().contains(relation)
64        || catalog.views().contains(relation)
65        || catalog.sequences().contains(relation)
66        || catalog.foreign_tables().contains(relation)
67        || catalog.indexes().contains(relation)
68        || catalog.composite_types().contains(relation)
69}
70
71pub fn temporary_creation_parts(
72    state: &dyn RelationCandidateState,
73    name: &str,
74) -> Result<(String, String), SQLError> {
75    let (schema, relation) =
76        RelationIdentity::parse_reference(name).map_err(SQLError::Unsupported)?;
77    let temporary_schema = state.temporary_schema_name();
78    if schema
79        .as_deref()
80        .is_some_and(|schema| schema != "pg_temp" && schema != temporary_schema)
81    {
82        return Err(SQLError::Unsupported(
83            "temporary relations cannot specify a schema name".into(),
84        ));
85    }
86    Ok((temporary_schema, relation))
87}
88
89pub fn api_relation_name(
90    state: &dyn RelationCandidateState,
91    catalog: &dyn SchemaPrivilegeCatalog,
92    name: &str,
93) -> Result<String, String> {
94    let (schema, relation) = RelationIdentity::parse_reference(name)?;
95    if let Some(schema) = schema {
96        if !catalog.schemas().contains_key(&schema) {
97            return Err(format!("schema `{schema}` does not exist"));
98        }
99        return Ok(RelationIdentity::new(schema, relation).qualified_name());
100    }
101    let search_path = state.search_path();
102    let schemas = catalog.schemas();
103    let schema = search_path
104        .iter()
105        .find(|schema| {
106            schema.as_str() != "pg_catalog"
107                && schema.as_str() != "information_schema"
108                && schemas.contains_key(schema.as_str())
109        })
110        .cloned()
111        .ok_or_else(|| "no schema has been selected to create in".to_string())?;
112    Ok(RelationIdentity::new(schema, relation).qualified_name())
113}
114
115pub fn sql_creation_schema(
116    state: &dyn RelationCandidateState,
117    privileges: &SchemaPrivilegeInquiry<'_>,
118    schema: Option<&str>,
119    current_user: &(impl crate::catalog::roles::identity::RoleSubject + ?Sized),
120) -> Option<String> {
121    if let Some(schema) = schema {
122        privileges
123            .schema_security_for_privilege(schema)
124            .is_some()
125            .then(|| schema.to_string())
126    } else {
127        let search_path = state.search_path().clone();
128        search_path.into_iter().find(|schema| {
129            privileges.schema_security_for_privilege(schema).is_some()
130                && privileges.schema_has_privilege_for_role(
131                    schema,
132                    current_user,
133                    SchemaAclPrivilege::Usage,
134                )
135        })
136    }
137}
138
139/// The schema `recomputeNamespacePath` makes the creation namespace of an unqualified relation: the first schema of the search path that exists and that the role may use, where `pg_temp` stands for the session's temporary namespace, which a relation created there creates when it does not exist yet.
140pub fn relation_creation_schema(
141    state: &dyn RelationCandidateState,
142    privileges: &SchemaPrivilegeInquiry<'_>,
143    current_user: &(impl crate::catalog::roles::identity::RoleSubject + ?Sized),
144) -> Option<String> {
145    let temporary = state.temporary_schema_name();
146    let search_path = state.search_path().clone();
147    search_path.into_iter().find_map(|schema| {
148        if schema == "pg_temp" {
149            return Some(temporary.clone());
150        }
151        (privileges.schema_security_for_privilege(&schema).is_some()
152            && privileges.schema_has_privilege_for_role(
153                &schema,
154                current_user,
155                SchemaAclPrivilege::Usage,
156            ))
157        .then_some(schema)
158    })
159}
160
161/// `RangeVarAdjustRelationPersistence`: the persistence a new relation takes in `schema`. A relation in the session's temporary namespace or its TOAST namespace is temporary; a temporary relation belongs in no other namespace, and an unlogged one in no temporary namespace.
162pub fn adjusted_relation_persistence(
163    schema: &str,
164    persistence: RelationPersistence,
165    temporary_schema: &str,
166) -> Result<RelationPersistence, SQLError> {
167    let own = schema == temporary_schema || schema == temporary_toast_schema_name(temporary_schema);
168    let any = own || is_temporary_schema_name(schema);
169    let invalid = |message: &str| {
170        Err(SQLError::Routine {
171            sqlstate: "42P16".into(),
172            message: message.into(),
173        })
174    };
175    match persistence {
176        RelationPersistence::Temporary | RelationPersistence::Permanent if own => {
177            Ok(RelationPersistence::Temporary)
178        }
179        RelationPersistence::Temporary | RelationPersistence::Permanent if any => {
180            invalid("cannot create relations in temporary schemas of other sessions")
181        }
182        RelationPersistence::Temporary => {
183            invalid("cannot create temporary relation in non-temporary schema")
184        }
185        RelationPersistence::Unlogged if any => {
186            invalid("only temporary relations may be created in temporary schemas")
187        }
188        persistence => Ok(persistence),
189    }
190}
191
192/// `heap_create`'s refusal of a new relation in `pg_catalog`, `pg_toast` or the session's temporary TOAST namespace, which only system catalog modifications may create relations in.
193pub fn ensure_relation_namespace_writable(
194    relation: &RelationIdentity,
195    temporary_schema: &str,
196) -> Result<(), SQLError> {
197    let schema = relation.schema.as_str();
198    if schema == "pg_catalog"
199        || schema == "pg_toast"
200        || schema == temporary_toast_schema_name(temporary_schema)
201    {
202        return Err(SQLError::Diagnostic {
203            sqlstate: "42501".into(),
204            message: format!("permission denied to create \"{schema}.{}\"", relation.name),
205            detail: Some("System catalog modifications are currently disallowed.".into()),
206            hint: None,
207        });
208    }
209    Ok(())
210}
211
212pub fn missing_creation_schema(schema: Option<String>) -> SQLError {
213    SQLError::Routine {
214        sqlstate: "3F000".into(),
215        message: schema.map_or_else(
216            || "no schema has been selected to create in".into(),
217            |schema| format!("schema \"{schema}\" does not exist"),
218        ),
219    }
220}
221
222pub fn ensure_creation_privilege(
223    names: &dyn RoleReferenceNames,
224    privileges: &SchemaPrivilegeInquiry<'_>,
225    canonical_name: &str,
226) -> Result<(), SQLError> {
227    let relation =
228        RelationIdentity::from_legacy_name(canonical_name).map_err(SQLError::Unsupported)?;
229    let current_user = names.current_role();
230    privileges.require_schema_privilege(&relation.schema, &current_user, SchemaAclPrivilege::Create)
231}
232
233pub fn resolve_index_table_name(
234    names: &dyn RoleReferenceNames,
235    state: &dyn RelationCandidateState,
236    privileges: &SchemaPrivilegeInquiry<'_>,
237    catalog: &dyn CreationRelationGuards,
238    name: &str,
239) -> Result<Option<String>, SQLError> {
240    let (qualified_schema, _) =
241        RelationIdentity::parse_reference(name).map_err(SQLError::Unsupported)?;
242    if let Some(schema) = qualified_schema.as_deref() {
243        if schema != "pg_temp" && schema != state.temporary_schema_name() {
244            if privileges.schema_security_for_privilege(schema).is_none() {
245                return Err(SQLError::Routine {
246                    sqlstate: "3F000".into(),
247                    message: format!("schema \"{schema}\" does not exist"),
248                });
249            }
250            let current_user = names.current_role();
251            privileges.require_schema_privilege(
252                schema,
253                &current_user,
254                SchemaAclPrivilege::Usage,
255            )?;
256        }
257    }
258    let current_user = names.current_role();
259    for relation in relation_lookup_candidates(state, name)
260        .map_err(|error| SQLError::Internal(format!("resolve index table `{name}`: {error}")))?
261    {
262        if qualified_schema.is_none()
263            && relation.schema != state.temporary_schema_name()
264            && !privileges.schema_has_privilege_for_role(
265                &relation.schema,
266                &current_user,
267                SchemaAclPrivilege::Usage,
268            )
269        {
270            continue;
271        }
272        if catalog.tables().contains(&relation) {
273            return Ok(Some(relation.qualified_name()));
274        }
275        // `table_open` refuses indexes and composite types before `DefineIndex` names the relations it cannot index.
276        let unopenable = if catalog.indexes().contains(&relation) {
277            Some("indexes")
278        } else if catalog.composite_types().contains(&relation) {
279            Some("composite types")
280        } else {
281            None
282        };
283        if let Some(kinds) = unopenable {
284            return Err(crate::catalog::analysis::UnopenableRelation {
285                name: relation.name,
286                kinds,
287            }
288            .error());
289        }
290        let unindexable = if catalog.sequences().contains(&relation) {
291            Some("sequence")
292        } else if catalog.foreign_tables().contains(&relation) {
293            Some("foreign table")
294        } else {
295            None
296        };
297        if let Some(kind) = unindexable {
298            return Err(SQLError::Diagnostic {
299                sqlstate: "42809".into(),
300                message: format!("cannot create index on relation \"{}\"", relation.name),
301                detail: crate::catalog::analysis::relkind_not_supported_detail(kind),
302                hint: None,
303            });
304        }
305        if catalog.views().contains(&relation) {
306            return Ok(None);
307        }
308    }
309    Ok(None)
310}
311
312#[cfg(test)]
313mod tests;
uqa_sql/catalog/resolution/creation.rs

uqa_sql/catalog/resolution/
creation.rs
