rust_ef/change_executor/
executor.rs

1//! `ChangeExecutor` — INSERT and UPSERT execution.
2
3use crate::entity::{IEntitySnapshot, IEntityType, IGetKeyValues};
4use crate::error::EFResult;
5use crate::metadata::EntityTypeMeta;
6use crate::provider::{DbValue, IAsyncConnection, IDatabaseProvider};
7
8/// Executes INSERT/UPDATE/DELETE for tracked entities within a transaction.
9pub struct ChangeExecutor;
10
11impl ChangeExecutor {
12    /// Executes INSERT statements for all added entities.
13    ///
14    /// Rows are batched into multi-value `INSERT INTO ... VALUES (...), (...)`
15    /// statements to minimize round trips. Batches are sized so that the total
16    /// parameter count stays ≤ 900 (SQLite's variable limit is 999; we use a
17    /// conservative ceiling that also fits MySQL/PG). Auto-increment columns
18    /// are excluded from the column list (the DB assigns them).
19    ///
20    /// `on_key_backfill` is invoked once per inserted row with the
21    /// database-generated auto-increment PK (when present). For PostgreSQL the
22    /// PKs are read from the `RETURNING *` result set; for SQLite/MySQL a
23    /// follow-up `last_insert_rowid()` / `LAST_INSERT_ID()` query retrieves
24    /// the first/last ID and the remaining batch keys are computed by
25    /// sequential increment.
26    pub async fn execute_inserts<E, F>(
27        conn: &mut dyn IAsyncConnection,
28        provider: &dyn IDatabaseProvider,
29        entities: &[(&E, &EntityTypeMeta)],
30        mut on_key_backfill: F,
31    ) -> EFResult<usize>
32    where
33        E: IEntityType + IEntitySnapshot + IGetKeyValues,
34        F: FnMut(usize, i64),
35    {
36        if entities.is_empty() {
37            return Ok(0);
38        }
39        let gen = provider.sql_generator();
40        let meta = entities[0].1;
41        let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
42        if scalar_props.is_empty() {
43            return Ok(0);
44        }
45        let insert_cols: Vec<&str> = scalar_props
46            .iter()
47            .filter(|p| !p.is_primary_key || (!p.is_auto_increment && !p.is_sequence))
48            .map(|p| p.column_name.as_ref())
49            .collect();
50        if insert_cols.is_empty() {
51            return Ok(0);
52        }
53
54        // Identify the auto-increment/sequence PK column (if any) for key backfill.
55        let auto_inc_pk = scalar_props
56            .iter()
57            .find(|p| (p.is_auto_increment || p.is_sequence) && p.is_primary_key);
58
59        // Conservative per-statement parameter ceiling (SQLite limit 999).
60        const MAX_PARAMS: usize = 900;
61        let batch_size = (MAX_PARAMS / insert_cols.len()).max(1);
62
63        let mut inserted = 0usize;
64        let mut start = 0usize;
65        while start < entities.len() {
66            let end = (start + batch_size).min(entities.len());
67            let batch = &entities[start..end];
68            let row_count = batch.len();
69
70            let sql = gen.insert_batch(meta.table_name.as_ref(), &insert_cols, row_count);
71            let mut params: Vec<DbValue> = Vec::with_capacity(row_count * insert_cols.len());
72            for (entity, _) in batch {
73                let snap = entity.snapshot();
74                for p in &scalar_props {
75                    if !p.is_primary_key || (!p.is_auto_increment && !p.is_sequence) {
76                        params.push(
77                            snap.get(p.field_name.as_ref())
78                                .cloned()
79                                .unwrap_or(DbValue::Null),
80                        );
81                    }
82                }
83            }
84
85            if let Some(_pk_prop) = auto_inc_pk {
86                if gen.supports_returning() {
87                    // PostgreSQL: INSERT ... RETURNING * returns all generated
88                    // rows. Extract the PK column from each returned row.
89                    let rows = conn.query(&sql, &params).await?;
90                    let affected = rows.len().min(row_count);
91                    for (i, row) in rows.iter().enumerate().take(affected) {
92                        if let Some(pk_val) = row.first() {
93                            if let Ok(key) = pk_val.clone().try_into() {
94                                on_key_backfill(start + i, key);
95                            }
96                        }
97                    }
98                    inserted += affected;
99                } else if let Some(last_id_sql) = gen.last_insert_id_sql() {
100                    // SQLite/MySQL: execute INSERT, then query for the
101                    // generated ID and compute the batch key sequence.
102                    conn.execute(&sql, &params).await?;
103                    let id_rows = conn.query(last_id_sql, &[]).await?;
104                    if let Some(id_row) = id_rows.first() {
105                        if let Some(id_val) = id_row.first() {
106                            let raw_id: i64 = id_val.clone().try_into().unwrap_or(0);
107                            let first_id = if gen.last_insert_id_returns_first() {
108                                raw_id
109                            } else {
110                                raw_id - row_count as i64 + 1
111                            };
112                            for i in 0..row_count {
113                                on_key_backfill(start + i, first_id + i as i64);
114                            }
115                        }
116                    }
117                    inserted += row_count;
118                } else {
119                    // No RETURNING and no last_insert_id — cannot backfill.
120                    let rows = conn.execute(&sql, &params).await?;
121                    let affected = (rows as usize).min(row_count);
122                    for i in 0..affected {
123                        on_key_backfill(start + i, 0);
124                    }
125                    inserted += affected;
126                }
127            } else {
128                // No auto-increment PK — entity already has its key.
129                let rows = conn.execute(&sql, &params).await?;
130                let affected = (rows as usize).min(row_count);
131                for i in 0..affected {
132                    on_key_backfill(start + i, 0);
133                }
134                inserted += affected;
135            }
136
137            start = end;
138        }
139
140        Ok(inserted)
141    }
142
143    /// Executes UPSERT statements (INSERT ... ON CONFLICT DO UPDATE) for
144    /// entities marked via `DbSet::upsert`.
145    ///
146    /// Unlike `execute_inserts`, ALL mapped columns (including auto-increment
147    /// PKs) are included in the INSERT column list so the caller-provided PK
148    /// value participates in the conflict check. The conflict target is the PK
149    /// column(s). The UPDATE SET clause covers all non-PK columns.
150    /// No PK backfill is performed — upsert callers are expected to know the key.
151    pub async fn execute_upserts<E>(
152        conn: &mut dyn IAsyncConnection,
153        provider: &dyn IDatabaseProvider,
154        entities: &[(&E, &EntityTypeMeta)],
155    ) -> EFResult<usize>
156    where
157        E: IEntityType + IEntitySnapshot + IGetKeyValues,
158    {
159        if entities.is_empty() {
160            return Ok(0);
161        }
162        let gen = provider.sql_generator();
163        let meta = entities[0].1;
164        let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
165        if scalar_props.is_empty() {
166            return Ok(0);
167        }
168        // For upsert, include ALL mapped columns (including auto-increment PK)
169        // so the caller's key value is part of the INSERT and can trigger the
170        // ON CONFLICT path.
171        let insert_cols: Vec<&str> = scalar_props
172            .iter()
173            .map(|p| p.column_name.as_ref())
174            .collect();
175        if insert_cols.is_empty() {
176            return Ok(0);
177        }
178
179        // Conflict target: PK column names.
180        let conflict_cols: Vec<&str> = scalar_props
181            .iter()
182            .filter(|p| p.is_primary_key)
183            .map(|p| p.column_name.as_ref())
184            .collect();
185        if conflict_cols.is_empty() {
186            return Err(crate::error::EFError::configuration(
187                "Upsert requires at least one primary key column as conflict target.",
188            ));
189        }
190
191        const MAX_PARAMS: usize = 900;
192        let batch_size = (MAX_PARAMS / insert_cols.len()).max(1);
193
194        let mut upserted = 0usize;
195        let mut start = 0usize;
196        while start < entities.len() {
197            let end = (start + batch_size).min(entities.len());
198            let batch = &entities[start..end];
199            let row_count = batch.len();
200
201            let sql = gen.upsert_batch(
202                meta.table_name.as_ref(),
203                &insert_cols,
204                &conflict_cols,
205                row_count,
206            );
207            let mut params: Vec<DbValue> = Vec::with_capacity(row_count * insert_cols.len());
208            for (entity, _) in batch {
209                let snap = entity.snapshot();
210                for p in &scalar_props {
211                    params.push(
212                        snap.get(p.field_name.as_ref())
213                            .cloned()
214                            .unwrap_or(DbValue::Null),
215                    );
216                }
217            }
218
219            let affected = conn.execute(&sql, &params).await?;
220            upserted += (affected as usize).min(row_count);
221
222            start = end;
223        }
224
225        Ok(upserted)
226    }
227}
rust_ef/change_executor/executor.rs

rust_ef/change_executor/
executor.rs
