ahmedabu98 commented on code in PR #38834: URL: https://github.com/apache/beam/pull/38834#discussion_r3476743938
########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/WatchForSnapshotsSdf.java: ########## @@ -0,0 +1,289 @@ +/* + * 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 static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.io.iceberg.IcebergScanConfig; +import org.apache.beam.sdk.io.iceberg.ReadUtils; +import org.apache.beam.sdk.io.iceberg.SnapshotInfo; +import org.apache.beam.sdk.io.iceberg.TableCache; +import org.apache.beam.sdk.io.range.OffsetRange; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.Gauge; +import org.apache.beam.sdk.metrics.Metrics; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.splittabledofn.GrowableOffsetRangeTracker; +import org.apache.beam.sdk.transforms.splittabledofn.ManualWatermarkEstimator; +import org.apache.beam.sdk.transforms.splittabledofn.OffsetRangeTracker; +import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker; +import org.apache.beam.sdk.transforms.splittabledofn.WatermarkEstimators.Manual; +import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.apache.beam.sdk.util.Preconditions; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.util.SnapshotUtil; +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; + +/** + * SplittableDoFn that watches an Iceberg table for new {@link Snapshot}s and emits them one at a + * time, advancing the watermark per snapshot. Each snapshot is later processed by {@link + * ChangelogScanner}. + * + * <p>The restriction tracks Snapshots via their <b>sequence numbers</b>, which are monotonic, + * unlike snapshot IDs. The initial range starts at the sequence number of the user-configured + * starting snapshot (or one if none configured) and runs to {@link Long#MAX_VALUE}. Each call to + * {@code @ProcessElement} claims the sequence numbers of newly discovered snapshots in + * chronological order. + * + * <p>Uses a {@link Manual} watermark estimator. After emitting a snapshot, the watermark is set to + * that snapshot's commit time. On empty polls, the watermark is bumped to {@code now() - + * MAX_SNAPSHOT_DISCOVERY_DELAY} to prevent downstream windows from stalling indefinitely during + * quiet periods. + */ [email protected] +class WatchForSnapshotsSdf extends DoFn<String, Long> { + private static final Logger LOG = LoggerFactory.getLogger(WatchForSnapshotsSdf.class); + private static final Duration DEFAULT_POLL_INTERVAL = Duration.standardSeconds(60); + + private static final Counter snapshotsEmitted = + Metrics.counter(WatchForSnapshotsSdf.class, "snapshotsEmitted"); + private static final Gauge latestEmittedSnapshotId = + Metrics.gauge(WatchForSnapshotsSdf.class, "latestEmittedSnapshotId"); + // TODO(ahmedabu98): consider exposing this as a config option + private static final Duration MAX_SNAPSHOT_DISCOVERY_DELAY = Duration.standardMinutes(5); + private static final Long POLL_FOREVER = Long.MAX_VALUE; + + private final IcebergScanConfig scanConfig; + private final Duration pollInterval; + + WatchForSnapshotsSdf(IcebergScanConfig scanConfig) { + this.scanConfig = scanConfig; + this.pollInterval = + MoreObjects.firstNonNull(scanConfig.getPollInterval(), DEFAULT_POLL_INTERVAL); + } + + @GetInitialRestriction + public OffsetRange initialRestriction() { + TableCache.setup(scanConfig); + Table table = TableCache.getRefreshed(scanConfig.getTableIdentifier()); + + long toSnapshotExclusiveSeq = POLL_FOREVER; + @Nullable Long toSnapshotId = ReadUtils.getToSnapshot(table, scanConfig); + if (toSnapshotId != null) { + toSnapshotExclusiveSeq = + Preconditions.checkStateNotNull( + table.snapshot(toSnapshotId), + "Configured end snapshot %s does not exist", + toSnapshotId) + .sequenceNumber() + + 1; + } + + @Nullable Long fromSnapshotInclusiveId = ReadUtils.getFromSnapshotInclusive(table, scanConfig); + long fromSnapshotInclusiveSeq; + if (fromSnapshotInclusiveId == null) { + fromSnapshotInclusiveSeq = 1L; // sequence numbers start at 1 + } else { + Snapshot fromSnapshotInclusive = + Preconditions.checkArgumentNotNull( + table.snapshot(fromSnapshotInclusiveId), + "The specified starting snapshot %s does not exist", + fromSnapshotInclusiveId); + fromSnapshotInclusiveSeq = fromSnapshotInclusive.sequenceNumber(); + + boolean sameLineage = + toSnapshotId == null + ? SnapshotUtil.isAncestorOf(table, fromSnapshotInclusiveId) + : SnapshotUtil.isAncestorOf(table, toSnapshotId, fromSnapshotInclusiveId); + checkArgument( + sameLineage, + "Configured starting snapshot %s is not an ancestor of %s", + fromSnapshotInclusiveId, + toSnapshotId == null ? "the current table" : "end snapshot " + toSnapshotId); + } + + return new OffsetRange( + fromSnapshotInclusiveSeq, Math.max(fromSnapshotInclusiveSeq, toSnapshotExclusiveSeq)); + } + + @NewTracker + public RestrictionTracker<OffsetRange, Long> newTracker(@Restriction OffsetRange restriction) { + if (restriction.getTo() == POLL_FOREVER) { + return new GrowableOffsetRangeTracker( + restriction.getFrom(), this::estimateCurrentRangeEndExclusive); + } + + return new OffsetRangeTracker(restriction); + } + + private long estimateCurrentRangeEndExclusive() { + TableCache.setup(scanConfig); + Table table = TableCache.get(scanConfig.getTableIdentifier()); + + @Nullable Long toSnapshotId = ReadUtils.getToSnapshot(table, scanConfig); + if (toSnapshotId != null) { + @Nullable Snapshot toSnapshot = table.snapshot(toSnapshotId); + return toSnapshot == null ? Long.MIN_VALUE : toSnapshot.sequenceNumber() + 1; + } + + @Nullable Snapshot current = table.currentSnapshot(); + return current == null ? Long.MIN_VALUE : current.sequenceNumber() + 1; + } Review Comment: done, applied 1L instead because that's the actual starting sequence number -- 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]
