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 5b8323f88891 fix(spark): resolve nested ordering fields for Spark
deletes written to delete blocks (#20368)
5b8323f88891 is described below
commit 5b8323f88891be103668ce098a0aafcf38e0b208
Author: Vova Kolmakov <[email protected]>
AuthorDate: Wed Oct 21 09:40:41 2026 +0700
fix(spark): resolve nested ordering fields for Spark deletes written to
delete blocks (#20368)
* fix(spark): resolve nested ordering fields for Spark deletes written to
delete blocks
* addressed review comments: drop the repeated schema and path lookups in
getOrderingValueAsJava, rename the test
* addressed review comments: add a NestedFieldPath overload of
getFieldValueFromInternalRowInternal
---------
Co-authored-by: Vova Kolmakov <[email protected]>
---
.../hudi/common/model/HoodieSparkRecord.java | 8 ++++--
.../hudi/BaseSparkInternalRecordContext.java | 13 +++++++--
.../apache/hudi/functional/TestMORDataSource.scala | 32 ++++++++++++++++++++++
3 files changed, 48 insertions(+), 5 deletions(-)
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/common/model/HoodieSparkRecord.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/common/model/HoodieSparkRecord.java
index de30e4ef94a6..3443d9411cbf 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/common/model/HoodieSparkRecord.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/common/model/HoodieSparkRecord.java
@@ -389,11 +389,15 @@ public class HoodieSparkRecord extends
HoodieRecord<InternalRow> {
if (orderingFields == null) {
return OrderingValues.getDefault();
} else {
+ StructType structType =
HoodieInternalRowUtils.getCachedSchema(recordSchema);
return OrderingValues.create(orderingFields, field -> {
- if (recordSchema.getField(field).isEmpty()) {
+ // Ordering fields can be nested, so resolve them by path as
doGetOrderingValue does.
+ scala.Option<NestedFieldPath> cachedNestedFieldPath =
+ HoodieInternalRowUtils.getCachedPosList(structType, field);
+ if (cachedNestedFieldPath.isEmpty()) {
return OrderingValues.getDefault();
}
- return (Comparable<?>) getColumnValueAsJava(recordSchema, field,
props);
+ return (Comparable<?>) getFieldValueFromInternalRowAsJava(data,
cachedNestedFieldPath.get());
});
}
}
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 ef29b0849682..0fdc4dddbb36 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
@@ -76,6 +76,10 @@ public abstract class BaseSparkInternalRecordContext extends
RecordContext<Inter
return getFieldValueFromInternalRowInternal(row, recordSchema, fieldName,
true);
}
+ public static Object getFieldValueFromInternalRowAsJava(InternalRow row,
HoodieUnsafeRowUtils.NestedFieldPath nestedFieldPath) {
+ return getFieldValueFromInternalRowInternal(row, nestedFieldPath, true);
+ }
+
private static Object getFieldValueFromInternalRowInternal(InternalRow row,
HoodieSchema recordSchema, String fieldName, boolean convertToJavaType) {
return getFieldValueFromInternalRowInternal(row,
getCachedSchema(recordSchema), fieldName, convertToJavaType);
}
@@ -84,14 +88,17 @@ public abstract class BaseSparkInternalRecordContext
extends RecordContext<Inter
scala.Option<HoodieUnsafeRowUtils.NestedFieldPath> cachedNestedFieldPath =
HoodieInternalRowUtils.getCachedPosList(structType, fieldName);
if (cachedNestedFieldPath.isDefined()) {
- HoodieUnsafeRowUtils.NestedFieldPath nestedFieldPath =
cachedNestedFieldPath.get();
- Object value = HoodieUnsafeRowUtils.getNestedInternalRowValue(row,
nestedFieldPath);
- return convertToJavaType ? sparkTypeToJavaType(value) : value;
+ return getFieldValueFromInternalRowInternal(row,
cachedNestedFieldPath.get(), convertToJavaType);
} else {
return null;
}
}
+ private static Object getFieldValueFromInternalRowInternal(InternalRow row,
HoodieUnsafeRowUtils.NestedFieldPath nestedFieldPath, boolean
convertToJavaType) {
+ Object value = HoodieUnsafeRowUtils.getNestedInternalRowValue(row,
nestedFieldPath);
+ return convertToJavaType ? sparkTypeToJavaType(value) : value;
+ }
+
private static Object sparkTypeToJavaType(Object value) {
if (value == null) {
return null;
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 363cf5a3a4c6..f2db2ac68753 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
@@ -2627,6 +2627,38 @@ class TestMORDataSource extends
HoodieSparkClientTestBase with SparkDatasetMixin
def testNestedFieldPartition(): Unit = {
TestCOWDataSource.runNestedFieldPartitionTest(spark, basePath, storage,
"MOR")
}
+
+ @ParameterizedTest
+ @CsvSource(value = Array("AVRO;bigint;8", "SPARK;bigint;9",
"SPARK;string;8", "SPARK;decimal(10,2);9"), delimiter = ';')
+ def testDeleteWithNestedOrderingFieldRespectsEventTimeOrdering(recordType:
HoodieRecordType, orderingType: String, tableVersion: Int): 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 -> "meta.ts",
+ HoodieWriteConfig.RECORD_MERGE_MODE.key ->
RecordMergeMode.EVENT_TIME_ORDERING.name(),
+ HoodieWriteConfig.WRITE_TABLE_VERSION.key -> tableVersion.toString,
+ HoodieTableConfig.VERSION.key -> tableVersion.toString,
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false")
+
+ def write(rows: Seq[(String, String, Boolean)], mode: SaveMode): Unit = {
+ val localSpark = spark
+ import localSpark.implicits._
+ rows.toDF("id", "ts", "_hoodie_is_deleted")
+ .select(col("id"),
struct(col("ts").cast(orderingType).as("ts")).as("meta"),
col("_hoodie_is_deleted"))
+
.coalesce(1).write.format("hudi").options(opts).mode(mode).save(basePath)
+ }
+
+ write(Seq(("id1", "100", false), ("id2", "100", false)),
SaveMode.Overwrite)
+ // The delete of id1 is older than the stored row and must not remove it;
the delete of id2 is newer.
+ // "050" has a leading zero so that it also sorts before "100" as a string.
+ write(Seq(("id1", "050", true), ("id2", "150", true)), SaveMode.Append)
+
+ val rows =
spark.read.format("hudi").options(readOpts).load(basePath).select("id",
"meta.ts").collect()
+ assertEquals(Seq("id1"), rows.map(_.getString(0)).toSeq)
+ assertEquals(100, BigDecimal(rows.head.get(1).toString).toInt)
+ }
}
object TestMORDataSource {
