lrsb commented on code in PR #16011:
URL: https://github.com/apache/iceberg/pull/16011#discussion_r3264650252
##########
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:
That wouldn't help, JobOperatorKey contains the job uuid and therefore
doesn't correlate with commit order.
Alternatives are:
- Re-keying by firstCheckpointId into a NavigableMap<Long, ...>: equivalent
to the current sort, just less obvious, and forces re-keying as new items are
inserted into a group.
- Restructuring to NavigableMap<Long, Map<JobOperatorKey, ...>> keyed by
checkpoint id at the top: breaks the per-group dedup batching (one commit per
group instead of per checkpoint × group)
--
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]
