1use crate::actions::helpers;
2use crate::actions::index_capacity::{
3 WriteCapacity, aggregate_by_table, per_table_capacity, transactional_read_units,
4};
5use crate::errors::{CancellationReason, DynoxideError, Result};
6use crate::partiql;
7use crate::partiql::executor::ResolvedTable;
8use crate::storage_backend::StorageBackend;
9use crate::types::{AttributeValue, Item};
10use serde::{Deserialize, Serialize};
11use std::collections::{HashMap, HashSet};
12
13#[derive(Debug, Clone, Default, Deserialize)]
14pub struct ExecuteTransactionRequest {
15 #[serde(rename = "TransactStatements")]
16 pub transact_statements: Vec<ParameterizedStatement>,
17 #[serde(rename = "ClientRequestToken", default)]
18 pub client_request_token: Option<String>,
19 #[serde(rename = "ReturnConsumedCapacity", default)]
20 pub return_consumed_capacity: Option<String>,
21}
22
23#[derive(Debug, Clone, Default, Deserialize, Serialize)]
24pub struct ParameterizedStatement {
25 #[serde(rename = "Statement")]
26 pub statement: String,
27 #[serde(
28 rename = "Parameters",
29 default,
30 deserialize_with = "crate::types::deserialize_parameters"
31 )]
32 pub parameters: Option<Vec<AttributeValue>>,
33 #[serde(rename = "ReturnValuesOnConditionCheckFailure", default)]
37 pub return_values_on_condition_check_failure: Option<String>,
38}
39
40#[derive(Serialize)]
42pub(crate) struct TokenStatement<'a> {
43 #[serde(rename = "Statement")]
44 statement: &'a str,
45 #[serde(rename = "Parameters")]
46 parameters: Option<&'a [AttributeValue]>,
47 #[serde(rename = "ReturnValuesOnConditionCheckFailure")]
48 mode: &'a str,
49}
50
51pub(crate) fn token_statements(statements: &[ParameterizedStatement]) -> Vec<TokenStatement<'_>> {
59 statements
60 .iter()
61 .map(|s| TokenStatement {
62 statement: &s.statement,
63 parameters: s.parameters.as_deref(),
64 mode: s
65 .return_values_on_condition_check_failure
66 .as_deref()
67 .unwrap_or("NONE"),
68 })
69 .collect()
70}
71
72#[derive(Debug, Clone, Default, Serialize)]
75pub struct ExecuteTransactionResponse {
76 #[serde(rename = "Responses", skip_serializing_if = "Option::is_none")]
77 pub responses: Option<Vec<ItemResponse>>,
78 #[serde(rename = "ConsumedCapacity", skip_serializing_if = "Option::is_none")]
79 pub consumed_capacity: Option<Vec<crate::types::ConsumedCapacity>>,
80}
81
82#[derive(Debug, Clone, Default)]
89pub(crate) struct CachedTransaction {
90 pub(crate) response: ExecuteTransactionResponse,
91 pub(crate) replay_sizes: Vec<(String, usize)>,
93}
94
95#[derive(Debug, Clone, Default, Serialize)]
96pub struct ItemResponse {
97 #[serde(
98 rename = "Item",
99 skip_serializing_if = "Option::is_none",
100 serialize_with = "crate::set_order::item_in_hash_order"
101 )]
102 pub item: Option<Item>,
105}
106
107pub async fn execute<S: StorageBackend>(
112 storage: &S,
113 request: ExecuteTransactionRequest,
114) -> Result<ExecuteTransactionResponse> {
115 Ok(execute_cached(storage, request).await?.response)
116}
117
118fn statements_echo(statements: &[ParameterizedStatement]) -> String {
125 let mut out = crate::actions::batch_write_item::BoundedEcho::new();
126 let _ = write_statements(&mut out, statements);
127 out.finish()
128}
129
130fn write_statements(
131 out: &mut dyn std::fmt::Write,
132 statements: &[ParameterizedStatement],
133) -> std::fmt::Result {
134 out.write_str("[")?;
135 for (i, statement) in statements.iter().enumerate() {
136 if i > 0 {
137 out.write_str(", ")?;
138 }
139 write!(
140 out,
141 "ParameterizedStatement(statement={}, parameters=",
142 statement.statement
143 )?;
144 match statement.parameters {
145 Some(ref parameters) => {
146 out.write_str("[")?;
147 for (j, parameter) in parameters.iter().enumerate() {
148 if j > 0 {
149 out.write_str(", ")?;
150 }
151 crate::actions::batch_write_item::write_java_attribute_value(out, parameter)?;
152 }
153 out.write_str("]")?;
154 }
155 None => out.write_str("null")?,
156 }
157 out.write_str(", returnValuesOnConditionCheckFailure=")?;
158 out.write_str(
159 statement
160 .return_values_on_condition_check_failure
161 .as_deref()
162 .unwrap_or("null"),
163 )?;
164 out.write_str(")")?;
165 }
166 out.write_str("]")
167}
168
169fn member_violations(request: &ExecuteTransactionRequest) -> Vec<String> {
181 let mut violations = Vec::new();
182 violations.extend(crate::token_constraint_error(
183 request.client_request_token.as_deref(),
184 ));
185 if request.transact_statements.is_empty() {
186 violations.push(
187 "Value '[]' at 'transactStatements' failed to satisfy constraint: Member must \
188 have length greater than or equal to 1"
189 .to_string(),
190 );
191 }
192 if request.transact_statements.len() > 100 {
193 violations.push(format!(
194 "Value '{}' at 'transactStatements' failed to satisfy constraint: Member must \
195 have length less than or equal to 100",
196 statements_echo(&request.transact_statements)
197 ));
198 }
199 for (i, statement) in request.transact_statements.iter().enumerate() {
200 let member = format!("transactStatements.{}.member", i + 1);
201 violations.extend(
202 crate::validation::return_values_on_condition_check_failure_rejection(
203 statement
204 .return_values_on_condition_check_failure
205 .as_deref(),
206 &format!("{member}.returnValuesOnConditionCheckFailure"),
207 ),
208 );
209 violations.extend(crate::partiql::statement_length_violation(
210 &statement.statement,
211 &format!("{member}.statement"),
212 ));
213 violations.extend(crate::partiql::parameters_violation(
214 statement.parameters.as_deref(),
215 &format!("{member}.parameters"),
216 ));
217 }
218 violations.extend(crate::validation::return_consumed_capacity_rejection(
219 request.return_consumed_capacity.as_deref(),
220 ));
221 violations
222}
223
224pub(crate) fn check_members(request: &ExecuteTransactionRequest) -> Result<()> {
230 match crate::partiql::member_rejection(&member_violations(request)) {
231 Some(e) => Err(e),
232 None => Ok(()),
233 }
234}
235
236pub(crate) async fn execute_cached<S: StorageBackend>(
237 storage: &S,
238 request: ExecuteTransactionRequest,
239) -> Result<CachedTransaction> {
240 check_members(&request)?;
243
244 let statements = &request.transact_statements;
247
248 if statements
251 .iter()
252 .any(|s| crate::empty_names::parameters_have_empty_name(s.parameters.as_deref()))
253 {
254 return Err(crate::empty_names::bare());
255 }
256
257 for stmt in statements {
262 for parameter in stmt.parameters.iter().flatten() {
263 partiql::executor::check_parameter_value(parameter)?;
264 }
265 }
266
267 let mut parsed = Vec::with_capacity(statements.len());
269 for (index, stmt) in statements.iter().enumerate() {
270 if crate::value_nesting::parameters_nest_too_deep(
274 stmt.parameters.as_deref().unwrap_or_default(),
275 ) {
276 return Err(crate::value_nesting::deep_parameter_error());
277 }
278 if let Some(msg) = partiql::parser::count_projection_rejection(&stmt.statement) {
283 return Err(DynoxideError::ValidationException(format!(
284 "Validation failed in TransactStatements[{index}]: {msg}"
285 )));
286 }
287 let supplied = stmt.parameters.as_ref().map_or(0, Vec::len);
292 let ast = partiql::parser::parse_with_parameters(&stmt.statement, supplied).map_err(
293 |failure| {
294 let message = match failure {
295 partiql::parser::ParseFailure::Statement(e) => e.into_message(),
296 partiql::parser::ParseFailure::Literal(message) => message,
297 };
298 DynoxideError::ValidationException(format!(
299 "Validation failed in TransactStatements[{index}]: {message}"
300 ))
301 },
302 )?;
303 if partiql::parser::returning_variant(&ast).is_some() {
307 return Err(DynoxideError::ValidationException(format!(
308 "Validation failed in TransactStatements[{index}]: RETURNING clause is not supported in ExecuteTransaction."
309 )));
310 }
311 if matches!(ast, partiql::parser::Statement::Select { .. })
315 && partiql::parser::index_name(&ast).is_some()
316 {
317 return Err(DynoxideError::ValidationException(format!(
318 "Validation failed in TransactStatements[{index}]: Reads on indices are not supported within transactions."
319 )));
320 }
321 let params = stmt.parameters.clone().unwrap_or_default();
322 parsed.push((ast, params));
323 }
324
325 let reads = parsed
328 .iter()
329 .filter(|(stmt, _)| matches!(stmt, partiql::parser::Statement::Select { .. }))
330 .count();
331 if reads > 0 && reads < parsed.len() {
332 return Err(DynoxideError::ValidationException(
333 "ExecuteTransaction API does not support both read and write operations in the same request."
334 .to_string(),
335 ));
336 }
337
338 let mut seen_targets = HashSet::new();
342 let mut tables: HashMap<String, ResolvedTable> = HashMap::new();
345 for (index, (stmt, params)) in parsed.iter().enumerate() {
346 let Some(name) = partiql::parser::table_name(stmt) else {
347 continue;
348 };
349 if !tables.contains_key(name)
350 && let Ok(resolved) = ResolvedTable::load(storage, name).await
351 {
352 tables.insert(name.to_string(), resolved);
353 }
354 let Some(resolved) = tables.get(name) else {
357 continue;
358 };
359 if let Some(partiql::executor::KeyFault::NotAKeyType(message)) =
363 partiql::executor::write_key_fault(stmt, params, resolved)
364 {
365 return Err(DynoxideError::ValidationException(format!(
366 "Validation failed in TransactStatements[{index}]: {message}"
367 )));
368 }
369 if matches!(stmt, partiql::parser::Statement::Select { .. })
373 && let Err(e) = partiql::executor::reject_overlapping_select(stmt, params, resolved)
374 {
375 return Err(DynoxideError::ValidationException(format!(
376 "Validation failed in TransactStatements[{index}]: {e}"
377 )));
378 }
379 if matches!(stmt, partiql::parser::Statement::Select { .. })
380 && matches!(
381 partiql::executor::keyed_select(stmt, params, resolved),
382 partiql::executor::KeyedSelect::NotByPrimaryKey
383 )
384 {
385 return Err(DynoxideError::ValidationException(format!(
386 "Validation failed in TransactStatements[{index}]: Select statements within \
387 ExecuteTransaction must specify the primary key in the where clause."
388 )));
389 }
390 if let Some(target) = partiql::executor::statement_target_in(stmt, params, resolved)
391 && !seen_targets.insert(target)
392 {
393 return Err(DynoxideError::ValidationException(
394 "Transaction request cannot include multiple operations on one item".to_string(),
395 ));
396 }
397 }
398
399 let (responses, charges) = helpers::with_write_transaction(
401 storage,
402 execute_within_transaction(
403 storage,
404 statements,
405 &parsed,
406 &tables,
407 request.return_consumed_capacity.as_deref(),
408 ),
409 )
410 .await?;
411
412 let mode = &request.return_consumed_capacity;
417 let consumed_capacity = if is_read_set(&parsed) {
418 crate::types::build_transactional_capacity(
419 &transactional_read_units(&replay_sizes(&charges)),
420 mode,
421 crate::types::transactional_read_capacity,
422 )
423 } else {
424 build_write_capacity(&charges, mode)
425 };
426
427 let responses = if is_read_set(&parsed) {
432 responses
433 } else {
434 Vec::new()
435 };
436 Ok(CachedTransaction {
437 response: ExecuteTransactionResponse {
438 responses: Some(responses),
439 consumed_capacity,
440 },
441 replay_sizes: replay_sizes(&charges),
442 })
443}
444
445enum StatementCharge {
447 Read { table_name: String, size: usize },
449 Write(WriteCapacity),
451}
452
453impl StatementCharge {
454 fn table_name(&self) -> &str {
455 match self {
456 Self::Read { table_name, .. } => table_name,
457 Self::Write(capacity) => &capacity.table_name,
458 }
459 }
460
461 fn size(&self) -> usize {
463 match self {
464 Self::Read { size, .. } => *size,
465 Self::Write(capacity) => capacity
466 .old_size
467 .unwrap_or(0)
468 .max(capacity.new_size.unwrap_or(0)),
469 }
470 }
471
472 fn as_write(&self) -> WriteCapacity {
479 match self {
480 Self::Write(capacity) => capacity.clone(),
481 Self::Read { table_name, size } => WriteCapacity::new(
482 table_name,
483 None,
484 Some(*size),
485 HashMap::new(),
486 HashMap::new(),
487 ),
488 }
489 }
490}
491
492fn build_write_capacity(
494 charges: &[StatementCharge],
495 mode: &Option<String>,
496) -> Option<Vec<crate::types::ConsumedCapacity>> {
497 if !matches!(mode.as_deref(), Some("TOTAL") | Some("INDEXES")) {
501 return None;
502 }
503
504 let records: Vec<WriteCapacity> = charges.iter().map(StatementCharge::as_write).collect();
505 let by_table = aggregate_by_table(&records, crate::types::TRANSACTIONAL_CAPACITY_FACTOR);
506 per_table_capacity(
507 &by_table,
508 mode,
509 crate::types::transactional_write_capacity_with_indexes,
510 )
511}
512
513fn replay_sizes(charges: &[StatementCharge]) -> Vec<(String, usize)> {
515 charges
516 .iter()
517 .map(|charge| (charge.table_name().to_string(), charge.size()))
518 .collect()
519}
520
521fn is_read_set(parsed: &[(partiql::parser::Statement, Vec<AttributeValue>)]) -> bool {
528 parsed
529 .iter()
530 .all(|(stmt, _)| matches!(stmt, partiql::parser::Statement::Select { .. }))
531}
532
533pub(crate) fn replay_response(
542 cached: &CachedTransaction,
543 mode: &Option<String>,
544) -> CachedTransaction {
545 CachedTransaction {
546 response: ExecuteTransactionResponse {
547 responses: cached.response.responses.clone(),
548 consumed_capacity: crate::types::build_transactional_capacity(
549 &transactional_read_units(&cached.replay_sizes),
550 mode,
551 crate::types::transactional_read_capacity,
552 ),
553 },
554 replay_sizes: cached.replay_sizes.clone(),
555 }
556}
557
558async fn execute_within_transaction<S: StorageBackend>(
573 storage: &S,
574 statements: &[ParameterizedStatement],
575 parsed: &[(partiql::parser::Statement, Vec<AttributeValue>)],
576 tables: &HashMap<String, ResolvedTable>,
577 capacity_mode: Option<&str>,
578) -> Result<(Vec<ItemResponse>, Vec<StatementCharge>)> {
579 let resolved_for = |stmt: &partiql::parser::Statement| {
580 partiql::parser::table_name(stmt).and_then(|t| tables.get(t))
581 };
582
583 let wrong_key_type: Vec<bool> = parsed
591 .iter()
592 .map(|(stmt, params)| {
593 resolved_for(stmt).is_some_and(|resolved| match stmt {
594 partiql::parser::Statement::Select { .. } => matches!(
595 partiql::executor::keyed_select(stmt, params, resolved),
596 partiql::executor::KeyedSelect::KeyType(_)
597 ),
598 _ => partiql::executor::write_key_fault(stmt, params, resolved).is_some(),
599 })
600 })
601 .collect();
602 if wrong_key_type.contains(&true) {
603 return Err(cancelled(
604 wrong_key_type
605 .iter()
606 .map(|&wrong| {
607 if wrong {
608 CancellationReason {
609 code: "ValidationError".to_string(),
610 message: Some(
611 "The provided key element does not match the schema".to_string(),
612 ),
613 item: None,
614 }
615 } else {
616 no_reason()
617 }
618 })
619 .collect(),
620 ));
621 }
622
623 let mut responses = Vec::with_capacity(parsed.len());
624 let mut charges: Vec<StatementCharge> = Vec::with_capacity(parsed.len());
625 let mut cancellation_reasons: Vec<CancellationReason> = Vec::with_capacity(parsed.len());
626 let mut failed = false;
627
628 for ((stmt, params), written) in parsed.iter().zip(statements) {
630 let resolved = resolved_for(stmt);
631 let old_on_failure =
634 written.return_values_on_condition_check_failure.as_deref() == Some("ALL_OLD");
635 let insert_keys = match resolved {
638 Some(resolved) => {
639 partiql::executor::check_transaction_insert_keys(stmt, params, resolved)
640 }
641 None => Ok(()),
642 };
643 let outcome = match insert_keys {
644 Err(e) => Err(e),
645 Ok(()) => {
646 partiql::executor::execute_written_page(
647 storage,
648 &written.statement,
649 stmt,
650 params,
651 None,
652 None,
653 false,
654 capacity_mode,
655 resolved,
656 old_on_failure,
657 )
658 .await
659 }
660 };
661 match outcome {
662 Ok(page) => {
663 let missed_condition = matches!(stmt, partiql::parser::Statement::Delete { .. })
668 && page
669 .capacity
670 .as_ref()
671 .is_some_and(|capacity| capacity.old_size.is_none())
672 && resolved.is_some_and(|resolved| {
673 partiql::executor::delete_has_condition(stmt, resolved)
674 });
675 let table_name = partiql::parser::table_name(stmt).unwrap_or_default();
676 charges.push(match page.capacity {
677 Some(capacity) => StatementCharge::Write(capacity),
678 None => StatementCharge::Read {
679 table_name: table_name.to_string(),
680 size: page.size,
681 },
682 });
683 let item = page.items.and_then(|items| items.into_iter().next());
684 responses.push(ItemResponse { item });
685 cancellation_reasons.push(if missed_condition {
686 CancellationReason {
687 code: "ConditionalCheckFailed".to_string(),
688 message: Some("The conditional request failed".to_string()),
689 item: None,
690 }
691 } else {
692 no_reason()
693 });
694 }
695 Err(e) => {
696 let Some(code) = outcome_code(&e) else {
697 return Err(e);
698 };
699 failed = true;
700 responses.push(ItemResponse { item: None });
701 cancellation_reasons.push(failure_reason(code, e));
702 }
703 }
704 }
705
706 if failed {
707 return Err(cancelled(cancellation_reasons));
708 }
709 Ok((responses, charges))
710}
711
712fn no_reason() -> CancellationReason {
714 CancellationReason {
715 code: "None".to_string(),
716 message: None,
717 item: None,
718 }
719}
720
721fn outcome_code(e: &DynoxideError) -> Option<&'static str> {
725 match e {
726 DynoxideError::ConditionalCheckFailedException(..) => Some("ConditionalCheckFailed"),
727 DynoxideError::DuplicateItemException(_) => Some("DuplicateItem"),
728 DynoxideError::ResourceNotFoundException(_) => Some("ResourceNotFound"),
732 DynoxideError::ValidationException(_) | DynoxideError::KeyEmptyValueValidation(_) => {
735 Some("ValidationError")
736 }
737 _ => None,
738 }
739}
740
741fn failure_reason(code: &str, e: DynoxideError) -> CancellationReason {
745 let message = Some(e.to_string());
746 let item = match e {
747 DynoxideError::ConditionalCheckFailedException(_, item) => item,
748 _ => None,
749 };
750 CancellationReason {
751 code: code.to_string(),
752 message,
753 item,
754 }
755}
756
757fn cancelled(reasons: Vec<CancellationReason>) -> DynoxideError {
759 let codes: Vec<&str> = reasons.iter().map(|r| r.code.as_str()).collect();
760 let message = format!(
761 "Transaction cancelled, please refer cancellation reasons for specific reasons [{}]",
762 codes.join(", ")
763 );
764 DynoxideError::TransactionCanceledException(message, reasons)
765}