use alloc::boxed::Box;
use alloc::string::{String, ToString};
use alloc::vec::Vec;
use spg_sql::ast::Expr;
use spg_storage::{Catalog, ColumnSchema, Row, StorageError, Value};
use crate::aggregate;
use crate::eval::{self, EvalError};
use crate::{Engine, EngineError, check_unsigned_range, coerce_value, value_to_literal_expr};
type KeyStrFn<'a> = dyn Fn(&[Value<'static>]) -> Result<Option<String>, EngineError> + 'a;
pub(crate) fn resolve_foreign_key(
local_table_name: &str,
local_cols: &[ColumnSchema],
fk: spg_sql::ast::ForeignKeyConstraint,
catalog: &Catalog,
) -> Result<spg_storage::ForeignKeyConstraint, EngineError> {
let mut local_columns = Vec::with_capacity(fk.columns.len());
for name in &fk.columns {
let pos = local_cols
.iter()
.position(|c| c.name == *name)
.ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"FOREIGN KEY references unknown local column {name:?}"
))
})?;
local_columns.push(pos);
}
let is_self_ref = fk.parent_table == local_table_name;
let (parent_cols_for_lookup, parent_table_str): (&[ColumnSchema], &str) = if is_self_ref {
(local_cols, local_table_name)
} else {
let parent_table = catalog.get(&fk.parent_table).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: fk.parent_table.clone(),
})
})?;
(
parent_table.schema().columns.as_slice(),
fk.parent_table.as_str(),
)
};
let parent_columns: Vec<usize> = if fk.parent_columns.is_empty() {
if fk.columns.len() != 1 {
return Err(EngineError::Unsupported(
"composite FOREIGN KEY without explicit parent column list is not supported \
— list the parent columns explicitly"
.into(),
));
}
let pos = pick_pk_index_column(catalog, parent_table_str, is_self_ref, local_cols)
.ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"parent table {parent_table_str:?} has no PRIMARY-key / UNIQUE BTree index \
to default the FOREIGN KEY against"
))
})?;
alloc::vec![pos]
} else {
let mut out = Vec::with_capacity(fk.parent_columns.len());
for name in &fk.parent_columns {
let pos = parent_cols_for_lookup
.iter()
.position(|c| c.name == *name)
.ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"FOREIGN KEY references unknown parent column \
{name:?} on table {parent_table_str:?}"
))
})?;
out.push(pos);
}
out
};
if parent_columns.len() != local_columns.len() {
return Err(EngineError::Unsupported(alloc::format!(
"FOREIGN KEY arity mismatch: {} local columns vs {} parent columns",
local_columns.len(),
parent_columns.len()
)));
}
if !is_self_ref {
let parent_table = catalog.get(&fk.parent_table).expect("checked above");
let primary_parent_col = parent_columns[0];
let has_btree = parent_table
.schema()
.columns
.get(primary_parent_col)
.is_some()
&& parent_table.indices().iter().any(|idx| {
matches!(idx.kind, spg_storage::IndexKind::BTree(_))
&& idx.column_position == primary_parent_col
&& idx.partial_predicate.is_none()
});
if !has_btree {
return Err(EngineError::Unsupported(alloc::format!(
"FOREIGN KEY parent column on {:?} is not covered by an unconditional BTree \
index — create one with `CREATE INDEX ... ON {} ({})` first",
parent_table_str,
parent_table_str,
parent_table.schema().columns[primary_parent_col].name,
)));
}
}
let on_delete = fk_action_sql_to_storage(fk.on_delete);
let on_update = fk_action_sql_to_storage(fk.on_update);
let match_type = match fk.match_type {
spg_sql::ast::MatchType::Simple => spg_storage::MatchType::Simple,
spg_sql::ast::MatchType::Full => spg_storage::MatchType::Full,
};
Ok(spg_storage::ForeignKeyConstraint {
name: fk.name,
local_columns,
parent_table: fk.parent_table,
parent_columns,
on_delete,
on_update,
deferrable: fk.deferrable,
initially_deferred: fk.initially_deferred,
match_type,
})
}
fn pick_pk_index_column(
catalog: &Catalog,
parent_name: &str,
is_self_ref: bool,
local_cols: &[ColumnSchema],
) -> Option<usize> {
if is_self_ref {
let _ = local_cols;
return Some(0);
}
let parent = catalog.get(parent_name)?;
parent.indices().iter().find_map(|idx| {
if matches!(idx.kind, spg_storage::IndexKind::BTree(_))
&& idx.partial_predicate.is_none()
&& idx.included_columns.is_empty()
&& idx.expression.is_none()
{
Some(idx.column_position)
} else {
None
}
})
}
pub(crate) fn on_conflict_arbiters(
catalog: &Catalog,
table_name: &str,
target: &[String],
from_constraint_name: bool,
) -> Result<Vec<(Vec<usize>, bool)>, EngineError> {
let table = catalog.get(table_name).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: table_name.into(),
})
})?;
let schema = table.schema();
let unique_btree_cols: Vec<usize> = table
.indices()
.iter()
.filter(|idx| {
idx.is_unique
&& matches!(idx.kind, spg_storage::IndexKind::BTree(_))
&& idx.partial_predicate.is_none()
&& idx.expression.is_none()
})
.map(|idx| idx.column_position)
.collect();
if target.is_empty() {
let mut out: Vec<(Vec<usize>, bool)> = schema
.uniqueness_constraints
.iter()
.map(|uc| (uc.columns.clone(), uc.nulls_not_distinct))
.collect();
for &pos in &unique_btree_cols {
if !out.iter().any(|(cols, _)| cols == &alloc::vec![pos]) {
out.push((alloc::vec![pos], false));
}
}
if out.is_empty() {
for idx in table.indices() {
if matches!(idx.kind, spg_storage::IndexKind::BTree(_))
&& idx.partial_predicate.is_none()
&& idx.expression.is_none()
&& idx.included_columns.is_empty()
{
out.push((alloc::vec![idx.column_position], false));
}
}
}
return Ok(out);
}
let mut positions = Vec::with_capacity(target.len());
for name in target {
let pos = schema
.columns
.iter()
.position(|c| c.name == *name)
.ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"ON CONFLICT target column {name:?} not found on {table_name:?}"
))
})?;
positions.push(pos);
}
let mut sorted = positions.clone();
sorted.sort_unstable();
let matched_uc = schema.uniqueness_constraints.iter().find(|uc| {
let mut u = uc.columns.clone();
u.sort_unstable();
u == sorted
});
let _ = from_constraint_name;
let nnd = matched_uc.is_some_and(|uc| uc.nulls_not_distinct);
Ok(alloc::vec![(positions, nnd)])
}
fn locator_is_tombstoned(table: &spg_storage::Table, loc: &spg_storage::RowLocator) -> bool {
loc.as_hot()
.is_some_and(|i| table.headers().get(i).is_some_and(|h| h.is_deleted()))
}
fn on_conflict_key_exists(
catalog: &Catalog,
table_name: &str,
column_pos: usize,
key: &Value,
) -> bool {
let Some(table) = catalog.get(table_name) else {
return false;
};
let Some(idx_key) = spg_storage::IndexKey::from_value(key) else {
return false;
};
table.indices().iter().any(|idx| {
matches!(idx.kind, spg_storage::IndexKind::BTree(_))
&& idx.column_position == column_pos
&& idx.partial_predicate.is_none()
&& idx
.lookup_eq(&idx_key)
.iter()
.any(|loc| !locator_is_tombstoned(table, loc))
})
}
pub(crate) fn lookup_row_position_by_keys(
catalog: &Catalog,
table_name: &str,
column_positions: &[usize],
key: &[&Value],
) -> Option<usize> {
let table = catalog.get(table_name)?;
table.rows().iter().enumerate().position(|(row_idx, r)| {
!table.headers().get(row_idx).is_some_and(|h| h.is_deleted())
&& column_positions
.iter()
.enumerate()
.all(|(i, &pos)| r.values.get(pos) == Some(key[i]))
})
}
pub(crate) fn on_conflict_keys_exist(
catalog: &Catalog,
table_name: &str,
column_positions: &[usize],
key: &[&Value],
) -> bool {
if column_positions.len() == 1 {
return on_conflict_key_exists(catalog, table_name, column_positions[0], key[0]);
}
let Some(table) = catalog.get(table_name) else {
return false;
};
let matches = |r: &Row<'static>| {
column_positions
.iter()
.enumerate()
.all(|(i, &pos)| r.values.get(pos) == Some(key[i]))
};
let hot_hit = table.rows().iter().enumerate().any(|(row_idx, r)| {
!table.headers().get(row_idx).is_some_and(|h| h.is_deleted()) && matches(r)
});
if hot_hit {
return true;
}
iter_cold_rows_of_parent(catalog, table)
.iter()
.any(&matches)
}
pub(crate) fn apply_on_conflict_assignments(
catalog: &Catalog,
table_name: &str,
alias: Option<&str>,
target_pos: usize,
incoming: &[Value<'static>],
assignments: &[(String, Expr)],
where_: Option<&Expr>,
sess: Option<&crate::eval::DmlSession>,
) -> Result<Option<Vec<Value<'static>>>, EngineError> {
let table = catalog.get(table_name).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: table_name.into(),
})
})?;
let schema_cols = table.schema().columns.clone();
let existing = table
.rows()
.get(target_pos)
.ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"ON CONFLICT DO UPDATE: row position {target_pos} out of bounds on {table_name:?}"
))
})?
.clone();
let mut ctx = eval::EvalContext::new(&schema_cols, Some(alias.unwrap_or(table_name)));
if let Some(sv) = sess {
ctx = ctx.with_session(sv);
}
if let Some(w) = where_ {
let pred = w.clone();
let pred = substitute_excluded_refs(pred, &schema_cols, incoming);
let v = eval::eval_expr(&pred, &existing, &ctx)?;
if !matches!(v, Value::Bool(true)) {
return Ok(None);
}
}
if assignments.is_empty() {
return Ok(Some(incoming.to_vec()));
}
let mut new_values = existing.values.clone();
for (col_name, expr) in assignments {
let target_idx = schema_cols
.iter()
.position(|c| c.name == *col_name)
.ok_or_else(|| {
EngineError::Eval(EvalError::ColumnNotFound {
name: col_name.clone(),
})
})?;
let sub = substitute_excluded_refs(expr.clone(), &schema_cols, incoming);
let v = eval::eval_expr(&sub, &existing, &ctx)?;
let coerced = coerce_value(v, schema_cols[target_idx].ty, col_name, target_idx)?;
let coerced = crate::conversions::truncate_to_column_fsp(coerced, &schema_cols[target_idx]);
check_unsigned_range(&coerced, &schema_cols[target_idx], target_idx)?;
new_values[target_idx] = coerced;
}
Ok(Some(new_values))
}
fn substitute_excluded_refs(
expr: Expr,
schema_cols: &[ColumnSchema],
incoming: &[Value<'static>],
) -> Expr {
use spg_sql::ast::ColumnName;
match expr {
Expr::Column(ColumnName { qualifier, name })
if qualifier
.as_deref()
.is_some_and(|q| q.eq_ignore_ascii_case("excluded")) =>
{
let pos = schema_cols.iter().position(|c| c.name == name);
match pos {
Some(p) => {
let v = incoming.get(p).cloned().unwrap_or(Value::Null);
value_to_literal_expr(v)
.unwrap_or_else(|_| Expr::Literal(spg_sql::ast::Literal::Null))
}
None => Expr::Column(ColumnName { qualifier, name }),
}
}
Expr::Binary { op, lhs, rhs } => Expr::Binary {
op,
lhs: Box::new(substitute_excluded_refs(*lhs, schema_cols, incoming)),
rhs: Box::new(substitute_excluded_refs(*rhs, schema_cols, incoming)),
},
Expr::Unary { op, expr } => Expr::Unary {
op,
expr: Box::new(substitute_excluded_refs(*expr, schema_cols, incoming)),
},
Expr::FunctionCall { name, args } => Expr::FunctionCall {
name,
args: args
.into_iter()
.map(|a| substitute_excluded_refs(a, schema_cols, incoming))
.collect(),
},
Expr::Cast { expr, target } => Expr::Cast {
expr: Box::new(substitute_excluded_refs(*expr, schema_cols, incoming)),
target,
},
Expr::IsNull { expr, negated } => Expr::IsNull {
expr: Box::new(substitute_excluded_refs(*expr, schema_cols, incoming)),
negated,
},
Expr::Like {
expr,
pattern,
negated,
case_insensitive,
} => Expr::Like {
expr: Box::new(substitute_excluded_refs(*expr, schema_cols, incoming)),
pattern: Box::new(substitute_excluded_refs(*pattern, schema_cols, incoming)),
negated,
case_insensitive,
},
Expr::InList {
expr,
list,
negated,
} => Expr::InList {
expr: Box::new(substitute_excluded_refs(*expr, schema_cols, incoming)),
list: list
.into_iter()
.map(|e| substitute_excluded_refs(e, schema_cols, incoming))
.collect(),
negated,
},
Expr::Case {
operand,
branches,
else_branch,
} => Expr::Case {
operand: operand.map(|o| Box::new(substitute_excluded_refs(*o, schema_cols, incoming))),
branches: branches
.into_iter()
.map(|(w, t)| {
(
substitute_excluded_refs(w, schema_cols, incoming),
substitute_excluded_refs(t, schema_cols, incoming),
)
})
.collect(),
else_branch: else_branch
.map(|e| Box::new(substitute_excluded_refs(*e, schema_cols, incoming))),
},
other => other,
}
}
fn indexkeyable_type(ty: &spg_storage::DataType) -> bool {
use spg_storage::DataType as D;
matches!(
ty,
D::SmallInt
| D::Int
| D::BigInt
| D::Text
| D::Varchar(_)
| D::Char(_)
| D::Bool
| D::Uuid
| D::Date
| D::Timestamp
)
}
fn probe_btree(table: &spg_storage::Table, leading_pos: usize) -> Option<&spg_storage::Index> {
table.indices().iter().find(|i| {
matches!(i.kind, spg_storage::IndexKind::BTree(_))
&& i.column_position == leading_pos
&& i.expression.is_none()
&& i.partial_predicate.is_none()
})
}
fn uc_probe_choice<'t>(
table: &'t spg_storage::Table,
columns: &[usize],
nulls_not_distinct: bool,
mysql: bool,
sample: Option<&[Value<'static>]>,
batch_len: usize,
) -> Option<(usize, &'t spg_storage::Index)> {
let sample = sample?;
uc_probe_guards(table, columns, nulls_not_distinct, mysql)?;
let schema = table.schema();
let mut best: Option<(usize, usize, &spg_storage::Index)> = None;
for &col in columns {
if !schema
.columns
.get(col)
.is_some_and(|c| indexkeyable_type(&c.ty))
{
continue;
}
let Some(idx) = probe_btree(table, col) else {
continue;
};
let Some(ik) = sample.get(col).and_then(spg_storage::IndexKey::from_value) else {
continue;
};
let n = idx.lookup_eq(&ik).len();
if best.is_none_or(|(bn, _, _)| n < bn) {
best = Some((n, col, idx));
}
if n == 0 {
break;
}
}
let (locators, col, idx) = best?;
if locators.saturating_mul(batch_len) >= table.rows().len().saturating_add(batch_len) {
crate::bump_counter!(crate::constraints::UNIQ_FOLD_CHOSEN);
return None;
}
Some((col, idx))
}
fn uc_probe_guards(
table: &spg_storage::Table,
columns: &[usize],
nulls_not_distinct: bool,
mysql: bool,
) -> Option<()> {
if nulls_not_distinct || columns.is_empty() {
return None;
}
if mysql {
return None;
}
let schema = table.schema();
let collation_ok = columns.iter().all(|&i| {
schema
.columns
.get(i)
.is_some_and(|c| !matches!(c.collation, spg_storage::Collation::CaseInsensitive))
});
if !collation_ok {
return None;
}
Some(())
}
pub static UNIQ_PROBE_CALLS: core::sync::atomic::AtomicU64 = core::sync::atomic::AtomicU64::new(0);
pub static UNIQ_PROBE_LOCATORS: core::sync::atomic::AtomicU64 =
core::sync::atomic::AtomicU64::new(0);
pub static UNIQ_FOLD_CHOSEN: core::sync::atomic::AtomicU64 = core::sync::atomic::AtomicU64::new(0);
fn probe_key_conflict(
table: &spg_storage::Table,
idx: &spg_storage::Index,
leading_val: &Value<'static>,
key: &[Value<'static>],
fold: &dyn Fn(&[Value<'static>]) -> Vec<Value<'static>>,
) -> Option<usize> {
let ik = spg_storage::IndexKey::from_value(leading_val)?;
crate::bump_counter!(crate::constraints::UNIQ_PROBE_CALLS);
crate::bump_counter!(
crate::constraints::UNIQ_PROBE_LOCATORS,
idx.lookup_eq(&ik).len() as u64
);
for loc in idx.lookup_eq(&ik) {
let spg_storage::RowLocator::Hot(ri) = loc else {
continue;
};
if table.headers().get(*ri).is_some_and(|h| h.is_deleted()) {
continue;
}
let Some(prow) = table.rows().get(*ri) else {
continue;
};
if fold(&prow.values) == key {
return Some(*ri);
}
}
None
}
pub(crate) fn enforce_uniqueness_inserts(
catalog: &Catalog,
child_table: &str,
constraints: &[spg_storage::UniquenessConstraint],
rows: &[Vec<Value<'static>>],
mysql: bool,
) -> Result<(), EngineError> {
if constraints.is_empty() {
return Ok(());
}
let table = catalog.get(child_table).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: child_table.into(),
})
})?;
let schema = table.schema();
for uc in constraints {
let fold_key = |values: &[Value<'static>]| -> Vec<Value<'static>> {
uc.columns
.iter()
.map(|&i| {
let v = values.get(i).cloned().unwrap_or(Value::Null);
collated_key_cell(&v, i, schema, mysql)
})
.collect()
};
let sample = rows
.iter()
.find(|r| !fold_key(r).iter().any(|v| matches!(v, Value::Null)))
.map(alloc::vec::Vec::as_slice);
if let Some((probe_col, idx)) = uc_probe_choice(
table,
&uc.columns,
uc.nulls_not_distinct,
mysql,
sample,
rows.len(),
) {
let mut batch_seen: hashbrown::HashSet<String> =
hashbrown::HashSet::with_capacity(rows.len());
let mut probe_ok = true;
for row_values in rows.iter() {
let key = fold_key(row_values);
if key.iter().any(|v| matches!(v, Value::Null)) && !uc.nulls_not_distinct {
continue;
}
let leading = row_values.get(probe_col).cloned().unwrap_or(Value::Null);
if spg_storage::IndexKey::from_value(&leading).is_none() {
probe_ok = false;
break;
}
let dup_in_batch = !batch_seen.insert(aggregate::encode_key(&key));
if dup_in_batch
|| probe_key_conflict(table, idx, &leading, &key, &fold_key).is_some()
{
let conname = crate::system_catalog::pg_unique_conname(table, uc, child_table);
let detail = unique_key_detail(
&uc.columns
.iter()
.map(|&i| table.schema().columns[i].name.clone())
.collect::<Vec<_>>(),
&key,
);
return Err(EngineError::Unsupported(alloc::format!(
"duplicate key value violates unique constraint \"{conname}\" \
on table \"{child_table}\"{detail}"
)));
}
}
if probe_ok {
continue;
}
}
let mut seen: hashbrown::HashSet<String> =
hashbrown::HashSet::with_capacity(table.rows().len() + rows.len());
for (row_idx, prow) in table.rows().iter().enumerate() {
if table.headers().get(row_idx).is_some_and(|h| h.is_deleted()) {
continue;
}
let key = fold_key(&prow.values);
if key.iter().any(|v| matches!(v, Value::Null)) && !uc.nulls_not_distinct {
continue;
}
seen.insert(aggregate::encode_key(&key));
}
for (batch_idx, row_values) in rows.iter().enumerate() {
let key = fold_key(row_values);
if key.iter().any(|v| matches!(v, Value::Null)) && !uc.nulls_not_distinct {
continue;
}
if !seen.insert(aggregate::encode_key(&key)) {
let conname = crate::system_catalog::pg_unique_conname(table, uc, child_table);
let detail = unique_key_detail(
&uc.columns
.iter()
.map(|&i| table.schema().columns[i].name.clone())
.collect::<Vec<_>>(),
&key,
);
return Err(EngineError::Unsupported(alloc::format!(
"duplicate key value violates unique constraint \"{conname}\" \
on table \"{child_table}\"{detail}"
)));
}
}
}
Ok(())
}
fn exclude_op_binop(op: &str) -> Option<spg_sql::ast::BinOp> {
use spg_sql::ast::BinOp;
Some(match op {
"&&" => BinOp::InetOverlap,
"=" => BinOp::Eq,
"@>" => BinOp::JsonContains,
"<@" => BinOp::JsonContainedBy,
"&<" => BinOp::OverLeft,
"&>" => BinOp::OverRight,
_ => return None,
})
}
fn excl_rows_conflict(
ex: &spg_storage::ExclusionConstraint,
newr: &[Value<'static>],
oldr: &[Value<'static>],
) -> Result<bool, EngineError> {
for (pos, op) in &ex.elements {
let a = newr.get(*pos).cloned().unwrap_or(Value::Null);
let b = oldr.get(*pos).cloned().unwrap_or(Value::Null);
if matches!(a, Value::Null) || matches!(b, Value::Null) {
return Ok(false);
}
let binop = exclude_op_binop(op).ok_or_else(|| {
EngineError::Unsupported(alloc::format!("unsupported EXCLUDE operator {op:?}"))
})?;
match eval::apply_binary(binop, a, b)? {
Value::Bool(true) => {}
_ => return Ok(false),
}
}
Ok(true)
}
enum ExclProbe {
Conflict(Vec<Value<'static>>),
NoOverlap,
Inconclusive,
}
enum KeyProbe {
Conflict(Vec<Value<'static>>),
LiveClear,
AllDead,
Absent,
}
fn excl_probe_existing(
table: &spg_storage::Table,
ex: &spg_storage::ExclusionConstraint,
index_col: usize,
newr: &[Value<'static>],
exclude: Option<&hashbrown::HashSet<usize>>,
) -> Result<ExclProbe, EngineError> {
let Some(map) = table.excl_range_index(index_col) else {
return Ok(ExclProbe::Inconclusive);
};
let cand = newr.get(index_col).cloned().unwrap_or(Value::Null);
if matches!(cand, Value::Null) {
return Ok(ExclProbe::NoOverlap); }
let Some(cand_key) = spg_storage::range_excl_index_key(&cand) else {
return Ok(ExclProbe::Inconclusive); };
let probe_entry = |entry: Option<(&(i128, u8), &Vec<spg_storage::RowLocator>)>|
-> Result<KeyProbe, EngineError> {
let Some((_, locs)) = entry else {
return Ok(KeyProbe::Absent);
};
let mut saw_live = false;
for loc in locs {
if locator_is_tombstoned(table, loc) {
continue;
}
let spg_storage::RowLocator::Hot(ri) = loc else {
continue; };
if exclude.is_some_and(|s| s.contains(ri)) {
continue;
}
let Some(prow) = table.rows().get(*ri) else {
continue;
};
saw_live = true;
if excl_rows_conflict(ex, newr, &prow.values)? {
return Ok(KeyProbe::Conflict(prow.values.clone()));
}
}
Ok(if saw_live {
KeyProbe::LiveClear
} else {
KeyProbe::AllDead
})
};
let pred = probe_entry(map.predecessor(&cand_key))?;
if let KeyProbe::Conflict(old) = pred {
return Ok(ExclProbe::Conflict(old));
}
let succ = probe_entry(
map.range(
core::ops::Bound::Included(&cand_key),
core::ops::Bound::Unbounded,
)
.next(),
)?;
if let KeyProbe::Conflict(old) = succ {
return Ok(ExclProbe::Conflict(old));
}
if matches!(pred, KeyProbe::AllDead) || matches!(succ, KeyProbe::AllDead) {
Ok(ExclProbe::Inconclusive)
} else {
Ok(ExclProbe::NoOverlap)
}
}
pub(crate) fn enforce_exclusion_inserts(
catalog: &Catalog,
child_table: &str,
constraints: &[spg_storage::ExclusionConstraint],
rows: &[Vec<Value<'static>>],
) -> Result<(), EngineError> {
if constraints.is_empty() {
return Ok(());
}
let table = catalog.get(child_table).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: child_table.into(),
})
})?;
let conflicts = excl_rows_conflict;
for ex in constraints {
let idx_col = ex
.elements
.iter()
.find(|(pos, op)| op == "&&" && table.excl_range_index(*pos).is_some())
.map(|(pos, _)| *pos);
for newr in rows.iter() {
let mut proved_clear = false;
if let Some(col) = idx_col {
match excl_probe_existing(table, ex, col, newr, None)? {
ExclProbe::Conflict(old) => {
return Err(exclusion_violation(table, ex, child_table, newr, &old));
}
ExclProbe::NoOverlap => proved_clear = true,
ExclProbe::Inconclusive => {} }
}
if proved_clear {
continue;
}
for (row_idx, prow) in table.rows().iter().enumerate() {
if table.headers().get(row_idx).is_some_and(|h| h.is_deleted()) {
continue;
}
if conflicts(ex, newr, &prow.values)? {
return Err(exclusion_violation(
table,
ex,
child_table,
newr,
&prow.values,
));
}
}
}
if !(ex.elements.len() == 1
&& ex.elements[0].1 == "&&"
&& intra_batch_proven_disjoint(ex.elements[0].0, rows)?)
{
for i in 0..rows.len() {
for j in (i + 1)..rows.len() {
if conflicts(ex, &rows[j], &rows[i])? {
return Err(exclusion_violation(
table,
ex,
child_table,
&rows[j],
&rows[i],
));
}
}
}
}
}
Ok(())
}
fn range_lower_sort_key(v: &Value<'_>) -> Option<(i128, u8)> {
let Value::Range {
lower,
lower_inc,
empty,
..
} = v
else {
return None;
};
if *empty {
return None;
}
let key = match lower {
None => i128::MIN,
Some(b) => match b.as_ref() {
Value::SmallInt(n) => i128::from(*n),
Value::Int(n) => i128::from(*n),
Value::BigInt(n) => i128::from(*n),
Value::Date(n) => i128::from(*n),
Value::Timestamp(n) => i128::from(*n),
_ => return None,
},
};
Some((key, u8::from(!*lower_inc)))
}
fn intra_batch_proven_disjoint(
pos: usize,
rows: &[Vec<Value<'static>>],
) -> Result<bool, EngineError> {
let mut keyed: Vec<((i128, u8), usize)> = Vec::with_capacity(rows.len());
for (i, r) in rows.iter().enumerate() {
match r.get(pos) {
None => return Ok(false), Some(Value::Null) => continue, Some(v @ Value::Range { empty, .. }) => {
if *empty {
continue; }
match range_lower_sort_key(v) {
Some(k) => keyed.push((k, i)),
None => return Ok(false), }
}
Some(_) => return Ok(false), }
}
if keyed.len() < 2 {
return Ok(true); }
keyed.sort_by_key(|k| k.0);
for w in keyed.windows(2) {
let a = rows[w[0].1][pos].clone();
let b = rows[w[1].1][pos].clone();
if let Value::Bool(true) = eval::apply_binary(spg_sql::ast::BinOp::InetOverlap, a, b)? {
return Ok(false);
}
}
Ok(true) }
pub(crate) fn enforce_exclusion_updates(
catalog: &Catalog,
table_name: &str,
constraints: &[spg_storage::ExclusionConstraint],
planned: &[(usize, Vec<Value<'static>>)],
) -> Result<(), EngineError> {
if constraints.is_empty() || planned.is_empty() {
return Ok(());
}
let table = catalog.get(table_name).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: table_name.into(),
})
})?;
let updated: hashbrown::HashSet<usize> = planned.iter().map(|(p, _)| *p).collect();
let conflicts = excl_rows_conflict;
for ex in constraints {
let idx_col = ex
.elements
.iter()
.find(|(pos, op)| op == "&&" && table.excl_range_index(*pos).is_some())
.map(|(pos, _)| *pos);
for (_pos, newr) in planned {
let mut proved_clear = false;
if let Some(col) = idx_col {
match excl_probe_existing(table, ex, col, newr, Some(&updated))? {
ExclProbe::Conflict(old) => {
return Err(exclusion_violation(table, ex, table_name, newr, &old));
}
ExclProbe::NoOverlap => proved_clear = true,
ExclProbe::Inconclusive => {}
}
}
if proved_clear {
continue;
}
for (row_idx, prow) in table.rows().iter().enumerate() {
if table.headers().get(row_idx).is_some_and(|h| h.is_deleted()) {
continue;
}
if updated.contains(&row_idx) {
continue;
}
if conflicts(ex, newr, &prow.values)? {
return Err(exclusion_violation(
table,
ex,
table_name,
newr,
&prow.values,
));
}
}
}
for i in 0..planned.len() {
for j in (i + 1)..planned.len() {
if conflicts(ex, &planned[j].1, &planned[i].1)? {
return Err(exclusion_violation(
table,
ex,
table_name,
&planned[j].1,
&planned[i].1,
));
}
}
}
}
Ok(())
}
fn exclusion_violation(
table: &spg_storage::Table,
ex: &spg_storage::ExclusionConstraint,
child_table: &str,
newr: &[Value<'static>],
oldr: &[Value<'static>],
) -> EngineError {
let render = |vals: &[Value<'static>]| -> (String, String) {
let cols = ex
.elements
.iter()
.map(|(p, _)| table.schema().columns[*p].name.clone())
.collect::<Vec<_>>()
.join(", ");
let rendered = ex
.elements
.iter()
.map(|(p, _)| {
let v = vals.get(*p).cloned().unwrap_or(Value::Null);
match v {
Value::Text(s) => s.to_string(),
other => crate::eval::value_to_text(&other),
}
})
.collect::<Vec<_>>()
.join(", ");
(cols, rendered)
};
let (cols, new_vals) = render(newr);
let (_, old_vals) = render(oldr);
EngineError::Unsupported(alloc::format!(
"conflicting key value violates exclusion constraint \"{}\" \
on table \"{child_table}\" DETAIL: Key ({cols})=({new_vals}) \
conflicts with existing key ({cols})=({old_vals}).",
ex.name
))
}
fn unique_key_detail(cols: &[String], key: &[Value<'_>]) -> String {
let vals = key
.iter()
.map(|v| match v {
Value::Text(s) => s.to_string(),
Value::Null => alloc::string::String::from("null"),
other => crate::eval::value_to_text(other),
})
.collect::<Vec<_>>()
.join(", ");
alloc::format!(
" DETAIL: Key ({})=({vals}) already exists.",
cols.join(", ")
)
}
fn fk_violation_message(
child: &spg_storage::Table,
child_table: &str,
fk: &spg_storage::ForeignKeyConstraint,
key_vals: &[&Value<'_>],
) -> String {
let conname = crate::system_catalog::pg_fk_conname(child, fk, child_table);
let cols = fk
.local_columns
.iter()
.map(|&p| {
child
.schema()
.columns
.get(p)
.map_or_else(|| alloc::format!("col{p}"), |c| c.name.clone())
})
.collect::<Vec<_>>()
.join(", ");
let vals = key_vals
.iter()
.map(|v| match v {
Value::Text(s) => s.to_string(),
other => crate::eval::value_to_text(other),
})
.collect::<Vec<_>>()
.join(", ");
alloc::format!(
"insert or update on table \"{child_table}\" violates foreign key \
constraint \"{conname}\" DETAIL: Key ({cols})=({vals}) is not present \
in table \"{}\".",
fk.parent_table
)
}
fn fk_restrict_message(
catalog: &Catalog,
parent_name: &str,
child: &spg_storage::Table,
child_name: &str,
fk: &spg_storage::ForeignKeyConstraint,
parent_key: &[&Value<'_>],
action: spg_storage::FkAction,
) -> String {
let conname = crate::system_catalog::pg_fk_conname(child, fk, child_name);
let pcols = match catalog.get(parent_name) {
Some(parent) => fk
.parent_columns
.iter()
.map(|&p| {
parent
.schema()
.columns
.get(p)
.map_or_else(|| alloc::format!("col{p}"), |c| c.name.clone())
})
.collect::<Vec<_>>()
.join(", "),
None => "?".into(),
};
let vals = parent_key
.iter()
.map(|v| match v {
Value::Text(s) => s.to_string(),
other => crate::eval::value_to_text(other),
})
.collect::<Vec<_>>()
.join(", ");
if matches!(action, spg_storage::FkAction::Restrict) {
return alloc::format!(
"update or delete on table \"{parent_name}\" violates RESTRICT \
setting of foreign key constraint \"{conname}\" on table \"{child_name}\" \
DETAIL: Key ({pcols})=({vals}) is referenced from table \"{child_name}\"."
);
}
alloc::format!(
"update or delete on table \"{parent_name}\" violates foreign key \
constraint \"{conname}\" on table \"{child_name}\" \
DETAIL: Key ({pcols})=({vals}) is still referenced from table \"{child_name}\"."
)
}
fn collated_key_cell(
v: &spg_storage::Value,
column_position: usize,
schema: &spg_storage::TableSchema,
mysql: bool,
) -> spg_storage::Value<'static> {
let explicit_binary = schema
.columns
.get(column_position)
.is_some_and(|c| matches!(c.collation, spg_storage::Collation::Binary));
if mysql && !explicit_binary {
match v {
spg_storage::Value::Text(s) => {
return spg_storage::Value::text(spg_storage::mysql_compare_fold(s));
}
spg_storage::Value::BpChar(s) => {
return spg_storage::Value::text(spg_storage::mysql_ci_fold(
s.trim_end_matches(' '),
));
}
_ => return v.clone().into_owned(),
}
}
match (v, schema.columns.get(column_position).map(|c| c.collation)) {
(spg_storage::Value::Text(s), Some(spg_storage::Collation::CaseInsensitive)) => {
spg_storage::Value::text(s.to_ascii_lowercase())
}
_ => v.clone().into_owned(),
}
}
fn predicate_truthy(v: &spg_storage::Value) -> bool {
use spg_storage::Value as V;
match v {
V::Bool(b) => *b,
V::Int(n) => *n != 0,
V::BigInt(n) => *n != 0,
V::SmallInt(n) => *n != 0,
_ => false,
}
}
pub(crate) fn check_existing_unique_violation(
idx: &spg_storage::Index,
schema: &spg_storage::TableSchema,
rows: &[spg_storage::Row<'static>],
mysql: bool,
) -> Result<(), EngineError> {
let predicate_expr = match idx.partial_predicate.as_deref() {
Some(s) => Some(spg_sql::parser::parse_expression(s).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"stored partial predicate {s:?} failed to re-parse: {e:?}"
))
})?),
None => None,
};
let ctx = eval::EvalContext::new(&schema.columns, None);
let key_positions = unique_key_positions(idx);
let mut seen: alloc::vec::Vec<alloc::vec::Vec<spg_storage::Value<'static>>> =
alloc::vec::Vec::new();
for row in rows {
if let Some(expr) = &predicate_expr {
let v = eval::eval_expr(expr, row, &ctx).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"evaluating UNIQUE INDEX predicate against existing row: {e:?}"
))
})?;
if !predicate_truthy(&v) {
continue;
}
}
let key: alloc::vec::Vec<spg_storage::Value<'static>> = key_positions
.iter()
.map(|&p| {
let v = row
.values
.get(p)
.cloned()
.unwrap_or(spg_storage::Value::Null);
collated_key_cell(&v, p, schema, mysql)
})
.collect();
if !idx.nulls_not_distinct && key.iter().any(|v| matches!(v, spg_storage::Value::Null)) {
continue;
}
if seen.iter().any(|other| *other == key) {
return Err(EngineError::Unsupported(alloc::format!(
"could not create unique index {:?}",
idx.name
)));
}
seen.push(key);
}
Ok(())
}
fn unique_key_positions(idx: &spg_storage::Index) -> alloc::vec::Vec<usize> {
let mut out = alloc::vec::Vec::with_capacity(1 + idx.extra_column_positions.len());
out.push(idx.column_position);
out.extend_from_slice(&idx.extra_column_positions);
out
}
pub(crate) fn enforce_unique_index_inserts(
catalog: &Catalog,
table_name: &str,
rows: &[alloc::vec::Vec<spg_storage::Value<'static>>],
mysql: bool,
) -> Result<(), EngineError> {
let table = catalog.get(table_name).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: table_name.into(),
})
})?;
let schema = table.schema();
let ctx = eval::EvalContext::new(&schema.columns, None);
for idx in table.indices() {
if !idx.is_unique {
continue;
}
let predicate_expr = match idx.partial_predicate.as_deref() {
Some(s) => Some(spg_sql::parser::parse_expression(s).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"UNIQUE INDEX {:?} predicate {s:?} failed to re-parse: {e:?}",
idx.name
))
})?),
None => None,
};
let expr_key = match idx.expression.as_deref() {
Some(s) => Some(spg_sql::parser::parse_expression(s).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"UNIQUE INDEX {:?} expression {s:?} failed to re-parse: {e:?}",
idx.name
))
})?),
None => None,
};
let key_positions = unique_key_positions(idx);
let key_col_names: alloc::vec::Vec<alloc::string::String> = match &expr_key {
Some(_) => alloc::vec![idx.expression.clone().unwrap_or_else(|| idx.name.clone())],
None => key_positions
.iter()
.map(|&p| {
schema
.columns
.get(p)
.map_or_else(|| alloc::format!("col{p}"), |c| c.name.clone())
})
.collect(),
};
let key_of = |values: &[spg_storage::Value<'static>]| -> Result<alloc::vec::Vec<spg_storage::Value<'static>>, EngineError> {
if let Some(expr) = &expr_key {
let tmp_row = spg_storage::Row {
values: values.to_vec(),
};
let v = eval::eval_expr(expr, &tmp_row, &ctx).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"UNIQUE INDEX {:?} expression eval: {e:?}",
idx.name
))
})?;
return Ok(alloc::vec![v]);
}
Ok(key_positions
.iter()
.map(|&p| {
let v = values.get(p).cloned().unwrap_or(spg_storage::Value::Null);
collated_key_cell(&v, p, schema, mysql)
})
.collect())
};
let participates = |values: &[spg_storage::Value<'static>]| -> Result<bool, EngineError> {
let Some(expr) = &predicate_expr else {
return Ok(true);
};
let tmp_row = spg_storage::Row {
values: values.to_vec(),
};
let v = eval::eval_expr(expr, &tmp_row, &ctx).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"UNIQUE INDEX {:?} predicate eval: {e:?}",
idx.name
))
})?;
Ok(predicate_truthy(&v))
};
if idx.expression.is_none()
&& idx.partial_predicate.is_none()
&& !idx.nulls_not_distinct
&& matches!(idx.kind, spg_storage::IndexKind::BTree(_))
{
let positions = unique_key_positions(idx);
let schema_ok = !mysql
&& positions.iter().all(|&i| {
schema.columns.get(i).is_some_and(|c| {
!matches!(c.collation, spg_storage::Collation::CaseInsensitive)
})
})
&& schema
.columns
.get(idx.column_position)
.is_some_and(|c| indexkeyable_type(&c.ty));
if schema_ok {
let fold =
|values: &[spg_storage::Value<'static>]| -> Vec<spg_storage::Value<'static>> {
positions
.iter()
.map(|&p| {
let v = values.get(p).cloned().unwrap_or(spg_storage::Value::Null);
collated_key_cell(&v, p, schema, mysql)
})
.collect()
};
let mut batch_seen: hashbrown::HashSet<String> =
hashbrown::HashSet::with_capacity(rows.len());
let mut probe_ok = true;
for row_values in rows.iter() {
let key = fold(row_values);
if key.iter().any(|v| matches!(v, spg_storage::Value::Null)) {
continue;
}
let leading = row_values
.get(idx.column_position)
.cloned()
.unwrap_or(spg_storage::Value::Null);
if spg_storage::IndexKey::from_value(&leading).is_none() {
probe_ok = false;
break;
}
if !batch_seen.insert(aggregate::encode_key(&key))
|| probe_key_conflict(table, idx, &leading, &key, &fold).is_some()
{
let detail = unique_key_detail(&key_col_names, &key);
return Err(EngineError::Unsupported(alloc::format!(
"duplicate key value violates unique constraint \"{}\" \
on table \"{table_name}\"{detail}",
idx.name
)));
}
}
if probe_ok {
continue;
}
}
}
let mut seen: hashbrown::HashSet<String> =
hashbrown::HashSet::with_capacity(table.rows().len() + rows.len());
for (row_idx, prow) in table.rows().iter().enumerate() {
if table.headers().get(row_idx).is_some_and(|h| h.is_deleted()) {
continue;
}
if !participates(&prow.values)? {
continue;
}
let key = key_of(&prow.values)?;
if !idx.nulls_not_distinct && key.iter().any(|v| matches!(v, spg_storage::Value::Null))
{
continue;
}
seen.insert(aggregate::encode_key(&key));
}
for (batch_idx, row_values) in rows.iter().enumerate() {
if !participates(row_values)? {
continue;
}
let key = key_of(row_values)?;
if !idx.nulls_not_distinct && key.iter().any(|v| matches!(v, spg_storage::Value::Null))
{
continue;
}
if !seen.insert(aggregate::encode_key(&key)) {
let detail = unique_key_detail(&key_col_names, &key);
return Err(EngineError::Unsupported(alloc::format!(
"duplicate key value violates unique constraint \"{}\" \
on table \"{table_name}\"{detail}",
idx.name
)));
}
}
}
Ok(())
}
#[allow(clippy::too_many_lines)]
fn probe_replay(
table: &spg_storage::Table,
idx: &spg_storage::Index,
probe_col: usize,
columns: &[usize],
planned: &[(usize, Vec<Value<'static>>)],
schema: &spg_storage::TableSchema,
key_str: &KeyStrFn<'_>,
on_conflict: &dyn Fn(usize) -> EngineError,
mysql: bool,
) -> Result<bool, EngineError> {
let fold = |values: &[Value<'static>]| -> Vec<Value<'static>> {
columns
.iter()
.map(|&i| {
let v = values.get(i).cloned().unwrap_or(Value::Null);
collated_key_cell(&v, i, schema, mysql)
})
.collect()
};
let mut added: hashbrown::HashSet<String> = hashbrown::HashSet::new();
let mut removed: hashbrown::HashSet<String> = hashbrown::HashSet::new();
for (pos, new_vals) in planned {
let old_key = match table.rows().get(*pos) {
Some(r) => key_str(&r.values)?,
None => None,
};
let new_key = key_str(new_vals)?;
if old_key == new_key {
continue;
}
if let Some(ok) = old_key {
if !added.remove(&ok) {
removed.insert(ok);
}
}
if let Some(nk) = new_key {
if added.contains(&nk) {
return Err(on_conflict(*pos));
}
if !removed.contains(&nk) {
let key_vec = fold(new_vals);
let leading = new_vals.get(probe_col).cloned().unwrap_or(Value::Null);
if spg_storage::IndexKey::from_value(&leading).is_none() {
return Ok(false);
}
if let Some(ri) = probe_key_conflict(table, idx, &leading, &key_vec, &fold)
&& ri != *pos
{
return Err(on_conflict(*pos));
}
}
added.insert(nk);
}
}
Ok(true)
}
pub(crate) fn enforce_unique_updates(
catalog: &Catalog,
table_name: &str,
planned: &[(usize, Vec<Value<'static>>)],
changed_cols: &hashbrown::HashSet<usize>,
mysql: bool,
) -> Result<(), EngineError> {
if planned.is_empty() {
return Ok(());
}
let table = catalog.get(table_name).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: table_name.into(),
})
})?;
let schema = table.schema();
let replay = |key_str: &KeyStrFn<'_>,
on_conflict: &dyn Fn(usize) -> EngineError|
-> Result<(), EngineError> {
let mut index: hashbrown::HashSet<String> =
hashbrown::HashSet::with_capacity(table.rows().len());
for (row_idx, prow) in table.rows().iter().enumerate() {
if table.headers().get(row_idx).is_some_and(|h| h.is_deleted()) {
continue;
}
if let Some(k) = key_str(&prow.values)? {
index.insert(k);
}
}
for (pos, new_vals) in planned {
let old_key = match table.rows().get(*pos) {
Some(r) => key_str(&r.values)?,
None => None,
};
let new_key = key_str(new_vals)?;
if old_key == new_key {
continue; }
if let Some(ok) = &old_key {
index.remove(ok);
}
if let Some(nk) = new_key
&& !index.insert(nk)
{
return Err(on_conflict(*pos));
}
}
Ok(())
};
for uc in &schema.uniqueness_constraints {
if !uc.columns.iter().any(|c| changed_cols.contains(c)) {
continue;
}
let key_str = |values: &[Value<'static>]| -> Result<Option<String>, EngineError> {
let key: Vec<Value<'static>> = uc
.columns
.iter()
.map(|&i| {
let v = values.get(i).cloned().unwrap_or(Value::Null);
collated_key_cell(&v, i, schema, mysql)
})
.collect();
if key.iter().any(|v| matches!(v, Value::Null)) && !uc.nulls_not_distinct {
return Ok(None);
}
Ok(Some(aggregate::encode_key(&key)))
};
let on_conflict = |_pos: usize| -> EngineError {
let conname = if uc.is_primary_key {
alloc::format!("{table_name}_pkey")
} else {
let cols = uc
.columns
.iter()
.map(|&i| schema.columns[i].name.clone())
.collect::<Vec<_>>()
.join("_");
alloc::format!("{table_name}_{cols}_key")
};
EngineError::Unsupported(alloc::format!(
"duplicate key value violates unique constraint \"{conname}\" \
on table \"{table_name}\""
))
};
let sample = planned.iter().map(|(_, v)| v.as_slice()).find(|v| {
!uc.columns
.iter()
.any(|&i| matches!(v.get(i), Some(Value::Null) | None))
});
if let Some((probe_col, pidx)) = uc_probe_choice(
table,
&uc.columns,
uc.nulls_not_distinct,
mysql,
sample,
planned.len(),
) && probe_replay(
table,
pidx,
probe_col,
&uc.columns,
planned,
schema,
&key_str,
&on_conflict,
mysql,
)? {
continue;
}
replay(&key_str, &on_conflict)?;
}
let ctx = eval::EvalContext::new(&schema.columns, None);
for idx in table.indices() {
if !idx.is_unique {
continue;
}
let is_expr_or_partial = idx.expression.is_some() || idx.partial_predicate.is_some();
let key_positions = unique_key_positions(idx);
if !is_expr_or_partial && !key_positions.iter().any(|c| changed_cols.contains(c)) {
continue;
}
let predicate_expr = match idx.partial_predicate.as_deref() {
Some(s) => Some(spg_sql::parser::parse_expression(s).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"UNIQUE INDEX {:?} predicate {s:?} failed to re-parse: {e:?}",
idx.name
))
})?),
None => None,
};
let expr_key = match idx.expression.as_deref() {
Some(s) => Some(spg_sql::parser::parse_expression(s).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"UNIQUE INDEX {:?} expression {s:?} failed to re-parse: {e:?}",
idx.name
))
})?),
None => None,
};
let key_str = |values: &[Value<'static>]| -> Result<Option<String>, EngineError> {
if let Some(pred) = &predicate_expr {
let tmp_row = spg_storage::Row {
values: values.to_vec(),
};
let v = eval::eval_expr(pred, &tmp_row, &ctx).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"UNIQUE INDEX {:?} predicate eval: {e:?}",
idx.name
))
})?;
if !predicate_truthy(&v) {
return Ok(None);
}
}
let key: Vec<Value<'static>> = if let Some(expr) = &expr_key {
let tmp_row = spg_storage::Row {
values: values.to_vec(),
};
let v = eval::eval_expr(expr, &tmp_row, &ctx).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"UNIQUE INDEX {:?} expression eval: {e:?}",
idx.name
))
})?;
alloc::vec![v]
} else {
key_positions
.iter()
.map(|&p| {
let v = values.get(p).cloned().unwrap_or(Value::Null);
collated_key_cell(&v, p, schema, mysql)
})
.collect()
};
if key.iter().any(|v| matches!(v, Value::Null)) {
return Ok(None);
}
Ok(Some(aggregate::encode_key(&key)))
};
let on_conflict = |pos: usize| -> EngineError {
EngineError::Unsupported(alloc::format!(
"UNIQUE INDEX {:?} violation on {table_name:?}: \
UPDATE of row #{pos} duplicates an existing key",
idx.name
))
};
let sample = planned.iter().map(|(_, v)| v.as_slice()).find(|v| {
!key_positions
.iter()
.any(|&i| matches!(v.get(i), Some(Value::Null) | None))
});
if !is_expr_or_partial
&& matches!(idx.kind, spg_storage::IndexKind::BTree(_))
&& let Some((probe_col, pidx)) = uc_probe_choice(
table,
&key_positions,
idx.nulls_not_distinct,
mysql,
sample,
planned.len(),
)
&& probe_replay(
table,
pidx,
probe_col,
&key_positions,
planned,
schema,
&key_str,
&on_conflict,
mysql,
)?
{
continue;
}
replay(&key_str, &on_conflict)?;
}
Ok(())
}
pub(crate) fn any_column_changed(
filter_cols: &[String],
schema_cols: &[ColumnSchema],
old_row: &Row<'static>,
new_row: &Row<'static>,
) -> bool {
for col_name in filter_cols {
let Some(pos) = schema_cols
.iter()
.position(|c| c.name.eq_ignore_ascii_case(col_name))
else {
continue;
};
let old_v = old_row.values.get(pos);
let new_v = new_row.values.get(pos);
if old_v != new_v {
return true;
}
}
false
}
pub(crate) fn format_failing_row(row_values: &[Value<'static>]) -> String {
row_values
.iter()
.map(|v| match v {
Value::Null => "null".to_string(),
Value::Text(s) => s.to_string(),
other => crate::eval::value_to_text(other),
})
.collect::<Vec<_>>()
.join(", ")
}
pub(crate) fn enforce_not_null(
catalog: &Catalog,
table_name: &str,
rows: &[alloc::vec::Vec<Value<'static>>],
) -> Result<(), EngineError> {
let table = catalog.get(table_name).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: table_name.into(),
})
})?;
let cols = &table.schema().columns;
for row in rows {
for (val, col) in row.iter().zip(cols) {
if val.is_null() && !col.nullable {
if let Some(dname) = &col.user_domain_type
&& catalog
.domain_types()
.get(dname)
.is_some_and(|d| !d.nullable)
{
return Err(EngineError::Unsupported(alloc::format!(
"domain {dname} does not allow null values"
)));
}
return Err(EngineError::Unsupported(alloc::format!(
"null value in column \"{}\" of relation \"{table_name}\" \
violates not-null constraint DETAIL: Failing row contains ({}).",
col.name,
format_failing_row(row)
)));
}
}
}
Ok(())
}
pub(crate) fn enforce_check_constraints(
catalog: &Catalog,
table_name: &str,
rows: &[alloc::vec::Vec<spg_storage::Value<'static>>],
sess: Option<&crate::eval::DmlSession>,
) -> Result<(), EngineError> {
let table = catalog.get(table_name).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: table_name.into(),
})
})?;
let schema = table.schema();
let mut domain_checks_per_col: alloc::vec::Vec<(
usize,
String,
alloc::vec::Vec<(String, Expr)>,
)> = alloc::vec::Vec::new();
for (idx, col) in schema.columns.iter().enumerate() {
let Some(dname) = &col.user_domain_type else {
continue;
};
let Some(dom) = catalog.domain_types().get(dname) else {
continue;
};
let mut parsed_for_col: alloc::vec::Vec<(alloc::string::String, Expr)> =
alloc::vec::Vec::with_capacity(dom.checks.len());
for chk in &dom.checks {
let src = &chk.expr;
let expr = spg_sql::parser::parse_expression(src).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"DOMAIN {dname:?} CHECK ({src:?}) on column {:?}: re-parse failed: {e:?}",
col.name
))
})?;
parsed_for_col.push((chk.name.clone(), expr));
}
if !parsed_for_col.is_empty() {
domain_checks_per_col.push((idx, dname.clone(), parsed_for_col));
}
}
if schema.checks.is_empty() && domain_checks_per_col.is_empty() {
return Ok(());
}
let mut ctx = eval::EvalContext::new(&schema.columns, None);
if let Some(s) = sess {
ctx = ctx.with_session(s);
}
let mut parsed: alloc::vec::Vec<(usize, Expr)> = alloc::vec::Vec::new();
for (i, src) in schema.checks.iter().enumerate() {
let expr = spg_sql::parser::parse_expression(&src.expr).map_err(|e| {
let pred = &src.expr;
EngineError::Unsupported(alloc::format!(
"CHECK constraint #{i} on {table_name:?} ({pred:?}) failed to re-parse: {e:?}"
))
})?;
parsed.push((i, expr));
}
for (batch_idx, row_values) in rows.iter().enumerate() {
let tmp_row = spg_storage::Row {
values: row_values.clone(),
};
for (i, expr) in &parsed {
let v = eval::eval_expr(expr, &tmp_row, &ctx).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"CHECK constraint #{i} on {table_name:?} eval at row #{batch_idx}: {e:?}"
))
})?;
if matches!(v, spg_storage::Value::Bool(false)) {
let names =
crate::system_catalog::pg_check_connames(table, table_name, &schema.checks);
let conname = names
.get(*i)
.cloned()
.unwrap_or_else(|| alloc::format!("{table_name}_check"));
let failing = format_failing_row(row_values);
return Err(EngineError::Unsupported(alloc::format!(
"new row for relation \"{table_name}\" violates check constraint \
\"{conname}\" DETAIL: Failing row contains ({failing})."
)));
}
}
for (col_idx, dname, checks) in &domain_checks_per_col {
let cell = row_values
.get(*col_idx)
.cloned()
.unwrap_or(spg_storage::Value::Null);
let synth_cols = alloc::vec![spg_storage::ColumnSchema::new(
"value",
schema.columns[*col_idx].ty,
schema.columns[*col_idx].nullable,
)];
let mut synth_ctx = eval::EvalContext::new(&synth_cols, None);
if let Some(s) = sess {
synth_ctx = synth_ctx.with_session(s);
}
let synth_row = spg_storage::Row {
values: alloc::vec![cell],
};
for (ci, (cname, expr)) in checks.iter().enumerate() {
let v = eval::eval_expr(expr, &synth_row, &synth_ctx).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"DOMAIN CHECK #{ci} on column {:?} eval at row #{batch_idx}: {e:?}",
schema.columns[*col_idx].name
))
})?;
if matches!(v, spg_storage::Value::Bool(false)) {
return Err(EngineError::Unsupported(alloc::format!(
"value for domain {dname} violates check constraint \"{cname}\""
)));
}
}
}
}
Ok(())
}
pub(crate) fn iter_cold_rows_of_parent(
catalog: &Catalog,
parent: &spg_storage::Table,
) -> Vec<Row<'static>> {
let schema = parent.schema();
let Some(pk_col_pos) = schema
.uniqueness_constraints
.iter()
.find(|u| u.is_primary_key && u.columns.len() == 1)
.map(|u| u.columns[0])
else {
return Vec::new();
};
let Some(idx) = parent.indices().iter().find(|i| {
i.column_position == pk_col_pos && matches!(i.kind, spg_storage::IndexKind::BTree(_))
}) else {
return Vec::new();
};
let table_name = schema.name.as_str();
let mut out = Vec::new();
for (key, locators) in idx.iter_asc() {
for loc in locators {
if let spg_storage::RowLocator::Cold { segment_id, .. } = loc
&& let Some(row) = catalog.resolve_cold_locator(table_name, *segment_id, key)
{
out.push(row);
}
}
}
out
}
pub(crate) fn iter_cold_rows_with_locator_map(
catalog: &Catalog,
table: &spg_storage::Table,
) -> (Vec<Row<'static>>, hashbrown::HashMap<i64, usize>) {
let schema = table.schema();
let Some(pk_col_pos) = schema
.uniqueness_constraints
.iter()
.find(|u| u.is_primary_key && u.columns.len() == 1)
.map(|u| u.columns[0])
else {
return (Vec::new(), hashbrown::HashMap::new());
};
let Some(idx) = table.indices().iter().find(|i| {
i.column_position == pk_col_pos && matches!(i.kind, spg_storage::IndexKind::BTree(_))
}) else {
return (Vec::new(), hashbrown::HashMap::new());
};
let table_name = schema.name.as_str();
let mut rows = Vec::new();
let mut map: hashbrown::HashMap<i64, usize> = hashbrown::HashMap::new();
for (key, locators) in idx.iter_asc() {
let spg_storage::IndexKey::Int(pk_value) = key else {
continue;
};
for loc in locators {
if let spg_storage::RowLocator::Cold { segment_id, .. } = loc
&& let Some(row) = catalog.resolve_cold_locator(table_name, *segment_id, key)
{
let offset = rows.len();
rows.push(row);
map.insert(*pk_value, offset);
}
}
}
(rows, map)
}
pub(crate) fn iter_cold_rows_with_pk_key(
catalog: &Catalog,
table: &spg_storage::Table,
) -> Vec<(spg_storage::IndexKey, Row<'static>)> {
let schema = table.schema();
let Some(pk_col_pos) = schema
.uniqueness_constraints
.iter()
.find(|u| u.is_primary_key && u.columns.len() == 1)
.map(|u| u.columns[0])
else {
return Vec::new();
};
let Some(idx) = table.indices().iter().find(|i| {
i.column_position == pk_col_pos && matches!(i.kind, spg_storage::IndexKind::BTree(_))
}) else {
return Vec::new();
};
let table_name = schema.name.as_str();
let mut out = Vec::new();
for (key, locators) in idx.iter_asc() {
for loc in locators {
if let spg_storage::RowLocator::Cold { segment_id, .. } = loc
&& let Some(row) = catalog.resolve_cold_locator(table_name, *segment_id, key)
{
out.push((key.clone(), row));
}
}
}
out
}
pub(crate) fn pk_btree_index_name(table: &spg_storage::Table) -> Option<String> {
let schema = table.schema();
let pk_col_pos = schema
.uniqueness_constraints
.iter()
.find(|u| u.is_primary_key && u.columns.len() == 1)
.map(|u| u.columns[0])?;
table.indices().iter().find_map(|i| {
if i.column_position == pk_col_pos && matches!(i.kind, spg_storage::IndexKind::BTree(_)) {
Some(i.name.clone())
} else {
None
}
})
}
pub(crate) fn enforce_fk_inserts(
catalog: &Catalog,
child_table: &str,
fks: &[spg_storage::ForeignKeyConstraint],
rows: &[Vec<Value<'static>>],
) -> Result<(), EngineError> {
for fk in fks {
let parent_is_self = fk.parent_table == child_table;
let parent = if parent_is_self {
catalog.get(child_table).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: child_table.into(),
})
})?
} else {
catalog.get(&fk.parent_table).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: fk.parent_table.clone(),
})
})?
};
let cold_parent_rows: alloc::vec::Vec<Row<'static>> = if fk.local_columns.len() == 1 {
Vec::new()
} else {
iter_cold_rows_of_parent(catalog, parent)
};
for (batch_idx, row_values) in rows.iter().enumerate() {
if fk.local_columns.len() == 1 {
let v = &row_values[fk.local_columns[0]];
if matches!(v, Value::Null) {
continue;
}
let parent_col = fk.parent_columns[0];
let key = spg_storage::IndexKey::from_value(v).ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"FOREIGN KEY column value of type {} is not index-eligible",
crate::conversions::pg_type_name_for_error_opt(v.data_type())
))
})?;
let present_committed = parent.indices().iter().any(|idx| {
matches!(idx.kind, spg_storage::IndexKind::BTree(_))
&& idx.column_position == parent_col
&& idx.partial_predicate.is_none()
&& idx
.lookup_eq(&key)
.iter()
.any(|loc| !locator_is_tombstoned(parent, loc))
});
let present_in_batch = parent_is_self
&& rows[..batch_idx]
.iter()
.any(|earlier| earlier.get(parent_col) == Some(v));
if !(present_committed || present_in_batch) {
let child = catalog.get(child_table).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: child_table.into(),
})
})?;
return Err(EngineError::Unsupported(fk_violation_message(
child,
child_table,
fk,
&[v],
)));
}
} else {
let null_cnt = fk
.local_columns
.iter()
.filter(|&&i| matches!(row_values.get(i), Some(Value::Null)))
.count();
match fk.match_type {
spg_storage::MatchType::Simple => {
if null_cnt > 0 {
continue;
}
}
spg_storage::MatchType::Full => {
if null_cnt == fk.local_columns.len() {
continue;
}
if null_cnt > 0 {
return Err(EngineError::Unsupported(
"insert or update violates foreign key constraint: MATCH FULL \
does not allow mixing of null and nonnull key values"
.into(),
));
}
}
}
let local: Vec<&Value> = fk.local_columns.iter().map(|&i| &row_values[i]).collect();
let matches_parent_row = |prow: &Row<'static>| {
fk.parent_columns
.iter()
.enumerate()
.all(|(i, &pi)| prow.values.get(pi) == Some(local[i]))
};
let hot_parent_match = parent.rows().iter().enumerate().any(|(row_idx, prow)| {
!parent
.headers()
.get(row_idx)
.is_some_and(|h| h.is_deleted())
&& matches_parent_row(prow)
});
let parent_match_committed =
hot_parent_match || cold_parent_rows.iter().any(&matches_parent_row);
let parent_match_in_batch = parent_is_self
&& rows[..batch_idx].iter().any(|earlier| {
fk.parent_columns
.iter()
.enumerate()
.all(|(i, &pi)| earlier.get(pi) == Some(local[i]))
});
if !(parent_match_committed || parent_match_in_batch) {
let child = catalog.get(child_table).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: child_table.into(),
})
})?;
return Err(EngineError::Unsupported(fk_violation_message(
child,
child_table,
fk,
&local,
)));
}
}
}
}
Ok(())
}
#[derive(Debug, Clone)]
pub(crate) struct FkChildStep {
child_table: String,
action: FkChildAction,
}
#[derive(Debug, Clone)]
pub(crate) enum FkChildAction {
Delete { positions: Vec<usize> },
SetNull {
positions: Vec<usize>,
columns: Vec<usize>,
},
SetDefault {
positions: Vec<usize>,
columns: Vec<usize>,
defaults: Vec<Value<'static>>,
},
}
pub(crate) fn any_fk_child_references(catalog: &Catalog, table_name: &str) -> bool {
catalog.table_names().into_iter().any(|child_name| {
catalog.get(&child_name).is_some_and(|c| {
c.schema()
.foreign_keys
.iter()
.any(|fk| fk.parent_table == table_name)
})
})
}
pub(crate) fn plan_fk_parent_deletions(
catalog: &Catalog,
parent_table_name: &str,
to_delete_positions: &[usize],
to_delete_rows: &[Vec<Value<'static>>],
) -> Result<Vec<FkChildStep>, EngineError> {
use alloc::collections::{BTreeMap, BTreeSet};
if to_delete_rows.is_empty() {
return Ok(Vec::new());
}
let mut delete_plan: BTreeMap<String, BTreeSet<usize>> = BTreeMap::new();
let mut setnull_plan: BTreeMap<String, BTreeSet<(usize, usize)>> = BTreeMap::new();
let mut setdefault_plan: BTreeMap<String, BTreeMap<(usize, usize), Value>> = BTreeMap::new();
let mut visited: BTreeSet<(String, usize)> = BTreeSet::new();
for &p in to_delete_positions {
visited.insert((parent_table_name.to_string(), p));
}
let mut work: Vec<(String, Vec<Value<'static>>)> = to_delete_rows
.iter()
.map(|r| (parent_table_name.to_string(), r.clone()))
.collect();
while let Some((cur_parent, parent_row)) = work.pop() {
for child_name in catalog.table_names() {
let child = catalog
.get(&child_name)
.expect("table_names → catalog.get round-trip is total");
for fk in &child.schema().foreign_keys {
if fk.parent_table != cur_parent {
continue;
}
let parent_key: Vec<&Value> = fk
.parent_columns
.iter()
.map(|&pi| &parent_row[pi])
.collect();
if parent_key.iter().any(|v| matches!(v, Value::Null)) {
continue;
}
if iter_cold_rows_of_parent(catalog, child).iter().any(|crow| {
fk.local_columns
.iter()
.enumerate()
.all(|(i, &li)| crow.values.get(li) == Some(parent_key[i]))
}) {
return Err(EngineError::Unsupported(alloc::format!(
"DELETE on {cur_parent:?}: cold-tier child row in {child_name:?} \
references the doomed parent key; cold-tier mutation by this \
FK action is a v7.37 candidate. Run COMPACT or move the cold \
rows back to the hot tier and retry."
)));
}
for (child_row_idx, child_row) in child.rows().iter().enumerate() {
if child_name == cur_parent
&& visited.contains(&(child_name.clone(), child_row_idx))
{
continue;
}
let matches_key = fk
.local_columns
.iter()
.enumerate()
.all(|(i, &li)| child_row.values.get(li) == Some(parent_key[i]));
if !matches_key {
continue;
}
match fk.on_delete {
spg_storage::FkAction::Restrict | spg_storage::FkAction::NoAction => {
return Err(EngineError::Unsupported(fk_restrict_message(
catalog,
&cur_parent,
child,
&child_name,
fk,
&parent_key,
fk.on_delete,
)));
}
spg_storage::FkAction::Cascade => {
if visited.insert((child_name.clone(), child_row_idx)) {
delete_plan
.entry(child_name.clone())
.or_default()
.insert(child_row_idx);
work.push((child_name.clone(), child_row.values.clone()));
}
}
spg_storage::FkAction::SetNull => {
for &li in &fk.local_columns {
let col = child.schema().columns.get(li).ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"FK local column {li} missing in {child_name:?}"
))
})?;
if !col.nullable {
return Err(EngineError::Unsupported(alloc::format!(
"FOREIGN KEY ON DELETE SET NULL: column \
{child_name:?}.{:?} is NOT NULL — cannot SET NULL",
col.name,
)));
}
}
let entry = setnull_plan.entry(child_name.clone()).or_default();
for &li in &fk.local_columns {
entry.insert((child_row_idx, li));
}
}
spg_storage::FkAction::SetDefault => {
let entry = setdefault_plan.entry(child_name.clone()).or_default();
for &li in &fk.local_columns {
let col = child.schema().columns.get(li).ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"FK local column {li} missing in {child_name:?}"
))
})?;
let default = col.default.clone().ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"FOREIGN KEY ON DELETE SET DEFAULT: column \
{child_name:?}.{:?} has no DEFAULT declared",
col.name,
))
})?;
entry.insert((child_row_idx, li), default);
}
}
}
}
}
}
}
let mut steps: Vec<FkChildStep> = Vec::new();
for (child_table, entries) in setnull_plan {
let (positions, columns): (Vec<usize>, Vec<usize>) = entries.into_iter().unzip();
steps.push(FkChildStep {
child_table,
action: FkChildAction::SetNull { positions, columns },
});
}
for (child_table, entries) in setdefault_plan {
let mut positions = Vec::with_capacity(entries.len());
let mut columns = Vec::with_capacity(entries.len());
let mut defaults = Vec::with_capacity(entries.len());
for ((p, c), v) in entries {
positions.push(p);
columns.push(c);
defaults.push(v);
}
steps.push(FkChildStep {
child_table,
action: FkChildAction::SetDefault {
positions,
columns,
defaults,
},
});
}
for (child_table, positions) in delete_plan {
steps.push(FkChildStep {
child_table,
action: FkChildAction::Delete {
positions: positions.into_iter().collect(),
},
});
}
Ok(steps)
}
pub(crate) fn plan_fk_parent_updates(
catalog: &Catalog,
parent_table_name: &str,
plan_with_old: &[(usize, Vec<Value<'static>>, Vec<Value<'static>>)],
) -> Result<Vec<FkChildStep>, EngineError> {
use alloc::collections::BTreeMap;
if plan_with_old.is_empty() {
return Ok(Vec::new());
}
let delete_plan: BTreeMap<String, alloc::collections::BTreeSet<usize>> = BTreeMap::new();
let mut setnull_plan: BTreeMap<String, alloc::collections::BTreeSet<(usize, usize)>> =
BTreeMap::new();
let mut setdefault_plan: BTreeMap<String, BTreeMap<(usize, usize), Value>> = BTreeMap::new();
let mut cascade_plan: BTreeMap<String, BTreeMap<(usize, usize), Value>> = BTreeMap::new();
for child_name in catalog.table_names() {
let child = catalog
.get(&child_name)
.expect("table_names → catalog.get total");
for fk in &child.schema().foreign_keys {
if fk.parent_table != parent_table_name {
continue;
}
for (_pos, old_row, new_row) in plan_with_old {
let key_changed = fk
.parent_columns
.iter()
.any(|&pi| old_row.get(pi) != new_row.get(pi));
if !key_changed {
continue;
}
let old_key: Vec<&Value> =
fk.parent_columns.iter().map(|&pi| &old_row[pi]).collect();
if old_key.iter().any(|v| matches!(v, Value::Null)) {
continue;
}
let new_key: Vec<&Value> =
fk.parent_columns.iter().map(|&pi| &new_row[pi]).collect();
if iter_cold_rows_of_parent(catalog, child).iter().any(|crow| {
fk.local_columns
.iter()
.enumerate()
.all(|(i, &li)| crow.values.get(li) == Some(old_key[i]))
}) {
return Err(EngineError::Unsupported(alloc::format!(
"UPDATE on {parent_table_name:?}: cold-tier child row in \
{child_name:?} references the changing parent key; cold-tier \
mutation by this FK action is a v7.37 candidate. Run COMPACT \
or move the cold rows back to the hot tier and retry."
)));
}
for (child_row_idx, child_row) in child.rows().iter().enumerate() {
if child_name == parent_table_name
&& plan_with_old.iter().any(|(p, _, _)| *p == child_row_idx)
{
continue;
}
let matches_key = fk
.local_columns
.iter()
.enumerate()
.all(|(i, &li)| child_row.values.get(li) == Some(old_key[i]));
if !matches_key {
continue;
}
match fk.on_update {
spg_storage::FkAction::Restrict | spg_storage::FkAction::NoAction => {
return Err(EngineError::Unsupported(fk_restrict_message(
catalog,
parent_table_name,
child,
&child_name,
fk,
&old_key,
fk.on_update,
)));
}
spg_storage::FkAction::Cascade => {
let entry = cascade_plan.entry(child_name.clone()).or_default();
for (i, &li) in fk.local_columns.iter().enumerate() {
entry.insert((child_row_idx, li), new_key[i].clone());
}
}
spg_storage::FkAction::SetNull => {
for &li in &fk.local_columns {
let col = child.schema().columns.get(li).ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"FK local column {li} missing in {child_name:?}"
))
})?;
if !col.nullable {
return Err(EngineError::Unsupported(alloc::format!(
"FOREIGN KEY ON UPDATE SET NULL: column \
{child_name:?}.{:?} is NOT NULL",
col.name,
)));
}
}
let entry = setnull_plan.entry(child_name.clone()).or_default();
for &li in &fk.local_columns {
entry.insert((child_row_idx, li));
}
}
spg_storage::FkAction::SetDefault => {
let entry = setdefault_plan.entry(child_name.clone()).or_default();
for &li in &fk.local_columns {
let col = child.schema().columns.get(li).ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"FK local column {li} missing in {child_name:?}"
))
})?;
let default = col.default.clone().ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"FOREIGN KEY ON UPDATE SET DEFAULT: column \
{child_name:?}.{:?} has no DEFAULT",
col.name,
))
})?;
entry.insert((child_row_idx, li), default);
}
}
}
}
}
}
}
let mut steps: Vec<FkChildStep> = Vec::new();
for (child_table, entries) in cascade_plan {
let mut positions = Vec::with_capacity(entries.len());
let mut columns = Vec::with_capacity(entries.len());
let mut defaults = Vec::with_capacity(entries.len());
for ((p, c), v) in entries {
positions.push(p);
columns.push(c);
defaults.push(v);
}
steps.push(FkChildStep {
child_table,
action: FkChildAction::SetDefault {
positions,
columns,
defaults,
},
});
}
for (child_table, entries) in setnull_plan {
let (positions, columns): (Vec<usize>, Vec<usize>) = entries.into_iter().unzip();
steps.push(FkChildStep {
child_table,
action: FkChildAction::SetNull { positions, columns },
});
}
for (child_table, entries) in setdefault_plan {
let mut positions = Vec::with_capacity(entries.len());
let mut columns = Vec::with_capacity(entries.len());
let mut defaults = Vec::with_capacity(entries.len());
for ((p, c), v) in entries {
positions.push(p);
columns.push(c);
defaults.push(v);
}
steps.push(FkChildStep {
child_table,
action: FkChildAction::SetDefault {
positions,
columns,
defaults,
},
});
}
let _ = delete_plan; Ok(steps)
}
pub(crate) fn apply_fk_child_step(
catalog: &mut Catalog,
step: &FkChildStep,
) -> Result<(), EngineError> {
let child = catalog.get_mut(&step.child_table).ok_or_else(|| {
EngineError::Storage(StorageError::TableNotFound {
name: step.child_table.clone(),
})
})?;
match &step.action {
FkChildAction::Delete { positions } => {
let _ = child.delete_rows(positions);
}
FkChildAction::SetNull { positions, columns } => {
apply_per_cell_writes(child, positions, columns, |_| Value::Null)?;
}
FkChildAction::SetDefault {
positions,
columns,
defaults,
} => {
apply_per_cell_writes(child, positions, columns, |i| defaults[i].clone())?;
}
}
Ok(())
}
fn apply_per_cell_writes(
child: &mut spg_storage::Table,
positions: &[usize],
columns: &[usize],
mut value_for: impl FnMut(usize) -> Value<'static>,
) -> Result<(), EngineError> {
use alloc::collections::BTreeMap;
let mut by_row: BTreeMap<usize, Vec<(usize, Value<'static>)>> = BTreeMap::new();
for i in 0..positions.len() {
by_row
.entry(positions[i])
.or_default()
.push((columns[i], value_for(i)));
}
for (pos, mutations) in by_row {
let mut new_values = child.rows()[pos].values.clone();
for (col, v) in mutations {
if let Some(slot) = new_values.get_mut(col) {
*slot = v;
}
}
child
.update_row(pos, new_values)
.map_err(EngineError::Storage)?;
}
Ok(())
}
fn fk_action_sql_to_storage(a: spg_sql::ast::FkAction) -> spg_storage::FkAction {
match a {
spg_sql::ast::FkAction::Restrict => spg_storage::FkAction::Restrict,
spg_sql::ast::FkAction::Cascade => spg_storage::FkAction::Cascade,
spg_sql::ast::FkAction::SetNull => spg_storage::FkAction::SetNull,
spg_sql::ast::FkAction::SetDefault => spg_storage::FkAction::SetDefault,
spg_sql::ast::FkAction::NoAction => spg_storage::FkAction::NoAction,
}
}
impl Engine {
pub(crate) fn drain_pending_foreign_keys(&mut self) -> Result<(), EngineError> {
let pending = core::mem::take(&mut self.pending_foreign_keys);
for (child, fk) in pending {
let cols_snapshot = match self.active_catalog().get(&child) {
Some(t) => t.schema().columns.clone(),
None => continue,
};
let storage_fk =
resolve_foreign_key(&child, &cols_snapshot, fk, self.active_catalog())?;
let table = self
.active_catalog_mut()
.get_mut(&child)
.expect("checked above");
table.schema_mut().foreign_keys.push(storage_fk);
}
Ok(())
}
}
impl Engine {
pub(crate) fn fk_is_deferred_now(&self, fk: &spg_storage::ForeignKeyConstraint) -> bool {
if !fk.deferrable {
return false;
}
let Some(tx_id) = self.current_tx else {
return false;
};
let Some(st) = self.tx_catalogs.get(&tx_id) else {
return false;
};
fk_deferred_in(st, fk)
}
pub(crate) fn immediate_fks(
&self,
fks: &[spg_storage::ForeignKeyConstraint],
) -> alloc::vec::Vec<spg_storage::ForeignKeyConstraint> {
fks.iter()
.filter(|fk| !self.fk_is_deferred_now(fk))
.cloned()
.collect()
}
pub(crate) fn run_deferred_fk_checks(&mut self) -> Result<(), EngineError> {
self.run_deferred_fk_checks_inner(None)
}
pub(crate) fn run_deferred_fk_checks_for(
&mut self,
names: &[String],
) -> Result<(), EngineError> {
self.run_deferred_fk_checks_inner(Some(names))
}
fn run_deferred_fk_checks_inner(&mut self, only: Option<&[String]>) -> Result<(), EngineError> {
let Some(tx_id) = self.current_tx else {
return Ok(());
};
let Some(st) = self.tx_catalogs.get(&tx_id) else {
return Ok(());
};
let tables: alloc::vec::Vec<String> = st.touched_tables.iter().cloned().collect();
let deferred_now = |fk: &spg_storage::ForeignKeyConstraint| {
if let Some(names) = only
&& !fk
.name
.as_deref()
.is_some_and(|n| names.iter().any(|w| w == n))
{
return false;
}
fk.deferrable && fk_deferred_in(st, fk)
};
for tname in &tables {
let Some(t) = st.catalog.get(tname) else {
continue;
};
let fks: alloc::vec::Vec<_> = t
.schema()
.foreign_keys
.iter()
.filter(|f| deferred_now(f))
.cloned()
.collect();
if fks.is_empty() {
continue;
}
let rows: alloc::vec::Vec<alloc::vec::Vec<Value<'static>>> = t
.rows()
.iter()
.enumerate()
.filter(|(i, _)| !t.headers().get(*i).is_some_and(|h| h.is_deleted()))
.map(|(_, r)| r.values.clone())
.collect();
enforce_fk_inserts(&st.catalog, tname, &fks, &rows)?;
}
for tname in &tables {
let Some(t) = st.catalog.get(tname) else {
continue;
};
let deferred_ucs: alloc::vec::Vec<(
spg_storage::UniquenessConstraint,
alloc::string::String,
)> = t
.schema()
.uniqueness_constraints
.iter()
.filter(|uc| uc.deferrable)
.map(|uc| {
let conname = crate::system_catalog::pg_unique_conname(t, uc, tname);
(uc.clone(), conname)
})
.filter(|(uc, conname)| {
if let Some(names) = only
&& !names.iter().any(|w| w == conname)
{
return false;
}
uc_deferred_in(st, uc, conname)
})
.collect();
for (uc, _) in &deferred_ucs {
validate_uniqueness_whole_table(&st.catalog, tname, uc, self.backslash_escapes)?;
}
}
Ok(())
}
}
pub(crate) fn uc_deferred_in(
st: &crate::TxState,
uc: &spg_storage::UniquenessConstraint,
conname: &str,
) -> bool {
if let Some(explicit) = st.constraints_deferred_by_name.get(conname) {
return *explicit;
}
st.constraints_deferred.unwrap_or(uc.initially_deferred)
}
pub(crate) fn validate_uniqueness_whole_table(
catalog: &Catalog,
tname: &str,
uc: &spg_storage::UniquenessConstraint,
mysql: bool,
) -> Result<(), EngineError> {
let Some(table) = catalog.get(tname) else {
return Ok(());
};
let schema = table.schema();
let mut seen: hashbrown::HashSet<alloc::string::String> = hashbrown::HashSet::new();
for (i, row) in table.rows().iter().enumerate() {
if table.headers().get(i).is_some_and(|h| h.is_deleted()) {
continue;
}
let key: Vec<Value<'static>> = uc
.columns
.iter()
.map(|&ci| {
let v = row.values.get(ci).cloned().unwrap_or(Value::Null);
collated_key_cell(&v, ci, schema, mysql)
})
.collect();
if !uc.nulls_not_distinct && key.iter().any(Value::is_null) {
continue;
}
let encoded = alloc::format!("{key:?}");
if !seen.insert(encoded) {
let conname = crate::system_catalog::pg_unique_conname(table, uc, tname);
let detail = unique_key_detail(
&uc.columns
.iter()
.map(|&ci| schema.columns[ci].name.clone())
.collect::<Vec<_>>(),
&key,
);
return Err(EngineError::Unsupported(alloc::format!(
"duplicate key value violates unique constraint \"{conname}\" \
on table \"{tname}\"{detail}"
)));
}
}
Ok(())
}
pub(crate) fn fk_deferred_in(st: &crate::TxState, fk: &spg_storage::ForeignKeyConstraint) -> bool {
if let Some(name) = fk.name.as_deref()
&& let Some(explicit) = st.constraints_deferred_by_name.get(name)
{
return *explicit;
}
st.constraints_deferred.unwrap_or(fk.initially_deferred)
}
impl crate::Engine {
pub(crate) fn exec_set_constraints(
&mut self,
names: &[alloc::string::String],
deferred: bool,
) -> Result<crate::QueryResult, EngineError> {
if !self.current_tx.is_some_and(|tx| self.is_tx_open(tx)) {
self.warning(alloc::string::String::from(
"SET CONSTRAINTS can only be used in transaction blocks",
));
}
for n in names {
match self.find_fk_by_name(n) {
Some(fk) if fk.deferrable => {}
Some(_) => {
return Err(EngineError::Unsupported(alloc::format!(
"constraint \"{n}\" is not deferrable"
)));
}
None => match self.find_uc_by_name(n) {
Some(uc) if uc.deferrable => {}
Some(_) => {
return Err(EngineError::Unsupported(alloc::format!(
"constraint \"{n}\" is not deferrable"
)));
}
None => {
return Err(EngineError::Unsupported(alloc::format!(
"constraint \"{n}\" does not exist"
)));
}
},
}
}
if !deferred {
if names.is_empty() {
self.run_deferred_fk_checks()?;
} else {
self.run_deferred_fk_checks_for(names)?;
}
}
if let Some(tx_id) = self.current_tx
&& let Some(st) = self.tx_catalogs.get_mut(&tx_id)
{
if names.is_empty() {
st.constraints_deferred = Some(deferred);
st.constraints_deferred_by_name.clear();
} else {
for n in names {
st.constraints_deferred_by_name.insert(n.clone(), deferred);
}
}
}
Ok(crate::QueryResult::CommandOk {
affected: 0,
modified_catalog: false,
})
}
fn find_uc_by_name(&self, name: &str) -> Option<spg_storage::UniquenessConstraint> {
let cat = self.active_catalog();
cat.table_names().into_iter().find_map(|tname| {
let t = cat.get(&tname)?;
t.schema()
.uniqueness_constraints
.iter()
.find(|uc| crate::system_catalog::pg_unique_conname(t, uc, &tname) == name)
.cloned()
})
}
fn find_fk_by_name(&self, name: &str) -> Option<spg_storage::ForeignKeyConstraint> {
let cat = self.active_catalog();
cat.table_names().into_iter().find_map(|t| {
cat.get(&t).and_then(|tbl| {
tbl.schema()
.foreign_keys
.iter()
.find(|fk| fk.name.as_deref() == Some(name))
.cloned()
})
})
}
}
pub fn validate_check_against_existing_rows(
table: &spg_storage::Table,
table_name: &str,
conname: &str,
expr_src: &str,
) -> Result<(), EngineError> {
let expr = spg_sql::parser::parse_expression(expr_src).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"CHECK constraint {conname:?} on {table_name:?} ({expr_src:?}) failed to parse: {e:?}"
))
})?;
let schema = table.schema();
let ctx = eval::EvalContext::new(&schema.columns, None);
let headers = table.headers();
for (i, row) in table.rows().iter().enumerate() {
if headers
.get(i)
.is_some_and(|h| h.xmax != spg_storage::row_header::XMAX_ALIVE)
{
continue;
}
let v = eval::eval_expr(&expr, row, &ctx).map_err(|e| {
EngineError::Unsupported(alloc::format!(
"CHECK constraint {conname:?} on {table_name:?} eval at row #{i}: {e:?}"
))
})?;
if matches!(v, spg_storage::Value::Bool(false)) {
return Err(EngineError::Unsupported(alloc::format!(
"check constraint \"{conname}\" of relation \"{table_name}\" \
is violated by some row"
)));
}
}
Ok(())
}