dynoxide/actions/
execute_statement.rs

1use crate::errors::{DynoxideError, Result};
2use crate::partiql;
3use crate::storage_backend::StorageBackend;
4use crate::types::{AttributeValue, Item};
5use serde::{Deserialize, Serialize};
6
7#[derive(Debug, Default, Deserialize)]
8pub struct ExecuteStatementRequest {
9    #[serde(rename = "Statement")]
10    pub statement: String,
11    #[serde(rename = "Parameters", default)]
12    pub parameters: Option<Vec<AttributeValue>>,
13    #[serde(rename = "Limit", default)]
14    pub limit: Option<usize>,
15    #[serde(rename = "NextToken", default)]
16    pub next_token: Option<String>,
17    /// Does not change which rows come back, because every read against SQLite
18    /// is already strongly consistent. It does change two things: the rate the
19    /// read is charged at, and whether a select qualified by a GSI is rejected
20    /// at all.
21    #[serde(rename = "ConsistentRead", default)]
22    pub consistent_read: Option<bool>,
23    #[serde(rename = "ReturnConsumedCapacity", default)]
24    pub return_consumed_capacity: Option<String>,
25}
26
27#[derive(Debug, Default, Serialize)]
28pub struct ExecuteStatementResponse {
29    #[serde(rename = "Items", skip_serializing_if = "Option::is_none")]
30    pub items: Option<Vec<Item>>,
31    #[serde(rename = "NextToken", skip_serializing_if = "Option::is_none")]
32    pub next_token: Option<String>,
33    #[serde(rename = "ConsumedCapacity", skip_serializing_if = "Option::is_none")]
34    pub consumed_capacity: Option<crate::types::ConsumedCapacity>,
35}
36
37pub async fn execute<S: StorageBackend>(
38    storage: &S,
39    request: ExecuteStatementRequest,
40) -> Result<ExecuteStatementResponse> {
41    // Limit is checked before the statement is parsed. A zero limit would
42    // otherwise read nothing and mint no token, silently ending a paginated
43    // walk. The wording follows Scan's shape (value kept, lowercase 'limit'),
44    // not Query's, matching what ExecuteStatement itself returns (captured
45    // eu-west-2, 2026-07-29).
46    if request.limit == Some(0) {
47        return Err(DynoxideError::ValidationException(
48            crate::validation::envelope_message(
49                "Value '0' at 'limit' failed to satisfy constraint: \
50                 Member must have value greater than or equal to 1",
51            ),
52        ));
53    }
54
55    if let Some(msg) = crate::validation::return_consumed_capacity_rejection(
56        request.return_consumed_capacity.as_deref(),
57    ) {
58        return Err(DynoxideError::ValidationException(
59            crate::validation::envelope_message(&msg),
60        ));
61    }
62
63    // A COUNT projection is rejected before parsing, with DynamoDB's bare
64    // message rather than the wasn't-well-formed wrapper the parse errors
65    // below carry. ExecuteStatement is the captured surface for this shape.
66    if let Some(msg) = partiql::parser::count_projection_rejection(&request.statement) {
67        return Err(DynoxideError::ValidationException(msg));
68    }
69
70    // The parser says which envelope its rejection takes: a malformed statement
71    // gets DynamoDB's "wasn't well formed" wrapper, while a rejection DynamoDB
72    // reports on its own terms is passed through bare.
73    let stmt = partiql::parser::parse(&request.statement)
74        .map_err(|e| DynoxideError::ValidationException(e.into_message()))?;
75
76    let params = request.parameters.unwrap_or_default();
77    let run = partiql::executor::execute_page(
78        storage,
79        &stmt,
80        &params,
81        request.limit,
82        request.next_token.as_deref(),
83        request.consistent_read.unwrap_or(false),
84        request.return_consumed_capacity.as_deref(),
85        // A single statement has no preparation pass to share a resolution
86        // with, so the executor resolves the table itself, once.
87        None,
88    );
89    // Base write, index fan-out and stream record are one atomic unit, as they
90    // are on every other write path. Without it a fan-out that failed part way
91    // left the base row committed and the indexes disagreeing with it. A read
92    // opens no transaction: it has nothing to roll back.
93    let page = if matches!(stmt, partiql::parser::Statement::Select { .. }) {
94        run.await?
95    } else {
96        super::helpers::with_write_transaction(storage, run).await?
97    };
98    let partiql::executor::StatementPage {
99        items,
100        size,
101        capacity,
102        next_token,
103        read_index,
104        base_read_units,
105    } = page;
106
107    // ConsumedCapacity is returned whenever ReturnConsumedCapacity is requested,
108    // unlike some emulators that omit it. A SELECT is charged read units (an
109    // eventually consistent read unless ConsistentRead is set) against the rows
110    // it walked, which is before its WHERE clause and its projection narrow
111    // them. A write is charged the base table arm plus a per-index arm,
112    // with no transactional factor: a capture against real DynamoDB reports a
113    // PartiQL INSERT of an item in two indexes as total 3, table 1, one unit per
114    // index, exactly as the equivalent `PutItem`.
115    let consumed_capacity = partiql::parser::table_name(&stmt).and_then(|table| {
116        let Some(capacity) = capacity else {
117            let units = crate::types::read_capacity_units_with_consistency(
118                size,
119                request.consistent_read.unwrap_or(false),
120            );
121            // A read served from an index is charged against that index's arm
122            // with the table arm at zero, the same shape Query and Scan already
123            // report. Captured eu-west-2 2026-08-15: a keyed GSI select is
124            // total 0.5, table 0, gsi 0.5, where dynoxide charged the table.
125            return match read_index {
126                Some(index) if index.is_lsi => {
127                    // A reach-back read the base item for each row, and those
128                    // land on the table arm, leaving the index arm to cover the
129                    // index read. Each one is charged on its own bytes at the
130                    // request's consistency and the charges are summed, which
131                    // the executor works out because only it sees the sizes.
132                    // Captured eu-west-2 2026-08-15: three small rows served
133                    // this way reported total 2, table 1.5, lsi 0.5. Captured
134                    // again 2026-08-17 across item sizes: the same three rows at
135                    // ~9KB apiece report table 4.5, and 9 under ConsistentRead.
136                    let lsi_units = std::collections::HashMap::from([(index.name, units)]);
137                    crate::types::consumed_capacity_with_secondary_indexes(
138                        table,
139                        base_read_units,
140                        &std::collections::HashMap::new(),
141                        &lsi_units,
142                        &request.return_consumed_capacity,
143                    )
144                }
145                Some(index) => {
146                    let gsi_units = std::collections::HashMap::from([(index.name, units)]);
147                    crate::types::consumed_capacity_with_indexes(
148                        table,
149                        0.0,
150                        &gsi_units,
151                        &request.return_consumed_capacity,
152                    )
153                }
154                None => {
155                    crate::types::consumed_capacity(table, units, &request.return_consumed_capacity)
156                }
157            };
158        };
159        crate::types::consumed_capacity_with_vector_indexes(
160            table,
161            crate::types::table_write_capacity_units(capacity.old_size, capacity.new_size),
162            &capacity.gsi_units,
163            &capacity.lsi_units,
164            &capacity.vector_bytes,
165            &request.return_consumed_capacity,
166        )
167    });
168
169    Ok(ExecuteStatementResponse {
170        items,
171        next_token,
172        consumed_capacity,
173    })
174}
175
176#[cfg(test)]
177mod tests {
178    use super::*;
179    use crate::storage::Storage;
180
181    /// Two GSIs, so the fan-out has a second insert to fail on.
182    async fn table_with_two_gsis(storage: &Storage) {
183        let req = serde_json::from_value(serde_json::json!({
184            "TableName": "pq_tx",
185            "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}],
186            "AttributeDefinitions": [
187                {"AttributeName": "pk", "AttributeType": "S"},
188                {"AttributeName": "g1", "AttributeType": "S"},
189                {"AttributeName": "g2", "AttributeType": "S"}
190            ],
191            "BillingMode": "PAY_PER_REQUEST",
192            "GlobalSecondaryIndexes": [
193                {
194                    "IndexName": "gsi-one",
195                    "KeySchema": [{"AttributeName": "g1", "KeyType": "HASH"}],
196                    "Projection": {"ProjectionType": "ALL"}
197                },
198                {
199                    "IndexName": "gsi-two",
200                    "KeySchema": [{"AttributeName": "g2", "KeyType": "HASH"}],
201                    "Projection": {"ProjectionType": "ALL"}
202                }
203            ]
204        }))
205        .unwrap();
206        crate::actions::create_table::execute(storage, req)
207            .await
208            .unwrap();
209    }
210
211    fn s(v: &str) -> AttributeValue {
212        AttributeValue::S(v.to_string())
213    }
214
215    async fn val_of(storage: &Storage, pk: &str) -> String {
216        let req = crate::actions::get_item::GetItemRequest {
217            table_name: "pq_tx".to_string(),
218            key: [("pk".to_string(), s(pk))].into_iter().collect(),
219            ..Default::default()
220        };
221        let got = crate::actions::get_item::execute(storage, req)
222            .await
223            .unwrap();
224        match got.item.unwrap().get("val").unwrap() {
225            AttributeValue::S(v) => v.clone(),
226            other => panic!("unexpected val: {other:?}"),
227        }
228    }
229
230    #[tokio::test]
231    async fn a_failed_fan_out_rolls_the_base_row_back() {
232        // Every other write path wraps its base write and index fan-out in one
233        // transaction. These two did not, so a fan-out that failed part way
234        // left the base row committed and the indexes disagreeing with it, with
235        // nothing to signal it.
236        let storage = Storage::memory().unwrap();
237        table_with_two_gsis(&storage).await;
238
239        let put = crate::actions::put_item::PutItemRequest {
240            table_name: "pq_tx".to_string(),
241            item: [
242                ("pk".to_string(), s("a")),
243                ("g1".to_string(), s("x")),
244                ("g2".to_string(), s("y")),
245                ("val".to_string(), s("before")),
246            ]
247            .into_iter()
248            .collect(),
249            ..Default::default()
250        };
251        crate::actions::put_item::execute(&storage, put)
252            .await
253            .unwrap();
254        assert_eq!(val_of(&storage, "a").await, "before");
255
256        // Let the first index insert land, fail the second.
257        storage.fail_gsi_insert_after(1);
258        let err = execute(
259            &storage,
260            ExecuteStatementRequest {
261                statement: "UPDATE \"pq_tx\" SET val='after' WHERE pk='a'".to_string(),
262                ..Default::default()
263            },
264        )
265        .await
266        .expect_err("the injected failure must surface");
267        assert!(
268            format!("{err:?}").contains("injected GSI insert failure"),
269            "unexpected error: {err:?}"
270        );
271
272        assert_eq!(
273            val_of(&storage, "a").await,
274            "before",
275            "the base row must roll back with the fan-out that failed"
276        );
277    }
278}
dynoxide/actions/execute_statement.rs

dynoxide/actions/
execute_statement.rs
