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 9b7a7139bffb fix(spark): preserve delete payloads during record
reconstruction (#20341)
9b7a7139bffb is described below
commit 9b7a7139bffb8956de906a2ef0f71609e7be77ba
Author: Shuo Cheng <[email protected]>
AuthorDate: Sat Oct 17 10:02:08 2026 +0800
fix(spark): preserve delete payloads during record reconstruction (#20341)
---
.../hudi/BaseSparkInternalRecordContext.java | 2 +-
.../apache/hudi/functional/TestMORDataSource.scala | 33 ++++++++++++++++++++++
2 files changed, 34 insertions(+), 1 deletion(-)
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 7b236d3419ed..ef29b0849682 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
@@ -124,7 +124,7 @@ public abstract class BaseSparkInternalRecordContext
extends RecordContext<Inter
@Override
public HoodieRecord<InternalRow>
constructHoodieRecord(BufferedRecord<InternalRow> bufferedRecord, String
partitionPath) {
HoodieKey hoodieKey = new HoodieKey(bufferedRecord.getRecordKey(),
partitionPath);
- if (bufferedRecord.isDelete()) {
+ if (bufferedRecord.isDelete() && bufferedRecord.isEmpty()) {
return new HoodieEmptyRecord<>(
hoodieKey,
bufferedRecord.getHoodieOperation(),
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 c422693cde09..363cf5a3a4c6 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
@@ -736,6 +736,39 @@ class TestMORDataSource extends HoodieSparkClientTestBase
with SparkDatasetMixin
HoodieMemoryConfig.DEFAULT_MR_COMPACTION_MEMORY_FRACTION)
}
+ @ParameterizedTest
+ @CsvSource(value = Array("string;8", "string;9", "decimal(10,2);8",
"decimal(10,2);9"), delimiter = ';')
+ def testPrecombinedDeleteWithEngineOrdering(orderingType: String,
tableVersion: Int): Unit = {
+ val (writeOpts, readOpts) = getWriterReaderOpts(HoodieRecordType.SPARK)
+ val opts = writeOpts ++ Map(
+ TABLE_TYPE.key -> MOR_TABLE_TYPE_OPT_VAL,
+ RECORDKEY_FIELD.key -> "id",
+ PARTITIONPATH_FIELD.key -> "",
+ HoodieTableConfig.ORDERING_FIELDS.key -> "ts",
+ HoodieWriteConfig.RECORD_MERGE_MODE.key ->
RecordMergeMode.EVENT_TIME_ORDERING.name(),
+ HoodieWriteConfig.WRITE_TABLE_VERSION.key -> tableVersion.toString,
+ HoodieTableConfig.VERSION.key -> tableVersion.toString,
+ "hoodie.compact.inline" -> "false")
+
+ 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))
+
.coalesce(1).write.format("hudi").options(opts).mode(mode).save(basePath)
+ }
+
+ write(Seq(("id1", "100", false), ("id2", "100", false)),
SaveMode.Overwrite)
+ // Both keys go through precombine. The older delete must not remove id1's
base row.
+ write(Seq(("id1", "050", false), ("id1", "060", true),
+ ("id2", "050", false), ("id2", "110", true)), SaveMode.Append)
+
+ val rows =
spark.read.format("hudi").options(readOpts).load(basePath).select("id",
"ts").collect()
+ assertEquals(1, rows.length)
+ assertEquals("id1", rows.head.getString(0))
+ assertEquals(0,
BigDecimal(rows.head.get(1).toString).compare(BigDecimal(100)))
+ }
+
@ParameterizedTest
@CsvSource(value = Array("AVRO,6", "AVRO,8", "SPARK,6", "SPARK,8"))
def testPayloadDelete(recordType: HoodieRecordType, tableVersion: Int) {
