uqa_sql/schema/
constraint_metadata.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Normalize durable constraint names and identities through a caller-owned identity allocator.
8use std::collections::BTreeSet;
9use uqa_core::RelationIdentity;
10pub mod identity;
11
12/// Explicit names share their relation's event namespace; automatic names also avoid every constraint in the containing schema.
13#[derive(Default)]
14pub struct ConstraintNameScope {
15    pub events: BTreeSet<String>,
16    pub schema: BTreeSet<String>,
17}
18
19#[derive(Debug)]
20pub enum ConstraintMetadataError {
21    Invalid(String),
22    Execution(Box<crate::SQLError>),
23}
24
25impl std::fmt::Display for ConstraintMetadataError {
26    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
27        match self {
28            Self::Invalid(message) => formatter.write_str(message),
29            Self::Execution(error) => std::fmt::Display::fmt(error, formatter),
30        }
31    }
32}
33
34impl std::error::Error for ConstraintMetadataError {
35    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
36        match self {
37            Self::Invalid(_) => None,
38            Self::Execution(error) => Some(error.as_ref()),
39        }
40    }
41}
42impl ConstraintMetadataError {
43    /// The error as a statement reports it: an invalid identity is internal, an execution error is itself.
44    pub fn into_sql_error(self) -> crate::SQLError {
45        match self {
46            Self::Invalid(message) => crate::SQLError::Internal(message),
47            Self::Execution(error) => *error,
48        }
49    }
50}
51
52pub type ConstraintMetadataResult<T> = Result<T, ConstraintMetadataError>;
53pub type CatalogIdentityAllocator<'a> = dyn CatalogObjectAllocator + 'a;
54
55#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
56pub enum CatalogOidClass {
57    Constraint,
58    Relation,
59    /// `pg_type` rows, including generated array types.
60    Type,
61    /// `pg_enum` label rows.
62    EnumLabel,
63    /// `pg_rewrite` rows: user rules and the `_RETURN` rules of views.
64    Rewrite,
65    /// `pg_proc` rows.
66    Procedure,
67    /// `pg_attrdef` rows: column defaults and generation expressions.
68    AttributeDefault,
69    /// `pg_trigger` rows.
70    Trigger,
71}
72
73impl CatalogOidClass {
74    pub const fn class_id(self) -> u32 {
75        match self {
76            Self::Constraint => 2606,
77            Self::Relation => 1259,
78            Self::Type => 1247,
79            Self::EnumLabel => 3501,
80            Self::Rewrite => 2618,
81            Self::Procedure => 1255,
82            Self::AttributeDefault => 2604,
83            Self::Trigger => 2620,
84        }
85    }
86
87    pub const fn label(self) -> &'static str {
88        match self {
89            Self::Constraint => "constraint",
90            Self::Relation => "relation",
91            Self::Type => "type",
92            Self::EnumLabel => "enum label",
93            Self::Rewrite => "rule",
94            Self::Procedure => "function",
95            Self::AttributeDefault => "default",
96            Self::Trigger => "trigger",
97        }
98    }
99}
100
101/// Declaration normalization requests identities from its caller. Execution reserves public addresses; isolated declarations and initial migration can derive candidates before validating the complete catalog.
102pub trait CatalogObjectAllocator {
103    /// Validate the owning relation and reserve supplied addresses before allocating another row.
104    fn include_catalog_identity(
105        &mut self,
106        _relation: &RelationIdentity,
107        _class: CatalogOidClass,
108        _identity: crate::ast::ConstraintCatalogIdentity,
109    ) -> ConstraintMetadataResult<()> {
110        Ok(())
111    }
112
113    fn allocate_object_id(&mut self, kind: &str) -> ConstraintMetadataResult<[u8; 16]>;
114
115    fn allocate_catalog_oid(
116        &mut self,
117        class: CatalogOidClass,
118        object_id: &[u8; 16],
119    ) -> ConstraintMetadataResult<i64>;
120}
121
122impl<F> CatalogObjectAllocator for F
123where
124    F: FnMut(&str) -> ConstraintMetadataResult<[u8; 16]>,
125{
126    fn allocate_object_id(&mut self, kind: &str) -> ConstraintMetadataResult<[u8; 16]> {
127        self(kind)
128    }
129
130    fn allocate_catalog_oid(
131        &mut self,
132        class: CatalogOidClass,
133        object_id: &[u8; 16],
134    ) -> ConstraintMetadataResult<i64> {
135        Ok(crate::catalog::oids::stable_object_oid(
136            class.label(),
137            object_id,
138        ))
139    }
140}
141
142pub fn materialize_constraint_metadata(
143    relation: &RelationIdentity,
144    columns: &mut [crate::ast::ColumnDef],
145    constraints: &mut crate::ast::TableConstraintSet,
146    allocate: &mut CatalogIdentityAllocator<'_>,
147) -> ConstraintMetadataResult<bool> {
148    materialize_constraint_metadata_with_names(
149        relation,
150        columns,
151        constraints,
152        allocate,
153        &ConstraintNameScope::default(),
154    )
155}
156
157/// Validate explicit local names before excluding schema-wide names from automatic selection.
158pub fn materialize_constraint_metadata_with_names(
159    relation: &RelationIdentity,
160    columns: &mut [crate::ast::ColumnDef],
161    constraints: &mut crate::ast::TableConstraintSet,
162    allocate: &mut CatalogIdentityAllocator<'_>,
163    names: &ConstraintNameScope,
164) -> ConstraintMetadataResult<bool> {
165    identity::claims::validate_present_identities(columns, constraints)?;
166    for identity in identity::claims::identities(columns, constraints) {
167        allocate.include_catalog_identity(relation, CatalogOidClass::Constraint, identity)?;
168    }
169    // Releases predating typed table-key persistence stored column-level PRIMARY KEY and UNIQUE declarations only as ColumnDef flags. Promote those legacy flags before assigning names so catalog publication always sees named constraints.
170    let mut changed =
171        crate::catalog::relation_attributes::materialize(columns, &constraints.dropped_attributes)
172            .map_err(|error| ConstraintMetadataError::Execution(Box::new(error)))?;
173    changed |= materialize_column_key_constraints(columns, constraints);
174    let mut used = constraint_names_for_assignment(relation, columns, constraints, names)?;
175
176    let mut column_object_ids = BTreeSet::new();
177    for column in columns.iter_mut() {
178        if column
179            .object_id
180            .is_some_and(|object_id| !column_object_ids.insert(object_id))
181        {
182            column.object_id = None;
183        }
184        changed |= assign_catalog_object_id(&mut column.object_id, "column", allocate)?;
185        if let Some(object_id) = column.object_id {
186            column_object_ids.insert(object_id);
187        }
188    }
189    // `DefineRelation` stores the defaults and generation expressions, then the CHECK constraints of the columns and of the table, then the NOT NULL constraints; key indexes and foreign keys follow the relation.
190    for column in columns.iter_mut() {
191        changed |= identity::materialize_default_oid(column, allocate)?;
192    }
193    for column in columns.iter_mut() {
194        if let Some(check) = &column.check {
195            changed |= assign_check_name(&relation.name, check, &mut column.check_name, &mut used)?;
196            changed |= materialize_check_identity(
197                &mut column.check_object_id,
198                &mut column.check_catalog_oid,
199                allocate,
200            )?;
201        }
202    }
203    changed |= materialize_checks(relation, &mut constraints.checks, &mut used, allocate)?;
204    for column in columns.iter_mut() {
205        if column.not_null {
206            changed |= assign_constraint_name(
207                &mut column.not_null_name,
208                (&relation.name, &column.name, "not_null"),
209                &mut used,
210            )?;
211            changed |= identity::materialize_not_null_identity(column, allocate)?;
212        }
213    }
214    for constraint in &mut constraints.key_constraints {
215        let (component, label) = match constraint.kind {
216            crate::ast::TableKeyConstraintKind::PrimaryKey => (String::new(), "pkey"),
217            crate::ast::TableKeyConstraintKind::Unique => (
218                constraint_column_component(&constraint.columns, relation)?,
219                "key",
220            ),
221        };
222        changed |= assign_constraint_name(
223            &mut constraint.name,
224            (&relation.name, &component, label),
225            &mut used,
226        )?;
227        changed |= identity::materialize_key_identity(constraint, allocate)?;
228    }
229    for column in columns.iter_mut() {
230        if let Some(reference) = &mut column.references {
231            changed |= assign_constraint_name(
232                &mut reference.name,
233                (&relation.name, &column.name, "fkey"),
234                &mut used,
235            )?;
236            changed |= materialize_foreign_key_identity(
237                &mut reference.object_id,
238                &mut reference.catalog_identity,
239                allocate,
240            )?;
241        }
242    }
243    changed |= synchronize_partition_inherited_foreign_key_ids(constraints);
244    for constraint in &mut constraints.foreign_keys {
245        let component = constraint_column_component(&constraint.local_columns, relation)?;
246        changed |= assign_constraint_name(
247            &mut constraint.name,
248            (&relation.name, &component, "fkey"),
249            &mut used,
250        )?;
251        changed |= materialize_foreign_key_identity(
252            &mut constraint.object_id,
253            &mut constraint.catalog_identity,
254            allocate,
255        )?;
256    }
257    changed |= synchronize_partition_inherited_foreign_key_ids(constraints);
258    changed |= identity::keys::synchronize_provenance(constraints);
259    identity::claims::validate_constraint_identities(columns, constraints)?;
260    Ok(changed)
261}
262
263fn materialize_checks(
264    relation: &RelationIdentity,
265    checks: &mut [crate::ast::TableCheck],
266    used: &mut BTreeSet<String>,
267    allocate: &mut CatalogIdentityAllocator<'_>,
268) -> ConstraintMetadataResult<bool> {
269    let mut changed = false;
270    for constraint in checks {
271        changed |= assign_check_name(&relation.name, &constraint.expr, &mut constraint.name, used)?;
272        changed |= materialize_check_identity(
273            &mut constraint.object_id,
274            &mut constraint.catalog_oid,
275            allocate,
276        )?;
277    }
278    Ok(changed)
279}
280
281/// `StoreRelCheck`: a CHECK constraint takes its incarnation and its `pg_constraint` OID when it is stored and keeps them afterwards.
282pub fn materialize_check_identity(
283    object_id: &mut Option<[u8; 16]>,
284    catalog_oid: &mut Option<i64>,
285    allocate: &mut CatalogIdentityAllocator<'_>,
286) -> ConstraintMetadataResult<bool> {
287    let mut changed = assign_catalog_object_id(object_id, "CHECK constraint", allocate)?;
288    changed |= identity::materialize_check_oid(*object_id, catalog_oid, allocate)?;
289    Ok(changed)
290}
291
292/// `CreateConstraintEntry` for a foreign key: the constraint row's incarnation and OID, allocated when the constraint is created and kept afterwards.
293pub fn materialize_foreign_key_identity(
294    object_id: &mut Option<[u8; 16]>,
295    catalog_identity: &mut Option<crate::ast::ConstraintCatalogIdentity>,
296    allocate: &mut CatalogIdentityAllocator<'_>,
297) -> ConstraintMetadataResult<bool> {
298    let mut changed = assign_constraint_object_id(object_id, allocate)?;
299    changed |= identity::foreign_keys::materialize(catalog_identity, allocate)?;
300    Ok(changed)
301}
302
303fn constraint_names_for_assignment(
304    relation: &RelationIdentity,
305    columns: &[crate::ast::ColumnDef],
306    constraints: &crate::ast::TableConstraintSet,
307    names: &ConstraintNameScope,
308) -> ConstraintMetadataResult<BTreeSet<String>> {
309    let mut used = BTreeSet::new();
310    for column in columns {
311        record_constraint_name(relation, &mut used, column.not_null_name.as_deref())?;
312        record_constraint_name(relation, &mut used, column.check_name.as_deref())?;
313        record_constraint_name(
314            relation,
315            &mut used,
316            column
317                .references
318                .as_ref()
319                .and_then(|reference| reference.name.as_deref()),
320        )?;
321    }
322    for constraint in &constraints.key_constraints {
323        record_constraint_name(relation, &mut used, constraint.name.as_deref())?;
324    }
325    for constraint in &constraints.checks {
326        record_constraint_name(relation, &mut used, constraint.name.as_deref())?;
327    }
328    for constraint in &constraints.foreign_keys {
329        record_constraint_name(relation, &mut used, constraint.name.as_deref())?;
330    }
331    // The constraints a foreign key derives on referenced partitions share the relation's names.
332    for derived in columns
333        .iter()
334        .filter_map(|column| column.references.as_ref())
335        .flat_map(|reference| &reference.referenced_partitions)
336        .chain(
337            constraints
338                .foreign_keys
339                .iter()
340                .flat_map(|foreign_key| &foreign_key.referenced_partitions),
341        )
342    {
343        record_constraint_name(relation, &mut used, Some(&derived.name))?;
344    }
345    for name in &names.events {
346        if !used.insert(name.clone()) {
347            return Err(duplicate_constraint(relation, name));
348        }
349    }
350    used.extend(names.schema.iter().cloned());
351    Ok(used)
352}
353
354/// The first of `base_1`, `base_2`, ... that no constraint in `used` holds, which then holds it, as `PostgreSQL`'s `ChooseConstraintName` chooses a name with an empty label for a base that a constraint of the schema already holds.
355pub fn choose_suffixed_constraint_name(
356    base: &str,
357    used: &mut BTreeSet<String>,
358) -> ConstraintMetadataResult<String> {
359    for suffix in 1_u64.. {
360        let candidate = super::indexes::names::object_name(base, "", &suffix.to_string());
361        if used.insert(candidate.clone()) {
362            return Ok(candidate);
363        }
364    }
365    Err(ConstraintMetadataError::Invalid(format!(
366        "constraint name suffix space exhausted for `{base}`"
367    )))
368}
369
370pub fn materialize_column_key_constraints(
371    columns: &[crate::ast::ColumnDef],
372    constraints: &mut crate::ast::TableConstraintSet,
373) -> bool {
374    let mut changed = false;
375    for column in columns {
376        for (present, kind) in [
377            (
378                column.primary_key,
379                crate::ast::TableKeyConstraintKind::PrimaryKey,
380            ),
381            (column.unique, crate::ast::TableKeyConstraintKind::Unique),
382        ] {
383            if !present
384                || constraints.key_constraints.iter().any(|constraint| {
385                    constraint.kind == kind
386                        && constraint.columns.as_slice() == [column.name.as_str()]
387                })
388            {
389                continue;
390            }
391            constraints
392                .key_constraints
393                .push(crate::ast::TableKeyConstraint {
394                    catalog_identity: None,
395                    index_identity: None,
396                    name: None,
397                    kind,
398                    columns: vec![column.name.clone()],
399                    included_columns: Vec::new(),
400                    nulls_not_distinct: false,
401                    without_overlaps: false,
402                });
403            changed = true;
404        }
405    }
406    changed
407}
408
409pub fn foreign_keys_match_without_object_id(
410    left: &crate::ast::ForeignKey,
411    right: &crate::ast::ForeignKey,
412) -> bool {
413    let mut left = left.clone();
414    let mut right = right.clone();
415    left.object_id = None;
416    right.object_id = None;
417    left.catalog_identity = None;
418    right.catalog_identity = None;
419    left == right
420}
421
422/// Attachment provenance tracks one local row even after its name or enforcement flags change. Legacy entries may still lack the independent catalog identity.
423pub fn foreign_key_provenance_matches(
424    left: &crate::ast::ForeignKey,
425    right: &crate::ast::ForeignKey,
426) -> bool {
427    match (left.catalog_identity, right.catalog_identity) {
428        (Some(left), Some(right)) => left == right,
429        _ => {
430            (left.object_id.is_some() && left.object_id == right.object_id)
431                || foreign_keys_match_without_object_id(left, right)
432        }
433    }
434}
435
436pub fn synchronize_partition_inherited_foreign_key_ids(
437    constraints: &mut crate::ast::TableConstraintSet,
438) -> bool {
439    let mut changed = false;
440    for inherited_index in 0..constraints.hierarchy.partition_inherited_foreign_keys.len() {
441        let inherited = &constraints.hierarchy.partition_inherited_foreign_keys[inherited_index];
442        let Some(foreign_key_index) = constraints
443            .foreign_keys
444            .iter()
445            .position(|foreign_key| foreign_key_provenance_matches(foreign_key, inherited))
446        else {
447            continue;
448        };
449        let object_id = constraints.foreign_keys[foreign_key_index]
450            .object_id
451            .or(inherited.object_id);
452        if constraints.foreign_keys[foreign_key_index].object_id != object_id {
453            constraints.foreign_keys[foreign_key_index].object_id = object_id;
454            changed = true;
455        }
456        if constraints.hierarchy.partition_inherited_foreign_keys[inherited_index].object_id
457            != object_id
458        {
459            constraints.hierarchy.partition_inherited_foreign_keys[inherited_index].object_id =
460                object_id;
461            changed = true;
462        }
463        let catalog_identity = constraints.foreign_keys[foreign_key_index].catalog_identity;
464        if constraints.hierarchy.partition_inherited_foreign_keys[inherited_index].catalog_identity
465            != catalog_identity
466        {
467            constraints.hierarchy.partition_inherited_foreign_keys[inherited_index]
468                .catalog_identity = catalog_identity;
469            changed = true;
470        }
471    }
472    changed
473}
474
475fn assign_constraint_object_id(
476    target: &mut Option<[u8; 16]>,
477    allocate: &mut CatalogIdentityAllocator<'_>,
478) -> ConstraintMetadataResult<bool> {
479    assign_catalog_object_id(target, "foreign-key constraint", allocate)
480}
481
482fn assign_catalog_object_id(
483    target: &mut Option<[u8; 16]>,
484    object_kind: &str,
485    allocate: &mut CatalogIdentityAllocator<'_>,
486) -> ConstraintMetadataResult<bool> {
487    if target.is_some() {
488        return Ok(false);
489    }
490    *target = Some(allocate.allocate_object_id(object_kind)?);
491    Ok(true)
492}
493
494fn record_constraint_name(
495    relation: &RelationIdentity,
496    used: &mut BTreeSet<String>,
497    name: Option<&str>,
498) -> ConstraintMetadataResult<()> {
499    let Some(name) = name else {
500        return Ok(());
501    };
502    if name.is_empty() {
503        return Err(ConstraintMetadataError::Invalid(
504            "constraint name must not be empty".into(),
505        ));
506    }
507    if !used.insert(name.to_string()) {
508        return Err(duplicate_constraint(relation, name));
509    }
510    Ok(())
511}
512
513fn duplicate_constraint(relation: &RelationIdentity, name: &str) -> ConstraintMetadataError {
514    ConstraintMetadataError::Execution(Box::new(
515        crate::schema::constraint_changes::constraint_error(
516            "42710",
517            format!(
518                "constraint \"{name}\" for relation \"{}\" already exists",
519                relation.name
520            ),
521        ),
522    ))
523}
524
525/// Name an unnamed CHECK as `AddRelationNewConstraints` does, wherever the statement wrote it: after the relation, the one column the expression references when it references exactly one, then `check`, unique among `used` as `ChooseConstraintName` makes it.
526pub(super) fn assign_check_name(
527    relation: &str,
528    expression: &crate::ast::Expr,
529    target: &mut Option<String>,
530    used: &mut BTreeSet<String>,
531) -> ConstraintMetadataResult<bool> {
532    let mut referenced_columns = Vec::new();
533    collect_constraint_columns(expression, &mut referenced_columns);
534    let component = match referenced_columns.as_slice() {
535        [column] => column.as_str(),
536        _ => "",
537    };
538    assign_constraint_name(target, (relation, component, "check"), used)
539}
540
541pub(super) fn assign_constraint_name(
542    target: &mut Option<String>,
543    parts: (&str, &str, &str),
544    used: &mut BTreeSet<String>,
545) -> ConstraintMetadataResult<bool> {
546    if target.is_some() {
547        return Ok(false);
548    }
549    let base = super::indexes::names::object_name(parts.0, parts.1, parts.2);
550    if used.insert(base.clone()) {
551        *target = Some(base);
552        return Ok(true);
553    }
554    for suffix in 1_u64.. {
555        let label = format!("{}{suffix}", parts.2);
556        let candidate = super::indexes::names::object_name(parts.0, parts.1, &label);
557        if used.insert(candidate.clone()) {
558            *target = Some(candidate);
559            return Ok(true);
560        }
561    }
562    Err(ConstraintMetadataError::Invalid(format!(
563        "constraint name suffix space exhausted for `{base}`"
564    )))
565}
566
567fn constraint_column_component(
568    columns: &[String],
569    relation: &RelationIdentity,
570) -> ConstraintMetadataResult<String> {
571    if columns.is_empty() {
572        return Err(ConstraintMetadataError::Invalid(format!(
573            "constraint on table `{}` has no columns",
574            relation.qualified_name()
575        )));
576    }
577    Ok(columns.join("_"))
578}
579
580#[expect(
581    clippy::too_many_lines,
582    reason = "one traversal covers every constraint expression child"
583)]
584fn collect_constraint_columns(expression: &crate::ast::Expr, output: &mut Vec<String>) {
585    use crate::ast::{Expr, FrameBound};
586    match expression {
587        Expr::Column(name) | Expr::QualifiedColumn { column: name, .. } => {
588            if !output.contains(name) {
589                output.push(name.clone());
590            }
591        }
592        Expr::Func {
593            args,
594            order_by,
595            filter,
596            ..
597        } => {
598            for argument in args {
599                collect_constraint_columns(argument, output);
600            }
601            for order in order_by {
602                collect_constraint_columns(&order.expr, output);
603            }
604            if let Some(filter) = filter {
605                collect_constraint_columns(filter, output);
606            }
607        }
608        Expr::Array(items)
609        | Expr::Row(items)
610        | Expr::CompositeRow { items, .. }
611        | Expr::And(items)
612        | Expr::Or(items) => {
613            for item in items {
614                collect_constraint_columns(item, output);
615            }
616        }
617        Expr::Binary { lhs, rhs, .. } => {
618            collect_constraint_columns(lhs, output);
619            collect_constraint_columns(rhs, output);
620        }
621        Expr::Not(inner)
622        | Expr::UnaryMinus(inner)
623        | Expr::IsNull { expr: inner, .. }
624        | Expr::Cast { expr: inner, .. } => {
625            collect_constraint_columns(inner, output);
626        }
627        Expr::Between { expr, low, high } => {
628            collect_constraint_columns(expr, output);
629            collect_constraint_columns(low, output);
630            collect_constraint_columns(high, output);
631        }
632        Expr::InList { expr, list, .. } => {
633            collect_constraint_columns(expr, output);
634            for item in list {
635                collect_constraint_columns(item, output);
636            }
637        }
638        Expr::WindowCall {
639            args, spec, filter, ..
640        } => {
641            for expression in args
642                .iter()
643                .chain(filter.as_deref())
644                .chain(&spec.partition_by)
645                .chain(spec.order_by.iter().map(|order| &order.expr))
646            {
647                collect_constraint_columns(expression, output);
648            }
649            for bound in spec
650                .frame
651                .iter()
652                .flat_map(|frame| [&frame.start, &frame.end])
653            {
654                if let FrameBound::Preceding(expression) | FrameBound::Following(expression) = bound
655                {
656                    collect_constraint_columns(expression, output);
657                }
658            }
659        }
660        Expr::Case {
661            base,
662            when,
663            else_branch,
664        } => {
665            if let Some(base) = base {
666                collect_constraint_columns(base, output);
667            }
668            for (condition, result) in when {
669                collect_constraint_columns(condition, output);
670                collect_constraint_columns(result, output);
671            }
672            if let Some(else_branch) = else_branch {
673                collect_constraint_columns(else_branch, output);
674            }
675        }
676        Expr::InSubquery { expr, .. } => collect_constraint_columns(expr, output),
677        Expr::Default
678        | Expr::Star
679        | Expr::QualifiedStar(_)
680        | Expr::InternalColumn(_)
681        | Expr::Literal(_)
682        | Expr::TypedLiteral { .. }
683        | Expr::Param(_)
684        | Expr::ScalarSubquery(_)
685        | Expr::Exists { .. } => {}
686    }
687}
uqa_sql/schema/constraint_metadata.rs

uqa_sql/schema/
constraint_metadata.rs
