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.catalog_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    // `cookConstraint` reads the objects a CHECK names as `reg*` constants when the constraint is defined.
81    super::dependencies::oid_alias::read_oid_alias_constants(context.catalog, expression)?;
82    let bound = bind_expr(
83        expression,
84        &mut CheckConditionTypeResolver {
85            table,
86            qualifier,
87            columns,
88        },
89    )?;
90    let lowered = crate::plan::ExpressionPlan::lower(bound);
91    if !lowered.subqueries.is_empty() {
92        return Err(SQLError::Routine {
93            sqlstate: "0A000".into(),
94            message: "cannot use subquery in check constraint".into(),
95        });
96    }
97    match crate::type_resolution::common_context_expression_type(
98        &lowered.scalar,
99        &crate::RowSchema::default(),
100        &[],
101        Some(context.catalog),
102    )? {
103        Some(ty) if !is_boolean_type(&ty) => Err(SQLError::Routine {
104            sqlstate: "42804".into(),
105            message: format!(
106                "argument of CHECK must be type boolean, not type {}",
107                ty.sql_name()
108            ),
109        }),
110        None => {
111            if let Expr::Literal(value @ (Value::Str(_) | Value::FixedChar(_))) = expression {
112                *value = crate::expr::cast_value(value, "boolean")?;
113            } else {
114                *expression = Expr::Cast {
115                    implicit: true,
116                    expr: Box::new(expression.clone()),
117                    ty: "boolean".into(),
118                };
119            }
120            Ok(())
121        }
122        Some(_) => Ok(()),
123    }?;
124    bind_stored_check_expression(context, table, qualifier, columns, expression)?;
125    super::generated::bind_schema_column_references(expression, qualifier);
126    super::generated::bind_schema_column_references(expression, table);
127    let relation = crate::RelationIdentity::from_legacy_name(table).map_err(SQLError::Internal)?;
128    super::generated::bind_schema_column_references(expression, &relation.name);
129    super::check_inheritance::remove_identity_casts(expression, columns)
130}
131
132pub fn bind_stored_check_expression(
133    context: &SchemaBindingContext<'_, '_>,
134    table: &str,
135    qualifier: &str,
136    columns: &[ColumnDef],
137    expression: &mut Expr,
138) -> Result<bool, SQLError> {
139    let typed_expression = bind_expr(
140        expression,
141        &mut CheckConditionTypeResolver {
142            table,
143            qualifier,
144            columns,
145        },
146    )?;
147    super::defaults::bind_stored_schema_expression(context, expression, typed_expression)
148}
149
150pub fn validate_foreign_key_definition(
151    local_table: &str,
152    local_columns: &[ColumnDef],
153    parent_table: &str,
154    parent_columns: &[ColumnDef],
155    parent_keys: &[TableKeyConstraint],
156    foreign_key: &ForeignKey,
157) -> Result<(), SQLError> {
158    if foreign_key.local_columns.is_empty()
159        || foreign_key.local_columns.len() != foreign_key.ref_columns.len()
160    {
161        return Err(invalid_foreign_key(format!(
162            "foreign key on relation \"{local_table}\" has mismatched local and referenced columns"
163        )));
164    }
165
166    let local_types = foreign_key
167        .local_columns
168        .iter()
169        .map(|name| {
170            local_columns
171                .iter()
172                .find(|column| column.name == *name)
173                .map(|column| &column.ty)
174                .ok_or_else(|| SQLError::UnknownColumn(format!("{local_table}.{name}")))
175        })
176        .collect::<Result<Vec<_>, _>>()?;
177    let parent_types = foreign_key
178        .ref_columns
179        .iter()
180        .map(|name| {
181            parent_columns
182                .iter()
183                .find(|column| column.name == *name)
184                .map(|column| &column.ty)
185                .ok_or_else(|| SQLError::UnknownColumn(format!("{parent_table}.{name}")))
186        })
187        .collect::<Result<Vec<_>, _>>()?;
188
189    let target_key = parent_keys.iter().find(|constraint| {
190        constraint.columns == foreign_key.ref_columns && constraint.without_overlaps
191    });
192    if foreign_key.period && target_key.is_none() {
193        return Err(invalid_foreign_key(format!(
194            "there is no primary key or unique constraint declared WITH WITHOUT OVERLAPS matching the referenced columns for table \"{parent_table}\""
195        )));
196    }
197
198    if foreign_key.period {
199        if foreign_key.local_columns.len() < 2 {
200            return Err(invalid_foreign_key(
201                "PERIOD foreign key must contain at least one ordinary column and one period column",
202            ));
203        }
204        let local_period = local_types.last().expect("non-empty foreign key");
205        let parent_period = parent_types.last().expect("non-empty foreign key");
206        if !matches!(
207            local_period,
208            ColumnType::Range(_) | ColumnType::Multirange(_)
209        ) || local_period != parent_period
210        {
211            return Err(SQLError::Routine {
212                sqlstate: "42804".into(),
213                message: format!(
214                    "PERIOD columns \"{}\" and \"{}\" have incompatible types {} and {}",
215                    foreign_key
216                        .local_columns
217                        .last()
218                        .expect("non-empty foreign key"),
219                    foreign_key
220                        .ref_columns
221                        .last()
222                        .expect("non-empty foreign key"),
223                    local_period.sql_name(),
224                    parent_period.sql_name()
225                ),
226            });
227        }
228    }
229
230    Ok(())
231}
232
233fn invalid_foreign_key(message: impl Into<String>) -> SQLError {
234    SQLError::Routine {
235        sqlstate: "42830".into(),
236        message: message.into(),
237    }
238}
uqa_sql/schema/constraints.rs

uqa_sql/schema/
constraints.rs
