[Feature Request]: Expose the full Debezium source metadata (pos, row, ts_ms, ...) in DebeziumIO JSON output #40701

What would you like to happen?

SourceRecordJson (the mapper behind DebeziumIO.readAsJson() and the Python, Go and YAML ReadFromDebezium transforms) only keeps six fields from the Debezium source block: connector, version, name, db, schema and table. Everything else that Debezium emits there is dropped.

As @ilyasahsan123 pointed out in #39568 (comment), this makes it impossible to deduplicate MySQL change events downstream or to see when a change was committed. Beam today gives:

"metadata": {
  "connector": "mysql",
  "version": "3.5.2.Final",
  "name": "beam-debezium-connector",
  "database": "inventory",
  "schema": "mysql-bin.000003",
  "table": "products_on_hand"
}

while Debezium itself emits:

"source": {
  "version": "3.5.2.Final", "connector": "mysql", "name": "tutorial",
  "ts_ms": 1791689147000, "snapshot": "false", "db": "inventory",
  "sequence": null, "ts_us": 1791689147000000, "ts_ns": 1791689147000000000,
  "table": "customers", "server_id": 223344, "gtid": null,
  "file": "mysql-bin.000003", "pos": 796, "row": 0, "thread": 13, "query": null
}

Missing in particular: pos and row (needed to order and deduplicate MySQL events), gtid, snapshot and ts_ms (database commit time). The same applies to other connectors, e.g. lsn and txId for PostgreSQL or change_lsn and commit_lsn for SQL Server.

Proposal

Add the complete source struct as a map under metadata.source, built from the struct's own schema, the same way before and after are already built. This is purely additive: the six existing metadata fields keep their names and values, so existing consumers keep working, and it covers every connector (and future Debezium fields) without hardcoding connector specific names.

Side note: for MySQL the binlog file is currently reported under metadata.schema. This proposal leaves that as is for compatibility, and the real file value becomes available under metadata.source.file.

Issue Priority

Priority: 2 (default / most feature requests should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector