dynoxide/actions/
execute_statement.rs1use 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 #[serde(rename = "ConsistentRead", default)]
20 pub consistent_read: Option<bool>,
21 #[serde(rename = "ReturnConsumedCapacity", default)]
22 pub return_consumed_capacity: Option<String>,
23}
24
25#[derive(Debug, Default, Serialize)]
26pub struct ExecuteStatementResponse {
27 #[serde(rename = "Items", skip_serializing_if = "Option::is_none")]
28 pub items: Option<Vec<Item>>,
29 #[serde(rename = "NextToken", skip_serializing_if = "Option::is_none")]
30 pub next_token: Option<String>,
31 #[serde(rename = "ConsumedCapacity", skip_serializing_if = "Option::is_none")]
32 pub consumed_capacity: Option<crate::types::ConsumedCapacity>,
33}
34
35pub async fn execute<S: StorageBackend>(
36 storage: &S,
37 request: ExecuteStatementRequest,
38) -> Result<ExecuteStatementResponse> {
39 if request.limit == Some(0) {
45 return Err(DynoxideError::ValidationException(
46 crate::validation::envelope_message(
47 "Value '0' at 'limit' failed to satisfy constraint: \
48 Member must have value greater than or equal to 1",
49 ),
50 ));
51 }
52
53 if let Some(msg) = partiql::parser::count_projection_rejection(&request.statement) {
57 return Err(DynoxideError::ValidationException(msg));
58 }
59
60 let stmt = partiql::parser::parse(&request.statement).map_err(|e| {
61 DynoxideError::ValidationException(format!(
62 "Statement wasn't well formed, can't be processed: {e}"
63 ))
64 })?;
65
66 let params = request.parameters.unwrap_or_default();
67 let page = partiql::executor::execute_page(
68 storage,
69 &stmt,
70 ¶ms,
71 request.limit,
72 request.next_token.as_deref(),
73 )
74 .await?;
75 let partiql::executor::StatementPage {
76 items,
77 size,
78 next_token,
79 ..
80 } = page;
81
82 let consumed_capacity = partiql::parser::table_name(&stmt).and_then(|table| {
88 let units = if matches!(stmt, partiql::parser::Statement::Select { .. }) {
89 crate::types::read_capacity_units_with_consistency(
90 size,
91 request.consistent_read.unwrap_or(false),
92 )
93 } else {
94 crate::types::write_capacity_units(size)
95 };
96 crate::types::consumed_capacity(table, units, &request.return_consumed_capacity)
97 });
98
99 Ok(ExecuteStatementResponse {
100 items,
101 next_token,
102 consumed_capacity,
103 })
104}