Skip to main content

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#[derive(Debug, Clone, Default, Deserialize, Serialize)]
24pub struct ParameterizedStatement {
25    #[serde(rename = "Statement")]
26    pub statement: String,
27    #[serde(
28        rename = "Parameters",
29        default,
30        deserialize_with = "crate::types::deserialize_parameters"
31    )]
32    pub parameters: Option<Vec<AttributeValue>>,
33    /// `ALL_OLD` hands back the item this statement's UPDATE or DELETE failed
34    /// its condition on, on the statement's cancellation reason; `NONE`, or
35    /// leaving it out, does not.
36    #[serde(rename = "ReturnValuesOnConditionCheckFailure", default)]
37    pub return_values_on_condition_check_failure: Option<String>,
38}
39
40/// One statement as a `ClientRequestToken` match compares it.
41#[derive(Serialize)]
42pub(crate) struct TokenStatement<'a> {
43    #[serde(rename = "Statement")]
44    statement: &'a str,
45    #[serde(rename = "Parameters")]
46    parameters: Option<&'a [AttributeValue]>,
47    #[serde(rename = "ReturnValuesOnConditionCheckFailure")]
48    mode: &'a str,
49}
50
51/// The statements as the idempotency hash sees them: each one's text,
52/// parameters and mode, and not `ReturnConsumedCapacity`, so a same-token call
53/// differing only in that replays rather than mismatches. A mode left out is
54/// `NONE`, so a same-token retry that leaves it out replays a call that sent
55/// `NONE`, and one sending `ALL_OLD` mismatches (captured eu-west-2 and
56/// us-east-1, 2026-10-10). The statements' own `Serialize` writes a left-out
57/// mode as left out.
58pub(crate) fn token_statements(statements: &[ParameterizedStatement]) -> Vec<TokenStatement<'_>> {
59    statements
60        .iter()
61        .map(|s| TokenStatement {
62            statement: &s.statement,
63            parameters: s.parameters.as_deref(),
64            mode: s
65                .return_values_on_condition_check_failure
66                .as_deref()
67                .unwrap_or("NONE"),
68        })
69        .collect()
70}
71
72// `Clone` so the idempotency cache can store the first-call response and clone
73// its `Responses` for the replay.
74#[derive(Debug, Clone, Default, Serialize)]
75pub struct ExecuteTransactionResponse {
76    #[serde(rename = "Responses", skip_serializing_if = "Option::is_none")]
77    pub responses: Option<Vec<ItemResponse>>,
78    #[serde(rename = "ConsumedCapacity", skip_serializing_if = "Option::is_none")]
79    pub consumed_capacity: Option<Vec<crate::types::ConsumedCapacity>>,
80}
81
82/// A first-call result together with what a same-token replay needs to bill it.
83///
84/// The replay is charged against the image each statement touched, which the
85/// statement text cannot supply: a `DELETE` names a key, not the row it
86/// removed. Those sizes are internal bookkeeping, so they live here, on the type
87/// the idempotency cache holds, rather than on the response type callers see.
88#[derive(Debug, Clone, Default)]
89pub(crate) struct CachedTransaction {
90    pub(crate) response: ExecuteTransactionResponse,
91    /// Per-statement `(table, image size)`.
92    pub(crate) replay_sizes: Vec<(String, usize)>,
93}
94
95#[derive(Debug, Clone, Default, Serialize)]
96pub struct ItemResponse {
97    #[serde(
98        rename = "Item",
99        skip_serializing_if = "Option::is_none",
100        serialize_with = "crate::set_order::item_in_hash_order"
101    )]
102    // Number sets in Java hash-set order here, as DynamoDB returns them;
103    // pinned by set_order's every_response_item_field test.
104    pub item: Option<Item>,
105}
106
107/// Run a PartiQL transaction.
108///
109/// Callers driving idempotency want `execute_cached`, which also hands back
110/// the sizes a replay is billed against.
111pub async fn execute<S: StorageBackend>(
112    storage: &S,
113    request: ExecuteTransactionRequest,
114) -> Result<ExecuteTransactionResponse> {
115    Ok(execute_cached(storage, request).await?.response)
116}
117
118/// The statement list as DynamoDB echoes it in a constraint message: each
119/// statement's Java `toString` (captured eu-west-2 and us-east-1, 2026-10-10;
120/// some hosts print an object hash per statement instead, and the form most
121/// give is the one followed). A statement's
122/// `ReturnValuesOnConditionCheckFailure` is echoed as sent, `null` when left
123/// out. Bounded by [`crate::actions::batch_write_item::ECHO_LIMIT`].
124fn statements_echo(statements: &[ParameterizedStatement]) -> String {
125    let mut out = crate::actions::batch_write_item::BoundedEcho::new();
126    let _ = write_statements(&mut out, statements);
127    out.finish()
128}
129
130fn write_statements(
131    out: &mut dyn std::fmt::Write,
132    statements: &[ParameterizedStatement],
133) -> std::fmt::Result {
134    out.write_str("[")?;
135    for (i, statement) in statements.iter().enumerate() {
136        if i > 0 {
137            out.write_str(", ")?;
138        }
139        write!(
140            out,
141            "ParameterizedStatement(statement={}, parameters=",
142            statement.statement
143        )?;
144        match statement.parameters {
145            Some(ref parameters) => {
146                out.write_str("[")?;
147                for (j, parameter) in parameters.iter().enumerate() {
148                    if j > 0 {
149                        out.write_str(", ")?;
150                    }
151                    crate::actions::batch_write_item::write_java_attribute_value(out, parameter)?;
152                }
153                out.write_str("]")?;
154            }
155            None => out.write_str("null")?,
156        }
157        out.write_str(", returnValuesOnConditionCheckFailure=")?;
158        out.write_str(
159            statement
160                .return_values_on_condition_check_failure
161                .as_deref()
162                .unwrap_or("null"),
163        )?;
164        out.write_str(")")?;
165    }
166    out.write_str("]")
167}
168
169/// The request's member violations, listed as DynamoDB lists them: the
170/// token, then each statement's ReturnValuesOnConditionCheckFailure, text and
171/// parameters by index, then ReturnConsumedCapacity (captured eu-west-2 and
172/// us-east-1, 2026-10-10, the regions identical). Any of them refuses the
173/// whole request before a statement is parsed or the idempotency cache is
174/// consulted.
175///
176/// The statement list's own length (empty, or more than 100) is listed with
177/// them, after the token and ahead of the statements, echoing the list: `[]`
178/// when empty, each statement's Java `toString` when too long (captured
179/// eu-west-2 and us-east-1, 2026-10-10; the empty list in six regions).
180fn member_violations(request: &ExecuteTransactionRequest) -> Vec<String> {
181    let mut violations = Vec::new();
182    violations.extend(crate::token_constraint_error(
183        request.client_request_token.as_deref(),
184    ));
185    if request.transact_statements.is_empty() {
186        violations.push(
187            "Value '[]' at 'transactStatements' failed to satisfy constraint: Member must \
188             have length greater than or equal to 1"
189                .to_string(),
190        );
191    }
192    if request.transact_statements.len() > 100 {
193        violations.push(format!(
194            "Value '{}' at 'transactStatements' failed to satisfy constraint: Member must \
195             have length less than or equal to 100",
196            statements_echo(&request.transact_statements)
197        ));
198    }
199    for (i, statement) in request.transact_statements.iter().enumerate() {
200        let member = format!("transactStatements.{}.member", i + 1);
201        violations.extend(
202            crate::validation::return_values_on_condition_check_failure_rejection(
203                statement
204                    .return_values_on_condition_check_failure
205                    .as_deref(),
206                &format!("{member}.returnValuesOnConditionCheckFailure"),
207            ),
208        );
209        violations.extend(crate::partiql::statement_length_violation(
210            &statement.statement,
211            &format!("{member}.statement"),
212        ));
213        violations.extend(crate::partiql::parameters_violation(
214            statement.parameters.as_deref(),
215            &format!("{member}.parameters"),
216        ));
217    }
218    violations.extend(crate::validation::return_consumed_capacity_rejection(
219        request.return_consumed_capacity.as_deref(),
220    ));
221    violations
222}
223
224/// Refuse a request whose members DynamoDB refuses, every violation listed in
225/// one error. The entry points call this ahead of the idempotency cache,
226/// because a same-token replay never reaches [`execute_cached`] and the token
227/// is keyed on the statements alone, so a replay may carry a mode or token
228/// the first call never did.
229pub(crate) fn check_members(request: &ExecuteTransactionRequest) -> Result<()> {
230    match crate::partiql::member_rejection(&member_violations(request)) {
231        Some(e) => Err(e),
232        None => Ok(()),
233    }
234}
235
236pub(crate) async fn execute_cached<S: StorageBackend>(
237    storage: &S,
238    request: ExecuteTransactionRequest,
239) -> Result<CachedTransaction> {
240    // The entry points check the members ahead of the idempotency cache too;
241    // this stays for callers that come here directly.
242    check_members(&request)?;
243
244    // The statement list's length is checked with the other members, so a
245    // list that reaches here holds 1 to 100 statements.
246    let statements = &request.transact_statements;
247
248    // A parameter holding an empty attribute name is a top-level rejection,
249    // before any statement is parsed (captured in ten regions).
250    if statements
251        .iter()
252        .any(|s| crate::empty_names::parameters_have_empty_name(s.parameters.as_deref()))
253    {
254        return Err(crate::empty_names::bare());
255    }
256
257    // Every statement's parameters are checked before any statement is read,
258    // used or not and ahead of a missing table, and a failure refuses the
259    // whole request with the bare message (captured eu-west-2 and us-east-1,
260    // 2026-10-09).
261    for stmt in statements {
262        for parameter in stmt.parameters.iter().flatten() {
263            partiql::executor::check_parameter_value(parameter)?;
264        }
265    }
266
267    // Parse all statements before executing any, to fail fast on syntax errors
268    let mut parsed = Vec::with_capacity(statements.len());
269    for (index, stmt) in statements.iter().enumerate() {
270        // A parameter nested too deep fails the whole transaction at the top
271        // level, before this statement is parsed, but after the statements
272        // ahead of it have been checked.
273        if crate::value_nesting::parameters_nest_too_deep(
274            stmt.parameters.as_deref().unwrap_or_default(),
275        ) {
276            return Err(crate::value_nesting::deep_parameter_error());
277        }
278        // A call in a SELECT list is rejected before parsing, with the
279        // statement's index in front of the bare message, not the
280        // wasn't-well-formed wrapper (captured eu-west-2 and us-east-1,
281        // 2026-10-09).
282        if let Some(msg) = partiql::parser::count_projection_rejection(&stmt.statement) {
283            return Err(DynoxideError::ValidationException(format!(
284                "Validation failed in TransactStatements[{index}]: {msg}"
285            )));
286        }
287        // A statement that does not parse, or carries a literal DynamoDB
288        // refuses, is reported with its index in front, as the RETURNING and
289        // index-read refusals below are (captured eu-west-2 and us-east-1,
290        // 2026-10-09).
291        let supplied = stmt.parameters.as_ref().map_or(0, Vec::len);
292        let ast = partiql::parser::parse_with_parameters(&stmt.statement, supplied).map_err(
293            |failure| {
294                let message = match failure {
295                    partiql::parser::ParseFailure::Statement(e) => e.into_message(),
296                    partiql::parser::ParseFailure::Literal(message) => message,
297                };
298                DynoxideError::ValidationException(format!(
299                    "Validation failed in TransactStatements[{index}]: {message}"
300                ))
301            },
302        )?;
303        // DynamoDB rejects a RETURNING clause on any member of a transaction with
304        // a top-level ValidationException, before applying any write. This is a
305        // plain validation failure, not a TransactionCanceledException.
306        if partiql::parser::returning_variant(&ast).is_some() {
307            return Err(DynoxideError::ValidationException(format!(
308                "Validation failed in TransactStatements[{index}]: RETURNING clause is not supported in ExecuteTransaction."
309            )));
310        }
311        // An index-qualified read is rejected outright inside a transaction, so
312        // there is no arm for it to be charged to. Rejected up front like the
313        // RETURNING case, not as a cancellation. Captured eu-west-2 2026-08-15.
314        if matches!(ast, partiql::parser::Statement::Select { .. })
315            && partiql::parser::index_name(&ast).is_some()
316        {
317            return Err(DynoxideError::ValidationException(format!(
318                "Validation failed in TransactStatements[{index}]: Reads on indices are not supported within transactions."
319            )));
320        }
321        let params = stmt.parameters.clone().unwrap_or_default();
322        parsed.push((ast, params));
323    }
324
325    // A transaction is all-read or all-write, and may not touch one item twice.
326    // AWS rejects both up front, before any statement runs.
327    let reads = parsed
328        .iter()
329        .filter(|(stmt, _)| matches!(stmt, partiql::parser::Statement::Select { .. }))
330        .count();
331    if reads > 0 && reads < parsed.len() {
332        return Err(DynoxideError::ValidationException(
333            "ExecuteTransaction API does not support both read and write operations in the same request."
334                .to_string(),
335        ));
336    }
337
338    // Duplicate detection resolves each statement's table, and the executor
339    // needs the same metadata and key schema a moment later, so they are kept
340    // here rather than resolved twice per statement.
341    let mut seen_targets = HashSet::new();
342    // Keyed by table rather than by statement: the metadata and key schema are
343    // per table, and only the key is per statement.
344    let mut tables: HashMap<String, ResolvedTable> = HashMap::new();
345    for (index, (stmt, params)) in parsed.iter().enumerate() {
346        let Some(name) = partiql::parser::table_name(stmt) else {
347            continue;
348        };
349        if !tables.contains_key(name)
350            && let Ok(resolved) = ResolvedTable::load(storage, name).await
351        {
352            tables.insert(name.to_string(), resolved);
353        }
354        // Absent when the table could not be read, which leaves the statement to
355        // fail on its own terms during execution rather than as a duplicate.
356        let Some(resolved) = tables.get(name) else {
357            continue;
358        };
359        // A key value no key can have in an UPDATE or DELETE refuses the
360        // request; one of the wrong key type cancels it when it runs
361        // (captured eu-west-2 and us-east-1, 2026-10-09).
362        if let Some(partiql::executor::KeyFault::NotAKeyType(message)) =
363            partiql::executor::write_key_fault(stmt, params, resolved)
364        {
365            return Err(DynoxideError::ValidationException(format!(
366                "Validation failed in TransactStatements[{index}]: {message}"
367            )));
368        }
369        // A SELECT is served as a GetItem, so it may name the primary key and
370        // nothing else, once overlapping key conditions have been ruled out
371        // (captured eu-west-2 and us-east-1, 2026-10-09).
372        if matches!(stmt, partiql::parser::Statement::Select { .. })
373            && let Err(e) = partiql::executor::reject_overlapping_select(stmt, params, resolved)
374        {
375            return Err(DynoxideError::ValidationException(format!(
376                "Validation failed in TransactStatements[{index}]: {e}"
377            )));
378        }
379        if matches!(stmt, partiql::parser::Statement::Select { .. })
380            && matches!(
381                partiql::executor::keyed_select(stmt, params, resolved),
382                partiql::executor::KeyedSelect::NotByPrimaryKey
383            )
384        {
385            return Err(DynoxideError::ValidationException(format!(
386                "Validation failed in TransactStatements[{index}]: Select statements within \
387                 ExecuteTransaction must specify the primary key in the where clause."
388            )));
389        }
390        if let Some(target) = partiql::executor::statement_target_in(stmt, params, resolved)
391            && !seen_targets.insert(target)
392        {
393            return Err(DynoxideError::ValidationException(
394                "Transaction request cannot include multiple operations on one item".to_string(),
395            ));
396        }
397    }
398
399    // All statements run inside one SQLite transaction (all-or-nothing).
400    let (responses, charges) = helpers::with_write_transaction(
401        storage,
402        execute_within_transaction(
403            storage,
404            statements,
405            &parsed,
406            &tables,
407            request.return_consumed_capacity.as_deref(),
408        ),
409    )
410    .await?;
411
412    // Transactional capacity, split by statement kind: an all-SELECT read set
413    // reports read capacity, any INSERT/UPDATE/DELETE makes it a write set. Both
414    // are sized on the items each statement touched, and the transactional
415    // factor reaches the base table arm only.
416    let mode = &request.return_consumed_capacity;
417    let consumed_capacity = if is_read_set(&parsed) {
418        crate::types::build_transactional_capacity(
419            &transactional_read_units(&replay_sizes(&charges)),
420            mode,
421            crate::types::transactional_read_capacity,
422        )
423    } else {
424        build_write_capacity(&charges, mode)
425    };
426
427    // A transaction of writes answers with an empty `Responses` list, not one
428    // empty entry per statement; a transaction of reads has one entry per
429    // statement (captured eu-west-2, us-east-1, eu-west-1 and us-west-2,
430    // 2026-10-09).
431    let responses = if is_read_set(&parsed) {
432        responses
433    } else {
434        Vec::new()
435    };
436    Ok(CachedTransaction {
437        response: ExecuteTransactionResponse {
438            responses: Some(responses),
439            consumed_capacity,
440        },
441        replay_sizes: replay_sizes(&charges),
442    })
443}
444
445/// What one statement contributed to the transaction's capacity.
446enum StatementCharge {
447    /// A `SELECT`, charged on the rows it walked at read granularity.
448    Read { table_name: String, size: usize },
449    /// A write, charged on its images and its per-index units.
450    Write(WriteCapacity),
451}
452
453impl StatementCharge {
454    fn table_name(&self) -> &str {
455        match self {
456            Self::Read { table_name, .. } => table_name,
457            Self::Write(capacity) => &capacity.table_name,
458        }
459    }
460
461    /// The image size this statement is charged on, for a replay.
462    fn size(&self) -> usize {
463        match self {
464            Self::Read { size, .. } => *size,
465            Self::Write(capacity) => capacity
466                .old_size
467                .unwrap_or(0)
468                .max(capacity.new_size.unwrap_or(0)),
469        }
470    }
471
472    /// The write record for this statement.
473    ///
474    /// A mixed set is rejected before execution, so a `Read` charge never
475    /// reaches the write path in practice. It maps to a table-only record
476    /// rather than panicking, so a future relaxation of that check degrades to
477    /// an over-estimate instead of a wrong shape.
478    fn as_write(&self) -> WriteCapacity {
479        match self {
480            Self::Write(capacity) => capacity.clone(),
481            Self::Read { table_name, size } => WriteCapacity::new(
482                table_name,
483                None,
484                Some(*size),
485                HashMap::new(),
486                HashMap::new(),
487            ),
488        }
489    }
490}
491
492/// Fold the per-statement records into one `ConsumedCapacity` per table.
493fn build_write_capacity(
494    charges: &[StatementCharge],
495    mode: &Option<String>,
496) -> Option<Vec<crate::types::ConsumedCapacity>> {
497    // Checked before the clone, not just inside the shared builder. Most calls
498    // ask for no capacity, and `as_write` clones every record and its index
499    // maps before the builder could discard the lot.
500    if !matches!(mode.as_deref(), Some("TOTAL") | Some("INDEXES")) {
501        return None;
502    }
503
504    let records: Vec<WriteCapacity> = charges.iter().map(StatementCharge::as_write).collect();
505    let by_table = aggregate_by_table(&records, crate::types::TRANSACTIONAL_CAPACITY_FACTOR);
506    per_table_capacity(
507        &by_table,
508        mode,
509        crate::types::transactional_write_capacity_with_indexes,
510    )
511}
512
513/// The image size each statement was charged on, kept for a same-token replay.
514fn replay_sizes(charges: &[StatementCharge]) -> Vec<(String, usize)> {
515    charges
516        .iter()
517        .map(|charge| (charge.table_name().to_string(), charge.size()))
518        .collect()
519}
520
521/// A transaction is a read set only when every statement is a `SELECT`; any
522/// `INSERT`/`UPDATE`/`DELETE` makes it a write set. AWS requires a transaction
523/// to be all-read or all-write and rejects a mixed set before capacity is
524/// computed, but dynoxide does not enforce that, so a mixed set is classified
525/// here as a write set. Revisit the predicate if condition-only checks (which
526/// AWS counts in the write set) are ever parsed.
527fn is_read_set(parsed: &[(partiql::parser::Statement, Vec<AttributeValue>)]) -> bool {
528    parsed
529        .iter()
530        .all(|(stmt, _)| matches!(stmt, partiql::parser::Statement::Select { .. }))
531}
532
533/// Build the response for a same-token idempotent replay. The statements are
534/// identical to the first call (the idempotency hash matched), so `Responses`
535/// carry over from the cached first call and capacity is reported as a
536/// transactional READ, honouring the replay request's own
537/// `ReturnConsumedCapacity` mode (the original call's mode does not carry over).
538/// The statements are re-parsed to recover per-table units; they parsed
539/// successfully on the first call, so an unexpected parse error just drops that
540/// statement from the estimate rather than failing the replay.
541pub(crate) fn replay_response(
542    cached: &CachedTransaction,
543    mode: &Option<String>,
544) -> CachedTransaction {
545    CachedTransaction {
546        response: ExecuteTransactionResponse {
547            responses: cached.response.responses.clone(),
548            consumed_capacity: crate::types::build_transactional_capacity(
549                &transactional_read_units(&cached.replay_sizes),
550                mode,
551                crate::types::transactional_read_capacity,
552            ),
553        },
554        replay_sizes: cached.replay_sizes.clone(),
555    }
556}
557
558/// Run every statement, returning the responses and what each contributed to
559/// `ConsumedCapacity`. The charges are only meaningful when the whole
560/// transaction commits; a cancellation returns an error and reports nothing.
561///
562/// A cancellation lists every statement that failed, as DynamoDB's does
563/// (captured eu-west-2 and us-east-1, 2026-10-10: `[ConditionalCheckFailed,
564/// None, ConditionalCheckFailed]`, `[ConditionalCheckFailed, DuplicateItem]`),
565/// so every statement runs even after one has failed. The SQLite transaction
566/// is rolled back on the error, so nothing any of them wrote is kept.
567///
568/// A failure of the storage itself (a full disk, an I/O error) is the
569/// exception: it stops the transaction there and is returned as it is. SQLite
570/// may already have rolled the whole transaction back for it, and a statement
571/// run after that would commit on its own.
572async fn execute_within_transaction<S: StorageBackend>(
573    storage: &S,
574    statements: &[ParameterizedStatement],
575    parsed: &[(partiql::parser::Statement, Vec<AttributeValue>)],
576    tables: &HashMap<String, ResolvedTable>,
577    capacity_mode: Option<&str>,
578) -> Result<(Vec<ItemResponse>, Vec<StatementCharge>)> {
579    let resolved_for = |stmt: &partiql::parser::Statement| {
580        partiql::parser::table_name(stmt).and_then(|t| tables.get(t))
581    };
582
583    // A SELECT key of the wrong type, or of a type no key can have, and an
584    // UPDATE or DELETE key of the wrong type, cancel the transaction with
585    // GetItem's reason (captured eu-west-2 and us-east-1, 2026-10-09). They
586    // are checked of every statement before any runs: when one has such a
587    // key, every other statement is reported `None`, a statement whose
588    // condition is false included (captured eu-west-2 and us-east-1,
589    // 2026-10-10; inferred for the other failure kinds).
590    let wrong_key_type: Vec<bool> = parsed
591        .iter()
592        .map(|(stmt, params)| {
593            resolved_for(stmt).is_some_and(|resolved| match stmt {
594                partiql::parser::Statement::Select { .. } => matches!(
595                    partiql::executor::keyed_select(stmt, params, resolved),
596                    partiql::executor::KeyedSelect::KeyType(_)
597                ),
598                _ => partiql::executor::write_key_fault(stmt, params, resolved).is_some(),
599            })
600        })
601        .collect();
602    if wrong_key_type.contains(&true) {
603        return Err(cancelled(
604            wrong_key_type
605                .iter()
606                .map(|&wrong| {
607                    if wrong {
608                        CancellationReason {
609                            code: "ValidationError".to_string(),
610                            message: Some(
611                                "The provided key element does not match the schema".to_string(),
612                            ),
613                            item: None,
614                        }
615                    } else {
616                        no_reason()
617                    }
618                })
619                .collect(),
620        ));
621    }
622
623    let mut responses = Vec::with_capacity(parsed.len());
624    let mut charges: Vec<StatementCharge> = Vec::with_capacity(parsed.len());
625    let mut cancellation_reasons: Vec<CancellationReason> = Vec::with_capacity(parsed.len());
626    let mut failed = false;
627
628    // `parsed` holds each statement of `statements`, in order.
629    for ((stmt, params), written) in parsed.iter().zip(statements) {
630        let resolved = resolved_for(stmt);
631        // Whether the statement asked for the item a failed condition was
632        // evaluated against.
633        let old_on_failure =
634            written.return_values_on_condition_check_failure.as_deref() == Some("ALL_OLD");
635        // An INSERT key of the wrong type cancels the transaction with
636        // PutItem's reason (captured eu-west-2 and us-east-1, 2026-10-09).
637        let insert_keys = match resolved {
638            Some(resolved) => {
639                partiql::executor::check_transaction_insert_keys(stmt, params, resolved)
640            }
641            None => Ok(()),
642        };
643        let outcome = match insert_keys {
644            Err(e) => Err(e),
645            Ok(()) => {
646                partiql::executor::execute_written_page(
647                    storage,
648                    &written.statement,
649                    stmt,
650                    params,
651                    None,
652                    None,
653                    false,
654                    capacity_mode,
655                    resolved,
656                    old_on_failure,
657                )
658                .await
659            }
660        };
661        match outcome {
662            Ok(page) => {
663                // A DELETE with a condition, of an item that isn't there, does
664                // nothing when the transaction commits, and is reported as a
665                // failed condition, with no item, when it is cancelled
666                // (captured eu-west-2 and us-east-1, 2026-10-10).
667                let missed_condition = matches!(stmt, partiql::parser::Statement::Delete { .. })
668                    && page
669                        .capacity
670                        .as_ref()
671                        .is_some_and(|capacity| capacity.old_size.is_none())
672                    && resolved.is_some_and(|resolved| {
673                        partiql::executor::delete_has_condition(stmt, resolved)
674                    });
675                let table_name = partiql::parser::table_name(stmt).unwrap_or_default();
676                charges.push(match page.capacity {
677                    Some(capacity) => StatementCharge::Write(capacity),
678                    None => StatementCharge::Read {
679                        table_name: table_name.to_string(),
680                        size: page.size,
681                    },
682                });
683                let item = page.items.and_then(|items| items.into_iter().next());
684                responses.push(ItemResponse { item });
685                cancellation_reasons.push(if missed_condition {
686                    CancellationReason {
687                        code: "ConditionalCheckFailed".to_string(),
688                        message: Some("The conditional request failed".to_string()),
689                        item: None,
690                    }
691                } else {
692                    no_reason()
693                });
694            }
695            Err(e) => {
696                let Some(code) = outcome_code(&e) else {
697                    return Err(e);
698                };
699                failed = true;
700                responses.push(ItemResponse { item: None });
701                cancellation_reasons.push(failure_reason(code, e));
702            }
703        }
704    }
705
706    if failed {
707        return Err(cancelled(cancellation_reasons));
708    }
709    Ok((responses, charges))
710}
711
712/// The reason a statement that ran without failing is given.
713fn no_reason() -> CancellationReason {
714    CancellationReason {
715        code: "None".to_string(),
716        message: None,
717        item: None,
718    }
719}
720
721/// The cancellation code for a statement that failed, or `None` when the
722/// error is not the statement's outcome but a failure of the storage under it,
723/// which stops the transaction instead.
724fn outcome_code(e: &DynoxideError) -> Option<&'static str> {
725    match e {
726        DynoxideError::ConditionalCheckFailedException(..) => Some("ConditionalCheckFailed"),
727        DynoxideError::DuplicateItemException(_) => Some("DuplicateItem"),
728        // A statement against a table that does not exist cancels the
729        // transaction with this reason (captured eu-west-2 and us-east-1,
730        // 2026-10-09).
731        DynoxideError::ResourceNotFoundException(_) => Some("ResourceNotFound"),
732        // Group KeyEmptyValueValidation with ValidationException so an
733        // empty-value key keeps the "ValidationError" reason (#95).
734        DynoxideError::ValidationException(_) | DynoxideError::KeyEmptyValueValidation(_) => {
735            Some("ValidationError")
736        }
737        _ => None,
738    }
739}
740
741/// The reason a failed statement is given, under the code
742/// [`outcome_code`] found for it, with the item its condition was evaluated
743/// against when it asked for it.
744fn failure_reason(code: &str, e: DynoxideError) -> CancellationReason {
745    let message = Some(e.to_string());
746    let item = match e {
747        DynoxideError::ConditionalCheckFailedException(_, item) => item,
748        _ => None,
749    };
750    CancellationReason {
751        code: code.to_string(),
752        message,
753        item,
754    }
755}
756
757/// The cancellation for these reasons, its message listing their codes.
758fn cancelled(reasons: Vec<CancellationReason>) -> DynoxideError {
759    let codes: Vec<&str> = reasons.iter().map(|r| r.code.as_str()).collect();
760    let message = format!(
761        "Transaction cancelled, please refer cancellation reasons for specific reasons [{}]",
762        codes.join(", ")
763    );
764    DynoxideError::TransactionCanceledException(message, reasons)
765}