uqa_sql/schema/
constraints.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Shared CREATE/ALTER validation for table constraints.
8
9use super::SchemaBindingContext;
10use crate::ast::{ColumnDef, Expr, ForeignKey, TableKeyConstraint};
11use crate::plpgsql::{bind_expr, ResolvedVariable, VariableResolver};
12use crate::{ColumnType, SQLError};
13use uqa_core::Value;
14
15struct CheckConditionTypeResolver<'a> {
16    table: &'a str,
17    qualifier: &'a str,
18    columns: &'a [ColumnDef],
19}
20
21impl CheckConditionTypeResolver<'_> {
22    fn column(&self, name: &str) -> Result<ResolvedVariable, SQLError> {
23        let definition = self
24            .columns
25            .iter()
26            .find(|column| column.name == name)
27            .ok_or_else(|| SQLError::UnknownColumn(name.to_string()))?;
28        Ok(ResolvedVariable {
29            value: Value::Null,
30            declared_type: Some(definition.ty.sql_name()),
31        })
32    }
33
34    fn qualifier_matches(&self, qualifier: &str) -> bool {
35        qualifier == self.qualifier
36            || qualifier == self.table
37            || self
38                .table
39                .rsplit_once('.')
40                .is_some_and(|(_, local)| qualifier == local)
41    }
42}
43
44impl VariableResolver for CheckConditionTypeResolver<'_> {
45    fn resolve_name(&mut self, name: &str) -> Result<Option<ResolvedVariable>, SQLError> {
46        self.column(name).map(Some)
47    }
48
49    fn resolve_qualified(
50        &mut self,
51        qualifier: &str,
52        column: &str,
53    ) -> Result<Option<ResolvedVariable>, SQLError> {
54        if !self.qualifier_matches(qualifier) {
55            return Err(SQLError::UnknownTable(qualifier.to_string()));
56        }
57        self.column(column).map(Some)
58    }
59
60    fn resolve_param(&mut self, _index: usize) -> Result<Option<ResolvedVariable>, SQLError> {
61        Ok(None)
62    }
63}
64
65fn is_boolean_type(ty: &ColumnType) -> bool {
66    match ty {
67        ColumnType::Boolean => true,
68        ColumnType::Domain { base, .. } => is_boolean_type(base),
69        _ => false,
70    }
71}
72
73pub fn validate_check_expression(
74    context: &SchemaBindingContext<'_, '_>,
75    table: &str,
76    qualifier: &str,
77    columns: &[ColumnDef],
78    expression: &mut Expr,
79) -> Result<(), SQLError> {
80    let bound = bind_expr(
81        expression,
82        &mut CheckConditionTypeResolver {
83            table,
84            qualifier,
85            columns,
86        },
87    )?;
88    let lowered = crate::plan::ExpressionPlan::lower(bound);
89    if !lowered.subqueries.is_empty() {
90        return Err(SQLError::Routine {
91            sqlstate: "0A000".into(),
92            message: "cannot use subquery in check constraint".into(),
93        });
94    }
95    match crate::type_resolution::common_context_expression_type(
96        &lowered.scalar,
97        &crate::RowSchema::default(),
98        &[],
99        Some(context.catalog),
100    )? {
101        Some(ty) if !is_boolean_type(&ty) => Err(SQLError::TypeMismatch(format!(
102            "argument of CHECK must be type boolean, not type {}",
103            ty.sql_name()
104        ))),
105        None => {
106            if let Expr::Literal(value @ (Value::Str(_) | Value::FixedChar(_))) = expression {
107                *value = crate::expr::cast_value(value, "boolean")?;
108            } else {
109                *expression = Expr::Cast {
110                    expr: Box::new(expression.clone()),
111                    ty: "boolean".into(),
112                };
113            }
114            Ok(())
115        }
116        Some(_) => Ok(()),
117    }?;
118    bind_stored_check_expression_routines(context, table, qualifier, columns, expression)?;
119    super::generated::bind_schema_column_references(expression, qualifier);
120    super::generated::bind_schema_column_references(expression, table);
121    let relation = crate::RelationIdentity::from_legacy_name(table).map_err(SQLError::Internal)?;
122    super::generated::bind_schema_column_references(expression, &relation.name);
123    Ok(())
124}
125
126pub fn bind_stored_check_expression_routines(
127    context: &SchemaBindingContext<'_, '_>,
128    table: &str,
129    qualifier: &str,
130    columns: &[ColumnDef],
131    expression: &mut Expr,
132) -> Result<bool, SQLError> {
133    let typed_expression = bind_expr(
134        expression,
135        &mut CheckConditionTypeResolver {
136            table,
137            qualifier,
138            columns,
139        },
140    )?;
141    super::defaults::bind_stored_schema_expression_routines(context, expression, typed_expression)
142}
143
144pub fn validate_foreign_key_definition(
145    local_table: &str,
146    local_columns: &[ColumnDef],
147    parent_table: &str,
148    parent_columns: &[ColumnDef],
149    parent_keys: &[TableKeyConstraint],
150    foreign_key: &ForeignKey,
151) -> Result<(), SQLError> {
152    if foreign_key.local_columns.is_empty()
153        || foreign_key.local_columns.len() != foreign_key.ref_columns.len()
154    {
155        return Err(invalid_foreign_key(format!(
156            "foreign key on relation \"{local_table}\" has mismatched local and referenced columns"
157        )));
158    }
159
160    let local_types = foreign_key
161        .local_columns
162        .iter()
163        .map(|name| {
164            local_columns
165                .iter()
166                .find(|column| column.name == *name)
167                .map(|column| &column.ty)
168                .ok_or_else(|| SQLError::UnknownColumn(format!("{local_table}.{name}")))
169        })
170        .collect::<Result<Vec<_>, _>>()?;
171    let parent_types = foreign_key
172        .ref_columns
173        .iter()
174        .map(|name| {
175            parent_columns
176                .iter()
177                .find(|column| column.name == *name)
178                .map(|column| &column.ty)
179                .ok_or_else(|| SQLError::UnknownColumn(format!("{parent_table}.{name}")))
180        })
181        .collect::<Result<Vec<_>, _>>()?;
182
183    let target_key = parent_keys.iter().find(|constraint| {
184        constraint.columns == foreign_key.ref_columns && constraint.without_overlaps
185    });
186    if foreign_key.period && target_key.is_none() {
187        return Err(invalid_foreign_key(format!(
188            "there is no primary key or unique constraint declared WITH WITHOUT OVERLAPS matching the referenced columns for table \"{parent_table}\""
189        )));
190    }
191
192    if foreign_key.period {
193        if foreign_key.local_columns.len() < 2 {
194            return Err(invalid_foreign_key(
195                "PERIOD foreign key must contain at least one ordinary column and one period column",
196            ));
197        }
198        let local_period = local_types.last().expect("non-empty foreign key");
199        let parent_period = parent_types.last().expect("non-empty foreign key");
200        if !matches!(
201            local_period,
202            ColumnType::Range(_) | ColumnType::Multirange(_)
203        ) || local_period != parent_period
204        {
205            return Err(SQLError::Routine {
206                sqlstate: "42804".into(),
207                message: format!(
208                    "PERIOD columns \"{}\" and \"{}\" have incompatible types {} and {}",
209                    foreign_key
210                        .local_columns
211                        .last()
212                        .expect("non-empty foreign key"),
213                    foreign_key
214                        .ref_columns
215                        .last()
216                        .expect("non-empty foreign key"),
217                    local_period.sql_name(),
218                    parent_period.sql_name()
219                ),
220            });
221        }
222    }
223
224    Ok(())
225}
226
227fn invalid_foreign_key(message: impl Into<String>) -> SQLError {
228    SQLError::Routine {
229        sqlstate: "42830".into(),
230        message: message.into(),
231    }
232}
uqa_sql/schema/constraints.rs

uqa_sql/schema/
constraints.rs
