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 31f4a5ef2de3 fix(common): write ordering values to native delete logs
in their stored form (#20373)
31f4a5ef2de3 is described below
commit 31f4a5ef2de3023ef3374a42133d6679b06747e5
Author: Vova Kolmakov <[email protected]>
AuthorDate: Wed Oct 21 09:56:39 2026 +0700
fix(common): write ordering values to native delete logs in their stored
form (#20373)
* fix(common): write ordering values to native delete logs in their stored
form
---------
Co-authored-by: Vova Kolmakov <[email protected]>
---
.../hudi/io/cdc/HoodieNativeLogFormatWriter.java | 7 ++-
.../io/cdc/TestHoodieNativeLogFormatWriter.java | 52 ++++++++++++++++++++
.../hudi/BaseSparkInternalRecordContext.java | 4 ++
.../TestBaseSparkInternalRowReaderContext.java | 14 ++++++
.../apache/hudi/common/engine/RecordContext.java | 3 +-
.../apache/hudi/functional/TestMORDataSource.scala | 55 ++++++++++++++++++++++
6 files changed, 132 insertions(+), 3 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/cdc/HoodieNativeLogFormatWriter.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/cdc/HoodieNativeLogFormatWriter.java
index f1ed4a9e821e..7f0414f2cf39 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/cdc/HoodieNativeLogFormatWriter.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/cdc/HoodieNativeLogFormatWriter.java
@@ -193,11 +193,14 @@ public class HoodieNativeLogFormatWriter extends
HoodieLogFormat.Writer {
fieldValues[0] = recordContext.convertValueToEngineType(recordKey);
if (!OrderingValues.isCommitTimeOrderingValue(orderingValue)) {
if (orderingFieldNames.size() == 1) {
- fieldValues[1] = orderingValue;
+ fieldValues[1] = recordContext.convertValueToEngineType(orderingValue);
} else {
List<Comparable> orderingValues =
OrderingValues.getValues((ArrayComparable) orderingValue);
for (int i = 0; i < orderingValues.size(); i++) {
- fieldValues[i + 1] = orderingValues.get(i);
+ // A null ordering field comes as the default ordering value and is
written as NULL.
+ Comparable value = orderingValues.get(i);
+ fieldValues[i + 1] = OrderingValues.isCommitTimeOrderingValue(value)
+ ? null : recordContext.convertValueToEngineType(value);
}
}
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/cdc/TestHoodieNativeLogFormatWriter.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/cdc/TestHoodieNativeLogFormatWriter.java
index d343ca4c077f..490d7162125b 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/cdc/TestHoodieNativeLogFormatWriter.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/cdc/TestHoodieNativeLogFormatWriter.java
@@ -36,6 +36,7 @@ import
org.apache.hudi.common.table.log.NativeLogFooterMetadata;
import
org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.OrderingValues;
+import org.apache.hudi.common.util.collection.ArrayComparable;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.core.io.storage.HoodieFileWriter;
import org.apache.hudi.core.io.storage.HoodieFileWriterFactory;
@@ -58,10 +59,12 @@ import java.io.IOException;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
@@ -200,6 +203,20 @@ public class TestHoodieNativeLogFormatWriter {
verify(record, never()).getOrderingValue(eq(schema), any(), any());
}
+ @Test
+ public void testConvertsOrderingValueForDeleteLog() throws Exception {
+ assertArrayEquals(new Object[] {"stored-key-1", "stored-110"},
+ writeDeleteLogFieldValues(Collections.singletonList("ts"), "110"));
+ }
+
+ @Test
+ public void testConvertsEachOrderingValueForDeleteLog() throws Exception {
+ // The default ordering value stands for a null ordering field and is
written as NULL.
+ assertArrayEquals(new Object[] {"stored-key-1", "stored-110", null,
"stored-2"},
+ writeDeleteLogFieldValues(Arrays.asList("ts", "seq", "seq2"),
+ new ArrayComparable(new Comparable[] {"110",
OrderingValues.getDefault(), "2"})));
+ }
+
@Test
public void testSkipsUnsupportedDataFileFormatMetadata() throws Exception {
Option<Object> metadata = writeDataLogAndGetFormatMetadata(
@@ -525,6 +542,41 @@ public class TestHoodieNativeLogFormatWriter {
return NativeLogFooterMetadata.fromFooterMetadata(footerCaptor.getValue());
}
+ private static Object[] writeDeleteLogFieldValues(List<String>
orderingFieldNames, Comparable orderingValue) throws Exception {
+ String schemaString =
"{\"type\":\"record\",\"name\":\"test\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},"
+ +
"{\"name\":\"ts\",\"type\":\"string\"},{\"name\":\"seq\",\"type\":\"string\"},{\"name\":\"seq2\",\"type\":\"string\"}]}";
+ HoodieSchema schema = HoodieSchema.parse(schemaString);
+ HoodieWriteConfig config =
HoodieWriteConfig.newBuilder().withPath("/tmp").withSchema(schemaString).build();
+ HoodieStorage storage = mock(HoodieStorage.class);
+ TaskContextSupplier taskContextSupplier = mock(TaskContextSupplier.class);
+ HoodieRecord record = mock(HoodieRecord.class);
+ when(record.getRecordKey()).thenReturn("key-1");
+ doReturn(orderingValue).when(record).getOrderingValue(eq(schema), any(),
any());
+ RecordContext recordContext = mock(RecordContext.class);
+
when(recordContext.getEngineRecordType()).thenReturn(HoodieRecord.HoodieRecordType.AVRO);
+ // Marks every value that went through the conversion to the stored engine
type.
+ when(recordContext.convertValueToEngineType(any())).thenAnswer(invocation
-> "stored-" + invocation.getArgument(0));
+ when(recordContext.constructEngineRecord(any(HoodieSchema.class),
any(Object[].class))).thenReturn(new Object());
+
+ try (MockedStatic<HoodieFileWriterFactory> writerFactory =
mockStatic(HoodieFileWriterFactory.class)) {
+ writerFactory.when(() -> HoodieFileWriterFactory.getFileWriter(
+ eq("100"), any(StoragePath.class), eq(storage), eq(config),
any(HoodieSchema.class),
+ eq(taskContextSupplier), eq(HoodieRecord.HoodieRecordType.AVRO)))
+ .thenReturn(mock(HoodieFileWriter.class));
+
+ try (HoodieNativeLogFormatWriter writer = new
HoodieNativeLogFormatWriter(
+ 4096, storage, new StoragePath("/tmp/partition"), "file-1", "100",
1, "1-0-1", 1024L,
+ new LogFileCreationCallback() {
+ }, HoodieTableVersion.current(), config, HoodieFileFormat.PARQUET,
schema,
+ taskContextSupplier, recordContext, orderingFieldNames,
Option.empty())) {
+ writer.appendDeleteRecord(record, schema);
+ }
+ }
+ ArgumentCaptor<Object[]> fieldValuesCaptor =
ArgumentCaptor.forClass(Object[].class);
+ verify(recordContext).constructEngineRecord(any(HoodieSchema.class),
fieldValuesCaptor.capture());
+ return fieldValuesCaptor.getValue();
+ }
+
private static void
assertNoBufferedRecordPositions(HoodieNativeLogFormatWriter writer, String
fieldName) throws Exception {
Field field =
HoodieNativeLogFormatWriter.class.getDeclaredField(fieldName);
field.setAccessible(true);
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRecordContext.java
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRecordContext.java
index 0fdc4dddbb36..dff6b48141e1 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRecordContext.java
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRecordContext.java
@@ -182,6 +182,10 @@ public abstract class BaseSparkInternalRecordContext
extends RecordContext<Inter
// into HoodieUTF8String.
return UTF8String.fromString((String) value);
}
+ if (value instanceof HoodieUTF8String) {
+ // An ordering value wraps the UTF8String to make it comparable; an
engine record stores the raw UTF8String.
+ return ((HoodieUTF8String) value).getUtf8String();
+ }
return value;
}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/TestBaseSparkInternalRowReaderContext.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/TestBaseSparkInternalRowReaderContext.java
index 62dfdd1a1144..f665e5a83c14 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/TestBaseSparkInternalRowReaderContext.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/TestBaseSparkInternalRowReaderContext.java
@@ -86,6 +86,20 @@ class TestBaseSparkInternalRowReaderContext {
assertEquals(fieldVals[2], row.getBoolean(2));
}
+ @Test
+ void testConstructEngineRecordWithComparableStringValue() {
+ UTF8String name = UTF8String.fromString("Alice");
+ HoodieUTF8String comparableName = new HoodieUTF8String(name) {
+ @Override
+ public int compareTo(HoodieUTF8String other) {
+ return toString().compareTo(other.toString());
+ }
+ };
+ Object[] fieldVals = new Object[]{1,
readerContext.getRecordContext().convertValueToEngineType(comparableName),
true};
+ InternalRow row =
readerContext.getRecordContext().constructEngineRecord(SCHEMA, fieldVals);
+ assertEquals(name, row.getUTF8String(1));
+ }
+
@Test
void testConstructEngineRecordWithNoUpdate() {
InternalRow base = new GenericInternalRow(new Object[]{1,
UTF8String.fromString("Alice"), true});
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/engine/RecordContext.java
b/hudi-common/src/main/java/org/apache/hudi/common/engine/RecordContext.java
index 6a95e70a7b0e..257ea08f0d05 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/engine/RecordContext.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/engine/RecordContext.java
@@ -213,7 +213,8 @@ public abstract class RecordContext<T> implements
Serializable {
/**
* Converts the value to the engine-native type representation that can be
stored directly
* as a field value inside an engine record, e.g., {@code UTF8String} for a
{@link String}
- * value in Spark.
+ * value in Spark. An ordering value in its comparable form, see {@link
#ensureComparability},
+ * is converted back to the stored form, e.g., {@code HoodieUTF8String} to
{@code UTF8String} in Spark.
* <p>
* NOTE: The returned value is NOT guaranteed to support comparison on every
engine
* (e.g., Spark's {@code UTF8String} no longer supports {@code compareTo},
see SPARK-46832).
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala
index f2db2ac68753..359e1fd54fc6 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala
@@ -1827,6 +1827,61 @@ class TestMORDataSource extends
HoodieSparkClientTestBase with SparkDatasetMixin
assertEquals(metaClient.getTableConfig.getRecordMergeStrategyId,
mergerStrategyName)
}
+ @ParameterizedTest
+ @CsvSource(value = Array(
+ "AVRO;AVRO;string;ts", "SPARK;SPARK;string;ts",
"SPARK;SPARK;decimal(10,2);ts", "SPARK;SPARK;string;ts,seq",
+ "AVRO;SPARK;string;ts"), delimiter = ';')
+ def testNativeDeleteLogWithEngineOrdering(recordType: HoodieRecordType,
logCompactionRecordType: HoodieRecordType,
+ orderingType: String,
orderingFields: String): Unit = {
+ val (writeOpts, readOpts) = getWriterReaderOpts(recordType)
+ val opts = writeOpts ++ Map(
+ TABLE_TYPE.key -> MOR_TABLE_TYPE_OPT_VAL,
+ RECORDKEY_FIELD.key -> "id",
+ PARTITIONPATH_FIELD.key -> "",
+ HoodieTableConfig.ORDERING_FIELDS.key -> orderingFields,
+ HoodieWriteConfig.RECORD_MERGE_MODE.key ->
RecordMergeMode.EVENT_TIME_ORDERING.name(),
+ // Table version 10 writes deletes to native delete log files.
+ HoodieWriteConfig.WRITE_TABLE_VERSION.key ->
HoodieTableVersion.TEN.versionCode().toString,
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false",
+ HoodieCompactionConfig.ENABLE_LOG_COMPACTION.key -> "true",
+ HoodieCompactionConfig.LOG_COMPACTION_BLOCKS_THRESHOLD.key -> "1")
+
+ def write(rows: Seq[(String, String, Boolean)], mode: SaveMode): Unit = {
+ val localSpark = spark
+ import localSpark.implicits._
+ rows.toDF("id", "ts", "_hoodie_is_deleted")
+ .withColumn("ts", col("ts").cast(orderingType))
+ .withColumn("seq", when(col("_hoodie_is_deleted"),
lit(null)).otherwise(lit("1")).cast(orderingType))
+
.coalesce(1).write.format("hudi").options(opts).mode(mode).save(basePath)
+ }
+
+ write(Seq(("id1", "100", false), ("id2", "100", false)),
SaveMode.Overwrite)
+ // The delete leaves seq null, so with two ordering fields its second
value is missing.
+ write(Seq(("id1", "110", true)), SaveMode.Append)
+ write(Seq(("id2", "120", false)), SaveMode.Append)
+
+ // Log compaction reads the delete back and writes it into a new native
delete log. Its record type can differ
+ // from the writers', e.g. when a Spark-typed job compacts deletes that an
Avro-typed one wrote.
+ val logCompactionOpts = if (logCompactionRecordType ==
HoodieRecordType.SPARK) opts ++ sparkOpts else opts
+ val client = DataSourceUtils.createHoodieClient(spark.sparkContext, "",
basePath, tableName, logCompactionOpts.asJava)
+ .asInstanceOf[SparkRDDWriteClient[HoodieRecordPayload[Nothing]]]
+ try {
+ val logCompactionInstant =
client.scheduleLogCompaction(Option.empty()).get()
+ client.logCompact(logCompactionInstant, true)
+ } finally {
+ client.close()
+ }
+
+ def readOrdering(): Map[String, Int] =
spark.read.format("hudi").options(readOpts).load(basePath)
+ .select("id", "ts").collect().map(row => row.getString(0) ->
BigDecimal(row.get(1).toString).toInt).toMap
+
+ // The delete keeps its ordering value: an older update leaves id1
deleted, a newer one brings it back.
+ write(Seq(("id1", "105", false)), SaveMode.Append)
+ assertEquals(Map("id2" -> 120), readOrdering())
+ write(Seq(("id1", "115", false)), SaveMode.Append)
+ assertEquals(Map("id1" -> 115, "id2" -> 120), readOrdering())
+ }
+
/**
* Test Read-Optimized and time travel query on MOR table with RECORD_INDEX
enabled.
*/
