[BUG] Flink COW insert/upsert can hide existing base-file records after unaligned-checkpoint recovery #19763

Bug Description

What happened:

We can reproducibly trigger silent data loss in the Flink COW write path after restoring from an unaligned checkpoint. The problem is not limited to insert clustering: we reproduced it with write.operation=insert and write.insert.cluster=true, and independently observed the same physical overwrite pattern with write.operation=upsert.

An in-flight record tagged as INSERT (instantTime = "I") is restored from checkpoint channel state and reaches the writer before UPDATE records for the same partitionPath + fileId. The writer bucket therefore starts with an INSERT record even though the file group already has a committed base file.

The write path selects FlinkCreateHandle for the existing fileId instead of a merge/concat handle. A new parquet file is written with the same fileId but only the small set of records from the current batch. Snapshot reads select this newer base-file version and hide the records that remain physically present in the older parquet file.

There is no exception and the commit succeeds.

Observed evidence from a deterministic reproduction:

• Existing file group at instant A contained 82,182 records.
• After recovery, the next writer bucket contained 2,028 records:
  • 57 records tagged INSERT
  • 1,971 records tagged UPDATE
  • first record type was INSERT
• Hudi selected FlinkCreateHandle.
• No previous base-file path was opened by the handle.
• The next commit metadata reported prevCommit = null and numWrites = numInserts.
• The new parquet reused the existing fileId but did not carry forward the old base-file records.
• Snapshot row count decreased although numDeletes = 0 and there were no write errors.

The same pattern was independently observed on a COW table configured with write.operation=upsert. Three existing file groups were replaced as follows:

• 408 previous rows -> 3 rows in the new base file
• 569 previous rows -> 2 rows in the new base file
• 448 previous rows -> 3 rows in the new base file

For all three new versions, commit metadata reported prevCommit = null, numWrites = numInserts, numUpdates = 0, and no deletes or write errors. Direct physical parquet reads confirmed that the old files still contained the hidden records.

Therefore, write.insert.cluster increases exposure in the insert path but is not a necessary condition for the underlying handle-selection failure.

What you expected:

Restoring records from a valid Flink checkpoint must not cause an existing Hudi file group to be treated as a new file handle in either INSERT or UPSERT mode.

For a fileId that already has a committed base file, Hudi should either:

1. select the merge/concat handle and preserve the existing base-file records; or
2. fail fast when a writer bucket contains inconsistent INSERT/UPDATE tags for the same existing file group.

A successful commit with zero deletes must not reduce the visible records of an existing file group.

Steps to reproduce:

Scenario A — deterministic insert-path reproduction:

1. Create a COPY_ON_WRITE table with a primary key.
2. Configure:
   • write.operation = insert
   • write.insert.cluster = true
   • unaligned checkpoints enabled
3. Write enough data to create and commit a base file for a fileId.
4. Arrange for an INSERT-tagged record for that fileId to remain in channel state when an unaligned checkpoint completes.
5. Restart the job from that checkpoint.
6. After recovery, send UPDATE-tagged records routed to the same partitionPath + fileId, with the restored INSERT record arriving first.
7. Let the next Hudi instant commit.
8. Compare the old/new parquet files, commit metadata, and snapshot row count.

Scenario B — upsert path:

1. Create a COPY_ON_WRITE table with a primary key and write.operation = upsert.
2. Enable unaligned checkpoints; write.insert.cluster is not required.
3. Create and commit existing file groups.
4. Restore from a checkpoint that contains in-flight INSERT-tagged records.
5. Allow restored INSERT-tagged and current UPDATE-tagged records for the same fileId to enter the next writer bucket.
6. Inspect the next commit for an existing fileId with prevCommit = null and numWrites = numInserts.
7. Physically compare the old and new parquet files for that fileId.

The issue becomes easier to reproduce with Flink mini-batch enabled because more records are released together after recovery.

Likely code path:

In Hudi 0.15.x, StreamWriteFunction groups records by partitionPath + fileId. Handle selection ultimately depends on the first record passed to FlinkWriteHandleFactory:

if (loc.getInstantTime().equals("I")) {
  writeHandle = new FlinkCreateHandle<>(...);
} else {
  writeHandle = createMergeHandle(...);
}

This makes handle selection dependent on restored record ordering instead of whether the file group already has a committed base file. The condition is shared by the COW commit write path used by both insert and upsert.

Workaround:

Disabling unaligned checkpoints prevents the reproduction in our tests:

execution.checkpointing.unaligned.enabled = false

This is only a workaround. Unaligned checkpoints are a supported Flink exactly-once feature, and Hudi should safely handle restored in-flight records.

Environment

Hudi version: 0.15.0

Query engine: Flink 1.18.1 / Flink SQL

Table type: COPY_ON_WRITE

Storage: S3-compatible object storage

Write mode: Streaming

Reproduced/observed write configurations:

Insert path:

write.operation=insert
write.insert.cluster=true
write.precombine=true
execution.checkpointing.unaligned.enabled=true
table.exec.mini-batch.enabled=true
table.exec.mini-batch.size=5000
table.exec.mini-batch.allow-latency=5s

Upsert path:

write.operation=upsert
write.precombine=true
execution.checkpointing.unaligned.enabled=true

The mini-batch settings and write.insert.cluster increase reproduction probability in the insert scenario, but neither is required for the upsert observation.

Logs and Stack Trace

No exception is thrown. The job and Hudi commit both complete successfully.

Deterministic insert-path diagnostic sequence:

instant A:
  fileId=<same-file-id>
  baseFileRows=82182

restored writer bucket for instant B:
  records=2028
  insertTagged=57
  updateTagged=1971
  firstRecordType=I

handle selection:
  handleClass=FlinkCreateHandle
  previousWritePath=NONE

commit metadata for instant B:
  fileId=<same-file-id>
  prevCommit=null
  numWrites=2028
  numInserts=2028
  numUpdates=0
  numDeletes=0
  totalWriteErrors=0

Independent upsert-path observations:

file group A: previousWrites=408, currentWrites=3, currentPrevCommit=null
file group B: previousWrites=569, currentWrites=2, currentPrevCommit=null
file group C: previousWrites=448, currentWrites=3, currentPrevCommit=null

all current versions:
  numWrites=numInserts
  numUpdates=0
  numDeletes=0
  totalWriteErrors=0

Physical validation:

old parquet: same fileId, hidden records are present
new parquet: same fileId, hidden records are absent
snapshot: selects new parquet, so old records are no longer visible