pvary commented on code in PR #6299:
URL: https://github.com/apache/iceberg/pull/6299#discussion_r1035587241
##########
flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java:
##########
@@ -78,6 +79,22 @@ public ContinuousEnumerationResult
planSplits(IcebergEnumeratorPosition lastPosi
}
}
+ private Snapshot toSnapshotInclusive(
+ Long lastConsumedSnapshotId, Snapshot currentSnapshot, int
maxPlanningSnapshotCount) {
+ // snapshots are in reverse order (latest snapshot first)
+ List<Snapshot> snapshots =
+ Lists.newArrayList(
+ SnapshotUtil.ancestorsBetween(
+ table, currentSnapshot.snapshotId(), lastConsumedSnapshotId));
+ if (snapshots.size() <= maxPlanningSnapshotCount) {
+ return currentSnapshot;
+ } else {
+ // Because snapshots are in reverse order of commit history, this index
returns
+ // the max allowed number of snapshots from the lastConsumedSnapshotId.
+ return snapshots.get(snapshots.size() - maxPlanningSnapshotCount);
+ }
+ }
+
/** Discover incremental changes between {@code lastPosition} and current
table snapshot */
Review Comment:
Nit: Maybe update the comment?
--
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
