1use crate::errors::{DynoxideError, Result};
6use crate::partiql::parser::{
7 CompOp, PartiqlValue, ReturningVariant, SetValue, Statement, WhereClause, WhereCondition,
8};
9use crate::storage_backend::StorageBackend;
10use crate::types::{AttributeValue, Item};
11use std::collections::HashMap;
12
13pub async fn execute<S: StorageBackend>(
20 storage: &S,
21 stmt: &Statement,
22 parameters: &[AttributeValue],
23 limit: Option<usize>,
24) -> Result<Option<Vec<Item>>> {
25 Ok(execute_measured(storage, stmt, parameters, limit).await?.0)
26}
27
28pub async fn execute_measured<S: StorageBackend>(
33 storage: &S,
34 stmt: &Statement,
35 parameters: &[AttributeValue],
36 limit: Option<usize>,
37) -> Result<(Option<Vec<Item>>, usize)> {
38 match stmt {
39 Statement::Select {
40 table_name,
41 projections,
42 where_clause,
43 } => {
44 let items = execute_select(
45 storage,
46 table_name,
47 projections,
48 where_clause.as_ref(),
49 parameters,
50 limit,
51 )
52 .await?;
53 let size = items
54 .as_ref()
55 .map(|rows| rows.iter().map(crate::types::item_size).sum())
56 .unwrap_or(0);
57 Ok((items, size))
58 }
59 Statement::Insert {
60 table_name,
61 item,
62 if_not_exists,
63 } => {
64 let size =
65 execute_insert(storage, table_name, item, parameters, *if_not_exists).await?;
66 Ok((None, size))
67 }
68 Statement::Update {
69 table_name,
70 set_clauses,
71 remove_paths,
72 where_clause,
73 returning,
74 } => {
75 let (projection, size) = execute_update(
76 storage,
77 table_name,
78 set_clauses,
79 remove_paths,
80 where_clause.as_ref(),
81 parameters,
82 *returning,
83 )
84 .await?;
85 let items = projection.map(|item| {
90 if item.is_empty() {
91 Vec::new()
92 } else {
93 vec![item]
94 }
95 });
96 Ok((items, size))
97 }
98 Statement::Delete {
99 table_name,
100 where_clause,
101 returning,
102 } => {
103 if let Some(variant) = returning {
107 if *variant != ReturningVariant::AllOld {
108 return Err(DynoxideError::ValidationException(format!(
109 "Invalid returning clause: RETURNING {} *. Only RETURNING ALL OLD * is allowed in DELETE statements.",
110 variant.as_sql()
111 )));
112 }
113 }
114 let (old_item, size) =
115 execute_delete(storage, table_name, where_clause.as_ref(), parameters).await?;
116 let items = if returning.is_some() {
121 Some(old_item.map(|item| vec![item]).unwrap_or_default())
122 } else {
123 None
124 };
125 Ok((items, size))
126 }
127 }
128}
129
130fn insert_nested_projection(result: &mut Item, path: &str, val: AttributeValue) {
136 let parts: Vec<&str> = path.split('.').collect();
137 let key = parts.last().unwrap();
139 result.insert(key.to_string(), val);
140}
141
142async fn execute_select<S: StorageBackend>(
143 storage: &S,
144 table_name: &str,
145 projections: &[String],
146 where_clause: Option<&WhereClause>,
147 parameters: &[AttributeValue],
148 limit: Option<usize>,
149) -> Result<Option<Vec<Item>>> {
150 let meta = require_table(storage, table_name).await?;
151 let key_schema = crate::actions::helpers::parse_key_schema(&meta)?;
152
153 if projections.len() == 1 && projections[0] == "COUNT(*)" {
155 let items = collect_matching_items(
156 storage,
157 table_name,
158 where_clause,
159 parameters,
160 &key_schema,
161 None,
162 )
163 .await?;
164 let count = items.len();
165 let mut result = HashMap::new();
166 result.insert("Count".to_string(), AttributeValue::N(count.to_string()));
167 return Ok(Some(vec![result]));
168 }
169
170 let items = collect_matching_items(
171 storage,
172 table_name,
173 where_clause,
174 parameters,
175 &key_schema,
176 limit,
177 )
178 .await?;
179
180 let items = if projections.is_empty() {
182 items
183 } else {
184 items
185 .into_iter()
186 .map(|item| {
187 let mut projected = HashMap::new();
188 for proj in projections {
189 if let Some(val) = resolve_nested_path(&item, proj) {
190 insert_nested_projection(&mut projected, proj, val.clone());
191 }
192 }
193 projected
194 })
195 .collect()
196 };
197
198 Ok(Some(items))
199}
200
201async fn collect_matching_items<S: StorageBackend>(
203 storage: &S,
204 table_name: &str,
205 where_clause: Option<&WhereClause>,
206 parameters: &[AttributeValue],
207 key_schema: &crate::actions::helpers::KeySchema,
208 limit: Option<usize>,
209) -> Result<Vec<Item>> {
210 let pk_condition = where_clause.and_then(|wc| find_pk_condition(wc, &key_schema.partition_key));
212
213 let items: Vec<Item> = if let Some(pk_cond) = pk_condition {
214 let pk_val = resolve_value(&pk_cond.value, parameters)?;
215 let pk_str = pk_val
216 .to_key_string()
217 .ok_or_else(|| DynoxideError::ValidationException("Invalid key value".to_string()))?;
218
219 let rows = storage
220 .query_items(table_name, &pk_str, &Default::default())
221 .await?;
222
223 let iter = rows
224 .into_iter()
225 .filter_map(|(_, _, json)| serde_json::from_str::<Item>(&json).ok())
226 .filter(|item| matches_where(item, where_clause, parameters));
227
228 if let Some(lim) = limit {
229 iter.take(lim).collect()
230 } else {
231 iter.collect()
232 }
233 } else {
234 let rows = storage.scan_items(table_name, &Default::default()).await?;
235
236 let iter = rows
237 .into_iter()
238 .filter_map(|(_, _, json)| serde_json::from_str::<Item>(&json).ok())
239 .filter(|item| matches_where(item, where_clause, parameters));
240
241 if let Some(lim) = limit {
242 iter.take(lim).collect()
243 } else {
244 iter.collect()
245 }
246 };
247
248 Ok(items)
249}
250
251fn find_pk_condition<'a>(
253 wc: &'a WhereClause,
254 pk_name: &str,
255) -> Option<&'a crate::partiql::parser::Condition> {
256 if wc.groups.len() == 1 {
259 wc.groups[0].iter().find_map(|c| match c {
260 WhereCondition::Comparison(cond) if cond.path == pk_name && cond.op == CompOp::Eq => {
261 Some(cond)
262 }
263 _ => None,
264 })
265 } else {
266 None
267 }
268}
269
270async fn execute_insert<S: StorageBackend>(
273 storage: &S,
274 table_name: &str,
275 item_template: &HashMap<String, PartiqlValue>,
276 parameters: &[AttributeValue],
277 if_not_exists: bool,
278) -> Result<usize> {
279 let mut item = HashMap::new();
281 for (k, v) in item_template {
282 let resolved = match v {
283 PartiqlValue::Literal(av) => av.clone(),
284 PartiqlValue::Parameter(idx) => parameters.get(*idx).cloned().ok_or_else(|| {
285 DynoxideError::ValidationException(format!(
286 "Parameter index {idx} out of range (have {} parameters)",
287 parameters.len()
288 ))
289 })?,
290 };
291 item.insert(k.clone(), resolved);
292 }
293
294 let meta = require_table(storage, table_name).await?;
295 let key_schema = crate::actions::helpers::parse_key_schema(&meta)?;
296
297 crate::actions::helpers::validate_item_keys(&item, &key_schema, &meta)?;
299 crate::validation::validate_item_attribute_values(&item)?;
300
301 crate::validation::normalize_item_sets(&mut item);
303
304 let (pk, sk) = crate::actions::helpers::extract_key_strings(&item, &key_schema)?;
306
307 let existing = storage.get_item(table_name, &pk, &sk).await?;
309 if existing.is_some() {
310 if if_not_exists {
311 return Ok(0);
313 }
314 return Err(DynoxideError::DuplicateItemException(
315 "Duplicate primary key exists in table".to_string(),
316 ));
317 }
318
319 let item_json = serde_json::to_string(&item)
320 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
321 let item_size = crate::types::item_size(&item);
322
323 let hash_prefix = item
324 .get(&key_schema.partition_key)
325 .map(crate::storage::compute_hash_prefix)
326 .unwrap_or_default();
327 let old_json = storage
328 .put_item_with_hash(table_name, &pk, &sk, &item_json, item_size, &hash_prefix)
329 .await?;
330
331 let table_sk_attr = key_schema.sort_key.as_deref();
333 let _ = crate::actions::gsi::maintain_gsis_after_write(
334 storage,
335 table_name,
336 &meta,
337 &pk,
338 &sk,
339 &item,
340 &key_schema.partition_key,
341 table_sk_attr,
342 )
343 .await?;
344
345 crate::actions::lsi::maintain_lsis_after_write(
347 storage,
348 table_name,
349 &meta,
350 &pk,
351 &sk,
352 &item,
353 &key_schema.partition_key,
354 table_sk_attr,
355 )
356 .await?;
357
358 let old_item: Option<Item> = old_json.as_ref().and_then(|j| serde_json::from_str(j).ok());
360 crate::streams::record_stream_event(storage, &meta, old_item.as_ref(), Some(&item)).await?;
361
362 Ok(item_size)
363}
364
365async fn execute_update<S: StorageBackend>(
370 storage: &S,
371 table_name: &str,
372 set_clauses: &[crate::partiql::parser::SetClause],
373 remove_paths: &[String],
374 where_clause: Option<&WhereClause>,
375 parameters: &[AttributeValue],
376 returning: Option<ReturningVariant>,
377) -> Result<(Option<Item>, usize)> {
378 let meta = require_table(storage, table_name).await?;
379 let key_schema = crate::actions::helpers::parse_key_schema(&meta)?;
380
381 let wc = where_clause.ok_or_else(|| {
383 DynoxideError::ValidationException("UPDATE requires a WHERE clause".to_string())
384 })?;
385
386 if wc.groups.len() > 1 {
388 return Err(DynoxideError::ValidationException(
389 "UPDATE does not support OR conditions in WHERE clause".to_string(),
390 ));
391 }
392
393 let pk_cond =
395 find_comparison_in_groups(&wc.groups, &key_schema.partition_key).ok_or_else(|| {
396 DynoxideError::ValidationException(
397 "Where clause does not contain a mandatory equality on all key attributes"
398 .to_string(),
399 )
400 })?;
401
402 let pk_val = resolve_value(&pk_cond.value, parameters)?;
403 let pk_str = pk_val
404 .to_key_string()
405 .ok_or_else(|| DynoxideError::ValidationException("Invalid key value".to_string()))?;
406
407 let sk_str = if let Some(ref sk_name) = key_schema.sort_key {
408 let sk_cond = find_comparison_in_groups(&wc.groups, sk_name);
409 if sk_cond.is_none() {
410 return Err(DynoxideError::ValidationException(
411 "Where clause does not contain a mandatory equality on all key attributes"
412 .to_string(),
413 ));
414 }
415 sk_cond
416 .map(|c| resolve_value(&c.value, parameters))
417 .transpose()?
418 .and_then(|v| v.to_key_string())
419 .unwrap_or_default()
420 } else {
421 String::new()
422 };
423
424 let existing_json = storage.get_item(table_name, &pk_str, &sk_str).await?;
426 let mut item: Item = existing_json
427 .as_ref()
428 .and_then(|j| serde_json::from_str(j).ok())
429 .unwrap_or_default();
430
431 let old_item = item.clone();
432
433 if existing_json.is_none() || !matches_where(&old_item, where_clause, parameters) {
438 return Err(DynoxideError::ConditionalCheckFailedException(
439 "The conditional request failed".to_string(),
440 None,
441 ));
442 }
443
444 let before_item = item.clone();
445
446 for clause in set_clauses {
448 let val = resolve_set_value(&clause.value, &item, parameters)?;
449 set_nested_value(&mut item, &clause.path, val)?;
450 }
451
452 for path in remove_paths {
454 remove_nested_value(&mut item, path);
455 }
456
457 if item.is_empty() {
459 return Ok((None, 0));
460 }
461
462 crate::validation::validate_item_attribute_values(&item)?;
464 crate::validation::normalize_item_sets(&mut item);
465
466 crate::actions::helpers::validate_updated_index_keys(&before_item, &item, &meta)?;
468
469 let item_json = serde_json::to_string(&item)
470 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
471 let item_size = crate::types::item_size(&item);
472
473 let hash_prefix = item
474 .get(&key_schema.partition_key)
475 .map(crate::storage::compute_hash_prefix)
476 .unwrap_or_default();
477 storage
478 .put_item_with_hash(
479 table_name,
480 &pk_str,
481 &sk_str,
482 &item_json,
483 item_size,
484 &hash_prefix,
485 )
486 .await?;
487
488 let table_sk_attr = key_schema.sort_key.as_deref();
490 let _ = crate::actions::gsi::maintain_gsis_after_write(
491 storage,
492 table_name,
493 &meta,
494 &pk_str,
495 &sk_str,
496 &item,
497 &key_schema.partition_key,
498 table_sk_attr,
499 )
500 .await?;
501
502 crate::actions::lsi::maintain_lsis_after_write(
504 storage,
505 table_name,
506 &meta,
507 &pk_str,
508 &sk_str,
509 &item,
510 &key_schema.partition_key,
511 table_sk_attr,
512 )
513 .await?;
514
515 let old_ref = if existing_json.is_some() {
517 Some(&old_item)
518 } else {
519 None
520 };
521 crate::streams::record_stream_event(storage, &meta, old_ref, Some(&item)).await?;
522
523 let projection = returning.map(|variant| {
528 let modified: std::collections::BTreeSet<String> = set_clauses
529 .iter()
530 .map(|c| c.path.clone())
531 .chain(remove_paths.iter().cloned())
532 .collect();
533 project_returning(variant, &old_item, &item, &modified)
534 });
535
536 Ok((projection, item_size))
537}
538
539fn project_returning(
547 variant: ReturningVariant,
548 old_item: &Item,
549 new_item: &Item,
550 modified: &std::collections::BTreeSet<String>,
551) -> Item {
552 match variant {
553 ReturningVariant::AllOld => old_item.clone(),
554 ReturningVariant::AllNew => new_item.clone(),
555 ReturningVariant::ModifiedOld => project_modified(modified, old_item),
556 ReturningVariant::ModifiedNew => project_modified(modified, new_item),
557 }
558}
559
560enum ProjNode {
565 Leaf(AttributeValue),
566 Map(HashMap<String, ProjNode>),
567 List(std::collections::BTreeMap<usize, ProjNode>),
568}
569
570fn project_modified(paths: &std::collections::BTreeSet<String>, source: &Item) -> Item {
580 let mut root: HashMap<String, ProjNode> = HashMap::new();
581 for path in paths {
582 if let (Some(val), Some(segments)) =
583 (resolve_nested_path(source, path), split_path_segments(path))
584 {
585 if let Some((PathSegment::Key(key), rest)) = segments.split_first() {
587 let node = root
588 .entry((*key).to_string())
589 .or_insert_with(|| fresh_proj_node(rest));
590 insert_proj_node(node, rest, val.clone());
591 }
592 }
593 }
594 root.into_iter()
595 .map(|(k, node)| (k, proj_node_to_value(node)))
596 .collect()
597}
598
599fn fresh_proj_node(segments: &[PathSegment]) -> ProjNode {
603 match segments.first() {
604 Some(PathSegment::Index(_)) => ProjNode::List(std::collections::BTreeMap::new()),
605 _ => ProjNode::Map(HashMap::new()),
606 }
607}
608
609fn insert_proj_node(node: &mut ProjNode, segments: &[PathSegment], val: AttributeValue) {
612 let Some((seg, rest)) = segments.split_first() else {
613 *node = ProjNode::Leaf(val);
614 return;
615 };
616 match seg {
617 PathSegment::Key(k) => {
618 if let ProjNode::Map(map) = node {
619 let child = map
620 .entry((*k).to_string())
621 .or_insert_with(|| fresh_proj_node(rest));
622 insert_proj_node(child, rest, val);
623 }
624 }
625 PathSegment::Index(i) => {
626 if let ProjNode::List(list) = node {
627 let child = list.entry(*i).or_insert_with(|| fresh_proj_node(rest));
628 insert_proj_node(child, rest, val);
629 }
630 }
631 }
632}
633
634fn proj_node_to_value(node: ProjNode) -> AttributeValue {
637 match node {
638 ProjNode::Leaf(v) => v,
639 ProjNode::Map(map) => AttributeValue::M(
640 map.into_iter()
641 .map(|(k, n)| (k, proj_node_to_value(n)))
642 .collect(),
643 ),
644 ProjNode::List(list) => {
645 AttributeValue::L(list.into_values().map(proj_node_to_value).collect())
646 }
647 }
648}
649
650async fn execute_delete<S: StorageBackend>(
653 storage: &S,
654 table_name: &str,
655 where_clause: Option<&WhereClause>,
656 parameters: &[AttributeValue],
657) -> Result<(Option<Item>, usize)> {
658 let meta = require_table(storage, table_name).await?;
659 let key_schema = crate::actions::helpers::parse_key_schema(&meta)?;
660
661 let wc = where_clause.ok_or_else(|| {
662 DynoxideError::ValidationException("DELETE requires a WHERE clause".to_string())
663 })?;
664
665 if wc.groups.len() > 1 {
667 return Err(DynoxideError::ValidationException(
668 "DELETE does not support OR conditions in WHERE clause".to_string(),
669 ));
670 }
671
672 let pk_cond =
673 find_comparison_in_groups(&wc.groups, &key_schema.partition_key).ok_or_else(|| {
674 DynoxideError::ValidationException(
675 "Where clause does not contain a mandatory equality on all key attributes"
676 .to_string(),
677 )
678 })?;
679
680 let pk_val = resolve_value(&pk_cond.value, parameters)?;
681 let pk_str = pk_val
682 .to_key_string()
683 .ok_or_else(|| DynoxideError::ValidationException("Invalid key value".to_string()))?;
684
685 if let Some(ref sk_name) = key_schema.sort_key {
687 let has_sk_condition = wc.groups.iter().any(|group| {
688 group.iter().any(|c| match c {
689 WhereCondition::Comparison(comp) => comp.path == *sk_name && comp.op == CompOp::Eq,
690 _ => false,
691 })
692 });
693 if !has_sk_condition {
694 return Err(DynoxideError::ValidationException(
695 "Where clause does not contain a mandatory equality on all key attributes"
696 .to_string(),
697 ));
698 }
699 }
700
701 let sk_str = if let Some(ref sk_name) = key_schema.sort_key {
702 find_comparison_in_groups(&wc.groups, sk_name)
703 .map(|c| resolve_value(&c.value, parameters))
704 .transpose()?
705 .and_then(|v| v.to_key_string())
706 .unwrap_or_default()
707 } else {
708 String::new()
709 };
710
711 if let Some(json) = storage.get_item(table_name, &pk_str, &sk_str).await? {
718 let existing: Item = serde_json::from_str(&json)
719 .map_err(|e| DynoxideError::InternalServerError(format!("Bad item JSON: {e}")))?;
720 if !matches_where(&existing, where_clause, parameters) {
721 return Err(DynoxideError::ConditionalCheckFailedException(
722 "The conditional request failed".to_string(),
723 None,
724 ));
725 }
726 }
727
728 let old_json = storage.delete_item(table_name, &pk_str, &sk_str).await?;
729
730 let _ = crate::actions::gsi::maintain_gsis_after_delete(
732 storage, table_name, &meta, &pk_str, &sk_str,
733 )
734 .await?;
735
736 crate::actions::lsi::maintain_lsis_after_delete(storage, table_name, &meta, &pk_str, &sk_str)
738 .await?;
739
740 let old_item: Option<Item> = old_json.as_ref().and_then(|j| serde_json::from_str(j).ok());
742 if old_item.is_some() {
743 crate::streams::record_stream_event(storage, &meta, old_item.as_ref(), None).await?;
744 }
745
746 let deleted_size = old_item.as_ref().map(crate::types::item_size).unwrap_or(0);
749 Ok((old_item, deleted_size))
750}
751
752async fn require_table<S: StorageBackend>(
757 storage: &S,
758 table_name: &str,
759) -> Result<crate::storage::TableMetadata> {
760 crate::actions::helpers::require_table(storage, table_name).await
761}
762
763fn find_comparison_in_groups<'a>(
766 groups: &'a [Vec<WhereCondition>],
767 path: &str,
768) -> Option<&'a crate::partiql::parser::Condition> {
769 for group in groups {
770 if let Some(cond) = find_comparison(group, path) {
771 return Some(cond);
772 }
773 }
774 None
775}
776
777fn find_comparison<'a>(
779 conditions: &'a [WhereCondition],
780 path: &str,
781) -> Option<&'a crate::partiql::parser::Condition> {
782 conditions.iter().find_map(|c| match c {
783 WhereCondition::Comparison(cond) if cond.path == path && cond.op == CompOp::Eq => {
784 Some(cond)
785 }
786 _ => None,
787 })
788}
789
790fn resolve_value(val: &PartiqlValue, parameters: &[AttributeValue]) -> Result<AttributeValue> {
792 match val {
793 PartiqlValue::Literal(av) => Ok(av.clone()),
794 PartiqlValue::Parameter(idx) => parameters.get(*idx).cloned().ok_or_else(|| {
795 DynoxideError::ValidationException(format!(
796 "Parameter index {idx} out of range (have {} parameters)",
797 parameters.len()
798 ))
799 }),
800 }
801}
802
803fn resolve_set_value(
805 val: &SetValue,
806 item: &Item,
807 parameters: &[AttributeValue],
808) -> Result<AttributeValue> {
809 match val {
810 SetValue::Simple(pv) => resolve_value(pv, parameters),
811 SetValue::Add(attr, pv) => {
812 let current = resolve_nested_path(item, attr);
813 let operand = resolve_value(pv, parameters)?;
814 match (current, &operand) {
815 (Some(AttributeValue::N(cur)), AttributeValue::N(add)) => {
816 use bigdecimal::BigDecimal;
817 use std::str::FromStr;
818 let a = BigDecimal::from_str(cur).map_err(|e| {
819 DynoxideError::ValidationException(format!("Invalid number: {e}"))
820 })?;
821 let b = BigDecimal::from_str(add).map_err(|e| {
822 DynoxideError::ValidationException(format!("Invalid number: {e}"))
823 })?;
824 let result = a + b;
825 Ok(AttributeValue::N(format_bigdecimal(&result)))
826 }
827 (None, AttributeValue::N(_)) => {
828 Ok(operand)
830 }
831 _ => Err(DynoxideError::ValidationException(
832 "SET expression add requires numeric attribute and operand".to_string(),
833 )),
834 }
835 }
836 SetValue::Sub(attr, pv) => {
837 let current = resolve_nested_path(item, attr);
838 let operand = resolve_value(pv, parameters)?;
839 match (current, &operand) {
840 (Some(AttributeValue::N(cur)), AttributeValue::N(sub)) => {
841 use bigdecimal::BigDecimal;
842 use std::str::FromStr;
843 let a = BigDecimal::from_str(cur).map_err(|e| {
844 DynoxideError::ValidationException(format!("Invalid number: {e}"))
845 })?;
846 let b = BigDecimal::from_str(sub).map_err(|e| {
847 DynoxideError::ValidationException(format!("Invalid number: {e}"))
848 })?;
849 let result = a - b;
850 Ok(AttributeValue::N(format_bigdecimal(&result)))
851 }
852 (None, AttributeValue::N(sub)) => {
853 use bigdecimal::BigDecimal;
855 use std::str::FromStr;
856 let b = BigDecimal::from_str(sub).map_err(|e| {
857 DynoxideError::ValidationException(format!("Invalid number: {e}"))
858 })?;
859 let result = -b;
860 Ok(AttributeValue::N(format_bigdecimal(&result)))
861 }
862 _ => Err(DynoxideError::ValidationException(
863 "SET expression subtract requires numeric attribute and operand".to_string(),
864 )),
865 }
866 }
867 SetValue::ListAppend(first, second) => {
868 let a = resolve_value(first, parameters)?;
869 let b = resolve_value(second, parameters)?;
870 let list_a = match &a {
873 AttributeValue::S(name) => resolve_nested_path(item, name)
874 .cloned()
875 .unwrap_or(AttributeValue::L(Vec::new())),
876 other => other.clone(),
877 };
878 let list_b = match &b {
879 AttributeValue::S(name) => resolve_nested_path(item, name)
880 .cloned()
881 .unwrap_or(AttributeValue::L(Vec::new())),
882 other => other.clone(),
883 };
884 match (list_a, list_b) {
885 (AttributeValue::L(mut la), AttributeValue::L(lb)) => {
886 la.extend(lb);
887 Ok(AttributeValue::L(la))
888 }
889 _ => Err(DynoxideError::ValidationException(
890 "list_append requires list operands".to_string(),
891 )),
892 }
893 }
894 }
895}
896
897fn invalid_update_path() -> DynoxideError {
900 DynoxideError::ValidationException(
901 "The document path provided in the update expression is invalid for update".to_string(),
902 )
903}
904
905fn set_nested_value(item: &mut Item, path: &str, val: AttributeValue) -> Result<()> {
911 let segments = split_path_segments(path).ok_or_else(invalid_update_path)?;
912 let (first, rest) = segments.split_first().ok_or_else(invalid_update_path)?;
913 let key = match first {
914 PathSegment::Key(k) => (*k).to_string(),
915 PathSegment::Index(_) => return Err(invalid_update_path()),
917 };
918 if rest.is_empty() {
919 item.insert(key, val);
920 return Ok(());
921 }
922 let entry = item
923 .entry(key)
924 .or_insert_with(|| AttributeValue::M(HashMap::new()));
925 set_into_value(entry, rest, val)
926}
927
928fn set_into_value(
931 current: &mut AttributeValue,
932 segments: &[PathSegment],
933 val: AttributeValue,
934) -> Result<()> {
935 let (seg, rest) = segments.split_first().expect("segments is non-empty");
936 if rest.is_empty() {
937 return match seg {
938 PathSegment::Key(k) => match current {
939 AttributeValue::M(map) => {
940 map.insert((*k).to_string(), val);
941 Ok(())
942 }
943 _ => Err(invalid_update_path()),
944 },
945 PathSegment::Index(i) => match current {
946 AttributeValue::L(list) => {
947 if *i < list.len() {
948 list[*i] = val;
949 } else {
950 list.push(val);
951 }
952 Ok(())
953 }
954 _ => Err(invalid_update_path()),
955 },
956 };
957 }
958 match seg {
959 PathSegment::Key(k) => match current {
960 AttributeValue::M(map) => {
961 let next = map
962 .entry((*k).to_string())
963 .or_insert_with(|| AttributeValue::M(HashMap::new()));
964 set_into_value(next, rest, val)
965 }
966 _ => Err(invalid_update_path()),
967 },
968 PathSegment::Index(i) => match current {
969 AttributeValue::L(list) => match list.get_mut(*i) {
970 Some(next) => set_into_value(next, rest, val),
971 None => Err(invalid_update_path()),
972 },
973 _ => Err(invalid_update_path()),
974 },
975 }
976}
977
978fn remove_nested_value(item: &mut Item, path: &str) {
982 let Some(segments) = split_path_segments(path) else {
983 return;
984 };
985 let Some((first, rest)) = segments.split_first() else {
986 return;
987 };
988 let PathSegment::Key(key) = first else {
989 return; };
991 if rest.is_empty() {
992 item.remove(*key);
993 return;
994 }
995 if let Some(current) = item.get_mut(*key) {
996 remove_from_value(current, rest);
997 }
998}
999
1000fn remove_from_value(current: &mut AttributeValue, segments: &[PathSegment]) {
1003 let (seg, rest) = segments.split_first().expect("segments is non-empty");
1004 if rest.is_empty() {
1005 match seg {
1006 PathSegment::Key(k) => {
1007 if let AttributeValue::M(map) = current {
1008 map.remove(*k);
1009 }
1010 }
1011 PathSegment::Index(i) => {
1012 if let AttributeValue::L(list) = current {
1013 if *i < list.len() {
1014 list.remove(*i);
1015 }
1016 }
1017 }
1018 }
1019 return;
1020 }
1021 match seg {
1022 PathSegment::Key(k) => {
1023 if let AttributeValue::M(map) = current {
1024 if let Some(next) = map.get_mut(*k) {
1025 remove_from_value(next, rest);
1026 }
1027 }
1028 }
1029 PathSegment::Index(i) => {
1030 if let AttributeValue::L(list) = current {
1031 if let Some(next) = list.get_mut(*i) {
1032 remove_from_value(next, rest);
1033 }
1034 }
1035 }
1036 }
1037}
1038
1039fn matches_where(
1041 item: &Item,
1042 where_clause: Option<&WhereClause>,
1043 parameters: &[AttributeValue],
1044) -> bool {
1045 let wc = match where_clause {
1046 Some(wc) => wc,
1047 None => return true,
1048 };
1049
1050 wc.groups
1052 .iter()
1053 .any(|group| matches_conditions(item, group, parameters))
1054}
1055
1056fn matches_conditions(
1058 item: &Item,
1059 conditions: &[WhereCondition],
1060 parameters: &[AttributeValue],
1061) -> bool {
1062 for cond in conditions {
1063 match cond {
1064 WhereCondition::Comparison(c) => {
1065 let item_val = match resolve_nested_path(item, &c.path) {
1066 Some(v) => v,
1067 None => return false,
1068 };
1069 let target = match resolve_value(&c.value, parameters) {
1070 Ok(v) => v,
1071 Err(_) => return false,
1072 };
1073 if !compare_values(item_val, &c.op, &target) {
1074 return false;
1075 }
1076 }
1077 WhereCondition::Exists(path) | WhereCondition::IsNotMissing(path) => {
1078 if resolve_nested_path(item, path).is_none() {
1079 return false;
1080 }
1081 }
1082 WhereCondition::NotExists(path) | WhereCondition::IsMissing(path) => {
1083 if resolve_nested_path(item, path).is_some() {
1084 return false;
1085 }
1086 }
1087 WhereCondition::BeginsWith(path, prefix_val) => {
1088 let item_val = match resolve_nested_path(item, path) {
1089 Some(v) => v,
1090 None => return false,
1091 };
1092 let prefix = match resolve_value(prefix_val, parameters) {
1093 Ok(v) => v,
1094 Err(_) => return false,
1095 };
1096 match (item_val, &prefix) {
1097 (AttributeValue::S(s), AttributeValue::S(p)) => {
1098 if !s.starts_with(p.as_str()) {
1099 return false;
1100 }
1101 }
1102 _ => return false,
1103 }
1104 }
1105 WhereCondition::NotBeginsWith(path, prefix_val) => {
1106 if let Some(item_val) = resolve_nested_path(item, path) {
1111 let prefix = match resolve_value(prefix_val, parameters) {
1112 Ok(v) => v,
1113 Err(_) => return false,
1114 };
1115 if let (AttributeValue::S(s), AttributeValue::S(p)) = (item_val, &prefix) {
1116 if s.starts_with(p.as_str()) {
1117 return false;
1118 }
1119 }
1120 }
1121 }
1122 WhereCondition::Between(path, low, high) => {
1123 let item_val = match resolve_nested_path(item, path) {
1124 Some(v) => v,
1125 None => return false,
1126 };
1127 let low_val = match resolve_value(low, parameters) {
1128 Ok(v) => v,
1129 Err(_) => return false,
1130 };
1131 let high_val = match resolve_value(high, parameters) {
1132 Ok(v) => v,
1133 Err(_) => return false,
1134 };
1135 if !compare_values(item_val, &CompOp::Ge, &low_val)
1136 || !compare_values(item_val, &CompOp::Le, &high_val)
1137 {
1138 return false;
1139 }
1140 }
1141 WhereCondition::In(path, values) => {
1142 let item_val = match resolve_nested_path(item, path) {
1143 Some(v) => v,
1144 None => return false,
1145 };
1146 let matched = values.iter().any(|v| {
1147 resolve_value(v, parameters)
1148 .map(|target| compare_values(item_val, &CompOp::Eq, &target))
1149 .unwrap_or(false)
1150 });
1151 if !matched {
1152 return false;
1153 }
1154 }
1155 WhereCondition::Contains(path, substr_val) => {
1156 let item_val = match resolve_nested_path(item, path) {
1157 Some(v) => v,
1158 None => return false,
1159 };
1160 let substr = match resolve_value(substr_val, parameters) {
1161 Ok(v) => v,
1162 Err(_) => return false,
1163 };
1164 match (item_val, &substr) {
1165 (AttributeValue::S(s), AttributeValue::S(sub)) => {
1166 if !s.contains(sub.as_str()) {
1167 return false;
1168 }
1169 }
1170 (AttributeValue::SS(set), AttributeValue::S(val)) => {
1171 if !set.contains(val) {
1172 return false;
1173 }
1174 }
1175 (AttributeValue::NS(set), AttributeValue::N(val)) => {
1176 if !set.contains(val) {
1177 return false;
1178 }
1179 }
1180 (AttributeValue::L(list), target) => {
1181 if !list.contains(target) {
1182 return false;
1183 }
1184 }
1185 _ => return false,
1186 }
1187 }
1188 }
1189 }
1190
1191 true
1192}
1193
1194fn resolve_nested_path<'a>(item: &'a Item, path: &str) -> Option<&'a AttributeValue> {
1198 if !path.contains('.') && !path.contains('[') {
1200 return item.get(path);
1201 }
1202
1203 let segments = split_path_segments(path)?;
1204 if segments.is_empty() {
1205 return None;
1206 }
1207
1208 let mut current = match &segments[0] {
1210 PathSegment::Key(k) => item.get(*k)?,
1211 PathSegment::Index(_) => return None,
1212 };
1213
1214 for seg in &segments[1..] {
1215 current = match seg {
1216 PathSegment::Key(k) => match current {
1217 AttributeValue::M(map) => map.get(*k)?,
1218 _ => return None,
1219 },
1220 PathSegment::Index(idx) => match current {
1221 AttributeValue::L(list) => list.get(*idx)?,
1222 _ => return None,
1223 },
1224 };
1225 }
1226
1227 Some(current)
1228}
1229
1230enum PathSegment<'a> {
1231 Key(&'a str),
1232 Index(usize),
1233}
1234
1235fn split_path_segments(path: &str) -> Option<Vec<PathSegment<'_>>> {
1238 let mut segments = Vec::new();
1239 let bytes = path.as_bytes();
1240 let mut start = 0;
1241 let mut i = 0;
1242
1243 while i < bytes.len() {
1244 match bytes[i] {
1245 b'.' => {
1246 if start < i {
1247 segments.push(PathSegment::Key(&path[start..i]));
1248 }
1249 i += 1;
1250 start = i;
1251 }
1252 b'[' => {
1253 if start < i {
1254 segments.push(PathSegment::Key(&path[start..i]));
1255 }
1256 i += 1;
1257 let idx_start = i;
1258 while i < bytes.len() && bytes[i] != b']' {
1259 i += 1;
1260 }
1261 let idx = path[idx_start..i].parse::<usize>().ok()?;
1262 segments.push(PathSegment::Index(idx));
1263 if i < bytes.len() {
1264 i += 1; }
1266 start = i;
1267 if i < bytes.len() && bytes[i] == b'.' {
1269 i += 1;
1270 start = i;
1271 }
1272 }
1273 _ => {
1274 i += 1;
1275 }
1276 }
1277 }
1278
1279 if start < bytes.len() {
1280 segments.push(PathSegment::Key(&path[start..]));
1281 }
1282
1283 Some(segments)
1284}
1285
1286fn compare_values(left: &AttributeValue, op: &CompOp, right: &AttributeValue) -> bool {
1288 match (left, right) {
1289 (AttributeValue::S(a), AttributeValue::S(b)) => compare_ord(a, op, b),
1290 (AttributeValue::N(a), AttributeValue::N(b)) => {
1291 use bigdecimal::BigDecimal;
1292 use std::str::FromStr;
1293 match (BigDecimal::from_str(a), BigDecimal::from_str(b)) {
1294 (Ok(da), Ok(db)) => compare_ord(&da, op, &db),
1295 _ => false,
1296 }
1297 }
1298 (AttributeValue::BOOL(a), AttributeValue::BOOL(b)) => match op {
1299 CompOp::Eq => a == b,
1300 CompOp::Ne => a != b,
1301 _ => false,
1302 },
1303 _ => match op {
1304 CompOp::Eq => false,
1305 CompOp::Ne => true,
1306 _ => false,
1307 },
1308 }
1309}
1310
1311fn format_bigdecimal(n: &bigdecimal::BigDecimal) -> String {
1313 let normalized = n.normalized();
1314 if normalized.as_bigint_and_exponent().1 < 0 {
1315 normalized.with_scale(0).to_string()
1316 } else {
1317 normalized.to_string()
1318 }
1319}
1320
1321fn compare_ord<T: PartialOrd>(a: &T, op: &CompOp, b: &T) -> bool {
1322 match op {
1323 CompOp::Eq => a == b,
1324 CompOp::Ne => a != b,
1325 CompOp::Lt => a < b,
1326 CompOp::Le => a <= b,
1327 CompOp::Gt => a > b,
1328 CompOp::Ge => a >= b,
1329 }
1330}