uqa_sql/schema/
foreign_keys.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Bind foreign-key declarations, select referenced keys, and validate REFERENCES privileges.
8use crate::assignment::columns::{AssignmentColumnCatalog, ColumnCatalogError};
9use crate::ast::TableKeyConstraint;
10use crate::SQLError;
11/// Namespace, key declarations, and REFERENCES checks for one foreign-key declaration.
12pub trait ForeignKeyDefinitionCatalog {
13    fn resolve_table_reference(&self, name: &str) -> Result<String, SQLError>;
14    fn bound_table_name(&self, name: &str) -> Result<Option<String>, SQLError>;
15    fn referenceable_keys(
16        &self,
17        table: &str,
18    ) -> Result<Vec<TableKeyConstraint>, ColumnCatalogError>;
19    fn ensure_reference_privilege(&self, table: &str, column: &str) -> Result<(), SQLError>;
20}
21pub struct ForeignKeyDefinitionContext<'a> {
22    pub catalog: &'a dyn ForeignKeyDefinitionCatalog,
23    pub columns: &'a dyn AssignmentColumnCatalog,
24}
25fn ddl_storage_error(action: &str, error: ColumnCatalogError) -> SQLError {
26    crate::catalog::errors::storage_error(action, error.as_ref())
27}
28fn constraint_error(sqlstate: &str, message: impl Into<String>) -> SQLError {
29    SQLError::Routine {
30        sqlstate: sqlstate.into(),
31        message: message.into(),
32    }
33}
34pub fn validate_foreign_key_definition(
35    context: &ForeignKeyDefinitionContext<'_>,
36    table: &str,
37    foreign_key: &mut crate::ast::ForeignKey,
38) -> Result<(), SQLError> {
39    validate_foreign_key_definition_with_local_state(context, table, None, None, foreign_key)
40}
41
42pub fn validate_foreign_key_definition_with_local_state(
43    context: &ForeignKeyDefinitionContext<'_>,
44    table: &str,
45    local_columns: Option<&[crate::ast::ColumnDef]>,
46    local_keys: Option<&[crate::ast::TableKeyConstraint]>,
47    foreign_key: &mut crate::ast::ForeignKey,
48) -> Result<(), SQLError> {
49    foreign_key.ref_table = context
50        .catalog
51        .resolve_table_reference(&foreign_key.ref_table)?;
52    validate_bound_foreign_key_definition_with_local_state(
53        context,
54        table,
55        local_columns,
56        local_keys,
57        foreign_key,
58    )
59}
60
61#[expect(
62    clippy::too_many_lines,
63    reason = "preserves DDL dependency and action order"
64)]
65pub fn validate_bound_foreign_key_definition_with_local_state(
66    context: &ForeignKeyDefinitionContext<'_>,
67    table: &str,
68    local_columns: Option<&[crate::ast::ColumnDef]>,
69    local_keys: Option<&[crate::ast::TableKeyConstraint]>,
70    foreign_key: &mut crate::ast::ForeignKey,
71) -> Result<(), SQLError> {
72    let stored_columns;
73    let columns = if let Some(columns) = local_columns {
74        columns
75    } else {
76        stored_columns = context
77            .columns
78            .try_describe_table(table)
79            .map_err(|error| ddl_storage_error("FOREIGN KEY local table", error))?
80            .ok_or_else(|| SQLError::UnknownTable(table.to_string()))?;
81        &stored_columns
82    };
83    for column in &foreign_key.local_columns {
84        if !columns.iter().any(|definition| definition.name == *column) {
85            return Err(SQLError::UnknownColumn(format!("{table}.{column}")));
86        }
87    }
88    let referenced = context
89        .catalog
90        .bound_table_name(&foreign_key.ref_table)?
91        .ok_or_else(|| SQLError::UnknownTable(foreign_key.ref_table.clone()))?;
92    let referenced_columns = context
93        .columns
94        .try_describe_table(&referenced)
95        .map_err(|error| ddl_storage_error("FOREIGN KEY referenced columns", error))?
96        .ok_or_else(|| SQLError::UnknownTable(referenced.clone()))?;
97    let local = context
98        .catalog
99        .bound_table_name(table)?
100        .ok_or_else(|| SQLError::UnknownTable(table.to_string()))?;
101    let referenced_keys = if referenced == local {
102        match local_keys {
103            Some(keys) => keys.to_vec(),
104            None => context
105                .catalog
106                .referenceable_keys(&referenced)
107                .map_err(|error| ddl_storage_error("FOREIGN KEY referenced key", error))?,
108        }
109    } else {
110        context
111            .catalog
112            .referenceable_keys(&referenced)
113            .map_err(|error| ddl_storage_error("FOREIGN KEY referenced key", error))?
114    };
115    if foreign_key.ref_columns.is_empty() {
116        let primary_key = referenced_keys
117            .iter()
118            .find(|key| key.kind == crate::ast::TableKeyConstraintKind::PrimaryKey)
119            .ok_or_else(|| {
120                constraint_error(
121                    "42704",
122                    format!("there is no primary key for referenced table \"{referenced}\""),
123                )
124            })?;
125        foreign_key.ref_columns.clone_from(&primary_key.columns);
126    }
127    if foreign_key.local_columns.len() != foreign_key.ref_columns.len() {
128        return Err(constraint_error(
129            "42830",
130            "number of referencing and referenced columns for foreign key disagree",
131        ));
132    }
133    for (local_column, referenced_column) in foreign_key
134        .local_columns
135        .iter()
136        .zip(&foreign_key.ref_columns)
137    {
138        let local_definition = columns
139            .iter()
140            .find(|definition| definition.name == *local_column)
141            .ok_or_else(|| SQLError::UnknownColumn(format!("{table}.{local_column}")))?;
142        let referenced_definition = referenced_columns
143            .iter()
144            .find(|definition| definition.name == *referenced_column)
145            .ok_or_else(|| SQLError::UnknownColumn(format!("{referenced}.{referenced_column}")))?;
146        if crate::type_resolution::foreign_key_operand_type(
147            &local_definition.ty,
148            &referenced_definition.ty,
149        )
150        .is_err()
151        {
152            return Err(constraint_error(
153                "42804",
154                format!(
155                    "foreign key constraint cannot be implemented: key columns \"{local_column}\" and \"{referenced_column}\" are of incompatible types: {} and {}",
156                    local_definition.ty.sql_name(),
157                    referenced_definition.ty.sql_name()
158                ),
159            ));
160        }
161    }
162    if foreign_key.period {
163        super::constraints::validate_foreign_key_definition(
164            table,
165            columns,
166            &referenced,
167            &referenced_columns,
168            &referenced_keys,
169            foreign_key,
170        )?;
171    } else {
172        let referenced_column_set = foreign_key
173            .ref_columns
174            .iter()
175            .collect::<std::collections::BTreeSet<_>>();
176        let has_unique_key = referenced_column_set.len() == foreign_key.ref_columns.len()
177            && referenced_keys.iter().any(|key| {
178                key.columns.len() == foreign_key.ref_columns.len()
179                    && key
180                        .columns
181                        .iter()
182                        .collect::<std::collections::BTreeSet<_>>()
183                        == referenced_column_set
184            });
185        if !has_unique_key {
186            return Err(constraint_error(
187                "42830",
188                format!(
189                    "there is no unique constraint matching given keys for referenced table \"{referenced}\""
190                ),
191            ));
192        }
193    }
194    for column in &foreign_key.ref_columns {
195        context
196            .catalog
197            .ensure_reference_privilege(&referenced, column)?;
198    }
199    foreign_key.referenced_key = referenced_keys
200        .iter()
201        .find(|key| {
202            key.columns.len() == foreign_key.ref_columns.len()
203                && foreign_key
204                    .ref_columns
205                    .iter()
206                    .all(|column| key.columns.contains(column))
207                && (!foreign_key.period || key.without_overlaps)
208        })
209        .and_then(|key| key.name.clone());
210    foreign_key.ref_table = referenced;
211    Ok(())
212}
213
214pub fn column_foreign_key(
215    column: &crate::ast::ColumnDef,
216    reference: &crate::ast::ForeignKeyRef,
217) -> crate::ast::ForeignKey {
218    crate::ast::ForeignKey {
219        referenced_key: reference.referenced_key.clone(),
220        name: reference.name.clone(),
221        object_id: reference.object_id,
222        local_columns: vec![column.name.clone()],
223        ref_table: reference.table.clone(),
224        ref_columns: reference.column.iter().cloned().collect(),
225        on_update: reference.on_update,
226        on_delete: reference.on_delete,
227        on_delete_set_columns: Vec::new(),
228        match_type: reference.match_type,
229        enforced: reference.enforced,
230        validated: reference.validated,
231        deferrable: reference.deferrable,
232        initially_deferred: reference.initially_deferred,
233        period: reference.period,
234    }
235}
236
237pub fn resolve_foreign_key_parent(
238    context: &ForeignKeyDefinitionContext<'_>,
239    reference: &str,
240) -> Result<(String, Vec<crate::ast::ColumnDef>, Vec<TableKeyConstraint>), SQLError> {
241    let canonical = context
242        .catalog
243        .bound_table_name(reference)?
244        .ok_or_else(|| SQLError::UnknownTable(reference.to_string()))?;
245    let columns = context
246        .columns
247        .try_describe_table(&canonical)
248        .map_err(|error| SQLError::Internal(format!("describe FOREIGN KEY target: {error}")))?
249        .ok_or_else(|| SQLError::UnknownTable(canonical.clone()))?;
250    let keys = context
251        .catalog
252        .referenceable_keys(&canonical)
253        .map_err(|error| SQLError::Internal(format!("read FOREIGN KEY target keys: {error}")))?;
254    Ok((canonical, columns, keys))
255}
uqa_sql/schema/foreign_keys.rs

uqa_sql/schema/
foreign_keys.rs
