hudi-agent commented on code in PR #20288:
URL: https://github.com/apache/hudi/pull/20288#discussion_r4246019326
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java:
##########
@@ -507,6 +509,47 @@ public void testUpsertWithDelete() throws Exception {
.end();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testDeduplicationPreservesEventTimeDeleteOrdering(boolean
preCombine) throws Exception {
+ conf.set(FlinkOptions.RECORD_MERGE_MODE,
RecordMergeMode.EVENT_TIME_ORDERING.name());
+ conf.set(FlinkOptions.PRE_COMBINE, preCombine);
+
+ List<RowData> inserts = new ArrayList<>();
+ List<RowData> changes = new ArrayList<>();
+ for (String key : Arrays.asList("id1", "id2")) {
+ inserts.add(TestData.insertRow(
+ StringData.fromString(key), StringData.fromString("base"), 100,
+ TimestampData.fromEpochMillis(100), StringData.fromString("par1")));
+ changes.add(TestData.insertRow(
+ StringData.fromString(key), StringData.fromString("older"), 50,
+ TimestampData.fromEpochMillis(50), StringData.fromString("par1")));
+ RowData delete = TestData.insertRow(
+ StringData.fromString(key), StringData.fromString("delete"), 60,
+ TimestampData.fromEpochMillis(key.equals("id1") ? 60 : 110),
StringData.fromString("par1"));
+ delete.setRowKind(RowKind.DELETE);
+ changes.add(delete);
Review Comment:
🤖 nit: could you pull the `key.equals("id1") ? 60 : 110` into a named local
(e.g. `deleteTs`) so the stale vs. newer delete intent is visible where it is
set?
<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:
🤖 This now keeps the engine-native ordering value that
`FlinkRecordContext.constructHoodieRecord` puts on dedup'd deletes (e.g.
`TimestampData` for the default `ts` field, or `StringData`/`DecimalData`), and
`HoodieEmptyRecord` doesn't override `getOrderingValueAsJava`. On a MOR table
with `hoodie.write.table.version` < 10,
`HoodieInlineLogAppendHandle.writeDelete` passes that value to
`HoodieDeleteBlock`, and `wrapValueIntoAvro` throws
`UnsupportedOperationException`; before this change the value was reset to `0`,
so the write went through. Could you convert the ordering value to its Java
type on the inline delete path, and add a v8/v9 MOR case for this test? The new
sink test only runs on v10, which uses the native log path where engine types
are expected.
<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]
