use crate::schema::constraints::{
add_check_constraint, add_foreign_key_constraint, add_not_null_constraint, alter_constraint,
drop::drop_constraint, set_not_null_constraint, table_constraint_state, validate_constraint,
};
use uqa_sql::{
ast::{AlterTableAction, AlterTableStmt},
SQLError, SQLResult,
};
pub mod binding;
mod column_removal;
mod context;
pub mod entry;
pub mod identity;
mod locking;
mod recursion;
mod type_changes;
pub use context::*;
use recursion::{materialize_recursive_action_names, run_recursive_alter_action};
fn ddl_storage_error(action: &str, error: uqa_storage::StorageBackendError) -> SQLError {
uqa_sql::catalog::errors::storage_error(action, &error)
}
fn check_added_column_declaration<S: Clone + 'static>(
context: &TableAlterContext<'_, S>,
table: &str,
qualifier: &str,
action: &mut AlterTableAction,
) -> Result<bool, SQLError> {
use uqa_sql::schema::table_creation::column_declarations::{
check_column_declaration, check_serial_array, ColumnDeclarationTarget,
};
let AlterTableAction::AddColumn {
column,
declaration,
if_not_exists,
..
} = action
else {
return Ok(true);
};
let hierarchy = context
.hierarchy
.partitions
.catalog
.try_table_hierarchy(table)
.map_err(|error| SQLError::Internal(format!("read table hierarchy: {error}")))?;
let partitioned = hierarchy.partition_spec.is_some();
let partition = hierarchy.is_partition();
if partition {
return Err(SQLError::Routine {
sqlstate: "42809".into(),
message: "cannot add column to a partition".into(),
});
}
uqa_sql::schema::columns::validate_postgres_column_name(&column.name)?;
if crate::schema::columns::addition::column_exists(
&context.addition,
table,
&column.name,
*if_not_exists,
)? {
context.constraints.notices.push(
uqa_sql::SQLNotice::notice(format!(
"column \"{}\" of relation \"{qualifier}\" already exists, skipping",
column.name
))
.with_sqlstate("42701"),
);
return Ok(false);
}
check_serial_array(declaration)?;
column.ty = uqa_sql::type_resolution::resolve_alter_column_type(
context.hierarchy.partitions.types,
&column.ty,
)?;
let target = ColumnDeclarationTarget {
table: qualifier,
partitioned,
partition,
};
if check_column_declaration(declaration, &column.name, target)? {
return Err(SQLError::Unsupported(
"ALTER TABLE: DEFERRABLE PRIMARY KEY and UNIQUE constraints are not supported".into(),
));
}
Ok(true)
}
pub fn run_alter_table<S: Clone + 'static>(
context: &TableAlterContext<'_, S>,
stmt: AlterTableStmt,
) -> Result<SQLResult, SQLError> {
let mode = uqa_sql::schema::table_alteration::syntax::table_alter_lock_mode(&stmt).into();
let AlterTableStmt {
table,
qualifier,
if_exists,
recurse,
mut actions,
} = stmt;
context.constraints.access.ensure_table_owner(&table)?;
let prepared_types = type_changes::prepare(context, &table, &qualifier, &mut actions)?;
let mut rewrites = type_changes::TypeRewrites::capture(
context,
&table,
recurse,
mode,
&actions,
prepared_types,
)?;
let _deferred = rewrites.as_ref().map(|_| {
(
context.addition.pending_rows.deferral.defer(),
context.columns.deferred_rows.defer(),
context.constraints.deferred_rows.defer(),
)
});
let mut actions = actions.into_iter().enumerate().collect::<Vec<_>>();
if rewrites.is_some() {
actions.sort_by_key(|(_, action)| {
uqa_sql::schema::table_alteration::ordering::execution_order(action)
});
}
for (position, mut action) in actions {
if let AlterTableAction::AlterColumnType { name, ty, .. } = &action {
rewrites
.as_mut()
.ok_or_else(|| SQLError::Internal("type change has no original rows".into()))?
.apply(context, position, name, ty)?;
continue;
}
if !check_added_column_declaration(context, &table, &qualifier, &mut action)? {
continue;
}
let column_checks = prepare_alter_action(context, &table, recurse, &mut action, mode)?;
run_recursive_alter_action(
context,
AlterTableStmt {
table: table.clone(),
qualifier: qualifier.clone(),
if_exists,
recurse,
actions: Vec::new(),
},
action,
)?;
for mut action in column_checks {
if matches!(action, AlterTableAction::AddForeignKeyConstraint { .. }) {
locking::prepare_table_alter_action(
&context.binding,
context,
&table,
recurse,
&mut action,
mode,
)?;
}
materialize_recursive_action_names(context, &table, recurse, &mut action)?;
run_recursive_alter_action(
context,
AlterTableStmt {
table: table.clone(),
qualifier: qualifier.clone(),
if_exists,
recurse,
actions: Vec::new(),
},
action,
)?;
}
}
if let Some(rewrites) = rewrites {
rewrites.finish(context)?;
}
Ok(SQLResult::empty())
}
fn prepare_added_column_sequence<S: Clone + 'static>(
context: &TableAlterContext<'_, S>,
table: &str,
recurse: bool,
column: &mut uqa_sql::ast::ColumnDef,
) -> Result<(), SQLError> {
let Some(provenance) = column.auto_increment.as_ref() else {
return Ok(());
};
if provenance.is_identity() && recurse {
let catalog = context.hierarchy.partitions.catalog;
let partitioned = catalog
.try_table_hierarchy(table)
.map_err(SQLError::Internal)?
.partition_spec
.is_some();
if !partitioned && !catalog.direct_hierarchy_children(table)?.is_empty() {
return Err(SQLError::Routine {
sqlstate: "42P16".into(),
message: "cannot recursively add identity column to table that has child tables"
.into(),
});
}
}
crate::schema::columns::addition::create_added_column_sequence(&context.addition, table, column)
}
fn prepare_alter_action<S: Clone + 'static>(
context: &TableAlterContext<'_, S>,
table: &str,
recurse: bool,
action: &mut AlterTableAction,
mode: crate::row_locks::RelationLockMode,
) -> Result<Vec<AlterTableAction>, SQLError> {
if let AlterTableAction::AddColumn { column, .. } = action {
column.ty = uqa_sql::type_resolution::resolve_declared_column_type(
context.hierarchy.publication.types,
&column.ty,
)?;
prepare_added_column_sequence(context, table, recurse, column)?;
}
materialize_recursive_action_names(context, table, recurse, action)?;
let mut column_checks = if let AlterTableAction::AddColumn {
column,
checks,
foreign_keys,
..
} = action
{
uqa_sql::schema::constraint_changes::take_column_check(column)
.into_iter()
.chain(std::mem::take(checks))
.map(|constraint| AlterTableAction::AddCheckConstraint { constraint })
.chain(
std::mem::take(foreign_keys)
.into_iter()
.map(|constraint| AlterTableAction::AddForeignKeyConstraint { constraint }),
)
.collect()
} else {
Vec::new()
};
locking::prepare_table_alter_action(&context.binding, context, table, recurse, action, mode)?;
for check in &mut column_checks {
if matches!(check, AlterTableAction::AddForeignKeyConstraint { .. }) {
continue;
}
locking::prepare_table_alter_action(
&context.binding,
context,
table,
recurse,
check,
mode,
)?;
}
context.binding.locks.prepare_definition_write()?;
Ok(column_checks)
}
#[expect(
clippy::too_many_lines,
reason = "preserves DDL dependency and action order"
)]
fn run_alter_table_action<S: Clone + 'static>(
context: &TableAlterContext<'_, S>,
stmt: AlterTableStmt,
action: AlterTableAction,
recursing: bool,
inherited_not_null_name: Option<String>,
) -> Result<(), SQLError> {
if matches!(&action, AlterTableAction::AddKeyConstraint { .. }) {
let persistence = context
.hierarchy
.catalog
.table_persistence(&stmt.table)
.map_err(|error| ddl_storage_error("ALTER TABLE ADD CONSTRAINT", error))?;
if persistence == Some(uqa_sql::ast::RelationPersistence::Temporary) {
context.addition.namespace.ensure_temporary_privilege()?;
} else {
context
.addition
.namespace
.ensure_existing_create(&stmt.table)?;
}
}
if !matches!(
action,
AlterTableAction::RenameColumn { .. }
| AlterTableAction::RenameTable { .. }
| AlterTableAction::RenameTrigger { .. }
| AlterTableAction::RenameConstraint { .. }
| AlterTableAction::RenameRule { .. }
) {
context
.constraints
.access
.ensure_no_pending_events(&stmt.table, "ALTER TABLE")?;
}
match action {
AlterTableAction::ChangeOwner { owner } => {
context
.ownership
.alter_table_role_owner(&stmt.table, &owner)?;
}
action @ (AlterTableAction::AddInheritance { .. }
| AlterTableAction::DropInheritance { .. }
| AlterTableAction::AttachPartition { .. }
| AlterTableAction::DetachPartition { .. }) => {
crate::schema::hierarchy::run_alter_hierarchy_action(
&context.hierarchy,
&stmt.table,
action,
)?;
}
AlterTableAction::AddColumn {
column,
key_constraints,
if_not_exists,
..
} => {
crate::schema::columns::addition::add_column(
&context.addition,
&stmt.table,
&stmt.qualifier,
column,
&key_constraints,
if_not_exists,
)?;
}
AlterTableAction::AddKeyConstraint { constraint } => {
crate::schema::constraints::add_key_constraint(
&context.constraints,
&stmt.table,
constraint,
)?;
}
AlterTableAction::AddCheckConstraint { constraint } => {
add_check_constraint(
&context.constraints,
&stmt.table,
&stmt.qualifier,
constraint,
)?;
}
AlterTableAction::AddForeignKeyConstraint { constraint } => {
add_foreign_key_constraint(
&context.constraints,
&stmt.table,
&stmt.qualifier,
constraint,
)?;
}
AlterTableAction::AddNotNullConstraint {
name,
column,
validated,
no_inherit,
} => {
add_not_null_constraint(
&context.constraints,
&stmt.table,
name,
&column,
validated,
no_inherit,
!recursing,
)?;
}
AlterTableAction::ValidateConstraint { name } => {
validate_constraint(&context.constraints, &stmt.table, &name, stmt.recurse)?;
}
AlterTableAction::AlterConstraint {
name,
enforceability,
deferrability,
no_inherit,
} => {
alter_constraint(
&context.constraints,
&stmt.table,
&name,
uqa_sql::schema::constraint_changes::ConstraintAlterOptions {
enforceability,
deferrability,
no_inherit,
ancestor: None,
},
stmt.recurse,
)?;
}
AlterTableAction::DropConstraint {
name,
if_exists,
cascade,
} => {
drop_constraint(
&context.constraints,
&stmt.table,
&name,
if_exists,
cascade,
stmt.recurse,
)?;
}
AlterTableAction::DropColumn {
name,
if_exists,
cascade,
} => {
column_removal::drop_column(
context,
&stmt.table,
&name,
if_exists,
cascade,
stmt.recurse,
)?;
}
AlterTableAction::RenameColumn { from, to } => {
uqa_sql::schema::columns::validate_postgres_column_name(&to)?;
if !context
.addition
.state
.has_column(&stmt.table, &from)
.map_err(|err| ddl_storage_error("ALTER TABLE RENAME COLUMN", err))?
{
return Err(SQLError::Routine {
sqlstate: "42703".into(),
message: format!("column \"{from}\" does not exist"),
});
}
if context
.addition
.state
.has_column(&stmt.table, &to)
.map_err(|err| ddl_storage_error("ALTER TABLE RENAME COLUMN", err))?
{
let relation = uqa_core::RelationIdentity::from_legacy_name(&stmt.table)
.map_err(SQLError::Internal)?;
return Err(SQLError::Routine {
sqlstate: "42701".into(),
message: format!(
"column \"{to}\" of relation \"{}\" already exists",
relation.name
),
});
}
context
.lifecycle
.rename_column(&stmt.table, &from, &to)
.map_err(|e| ddl_storage_error("ALTER TABLE RENAME COLUMN", e))?;
}
AlterTableAction::RenameTable { to } => {
let source = uqa_core::RelationIdentity::from_legacy_name(&stmt.table)
.map_err(SQLError::Internal)?;
let target = uqa_sql::schema::relation_alteration::relation_rename_target(
context.binding.names,
&source,
&to,
"ALTER TABLE RENAME",
)?;
if !context
.lifecycle
.rename_table(&stmt.table, &target.qualified_name())
.map_err(|e| ddl_storage_error("ALTER TABLE RENAME", e))?
{
return Err(SQLError::Unsupported(format!(
"ALTER TABLE RENAME: rename of `{}` failed",
stmt.table
)));
}
}
AlterTableAction::RenameTrigger { from, to } => {
context.events.rename_trigger(&stmt.table, &from, &to)?;
}
AlterTableAction::RenameConstraint { from, to } => {
if !crate::schema::constraints::renaming::rename_constraint(
&context.constraints,
&stmt.table,
&from,
&to,
stmt.recurse,
)? {
context
.events
.rename_trigger_constraint(&stmt.table, &from, &to)?;
}
}
AlterTableAction::RenameRule { from, to } => {
context.events.rename_rule(&stmt.table, &from, &to)?;
}
AlterTableAction::SetTriggerEnableMode {
name,
user_only: _,
mode,
} => {
context
.events
.set_trigger_enable_mode(&stmt.table, name.as_deref(), mode)?;
}
AlterTableAction::SetRuleEnableMode { name, mode } => {
context
.events
.set_rule_enable_mode(&stmt.table, &name, mode)?;
}
AlterTableAction::SetPersistence { persistence } => {
return Err(SQLError::Unsupported(format!(
"ALTER TABLE SET {} is not supported for tables",
match persistence {
uqa_sql::ast::RelationPersistence::Permanent => "LOGGED",
uqa_sql::ast::RelationPersistence::Unlogged => "UNLOGGED",
uqa_sql::ast::RelationPersistence::Temporary => "TEMPORARY",
}
)));
}
AlterTableAction::SetSchema { schema } => {
let source = uqa_core::RelationIdentity::from_legacy_name(&stmt.table)
.map_err(SQLError::Internal)?;
let catalog = context
.schema_moves
.indexes
.identities
.catalog
.current_catalog_snapshot();
let table = catalog
.snapshot()
.tables
.get(&source)
.ok_or_else(|| SQLError::UnknownTable(stmt.table.clone()))?;
if let Some(target) =
context
.schema_moves
.target(&source, &schema, table.persistence)?
{
let relocation = context
.schema_moves
.prepare(&source, &target, table.object_id)?;
if !context
.lifecycle
.rename_table(&stmt.table, &target.qualified_name())
.map_err(|error| ddl_storage_error("ALTER TABLE SET SCHEMA", error))?
{
return Err(SQLError::Internal(
"table disappeared during schema movement".into(),
));
}
relocation.publish(&context.schema_moves)?;
}
}
AlterTableAction::SetDefault { name, default } => {
crate::schema::columns::alteration::set_default(
&context.columns,
&stmt.table,
&name,
default,
)?;
}
AlterTableAction::DropDefault { name } => {
crate::schema::columns::alteration::drop_default(&context.columns, &stmt.table, &name)?;
}
AlterTableAction::SetExpression { name, expression } => {
crate::schema::columns::alteration::set_expression(
&context.columns,
&stmt.table,
&stmt.qualifier,
&name,
expression,
)?;
}
AlterTableAction::DropExpression { name, if_exists } => {
if let Some(notice) = crate::schema::columns::alteration::drop_expression(
&context.columns,
&stmt.table,
&name,
if_exists,
)? {
context.constraints.notices.push(notice);
}
}
AlterTableAction::SetNotNull { name } => {
set_not_null_constraint(
&context.constraints,
&stmt.table,
&name,
stmt.recurse,
!recursing,
inherited_not_null_name,
)?;
}
AlterTableAction::DropNotNull { name } => {
let (columns, constraints) = table_constraint_state(&context.constraints, &stmt.table)?;
let column = uqa_sql::schema::columns::altered_column(&stmt.table, &columns, &name)?;
let parent_not_null = match constraints.hierarchy.parents.first() {
Some(parent) if constraints.hierarchy.partition_bound.is_some() => {
let (parent_columns, _) = table_constraint_state(&context.constraints, parent)?;
parent_columns
.iter()
.any(|parent_column| parent_column.name == name && parent_column.not_null)
}
_ => false,
};
uqa_sql::schema::constraint_changes::not_null_removal::validate_column_removal(
&stmt.table,
column,
parent_not_null,
)?;
if let Some(constraint_name) = column.not_null_name.as_deref() {
drop_constraint(
&context.constraints,
&stmt.table,
constraint_name,
false,
false,
stmt.recurse,
)?;
}
}
AlterTableAction::AlterColumnType { .. } => {
return Err(SQLError::Internal(
"type change bypassed its original-row queue".into(),
));
}
AlterTableAction::AddIdentity {
name,
kind,
declaration,
} => identity::add_identity(
context,
identity::IdentityTarget {
table: &stmt.table,
recurse: stmt.recurse,
recursing,
},
&name,
kind,
declaration,
)?,
AlterTableAction::SetIdentity {
name,
kind,
repeated_kind,
sequence,
error,
} => identity::set_identity(
context,
identity::IdentityTarget {
table: &stmt.table,
recurse: stmt.recurse,
recursing,
},
&name,
kind,
repeated_kind,
&sequence,
error.as_ref(),
)?,
AlterTableAction::DropIdentity { name, if_exists } => identity::drop_identity(
context,
identity::IdentityTarget {
table: &stmt.table,
recurse: stmt.recurse,
recursing,
},
&name,
if_exists,
)?,
}
Ok(())
}