use crate::errors::{DynoxideError, Result};
use crate::expressions::key_condition::{ResolvedSortKeyCondition, sk_conditions_to_sql};
use crate::partiql::parser::{
CompOp, PartiqlValue, ReturningVariant, SetValue, Statement, WhereClause, WhereCondition,
};
use crate::storage_backend::StorageBackend;
use crate::types::{AttributeValue, Item};
use std::collections::HashMap;
pub async fn execute<S: StorageBackend>(
storage: &S,
stmt: &Statement,
parameters: &[AttributeValue],
limit: Option<usize>,
) -> Result<Option<Vec<Item>>> {
Ok(execute_measured(storage, stmt, parameters, limit).await?.0)
}
pub async fn execute_measured<S: StorageBackend>(
storage: &S,
stmt: &Statement,
parameters: &[AttributeValue],
limit: Option<usize>,
) -> Result<(Option<Vec<Item>>, usize)> {
let page = execute_page(storage, stmt, parameters, limit, None).await?;
Ok((page.items, page.size))
}
#[derive(Debug, Default)]
#[non_exhaustive]
pub struct StatementPage {
pub items: Option<Vec<Item>>,
pub size: usize,
pub next_token: Option<String>,
}
pub async fn execute_page<S: StorageBackend>(
storage: &S,
stmt: &Statement,
parameters: &[AttributeValue],
limit: Option<usize>,
next_token: Option<&str>,
) -> Result<StatementPage> {
if next_token.is_some() && !matches!(stmt, Statement::Select { .. }) {
return Err(DynoxideError::ValidationException(
"NextToken is only valid on a SELECT statement".to_string(),
));
}
match stmt {
Statement::Select {
table_name,
projections,
where_clause,
} => {
let (items, token) = execute_select(
storage,
table_name,
projections,
where_clause.as_ref(),
parameters,
limit,
next_token,
)
.await?;
let size = items
.as_ref()
.map(|rows| rows.iter().map(crate::types::item_size).sum())
.unwrap_or(0);
Ok(StatementPage {
items,
size,
next_token: token,
})
}
Statement::Insert {
table_name,
item,
if_not_exists,
} => {
let size =
execute_insert(storage, table_name, item, parameters, *if_not_exists).await?;
Ok(StatementPage {
items: None,
size,
..Default::default()
})
}
Statement::Update {
table_name,
set_clauses,
remove_paths,
where_clause,
returning,
} => {
let (projection, size) = execute_update(
storage,
table_name,
set_clauses,
remove_paths,
where_clause.as_ref(),
parameters,
*returning,
)
.await?;
let items = projection.map(|item| {
if item.is_empty() {
Vec::new()
} else {
vec![item]
}
});
Ok(StatementPage {
items,
size,
..Default::default()
})
}
Statement::Delete {
table_name,
where_clause,
returning,
} => {
if let Some(variant) = returning {
if *variant != ReturningVariant::AllOld {
return Err(DynoxideError::ValidationException(format!(
"Invalid returning clause: RETURNING {} *. Only RETURNING ALL OLD * is allowed in DELETE statements.",
variant.as_sql()
)));
}
}
let (old_item, size) =
execute_delete(storage, table_name, where_clause.as_ref(), parameters).await?;
let items = if returning.is_some() {
Some(old_item.map(|item| vec![item]).unwrap_or_default())
} else {
None
};
Ok(StatementPage {
items,
size,
..Default::default()
})
}
}
}
fn insert_nested_projection(result: &mut Item, path: &str, val: AttributeValue) {
let parts: Vec<&str> = path.split('.').collect();
let key = parts.last().unwrap();
result.insert(key.to_string(), val);
}
async fn execute_select<S: StorageBackend>(
storage: &S,
table_name: &str,
projections: &[String],
where_clause: Option<&WhereClause>,
parameters: &[AttributeValue],
limit: Option<usize>,
next_token: Option<&str>,
) -> Result<(Option<Vec<Item>>, Option<String>)> {
let meta = require_table(storage, table_name).await?;
let key_schema = crate::actions::helpers::parse_key_schema(&meta)?;
let fingerprint = statement_fingerprint(where_clause, parameters);
let cursor = next_token
.map(|token| decode_next_token(token, table_name, fingerprint))
.transpose()?;
let window = evaluate_window(
storage,
table_name,
where_clause,
parameters,
&key_schema,
cursor.as_ref(),
limit,
)
.await?;
let token = match (limit, &window.last_evaluated) {
(Some(lim), Some((pk, sk))) if window.evaluated >= lim => {
Some(encode_next_token(table_name, fingerprint, pk, sk))
}
_ => None,
};
let items = window
.matched
.into_iter()
.map(|item| {
if projections.is_empty() {
item
} else {
let mut projected = HashMap::new();
for proj in projections {
if let Some(val) = resolve_nested_path(&item, proj) {
insert_nested_projection(&mut projected, proj, val.clone());
}
}
projected
}
})
.collect();
Ok((Some(items), token))
}
struct Window {
matched: Vec<Item>,
last_evaluated: Option<(String, String)>,
evaluated: usize,
}
async fn evaluate_window<S: StorageBackend>(
storage: &S,
table_name: &str,
where_clause: Option<&WhereClause>,
parameters: &[AttributeValue],
key_schema: &crate::actions::helpers::KeySchema,
cursor: Option<&Cursor>,
limit: Option<usize>,
) -> Result<Window> {
let pk_condition = where_clause.and_then(|wc| find_pk_condition(wc, &key_schema.partition_key));
let rows: Vec<(String, String, String)> = if let Some(pk_cond) = pk_condition {
let pk_val = resolve_value(&pk_cond.value, parameters)?;
let pk_str = pk_val
.to_key_string()
.ok_or_else(|| DynoxideError::ValidationException("Invalid key value".to_string()))?;
let sk_conditions = match (key_schema.sort_key.as_deref(), where_clause) {
(Some(sk_name), Some(wc)) => {
translate_sk_conditions(&wc.groups[0], sk_name, parameters)
}
_ => None,
}
.unwrap_or_default();
let (sk_condition_sql, sk_param_values) = sk_conditions_to_sql(&sk_conditions);
let sk_params_refs: Vec<&str> = sk_param_values.iter().map(|s| s.as_str()).collect();
let params = crate::storage::QueryParams {
sk_condition: sk_condition_sql.as_deref(),
sk_params: &sk_params_refs,
forward: true,
limit,
exclusive_start_sk: cursor.map(|c| c.sk.as_str()),
..Default::default()
};
storage.query_items(table_name, &pk_str, ¶ms).await?
} else {
let params = crate::storage::ScanParams {
limit,
exclusive_start_pk: cursor.map(|c| c.pk.as_str()),
exclusive_start_sk: cursor.map(|c| c.sk.as_str()),
..Default::default()
};
storage.scan_items(table_name, ¶ms).await?
};
let evaluated = rows.len();
let last_evaluated = rows.last().map(|(pk, sk, _)| (pk.clone(), sk.clone()));
let matched = rows
.into_iter()
.filter_map(|(_, _, json)| serde_json::from_str::<Item>(&json).ok())
.filter(|item| matches_where(item, where_clause, parameters))
.collect();
Ok(Window {
matched,
last_evaluated,
evaluated,
})
}
struct Cursor {
pk: String,
sk: String,
}
fn statement_fingerprint(where_clause: Option<&WhereClause>, parameters: &[AttributeValue]) -> u64 {
use std::hash::{Hash, Hasher};
let mut hasher = std::collections::hash_map::DefaultHasher::new();
format!("{where_clause:?}").hash(&mut hasher);
serde_json::to_string(parameters)
.unwrap_or_default()
.hash(&mut hasher);
hasher.finish()
}
fn encode_next_token(table_name: &str, fingerprint: u64, pk: &str, sk: &str) -> String {
use base64::Engine;
let payload =
serde_json::json!({ "t": table_name, "f": fingerprint, "pk": pk, "sk": sk }).to_string();
base64::engine::general_purpose::STANDARD.encode(payload)
}
fn token_mismatch() -> DynoxideError {
DynoxideError::ValidationException("NextToken does not match request".to_string())
}
fn decode_next_token(token: &str, table_name: &str, fingerprint: u64) -> Result<Cursor> {
use base64::Engine;
let invalid = || DynoxideError::ValidationException("Invalid NextToken".to_string());
let raw = base64::engine::general_purpose::STANDARD
.decode(token)
.map_err(|_| invalid())?;
let value: serde_json::Value = serde_json::from_slice(&raw).map_err(|_| invalid())?;
if value["t"].as_str() != Some(table_name) || value["f"].as_u64() != Some(fingerprint) {
return Err(token_mismatch());
}
Ok(Cursor {
pk: value["pk"].as_str().ok_or_else(invalid)?.to_string(),
sk: value["sk"].as_str().ok_or_else(invalid)?.to_string(),
})
}
fn translate_sk_conditions(
group: &[WhereCondition],
sk_name: &str,
parameters: &[AttributeValue],
) -> Option<Vec<ResolvedSortKeyCondition>> {
let mut resolved = Vec::new();
for cond in group {
match cond {
WhereCondition::Comparison(c) if c.path == sk_name => {
let value = resolve_value(&c.value, parameters).ok()?;
value.to_key_string()?;
let sk = sk_name.to_string();
resolved.push(match c.op {
CompOp::Eq => ResolvedSortKeyCondition::Eq(sk, value),
CompOp::Lt => ResolvedSortKeyCondition::Lt(sk, value),
CompOp::Le => ResolvedSortKeyCondition::Le(sk, value),
CompOp::Gt => ResolvedSortKeyCondition::Gt(sk, value),
CompOp::Ge => ResolvedSortKeyCondition::Ge(sk, value),
CompOp::Ne => return None,
});
}
WhereCondition::Between(path, lo, hi) if path == sk_name => {
let lo = resolve_value(lo, parameters).ok()?;
let hi = resolve_value(hi, parameters).ok()?;
lo.to_key_string()?;
hi.to_key_string()?;
resolved.push(ResolvedSortKeyCondition::Between(
sk_name.to_string(),
lo,
hi,
));
}
WhereCondition::BeginsWith(path, prefix) if path == sk_name => {
let prefix = resolve_value(prefix, parameters).ok()?;
prefix.to_key_string()?;
resolved.push(ResolvedSortKeyCondition::BeginsWith(
sk_name.to_string(),
prefix,
));
}
WhereCondition::NotBeginsWith(path, _)
| WhereCondition::In(path, _)
| WhereCondition::Contains(path, _)
| WhereCondition::Exists(path)
| WhereCondition::NotExists(path)
| WhereCondition::IsMissing(path)
| WhereCondition::IsNotMissing(path)
if path == sk_name =>
{
return None;
}
_ => {}
}
}
Some(resolved)
}
fn find_pk_condition<'a>(
wc: &'a WhereClause,
pk_name: &str,
) -> Option<&'a crate::partiql::parser::Condition> {
if wc.groups.len() == 1 {
wc.groups[0].iter().find_map(|c| match c {
WhereCondition::Comparison(cond) if cond.path == pk_name && cond.op == CompOp::Eq => {
Some(cond)
}
_ => None,
})
} else {
None
}
}
async fn execute_insert<S: StorageBackend>(
storage: &S,
table_name: &str,
item_template: &HashMap<String, PartiqlValue>,
parameters: &[AttributeValue],
if_not_exists: bool,
) -> Result<usize> {
let mut item = HashMap::new();
for (k, v) in item_template {
let resolved = match v {
PartiqlValue::Literal(av) => av.clone(),
PartiqlValue::Parameter(idx) => parameters.get(*idx).cloned().ok_or_else(|| {
DynoxideError::ValidationException(format!(
"Parameter index {idx} out of range (have {} parameters)",
parameters.len()
))
})?,
};
item.insert(k.clone(), resolved);
}
let meta = require_table(storage, table_name).await?;
let key_schema = crate::actions::helpers::parse_key_schema(&meta)?;
crate::actions::helpers::validate_item_keys(&item, &key_schema, &meta)?;
crate::validation::validate_item_attribute_values(&item)?;
crate::validation::normalize_item_sets(&mut item);
let (pk, sk) = crate::actions::helpers::extract_key_strings(&item, &key_schema)?;
let existing = storage.get_item(table_name, &pk, &sk).await?;
if existing.is_some() {
if if_not_exists {
return Ok(0);
}
return Err(DynoxideError::DuplicateItemException(
"Duplicate primary key exists in table".to_string(),
));
}
let item_json = serde_json::to_string(&item)
.map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
let item_size = crate::types::item_size(&item);
let hash_prefix = item
.get(&key_schema.partition_key)
.map(crate::storage::compute_hash_prefix)
.unwrap_or_default();
let old_json = storage
.put_item_with_hash(table_name, &pk, &sk, &item_json, item_size, &hash_prefix)
.await?;
let table_sk_attr = key_schema.sort_key.as_deref();
let _ = crate::actions::gsi::maintain_gsis_after_write(
storage,
table_name,
&meta,
&pk,
&sk,
&item,
&key_schema.partition_key,
table_sk_attr,
)
.await?;
crate::actions::lsi::maintain_lsis_after_write(
storage,
table_name,
&meta,
&pk,
&sk,
&item,
&key_schema.partition_key,
table_sk_attr,
)
.await?;
let old_item: Option<Item> = old_json.as_ref().and_then(|j| serde_json::from_str(j).ok());
crate::streams::record_stream_event(storage, &meta, old_item.as_ref(), Some(&item)).await?;
Ok(item_size)
}
async fn execute_update<S: StorageBackend>(
storage: &S,
table_name: &str,
set_clauses: &[crate::partiql::parser::SetClause],
remove_paths: &[String],
where_clause: Option<&WhereClause>,
parameters: &[AttributeValue],
returning: Option<ReturningVariant>,
) -> Result<(Option<Item>, usize)> {
let meta = require_table(storage, table_name).await?;
let key_schema = crate::actions::helpers::parse_key_schema(&meta)?;
let wc = where_clause.ok_or_else(|| {
DynoxideError::ValidationException("UPDATE requires a WHERE clause".to_string())
})?;
if wc.groups.len() > 1 {
return Err(DynoxideError::ValidationException(
"UPDATE does not support OR conditions in WHERE clause".to_string(),
));
}
let pk_cond =
find_comparison_in_groups(&wc.groups, &key_schema.partition_key).ok_or_else(|| {
DynoxideError::ValidationException(
"Where clause does not contain a mandatory equality on all key attributes"
.to_string(),
)
})?;
let pk_val = resolve_value(&pk_cond.value, parameters)?;
let pk_str = pk_val
.to_key_string()
.ok_or_else(|| DynoxideError::ValidationException("Invalid key value".to_string()))?;
let sk_str = if let Some(ref sk_name) = key_schema.sort_key {
let sk_cond = find_comparison_in_groups(&wc.groups, sk_name);
if sk_cond.is_none() {
return Err(DynoxideError::ValidationException(
"Where clause does not contain a mandatory equality on all key attributes"
.to_string(),
));
}
sk_cond
.map(|c| resolve_value(&c.value, parameters))
.transpose()?
.and_then(|v| v.to_key_string())
.unwrap_or_default()
} else {
String::new()
};
let existing_json = storage.get_item(table_name, &pk_str, &sk_str).await?;
let mut item: Item = existing_json
.as_ref()
.and_then(|j| serde_json::from_str(j).ok())
.unwrap_or_default();
let old_item = item.clone();
if existing_json.is_none() || !matches_where(&old_item, where_clause, parameters) {
return Err(DynoxideError::ConditionalCheckFailedException(
"The conditional request failed".to_string(),
None,
));
}
let before_item = item.clone();
for clause in set_clauses {
let val = resolve_set_value(&clause.value, &item, parameters)?;
set_nested_value(&mut item, &clause.path, val)?;
}
for path in remove_paths {
remove_nested_value(&mut item, path);
}
if item.is_empty() {
return Ok((None, 0));
}
crate::validation::validate_item_attribute_values(&item)?;
crate::validation::normalize_item_sets(&mut item);
crate::actions::helpers::validate_updated_index_keys(&before_item, &item, &meta)?;
let item_json = serde_json::to_string(&item)
.map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
let item_size = crate::types::item_size(&item);
let hash_prefix = item
.get(&key_schema.partition_key)
.map(crate::storage::compute_hash_prefix)
.unwrap_or_default();
storage
.put_item_with_hash(
table_name,
&pk_str,
&sk_str,
&item_json,
item_size,
&hash_prefix,
)
.await?;
let table_sk_attr = key_schema.sort_key.as_deref();
let _ = crate::actions::gsi::maintain_gsis_after_write(
storage,
table_name,
&meta,
&pk_str,
&sk_str,
&item,
&key_schema.partition_key,
table_sk_attr,
)
.await?;
crate::actions::lsi::maintain_lsis_after_write(
storage,
table_name,
&meta,
&pk_str,
&sk_str,
&item,
&key_schema.partition_key,
table_sk_attr,
)
.await?;
let old_ref = if existing_json.is_some() {
Some(&old_item)
} else {
None
};
crate::streams::record_stream_event(storage, &meta, old_ref, Some(&item)).await?;
let projection = returning.map(|variant| {
let modified: std::collections::BTreeSet<String> = set_clauses
.iter()
.map(|c| c.path.clone())
.chain(remove_paths.iter().cloned())
.collect();
project_returning(variant, &old_item, &item, &modified)
});
Ok((projection, item_size))
}
fn project_returning(
variant: ReturningVariant,
old_item: &Item,
new_item: &Item,
modified: &std::collections::BTreeSet<String>,
) -> Item {
match variant {
ReturningVariant::AllOld => old_item.clone(),
ReturningVariant::AllNew => new_item.clone(),
ReturningVariant::ModifiedOld => project_modified(modified, old_item),
ReturningVariant::ModifiedNew => project_modified(modified, new_item),
}
}
enum ProjNode {
Leaf(AttributeValue),
Map(HashMap<String, ProjNode>),
List(std::collections::BTreeMap<usize, ProjNode>),
}
fn project_modified(paths: &std::collections::BTreeSet<String>, source: &Item) -> Item {
let mut root: HashMap<String, ProjNode> = HashMap::new();
for path in paths {
if let (Some(val), Some(segments)) =
(resolve_nested_path(source, path), split_path_segments(path))
{
if let Some((PathSegment::Key(key), rest)) = segments.split_first() {
let node = root
.entry((*key).to_string())
.or_insert_with(|| fresh_proj_node(rest));
insert_proj_node(node, rest, val.clone());
}
}
}
root.into_iter()
.map(|(k, node)| (k, proj_node_to_value(node)))
.collect()
}
fn fresh_proj_node(segments: &[PathSegment]) -> ProjNode {
match segments.first() {
Some(PathSegment::Index(_)) => ProjNode::List(std::collections::BTreeMap::new()),
_ => ProjNode::Map(HashMap::new()),
}
}
fn insert_proj_node(node: &mut ProjNode, segments: &[PathSegment], val: AttributeValue) {
let Some((seg, rest)) = segments.split_first() else {
*node = ProjNode::Leaf(val);
return;
};
match seg {
PathSegment::Key(k) => {
if let ProjNode::Map(map) = node {
let child = map
.entry((*k).to_string())
.or_insert_with(|| fresh_proj_node(rest));
insert_proj_node(child, rest, val);
}
}
PathSegment::Index(i) => {
if let ProjNode::List(list) = node {
let child = list.entry(*i).or_insert_with(|| fresh_proj_node(rest));
insert_proj_node(child, rest, val);
}
}
}
}
fn proj_node_to_value(node: ProjNode) -> AttributeValue {
match node {
ProjNode::Leaf(v) => v,
ProjNode::Map(map) => AttributeValue::M(
map.into_iter()
.map(|(k, n)| (k, proj_node_to_value(n)))
.collect(),
),
ProjNode::List(list) => {
AttributeValue::L(list.into_values().map(proj_node_to_value).collect())
}
}
}
async fn execute_delete<S: StorageBackend>(
storage: &S,
table_name: &str,
where_clause: Option<&WhereClause>,
parameters: &[AttributeValue],
) -> Result<(Option<Item>, usize)> {
let meta = require_table(storage, table_name).await?;
let key_schema = crate::actions::helpers::parse_key_schema(&meta)?;
let wc = where_clause.ok_or_else(|| {
DynoxideError::ValidationException("DELETE requires a WHERE clause".to_string())
})?;
if wc.groups.len() > 1 {
return Err(DynoxideError::ValidationException(
"DELETE does not support OR conditions in WHERE clause".to_string(),
));
}
let pk_cond =
find_comparison_in_groups(&wc.groups, &key_schema.partition_key).ok_or_else(|| {
DynoxideError::ValidationException(
"Where clause does not contain a mandatory equality on all key attributes"
.to_string(),
)
})?;
let pk_val = resolve_value(&pk_cond.value, parameters)?;
let pk_str = pk_val
.to_key_string()
.ok_or_else(|| DynoxideError::ValidationException("Invalid key value".to_string()))?;
if let Some(ref sk_name) = key_schema.sort_key {
let has_sk_condition = wc.groups.iter().any(|group| {
group.iter().any(|c| match c {
WhereCondition::Comparison(comp) => comp.path == *sk_name && comp.op == CompOp::Eq,
_ => false,
})
});
if !has_sk_condition {
return Err(DynoxideError::ValidationException(
"Where clause does not contain a mandatory equality on all key attributes"
.to_string(),
));
}
}
let sk_str = if let Some(ref sk_name) = key_schema.sort_key {
find_comparison_in_groups(&wc.groups, sk_name)
.map(|c| resolve_value(&c.value, parameters))
.transpose()?
.and_then(|v| v.to_key_string())
.unwrap_or_default()
} else {
String::new()
};
if let Some(json) = storage.get_item(table_name, &pk_str, &sk_str).await? {
let existing: Item = serde_json::from_str(&json)
.map_err(|e| DynoxideError::InternalServerError(format!("Bad item JSON: {e}")))?;
if !matches_where(&existing, where_clause, parameters) {
return Err(DynoxideError::ConditionalCheckFailedException(
"The conditional request failed".to_string(),
None,
));
}
}
let old_json = storage.delete_item(table_name, &pk_str, &sk_str).await?;
let _ = crate::actions::gsi::maintain_gsis_after_delete(
storage, table_name, &meta, &pk_str, &sk_str,
)
.await?;
crate::actions::lsi::maintain_lsis_after_delete(storage, table_name, &meta, &pk_str, &sk_str)
.await?;
let old_item: Option<Item> = old_json.as_ref().and_then(|j| serde_json::from_str(j).ok());
if old_item.is_some() {
crate::streams::record_stream_event(storage, &meta, old_item.as_ref(), None).await?;
}
let deleted_size = old_item.as_ref().map(crate::types::item_size).unwrap_or(0);
Ok((old_item, deleted_size))
}
async fn require_table<S: StorageBackend>(
storage: &S,
table_name: &str,
) -> Result<crate::storage::TableMetadata> {
crate::actions::helpers::require_table(storage, table_name).await
}
fn find_comparison_in_groups<'a>(
groups: &'a [Vec<WhereCondition>],
path: &str,
) -> Option<&'a crate::partiql::parser::Condition> {
for group in groups {
if let Some(cond) = find_comparison(group, path) {
return Some(cond);
}
}
None
}
fn find_comparison<'a>(
conditions: &'a [WhereCondition],
path: &str,
) -> Option<&'a crate::partiql::parser::Condition> {
conditions.iter().find_map(|c| match c {
WhereCondition::Comparison(cond) if cond.path == path && cond.op == CompOp::Eq => {
Some(cond)
}
_ => None,
})
}
fn resolve_value(val: &PartiqlValue, parameters: &[AttributeValue]) -> Result<AttributeValue> {
match val {
PartiqlValue::Literal(av) => Ok(av.clone()),
PartiqlValue::Parameter(idx) => parameters.get(*idx).cloned().ok_or_else(|| {
DynoxideError::ValidationException(format!(
"Parameter index {idx} out of range (have {} parameters)",
parameters.len()
))
}),
}
}
fn resolve_set_value(
val: &SetValue,
item: &Item,
parameters: &[AttributeValue],
) -> Result<AttributeValue> {
match val {
SetValue::Simple(pv) => resolve_value(pv, parameters),
SetValue::Add(attr, pv) => {
let current = resolve_nested_path(item, attr);
let operand = resolve_value(pv, parameters)?;
match (current, &operand) {
(Some(AttributeValue::N(cur)), AttributeValue::N(add)) => {
use bigdecimal::BigDecimal;
use std::str::FromStr;
let a = BigDecimal::from_str(cur).map_err(|e| {
DynoxideError::ValidationException(format!("Invalid number: {e}"))
})?;
let b = BigDecimal::from_str(add).map_err(|e| {
DynoxideError::ValidationException(format!("Invalid number: {e}"))
})?;
let result = a + b;
Ok(AttributeValue::N(format_bigdecimal(&result)))
}
(None, AttributeValue::N(_)) => {
Ok(operand)
}
_ => Err(DynoxideError::ValidationException(
"SET expression add requires numeric attribute and operand".to_string(),
)),
}
}
SetValue::Sub(attr, pv) => {
let current = resolve_nested_path(item, attr);
let operand = resolve_value(pv, parameters)?;
match (current, &operand) {
(Some(AttributeValue::N(cur)), AttributeValue::N(sub)) => {
use bigdecimal::BigDecimal;
use std::str::FromStr;
let a = BigDecimal::from_str(cur).map_err(|e| {
DynoxideError::ValidationException(format!("Invalid number: {e}"))
})?;
let b = BigDecimal::from_str(sub).map_err(|e| {
DynoxideError::ValidationException(format!("Invalid number: {e}"))
})?;
let result = a - b;
Ok(AttributeValue::N(format_bigdecimal(&result)))
}
(None, AttributeValue::N(sub)) => {
use bigdecimal::BigDecimal;
use std::str::FromStr;
let b = BigDecimal::from_str(sub).map_err(|e| {
DynoxideError::ValidationException(format!("Invalid number: {e}"))
})?;
let result = -b;
Ok(AttributeValue::N(format_bigdecimal(&result)))
}
_ => Err(DynoxideError::ValidationException(
"SET expression subtract requires numeric attribute and operand".to_string(),
)),
}
}
SetValue::ListAppend(first, second) => {
let a = resolve_value(first, parameters)?;
let b = resolve_value(second, parameters)?;
let list_a = match &a {
AttributeValue::S(name) => resolve_nested_path(item, name)
.cloned()
.unwrap_or(AttributeValue::L(Vec::new())),
other => other.clone(),
};
let list_b = match &b {
AttributeValue::S(name) => resolve_nested_path(item, name)
.cloned()
.unwrap_or(AttributeValue::L(Vec::new())),
other => other.clone(),
};
match (list_a, list_b) {
(AttributeValue::L(mut la), AttributeValue::L(lb)) => {
la.extend(lb);
Ok(AttributeValue::L(la))
}
_ => Err(DynoxideError::ValidationException(
"list_append requires list operands".to_string(),
)),
}
}
}
}
fn invalid_update_path() -> DynoxideError {
DynoxideError::ValidationException(
"The document path provided in the update expression is invalid for update".to_string(),
)
}
fn set_nested_value(item: &mut Item, path: &str, val: AttributeValue) -> Result<()> {
let segments = split_path_segments(path).ok_or_else(invalid_update_path)?;
let (first, rest) = segments.split_first().ok_or_else(invalid_update_path)?;
let key = match first {
PathSegment::Key(k) => (*k).to_string(),
PathSegment::Index(_) => return Err(invalid_update_path()),
};
if rest.is_empty() {
item.insert(key, val);
return Ok(());
}
let entry = item
.entry(key)
.or_insert_with(|| AttributeValue::M(HashMap::new()));
set_into_value(entry, rest, val)
}
fn set_into_value(
current: &mut AttributeValue,
segments: &[PathSegment],
val: AttributeValue,
) -> Result<()> {
let (seg, rest) = segments.split_first().expect("segments is non-empty");
if rest.is_empty() {
return match seg {
PathSegment::Key(k) => match current {
AttributeValue::M(map) => {
map.insert((*k).to_string(), val);
Ok(())
}
_ => Err(invalid_update_path()),
},
PathSegment::Index(i) => match current {
AttributeValue::L(list) => {
if *i < list.len() {
list[*i] = val;
} else {
list.push(val);
}
Ok(())
}
_ => Err(invalid_update_path()),
},
};
}
match seg {
PathSegment::Key(k) => match current {
AttributeValue::M(map) => {
let next = map
.entry((*k).to_string())
.or_insert_with(|| AttributeValue::M(HashMap::new()));
set_into_value(next, rest, val)
}
_ => Err(invalid_update_path()),
},
PathSegment::Index(i) => match current {
AttributeValue::L(list) => match list.get_mut(*i) {
Some(next) => set_into_value(next, rest, val),
None => Err(invalid_update_path()),
},
_ => Err(invalid_update_path()),
},
}
}
fn remove_nested_value(item: &mut Item, path: &str) {
let Some(segments) = split_path_segments(path) else {
return;
};
let Some((first, rest)) = segments.split_first() else {
return;
};
let PathSegment::Key(key) = first else {
return; };
if rest.is_empty() {
item.remove(*key);
return;
}
if let Some(current) = item.get_mut(*key) {
remove_from_value(current, rest);
}
}
fn remove_from_value(current: &mut AttributeValue, segments: &[PathSegment]) {
let (seg, rest) = segments.split_first().expect("segments is non-empty");
if rest.is_empty() {
match seg {
PathSegment::Key(k) => {
if let AttributeValue::M(map) = current {
map.remove(*k);
}
}
PathSegment::Index(i) => {
if let AttributeValue::L(list) = current {
if *i < list.len() {
list.remove(*i);
}
}
}
}
return;
}
match seg {
PathSegment::Key(k) => {
if let AttributeValue::M(map) = current {
if let Some(next) = map.get_mut(*k) {
remove_from_value(next, rest);
}
}
}
PathSegment::Index(i) => {
if let AttributeValue::L(list) = current {
if let Some(next) = list.get_mut(*i) {
remove_from_value(next, rest);
}
}
}
}
}
fn matches_where(
item: &Item,
where_clause: Option<&WhereClause>,
parameters: &[AttributeValue],
) -> bool {
let wc = match where_clause {
Some(wc) => wc,
None => return true,
};
wc.groups
.iter()
.any(|group| matches_conditions(item, group, parameters))
}
fn matches_conditions(
item: &Item,
conditions: &[WhereCondition],
parameters: &[AttributeValue],
) -> bool {
for cond in conditions {
match cond {
WhereCondition::Comparison(c) => {
let item_val = match resolve_nested_path(item, &c.path) {
Some(v) => v,
None => return false,
};
let target = match resolve_value(&c.value, parameters) {
Ok(v) => v,
Err(_) => return false,
};
if !compare_values(item_val, &c.op, &target) {
return false;
}
}
WhereCondition::Exists(path) | WhereCondition::IsNotMissing(path) => {
if resolve_nested_path(item, path).is_none() {
return false;
}
}
WhereCondition::NotExists(path) | WhereCondition::IsMissing(path) => {
if resolve_nested_path(item, path).is_some() {
return false;
}
}
WhereCondition::BeginsWith(path, prefix_val) => {
let item_val = match resolve_nested_path(item, path) {
Some(v) => v,
None => return false,
};
let prefix = match resolve_value(prefix_val, parameters) {
Ok(v) => v,
Err(_) => return false,
};
match (item_val, &prefix) {
(AttributeValue::S(s), AttributeValue::S(p)) => {
if !s.starts_with(p.as_str()) {
return false;
}
}
_ => return false,
}
}
WhereCondition::NotBeginsWith(path, prefix_val) => {
if let Some(item_val) = resolve_nested_path(item, path) {
let prefix = match resolve_value(prefix_val, parameters) {
Ok(v) => v,
Err(_) => return false,
};
if let (AttributeValue::S(s), AttributeValue::S(p)) = (item_val, &prefix) {
if s.starts_with(p.as_str()) {
return false;
}
}
}
}
WhereCondition::Between(path, low, high) => {
let item_val = match resolve_nested_path(item, path) {
Some(v) => v,
None => return false,
};
let low_val = match resolve_value(low, parameters) {
Ok(v) => v,
Err(_) => return false,
};
let high_val = match resolve_value(high, parameters) {
Ok(v) => v,
Err(_) => return false,
};
if !compare_values(item_val, &CompOp::Ge, &low_val)
|| !compare_values(item_val, &CompOp::Le, &high_val)
{
return false;
}
}
WhereCondition::In(path, values) => {
let item_val = match resolve_nested_path(item, path) {
Some(v) => v,
None => return false,
};
let matched = values.iter().any(|v| {
resolve_value(v, parameters)
.map(|target| compare_values(item_val, &CompOp::Eq, &target))
.unwrap_or(false)
});
if !matched {
return false;
}
}
WhereCondition::Contains(path, substr_val) => {
let item_val = match resolve_nested_path(item, path) {
Some(v) => v,
None => return false,
};
let substr = match resolve_value(substr_val, parameters) {
Ok(v) => v,
Err(_) => return false,
};
match (item_val, &substr) {
(AttributeValue::S(s), AttributeValue::S(sub)) => {
if !s.contains(sub.as_str()) {
return false;
}
}
(AttributeValue::SS(set), AttributeValue::S(val)) => {
if !set.contains(val) {
return false;
}
}
(AttributeValue::NS(set), AttributeValue::N(val)) => {
if !set.contains(val) {
return false;
}
}
(AttributeValue::L(list), target) => {
if !list.contains(target) {
return false;
}
}
_ => return false,
}
}
}
}
true
}
fn resolve_nested_path<'a>(item: &'a Item, path: &str) -> Option<&'a AttributeValue> {
if !path.contains('.') && !path.contains('[') {
return item.get(path);
}
let segments = split_path_segments(path)?;
if segments.is_empty() {
return None;
}
let mut current = match &segments[0] {
PathSegment::Key(k) => item.get(*k)?,
PathSegment::Index(_) => return None,
};
for seg in &segments[1..] {
current = match seg {
PathSegment::Key(k) => match current {
AttributeValue::M(map) => map.get(*k)?,
_ => return None,
},
PathSegment::Index(idx) => match current {
AttributeValue::L(list) => list.get(*idx)?,
_ => return None,
},
};
}
Some(current)
}
enum PathSegment<'a> {
Key(&'a str),
Index(usize),
}
fn split_path_segments(path: &str) -> Option<Vec<PathSegment<'_>>> {
let mut segments = Vec::new();
let bytes = path.as_bytes();
let mut start = 0;
let mut i = 0;
while i < bytes.len() {
match bytes[i] {
b'.' => {
if start < i {
segments.push(PathSegment::Key(&path[start..i]));
}
i += 1;
start = i;
}
b'[' => {
if start < i {
segments.push(PathSegment::Key(&path[start..i]));
}
i += 1;
let idx_start = i;
while i < bytes.len() && bytes[i] != b']' {
i += 1;
}
let idx = path[idx_start..i].parse::<usize>().ok()?;
segments.push(PathSegment::Index(idx));
if i < bytes.len() {
i += 1; }
start = i;
if i < bytes.len() && bytes[i] == b'.' {
i += 1;
start = i;
}
}
_ => {
i += 1;
}
}
}
if start < bytes.len() {
segments.push(PathSegment::Key(&path[start..]));
}
Some(segments)
}
fn compare_values(left: &AttributeValue, op: &CompOp, right: &AttributeValue) -> bool {
match (left, right) {
(AttributeValue::S(a), AttributeValue::S(b)) => compare_ord(a, op, b),
(AttributeValue::N(a), AttributeValue::N(b)) => {
use bigdecimal::BigDecimal;
use std::str::FromStr;
match (BigDecimal::from_str(a), BigDecimal::from_str(b)) {
(Ok(da), Ok(db)) => compare_ord(&da, op, &db),
_ => false,
}
}
(AttributeValue::BOOL(a), AttributeValue::BOOL(b)) => match op {
CompOp::Eq => a == b,
CompOp::Ne => a != b,
_ => false,
},
_ => match op {
CompOp::Eq => false,
CompOp::Ne => true,
_ => false,
},
}
}
fn format_bigdecimal(n: &bigdecimal::BigDecimal) -> String {
let normalized = n.normalized();
if normalized.as_bigint_and_exponent().1 < 0 {
normalized.with_scale(0).to_string()
} else {
normalized.to_string()
}
}
fn compare_ord<T: PartialOrd>(a: &T, op: &CompOp, b: &T) -> bool {
match op {
CompOp::Eq => a == b,
CompOp::Ne => a != b,
CompOp::Lt => a < b,
CompOp::Le => a <= b,
CompOp::Gt => a > b,
CompOp::Ge => a >= b,
}
}