rust_ef/change_executor/

executor_dml.rs

//! `ChangeExecutor` — UPDATE and DELETE execution (batched + per-row fallback).

use crate::entity::{IEntitySnapshot, IEntityType, IGetKeyValues};
use crate::entity_snapshot::EntitySnapshot;
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 super::executor::ChangeExecutor;

impl ChangeExecutor {
    /// Executes UPDATE statements for all modified entities.
    ///
    /// When no concurrency tokens are present and the entity has a single-column
    /// primary key, rows are batched into a single `UPDATE ... SET col = CASE
    /// pk WHEN ? THEN ? ... END WHERE pk IN (...)` statement (≤900 params per
    /// batch) to minimize round trips. Otherwise (concurrency tokens, composite
    /// PK), falls back to per-row UPDATE so optimistic-concurrency checks run
    /// on each row.
    ///
    /// When `modified_properties` is populated (via `detect_changes`), only the
    /// dirty columns are SET. When empty (entity marked Modified via `update()`
    /// without detection), all non-PK columns are SET (backward compatible).
    ///
    /// When `query_filter` is `Some`, the filter is AND-ed into the WHERE
    /// clause so updates cannot cross the filter boundary (multi-tenant /
    /// soft-delete isolation).
    #[allow(clippy::type_complexity)]
    pub async fn execute_updates<E>(
        conn: &mut dyn IAsyncConnection,
        provider: &dyn IDatabaseProvider,
        entities: &[(&E, &EntityTypeMeta, Option<&EntitySnapshot>, &[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);

        // Compute the union of modified field names across all entities. When
        // non-empty, only those columns are SET (partial update). When empty
        // (no change detection ran), all non-PK columns are SET.
        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();

        // Fall back to per-row UPDATE when optimistic concurrency tokens are
        // present (each row needs its own WHERE to check the token) or when
        // the PK is composite (CASE WHEN only works with a single column).
        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();

        // Pre-compute snapshots and keys to avoid re-hashing per batch.
        let entity_data: Vec<(EntitySnapshot, EntitySnapshot)> = entities
            .iter()
            .map(|(e, _, _, _)| (e.snapshot(), e.key_values()))
            .collect();

        // Filter params are constant across batches.
        let filter_params: Vec<DbValue> = match query_filter {
            Some(filter) => collect_bool_expr_values(filter),
            None => Vec::new(),
        };

        // For non-numbered placeholder dialects (SQLite/MySQL `?`), the filter
        // SQL is index-independent — compile once and reuse across batches.
        // PostgreSQL (`$N`) must recompile per batch because numbering shifts.
        let cached_filter_sql: Option<String> = if !gen.uses_numbered_placeholders() {
            query_filter.map(|f| {
                let mut idx = 1;
                compile_bool_expr(f, gen, &mut idx)
            })
        } else {
            None
        };

        // Each row consumes 2 * set_cols params (CASE WHEN pk/value pairs)
        // plus 1 param in the WHERE IN-list.
        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;

            // SET clause consumes 2 * set_cols * row_count placeholders.
            let set_param_count = 2 * set_cols.len() * row_count;
            let mut idx = set_param_count + 1;

            // Build WHERE clause: pk_col IN (?, ...) [AND (filter)]
            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(", ")
            );

            // CASE WHEN params: for each col, for each entity: [pk_value, col_value]
            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);
                }
            }
            // WHERE IN params: pk values
            for (_, keys) in entity_data[start..end].iter() {
                let pk_val = keys.get(pk_field).cloned().unwrap_or(DbValue::Null);
                params.push(pk_val);
            }

            // Append filter to WHERE clause.
            if let Some(filter) = query_filter {
                let filter_sql = match &cached_filter_sql {
                    Some(cached) => cached.clone(),
                    None => 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, &params).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)
    }

    /// Per-row UPDATE fallback used when concurrency tokens or composite PKs
    /// prevent batching. Each row gets its own `UPDATE ... SET ... WHERE
    /// pk = ? AND ...`. When `modified_properties` is non-empty for an entity,
    /// only those columns are SET (partial update); otherwise all non-PK
    /// columns are SET (backward compatible).
    #[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<&EntitySnapshot>, &[String])],
        query_filter: Option<&BoolExpr>,
    ) -> EFResult<usize>
    where
        E: IEntityType + IEntitySnapshot + IGetKeyValues,
    {
        let mut updated = 0;

        // Hoist metadata-derived collections outside the per-entity loop —
        // all entities share the same EntityTypeMeta (same type E), so
        // scalar_props and concurrency_tokens are identical for every row.
        let meta0 = entities[0].1;
        let scalar_props: Vec<&PropertyMeta> = meta0.mapped_scalar_properties().collect();
        let concurrency_tokens: Vec<&PropertyMeta> = scalar_props
            .iter()
            .copied()
            .filter(|p| p.is_concurrency_token)
            .collect();
        let table_name = meta0.table_name.as_ref();

        for (entity, _meta, original, modified_props) in entities {
            let snap = entity.snapshot();
            let keys = entity.key_values();

            // When modified_properties is populated, SET only those columns
            // (partial update). When empty, SET all non-PK columns.
            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 (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 = gen.update(table_name, &set_cols, &where_clause);

            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, &params).await?;
            if rows == 0 {
                return Err(EFError::concurrency_conflict(format!(
                    "update affected 0 rows on {} (row may have been modified or deleted)",
                    table_name
                )));
            }
            updated += 1;
        }

        Ok(updated)
    }

    /// Executes DELETE statements for all deleted entities.
    ///
    /// When no concurrency tokens are present and the entity has a single-column
    /// primary key, rows are batched into `DELETE ... WHERE pk IN (?, ?, ...)`
    /// statements (≤900 params per batch) to minimize round trips. Otherwise
    /// (concurrency tokens, composite PK), falls back to per-row DELETE so
    /// optimistic-concurrency checks run on each row.
    ///
    /// When `query_filter` is `Some`, the filter is AND-ed into the WHERE
    /// clause so deletes cannot cross the filter boundary.
    #[allow(clippy::type_complexity)]
    pub async fn execute_deletes<E>(
        conn: &mut dyn IAsyncConnection,
        provider: &dyn IDatabaseProvider,
        entities: &[(&E, &EntityTypeMeta, Option<&EntitySnapshot>)],
        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();

        // Fall back to per-row DELETE when optimistic concurrency tokens are
        // present (each row needs its own WHERE to check the token) or when
        // the PK is composite (IN clause only works for a single column).
        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();

        // Collect PK values; entities missing the PK value are skipped.
        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);
        }

        // Filter params are constant across batches; their SQL is recomputed
        // per batch for numbered-placeholder dialects (Postgres `$N`).
        // For `?` dialects (SQLite/MySQL), the filter SQL is index-independent
        // and can be compiled once and reused.
        let filter_params: Vec<DbValue> = match query_filter {
            Some(filter) => collect_bool_expr_values(filter),
            None => Vec::new(),
        };
        let cached_filter_sql: Option<String> = if !gen.uses_numbered_placeholders() {
            query_filter.map(|f| {
                let mut idx = 1;
                compile_bool_expr(f, gen, &mut idx)
            })
        } else {
            None
        };
        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 filter_sql = match &cached_filter_sql {
                    Some(cached) => cached.clone(),
                    None => {
                        let mut idx = pk_count + 1;
                        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, &params).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)
    }

    /// Per-row DELETE fallback used when concurrency tokens or composite PKs
    /// prevent batching. Each row gets its own `DELETE ... WHERE pk = ? AND ...`.
    #[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<&EntitySnapshot>)],
        query_filter: Option<&BoolExpr>,
    ) -> EFResult<usize>
    where
        E: IEntityType + IGetKeyValues,
    {
        let mut deleted = 0;

        // Hoist metadata-derived collections outside the per-entity loop —
        // all entities share the same EntityTypeMeta (same type E).
        let meta0 = entities[0].1;
        let concurrency_tokens: Vec<&PropertyMeta> = meta0
            .mapped_scalar_properties()
            .filter(|p| p.is_concurrency_token)
            .collect();
        let table_name = meta0.table_name.as_ref();

        for (entity, _meta, original) in entities {
            let keys = entity.key_values();
            if keys.is_empty() {
                continue;
            }

            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(table_name, &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)",
                    table_name
                )));
            }
            deleted += 1;
        }
        Ok(deleted)
    }
}

fn build_where_with_concurrency(
    gen: &dyn crate::provider::ISqlGenerator,
    keys: &EntitySnapshot,
    concurrency_tokens: &[&PropertyMeta],
    original: Option<&EntitySnapshot>,
    start_param_idx: usize,
) -> EFResult<(String, Vec<DbValue>)> {
    let mut where_parts: Vec<String> = keys
        .iter()
        .enumerate()
        .map(|(i, (k, _))| {
            format!(
                "{} = {}",
                gen.quote_identifier(k),
                gen.parameter_placeholder(start_param_idx + i)
            )
        })
        .collect();

    let mut params: Vec<DbValue> = keys.iter().map(|(_, v)| v.clone()).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))
}