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        let ast = partiql::parser::parse(&stmt.statement).map_err(|e| {
67            DynoxideError::ValidationException(format!(
68                "Statement wasn't well formed, can't be processed: {e}"
69            ))
70        })?;
71        // DynamoDB rejects a RETURNING clause on any member of a transaction with
72        // a top-level ValidationException, before applying any write. This is a
73        // plain validation failure, not a TransactionCanceledException.
74        if partiql::parser::returning_variant(&ast).is_some() {
75            return Err(DynoxideError::ValidationException(format!(
76                "Validation failed in TransactStatements[{index}]: RETURNING clause is not supported in ExecuteTransaction."
77            )));
78        }
79        let params = stmt.parameters.clone().unwrap_or_default();
80        parsed.push((ast, params));
81    }
82
83    // All statements run inside one SQLite transaction (all-or-nothing).
84    let responses =
85        helpers::with_write_transaction(storage, execute_within_transaction(storage, &parsed))
86            .await?;
87
88    // Transactional capacity, split by statement kind: an all-SELECT read set
89    // reports read capacity, any INSERT/UPDATE/DELETE makes it a write set.
90    //
91    // TODO: capacity is charged a flat per-statement transactional unit, not an
92    // item-size computation, so it under-counts PartiQL items above 1KB (writes
93    // round at 1KB, reads at 4KB). Correct for the small items conformance pins;
94    // size-accurate rounding for large PartiQL statements is a tracked follow-up.
95    let builder = if is_read_set(&parsed) {
96        crate::types::transactional_read_capacity
97    } else {
98        crate::types::transactional_write_capacity
99    };
100    let consumed_capacity = crate::types::build_transactional_capacity(
101        &statement_table_units(parsed.iter().map(|(stmt, _)| stmt)),
102        &request.return_consumed_capacity,
103        builder,
104    );
105
106    Ok(ExecuteTransactionResponse {
107        responses: Some(responses),
108        consumed_capacity,
109    })
110}
111
112/// A transaction is a read set only when every statement is a `SELECT`; any
113/// `INSERT`/`UPDATE`/`DELETE` makes it a write set. AWS requires a transaction
114/// to be all-read or all-write and rejects a mixed set before capacity is
115/// computed, but dynoxide does not enforce that, so a mixed set is classified
116/// here as a write set. Revisit the predicate if condition-only checks (which
117/// AWS counts in the write set) are ever parsed.
118fn is_read_set(parsed: &[(partiql::parser::Statement, Vec<AttributeValue>)]) -> bool {
119    parsed
120        .iter()
121        .all(|(stmt, _)| matches!(stmt, partiql::parser::Statement::Select { .. }))
122}
123
124/// Per-table transactional units for a set of parsed statements. Each statement
125/// costs the per-statement base (1 unit) doubled by the transactional factor,
126/// summed by target table (matching `TransactWriteItems`, which doubles per
127/// item). Item-size rounding is not applied here (see the TODO in `execute`).
128fn statement_table_units<'a>(
129    statements: impl Iterator<Item = &'a partiql::parser::Statement>,
130) -> std::collections::HashMap<String, f64> {
131    let mut table_units: std::collections::HashMap<String, f64> = std::collections::HashMap::new();
132    for stmt in statements {
133        if let Some(tbl) = partiql::parser::table_name(stmt) {
134            *table_units.entry(tbl.to_string()).or_default() +=
135                crate::types::TRANSACTIONAL_CAPACITY_FACTOR;
136        }
137    }
138    table_units
139}
140
141/// Build the response for a same-token idempotent replay. The statements are
142/// identical to the first call (the idempotency hash matched), so `Responses`
143/// carry over from the cached first call and capacity is reported as a
144/// transactional READ, honouring the replay request's own
145/// `ReturnConsumedCapacity` mode (the original call's mode does not carry over).
146/// The statements are re-parsed to recover per-table units; they parsed
147/// successfully on the first call, so an unexpected parse error just drops that
148/// statement from the estimate rather than failing the replay.
149pub(crate) fn replay_response(
150    statements: &[ParameterizedStatement],
151    mode: &Option<String>,
152    cached_responses: Option<Vec<ItemResponse>>,
153) -> ExecuteTransactionResponse {
154    let parsed: Vec<partiql::parser::Statement> = statements
155        .iter()
156        .filter_map(|s| partiql::parser::parse(&s.statement).ok())
157        .collect();
158    ExecuteTransactionResponse {
159        responses: cached_responses,
160        consumed_capacity: crate::types::build_transactional_capacity(
161            &statement_table_units(parsed.iter()),
162            mode,
163            crate::types::transactional_read_capacity,
164        ),
165    }
166}
167
168async fn execute_within_transaction<S: StorageBackend>(
169    storage: &S,
170    parsed: &[(partiql::parser::Statement, Vec<AttributeValue>)],
171) -> Result<Vec<ItemResponse>> {
172    let mut responses = Vec::with_capacity(parsed.len());
173    let mut cancellation_reasons: Vec<CancellationReason> = Vec::with_capacity(parsed.len());
174
175    for (stmt, params) in parsed {
176        match partiql::executor::execute(storage, stmt, params, None).await {
177            Ok(result) => {
178                let item = result.and_then(|items| items.into_iter().next());
179                responses.push(ItemResponse { item });
180                cancellation_reasons.push(CancellationReason {
181                    code: "None".to_string(),
182                    message: None,
183                    item: None,
184                });
185            }
186            Err(e) => {
187                // Record the failure reason
188                let message = Some(e.to_string());
189                let (code, item) = match e {
190                    DynoxideError::ConditionalCheckFailedException(_, item) => {
191                        ("ConditionalCheckFailed".to_string(), item)
192                    }
193                    DynoxideError::DuplicateItemException(_) => ("DuplicateItem".to_string(), None),
194                    // Group KeyEmptyValueValidation with ValidationException so an empty-value
195                    // key keeps the "ValidationError" reason instead of falling through to
196                    // InternalError (#95).
197                    DynoxideError::ValidationException(_)
198                    | DynoxideError::KeyEmptyValueValidation(_) => {
199                        ("ValidationError".to_string(), None)
200                    }
201                    _ => ("InternalError".to_string(), None),
202                };
203                responses.push(ItemResponse { item: None });
204                cancellation_reasons.push(CancellationReason {
205                    code,
206                    message,
207                    item,
208                });
209
210                // Fill remaining slots with None and stop — don't execute
211                // statements that will be rolled back.
212                for _ in responses.len()..parsed.len() {
213                    responses.push(ItemResponse { item: None });
214                    cancellation_reasons.push(CancellationReason {
215                        code: "None".to_string(),
216                        message: None,
217                        item: None,
218                    });
219                }
220
221                let codes: Vec<&str> = cancellation_reasons
222                    .iter()
223                    .map(|r| r.code.as_str())
224                    .collect();
225                let message = format!(
226                    "Transaction cancelled, please refer cancellation reasons for specific reasons [{}]",
227                    codes.join(", ")
228                );
229                return Err(DynoxideError::TransactionCanceledException(
230                    message,
231                    cancellation_reasons,
232                ));
233            }
234        }
235    }
236
237    Ok(responses)
238}
dynoxide/actions/execute_transaction.rs

dynoxide/actions/
execute_transaction.rs
