use crate::entity::{IEntitySnapshot, IEntityType, IGetKeyValues};
use crate::error::{EFError, EFResult};
use crate::metadata::{EntityTypeMeta, PropertyMeta};
use crate::provider::{DbValue, IAsyncConnection, IDatabaseProvider};
use crate::query::{collect_bool_expr_values, compile_bool_expr, BoolExpr};
use std::collections::HashMap;
pub struct ChangeExecutor;
impl ChangeExecutor {
pub async fn execute_inserts<E, F>(
conn: &mut dyn IAsyncConnection,
provider: &dyn IDatabaseProvider,
entities: &[(&E, &EntityTypeMeta)],
mut on_key_backfill: F,
) -> EFResult<usize>
where
E: IEntityType + IEntitySnapshot + IGetKeyValues,
F: FnMut(usize, i64),
{
if entities.is_empty() {
return Ok(0);
}
let gen = provider.sql_generator();
let meta = entities[0].1;
let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
if scalar_props.is_empty() {
return Ok(0);
}
let insert_cols: Vec<&str> = scalar_props
.iter()
.filter(|p| !p.is_primary_key || (!p.is_auto_increment && !p.is_sequence))
.map(|p| p.column_name.as_ref())
.collect();
if insert_cols.is_empty() {
return Ok(0);
}
let auto_inc_pk = scalar_props
.iter()
.find(|p| (p.is_auto_increment || p.is_sequence) && p.is_primary_key);
const MAX_PARAMS: usize = 900;
let batch_size = (MAX_PARAMS / insert_cols.len()).max(1);
let mut inserted = 0usize;
let mut start = 0usize;
while start < entities.len() {
let end = (start + batch_size).min(entities.len());
let batch = &entities[start..end];
let row_count = batch.len();
let sql = gen.insert_batch(meta.table_name.as_ref(), &insert_cols, row_count);
let mut params: Vec<DbValue> = Vec::with_capacity(row_count * insert_cols.len());
for (entity, _) in batch {
let snap = entity.snapshot();
for p in &scalar_props {
if !p.is_primary_key || (!p.is_auto_increment && !p.is_sequence) {
params.push(
snap.get(p.field_name.as_ref())
.cloned()
.unwrap_or(DbValue::Null),
);
}
}
}
if let Some(_pk_prop) = auto_inc_pk {
if gen.supports_returning() {
let rows = conn.query(&sql, ¶ms).await?;
let affected = rows.len().min(row_count);
for (i, row) in rows.iter().enumerate().take(affected) {
if let Some(pk_val) = row.first() {
if let Ok(key) = pk_val.clone().try_into() {
on_key_backfill(start + i, key);
}
}
}
inserted += affected;
} else if let Some(last_id_sql) = gen.last_insert_id_sql() {
conn.execute(&sql, ¶ms).await?;
let id_rows = conn.query(last_id_sql, &[]).await?;
if let Some(id_row) = id_rows.first() {
if let Some(id_val) = id_row.first() {
let raw_id: i64 = id_val.clone().try_into().unwrap_or(0);
let first_id = if gen.last_insert_id_returns_first() {
raw_id
} else {
raw_id - row_count as i64 + 1
};
for i in 0..row_count {
on_key_backfill(start + i, first_id + i as i64);
}
}
}
inserted += row_count;
} else {
let rows = conn.execute(&sql, ¶ms).await?;
let affected = (rows as usize).min(row_count);
for i in 0..affected {
on_key_backfill(start + i, 0);
}
inserted += affected;
}
} else {
let rows = conn.execute(&sql, ¶ms).await?;
let affected = (rows as usize).min(row_count);
for i in 0..affected {
on_key_backfill(start + i, 0);
}
inserted += affected;
}
start = end;
}
Ok(inserted)
}
pub async fn execute_upserts<E>(
conn: &mut dyn IAsyncConnection,
provider: &dyn IDatabaseProvider,
entities: &[(&E, &EntityTypeMeta)],
) -> EFResult<usize>
where
E: IEntityType + IEntitySnapshot + IGetKeyValues,
{
if entities.is_empty() {
return Ok(0);
}
let gen = provider.sql_generator();
let meta = entities[0].1;
let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
if scalar_props.is_empty() {
return Ok(0);
}
let insert_cols: Vec<&str> = scalar_props
.iter()
.map(|p| p.column_name.as_ref())
.collect();
if insert_cols.is_empty() {
return Ok(0);
}
let conflict_cols: Vec<&str> = scalar_props
.iter()
.filter(|p| p.is_primary_key)
.map(|p| p.column_name.as_ref())
.collect();
if conflict_cols.is_empty() {
return Err(EFError::configuration(
"Upsert requires at least one primary key column as conflict target.",
));
}
const MAX_PARAMS: usize = 900;
let batch_size = (MAX_PARAMS / insert_cols.len()).max(1);
let mut upserted = 0usize;
let mut start = 0usize;
while start < entities.len() {
let end = (start + batch_size).min(entities.len());
let batch = &entities[start..end];
let row_count = batch.len();
let sql = gen.upsert_batch(
meta.table_name.as_ref(),
&insert_cols,
&conflict_cols,
row_count,
);
let mut params: Vec<DbValue> = Vec::with_capacity(row_count * insert_cols.len());
for (entity, _) in batch {
let snap = entity.snapshot();
for p in &scalar_props {
params.push(
snap.get(p.field_name.as_ref())
.cloned()
.unwrap_or(DbValue::Null),
);
}
}
let affected = conn.execute(&sql, ¶ms).await?;
upserted += (affected as usize).min(row_count);
start = end;
}
Ok(upserted)
}
#[allow(clippy::type_complexity)]
pub async fn execute_updates<E>(
conn: &mut dyn IAsyncConnection,
provider: &dyn IDatabaseProvider,
entities: &[(
&E,
&EntityTypeMeta,
Option<&HashMap<String, DbValue>>,
&[String],
)],
query_filter: Option<&BoolExpr>,
) -> EFResult<usize>
where
E: IEntityType + IEntitySnapshot + IGetKeyValues,
{
if entities.is_empty() {
return Ok(0);
}
let gen = provider.sql_generator();
let meta = entities[0].1;
let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
let pk_props: Vec<_> = scalar_props.iter().filter(|p| p.is_primary_key).collect();
let has_concurrency_tokens = scalar_props.iter().any(|p| p.is_concurrency_token);
let modified_fields: std::collections::HashSet<&str> = entities
.iter()
.flat_map(|(_, _, _, mods)| mods.iter().map(|s| s.as_str()))
.collect();
let set_props: Vec<&PropertyMeta> = if modified_fields.is_empty() {
scalar_props
.iter()
.copied()
.filter(|p| !p.is_primary_key)
.collect()
} else {
scalar_props
.iter()
.copied()
.filter(|p| !p.is_primary_key && modified_fields.contains(p.field_name.as_ref()))
.collect()
};
let set_cols: Vec<&str> = set_props.iter().map(|p| p.column_name.as_ref()).collect();
if has_concurrency_tokens || pk_props.len() != 1 || set_cols.is_empty() {
return Self::execute_updates_per_row(conn, gen, entities, query_filter).await;
}
let pk_col = pk_props[0].column_name.as_ref();
let pk_field = pk_props[0].field_name.as_ref();
let entity_data: Vec<(HashMap<String, DbValue>, HashMap<String, DbValue>)> = entities
.iter()
.map(|(e, _, _, _)| (e.snapshot(), e.key_values()))
.collect();
let filter_params: Vec<DbValue> = match query_filter {
Some(filter) => collect_bool_expr_values(filter),
None => Vec::new(),
};
const MAX_PARAMS: usize = 900;
let params_per_row = 2 * set_cols.len() + 1;
let batch_size =
((MAX_PARAMS.saturating_sub(filter_params.len())) / params_per_row.max(1)).max(1);
let mut updated = 0usize;
let mut start = 0usize;
while start < entity_data.len() {
let end = (start + batch_size).min(entity_data.len());
let row_count = end - start;
let set_param_count = 2 * set_cols.len() * row_count;
let mut idx = set_param_count + 1;
let pk_placeholders: Vec<String> = (0..row_count)
.map(|_| {
let ph = gen.parameter_placeholder(idx);
idx += 1;
ph
})
.collect();
let mut where_clause = format!(
"{} IN ({})",
gen.quote_identifier(pk_col),
pk_placeholders.join(", ")
);
let mut params: Vec<DbValue> = Vec::with_capacity(set_param_count + row_count);
for col_prop in &set_props {
for (snap, keys) in entity_data[start..end].iter() {
let pk_val = keys.get(pk_field).cloned().unwrap_or(DbValue::Null);
let col_val = snap
.get(col_prop.field_name.as_ref())
.cloned()
.unwrap_or(DbValue::Null);
params.push(pk_val);
params.push(col_val);
}
}
for (_, keys) in entity_data[start..end].iter() {
let pk_val = keys.get(pk_field).cloned().unwrap_or(DbValue::Null);
params.push(pk_val);
}
if let Some(filter) = query_filter {
let filter_sql = compile_bool_expr(filter, gen, &mut idx);
params.extend(filter_params.iter().cloned());
where_clause = format!("({}) AND ({})", where_clause, filter_sql);
}
let sql = gen.update_batch(
meta.table_name.as_ref(),
&set_cols,
pk_col,
row_count,
&where_clause,
);
let rows = conn.execute(&sql, ¶ms).await?;
if rows == 0 && row_count > 0 {
return Err(EFError::concurrency_conflict(format!(
"batch update affected 0 rows on {} (rows may have been modified or deleted)",
meta.table_name
)));
}
updated += (rows as usize).min(row_count);
start = end;
}
Ok(updated)
}
#[allow(clippy::type_complexity)]
async fn execute_updates_per_row<E>(
conn: &mut dyn IAsyncConnection,
gen: &'static dyn crate::provider::ISqlGenerator,
entities: &[(
&E,
&EntityTypeMeta,
Option<&HashMap<String, DbValue>>,
&[String],
)],
query_filter: Option<&BoolExpr>,
) -> EFResult<usize>
where
E: IEntityType + IEntitySnapshot + IGetKeyValues,
{
let mut updated = 0;
let mut sql_cache: HashMap<(String, Vec<String>, String), String> = HashMap::new();
for (entity, meta, original, modified_props) in entities {
let snap = entity.snapshot();
let keys = entity.key_values();
let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
let set_props: Vec<&PropertyMeta> = if modified_props.is_empty() {
scalar_props
.iter()
.copied()
.filter(|p| !p.is_primary_key)
.collect()
} else {
let modified_set: std::collections::HashSet<&str> =
modified_props.iter().map(|s| s.as_str()).collect();
scalar_props
.iter()
.copied()
.filter(|p| !p.is_primary_key && modified_set.contains(p.field_name.as_ref()))
.collect()
};
let set_cols: Vec<&str> = set_props.iter().map(|p| p.column_name.as_ref()).collect();
if set_cols.is_empty() || keys.is_empty() {
continue;
}
let concurrency_tokens: Vec<&PropertyMeta> = scalar_props
.iter()
.copied()
.filter(|p| p.is_concurrency_token)
.collect();
let (mut where_clause, mut where_params) = build_where_with_concurrency(
gen,
&keys,
&concurrency_tokens,
*original,
set_cols.len() + 1,
)?;
if let Some(filter) = query_filter {
let mut idx = set_cols.len() + where_params.len() + 1;
let filter_sql = compile_bool_expr(filter, gen, &mut idx);
where_params.extend(collect_bool_expr_values(filter));
where_clause = format!("({}) AND ({})", where_clause, filter_sql);
}
let sql = sql_cache
.entry((
meta.table_name.to_string(),
set_cols.iter().map(|s| (*s).to_string()).collect(),
where_clause.clone(),
))
.or_insert_with(|| gen.update(meta.table_name.as_ref(), &set_cols, &where_clause))
.clone();
let mut params: Vec<DbValue> = set_props
.iter()
.map(|p| {
snap.get(p.field_name.as_ref())
.cloned()
.unwrap_or(DbValue::Null)
})
.collect();
params.extend(where_params);
let rows = conn.execute(&sql, ¶ms).await?;
if rows == 0 {
return Err(EFError::concurrency_conflict(format!(
"update affected 0 rows on {} (row may have been modified or deleted)",
meta.table_name
)));
}
updated += 1;
}
Ok(updated)
}
#[allow(clippy::type_complexity)]
pub async fn execute_deletes<E>(
conn: &mut dyn IAsyncConnection,
provider: &dyn IDatabaseProvider,
entities: &[(&E, &EntityTypeMeta, Option<&HashMap<String, DbValue>>)],
query_filter: Option<&BoolExpr>,
) -> EFResult<usize>
where
E: IEntityType + IGetKeyValues,
{
if entities.is_empty() {
return Ok(0);
}
let gen = provider.sql_generator();
let meta = entities[0].1;
let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
let has_concurrency_tokens = scalar_props.iter().any(|p| p.is_concurrency_token);
let pk_props: Vec<_> = scalar_props.iter().filter(|p| p.is_primary_key).collect();
if has_concurrency_tokens || pk_props.len() != 1 {
return Self::execute_deletes_per_row(conn, gen, entities, query_filter).await;
}
let pk_col = pk_props[0].column_name.as_ref();
let pk_field = pk_props[0].field_name.as_ref();
let pk_values: Vec<DbValue> = entities
.iter()
.filter_map(|(e, _, _)| e.key_values().get(pk_field).cloned())
.collect();
if pk_values.is_empty() {
return Ok(0);
}
let filter_params: Vec<DbValue> = match query_filter {
Some(filter) => collect_bool_expr_values(filter),
None => Vec::new(),
};
const MAX_PARAMS: usize = 900;
let batch_size = MAX_PARAMS.saturating_sub(filter_params.len()).max(1);
let mut deleted = 0usize;
let mut start = 0usize;
while start < pk_values.len() {
let end = (start + batch_size).min(pk_values.len());
let batch = &pk_values[start..end];
let pk_count = batch.len();
let pk_placeholders: Vec<String> = (1..=pk_count)
.map(|i| gen.parameter_placeholder(i))
.collect();
let mut where_clause = format!(
"{} IN ({})",
gen.quote_identifier(pk_col),
pk_placeholders.join(", "),
);
let mut params: Vec<DbValue> = batch.to_vec();
if let Some(filter) = query_filter {
let mut idx = pk_count + 1;
let filter_sql = compile_bool_expr(filter, gen, &mut idx);
params.extend(filter_params.iter().cloned());
where_clause = format!("({}) AND ({})", where_clause, filter_sql);
}
let sql = gen.delete(meta.table_name.as_ref(), &where_clause);
let rows = conn.execute(&sql, ¶ms).await?;
if rows == 0 && pk_count > 0 {
return Err(EFError::concurrency_conflict(format!(
"batch delete affected 0 rows on {} (rows may have been modified or deleted)",
meta.table_name
)));
}
deleted += (rows as usize).min(pk_count);
start = end;
}
Ok(deleted)
}
#[allow(clippy::type_complexity)]
async fn execute_deletes_per_row<E>(
conn: &mut dyn IAsyncConnection,
gen: &'static dyn crate::provider::ISqlGenerator,
entities: &[(&E, &EntityTypeMeta, Option<&HashMap<String, DbValue>>)],
query_filter: Option<&BoolExpr>,
) -> EFResult<usize>
where
E: IEntityType + IGetKeyValues,
{
let mut deleted = 0;
for (entity, meta, original) in entities {
let keys = entity.key_values();
if keys.is_empty() {
continue;
}
let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
let concurrency_tokens: Vec<&PropertyMeta> = scalar_props
.iter()
.copied()
.filter(|p| p.is_concurrency_token)
.collect();
let (mut where_clause, mut where_params) =
build_where_with_concurrency(gen, &keys, &concurrency_tokens, *original, 1)?;
if let Some(filter) = query_filter {
let mut idx = where_params.len() + 1;
let filter_sql = compile_bool_expr(filter, gen, &mut idx);
where_params.extend(collect_bool_expr_values(filter));
where_clause = format!("({}) AND ({})", where_clause, filter_sql);
}
let sql = gen.delete(meta.table_name.as_ref(), &where_clause);
let rows = conn.execute(&sql, &where_params).await?;
if rows == 0 {
return Err(EFError::concurrency_conflict(format!(
"delete affected 0 rows on {} (row may have been modified or deleted)",
meta.table_name
)));
}
deleted += 1;
}
Ok(deleted)
}
}
fn build_where_with_concurrency(
gen: &dyn crate::provider::ISqlGenerator,
keys: &HashMap<String, DbValue>,
concurrency_tokens: &[&PropertyMeta],
original: Option<&HashMap<String, DbValue>>,
start_param_idx: usize,
) -> EFResult<(String, Vec<DbValue>)> {
let mut where_parts: Vec<String> = keys
.keys()
.enumerate()
.map(|(i, k)| {
format!(
"{} = {}",
gen.quote_identifier(k),
gen.parameter_placeholder(start_param_idx + i)
)
})
.collect();
let mut params: Vec<DbValue> = keys.values().cloned().collect();
for (next_idx, token) in (start_param_idx + keys.len()..).zip(concurrency_tokens.iter()) {
where_parts.push(format!(
"{} = {}",
gen.quote_identifier(token.column_name.as_ref()),
gen.parameter_placeholder(next_idx)
));
let original_val = original
.and_then(|o| o.get(token.field_name.as_ref()))
.ok_or_else(|| {
EFError::change_tracking(format!(
"missing original concurrency token for '{}'",
token.field_name
))
})?;
params.push(original_val.clone());
}
Ok((where_parts.join(" AND "), params))
}
pub fn generate_insert_sql(
provider: &dyn IDatabaseProvider,
meta: &EntityTypeMeta,
_property_values: &HashMap<String, DbValue>,
) -> String {
let gen = provider.sql_generator();
let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
let columns: Vec<&str> = scalar_props
.iter()
.map(|p| p.column_name.as_ref())
.collect();
if columns.is_empty() {
return String::new();
}
gen.insert(meta.table_name.as_ref(), &columns, true)
}
pub fn generate_update_sql(
provider: &dyn IDatabaseProvider,
meta: &EntityTypeMeta,
property_values: &HashMap<String, DbValue>,
primary_key_values: &HashMap<String, DbValue>,
) -> String {
let gen = provider.sql_generator();
let set_columns: Vec<&str> = property_values
.keys()
.filter(|k| !primary_key_values.contains_key(*k))
.map(|k| k.as_str())
.collect();
if set_columns.is_empty() || primary_key_values.is_empty() {
return String::new();
}
let where_parts: Vec<String> = primary_key_values
.keys()
.enumerate()
.map(|(i, k)| {
format!(
"{} = {}",
gen.quote_identifier(k),
gen.parameter_placeholder(i + 1)
)
})
.collect();
gen.update(
meta.table_name.as_ref(),
&set_columns,
&where_parts.join(" AND "),
)
}
pub fn generate_delete_sql(
provider: &dyn IDatabaseProvider,
meta: &EntityTypeMeta,
primary_key_values: &HashMap<String, DbValue>,
) -> String {
let gen = provider.sql_generator();
if primary_key_values.is_empty() {
return String::new();
}
let where_parts: Vec<String> = primary_key_values
.keys()
.enumerate()
.map(|(i, k)| {
format!(
"{} = {}",
gen.quote_identifier(k),
gen.parameter_placeholder(i + 1)
)
})
.collect();
gen.delete(meta.table_name.as_ref(), &where_parts.join(" AND "))
}
pub fn collect_insert_params(
meta: &EntityTypeMeta,
property_values: &HashMap<String, DbValue>,
) -> Vec<DbValue> {
meta.mapped_scalar_properties()
.map(|p| {
property_values
.get(p.field_name.as_ref())
.cloned()
.unwrap_or(DbValue::Null)
})
.collect()
}
pub fn collect_update_params(
property_values: &HashMap<String, DbValue>,
primary_key_values: &HashMap<String, DbValue>,
set_keys: &[String],
) -> Vec<DbValue> {
let mut params: Vec<DbValue> = set_keys
.iter()
.filter(|k| !primary_key_values.contains_key(*k))
.map(|k| property_values.get(k).cloned().unwrap_or(DbValue::Null))
.collect();
for v in primary_key_values.values() {
params.push(v.clone());
}
params
}
pub fn collect_delete_params(primary_key_values: &HashMap<String, DbValue>) -> Vec<DbValue> {
primary_key_values.values().cloned().collect()
}