[Iceberg CDC sink] Commit files stage

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

@@ -0,0 +1,195 @@
+/*
+ * 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.sink;
+
+import com.google.auto.value.AutoValue;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.BiConsumer;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig;
+import org.apache.beam.sdk.io.iceberg.SnapshotInfo;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.NoSuchSchemaException;
+import org.apache.beam.sdk.schemas.SchemaRegistry;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.WithKeys;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * The CDC sink's commit stage: commits each {@code (destination, window)}'s merged writer outputs
+ * (represented as {@link ShardDeltaFiles}) as a single Iceberg snapshot, in ascending window-end
+ * order. Re-keys by destination, gathers all shards per {@code (dest, window)}, captures the window
+ * end, then re-windows into the global window for the stateful {@link OrderedCommitFn}.
+ *
+ * <p>Each commit writes the window's end millis to the snapshot summary as an idempotency token,
+ * keyed by the sink's unique {@code sinkId}. The committer recovers it by scanning snapshot
+ * ancestry: once on first touch of a destination, and again on every commit fire. Any window whose
+ * end is at or below the recovered token has already been committed, so it is skipped.
+ */
+class CommitDeltas
+    extends PTransform<PCollection<ShardDeltaFiles>, PCollection<KV<String, SnapshotInfo>>> {
+
+  // test-only attributes
+  @VisibleForTesting static volatile @Nullable Runnable preCommitHookForTest = null;
+  @VisibleForTesting static volatile @Nullable BiConsumer<Long, List<Long>> onFireForTest = null;
+
+  private final IcebergCatalogConfig catalogConfig;
+  private final String sinkId;
+  private final Map<String, String> snapshotProperties;
+  private final long heartbeatMillis;
+
+  /** The expansion's runId. */
+  private final String runId;
+
+  private OrderedCommitFn.Clock clock = System::currentTimeMillis;
+
+  @VisibleForTesting
+  CommitDeltas(IcebergCatalogConfig catalogConfig, String sinkId) {
+    this(catalogConfig, sinkId, null, null);
+  }
+
+  @VisibleForTesting
+  CommitDeltas(
+      IcebergCatalogConfig catalogConfig,
+      String sinkId,
+      @Nullable Map<String, String> snapshotProperties,
+      @Nullable Long tokenHeartbeatMillis) {
+    this(
+        catalogConfig,
+        sinkId,
+        snapshotProperties,
+        tokenHeartbeatMillis,
+        UUID.randomUUID().toString());
+  }
+
+  CommitDeltas(
+      IcebergCatalogConfig catalogConfig,
+      String sinkId,
+      @Nullable Map<String, String> snapshotProperties,
+      @Nullable Long tokenHeartbeatMillis,
+      String runId) {
+    this.catalogConfig = catalogConfig;
+    this.sinkId = sinkId;
+    this.snapshotProperties =
+        snapshotProperties == null ? Collections.emptyMap() : snapshotProperties;
+    this.heartbeatMillis = tokenHeartbeatMillis == null ? 0L : tokenHeartbeatMillis;
+    this.runId = runId;
+  }
+
+  /** Overrides the committer's clock to test skew deterministically. */
+  @VisibleForTesting
+  CommitDeltas withClockForTest(OrderedCommitFn.Clock clock) {
+    this.clock = clock;
+    return this;
+  }
+
+  @Override
+  public PCollection<KV<String, SnapshotInfo>> expand(PCollection<ShardDeltaFiles> input) {
+    boolean streaming = input.isBounded() == PCollection.IsBounded.UNBOUNDED;
+    return input
+        .apply("KeyByDestination", WithKeys.of(ShardDeltaFiles::getTableIdentifierString))
+        .setCoder(KvCoder.of(StringUtf8Coder.of(), ShardDeltaFiles.coder()))
+        // One element per (dest, window): every shard's output for the pair.
+        // Late panes are handled upstream by SplitLateData so no need to handle again here.
+        .apply("GatherShardsPerWindow", GroupByKey.create())
+        .apply("CaptureWindowEnd", ParDo.of(new CaptureWindowEndFn()))
+        .setCoder(KvCoder.of(StringUtf8Coder.of(), windowedCommitCoder()))
+        .apply("ToGlobalWindow", Window.into(new GlobalWindows()))
+        .apply(
+            "OrderedCommit",
+            ParDo.of(
+                new OrderedCommitFn(
+                    catalogConfig,
+                    sinkId,
+                    runId,
+                    snapshotProperties,
+                    heartbeatMillis,
+                    clock,
+                    streaming)))
+        .setCoder(KvCoder.of(StringUtf8Coder.of(), snapshotInfoCoder()));
+  }
+
+  private static Coder<SnapshotInfo> snapshotInfoCoder() {
+    try {
+      return SchemaRegistry.createDefault().getSchemaCoder(SnapshotInfo.class);
+    } catch (NoSuchSchemaException e) {
+      throw new RuntimeException("Could not build a coder for SnapshotInfo.", e);
+    }
+  }
+
+  static Coder<WindowedCommit> windowedCommitCoder() {
+    try {
+      return SchemaRegistry.createDefault().getSchemaCoder(WindowedCommit.class);
+    } catch (NoSuchSchemaException e) {
+      throw new RuntimeException("Could not build a coder for WindowedCommit.", e);
+    }
+  }
+
+  /**
+   * One {@code (destination, window)}'s merged writer outputs, tagged with the window's end.
+   *
+   * <p>The window end is a deterministic {@code FixedWindows} boundary and doubles as the
+   * restart-safe idempotency token. A window whose end is at or below the recovered
+   * committed-through token is skipped.
+   */
+  @AutoValue
+  @DefaultSchema(AutoValueSchema.class)
+  abstract static class WindowedCommit {
+    @SchemaFieldNumber("0")
+    public abstract long getWindowEndMs();
+
+    @SchemaFieldNumber("1")
+    public abstract List<ShardDeltaFiles> getFiles();
+
+    public static WindowedCommit of(long windowEndMs, List<ShardDeltaFiles> files) {
+      return new AutoValue_CommitDeltas_WindowedCommit(windowEndMs, files);
+    }
+  }
+
+  /** Folds a {@code (dest, window)}'s shard outputs into one {@link WindowedCommit}. */
+  static class CaptureWindowEndFn
+      extends DoFn<KV<String, Iterable<ShardDeltaFiles>>, KV<String, WindowedCommit>> {
+    @ProcessElement
+    public void process(
+        @Element KV<String, Iterable<ShardDeltaFiles>> element,
+        BoundedWindow window,
+        OutputReceiver<KV<String, WindowedCommit>> out) {
+      long windowEndMs = window.maxTimestamp().getMillis();
+      List<ShardDeltaFiles> files = Lists.newArrayList(element.getValue());
+      out.outputWithTimestamp(
+          KV.of(element.getKey(), WindowedCommit.of(windowEndMs, files)), window.maxTimestamp());
+    }
+  }
+}

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

@@ -0,0 +1,135 @@
+/*
+ * 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.sink;
+
+import java.io.Serializable;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileContent;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/** The committer's metrics, all namespaced under {@link CommitDeltas}. */
+final class CommitterMetrics implements Serializable {
+
+  final Counter snapshotsCreated = Metrics.counter(CommitDeltas.class, "snapshotsCreated");
+  final Counter committedDataFiles = Metrics.counter(CommitDeltas.class, "committedDataFiles");
+  final Counter committedDeleteFiles = Metrics.counter(CommitDeltas.class, "committedDeleteFiles");
+  final Counter committedRecords = Metrics.counter(CommitDeltas.class, "committedRecords");
+  final Counter committedEqualityDeleteRecords =
+      Metrics.counter(CommitDeltas.class, "committedEqualityDeleteRecords");
+  final Counter committedBytes = Metrics.counter(CommitDeltas.class, "committedBytes");
+  final Distribution commitDurationMs =
+      Metrics.distribution(CommitDeltas.class, "commitDurationMs");
+  final Counter commitFailures = Metrics.counter(CommitDeltas.class, "commitFailures");
+
+  final Counter alreadyCommittedWindowsSkipped =
+      Metrics.counter(CommitDeltas.class, "alreadyCommittedWindowsSkipped");
+  final Counter orphanFiles = Metrics.counter(CommitDeltas.class, "orphanFiles");
+  final Counter tokenParseFailures = Metrics.counter(CommitDeltas.class, "tokenParseFailures");
+  final Counter suspectedTokenExpiry = Metrics.counter(CommitDeltas.class, "suspectedTokenExpiry");
+  final Counter crossWindowSequenceInversions =
+      Metrics.counter(CommitDeltas.class, "crossWindowSequenceInversions");
+  final Counter specMismatchedWindows =
+      Metrics.counter(CommitDeltas.class, "specMismatchedWindows");
+
+  final Counter heartbeatCommits = Metrics.counter(CommitDeltas.class, "heartbeatCommits");
+
+  /** Records a successful commit's volume metrics. */
+  void recordCommit(CommitSummary summary) {
+    committedDataFiles.inc(summary.dataFileCount);
+    committedDeleteFiles.inc(summary.deleteFileCount);
+    committedRecords.inc(summary.dataRecords);
+    committedEqualityDeleteRecords.inc(summary.equalityDeleteRecords);
+    committedBytes.inc(summary.bytes);
+    snapshotsCreated.inc();
+  }
+
+  /** One commit's volume and partition-spec summary, computed from its files. */
+  static final class CommitSummary {
+    final long dataFileCount;
+    final long deleteFileCount;
+    final long dataRecords;
+    final long equalityDeleteRecords;
+    final long bytes;
+    final @Nullable Integer firstSpecId;
+    final Set<Integer> specIds;
+    final boolean hasEqualityDeletes;
+
+    private CommitSummary(
+        long dataFileCount,
+        long deleteFileCount,
+        long dataRecords,
+        long equalityDeleteRecords,
+        long bytes,
+        @Nullable Integer firstSpecId,
+        Set<Integer> specIds,
+        boolean hasEqualityDeletes) {
+      this.dataFileCount = dataFileCount;
+      this.deleteFileCount = deleteFileCount;
+      this.dataRecords = dataRecords;
+      this.equalityDeleteRecords = equalityDeleteRecords;
+      this.bytes = bytes;
+      this.firstSpecId = firstSpecId;
+      this.specIds = specIds;
+      this.hasEqualityDeletes = hasEqualityDeletes;
+    }
+
+    static CommitSummary of(List<DataFile> dataFiles, List<DeleteFile> deleteFiles) {
+      long dataRecords = 0;
+      long bytes = 0;
+      @Nullable Integer firstSpecId = null;
+      Set<Integer> specIds = new HashSet<>();
+      for (DataFile f : dataFiles) {
+        dataRecords += f.recordCount();
+        bytes += f.fileSizeInBytes();
+        if (firstSpecId == null) {
+          firstSpecId = f.specId();
+        }
+        specIds.add(f.specId());
+      }
+      long equalityDeleteRecords = 0;
+      boolean hasEqualityDeletes = false;
+      for (DeleteFile f : deleteFiles) {
+        bytes += f.fileSizeInBytes();
+        if (firstSpecId == null) {
+          firstSpecId = f.specId();
+        }
+        specIds.add(f.specId());
+        if (f.content() == FileContent.EQUALITY_DELETES) {
+          equalityDeleteRecords += f.recordCount();
+          hasEqualityDeletes = true;
+        }
+      }
+      return new CommitSummary(
+          dataFiles.size(),
+          deleteFiles.size(),
+          dataRecords,
+          equalityDeleteRecords,
+          bytes,
+          firstSpecId,
+          specIds,
+          hasEqualityDeletes);
+    }
+  }
+}