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        // Each row consumes 2 * set_cols params (CASE WHEN pk/value pairs)
98        // plus 1 param in the WHERE IN-list.
99        const MAX_PARAMS: usize = 900;
100        let params_per_row = 2 * set_cols.len() + 1;
101        let batch_size =
102            ((MAX_PARAMS.saturating_sub(filter_params.len())) / params_per_row.max(1)).max(1);
103
104        let mut updated = 0usize;
105        let mut start = 0usize;
106        while start < entity_data.len() {
107            let end = (start + batch_size).min(entity_data.len());
108            let row_count = end - start;
109
110            // SET clause consumes 2 * set_cols * row_count placeholders.
111            let set_param_count = 2 * set_cols.len() * row_count;
112            let mut idx = set_param_count + 1;
113
114            // Build WHERE clause: pk_col IN (?, ...) [AND (filter)]
115            let pk_placeholders: Vec<String> = (0..row_count)
116                .map(|_| {
117                    let ph = gen.parameter_placeholder(idx);
118                    idx += 1;
119                    ph
120                })
121                .collect();
122            let mut where_clause = format!(
123                "{} IN ({})",
124                gen.quote_identifier(pk_col),
125                pk_placeholders.join(", ")
126            );
127
128            // CASE WHEN params: for each col, for each entity: [pk_value, col_value]
129            let mut params: Vec<DbValue> = Vec::with_capacity(set_param_count + row_count);
130            for col_prop in &set_props {
131                for (snap, keys) in entity_data[start..end].iter() {
132                    let pk_val = keys.get(pk_field).cloned().unwrap_or(DbValue::Null);
133                    let col_val = snap
134                        .get(col_prop.field_name.as_ref())
135                        .cloned()
136                        .unwrap_or(DbValue::Null);
137                    params.push(pk_val);
138                    params.push(col_val);
139                }
140            }
141            // WHERE IN params: pk values
142            for (_, keys) in entity_data[start..end].iter() {
143                let pk_val = keys.get(pk_field).cloned().unwrap_or(DbValue::Null);
144                params.push(pk_val);
145            }
146
147            // Append filter to WHERE clause.
148            if let Some(filter) = query_filter {
149                let filter_sql = compile_bool_expr(filter, gen, &mut idx);
150                params.extend(filter_params.iter().cloned());
151                where_clause = format!("({}) AND ({})", where_clause, filter_sql);
152            }
153
154            let sql = gen.update_batch(
155                meta.table_name.as_ref(),
156                &set_cols,
157                pk_col,
158                row_count,
159                &where_clause,
160            );
161            let rows = conn.execute(&sql, &params).await?;
162            if rows == 0 && row_count > 0 {
163                return Err(EFError::concurrency_conflict(format!(
164                    "batch update affected 0 rows on {} (rows may have been modified or deleted)",
165                    meta.table_name
166                )));
167            }
168            updated += (rows as usize).min(row_count);
169            start = end;
170        }
171
172        Ok(updated)
173    }
174
175    /// Per-row UPDATE fallback used when concurrency tokens or composite PKs
176    /// prevent batching. Each row gets its own `UPDATE ... SET ... WHERE
177    /// pk = ? AND ...`. When `modified_properties` is non-empty for an entity,
178    /// only those columns are SET (partial update); otherwise all non-PK
179    /// columns are SET (backward compatible).
180    #[allow(clippy::type_complexity)]
181    async fn execute_updates_per_row<E>(
182        conn: &mut dyn IAsyncConnection,
183        gen: &'static dyn crate::provider::ISqlGenerator,
184        entities: &[(
185            &E,
186            &EntityTypeMeta,
187            Option<&HashMap<String, DbValue>>,
188            &[String],
189        )],
190        query_filter: Option<&BoolExpr>,
191    ) -> EFResult<usize>
192    where
193        E: IEntityType + IEntitySnapshot + IGetKeyValues,
194    {
195        let mut updated = 0;
196        let mut sql_cache: HashMap<(String, Vec<String>, String), String> = HashMap::new();
197
198        for (entity, meta, original, modified_props) in entities {
199            let snap = entity.snapshot();
200            let keys = entity.key_values();
201            let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
202
203            // When modified_properties is populated, SET only those columns
204            // (partial update). When empty, SET all non-PK columns.
205            let set_props: Vec<&PropertyMeta> = if modified_props.is_empty() {
206                scalar_props
207                    .iter()
208                    .copied()
209                    .filter(|p| !p.is_primary_key)
210                    .collect()
211            } else {
212                let modified_set: std::collections::HashSet<&str> =
213                    modified_props.iter().map(|s| s.as_str()).collect();
214                scalar_props
215                    .iter()
216                    .copied()
217                    .filter(|p| !p.is_primary_key && modified_set.contains(p.field_name.as_ref()))
218                    .collect()
219            };
220            let set_cols: Vec<&str> = set_props.iter().map(|p| p.column_name.as_ref()).collect();
221
222            if set_cols.is_empty() || keys.is_empty() {
223                continue;
224            }
225
226            let concurrency_tokens: Vec<&PropertyMeta> = scalar_props
227                .iter()
228                .copied()
229                .filter(|p| p.is_concurrency_token)
230                .collect();
231
232            let (mut where_clause, mut where_params) = build_where_with_concurrency(
233                gen,
234                &keys,
235                &concurrency_tokens,
236                *original,
237                set_cols.len() + 1,
238            )?;
239
240            if let Some(filter) = query_filter {
241                let mut idx = set_cols.len() + where_params.len() + 1;
242                let filter_sql = compile_bool_expr(filter, gen, &mut idx);
243                where_params.extend(collect_bool_expr_values(filter));
244                where_clause = format!("({}) AND ({})", where_clause, filter_sql);
245            }
246
247            let sql = sql_cache
248                .entry((
249                    meta.table_name.to_string(),
250                    set_cols.iter().map(|s| (*s).to_string()).collect(),
251                    where_clause.clone(),
252                ))
253                .or_insert_with(|| gen.update(meta.table_name.as_ref(), &set_cols, &where_clause))
254                .clone();
255
256            let mut params: Vec<DbValue> = set_props
257                .iter()
258                .map(|p| {
259                    snap.get(p.field_name.as_ref())
260                        .cloned()
261                        .unwrap_or(DbValue::Null)
262                })
263                .collect();
264            params.extend(where_params);
265
266            let rows = conn.execute(&sql, &params).await?;
267            if rows == 0 {
268                return Err(EFError::concurrency_conflict(format!(
269                    "update affected 0 rows on {} (row may have been modified or deleted)",
270                    meta.table_name
271                )));
272            }
273            updated += 1;
274        }
275
276        Ok(updated)
277    }
278
279    /// Executes DELETE statements for all deleted entities.
280    ///
281    /// When no concurrency tokens are present and the entity has a single-column
282    /// primary key, rows are batched into `DELETE ... WHERE pk IN (?, ?, ...)`
283    /// statements (≤900 params per batch) to minimize round trips. Otherwise
284    /// (concurrency tokens, composite PK), falls back to per-row DELETE so
285    /// optimistic-concurrency checks run on each row.
286    ///
287    /// When `query_filter` is `Some`, the filter is AND-ed into the WHERE
288    /// clause so deletes cannot cross the filter boundary.
289    #[allow(clippy::type_complexity)]
290    pub async fn execute_deletes<E>(
291        conn: &mut dyn IAsyncConnection,
292        provider: &dyn IDatabaseProvider,
293        entities: &[(&E, &EntityTypeMeta, Option<&HashMap<String, DbValue>>)],
294        query_filter: Option<&BoolExpr>,
295    ) -> EFResult<usize>
296    where
297        E: IEntityType + IGetKeyValues,
298    {
299        if entities.is_empty() {
300            return Ok(0);
301        }
302        let gen = provider.sql_generator();
303        let meta = entities[0].1;
304        let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
305        let has_concurrency_tokens = scalar_props.iter().any(|p| p.is_concurrency_token);
306        let pk_props: Vec<_> = scalar_props.iter().filter(|p| p.is_primary_key).collect();
307
308        // Fall back to per-row DELETE when optimistic concurrency tokens are
309        // present (each row needs its own WHERE to check the token) or when
310        // the PK is composite (IN clause only works for a single column).
311        if has_concurrency_tokens || pk_props.len() != 1 {
312            return Self::execute_deletes_per_row(conn, gen, entities, query_filter).await;
313        }
314
315        let pk_col = pk_props[0].column_name.as_ref();
316        let pk_field = pk_props[0].field_name.as_ref();
317
318        // Collect PK values; entities missing the PK value are skipped.
319        let pk_values: Vec<DbValue> = entities
320            .iter()
321            .filter_map(|(e, _, _)| e.key_values().get(pk_field).cloned())
322            .collect();
323        if pk_values.is_empty() {
324            return Ok(0);
325        }
326
327        // Filter params are constant across batches; their SQL is recomputed
328        // per batch because Postgres numbered placeholders depend on batch size.
329        let filter_params: Vec<DbValue> = match query_filter {
330            Some(filter) => collect_bool_expr_values(filter),
331            None => Vec::new(),
332        };
333        const MAX_PARAMS: usize = 900;
334        let batch_size = MAX_PARAMS.saturating_sub(filter_params.len()).max(1);
335
336        let mut deleted = 0usize;
337        let mut start = 0usize;
338        while start < pk_values.len() {
339            let end = (start + batch_size).min(pk_values.len());
340            let batch = &pk_values[start..end];
341            let pk_count = batch.len();
342
343            let pk_placeholders: Vec<String> = (1..=pk_count)
344                .map(|i| gen.parameter_placeholder(i))
345                .collect();
346            let mut where_clause = format!(
347                "{} IN ({})",
348                gen.quote_identifier(pk_col),
349                pk_placeholders.join(", "),
350            );
351            let mut params: Vec<DbValue> = batch.to_vec();
352            if let Some(filter) = query_filter {
353                // Filter placeholders are numbered after the PK IN-list.
354                let mut idx = pk_count + 1;
355                let filter_sql = compile_bool_expr(filter, gen, &mut idx);
356                params.extend(filter_params.iter().cloned());
357                where_clause = format!("({}) AND ({})", where_clause, filter_sql);
358            }
359
360            let sql = gen.delete(meta.table_name.as_ref(), &where_clause);
361            let rows = conn.execute(&sql, &params).await?;
362            if rows == 0 && pk_count > 0 {
363                return Err(EFError::concurrency_conflict(format!(
364                    "batch delete affected 0 rows on {} (rows may have been modified or deleted)",
365                    meta.table_name
366                )));
367            }
368            deleted += (rows as usize).min(pk_count);
369            start = end;
370        }
371
372        Ok(deleted)
373    }
374
375    /// Per-row DELETE fallback used when concurrency tokens or composite PKs
376    /// prevent batching. Each row gets its own `DELETE ... WHERE pk = ? AND ...`.
377    #[allow(clippy::type_complexity)]
378    async fn execute_deletes_per_row<E>(
379        conn: &mut dyn IAsyncConnection,
380        gen: &'static dyn crate::provider::ISqlGenerator,
381        entities: &[(&E, &EntityTypeMeta, Option<&HashMap<String, DbValue>>)],
382        query_filter: Option<&BoolExpr>,
383    ) -> EFResult<usize>
384    where
385        E: IEntityType + IGetKeyValues,
386    {
387        let mut deleted = 0;
388        for (entity, meta, original) in entities {
389            let keys = entity.key_values();
390            if keys.is_empty() {
391                continue;
392            }
393            let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
394            let concurrency_tokens: Vec<&PropertyMeta> = scalar_props
395                .iter()
396                .copied()
397                .filter(|p| p.is_concurrency_token)
398                .collect();
399
400            let (mut where_clause, mut where_params) =
401                build_where_with_concurrency(gen, &keys, &concurrency_tokens, *original, 1)?;
402
403            if let Some(filter) = query_filter {
404                let mut idx = where_params.len() + 1;
405                let filter_sql = compile_bool_expr(filter, gen, &mut idx);
406                where_params.extend(collect_bool_expr_values(filter));
407                where_clause = format!("({}) AND ({})", where_clause, filter_sql);
408            }
409
410            let sql = gen.delete(meta.table_name.as_ref(), &where_clause);
411            let rows = conn.execute(&sql, &where_params).await?;
412            if rows == 0 {
413                return Err(EFError::concurrency_conflict(format!(
414                    "delete affected 0 rows on {} (row may have been modified or deleted)",
415                    meta.table_name
416                )));
417            }
418            deleted += 1;
419        }
420        Ok(deleted)
421    }
422}
423
424fn build_where_with_concurrency(
425    gen: &dyn crate::provider::ISqlGenerator,
426    keys: &HashMap<String, DbValue>,
427    concurrency_tokens: &[&PropertyMeta],
428    original: Option<&HashMap<String, DbValue>>,
429    start_param_idx: usize,
430) -> EFResult<(String, Vec<DbValue>)> {
431    let mut where_parts: Vec<String> = keys
432        .keys()
433        .enumerate()
434        .map(|(i, k)| {
435            format!(
436                "{} = {}",
437                gen.quote_identifier(k),
438                gen.parameter_placeholder(start_param_idx + i)
439            )
440        })
441        .collect();
442
443    let mut params: Vec<DbValue> = keys.values().cloned().collect();
444
445    for (next_idx, token) in (start_param_idx + keys.len()..).zip(concurrency_tokens.iter()) {
446        where_parts.push(format!(
447            "{} = {}",
448            gen.quote_identifier(token.column_name.as_ref()),
449            gen.parameter_placeholder(next_idx)
450        ));
451
452        let original_val = original
453            .and_then(|o| o.get(token.field_name.as_ref()))
454            .ok_or_else(|| {
455                EFError::change_tracking(format!(
456                    "missing original concurrency token for '{}'",
457                    token.field_name
458                ))
459            })?;
460        params.push(original_val.clone());
461    }
462
463    Ok((where_parts.join(" AND "), params))
464}
rust_ef/change_executor/executor_dml.rs

rust_ef/change_executor/
executor_dml.rs
