DynamoDB CDC: Make MODIFY operation also propagate deleted attributes

CHANGES.md

@@ -3,6 +3,7 @@
 
 ## Unreleased
 - MongoDB: Configure `MongoDBCrateDBConverter` after updating to commons-codec 0.0.18
+- DynamoDB CDC: Fix `MODIFY` operation to also propagate deleted attributes
 
 ## 2024/09/22 v0.0.25
 - Table Loader: Improved conditional handling of "transformation" parameter

cratedb_toolkit/io/kinesis/relay.py

@@ -84,7 +84,7 @@
 
             self.connection.commit()
         except sa.exc.ProgrammingError as ex:
-            logger.warning(f"Running query failed: {ex}")
+            logger.exception(f"Executing query failed: {ex}")
         self.progress_bar.update()
 
     def __del__(self):

pyproject.toml

@@ -143,7 +143,7 @@ docs = [
 ]
 dynamodb = [
   "boto3",
-  "commons-codec>=0.0.14",
+  "commons-codec>=0.0.19",
 ]
 full = [
   "cratedb-toolkit[cfr,cloud,datasets,io,service]",
@@ -163,11 +163,11 @@ io = [
 kinesis = [
   "aiobotocore<2.16",
   "async-kinesis<1.2",
-  "commons-codec>=0.0.14",
+  "commons-codec>=0.0.19",
   "lorrystream[carabas]>=0.0.6",
 ]
 mongodb = [
-  "commons-codec[mongodb,zyp]>=0.0.18",
+  "commons-codec[mongodb,zyp]>=0.0.19",
   "cratedb-toolkit[io]",
   "orjson<4,>=3.3.1",
   "pymongo<4.9,>=3.10.1",

tests/io/dynamodb/test_relay.py

@@ -55,6 +55,7 @@ def test_kinesis_earliest_dynamodb_cdc_insert_update(caplog, cratedb, dynamodb):
     assert cratedb.database.count_records(table_name) == 1
     results = cratedb.database.run_sql(f"SELECT * FROM {table_name}", records=True)  # noqa: S608
     assert results[0]["data"]["list_of_objects"] == [{"foo": "bar"}, {"baz": "qux"}]
+    assert "tombstone" not in results[0]["data"]
 
 
 def test_kinesis_latest_dynamodb_cdc_insert_update(caplog, cratedb, dynamodb):
@@ -104,3 +105,4 @@ def test_kinesis_latest_dynamodb_cdc_insert_update(caplog, cratedb, dynamodb):
     assert cratedb.database.count_records(table_name) == 1
     results = cratedb.database.run_sql(f"SELECT * FROM {table_name}", records=True)  # noqa: S608
     assert results[0]["data"]["list_of_objects"] == [{"foo": "bar"}, {"baz": "qux"}]
+    assert "tombstone" not in results[0]["data"]

tests/io/test_processor.py

@@ -34,6 +34,8 @@
                     "test2": {"N": 2},
                 }
             },
+            # An item that should be removed by the subsequent MODIFY operation.
+            "tombstone": {"S": "foo"},
         },
         "SizeBytes": 156,
         "ApproximateCreationDateTimePrecision": "MICROSECOND",