hudi-agent commented on code in PR #20288:
URL: https://github.com/apache/hudi/pull/20288#discussion_r4253901775
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/format/FlinkRecordContext.java:
##########
@@ -139,8 +139,7 @@ public RowData convertAvroRecord(IndexedRecord avroRecord) {
@Override
public HoodieRecord<RowData> constructHoodieRecord(BufferedRecord<RowData>
bufferedRecord, String partitionPath) {
HoodieKey hoodieKey = new HoodieKey(bufferedRecord.getRecordKey(),
partitionPath);
- // delete record
- if (bufferedRecord.isDelete()) {
+ if (bufferedRecord.isDelete() && bufferedRecord.isEmpty()) {
Review Comment:
🤖 nit: could you add a short comment on why a delete with row data is not
turned into a `HoodieEmptyRecord`? The removed comment was obvious, but the new
`isDelete() && isEmpty()` condition is subtle.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/model/HoodieEmptyRecord.java:
##########
@@ -75,7 +75,7 @@ public HoodieRecord<T> newInstance(HoodieKey key,
HoodieOperation op) {
@Override
public HoodieRecord<T> newInstance(HoodieKey key) {
- return new HoodieEmptyRecord<>(key, type);
+ return new HoodieEmptyRecord<>(key, operation, orderingVal, type);
Review Comment:
🤖 The `FlinkRecordContext` change here narrows this a lot: deletes that
still carry a row now come back as `HoodieFlinkRecord`, and its
`getOrderingValueAsJava` converts the value from the row via
`getColumnValueAsJava`. So the native `TimestampData`/`StringData` value only
reaches `HoodieInlineLogAppendHandle.writeDelete` for row-less deletes
(`bufferedRecord.isEmpty()`, e.g. a merger result that is an empty delete), and
those now skip the reset to `0` in `newInstance(key)`. Overriding
`getOrderingValueAsJava` in `HoodieEmptyRecord` (or converting just in that
branch) should cover the v8/v9 MOR case without touching the common path.
##########
hudi-common/src/main/java/org/apache/hudi/common/model/HoodieEmptyRecord.java:
##########
@@ -75,7 +75,7 @@ public HoodieRecord<T> newInstance(HoodieKey key,
HoodieOperation op) {
@Override
public HoodieRecord<T> newInstance(HoodieKey key) {
- return new HoodieEmptyRecord<>(key, type);
+ return new HoodieEmptyRecord<>(key, operation, orderingVal, type);
Review Comment:
🤖 The new `FlinkRecordContext` check fixes this for Flink, but
`BaseSparkInternalRecordContext.constructHoodieRecord` still turns every delete
that wins dedup into a `HoodieEmptyRecord` holding the Spark-native ordering
value (`HoodieUTF8String`, Spark `Decimal`), and this line now keeps that
value. Before, it was reset to `0`. Take a v8/v9 MOR upsert with
`HoodieSparkRecordMerger` and a string ordering field, where a delete wins
dedup: I think it would now throw `UnsupportedOperationException` in
`wrapValueIntoAvro` via `HoodieInlineLogAppendHandle.writeDelete`. Could you
apply the same `isEmpty()` split in the Spark record context, or convert the
value on the inline delete path, and add a Spark MOR v8/v9 case?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
