This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new b86a34b59e01 fix(flink): preserve event-time delete ordering during
precombine (#20288)
b86a34b59e01 is described below
commit b86a34b59e015b68796d88de8b18eea38c1739ae
Author: Shuo Cheng <[email protected]>
AuthorDate: Wed Oct 14 10:47:04 2026 +0800
fix(flink): preserve event-time delete ordering during precombine (#20288)
---
.../hudi/table/format/FlinkRecordContext.java | 3 +-
.../table/action/commit/TestFlinkWriteHelper.java | 30 ++++++++
.../hudi/table/format/TestFlinkRecordContext.java | 39 ++++++++++
.../hudi/common/model/HoodieEmptyRecord.java | 2 +-
.../hudi/common/model/TestHoodieEmptyRecord.java | 88 ++++++++++++++++++++++
.../org/apache/hudi/sink/TestWriteCopyOnWrite.java | 46 +++++++++++
6 files changed, 205 insertions(+), 3 deletions(-)
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/format/FlinkRecordContext.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/format/FlinkRecordContext.java
index 3d3d06430be4..f3421cd4aefd 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/format/FlinkRecordContext.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/format/FlinkRecordContext.java
@@ -139,8 +139,7 @@ public class FlinkRecordContext extends
RecordContext<RowData> {
@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()) {
return new HoodieEmptyRecord<>(hoodieKey,
bufferedRecord.getHoodieOperation(), bufferedRecord.getOrderingValue(),
HoodieRecord.HoodieRecordType.FLINK);
}
RowData rowData = bufferedRecord.getRecord();
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/table/action/commit/TestFlinkWriteHelper.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/table/action/commit/TestFlinkWriteHelper.java
index 1751cee3c4a9..808a404c9373 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/table/action/commit/TestFlinkWriteHelper.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/table/action/commit/TestFlinkWriteHelper.java
@@ -42,6 +42,7 @@ import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.function.Executable;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
@@ -53,6 +54,7 @@ import java.util.NoSuchElementException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
+import static org.junit.jupiter.api.Assertions.assertAll;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
@@ -140,6 +142,34 @@ class TestFlinkWriteHelper {
}
}
+ @ParameterizedTest
+ @CsvSource({"false", "true"})
+ void testDeduplicationPreservesDeleteOrdering(boolean sortedByRecordKey) {
+ HoodieReaderContext<RowData> readerContext = readerContext();
+ TypedProperties props = new TypedProperties();
+ HoodieSchema schema = HoodieSchema.parse(ROW_SCHEMA);
+ BufferedRecordMerger<RowData> merger = BufferedRecordMergerFactory.create(
+ readerContext, RecordMergeMode.EVENT_TIME_ORDERING, false,
+ Option.empty(), Option.empty(), schema, props, Option.empty());
+ List<HoodieRecord<RowData>> records = Arrays.asList(
+ row("a", 50, "older", HoodieOperation.INSERT),
+ row("a", 60, null, HoodieOperation.DELETE),
+ row("b", 50, "older", HoodieOperation.INSERT),
+ row("b", 60, null, HoodieOperation.DELETE),
+ row("b", 55, "obsolete-reinsert", HoodieOperation.INSERT));
+ List<HoodieRecord<RowData>> actual = CollectionUtils.toList(
+ rowWriteHelper.deduplicateSortedRecords(
+ records.iterator(), sortedByRecordKey, ROW_SCHEMA,
+ props, merger, readerContext, ORDERING_FIELDS));
+ assertEquals(2, actual.size());
+ DeleteContext deleteContext = DeleteContext.fromRecordSchema(props,
schema);
+ assertAll(actual.stream().map(record -> (Executable) () ->
+ assertAll(record.getRecordKey(),
+ () -> assertEquals(60L, record.getOrderingValue(schema, props,
ORDERING_FIELDS)),
+ () -> assertEquals(HoodieOperation.DELETE, record.getOperation()),
+ () -> assertTrue(record.isDelete(deleteContext, props)))));
+ }
+
@Test
@SuppressWarnings("unchecked")
void testSortedDeduplicationPreservesCustomMergeOrderAndEmptyResult() throws
Exception {
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/table/format/TestFlinkRecordContext.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/table/format/TestFlinkRecordContext.java
index 847401887fd3..e318cdc3b69f 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/table/format/TestFlinkRecordContext.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/table/format/TestFlinkRecordContext.java
@@ -19,25 +19,38 @@
package org.apache.hudi.table.format;
+import org.apache.hudi.client.model.HoodieFlinkRecord;
+import org.apache.hudi.common.model.HoodieEmptyRecord;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieOperation;
+import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.table.read.BufferedRecord;
+import org.apache.hudi.common.table.read.BufferedRecords;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.data.columnar.ColumnarRowData;
import org.apache.flink.table.data.columnar.vector.ColumnVector;
import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
import org.apache.flink.table.data.columnar.vector.heap.HeapBytesVector;
import org.apache.flink.table.data.columnar.vector.heap.HeapIntVector;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.Properties;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
class TestFlinkRecordContext {
@@ -45,6 +58,32 @@ class TestFlinkRecordContext {
HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.STRING)),
HoodieSchemaField.of("value",
HoodieSchema.create(HoodieSchemaType.INT))));
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testConstructDeleteRecord(boolean hasRowData) {
+ HoodieSchema schema = HoodieSchema.createRecord("delete_record", null,
null,
+ Collections.singletonList(HoodieSchemaField.of("ts",
HoodieSchema.createTimestampMillis())));
+ TimestampData orderingValue = TimestampData.fromEpochMillis(60);
+ RowData row = hasRowData ? GenericRowData.of(orderingValue) : null;
+ FlinkRecordContext context =
FlinkRecordContext.getDeleteCheckingInstance();
+ BufferedRecord<RowData> buffered = new BufferedRecord<>("id1",
orderingValue, row, context.encodeSchema(schema), HoodieOperation.DELETE);
+
+ HoodieRecord<RowData> record = context.constructHoodieRecord(buffered,
"par1").newInstance(new HoodieKey("id1", "par2"));
+
+ assertEquals(hasRowData ? HoodieFlinkRecord.class :
HoodieEmptyRecord.class, record.getClass());
+ assertSame(row, record.getData());
+ assertEquals(HoodieOperation.DELETE, record.getOperation());
+ assertTrue(record.isDelete(null, new Properties()));
+ if (hasRowData) {
+ assertEquals(60L, record.getOrderingValueAsJava(schema, new
Properties(), new String[] {"ts"}));
+ }
+ assertSame(orderingValue, record.getOrderingValue(schema, new
Properties(), new String[] {"ts"}));
+ BufferedRecord<RowData> roundTrip =
BufferedRecords.fromHoodieRecord(record, schema, context, new Properties(), new
String[] {"ts"}, true);
+ assertTrue(roundTrip.isDelete());
+ assertSame(row, roundTrip.getRecord());
+ assertSame(orderingValue, roundTrip.getOrderingValue());
+ }
+
@Test
void testSealDoesNotCopyGenericRowData() {
GenericRowData rowData = new GenericRowData(2);
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieEmptyRecord.java
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieEmptyRecord.java
index 444ac3a0323e..dd58b05352b2 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieEmptyRecord.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieEmptyRecord.java
@@ -75,7 +75,7 @@ public class HoodieEmptyRecord<T> extends HoodieRecord<T> {
@Override
public HoodieRecord<T> newInstance(HoodieKey key) {
- return new HoodieEmptyRecord<>(key, type);
+ return new HoodieEmptyRecord<>(key, operation, orderingVal, type);
}
@Override
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieEmptyRecord.java
b/hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieEmptyRecord.java
new file mode 100644
index 000000000000..f276ebf8f86e
--- /dev/null
+++
b/hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieEmptyRecord.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.model;
+
+import org.apache.hudi.common.util.OrderingValues;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.Properties;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestHoodieEmptyRecord {
+
+ @ParameterizedTest
+ @MethodSource("orderedDeletes")
+ void testKeyCopyPreservesDeleteOrdering(HoodieOperation operation,
Comparable<?> orderingValue,
+ HoodieRecord.HoodieRecordType
recordType) {
+ HoodieKey originalKey = new HoodieKey("old-key", "old-partition");
+ HoodieKey newKey = new HoodieKey("new-key", "new-partition");
+ Properties props = new Properties();
+ String[] orderingFields = {"ts"};
+ HoodieEmptyRecord<Object> original = new HoodieEmptyRecord<>(originalKey,
operation, orderingValue, recordType);
+ assertEquals(orderingValue, original.getOrderingValue(null, props,
orderingFields));
+
+ HoodieRecord<Object> copy = original.newInstance(newKey);
+
+ assertAll(
+ () -> assertNotSame(original, copy),
+ () -> assertEquals(newKey, copy.getKey()),
+ () -> assertEquals(originalKey, original.getKey()),
+ () -> assertEquals(operation, copy.getOperation()),
+ () -> assertEquals(orderingValue, copy.getOrderingValue(null, props,
orderingFields)),
+ () -> assertEquals(recordType, copy.getRecordType()),
+ () -> assertTrue(copy.isDelete(null, props)),
+ () -> assertNull(copy.getData()));
+ }
+
+ private static Stream<Arguments> orderedDeletes() {
+ return Stream.of(
+ Arguments.of(HoodieOperation.DELETE, 60L,
HoodieRecord.HoodieRecordType.FLINK),
+ Arguments.of(HoodieOperation.UPDATE_BEFORE, 60L,
HoodieRecord.HoodieRecordType.FLINK),
+ Arguments.of(HoodieOperation.DELETE, "060",
HoodieRecord.HoodieRecordType.AVRO),
+ Arguments.of(HoodieOperation.UPDATE_BEFORE, "060",
HoodieRecord.HoodieRecordType.SPARK),
+ Arguments.of(HoodieOperation.DELETE, OrderingValues.getDefault(),
HoodieRecord.HoodieRecordType.HIVE),
+ Arguments.of(HoodieOperation.UPDATE_BEFORE,
OrderingValues.getDefault(), HoodieRecord.HoodieRecordType.FLINK));
+ }
+
+ @ParameterizedTest
+ @EnumSource(HoodieRecord.HoodieRecordType.class)
+ void testKeyCopyPreservesNaturalOrderDelete(HoodieRecord.HoodieRecordType
recordType) {
+ HoodieEmptyRecord<Object> original = new HoodieEmptyRecord<>(new
HoodieKey("key", "partition"), recordType);
+ HoodieKey newKey = new HoodieKey("key", "new-partition");
+ Properties props = new Properties();
+
+ HoodieRecord<Object> copy = original.newInstance(newKey);
+
+ assertEquals(newKey, copy.getKey());
+ assertEquals(recordType, copy.getRecordType());
+ assertNull(copy.getOperation());
+ assertTrue(OrderingValues.isDefault(copy.getOrderingValue(null, props, new
String[0])));
+ assertTrue(copy.isDelete(null, props));
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
index 7f53f9fb8efc..8c263f38c9c4 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
@@ -22,6 +22,7 @@ import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.common.HoodieFlinkEngineContext;
import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.RecordMergeMode;
import org.apache.hudi.common.data.HoodieListData;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieKey;
@@ -63,9 +64,11 @@ import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.types.RowKind;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
@@ -507,6 +510,49 @@ public class TestWriteCopyOnWrite extends TestWriteBase {
.end();
}
+ @ParameterizedTest
+ @CsvSource({"8,false", "8,true", "9,false", "9,true", "10,false", "10,true"})
+ public void testDeduplicationPreservesEventTimeDeleteOrdering(int
tableVersion, boolean preCombine) throws Exception {
+ conf.set(FlinkOptions.WRITE_TABLE_VERSION, tableVersion);
+ 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")));
+ long deleteTs = key.equals("id1") ? 60 : 110;
+ RowData delete = TestData.insertRow(
+ StringData.fromString(key), StringData.fromString("delete"), 60,
+ TimestampData.fromEpochMillis(deleteTs),
StringData.fromString("par1"));
+ delete.setRowKind(RowKind.DELETE);
+ changes.add(delete);
+ }
+
+ preparePipeline()
+ .consume(inserts)
+ .checkpoint(1)
+ .assertNextEvent()
+ .checkpointComplete(1)
+ .assertGlobalFailure(false)
+ .checkWrittenData(Collections.singletonMap(
+ "par1", "[id1,par1,id1,base,100,100,par1,
id2,par1,id2,base,100,100,par1]"), 1)
+ .consume(changes)
+ .assertDataBuffer(1, 4)
+ .checkpoint(2)
+ .assertNextEvent()
+ .checkpointComplete(2)
+ .assertGlobalFailure(false)
+ // The stale delete must not remove id1 at ts=100; the newer delete
must remove id2.
+ .checkWrittenData(Collections.singletonMap("par1",
"[id1,par1,id1,base,100,100,par1]"), 1)
+ .end();
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testInsertWithMiniBatches(boolean
useFileGroupReaderBasedMergeHandle) throws Exception {
