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}
42pub type ConstraintMetadataResult<T> = Result<T, ConstraintMetadataError>;
43pub type CatalogIdentityAllocator<'a> = dyn CatalogObjectAllocator + 'a;
44
45#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
46pub enum CatalogOidClass {
47    Constraint,
48    Relation,
49}
50
51impl CatalogOidClass {
52    pub const fn class_id(self) -> u32 {
53        match self {
54            Self::Constraint => 2606,
55            Self::Relation => 1259,
56        }
57    }
58
59    pub const fn label(self) -> &'static str {
60        match self {
61            Self::Constraint => "constraint",
62            Self::Relation => "relation",
63        }
64    }
65}
66
67/// Declaration normalization requests identities from its caller. Execution reserves public addresses; isolated declarations and initial migration can derive candidates before validating the complete catalog.
68pub trait CatalogObjectAllocator {
69    /// Validate the owning relation and reserve supplied addresses before allocating another row.
70    fn include_catalog_identity(
71        &mut self,
72        _relation: &RelationIdentity,
73        _class: CatalogOidClass,
74        _identity: crate::ast::ConstraintCatalogIdentity,
75    ) -> ConstraintMetadataResult<()> {
76        Ok(())
77    }
78
79    fn allocate_object_id(&mut self, kind: &str) -> ConstraintMetadataResult<[u8; 16]>;
80
81    fn allocate_catalog_oid(
82        &mut self,
83        class: CatalogOidClass,
84        object_id: &[u8; 16],
85    ) -> ConstraintMetadataResult<i64>;
86}
87
88impl<F> CatalogObjectAllocator for F
89where
90    F: FnMut(&str) -> ConstraintMetadataResult<[u8; 16]>,
91{
92    fn allocate_object_id(&mut self, kind: &str) -> ConstraintMetadataResult<[u8; 16]> {
93        self(kind)
94    }
95
96    fn allocate_catalog_oid(
97        &mut self,
98        class: CatalogOidClass,
99        object_id: &[u8; 16],
100    ) -> ConstraintMetadataResult<i64> {
101        Ok(crate::catalog::oids::stable_object_oid(
102            class.label(),
103            object_id,
104        ))
105    }
106}
107
108pub fn materialize_constraint_metadata(
109    relation: &RelationIdentity,
110    columns: &mut [crate::ast::ColumnDef],
111    constraints: &mut crate::ast::TableConstraintSet,
112    allocate: &mut CatalogIdentityAllocator<'_>,
113) -> ConstraintMetadataResult<bool> {
114    materialize_constraint_metadata_with_names(
115        relation,
116        columns,
117        constraints,
118        allocate,
119        &ConstraintNameScope::default(),
120    )
121}
122
123/// Validate explicit local names before excluding schema-wide names from automatic selection.
124pub fn materialize_constraint_metadata_with_names(
125    relation: &RelationIdentity,
126    columns: &mut [crate::ast::ColumnDef],
127    constraints: &mut crate::ast::TableConstraintSet,
128    allocate: &mut CatalogIdentityAllocator<'_>,
129    names: &ConstraintNameScope,
130) -> ConstraintMetadataResult<bool> {
131    identity::claims::validate_present_identities(columns, constraints)?;
132    for identity in identity::claims::identities(columns, constraints) {
133        allocate.include_catalog_identity(relation, CatalogOidClass::Constraint, identity)?;
134    }
135    // 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.
136    let mut changed = materialize_column_key_constraints(columns, constraints);
137    let mut used = constraint_names_for_assignment(relation, columns, constraints, names)?;
138
139    let mut column_object_ids = BTreeSet::new();
140    for column in columns.iter_mut() {
141        if column
142            .object_id
143            .is_some_and(|object_id| !column_object_ids.insert(object_id))
144        {
145            column.object_id = None;
146        }
147        changed |= assign_catalog_object_id(&mut column.object_id, "column", allocate)?;
148        if let Some(object_id) = column.object_id {
149            column_object_ids.insert(object_id);
150        }
151        if column.not_null {
152            changed |= assign_constraint_name(
153                &mut column.not_null_name,
154                (&relation.name, &column.name, "not_null"),
155                &mut used,
156            )?;
157            changed |= identity::materialize_not_null_identity(column, allocate)?;
158        }
159        if column.check.is_some() {
160            changed |= assign_constraint_name(
161                &mut column.check_name,
162                (&relation.name, &column.name, "check"),
163                &mut used,
164            )?;
165            changed |= assign_catalog_object_id(
166                &mut column.check_object_id,
167                "CHECK constraint",
168                allocate,
169            )?;
170            changed |= identity::materialize_check_oid(
171                column.check_object_id,
172                &mut column.check_catalog_oid,
173                allocate,
174            )?;
175        }
176        if let Some(reference) = &mut column.references {
177            changed |= assign_constraint_name(
178                &mut reference.name,
179                (&relation.name, &column.name, "fkey"),
180                &mut used,
181            )?;
182            changed |= assign_constraint_object_id(&mut reference.object_id, allocate)?;
183            changed |=
184                identity::foreign_keys::materialize(&mut reference.catalog_identity, allocate)?;
185        }
186    }
187    for constraint in &mut constraints.key_constraints {
188        let (component, label) = match constraint.kind {
189            crate::ast::TableKeyConstraintKind::PrimaryKey => (String::new(), "pkey"),
190            crate::ast::TableKeyConstraintKind::Unique => (
191                constraint_column_component(&constraint.columns, relation)?,
192                "key",
193            ),
194        };
195        changed |= assign_constraint_name(
196            &mut constraint.name,
197            (&relation.name, &component, label),
198            &mut used,
199        )?;
200        changed |= identity::materialize_key_identity(constraint, allocate)?;
201    }
202    changed |= materialize_checks(relation, &mut constraints.checks, &mut used, allocate)?;
203    changed |= synchronize_partition_inherited_foreign_key_ids(constraints);
204    for constraint in &mut constraints.foreign_keys {
205        let component = constraint_column_component(&constraint.local_columns, relation)?;
206        changed |= assign_constraint_name(
207            &mut constraint.name,
208            (&relation.name, &component, "fkey"),
209            &mut used,
210        )?;
211        changed |= assign_constraint_object_id(&mut constraint.object_id, allocate)?;
212        changed |= identity::foreign_keys::materialize(&mut constraint.catalog_identity, allocate)?;
213    }
214    changed |= synchronize_partition_inherited_foreign_key_ids(constraints);
215    changed |= identity::keys::synchronize_provenance(constraints);
216    identity::claims::validate_constraint_identities(columns, constraints)?;
217    Ok(changed)
218}
219
220fn materialize_checks(
221    relation: &RelationIdentity,
222    checks: &mut [crate::ast::TableCheck],
223    used: &mut BTreeSet<String>,
224    allocate: &mut CatalogIdentityAllocator<'_>,
225) -> ConstraintMetadataResult<bool> {
226    let mut changed = false;
227    for constraint in checks {
228        let mut referenced_columns = Vec::new();
229        collect_constraint_columns(&constraint.expr, &mut referenced_columns);
230        let component = if referenced_columns.len() == 1 {
231            referenced_columns[0].as_str()
232        } else {
233            ""
234        };
235        changed |= assign_constraint_name(
236            &mut constraint.name,
237            (&relation.name, component, "check"),
238            used,
239        )?;
240        changed |=
241            assign_catalog_object_id(&mut constraint.object_id, "CHECK constraint", allocate)?;
242        changed |= identity::materialize_check_oid(
243            constraint.object_id,
244            &mut constraint.catalog_oid,
245            allocate,
246        )?;
247    }
248    Ok(changed)
249}
250
251fn constraint_names_for_assignment(
252    relation: &RelationIdentity,
253    columns: &[crate::ast::ColumnDef],
254    constraints: &crate::ast::TableConstraintSet,
255    names: &ConstraintNameScope,
256) -> ConstraintMetadataResult<BTreeSet<String>> {
257    let mut used = BTreeSet::new();
258    for column in columns {
259        record_constraint_name(relation, &mut used, column.not_null_name.as_deref())?;
260        record_constraint_name(relation, &mut used, column.check_name.as_deref())?;
261        record_constraint_name(
262            relation,
263            &mut used,
264            column
265                .references
266                .as_ref()
267                .and_then(|reference| reference.name.as_deref()),
268        )?;
269    }
270    for constraint in &constraints.key_constraints {
271        record_constraint_name(relation, &mut used, constraint.name.as_deref())?;
272    }
273    for constraint in &constraints.checks {
274        record_constraint_name(relation, &mut used, constraint.name.as_deref())?;
275    }
276    for constraint in &constraints.foreign_keys {
277        record_constraint_name(relation, &mut used, constraint.name.as_deref())?;
278    }
279    // The constraints a foreign key derives on referenced partitions share the relation's names.
280    for derived in columns
281        .iter()
282        .filter_map(|column| column.references.as_ref())
283        .flat_map(|reference| &reference.referenced_partitions)
284        .chain(
285            constraints
286                .foreign_keys
287                .iter()
288                .flat_map(|foreign_key| &foreign_key.referenced_partitions),
289        )
290    {
291        record_constraint_name(relation, &mut used, Some(&derived.name))?;
292    }
293    for name in &names.events {
294        if !used.insert(name.clone()) {
295            return Err(duplicate_constraint(relation, name));
296        }
297    }
298    used.extend(names.schema.iter().cloned());
299    Ok(used)
300}
301
302/// 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.
303pub fn choose_suffixed_constraint_name(
304    base: &str,
305    used: &mut BTreeSet<String>,
306) -> ConstraintMetadataResult<String> {
307    for suffix in 1_u64.. {
308        let candidate = super::indexes::names::object_name(base, "", &suffix.to_string());
309        if used.insert(candidate.clone()) {
310            return Ok(candidate);
311        }
312    }
313    Err(ConstraintMetadataError::Invalid(format!(
314        "constraint name suffix space exhausted for `{base}`"
315    )))
316}
317
318pub fn materialize_column_key_constraints(
319    columns: &[crate::ast::ColumnDef],
320    constraints: &mut crate::ast::TableConstraintSet,
321) -> bool {
322    let mut changed = false;
323    for column in columns {
324        for (present, kind) in [
325            (
326                column.primary_key,
327                crate::ast::TableKeyConstraintKind::PrimaryKey,
328            ),
329            (column.unique, crate::ast::TableKeyConstraintKind::Unique),
330        ] {
331            if !present
332                || constraints.key_constraints.iter().any(|constraint| {
333                    constraint.kind == kind
334                        && constraint.columns.as_slice() == [column.name.as_str()]
335                })
336            {
337                continue;
338            }
339            constraints
340                .key_constraints
341                .push(crate::ast::TableKeyConstraint {
342                    catalog_identity: None,
343                    name: None,
344                    kind,
345                    columns: vec![column.name.clone()],
346                    included_columns: Vec::new(),
347                    nulls_not_distinct: false,
348                    without_overlaps: false,
349                });
350            changed = true;
351        }
352    }
353    changed
354}
355
356pub fn foreign_keys_match_without_object_id(
357    left: &crate::ast::ForeignKey,
358    right: &crate::ast::ForeignKey,
359) -> bool {
360    let mut left = left.clone();
361    let mut right = right.clone();
362    left.object_id = None;
363    right.object_id = None;
364    left.catalog_identity = None;
365    right.catalog_identity = None;
366    left == right
367}
368
369/// Attachment provenance tracks one local row even after its name or enforcement flags change. Legacy entries may still lack the independent catalog identity.
370pub fn foreign_key_provenance_matches(
371    left: &crate::ast::ForeignKey,
372    right: &crate::ast::ForeignKey,
373) -> bool {
374    match (left.catalog_identity, right.catalog_identity) {
375        (Some(left), Some(right)) => left == right,
376        _ => {
377            (left.object_id.is_some() && left.object_id == right.object_id)
378                || foreign_keys_match_without_object_id(left, right)
379        }
380    }
381}
382
383pub fn synchronize_partition_inherited_foreign_key_ids(
384    constraints: &mut crate::ast::TableConstraintSet,
385) -> bool {
386    let mut changed = false;
387    for inherited_index in 0..constraints.hierarchy.partition_inherited_foreign_keys.len() {
388        let inherited = &constraints.hierarchy.partition_inherited_foreign_keys[inherited_index];
389        let Some(foreign_key_index) = constraints
390            .foreign_keys
391            .iter()
392            .position(|foreign_key| foreign_key_provenance_matches(foreign_key, inherited))
393        else {
394            continue;
395        };
396        let object_id = constraints.foreign_keys[foreign_key_index]
397            .object_id
398            .or(inherited.object_id);
399        if constraints.foreign_keys[foreign_key_index].object_id != object_id {
400            constraints.foreign_keys[foreign_key_index].object_id = object_id;
401            changed = true;
402        }
403        if constraints.hierarchy.partition_inherited_foreign_keys[inherited_index].object_id
404            != object_id
405        {
406            constraints.hierarchy.partition_inherited_foreign_keys[inherited_index].object_id =
407                object_id;
408            changed = true;
409        }
410        let catalog_identity = constraints.foreign_keys[foreign_key_index].catalog_identity;
411        if constraints.hierarchy.partition_inherited_foreign_keys[inherited_index].catalog_identity
412            != catalog_identity
413        {
414            constraints.hierarchy.partition_inherited_foreign_keys[inherited_index]
415                .catalog_identity = catalog_identity;
416            changed = true;
417        }
418    }
419    changed
420}
421
422fn assign_constraint_object_id(
423    target: &mut Option<[u8; 16]>,
424    allocate: &mut CatalogIdentityAllocator<'_>,
425) -> ConstraintMetadataResult<bool> {
426    assign_catalog_object_id(target, "foreign-key constraint", allocate)
427}
428
429fn assign_catalog_object_id(
430    target: &mut Option<[u8; 16]>,
431    object_kind: &str,
432    allocate: &mut CatalogIdentityAllocator<'_>,
433) -> ConstraintMetadataResult<bool> {
434    if target.is_some() {
435        return Ok(false);
436    }
437    *target = Some(allocate.allocate_object_id(object_kind)?);
438    Ok(true)
439}
440
441fn record_constraint_name(
442    relation: &RelationIdentity,
443    used: &mut BTreeSet<String>,
444    name: Option<&str>,
445) -> ConstraintMetadataResult<()> {
446    let Some(name) = name else {
447        return Ok(());
448    };
449    if name.is_empty() {
450        return Err(ConstraintMetadataError::Invalid(
451            "constraint name must not be empty".into(),
452        ));
453    }
454    if !used.insert(name.to_string()) {
455        return Err(duplicate_constraint(relation, name));
456    }
457    Ok(())
458}
459
460fn duplicate_constraint(relation: &RelationIdentity, name: &str) -> ConstraintMetadataError {
461    ConstraintMetadataError::Execution(Box::new(
462        crate::schema::constraint_changes::constraint_error(
463            "42710",
464            format!(
465                "constraint \"{name}\" for relation \"{}\" already exists",
466                relation.name
467            ),
468        ),
469    ))
470}
471
472pub(super) fn assign_constraint_name(
473    target: &mut Option<String>,
474    parts: (&str, &str, &str),
475    used: &mut BTreeSet<String>,
476) -> ConstraintMetadataResult<bool> {
477    if target.is_some() {
478        return Ok(false);
479    }
480    let base = super::indexes::names::object_name(parts.0, parts.1, parts.2);
481    if used.insert(base.clone()) {
482        *target = Some(base);
483        return Ok(true);
484    }
485    for suffix in 1_u64.. {
486        let label = format!("{}{suffix}", parts.2);
487        let candidate = super::indexes::names::object_name(parts.0, parts.1, &label);
488        if used.insert(candidate.clone()) {
489            *target = Some(candidate);
490            return Ok(true);
491        }
492    }
493    Err(ConstraintMetadataError::Invalid(format!(
494        "constraint name suffix space exhausted for `{base}`"
495    )))
496}
497
498fn constraint_column_component(
499    columns: &[String],
500    relation: &RelationIdentity,
501) -> ConstraintMetadataResult<String> {
502    if columns.is_empty() {
503        return Err(ConstraintMetadataError::Invalid(format!(
504            "constraint on table `{}` has no columns",
505            relation.qualified_name()
506        )));
507    }
508    Ok(columns.join("_"))
509}
510
511fn collect_constraint_columns(expression: &crate::ast::Expr, output: &mut Vec<String>) {
512    use crate::ast::{Expr, FrameBound};
513    match expression {
514        Expr::Column(name) | Expr::QualifiedColumn { column: name, .. } => {
515            if !output.contains(name) {
516                output.push(name.clone());
517            }
518        }
519        Expr::Func {
520            args,
521            order_by,
522            filter,
523            ..
524        } => {
525            for argument in args {
526                collect_constraint_columns(argument, output);
527            }
528            for order in order_by {
529                collect_constraint_columns(&order.expr, output);
530            }
531            if let Some(filter) = filter {
532                collect_constraint_columns(filter, output);
533            }
534        }
535        Expr::Array(items) | Expr::Row(items) | Expr::And(items) | Expr::Or(items) => {
536            for item in items {
537                collect_constraint_columns(item, output);
538            }
539        }
540        Expr::Binary { lhs, rhs, .. } => {
541            collect_constraint_columns(lhs, output);
542            collect_constraint_columns(rhs, output);
543        }
544        Expr::Not(inner)
545        | Expr::UnaryMinus(inner)
546        | Expr::IsNull { expr: inner, .. }
547        | Expr::Cast { expr: inner, .. } => {
548            collect_constraint_columns(inner, output);
549        }
550        Expr::Between { expr, low, high } => {
551            collect_constraint_columns(expr, output);
552            collect_constraint_columns(low, output);
553            collect_constraint_columns(high, output);
554        }
555        Expr::InList { expr, list, .. } => {
556            collect_constraint_columns(expr, output);
557            for item in list {
558                collect_constraint_columns(item, output);
559            }
560        }
561        Expr::WindowCall { args, spec, .. } => {
562            for argument in args {
563                collect_constraint_columns(argument, output);
564            }
565            for expression in &spec.partition_by {
566                collect_constraint_columns(expression, output);
567            }
568            for order in &spec.order_by {
569                collect_constraint_columns(&order.expr, output);
570            }
571            if let Some(frame) = &spec.frame {
572                for bound in [&frame.start, &frame.end] {
573                    if let FrameBound::Preceding(expression) | FrameBound::Following(expression) =
574                        bound
575                    {
576                        collect_constraint_columns(expression, output);
577                    }
578                }
579            }
580        }
581        Expr::Case {
582            base,
583            when,
584            else_branch,
585        } => {
586            if let Some(base) = base {
587                collect_constraint_columns(base, output);
588            }
589            for (condition, result) in when {
590                collect_constraint_columns(condition, output);
591                collect_constraint_columns(result, output);
592            }
593            if let Some(else_branch) = else_branch {
594                collect_constraint_columns(else_branch, output);
595            }
596        }
597        Expr::InSubquery { expr, .. } => collect_constraint_columns(expr, output),
598        Expr::Default
599        | Expr::Star
600        | Expr::QualifiedStar(_)
601        | Expr::InternalColumn(_)
602        | Expr::Literal(_)
603        | Expr::TypedLiteral { .. }
604        | Expr::Param(_)
605        | Expr::ScalarSubquery(_)
606        | Expr::Exists { .. } => {}
607    }
608}
uqa_sql/schema/constraint_metadata.rs

uqa_sql/schema/
constraint_metadata.rs
