heshshark opened a new issue, #19763:
URL: https://github.com/apache/hudi/issues/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`:
```java
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:
```properties
write.operation=insert
write.insert.cluster=true
write.precombine=false
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:
```properties
write.operation=upsert
write.precombine=false
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:
```text
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:
```text
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:
```text
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
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
