DMS: Serialization flaw: DDL/DML events are processed out of order

[amotl]

Problem

We are using an AWS Lambda for processing DMS events from a Kinesis stream, and observed out-of-order processing behaviour. In our case, there seems to be a race condition at the beginning of the load where INSERT is attempted before CREATE TABLE.

Thoughts

@hammerhead said:

This makes me think if we have any issue with the serialization of CDC events. For plain INSERTs, the order in which we receive those CDC events from Kinesis and execute them doesn’t matter. But as soon as we have let’s say two UPDATE statements changing the same record, we need to be sure to execute them in the right order (latest should win). Not sure yet how Lambda behaves in terms of concurrency and if Kinesis maintains any order of events.

[amotl]

We consulted Google Gemini, and it said

to use the table name as partition key in Kinesis. Each table will then go to its own shard, and ordering is guaranteed within shards. If the Lambda function is set to ParallelizationFactor=1, it will only run once per Kinesis shard. It may receive batches from Kinesis, but all should be fine as long as batches are executed sequentially based on the sequenceNumber that is included in the metadata.

Bonus challenge: What about retries? How can a retry know if it should still attempt to re-execute or if it has been superseded by another change? 😵‍💫

The full wisdom of Google Gemini for reference: https://g.co/gemini/share/ce9684c79777

[amotl]

Wow, thanks. So, making this outcome tangible will be about, citing Gemini:

• adjusting the DMS task configuration, specifically the Kinesis partition key in the DMS target endpoint settings. => How?
• making sure the Lambda's ParallelizationFactor setting uses its default value (1) to ensure sequential processing of batches from a single shard. => We should include it into the IaC configuration, right?

[amotl]

We figured out the DMS table-/column mappings would be the right spot to adjust the partition key in Kinesis, so that operations concerning a specific table will go to a dedicated shard. @hammerhead tried to work out a configuration, and came up with those DMS mapping rules, suitable for two tables public.foo and public.boo. Thanks!

boo and foo go to different shards, using their respective values of schema.table as the partitioning key.

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "select-all-tables",
            "object-locator": {
                "schema-name": "public",
                "table-name": "%"
            },
            "rule-action": "include"
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "2",
            "rule-name": "DefaultMapToKinesis",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "public",
                "table-name": "foo"
            },
            "filters": []
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "3",
            "rule-name": "DefaultMapToKinesisTwo",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "public",
                "table-name": "boo"
            },
            "filters": []
        }
    ]
}

[amotl]

^^ Regarding the observation above, it looks like object-mapping rules are not strictly required, and also do not improve the situation with partition key assignments.

[amotl]

Problem

In full-load operation mode, AWS DMS seems to select partition-key-type: primary-key when converging a source table from RDS PostgreSQL using pglogical into a Kinesis stream.

Observations

This leads to additional Kinesis attributes like partition-key-value: public.arbitrary.34, including the primary key value of the corresponding record. That might be responsible that CDC events are processed out of order, because each record could be addressed into a different partition after hashing the partition key.

After another round of explorations, we discovered the data transfer goes south when DMS selects 'partition-key-type': 'primary-key' for a table. On the other hand, it works well when DMS selects 'partition-key-type': 'schema-table'.

Details

Based on our explorations, the former (selecting the primary-key partition key type) happens if a table already exists when starting the DMS replication task (full-load mode), while the latter (selecting the schema-table partition key type) happens if a table is created while the replication task is running already (cdc mode).

The difference between both variants looks like this when inspecting the DMS events on the Kinesis wire:

{'data': {'data': 'foo', 'id': 34},
 'metadata': {'operation': 'load',
              'partition-key-type': 'primary-key',
              'partition-key-value': 'public.arbitrary.34',
              'record-type': 'data',
              'schema-name': 'public',
              'table-name': 'arbitrary',
              'timestamp': '2025-06-20T14:49:57.211961Z'}}

{'data': {'data': 'foo', 'id': 34},
 'metadata': {'commit-timestamp': '2025-06-20T14:53:52.828993Z',
              'operation': 'insert',
              'partition-key-type': 'schema-table',
              'prev-transaction-id': 8,
              'prev-transaction-record-id': 1,
              'record-type': 'data',
              'schema-name': 'public',
              'stream-position': '00000037/F0015020.3.00000037/F0015020',
              'table-name': 'anything',
              'timestamp': '2025-06-20T14:53:52.869780Z',
              'transaction-id': 8503,
              'transaction-record-id': 1}}

In this spirit, it looks like only the DMS full-load operation mode is impacted by this obstacle. It also looks like we are not the only ones who would like to control this setting, but can't discover how:

See also

Asked on SO, 11 months ago:

How can I use SchemaName.TableName partition key in AWS DMS to migrate data to an Amazon Kinesis data stream?

I'm trying to set up DMS to push data to kinesis stream, with partition keys based on the schema and table name. The intent is to get all cdc events for a given table hitting a single shard, to better guarantee on ordering of events for a given table (though I know it's not a 100% guarantee).

I'd like to use the setting that the docs says is supported (and in some other places says is the default!):

AWS DMS supports the following two forms for partition keys:

• SchemaName.TableName: A combination of the schema and table name.
• ${AttributeName}: The value of one of the fields in the JSON, or the primary key of the table in the source database.

But, I do not see any setting in the AWS console or aws cdk to use the former, and my partition keys are showing as the primary key of the data within the source db (which is aurora postgresSQL, if it matters). I can append the schema and table to the start but that doesn't solve the issue of getting 1 table to 1 shard.

[amotl]

mapping-parameters.partition-key-type: schema-table FTW

It is clearly an advantage to read the documentation right:

When using ParallelApply* task settings, the partition-key-type default is the primary-key of the table, not schema-name.table-name.
-- https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Kinesis.html

We are now probing the machinery using those table mapping settings:

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "select-all-tables",
            "object-locator": {
                "schema-name": "public",
                "table-name": "%"
            },
            "rule-action": "include"
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "2",
            "rule-name": "2",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "public",
                "table-name": "diamonds"
            },
            "mapping-parameters": {
                "partition-key-type": "schema-table"
            }
        }        
    ]
}

Others have been struggling with the mapping-parameters setting as well.

-- https://stackoverflow.com/questions/54535540/amazon-dms-task-with-custom-rules-fails-when-sinking-to-kinesis

However, it works for us. This is a load DMS event (presumably originating from processing in full-load operation mode), and it still uses the partition-key-type: schema-table.

{'data': {'carat': 0.23,
          'depth': 61.5,
          'price': 326,
          'table': 55,
          'x': 3.95,
          'y': 3.98,
          'z': 2.43},
 'metadata': {'operation': 'load',
              'partition-key-type': 'schema-table',
              'record-type': 'data',
              'schema-name': 'public',
              'table-name': 'diamonds',
              'timestamp': '2025-06-20T17:46:43.388001Z'}}

Yay?

[amotl]

Solution?

The solution is to use this configuration snippet within the table mapping settings:

"mapping-parameters": {
    "partition-key-type": "schema-table"
}

Based on our evaluations around needing object-mapping.map-record-to-record, and how to configure it, we found that you really need to enumerate each table by using a dedicated item.

When using ParallelApply* task settings, the partition-key-type default is the primary-key of the table, not schema-name.table-name.
-- https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Kinesis.html

Our observations are that even when not using those settings, DMS will use partition-key-type: primary-key by default, but needs explicit configuration for each table to use partition-key-type: schema-table instead. We think using the former option by default, as suggested by the documentation, would be a safer bet.

[amotl]

No way, still borked!

Example

On behalf of the public.astronauts table, you can observe the events are not processed in order, probably because they are relayed on different Kinesis shards.

shard_id	ts	op	partition-key-type	partition-key-value	record-type	table
shardId-000000000002	2025-06-20T21:26:19.819180Z	load	schema-table		data	public.astronauts
shardId-000000000002	2025-06-20T21:26:19.833024Z	load	schema-table		data	public.astronauts
...	...	...	...		...	...
shardId-000000000003	2025-06-20T21:26:19.402581Z	drop-table	task-id	DZ6OKLFYA5BWVG4RYROQUHXU4E	control	public.astronauts
shardId-000000000003	2025-06-20T21:26:19.413246Z	create-table	task-id	DZ6OKLFYA5BWVG4RYROQUHXU4E	control	public.astronauts

Results

We are using an example dataset from the Ibis project, replayed its DDL/DML statements into RDS PostgreSQL, ...

• CDC: Add canonical SQL example for PostgreSQL from Ibis cratedb-toolkit#457

... then conducting a DMS full-load operation on it.

• Kinesis stream dump: dms-over-kinesis-20250620c.json (JSONL/NDJSON format)
• Metadata overview: dms-over-kinesis-20250620c.metadata.txt (ASCII format)

NB: Instructions how to create these files

LORRY_BACKEND=ingestr python examples/aws/kinesis_subscribe.py arn:aws:kinesis:eu-central-1:931394475905:stream/testdrive-dms-postgresql-dev-stream > dms-over-kinesis-20250620c.json

qv dms-over-kinesis-20250620c.json -l 999 -q "select kinesis['shard_id'] as shard_id, metadata['timestamp'] as ts, metadata['operation'] as op, metadata['partition-key-type'] as 'partition-key-type', metadata['partition-key-value'] as 'partition-key-value', metadata['record-type'] as 'record-type', metadata['schema-name'] || '.' || metadata['table-name'] as table from tbl" > dms-over-kinesis-20250620c.metadata.txt

[amotl]

Twilight zone?

Question

There must be a way to make DMS force all events concerning a specific table into a single outbound Kinesis stream shard, in order to sustain message order when processing events on the receiver's end?

Research

The document about using object mapping to migrate data to a Kinesis data stream includes a few valuable bits of information:

Kinesis data streams don't have a preset structure other than having a partition key. In an object mapping rule, the possible values of a partition-key-type for data records are schema-table, transaction-id, primary-key, constant, and attribute-name.

Primarily at the second example within the "Restructuring data with attribute mapping" section:

To set a constant value for partition-key, specify a partition-key value. For example, you might do this to force all the data to be stored in a single shard. The following mapping illustrates this approach.

However, there is a note below:

Note

The partition-key value for a control record that is for a specific table is TaskId.SchemaName.TableName. The partition-key value for a control record that is for a specific task is that record's TaskId. Specifying a partition-key value in the object mapping has no impact on the partition-key for a control record.

[amotl]

Workaround

Solution

Let's tune down the Kinesis stream by switching from capacity mode ON_DEMAND to PROVISIONED, with a fixed number of provisioned shards, 1. See also daq-tools/lorrystream@e3e501879.

Result

Using a single shard naturally keeps events in the same order when consuming them.

• dms-over-kinesis-20250621a.json
• dms-over-kinesis-20250621a.metadata.txt

shard_id	ts	op	partition-key-type	partition-key-value	record-type	table
shardId-000000000000	2025-06-20T23:20:48.943135Z	drop-table	task-id	DZ6OKLFYA5BWVG4RYROQUHXU4E	control	public.astronauts
shardId-000000000000	2025-06-20T23:20:48.953985Z	create-table	task-id	DZ6OKLFYA5BWVG4RYROQUHXU4E	control	public.astronauts
...	...	...	...		...	...
shardId-000000000000	2025-06-20T23:20:49.699299Z	load	schema-table		data	public.astronauts
shardId-000000000000	2025-06-20T23:20:49.699038Z	load	schema-table		data	public.astronauts

[amotl]

@hammerhead suggests:

Ok, understand. For a proof-of-concept, we can create tables manually upfront, that’s fine. Potential ALTER TABLE statements during runtime we can park for now.

[amotl]

Alternative workaround

We've added an option to ignore DDL statements.

• DMS: Add option ignore_ddl to ignore DDL events #119

If you want to provide the SQL DDL manually, please toggle settings.ignore_ddl: true, and provide the primary key information in the pk section instead.

-- https://cratedb-toolkit.readthedocs.io/io/dms/standalone.html#configuration

[amotl]

Question: Let's elaborate this very obstacle on the documentation, and refer to the two available workarounds to cope with the situation in order to wrap this up?