Updates the Delta Lake source to support reading bounded change data by chamikaramj · Pull Request #39426 · apache/beam

chamikaramj · 2026-07-22T05:53:42Z

Please see here for a short design description.

Github issue: #39492.

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-07-22T05:53:44Z

Caution

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

gemini-code-assist · 2026-07-25T19:21:22Z

Caution

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

github-actions · 2026-07-25T19:29:55Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

chamikaramj · 2026-07-25T23:29:08Z

assign set of reviewers

github-actions · 2026-07-25T23:30:25Z

Assigning reviewers:

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

Abacn · 2026-07-27T20:25:04Z

+
+  @Override
+  public String toString() {
+    return "DeltaCDCReadTask{"


nit: prefer String.format

Abacn · 2026-07-27T22:01:48Z

+      }
+    }
+
+    return ProcessContinuation.stop();


This is the only place where DoFn exits. if there is error throws, will it retry the whole element? In other words, would bounded SDF just resume from a claimed restriction even though previously crashed in the middle of a DoFn.process invocation?

I wonder if we need ProcessContinuation here at all, since we never call resume() ?

It should retry the whole workitem (including the whole element) if there's a crash.

Agree that we can just return here instead of using ProcessContinuation. Updated.

Abacn · 2026-07-27T22:03:09Z

+
+  private static StructType appendCDFColumns(StructType schema) {
+    return schema
+        .add("_change_type", StringType.STRING, false)


nit: consider delcare these field name as constants. There are used in multiple places

Abacn · 2026-07-27T22:07:10Z

+
+        while (logicalBatches.hasNext()) {
+          FilteredColumnarBatch batch = logicalBatches.next();
+          ColumnarBatch logicalBatch = batch.getData();


(based on AI comment): batch.getData() return raw data, ignoring selectionVector and causing filtered-out or deleted rows to be iterated over and emitted into the output PCollection.

Suggestion, call batch. getRows which handles selectionVector

while (logicalBatches.hasNext()) { FilteredColumnarBatch batch = logicalBatches.next(); if (!task.isCDC()) { ColumnarBatch logicalBatch = appendConstantCDFColumns( currentEngine, batch.getData(), task.getVersion(), task.getTimestamp()); batch = new FilteredColumnarBatch(logicalBatch, batch.getSelectionVector()); } try (CloseableIterator<io.delta.kernel.data.Row> logicalRows = batch.getRows()) { // <-- Uses FilteredColumnarBatch.getRows() while (logicalRows.hasNext()) { io.delta.kernel.data.Row deltaRow = logicalRows.next(); ... } } }

Abacn · 2026-07-27T22:15:16Z

Not related to the change. But running the test realize there are System.out.println and System.err.println debugging leftovers. Consider removing them or using slf4j LOGGER.

Abacn · 2026-07-27T22:16:33Z

+    }
+  }
+
+  private byte[] writeParquetFile(File file, Schema schema, java.util.List<Row> rows)


There is already a writeParquetFile

https://github.com/chamikaramj/beam/blob/82ab11edcafe9251d75af1185daf33f247ca261c/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java#L379

ahmedabu98 · 2026-07-27T22:46:25Z

+      if (getStartVersion() == null && getStartTimestamp() == null) {
+        throw new IllegalArgumentException("Either startVersion or startTimestamp must be set.");
+      }


If there's no start version specified, we can have a default starting strategy. Either start at the very beginning, or at the very latest change.

Not blocking though, can add this in a future change

The IcebergIO default is we read from the beginning for batch and from the very end for streaming

I think given that this is a bounded source, it's cleaner for users to provide a starting point ? Starting from current doesn't make sense since that would not result in anything (makes sense to extend this after we support unbounded reads). Added a TODO.

ahmedabu98 · 2026-07-27T22:47:08Z

+      if (getEndVersion() != null && getEndTimestamp() != null) {
+        throw new IllegalArgumentException("Cannot set both endVersion and endTimestamp.");
+      }


Same as above. If no end is specified, we can just stop at the very latest change

I think your logic in the Create DoFn already handles this

Yeah, this is handled. We only fail if end is specified both as a version and a timestampl.

ahmedabu98 · 2026-07-27T23:14:32Z

+        boolean hasCDC = !info.cdcRows.isEmpty();
+
+        List<Row> rowsToProcess = hasCDC ? info.cdcRows : info.addRows;
+        boolean isCDC = hasCDC;


nit: unnecessary assignment to another variable?

ahmedabu98 · 2026-07-27T23:19:38Z

+    final List<Row> cdcRows = new ArrayList<>();
+    final List<Row> addRows = new ArrayList<>();


Naming is throwing me off a little. So these aren't data rows right? Each row is metadata about a particular CDC or INSERT task?

Yeah, renamed to better convey this.

ahmedabu98 · 2026-07-27T23:30:11Z

+ * files, converting rows to Beam Rows.
+ */
+@DoFn.BoundedPerElement
+class DeltaCDCSourceDoFn extends DoFn<DeltaCDCReadTask, Row> {


This DoFn parallelizes at the file level right? I think this should be fine for batch, but note that it may not be scalable in DF streaming mode because the runner opens too many threads per VM, each thread processing a file and competing for memory leading to OOM.

Yeah, for streaming, we could consider batching files. This just supports batch reads for now. Added a TODO.

ahmedabu98 · 2026-07-27T23:33:45Z

+        .add("_change_type", StringType.STRING, false)
+        .add("_commit_version", LongType.LONG, false)
+        .add("_commit_timestamp", TimestampType.TIMESTAMP, false);


Can we have these field names be final static variables at the top?

What values can _change_type take?

Also is _commit_version increasingly monotonic?

hmmm I see that delta uses update_preimage and update_postimage for UPDATE_BEFORE/AFTER. This might not play too well with Iceberg CDC sink but I think should be okay because we can rely on the ValueKind metadata instead

Done. Seems like these map well to what's defined in ValueKind. What's the incompatibility you expect for Iceberg sink ?

ahmedabu98 · 2026-07-27T23:49:55Z

+      }
+    }
+
+    return ProcessContinuation.stop();


I wonder if we need ProcessContinuation here at all, since we never call resume() ?

ahmedabu98 · 2026-07-27T23:51:33Z

+  private static Row projectRow(Row row, Schema targetSchema) {
+    Row.Builder builder = Row.withSchema(targetSchema);
+    for (Schema.Field field : targetSchema.getFields()) {
+      builder.addValue(row.getValue(field.getName()));
+    }
+    return builder.build();
+  }


If schemas are equal we can just return the original row

ahmedabu98 · 2026-07-27T23:54:21Z

+  private void writeCommit(
+      File logDir,
+      long version,
+      long timestamp,
+      @Nullable String addPath,
+      long addSize,
+      @Nullable String removePath,
+      @Nullable String cdcPath,
+      long cdcSize,
+      boolean writeMetadata)
+      throws IOException {


Can we commit using the DeltaLake API? to make sure it works against the actual API and get notified if a library upgrade breaks things

ahmedabu98 · 2026-07-28T00:01:59Z

+    // Test 1: Read changes between start version 0 and end version 2
+    PCollection<Row> outputVersions =
+        readPipeline.apply(
+            "Read Changes Version Range",
+            DeltaIO.readChanges()
+                .from(tableDir.getAbsolutePath())
+                .withStartVersion(0L)
+                .withEndVersion(2L));


This test reads the whole range, so it can still pass if the range handling logic happened to be incorrect. We should have start version > 0 and/or end version < 2

Added a new test for this.

chamikaramj

Thanks. PTAL.

chamikaramj · 2026-07-28T01:06:33Z

+        boolean hasCDC = !info.cdcRows.isEmpty();
+
+        List<Row> rowsToProcess = hasCDC ? info.cdcRows : info.addRows;
+        boolean isCDC = hasCDC;


chamikaramj · 2026-07-28T01:20:23Z

+    final List<Row> cdcRows = new ArrayList<>();
+    final List<Row> addRows = new ArrayList<>();


Yeah, renamed to better convey this.

chamikaramj · 2026-07-28T01:28:50Z

+
+  @Override
+  public String toString() {
+    return "DeltaCDCReadTask{"


chamikaramj · 2026-07-28T01:32:13Z

+ * files, converting rows to Beam Rows.
+ */
+@DoFn.BoundedPerElement
+class DeltaCDCSourceDoFn extends DoFn<DeltaCDCReadTask, Row> {


Yeah, for streaming, we could consider batching files. This just supports batch reads for now. Added a TODO.

chamikaramj · 2026-07-28T17:09:08Z

+
+        while (logicalBatches.hasNext()) {
+          FilteredColumnarBatch batch = logicalBatches.next();
+          ColumnarBatch logicalBatch = batch.getData();


chamikaramj · 2026-07-28T22:03:37Z

+      if (getStartVersion() == null && getStartTimestamp() == null) {
+        throw new IllegalArgumentException("Either startVersion or startTimestamp must be set.");
+      }


I think given that this is a bounded source, it's cleaner for users to provide a starting point ? Starting from current doesn't make sense since that would not result in anything (makes sense to extend this after we support unbounded reads). Added a TODO.

chamikaramj · 2026-07-29T04:54:27Z

chamikaramj · 2026-07-29T04:58:08Z

+    // Test 1: Read changes between start version 0 and end version 2
+    PCollection<Row> outputVersions =
+        readPipeline.apply(
+            "Read Changes Version Range",
+            DeltaIO.readChanges()
+                .from(tableDir.getAbsolutePath())
+                .withStartVersion(0L)
+                .withEndVersion(2L));


Added a new test for this.

chamikaramj · 2026-07-29T05:14:50Z

+  private void writeCommit(
+      File logDir,
+      long version,
+      long timestamp,
+      @Nullable String addPath,
+      long addSize,
+      @Nullable String removePath,
+      @Nullable String cdcPath,
+      long cdcSize,
+      boolean writeMetadata)
+      throws IOException {


chamikaramj · 2026-07-29T05:15:59Z

+    }
+  }
+
+  private byte[] writeParquetFile(File file, Schema schema, java.util.List<Row> rows)


github-actions Bot added java io delta labels Jul 22, 2026

chamikaramj marked this pull request as ready for review July 25, 2026 19:21

chamikaramj force-pushed the delta_lake_cdc branch from ed28244 to 82ab11e Compare July 25, 2026 22:05

github-actions Bot added the Next Action: Reviewers label Jul 25, 2026

Abacn reviewed Jul 27, 2026

View reviewed changes

ahmedabu98 reviewed Jul 28, 2026

View reviewed changes

chamikaramj added 3 commits July 28, 2026 10:54

Updates the Delta Lake source to support reading bounded change data

0c74b47

Update and add a test

4b75dc0

Address reviewer comments

b85ccc2

chamikaramj force-pushed the delta_lake_cdc branch from 82ab11e to b85ccc2 Compare July 29, 2026 05:20

chamikaramj commented Jul 29, 2026

View reviewed changes

		final List<Row> cdcRows = new ArrayList<>();
		final List<Row> addRows = new ArrayList<>();

Uh oh!

Conversation

chamikaramj commented Jul 22, 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 Jul 22, 2026

Uh oh!

gemini-code-assist Bot commented Jul 25, 2026

Uh oh!

github-actions Bot commented Jul 25, 2026

Uh oh!

chamikaramj commented Jul 25, 2026

Uh oh!

github-actions Bot commented Jul 25, 2026

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!

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 commented Jul 22, 2026 •

edited

Loading