dynoxide/actions/
execute_transaction.rs

1use crate::actions::helpers;
2use crate::errors::{CancellationReason, DynoxideError, Result};
3use crate::partiql;
4use crate::storage_backend::StorageBackend;
5use crate::types::{AttributeValue, Item};
6use serde::{Deserialize, Serialize};
7
8#[derive(Debug, Clone, Default, Deserialize)]
9pub struct ExecuteTransactionRequest {
10    #[serde(rename = "TransactStatements")]
11    pub transact_statements: Vec<ParameterizedStatement>,
12    #[serde(rename = "ClientRequestToken", default)]
13    pub client_request_token: Option<String>,
14    #[serde(rename = "ReturnConsumedCapacity", default)]
15    pub return_consumed_capacity: Option<String>,
16}
17
18// `Serialize` backs the idempotency request hash (the statements and their
19// parameters are serialised via `serde_json`), so a same-token call differing
20// only in `ReturnConsumedCapacity` replays rather than mismatches.
21#[derive(Debug, Clone, Default, Deserialize, Serialize)]
22pub struct ParameterizedStatement {
23    #[serde(rename = "Statement")]
24    pub statement: String,
25    #[serde(rename = "Parameters", default)]
26    pub parameters: Option<Vec<AttributeValue>>,
27}
28
29// `Clone` so the idempotency cache can store the first-call response and clone
30// its `Responses` for the replay.
31#[derive(Debug, Clone, Default, Serialize)]
32pub struct ExecuteTransactionResponse {
33    #[serde(rename = "Responses", skip_serializing_if = "Option::is_none")]
34    pub responses: Option<Vec<ItemResponse>>,
35    #[serde(rename = "ConsumedCapacity", skip_serializing_if = "Option::is_none")]
36    pub consumed_capacity: Option<Vec<crate::types::ConsumedCapacity>>,
37}
38
39#[derive(Debug, Clone, Default, Serialize)]
40pub struct ItemResponse {
41    #[serde(rename = "Item", skip_serializing_if = "Option::is_none")]
42    pub item: Option<Item>,
43}
44
45pub async fn execute<S: StorageBackend>(
46    storage: &S,
47    request: ExecuteTransactionRequest,
48) -> Result<ExecuteTransactionResponse> {
49    let statements = &request.transact_statements;
50
51    // Validate: must have between 1 and 100 statements
52    if statements.is_empty() {
53        return Err(DynoxideError::ValidationException(
54            "1 validation error detected: Value at 'transactStatements' failed to satisfy constraint: Member must have length greater than or equal to 1".to_string(),
55        ));
56    }
57    if statements.len() > 100 {
58        return Err(DynoxideError::ValidationException(
59            "Member must have length less than or equal to 100".to_string(),
60        ));
61    }
62
63    // Parse all statements before executing any, to fail fast on syntax errors
64    let mut parsed = Vec::with_capacity(statements.len());
65    for (index, stmt) in statements.iter().enumerate() {
66        // A COUNT projection is rejected before parsing, at the same top level
67        // as a parse error but with the bare message captured on
68        // ExecuteStatement, not the wasn't-well-formed wrapper.
69        if let Some(msg) = partiql::parser::count_projection_rejection(&stmt.statement) {
70            return Err(DynoxideError::ValidationException(msg));
71        }
72        let ast = partiql::parser::parse(&stmt.statement).map_err(|e| {
73            DynoxideError::ValidationException(format!(
74                "Statement wasn't well formed, can't be processed: {e}"
75            ))
76        })?;
77        // DynamoDB rejects a RETURNING clause on any member of a transaction with
78        // a top-level ValidationException, before applying any write. This is a
79        // plain validation failure, not a TransactionCanceledException.
80        if partiql::parser::returning_variant(&ast).is_some() {
81            return Err(DynoxideError::ValidationException(format!(
82                "Validation failed in TransactStatements[{index}]: RETURNING clause is not supported in ExecuteTransaction."
83            )));
84        }
85        let params = stmt.parameters.clone().unwrap_or_default();
86        parsed.push((ast, params));
87    }
88
89    // All statements run inside one SQLite transaction (all-or-nothing).
90    let responses =
91        helpers::with_write_transaction(storage, execute_within_transaction(storage, &parsed))
92            .await?;
93
94    // Transactional capacity, split by statement kind: an all-SELECT read set
95    // reports read capacity, any INSERT/UPDATE/DELETE makes it a write set.
96    //
97    // TODO: capacity is charged a flat per-statement transactional unit, not an
98    // item-size computation, so it under-counts PartiQL items above 1KB (writes
99    // round at 1KB, reads at 4KB). Correct for the small items conformance pins;
100    // size-accurate rounding for large PartiQL statements is a tracked follow-up.
101    let builder = if is_read_set(&parsed) {
102        crate::types::transactional_read_capacity
103    } else {
104        crate::types::transactional_write_capacity
105    };
106    let consumed_capacity = crate::types::build_transactional_capacity(
107        &statement_table_units(parsed.iter().map(|(stmt, _)| stmt)),
108        &request.return_consumed_capacity,
109        builder,
110    );
111
112    Ok(ExecuteTransactionResponse {
113        responses: Some(responses),
114        consumed_capacity,
115    })
116}
117
118/// A transaction is a read set only when every statement is a `SELECT`; any
119/// `INSERT`/`UPDATE`/`DELETE` makes it a write set. AWS requires a transaction
120/// to be all-read or all-write and rejects a mixed set before capacity is
121/// computed, but dynoxide does not enforce that, so a mixed set is classified
122/// here as a write set. Revisit the predicate if condition-only checks (which
123/// AWS counts in the write set) are ever parsed.
124fn is_read_set(parsed: &[(partiql::parser::Statement, Vec<AttributeValue>)]) -> bool {
125    parsed
126        .iter()
127        .all(|(stmt, _)| matches!(stmt, partiql::parser::Statement::Select { .. }))
128}
129
130/// Per-table transactional units for a set of parsed statements. Each statement
131/// costs the per-statement base (1 unit) doubled by the transactional factor,
132/// summed by target table (matching `TransactWriteItems`, which doubles per
133/// item). Item-size rounding is not applied here (see the TODO in `execute`).
134fn statement_table_units<'a>(
135    statements: impl Iterator<Item = &'a partiql::parser::Statement>,
136) -> std::collections::HashMap<String, f64> {
137    let mut table_units: std::collections::HashMap<String, f64> = std::collections::HashMap::new();
138    for stmt in statements {
139        if let Some(tbl) = partiql::parser::table_name(stmt) {
140            *table_units.entry(tbl.to_string()).or_default() +=
141                crate::types::TRANSACTIONAL_CAPACITY_FACTOR;
142        }
143    }
144    table_units
145}
146
147/// Build the response for a same-token idempotent replay. The statements are
148/// identical to the first call (the idempotency hash matched), so `Responses`
149/// carry over from the cached first call and capacity is reported as a
150/// transactional READ, honouring the replay request's own
151/// `ReturnConsumedCapacity` mode (the original call's mode does not carry over).
152/// The statements are re-parsed to recover per-table units; they parsed
153/// successfully on the first call, so an unexpected parse error just drops that
154/// statement from the estimate rather than failing the replay.
155pub(crate) fn replay_response(
156    statements: &[ParameterizedStatement],
157    mode: &Option<String>,
158    cached_responses: Option<Vec<ItemResponse>>,
159) -> ExecuteTransactionResponse {
160    let parsed: Vec<partiql::parser::Statement> = statements
161        .iter()
162        .filter_map(|s| partiql::parser::parse(&s.statement).ok())
163        .collect();
164    ExecuteTransactionResponse {
165        responses: cached_responses,
166        consumed_capacity: crate::types::build_transactional_capacity(
167            &statement_table_units(parsed.iter()),
168            mode,
169            crate::types::transactional_read_capacity,
170        ),
171    }
172}
173
174async fn execute_within_transaction<S: StorageBackend>(
175    storage: &S,
176    parsed: &[(partiql::parser::Statement, Vec<AttributeValue>)],
177) -> Result<Vec<ItemResponse>> {
178    let mut responses = Vec::with_capacity(parsed.len());
179    let mut cancellation_reasons: Vec<CancellationReason> = Vec::with_capacity(parsed.len());
180
181    for (stmt, params) in parsed {
182        match partiql::executor::execute(storage, stmt, params, None).await {
183            Ok(result) => {
184                let item = result.and_then(|items| items.into_iter().next());
185                responses.push(ItemResponse { item });
186                cancellation_reasons.push(CancellationReason {
187                    code: "None".to_string(),
188                    message: None,
189                    item: None,
190                });
191            }
192            Err(e) => {
193                // Record the failure reason
194                let message = Some(e.to_string());
195                let (code, item) = match e {
196                    DynoxideError::ConditionalCheckFailedException(_, item) => {
197                        ("ConditionalCheckFailed".to_string(), item)
198                    }
199                    DynoxideError::DuplicateItemException(_) => ("DuplicateItem".to_string(), None),
200                    // Group KeyEmptyValueValidation with ValidationException so an empty-value
201                    // key keeps the "ValidationError" reason instead of falling through to
202                    // InternalError (#95).
203                    DynoxideError::ValidationException(_)
204                    | DynoxideError::KeyEmptyValueValidation(_) => {
205                        ("ValidationError".to_string(), None)
206                    }
207                    _ => ("InternalError".to_string(), None),
208                };
209                responses.push(ItemResponse { item: None });
210                cancellation_reasons.push(CancellationReason {
211                    code,
212                    message,
213                    item,
214                });
215
216                // Fill remaining slots with None and stop — don't execute
217                // statements that will be rolled back.
218                for _ in responses.len()..parsed.len() {
219                    responses.push(ItemResponse { item: None });
220                    cancellation_reasons.push(CancellationReason {
221                        code: "None".to_string(),
222                        message: None,
223                        item: None,
224                    });
225                }
226
227                let codes: Vec<&str> = cancellation_reasons
228                    .iter()
229                    .map(|r| r.code.as_str())
230                    .collect();
231                let message = format!(
232                    "Transaction cancelled, please refer cancellation reasons for specific reasons [{}]",
233                    codes.join(", ")
234                );
235                return Err(DynoxideError::TransactionCanceledException(
236                    message,
237                    cancellation_reasons,
238                ));
239            }
240        }
241    }
242
243    Ok(responses)
244}
dynoxide/actions/execute_transaction.rs

dynoxide/actions/
execute_transaction.rs
