Skip to content

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

Open
hbgstc123 wants to merge 11 commits into
apache:masterfrom
hbgstc123:fix_sort_compact_append_only
Open

[core] Refactor sort compact to produce COMPACT commits#8631
hbgstc123 wants to merge 11 commits into
apache:masterfrom
hbgstc123:fix_sort_compact_append_only

Conversation

@hbgstc123

@hbgstc123 hbgstc123 commented Jul 15, 2026 •
edited
Loading

Copy link
Copy Markdown

Purpose

Follow up on #7595 per this review comment: sort compact should not be modeled as an OVERWRITE commit with extra base-snapshot handling. Instead, it should produce a normal COMPACT commit so the existing commit protocol can perform compact validation and conflict detection.

This PR fixes the concurrent-write data loss risk in sort compact by refactoring Flink/Spark sort compact for append tables to:

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

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

This is a follow-up to #8571, scoped to append tables only per review feedback.

Additional hardening on top of the COMPACT-commit refactor:

  • Capture base-snapshot deletion-vector metadata in SortCompactPlanMetadata at planning time, so Flink commit recovery can still clean up DVs even if the base snapshot has expired.
  • At rewrite time, merge latest-snapshot DV metadata so concurrent DV writes are also cleaned up.
  • Align Flink commit failure behavior with Spark: abort newly written files when commit fails and the COMPACT snapshot is not yet visible.
  • Add sort-compaction.warn-input-files / sort-compaction.max-input-files to guard large sort compact plans whose input files are embedded in the Flink job graph.

Tests

Core

  • SortCompactCommitMessageRewriterTest
    • rewrite to compact messages
    • multi-bucket / cross-bucket rewrite
    • deletion-vector cleanup, including concurrent DV writes and expired base snapshots
    • captured plan metadata serialization / round trip
    • compact commit success detection, including expired snapshot gaps
    • inline compaction rejection

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

@hbgstc123 hbgstc123 mentioned this pull request Jul 15, 2026
Closed
@hbgstc123
hbgstc123 force-pushed the fix_sort_compact_append_only branch from c8a1b81 to de2e634 Compare July 15, 2026 08:04
hbg and others added 5 commits July 16, 2026 09:58
hbg
[core] Refactor sort compact to produce COMPACT commits
6d4da26
hbg @cursoragent
[core][flink] Harden sort compact DV cleanup and commit failure handling
b7154e6 
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>
hbg @cursoragent
[hotfix] Restore protected visibility for StoreSinkWriteImpl fields u…
4cec595 
…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>
hbg @cursoragent
[flink] Remove SortCompactActionForDynamicBucketITCase after sort com…
aa4f4bf 
…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>
hbg @cursoragent
[spark] Skip concurrent sort compact test for V2 DV append tables
d462567 
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>
@hbgstc123
hbgstc123 force-pushed the fix_sort_compact_append_only branch from de2e634 to d462567 Compare July 16, 2026 02:01
@JingsongLi

JingsongLi commented Jul 16, 2026

Copy link
Copy Markdown
Contributor

There is a data correctness issue with “Sort compact” and concurrent DV updates. Since “Sort compact” generates output based on an old snapshot but deletes the latest DV from the input file upon commit, this can cause deleted data to be restored.

@hbgstc123

hbgstc123 commented Jul 17, 2026

Copy link
Copy Markdown
Author

There is a data correctness issue with “Sort compact” and concurrent DV updates. Since “Sort compact” generates output based on an old snapshot but deletes the latest DV from the input file upon commit, this can cause deleted data to be restored.

Right, just pushed a fix

hbg and others added 3 commits July 17, 2026 14:02
hbg
Merge remote-tracking branch 'apache/master' into fix_sort_compact_ap…
dcd3ba7 
…pend_only
hbg
report correct error message in ConflictDetection
13ba744
hbg @cursoragent
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>
JingsongLi
JingsongLi reviewed Jul 23, 2026
View reviewed changes
Comment thread ...k/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SortCompactSinkBuilder.java Outdated
return this;
}

private void validateSortCompactInput(List<DataSplit> compactInputSplits) {

@JingsongLi JingsongLi Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Remove this method and two options. We don't need to restrict this.

@hbgstc123 hbgstc123 Jul 23, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

hbg and others added 2 commits July 24, 2026 00:49
hbg @cursoragent
[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>
hbg @cursoragent
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>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@JingsongLi JingsongLi JingsongLi left review comments

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@hbgstc123 @JingsongLi