uqa_sql/schema/
check_inheritance.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! CHECK definition merging at CREATE and ALTER inheritance boundaries.
8
9use crate::ast::{ColumnDef, ColumnType, Expr, TableCheck};
10use crate::SQLError;
11use crate::ScalarExpr;
12use uqa_core::Value;
13
14pub fn bind_parent_check_columns(parent: &str, check: &mut TableCheck) -> Result<(), SQLError> {
15    let relation =
16        uqa_core::RelationIdentity::from_legacy_name(parent).map_err(SQLError::Internal)?;
17    crate::catalog::stored_ast::visit_stored_expression(&mut check.expr, &mut |node| {
18        if matches!(node, Expr::Star | Expr::QualifiedStar(_)) {
19            return Err(SQLError::Diagnostic {
20                sqlstate: "0A000".into(),
21                message: "cannot convert whole-row table reference".into(),
22                detail: Some(format!(
23                    "Constraint \"{}\" contains a whole-row reference to table \"{}\".",
24                    check.name.as_deref().unwrap_or("<unnamed>"),
25                    relation.name,
26                )),
27                hint: None,
28            });
29        }
30        Ok(())
31    })?;
32    crate::schema::generated::bind_schema_column_references(&mut check.expr, parent);
33    crate::schema::generated::bind_schema_column_references(&mut check.expr, &relation.name);
34    Ok(())
35}
36
37/// Remove casts that parse analysis treats as an identity before storing or comparing CHECK syntax.
38pub(super) fn remove_identity_casts(
39    expression: &mut Expr,
40    columns: &[ColumnDef],
41) -> Result<(), SQLError> {
42    crate::catalog::stored_ast::visit_stored_expression(expression, &mut |node| {
43        while let Expr::Cast { expr, ty, .. } = node {
44            let Ok(target) = ColumnType::from_sql_name(ty) else {
45                break;
46            };
47            let source = match expr.as_ref() {
48                Expr::Column(name) => columns
49                    .iter()
50                    .find(|column| column.name == *name)
51                    .map(|column| column.ty.clone()),
52                Expr::TypedLiteral { ty, .. } => ColumnType::from_sql_name(ty).ok(),
53                Expr::Literal(_) => crate::scalar_type(
54                    &crate::plan::ExpressionPlan::lower(*expr.clone()).scalar,
55                    &crate::RowSchema::default(),
56                    &[],
57                )?,
58                _ => None,
59            };
60            if source.as_ref() != Some(&target) {
61                break;
62            }
63            *node = *expr.clone();
64        }
65        Ok(())
66    })
67}
68
69pub fn same_check_expression(
70    left: &Expr,
71    right: &Expr,
72    columns: &[ColumnDef],
73) -> Result<bool, SQLError> {
74    fn canonical(expression: &Expr, columns: &[ColumnDef]) -> Result<ScalarExpr, SQLError> {
75        let mut expression = expression.clone();
76        remove_identity_casts(&mut expression, columns)?;
77        let mut scalar = crate::plan::ExpressionPlan::lower(expression).scalar;
78        let mut failure = None;
79        crate::plan::rewrite_scalar_expression(&mut scalar, &mut |node| {
80            if let ScalarExpr::TypedLiteral {
81                value: Value::Int(value),
82                ty,
83                parameter_index: None,
84                ..
85            } = node
86            {
87                if matches!(ColumnType::from_sql_name(ty), Ok(ColumnType::Integer))
88                    && i32::try_from(*value).is_ok()
89                {
90                    *node = ScalarExpr::Literal(Value::Int(*value));
91                }
92                return;
93            }
94            let ScalarExpr::Cast { expr, ty, implicit } = node else {
95                return;
96            };
97            // PostgreSQL expression equality ignores the display form of a coercion.
98            *implicit = false;
99            let Ok(target) = ColumnType::from_sql_name(ty) else {
100                return;
101            };
102            if let ScalarExpr::Column(name) = expr.as_ref() {
103                if columns
104                    .iter()
105                    .any(|column| column.name == *name && column.ty == target)
106                {
107                    *node = *expr.clone();
108                }
109            } else if let ScalarExpr::Literal(value @ Value::Str(_)) = expr.as_ref() {
110                // PostgreSQL resolves an unknown string to an integer constant during analysis. Keep wider and narrower integer coercions distinct from the ordinary int4 literal.
111                if target == ColumnType::Integer {
112                    match crate::expr::cast_value(value, ty) {
113                        Ok(value) => *node = ScalarExpr::Literal(value),
114                        Err(error) => failure = Some(error),
115                    }
116                }
117            }
118        });
119        if let Some(error) = failure {
120            return Err(error);
121        }
122        Ok(scalar)
123    }
124    Ok(canonical(left, columns)? == canonical(right, columns)?)
125}
126
127pub fn duplicate_check(table: &str, name: &str) -> SQLError {
128    error(
129        "42710",
130        format!("constraint \"{name}\" for relation \"{table}\" already exists"),
131    )
132}
133
134fn error(sqlstate: &str, message: String) -> SQLError {
135    SQLError::Routine {
136        sqlstate: sqlstate.into(),
137        message,
138    }
139}
140
141/// The caller decides whether a local or inherited duplicate is eligible to merge. Existing validation and enforcement states follow `PostgreSQL`'s directional merge rules.
142pub fn validate_check_merge(
143    table: &str,
144    existing: &TableCheck,
145    incoming: &TableCheck,
146    columns: &[ColumnDef],
147) -> Result<(), SQLError> {
148    let name = incoming.name.as_deref().unwrap_or("<unnamed>");
149    if !same_check_expression(&existing.expr, &incoming.expr, columns)? {
150        return Err(duplicate_check(table, name));
151    }
152    let conflict = if existing.no_inherit {
153        Some("non-inherited")
154    } else if incoming.no_inherit {
155        Some("inherited")
156    } else if incoming.validated && existing.enforced && !existing.validated {
157        Some("NOT VALID")
158    } else if (!incoming.is_local && incoming.enforced && !existing.enforced)
159        || (incoming.is_local && !incoming.enforced && existing.enforced)
160    {
161        Some("NOT ENFORCED")
162    } else {
163        None
164    };
165    if let Some(conflict) = conflict {
166        return Err(error("42P17", format!("constraint \"{name}\" conflicts with {conflict} constraint on relation \"{table}\"")));
167    }
168    Ok(())
169}
170
171/// `MergeCheckConstraint`: a parent's CHECK joins the constraints the new table inherits, merging with an earlier parent's CHECK of the same name when their expressions match, an enforced copy making the merged constraint enforced.
172pub fn merge_inherited_check(
173    inherited: &mut Vec<TableCheck>,
174    check: TableCheck,
175    columns: &[ColumnDef],
176) -> Result<(), SQLError> {
177    let Some(existing) = inherited
178        .iter_mut()
179        .find(|existing| existing.name.is_some() && existing.name == check.name)
180    else {
181        inherited.push(check);
182        return Ok(());
183    };
184    if !same_check_expression(&existing.expr, &check.expr, columns)? {
185        return Err(error(
186            "42710",
187            format!(
188                "check constraint name \"{}\" appears multiple times but with different expressions",
189                check.name.as_deref().unwrap_or("<unnamed>")
190            ),
191        ));
192    }
193    existing.enforced |= check.enforced;
194    existing.validated = existing.enforced;
195    Ok(())
196}
197
198#[cfg(test)]
199mod tests {
200    use super::*;
201
202    #[test]
203    fn inherited_checks_compare_cooked_int4_constants_without_erasing_other_types() {
204        let untyped = Expr::Literal(Value::Int(0));
205        for ty in ["integer", "int4"] {
206            let cooked = Expr::TypedLiteral {
207                collation_oid: None,
208                composite_source: None,
209                value: Value::Int(0),
210                ty: ty.into(),
211            };
212            assert!(same_check_expression(&untyped, &cooked, &[]).unwrap());
213            assert!(same_check_expression(&cooked, &untyped, &[]).unwrap());
214            let mut cast = Expr::Cast {
215                implicit: false,
216                expr: Box::new(cooked.clone()),
217                ty: "integer".into(),
218            };
219            assert!(same_check_expression(&cast, &untyped, &[]).unwrap());
220            remove_identity_casts(&mut cast, &[]).unwrap();
221            assert_eq!(cast, cooked);
222        }
223        for ty in ["smallint", "bigint", "oid"] {
224            let cooked = Expr::TypedLiteral {
225                collation_oid: None,
226                composite_source: None,
227                value: Value::Int(0),
228                ty: ty.into(),
229            };
230            assert!(!same_check_expression(&untyped, &cooked, &[]).unwrap());
231        }
232    }
233    #[test]
234    fn inherited_check_equality_ignores_coercion_display_origin() {
235        let implicit = Expr::Cast {
236            implicit: true,
237            expr: Box::new(Expr::Column("value".into())),
238            ty: "bigint".into(),
239        };
240        let mut explicit = implicit.clone();
241        let Expr::Cast {
242            implicit: origin, ..
243        } = &mut explicit
244        else {
245            unreachable!()
246        };
247        *origin = false;
248        assert!(same_check_expression(&implicit, &explicit, &[]).unwrap());
249    }
250}
uqa_sql/schema/check_inheritance.rs

uqa_sql/schema/
check_inheritance.rs
