[flink][spark] Wire sort compact to produce COMPACT commits by hbgstc123 · Pull Request #9790 · apache/paimon

hbgstc123 · 2026-09-13T12:58:56Z

Purpose

Stacked follow-up to #8631. Please review this PR after (or together with) the core rewriter.

Until #8631 merges, GitHub's Files changed view includes the core commits. Engine-only diff:
hbgstc123/paimon@fix_sort_compact_append_only...fix_sort_compact_append_engines

This PR wires Flink and Spark sort compact for append tables to the core rewriter:

Write sorted output in write-only mode.
Rewrite the written append commit messages into CompactIncrement messages.
Commit them through the normal compact commit path.

Additional engine hardening:

Flink SortCompactCommitter merge/rewrite/abort and commit recovery.
Align Flink commit failure abort with Spark: abort newly written files when commit fails and the COMPACT snapshot is not yet visible.
Remove SortCompactActionForDynamicBucketITCase because sort compact is append-table only.

No new overwritePartition(baseSnapshotId) APIs or special overwrite conflict logic are introduced.

Tests

Flink

SortCompactCommitterTest
- merge partial committables before rewrite
- delete-only compact when all input rows are filtered out
- commit recovery does not delete planned input or create DV index files
- commit failure aborts written messages only when COMPACT snapshot is not yet visible
SortCompactActionForAppendTableITCase
- latest snapshot kind is COMPACT
- concurrent append between read and commit does not lose data
CommitterOperatorTest
CompactProcedureITCase updated to expect COMPACT

Spark

SortCompactSparkCommitTest
- commit failure aborts written messages only when COMPACT snapshot is not yet visible
CompactProcedureTestBase updated to expect COMPACT

Merge latest-snapshot deletion-vector metadata at rewrite time so concurrent DV writes are cleaned up, while keeping captured plan metadata for expired snapshots. Align Flink commit failure behavior with Spark by aborting write output when the COMPACT snapshot is not yet visible. Co-authored-by: Cursor <cursoragent@cursor.com>

…sed by subclasses Commit 0f5aecf changed commitUser, state, and write from protected to private, but GlobalFullCompactionSinkWrite and LookupSinkWrite still access them directly. This causes JDK 8 compilation failures. Restore protected visibility to fix the build. Co-authored-by: Cursor <cursoragent@cursor.com>

…pact restricts to append tables SortCompactAction now explicitly rejects primary-key tables and only supports bucket-unaware append tables. The dynamic-bucket test case exercises unsupported behavior and fails on Flink 2.x CI, so remove it. Co-authored-by: Cursor <cursoragent@cursor.com>

Sort compact now commits as COMPACT instead of OVERWRITE. Concurrent merge and sort compact on V2 delta row-level DV append tables can leave duplicate visible rows, so skip that combination and keep coverage on the V1 write path. Co-authored-by: Cursor <cursoragent@cursor.com>

…pend_only

Resolve conflicts in FlinkSink and FlinkSinkBuilder by keeping both sort-compact writeProviderOverride/createAppendTableSink hooks and master's BLOB descriptor reader factory wiring. Co-authored-by: Cursor <cursoragent@cursor.com>

Drop validateSortCompactInput and the warn/max input file options per review feedback. Co-authored-by: Cursor <cursoragent@cursor.com>

…ct_append_only Resolve conflict in BaseAppendDeleteFileMaintainer by keeping upstream javadoc. Co-authored-by: Cursor <cursoragent@cursor.com>

Co-authored-by: Cursor <cursoragent@cursor.com>

…append_only

After making DataFileMeta#assignFileSource abstract, BinaryDataFileMeta from master must provide the method explicitly.

… compact Matching success by removed input files can skip abort after a failed delete-only commit when another compact already deleted the same inputs, leaking the DV index files produced by rewrite.

# Conflicts: # paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java # paimon-core/src/main/java/org/apache/paimon/io/ProjectedDataFileMeta.java # paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/SortCompactAction.java # paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java # paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java # paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java # paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala

…rt_compact_append_only

… refactors from master

…ct_append_only

Move Flink/Spark wiring to a follow-up PR so the COMPACT-commit protocol can be reviewed independently.

Use the core rewriter from the stacked core PR so Flink and Spark sort compact produce COMPACT snapshots instead of OVERWRITE.

JingsongLi

Reviewed 09b2a56. Requirement fit: SUPPORTED. Implementation: FINDINGS.

Preserving concurrent appends by publishing sort compaction as COMPACT has clear value, and this is the engine wiring needed to make the core rewriter useful. There is a recovery regression in the new Flink abort path: one transient failure before publishing the snapshot can make committer-only batch recovery fail permanently. Details are inline.

I reproduced this with the exact changed committer/rewriter sources and real Paimon metadata/files: replaying the same committable after one injected failure reports the sorted output as deleted; a control retaining the output successfully publishes snapshot 2. This was a focused committer replay probe, not a full Flink cluster restart. The current head also has a failing Flink 2 CI lane.

JingsongLi · 2026-09-14T02:12:11Z

+        // writtenMessages is empty but compactMessages still carries the new DV index files, so it
+        // must be aborted (the previous writtenMessages.isEmpty() early return skipped cleanup
+        // entirely in that case).
+        abortWrittenQuietly(writtenMessages, cause);


[P1] Keep writer output available for Flink batch commit replay

These two abort calls delete the sorted data files when a commit fails before the COMPACT snapshot is visible. Flink batch recovery can restart only the committer and replay the same upstream committables; CommitterOperator#commitUpToCheckpoint explicitly documents and implements that contract. After one transient pre-snapshot failure, the replay therefore references files deleted here, and filterAndCommitMultiple fails with "Cannot recover ... files ... have been deleted" on every retry. I reproduced this with the same commit user, captured plan and Long.MAX_VALUE committable; suppressing the abort calls alone lets the identical replay succeed.

Preserve writer output while these committables can be replayed. Cleanup of newly generated DV files must be separated from cleanup of compactAfter/written files, or recovery must explicitly regenerate the writer output. Please add a failure-then-committer-only-replay regression, rather than only asserting that failed commits remove files.

Thanks, agreed.

The pre-snapshot abort path was deleting writer output (compactAfter is the same files). Flink batch recovery can restart only the committer and replay the same committables (CommitterOperator#commitUpToCheckpoint), so one transient failure made every retry fail with Cannot recover ... files ... have been deleted.

Fix in 30fa106:

Keep writer output / compactAfter after a failed commit so the same committable can be replayed.

Clean only the new DV index files produced by rewrite (abortNewIndexFiles); they are regenerated on the next rewrite.

Add testCommitFailureThenCommitterOnlyReplaySucceeds: fail filterAndCommit once, then replay the same Long.MAX_VALUE committable with the same commit user.

Spark is unchanged: a Spark procedure re-runs writers, so aborting written files there is still correct.

Flink batch recovery can restart only the committer and replay the same committables. Aborting compactAfter on a pre-snapshot failure deleted those files and made every retry fail.

hbg and others added 24 commits July 16, 2026 09:58

[core] Refactor sort compact to produce COMPACT commits

6d4da26

address comment, detect concurrent dv conflict

92e4afa

Merge remote-tracking branch 'apache/master' into fix_sort_compact_ap…

dcd3ba7

…pend_only

report correct error message in ConflictDetection

13ba744

Merge branch 'master' into fix_sort_compact_append_only

66235f7

Resolve conflicts in FlinkSink and FlinkSinkBuilder by keeping both sort-compact writeProviderOverride/createAppendTableSink hooks and master's BLOB descriptor reader factory wiring. Co-authored-by: Cursor <cursoragent@cursor.com>

[flink] Remove sort compact input file count limits

626bb3f

Drop validateSortCompactInput and the warn/max input file options per review feedback. Co-authored-by: Cursor <cursoragent@cursor.com>

Merge branch 'master' of github.com:apache/paimon into fix_sort_compa…

0ab8ac0

…ct_append_only Resolve conflict in BaseAppendDeleteFileMaintainer by keeping upstream javadoc. Co-authored-by: Cursor <cursoragent@cursor.com>

delete new dv index files when error thrown in rewrite.

1423d4f

Retrigger CI for Flink 1.x Others workflow.

ffc40c2

Co-authored-by: Cursor <cursoragent@cursor.com>

address comment, make DataFileMeta#assignFileSource abstract

5a39d9c

Merge remote-tracking branch 'upstream/master' into fix_sort_compact_…

0231843

…append_only

[core] Implement assignFileSource for BinaryDataFileMeta

5830cab

After making DataFileMeta#assignFileSource abstract, BinaryDataFileMeta from master must provide the method explicitly.

Trigger CI re-run.

604c6ab

[core] Fix delete-only sort compact success detection with concurrent…

76878c4

… compact Matching success by removed input files can skip abort after a failed delete-only commit when another compact already deleted the same inputs, leaking the DV index files produced by rewrite.

Merge remote-tracking branch 'refs/remotes/apache/master' into fix_so…

68231c2

…rt_compact_append_only

[core] Fix sort compact tests for SchemaManager and ConflictDetection…

68290a5

… refactors from master

Merge branch 'master' of github.com:apache/paimon into fix_sort_compa…

51b6fe3

…ct_append_only

[core] Split sort compact PR: keep core rewriter only

5b5dc50

Move Flink/Spark wiring to a follow-up PR so the COMPACT-commit protocol can be reviewed independently.

[flink][spark] Wire sort compact to COMPACT commits

09b2a56

Use the core rewriter from the stacked core PR so Flink and Spark sort compact produce COMPACT snapshots instead of OVERWRITE.

hbgstc123 mentioned this pull request Sep 13, 2026

[core] Refactor sort compact to produce COMPACT commits #8631

Open

JingsongLi reviewed Sep 14, 2026

View reviewed changes

hbg added 2 commits September 20, 2026 17:01

[flink] Preserve sort compact writer output for committer-only replay

30fa106

Flink batch recovery can restart only the committer and replay the same committables. Aborting compactAfter on a pre-snapshot failure deleted those files and made every retry fail.

[core] Fix spotless formatting in sort compact rewriter test

8762fb6

Navigation Menu

[flink][spark] Wire sort compact to produce COMPACT commits - #9790

[flink][spark] Wire sort compact to produce COMPACT commits#9790
hbgstc123 wants to merge 26 commits into
apache:masterfrom
hbgstc123:fix_sort_compact_append_engines

hbgstc123 commented Sep 13, 2026

Uh oh!

JingsongLi left a comment

Uh oh!

JingsongLi Sep 14, 2026

Uh oh!

hbgstc123 Sep 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

hbgstc123 commented Sep 13, 2026

Purpose

Tests

Uh oh!

JingsongLi left a comment

Choose a reason for hiding this comment

Uh oh!

JingsongLi Sep 14, 2026

Choose a reason for hiding this comment

Uh oh!

hbgstc123 Sep 20, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants