Is _dynamo_op_type needed to be filled in and written for DynamoDB migration with streaming ?

[tarzanek]

Can the additional field _dynamo_op_type be avoided when writing DELETES ?

https://github.com/scylladb/scylla-migrator/blob/master/migrator/src/main/scala/com/scylladb/migrator/writers/DynamoStreamReplication.scala#L56
populates _dynamo_op_type
and this gets later written to DB

even for delete, which of course leaves just rows with the key and just this attribute around

[tarzanek]

old audienceproject/kinesis library used _dynamo_op_type as hint to replay writes
https://github.com/scylladb/scylla-migrator/tree/c8a05d3086d59ab6b00e1274ea0632117df97901

but the question is if it didn't silently remove it from row and didn't write it to db (if it did, then this is not a bug, though it's still inconvenient to have leftovers in DB after delete)

[tarzanek]

We executed the migrator on Dataproc Serverless Spark (v2.2) which runs spark version 3.5.1 and scala version 2.13. However observation is the same while running on the docker container also, i.e. it enabled DynamoDB streams on the DynamoDB table.

[tarzanek]

older version of migrator doesn't have this problem, so this looks like a bug

[tarzanek]

we can most probably filter the column introduced by 5bba427#diff-6fdbb9649ce553c8b808e661105610309c4ba1e466c2e06fcd092c1418e48a73

since https://github.com/search?q=repo%3Aaudienceproject%2Fspark-dynamodb+ColumnSchema&type=code translated it into class that was doing writes, while now we write directly, so this column needs to be filtered

[tarzanek]

--- a/migrator/src/main/scala/com/scylladb/migrator/writers/DynamoStreamReplication.scala
+++ b/migrator/src/main/scala/com/scylladb/migrator/writers/DynamoStreamReplication.scala
@@ -108,6 +108,7 @@ object DynamoStreamReplication {
               item
                 .entrySet()
                 .stream()
+                .filter(e => !e.getKey.equals(operationTypeColumn))
                 .collect(
                   Collectors.toMap(
                     (e: util.Map.Entry[String, AttributeValueV1]) => e.getKey,

should be quick fix, need to test it

[pizzaeueu]

Hey @tarzanek
Could you please take a look at #239 ?

this column needs to be filtered

It's filtered in the PR, but separate deleteRDD was introduced for delete operation
Does it make sense to you?