Repository navigation
[DebeziumIO] Add opt-in full metadata (op and Debezium source fields) to readAsJson - #40702
Conversation
There was a problem hiding this comment.
📋 Review Summary
This PR exposes the complete Debezium source block under metadata.source in SourceRecordJson while preserving the existing top-level metadata fields for backward compatibility. The implementation is clean, preserves field ordering via LinkedHashMap, and includes comprehensive updates to unit tests, multi-language (Java, Python, Go) integration tests, documentation, and CHANGES.md.
🔍 General Feedback
- The additive approach (
metadata.source) is well-designed and minimizes disruption while providing necessary connector-specific metadata (pos,row,lsn,ts_ms, etc.) for deduplication and commit-time tracking. - Integration tests across Java, Python, and Go properly validate the presence of dynamic source fields before stripping
metadata.sourcefor deterministic event comparison.
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
b895b27 to
66195eb
Compare
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #40702 +/- ##
===========================================
- Coverage 80.04% 56.45% -23.60%
===========================================
Files 490 871 +381
Lines 82916 154217 +71301
===========================================
+ Hits 66373 87058 +20685
- Misses 16543 65303 +48760
- Partials 0 1856 +1856
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Assigning reviewers: R: @shunping for label python. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
Abacn
left a comment
There was a problem hiding this comment.
We can collect feedbacks and polish the user exposing API as well as the row format
cc: @ilyasahsan123
| source.getString("db"), | ||
| schema, | ||
| source.getString("table")); | ||
| source.getString("table"), |
There was a problem hiding this comment.
We can add "table" field. For full source field, consider add a transform option that control whether preserve full "source"? We can either
- default it to "true" while in existing IT tests we set it to "false" so no need to add custom filter for tests in every SDK
- default to "false" and testing "true" in one of the test cases
… to readAsJson (apache#40701) SourceRecordJson only copies six fields of the Debezium `source` struct into `metadata` (connector, version, name, database, schema, table), so values such as MySQL `file`, `pos`, `row` and `ts_ms`, PostgreSQL `lsn` and `txId`, and the change event's `op` are dropped. Add `DebeziumIO.Read#withFullMetadata(boolean)`, off by default. With the option off, the JSON output is byte for byte identical to before. With it on, `metadata` additionally gets, after `table`: - `op`, if the record envelope has an `op` field - every other field of the `source` struct in schema order, skipping those `metadata` already carries (connector, version, name, db, schema, table) so nothing is duplicated and no key is overwritten. The fields are flattened into `metadata`, there is no nested object. For MySQL this includes `file`, the real binlog file name that the default output exposes under `schema`. The option only applies to readAsJson() / SourceRecordJsonMapper; with a custom format function it fails with an IllegalArgumentException when the transform is expanded. SourceRecordJson and SourceRecordJsonMapper get boolean constructor overloads, and the existing constructors keep the default behaviour. This follows the review on apache#40702: Abacn suggested making the extra metadata a transform option that is off by default and tested in one case, and Ilyas Ahsan asked for `op` to count, route and soft delete events per operation.
88780ea to
0692974
Compare
|
Thanks @Abacn and @ilyasahsan123, great feedback! I reworked the PR:
@Abacn one clarification regarding the "table" field: |
…ezium (apache#40701) Wire the new DebeziumIO.Read#withFullMetadata option through the cross-language registrar (`setFullMetadata`, boxed and nullable like the other optional settings; only an explicit true enables it), the Python `ReadFromDebezium(full_metadata=...)` keyword argument (appended at the end of ReadFromDebeziumSchema) and the Go `debeziumio.FullMetadata()` read option. The default output is unchanged, so the Python and Go xlang ITs stay as they are. Touch the PostCommit Python and PostCommit XVR Direct trigger files because those suites run the Python and Go Debezium xlang tests.
0692974 to
a28e768
Compare


Fixes #40701.
SourceRecordJson(behindreadAsJson()and the Python/GoReadFromDebezium) only keeps six fields of the Debeziumsourceblock and drops the envelope'sop, so users can't deduplicate MySQL events (pos/row), get the commit time (ts_ms) or tell creates, updates and deletes apart (see #39568 (comment)).This adds an opt-in option, off by default:
DebeziumIO.readAsJson().withFullMetadata(true)ReadFromDebezium(..., full_metadata=True)debeziumio.Read(..., debeziumio.FullMetadata())When enabled,
metadataadditionally containsopand every Debeziumsourcefield that isn't already represented (connector,version,name,db/database,schemaandtableare not repeated). Examples: MySQLfile,pos,row,gtid,ts_ms; PostgreSQLlsn,txId,ts_ms. For MySQLfileis kept explicitly, since the binlog file name currently shows up undermetadata.schema.With the option off (default) the JSON output is byte for byte identical to before, so existing pipelines and the existing tests in all SDKs are unchanged. A custom format function combined with
withFullMetadata(true)is rejected with anIllegalArgumentException. YAMLReadFromDebeziumuses the Row path and is not affected.Commits:
testDebeziumIOMySqlFullMetadata) that runs with the option on.Testing:
:sdks:java:io:debezium:testandcheckpass;DebeziumIOMySqlConnectorIT(including the new full metadata case) andDebeziumIOPostgresSqlConnectorITpass locally against Docker. I'll add the CHANGES.md entry once the change is approved.R: @Abacn
cc @ilyasahsan123
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes. (will follow after approval)