amogh-jahagirdar commented on code in PR #5029:
URL: https://github.com/apache/iceberg/pull/5029#discussion_r1093893207
##########
flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/source/StreamingMonitorFunction.java:
##########
@@ -124,11 +126,33 @@ public void initializeState(FunctionInitializationContext
context) throws Except
if (context.isRestored()) {
LOG.info("Restoring state for the {}.", getClass().getSimpleName());
lastSnapshotId = lastSnapshotIdState.get().iterator().next();
+ } else if (scanContext.startTag() != null) {
+ Snapshot currentSnapshot =
+ scanContext.branch() == null
+ ? table.currentSnapshot()
+ : table.snapshot(scanContext.branch());
+ Preconditions.checkNotNull(
Review Comment:
I think it'll help if at some point (probably options/conf default) we set
the default branch to main and always assume a branch is set in this code. Then
we can just do table.snapshot(branch).
Nit on log message:
"No snapshots on branch %s in table %s"
##########
flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/source/StreamingMonitorFunction.java:
##########
@@ -124,11 +126,33 @@ public void initializeState(FunctionInitializationContext
context) throws Except
if (context.isRestored()) {
LOG.info("Restoring state for the {}.", getClass().getSimpleName());
lastSnapshotId = lastSnapshotIdState.get().iterator().next();
+ } else if (scanContext.startTag() != null) {
+ Snapshot currentSnapshot =
+ scanContext.branch() == null
+ ? table.currentSnapshot()
+ : table.snapshot(scanContext.branch());
+ Preconditions.checkNotNull(
+ currentSnapshot,
+ "Don't have any available snapshot for branch " +
scanContext.branch() + " in table.");
+
+ long currentSnapshotId = currentSnapshot.snapshotId();
+ long startSnapshotId =
table.snapshot(scanContext.startTag()).snapshotId();
+ Preconditions.checkState(
+ SnapshotUtil.isAncestorOf(table, currentSnapshotId, startSnapshotId),
+ "The option start-snapshot-id %s is not an ancestor of the current
snapshot.",
+ startSnapshotId);
+
+ lastSnapshotId = startSnapshotId;
} else if (scanContext.startSnapshotId() != null) {
+ Snapshot currentSnapshot =
+ scanContext.branch() == null
+ ? table.currentSnapshot()
+ : table.snapshot(scanContext.branch());
Preconditions.checkNotNull(
- table.currentSnapshot(), "Don't have any available snapshot in
table.");
+ currentSnapshot,
+ "Don't have any available snapshot for branch " +
scanContext.branch() + " in table.");
Review Comment:
Also sorry if I missed it, but at some point shouldn't there be validation
that the start tag is an ancestor of the end tag?
--
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: issues-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
