JingsongLi commented on code in PR #10625:
URL: https://github.com/apache/paimon/pull/10625#discussion_r4313658468
##########
paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java:
##########
@@ -217,62 +251,105 @@ private int innerExpireUntil(long earliestId, long
endExclusiveId)
// tags to create data file skipper
List<Snapshot> taggedSnapshots = collectTaggedSnapshots();
+ Runnable collectedTags = afterTagCollection;
+ if (collectedTags != null) {
+ collectedTags.run();
+ }
+ // Another JVM can create a tag without this process's lock. Mark
untagged snapshots, scan
+ // tags again, and keep any snapshot that published a tag before the
marker. Snapshot files
+ // are still deleted at the end, same as before.
+ List<Long> marked = new ArrayList<>();
+ try {
+ retainTagsCreatedDuringExpiration(snapshotsExcludingEnd,
taggedSnapshots, marked);
- // delete merge tree files
- // deleted merge tree files in a snapshot are not used by the next
snapshot, so the range of
- // id should be (beginInclusiveId, endExclusiveId]
- cleanDataFiles(snapshotsIncludingEnd, taggedSnapshots,
beginInclusiveId);
+ // delete merge tree files
+ // deleted merge tree files in a snapshot are not used by the next
snapshot, so the
+ // range of id should be (beginInclusiveId, endExclusiveId]
+ cleanDataFiles(snapshotsIncludingEnd, taggedSnapshots,
beginInclusiveId);
- // delete changelog files
- if (!expireConfig.isChangelogDecoupled()) {
- cleanChangelogFiles(snapshotsExcludingEnd);
- }
+ // delete changelog files
+ if (!expireConfig.isChangelogDecoupled()) {
+ cleanChangelogFiles(snapshotsExcludingEnd);
+ }
- // data files and changelog files in bucket directories has been
deleted
- // then delete changed bucket directories if they are empty
- snapshotDeletion.cleanEmptyDirectories();
+ // data files and changelog files in bucket directories has been
deleted
+ // then delete changed bucket directories if they are empty
+ snapshotDeletion.cleanEmptyDirectories();
- // delete manifests and indexFiles
- List<Snapshot> skippingSnapshots =
- findSkippingTags(taggedSnapshots, beginInclusiveId,
endExclusiveId);
+ // delete manifests and indexFiles
+ List<Snapshot> skippingSnapshots =
+ findSkippingTags(taggedSnapshots, beginInclusiveId,
endExclusiveId);
- Snapshot lastSnapshot =
snapshotsIncludingEnd.get(snapshotsIncludingEnd.size() - 1);
- if (lastSnapshot.id() != endExclusiveId) {
- // the end exclusive snapshot is gone
- // there is no need to proceed
- return 0;
- }
- skippingSnapshots.add(lastSnapshot);
+ Snapshot lastSnapshot =
snapshotsIncludingEnd.get(snapshotsIncludingEnd.size() - 1);
+ if (lastSnapshot.id() != endExclusiveId) {
+ // the end exclusive snapshot is gone
+ // there is no need to proceed
+ return 0;
+ }
+ skippingSnapshots.add(lastSnapshot);
- Set<String> skippingSet = null;
- try {
- Set<String> builtSkippingSet = ConcurrentHashMap.newKeySet();
-
builtSkippingSet.addAll(snapshotDeletion.manifestSkippingSet(skippingSnapshots));
- skippingSet = builtSkippingSet;
- } catch (Exception e) {
- LOG.info("Skip cleaning manifest files due to failed to build
skipping set.", e);
- }
- if (skippingSet != null) {
- snapshotDeletion.executeAll(
- collectManifestDeletionTasks(snapshotsExcludingEnd,
skippingSet));
- }
+ Set<String> skippingSet = null;
+ try {
+ Set<String> builtSkippingSet = ConcurrentHashMap.newKeySet();
+
builtSkippingSet.addAll(snapshotDeletion.manifestSkippingSet(skippingSnapshots));
+ skippingSet = builtSkippingSet;
+ } catch (Exception e) {
+ LOG.info("Skip cleaning manifest files due to failed to build
skipping set.", e);
+ }
+ if (skippingSet != null) {
+ snapshotDeletion.executeAll(
+ collectManifestDeletionTasks(snapshotsExcludingEnd,
skippingSet));
+ }
- // delete snapshot file finally
- for (Snapshot snapshot : snapshotsExcludingEnd) {
- if (expireConfig.isChangelogDecoupled()) {
- commitChangelog(new Changelog(snapshot));
+ // delete snapshot file finally
+ for (Snapshot snapshot : snapshotsExcludingEnd) {
+ if (expireConfig.isChangelogDecoupled()) {
+ commitChangelog(new Changelog(snapshot));
+ }
+ snapshotManager.deleteSnapshot(snapshot.id());
+ }
+
+ writeEarliestHint(endExclusiveId);
+ long duration = currentTimeMillis.get() - startTime;
+ LOG.info(
+ "Finished expire snapshots, duration {} ms, range is [{},
{})",
+ duration,
+ beginInclusiveId,
+ endExclusiveId);
+ return snapshotsExcludingEnd.size();
+ } finally {
+ for (Long snapshotId : marked) {
+ tagManager.deleteExpirationMarker(snapshotId);
}
- snapshotManager.deleteSnapshot(snapshot.id());
}
+ }
- writeEarliestHint(endExclusiveId);
- long duration = currentTimeMillis.get() - startTime;
- LOG.info(
- "Finished expire snapshots, duration {} ms, range is [{}, {})",
- duration,
- beginInclusiveId,
- endExclusiveId);
- return snapshotsExcludingEnd.size();
+ private void retainTagsCreatedDuringExpiration(
+ List<Snapshot> expiring, List<Snapshot> taggedSnapshots,
List<Long> marked)
+ throws ExecutionException, InterruptedException {
+ Set<Long> taggedIds = new HashSet<>();
+ for (Snapshot tagged : taggedSnapshots) {
+ taggedIds.add(tagged.id());
+ }
+ for (Snapshot snapshot : expiring) {
+ if (taggedIds.contains(snapshot.id())) {
+ tagManager.deleteExpirationMarker(snapshot.id());
Review Comment:
[P1] Preserve fences owned by other active expirations
This deletes a marker that may belong to another still-running expirer. I
reproduced this with three independent JDK 8 JVMs: A marks snapshot 2 and
pauses after its final tag scan; C publishes a tag for 2 and pauses before its
post-write check; B sees C's tag, deletes A's marker here, preserves the tag's
files, and successfully expires [1,3). C then sees no marker and surviving
manifests, so tag creation succeeds. A resumes with its earlier tag set and
deletes those manifests, leaving the successfully created tag unreadable. Both
expirations returned successfully, with no injected failure. Automatic writer
expiration can overlap an independent Spark/Flink expire_snapshots call. Please
serialize expirers across processes or give each run its own fence ownership,
keeping creators fenced until every expirer that missed the tag has finished.
The deletion at line 345 and cleanup of borrowed markers in finally also need
the same ownership guarantee.
##########
paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java:
##########
@@ -181,6 +317,15 @@ private void createOrReplaceTag(
e);
}
+ if (rollbackIfSnapshotExpired) {
+ try {
+ ensureSnapshotRetained(snapshot, tagName, true);
+ } catch (SnapshotNotExistException e) {
Review Comment:
[P1] Roll back a published tag when retention validation fails with I/O
Only SnapshotNotExistException removes the tag written at line 310. If the
new post-write marker/manifest existence check throws IOException,
ensureSnapshotRetained wraps it as UncheckedIOException and the published tag
survives without completing the retention handshake. I reproduced this with a
creator in a separate JVM started after the expirer’s final tag scan: inject
one transient IOException on exists(expire-marker/2), let expiration finish,
then retry createTag(..., ignoreIfExists=true). The first call fails but leaves
tagExists=true; expiration deletes snapshot 2's manifests; the retry returns
successfully because the tag already exists; reading that tag fails with a
missing manifest-list. Auto/batch tag creators use ignoreIfExists=true. Please
clean up this call's newly published tag on every failed post-write retention
validation (preserving the original error), so a retry cannot acknowledge an
unvalidated dangling 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
