dynoxide/actions/
execute_transaction.rs

1use crate::actions::helpers;
2use crate::actions::index_capacity::{
3    WriteCapacity, aggregate_by_table, per_table_capacity, transactional_read_units,
4};
5use crate::errors::{CancellationReason, DynoxideError, Result};
6use crate::partiql;
7use crate::partiql::executor::ResolvedTable;
8use crate::storage_backend::StorageBackend;
9use crate::types::{AttributeValue, Item};
10use serde::{Deserialize, Serialize};
11use std::collections::{HashMap, HashSet};
12
13#[derive(Debug, Clone, Default, Deserialize)]
14pub struct ExecuteTransactionRequest {
15    #[serde(rename = "TransactStatements")]
16    pub transact_statements: Vec<ParameterizedStatement>,
17    #[serde(rename = "ClientRequestToken", default)]
18    pub client_request_token: Option<String>,
19    #[serde(rename = "ReturnConsumedCapacity", default)]
20    pub return_consumed_capacity: Option<String>,
21}
22
23// `Serialize` backs the idempotency request hash (the statements and their
24// parameters are serialised via `serde_json`), so a same-token call differing
25// only in `ReturnConsumedCapacity` replays rather than mismatches.
26#[derive(Debug, Clone, Default, Deserialize, Serialize)]
27pub struct ParameterizedStatement {
28    #[serde(rename = "Statement")]
29    pub statement: String,
30    #[serde(rename = "Parameters", default)]
31    pub parameters: Option<Vec<AttributeValue>>,
32}
33
34// `Clone` so the idempotency cache can store the first-call response and clone
35// its `Responses` for the replay.
36#[derive(Debug, Clone, Default, Serialize)]
37pub struct ExecuteTransactionResponse {
38    #[serde(rename = "Responses", skip_serializing_if = "Option::is_none")]
39    pub responses: Option<Vec<ItemResponse>>,
40    #[serde(rename = "ConsumedCapacity", skip_serializing_if = "Option::is_none")]
41    pub consumed_capacity: Option<Vec<crate::types::ConsumedCapacity>>,
42}
43
44/// A first-call result together with what a same-token replay needs to bill it.
45///
46/// The replay is charged against the image each statement touched, which the
47/// statement text cannot supply: a `DELETE` names a key, not the row it
48/// removed. Those sizes are internal bookkeeping, so they live here, on the type
49/// the idempotency cache holds, rather than on the response type callers see.
50#[derive(Debug, Clone, Default)]
51pub(crate) struct CachedTransaction {
52    pub(crate) response: ExecuteTransactionResponse,
53    /// Per-statement `(table, image size)`.
54    pub(crate) replay_sizes: Vec<(String, usize)>,
55}
56
57#[derive(Debug, Clone, Default, Serialize)]
58pub struct ItemResponse {
59    #[serde(rename = "Item", skip_serializing_if = "Option::is_none")]
60    pub item: Option<Item>,
61}
62
63/// Run a PartiQL transaction.
64///
65/// Callers driving idempotency want `execute_cached`, which also hands back
66/// the sizes a replay is billed against.
67pub async fn execute<S: StorageBackend>(
68    storage: &S,
69    request: ExecuteTransactionRequest,
70) -> Result<ExecuteTransactionResponse> {
71    Ok(execute_cached(storage, request).await?.response)
72}
73
74/// Reject a `ReturnConsumedCapacity` outside the enum.
75///
76/// Called at the entry points ahead of the idempotency cache as well as inside
77/// `execute_cached`. A same-token replay never reaches `execute_cached`, and
78/// the token is keyed on the statements alone, so a second call may carry a
79/// capacity mode the first one never did. A check living only in
80/// `execute_cached` lets that one through. It stays there for callers that reach
81/// it directly.
82pub(crate) fn reject_bad_capacity_mode(mode: Option<&str>) -> Result<()> {
83    match crate::validation::return_consumed_capacity_rejection(mode) {
84        Some(msg) => Err(DynoxideError::ValidationException(
85            crate::validation::envelope_message(&msg),
86        )),
87        None => Ok(()),
88    }
89}
90
91pub(crate) async fn execute_cached<S: StorageBackend>(
92    storage: &S,
93    request: ExecuteTransactionRequest,
94) -> Result<CachedTransaction> {
95    reject_bad_capacity_mode(request.return_consumed_capacity.as_deref())?;
96
97    let statements = &request.transact_statements;
98
99    // Validate: must have between 1 and 100 statements
100    if statements.is_empty() {
101        return Err(DynoxideError::ValidationException(
102            "1 validation error detected: Value at 'transactStatements' failed to satisfy constraint: Member must have length greater than or equal to 1".to_string(),
103        ));
104    }
105    if statements.len() > 100 {
106        return Err(DynoxideError::ValidationException(
107            "Member must have length less than or equal to 100".to_string(),
108        ));
109    }
110
111    // Parse all statements before executing any, to fail fast on syntax errors
112    let mut parsed = Vec::with_capacity(statements.len());
113    for (index, stmt) in statements.iter().enumerate() {
114        // A COUNT projection is rejected before parsing, at the same top level
115        // as a parse error but with the bare message captured on
116        // ExecuteStatement, not the wasn't-well-formed wrapper.
117        if let Some(msg) = partiql::parser::count_projection_rejection(&stmt.statement) {
118            return Err(DynoxideError::ValidationException(msg));
119        }
120        let ast = partiql::parser::parse(&stmt.statement)
121            .map_err(|e| DynoxideError::ValidationException(e.into_message()))?;
122        // DynamoDB rejects a RETURNING clause on any member of a transaction with
123        // a top-level ValidationException, before applying any write. This is a
124        // plain validation failure, not a TransactionCanceledException.
125        if partiql::parser::returning_variant(&ast).is_some() {
126            return Err(DynoxideError::ValidationException(format!(
127                "Validation failed in TransactStatements[{index}]: RETURNING clause is not supported in ExecuteTransaction."
128            )));
129        }
130        // An index-qualified read is rejected outright inside a transaction, so
131        // there is no arm for it to be charged to. Rejected up front like the
132        // RETURNING case, not as a cancellation. Captured eu-west-2 2026-08-15.
133        if matches!(ast, partiql::parser::Statement::Select { .. })
134            && partiql::parser::index_name(&ast).is_some()
135        {
136            return Err(DynoxideError::ValidationException(format!(
137                "Validation failed in TransactStatements[{index}]: Reads on indices are not supported within transactions."
138            )));
139        }
140        let params = stmt.parameters.clone().unwrap_or_default();
141        parsed.push((ast, params));
142    }
143
144    // A transaction is all-read or all-write, and may not touch one item twice.
145    // AWS rejects both up front, before any statement runs.
146    let reads = parsed
147        .iter()
148        .filter(|(stmt, _)| matches!(stmt, partiql::parser::Statement::Select { .. }))
149        .count();
150    if reads > 0 && reads < parsed.len() {
151        return Err(DynoxideError::ValidationException(
152            "ExecuteTransaction API does not support both read and write operations in the same request."
153                .to_string(),
154        ));
155    }
156
157    // Duplicate detection resolves each statement's table, and the executor
158    // needs the same metadata and key schema a moment later, so they are kept
159    // here rather than resolved twice per statement.
160    let mut seen_targets = HashSet::new();
161    // Keyed by table rather than by statement: the metadata and key schema are
162    // per table, and only the key is per statement.
163    let mut tables: HashMap<String, ResolvedTable> = HashMap::new();
164    for (stmt, params) in &parsed {
165        let Some(name) = partiql::parser::table_name(stmt) else {
166            continue;
167        };
168        if !tables.contains_key(name)
169            && let Ok(resolved) = ResolvedTable::load(storage, name).await
170        {
171            tables.insert(name.to_string(), resolved);
172        }
173        // Absent when the table could not be read, which leaves the statement to
174        // fail on its own terms during execution rather than as a duplicate.
175        let Some(resolved) = tables.get(name) else {
176            continue;
177        };
178        if let Some(target) = partiql::executor::statement_target_in(stmt, params, resolved)
179            && !seen_targets.insert(target)
180        {
181            return Err(DynoxideError::ValidationException(
182                "Transaction request cannot include multiple operations on one item".to_string(),
183            ));
184        }
185    }
186
187    // All statements run inside one SQLite transaction (all-or-nothing).
188    let (responses, charges) = helpers::with_write_transaction(
189        storage,
190        execute_within_transaction(
191            storage,
192            &parsed,
193            &tables,
194            request.return_consumed_capacity.as_deref(),
195        ),
196    )
197    .await?;
198
199    // Transactional capacity, split by statement kind: an all-SELECT read set
200    // reports read capacity, any INSERT/UPDATE/DELETE makes it a write set. Both
201    // are sized on the items each statement touched, and the transactional
202    // factor reaches the base table arm only.
203    let mode = &request.return_consumed_capacity;
204    let consumed_capacity = if is_read_set(&parsed) {
205        crate::types::build_transactional_capacity(
206            &transactional_read_units(&replay_sizes(&charges)),
207            mode,
208            crate::types::transactional_read_capacity,
209        )
210    } else {
211        build_write_capacity(&charges, mode)
212    };
213
214    Ok(CachedTransaction {
215        response: ExecuteTransactionResponse {
216            responses: Some(responses),
217            consumed_capacity,
218        },
219        replay_sizes: replay_sizes(&charges),
220    })
221}
222
223/// What one statement contributed to the transaction's capacity.
224enum StatementCharge {
225    /// A `SELECT`, charged on the rows it walked at read granularity.
226    Read { table_name: String, size: usize },
227    /// A write, charged on its images and its per-index units.
228    Write(WriteCapacity),
229}
230
231impl StatementCharge {
232    fn table_name(&self) -> &str {
233        match self {
234            Self::Read { table_name, .. } => table_name,
235            Self::Write(capacity) => &capacity.table_name,
236        }
237    }
238
239    /// The image size this statement is charged on, for a replay.
240    fn size(&self) -> usize {
241        match self {
242            Self::Read { size, .. } => *size,
243            Self::Write(capacity) => capacity
244                .old_size
245                .unwrap_or(0)
246                .max(capacity.new_size.unwrap_or(0)),
247        }
248    }
249
250    /// The write record for this statement.
251    ///
252    /// A mixed set is rejected before execution, so a `Read` charge never
253    /// reaches the write path in practice. It maps to a table-only record
254    /// rather than panicking, so a future relaxation of that check degrades to
255    /// an over-estimate instead of a wrong shape.
256    fn as_write(&self) -> WriteCapacity {
257        match self {
258            Self::Write(capacity) => capacity.clone(),
259            Self::Read { table_name, size } => WriteCapacity::new(
260                table_name,
261                None,
262                Some(*size),
263                HashMap::new(),
264                HashMap::new(),
265            ),
266        }
267    }
268}
269
270/// Fold the per-statement records into one `ConsumedCapacity` per table.
271fn build_write_capacity(
272    charges: &[StatementCharge],
273    mode: &Option<String>,
274) -> Option<Vec<crate::types::ConsumedCapacity>> {
275    // Checked before the clone, not just inside the shared builder. Most calls
276    // ask for no capacity, and `as_write` clones every record and its index
277    // maps before the builder could discard the lot.
278    if !matches!(mode.as_deref(), Some("TOTAL") | Some("INDEXES")) {
279        return None;
280    }
281
282    let records: Vec<WriteCapacity> = charges.iter().map(StatementCharge::as_write).collect();
283    let by_table = aggregate_by_table(&records, crate::types::TRANSACTIONAL_CAPACITY_FACTOR);
284    per_table_capacity(
285        &by_table,
286        mode,
287        crate::types::transactional_write_capacity_with_indexes,
288    )
289}
290
291/// The image size each statement was charged on, kept for a same-token replay.
292fn replay_sizes(charges: &[StatementCharge]) -> Vec<(String, usize)> {
293    charges
294        .iter()
295        .map(|charge| (charge.table_name().to_string(), charge.size()))
296        .collect()
297}
298
299/// A transaction is a read set only when every statement is a `SELECT`; any
300/// `INSERT`/`UPDATE`/`DELETE` makes it a write set. AWS requires a transaction
301/// to be all-read or all-write and rejects a mixed set before capacity is
302/// computed, but dynoxide does not enforce that, so a mixed set is classified
303/// here as a write set. Revisit the predicate if condition-only checks (which
304/// AWS counts in the write set) are ever parsed.
305fn is_read_set(parsed: &[(partiql::parser::Statement, Vec<AttributeValue>)]) -> bool {
306    parsed
307        .iter()
308        .all(|(stmt, _)| matches!(stmt, partiql::parser::Statement::Select { .. }))
309}
310
311/// Build the response for a same-token idempotent replay. The statements are
312/// identical to the first call (the idempotency hash matched), so `Responses`
313/// carry over from the cached first call and capacity is reported as a
314/// transactional READ, honouring the replay request's own
315/// `ReturnConsumedCapacity` mode (the original call's mode does not carry over).
316/// The statements are re-parsed to recover per-table units; they parsed
317/// successfully on the first call, so an unexpected parse error just drops that
318/// statement from the estimate rather than failing the replay.
319pub(crate) fn replay_response(
320    cached: &CachedTransaction,
321    mode: &Option<String>,
322) -> CachedTransaction {
323    CachedTransaction {
324        response: ExecuteTransactionResponse {
325            responses: cached.response.responses.clone(),
326            consumed_capacity: crate::types::build_transactional_capacity(
327                &transactional_read_units(&cached.replay_sizes),
328                mode,
329                crate::types::transactional_read_capacity,
330            ),
331        },
332        replay_sizes: cached.replay_sizes.clone(),
333    }
334}
335
336/// Run every statement, returning the responses and what each contributed to
337/// `ConsumedCapacity`. The charges are only meaningful when the whole
338/// transaction commits; a cancellation returns an error and reports nothing.
339async fn execute_within_transaction<S: StorageBackend>(
340    storage: &S,
341    parsed: &[(partiql::parser::Statement, Vec<AttributeValue>)],
342    tables: &HashMap<String, ResolvedTable>,
343    capacity_mode: Option<&str>,
344) -> Result<(Vec<ItemResponse>, Vec<StatementCharge>)> {
345    let mut responses = Vec::with_capacity(parsed.len());
346    let mut charges: Vec<StatementCharge> = Vec::with_capacity(parsed.len());
347    let mut cancellation_reasons: Vec<CancellationReason> = Vec::with_capacity(parsed.len());
348
349    for (stmt, params) in parsed {
350        match partiql::executor::execute_page(
351            storage,
352            stmt,
353            params,
354            None,
355            None,
356            false,
357            capacity_mode,
358            partiql::parser::table_name(stmt).and_then(|t| tables.get(t)),
359        )
360        .await
361        {
362            Ok(page) => {
363                let table_name = partiql::parser::table_name(stmt).unwrap_or_default();
364                charges.push(match page.capacity {
365                    Some(capacity) => StatementCharge::Write(capacity),
366                    None => StatementCharge::Read {
367                        table_name: table_name.to_string(),
368                        size: page.size,
369                    },
370                });
371                let item = page.items.and_then(|items| items.into_iter().next());
372                responses.push(ItemResponse { item });
373                cancellation_reasons.push(CancellationReason {
374                    code: "None".to_string(),
375                    message: None,
376                    item: None,
377                });
378            }
379            Err(e) => {
380                // Record the failure reason
381                let message = Some(e.to_string());
382                let (code, item) = match e {
383                    DynoxideError::ConditionalCheckFailedException(_, item) => {
384                        ("ConditionalCheckFailed".to_string(), item)
385                    }
386                    DynoxideError::DuplicateItemException(_) => ("DuplicateItem".to_string(), None),
387                    // Group KeyEmptyValueValidation with ValidationException so an empty-value
388                    // key keeps the "ValidationError" reason instead of falling through to
389                    // InternalError (#95).
390                    DynoxideError::ValidationException(_)
391                    | DynoxideError::KeyEmptyValueValidation(_) => {
392                        ("ValidationError".to_string(), None)
393                    }
394                    _ => ("InternalError".to_string(), None),
395                };
396                responses.push(ItemResponse { item: None });
397                cancellation_reasons.push(CancellationReason {
398                    code,
399                    message,
400                    item,
401                });
402
403                // Fill remaining slots with None and stop: don't execute
404                // statements that will be rolled back.
405                for _ in responses.len()..parsed.len() {
406                    responses.push(ItemResponse { item: None });
407                    cancellation_reasons.push(CancellationReason {
408                        code: "None".to_string(),
409                        message: None,
410                        item: None,
411                    });
412                }
413
414                let codes: Vec<&str> = cancellation_reasons
415                    .iter()
416                    .map(|r| r.code.as_str())
417                    .collect();
418                let message = format!(
419                    "Transaction cancelled, please refer cancellation reasons for specific reasons [{}]",
420                    codes.join(", ")
421                );
422                return Err(DynoxideError::TransactionCanceledException(
423                    message,
424                    cancellation_reasons,
425                ));
426            }
427        }
428    }
429
430    Ok((responses, charges))
431}
dynoxide/actions/execute_transaction.rs

dynoxide/actions/
execute_transaction.rs
