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

uqa_sql/schema/inheritance/
alter.rs
