[core] improve insert overwrite conflict detection

[hbgstc123]

Purpose

Problem

Sort Compact reads data from snapshot S_read, sorts it, then commits as OVERWRITE. However, tryOverwritePartition builds the DELETE list from latestSnapshot (which may be
S_read+N). If concurrent writes add new files between S_read and latestSnapshot, those files are deleted without their data being included in the sorted output, causing silent
data loss.

Root Cause

Timeline:

1. SortCompact reads from snapshot S0 (files A, B, C)
2. Concurrent writer adds file D -> S1
3. SortCompact sorts A+B+C -> produces X, Y
4. SortCompact calls overwritePartition(new files X, Y)
5. Commit builds DELETE list from latestSnapshot = S1 (contains A, B, C, D)
6. Commit issues: DELETE A, B, C, D + ADD X, Y
7. File D is deleted but X, Y don't contain D's data -> DATA LOSS

Solution

Part 1: Pin DELETE list to base snapshot

Build the DELETE list from the snapshot the sort compact actually read from (base snapshot) instead of latestSnapshot. This ensures we only delete files that were included in the
sorted output.

Part 2: Concurrent write detection

When the base snapshot differs from the latest snapshot, use readIncrementalChanges(baseSnapshot, latestSnapshot) to detect new files added to the overwritten partitions. If new
files exist, fail the commit with a clear conflict error instead of silently losing data.

Changes

Core

• FileStoreCommit.java: Add @Nullable Long baseSnapshotId parameter to overwritePartition()
• FileStoreCommitImpl.java:
  • Modify tryOverwritePartition() to build DELETE list from baseSnapshot
  • Add concurrent write detection that fails fast if new files were added
• InnerTableCommit.java: Add withOverwriteBaseSnapshot(@Nullable Long snapshotId) method
• TableCommitImpl.java: Implement withOverwriteBaseSnapshot() and pass baseSnapshotId to commit

Flink

• SortCompactAction.java: Capture read snapshot ID before building source, pass to sink builder
• FlinkSinkBuilder.java: Add overwriteBaseSnapshotId field and setter
• FlinkWriteSink.java: Add overwriteBaseSnapshotId field and setter, pass to committer

Spark

• CompactProcedure.java: Capture read snapshot ID for sort compact operations
• PaimonSparkWriter.scala: Add withOverwriteBaseSnapshot() method

Backward Compatibility

When baseSnapshotId == null, tryOverwritePartition falls back to current behavior (latestSnapshot), so existing code paths (SQL INSERT OVERWRITE, DROP PARTITION, TRUNCATE,
etc.) are unaffected.

The concurrent write detection only activates when:

1. baseSnapshotId != null (explicitly set by sort compact)
2. There are actual new files added between base and latest snapshot

Data Flow After Fix

1. SortCompact reads snapshot S0 (files A,B,C), records S0.id
2. Concurrent writer adds file D -> S1
3. SortCompact sorts A+B+C -> produce X,Y
4. SortCompact calls overwritePartition(files X,Y, baseSnapshotId=S0)
5. Commit reads S0 (base snapshot), issues: DELETE A,B,C + ADD X,Y
6. Commit reads S1 (latest snapshot), detects D was added between S0 and S1
7. Commit FAILS with clear error: "concurrent write detected"
8. User retries sort compact safely (D will be included in next read)

Tests