diff --git a/crates/core/src/expression/ast.rs b/crates/core/src/expression/ast.rs index 4b2cbe751..3ec2d3aef 100755 --- a/crates/core/src/expression/ast.rs +++ b/crates/core/src/expression/ast.rs @@ -82,3 +82,53 @@ pub enum UpdateAction { /// `DELETE path value` — removes elements from a set. Delete { path: Vec, 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 { + 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"))); + } +} diff --git a/crates/core/src/expression/mod.rs b/crates/core/src/expression/mod.rs index db2613a7e..55febd0ad 100755 --- a/crates/core/src/expression/mod.rs +++ b/crates/core/src/expression/mod.rs @@ -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; diff --git a/crates/storage-cassandra/src/data/update_item.rs b/crates/storage-cassandra/src/data/update_item.rs index f90dd4bba..cbd7aeea8 100644 --- a/crates/storage-cassandra/src/data/update_item.rs +++ b/crates/storage-cassandra/src/data/update_item.rs @@ -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 = diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index 1b8bd984d..00cadc4cd 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -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 @@ -894,6 +918,9 @@ impl MongoEngine { enum AttemptOk { Committed(Option, Option), Stale, + /// The key holds no document and the update only removes: + /// nothing to write, nothing to return (issue #478). + NoWrite, } let attempt_res: Result = async { @@ -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, @@ -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; @@ -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, diff --git a/crates/storage-mongodb/src/data_engine_live_tests.rs b/crates/storage-mongodb/src/data_engine_live_tests.rs index f15761132..c7cb2a5b1 100644 --- a/crates/storage-mongodb/src/data_engine_live_tests.rs +++ b/crates/storage-mongodb/src/data_engine_live_tests.rs @@ -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 @@ -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, AttributeValue, Item); 3] = [ + let cases: [(&str, Vec, AttributeValue, Item); 2] = [ ( "set", vec![UpdateAction::Set { @@ -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 { @@ -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::(&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 diff --git a/crates/storage-postgres/src/data/update_item.rs b/crates/storage-postgres/src/data/update_item.rs index 72ce6edb9..6216b773b 100755 --- a/crates/storage-postgres/src/data/update_item.rs +++ b/crates/storage-postgres/src/data/update_item.rs @@ -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 = diff --git a/crates/storage-sqlite/src/data/update_item.rs b/crates/storage-sqlite/src/data/update_item.rs index c0875a673..22ed86b3f 100644 --- a/crates/storage-sqlite/src/data/update_item.rs +++ b/crates/storage-sqlite/src/data/update_item.rs @@ -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). @@ -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() { @@ -153,3 +162,7 @@ impl SqliteEngine { Ok((old_ret, new_ret)) } } + +#[cfg(test)] +#[path = "update_item_tests.rs"] +mod tests; diff --git a/crates/storage-sqlite/src/data/update_item_tests.rs b/crates/storage-sqlite/src/data/update_item_tests.rs new file mode 100644 index 000000000..f0f02a64e --- /dev/null +++ b/crates/storage-sqlite/src/data/update_item_tests.rs @@ -0,0 +1,150 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `update_item` on a key that holds no item (issue #478). + +use std::collections::HashMap; + +use extenddb_core::expression::{ExpressionMaps, UpdateAction, parse_update, tokenize}; +use extenddb_core::types::{AttributeValue, CreateTableInput, Item, TableKeyInfo}; +use extenddb_storage::DataEngine; +use serde_json::json; + +use crate::SqliteEngine; + +const ACCOUNT: &str = "000000000000"; + +async fn engine() -> SqliteEngine { + let engine = SqliteEngine::new(":memory:", 1, "us-east-1", 409_600) + .await + .expect("engine"); + crate::schema::apply(&engine.pool).await.expect("schema"); + sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'") + .execute(&engine.pool) + .await + .expect("delay"); + sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES (?, 'default')") + .bind(ACCOUNT) + .execute(&engine.pool) + .await + .expect("account"); + engine +} + +async fn hash_table(engine: &SqliteEngine) -> TableKeyInfo { + let input: CreateTableInput = serde_json::from_value(json!({ + "TableName": "t", + "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}], + "AttributeDefinitions": [{"AttributeName": "pk", "AttributeType": "S"}], + "BillingMode": "PAY_PER_REQUEST" + })) + .expect("input"); + let desc = engine + .create_table_impl(ACCOUNT, input, false) + .await + .expect("create"); + TableKeyInfo { + table_name: "t".to_owned(), + account_id: ACCOUNT.to_owned(), + table_id: desc.table_id.clone(), + key_schema: desc.key_schema.clone(), + base_key_schema: desc.key_schema.clone(), + attribute_definitions: desc.attribute_definitions.clone(), + ..Default::default() + } +} + +fn key(name: &str) -> Item { + let mut key = Item::new(); + key.insert("pk".to_owned(), AttributeValue::S(name.to_owned())); + key +} + +fn actions(expr: &str) -> Vec { + parse_update(&tokenize(expr).expect("tokens")).expect("actions") +} + +fn maps_with_value(value: AttributeValue) -> ExpressionMaps { + let mut values = HashMap::new(); + values.insert("v".to_owned(), value); + ExpressionMaps::new(HashMap::new(), values) +} + +async fn update( + engine: &SqliteEngine, + key_info: &TableKeyInfo, + key: &Item, + expr: &str, + maps: &ExpressionMaps, +) -> (Option, Option) { + engine + .update_item(key_info, key, &actions(expr), true, true, None, maps, None) + .await + .expect("update") +} + +#[tokio::test] +async fn removal_only_update_on_a_missing_key_writes_nothing() { + let engine = engine().await; + let key_info = hash_table(&engine).await; + let maps = maps_with_value(AttributeValue::SS(["x".to_owned()].into_iter().collect())); + + for expr in [ + "REMOVE a", + "REMOVE a, b", + "REMOVE a.b[0]", + "DELETE s :v", + "REMOVE a DELETE s :v", + ] { + let k = key(expr); + let (old, new) = update(&engine, &key_info, &k, expr, &maps).await; + assert_eq!((old, new), (None, None), "{expr}: no image to return"); + let stored = engine.get_item(&key_info, &k).await.expect("get"); + assert_eq!(stored, None, "{expr}: no item may be created"); + } +} + +#[tokio::test] +async fn set_add_and_mixed_updates_on_a_missing_key_still_create() { + let engine = engine().await; + let key_info = hash_table(&engine).await; + let maps = maps_with_value(AttributeValue::N("1".to_owned())); + + for expr in ["SET a = :v", "ADD n :v", "REMOVE a SET b = :v"] { + let k = key(expr); + let (old, new) = update(&engine, &key_info, &k, expr, &maps).await; + assert_eq!(old, None, "{expr}"); + let new = new.expect("created image"); + assert_eq!(new.len(), 2, "{expr}: key plus one attribute, got {new:?}"); + let stored = engine.get_item(&key_info, &k).await.expect("get"); + assert_eq!(stored, Some(new), "{expr}"); + } + + // No actions at all: the key-only item is created, as the service does. + let k = key("no-actions"); + let (_, new) = engine + .update_item(&key_info, &k, &[], true, true, None, &maps, None) + .await + .expect("update"); + assert_eq!(new, Some(k.clone())); + assert_eq!(engine.get_item(&key_info, &k).await.expect("get"), Some(k)); +} + +#[tokio::test] +async fn removal_on_an_existing_item_still_removes() { + let engine = engine().await; + let key_info = hash_table(&engine).await; + let maps = ExpressionMaps::default(); + let k = key("existing"); + let mut item = k.clone(); + item.insert("a".to_owned(), AttributeValue::S("1".to_owned())); + engine + .put_item(&key_info, item.clone(), false, None, &maps, None) + .await + .expect("put"); + + let (old, new) = update(&engine, &key_info, &k, "REMOVE a", &maps).await; + assert_eq!(old, Some(item)); + assert_eq!(new, Some(k.clone())); + assert_eq!(engine.get_item(&key_info, &k).await.expect("get"), Some(k)); +} diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index c760583a6..02c7c14ed 100755 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -316,7 +316,11 @@ pub trait DataEngine: Send + Sync { /// Update an item by primary key using update actions. /// /// `UpdateItem` is an upsert: if the item doesn't exist, a new item is created - /// containing the key attributes plus the SET values. + /// containing the key attributes plus the SET values. The exception is an + /// update whose actions all remove data (`REMOVE`, `DELETE`): on a missing + /// item it writes nothing, emits no stream record, and returns + /// `(None, None)` whatever `return_old` and `return_new` say + /// (`extenddb_core::expression::removes_only`). /// /// If `condition` is `Some`, evaluates the condition against the existing item /// (or empty item for new) inside a transaction. diff --git a/tests/test_update_remove_only_missing_item.py b/tests/test_update_remove_only_missing_item.py new file mode 100644 index 000000000..86c598a61 --- /dev/null +++ b/tests/test_update_remove_only_missing_item.py @@ -0,0 +1,554 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""UpdateItem whose actions only remove data, on a key that holds no item. + +Amazon DynamoDB writes nothing for such a request: no item is created, no +stream record is emitted, and the response carries no Attributes under any +ReturnValues. SET and ADD, alone or mixed with a removal, and an update with +no actions at all, still create the item. On a table with a GSI and an LSI no +index entry appears, ItemCollectionMetrics is reported from the key, and +ReturnConsumedCapacity INDEXES charges the table only. Measured 2026-10-24 in +us-east-1 on a hash table, a hash and range table, and a hash and range table +with a GSI and an LSI (issue #478). +""" + +from __future__ import annotations + +import os +import time +import uuid + +import boto3 +import pytest +from botocore.exceptions import ClientError + +from conftest import _poll_interval, scoped_table, wait_for_active, wait_for_deleted + +ALL_RETURN_VALUES = ["NONE", "ALL_OLD", "UPDATED_OLD", "ALL_NEW", "UPDATED_NEW"] + + +@pytest.fixture(scope="module") +def hash_table(dynamodb_client): + with scoped_table(dynamodb_client) as name: + yield name + + +@pytest.fixture(scope="module") +def hash_range_table(dynamodb_client): + with scoped_table( + dynamodb_client, + attribute_definitions=[ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "S"}, + ], + key_schema=[ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"}, + ], + ) as name: + yield name + + +@pytest.fixture(params=["hash", "hash_range"]) +def table(request, hash_table, hash_range_table): + """(table name, key factory) for both key shapes.""" + if request.param == "hash": + return hash_table, lambda: {"pk": {"S": f"r-{uuid.uuid4().hex[:12]}"}} + return hash_range_table, lambda: { + "pk": {"S": f"r-{uuid.uuid4().hex[:12]}"}, + "sk": {"S": "s1"}, + } + + +def _get(client, name, key): + return client.get_item(TableName=name, Key=key, ConsistentRead=True) + + +def _assert_nothing_written(client, name, key, resp): + assert "Attributes" not in resp, resp + assert "Item" not in _get(client, name, key), "the item was created" + + +class TestRemoveOnlyOnMissingKey: + @pytest.mark.parametrize("return_values", ALL_RETURN_VALUES) + def test_remove_creates_no_item(self, dynamodb_client, table, return_values): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="REMOVE a", + ReturnValues=return_values, + ) + _assert_nothing_written(dynamodb_client, name, key, resp) + + @pytest.mark.parametrize("return_values", ["NONE", "ALL_NEW", "UPDATED_NEW"]) + def test_remove_two_attributes(self, dynamodb_client, table, return_values): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="REMOVE a, b", + ReturnValues=return_values, + ) + _assert_nothing_written(dynamodb_client, name, key, resp) + + def test_remove_nested_path(self, dynamodb_client, table): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, Key=key, UpdateExpression="REMOVE a.b[0]" + ) + _assert_nothing_written(dynamodb_client, name, key, resp) + + @pytest.mark.parametrize("return_values", ["NONE", "ALL_NEW"]) + def test_delete_from_set(self, dynamodb_client, table, return_values): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="DELETE s :v", + ExpressionAttributeValues={":v": {"SS": ["x"]}}, + ReturnValues=return_values, + ) + _assert_nothing_written(dynamodb_client, name, key, resp) + + def test_remove_and_delete_together(self, dynamodb_client, table): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="REMOVE a DELETE s :v", + ExpressionAttributeValues={":v": {"SS": ["x"]}}, + ReturnValues="ALL_NEW", + ) + _assert_nothing_written(dynamodb_client, name, key, resp) + + def test_remove_with_attribute_not_exists_condition(self, dynamodb_client, table): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="REMOVE a", + ConditionExpression="attribute_not_exists(pk)", + ReturnValues="ALL_NEW", + ) + _assert_nothing_written(dynamodb_client, name, key, resp) + + def test_remove_with_attribute_exists_condition_fails(self, dynamodb_client, table): + name, new_key = table + key = new_key() + with pytest.raises(ClientError) as ei: + dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="REMOVE a", + ConditionExpression="attribute_exists(pk)", + ) + assert ei.value.response["Error"]["Code"] == "ConditionalCheckFailedException" + assert "Item" not in _get(dynamodb_client, name, key) + + def test_consumed_capacity_is_still_charged(self, dynamodb_client, table): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="REMOVE a", + ReturnConsumedCapacity="TOTAL", + ) + assert resp["ConsumedCapacity"] == {"CapacityUnits": 1.0, "TableName": name} + _assert_nothing_written(dynamodb_client, name, key, resp) + + +class TestLegacyDeleteOnMissingKey: + @pytest.mark.parametrize("return_values", ["NONE", "ALL_NEW"]) + def test_delete_without_value(self, dynamodb_client, table, return_values): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + AttributeUpdates={"a": {"Action": "DELETE"}}, + ReturnValues=return_values, + ) + _assert_nothing_written(dynamodb_client, name, key, resp) + + def test_two_deletes_without_value(self, dynamodb_client, table): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + AttributeUpdates={"a": {"Action": "DELETE"}, "b": {"Action": "DELETE"}}, + ) + _assert_nothing_written(dynamodb_client, name, key, resp) + + @pytest.mark.parametrize("return_values", ["NONE", "ALL_NEW"]) + def test_delete_with_set_value(self, dynamodb_client, table, return_values): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + AttributeUpdates={"s": {"Action": "DELETE", "Value": {"SS": ["x"]}}}, + ReturnValues=return_values, + ) + _assert_nothing_written(dynamodb_client, name, key, resp) + + +class TestUpdatesThatStillCreate: + """Controls: every other update shape creates the item as before.""" + + def test_set_only(self, dynamodb_client, table): + name, new_key = table + key = new_key() + dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="SET a = :v", + ExpressionAttributeValues={":v": {"S": "x"}}, + ) + assert _get(dynamodb_client, name, key)["Item"] == {**key, "a": {"S": "x"}} + + def test_add_only(self, dynamodb_client, table): + name, new_key = table + key = new_key() + dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="ADD n :v", + ExpressionAttributeValues={":v": {"N": "1"}}, + ) + assert _get(dynamodb_client, name, key)["Item"] == {**key, "n": {"N": "1"}} + + def test_remove_plus_set(self, dynamodb_client, table): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="REMOVE a SET b = :v", + ExpressionAttributeValues={":v": {"S": "x"}}, + ReturnValues="ALL_NEW", + ) + assert resp["Attributes"] == {**key, "b": {"S": "x"}} + assert _get(dynamodb_client, name, key)["Item"] == {**key, "b": {"S": "x"}} + + def test_remove_plus_add(self, dynamodb_client, table): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="REMOVE a ADD n :v", + ExpressionAttributeValues={":v": {"N": "1"}}, + ReturnValues="ALL_NEW", + ) + assert resp["Attributes"] == {**key, "n": {"N": "1"}} + assert _get(dynamodb_client, name, key)["Item"] == {**key, "n": {"N": "1"}} + + def test_delete_from_set_plus_add(self, dynamodb_client, table): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="DELETE s :s ADD n :n", + ExpressionAttributeValues={":s": {"SS": ["x"]}, ":n": {"N": "1"}}, + ReturnValues="ALL_NEW", + ) + assert resp["Attributes"] == {**key, "n": {"N": "1"}} + assert _get(dynamodb_client, name, key)["Item"] == {**key, "n": {"N": "1"}} + + def test_legacy_delete_plus_put(self, dynamodb_client, table): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + AttributeUpdates={ + "a": {"Action": "DELETE"}, + "b": {"Action": "PUT", "Value": {"S": "x"}}, + }, + ReturnValues="ALL_NEW", + ) + assert resp["Attributes"] == {**key, "b": {"S": "x"}} + assert _get(dynamodb_client, name, key)["Item"] == {**key, "b": {"S": "x"}} + + def test_empty_attribute_updates_creates_key_only_item(self, dynamodb_client, table): + name, new_key = table + key = new_key() + resp = dynamodb_client.update_item(TableName=name, Key=key, AttributeUpdates={}) + assert "Attributes" not in resp + assert _get(dynamodb_client, name, key)["Item"] == key + + def test_remove_on_existing_item_still_removes(self, dynamodb_client, table): + name, new_key = table + key = new_key() + dynamodb_client.put_item( + TableName=name, Item={**key, "a": {"S": "1"}, "b": {"S": "2"}} + ) + resp = dynamodb_client.update_item( + TableName=name, Key=key, UpdateExpression="REMOVE a", ReturnValues="ALL_NEW" + ) + assert resp["Attributes"] == {**key, "b": {"S": "2"}} + resp = dynamodb_client.update_item( + TableName=name, Key=key, UpdateExpression="REMOVE b", ReturnValues="ALL_NEW" + ) + assert resp["Attributes"] == key + assert _get(dynamodb_client, name, key)["Item"] == key + + +class TestAfterAWriteToTheTable: + """The same requests on a table that has already taken a write. + + A backend may serve updates on a table it has written to through a + different path from the first one, so each test writes a control item + first, then sends the removal-only update on a missing key.""" + + def _write_control(self, client, name, new_key): + control = new_key() + client.update_item( + TableName=name, + Key=control, + UpdateExpression="SET a = :v", + ExpressionAttributeValues={":v": {"S": "x"}}, + ) + return control + + def _scan_keys(self, client, name, keys): + pks = [k["pk"]["S"] for k in keys] + resp = client.scan( + TableName=name, + FilterExpression="pk IN (:k0, :k1)", + ExpressionAttributeValues={":k0": {"S": pks[0]}, ":k1": {"S": pks[1]}}, + ConsistentRead=True, + ) + return sorted(item["pk"]["S"] for item in resp["Items"]) + + def test_remove_after_a_write(self, dynamodb_client, table): + name, new_key = table + control = self._write_control(dynamodb_client, name, new_key) + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, Key=key, UpdateExpression="REMOVE a", ReturnValues="ALL_NEW" + ) + _assert_nothing_written(dynamodb_client, name, key, resp) + assert self._scan_keys(dynamodb_client, name, [control, key]) == [control["pk"]["S"]] + + def test_conditional_remove_after_a_write(self, dynamodb_client, table): + name, new_key = table + control = self._write_control(dynamodb_client, name, new_key) + key = new_key() + resp = dynamodb_client.update_item( + TableName=name, + Key=key, + UpdateExpression="REMOVE a", + ConditionExpression="attribute_not_exists(pk)", + ReturnValues="ALL_NEW", + ) + _assert_nothing_written(dynamodb_client, name, key, resp) + assert self._scan_keys(dynamodb_client, name, [control, key]) == [control["pk"]["S"]] + + +# --------------------------------------------------------------------------- +# A table with a GSI and an LSI: no index entry, and the capacity and item +# collection metrics the service returns for a removal-only update. +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def index_table(dynamodb_client): + with scoped_table( + dynamodb_client, + attribute_definitions=[ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "S"}, + {"AttributeName": "g1pk", "AttributeType": "S"}, + {"AttributeName": "l1sk", "AttributeType": "S"}, + ], + key_schema=[ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"}, + ], + GlobalSecondaryIndexes=[ + { + "IndexName": "gsi1", + "KeySchema": [{"AttributeName": "g1pk", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + } + ], + LocalSecondaryIndexes=[ + { + "IndexName": "lsi1", + "KeySchema": [ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "l1sk", "KeyType": "RANGE"}, + ], + "Projection": {"ProjectionType": "ALL"}, + } + ], + ) as name: + dynamodb_client.put_item( + TableName=name, + Item={ + "pk": {"S": "present"}, + "sk": {"S": "s1"}, + "g1pk": {"S": "g-present"}, + "l1sk": {"S": "l-present"}, + }, + ) + yield name + + +def _index_key(): + return {"pk": {"S": f"r-{uuid.uuid4().hex[:12]}"}, "sk": {"S": "s1"}} + + +class TestTableWithSecondaryIndexes: + def test_removing_index_keys_creates_no_item_or_index_entry( + self, dynamodb_client, index_table + ): + key = _index_key() + resp = dynamodb_client.update_item( + TableName=index_table, + Key=key, + UpdateExpression="REMOVE g1pk, l1sk", + ReturnValues="ALL_NEW", + ) + _assert_nothing_written(dynamodb_client, index_table, key, resp) + lsi = dynamodb_client.query( + TableName=index_table, + IndexName="lsi1", + KeyConditionExpression="pk = :p", + ExpressionAttributeValues={":p": key["pk"]}, + ) + assert lsi["Count"] == 0 + gsi = dynamodb_client.scan( + TableName=index_table, + IndexName="gsi1", + FilterExpression="pk = :p", + ExpressionAttributeValues={":p": key["pk"]}, + ) + assert gsi["Count"] == 0 + + def test_item_collection_metrics_come_from_the_key(self, dynamodb_client, index_table): + key = _index_key() + resp = dynamodb_client.update_item( + TableName=index_table, + Key=key, + UpdateExpression="REMOVE a", + ReturnItemCollectionMetrics="SIZE", + ) + assert resp["ItemCollectionMetrics"] == { + "ItemCollectionKey": {"pk": key["pk"]}, + "SizeEstimateRangeGB": [0.0, 1.0], + } + _assert_nothing_written(dynamodb_client, index_table, key, resp) + + def test_consumed_capacity_indexes_charges_the_table_only( + self, dynamodb_client, index_table + ): + key = _index_key() + resp = dynamodb_client.update_item( + TableName=index_table, + Key=key, + UpdateExpression="REMOVE g1pk", + ReturnConsumedCapacity="INDEXES", + ) + assert resp["ConsumedCapacity"] == { + "CapacityUnits": 1.0, + "Table": {"CapacityUnits": 1.0}, + "TableName": index_table, + } + _assert_nothing_written(dynamodb_client, index_table, key, resp) + + +# --------------------------------------------------------------------------- +# Streams: a REMOVE-only update on a missing key emits no record. +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def streams_client(endpoint_url): + kwargs: dict = { + "service_name": "dynamodbstreams", + "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), + } + if endpoint_url: + kwargs["endpoint_url"] = endpoint_url + if endpoint_url.startswith("https://"): + kwargs["verify"] = False + return boto3.client(**kwargs) + + +@pytest.fixture(scope="module") +def stream_table(dynamodb_client): + name = f"extenddb-test-{uuid.uuid4().hex[:12]}" + dynamodb_client.create_table( + TableName=name, + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + BillingMode="PAY_PER_REQUEST", + StreamSpecification={"StreamEnabled": True, "StreamViewType": "NEW_AND_OLD_IMAGES"}, + ) + wait_for_active(dynamodb_client, name) + arn = dynamodb_client.describe_table(TableName=name)["Table"]["LatestStreamArn"] + yield name, arn + dynamodb_client.delete_table(TableName=name) + wait_for_deleted(dynamodb_client, name) + + +def _records_for(streams_client, arn, pks: set[str]) -> list[dict]: + desc = streams_client.describe_stream(StreamArn=arn)["StreamDescription"] + found = [] + for shard in desc["Shards"]: + it = streams_client.get_shard_iterator( + StreamArn=arn, ShardId=shard["ShardId"], ShardIteratorType="TRIM_HORIZON" + )["ShardIterator"] + for _ in range(20): + if not it: + break + page = streams_client.get_records(ShardIterator=it, Limit=100) + found.extend( + r for r in page.get("Records", []) if r["dynamodb"]["Keys"]["pk"]["S"] in pks + ) + it = page.get("NextShardIterator") + if not page.get("Records"): + break + return found + + +def test_remove_only_on_missing_key_emits_no_stream_record( + dynamodb_client, streams_client, stream_table +): + name, arn = stream_table + removed = f"r-{uuid.uuid4().hex[:12]}" + control = f"r-{uuid.uuid4().hex[:12]}" + dynamodb_client.update_item( + TableName=name, Key={"pk": {"S": removed}}, UpdateExpression="REMOVE a" + ) + dynamodb_client.update_item( + TableName=name, + Key={"pk": {"S": control}}, + UpdateExpression="SET a = :v", + ExpressionAttributeValues={":v": {"S": "x"}}, + ) + # The control record proves the stream is delivering before the absence + # of a record for the REMOVE-only key means anything. + deadline = time.monotonic() + 10.0 + records: list[dict] = [] + while time.monotonic() < deadline: + records = _records_for(streams_client, arn, {removed, control}) + if any(r["dynamodb"]["Keys"]["pk"]["S"] == control for r in records): + break + time.sleep(_poll_interval()) + events = [(r["dynamodb"]["Keys"]["pk"]["S"], r["eventName"]) for r in records] + assert events == [(control, "INSERT")], events + assert "Item" not in _get(dynamodb_client, name, {"pk": {"S": removed}})