Speed up DynamoDB query deepcopy while preserving isolation (getmoto#9670)

Author: Will282

moto/dynamodb/models/dynamo_type.py

@@ -1,7 +1,7 @@
 import base64
 import copy
 from decimal import Decimal
-from typing import Any, Optional, Union
+from typing import Any, Optional, Union, cast
 
 from boto3.dynamodb.types import TypeDeserializer, TypeSerializer
 from botocore.utils import merge_dicts
@@ -97,6 +97,25 @@ def __ge__(self, other: "DynamoType") -> bool:
     def __repr__(self) -> str:
         return f"DynamoType: {self.to_json()}"
 
+    def __deepcopy__(self, memo: dict[int, Any]) -> "DynamoType":
+        """Custom deepcopy that bypasses the expensive default pickle-based
+        __reduce_ex__ protocol. Safe for DynamoDB AttributeValue trees,
+        which are acyclic."""
+        if id(self) in memo:  # pragma: no cover
+            return memo[id(self)]  # Circular refs can't occur in DynamoDB data
+        result = self.__class__.__new__(self.__class__)
+        memo[id(self)] = result
+        result.type = self.type
+        if self.is_list():
+            result.value = [copy.deepcopy(v, memo) for v in self.value]
+        elif self.is_map():
+            result.value = {k: copy.deepcopy(v, memo) for k, v in self.value.items()}
+        elif self.is_set():
+            result.value = list(self.value)
+        else:
+            result.value = self.value
+        return result
+
     def __add__(self, other: "DynamoType") -> "DynamoType":
         if self.type != other.type:
             raise TypeError("Different types of operandi is not allowed.")
@@ -323,6 +342,26 @@ def __eq__(self, other: "Item") -> bool:  # type: ignore[override]
     def __repr__(self) -> str:
         return f"Item: {self.to_json()}"
 
+    def __deepcopy__(self, memo: dict[int, Any]) -> "Item":
+        """Custom deepcopy that bypasses the expensive default pickle-based
+        __reduce_ex__ protocol, and skips LimitedSizeDict size validation
+        during copy (the source item already passed validation on write)."""
+        if id(self) in memo:  # pragma: no cover
+            return memo[id(self)]  # Circular refs can't occur in DynamoDB data
+        result = cast(Item, self.__class__.__new__(self.__class__))
+        memo[id(self)] = result
+        result.hash_key = copy.deepcopy(self.hash_key, memo)
+        result.range_key = copy.deepcopy(self.range_key, memo)
+        # Bypass LimitedSizeDict.__setitem__ which runs O(n) size validation on
+        # every insert. The source item already passed validation, so re-checking
+        # during copy is unnecessary and would make the overall copy O(n²).
+        attrs_copy = LimitedSizeDict.__new__(LimitedSizeDict)
+        dict.__init__(attrs_copy)
+        for key, value in self.attrs.items():
+            dict.__setitem__(attrs_copy, key, copy.deepcopy(value, memo))
+        result.attrs = attrs_copy
+        return result
+
     def size(self) -> int:
         return sum(bytesize(key) + value.size() for key, value in self.attrs.items())
 

tests/test_dynamodb/test_dynamodb_query.py

@@ -1,11 +1,15 @@
+import copy
 from decimal import Decimal
 from uuid import uuid4
 
 import boto3
 import pytest
 from boto3.dynamodb.conditions import Attr, Key
 
-from moto import mock_aws
+from moto import mock_aws, settings
+from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
+from moto.dynamodb import dynamodb_backends
+from moto.dynamodb.models import DynamoType
 
 from . import dynamodb_aws_verified
 
@@ -153,6 +157,91 @@ def test_key_condition_expressions():
     assert results["Count"] == 1
 
 
+@mock_aws
+@pytest.mark.requires_clean_slate
+def test_query_returns_detached_items():
+    if settings.TEST_SERVER_MODE:
+        pytest.skip("Can't access backend directly in server mode")
+    dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
+    table_name = f"T{uuid4()}"
+    table = dynamodb.create_table(
+        TableName=table_name,
+        KeySchema=[
+            {"AttributeName": "pk", "KeyType": "HASH"},
+            {"AttributeName": "sk", "KeyType": "RANGE"},
+        ],
+        AttributeDefinitions=[
+            {"AttributeName": "pk", "AttributeType": "S"},
+            {"AttributeName": "sk", "AttributeType": "S"},
+        ],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )
+
+    table.put_item(
+        Item={
+            "pk": "hash",
+            "sk": "1",
+            "payload": {"nested": "original"},
+            "tags": ["a", "b"],
+        }
+    )
+
+    # Use backend API to verify Item isolation (boto3 returns plain dicts).
+    backend = dynamodb_backends[ACCOUNT_ID]["us-east-1"]
+    items, _, _ = backend.query(
+        table_name,
+        {"S": "hash"},
+        None,
+        [],
+        limit=10,
+        exclusive_start_key=None,
+        scan_index_forward=True,
+        projection_expressions=None,
+        index_name=None,
+        consistent_read=False,
+        expr_names=None,
+        expr_values=None,
+        filter_expression=None,
+    )
+
+    item = items[0]
+    item.attrs["payload"].value["nested"].value = "changed"
+    item.attrs["tags"].value[0].value = "changed"
+
+    stored_item = backend.get_item(table_name, {"pk": {"S": "hash"}, "sk": {"S": "1"}})
+    assert stored_item is not None
+    assert stored_item is not item
+    assert stored_item.attrs["payload"].value["nested"].value == "original"
+    assert stored_item.attrs["tags"].value[0].value == "a"
+
+
+@pytest.mark.parametrize(
+    "attr_value",
+    [
+        pytest.param({"S": "hello"}, id="string"),
+        pytest.param({"N": "123"}, id="number"),
+        pytest.param({"B": "aGVsbG8="}, id="binary"),
+        pytest.param({"BOOL": True}, id="boolean"),
+        pytest.param({"NULL": True}, id="null"),
+        pytest.param({"SS": ["a", "b", "c"]}, id="string_set"),
+        pytest.param({"NS": ["1", "2", "3"]}, id="number_set"),
+        pytest.param({"BS": ["aGVsbG8=", "d29ybGQ="]}, id="binary_set"),
+        pytest.param({"L": [{"S": "a"}, {"N": "1"}]}, id="list"),
+        pytest.param({"M": {"key": {"S": "value"}}}, id="map"),
+    ],
+)
+def test_dynamotype_deepcopy_all_types(attr_value):
+    original = DynamoType(attr_value)
+    copied = copy.deepcopy(original)
+
+    assert copied == original
+    assert copied is not original
+    assert copied.type == original.type
+    # For mutable containers, verify the value is a different object
+    if original.is_list() or original.is_map() or original.is_set():
+        assert copied.value is not original.value
+
+
 @pytest.mark.aws_verified
 @dynamodb_aws_verified(add_range=True)
 def test_query_pagination(table_name=None):