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

uqa_sql/schema/inheritance/
alter.rs
