rust_ef/
change_executor.rs

1//! Change executor — generates and executes SQL for entity state changes.
2//!
3//! The `ChangeExecutor` takes a collection of tracked entities grouped by
4//! state (Added/Modified/Deleted), generates the appropriate parameterized
5//! DML, and executes it against the database via the provider.
6
7use crate::entity::{IEntitySnapshot, IEntityType, IGetKeyValues};
8use crate::error::{EFError, EFResult};
9use crate::metadata::{EntityTypeMeta, PropertyMeta};
10use crate::provider::{DbValue, IAsyncConnection, IDatabaseProvider};
11use crate::query::{collect_bool_expr_values, compile_bool_expr, BoolExpr};
12use std::collections::HashMap;
13
14/// Executes INSERT/UPDATE/DELETE for tracked entities within a transaction.
15pub struct ChangeExecutor;
16
17impl ChangeExecutor {
18    /// Executes INSERT statements for all added entities.
19    ///
20    /// Rows are batched into multi-value `INSERT INTO ... VALUES (...), (...)`
21    /// statements to minimize round trips. Batches are sized so that the total
22    /// parameter count stays ≤ 900 (SQLite's variable limit is 999; we use a
23    /// conservative ceiling that also fits MySQL/PG). Auto-increment columns
24    /// are excluded from the column list (the DB assigns them).
25    ///
26    /// `on_key_backfill` is invoked once per inserted row. In batch mode the
27    /// per-row generated key is not available, so `0` is passed as the key
28    /// (the callback is currently a no-op in the framework).
29    pub async fn execute_inserts<E, F>(
30        conn: &mut dyn IAsyncConnection,
31        provider: &dyn IDatabaseProvider,
32        entities: &[(&E, &EntityTypeMeta)],
33        mut on_key_backfill: F,
34    ) -> EFResult<usize>
35    where
36        E: IEntityType + IEntitySnapshot + IGetKeyValues,
37        F: FnMut(usize, i64),
38    {
39        if entities.is_empty() {
40            return Ok(0);
41        }
42        let gen = provider.sql_generator();
43        let meta = entities[0].1;
44        let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
45        if scalar_props.is_empty() {
46            return Ok(0);
47        }
48        let insert_cols: Vec<&str> = scalar_props
49            .iter()
50            .filter(|p| !p.is_auto_increment || !p.is_primary_key)
51            .map(|p| p.column_name.as_ref())
52            .collect();
53        if insert_cols.is_empty() {
54            return Ok(0);
55        }
56
57        // Conservative per-statement parameter ceiling (SQLite limit 999).
58        const MAX_PARAMS: usize = 900;
59        let batch_size = (MAX_PARAMS / insert_cols.len()).max(1);
60
61        let mut inserted = 0usize;
62        let mut start = 0usize;
63        while start < entities.len() {
64            let end = (start + batch_size).min(entities.len());
65            let batch = &entities[start..end];
66            let row_count = batch.len();
67
68            let sql = gen.insert_batch(meta.table_name.as_ref(), &insert_cols, row_count);
69            let mut params: Vec<DbValue> = Vec::with_capacity(row_count * insert_cols.len());
70            for (entity, _) in batch {
71                let snap = entity.snapshot();
72                for p in &scalar_props {
73                    if !p.is_auto_increment || !p.is_primary_key {
74                        params.push(
75                            snap.get(p.field_name.as_ref())
76                                .cloned()
77                                .unwrap_or(DbValue::Null),
78                        );
79                    }
80                }
81            }
82
83            let rows = conn.execute(&sql, &params).await?;
84            let affected = (rows as usize).min(row_count);
85            for i in 0..affected {
86                on_key_backfill(start + i, 0);
87            }
88            inserted += affected;
89            start = end;
90        }
91
92        Ok(inserted)
93    }
94
95    /// Executes UPDATE statements for all modified entities.
96    /// Uses original snapshots for optimistic concurrency tokens in the WHERE clause.
97    ///
98    /// When `query_filter` is `Some`, the filter (e.g. a tenant-id predicate)
99    /// is AND-ed into the WHERE clause so updates cannot cross the filter
100    /// boundary (multi-tenant / soft-delete isolation).
101    #[allow(clippy::type_complexity)]
102    pub async fn execute_updates<E>(
103        conn: &mut dyn IAsyncConnection,
104        provider: &dyn IDatabaseProvider,
105        entities: &[(&E, &EntityTypeMeta, Option<&HashMap<String, DbValue>>)],
106        query_filter: Option<&BoolExpr>,
107    ) -> EFResult<usize>
108    where
109        E: IEntityType + IEntitySnapshot + IGetKeyValues,
110    {
111        let gen = provider.sql_generator();
112        let mut updated = 0;
113        // Cache of `(table, set_cols, where_clause) -> UPDATE SQL template`.
114        // For N rows of the same entity type the SQL string is identical (only
115        // the bound parameter values differ), so this avoids N-1 redundant
116        // `format!` calls. Scoped to this call (DbContext lifetime).
117        let mut sql_cache: HashMap<(String, Vec<String>, String), String> = HashMap::new();
118
119        for (entity, meta, original) in entities {
120            let snap = entity.snapshot();
121            let keys = entity.key_values();
122            let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
123
124            let set_cols: Vec<&str> = scalar_props
125                .iter()
126                .filter(|p| !p.is_primary_key)
127                .map(|p| p.column_name.as_ref())
128                .collect();
129
130            if set_cols.is_empty() || keys.is_empty() {
131                continue;
132            }
133
134            let concurrency_tokens: Vec<&PropertyMeta> = scalar_props
135                .iter()
136                .copied()
137                .filter(|p| p.is_concurrency_token)
138                .collect();
139
140            let (mut where_clause, mut where_params) = build_where_with_concurrency(
141                gen,
142                &keys,
143                &concurrency_tokens,
144                *original,
145                set_cols.len() + 1,
146            )?;
147
148            // Append the query filter (e.g. tenant_id = ?) to the WHERE clause.
149            // Filter param placeholders are indexed after SET cols + existing WHERE params.
150            if let Some(filter) = query_filter {
151                let mut idx = set_cols.len() + where_params.len() + 1;
152                let filter_sql = compile_bool_expr(filter, gen, &mut idx);
153                where_params.extend(collect_bool_expr_values(filter));
154                where_clause = format!("({}) AND ({})", where_clause, filter_sql);
155            }
156
157            let sql = sql_cache
158                .entry((
159                    meta.table_name.to_string(),
160                    set_cols.iter().map(|s| (*s).to_string()).collect(),
161                    where_clause.clone(),
162                ))
163                .or_insert_with(|| gen.update(meta.table_name.as_ref(), &set_cols, &where_clause))
164                .clone();
165
166            let mut params: Vec<DbValue> = set_cols
167                .iter()
168                .map(|col| {
169                    let prop = scalar_props.iter().find(|p| p.column_name.as_ref() == *col);
170                    match prop {
171                        Some(p) => snap
172                            .get(p.field_name.as_ref())
173                            .cloned()
174                            .unwrap_or(DbValue::Null),
175                        None => DbValue::Null,
176                    }
177                })
178                .collect();
179            params.extend(where_params);
180
181            let rows = conn.execute(&sql, &params).await?;
182            if rows == 0 {
183                return Err(EFError::concurrency_conflict(format!(
184                    "update affected 0 rows on {} (row may have been modified or deleted)",
185                    meta.table_name
186                )));
187            }
188            updated += 1;
189        }
190
191        Ok(updated)
192    }
193
194    /// Executes DELETE statements for all deleted entities.
195    ///
196    /// When no concurrency tokens are present and the entity has a single-column
197    /// primary key, rows are batched into `DELETE ... WHERE pk IN (?, ?, ...)`
198    /// statements (≤900 params per batch) to minimize round trips. Otherwise
199    /// (concurrency tokens, composite PK), falls back to per-row DELETE so
200    /// optimistic-concurrency checks run on each row.
201    ///
202    /// When `query_filter` is `Some`, the filter is AND-ed into the WHERE
203    /// clause so deletes cannot cross the filter boundary.
204    #[allow(clippy::type_complexity)]
205    pub async fn execute_deletes<E>(
206        conn: &mut dyn IAsyncConnection,
207        provider: &dyn IDatabaseProvider,
208        entities: &[(&E, &EntityTypeMeta, Option<&HashMap<String, DbValue>>)],
209        query_filter: Option<&BoolExpr>,
210    ) -> EFResult<usize>
211    where
212        E: IEntityType + IGetKeyValues,
213    {
214        if entities.is_empty() {
215            return Ok(0);
216        }
217        let gen = provider.sql_generator();
218        let meta = entities[0].1;
219        let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
220        let has_concurrency_tokens = scalar_props.iter().any(|p| p.is_concurrency_token);
221        let pk_props: Vec<_> = scalar_props.iter().filter(|p| p.is_primary_key).collect();
222
223        // Fall back to per-row DELETE when optimistic concurrency tokens are
224        // present (each row needs its own WHERE to check the token) or when
225        // the PK is composite (IN clause only works for a single column).
226        if has_concurrency_tokens || pk_props.len() != 1 {
227            return Self::execute_deletes_per_row(conn, gen, entities, query_filter).await;
228        }
229
230        let pk_col = pk_props[0].column_name.as_ref();
231        let pk_field = pk_props[0].field_name.as_ref();
232
233        // Collect PK values; entities missing the PK value are skipped.
234        let pk_values: Vec<DbValue> = entities
235            .iter()
236            .filter_map(|(e, _, _)| e.key_values().get(pk_field).cloned())
237            .collect();
238        if pk_values.is_empty() {
239            return Ok(0);
240        }
241
242        // Filter params are constant across batches; their SQL is recomputed
243        // per batch because Postgres numbered placeholders depend on batch size.
244        let filter_params: Vec<DbValue> = match query_filter {
245            Some(filter) => collect_bool_expr_values(filter),
246            None => Vec::new(),
247        };
248        const MAX_PARAMS: usize = 900;
249        let batch_size = MAX_PARAMS.saturating_sub(filter_params.len()).max(1);
250
251        let mut deleted = 0usize;
252        let mut start = 0usize;
253        while start < pk_values.len() {
254            let end = (start + batch_size).min(pk_values.len());
255            let batch = &pk_values[start..end];
256            let pk_count = batch.len();
257
258            let pk_placeholders: Vec<String> = (1..=pk_count)
259                .map(|i| gen.parameter_placeholder(i))
260                .collect();
261            let mut where_clause = format!(
262                "{} IN ({})",
263                gen.quote_identifier(pk_col),
264                pk_placeholders.join(", "),
265            );
266            let mut params: Vec<DbValue> = batch.to_vec();
267            if let Some(filter) = query_filter {
268                // Filter placeholders are numbered after the PK IN-list.
269                let mut idx = pk_count + 1;
270                let filter_sql = compile_bool_expr(filter, gen, &mut idx);
271                params.extend(filter_params.iter().cloned());
272                where_clause = format!("({}) AND ({})", where_clause, filter_sql);
273            }
274
275            let sql = gen.delete(meta.table_name.as_ref(), &where_clause);
276            let rows = conn.execute(&sql, &params).await?;
277            if rows == 0 && pk_count > 0 {
278                return Err(EFError::concurrency_conflict(format!(
279                    "batch delete affected 0 rows on {} (rows may have been modified or deleted)",
280                    meta.table_name
281                )));
282            }
283            deleted += (rows as usize).min(pk_count);
284            start = end;
285        }
286
287        Ok(deleted)
288    }
289
290    /// Per-row DELETE fallback used when concurrency tokens or composite PKs
291    /// prevent batching. Each row gets its own `DELETE ... WHERE pk = ? AND ...`.
292    #[allow(clippy::type_complexity)]
293    async fn execute_deletes_per_row<E>(
294        conn: &mut dyn IAsyncConnection,
295        gen: &'static dyn crate::provider::ISqlGenerator,
296        entities: &[(&E, &EntityTypeMeta, Option<&HashMap<String, DbValue>>)],
297        query_filter: Option<&BoolExpr>,
298    ) -> EFResult<usize>
299    where
300        E: IEntityType + IGetKeyValues,
301    {
302        let mut deleted = 0;
303        for (entity, meta, original) in entities {
304            let keys = entity.key_values();
305            if keys.is_empty() {
306                continue;
307            }
308            let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
309            let concurrency_tokens: Vec<&PropertyMeta> = scalar_props
310                .iter()
311                .copied()
312                .filter(|p| p.is_concurrency_token)
313                .collect();
314
315            let (mut where_clause, mut where_params) =
316                build_where_with_concurrency(gen, &keys, &concurrency_tokens, *original, 1)?;
317
318            if let Some(filter) = query_filter {
319                let mut idx = where_params.len() + 1;
320                let filter_sql = compile_bool_expr(filter, gen, &mut idx);
321                where_params.extend(collect_bool_expr_values(filter));
322                where_clause = format!("({}) AND ({})", where_clause, filter_sql);
323            }
324
325            let sql = gen.delete(meta.table_name.as_ref(), &where_clause);
326            let rows = conn.execute(&sql, &where_params).await?;
327            if rows == 0 {
328                return Err(EFError::concurrency_conflict(format!(
329                    "delete affected 0 rows on {} (row may have been modified or deleted)",
330                    meta.table_name
331                )));
332            }
333            deleted += 1;
334        }
335        Ok(deleted)
336    }
337}
338
339fn build_where_with_concurrency(
340    gen: &dyn crate::provider::ISqlGenerator,
341    keys: &HashMap<String, DbValue>,
342    concurrency_tokens: &[&PropertyMeta],
343    original: Option<&HashMap<String, DbValue>>,
344    start_param_idx: usize,
345) -> EFResult<(String, Vec<DbValue>)> {
346    let mut where_parts: Vec<String> = keys
347        .keys()
348        .enumerate()
349        .map(|(i, k)| {
350            format!(
351                "{} = {}",
352                gen.quote_identifier(k),
353                gen.parameter_placeholder(start_param_idx + i)
354            )
355        })
356        .collect();
357
358    let mut params: Vec<DbValue> = keys.values().cloned().collect();
359
360    for (next_idx, token) in (start_param_idx + keys.len()..).zip(concurrency_tokens.iter()) {
361        where_parts.push(format!(
362            "{} = {}",
363            gen.quote_identifier(token.column_name.as_ref()),
364            gen.parameter_placeholder(next_idx)
365        ));
366
367        let original_val = original
368            .and_then(|o| o.get(token.field_name.as_ref()))
369            .ok_or_else(|| {
370                EFError::change_tracking(format!(
371                    "missing original concurrency token for '{}'",
372                    token.field_name
373                ))
374            })?;
375        params.push(original_val.clone());
376    }
377
378    Ok((where_parts.join(" AND "), params))
379}
380
381// ---------------------------------------------------------------------------
382// Standalone SQL generation helpers (for use by simplified callers)
383// ---------------------------------------------------------------------------
384
385pub fn generate_insert_sql(
386    provider: &dyn IDatabaseProvider,
387    meta: &EntityTypeMeta,
388    _property_values: &HashMap<String, DbValue>,
389) -> String {
390    let gen = provider.sql_generator();
391    let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
392    let columns: Vec<&str> = scalar_props
393        .iter()
394        .map(|p| p.column_name.as_ref())
395        .collect();
396    if columns.is_empty() {
397        return String::new();
398    }
399    gen.insert(meta.table_name.as_ref(), &columns, true)
400}
401
402pub fn generate_update_sql(
403    provider: &dyn IDatabaseProvider,
404    meta: &EntityTypeMeta,
405    property_values: &HashMap<String, DbValue>,
406    primary_key_values: &HashMap<String, DbValue>,
407) -> String {
408    let gen = provider.sql_generator();
409    let set_columns: Vec<&str> = property_values
410        .keys()
411        .filter(|k| !primary_key_values.contains_key(*k))
412        .map(|k| k.as_str())
413        .collect();
414    if set_columns.is_empty() || primary_key_values.is_empty() {
415        return String::new();
416    }
417    let where_parts: Vec<String> = primary_key_values
418        .keys()
419        .enumerate()
420        .map(|(i, k)| {
421            format!(
422                "{} = {}",
423                gen.quote_identifier(k),
424                gen.parameter_placeholder(i + 1)
425            )
426        })
427        .collect();
428    gen.update(
429        meta.table_name.as_ref(),
430        &set_columns,
431        &where_parts.join(" AND "),
432    )
433}
434
435pub fn generate_delete_sql(
436    provider: &dyn IDatabaseProvider,
437    meta: &EntityTypeMeta,
438    primary_key_values: &HashMap<String, DbValue>,
439) -> String {
440    let gen = provider.sql_generator();
441    if primary_key_values.is_empty() {
442        return String::new();
443    }
444    let where_parts: Vec<String> = primary_key_values
445        .keys()
446        .enumerate()
447        .map(|(i, k)| {
448            format!(
449                "{} = {}",
450                gen.quote_identifier(k),
451                gen.parameter_placeholder(i + 1)
452            )
453        })
454        .collect();
455    gen.delete(meta.table_name.as_ref(), &where_parts.join(" AND "))
456}
457
458pub fn collect_insert_params(
459    meta: &EntityTypeMeta,
460    property_values: &HashMap<String, DbValue>,
461) -> Vec<DbValue> {
462    meta.mapped_scalar_properties()
463        .map(|p| {
464            property_values
465                .get(p.field_name.as_ref())
466                .cloned()
467                .unwrap_or(DbValue::Null)
468        })
469        .collect()
470}
471
472pub fn collect_update_params(
473    property_values: &HashMap<String, DbValue>,
474    primary_key_values: &HashMap<String, DbValue>,
475    set_keys: &[String],
476) -> Vec<DbValue> {
477    let mut params: Vec<DbValue> = set_keys
478        .iter()
479        .filter(|k| !primary_key_values.contains_key(*k))
480        .map(|k| property_values.get(k).cloned().unwrap_or(DbValue::Null))
481        .collect();
482    for v in primary_key_values.values() {
483        params.push(v.clone());
484    }
485    params
486}
487
488pub fn collect_delete_params(primary_key_values: &HashMap<String, DbValue>) -> Vec<DbValue> {
489    primary_key_values.values().cloned().collect()
490}
rust_ef/change_executor.rs

rust_ef/
change_executor.rs
