rust_ef/change_executor/
executor_dml.rs

1//! `ChangeExecutor` — UPDATE and DELETE execution (batched + per-row fallback).
2
3use crate::entity::{IEntitySnapshot, IEntityType, IGetKeyValues};
4use crate::error::{EFError, EFResult};
5use crate::metadata::{EntityTypeMeta, PropertyMeta};
6use crate::provider::{DbValue, IAsyncConnection, IDatabaseProvider};
7use crate::query::{collect_bool_expr_values, compile_bool_expr, BoolExpr};
8use std::collections::HashMap;
9
10use super::executor::ChangeExecutor;
11
12impl ChangeExecutor {
13    /// Executes UPDATE statements for all modified entities.
14    ///
15    /// When no concurrency tokens are present and the entity has a single-column
16    /// primary key, rows are batched into a single `UPDATE ... SET col = CASE
17    /// pk WHEN ? THEN ? ... END WHERE pk IN (...)` statement (≤900 params per
18    /// batch) to minimize round trips. Otherwise (concurrency tokens, composite
19    /// PK), falls back to per-row UPDATE so optimistic-concurrency checks run
20    /// on each row.
21    ///
22    /// When `modified_properties` is populated (via `detect_changes`), only the
23    /// dirty columns are SET. When empty (entity marked Modified via `update()`
24    /// without detection), all non-PK columns are SET (backward compatible).
25    ///
26    /// When `query_filter` is `Some`, the filter is AND-ed into the WHERE
27    /// clause so updates cannot cross the filter boundary (multi-tenant /
28    /// soft-delete isolation).
29    #[allow(clippy::type_complexity)]
30    pub async fn execute_updates<E>(
31        conn: &mut dyn IAsyncConnection,
32        provider: &dyn IDatabaseProvider,
33        entities: &[(
34            &E,
35            &EntityTypeMeta,
36            Option<&HashMap<String, DbValue>>,
37            &[String],
38        )],
39        query_filter: Option<&BoolExpr>,
40    ) -> EFResult<usize>
41    where
42        E: IEntityType + IEntitySnapshot + IGetKeyValues,
43    {
44        if entities.is_empty() {
45            return Ok(0);
46        }
47        let gen = provider.sql_generator();
48        let meta = entities[0].1;
49        let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
50        let pk_props: Vec<_> = scalar_props.iter().filter(|p| p.is_primary_key).collect();
51        let has_concurrency_tokens = scalar_props.iter().any(|p| p.is_concurrency_token);
52
53        // Compute the union of modified field names across all entities. When
54        // non-empty, only those columns are SET (partial update). When empty
55        // (no change detection ran), all non-PK columns are SET.
56        let modified_fields: std::collections::HashSet<&str> = entities
57            .iter()
58            .flat_map(|(_, _, _, mods)| mods.iter().map(|s| s.as_str()))
59            .collect();
60        let set_props: Vec<&PropertyMeta> = if modified_fields.is_empty() {
61            scalar_props
62                .iter()
63                .copied()
64                .filter(|p| !p.is_primary_key)
65                .collect()
66        } else {
67            scalar_props
68                .iter()
69                .copied()
70                .filter(|p| !p.is_primary_key && modified_fields.contains(p.field_name.as_ref()))
71                .collect()
72        };
73        let set_cols: Vec<&str> = set_props.iter().map(|p| p.column_name.as_ref()).collect();
74
75        // Fall back to per-row UPDATE when optimistic concurrency tokens are
76        // present (each row needs its own WHERE to check the token) or when
77        // the PK is composite (CASE WHEN only works with a single column).
78        if has_concurrency_tokens || pk_props.len() != 1 || set_cols.is_empty() {
79            return Self::execute_updates_per_row(conn, gen, entities, query_filter).await;
80        }
81
82        let pk_col = pk_props[0].column_name.as_ref();
83        let pk_field = pk_props[0].field_name.as_ref();
84
85        // Pre-compute snapshots and keys to avoid re-hashing per batch.
86        let entity_data: Vec<(HashMap<String, DbValue>, HashMap<String, DbValue>)> = entities
87            .iter()
88            .map(|(e, _, _, _)| (e.snapshot(), e.key_values()))
89            .collect();
90
91        // Filter params are constant across batches.
92        let filter_params: Vec<DbValue> = match query_filter {
93            Some(filter) => collect_bool_expr_values(filter),
94            None => Vec::new(),
95        };
96
97        // For non-numbered placeholder dialects (SQLite/MySQL `?`), the filter
98        // SQL is index-independent — compile once and reuse across batches.
99        // PostgreSQL (`$N`) must recompile per batch because numbering shifts.
100        let cached_filter_sql: Option<String> = if !gen.uses_numbered_placeholders() {
101            query_filter.map(|f| {
102                let mut idx = 1;
103                compile_bool_expr(f, gen, &mut idx)
104            })
105        } else {
106            None
107        };
108
109        // Each row consumes 2 * set_cols params (CASE WHEN pk/value pairs)
110        // plus 1 param in the WHERE IN-list.
111        const MAX_PARAMS: usize = 900;
112        let params_per_row = 2 * set_cols.len() + 1;
113        let batch_size =
114            ((MAX_PARAMS.saturating_sub(filter_params.len())) / params_per_row.max(1)).max(1);
115
116        let mut updated = 0usize;
117        let mut start = 0usize;
118        while start < entity_data.len() {
119            let end = (start + batch_size).min(entity_data.len());
120            let row_count = end - start;
121
122            // SET clause consumes 2 * set_cols * row_count placeholders.
123            let set_param_count = 2 * set_cols.len() * row_count;
124            let mut idx = set_param_count + 1;
125
126            // Build WHERE clause: pk_col IN (?, ...) [AND (filter)]
127            let pk_placeholders: Vec<String> = (0..row_count)
128                .map(|_| {
129                    let ph = gen.parameter_placeholder(idx);
130                    idx += 1;
131                    ph
132                })
133                .collect();
134            let mut where_clause = format!(
135                "{} IN ({})",
136                gen.quote_identifier(pk_col),
137                pk_placeholders.join(", ")
138            );
139
140            // CASE WHEN params: for each col, for each entity: [pk_value, col_value]
141            let mut params: Vec<DbValue> = Vec::with_capacity(set_param_count + row_count);
142            for col_prop in &set_props {
143                for (snap, keys) in entity_data[start..end].iter() {
144                    let pk_val = keys.get(pk_field).cloned().unwrap_or(DbValue::Null);
145                    let col_val = snap
146                        .get(col_prop.field_name.as_ref())
147                        .cloned()
148                        .unwrap_or(DbValue::Null);
149                    params.push(pk_val);
150                    params.push(col_val);
151                }
152            }
153            // WHERE IN params: pk values
154            for (_, keys) in entity_data[start..end].iter() {
155                let pk_val = keys.get(pk_field).cloned().unwrap_or(DbValue::Null);
156                params.push(pk_val);
157            }
158
159            // Append filter to WHERE clause.
160            if let Some(filter) = query_filter {
161                let filter_sql = match &cached_filter_sql {
162                    Some(cached) => cached.clone(),
163                    None => compile_bool_expr(filter, gen, &mut idx),
164                };
165                params.extend(filter_params.iter().cloned());
166                where_clause = format!("({}) AND ({})", where_clause, filter_sql);
167            }
168
169            let sql = gen.update_batch(
170                meta.table_name.as_ref(),
171                &set_cols,
172                pk_col,
173                row_count,
174                &where_clause,
175            );
176            let rows = conn.execute(&sql, &params).await?;
177            if rows == 0 && row_count > 0 {
178                return Err(EFError::concurrency_conflict(format!(
179                    "batch update affected 0 rows on {} (rows may have been modified or deleted)",
180                    meta.table_name
181                )));
182            }
183            updated += (rows as usize).min(row_count);
184            start = end;
185        }
186
187        Ok(updated)
188    }
189
190    /// Per-row UPDATE fallback used when concurrency tokens or composite PKs
191    /// prevent batching. Each row gets its own `UPDATE ... SET ... WHERE
192    /// pk = ? AND ...`. When `modified_properties` is non-empty for an entity,
193    /// only those columns are SET (partial update); otherwise all non-PK
194    /// columns are SET (backward compatible).
195    #[allow(clippy::type_complexity)]
196    async fn execute_updates_per_row<E>(
197        conn: &mut dyn IAsyncConnection,
198        gen: &'static dyn crate::provider::ISqlGenerator,
199        entities: &[(
200            &E,
201            &EntityTypeMeta,
202            Option<&HashMap<String, DbValue>>,
203            &[String],
204        )],
205        query_filter: Option<&BoolExpr>,
206    ) -> EFResult<usize>
207    where
208        E: IEntityType + IEntitySnapshot + IGetKeyValues,
209    {
210        let mut updated = 0;
211        let mut sql_cache: HashMap<(String, Vec<String>, String), String> = HashMap::new();
212
213        // Hoist metadata-derived collections outside the per-entity loop —
214        // all entities share the same EntityTypeMeta (same type E), so
215        // scalar_props and concurrency_tokens are identical for every row.
216        let meta0 = entities[0].1;
217        let scalar_props: Vec<&PropertyMeta> = meta0.mapped_scalar_properties().collect();
218        let concurrency_tokens: Vec<&PropertyMeta> = scalar_props
219            .iter()
220            .copied()
221            .filter(|p| p.is_concurrency_token)
222            .collect();
223        let table_name = meta0.table_name.as_ref();
224
225        for (entity, _meta, original, modified_props) in entities {
226            let snap = entity.snapshot();
227            let keys = entity.key_values();
228
229            // When modified_properties is populated, SET only those columns
230            // (partial update). When empty, SET all non-PK columns.
231            let set_props: Vec<&PropertyMeta> = if modified_props.is_empty() {
232                scalar_props
233                    .iter()
234                    .copied()
235                    .filter(|p| !p.is_primary_key)
236                    .collect()
237            } else {
238                let modified_set: std::collections::HashSet<&str> =
239                    modified_props.iter().map(|s| s.as_str()).collect();
240                scalar_props
241                    .iter()
242                    .copied()
243                    .filter(|p| !p.is_primary_key && modified_set.contains(p.field_name.as_ref()))
244                    .collect()
245            };
246            let set_cols: Vec<&str> = set_props.iter().map(|p| p.column_name.as_ref()).collect();
247
248            if set_cols.is_empty() || keys.is_empty() {
249                continue;
250            }
251
252            let (mut where_clause, mut where_params) = build_where_with_concurrency(
253                gen,
254                &keys,
255                &concurrency_tokens,
256                *original,
257                set_cols.len() + 1,
258            )?;
259
260            if let Some(filter) = query_filter {
261                let mut idx = set_cols.len() + where_params.len() + 1;
262                let filter_sql = compile_bool_expr(filter, gen, &mut idx);
263                where_params.extend(collect_bool_expr_values(filter));
264                where_clause = format!("({}) AND ({})", where_clause, filter_sql);
265            }
266
267            let sql = sql_cache
268                .entry((
269                    table_name.to_string(),
270                    set_cols.iter().map(|s| (*s).to_string()).collect(),
271                    where_clause.clone(),
272                ))
273                .or_insert_with(|| gen.update(table_name, &set_cols, &where_clause))
274                .clone();
275
276            let mut params: Vec<DbValue> = set_props
277                .iter()
278                .map(|p| {
279                    snap.get(p.field_name.as_ref())
280                        .cloned()
281                        .unwrap_or(DbValue::Null)
282                })
283                .collect();
284            params.extend(where_params);
285
286            let rows = conn.execute(&sql, &params).await?;
287            if rows == 0 {
288                return Err(EFError::concurrency_conflict(format!(
289                    "update affected 0 rows on {} (row may have been modified or deleted)",
290                    table_name
291                )));
292            }
293            updated += 1;
294        }
295
296        Ok(updated)
297    }
298
299    /// Executes DELETE statements for all deleted entities.
300    ///
301    /// When no concurrency tokens are present and the entity has a single-column
302    /// primary key, rows are batched into `DELETE ... WHERE pk IN (?, ?, ...)`
303    /// statements (≤900 params per batch) to minimize round trips. Otherwise
304    /// (concurrency tokens, composite PK), falls back to per-row DELETE so
305    /// optimistic-concurrency checks run on each row.
306    ///
307    /// When `query_filter` is `Some`, the filter is AND-ed into the WHERE
308    /// clause so deletes cannot cross the filter boundary.
309    #[allow(clippy::type_complexity)]
310    pub async fn execute_deletes<E>(
311        conn: &mut dyn IAsyncConnection,
312        provider: &dyn IDatabaseProvider,
313        entities: &[(&E, &EntityTypeMeta, Option<&HashMap<String, DbValue>>)],
314        query_filter: Option<&BoolExpr>,
315    ) -> EFResult<usize>
316    where
317        E: IEntityType + IGetKeyValues,
318    {
319        if entities.is_empty() {
320            return Ok(0);
321        }
322        let gen = provider.sql_generator();
323        let meta = entities[0].1;
324        let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
325        let has_concurrency_tokens = scalar_props.iter().any(|p| p.is_concurrency_token);
326        let pk_props: Vec<_> = scalar_props.iter().filter(|p| p.is_primary_key).collect();
327
328        // Fall back to per-row DELETE when optimistic concurrency tokens are
329        // present (each row needs its own WHERE to check the token) or when
330        // the PK is composite (IN clause only works for a single column).
331        if has_concurrency_tokens || pk_props.len() != 1 {
332            return Self::execute_deletes_per_row(conn, gen, entities, query_filter).await;
333        }
334
335        let pk_col = pk_props[0].column_name.as_ref();
336        let pk_field = pk_props[0].field_name.as_ref();
337
338        // Collect PK values; entities missing the PK value are skipped.
339        let pk_values: Vec<DbValue> = entities
340            .iter()
341            .filter_map(|(e, _, _)| e.key_values().get(pk_field).cloned())
342            .collect();
343        if pk_values.is_empty() {
344            return Ok(0);
345        }
346
347        // Filter params are constant across batches; their SQL is recomputed
348        // per batch for numbered-placeholder dialects (Postgres `$N`).
349        // For `?` dialects (SQLite/MySQL), the filter SQL is index-independent
350        // and can be compiled once and reused.
351        let filter_params: Vec<DbValue> = match query_filter {
352            Some(filter) => collect_bool_expr_values(filter),
353            None => Vec::new(),
354        };
355        let cached_filter_sql: Option<String> = if !gen.uses_numbered_placeholders() {
356            query_filter.map(|f| {
357                let mut idx = 1;
358                compile_bool_expr(f, gen, &mut idx)
359            })
360        } else {
361            None
362        };
363        const MAX_PARAMS: usize = 900;
364        let batch_size = MAX_PARAMS.saturating_sub(filter_params.len()).max(1);
365
366        let mut deleted = 0usize;
367        let mut start = 0usize;
368        while start < pk_values.len() {
369            let end = (start + batch_size).min(pk_values.len());
370            let batch = &pk_values[start..end];
371            let pk_count = batch.len();
372
373            let pk_placeholders: Vec<String> = (1..=pk_count)
374                .map(|i| gen.parameter_placeholder(i))
375                .collect();
376            let mut where_clause = format!(
377                "{} IN ({})",
378                gen.quote_identifier(pk_col),
379                pk_placeholders.join(", "),
380            );
381            let mut params: Vec<DbValue> = batch.to_vec();
382            if let Some(filter) = query_filter {
383                let filter_sql = match &cached_filter_sql {
384                    Some(cached) => cached.clone(),
385                    None => {
386                        let mut idx = pk_count + 1;
387                        compile_bool_expr(filter, gen, &mut idx)
388                    }
389                };
390                params.extend(filter_params.iter().cloned());
391                where_clause = format!("({}) AND ({})", where_clause, filter_sql);
392            }
393
394            let sql = gen.delete(meta.table_name.as_ref(), &where_clause);
395            let rows = conn.execute(&sql, &params).await?;
396            if rows == 0 && pk_count > 0 {
397                return Err(EFError::concurrency_conflict(format!(
398                    "batch delete affected 0 rows on {} (rows may have been modified or deleted)",
399                    meta.table_name
400                )));
401            }
402            deleted += (rows as usize).min(pk_count);
403            start = end;
404        }
405
406        Ok(deleted)
407    }
408
409    /// Per-row DELETE fallback used when concurrency tokens or composite PKs
410    /// prevent batching. Each row gets its own `DELETE ... WHERE pk = ? AND ...`.
411    #[allow(clippy::type_complexity)]
412    async fn execute_deletes_per_row<E>(
413        conn: &mut dyn IAsyncConnection,
414        gen: &'static dyn crate::provider::ISqlGenerator,
415        entities: &[(&E, &EntityTypeMeta, Option<&HashMap<String, DbValue>>)],
416        query_filter: Option<&BoolExpr>,
417    ) -> EFResult<usize>
418    where
419        E: IEntityType + IGetKeyValues,
420    {
421        let mut deleted = 0;
422
423        // Hoist metadata-derived collections outside the per-entity loop —
424        // all entities share the same EntityTypeMeta (same type E).
425        let meta0 = entities[0].1;
426        let concurrency_tokens: Vec<&PropertyMeta> = meta0
427            .mapped_scalar_properties()
428            .filter(|p| p.is_concurrency_token)
429            .collect();
430        let table_name = meta0.table_name.as_ref();
431
432        for (entity, _meta, original) in entities {
433            let keys = entity.key_values();
434            if keys.is_empty() {
435                continue;
436            }
437
438            let (mut where_clause, mut where_params) =
439                build_where_with_concurrency(gen, &keys, &concurrency_tokens, *original, 1)?;
440
441            if let Some(filter) = query_filter {
442                let mut idx = where_params.len() + 1;
443                let filter_sql = compile_bool_expr(filter, gen, &mut idx);
444                where_params.extend(collect_bool_expr_values(filter));
445                where_clause = format!("({}) AND ({})", where_clause, filter_sql);
446            }
447
448            let sql = gen.delete(table_name, &where_clause);
449            let rows = conn.execute(&sql, &where_params).await?;
450            if rows == 0 {
451                return Err(EFError::concurrency_conflict(format!(
452                    "delete affected 0 rows on {} (row may have been modified or deleted)",
453                    table_name
454                )));
455            }
456            deleted += 1;
457        }
458        Ok(deleted)
459    }
460}
461
462fn build_where_with_concurrency(
463    gen: &dyn crate::provider::ISqlGenerator,
464    keys: &HashMap<String, DbValue>,
465    concurrency_tokens: &[&PropertyMeta],
466    original: Option<&HashMap<String, DbValue>>,
467    start_param_idx: usize,
468) -> EFResult<(String, Vec<DbValue>)> {
469    let mut where_parts: Vec<String> = keys
470        .keys()
471        .enumerate()
472        .map(|(i, k)| {
473            format!(
474                "{} = {}",
475                gen.quote_identifier(k),
476                gen.parameter_placeholder(start_param_idx + i)
477            )
478        })
479        .collect();
480
481    let mut params: Vec<DbValue> = keys.values().cloned().collect();
482
483    for (next_idx, token) in (start_param_idx + keys.len()..).zip(concurrency_tokens.iter()) {
484        where_parts.push(format!(
485            "{} = {}",
486            gen.quote_identifier(token.column_name.as_ref()),
487            gen.parameter_placeholder(next_idx)
488        ));
489
490        let original_val = original
491            .and_then(|o| o.get(token.field_name.as_ref()))
492            .ok_or_else(|| {
493                EFError::change_tracking(format!(
494                    "missing original concurrency token for '{}'",
495                    token.field_name
496                ))
497            })?;
498        params.push(original_val.clone());
499    }
500
501    Ok((where_parts.join(" AND "), params))
502}
rust_ef/change_executor/executor_dml.rs

rust_ef/change_executor/
executor_dml.rs
