Fix merging records for property chunking

Author: tolik0

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -10,6 +10,7 @@
 from typing import (
     Any,
     Callable,
+    Dict,
     Iterable,
     List,
     Mapping,
@@ -404,7 +405,7 @@ def _read_pages(
                             )
                         )
                         if merge_key:
-                            merged_records[merge_key].update(current_record)
+                            deep_merge(merged_records[merge_key], current_record)
                         else:
                             # We should still emit records even if the record did not have a merge key
                             last_page_size += 1
@@ -623,6 +624,20 @@ def _to_partition_key(to_serialize: Any) -> str:
         return json.dumps(to_serialize, indent=None, separators=(",", ":"), sort_keys=True)
 
 
+def deep_merge(target: Dict[str, Any], source: Union[Record, Dict[str, Any]]) -> None:
+    """
+    Recursively merge two dictionaries, combining nested dictionaries instead of overwriting them.
+
+    :param target: The dictionary to merge into (modified in place)
+    :param source: The dictionary to merge from
+    """
+    for key, value in source.items():
+        if key in target and isinstance(target[key], dict) and isinstance(value, dict):
+            deep_merge(target[key], value)
+        else:
+            target[key] = value
+
+
 @dataclass
 class SimpleRetrieverTestReadDecorator(SimpleRetriever):
     """

unit_tests/sources/declarative/retrievers/test_simple_retriever.py

@@ -1148,6 +1148,10 @@ def test_simple_retriever_with_additional_query_properties():
                 "last_name": "hongou",
                 "nonary": "second",
                 "bracelet": "1",
+                "dict_field": {
+                    "key1": "value1",
+                    "key2": "value2",
+                },
             },
             associated_slice=None,
             stream_name=stream_name,
@@ -1216,7 +1220,12 @@ def test_simple_retriever_with_additional_query_properties():
     record_selector.select_records.side_effect = [
         [
             Record(
-                data={"id": "a", "first_name": "gentarou", "last_name": "hongou"},
+                data={
+                    "id": "a",
+                    "first_name": "gentarou",
+                    "last_name": "hongou",
+                    "dict_field": {"key1": "value1"},
+                },
                 associated_slice=None,
                 stream_name=stream_name,
             ),
@@ -1263,7 +1272,12 @@ def test_simple_retriever_with_additional_query_properties():
                 stream_name=stream_name,
             ),
             Record(
-                data={"id": "a", "nonary": "second", "bracelet": "1"},
+                data={
+                    "id": "a",
+                    "nonary": "second",
+                    "bracelet": "1",
+                    "dict_field": {"key2": "value2"},
+                },
                 associated_slice=None,
                 stream_name=stream_name,
             ),