[Iceberg CDC] Finish wiring CDC source together and add external API by ahmedabu98 · Pull Request #39600 · apache/beam

ahmedabu98 · 2026-08-04T00:38:08Z

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:

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.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

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.

gemini-code-assist · 2026-08-04T00:38:22Z

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

github-actions · 2026-08-04T03:09:05Z

Assigning reviewers:

R: @chamikaramj for label java.

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).

chamikaramj

Thanks!

chamikaramj · 2026-08-04T23:13:12Z

+
+    // 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


How about deletes ?
Did you mean "these only contain inserts" ?

"unidirectional tasks" refers to a group of tasks where we are certain there is no insert-delete collision.

For example, it may contain ONLY inserts, or ONLY deletes. or it may contain inserts and deletes that don't overlap in PK

chamikaramj · 2026-08-04T23:19:41Z

+    PCollection<Row> smallBidirectionalCdcRows =
+        changelogTasks
+            .get(SMALL_BIDIRECTIONAL_TASKS)
+            .apply("Redistribute Small Bidirectional Changes", Redistribute.arbitrarily())


Seems like we have a shuffle here ?

Yeah but this is before we apply any ValueKind metadata. Same with the CoGroupByKey in the large bi-directional path -- it's before any ValueKind metadata gets added

chamikaramj · 2026-08-05T00:13:37Z

+    Window<KV<CdcRowDescriptor, Row>> keyedWindowing =
+        Window.<KV<CdcRowDescriptor, Row>>into(new SnapshotWindowFn())
+            .triggering(AfterWatermark.pastEndOfWindow())
+            .withAllowedLateness(Duration.ZERO)


Is dropping late data the correct approach here ?

The watermark here is actually controlled by us (in WatchForSnapshotsSdf). We only advance it past a snapshot's timestamp after emitting that snapshot, so in the normal flow there's no "late" data. Timestamps represent snapshot commit time

There was one exception I just fixed. On empty polls we bump the watermark to now() - pollInterval so downstream windows don't stall while the table is quiet. But if snapshots are committed and only discovered much later (e.g. a catalog outage), that bump could run ahead of their commit times, and they'd arrive late and get silently dropped. Fixed this edge case by setting output timestamp = max(commitTs, currentWatermark).

chamikaramj · 2026-08-05T00:15:56Z

+        KeyedPCollectionTuple.of(INSERTS, keyedInserts)
+            .and(DELETES, keyedDeletes)
+            .apply("CoGroupBy Primary Key", CoGroupByKey.create())
+            .apply("Resolve Delete-Insert Pairs", ParDo.of(new ResolveChanges(scanConfig)))


I suspect this is something we can do at the framework level later right ? (Without having to do this kind of optimizations per source)

This logic is written just for Iceberg actually. Iceberg doesn't have any native concept of an "update", so we (the engine) have to figure it out ourselves

But it can be mostly re-used if we encounter another source with similar behavior

chamikaramj · 2026-08-05T05:46:19Z

+    }
+
+    @Override
+    protected int nonPkHash(Row element) {


How expensive is this ? Seems like we are hashing pretty much all fields here of all elements ? (not a blocker)

Well we need to compare all fields to correctly determine if it's an update. This is only done for records that fall in the overlap range. Other records skip this entirely

chamikaramj · 2026-08-05T05:46:42Z

+    }
+
+    @Override
+    protected boolean nonPkEquals(Row delete, Row insert) {


Ditto (might be good to do a perf analysis after the PR is submitted)

See comment above. We only do this equality comparison on hash collision, so it's even more rare.

Perf analysis could be warranted though

chamikaramj · 2026-08-05T05:52:24Z

+     * 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()}.


Let's fail if the incorrect combination is set.

It does in IcebergScanConfig#validate

…erg-full-cdc-read

chamikaramj

Thanks!

chamikaramj · 2026-08-05T17:51:05Z

+      if (ts.isBefore(watermark.currentWatermark())) {
+        // The watermark already moved past this snapshot's commit time (e.g. the idle bump ran
+        // ahead of a slow discovery). Use the current watermark so the snapshot is not dropped
+        ts = watermark.currentWatermark();


Probably there still a race (extremely rare) between we setting this here and runner firing for the current watermark, right ?

I don't think so, IIUC the watermark only advances when the bundle commits, along with the emmitted elements. and those same in-flight elements hold back the watermark until they arrive downstream, so a downstream window can't fire until they all arrive

chamikaramj · 2026-08-05T17:57:09Z

    abstract @Nullable String getWatermarkColumnTimeUnit();

-    @SchemaFieldDescription(
-        "Maximum expected snapshot discovery delay in seconds. While idle, the source may advance "


Did you intend to remove this knob ?

Yeah we no longer need it after the fix mentioned in #39600 (comment)

Only had it to allow customers to control what we would consider "late". but now even if the catalog is slow/down, we wouldn't consider the missed snapshots as late. We just weave them in and attach current watermark as timestamp

…pache#39600) * wrap up * trigger ITs and add to CHANGES.md * fix late snapshot edge case * update resolution optimization --------- Co-authored-by: Ahmed Abualsaud <ahmedabualsaud@MacBook-Pro-2.local>

wrap up

0d5daaf

github-actions Bot added java io labels Aug 4, 2026

trigger ITs and add to CHANGES.md

76ad078

github-actions Bot added the Next Action: Reviewers label Aug 4, 2026

chamikaramj added this to the 2.76.0 Release milestone Aug 4, 2026

chamikaramj reviewed Aug 5, 2026

View reviewed changes

Ahmed Abualsaud added 3 commits August 5, 2026 09:44

fix late snapshot edge case

fb86f24

Merge branch 'master' of https://github.com/ahmedabu98/beam into iceb…

0b87b54

…erg-full-cdc-read

update resolution optimization

e2a7676

ahmedabu98 requested a review from chamikaramj August 5, 2026 17:25

chamikaramj approved these changes Aug 5, 2026

View reviewed changes

ahmedabu98 merged commit c60b021 into apache:master Aug 5, 2026
24 checks passed

Uh oh!

Conversation

ahmedabu98 commented Aug 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

GitHub Actions Tests Status (on master branch)

Uh oh!

gemini-code-assist Bot commented Aug 4, 2026

Uh oh!

github-actions Bot commented Aug 4, 2026

Uh oh!

chamikaramj left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chamikaramj left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

ahmedabu98 commented Aug 4, 2026 •

edited

Loading