uqa_sql/schema/inheritance/
alter.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! ALTER inheritance and partition declaration rules over immutable column and constraint definitions.
8use crate::ast::{
9    AutoIncrement, BinaryOp, ColumnDef, DetachedPartitionConstraint, Expr, ForeignKey,
10    PartitionBound, PartitionIdentityOverride, PartitionRangeDatum, PartitionSpec,
11    RelationPersistence, TableCheck, TableHierarchy, TableKeyConstraint,
12};
13use crate::SQLError;
14use uqa_core::Value;
15pub fn validate_row_type(
16    parent_columns: &[ColumnDef],
17    child_columns: &[ColumnDef],
18    parent: &str,
19    child: &str,
20    exact_columns: bool,
21    reject_child_identity: bool,
22) -> Result<(), SQLError> {
23    if reject_child_identity {
24        if let Some(column) = child_columns.iter().find(|column| {
25            column
26                .auto_increment
27                .as_ref()
28                .is_some_and(AutoIncrement::is_identity)
29        }) {
30            return Err(routine(
31                "55000",
32                format!(
33                    "table \"{}\" being attached contains an identity column \"{}\"\nDETAIL: The new partition may not contain an identity column.",
34                    local_relation_name(child),
35                    column.name
36                ),
37            ));
38        }
39    }
40    for parent_column in parent_columns {
41        let Some(child_column) = child_columns
42            .iter()
43            .find(|column| column.name == parent_column.name)
44        else {
45            return Err(routine(
46                "42804",
47                format!("child table is missing column \"{}\"", parent_column.name),
48            ));
49        };
50        if parent_column.ty != child_column.ty {
51            return Err(routine(
52                "42804",
53                format!(
54                    "child table \"{}\" has different type for column \"{}\"",
55                    local_relation_name(child),
56                    parent_column.name
57                ),
58            ));
59        }
60        if parent_column.not_null && !child_column.not_null {
61            return Err(routine(
62                "42804",
63                format!(
64                    "column \"{}\" in child table \"{}\" must be marked NOT NULL",
65                    parent_column.name,
66                    local_relation_name(child)
67                ),
68            ));
69        }
70        match (&parent_column.generated, &child_column.generated) {
71            (None, None) | (Some(_), Some(_)) => {}
72            (Some(_), None) => {
73                return Err(routine(
74                    "42804",
75                    format!(
76                        "column \"{}\" in child table must be a generated column",
77                        parent_column.name
78                    ),
79                ))
80            }
81            (None, Some(_)) => {
82                return Err(routine(
83                    "42804",
84                    format!(
85                        "column \"{}\" in child table must not be a generated column",
86                        parent_column.name
87                    ),
88                ))
89            }
90        }
91        if let (Some(parent_generated), Some(child_generated)) =
92            (&parent_column.generated, &child_column.generated)
93        {
94            if parent_generated.kind != child_generated.kind {
95                return Err(routine(
96                    "42804",
97                    format!(
98                        "column \"{}\" inherits from generated column of different kind",
99                        parent_column.name
100                    ),
101                ));
102            }
103        }
104    }
105    if exact_columns {
106        if let Some(extra) = child_columns.iter().find(|child_column| {
107            !parent_columns
108                .iter()
109                .any(|parent_column| parent_column.name == child_column.name)
110        }) {
111            return Err(routine(
112                "42804",
113                format!(
114                    "table \"{}\" contains column \"{}\" not found in parent \"{}\"\nDETAIL: The new partition may contain only the columns present in parent.",
115                    local_relation_name(child),
116                    extra.name,
117                    local_relation_name(parent)
118                ),
119            ));
120        }
121    }
122    Ok(())
123}
124
125pub fn validate_inherited_checks(
126    child: &str,
127    child_columns: &[ColumnDef],
128    parent_checks: &[TableCheck],
129    child_checks: &[TableCheck],
130) -> Result<(), SQLError> {
131    for parent_check in parent_checks
132        .iter()
133        .filter(|constraint| !constraint.no_inherit)
134    {
135        let Some(name) = parent_check.name.as_deref() else {
136            return Err(SQLError::Internal(
137                "persisted parent CHECK constraint has no name".into(),
138            ));
139        };
140        let Some(child_check) = child_checks
141            .iter()
142            .find(|constraint| constraint.name.as_deref() == Some(name))
143        else {
144            return Err(routine(
145                "42804",
146                format!("child table is missing constraint \"{name}\""),
147            ));
148        };
149        if !crate::schema::check_inheritance::same_check_expression(
150            &child_check.expr,
151            &parent_check.expr,
152            child_columns,
153        )? {
154            return Err(routine(
155                "42804",
156                format!(
157                    "child table \"{}\" has different definition for check constraint \"{name}\"",
158                    local_relation_name(child)
159                ),
160            ));
161        }
162        let conflict = if child_check.no_inherit {
163            Some("non-inherited")
164        } else if parent_check.validated && child_check.enforced && !child_check.validated {
165            Some("NOT VALID")
166        } else if parent_check.enforced && !child_check.enforced {
167            Some("NOT ENFORCED")
168        } else {
169            None
170        };
171        if let Some(conflict) = conflict {
172            return Err(routine("42P17", format!("constraint \"{name}\" conflicts with {conflict} constraint on child table \"{}\"", local_relation_name(child))));
173        }
174    }
175    Ok(())
176}
177
178pub fn install_inherited_identity(
179    columns: &mut [ColumnDef],
180    inherited: &[(String, AutoIncrement)],
181) -> Result<Vec<PartitionIdentityOverride>, SQLError> {
182    let mut overrides = Vec::with_capacity(inherited.len());
183    for (name, increment) in inherited {
184        let column = columns
185            .iter_mut()
186            .find(|column| column.name == *name)
187            .ok_or_else(|| SQLError::Internal(format!("partition lost column `{name}`")))?;
188        overrides.push(PartitionIdentityOverride {
189            column: name.clone(),
190            original: column.auto_increment.clone(),
191        });
192        column.auto_increment = Some(increment.clone());
193    }
194    Ok(overrides)
195}
196
197pub fn restore_identity_overrides(
198    columns: &mut [ColumnDef],
199    inherited: &[(String, AutoIncrement)],
200    overrides: &[PartitionIdentityOverride],
201) {
202    for (name, _) in inherited {
203        let Some(column) = columns.iter_mut().find(|column| column.name == *name) else {
204            continue;
205        };
206        column.auto_increment = overrides
207            .iter()
208            .find(|identity_override| identity_override.column == *name)
209            .and_then(|identity_override| identity_override.original.clone());
210    }
211}
212
213pub fn append_inherited_keys(
214    target: &mut Vec<TableKeyConstraint>,
215    inherited: &[TableKeyConstraint],
216) -> Vec<TableKeyConstraint> {
217    append_inherited_keys_matching(target, inherited, |_, _| true)
218}
219
220/// Each parent index requires a distinct child. The caller supplies attachment eligibility independently of SQL key equivalence.
221pub fn append_inherited_keys_matching(
222    target: &mut Vec<TableKeyConstraint>,
223    inherited: &[TableKeyConstraint],
224    can_attach: impl Fn(&TableKeyConstraint, &TableKeyConstraint) -> bool,
225) -> Vec<TableKeyConstraint> {
226    let mut appended = Vec::new();
227    let mut used = std::collections::BTreeSet::new();
228    for constraint in inherited {
229        if let Some((position, _)) = target.iter().enumerate().find(|(position, candidate)| {
230            !used.contains(position)
231                && key_equivalent(candidate, constraint)
232                && can_attach(candidate, constraint)
233        }) {
234            used.insert(position);
235            continue;
236        }
237        let mut constraint = constraint.clone();
238        constraint.name = None;
239        constraint.catalog_identity = None;
240        used.insert(target.len());
241        target.push(constraint.clone());
242        appended.push(constraint);
243    }
244    appended
245}
246
247pub fn key_equivalent(left: &TableKeyConstraint, right: &TableKeyConstraint) -> bool {
248    left.kind == right.kind
249        && left.columns == right.columns
250        && left.included_columns == right.included_columns
251        && left.nulls_not_distinct == right.nulls_not_distinct
252        && left.without_overlaps == right.without_overlaps
253}
254
255pub fn append_inherited_foreign_keys(
256    target: &mut Vec<ForeignKey>,
257    inherited: &[ForeignKey],
258) -> Vec<ForeignKey> {
259    let mut appended = Vec::new();
260    for constraint in inherited {
261        if !target
262            .iter()
263            .any(|candidate| foreign_key_equivalent(candidate, constraint))
264        {
265            let mut clone = constraint.clone();
266            clone.catalog_identity = None;
267            target.push(clone.clone());
268            appended.push(clone);
269        }
270    }
271    appended
272}
273
274pub fn clear_partition_constraint_provenance(constraints: &mut crate::ast::TableConstraintSet) {
275    constraints
276        .hierarchy
277        .partition_inherited_key_constraints
278        .clear();
279    constraints
280        .hierarchy
281        .partition_inherited_foreign_keys
282        .clear();
283}
284
285fn foreign_key_equivalent(left: &ForeignKey, right: &ForeignKey) -> bool {
286    left.local_columns == right.local_columns
287        && left.ref_table == right.ref_table
288        && left.ref_columns == right.ref_columns
289        && left.on_update == right.on_update
290        && left.on_delete == right.on_delete
291        && left.on_delete_set_columns == right.on_delete_set_columns
292        && left.match_type == right.match_type
293        && left.enforced == right.enforced
294}
295
296pub fn detached_bound_check(
297    table: &str,
298    spec: &PartitionSpec,
299    bound: &PartitionBound,
300    existing: &[TableCheck],
301) -> TableCheck {
302    let expr = renderable_bound_expression(spec, bound);
303    let relation = local_relation_name(table);
304    let key = spec.keys.first().and_then(|key| match key {
305        Expr::Column(column) => Some(column.as_str()),
306        _ => None,
307    });
308    let base = key.map_or_else(
309        || format!("{relation}_check"),
310        |column| format!("{relation}_{column}_check"),
311    );
312    let name = unique_constraint_name(&base, existing);
313    TableCheck {
314        catalog_oid: None,
315        name: Some(name),
316        expr,
317        enforced: true,
318        validated: true,
319        no_inherit: false,
320        object_id: None,
321        is_local: true,
322        partition_constraint: Some(DetachedPartitionConstraint {
323            spec: spec.clone(),
324            bound: bound.clone(),
325        }),
326    }
327}
328
329fn unique_constraint_name(base: &str, existing: &[TableCheck]) -> String {
330    if !existing
331        .iter()
332        .any(|constraint| constraint.name.as_deref() == Some(base))
333    {
334        return base.to_string();
335    }
336    for suffix in 1_u64.. {
337        let candidate = format!("{base}{suffix}");
338        if !existing
339            .iter()
340            .any(|constraint| constraint.name.as_deref() == Some(candidate.as_str()))
341        {
342            return candidate;
343        }
344    }
345    unreachable!("u64 constraint suffix space is exhaustive")
346}
347
348fn renderable_bound_expression(spec: &PartitionSpec, bound: &PartitionBound) -> Expr {
349    let Some(key) = spec.keys.first().cloned().filter(|_| spec.keys.len() == 1) else {
350        return Expr::Literal(Value::Bool(true));
351    };
352    match bound {
353        PartitionBound::List(values) => {
354            let mut terms = Vec::new();
355            let mut non_null = Vec::new();
356            for value in values {
357                if matches!(value, Expr::Literal(Value::Null)) {
358                    terms.push(Expr::IsNull {
359                        expr: Box::new(key.clone()),
360                        negated: false,
361                    });
362                } else {
363                    non_null.push(value.clone());
364                }
365            }
366            if !non_null.is_empty() {
367                terms.push(Expr::InList {
368                    expr: Box::new(key),
369                    list: non_null,
370                    negated: false,
371                });
372            }
373            if terms.len() == 1 {
374                terms.pop().unwrap_or(Expr::Literal(Value::Bool(true)))
375            } else {
376                Expr::Or(terms)
377            }
378        }
379        PartitionBound::Range { lower, upper } if lower.len() == 1 && upper.len() == 1 => {
380            let mut terms = vec![Expr::IsNull {
381                expr: Box::new(key.clone()),
382                negated: true,
383            }];
384            if let PartitionRangeDatum::Value(lower) = &lower[0] {
385                terms.push(Expr::Binary {
386                    op: BinaryOp::GreaterEqual,
387                    lhs: Box::new(key.clone()),
388                    rhs: Box::new(lower.clone()),
389                });
390            }
391            if let PartitionRangeDatum::Value(upper) = &upper[0] {
392                terms.push(Expr::Binary {
393                    op: BinaryOp::Less,
394                    lhs: Box::new(key),
395                    rhs: Box::new(upper.clone()),
396                });
397            }
398            Expr::And(terms)
399        }
400        PartitionBound::Hash { .. } | PartitionBound::Range { .. } | PartitionBound::Default => {
401            Expr::Literal(Value::Bool(true))
402        }
403    }
404}
405
406pub fn validate_matching_persistence(
407    child: &str,
408    parent: &str,
409    operation: &str,
410    child_persistence: RelationPersistence,
411    parent_persistence: RelationPersistence,
412) -> Result<(), SQLError> {
413    if (child_persistence == RelationPersistence::Temporary)
414        != (parent_persistence == RelationPersistence::Temporary)
415    {
416        return Err(wrong_object(format!(
417            "cannot {operation} {} relation \"{}\" from {} relation \"{}\"",
418            persistence_label(child_persistence),
419            local_relation_name(child),
420            persistence_label(parent_persistence),
421            local_relation_name(parent)
422        )));
423    }
424    Ok(())
425}
426
427fn persistence_label(persistence: RelationPersistence) -> &'static str {
428    match persistence {
429        RelationPersistence::Temporary => "temporary",
430        RelationPersistence::Unlogged => "unlogged",
431        RelationPersistence::Permanent => "permanent",
432    }
433}
434
435pub fn normalize_parent_sequence_numbers(hierarchy: &mut TableHierarchy) {
436    if hierarchy.parent_sequence_numbers.len() == hierarchy.parents.len() {
437        return;
438    }
439    hierarchy.parent_sequence_numbers = hierarchy
440        .parents
441        .iter()
442        .enumerate()
443        .map(|(index, _)| i32::try_from(index + 1).unwrap_or(i32::MAX))
444        .collect();
445}
446
447fn local_relation_name(name: &str) -> &str {
448    name.rsplit('.').next().unwrap_or(name)
449}
450fn wrong_object(message: impl Into<String>) -> SQLError {
451    routine("42809", message)
452}
453fn routine(sqlstate: &str, message: impl Into<String>) -> SQLError {
454    SQLError::Routine {
455        sqlstate: sqlstate.into(),
456        message: message.into(),
457    }
458}
uqa_sql/schema/inheritance/alter.rs

uqa_sql/schema/inheritance/
alter.rs
