Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 50 additions & 0 deletions crates/core/src/expression/ast.rs
Original file line number Diff line number Diff line change
Expand Up @@ -82,3 +82,53 @@ pub enum UpdateAction {
/// `DELETE path value` — removes elements from a set.
Delete { path: Vec<PathElement>, value: Expr },
}

impl UpdateAction {
/// Whether this action can only take data away: `REMOVE` or `DELETE`.
#[must_use]
pub fn removes_only(&self) -> bool {
matches!(self, Self::Remove { .. } | Self::Delete { .. })
}
}

/// Whether an update made of these actions writes nothing when the item does
/// not exist.
///
/// Amazon DynamoDB creates the item for `SET`, `ADD`, and for an update with
/// no actions at all (a key-only item), but an update whose actions are all
/// `REMOVE` or `DELETE` on a missing key leaves the table unchanged: no item,
/// no stream record, no `Attributes` in the response for any `ReturnValues`
/// (measured 2026-10-24, us-east-1, hash and hash-plus-range tables).
#[must_use]
pub fn removes_only(actions: &[UpdateAction]) -> bool {
!actions.is_empty() && actions.iter().all(UpdateAction::removes_only)
}

#[cfg(test)]
mod tests {
use super::removes_only;
use crate::expression::tokenizer::tokenize;
use crate::expression::update_parser::parse_update;

fn actions(expr: &str) -> Vec<super::UpdateAction> {
parse_update(&tokenize(expr).expect("tokens")).expect("actions")
}

#[test]
fn remove_and_delete_actions_remove_only() {
assert!(removes_only(&actions("REMOVE a")));
assert!(removes_only(&actions("REMOVE a, b")));
assert!(removes_only(&actions("REMOVE a.b[0]")));
assert!(removes_only(&actions("DELETE s :v")));
assert!(removes_only(&actions("REMOVE a DELETE s :v")));
}

#[test]
fn set_add_or_no_actions_create_the_item() {
assert!(!removes_only(&[]));
assert!(!removes_only(&actions("SET a = :v")));
assert!(!removes_only(&actions("ADD n :v")));
assert!(!removes_only(&actions("REMOVE a SET b = :v")));
assert!(!removes_only(&actions("DELETE s :v ADD n :w")));
}
}
2 changes: 1 addition & 1 deletion crates/core/src/expression/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ mod tokenizer;
mod update_evaluator;
mod update_parser;

pub use ast::{ArithOp, CompareOp, Expr, PathElement, UpdateAction};
pub use ast::{ArithOp, CompareOp, Expr, PathElement, UpdateAction, removes_only};
pub use evaluator::evaluate_condition;
pub use key_condition::{KeyCondition, SortKeyCondition, parse_key_condition};
pub use kind::ExpressionKind;
Expand Down
8 changes: 8 additions & 0 deletions crates/storage-cassandra/src/data/update_item.rs
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,14 @@ impl CassandraEngine {
validation::validate_item_size(&item, MAX_ITEM_SIZE_BYTES)
.map_err(|e| StorageError::Validation(e.to_string()))?;

// REMOVE and DELETE on a key that holds no item change nothing: the
// service writes no item and emits no stream record (issue #478).
// The actions were still applied above so that their validation
// errors surface the same way as on an existing item.
if !item_existed && expression::removes_only(actions) {
return Ok((None, None));
}

let new_item = if return_new { Some(item.clone()) } else { None };

let item_json =
Expand Down
44 changes: 44 additions & 0 deletions crates/storage-mongodb/src/data_engine.rs
Original file line number Diff line number Diff line change
Expand Up @@ -807,6 +807,30 @@ impl MongoEngine {
&& self.gsi_cache_get_fresh(&key_info.table_id) == Some(false)
&& let Some(mongo_update) = self.try_build_native_update(actions, maps)
{
// REMOVE and DELETE on a key that holds no document change
// nothing (issue #478), so the native update runs without
// upsert: a missing document is the whole answer, and the
// service returns no Attributes for it under any ReturnValues.
if let NativeUpdate::Doc(d) = &mongo_update
&& expression::removes_only(actions)
{
let opts = mongodb::options::FindOneAndUpdateOptions::builder()
.upsert(false)
.return_document(ReturnDocument::After)
.build();
let result = coll
.find_one_and_update(key_filter.clone(), d.clone())
.with_options(opts)
.await
.map_err(|e| StorageError::Internal(e.to_string()))?;
let new_item = if return_new {
result.as_ref().map(document_to_item).transpose()?
} else {
None
};
return Ok((None, new_item));
}

let (fast_filter, took_fast) = match mongo_update {
NativeUpdate::Doc(d) => {
// The upsert's insert branch must write the document
Expand Down Expand Up @@ -894,6 +918,9 @@ impl MongoEngine {
enum AttemptOk {
Committed(Option<Item>, Option<Item>),
Stale,
/// The key holds no document and the update only removes:
/// nothing to write, nothing to return (issue #478).
NoWrite,
}

let attempt_res: Result<AttemptOk, TxErr> = async {
Expand Down Expand Up @@ -958,6 +985,13 @@ impl MongoEngine {
.await
.map_err(TxErr::Fatal)?;

// The actions were applied above so that their validation
// errors surface the same way as on an existing document;
// with no document to change, the write is skipped.
if existing_doc.is_none() && expression::removes_only(actions) {
return Ok(AttemptOk::NoWrite);
}

let mut new_doc = item_to_document(
&new_item,
&key_info.key_schema,
Expand Down Expand Up @@ -1055,6 +1089,10 @@ impl MongoEngine {
backoff_sleep(attempt).await;
continue;
}
Ok(AttemptOk::NoWrite) => {
let _ = session.abort_transaction().await;
return Ok((None, None));
}
Err(TxErr::Transient) => {
let _ = session.abort_transaction().await;
backoff_sleep(attempt).await;
Expand Down Expand Up @@ -2946,6 +2984,12 @@ impl MongoEngine {
&key_info.attribute_definitions,
)
.map_err(|e| StorageError::Validation(e.to_string()))?;
// REMOVE and DELETE on a missing key write nothing
// (issue #478); the actions ran above for their
// validation errors only.
if expression::removes_only(actions) {
return Ok((None, None));
}
let new_doc = item_to_document(
&new_item,
&key_info.key_schema,
Expand Down
69 changes: 61 additions & 8 deletions crates/storage-mongodb/src/data_engine_live_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,7 @@ fn expected_document(item: &Item, key_info: &TableKeyInfo, v: i64) -> Document {
}

/// An UpdateItem that creates the item, on the native fast path (SET with a
/// placeholder, REMOVE) and on the transaction path (numeric ADD), stores the
/// placeholder) and on the transaction path (numeric ADD), stores the
/// same document PutItem would store: `_id` as the key text, the key
/// attributes in `item_data`, version 1. ALL_NEW returns the item with its
/// key. A PutItem on the key then replaces the item and returns it as the old
Expand All @@ -152,7 +152,7 @@ fn expected_document(item: &Item, key_info: &TableKeyInfo, v: i64) -> Document {
#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"]
async fn live_update_item_that_creates_an_item_stores_its_key_like_put_item() {
let engine = scratch_engine().await;
let cases: [(&str, Vec<UpdateAction>, AttributeValue, Item); 3] = [
let cases: [(&str, Vec<UpdateAction>, AttributeValue, Item); 2] = [
(
"set",
vec![UpdateAction::Set {
Expand All @@ -162,12 +162,6 @@ async fn live_update_item_that_creates_an_item_stores_its_key_like_put_item() {
s("upd"),
with(Item::new(), "w", s("upd")),
),
(
"remove",
vec![UpdateAction::Remove { path: path("w") }],
s("unused"),
Item::new(),
),
(
"add",
vec![UpdateAction::Add {
Expand Down Expand Up @@ -235,6 +229,65 @@ async fn live_update_item_that_creates_an_item_stores_its_key_like_put_item() {
finish(engine).await;
}

/// A REMOVE-only UpdateItem on a missing key, on the native fast path (the
/// table helper records that the table has no GSI), writes nothing: no old
/// or new image and no document (issue #478). A PutItem on the key then
/// creates the item.
#[tokio::test]
#[ignore = "needs a MongoDB replica set in EXTENDDB_TEST_MONGODB_URI"]
async fn live_remove_only_update_on_a_missing_key_writes_nothing() {
let engine = scratch_engine().await;
for range in [false, true] {
let table_id = format!("remove-{}", if range { "range" } else { "hash" });
let key_info = table(&engine, &table_id, range);
let key = item_key(range);
let remove = vec![UpdateAction::Remove { path: path("w") }];
let images = engine
.update_item_impl(
&key_info,
&key,
&remove,
false,
true,
None,
&maps(s("x")),
None,
)
.await
.unwrap_or_else(|e| panic!("{table_id}: update: {e:?}"));
assert_eq!(images, (None, None), "{table_id}: no image");
let filter =
pk_filter(&key, &key_info.key_schema, &key_info.attribute_definitions).expect("filter");
let doc = engine
.data_db
.collection::<Document>(&data_collection_name(&key_info.table_id))
.find_one(filter)
.await
.expect("read back");
assert_eq!(doc, None, "{table_id}: no document");

let put = with(key.clone(), "w", s("put"));
engine
.put_item_impl(
&key_info,
put.clone(),
true,
None,
&ExpressionMaps::default(),
None,
)
.await
.unwrap_or_else(|e| panic!("{table_id}: put: {e:?}"));
let after = stored(&engine, &key_info, &key).await;
assert_eq!(
document_to_item(&after).expect("item"),
put,
"{table_id}: put"
);
}
finish(engine).await;
}

/// Sixty-four UpdateItem calls creating the same item at once, on the fast
/// path, from eight runtime threads: every call succeeds, the stored
/// document has the shape PutItem writes, and `_v` counts every call, so
Expand Down
8 changes: 8 additions & 0 deletions crates/storage-postgres/src/data/update_item.rs
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,14 @@ impl PostgresEngine {
validation::validate_item_size(&item, self.max_item_size_bytes)
.map_err(|e| StorageError::Validation(e.to_string()))?;

// REMOVE and DELETE on a key that holds no row change nothing: the
// service writes no item and emits no stream record (issue #478).
// The actions were still applied above so that their validation
// errors surface the same way as on an existing row.
if old_json.is_none() && expression::removes_only(actions) {
return Ok((None, None));
}

let new_item = if return_new { Some(item.clone()) } else { None };

let item_json =
Expand Down
15 changes: 14 additions & 1 deletion crates/storage-sqlite/src/data/update_item.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,8 @@
//! `update_item` for the SQLite backend.
//!
//! UpdateItem is an upsert: when the item is absent a new one is created from
//! the key plus the applied actions. The condition is evaluated against the
//! the key plus the applied actions, unless every action is a REMOVE or DELETE,
//! in which case nothing is written. The condition is evaluated against the
//! pre-update image (or an empty item), then update actions are applied and the
//! result is validated and written — all under the engine write lock (D1).

Expand Down Expand Up @@ -90,6 +91,14 @@ impl SqliteEngine {
.map_err(|e| StorageError::Validation(e.to_string()))?;
}

// REMOVE and DELETE on a key that holds no item change nothing: the
// service writes no item and emits no stream record (issue #478). The
// actions were still applied above so that their validation errors
// surface the same way as on an existing item.
if old.is_none() && expression::removes_only(actions) {
return Ok((None, None));
}

upsert_item_in_tx(&mut tx, key_info, &item).await?;

if !indexes.is_empty() {
Expand Down Expand Up @@ -153,3 +162,7 @@ impl SqliteEngine {
Ok((old_ret, new_ret))
}
}

#[cfg(test)]
#[path = "update_item_tests.rs"]
mod tests;
Loading
Loading