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(missing_foreign_key_column(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    let referenced_name = uqa_core::RelationIdentity::from_legacy_name(&referenced)
124        .map_err(SQLError::Internal)?
125        .name;
126    if foreign_key.ref_columns.is_empty() {
127        let primary_key = referenced_keys
128            .iter()
129            .find(|key| key.kind == crate::ast::TableKeyConstraintKind::PrimaryKey)
130            .ok_or_else(|| {
131                constraint_error(
132                    "42704",
133                    format!("there is no primary key for referenced table \"{referenced_name}\""),
134                )
135            })?;
136        foreign_key.ref_columns.clone_from(&primary_key.columns);
137    }
138    if foreign_key.local_columns.len() != foreign_key.ref_columns.len() {
139        return Err(constraint_error(
140            "42830",
141            "number of referencing and referenced columns for foreign key disagree",
142        ));
143    }
144    for (local_column, referenced_column) in foreign_key
145        .local_columns
146        .iter()
147        .zip(&foreign_key.ref_columns)
148    {
149        let local_definition = columns
150            .iter()
151            .find(|definition| definition.name == *local_column)
152            .ok_or_else(|| missing_foreign_key_column(local_column))?;
153        let referenced_definition = referenced_columns
154            .iter()
155            .find(|definition| definition.name == *referenced_column)
156            .ok_or_else(|| missing_foreign_key_column(referenced_column))?;
157        if crate::type_resolution::foreign_key_operand_type(
158            &local_definition.ty,
159            &referenced_definition.ty,
160        )
161        .is_err()
162        {
163            return Err(constraint_error(
164                "42804",
165                format!(
166                    "foreign key constraint cannot be implemented: key columns \"{local_column}\" and \"{referenced_column}\" are of incompatible types: {} and {}",
167                    local_definition.ty.sql_name(),
168                    referenced_definition.ty.sql_name()
169                ),
170            ));
171        }
172    }
173    if foreign_key.period {
174        super::constraints::validate_foreign_key_definition(
175            table,
176            columns,
177            &referenced,
178            &referenced_columns,
179            &referenced_keys,
180            foreign_key,
181        )?;
182    } else {
183        let referenced_column_set = foreign_key
184            .ref_columns
185            .iter()
186            .collect::<std::collections::BTreeSet<_>>();
187        let has_unique_key = referenced_column_set.len() == foreign_key.ref_columns.len()
188            && referenced_keys.iter().any(|key| {
189                key.columns.len() == foreign_key.ref_columns.len()
190                    && key
191                        .columns
192                        .iter()
193                        .collect::<std::collections::BTreeSet<_>>()
194                        == referenced_column_set
195            });
196        if !has_unique_key {
197            return Err(constraint_error(
198                "42830",
199                format!(
200                    "there is no unique constraint matching given keys for referenced table \"{referenced_name}\""
201                ),
202            ));
203        }
204    }
205    for column in &foreign_key.ref_columns {
206        context
207            .catalog
208            .ensure_reference_privilege(&referenced, column)?;
209    }
210    foreign_key.referenced_key = referenced_keys
211        .iter()
212        .find(|key| {
213            key.columns.len() == foreign_key.ref_columns.len()
214                && foreign_key
215                    .ref_columns
216                    .iter()
217                    .all(|column| key.columns.contains(column))
218                && (!foreign_key.period || key.without_overlaps)
219        })
220        .and_then(|key| key.name.clone());
221    foreign_key.ref_table = referenced;
222    Ok(())
223}
224
225pub fn column_foreign_key(
226    column: &crate::ast::ColumnDef,
227    reference: &crate::ast::ForeignKeyRef,
228) -> crate::ast::ForeignKey {
229    crate::ast::ForeignKey {
230        referenced_key: reference.referenced_key.clone(),
231        referenced_index: reference.referenced_index,
232        name: reference.name.clone(),
233        object_id: reference.object_id,
234        catalog_identity: reference.catalog_identity,
235        local_columns: vec![column.name.clone()],
236        ref_table: reference.table.clone(),
237        ref_columns: reference.column.iter().cloned().collect(),
238        on_update: reference.on_update,
239        on_delete: reference.on_delete,
240        on_delete_set_columns: Vec::new(),
241        match_type: reference.match_type,
242        enforced: reference.enforced,
243        validated: reference.validated,
244        deferrable: reference.deferrable,
245        initially_deferred: reference.initially_deferred,
246        period: reference.period,
247        referenced_partitions: reference.referenced_partitions.clone(),
248    }
249}
250
251pub fn resolve_foreign_key_parent(
252    context: &ForeignKeyDefinitionContext<'_>,
253    reference: &str,
254) -> Result<(String, Vec<crate::ast::ColumnDef>, Vec<TableKeyConstraint>), SQLError> {
255    let canonical = context
256        .catalog
257        .bound_table_name(reference)?
258        .ok_or_else(|| SQLError::UnknownTable(reference.to_string()))?;
259    let columns = context
260        .columns
261        .try_describe_table(&canonical)
262        .map_err(|error| SQLError::Internal(format!("describe FOREIGN KEY target: {error}")))?
263        .ok_or_else(|| SQLError::UnknownTable(canonical.clone()))?;
264    let keys = context
265        .catalog
266        .referenceable_keys(&canonical)
267        .map_err(|error| SQLError::Internal(format!("read FOREIGN KEY target keys: {error}")))?;
268    Ok((canonical, columns, keys))
269}
270
271/// A foreign key using `PERIOD` supports no referential action but `NO ACTION`; `PostgreSQL` checks the update action before the delete action and names the one it rejects.
272pub fn validate_period_foreign_key_actions(
273    foreign_key: &crate::ast::ForeignKey,
274) -> Result<(), SQLError> {
275    if !foreign_key.period {
276        return Ok(());
277    }
278    for (action, clause) in [
279        (foreign_key.on_update, "ON UPDATE"),
280        (foreign_key.on_delete, "ON DELETE"),
281    ] {
282        if action != crate::ast::ForeignKeyAction::NoAction {
283            return Err(SQLError::Routine {
284                sqlstate: "0A000".into(),
285                message: format!(
286                    "unsupported {clause} action for foreign key constraint using PERIOD"
287                ),
288            });
289        }
290    }
291    Ok(())
292}
293
294fn missing_foreign_key_column(column: &str) -> SQLError {
295    constraint_error(
296        "42703",
297        format!("column \"{column}\" referenced in foreign key constraint does not exist"),
298    )
299}
uqa_sql/schema/foreign_keys.rs

uqa_sql/schema/
foreign_keys.rs
