[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

[wenshao]

[Critical] 这里把 readIncrementalChanges(...) 调成了当前类方法，但实际定义在 scanner 上；同时 Collection<SimpleFileEntry> 缺少 java.util.Collection import。这两处都会导致新增冲突检测逻辑无法编译。

建议改为 scanner.readIncrementalChanges(...)，并补充 java.util.Collection import。

— gpt-5.4 via Qwen Code /review

[wenshao]

Request changes — one inline comment was posted successfully, and two findings could not be attached to diff lines so they are included below.

• [Critical] paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java:620,626
tryOverwritePartition 新增 baseSnapshotId 参数后，这两处内部调用仍沿用旧签名，当前参数个数不匹配，paimon-core 编译会直接失败。
  建议在 dropPartitions 和 truncateTable 里的调用末尾补传 null，保持非 sort compact 路径的现有行为，例如：tryOverwritePartition(..., new HashMap<>(), null);

• [Critical] paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java:343
SortCompactAction 已经设置了 withOverwriteBaseSnapshot(readSnapshotId)，但 unaware-bucket 路径这里没有继续把 overwriteBaseSnapshotId 传给最终 sink/committer。这样在 bucket = -1 的 Flink sort compact 场景下，新增冲突检测实际上不会生效。
  建议让 unaware-bucket 路径和 dynamic-bucket 路径一样透传 overwriteBaseSnapshotId，并补一个对应的 Flink 测试覆盖该场景。

Reviewed by gpt-5.4 via Qwen Code /review