[Iceberg CDC] Add Changelog readers and update resolver - #38837
Conversation
…erg_changelog_readers
…erg_changelog_readers
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces the core infrastructure for reading and resolving Iceberg CDC changelogs within Apache Beam. It implements a routing mechanism that distinguishes between unidirectional and bidirectional changelog tasks, ensuring that records are either emitted directly or routed for update resolution based on primary key overlap. By centralizing the reconciliation logic, this change enables consistent handling of CDC data, including the identification of update pairs and the filtering of redundant operations. Highlights
New Features🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces support for Iceberg Change Data Capture (CDC) processing in Apache Beam, adding classes like CdcResolver, CdcRowDescriptor, LocalResolveDoFn, OverlapRange, and ReadFromChangelogs to handle bi-directional changelog resolution, Copy-on-Write deduplication, and primary key overlap detection. The review feedback highlights several critical issues: a potential ClassCastException in IcebergUtils when casting nested StructLike objects directly to Record, and multiple schema mismatches in OverlapRange, ReadFromChangelogs, and LocalResolveDoFn where projectors are constructed without accounting for metadata columns, which would cause position mismatches during record processing.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
|
R: @talatuyarer |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
…erg_changelog_readers
| protected int nonPkHash(Record rec) { | ||
| int hash = 1; | ||
| for (Types.NestedField field : nonPkFields) { | ||
| hash = 31 * hash + Objects.hashCode(rec.getField(field.name())); |
There was a problem hiding this comment.
On a table with a non-PK fixed column, a copy-on-write rewrite's delete/insert pair never matches in the hash index, so unchanged rows are emitted as spurious UPDATE_BEFORE/UPDATE_AFTER pairs.
Can we derive both hash and equals from one mechanism
There was a problem hiding this comment.
Can we derive both hash and equals from one mechanism
We could build a single content-based key per overlapping record and do a equals/hashCode for each lookup, but that would be less performant
Correct that the old code was problematic for fixed types cuz Objects.hashCode() would be identity based for byte[]. I covered this by making the hash logic array-aware. This way, we only check deep-equality on a hash collision. I also added tests that a CoW pair with a non-PK fixed column is recognized and dropped.
I'd rather keep it this way cuz it's more performant, but let me know what you think.
|
|
||
| // {PK: (inserts | deletes)} for in-overlap records that need resolution. | ||
| // Records outside the overlap are emitted directly | ||
| StructLikeMap<PkGroup> pkGroups = StructLikeMap.create(ovl.recordIdSchema().asStruct()); |
There was a problem hiding this comment.
Consider noting that while records are buffered based on the 128MB compressed SPLIT_SIZE, the current implementation lacks an inflation factor for decoded data. With no safety check for the actual memory footprint in LocalResolveDoFn, a single process() call can exceed 500MB of heap usage. Furthermore, since the overlap bounds can be null when metrics are missing, the scanner defaults to buffering everything, which risks OOM errors.
There was a problem hiding this comment.
I added a scaling factor in ChangelogScanner to make a rough estimate of the decoded bytes. This reduces the byte size threshold of batches getting routed to LocalResolveDoFn, so we won't see big batches making it there. Should help us avoid OOMs, and we can adjust the factor when we see how it's handled with real usage
…hash logic array-aware
…erg_changelog_readers
…erg_changelog_readers
chamikaramj
left a comment
chamikaramj
left a comment
There was a problem hiding this comment.
Thanks! LGTM.
Just nits.
| * <li>Hash-index inserts by their non-PK field hash, for efficient Copy-on-Write detection. | ||
| * <li>Skip matching (delete, insert) pairs with identical non-PK columns. A CoW operation deletes | ||
| * and rewrites the whole file (minus some records that are actually marked for deletion). | ||
| * Unchanged records are no-ops and should not be mistaken for updates. |
There was a problem hiding this comment.
"Unchanged records are no-ops"
Is this a quirk in Iceberg CDC ? Ideally we never produce such records.
There was a problem hiding this comment.
A primary key can have identical (delete, insert) pairs in two scenarios:
- There is a Copy-on-Write, where a new DataFile is added to replace an old DataFile now marked deleted. The new DataFile's rows are identical to the old DataFile, except for some missing rows that are interpreted to be deleted. The overlapping rows, on the surface, are identical (delete, insert) pairs.
- Some records are inserted and then subsequently deleted in the same snapshot. Reading this produces identical (delete, insert) pairs. Our policy is to only look at the delta so we ignore them
| } | ||
|
|
||
| if (d < deletes.size() && i < inserts.size()) { | ||
| emit.accept(ValueKind.UPDATE_BEFORE, deletes.get(d)); |
There was a problem hiding this comment.
Seems like deletes and inserts have to be in a particular order to match here correctly ?
There was a problem hiding this comment.
Only when a PK group has multiple inserts or deletes. This would be unusual because it would mean one commit inserted or deleted a record multiple times. It's still possible though because Iceberg doesn't enforce uniqueness.
We unfortunately can't gather any further insight on intended record ordering within a commit, so pairing can be arbitrary. I've documented the case in the javadoc and now sort both sides by hash before pairing so the result is at least deterministic regardless of input order.
| /** | ||
| * Primary-key-projection and overlap-range comparison helper. | ||
| * | ||
| * <p>Used by {@link LocalResolveDoFn} and {@link ReadFromChangelogs} to decide whether a record's |
There was a problem hiding this comment.
In general, ranges where both sides are inclusive can lead to errors at the boundary. For example, this is why our source split ranges are non-inclusive at the upper bound "[...)". I'm not sure if it's an issue here but if we keep range as is, let's add assertions/tests to make sure that we do not run into issues at the boundary.
There was a problem hiding this comment.
It's different from split ranges because these are actual data values, not partitions. If we make the range open on one end, we'd miss a collision and accidentally process an update pair as separate delete + insert.
I added some language to clarify. We already have a few boundary tests in OverlapRangeTest but I added a couple more to cover edge cases
| case DATETIME: | ||
| // Iceberg uses a long for micros. | ||
| // Beam DATETIME uses joda's DateTime, which only supports millis, | ||
| // so we do lose some precision here |
There was a problem hiding this comment.
Can't really fix this, it's a known shortcoming of Joda which is what Beam DATETIME uses.
The workaround is to use the new Timestamp.MICROS logical type (backed by Java time), which has just been made the new default (#39344).
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #38837 +/- ##
============================================
- Coverage 58.17% 58.15% -0.02%
- Complexity 13086 13087 +1
============================================
Files 2521 2521
Lines 264569 264573 +4
Branches 10788 10788
============================================
- Hits 153902 153872 -30
- Misses 104900 104929 +29
- Partials 5767 5772 +5
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:
|
4 participants
Adds read transforms that consume planned Iceberg changelog tasks and turn them into Beam CDC rows.
ReadFromChangelogsreads task batches produced byChangelogScanner:INSERTorDELETErowsLocalResolveDoFnreads and resolves small bidirectional task batches in memory.CdcResolvercentralizes the logic for reconciling deletes and inserts for the same primary key. It emits changed pairs asUPDATE_BEFORE/UPDATE_AFTER, and leaves unmatched rows asDELETEorINSERT. Duplicate no-op pairs with identical non-PK fields are dropped.Part of #38831
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.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.