diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index bb75560d3..dd8f4cc9f 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -805,7 +805,7 @@ impl MongoEngine { && !return_old && stream.is_none() && self.gsi_cache_get_fresh(&key_info.table_id) == Some(false) - && let Some(mongo_update) = self.try_build_native_update(actions, maps) + && let Some(mongo_update) = Self::try_build_native_update(actions, maps) { let (fast_filter, took_fast) = match mongo_update { NativeUpdate::Doc(d) => { @@ -1630,7 +1630,6 @@ impl MongoEngine { /// because mongo serializes doc-scoped write locks around the /// pipeline's read + compute + write, no OCC retry needed. fn try_build_native_update( - &self, actions: &[UpdateAction], maps: &ExpressionMaps, ) -> Option { @@ -1705,10 +1704,13 @@ impl MongoEngine { // Each numeric ADD is ` = toString(toDecimal(field // or 0) + delta)`; `$unset` is expressed as ` = // "$$REMOVE"`; SET actions are literal assignments. `_v` - // is bumped in the same stage. + // is bumped in the same stage. `$literal` makes each SET value a + // constant: a `$` string or member name is not read as an + // expression, and a document value replaces the attribute instead + // of merging into the embedded document it replaces. let mut stage: Document = Document::new(); for (k, v) in &set_doc { - stage.insert(k, v.clone()); + stage.insert(k, doc! { "$literal": v.clone() }); } for k in unset_doc.keys() { stage.insert(k, "$$REMOVE"); @@ -4428,4 +4430,38 @@ transaction_read_concern = "majority""#, assert_eq!(native_attribute_name(&path, &maps), Some("name".to_owned())); } + + #[test] + fn native_update_pipeline_sets_each_value_as_a_literal() { + let names = std::collections::HashMap::from([ + ("s".to_owned(), "s".to_owned()), + ("p".to_owned(), "p".to_owned()), + ("n".to_owned(), "n".to_owned()), + ]); + let values = std::collections::HashMap::from([ + ("dollar".to_owned(), AttributeValue::S("$item_data".into())), + ("plain".to_owned(), AttributeValue::S("x".into())), + ("one".to_owned(), AttributeValue::N("1".into())), + ]); + let maps = ExpressionMaps::new(names, values); + let tokens = expression::tokenize("SET #s = :dollar, #p = :plain ADD #n :one").unwrap(); + let actions = expression::parse_update(&tokens).unwrap(); + + let Some(NativeUpdate::Pipeline { pipeline, .. }) = + MongoEngine::try_build_native_update(&actions, &maps) + else { + panic!("a SET with a numeric ADD builds a pipeline update"); + }; + let stage = pipeline[0].get_document("$set").unwrap(); + // Every SET value is wrapped, not only the ones whose text reads as a + // field path: a plain string under a `$literal` stays a string, and + // without the wrapper a later read of a non-string attribute fails. + let want = doc! { "$literal": { "S": "$item_data" } }; + assert_eq!(stage.get("item_data.s"), Some(&bson::Bson::Document(want))); + let want_plain = doc! { "$literal": { "S": "x" } }; + assert_eq!( + stage.get("item_data.p"), + Some(&bson::Bson::Document(want_plain)) + ); + } } diff --git a/tests/test_update_item_dollar_values.py b/tests/test_update_item_dollar_values.py new file mode 100644 index 000000000..06174a959 --- /dev/null +++ b/tests/test_update_item_dollar_values.py @@ -0,0 +1,158 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""UpdateItem SET of values that hold `$` strings or `$` member names. + +Amazon DynamoDB stores every SET value as sent, with or without a numeric ADD +in the same update: strings such as ``$item_data``, ``$$ROOT``, ``$`` and +``$$REMOVE``, string sets and lists holding them, maps whose values hold +them, maps with members named ``$k`` or ``$literal``, and maps with members +whose names hold ``.`` (measured with the ADD on 2026-10-25 at 23:15 UTC and without +it on 2026-10-26 at 00:55 UTC, us-east-1, each case twice). A SET over an +attribute the item already holds replaces it, and a REMOVE in the same update +as the ADD removes its attribute (2026-10-26 at 00:55 UTC). + +On MongoDB an unconditional update that asks for no old values and no +consumed capacity, on a table with no index and no stream, is built as a +native update while the server records that the table has no index. The +first write on the session-scoped read-modify-write path that finds no such +record, or one older than 60 seconds, reads the index catalog and records the +answer; no write renews the record during its 60 seconds. With a numeric ADD +the native update is an aggregation pipeline, where a `$` string reads as a +field path. Each test gets a table of its own and writes an item before its +update. That write takes the session-scoped path and records that the table +has no index, so the update runs as a native update, which is the pipeline +when it has the ADD. + +Dual-target against Amazon DynamoDB and extenddb. +""" + +from __future__ import annotations + +import pytest + +from conftest import scoped_table + +CASES = { + "s_dollar_field": {"S": "$item_data"}, + "s_dollar_dollar_root": {"S": "$$ROOT"}, + "s_dollar_only": {"S": "$"}, + "s_dollar_remove": {"S": "$$REMOVE"}, + "s_literal_text": {"S": '{"$literal": 1}'}, + "ss_dollar": {"SS": ["$a", "$$ROOT", "b"]}, + "l_dollar": {"L": [{"S": "$a"}, {"N": "1"}, {"M": {"k": {"S": "$b"}}}]}, + "m_dollar_value": {"M": {"k": {"S": "$item_data.pk"}}}, + "m_dollar_member_name": {"M": {"$k": {"S": "v"}}}, + "m_member_named_literal": {"M": {"$literal": {"S": "v"}}}, + "m_dotted_member_names": { + "M": {"a.b": {"S": "v"}, "$c.d": {"S": "$e"}, "f": {"M": {"g.h": {"N": "2"}}}} + }, + # Unlike m_dotted_member_names, this map holds no `$`. In the pipeline stage + # MongoDB 7.0.45 refuses it as it is (error 40183, "cannot use dotted field + # name 'a.b' in a sub object") and stores it as sent inside `$literal` + # (measured on 2026-10-27 with the stage this server builds). + "m_dotted_member_name_only": {"M": {"a.b": {"S": "v"}}}, + "n_value": {"N": "5"}, + "bool_value": {"BOOL": True}, + "null_value": {"NULL": True}, + "b_value": {"B": b"$x"}, +} + + +@pytest.fixture +def table(dynamodb_client): + with scoped_table(dynamodb_client) as name: + yield name + + +def _normalize(value): + """String sets come back in no fixed order.""" + if "SS" in value: + return {"SS": sorted(value["SS"])} + return value + + +@pytest.mark.parametrize("with_add", [True, False], ids=["set_and_add", "set_alone"]) +@pytest.mark.parametrize("case", sorted(CASES)) +def test_set_stores_the_value_as_sent(dynamodb_client, table, case, with_add): + value = CASES[case] + key = {"pk": {"S": case}} + dynamodb_client.put_item(TableName=table, Item={**key, "v": {"S": "x"}}) + names = {"#s": "s"} + values = {":s": value} + expression = "SET #s = :s" + if with_add: + names["#n"] = "n" + values[":one"] = {"N": "1"} + expression += " ADD #n :one" + dynamodb_client.update_item( + TableName=table, + Key=key, + UpdateExpression=expression, + ExpressionAttributeNames=names, + ExpressionAttributeValues=values, + ) + item = dynamodb_client.get_item(TableName=table, Key=key, ConsistentRead=True)["Item"] + want = {**key, "v": {"S": "x"}, "s": value} + if with_add: + want["n"] = {"N": "1"} + assert {k: _normalize(v) for k, v in item.items()} == { + k: _normalize(v) for k, v in want.items() + } + + +# A SET over an attribute the item already holds replaces it, with or without a +# numeric ADD in the same update (measured on 2026-10-26 at 00:55 UTC). In a +# MongoDB pipeline a document value merges into the embedded document it +# replaces unless it is a constant. +OVER = { + "over_s_to_n": ({"S": "x"}, {"N": "5"}), + "over_m_to_m": ({"M": {"a": {"S": "1"}}}, {"M": {"b": {"S": "2"}}}), + "over_m_nested": ( + {"M": {"a": {"M": {"x": {"S": "1"}}}}}, + {"M": {"a": {"M": {"y": {"S": "2"}}}}}, + ), + "over_l_to_m": ({"L": [{"S": "1"}]}, {"M": {"b": {"S": "2"}}}), +} + + +@pytest.mark.parametrize("with_add", [True, False], ids=["set_and_add", "set_alone"]) +@pytest.mark.parametrize("case", sorted(OVER)) +def test_set_replaces_the_value_the_item_holds(dynamodb_client, table, case, with_add): + old, new = OVER[case] + key = {"pk": {"S": case}} + dynamodb_client.put_item(TableName=table, Item={**key, "s": old}) + names = {"#s": "s"} + values = {":s": new} + expression = "SET #s = :s" + if with_add: + names["#n"] = "n" + values[":one"] = {"N": "1"} + expression += " ADD #n :one" + dynamodb_client.update_item( + TableName=table, + Key=key, + UpdateExpression=expression, + ExpressionAttributeNames=names, + ExpressionAttributeValues=values, + ) + item = dynamodb_client.get_item(TableName=table, Key=key, ConsistentRead=True)["Item"] + want = {**key, "s": new} + if with_add: + want["n"] = {"N": "1"} + assert item == want + + +def test_remove_in_the_same_update_as_add_removes_the_attribute(dynamodb_client, table): + """Measured on 2026-10-26 at 00:55 UTC: r is removed, s and n are written.""" + key = {"pk": {"S": "remove_with_add"}} + dynamodb_client.put_item(TableName=table, Item={**key, "r": {"S": "gone"}}) + dynamodb_client.update_item( + TableName=table, + Key=key, + UpdateExpression="SET #s = :s REMOVE #r ADD #n :one", + ExpressionAttributeNames={"#s": "s", "#r": "r", "#n": "n"}, + ExpressionAttributeValues={":s": {"S": "$$REMOVE"}, ":one": {"N": "1"}}, + ) + item = dynamodb_client.get_item(TableName=table, Key=key, ConsistentRead=True)["Item"] + assert item == {**key, "s": {"S": "$$REMOVE"}, "n": {"N": "1"}}