Skip to content

[DebeziumIO] Add opt-in full metadata (op and Debezium source fields) to readAsJson - #40702

Merged
Abacn merged 2 commits into
apache:masterfrom
tkaymak:debezium-source-metadata-40701
Oct 22, 2026
Merged

Abacn merged 2 commits into
apache:masterfrom
tkaymak:debezium-source-metadata-40701

Conversation

@tkaymak

@tkaymak tkaymak commented Oct 21, 2026 •

Copy link
Copy Markdown
Contributor

Fixes #40701.

SourceRecordJson (behind readAsJson() and the Python/Go ReadFromDebezium) only keeps six fields of the Debezium source block and drops the envelope's op, 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:

  • Java: DebeziumIO.readAsJson().withFullMetadata(true)
  • Python: ReadFromDebezium(..., full_metadata=True)
  • Go: debeziumio.Read(..., debeziumio.FullMetadata())

When enabled, metadata additionally contains op and every Debezium source field that isn't already represented (connector, version, name, db/database, schema and table are not repeated). Examples: MySQL file, pos, row, gtid, ts_ms; PostgreSQL lsn, txId, ts_ms. For MySQL file is kept explicitly, since the binlog file name currently shows up under metadata.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 an IllegalArgumentException. YAML ReadFromDebezium uses the Row path and is not affected.

Commits:

  1. Java option, unit tests and one new MySQL IT case (testDebeziumIOMySqlFullMetadata) that runs with the option on.
  2. Python/Go exposure through the expansion service, and the PostCommit Python / XVR Direct trigger files so the xlang tests run.

Testing: :sdks:java:io:debezium:test and check pass; DebeziumIOMySqlConnectorIT (including the new full metadata case) and DebeziumIOPostgresSqlConnectorIT pass 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:

  • Mention the appropriate issue in your description (for example: 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, comment fixes #<ISSUE NUMBER> instead.
  • Update CHANGES.md with noteworthy changes. (will follow after approval)
  • If this contribution is large, please file an Apache Individual Contributor License Agreement.

@github-actions github-actions Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📋 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.source for deterministic event comparison.

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

@tkaymak
tkaymak force-pushed the debezium-source-metadata-40701 branch 2 times, most recently from b895b27 to 66195eb Compare October 21, 2026 13:08
@codecov

codecov Bot commented Oct 21, 2026 •

Copy link
Copy Markdown

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 56.45%. Comparing base (a8d6afd) to head (a28e768).
⚠️ Report is 35 commits behind head on master.

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     
Flag Coverage Δ
go 28.91% <100.00%> (?)
python 80.07% <ø> (+0.02%) ⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

@github-actions

Copy link
Copy Markdown
Contributor

Assigning reviewers:

R: @shunping for label python.
Backup reviewer: jrmccluskey (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

  • stop reviewer notifications - opt out of the automated review tooling
  • remind me after tests pass - tag the comment author after tests pass
  • waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

@Abacn Abacn left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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"),

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@ilyasahsan123

ilyasahsan123 commented Oct 22, 2026 •

Copy link
Copy Markdown
Contributor

Great! the metadata now includes every source property. 🎊

image

Furthermore, could the op property be included in this pull request?

image

This property helped us a lot for the following reasons:

  1. monitor the numbers of events from specific operations (create, update, delete)
  2. mark the rows within the data warehouse as deleted
  3. edge case, there is company differentiate destination tables per operations

I recognize that we can establish conditions using values in the after and before properties. However, if the op is present, it would assist us considerably.

P.S. If a separate pull request is preferable, I am glad to assist with creating the issue and opening the pull request. 😄

Thanks

… 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.
@tkaymak tkaymak changed the title [DebeziumIO] Expose the full Debezium source block in SourceRecordJson metadata [DebeziumIO] Add opt-in full metadata (op and Debezium source fields) to readAsJson Oct 22, 2026
@tkaymak
tkaymak force-pushed the debezium-source-metadata-40701 branch from 88780ea to 0692974 Compare October 22, 2026 07:38
@tkaymak

tkaymak commented Oct 22, 2026 •

Copy link
Copy Markdown
Contributor Author

Thanks @Abacn and @ilyasahsan123, great feedback! I reworked the PR:

  • The extra metadata is now opt-in via DebeziumIO.readAsJson().withFullMetadata(true) and off by default, so the default output is unchanged and the existing tests in all SDKs stay as they are. The full mode is covered by unit tests and one new MySQL IT case.
  • Instead of a nested metadata.source that repeats connector, version, name, db and table, the additional source fields are now added directly to metadata, and only the ones that aren't there yet (e.g. MySQL file, pos, row, gtid, ts_ms; PostgreSQL lsn, txId, ts_ms). For MySQL I kept file explicitly, since the binlog file currently shows up under schema.
  • @ilyasahsan123 op is included as metadata.op behind the same option.
  • The second commit exposes the option in Python (full_metadata=True) and Go (debeziumio.FullMetadata()), and touches the PostCommit trigger files so the xlang tests run again.

@Abacn one clarification regarding the "table" field: metadata.table is already part of the default output. Did you mean something else, maybe op? Happy to adjust.

…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.
@tkaymak
tkaymak force-pushed the debezium-source-metadata-40701 branch from 0692974 to a28e768 Compare October 22, 2026 07:56

@Abacn Abacn left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks!

@Abacn
Abacn merged commit 9811046 into apache:master Oct 22, 2026
109 of 110 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

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

3 participants