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

dynoxide/actions/
execute_transaction.rs
