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.nulls_not_distinct == right.nulls_not_distinct
251        && left.without_overlaps == right.without_overlaps
252}
253
254pub fn append_inherited_foreign_keys(
255    target: &mut Vec<ForeignKey>,
256    inherited: &[ForeignKey],
257) -> Vec<ForeignKey> {
258    let mut appended = Vec::new();
259    for constraint in inherited {
260        if !target
261            .iter()
262            .any(|candidate| foreign_key_equivalent(candidate, constraint))
263        {
264            let mut clone = constraint.clone();
265            clone.catalog_identity = None;
266            target.push(clone.clone());
267            appended.push(clone);
268        }
269    }
270    appended
271}
272
273pub fn clear_partition_constraint_provenance(constraints: &mut crate::ast::TableConstraintSet) {
274    constraints
275        .hierarchy
276        .partition_inherited_key_constraints
277        .clear();
278    constraints
279        .hierarchy
280        .partition_inherited_foreign_keys
281        .clear();
282}
283
284fn foreign_key_equivalent(left: &ForeignKey, right: &ForeignKey) -> bool {
285    left.local_columns == right.local_columns
286        && left.ref_table == right.ref_table
287        && left.ref_columns == right.ref_columns
288        && left.on_update == right.on_update
289        && left.on_delete == right.on_delete
290        && left.on_delete_set_columns == right.on_delete_set_columns
291        && left.match_type == right.match_type
292        && left.enforced == right.enforced
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
