Repository navigation
Conversation
When the committable identifier equals that latest snapshot and the snapshot kind is APPEND or OVERWRITE, filterCommitted adds a separate committable that copies the watermark and properties and keeps only the non-empty compact increments, each with an empty data increment. CompactRecoveryCommittable marks that copy so commit writes no further append snapshot and skips overwrite, and the replay publishes one COMPACT snapshot even when an empty append would otherwise be written. Before replay, every file from the compact-after and changelog metas and every new index file is resolved through CompactedChangelogPathResolver and checked for existence; a missing output skips the whole identifier, and an I/O error from the existence check fails recovery as UncheckedIOException. Fixes apache#11071
| commitCallbacks.forEach(callback -> callback.retry(committable)); | ||
| Snapshot snapshot = latestSnapshot.get(); | ||
| if (committable.identifier() == snapshot.commitIdentifier() | ||
| && (snapshot.commitKind() == CommitKind.APPEND |
There was a problem hiding this comment.
Do not replay a compact increment whose inputs have been concurrently removed
This branch recovers the COMPACT whenever its output files exist, but that is not sufficient to make the increment replayable. A concurrent compaction may have already deleted one of its compactBefore files. In that case conflict detection rejects this stale delete; after restart filterCommitted reconstructs the same recovery committable again, so the job can never progress. The existing PrimaryKeyFileStoreTableITCase.testBatchJobWithConflictAndRestart demonstrates this in the failed Flink 2 CI job and in the local run: all 10 restart attempts end in File deletion conflicts detected.
Please validate the compact inputs against the current snapshot and discard/rebase an invalid recovery (with coverage for this concurrent-compaction case) instead of unconditionally replaying it based only on output existence + adding a test case might help.
There was a problem hiding this comment.
Good catch, thanks. 49ef4c4 adds a check in recoverCompact: before replaying a recovered COMPACT, compactInputsStillLive verifies that every compactBefore file is still an ADD entry in the latest snapshot, and filterCommitted drops the recovery when one was removed by a concurrent compaction. FileStoreCommitTest gets testFilterHalfCommittedIdentifierWithConcurrentlyCompactedInput for that case.
A concurrent compaction may delete a compact-before file of the recovered increment. Replaying it would fail conflict detection on every restart, so filterCommitted now discards the recovery unless all its inputs are still live in the latest snapshot.
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Requirement fit: SUPPORTED. Implementation: FINDINGS.
The partial-commit recovery problem is real. On this head, 43 focused existing core tests passed; two added deterministic recovery regressions failed. The existing Flink 2 / JDK 17 PrimaryKeyFileStoreTableITCase#testBatchJobWithConflictAndRestart passed. Two stale-input cases below still prevent recovery.
Verification limit: the full FileStoreCommitTest run was stopped while its pre-existing random multi-writer stress test was still running; no full-suite pass is claimed. The current CI run was cancelled and its aggregate CI-result failed, so that status alone is not attributed to the patch.
| for (CommitMessage message : committable.fileCommittables()) { | ||
| for (DataFileMeta file : | ||
| ((CommitMessageImpl) message).compactIncrement().compactBefore()) { | ||
| inputs.add(Arrays.asList(message.partition(), message.bucket(), file.fileName())); |
There was a problem hiding this comment.
[P1] Compare complete manifest identities when checking live compact inputs
A concurrent compaction can upgrade a file from level 0 to level 1 without changing its physical filename (MergeTreeCompactTask.upgrade / AbstractCompactRewriter.upgrade). This tuple still considers the old input live, but conflict detection uses FileEntry.Identifier, including level and auxiliary metadata. Replaying the old level-0 deletion then fails with File deletion conflicts detected. The latest snapshot still contains the same filename, so every restart reconstructs the same invalid recovery instead of skipping it.
I reproduced this on 49ef4c4: publish the APPEND phase of identifier 2, commit DELETE(old level 0)/ADD(same file upgraded to level 1) as another user, then restore identifier 2. The new regression testFilterHalfCommittedIdentifierWithConcurrentlyUpgradedInput confirms the replay conflicts and fails the expected-empty recovery assertion. Please compare the full manifest identity (or explicitly rebase), rather than just partition/bucket/name.
There was a problem hiding this comment.
Agreed, matching on partition/bucket/name misses the in-place level upgrade. I'll change compactInputsStillLive to compare the full FileEntry.Identifier (level and extra metadata included) and add your testFilterHalfCommittedIdentifierWithConcurrentlyUpgradedInput regression to the PR.
There was a problem hiding this comment.
Done in 4820c4a. compactInputsStillLive now builds the full FileEntry.Identifier for each compact-before file and checks it against the identifiers of the live ADD entries, so a file that a concurrent compaction upgraded in place no longer counts as live and the stale recovery is skipped. Your testFilterHalfCommittedIdentifierWithConcurrentlyUpgradedInput regression is in FileStoreCommitTest and passes along with the other testFilterHalfCommitted cases.
| partitions.add(message.partition()); | ||
| } | ||
| } | ||
| if (inputs.isEmpty()) { |
There was a problem hiding this comment.
[P1] Validate pending index deletions before accepting recovery
Compact increments can also contain deletedIndexFiles, including an index-only message produced by DataEvolutionCompactGlobalIndexDropper. If a concurrent DROP GLOBAL INDEX removes that index after the APPEND phase, this check still succeeds: the index-only message has no compactBefore, and index liveness is never checked. GlobalIndexCombiner then rejects the stale deletion, while each subsequent restart reconstructs it again.
The added testRecoveryAfterConcurrentGlobalIndexDrop reproduces two fresh writer recoveries on this head; both explicitly assert the root cause IllegalStateException: Trying to delete global index file recovery-index which does not exist. Spark/Flink DROP GLOBAL INDEX perform the same metadata-only removal. Please validate the compact increment's deleted index identities against the current index manifest as well, and discard/rebase an invalid replay.
There was a problem hiding this comment.
Good catch. In 6cd7bd9, recovery now also checks every deleted index file in the compact increment against the latest index manifest (same partition, bucket and file meta) and discards the replay when one is missing or changed, so a concurrent DROP GLOBAL INDEX no longer makes each restart fail in the combiner. The new FileStoreCommitTest case replays an index-only message after the index was dropped by another commit.
A concurrent compaction can upgrade a compact-before file to a higher level without renaming it. Matching inputs only on partition, bucket and file name treated such a file as still live, but conflict detection uses the full FileEntry.Identifier, so the replayed level 0 deletion failed on every restart. compactInputsStillLive now compares the full identifier, and a regression test covers the in-place upgrade.
A compact increment can delete index files without any compact-before file, for example the index-only message written when dropping a global index. If a concurrent DROP GLOBAL INDEX removed that index after the first phase, the replayed deletion failed in the index manifest combiner on every restart. Recovery now also checks each deleted index file against the latest index manifest and discards the increment when one is missing or changed.
Purpose
If a job stops after one commit identifier has published its first snapshot and before the compaction snapshot for that same identifier, restore drops the compaction. The next compaction then deletes an output file that is absent from the table, conflict detection aborts the commit, and the job restarts. Each table left in that window costs another restart, and the changelog the dropped compaction would have published is missing. One commit identifier is published as two snapshots, first APPEND or OVERWRITE and then COMPACT.
On restore a committable is kept only when its identifier is greater than the latest snapshot identifier for this commit user, so an identifier whose first snapshot already exists is dropped together with its still unpublished compact increments. The writer of the next identifier already had those compaction outputs in memory and emits deletes for them, which conflict detection rejects because the files were never added.
When the committable identifier equals that latest snapshot and the snapshot kind is APPEND or OVERWRITE, filterCommitted adds a separate committable that copies the watermark and properties and keeps only the non-empty compact increments, each with an empty data increment. CompactRecoveryCommittable marks that copy so commit writes no further append snapshot and skips overwrite, and the replay publishes one COMPACT snapshot even when an empty append would otherwise be written.
Before replay, every file from the compact-after and changelog metas and every new index file is resolved through CompactedChangelogPathResolver and checked for existence; a missing output skips the whole identifier, and an I/O error from the existence check fails recovery as UncheckedIOException.
Fixes #11071
Tests
mvn -B -ntp -pl paimon-core -am test -Dtest=FileStoreCommitTest,TableCommitTest -Dsurefire.failIfNoSpecifiedTests=false finished TableCommitTest with 30 tests, 0 failures, 0 errors, and 0 skipped, then timed out at 300s.
No test was run with the paimon FileStoreCommitImpl change reverted, because no test was named as covering the reported bug, so none is shown to catch this bug.
The changed test paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java did not run here: no test runner for it was available.
The changed test paimon-core/src/test/java/org/apache/paimon/table/sink/TableCommitTest.java did not run here: no test runner for it was available.