Spark: Add native CDC for copy-on-write tables

spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java

@@ -73,6 +73,8 @@
 import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException;
 import org.apache.spark.sql.catalyst.analysis.ViewAlreadyExistsException;
 import org.apache.spark.sql.catalyst.analysis.ViewUtil;
+import org.apache.spark.sql.connector.catalog.Changelog;
+import org.apache.spark.sql.connector.catalog.ChangelogContext;
 import org.apache.spark.sql.connector.catalog.Identifier;
 import org.apache.spark.sql.connector.catalog.NamespaceChange;
 import org.apache.spark.sql.connector.catalog.StagedTable;
@@ -192,6 +194,17 @@ public Table loadTable(Identifier ident, long timestampMicros) throws NoSuchTabl
     return load(ident, TimeTravel.timestampMicros(timestampMicros));
   }
 
+  @Override
+  public Changelog loadChangelog(
+      Identifier ident, ChangelogContext context, CaseInsensitiveStringMap options)
+      throws NoSuchTableException {
+    try {
+      return new SparkChangelogTable(icebergCatalog.loadTable(buildIdentifier(ident)), context);
+    } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
+      throw new NoSuchTableException(ident);
+    }
+  }
+
   @Override
   public boolean tableExists(Identifier ident) {
     if (isPathIdentifier(ident)) {

spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java

@@ -43,6 +43,8 @@
 import org.apache.spark.sql.catalyst.analysis.ViewAlreadyExistsException;
 import org.apache.spark.sql.connector.catalog.CatalogExtension;
 import org.apache.spark.sql.connector.catalog.CatalogPlugin;
+import org.apache.spark.sql.connector.catalog.Changelog;
+import org.apache.spark.sql.connector.catalog.ChangelogContext;
 import org.apache.spark.sql.connector.catalog.FunctionCatalog;
 import org.apache.spark.sql.connector.catalog.Identifier;
 import org.apache.spark.sql.connector.catalog.NamespaceChange;
@@ -224,6 +226,17 @@ public Table loadTable(Identifier ident, long timestamp) throws NoSuchTableExcep
     }
   }
 
+  @Override
+  public Changelog loadChangelog(
+      Identifier ident, ChangelogContext context, CaseInsensitiveStringMap options)
+      throws NoSuchTableException {
+    try {
+      return icebergCatalog.loadChangelog(ident, context, options);
+    } catch (NoSuchTableException e) {
+      return getSessionCatalog().loadChangelog(ident, context, options);
+    }
+  }
+
   @Override
   public void invalidateTable(Identifier ident) {
     // We do not need to check whether the table exists and whether

spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/ChangelogRowReader.java

@@ -32,6 +32,7 @@
 import org.apache.iceberg.DeletedRowsScanTask;
 import org.apache.iceberg.ScanTaskGroup;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.CloseableIterator;
@@ -42,12 +43,15 @@
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
 import org.apache.spark.sql.catalyst.expressions.JoinedRow;
+import org.apache.spark.sql.connector.catalog.Changelog;
 import org.apache.spark.sql.connector.read.PartitionReader;
 import org.apache.spark.unsafe.types.UTF8String;
 
 class ChangelogRowReader extends BaseRowReader<ChangelogScanTask>
     implements PartitionReader<InternalRow> {
 
+  private final SparkChangelogReadMode readMode;
+
   ChangelogRowReader(SparkInputPartition partition) {
     this(
         partition.table(),
@@ -69,9 +73,13 @@ class ChangelogRowReader extends BaseRowReader<ChangelogScanTask>
         table,
         fileIO,
         taskGroup,
-        ChangelogUtil.dropChangelogMetadata(expectedSchema),
+        dataSchema(table, expectedSchema),
         caseSensitive,
         cacheDeleteFilesOnExecutors);
+    this.readMode =
+        expectedSchema.findField(SparkChangelogTable.COMMIT_VERSION) != null
+            ? SparkChangelogReadMode.SPARK_CDC
+            : SparkChangelogReadMode.ICEBERG_CHANGELOG;
   }
 
   @Override
@@ -86,16 +94,40 @@ protected CloseableIterator<InternalRow> open(ChangelogScanTask task) {
     return cdcRows.iterator();
   }
 
-  private static InternalRow changelogMetadata(ChangelogScanTask task) {
+  private InternalRow changelogMetadata(ChangelogScanTask task) {
     InternalRow metadataRow = new GenericInternalRow(3);
 
-    metadataRow.update(0, UTF8String.fromString(task.operation().name()));
-    metadataRow.update(1, task.changeOrdinal());
-    metadataRow.update(2, task.commitSnapshotId());
+    if (readMode.isSparkCdc()) {
+      Snapshot snapshot = table().snapshot(task.commitSnapshotId());
+      Preconditions.checkNotNull(
+          snapshot, "Cannot find snapshot for changelog task: %s", task.commitSnapshotId());
+      metadataRow.update(0, UTF8String.fromString(changeType(task)));
+      metadataRow.update(1, snapshot.sequenceNumber());
+      metadataRow.update(2, snapshot.timestampMillis() * 1000);
+    } else {
+      metadataRow.update(0, UTF8String.fromString(task.operation().name()));
+      metadataRow.update(1, task.changeOrdinal());
+      metadataRow.update(2, task.commitSnapshotId());
+    }
 
     return metadataRow;
   }
 
+  private static Schema dataSchema(Table table, Schema expectedSchema) {
+    return expectedSchema.findField(SparkChangelogTable.COMMIT_VERSION) != null
+        ? SparkChangelogTable.cdcDataSchema(table)
+        : ChangelogUtil.dropChangelogMetadata(expectedSchema);
+  }
+
+  private static String changeType(ChangelogScanTask task) {
+    return switch (task.operation()) {
+      case INSERT -> Changelog.CHANGE_TYPE_INSERT;
+      case DELETE -> Changelog.CHANGE_TYPE_DELETE;
+      case UPDATE_BEFORE -> Changelog.CHANGE_TYPE_UPDATE_PREIMAGE;
+      case UPDATE_AFTER -> Changelog.CHANGE_TYPE_UPDATE_POSTIMAGE;
+    };
+  }
+
   private CloseableIterable<InternalRow> openChangelogScanTask(ChangelogScanTask task) {
     if (task instanceof AddedRowsScanTask) {
       return openAddedRowsScanTask((AddedRowsScanTask) task);

spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogMicroBatchStream.java

@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.List;
+import org.apache.iceberg.ChangelogScanTask;
+import org.apache.iceberg.ChangelogUtil;
+import org.apache.iceberg.IncrementalChangelogScan;
+import org.apache.iceberg.ScanTaskGroup;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.SparkReadConf;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.broadcast.Broadcast;
+import org.apache.spark.sql.connector.read.streaming.Offset;
+import org.apache.spark.sql.connector.read.streaming.ReadLimit;
+
+/**
+ * A minimal changelog stream that advances at Iceberg snapshot boundaries.
+ *
+ * <p>Each planned range contains complete snapshots, ensuring that all rows from a commit remain in
+ * the same Spark micro-batch.
+ */
+class SparkChangelogMicroBatchStream extends SparkMicroBatchStreamBase {
+
+  private Broadcast<Table> plannedTableBroadcast = null;
+
+  SparkChangelogMicroBatchStream(
+      JavaSparkContext sparkContext,
+      Table table,
+      SparkReadConf readConf,
+      Schema projection,
+      String checkpointLocation) {
+    super(
+        sparkContext,
+        table,
+        table::io,
+        readConf,
+        projection,
+        checkpointLocation,
+        () -> configuredInitialOffset(readConf));
+  }
+
+  private static StreamingOffset configuredInitialOffset(SparkReadConf readConf) {
+    return readConf.startSnapshotId() != null
+        ? new StreamingOffset(readConf.startSnapshotId(), 0, false)
+        : StreamingOffset.START_OFFSET;
+  }
+
+  @Override
+  protected StreamingOffset latestStreamingOffset() {
+    table().refresh();
+    Snapshot latest = table().currentSnapshot();
+    Long configuredEndSnapshotId = readConf().endSnapshotId();
+    if (configuredEndSnapshotId != null) {
+      latest = table().snapshot(configuredEndSnapshotId);
+    }
+
+    return latest != null
+        ? new StreamingOffset(latest.snapshotId(), 0, false)
+        : StreamingOffset.START_OFFSET;
+  }
+
+  @Override
+  public Offset latestOffset(Offset startOffset, ReadLimit limit) {
+    Preconditions.checkArgument(
+        startOffset instanceof StreamingOffset, "Invalid start offset: %s", startOffset);
+
+    StreamingOffset latestOffset = (StreamingOffset) latestOffset();
+    return latestOffset.equals(StreamingOffset.START_OFFSET) || latestOffset.equals(startOffset)
+        ? null
+        : latestOffset;
+  }
+
+  @Override
+  protected List<ScanTaskGroup<ChangelogScanTask>> planTaskGroups(
+      StreamingOffset startOffset, StreamingOffset endOffset) {
+    if (endOffset.equals(StreamingOffset.START_OFFSET) || startOffset.equals(endOffset)) {
+      return Lists.newArrayList();
+    }
+
+    table().refresh();
+    if (!startOffset.equals(StreamingOffset.START_OFFSET)) {
+      Preconditions.checkState(
+          table().snapshot(startOffset.snapshotId()) != null,
+          "Cannot load changelog start offset at expired or removed snapshot: %s",
+          startOffset.snapshotId());
+    }
+
+    Preconditions.checkState(
+        table().snapshot(endOffset.snapshotId()) != null,
+        "Cannot load changelog end offset at expired or removed snapshot: %s",
+        endOffset.snapshotId());
+
+    IncrementalChangelogScan scan =
+        table()
+            .newIncrementalChangelogScan()
+            .caseSensitive(readConf().caseSensitive())
+            .project(ChangelogUtil.changelogSchema(SparkChangelogTable.cdcDataSchema(table())));
+    if (!startOffset.equals(StreamingOffset.START_OFFSET)) {
+      scan = scan.fromSnapshotExclusive(startOffset.snapshotId());
+    }
+    scan = scan.toSnapshot(endOffset.snapshotId());
+
+    List<ScanTaskGroup<ChangelogScanTask>> taskGroups;
+    try (CloseableIterable<ScanTaskGroup<ChangelogScanTask>> groups = scan.planTasks()) {
+      taskGroups = Lists.newArrayList(groups);
+    } catch (IOException e) {
+      throw new UncheckedIOException("Failed to close Iceberg changelog task groups", e);
+    }
+
+    return taskGroups;
+  }
+
+  @Override
+  protected Broadcast<Table> tableBroadcast() {
+    if (plannedTableBroadcast != null) {
+      plannedTableBroadcast.unpersist(false);
+    }
+
+    this.plannedTableBroadcast =
+        sparkContext().broadcast(SerializableTableWithSize.copyOf(table()));
+    return plannedTableBroadcast;
+  }
+
+  @Override
+  protected void stopStream() {
+    if (plannedTableBroadcast != null) {
+      plannedTableBroadcast.unpersist(false);
+      plannedTableBroadcast = null;
+    }
+  }
+}

spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogReadMode.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source;
+
+enum SparkChangelogReadMode {
+  ICEBERG_CHANGELOG,
+  SPARK_CDC,
+  EMPTY_SPARK_CDC;
+
+  boolean isSparkCdc() {
+    return this != ICEBERG_CHANGELOG;
+  }
+
+  boolean isEmpty() {
+    return this == EMPTY_SPARK_CDC;
+  }
+}

spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScan.java

@@ -42,6 +42,7 @@
 import org.apache.spark.sql.connector.read.Scan;
 import org.apache.spark.sql.connector.read.Statistics;
 import org.apache.spark.sql.connector.read.SupportsReportStatistics;
+import org.apache.spark.sql.connector.read.streaming.MicroBatchStream;
 import org.apache.spark.sql.types.StructType;
 
 class SparkChangelogScan implements Scan, SupportsReportStatistics {
@@ -56,6 +57,7 @@ class SparkChangelogScan implements Scan, SupportsReportStatistics {
   private final List<Expression> filters;
   private final Long startSnapshotId;
   private final Long endSnapshotId;
+  private final SparkChangelogReadMode readMode;
 
   // lazy variables
   private List<ScanTaskGroup<ChangelogScanTask>> taskGroups = null;
@@ -68,7 +70,24 @@ class SparkChangelogScan implements Scan, SupportsReportStatistics {
       SparkReadConf readConf,
       Schema projection,
       List<Expression> filters) {
+    this(
+        spark,
+        table,
+        scan,
+        readConf,
+        projection,
+        filters,
+        SparkChangelogReadMode.ICEBERG_CHANGELOG);
+  }
 
+  SparkChangelogScan(
+      SparkSession spark,
+      Table table,
+      IncrementalChangelogScan scan,
+      SparkReadConf readConf,
+      Schema projection,
+      List<Expression> filters,
+      SparkChangelogReadMode readMode) {
     SparkSchemaUtil.validateMetadataColumnReferences(table.schema(), projection);
 
     this.sparkContext = JavaSparkContext.fromSparkContext(spark.sparkContext());
@@ -79,6 +98,7 @@ class SparkChangelogScan implements Scan, SupportsReportStatistics {
     this.filters = filters != null ? filters : Collections.emptyList();
     this.startSnapshotId = readConf.startSnapshotId();
     this.endSnapshotId = readConf.endSnapshotId();
+    this.readMode = readMode;
     if (scan == null) {
       this.taskGroups = Collections.emptyList();
     }
@@ -113,6 +133,16 @@ public Batch toBatch() {
         hashCode());
   }
 
+  @Override
+  public MicroBatchStream toMicroBatchStream(String checkpointLocation) {
+    if (!readMode.isSparkCdc()) {
+      throw new UnsupportedOperationException("Changelog streaming is only supported through CDC");
+    }
+
+    return new SparkChangelogMicroBatchStream(
+        sparkContext, table, readConf, projection, checkpointLocation);
+  }
+
   private List<ScanTaskGroup<ChangelogScanTask>> taskGroups() {
     if (taskGroups == null) {
       try (CloseableIterable<ScanTaskGroup<ChangelogScanTask>> groups = scan.planTasks()) {