cshuo opened a new issue, #20287:
URL: https://github.com/apache/hudi/issues/20287
### Bug Description
With Flink upsert, `EVENT_TIME_ORDERING`, and `write.precombine=true`,
deduplication can turn an event-time delete into a natural-order delete. A
stale delete then removes a newer persisted record. The write can complete
successfully without record-level or global WriteStatus errors.
**Steps to reproduce:**
1. Use `ordering.fields=ts` and `write.precombine=true` with a COW or MOR
table.
2. In checkpoint 1, write `id1` and `id2` in `par1`, both with `ts=100`.
3. In checkpoint 2, put these records in the same flushed bucket:
| Record key | Operations in arrival order |
| --- | --- |
| `id1` | INSERT at `ts=50`, DELETE at `ts=60` |
| `id2` | INSERT at `ts=50`, DELETE at `ts=110` |
4. Complete checkpoint 2 and read the table snapshot.
**Expected:** Retain the complete `id1` row from checkpoint 1 and delete
`id2`.
**Actual before the fix:** The snapshot is empty for both COW and MOR. The
same workload with precombine disabled produces the expected result.
A helper-level reproduction also shows that INSERT(50), DELETE(60),
INSERT(55) can retain the obsolete INSERT(55), rather than DELETE(60).
### Root cause
[`BaseWriteHelper.reduceRecords`](https://github.com/apache/hudi/blob/3aa7fe2a9f70ef52064c977d28d28b4f4c95d00b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseWriteHelper.java#L147-L162)
reconstructs the merged record and then calls
`reducedRecord.newInstance(reducedKey)`. The Flink record context correctly
constructs a `HoodieEmptyRecord` with the delete's operation and ordering value.
However,
[`HoodieEmptyRecord.newInstance(HoodieKey)`](https://github.com/apache/hudi/blob/3aa7fe2a9f70ef52064c977d28d28b4f4c95d00b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieEmptyRecord.java#L76-L79)
uses the key-only constructor. It drops the operation and resets ordering to
the default `Integer(0)` sentinel. Downstream merge logic interprets that
sentinel as a commit-time/natural-order delete, bypassing the intended
event-time comparison. The loss occurs even when the copied key is equal to the
original key.
The proposed fix is to use the existing constructor accepting key,
operation, ordering value, and record type. Deletes that originally use natural
ordering should retain that behavior.
### Environment
- Hudi: `1.3.0-SNAPSHOT`, reproduced at
`3aa7fe2a9f70ef52064c977d28d28b4f4c95d00b`.
- Query engine: Flink `1.20.1`, Java `11.0.27`, Maven profile `flink1.20`.
- Physical reproduction: COW and MOR, table version 10; MOR writes native
Parquet logs.
- Relevant configuration:
```text
write.operation=upsert
hoodie.record.merge.mode=EVENT_TIME_ORDERING
ordering.fields=ts
write.precombine=true
metadata.enabled=false
compaction.async.enabled=false
```
The ordinary Flink SQL COW upsert setup enables precombine automatically.
Explicitly setting it to false is therefore not a universal workaround for that
factory path. MOR also exposes the problem when precombine is enabled.
### Logs and regression evidence
Before the fix:
```text
Deduplication: expected ordering=60, operation=DELETE; actual ordering=0,
operation=null
Older reinsert: expected DELETE(60); actual INSERT(55)
COW/MOR snapshot with precombine=true: expected [id1 at ts=100]; actual []
```
Both grouped and sorted helper reduction reproduce the metadata loss.
Physical write/read tests reproduce the incorrect COW and MOR snapshots, while
precombine-disabled controls pass.
This differs from
[apache/hudi#16447](https://github.com/apache/hudi/issues/16447), which
concerns the key-only delete API not accepting ordering fields. Here, Flink
already supplies the correct delete ordering, and the subsequent key copy
erases it.
--
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]
