aws_dynamodb_cdc: add checkpoint_namespace for checkpoint table sharing

[squiidz]

Summary

Adds an optional checkpoint_namespace field to the aws_dynamodb_cdc input so multiple independent pipelines (e.g. one per developer or environment) can share a single checkpoint table without overwriting each other's checkpoints.

Checkpoints were keyed only by (StreamArn, ShardID) — or (TableId, ShardID) in global_table mode — so concurrent consumers of the same stream pointing at the same checkpoint_table raced on the same rows, causing skipped/duplicated events. The workaround (a checkpoint table per consumer) means IAM and infrastructure sprawl.

Behavior

• When checkpoint_namespace is set, the namespace is prefixed to the hash key value with a # delimiter: StreamArn = "dev-alice#arn:aws:dynamodb:..." (default mode), TableId = "dev-alice#my-table" (global_table mode). Each namespace occupies its own partition; shard checkpoints, snapshot progress rows, the snapshot completion sentinel, and the global-mode resume query are all scoped by it.
• The default (empty) leaves checkpoint keys byte-identical to previous behavior — no schema change, no migration, existing checkpoint tables are reused as-is.
• Namespaces are mutually invisible: changing (or removing) the namespace makes the pipeline restart from start_from. Namespaces do not coordinate consumers — two pipelines sharing the same namespace still race, exactly like today.
• A namespace containing # is rejected at config validation time. Since DynamoDB table names and stream ARNs cannot contain #, the encoding is unambiguous.

Testing

• Key construction across all four mode × namespace combinations, including that the range key is never namespaced.
• Cross-namespace isolation round-trips through a shared in-memory fake (two namespaces plus an un-namespaced reader; no crosstalk in either direction).
• Snapshot progress reads/writes and the global-mode resume partition query use the namespaced hash value.
• Config parsing (set + default-empty) and validation (# rejected).
• All pre-existing checkpointer tests pass unmodified, which pins the empty-namespace backward-compatibility guarantee.

[claude]

Commits
LGTM

Review
Reviewed the addition of the optional checkpoint_namespace field to the aws_dynamodb_cdc input. The namespace is prefixed to the checkpoint hash key value uniformly via hashKeyValue() across shard checkpoints, snapshot progress, and the global-mode resume query, while the hash key attribute name and the range key remain unchanged. The empty default preserves byte-identical behavior for existing deployments, the # delimiter is rejected at config validation, and the raw StreamArn discriminator in global-mode prepareResume stays correct because the partition query is already namespace-scoped. Config, docs, changelog, and tests are consistent with project patterns and thorough.

LGTM

[claude]

Commits
LGTM

Review
Reviewed the addition of the optional checkpoint_namespace field to the aws_dynamodb_cdc input. The change prefixes the namespace to the checkpoint table hash key value (StreamArn in default mode, TableId in global mode) via the new hashKeyValue()/hashAttrName() helpers.

I verified that every hash-key construction site (Get, Set, the snapshot completion sentinel, the snapshot segment query/write, MarkSnapshotComplete, and the global-mode prepareResume query) consistently routes through these helpers, so reads and writes stay in sync. The empty-namespace path returns the raw value unchanged, preserving byte-identical backward compatibility. In global mode the StreamArn non-key attribute used for own/foreign classification in prepareResume correctly remains un-namespaced. The # delimiter is rejected at config validation time, and the encoding is unambiguous since DynamoDB table names and stream ARNs cannot contain #.

The CDC conformance gate at internal/plugins/cdctest only asserts presence of canonical fields and does not reject additional ones, so the new field does not break it; the connector's pre-existing non-canonical checkpoint naming is waived there and out of scope for this change. Unit tests cover all four mode × namespace combinations, cross-namespace isolation, snapshot paths, and config parse/validation.

LGTM

[claude]

Commits

LGTM

Review

This PR adds an optional checkpoint_namespace field to the aws_dynamodb_cdc input, prefixing it to the checkpoint hash key value so multiple pipelines can safely share a single checkpoint table. The change is well-scoped and cleanly implemented: the hash-value/attr-name logic is consolidated into hashKeyValue()/hashAttrName() and consistently threaded through the key builder, error messages, resume query, and snapshot progress paths. The raw StreamArn non-key attribute correctly stays un-namespaced (it disambiguates own vs foreign region rows within an already-namespaced partition). The empty default preserves byte-identical keys for existing deployments, # is rejected at validation time, and the field follows established config patterns. Test coverage is thorough (prefixing, cross-namespace isolation, snapshot/resume namespacing, config parsing, and validation).

LGTM

[josephwoodward]

thought (non-blocking): would it be worth moving this into a lint rule for the connector?