pvary commented on code in PR #16011:
URL: https://github.com/apache/iceberg/pull/16011#discussion_r3180743305
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -119,43 +131,58 @@ public void
commit(Collection<CommitRequest<DynamicCommittable>> commitRequests)
DynamicWriteResultAggregator. Iceberg 1.12 will remove this, and users
should upgrade to the 1.11 release first
to migrate their state to a single commit request per checkpoint.
*/
- Map<TableKey, NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>>
commitRequestMap =
- Maps.newHashMap();
+ Map<TableKey, Map<JobOperatorKey, NavigableMap<Long,
List<CommitRequest<DynamicCommittable>>>>>
+ commitRequestMap = Maps.newHashMap();
for (CommitRequest<DynamicCommittable> request : commitRequests) {
- NavigableMap<Long, List<CommitRequest<DynamicCommittable>>> committables
=
- commitRequestMap.computeIfAbsent(
- new TableKey(request.getCommittable()), unused ->
Maps.newTreeMap());
- committables
- .computeIfAbsent(request.getCommittable().checkpointId(), unused ->
Lists.newArrayList())
+ DynamicCommittable committable = request.getCommittable();
+ commitRequestMap
+ .computeIfAbsent(committable.key(), unused -> Maps.newHashMap())
+ .computeIfAbsent(new JobOperatorKey(committable), unused ->
Maps.newTreeMap())
+ .computeIfAbsent(committable.checkpointId(), unused ->
Lists.newArrayList())
.add(request);
}
- for (Map.Entry<TableKey, NavigableMap<Long,
List<CommitRequest<DynamicCommittable>>>> entry :
- commitRequestMap.entrySet()) {
- Table table =
catalog.loadTable(TableIdentifier.parse(entry.getKey().tableName()));
- DynamicCommittable last =
entry.getValue().lastEntry().getValue().get(0).getCommittable();
- Snapshot latestSnapshot = table.snapshot(entry.getKey().branch());
+ for (Map.Entry<
+ TableKey,
+ Map<JobOperatorKey, NavigableMap<Long,
List<CommitRequest<DynamicCommittable>>>>>
+ tableEntry : commitRequestMap.entrySet()) {
+ TableKey tableKey = tableEntry.getKey();
+ Table table =
catalog.loadTable(TableIdentifier.parse(tableKey.tableName()));
+ Snapshot latestSnapshot = table.snapshot(tableKey.branch());
Iterable<Snapshot> ancestors =
latestSnapshot != null
? SnapshotUtil.ancestorsOf(latestSnapshot.snapshotId(),
table::snapshot)
: List.of();
- long maxCommittedCheckpointId =
- getMaxCommittedCheckpointId(ancestors, last.jobId(),
last.operatorId());
-
- NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>
skippedCommitRequests =
- entry.getValue().headMap(maxCommittedCheckpointId, true);
- LOG.debug(
- "Skipping {} commit requests: {}", skippedCommitRequests.size(),
skippedCommitRequests);
- // Mark the already committed FilesCommittable(s) as finished
- skippedCommitRequests
- .values()
- .forEach(list ->
list.forEach(CommitRequest::signalAlreadyCommitted));
-
- NavigableMap<Long, List<CommitRequest<DynamicCommittable>>> uncommitted =
- entry.getValue().tailMap(maxCommittedCheckpointId, false);
- if (!uncommitted.isEmpty()) {
- commitPendingRequests(
- table, entry.getKey().branch(), uncommitted, last.jobId(),
last.operatorId());
+
+ List<Map.Entry<JobOperatorKey, NavigableMap<Long,
List<CommitRequest<DynamicCommittable>>>>>
+ jobEntries = Lists.newArrayList(tableEntry.getValue().entrySet());
+ // Preserve checkpoint order across groups so that older-jobId commits
land before newer-jobId
+ // ones when the batch mixes committables from different jobIds (e.g.
state replay after a
+ // restart). Within a (jobId, operatorId) group, checkpoint order is
already guaranteed by
+ // the inner NavigableMap.
+ jobEntries.sort(Comparator.comparingLong(entry ->
entry.getValue().firstKey()));
Review Comment:
Shouldn't we just use NavigableMap for the map keyed by the `JobOperatorKey`?
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
