chamikaramj commented on code in PR #40134: URL: https://github.com/apache/beam/pull/40134#discussion_r4050641948
########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltas.java: ########## @@ -0,0 +1,1015 @@ +/* + * 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 static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects.firstNonNull; + +import com.google.auto.value.AutoValue; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.function.BiConsumer; +import java.util.function.LongSupplier; +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.coders.VarIntCoder; +import org.apache.beam.sdk.coders.VarLongCoder; +import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; +import org.apache.beam.sdk.io.iceberg.SerializableDataFile; +import org.apache.beam.sdk.io.iceberg.SerializableDeleteFile; +import org.apache.beam.sdk.io.iceberg.SnapshotInfo; +import org.apache.beam.sdk.io.iceberg.TableCache; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.Distribution; +import org.apache.beam.sdk.metrics.Metrics; +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.state.BagState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.TimeDomain; +import org.apache.beam.sdk.state.Timer; +import org.apache.beam.sdk.state.TimerSpec; +import org.apache.beam.sdk.state.TimerSpecs; +import org.apache.beam.sdk.state.ValueState; +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.GlobalWindow; +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.apache.iceberg.AppendFiles; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.RowDelta; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SnapshotUpdate; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.Table; +import org.apache.iceberg.util.ThreadPools; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.Duration; +import org.joda.time.Instant; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 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>>> { + + private static final Logger LOG = LoggerFactory.getLogger(CommitDeltas.class); + + // 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 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(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. + .apply("GatherShardsPerWindow", GroupByKey.create()) Review Comment: Do we have to handle late data via after this GBK similar to `CommitWindows` ? ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltas.java: ########## @@ -0,0 +1,1015 @@ +/* + * 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 static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects.firstNonNull; + +import com.google.auto.value.AutoValue; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.function.BiConsumer; +import java.util.function.LongSupplier; +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.coders.VarIntCoder; +import org.apache.beam.sdk.coders.VarLongCoder; +import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; +import org.apache.beam.sdk.io.iceberg.SerializableDataFile; +import org.apache.beam.sdk.io.iceberg.SerializableDeleteFile; +import org.apache.beam.sdk.io.iceberg.SnapshotInfo; +import org.apache.beam.sdk.io.iceberg.TableCache; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.Distribution; +import org.apache.beam.sdk.metrics.Metrics; +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.state.BagState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.TimeDomain; +import org.apache.beam.sdk.state.Timer; +import org.apache.beam.sdk.state.TimerSpec; +import org.apache.beam.sdk.state.TimerSpecs; +import org.apache.beam.sdk.state.ValueState; +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.GlobalWindow; +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.apache.iceberg.AppendFiles; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.RowDelta; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SnapshotUpdate; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.Table; +import org.apache.iceberg.util.ThreadPools; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.Duration; +import org.joda.time.Instant; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 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>>> { + + private static final Logger LOG = LoggerFactory.getLogger(CommitDeltas.class); + + // 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 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(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. + .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); + } + } + + /** Max number of file paths listed in a skip-path WARN before truncating. */ + private static final int SKIP_PATHS_LOGGED = 5; + + /** Every file path a window carries: data files first then delete files. */ + @VisibleForTesting + static List<String> filePaths(WindowedCommit wc) { + List<String> paths = new ArrayList<>(); + List<String> deletePaths = new ArrayList<>(); + for (ShardDeltaFiles shard : wc.getFiles()) { + for (SerializableDataFile dataFile : shard.getDataFiles()) { + paths.add(dataFile.getPath()); + } + for (SerializableDeleteFile deleteFile : shard.getDeleteFiles()) { + deletePaths.add(deleteFile.getLocation()); + } + } + paths.addAll(deletePaths); + return paths; + } + + /** Renders up to {@link #SKIP_PATHS_LOGGED} of a skipped window's file paths, plus a count. */ + @VisibleForTesting + static String describeSkippedFiles(WindowedCommit wc) { + return describePaths(filePaths(wc)); + } + + private static String describePaths(List<String> paths) { + StringBuilder sb = new StringBuilder(); + int shown = Math.min(SKIP_PATHS_LOGGED, paths.size()); + for (int i = 0; i < shown; i++) { + if (i > 0) { + sb.append(", "); + } + sb.append(paths.get(i)); + } + if (paths.size() > SKIP_PATHS_LOGGED) { + sb.append(" (… ").append(paths.size() - SKIP_PATHS_LOGGED).append(" more)"); + } + return sb.toString(); + } + + /** + * 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()); + } + } + + /** + * A serializable millisecond clock; injectable via {@link #withClockForTest} so heartbeat tests + * can skew "now" deterministically. + */ + @FunctionalInterface + interface Clock extends LongSupplier, Serializable {} + + /** The committer's metrics, all namespaced under {@link CommitDeltas}. */ + static 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"); + } + + /** + * The ordered, idempotent committer. Keyed by destination; keeps the last-committed window-end + * and a bag of pending windows, plus one event-time timer armed at the earliest pending end. On + * fire, it commits every pending window at or below the input watermark, in ascending order, each + * as its own single-snapshot commit. A configured {@code tokenHeartbeatMillis} adds an idle + * token-refresh timer ({@link #onHeartbeat}). + */ + static class OrderedCommitFn extends DoFn<KV<String, WindowedCommit>, KV<String, SnapshotInfo>> { + + private final CommitterMetrics metrics = new CommitterMetrics(); + + private final IcebergCatalogConfig catalogConfig; + private final String sinkId; + private final String runId; + private final Map<String, String> snapshotProperties; + private final CommitToken token; + + /** Idle token-refresh heartbeat interval in millis; {@code 0} = disabled. */ + private final long heartbeatMillis; + + private final Clock clock; + private final boolean streaming; + + @StateId("lastCommittedEndMs") + private final StateSpec<ValueState<Long>> lastCommittedEndMsSpec = + StateSpecs.value(VarLongCoder.of()); + + /** The max source sequence number committed so far. */ + @StateId("lastCommittedMaxSeq") + private final StateSpec<ValueState<Long>> lastCommittedMaxSeqSpec = + StateSpecs.value(VarLongCoder.of()); + + @StateId("pending") + private final StateSpec<BagState<WindowedCommit>> pendingSpec; + + /** + * The earliest uncommitted window-end in {@link #pendingSpec} (what the commit timer must be + * armed at). + */ + @StateId("earliestPending") + private final StateSpec<ValueState<Long>> earliestPendingSpec = + StateSpecs.value(VarLongCoder.of()); + + /** The partition-spec id pinned for this destination under {@link #pinnedRunIdSpec}'s runId. */ + @StateId("pinnedSpecId") + private final StateSpec<ValueState<Integer>> pinnedSpecIdSpec = + StateSpecs.value(VarIntCoder.of()); + + /** The runId the spec pin was taken under; a different live runId re-pins. */ + @StateId("pinnedRunId") + private final StateSpec<ValueState<String>> pinnedRunIdSpec = + StateSpecs.value(StringUtf8Coder.of()); + + @TimerId("commit") + private final TimerSpec commitTimerSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME); + + /** Processing-time timer that fires the idle token-refresh heartbeat (when configured). */ + @TimerId("heartbeat") + private final TimerSpec heartbeatTimerSpec = TimerSpecs.timer(TimeDomain.PROCESSING_TIME); + + OrderedCommitFn( + IcebergCatalogConfig catalogConfig, + String sinkId, + String runId, + Map<String, String> snapshotProperties, + long heartbeatMillis, + Clock clock, + boolean streaming) { + this.catalogConfig = catalogConfig; + this.sinkId = sinkId; + this.runId = runId; + this.snapshotProperties = snapshotProperties; + this.heartbeatMillis = heartbeatMillis; + this.clock = clock; + this.streaming = streaming; + this.token = + new CommitToken(sinkId, runId, metrics.tokenParseFailures, metrics.suspectedTokenExpiry); + this.pendingSpec = StateSpecs.bag(windowedCommitCoder()); + } + + @RequiresStableInput + @ProcessElement + public void process( + @Element KV<String, WindowedCommit> element, + @StateId("lastCommittedEndMs") ValueState<Long> lastCommittedEndMs, + @StateId("lastCommittedMaxSeq") ValueState<Long> lastMaxSeq, + @StateId("pending") BagState<WindowedCommit> pending, + @StateId("earliestPending") ValueState<Long> earliestPending, + @TimerId("commit") Timer commitTimer, + @TimerId("heartbeat") Timer heartbeatTimer) { + String dest = element.getKey(); + WindowedCommit wc = element.getValue(); + + long lastCommittedMs = recoverOrReadCommitted(dest, lastCommittedEndMs, lastMaxSeq); + // Sets the idle heartbeat timer on every element; it fires after a full interval of idleness. + setHeartbeat(heartbeatTimer); + if (wc.getWindowEndMs() <= lastCommittedMs) { + // Already committed (retry/duplicate, or a rerun under a stable sink_id). + skipAlreadyCommitted(dest, wc, lastCommittedMs); + return; + } + + long earliest = + Math.min(firstNonNull(earliestPending.read(), Long.MAX_VALUE), wc.getWindowEndMs()); + pending.add(wc); + setCommitTimer(earliest, earliestPending, commitTimer); + } + + /** Returns the last committed window-end for {@code dest}. */ + private long recoverOrReadCommitted( + String dest, ValueState<Long> lastCommittedEndMs, ValueState<Long> lastMaxSeq) { + @Nullable Long stored = lastCommittedEndMs.read(); + if (stored != null) { + return stored; + } + CommitToken.Recovered recovered = token.recoverFromTable(catalogConfig, dest); + // Throw before writing to state + checkRecoveredTokenNotBatchEnd(dest, recovered.committedThroughMs); + lastCommittedEndMs.write(recovered.committedThroughMs); + lastMaxSeq.write(recovered.maxCommittedSeq); + return recovered.committedThroughMs; + } + + /** + * Fails a streaming destination whose recovered token is the batch token (the global-window end + * every bounded load commits under). Every real-time window's end falls below it, so the run + * would silently skip every window forever. + */ + private void checkRecoveredTokenNotBatchEnd(String dest, long recoveredMs) { + if (!streaming || recoveredMs != GlobalWindow.INSTANCE.maxTimestamp().getMillis()) { + return; + } + throw new IllegalStateException( + "CDC sink '" + + sinkId + + "' recovered a committed-through token for table '" + + dest + + "' equal to the global-window end (" + + recoveredMs + + " ms): this sink_id was last used by a batch (bounded) load, whose single " + + "global-window commit claims every event-time window. A streaming run reusing it " + + "would skip every window forever. Use a different sink_id for the streaming " + + "continuation."); + } + + @RequiresStableInput + @OnTimer("commit") + public void onCommit( + OnTimerContext c, + @Key String dest, + @StateId("lastCommittedEndMs") ValueState<Long> lastCommittedEndMs, + @StateId("lastCommittedMaxSeq") ValueState<Long> lastMaxSeq, + @StateId("pending") BagState<WindowedCommit> pending, + @StateId("earliestPending") ValueState<Long> earliestPending, + @StateId("pinnedSpecId") ValueState<Integer> pinnedSpecId, + @StateId("pinnedRunId") ValueState<String> pinnedRunId, + @TimerId("commit") Timer timer, + OutputReceiver<KV<String, SnapshotInfo>> out) { + // The "commit" timer is always armed at the earliest pending window's end. + long earliestPendingWindowEnd = c.timestamp().getMillis(); + // Current input watermark tells us that all past pending windows are safe to commit. + long inputWatermark = timer.getCurrentRelativeTime().getMillis(); Review Comment: This might be runner dependent. Let's adjust or document (if the correctness is preserved). ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltas.java: ########## @@ -0,0 +1,1015 @@ +/* + * 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 static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects.firstNonNull; + +import com.google.auto.value.AutoValue; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.function.BiConsumer; +import java.util.function.LongSupplier; +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.coders.VarIntCoder; +import org.apache.beam.sdk.coders.VarLongCoder; +import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; +import org.apache.beam.sdk.io.iceberg.SerializableDataFile; +import org.apache.beam.sdk.io.iceberg.SerializableDeleteFile; +import org.apache.beam.sdk.io.iceberg.SnapshotInfo; +import org.apache.beam.sdk.io.iceberg.TableCache; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.Distribution; +import org.apache.beam.sdk.metrics.Metrics; +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.state.BagState; +import org.apache.beam.sdk.state.StateSpec; +import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.state.TimeDomain; +import org.apache.beam.sdk.state.Timer; +import org.apache.beam.sdk.state.TimerSpec; +import org.apache.beam.sdk.state.TimerSpecs; +import org.apache.beam.sdk.state.ValueState; +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.GlobalWindow; +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.apache.iceberg.AppendFiles; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.RowDelta; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SnapshotUpdate; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.Table; +import org.apache.iceberg.util.ThreadPools; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.Duration; +import org.joda.time.Instant; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 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>>> { + + private static final Logger LOG = LoggerFactory.getLogger(CommitDeltas.class); + + // 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 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(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. + .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); + } + } + + /** Max number of file paths listed in a skip-path WARN before truncating. */ + private static final int SKIP_PATHS_LOGGED = 5; + + /** Every file path a window carries: data files first then delete files. */ + @VisibleForTesting + static List<String> filePaths(WindowedCommit wc) { + List<String> paths = new ArrayList<>(); + List<String> deletePaths = new ArrayList<>(); + for (ShardDeltaFiles shard : wc.getFiles()) { + for (SerializableDataFile dataFile : shard.getDataFiles()) { + paths.add(dataFile.getPath()); + } + for (SerializableDeleteFile deleteFile : shard.getDeleteFiles()) { + deletePaths.add(deleteFile.getLocation()); + } + } + paths.addAll(deletePaths); + return paths; + } + + /** Renders up to {@link #SKIP_PATHS_LOGGED} of a skipped window's file paths, plus a count. */ + @VisibleForTesting + static String describeSkippedFiles(WindowedCommit wc) { + return describePaths(filePaths(wc)); + } + + private static String describePaths(List<String> paths) { + StringBuilder sb = new StringBuilder(); + int shown = Math.min(SKIP_PATHS_LOGGED, paths.size()); + for (int i = 0; i < shown; i++) { + if (i > 0) { + sb.append(", "); + } + sb.append(paths.get(i)); + } + if (paths.size() > SKIP_PATHS_LOGGED) { + sb.append(" (… ").append(paths.size() - SKIP_PATHS_LOGGED).append(" more)"); + } + return sb.toString(); + } + + /** + * 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()); + } + } + + /** + * A serializable millisecond clock; injectable via {@link #withClockForTest} so heartbeat tests + * can skew "now" deterministically. + */ + @FunctionalInterface + interface Clock extends LongSupplier, Serializable {} + + /** The committer's metrics, all namespaced under {@link CommitDeltas}. */ + static 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"); + } + + /** + * The ordered, idempotent committer. Keyed by destination; keeps the last-committed window-end + * and a bag of pending windows, plus one event-time timer armed at the earliest pending end. On + * fire, it commits every pending window at or below the input watermark, in ascending order, each + * as its own single-snapshot commit. A configured {@code tokenHeartbeatMillis} adds an idle + * token-refresh timer ({@link #onHeartbeat}). + */ + static class OrderedCommitFn extends DoFn<KV<String, WindowedCommit>, KV<String, SnapshotInfo>> { Review Comment: This file is extremely long. Can we move some of the inner classes and interfaces into separate files ? Also add dedicated test classes when appropriate. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
