use super::publication::{hierarchy::HierarchySchemaChange, SchemaPublicationContext};
use crate::catalog::RelationResolution;
use crate::mutation::constraints::context::ConstraintContext;
use uqa_sql::schema::inheritance::alter::{
append_inherited_foreign_keys, append_inherited_keys, detached_bound_check,
install_inherited_identity, normalize_parent_sequence_numbers,
remove_partition_inherited_constraints, restore_identity_overrides,
};
use uqa_sql::semantics::partition::PartitionContext;
use uqa_sql::{
ast::{
AlterTableAction, AutoIncrement, ColumnDef, ForeignKey, PartitionBound,
RelationPersistence, TableCheck, TableConstraintSet, TableHierarchy, TableKeyConstraint,
},
SQLError,
};
use uqa_storage::StorageBackendResult;
pub trait HierarchyCatalog {
fn try_describe_table(&self, table: &str) -> StorageBackendResult<Option<Vec<ColumnDef>>>;
fn try_check_constraint_definitions(
&self,
table: &str,
) -> StorageBackendResult<Vec<TableCheck>>;
fn try_key_constraints(&self, table: &str) -> StorageBackendResult<Vec<TableKeyConstraint>>;
fn try_foreign_keys(&self, table: &str) -> StorageBackendResult<Vec<ForeignKey>>;
fn try_declared_table_constraints(
&self,
table: &str,
) -> StorageBackendResult<TableConstraintSet>;
fn table_persistence(&self, table: &str) -> StorageBackendResult<Option<RelationPersistence>>;
}
pub trait HierarchyNamespace {
fn resolve_visible_relation_kind(&self, name: &str) -> Result<RelationResolution, SQLError>;
fn lock_exclusive(&self, table: &str) -> Result<(), SQLError>;
}
pub struct HierarchyContext<'a> {
pub catalog: &'a dyn HierarchyCatalog,
pub namespace: &'a dyn HierarchyNamespace,
pub constraints: ConstraintContext<'a>,
pub partitions: PartitionContext<'a>,
pub publication: SchemaPublicationContext<'a>,
}
fn ddl_storage_error(action: &str, error: uqa_storage::StorageBackendError) -> SQLError {
uqa_sql::catalog::errors::storage_error(action, &error)
}
pub fn run_alter_hierarchy_action(
context: &HierarchyContext<'_>,
table: &str,
action: AlterTableAction,
) -> Result<(), SQLError> {
match action {
AlterTableAction::AddInheritance { parent } => add_inheritance(context, table, &parent),
AlterTableAction::DropInheritance { parent } => drop_inheritance(context, table, &parent),
AlterTableAction::AttachPartition { partition, bound } => {
attach_partition(context, table, &partition, bound)
}
AlterTableAction::DetachPartition {
partition,
concurrently,
finalize,
} => detach_partition(context, table, &partition, concurrently, finalize),
_ => Err(SQLError::Internal(
"non-hierarchy ALTER action reached hierarchy executor".into(),
)),
}
}
fn add_inheritance(
context: &HierarchyContext<'_>,
child: &str,
requested_parent: &str,
) -> Result<(), SQLError> {
let parent = resolve_table(context, requested_parent)?;
lock_secondary_relation(context, child, &parent)?;
validate_matching_persistence(context, child, &parent, "inherit from")?;
let mut hierarchy = read_hierarchy(context, child)?;
let parent_hierarchy = read_hierarchy(context, &parent)?;
if hierarchy.is_partition() {
return Err(wrong_object("cannot change inheritance of a partition"));
}
if hierarchy.partition_spec.is_some() {
return Err(wrong_object(
"cannot change inheritance of partitioned table",
));
}
if parent_hierarchy.is_partition() {
return Err(wrong_object("cannot inherit from a partition"));
}
if parent_hierarchy.partition_spec.is_some() {
return Err(wrong_object(format!(
"cannot inherit from partitioned table \"{requested_parent}\""
)));
}
if hierarchy
.parents
.iter()
.any(|candidate| candidate == &parent)
{
return Err(routine(
"42P07",
format!("relation \"{requested_parent}\" would be inherited from more than once"),
));
}
if child == parent
|| context
.constraints
.catalog
.hierarchy_scan_tables(child, true)?
.iter()
.any(|descendant| descendant == &parent)
{
return Err(routine("42P07", "circular inheritance not allowed"));
}
validate_row_type(context, &parent, child, false, false)?;
validate_inherited_checks(context, &parent, child)?;
normalize_parent_sequence_numbers(&mut hierarchy);
let sequence_number = hierarchy.next_parent_sequence_number();
hierarchy.parents.push(parent);
hierarchy.parent_sequence_numbers.push(sequence_number);
replace_hierarchy_only(context, child, hierarchy, "ALTER TABLE INHERIT")
}
fn drop_inheritance(
context: &HierarchyContext<'_>,
child: &str,
requested_parent: &str,
) -> Result<(), SQLError> {
let parent = resolve_table(context, requested_parent)?;
lock_secondary_relation(context, child, &parent)?;
let mut hierarchy = read_hierarchy(context, child)?;
if hierarchy.is_partition() {
return Err(wrong_object("cannot change inheritance of a partition"));
}
if hierarchy.partition_spec.is_some() {
return Err(wrong_object(
"cannot change inheritance of partitioned table",
));
}
let Some(index) = hierarchy
.parents
.iter()
.position(|candidate| candidate == &parent)
else {
return Err(routine(
"42P01",
format!(
"relation \"{requested_parent}\" is not a parent of relation \"{}\"",
local_relation_name(child)
),
));
};
normalize_parent_sequence_numbers(&mut hierarchy);
hierarchy.parents.remove(index);
hierarchy.parent_sequence_numbers.remove(index);
replace_hierarchy_only(context, child, hierarchy, "ALTER TABLE NO INHERIT")
}
fn attach_partition(
context: &HierarchyContext<'_>,
parent: &str,
requested_partition: &str,
bound: PartitionBound,
) -> Result<(), SQLError> {
let partition = resolve_table(context, requested_partition)?;
lock_secondary_relation(context, parent, &partition)?;
validate_matching_persistence(context, &partition, parent, "attach to")?;
let parent_hierarchy = read_hierarchy(context, parent)?;
let Some(parent_spec) = parent_hierarchy.partition_spec.as_ref() else {
return Err(wrong_object(format!(
"ALTER action ATTACH PARTITION cannot be performed on relation \"{}\"",
local_relation_name(parent)
)));
};
let partition_hierarchy = read_hierarchy(context, &partition)?;
if partition_hierarchy.is_partition() {
return Err(wrong_object(format!(
"\"{requested_partition}\" is already a partition"
)));
}
if !partition_hierarchy.parents.is_empty() {
return Err(wrong_object("cannot attach inheritance child as partition"));
}
let direct_children = context
.partitions
.catalog
.direct_hierarchy_children(&partition)?;
if partition_hierarchy.partition_spec.is_none() && !direct_children.is_empty() {
return Err(wrong_object(
"cannot attach inheritance parent as partition",
));
}
if parent == partition
|| context
.constraints
.catalog
.hierarchy_scan_tables(&partition, true)?
.iter()
.any(|descendant| descendant == parent)
{
return Err(routine("42P07", "circular inheritance not allowed"));
}
validate_row_type(context, parent, &partition, true, true)?;
validate_inherited_checks(context, parent, &partition)?;
uqa_sql::semantics::partition::validate_new_partition_bound(
&context.partitions,
parent,
&bound,
)?;
validate_attached_rows(context, parent, &partition, &bound)?;
validate_default_partition_exclusion(context, parent, &bound)?;
let subtree = inherit_partition_schema(context, parent, &partition, &bound)?;
for target in subtree {
validate_existing_constraints(context, &target)?;
}
let _ = parent_spec;
Ok(())
}
fn inherit_partition_schema(
context: &HierarchyContext<'_>,
parent: &str,
partition: &str,
bound: &PartitionBound,
) -> Result<Vec<String>, SQLError> {
let parent_columns = table_columns(context, parent, "ATTACH PARTITION")?;
let inherited_identity = parent_columns
.iter()
.filter_map(|column| {
column
.auto_increment
.as_ref()
.filter(|increment| increment.is_identity())
.map(|increment| (column.name.clone(), increment.clone()))
})
.collect::<Vec<_>>();
let parent_keys = context
.catalog
.try_key_constraints(parent)
.map_err(|error| ddl_storage_error("ATTACH PARTITION constraints", error))?;
let parent_foreign_keys = context
.catalog
.try_foreign_keys(parent)
.map_err(|error| ddl_storage_error("ATTACH PARTITION constraints", error))?;
let subtree = context
.constraints
.catalog
.hierarchy_scan_tables(partition, true)?;
for target in &subtree {
let mut columns = table_columns(context, target, "ATTACH PARTITION")?;
let identity_overrides = install_inherited_identity(&mut columns, &inherited_identity)?;
let mut constraints = declared_constraints(context, target, "ATTACH PARTITION")?;
let inherited_keys = append_inherited_keys(&mut constraints.key_constraints, &parent_keys);
let inherited_foreign_keys =
append_inherited_foreign_keys(&mut constraints.foreign_keys, &parent_foreign_keys);
let mut hierarchy = constraints.hierarchy.clone();
hierarchy.partition_identity_overrides = identity_overrides;
hierarchy.partition_inherited_key_constraints = inherited_keys;
hierarchy.partition_inherited_foreign_keys = inherited_foreign_keys;
if target == partition {
hierarchy.parents = vec![parent.to_string()];
hierarchy.parent_sequence_numbers = vec![1];
hierarchy.partition_bound = Some(bound.clone());
}
super::publication::hierarchy::replace_hierarchy_components(
&context.publication,
context.catalog,
target,
HierarchySchemaChange {
columns,
checks: constraints.checks,
foreign_keys: constraints.foreign_keys,
key_constraints: constraints.key_constraints,
hierarchy,
},
)
.map_err(|error| ddl_storage_error("ATTACH PARTITION", error))?;
}
Ok(subtree)
}
fn detach_partition(
context: &HierarchyContext<'_>,
parent: &str,
requested_partition: &str,
concurrently: bool,
finalize: bool,
) -> Result<(), SQLError> {
let partition = resolve_table(context, requested_partition)?;
lock_secondary_relation(context, parent, &partition)?;
let parent_hierarchy = read_hierarchy(context, parent)?;
let Some(parent_spec) = parent_hierarchy.partition_spec.as_ref() else {
return Err(wrong_object(format!(
"ALTER action DETACH PARTITION cannot be performed on relation \"{}\"",
local_relation_name(parent)
)));
};
let partition_hierarchy = read_hierarchy(context, &partition)?;
let attached = partition_hierarchy.is_partition()
&& partition_hierarchy
.parents
.first()
.is_some_and(|edge| edge == parent);
if finalize {
return Err(routine(
"55000",
format!(
"cannot complete detaching partition \"{}\"\nDETAIL: There's no pending concurrent detach.",
local_relation_name(&partition)
),
));
}
if !attached {
return Err(routine(
"42P01",
format!(
"relation \"{requested_partition}\" is not a partition of relation \"{}\"",
local_relation_name(parent)
),
));
}
if concurrently && direct_default_partition(context, parent)?.is_some() {
return Err(routine(
"55000",
"cannot detach partitions concurrently when a default partition exists",
));
}
let bound = partition_hierarchy
.partition_bound
.as_ref()
.ok_or_else(|| SQLError::Internal("attached partition lost its bound".into()))?
.clone();
let inherited_identity = table_columns(context, parent, "DETACH PARTITION")?
.into_iter()
.filter_map(|column| {
column
.auto_increment
.filter(AutoIncrement::is_identity)
.map(|increment| (column.name, increment))
})
.collect::<Vec<_>>();
let subtree = context
.constraints
.catalog
.hierarchy_scan_tables(&partition, true)?;
for target in &subtree {
let mut columns = table_columns(context, target, "DETACH PARTITION")?;
let mut constraints = declared_constraints(context, target, "DETACH PARTITION")?;
restore_identity_overrides(
&mut columns,
&inherited_identity,
&constraints.hierarchy.partition_identity_overrides,
);
remove_partition_inherited_constraints(&mut constraints);
if concurrently {
constraints.checks.push(detached_bound_check(
target,
parent_spec,
&bound,
&constraints.checks,
));
}
let mut hierarchy = constraints.hierarchy.clone();
hierarchy.partition_identity_overrides.clear();
hierarchy.partition_inherited_key_constraints.clear();
hierarchy.partition_inherited_foreign_keys.clear();
if target == &partition {
hierarchy.parents.clear();
hierarchy.parent_sequence_numbers.clear();
hierarchy.partition_bound = None;
}
super::publication::hierarchy::replace_hierarchy_components(
&context.publication,
context.catalog,
target,
HierarchySchemaChange {
columns,
checks: constraints.checks,
foreign_keys: constraints.foreign_keys,
key_constraints: constraints.key_constraints,
hierarchy,
},
)
.map_err(|error| ddl_storage_error("DETACH PARTITION", error))?;
}
Ok(())
}
fn validate_row_type(
context: &HierarchyContext<'_>,
parent: &str,
child: &str,
exact_columns: bool,
reject_child_identity: bool,
) -> Result<(), SQLError> {
let parent_columns = table_columns(context, parent, "ALTER TABLE hierarchy")?;
let child_columns = table_columns(context, child, "ALTER TABLE hierarchy")?;
uqa_sql::schema::inheritance::alter::validate_row_type(
&parent_columns,
&child_columns,
parent,
child,
exact_columns,
reject_child_identity,
)
}
fn validate_inherited_checks(
context: &HierarchyContext<'_>,
parent: &str,
child: &str,
) -> Result<(), SQLError> {
let child_columns = context
.catalog
.try_describe_table(child)
.map_err(|error| ddl_storage_error("read child CHECK columns", error))?
.ok_or_else(|| SQLError::UnknownTable(child.to_string()))?;
let parent_checks = context
.catalog
.try_check_constraint_definitions(parent)
.map_err(|error| ddl_storage_error("read parent CHECK constraints", error))?;
let child_checks = context
.catalog
.try_check_constraint_definitions(child)
.map_err(|error| ddl_storage_error("read child CHECK constraints", error))?;
uqa_sql::schema::inheritance::alter::validate_inherited_checks(
child,
&child_columns,
&parent_checks,
&child_checks,
)
}
fn validate_attached_rows(
context: &HierarchyContext<'_>,
parent: &str,
partition: &str,
bound: &PartitionBound,
) -> Result<(), SQLError> {
for physical_table in context
.constraints
.catalog
.hierarchy_scan_tables(partition, true)?
{
for doc_id in context
.constraints
.reads
.live_table_doc_ids(&physical_table)?
{
let Some(document) = context
.constraints
.reads
.get_document(&physical_table, doc_id)?
else {
continue;
};
if !uqa_sql::semantics::partition::prospective_partition_bound_accepts_document(
&context.partitions,
parent,
bound,
&document,
)? {
return Err(routine(
"23514",
format!(
"partition constraint of relation \"{}\" is violated by some row",
local_relation_name(&physical_table)
),
));
}
}
}
Ok(())
}
fn validate_default_partition_exclusion(
context: &HierarchyContext<'_>,
parent: &str,
new_bound: &PartitionBound,
) -> Result<(), SQLError> {
if matches!(new_bound, PartitionBound::Default) {
return Ok(());
}
let Some(default) = direct_default_partition(context, parent)? else {
return Ok(());
};
for physical_table in context
.constraints
.catalog
.hierarchy_scan_tables(&default, true)?
{
for doc_id in context
.constraints
.reads
.live_table_doc_ids(&physical_table)?
{
let Some(document) = context
.constraints
.reads
.get_document(&physical_table, doc_id)?
else {
continue;
};
if uqa_sql::semantics::partition::prospective_partition_bound_accepts_document(
&context.partitions,
parent,
new_bound,
&document,
)? {
return Err(routine(
"23514",
format!(
"updated partition constraint for default partition \"{}\" would be violated by some row",
local_relation_name(&default)
),
));
}
}
}
Ok(())
}
fn validate_existing_constraints(
context: &HierarchyContext<'_>,
table: &str,
) -> Result<(), SQLError> {
for doc_id in context.constraints.reads.live_table_doc_ids(table)? {
let Some(document) = context.constraints.reads.get_document(table, doc_id)? else {
continue;
};
crate::mutation::constraints::validate_document_constraints(
context.constraints,
table,
&document,
&[],
Some(doc_id),
)?;
}
Ok(())
}
fn direct_default_partition(
context: &HierarchyContext<'_>,
parent: &str,
) -> Result<Option<String>, SQLError> {
for child in context
.partitions
.catalog
.direct_hierarchy_children(parent)?
{
if matches!(
read_hierarchy(context, &child)?.partition_bound,
Some(PartitionBound::Default)
) {
return Ok(Some(child));
}
}
Ok(None)
}
fn replace_hierarchy_only(
context: &HierarchyContext<'_>,
table: &str,
hierarchy: TableHierarchy,
action: &str,
) -> Result<(), SQLError> {
let columns = table_columns(context, table, action)?;
let constraints = declared_constraints(context, table, action)?;
super::publication::hierarchy::replace_hierarchy_components(
&context.publication,
context.catalog,
table,
HierarchySchemaChange {
columns,
checks: constraints.checks,
foreign_keys: constraints.foreign_keys,
key_constraints: constraints.key_constraints,
hierarchy,
},
)
.map_err(|error| ddl_storage_error(action, error))
}
fn table_columns(
context: &HierarchyContext<'_>,
table: &str,
action: &str,
) -> Result<Vec<ColumnDef>, SQLError> {
context
.catalog
.try_describe_table(table)
.map_err(|error| ddl_storage_error(action, error))?
.ok_or_else(|| SQLError::UnknownTable(table.to_string()))
}
fn declared_constraints(
context: &HierarchyContext<'_>,
table: &str,
action: &str,
) -> Result<uqa_sql::ast::TableConstraintSet, SQLError> {
context
.catalog
.try_declared_table_constraints(table)
.map_err(|error| ddl_storage_error(action, error))
}
fn read_hierarchy(context: &HierarchyContext<'_>, table: &str) -> Result<TableHierarchy, SQLError> {
context
.partitions
.catalog
.try_table_hierarchy(table)
.map_err(|error| SQLError::Internal(format!("read table hierarchy: {error}")))
}
fn resolve_table(context: &HierarchyContext<'_>, requested: &str) -> Result<String, SQLError> {
match context.namespace.resolve_visible_relation_kind(requested)? {
RelationResolution::Found(canonical, "table") => Ok(canonical),
RelationResolution::Found(canonical, kind) => Err(wrong_object(format!(
"relation \"{canonical}\" is a {kind}, not a table"
))),
RelationResolution::MissingSchema(schema) => Err(routine(
"3F000",
format!("schema \"{schema}\" does not exist"),
)),
RelationResolution::MissingRelation => Err(routine(
"42P01",
format!("relation \"{requested}\" does not exist"),
)),
}
}
fn lock_secondary_relation(
context: &HierarchyContext<'_>,
primary: &str,
secondary: &str,
) -> Result<(), SQLError> {
if primary != secondary {
context.namespace.lock_exclusive(secondary)?;
}
Ok(())
}
fn validate_matching_persistence(
context: &HierarchyContext<'_>,
child: &str,
parent: &str,
operation: &str,
) -> Result<(), SQLError> {
let child_persistence = context
.catalog
.table_persistence(child)
.map_err(|error| ddl_storage_error("read child persistence", error))?
.ok_or_else(|| SQLError::UnknownTable(child.to_string()))?;
let parent_persistence = context
.catalog
.table_persistence(parent)
.map_err(|error| ddl_storage_error("read parent persistence", error))?
.ok_or_else(|| SQLError::UnknownTable(parent.to_string()))?;
uqa_sql::schema::inheritance::alter::validate_matching_persistence(
child,
parent,
operation,
child_persistence,
parent_persistence,
)
}
fn local_relation_name(name: &str) -> &str {
name.rsplit('.').next().unwrap_or(name)
}
fn wrong_object(message: impl Into<String>) -> SQLError {
routine("42809", message)
}
fn routine(sqlstate: &str, message: impl Into<String>) -> SQLError {
SQLError::Routine {
sqlstate: sqlstate.into(),
message: message.into(),
}
}