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 local = context
93        .catalog
94        .bound_table_name(table)?
95        .ok_or_else(|| SQLError::UnknownTable(table.to_string()))?;
96    let referenced_columns = if referenced == local {
97        local_columns.map(<[crate::ast::ColumnDef]>::to_vec)
98    } else {
99        None
100    }
101    .map(Ok)
102    .unwrap_or_else(|| {
103        context
104            .columns
105            .try_describe_table(&referenced)
106            .map_err(|error| ddl_storage_error("FOREIGN KEY referenced columns", error))?
107            .ok_or_else(|| SQLError::UnknownTable(referenced.clone()))
108    })?;
109    let referenced_keys = if referenced == local {
110        match local_keys {
111            Some(keys) => keys.to_vec(),
112            None => context
113                .catalog
114                .referenceable_keys(&referenced)
115                .map_err(|error| ddl_storage_error("FOREIGN KEY referenced key", error))?,
116        }
117    } else {
118        context
119            .catalog
120            .referenceable_keys(&referenced)
121            .map_err(|error| ddl_storage_error("FOREIGN KEY referenced key", error))?
122    };
123    if foreign_key.ref_columns.is_empty() {
124        let primary_key = referenced_keys
125            .iter()
126            .find(|key| key.kind == crate::ast::TableKeyConstraintKind::PrimaryKey)
127            .ok_or_else(|| {
128                constraint_error(
129                    "42704",
130                    format!("there is no primary key for referenced table \"{referenced}\""),
131                )
132            })?;
133        foreign_key.ref_columns.clone_from(&primary_key.columns);
134    }
135    if foreign_key.local_columns.len() != foreign_key.ref_columns.len() {
136        return Err(constraint_error(
137            "42830",
138            "number of referencing and referenced columns for foreign key disagree",
139        ));
140    }
141    for (local_column, referenced_column) in foreign_key
142        .local_columns
143        .iter()
144        .zip(&foreign_key.ref_columns)
145    {
146        let local_definition = columns
147            .iter()
148            .find(|definition| definition.name == *local_column)
149            .ok_or_else(|| SQLError::UnknownColumn(format!("{table}.{local_column}")))?;
150        let referenced_definition = referenced_columns
151            .iter()
152            .find(|definition| definition.name == *referenced_column)
153            .ok_or_else(|| SQLError::UnknownColumn(format!("{referenced}.{referenced_column}")))?;
154        if crate::type_resolution::foreign_key_operand_type(
155            &local_definition.ty,
156            &referenced_definition.ty,
157        )
158        .is_err()
159        {
160            return Err(constraint_error(
161                "42804",
162                format!(
163                    "foreign key constraint cannot be implemented: key columns \"{local_column}\" and \"{referenced_column}\" are of incompatible types: {} and {}",
164                    local_definition.ty.sql_name(),
165                    referenced_definition.ty.sql_name()
166                ),
167            ));
168        }
169    }
170    if foreign_key.period {
171        super::constraints::validate_foreign_key_definition(
172            table,
173            columns,
174            &referenced,
175            &referenced_columns,
176            &referenced_keys,
177            foreign_key,
178        )?;
179    } else {
180        let referenced_column_set = foreign_key
181            .ref_columns
182            .iter()
183            .collect::<std::collections::BTreeSet<_>>();
184        let has_unique_key = referenced_column_set.len() == foreign_key.ref_columns.len()
185            && referenced_keys.iter().any(|key| {
186                key.columns.len() == foreign_key.ref_columns.len()
187                    && key
188                        .columns
189                        .iter()
190                        .collect::<std::collections::BTreeSet<_>>()
191                        == referenced_column_set
192            });
193        if !has_unique_key {
194            return Err(constraint_error(
195                "42830",
196                format!(
197                    "there is no unique constraint matching given keys for referenced table \"{referenced}\""
198                ),
199            ));
200        }
201    }
202    for column in &foreign_key.ref_columns {
203        context
204            .catalog
205            .ensure_reference_privilege(&referenced, column)?;
206    }
207    foreign_key.referenced_key = referenced_keys
208        .iter()
209        .find(|key| {
210            key.columns.len() == foreign_key.ref_columns.len()
211                && foreign_key
212                    .ref_columns
213                    .iter()
214                    .all(|column| key.columns.contains(column))
215                && (!foreign_key.period || key.without_overlaps)
216        })
217        .and_then(|key| key.name.clone());
218    foreign_key.ref_table = referenced;
219    Ok(())
220}
221
222pub fn column_foreign_key(
223    column: &crate::ast::ColumnDef,
224    reference: &crate::ast::ForeignKeyRef,
225) -> crate::ast::ForeignKey {
226    crate::ast::ForeignKey {
227        referenced_key: reference.referenced_key.clone(),
228        referenced_index: reference.referenced_index,
229        name: reference.name.clone(),
230        object_id: reference.object_id,
231        catalog_identity: reference.catalog_identity,
232        local_columns: vec![column.name.clone()],
233        ref_table: reference.table.clone(),
234        ref_columns: reference.column.iter().cloned().collect(),
235        on_update: reference.on_update,
236        on_delete: reference.on_delete,
237        on_delete_set_columns: Vec::new(),
238        match_type: reference.match_type,
239        enforced: reference.enforced,
240        validated: reference.validated,
241        deferrable: reference.deferrable,
242        initially_deferred: reference.initially_deferred,
243        period: reference.period,
244    }
245}
246
247pub fn resolve_foreign_key_parent(
248    context: &ForeignKeyDefinitionContext<'_>,
249    reference: &str,
250) -> Result<(String, Vec<crate::ast::ColumnDef>, Vec<TableKeyConstraint>), SQLError> {
251    let canonical = context
252        .catalog
253        .bound_table_name(reference)?
254        .ok_or_else(|| SQLError::UnknownTable(reference.to_string()))?;
255    let columns = context
256        .columns
257        .try_describe_table(&canonical)
258        .map_err(|error| SQLError::Internal(format!("describe FOREIGN KEY target: {error}")))?
259        .ok_or_else(|| SQLError::UnknownTable(canonical.clone()))?;
260    let keys = context
261        .catalog
262        .referenceable_keys(&canonical)
263        .map_err(|error| SQLError::Internal(format!("read FOREIGN KEY target keys: {error}")))?;
264    Ok((canonical, columns, keys))
265}
uqa_sql/schema/foreign_keys.rs

uqa_sql/schema/
foreign_keys.rs
