use super::{
dml_storage_error, missing_document_error, ConstraintContext, Document, ForeignKey,
PhysicalDocumentIdentity, SQLError, Value,
};
use uqa_sql::{expr::multirange_from_ranges, semantics::period::period_ranges};
pub fn period_foreign_key_coverage(
context: ConstraintContext<'_>,
foreign_key: &ForeignKey,
local_values: &[Value],
excluded_parents: &[PhysicalDocumentIdentity],
replacement_parent: Option<(&PhysicalDocumentIdentity, &Document)>,
) -> Result<(bool, Vec<PhysicalDocumentIdentity>), SQLError> {
super::authorize_foreign_key_parent_namespace(context, foreign_key)?;
let Some(period_column) = foreign_key.ref_columns.last() else {
return Err(SQLError::Internal(
"PERIOD foreign key has no referenced period column".into(),
));
};
let parent_type = context
.catalog
.column_type(&foreign_key.ref_table, period_column)
.map_err(|error| dml_storage_error("PERIOD foreign-key type lookup", error))?
.ok_or_else(|| {
SQLError::UnknownColumn(format!("{}.{period_column}", foreign_key.ref_table))
})?;
let Some(child_period) = local_values.last() else {
return Err(SQLError::Internal(
"PERIOD foreign key has no local period value".into(),
));
};
let (child_subtype, child_ranges) = period_ranges(child_period, &parent_type)?;
if child_ranges.is_empty() {
return Ok((false, Vec::new()));
}
let ordinary_values = &local_values[..local_values.len() - 1];
let ordinary_columns = &foreign_key.ref_columns[..foreign_key.ref_columns.len() - 1];
let mut parent_ranges = Vec::new();
let mut parent_ids = Vec::new();
for physical_table in context
.catalog
.hierarchy_scan_tables(&foreign_key.ref_table, true)?
{
for doc_id in context.reads.table_doc_ids(&physical_table)? {
let identity = PhysicalDocumentIdentity {
table: physical_table.clone(),
doc_id,
};
let replacement = replacement_parent
.filter(|(replacement_identity, _)| *replacement_identity == &identity)
.map(|(_, document)| document);
if replacement.is_none() && excluded_parents.contains(&identity) {
continue;
}
let owned_parent = if replacement.is_some() {
None
} else {
Some(context.reads.get_document(&physical_table, doc_id)?)
};
let parent =
match replacement.or_else(|| owned_parent.as_ref().and_then(Option::as_ref)) {
Some(parent) => parent,
None => {
return Err(missing_document_error(
"PERIOD foreign-key parent scan",
&physical_table,
doc_id,
));
}
};
if !ordinary_columns
.iter()
.zip(ordinary_values)
.all(|(column, value)| parent.get(column).cloned().unwrap_or(Value::Null) == *value)
{
continue;
}
let parent_period = parent.get(period_column).cloned().unwrap_or(Value::Null);
if matches!(parent_period, Value::Null) {
continue;
}
let (parent_subtype, mut ranges) = period_ranges(&parent_period, &parent_type)?;
if parent_subtype != child_subtype {
return Err(SQLError::TypeMismatch(
"PERIOD foreign-key range subtypes do not match".into(),
));
}
parent_ranges.append(&mut ranges);
parent_ids.push(identity);
}
}
let coverage = multirange_from_ranges(child_subtype, parent_ranges);
Ok((
child_ranges
.iter()
.all(|range| coverage.contains_range(range)),
parent_ids,
))
}