[Iceberg CDC] Add Snapshot Changelog Scanner (#38836)

* add changelog scanner

* sync and spotless

* address comments

Author: ahmedabu98

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergScanConfig.java

@@ -17,6 +17,8 @@
  */
 package org.apache.beam.sdk.io.iceberg;
 
+import static org.apache.beam.sdk.io.iceberg.IcebergUtils.icebergSchemaToBeamSchema;
+import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
@@ -26,6 +28,7 @@
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -37,12 +40,15 @@
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableUtil;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.expressions.Evaluator;
 import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.types.Comparators;
 import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.SnapshotUtil;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.checkerframework.dataflow.qual.Pure;
@@ -54,6 +60,8 @@ public abstract class IcebergScanConfig implements Serializable {
   private transient org.apache.iceberg.@MonotonicNonNull Schema cachedProjectedSchema;
   private transient org.apache.iceberg.@MonotonicNonNull Schema cachedRequiredSchema;
   private transient @MonotonicNonNull Expression cachedFilter;
+  private transient org.apache.iceberg.@MonotonicNonNull Schema cachedRecordIdSchema;
+  private transient @MonotonicNonNull Schema cachedRowIdBeamSchema;
 
   public enum ScanType {
     TABLE,
@@ -144,6 +152,25 @@ public org.apache.iceberg.Schema getRequiredSchema() {
     return cachedRequiredSchema;
   }
 
+  public org.apache.iceberg.Schema recordIdSchema() {
+    if (cachedRecordIdSchema == null) {
+      org.apache.iceberg.Schema fullSchema = getTable().schema();
+      cachedRecordIdSchema = TypeUtil.select(fullSchema, fullSchema.identifierFieldIds());
+    }
+    return cachedRecordIdSchema;
+  }
+
+  public Schema rowIdBeamSchema() {
+    if (cachedRowIdBeamSchema == null) {
+      cachedRowIdBeamSchema = icebergSchemaToBeamSchema(recordIdSchema());
+    }
+    return cachedRowIdBeamSchema;
+  }
+
+  public Comparator<StructLike> recordIdComparator() {
+    return Comparators.forType(recordIdSchema().asStruct());
+  }
+
   @Pure
   @Nullable
   public Evaluator getEvaluator(org.apache.iceberg.Schema requiredSchema) {
@@ -408,6 +435,28 @@ void validate(Table table) {
         getToTimestamp() == null || getToSnapshot() == null,
         error("only one of 'to_timestamp' or 'to_snapshot' can be set"));
 
+    @Nullable Long fromSnapshotId = ReadUtils.getFromSnapshotInclusive(table, this);
+    @Nullable Long toSnapshotId = ReadUtils.getToSnapshot(table, this);
+    if (fromSnapshotId != null) {
+      checkArgumentNotNull(
+          table.snapshot(fromSnapshotId),
+          error("configured starting snapshot does not exist: '%s'"),
+          fromSnapshotId);
+    }
+    if (toSnapshotId != null) {
+      checkArgumentNotNull(
+          table.snapshot(toSnapshotId),
+          error("configured end snapshot does not exist: '%s'"),
+          toSnapshotId);
+    }
+    if (fromSnapshotId != null && toSnapshotId != null) {
+      checkArgument(
+          SnapshotUtil.isAncestorOf(table, toSnapshotId, fromSnapshotId),
+          error("fromSnapshot '%s' is not an ancestor of toSnapshot '%s'"),
+          fromSnapshotId,
+          toSnapshotId);
+    }
+
     if (getPollInterval() != null) {
       checkArgument(
           Boolean.TRUE.equals(getStreaming()),

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/TableCache.java

@@ -40,30 +40,30 @@
  * refreshed after a caller started its request, the caller reuses that refresh instead of making
  * another catalog call.
  */
-class TableCache {
+public class TableCache {
   static final Duration DEFAULT_REFRESH_INTERVAL = Duration.ofMinutes(2);
 
   private static final Cache<CacheKey, CachedTable> TABLES =
       CacheBuilder.newBuilder().expireAfterAccess(1, TimeUnit.HOURS).build();
 
   /** Returns the cached table, loading it from the catalog on a cache miss. */
-  static Table get(IcebergCatalogConfig catalogConfig, TableIdentifier identifier) {
+  public static Table get(IcebergCatalogConfig catalogConfig, TableIdentifier identifier) {
     return get(catalogConfig, identifier, () -> catalogConfig.catalog().loadTable(identifier));
   }
 
   /** Returns the cached table for a string identifier, loading it on a cache miss. */
-  static Table get(IcebergCatalogConfig catalogConfig, String identifier) {
+  public static Table get(IcebergCatalogConfig catalogConfig, String identifier) {
     return get(catalogConfig, TableIdentifier.parse(identifier));
   }
 
   /** Returns the cached table, using the given loader only on a cache miss. */
-  static Table get(
+  public static Table get(
       IcebergCatalogConfig catalogConfig, TableIdentifier identifier, Callable<Table> loader) {
     return getEntry(catalogConfig, identifier, loader).table;
   }
 
   /** Returns the cached table after forcing a refresh of any pre-existing cache entry. */
-  static Table getRefreshed(IcebergCatalogConfig catalogConfig, TableIdentifier identifier) {
+  public static Table getRefreshed(IcebergCatalogConfig catalogConfig, TableIdentifier identifier) {
     Instant refreshRequestTime = Instant.now();
     CachedTable cachedTable =
         getEntry(catalogConfig, identifier, () -> catalogConfig.catalog().loadTable(identifier));
@@ -72,22 +72,26 @@ static Table getRefreshed(IcebergCatalogConfig catalogConfig, TableIdentifier id
   }
 
   /** Returns the cached table for a string identifier after refreshing any pre-existing entry. */
-  static Table getRefreshed(IcebergCatalogConfig catalogConfig, String identifier) {
+  public static Table getRefreshed(IcebergCatalogConfig catalogConfig, String identifier) {
     return getRefreshed(catalogConfig, TableIdentifier.parse(identifier));
   }
 
+  public static Table getAndRefreshIfStale(IcebergCatalogConfig catalogConfig, String identifier) {
+    return getAndRefreshIfStale(catalogConfig, TableIdentifier.parse(identifier));
+  }
+
   /**
    * Returns the cached table, refreshing it only if it is older than {@link
    * #DEFAULT_REFRESH_INTERVAL}.
    */
-  static Table getAndRefreshIfStale(
+  public static Table getAndRefreshIfStale(
       IcebergCatalogConfig catalogConfig, TableIdentifier identifier) {
     return getAndRefreshIfStale(
         catalogConfig, identifier, () -> catalogConfig.catalog().loadTable(identifier));
   }
 
   /** Returns the cached table, using the loader on a miss and refreshing stale entries. */
-  static Table getAndRefreshIfStale(
+  public static Table getAndRefreshIfStale(
       IcebergCatalogConfig catalogConfig, TableIdentifier identifier, Callable<Table> loader) {
     CachedTable cachedTable = getEntry(catalogConfig, identifier, loader);
     cachedTable.refreshIfOlderThan(Instant.now().minus(DEFAULT_REFRESH_INTERVAL));

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/ChangelogDescriptor.java

@@ -0,0 +1,106 @@
+/*
+ * 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.beam.sdk.io.iceberg.cdc;
+
+import com.google.auto.value.AutoValue;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.SchemaCoder;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Descriptor for a set of {@link SerializableChangelogTask}s.
+ *
+ * <p>This carries commit-sourced metadata for all rows produced from the task group. These values
+ * are not read from data files; they are appended to final CDC output rows by {@link
+ * CdcOutputUtils#outputRow} when requested.
+ */
+@DefaultSchema(AutoValueSchema.class)
+@AutoValue
+public abstract class ChangelogDescriptor {
+  public static Builder builder() {
+    return new AutoValue_ChangelogDescriptor.Builder();
+  }
+
+  @SuppressWarnings("nullness")
+  public static SchemaCoder<ChangelogDescriptor> coder(Schema overlapSchema) {
+    Schema descriptorSchema =
+        Schema.builder()
+            .addStringField("tableIdentifierString")
+            .addInt64Field("snapshotSequenceNumber")
+            .addInt64Field("commitSnapshotId")
+            .addNullableField("overlapLower", Schema.FieldType.row(overlapSchema))
+            .addNullableField("overlapUpper", Schema.FieldType.row(overlapSchema))
+            .build();
+
+    return SchemaCoder.of(
+        descriptorSchema,
+        TypeDescriptor.of(ChangelogDescriptor.class),
+        descriptor ->
+            Row.withSchema(descriptorSchema)
+                .addValues(
+                    descriptor.getTableIdentifierString(),
+                    descriptor.getSnapshotSequenceNumber(),
+                    descriptor.getCommitSnapshotId(),
+                    descriptor.getOverlapLower(),
+                    descriptor.getOverlapUpper())
+                .build(),
+        row ->
+            ChangelogDescriptor.builder()
+                .setTableIdentifierString(row.getString("tableIdentifierString"))
+                .setSnapshotSequenceNumber(row.getInt64("snapshotSequenceNumber"))
+                .setCommitSnapshotId(row.getInt64("commitSnapshotId"))
+                .setOverlapLower(row.getRow("overlapLower"))
+                .setOverlapUpper(row.getRow("overlapUpper"))
+                .build());
+  }
+
+  @SchemaFieldNumber("0")
+  public abstract String getTableIdentifierString();
+
+  @SchemaFieldNumber("1")
+  public abstract long getSnapshotSequenceNumber();
+
+  @SchemaFieldNumber("2")
+  public abstract long getCommitSnapshotId();
+
+  @SchemaFieldNumber("3")
+  public abstract @Nullable Row getOverlapLower();
+
+  @SchemaFieldNumber("4")
+  public abstract @Nullable Row getOverlapUpper();
+
+  @AutoValue.Builder
+  public abstract static class Builder {
+    abstract Builder setTableIdentifierString(String table);
+
+    abstract Builder setSnapshotSequenceNumber(long sequenceNumber);
+
+    abstract Builder setCommitSnapshotId(long snapshotId);
+
+    abstract Builder setOverlapLower(@Nullable Row overlapLower);
+
+    abstract Builder setOverlapUpper(@Nullable Row overlapUpper);
+
+    abstract ChangelogDescriptor build();
+  }
+}