fix(mongodb): store SET values as sent in the numeric ADD pipeline update

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<NativeUpdate> {
@@ -1705,10 +1704,13 @@ impl MongoEngine {
             // Each numeric ADD is `<field> = toString(toDecimal(field
             // or 0) + delta)`; `$unset` is expressed as `<field> =
             // "$$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))
+        );
+    }
 }

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"}}