[Iceberg] CDC config and writer by ahmedabu98 · Pull Request #39997 · apache/beam

ahmedabu98 · 2026-09-03T18:55:55Z

Adds the CDC sink's delta writer, which receives sorted rows (by PK, sequence number, change kind). The writer collapses changes and only writes one final state per key per window (an equality delete plus the surviving row). Because it collapses changes, it never needs to write a deletion vector.

Also adds the sink's CdcWriteConfig.

Need to merge #39981 first.

Part of #39979

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.

Copilot

🟡 Changes recommended

The current CdcWriteConfig defaults/validation can produce invalid configs (notably around shards-per-partition defaulting) and there are a couple of robustness/determinism issues that should be addressed before merging.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

This PR introduces core components for an Iceberg CDC sink in the Java SDK: a delta task writer that consumes sorted CDC records and collapses per-key changes into at most one equality delete plus one final row per window, and a corresponding CdcWriteConfig configuration object. It also adds targeted tests validating both the writer’s flush/partition-routing behavior and the config’s validation/serialization behavior.

Changes:

Added RecordDeltaTaskWriter to collapse sorted CDC streams into Iceberg row-delta writes (equality deletes + final rows) without DV/position deletes.
Added CdcWriteConfig (AutoValue) with validation for CDC sink options.
Added comprehensive unit tests for writer behavior and config validation/serialization.

File summaries

File	Description
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriter.java	Adds the delta writer that collapses CDC changes per PK and writes equality deletes + final rows, including partition fanout support.
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfig.java	Adds sink configuration with defaults and validation for CDC-specific options.
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriterTest.java	Adds a flush “truth table” and partition-routing tests validating the writer’s semantics.
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfigTest.java	Adds tests covering defaults, validation rejection matrix, and Java serialization of the config.

Review details

Suppressed comments (3)

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfig.java:135

Given the intended semantics ("unset shards_per_partition resolves to num_shards"), setting shards_per_partition to a fixed default in builder() prevents that and can cause validate() to fail when num_shards is changed. With a resolved getter, this default should be removed so the value can remain unset.

  static Builder builder() {
    return new AutoValue_CdcWriteConfig.Builder()
        .setSequenceNumberColumn(DEFAULT_SEQUENCE_NUMBER_COLUMN)
        .setNumShards(DEFAULT_NUM_SHARDS)
        .setShardsPerPartition(DEFAULT_NUM_SHARDS)
        .setSorterMemoryMB(DEFAULT_SORTER_MEMORY_MB)
        .setUpsert(false)
        .setErrorHandling(false);

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfig.java:188

validate() will throw a NullPointerException if snapshot_properties contains a null key (due to key.startsWith(...)). Consider rejecting null keys explicitly so invalid user input fails with a clear IllegalArgumentException.

    @Nullable Map<String, String> snapshotProperties = getSnapshotProperties();
    if (snapshotProperties != null) {
      for (String key : snapshotProperties.keySet()) {
        checkArgument(
            !key.startsWith("beam.cdc."),
            "snapshot_properties key '%s' uses the reserved 'beam.cdc.' prefix; choose a "
                + "different key.",
            key);

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfig.java:216

After introducing a resolved getShardsPerPartition() method, the AutoValue property should be a nullable "override" (unset vs explicitly set). The builder currently only exposes setShardsPerPartition(int), which can't represent "unset"; adding an override setter keeps the external API while enabling correct defaulting behavior.

    abstract Builder setNumShards(int numShards);

    abstract Builder setShardsPerPartition(int shardsPerPartition);

Files reviewed: 4/4 changed files
Comments generated: 3
Review effort level: Lite

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

…delta-writer-and-config

chamikaramj

Thanks!

chamikaramj · 2026-09-04T20:15:58Z

+   * Columns that define a row's identity (the Iceberg equality-delete fields). If unspecified, will
+   * try to use the destination table's identifier fields.
+   */
+  abstract @Nullable List<String> getEqualityColumns();


Is this the same as identifier fields (primary-key fields) ? I guess the difference is whether you are talking about the table in general or a specific delete file. If it's the prior probably we should name accordingly.

"primary key" isn't in the Iceberg vocabulary, although the concept is similar.
"Equality columns" is a write property. It defaults to the table's "identifier fields" by default, so it's similar in that sense, but it can be overridden for a given write operation.

I'd rather keep it this way, it matches Spark and Flink too. Users will be familiar with it

chamikaramj · 2026-09-04T20:17:44Z

+   * If set, the change kind is read from this string column instead of the element's native {@link
+   * ValueKind}. The column is stripped from the data row and never written to Iceberg.
+   */
+  abstract @Nullable String getChangeTypeColumn();


Probably mention what values this column can include.

chamikaramj · 2026-09-04T20:28:51Z

+  /**
+   * If {@code true}, {@code UPDATE_BEFORE} records are dropped and {@code INSERT}/{@code
+   * UPDATE_AFTER} are applied as upserts (equality-delete-then-insert on the primary key). Defaults
+   * to {@code false}.


Also document the behavior if false.

chamikaramj · 2026-09-04T22:17:53Z

+  abstract @Nullable Map<String, String> getSnapshotProperties();
+
+  /**
+   * If {@code true}, a poison record (unknown change type, missing/null sequence number, null


Is "poison record" the correct term ? To me this sounds like valid data regarding errors that occurred.

The mentioned error-cases wouldn't be considered valid for this sink

chamikaramj · 2026-09-04T22:19:58Z

+    @Nullable List<String> equalityColumns = getEqualityColumns();
+    checkArgument(
+        equalityColumns == null || !equalityColumns.isEmpty(),
+        "equality_columns must be non-empty or unset (leave unset to use the table's identifier "


Ah, seems like we just default to identifier fields ? Probably we should just document this in 'getEqualityColumns' call above.

Already documented in getEqualityColumns: "If unspecified, will try to use the destination table's identifier fields"

chamikaramj · 2026-09-04T23:00:09Z

+
+  /** Closes every file and deletes it: a failed group must leave nothing behind. */
+  public void abort() throws IOException {
+    close();


Seems like there might be a leak here if the close() invocation throws an exception ?

Let's also add a test for this.

chamikaramj · 2026-09-04T23:02:29Z

+        // The shared partitionKey is mutated on every route() call; copy before keying the map.
+        PartitionKey copiedKey = partitionKey.copy();
+        writer = newPartitionWriter(copiedKey);
+        writers.put(copiedKey, writer);


Probably this can cause an OOM if a bundle includes many primary keys and we hold that many writes in memory ?

We create one writer per partition, not per primary key. There is a risk for high partition fanout, but it's much less severe and it's the same risk Flink has in its sink.

We have an advantage here because of shardsPerPartition, reducing it will limit the number of open writers per partition

chamikaramj · 2026-09-04T23:04:05Z

+    List<Types.NestedField> pkFields = deleteSchema.columns();
+    this.pkPos = new int[pkFields.size()];
+    List<Types.NestedField> allFields = schema.columns();
+    for (int i = 0; i < pkFields.size(); i++) {


Nit suggestion from AI:

for (int i = 0; i < pkFields.size(); i++) {
Types.NestedField field = schema.findField(pkFields.get(i).fieldId());
if (field == null) {
throw new IllegalStateException("...");
}
this.pkPos[i] = schema.columns().indexOf(field);
}

schema.findField searches for nested fields too. We're purposely only looking for top-level fields with schema.columns()

chamikaramj · 2026-09-04T23:14:16Z

+
+  @Rule public transient TemporaryFolder tmp = new TemporaryFolder();
+
+  private static final Schema SCHEMA =


Also test for composite primary keys, other data types in the primary key.

chamikaramj · 2026-09-04T23:14:59Z

+
+  private static final Schema SCHEMA =
+      new Schema(
+          Types.NestedField.required(1, "id", Types.IntegerType.get()),


Also test for cases where the primary key is not the first field ?

github-actions · 2026-09-04T23:36:32Z

Assigning reviewers:

R: @kennknowles 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.

config and writer

bf0eb84

ahmedabu98 requested a lite review from Copilot September 3, 2026 18:56

github-actions Bot added java io labels Sep 3, 2026

ahmedabu98 requested a review from chamikaramj September 3, 2026 18:56

Copilot started reviewing on behalf of ahmedabu98 September 3, 2026 18:56 View session

Copilot AI reviewed Sep 3, 2026

View reviewed changes

ahmedabu98 added 2 commits September 3, 2026 15:55

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

19afeff

…delta-writer-and-config

test utils

f0a51d1

github-actions Bot added the Next Action: Reviewers label Sep 3, 2026

ahmedabu98 mentioned this pull request Sep 3, 2026

[Iceberg CDC sink] Split late data and create commit windows #40006

Open

3 tasks

spotless

60e6ccf

ahmedabu98 mentioned this pull request Sep 3, 2026

[Iceberg CDC sink] table setup #40007

Open

3 tasks

ahmedabu98 added 4 commits September 3, 2026 17:31

spotless

9399146

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

f935511

…delta-writer-and-config

pull test fix

b9fd7e8

remove dep

d8e8276

chamikaramj requested changes Sep 4, 2026

View reviewed changes

address comments

50b35d7

ahmedabu98 requested a review from chamikaramj September 5, 2026 00:15

chamikaramj approved these changes Sep 5, 2026

View reviewed changes


		@Rule public transient TemporaryFolder tmp = new TemporaryFolder();

		private static final Schema SCHEMA =

Uh oh!

Conversation

ahmedabu98 commented Sep 3, 2026

GitHub Actions Tests Status (on master branch)

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Suppressed comments (3)

Uh oh!

Uh oh!

Uh oh!

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!

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!

github-actions Bot commented Sep 4, 2026

Uh oh!

chamikaramj left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants