1use crate::errors::{DynoxideError, Result};
6use crate::expressions::key_condition::{ResolvedSortKeyCondition, sk_conditions_to_sql};
7use crate::partiql::parser::{
8 CompOp, PartiqlValue, ReturningVariant, SetValue, Statement, WhereClause, WhereCondition,
9};
10use crate::storage_backend::StorageBackend;
11use crate::types::{AttributeValue, Item};
12use std::collections::HashMap;
13
14pub async fn execute<S: StorageBackend>(
22 storage: &S,
23 stmt: &Statement,
24 parameters: &[AttributeValue],
25 limit: Option<usize>,
26) -> Result<Option<Vec<Item>>> {
27 Ok(execute_measured(storage, stmt, parameters, limit).await?.0)
28}
29
30pub async fn execute_measured<S: StorageBackend>(
35 storage: &S,
36 stmt: &Statement,
37 parameters: &[AttributeValue],
38 limit: Option<usize>,
39) -> Result<(Option<Vec<Item>>, usize)> {
40 let page = execute_page(storage, stmt, parameters, limit, None).await?;
41 Ok((page.items, page.size))
42}
43
44#[derive(Debug, Default)]
49#[non_exhaustive]
50pub struct StatementPage {
51 pub items: Option<Vec<Item>>,
54 pub size: usize,
56 pub next_token: Option<String>,
58}
59
60pub async fn execute_page<S: StorageBackend>(
66 storage: &S,
67 stmt: &Statement,
68 parameters: &[AttributeValue],
69 limit: Option<usize>,
70 next_token: Option<&str>,
71) -> Result<StatementPage> {
72 if next_token.is_some() && !matches!(stmt, Statement::Select { .. }) {
73 return Err(DynoxideError::ValidationException(
74 "NextToken is only valid on a SELECT statement".to_string(),
75 ));
76 }
77 match stmt {
78 Statement::Select {
79 table_name,
80 projections,
81 where_clause,
82 } => {
83 let (items, token) = execute_select(
84 storage,
85 table_name,
86 projections,
87 where_clause.as_ref(),
88 parameters,
89 limit,
90 next_token,
91 )
92 .await?;
93 let size = items
94 .as_ref()
95 .map(|rows| rows.iter().map(crate::types::item_size).sum())
96 .unwrap_or(0);
97 Ok(StatementPage {
98 items,
99 size,
100 next_token: token,
101 })
102 }
103 Statement::Insert {
104 table_name,
105 item,
106 if_not_exists,
107 } => {
108 let size =
109 execute_insert(storage, table_name, item, parameters, *if_not_exists).await?;
110 Ok(StatementPage {
111 items: None,
112 size,
113 ..Default::default()
114 })
115 }
116 Statement::Update {
117 table_name,
118 set_clauses,
119 remove_paths,
120 where_clause,
121 returning,
122 } => {
123 let (projection, size) = execute_update(
124 storage,
125 table_name,
126 set_clauses,
127 remove_paths,
128 where_clause.as_ref(),
129 parameters,
130 *returning,
131 )
132 .await?;
133 let items = projection.map(|item| {
138 if item.is_empty() {
139 Vec::new()
140 } else {
141 vec![item]
142 }
143 });
144 Ok(StatementPage {
145 items,
146 size,
147 ..Default::default()
148 })
149 }
150 Statement::Delete {
151 table_name,
152 where_clause,
153 returning,
154 } => {
155 if let Some(variant) = returning {
159 if *variant != ReturningVariant::AllOld {
160 return Err(DynoxideError::ValidationException(format!(
161 "Invalid returning clause: RETURNING {} *. Only RETURNING ALL OLD * is allowed in DELETE statements.",
162 variant.as_sql()
163 )));
164 }
165 }
166 let (old_item, size) =
167 execute_delete(storage, table_name, where_clause.as_ref(), parameters).await?;
168 let items = if returning.is_some() {
173 Some(old_item.map(|item| vec![item]).unwrap_or_default())
174 } else {
175 None
176 };
177 Ok(StatementPage {
178 items,
179 size,
180 ..Default::default()
181 })
182 }
183 }
184}
185
186fn insert_nested_projection(result: &mut Item, path: &str, val: AttributeValue) {
192 let parts: Vec<&str> = path.split('.').collect();
193 let key = parts.last().unwrap();
195 result.insert(key.to_string(), val);
196}
197
198async fn execute_select<S: StorageBackend>(
199 storage: &S,
200 table_name: &str,
201 projections: &[String],
202 where_clause: Option<&WhereClause>,
203 parameters: &[AttributeValue],
204 limit: Option<usize>,
205 next_token: Option<&str>,
206) -> Result<(Option<Vec<Item>>, Option<String>)> {
207 let meta = require_table(storage, table_name).await?;
208 let key_schema = crate::actions::helpers::parse_key_schema(&meta)?;
209
210 let fingerprint = statement_fingerprint(where_clause, parameters);
211 let cursor = next_token
212 .map(|token| decode_next_token(token, table_name, fingerprint))
213 .transpose()?;
214
215 let window = evaluate_window(
216 storage,
217 table_name,
218 where_clause,
219 parameters,
220 &key_schema,
221 cursor.as_ref(),
222 limit,
223 )
224 .await?;
225
226 let token = match (limit, &window.last_evaluated) {
230 (Some(lim), Some((pk, sk))) if window.evaluated >= lim => {
231 Some(encode_next_token(table_name, fingerprint, pk, sk))
232 }
233 _ => None,
234 };
235
236 let items = window
239 .matched
240 .into_iter()
241 .map(|item| {
242 if projections.is_empty() {
243 item
244 } else {
245 let mut projected = HashMap::new();
246 for proj in projections {
247 if let Some(val) = resolve_nested_path(&item, proj) {
248 insert_nested_projection(&mut projected, proj, val.clone());
249 }
250 }
251 projected
252 }
253 })
254 .collect();
255
256 Ok((Some(items), token))
257}
258
259struct Window {
261 matched: Vec<Item>,
262 last_evaluated: Option<(String, String)>,
265 evaluated: usize,
267}
268
269async fn evaluate_window<S: StorageBackend>(
277 storage: &S,
278 table_name: &str,
279 where_clause: Option<&WhereClause>,
280 parameters: &[AttributeValue],
281 key_schema: &crate::actions::helpers::KeySchema,
282 cursor: Option<&Cursor>,
283 limit: Option<usize>,
284) -> Result<Window> {
285 let pk_condition = where_clause.and_then(|wc| find_pk_condition(wc, &key_schema.partition_key));
286
287 let rows: Vec<(String, String, String)> = if let Some(pk_cond) = pk_condition {
288 let pk_val = resolve_value(&pk_cond.value, parameters)?;
289 let pk_str = pk_val
290 .to_key_string()
291 .ok_or_else(|| DynoxideError::ValidationException("Invalid key value".to_string()))?;
292
293 let sk_conditions = match (key_schema.sort_key.as_deref(), where_clause) {
299 (Some(sk_name), Some(wc)) => {
300 translate_sk_conditions(&wc.groups[0], sk_name, parameters)
301 }
302 _ => None,
303 }
304 .unwrap_or_default();
305
306 let (sk_condition_sql, sk_param_values) = sk_conditions_to_sql(&sk_conditions);
311 let sk_params_refs: Vec<&str> = sk_param_values.iter().map(|s| s.as_str()).collect();
312
313 let params = crate::storage::QueryParams {
316 sk_condition: sk_condition_sql.as_deref(),
317 sk_params: &sk_params_refs,
318 forward: true,
319 limit,
320 exclusive_start_sk: cursor.map(|c| c.sk.as_str()),
321 ..Default::default()
322 };
323 storage.query_items(table_name, &pk_str, ¶ms).await?
324 } else {
325 let params = crate::storage::ScanParams {
326 limit,
327 exclusive_start_pk: cursor.map(|c| c.pk.as_str()),
328 exclusive_start_sk: cursor.map(|c| c.sk.as_str()),
329 ..Default::default()
330 };
331 storage.scan_items(table_name, ¶ms).await?
332 };
333
334 let evaluated = rows.len();
335 let last_evaluated = rows.last().map(|(pk, sk, _)| (pk.clone(), sk.clone()));
336 let matched = rows
337 .into_iter()
338 .filter_map(|(_, _, json)| serde_json::from_str::<Item>(&json).ok())
339 .filter(|item| matches_where(item, where_clause, parameters))
340 .collect();
341
342 Ok(Window {
343 matched,
344 last_evaluated,
345 evaluated,
346 })
347}
348
349struct Cursor {
351 pk: String,
352 sk: String,
353}
354
355fn statement_fingerprint(where_clause: Option<&WhereClause>, parameters: &[AttributeValue]) -> u64 {
360 use std::hash::{Hash, Hasher};
361 let mut hasher = std::collections::hash_map::DefaultHasher::new();
362 format!("{where_clause:?}").hash(&mut hasher);
363 serde_json::to_string(parameters)
364 .unwrap_or_default()
365 .hash(&mut hasher);
366 hasher.finish()
367}
368
369fn encode_next_token(table_name: &str, fingerprint: u64, pk: &str, sk: &str) -> String {
375 use base64::Engine;
376 let payload =
377 serde_json::json!({ "t": table_name, "f": fingerprint, "pk": pk, "sk": sk }).to_string();
378 base64::engine::general_purpose::STANDARD.encode(payload)
379}
380
381fn token_mismatch() -> DynoxideError {
387 DynoxideError::ValidationException("NextToken does not match request".to_string())
388}
389
390fn decode_next_token(token: &str, table_name: &str, fingerprint: u64) -> Result<Cursor> {
397 use base64::Engine;
398 let invalid = || DynoxideError::ValidationException("Invalid NextToken".to_string());
399 let raw = base64::engine::general_purpose::STANDARD
400 .decode(token)
401 .map_err(|_| invalid())?;
402 let value: serde_json::Value = serde_json::from_slice(&raw).map_err(|_| invalid())?;
403 if value["t"].as_str() != Some(table_name) || value["f"].as_u64() != Some(fingerprint) {
404 return Err(token_mismatch());
405 }
406 Ok(Cursor {
407 pk: value["pk"].as_str().ok_or_else(invalid)?.to_string(),
408 sk: value["sk"].as_str().ok_or_else(invalid)?.to_string(),
409 })
410}
411
412fn translate_sk_conditions(
424 group: &[WhereCondition],
425 sk_name: &str,
426 parameters: &[AttributeValue],
427) -> Option<Vec<ResolvedSortKeyCondition>> {
428 let mut resolved = Vec::new();
429 for cond in group {
430 match cond {
431 WhereCondition::Comparison(c) if c.path == sk_name => {
432 let value = resolve_value(&c.value, parameters).ok()?;
433 value.to_key_string()?;
434 let sk = sk_name.to_string();
435 resolved.push(match c.op {
436 CompOp::Eq => ResolvedSortKeyCondition::Eq(sk, value),
437 CompOp::Lt => ResolvedSortKeyCondition::Lt(sk, value),
438 CompOp::Le => ResolvedSortKeyCondition::Le(sk, value),
439 CompOp::Gt => ResolvedSortKeyCondition::Gt(sk, value),
440 CompOp::Ge => ResolvedSortKeyCondition::Ge(sk, value),
441 CompOp::Ne => return None,
444 });
445 }
446 WhereCondition::Between(path, lo, hi) if path == sk_name => {
447 let lo = resolve_value(lo, parameters).ok()?;
448 let hi = resolve_value(hi, parameters).ok()?;
449 lo.to_key_string()?;
450 hi.to_key_string()?;
451 resolved.push(ResolvedSortKeyCondition::Between(
452 sk_name.to_string(),
453 lo,
454 hi,
455 ));
456 }
457 WhereCondition::BeginsWith(path, prefix) if path == sk_name => {
458 let prefix = resolve_value(prefix, parameters).ok()?;
459 prefix.to_key_string()?;
460 resolved.push(ResolvedSortKeyCondition::BeginsWith(
461 sk_name.to_string(),
462 prefix,
463 ));
464 }
465 WhereCondition::NotBeginsWith(path, _)
469 | WhereCondition::In(path, _)
470 | WhereCondition::Contains(path, _)
471 | WhereCondition::Exists(path)
472 | WhereCondition::NotExists(path)
473 | WhereCondition::IsMissing(path)
474 | WhereCondition::IsNotMissing(path)
475 if path == sk_name =>
476 {
477 return None;
478 }
479 _ => {}
480 }
481 }
482 Some(resolved)
483}
484
485fn find_pk_condition<'a>(
487 wc: &'a WhereClause,
488 pk_name: &str,
489) -> Option<&'a crate::partiql::parser::Condition> {
490 if wc.groups.len() == 1 {
493 wc.groups[0].iter().find_map(|c| match c {
494 WhereCondition::Comparison(cond) if cond.path == pk_name && cond.op == CompOp::Eq => {
495 Some(cond)
496 }
497 _ => None,
498 })
499 } else {
500 None
501 }
502}
503
504async fn execute_insert<S: StorageBackend>(
507 storage: &S,
508 table_name: &str,
509 item_template: &HashMap<String, PartiqlValue>,
510 parameters: &[AttributeValue],
511 if_not_exists: bool,
512) -> Result<usize> {
513 let mut item = HashMap::new();
515 for (k, v) in item_template {
516 let resolved = match v {
517 PartiqlValue::Literal(av) => av.clone(),
518 PartiqlValue::Parameter(idx) => parameters.get(*idx).cloned().ok_or_else(|| {
519 DynoxideError::ValidationException(format!(
520 "Parameter index {idx} out of range (have {} parameters)",
521 parameters.len()
522 ))
523 })?,
524 };
525 item.insert(k.clone(), resolved);
526 }
527
528 let meta = require_table(storage, table_name).await?;
529 let key_schema = crate::actions::helpers::parse_key_schema(&meta)?;
530
531 crate::actions::helpers::validate_item_keys(&item, &key_schema, &meta)?;
533 crate::validation::validate_item_attribute_values(&item)?;
534
535 crate::validation::normalize_item_sets(&mut item);
537
538 let (pk, sk) = crate::actions::helpers::extract_key_strings(&item, &key_schema)?;
540
541 let existing = storage.get_item(table_name, &pk, &sk).await?;
543 if existing.is_some() {
544 if if_not_exists {
545 return Ok(0);
547 }
548 return Err(DynoxideError::DuplicateItemException(
549 "Duplicate primary key exists in table".to_string(),
550 ));
551 }
552
553 let item_json = serde_json::to_string(&item)
554 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
555 let item_size = crate::types::item_size(&item);
556
557 let hash_prefix = item
558 .get(&key_schema.partition_key)
559 .map(crate::storage::compute_hash_prefix)
560 .unwrap_or_default();
561 let old_json = storage
562 .put_item_with_hash(table_name, &pk, &sk, &item_json, item_size, &hash_prefix)
563 .await?;
564
565 let table_sk_attr = key_schema.sort_key.as_deref();
567 let _ = crate::actions::gsi::maintain_gsis_after_write(
568 storage,
569 table_name,
570 &meta,
571 &pk,
572 &sk,
573 &item,
574 &key_schema.partition_key,
575 table_sk_attr,
576 )
577 .await?;
578
579 crate::actions::lsi::maintain_lsis_after_write(
581 storage,
582 table_name,
583 &meta,
584 &pk,
585 &sk,
586 &item,
587 &key_schema.partition_key,
588 table_sk_attr,
589 )
590 .await?;
591
592 let old_item: Option<Item> = old_json.as_ref().and_then(|j| serde_json::from_str(j).ok());
594 crate::streams::record_stream_event(storage, &meta, old_item.as_ref(), Some(&item)).await?;
595
596 Ok(item_size)
597}
598
599async fn execute_update<S: StorageBackend>(
604 storage: &S,
605 table_name: &str,
606 set_clauses: &[crate::partiql::parser::SetClause],
607 remove_paths: &[String],
608 where_clause: Option<&WhereClause>,
609 parameters: &[AttributeValue],
610 returning: Option<ReturningVariant>,
611) -> Result<(Option<Item>, usize)> {
612 let meta = require_table(storage, table_name).await?;
613 let key_schema = crate::actions::helpers::parse_key_schema(&meta)?;
614
615 let wc = where_clause.ok_or_else(|| {
617 DynoxideError::ValidationException("UPDATE requires a WHERE clause".to_string())
618 })?;
619
620 if wc.groups.len() > 1 {
622 return Err(DynoxideError::ValidationException(
623 "UPDATE does not support OR conditions in WHERE clause".to_string(),
624 ));
625 }
626
627 let pk_cond =
629 find_comparison_in_groups(&wc.groups, &key_schema.partition_key).ok_or_else(|| {
630 DynoxideError::ValidationException(
631 "Where clause does not contain a mandatory equality on all key attributes"
632 .to_string(),
633 )
634 })?;
635
636 let pk_val = resolve_value(&pk_cond.value, parameters)?;
637 let pk_str = pk_val
638 .to_key_string()
639 .ok_or_else(|| DynoxideError::ValidationException("Invalid key value".to_string()))?;
640
641 let sk_str = if let Some(ref sk_name) = key_schema.sort_key {
642 let sk_cond = find_comparison_in_groups(&wc.groups, sk_name);
643 if sk_cond.is_none() {
644 return Err(DynoxideError::ValidationException(
645 "Where clause does not contain a mandatory equality on all key attributes"
646 .to_string(),
647 ));
648 }
649 sk_cond
650 .map(|c| resolve_value(&c.value, parameters))
651 .transpose()?
652 .and_then(|v| v.to_key_string())
653 .unwrap_or_default()
654 } else {
655 String::new()
656 };
657
658 let existing_json = storage.get_item(table_name, &pk_str, &sk_str).await?;
660 let mut item: Item = existing_json
661 .as_ref()
662 .and_then(|j| serde_json::from_str(j).ok())
663 .unwrap_or_default();
664
665 let old_item = item.clone();
666
667 if existing_json.is_none() || !matches_where(&old_item, where_clause, parameters) {
672 return Err(DynoxideError::ConditionalCheckFailedException(
673 "The conditional request failed".to_string(),
674 None,
675 ));
676 }
677
678 let before_item = item.clone();
679
680 for clause in set_clauses {
682 let val = resolve_set_value(&clause.value, &item, parameters)?;
683 set_nested_value(&mut item, &clause.path, val)?;
684 }
685
686 for path in remove_paths {
688 remove_nested_value(&mut item, path);
689 }
690
691 if item.is_empty() {
693 return Ok((None, 0));
694 }
695
696 crate::validation::validate_item_attribute_values(&item)?;
698 crate::validation::normalize_item_sets(&mut item);
699
700 crate::actions::helpers::validate_updated_index_keys(&before_item, &item, &meta)?;
702
703 let item_json = serde_json::to_string(&item)
704 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
705 let item_size = crate::types::item_size(&item);
706
707 let hash_prefix = item
708 .get(&key_schema.partition_key)
709 .map(crate::storage::compute_hash_prefix)
710 .unwrap_or_default();
711 storage
712 .put_item_with_hash(
713 table_name,
714 &pk_str,
715 &sk_str,
716 &item_json,
717 item_size,
718 &hash_prefix,
719 )
720 .await?;
721
722 let table_sk_attr = key_schema.sort_key.as_deref();
724 let _ = crate::actions::gsi::maintain_gsis_after_write(
725 storage,
726 table_name,
727 &meta,
728 &pk_str,
729 &sk_str,
730 &item,
731 &key_schema.partition_key,
732 table_sk_attr,
733 )
734 .await?;
735
736 crate::actions::lsi::maintain_lsis_after_write(
738 storage,
739 table_name,
740 &meta,
741 &pk_str,
742 &sk_str,
743 &item,
744 &key_schema.partition_key,
745 table_sk_attr,
746 )
747 .await?;
748
749 let old_ref = if existing_json.is_some() {
751 Some(&old_item)
752 } else {
753 None
754 };
755 crate::streams::record_stream_event(storage, &meta, old_ref, Some(&item)).await?;
756
757 let projection = returning.map(|variant| {
762 let modified: std::collections::BTreeSet<String> = set_clauses
763 .iter()
764 .map(|c| c.path.clone())
765 .chain(remove_paths.iter().cloned())
766 .collect();
767 project_returning(variant, &old_item, &item, &modified)
768 });
769
770 Ok((projection, item_size))
771}
772
773fn project_returning(
781 variant: ReturningVariant,
782 old_item: &Item,
783 new_item: &Item,
784 modified: &std::collections::BTreeSet<String>,
785) -> Item {
786 match variant {
787 ReturningVariant::AllOld => old_item.clone(),
788 ReturningVariant::AllNew => new_item.clone(),
789 ReturningVariant::ModifiedOld => project_modified(modified, old_item),
790 ReturningVariant::ModifiedNew => project_modified(modified, new_item),
791 }
792}
793
794enum ProjNode {
799 Leaf(AttributeValue),
800 Map(HashMap<String, ProjNode>),
801 List(std::collections::BTreeMap<usize, ProjNode>),
802}
803
804fn project_modified(paths: &std::collections::BTreeSet<String>, source: &Item) -> Item {
814 let mut root: HashMap<String, ProjNode> = HashMap::new();
815 for path in paths {
816 if let (Some(val), Some(segments)) =
817 (resolve_nested_path(source, path), split_path_segments(path))
818 {
819 if let Some((PathSegment::Key(key), rest)) = segments.split_first() {
821 let node = root
822 .entry((*key).to_string())
823 .or_insert_with(|| fresh_proj_node(rest));
824 insert_proj_node(node, rest, val.clone());
825 }
826 }
827 }
828 root.into_iter()
829 .map(|(k, node)| (k, proj_node_to_value(node)))
830 .collect()
831}
832
833fn fresh_proj_node(segments: &[PathSegment]) -> ProjNode {
837 match segments.first() {
838 Some(PathSegment::Index(_)) => ProjNode::List(std::collections::BTreeMap::new()),
839 _ => ProjNode::Map(HashMap::new()),
840 }
841}
842
843fn insert_proj_node(node: &mut ProjNode, segments: &[PathSegment], val: AttributeValue) {
846 let Some((seg, rest)) = segments.split_first() else {
847 *node = ProjNode::Leaf(val);
848 return;
849 };
850 match seg {
851 PathSegment::Key(k) => {
852 if let ProjNode::Map(map) = node {
853 let child = map
854 .entry((*k).to_string())
855 .or_insert_with(|| fresh_proj_node(rest));
856 insert_proj_node(child, rest, val);
857 }
858 }
859 PathSegment::Index(i) => {
860 if let ProjNode::List(list) = node {
861 let child = list.entry(*i).or_insert_with(|| fresh_proj_node(rest));
862 insert_proj_node(child, rest, val);
863 }
864 }
865 }
866}
867
868fn proj_node_to_value(node: ProjNode) -> AttributeValue {
871 match node {
872 ProjNode::Leaf(v) => v,
873 ProjNode::Map(map) => AttributeValue::M(
874 map.into_iter()
875 .map(|(k, n)| (k, proj_node_to_value(n)))
876 .collect(),
877 ),
878 ProjNode::List(list) => {
879 AttributeValue::L(list.into_values().map(proj_node_to_value).collect())
880 }
881 }
882}
883
884async fn execute_delete<S: StorageBackend>(
887 storage: &S,
888 table_name: &str,
889 where_clause: Option<&WhereClause>,
890 parameters: &[AttributeValue],
891) -> Result<(Option<Item>, usize)> {
892 let meta = require_table(storage, table_name).await?;
893 let key_schema = crate::actions::helpers::parse_key_schema(&meta)?;
894
895 let wc = where_clause.ok_or_else(|| {
896 DynoxideError::ValidationException("DELETE requires a WHERE clause".to_string())
897 })?;
898
899 if wc.groups.len() > 1 {
901 return Err(DynoxideError::ValidationException(
902 "DELETE does not support OR conditions in WHERE clause".to_string(),
903 ));
904 }
905
906 let pk_cond =
907 find_comparison_in_groups(&wc.groups, &key_schema.partition_key).ok_or_else(|| {
908 DynoxideError::ValidationException(
909 "Where clause does not contain a mandatory equality on all key attributes"
910 .to_string(),
911 )
912 })?;
913
914 let pk_val = resolve_value(&pk_cond.value, parameters)?;
915 let pk_str = pk_val
916 .to_key_string()
917 .ok_or_else(|| DynoxideError::ValidationException("Invalid key value".to_string()))?;
918
919 if let Some(ref sk_name) = key_schema.sort_key {
921 let has_sk_condition = wc.groups.iter().any(|group| {
922 group.iter().any(|c| match c {
923 WhereCondition::Comparison(comp) => comp.path == *sk_name && comp.op == CompOp::Eq,
924 _ => false,
925 })
926 });
927 if !has_sk_condition {
928 return Err(DynoxideError::ValidationException(
929 "Where clause does not contain a mandatory equality on all key attributes"
930 .to_string(),
931 ));
932 }
933 }
934
935 let sk_str = if let Some(ref sk_name) = key_schema.sort_key {
936 find_comparison_in_groups(&wc.groups, sk_name)
937 .map(|c| resolve_value(&c.value, parameters))
938 .transpose()?
939 .and_then(|v| v.to_key_string())
940 .unwrap_or_default()
941 } else {
942 String::new()
943 };
944
945 if let Some(json) = storage.get_item(table_name, &pk_str, &sk_str).await? {
952 let existing: Item = serde_json::from_str(&json)
953 .map_err(|e| DynoxideError::InternalServerError(format!("Bad item JSON: {e}")))?;
954 if !matches_where(&existing, where_clause, parameters) {
955 return Err(DynoxideError::ConditionalCheckFailedException(
956 "The conditional request failed".to_string(),
957 None,
958 ));
959 }
960 }
961
962 let old_json = storage.delete_item(table_name, &pk_str, &sk_str).await?;
963
964 let _ = crate::actions::gsi::maintain_gsis_after_delete(
966 storage, table_name, &meta, &pk_str, &sk_str,
967 )
968 .await?;
969
970 crate::actions::lsi::maintain_lsis_after_delete(storage, table_name, &meta, &pk_str, &sk_str)
972 .await?;
973
974 let old_item: Option<Item> = old_json.as_ref().and_then(|j| serde_json::from_str(j).ok());
976 if old_item.is_some() {
977 crate::streams::record_stream_event(storage, &meta, old_item.as_ref(), None).await?;
978 }
979
980 let deleted_size = old_item.as_ref().map(crate::types::item_size).unwrap_or(0);
983 Ok((old_item, deleted_size))
984}
985
986async fn require_table<S: StorageBackend>(
991 storage: &S,
992 table_name: &str,
993) -> Result<crate::storage::TableMetadata> {
994 crate::actions::helpers::require_table(storage, table_name).await
995}
996
997fn find_comparison_in_groups<'a>(
1000 groups: &'a [Vec<WhereCondition>],
1001 path: &str,
1002) -> Option<&'a crate::partiql::parser::Condition> {
1003 for group in groups {
1004 if let Some(cond) = find_comparison(group, path) {
1005 return Some(cond);
1006 }
1007 }
1008 None
1009}
1010
1011fn find_comparison<'a>(
1013 conditions: &'a [WhereCondition],
1014 path: &str,
1015) -> Option<&'a crate::partiql::parser::Condition> {
1016 conditions.iter().find_map(|c| match c {
1017 WhereCondition::Comparison(cond) if cond.path == path && cond.op == CompOp::Eq => {
1018 Some(cond)
1019 }
1020 _ => None,
1021 })
1022}
1023
1024fn resolve_value(val: &PartiqlValue, parameters: &[AttributeValue]) -> Result<AttributeValue> {
1026 match val {
1027 PartiqlValue::Literal(av) => Ok(av.clone()),
1028 PartiqlValue::Parameter(idx) => parameters.get(*idx).cloned().ok_or_else(|| {
1029 DynoxideError::ValidationException(format!(
1030 "Parameter index {idx} out of range (have {} parameters)",
1031 parameters.len()
1032 ))
1033 }),
1034 }
1035}
1036
1037fn resolve_set_value(
1039 val: &SetValue,
1040 item: &Item,
1041 parameters: &[AttributeValue],
1042) -> Result<AttributeValue> {
1043 match val {
1044 SetValue::Simple(pv) => resolve_value(pv, parameters),
1045 SetValue::Add(attr, pv) => {
1046 let current = resolve_nested_path(item, attr);
1047 let operand = resolve_value(pv, parameters)?;
1048 match (current, &operand) {
1049 (Some(AttributeValue::N(cur)), AttributeValue::N(add)) => {
1050 use bigdecimal::BigDecimal;
1051 use std::str::FromStr;
1052 let a = BigDecimal::from_str(cur).map_err(|e| {
1053 DynoxideError::ValidationException(format!("Invalid number: {e}"))
1054 })?;
1055 let b = BigDecimal::from_str(add).map_err(|e| {
1056 DynoxideError::ValidationException(format!("Invalid number: {e}"))
1057 })?;
1058 let result = a + b;
1059 Ok(AttributeValue::N(format_bigdecimal(&result)))
1060 }
1061 (None, AttributeValue::N(_)) => {
1062 Ok(operand)
1064 }
1065 _ => Err(DynoxideError::ValidationException(
1066 "SET expression add requires numeric attribute and operand".to_string(),
1067 )),
1068 }
1069 }
1070 SetValue::Sub(attr, pv) => {
1071 let current = resolve_nested_path(item, attr);
1072 let operand = resolve_value(pv, parameters)?;
1073 match (current, &operand) {
1074 (Some(AttributeValue::N(cur)), AttributeValue::N(sub)) => {
1075 use bigdecimal::BigDecimal;
1076 use std::str::FromStr;
1077 let a = BigDecimal::from_str(cur).map_err(|e| {
1078 DynoxideError::ValidationException(format!("Invalid number: {e}"))
1079 })?;
1080 let b = BigDecimal::from_str(sub).map_err(|e| {
1081 DynoxideError::ValidationException(format!("Invalid number: {e}"))
1082 })?;
1083 let result = a - b;
1084 Ok(AttributeValue::N(format_bigdecimal(&result)))
1085 }
1086 (None, AttributeValue::N(sub)) => {
1087 use bigdecimal::BigDecimal;
1089 use std::str::FromStr;
1090 let b = BigDecimal::from_str(sub).map_err(|e| {
1091 DynoxideError::ValidationException(format!("Invalid number: {e}"))
1092 })?;
1093 let result = -b;
1094 Ok(AttributeValue::N(format_bigdecimal(&result)))
1095 }
1096 _ => Err(DynoxideError::ValidationException(
1097 "SET expression subtract requires numeric attribute and operand".to_string(),
1098 )),
1099 }
1100 }
1101 SetValue::ListAppend(first, second) => {
1102 let a = resolve_value(first, parameters)?;
1103 let b = resolve_value(second, parameters)?;
1104 let list_a = match &a {
1107 AttributeValue::S(name) => resolve_nested_path(item, name)
1108 .cloned()
1109 .unwrap_or(AttributeValue::L(Vec::new())),
1110 other => other.clone(),
1111 };
1112 let list_b = match &b {
1113 AttributeValue::S(name) => resolve_nested_path(item, name)
1114 .cloned()
1115 .unwrap_or(AttributeValue::L(Vec::new())),
1116 other => other.clone(),
1117 };
1118 match (list_a, list_b) {
1119 (AttributeValue::L(mut la), AttributeValue::L(lb)) => {
1120 la.extend(lb);
1121 Ok(AttributeValue::L(la))
1122 }
1123 _ => Err(DynoxideError::ValidationException(
1124 "list_append requires list operands".to_string(),
1125 )),
1126 }
1127 }
1128 }
1129}
1130
1131fn invalid_update_path() -> DynoxideError {
1134 DynoxideError::ValidationException(
1135 "The document path provided in the update expression is invalid for update".to_string(),
1136 )
1137}
1138
1139fn set_nested_value(item: &mut Item, path: &str, val: AttributeValue) -> Result<()> {
1145 let segments = split_path_segments(path).ok_or_else(invalid_update_path)?;
1146 let (first, rest) = segments.split_first().ok_or_else(invalid_update_path)?;
1147 let key = match first {
1148 PathSegment::Key(k) => (*k).to_string(),
1149 PathSegment::Index(_) => return Err(invalid_update_path()),
1151 };
1152 if rest.is_empty() {
1153 item.insert(key, val);
1154 return Ok(());
1155 }
1156 let entry = item
1157 .entry(key)
1158 .or_insert_with(|| AttributeValue::M(HashMap::new()));
1159 set_into_value(entry, rest, val)
1160}
1161
1162fn set_into_value(
1165 current: &mut AttributeValue,
1166 segments: &[PathSegment],
1167 val: AttributeValue,
1168) -> Result<()> {
1169 let (seg, rest) = segments.split_first().expect("segments is non-empty");
1170 if rest.is_empty() {
1171 return match seg {
1172 PathSegment::Key(k) => match current {
1173 AttributeValue::M(map) => {
1174 map.insert((*k).to_string(), val);
1175 Ok(())
1176 }
1177 _ => Err(invalid_update_path()),
1178 },
1179 PathSegment::Index(i) => match current {
1180 AttributeValue::L(list) => {
1181 if *i < list.len() {
1182 list[*i] = val;
1183 } else {
1184 list.push(val);
1185 }
1186 Ok(())
1187 }
1188 _ => Err(invalid_update_path()),
1189 },
1190 };
1191 }
1192 match seg {
1193 PathSegment::Key(k) => match current {
1194 AttributeValue::M(map) => {
1195 let next = map
1196 .entry((*k).to_string())
1197 .or_insert_with(|| AttributeValue::M(HashMap::new()));
1198 set_into_value(next, rest, val)
1199 }
1200 _ => Err(invalid_update_path()),
1201 },
1202 PathSegment::Index(i) => match current {
1203 AttributeValue::L(list) => match list.get_mut(*i) {
1204 Some(next) => set_into_value(next, rest, val),
1205 None => Err(invalid_update_path()),
1206 },
1207 _ => Err(invalid_update_path()),
1208 },
1209 }
1210}
1211
1212fn remove_nested_value(item: &mut Item, path: &str) {
1216 let Some(segments) = split_path_segments(path) else {
1217 return;
1218 };
1219 let Some((first, rest)) = segments.split_first() else {
1220 return;
1221 };
1222 let PathSegment::Key(key) = first else {
1223 return; };
1225 if rest.is_empty() {
1226 item.remove(*key);
1227 return;
1228 }
1229 if let Some(current) = item.get_mut(*key) {
1230 remove_from_value(current, rest);
1231 }
1232}
1233
1234fn remove_from_value(current: &mut AttributeValue, segments: &[PathSegment]) {
1237 let (seg, rest) = segments.split_first().expect("segments is non-empty");
1238 if rest.is_empty() {
1239 match seg {
1240 PathSegment::Key(k) => {
1241 if let AttributeValue::M(map) = current {
1242 map.remove(*k);
1243 }
1244 }
1245 PathSegment::Index(i) => {
1246 if let AttributeValue::L(list) = current {
1247 if *i < list.len() {
1248 list.remove(*i);
1249 }
1250 }
1251 }
1252 }
1253 return;
1254 }
1255 match seg {
1256 PathSegment::Key(k) => {
1257 if let AttributeValue::M(map) = current {
1258 if let Some(next) = map.get_mut(*k) {
1259 remove_from_value(next, rest);
1260 }
1261 }
1262 }
1263 PathSegment::Index(i) => {
1264 if let AttributeValue::L(list) = current {
1265 if let Some(next) = list.get_mut(*i) {
1266 remove_from_value(next, rest);
1267 }
1268 }
1269 }
1270 }
1271}
1272
1273fn matches_where(
1275 item: &Item,
1276 where_clause: Option<&WhereClause>,
1277 parameters: &[AttributeValue],
1278) -> bool {
1279 let wc = match where_clause {
1280 Some(wc) => wc,
1281 None => return true,
1282 };
1283
1284 wc.groups
1286 .iter()
1287 .any(|group| matches_conditions(item, group, parameters))
1288}
1289
1290fn matches_conditions(
1292 item: &Item,
1293 conditions: &[WhereCondition],
1294 parameters: &[AttributeValue],
1295) -> bool {
1296 for cond in conditions {
1297 match cond {
1298 WhereCondition::Comparison(c) => {
1299 let item_val = match resolve_nested_path(item, &c.path) {
1300 Some(v) => v,
1301 None => return false,
1302 };
1303 let target = match resolve_value(&c.value, parameters) {
1304 Ok(v) => v,
1305 Err(_) => return false,
1306 };
1307 if !compare_values(item_val, &c.op, &target) {
1308 return false;
1309 }
1310 }
1311 WhereCondition::Exists(path) | WhereCondition::IsNotMissing(path) => {
1312 if resolve_nested_path(item, path).is_none() {
1313 return false;
1314 }
1315 }
1316 WhereCondition::NotExists(path) | WhereCondition::IsMissing(path) => {
1317 if resolve_nested_path(item, path).is_some() {
1318 return false;
1319 }
1320 }
1321 WhereCondition::BeginsWith(path, prefix_val) => {
1322 let item_val = match resolve_nested_path(item, path) {
1323 Some(v) => v,
1324 None => return false,
1325 };
1326 let prefix = match resolve_value(prefix_val, parameters) {
1327 Ok(v) => v,
1328 Err(_) => return false,
1329 };
1330 match (item_val, &prefix) {
1331 (AttributeValue::S(s), AttributeValue::S(p)) => {
1332 if !s.starts_with(p.as_str()) {
1333 return false;
1334 }
1335 }
1336 _ => return false,
1337 }
1338 }
1339 WhereCondition::NotBeginsWith(path, prefix_val) => {
1340 if let Some(item_val) = resolve_nested_path(item, path) {
1345 let prefix = match resolve_value(prefix_val, parameters) {
1346 Ok(v) => v,
1347 Err(_) => return false,
1348 };
1349 if let (AttributeValue::S(s), AttributeValue::S(p)) = (item_val, &prefix) {
1350 if s.starts_with(p.as_str()) {
1351 return false;
1352 }
1353 }
1354 }
1355 }
1356 WhereCondition::Between(path, low, high) => {
1357 let item_val = match resolve_nested_path(item, path) {
1358 Some(v) => v,
1359 None => return false,
1360 };
1361 let low_val = match resolve_value(low, parameters) {
1362 Ok(v) => v,
1363 Err(_) => return false,
1364 };
1365 let high_val = match resolve_value(high, parameters) {
1366 Ok(v) => v,
1367 Err(_) => return false,
1368 };
1369 if !compare_values(item_val, &CompOp::Ge, &low_val)
1370 || !compare_values(item_val, &CompOp::Le, &high_val)
1371 {
1372 return false;
1373 }
1374 }
1375 WhereCondition::In(path, values) => {
1376 let item_val = match resolve_nested_path(item, path) {
1377 Some(v) => v,
1378 None => return false,
1379 };
1380 let matched = values.iter().any(|v| {
1381 resolve_value(v, parameters)
1382 .map(|target| compare_values(item_val, &CompOp::Eq, &target))
1383 .unwrap_or(false)
1384 });
1385 if !matched {
1386 return false;
1387 }
1388 }
1389 WhereCondition::Contains(path, substr_val) => {
1390 let item_val = match resolve_nested_path(item, path) {
1391 Some(v) => v,
1392 None => return false,
1393 };
1394 let substr = match resolve_value(substr_val, parameters) {
1395 Ok(v) => v,
1396 Err(_) => return false,
1397 };
1398 match (item_val, &substr) {
1399 (AttributeValue::S(s), AttributeValue::S(sub)) => {
1400 if !s.contains(sub.as_str()) {
1401 return false;
1402 }
1403 }
1404 (AttributeValue::SS(set), AttributeValue::S(val)) => {
1405 if !set.contains(val) {
1406 return false;
1407 }
1408 }
1409 (AttributeValue::NS(set), AttributeValue::N(val)) => {
1410 if !set.contains(val) {
1411 return false;
1412 }
1413 }
1414 (AttributeValue::L(list), target) => {
1415 if !list.contains(target) {
1416 return false;
1417 }
1418 }
1419 _ => return false,
1420 }
1421 }
1422 }
1423 }
1424
1425 true
1426}
1427
1428fn resolve_nested_path<'a>(item: &'a Item, path: &str) -> Option<&'a AttributeValue> {
1432 if !path.contains('.') && !path.contains('[') {
1434 return item.get(path);
1435 }
1436
1437 let segments = split_path_segments(path)?;
1438 if segments.is_empty() {
1439 return None;
1440 }
1441
1442 let mut current = match &segments[0] {
1444 PathSegment::Key(k) => item.get(*k)?,
1445 PathSegment::Index(_) => return None,
1446 };
1447
1448 for seg in &segments[1..] {
1449 current = match seg {
1450 PathSegment::Key(k) => match current {
1451 AttributeValue::M(map) => map.get(*k)?,
1452 _ => return None,
1453 },
1454 PathSegment::Index(idx) => match current {
1455 AttributeValue::L(list) => list.get(*idx)?,
1456 _ => return None,
1457 },
1458 };
1459 }
1460
1461 Some(current)
1462}
1463
1464enum PathSegment<'a> {
1465 Key(&'a str),
1466 Index(usize),
1467}
1468
1469fn split_path_segments(path: &str) -> Option<Vec<PathSegment<'_>>> {
1472 let mut segments = Vec::new();
1473 let bytes = path.as_bytes();
1474 let mut start = 0;
1475 let mut i = 0;
1476
1477 while i < bytes.len() {
1478 match bytes[i] {
1479 b'.' => {
1480 if start < i {
1481 segments.push(PathSegment::Key(&path[start..i]));
1482 }
1483 i += 1;
1484 start = i;
1485 }
1486 b'[' => {
1487 if start < i {
1488 segments.push(PathSegment::Key(&path[start..i]));
1489 }
1490 i += 1;
1491 let idx_start = i;
1492 while i < bytes.len() && bytes[i] != b']' {
1493 i += 1;
1494 }
1495 let idx = path[idx_start..i].parse::<usize>().ok()?;
1496 segments.push(PathSegment::Index(idx));
1497 if i < bytes.len() {
1498 i += 1; }
1500 start = i;
1501 if i < bytes.len() && bytes[i] == b'.' {
1503 i += 1;
1504 start = i;
1505 }
1506 }
1507 _ => {
1508 i += 1;
1509 }
1510 }
1511 }
1512
1513 if start < bytes.len() {
1514 segments.push(PathSegment::Key(&path[start..]));
1515 }
1516
1517 Some(segments)
1518}
1519
1520fn compare_values(left: &AttributeValue, op: &CompOp, right: &AttributeValue) -> bool {
1522 match (left, right) {
1523 (AttributeValue::S(a), AttributeValue::S(b)) => compare_ord(a, op, b),
1524 (AttributeValue::N(a), AttributeValue::N(b)) => {
1525 use bigdecimal::BigDecimal;
1526 use std::str::FromStr;
1527 match (BigDecimal::from_str(a), BigDecimal::from_str(b)) {
1528 (Ok(da), Ok(db)) => compare_ord(&da, op, &db),
1529 _ => false,
1530 }
1531 }
1532 (AttributeValue::BOOL(a), AttributeValue::BOOL(b)) => match op {
1533 CompOp::Eq => a == b,
1534 CompOp::Ne => a != b,
1535 _ => false,
1536 },
1537 _ => match op {
1538 CompOp::Eq => false,
1539 CompOp::Ne => true,
1540 _ => false,
1541 },
1542 }
1543}
1544
1545fn format_bigdecimal(n: &bigdecimal::BigDecimal) -> String {
1547 let normalized = n.normalized();
1548 if normalized.as_bigint_and_exponent().1 < 0 {
1549 normalized.with_scale(0).to_string()
1550 } else {
1551 normalized.to_string()
1552 }
1553}
1554
1555fn compare_ord<T: PartialOrd>(a: &T, op: &CompOp, b: &T) -> bool {
1556 match op {
1557 CompOp::Eq => a == b,
1558 CompOp::Ne => a != b,
1559 CompOp::Lt => a < b,
1560 CompOp::Le => a <= b,
1561 CompOp::Gt => a > b,
1562 CompOp::Ge => a >= b,
1563 }
1564}