dynoxide/actions/
batch_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 BatchExecuteStatementRequest {
9    #[serde(rename = "Statements")]
10    pub statements: Vec<BatchStatementRequest>,
11}
12
13#[derive(Debug, Default, Deserialize)]
14pub struct BatchStatementRequest {
15    #[serde(rename = "Statement")]
16    pub statement: String,
17    #[serde(rename = "Parameters", default)]
18    pub parameters: Option<Vec<AttributeValue>>,
19}
20
21#[derive(Debug, Default, Serialize)]
22pub struct BatchExecuteStatementResponse {
23    #[serde(rename = "Responses")]
24    pub responses: Vec<BatchStatementResponse>,
25}
26
27#[derive(Debug, Default, Serialize)]
28#[non_exhaustive]
29pub struct BatchStatementResponse {
30    #[serde(rename = "Error", skip_serializing_if = "Option::is_none")]
31    pub error: Option<BatchStatementError>,
32    #[serde(rename = "Item", skip_serializing_if = "Option::is_none")]
33    pub item: Option<Item>,
34    /// The member statement's target table, echoed on a successful response
35    /// (with or without an `Item`), matching DynamoDB. Omitted on a
36    /// per-statement error and when the statement fails to parse.
37    #[serde(rename = "TableName", skip_serializing_if = "Option::is_none")]
38    pub table_name: Option<String>,
39}
40
41#[derive(Debug, Default, Serialize)]
42pub struct BatchStatementError {
43    #[serde(rename = "Code")]
44    pub code: String,
45    #[serde(rename = "Message")]
46    pub message: String,
47}
48
49pub async fn execute<S: StorageBackend>(
50    storage: &S,
51    request: BatchExecuteStatementRequest,
52) -> Result<BatchExecuteStatementResponse> {
53    if request.statements.is_empty() {
54        return Err(DynoxideError::ValidationException(
55            "1 validation error detected: Value '[]' at 'statements' failed to satisfy constraint: Member must have length greater than or equal to 1".to_string(),
56        ));
57    }
58
59    if request.statements.len() > 25 {
60        return Err(DynoxideError::ValidationException(
61            "Too many statements in BatchExecuteStatement; limit is 25".to_string(),
62        ));
63    }
64
65    let mut responses = Vec::with_capacity(request.statements.len());
66
67    for stmt_req in &request.statements {
68        // A COUNT projection is rejected before parsing with the bare message
69        // captured on ExecuteStatement, carried here under the same
70        // per-statement ValidationError code a parse failure uses.
71        if let Some(msg) = partiql::parser::count_projection_rejection(&stmt_req.statement) {
72            responses.push(BatchStatementResponse {
73                error: Some(BatchStatementError {
74                    code: "ValidationError".to_string(),
75                    message: msg,
76                }),
77                item: None,
78                table_name: None,
79            });
80            continue;
81        }
82
83        let parsed = partiql::parser::parse(&stmt_req.statement);
84
85        let response = match parsed {
86            Err(e) => BatchStatementResponse {
87                error: Some(BatchStatementError {
88                    // A per-statement parse failure carries the short-form
89                    // `ValidationError` code, the same as an execution error,
90                    // matching DynamoDB.
91                    code: "ValidationError".to_string(),
92                    message: format!("Statement wasn't well formed, can't be processed: {e}"),
93                }),
94                item: None,
95                table_name: None,
96            },
97            Ok(stmt) => {
98                // DynamoDB echoes the target table on a successful response, but
99                // not on a per-statement error.
100                let table = partiql::parser::table_name(&stmt).map(str::to_string);
101                let params = stmt_req.parameters.as_deref().unwrap_or_default();
102                match partiql::executor::execute(storage, &stmt, params, None).await {
103                    Ok(Some(items)) => {
104                        // A SELECT yields its row here; a DELETE/UPDATE carrying a
105                        // RETURNING clause yields the returned item. Batch surfaces
106                        // a single item per statement, so take the first.
107                        BatchStatementResponse {
108                            error: None,
109                            item: items.into_iter().next(),
110                            table_name: table,
111                        }
112                    }
113                    Ok(None) => BatchStatementResponse {
114                        error: None,
115                        item: None,
116                        table_name: table,
117                    },
118                    Err(e) => BatchStatementResponse {
119                        error: Some(BatchStatementError {
120                            code: e.short_error_code().to_string(),
121                            message: e.to_string(),
122                        }),
123                        item: None,
124                        table_name: None,
125                    },
126                }
127            }
128        };
129
130        responses.push(response);
131    }
132
133    Ok(BatchExecuteStatementResponse { responses })
134}
dynoxide/actions/batch_execute_statement.rs

dynoxide/actions/
batch_execute_statement.rs
