You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2025/08/03 07:00:12 UTC
(hudi) branch master updated: [HUDI-9683] Fix field renames in HoodieInternalRowUtils.genUnsafeRowWriter (#13672)
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 f51c36fd6439 [HUDI-9683] Fix field renames in HoodieInternalRowUtils.genUnsafeRowWriter (#13672)
f51c36fd6439 is described below
commit f51c36fd64398718c039b26c15767e793b82904e
Author: Jon Vexler <jb...@gmail.com>
AuthorDate: Sun Aug 3 03:00:06 2025 -0400
[HUDI-9683] Fix field renames in HoodieInternalRowUtils.genUnsafeRowWriter (#13672)
* fix field renaming in spark projection
* should not use full path so that we match the output of the merger
* update comments that were wrong
* avoid string handling when field is not renamed
---------
Co-authored-by: Jonathan Vexler <=>
Co-authored-by: danny0405 <yu...@gmail.com>
---
.../apache/spark/sql/HoodieInternalRowUtils.scala | 28 +++---
.../hudi/common/TestHoodieInternalRowUtils.scala | 102 +++++++++++++++++++++
2 files changed, 113 insertions(+), 17 deletions(-)
diff --git a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/HoodieInternalRowUtils.scala b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/HoodieInternalRowUtils.scala
index 4a91609e19ba..e5c5761c73b3 100644
--- a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/HoodieInternalRowUtils.scala
+++ b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/HoodieInternalRowUtils.scala
@@ -188,25 +188,14 @@ object HoodieInternalRowUtils {
fieldNamesStack.push(newField.name)
val (fieldWriter, prevFieldPos): (RowFieldUpdater, Int) =
- prevStructType.getFieldIndex(newField.name) match {
+ prevStructType.getFieldIndex(lookupRenamedField(newField.name, createFullName(fieldNamesStack), renamedColumnsMap)) match {
case Some(prevFieldPos) =>
val prevField = prevStructType(prevFieldPos)
(newWriterRenaming(prevField.dataType, newField.dataType, renamedColumnsMap, fieldNamesStack), prevFieldPos)
case None =>
- val newFieldQualifiedName = createFullName(fieldNamesStack)
- val prevFieldName: String = lookupRenamedField(newFieldQualifiedName, renamedColumnsMap)
-
- // Handle rename
- prevStructType.getFieldIndex(prevFieldName) match {
- case Some(prevFieldPos) =>
- val prevField = prevStructType.fields(prevFieldPos)
- (newWriterRenaming(prevField.dataType, newField.dataType, renamedColumnsMap, fieldNamesStack), prevFieldPos)
-
- case None =>
- val updater: RowFieldUpdater = (fieldUpdater, ordinal, _) => fieldUpdater.setNullAt(ordinal)
- (updater, -1)
- }
+ val updater: RowFieldUpdater = (fieldUpdater, ordinal, _) => fieldUpdater.setNullAt(ordinal)
+ (updater, -1)
}
fieldWriters += fieldWriter
@@ -415,9 +404,14 @@ object HoodieInternalRowUtils {
}
}
- private def lookupRenamedField(newFieldQualifiedName: String, renamedColumnsMap: JMap[String, String]) = {
- val prevFieldQualifiedName = renamedColumnsMap.getOrDefault(newFieldQualifiedName, "")
- val prevFieldQualifiedNameParts = prevFieldQualifiedName.split("\\.")
+ private def lookupRenamedField(newFieldName: String,
+ newFieldQualifiedName: String,
+ renamedColumnsMap: JMap[String, String]): String = {
+ val renamed = renamedColumnsMap.get(newFieldQualifiedName)
+ if (renamed == null) {
+ return newFieldName
+ }
+ val prevFieldQualifiedNameParts = renamed.split("\\.")
val prevFieldName = prevFieldQualifiedNameParts(prevFieldQualifiedNameParts.length - 1)
prevFieldName
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieInternalRowUtils.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieInternalRowUtils.scala
index d0d045da5931..989b2b19c225 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieInternalRowUtils.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieInternalRowUtils.scala
@@ -124,6 +124,108 @@ class TestHoodieInternalRowUtils extends FunSuite with Matchers with BeforeAndAf
assertEquals(serDe.deserializeRow(newRow2), Row("Andrew", 25, Row("Mission st", "SF")));
}
+ test("Test rewrite row with renamed columns") {
+ // Original schema
+ val oldSchema = StructType(Seq(
+ StructField("first_name", StringType),
+ StructField("years_old", IntegerType)
+ ))
+
+ // Renamed schema
+ val newSchema = StructType(Seq(
+ StructField("name", StringType),
+ StructField("age", IntegerType)
+ ))
+
+ // Rename mapping: new -> old
+ val renameMap: java.util.Map[String, String] = new java.util.HashMap()
+ renameMap.put("name", "first_name")
+ renameMap.put("age", "years_old")
+
+ // Sample row
+ val oldRowData = sparkSession.sparkContext.parallelize(Seq(Row("Alice", 30)))
+ val oldRow = sparkSession.createDataFrame(oldRowData, oldSchema).queryExecution.toRdd.first()
+
+ // Generate writer with rename map
+ val rowWriter = HoodieInternalRowUtils.genUnsafeRowWriter(oldSchema, newSchema, renameMap, JCollections.emptyMap())
+ val newRow = rowWriter(oldRow)
+
+ val serDe = sparkAdapter.createSparkRowSerDe(newSchema)
+ assertEquals(Row("Alice", 30), serDe.deserializeRow(newRow))
+ }
+
+ test("Test rewrite row with columns swap") {
+ // Original schema
+ val oldSchema = StructType(Seq(
+ StructField("first_name", StringType),
+ StructField("years_old", IntegerType)
+ ))
+
+ // Renamed schema
+ val newSchema = StructType(Seq(
+ StructField("years_old", StringType),
+ StructField("first_name", IntegerType)
+ ))
+
+ // Rename mapping: new -> old
+ val renameMap: java.util.Map[String, String] = new java.util.HashMap()
+ renameMap.put("years_old", "first_name")
+ renameMap.put("first_name", "years_old")
+
+ // Sample row
+ val oldRowData = sparkSession.sparkContext.parallelize(Seq(Row("Alice", 30)))
+ val oldRow = sparkSession.createDataFrame(oldRowData, oldSchema).queryExecution.toRdd.first()
+
+ // Generate writer with rename map
+ val rowWriter = HoodieInternalRowUtils.genUnsafeRowWriter(oldSchema, newSchema, renameMap, JCollections.emptyMap())
+ val newRow = rowWriter(oldRow)
+
+ val serDe = sparkAdapter.createSparkRowSerDe(newSchema)
+ assertEquals(Row("Alice", 30), serDe.deserializeRow(newRow))
+ }
+
+ test("Test rewrite row with columns swap nested") {
+ // Original schema
+ val oldSchema = StructType(Seq(
+ StructField("first_name", StringType),
+ StructField("years_old", IntegerType),
+ StructField("address",
+ StructType(Seq(
+ StructField("city", StringType),
+ StructField("street", StringType)
+ )
+ ))))
+
+ // Renamed schema
+ val newSchema = StructType(Seq(
+ StructField("years_old", StringType),
+ StructField("first_name", IntegerType),
+ StructField("address",
+ StructType(Seq(
+ StructField("street", StringType),
+ StructField("city", StringType)
+ )
+ ))))
+
+ // Rename mapping: new -> old
+ val renameMap: java.util.Map[String, String] = new java.util.HashMap()
+ renameMap.put("years_old", "first_name")
+ renameMap.put("first_name", "years_old")
+ renameMap.put("address.city", "street")
+ renameMap.put("address.street", "city")
+
+ // Sample row
+ val oldRowData = sparkSession.sparkContext.parallelize(Seq(Row("Alice", 30, Row("SF", "Mission st"))))
+ val oldRow = sparkSession.createDataFrame(oldRowData, oldSchema).queryExecution.toRdd.first()
+
+ // Generate writer with rename map
+ val rowWriter = HoodieInternalRowUtils.genUnsafeRowWriter(oldSchema, newSchema, renameMap, JCollections.emptyMap())
+ val newRow = rowWriter(oldRow)
+
+ val serDe = sparkAdapter.createSparkRowSerDe(newSchema)
+ assertEquals(Row("Alice", 30, Row("SF", "Mission st")), serDe.deserializeRow(newRow))
+ }
+
/**
* test record data type changes.
* int => long/float/double/string