uqa_sql/schema/sequences/
ownership.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Bind sequence ownership to stable table and column identities.
8use crate::ast::{ColumnDef, SequenceOwnership};
9use crate::SQLError;
10use uqa_core::RelationIdentity;
11
12pub type SequenceOwnerColumns = ([u8; 16], Vec<ColumnDef>);
13
14pub trait SequenceOwnerCatalog {
15    fn resolve_owner_relation(
16        &self,
17        name: &str,
18    ) -> Result<Option<(String, &'static str)>, SQLError>;
19    fn owner_table_columns(
20        &self,
21        canonical: &str,
22    ) -> Result<Option<SequenceOwnerColumns>, SQLError>;
23    fn owner_foreign_columns(&self, relation: &RelationIdentity) -> Option<SequenceOwnerColumns>;
24}
25
26#[derive(Debug, Clone, Copy)]
27pub struct SequenceOwnerColumnIdentity {
28    pub table_object_id: [u8; 16],
29    pub column_object_id: [u8; 16],
30}
31
32pub fn sequence_owner_column_identity(
33    catalog: &dyn SequenceOwnerCatalog,
34    canonical: &str,
35    relation_kind: &str,
36    column_name: &str,
37) -> Result<Option<SequenceOwnerColumnIdentity>, SQLError> {
38    let relation = RelationIdentity::from_legacy_name(canonical).map_err(|error| {
39        SQLError::Internal(format!("resolve sequence owner `{canonical}`: {error}"))
40    })?;
41    let columns = match relation_kind {
42        "table" => catalog
43            .owner_table_columns(canonical)?
44            .ok_or_else(|| SQLError::Internal(format!("table `{canonical}` disappeared")))?,
45        "foreign table" => catalog.owner_foreign_columns(&relation).ok_or_else(|| {
46            SQLError::Internal(format!("foreign table `{canonical}` disappeared"))
47        })?,
48        _ => return Ok(None),
49    };
50    let column_object_id = columns
51        .1
52        .iter()
53        .find(|column| column.name == column_name)
54        .ok_or_else(|| SQLError::Routine {
55            sqlstate: "42703".into(),
56            message: format!(
57                "column \"{column_name}\" of relation \"{}\" does not exist",
58                relation.name
59            ),
60        })?
61        .object_id
62        .ok_or_else(|| {
63            SQLError::Internal(format!(
64                "column `{canonical}`.`{column_name}` has no object identity"
65            ))
66        })?;
67    Ok(Some(SequenceOwnerColumnIdentity {
68        table_object_id: columns.0,
69        column_object_id,
70    }))
71}
72
73pub fn bind_sequence_owner(
74    catalog: &dyn SequenceOwnerCatalog,
75    sequence_name: &str,
76    ownership: &SequenceOwnership,
77) -> Result<Option<SequenceOwnerColumnIdentity>, SQLError> {
78    let SequenceOwnership::Column { table, column } = ownership else {
79        return Ok(None);
80    };
81    let (table_name, kind) =
82        catalog
83            .resolve_owner_relation(table)?
84            .ok_or_else(|| SQLError::Routine {
85                sqlstate: "42P01".into(),
86                message: format!("relation \"{table}\" does not exist"),
87            })?;
88    let Some(owner_column) = sequence_owner_column_identity(catalog, &table_name, kind, column)?
89    else {
90        return Err(SQLError::Routine {
91            sqlstate: "42809".into(),
92            message: format!("sequence cannot be owned by relation \"{table_name}\""),
93        });
94    };
95    let sequence_relation = RelationIdentity::from_legacy_name(sequence_name).map_err(|error| {
96        SQLError::Internal(format!(
97            "resolve sequence `{sequence_name}` ownership: {error}"
98        ))
99    })?;
100    let table_relation = RelationIdentity::from_legacy_name(&table_name).map_err(|error| {
101        SQLError::Internal(format!("resolve table `{table_name}` ownership: {error}"))
102    })?;
103    if sequence_relation.schema != table_relation.schema {
104        return Err(SQLError::Routine {
105            sqlstate: "55000".into(),
106            message: "sequence must be in same schema as table it is linked to".into(),
107        });
108    }
109    Ok(Some(owner_column))
110}
111
112pub fn require_sequence_ownership(
113    local_name: &str,
114    has_owner_privileges: bool,
115) -> Result<(), SQLError> {
116    if has_owner_privileges {
117        return Ok(());
118    }
119    Err(SQLError::Routine {
120        sqlstate: "42501".into(),
121        message: format!("must be owner of sequence {local_name}"),
122    })
123}
124pub fn reject_owned_sequence_role_change(
125    local_name: &str,
126    has_column_owner: bool,
127) -> Result<(), SQLError> {
128    if !has_column_owner {
129        return Ok(());
130    }
131    Err(SQLError::Routine {
132        sqlstate: "0A000".into(),
133        message: format!("cannot change owner of sequence \"{local_name}\""),
134    })
135}
uqa_sql/schema/sequences/ownership.rs

uqa_sql/schema/sequences/
ownership.rs
