[Iceberg CDC] Finish wiring CDC source together and add external API - #39600
[Iceberg CDC] Finish wiring CDC source together and add external API#39600ahmedabu98 wants to merge 2 commits into
Conversation
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
|
Assigning reviewers: R: @chamikaramj for label java. 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). |
|
|
||
| // process one snapshot at a time and produce batches of changelog scan tasks. | ||
| // tasks are emitted to three outputs: | ||
| // 1. unidirectional tasks: we know these won't have any updates |
There was a problem hiding this comment.
How about deletes ?
Did you mean "these only contain inserts" ?
| PCollection<Row> smallBidirectionalCdcRows = | ||
| changelogTasks | ||
| .get(SMALL_BIDIRECTIONAL_TASKS) | ||
| .apply("Redistribute Small Bidirectional Changes", Redistribute.arbitrarily()) |
There was a problem hiding this comment.
Seems like we have a shuffle here ?
| Window<KV<CdcRowDescriptor, Row>> keyedWindowing = | ||
| Window.<KV<CdcRowDescriptor, Row>>into(new SnapshotWindowFn()) | ||
| .triggering(AfterWatermark.pastEndOfWindow()) | ||
| .withAllowedLateness(Duration.ZERO) |
There was a problem hiding this comment.
Is dropping late data the correct approach here ?
| KeyedPCollectionTuple.of(INSERTS, keyedInserts) | ||
| .and(DELETES, keyedDeletes) | ||
| .apply("CoGroupBy Primary Key", CoGroupByKey.create()) | ||
| .apply("Resolve Delete-Insert Pairs", ParDo.of(new ResolveChanges(scanConfig))) |
There was a problem hiding this comment.
I suspect this is something we can do at the framework level later right ? (Without having to do this kind of optimizations per source)
| } | ||
|
|
||
| @Override | ||
| protected int nonPkHash(Row element) { |
There was a problem hiding this comment.
How expensive is this ? Seems like we are hashing pretty much all fields here of all elements ? (not a blocker)
| } | ||
|
|
||
| @Override | ||
| protected boolean nonPkEquals(Row delete, Row insert) { |
There was a problem hiding this comment.
Ditto (might be good to do a perf analysis after the PR is submitted)
| * require a row-lineage table. The changelog metadata columns come from the emitted change kind | ||
| * and snapshot context and are appended when final Beam rows are emitted. | ||
| * | ||
| * <p>This option is only valid {@link #withCdc()}. |
There was a problem hiding this comment.
Let's fail if the incorrect combination is set.
2 participants
Finalizes the IcebergIO CDC streaming source. Introduces top-level IncrementalChangelogSource to replace IncrementalAppendSource.
Reads a table's changelog one snapshot at a time and emits net per-snapshot changes as CDC rows. Bi-directional groups too large to resolve in memory are windowed per snapshot (SnapshotWindowFn), grouped by primary key, and reconciled by ResolveChanges, while small groups continue to resolve locally (LocalResolveDoFn) without a shuffle.
Adds user-facing options for CDC metadata columns, a custom watermark column with its time unit, and a maximum snapshot discovery delay.
Removes the now-unused IncrementalScanSource, ReadFromTasks, and WatchForSnapshots.
Fixes #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.