use super::{
apply_delete_internal_encoded, extract_primary_key_from_row, split_namespace,
validate_row_constraints,
};
use crate::catalog::schema::{Constraint, ForeignKeyAction, TableSchema};
use crate::catalog::types::{Row, Value};
use crate::catalog::{Catalog, namespace_key};
use crate::error::AedbError;
use crate::error::ResourceType as ErrorResourceType;
use crate::storage::encoded_key::EncodedKey;
use crate::storage::keyspace::Keyspace;
const MAX_CASCADE_DELETE_DEPTH: usize = 8;
pub(super) fn ensure_cascade_delete_depth(cascade_depth: usize) -> Result<(), AedbError> {
if cascade_depth > MAX_CASCADE_DELETE_DEPTH {
return Err(AedbError::Validation(format!(
"cascade delete depth exceeded: depth={}, max_depth={}",
cascade_depth, MAX_CASCADE_DELETE_DEPTH
)));
}
Ok(())
}
pub(super) fn validate_foreign_keys(
catalog: &Catalog,
keyspace: &Keyspace,
schema: &TableSchema,
row: &Row,
_old_row: Option<&Row>,
) -> Result<(), AedbError> {
for fk in &schema.foreign_keys {
let Some(ref_schema) = catalog.tables.get(&(
namespace_key(&fk.references_project_id, &fk.references_scope_id),
fk.references_table.clone(),
)) else {
return Err(AedbError::NotFound {
resource_type: ErrorResourceType::Table,
resource_id: format!(
"{}.{}.{}",
fk.references_project_id, fk.references_scope_id, fk.references_table
),
});
};
let mut values = Vec::with_capacity(fk.columns.len());
let mut any_null = false;
for col in &fk.columns {
let column_index = schema
.columns
.iter()
.position(|c| c.name == *col)
.ok_or_else(|| AedbError::UnknownColumn {
table: schema.table_name.clone(),
column: col.clone(),
})?;
let value = row.values.get(column_index).cloned().unwrap_or(Value::Null);
if matches!(value, Value::Null) {
any_null = true;
}
values.push(value);
}
if any_null {
continue;
}
if ref_schema.primary_key == fk.references_columns {
let exists = keyspace
.row_slot(
&fk.references_project_id,
&fk.references_scope_id,
&fk.references_table,
&EncodedKey::from_values(&values),
)
.is_some();
if !exists {
return Err(AedbError::ForeignKeyViolation {
fk_name: fk.name.clone(),
table: schema.table_name.clone(),
ref_table: fk.references_table.clone(),
ref_key: format!("{values:?}"),
});
}
} else {
let ref_ns = namespace_key(&fk.references_project_id, &fk.references_scope_id);
let lookup_key = EncodedKey::from_values(&values);
let index_name = catalog
.indexes
.iter()
.find(|((idx_ns, idx_table, _), idx_def)| {
idx_ns == &ref_ns
&& idx_table == &fk.references_table
&& idx_def.columns == fk.references_columns
&& matches!(
idx_def.index_type,
crate::catalog::schema::IndexType::UniqueHash
)
})
.map(|((_, _, idx_name), _)| idx_name.clone())
.ok_or_else(|| AedbError::IntegrityError {
message: format!(
"non-primary-key foreign key {} references {} without a unique runtime index",
fk.name, fk.references_table
),
})?;
let ref_table = keyspace
.table_by_namespace_key(&ref_ns, &fk.references_table)
.ok_or_else(|| AedbError::NotFound {
resource_type: ErrorResourceType::Table,
resource_id: format!(
"{}.{}.{}",
fk.references_project_id, fk.references_scope_id, fk.references_table
),
})?;
let reference_unique_index = ref_table
.indexes
.get(&index_name)
.ok_or_else(|| AedbError::IntegrityError {
message: format!(
"missing runtime unique index {index_name} for non-primary-key foreign key {}",
fk.name
),
})?;
if reference_unique_index
.tier_unique_existing(&lookup_key, keyspace.kv_segment_store.as_deref())?
.is_none()
{
return Err(AedbError::ForeignKeyViolation {
fk_name: fk.name.clone(),
table: schema.table_name.clone(),
ref_table: fk.references_table.clone(),
ref_key: format!("{values:?}"),
});
}
continue;
}
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub(super) fn handle_referencing_foreign_keys(
catalog: &Catalog,
keyspace: &mut Keyspace,
ref_project_id: &str,
ref_scope_id: &str,
ref_table_name: &str,
ref_row: &Row,
commit_seq: u64,
cascade_depth: usize,
) -> Result<(), AedbError> {
ensure_cascade_delete_depth(cascade_depth)?;
let target_ns = namespace_key(ref_project_id, ref_scope_id);
for ((dep_ns, dep_table_name), dep_schema) in &catalog.tables {
for fk in &dep_schema.foreign_keys {
if namespace_key(&fk.references_project_id, &fk.references_scope_id) != target_ns
|| fk.references_table != ref_table_name
{
continue;
}
let mut referenced_vals = Vec::with_capacity(fk.references_columns.len());
for ref_col in &fk.references_columns {
let reference_column_index = catalog
.tables
.get(&(target_ns.clone(), ref_table_name.to_string()))
.and_then(|s| s.columns.iter().position(|c| c.name == *ref_col))
.ok_or_else(|| {
AedbError::Validation(format!(
"referenced column missing for fk {}",
fk.name
))
})?;
referenced_vals.push(ref_row.values[reference_column_index].clone());
}
let Some(dep_table) = keyspace
.table_by_namespace_key(dep_ns, dep_table_name)
.cloned()
else {
continue;
};
let mut matched_pks: Vec<EncodedKey> = Vec::new();
for (pk, stored) in dep_table.rows {
let row = keyspace.materialize_row(&stored)?;
let mut matched = true;
for (i, fk_col) in fk.columns.iter().enumerate() {
let fk_idx = dep_schema
.columns
.iter()
.position(|c| c.name == *fk_col)
.ok_or_else(|| AedbError::Validation("fk column missing".into()))?;
if row.values.get(fk_idx) != referenced_vals.get(i) {
matched = false;
break;
}
}
if matched {
matched_pks.push(pk);
}
}
if matched_pks.is_empty() {
continue;
}
match fk.on_delete {
ForeignKeyAction::Restrict | ForeignKeyAction::NoAction => {
return Err(AedbError::ForeignKeyViolation {
fk_name: fk.name.clone(),
table: dep_table_name.clone(),
ref_table: ref_table_name.to_string(),
ref_key: "restrict".into(),
});
}
ForeignKeyAction::Cascade => {
for pk_encoded in matched_pks {
let (dep_project_id, dep_scope_id) = split_namespace(dep_ns)
.ok_or_else(|| AedbError::Validation("invalid namespace key".into()))?;
apply_delete_internal_encoded(
catalog,
keyspace,
&dep_project_id,
&dep_scope_id,
dep_table_name,
&pk_encoded,
commit_seq,
cascade_depth.saturating_add(1),
)?;
}
}
ForeignKeyAction::SetNull | ForeignKeyAction::SetDefault => {
for pk_encoded in matched_pks {
let (dep_project_id, dep_scope_id) = split_namespace(dep_ns)
.ok_or_else(|| AedbError::Validation("invalid namespace key".into()))?;
let dep_ns_key = namespace_key(&dep_project_id, &dep_scope_id);
let dep_schema = catalog
.tables
.get(&(dep_ns_key.clone(), dep_table_name.clone()))
.ok_or_else(|| {
AedbError::Validation("dependent table missing".into())
})?;
let mut row = keyspace
.get_row_by_encoded(
&dep_project_id,
&dep_scope_id,
dep_table_name,
&pk_encoded,
)?
.map(|row| row.into_owned())
.ok_or_else(|| AedbError::Validation("dependent row missing".into()))?;
for fk_col in &fk.columns {
if let Some(column_index) =
dep_schema.columns.iter().position(|c| c.name == *fk_col)
{
match fk.on_delete {
ForeignKeyAction::SetNull => {
row.values[column_index] = Value::Null
}
ForeignKeyAction::SetDefault => {
if let Some(default_value) =
dep_schema.constraints.iter().find_map(|c| match c {
Constraint::Default { column, value }
if column == fk_col =>
{
Some(value.clone())
}
_ => None,
})
{
row.values[column_index] = default_value;
} else {
row.values[column_index] = Value::Null;
}
}
_ => {}
}
}
}
validate_row_constraints(
catalog,
keyspace,
dep_schema,
&dep_project_id,
&dep_scope_id,
dep_table_name,
&extract_primary_key_from_row(dep_schema, &row)?,
&row,
None,
)?;
keyspace.upsert_row_by_encoded_pk(
&dep_project_id,
&dep_scope_id,
dep_table_name,
pk_encoded.clone(),
row.clone(),
commit_seq,
);
}
}
}
}
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub(super) fn handle_referencing_foreign_keys_on_update(
catalog: &Catalog,
keyspace: &mut Keyspace,
ref_project_id: &str,
ref_scope_id: &str,
ref_table_name: &str,
ref_schema: &TableSchema,
old_row: &Row,
new_row: &Row,
commit_seq: u64,
) -> Result<(), AedbError> {
let target_ns = namespace_key(ref_project_id, ref_scope_id);
for ((dep_ns, dep_table_name), dep_schema) in &catalog.tables {
for fk in &dep_schema.foreign_keys {
if namespace_key(&fk.references_project_id, &fk.references_scope_id) != target_ns
|| fk.references_table != ref_table_name
{
continue;
}
let mut old_values = Vec::with_capacity(fk.references_columns.len());
let mut new_values = Vec::with_capacity(fk.references_columns.len());
for ref_col in &fk.references_columns {
let reference_column_index = ref_schema
.columns
.iter()
.position(|c| c.name == *ref_col)
.ok_or_else(|| AedbError::Validation("referenced column missing".into()))?;
old_values.push(old_row.values[reference_column_index].clone());
new_values.push(new_row.values[reference_column_index].clone());
}
if old_values == new_values {
continue;
}
let Some(dep_table) = keyspace
.table_by_namespace_key(dep_ns, dep_table_name)
.cloned()
else {
continue;
};
let mut matched_pks: Vec<EncodedKey> = Vec::new();
for (pk, stored) in dep_table.rows {
let row = keyspace.materialize_row(&stored)?;
let mut matched = true;
for (i, fk_col) in fk.columns.iter().enumerate() {
let fk_idx = dep_schema
.columns
.iter()
.position(|c| c.name == *fk_col)
.ok_or_else(|| AedbError::Validation("fk column missing".into()))?;
if row.values.get(fk_idx) != old_values.get(i) {
matched = false;
break;
}
}
if matched {
matched_pks.push(pk);
}
}
if matched_pks.is_empty() {
continue;
}
match fk.on_update {
ForeignKeyAction::Restrict | ForeignKeyAction::NoAction => {
return Err(AedbError::Validation(format!(
"foreign key update restrict violation: {}",
fk.name
)));
}
ForeignKeyAction::Cascade
| ForeignKeyAction::SetNull
| ForeignKeyAction::SetDefault => {
for pk_encoded in matched_pks {
let (dep_project_id, dep_scope_id) = split_namespace(dep_ns)
.ok_or_else(|| AedbError::Validation("invalid namespace key".into()))?;
let dep_schema = catalog
.tables
.get(&(
namespace_key(&dep_project_id, &dep_scope_id),
dep_table_name.clone(),
))
.ok_or_else(|| {
AedbError::Validation("dependent table missing".into())
})?;
let mut row = keyspace
.get_row_by_encoded(
&dep_project_id,
&dep_scope_id,
dep_table_name,
&pk_encoded,
)?
.map(|row| row.into_owned())
.ok_or_else(|| AedbError::Validation("dependent row missing".into()))?;
for (i, fk_col) in fk.columns.iter().enumerate() {
if let Some(column_index) =
dep_schema.columns.iter().position(|c| c.name == *fk_col)
{
match fk.on_update {
ForeignKeyAction::Cascade => {
row.values[column_index] = new_values[i].clone()
}
ForeignKeyAction::SetNull => {
row.values[column_index] = Value::Null
}
ForeignKeyAction::SetDefault => {
row.values[column_index] = dep_schema
.constraints
.iter()
.find_map(|c| match c {
Constraint::Default { column, value }
if column == fk_col =>
{
Some(value.clone())
}
_ => None,
})
.unwrap_or(Value::Null);
}
_ => {}
}
}
}
validate_row_constraints(
catalog,
keyspace,
dep_schema,
&dep_project_id,
&dep_scope_id,
dep_table_name,
&extract_primary_key_from_row(dep_schema, &row)?,
&row,
None,
)?;
keyspace.upsert_row_by_encoded_pk(
&dep_project_id,
&dep_scope_id,
dep_table_name,
pk_encoded,
row,
commit_seq,
);
}
}
}
}
}
Ok(())
}
pub(super) fn validate_foreign_key_definition(
catalog: &Catalog,
project_id: &str,
fk: &crate::catalog::schema::ForeignKey,
) -> Result<(), AedbError> {
if fk.columns.is_empty() || fk.references_columns.is_empty() {
return Err(AedbError::Validation(
"foreign key columns cannot be empty".into(),
));
}
if fk.columns.len() != fk.references_columns.len() {
return Err(AedbError::Validation(
"foreign key columns length mismatch".into(),
));
}
if fk.references_project_id != project_id {
if fk.references_project_id != "_global" {
return Err(AedbError::Validation(
"project foreign keys may only reference _global or same project".into(),
));
}
if matches!(fk.on_delete, ForeignKeyAction::Cascade) {
return Err(AedbError::Validation(
"project -> _global foreign keys cannot use ON DELETE CASCADE".into(),
));
}
}
if project_id == "_global" && fk.references_project_id != "_global" {
return Err(AedbError::Validation(
"_global tables cannot reference project tables".into(),
));
}
let ref_key = (
namespace_key(&fk.references_project_id, &fk.references_scope_id),
fk.references_table.clone(),
);
let ref_table = catalog
.tables
.get(&ref_key)
.ok_or_else(|| AedbError::NotFound {
resource_type: ErrorResourceType::Table,
resource_id: format!(
"{}.{}.{}",
fk.references_project_id, fk.references_scope_id, fk.references_table
),
})?;
for col in &fk.references_columns {
if !ref_table.columns.iter().any(|c| c.name == *col) {
return Err(AedbError::UnknownColumn {
table: fk.references_table.clone(),
column: col.clone(),
});
}
}
let refs_pk = ref_table.primary_key == fk.references_columns;
let refs_unique = catalog.indexes.values().any(|idx| {
idx.project_id == fk.references_project_id
&& idx.scope_id == fk.references_scope_id
&& idx.table_name == fk.references_table
&& idx.columns == fk.references_columns
&& matches!(
idx.index_type,
crate::catalog::schema::IndexType::UniqueHash
)
});
if !refs_pk && !refs_unique {
return Err(AedbError::Validation(
"foreign key references must target PK or unique index".into(),
));
}
Ok(())
}