This is an automated email from the ASF dual-hosted git repository.
voonhous 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 a18f22126dfa [HUDI-18060] Improve error message when ordering field
value is null (#18061)
a18f22126dfa is described below
commit a18f22126dfa029b59d68cf53c4d1a3b75be0e33
Author: Prashant Wason <[email protected]>
AuthorDate: Wed Jul 22 02:38:55 2026 -0700
[HUDI-18060] Improve error message when ordering field value is null
(#18061)
When records have a null value in the ordering (precombine) field, Hudi
jobs previously failed with a cryptic "Ordering value is null for record"
error that gave no actionable context. This change fails fast in
HoodieCreateRecordUtils with a clear message identifying the offending
ordering field and record key, and suggesting remediation.
Merge modes that do not depend on the ordering value are exempted:
COMMIT_TIME_ORDERING and OverwriteWithLatestAvroPayload fall back to
OrderingValues.getDefault() instead of failing. The payload-class check
is retained so table version 6 (which may not have a merge mode set)
still bypasses the failure for OverwriteWithLatestAvroPayload.
The gating flag is computed once at driver scope rather than per record
to avoid repeated work on the per-record hot path.
Closes #18060
---
.../org/apache/hudi/HoodieCreateRecordUtils.scala | 45 +++-
.../apache/hudi/TestHoodieCreateRecordUtils.scala | 291 +++++++++++++++++++++
2 files changed, 330 insertions(+), 6 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCreateRecordUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCreateRecordUtils.scala
index d942657a463a..bd96649c19af 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCreateRecordUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCreateRecordUtils.scala
@@ -20,7 +20,7 @@ package org.apache.hudi
import org.apache.hudi.DataSourceWriteOptions.INSERT_DROP_DUPS
import org.apache.hudi.common.avro.{AvroRecordContext, AvroSchemaCache,
HoodieAvroUtils}
-import org.apache.hudi.common.config.TypedProperties
+import org.apache.hudi.common.config.{RecordMergeMode, TypedProperties}
import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.model._
import org.apache.hudi.common.model.WriteOperationType.isChangingRecords
@@ -78,6 +78,12 @@ object HoodieCreateRecordUtils {
val preppedSparkSqlMergeInto = args.preppedSparkSqlMergeInto
val preppedWriteOperation = args.preppedWriteOperation
val orderingFields = args.tableConfig.getOrderingFields
+ val recordMergeMode = args.tableConfig.getRecordMergeMode
+ val payloadClass = config.getPayloadClass
+ // Ordering values are not required for COMMIT_TIME_ORDERING or
OverwriteWithLatestAvroPayload.
+ // Table version 6 may not have a merge mode set, so the payload class
check is still needed.
+ val requiresOrderingValue = !((recordMergeMode ==
RecordMergeMode.COMMIT_TIME_ORDERING)
+ || classOf[OverwriteWithLatestAvroPayload].getName.equals(payloadClass))
val shouldDropPartitionColumns =
config.getBoolean(DataSourceWriteOptions.DROP_PARTITION_COLUMNS)
val recordType = config.getRecordMerger.getRecordType
@@ -150,11 +156,8 @@ object HoodieCreateRecordUtils {
avroRecWithoutMeta
}
val hoodieRecord = if (shouldCombine && !orderingFields.isEmpty) {
- val orderingVal = OrderingValues.create(
- orderingFields,
- JFunction.toJavaFunction[String, Comparable[_]](
- field => HoodieAvroUtils.getNestedFieldVal(avroRec, field,
false,
-
consistentLogicalTimestampEnabled).asInstanceOf[Comparable[_]]))
+ val orderingVal = getOrderingValue(orderingFields, avroRec,
hoodieKey.getRecordKey,
+ consistentLogicalTimestampEnabled, requiresOrderingValue)
HoodieRecordUtils.createHoodieRecord(processedRecord,
orderingVal, hoodieKey,
config.getPayloadClass, null, recordLocation, requiresPayload,
isDelete)
} else {
@@ -282,4 +285,34 @@ object HoodieCreateRecordUtils {
(new HoodieKey(recordKey, partitionPath), recordLocation)
}
+
+ /**
+ * Gets the ordering value from the ordering fields of an Avro record.
+ * When `requiresOrderingValue` is false (e.g., COMMIT_TIME_ORDERING or
OverwriteWithLatestAvroPayload),
+ * null values are allowed and a default ordering value is used.
+ * Otherwise, throws IllegalArgumentException if any ordering field has a
null value.
+ */
+ private def getOrderingValue(orderingFields: java.util.List[String],
+ avroRec: GenericRecord,
+ recordKey: String,
+ consistentLogicalTimestampEnabled: Boolean,
+ requiresOrderingValue: Boolean): Comparable[_]
= {
+ OrderingValues.create(
+ orderingFields,
+ JFunction.toJavaFunction[String, Comparable[_]](field => {
+ val fieldVal = HoodieAvroUtils.getNestedFieldVal(avroRec, field,
false, consistentLogicalTimestampEnabled)
+ if (fieldVal == null) {
+ if (requiresOrderingValue) {
+ throw new IllegalArgumentException(
+ s"Ordering field '$field' has null value for record key
'$recordKey'. " +
+ s"Please ensure all records have non-null values for the
ordering field, " +
+ s"or use a payload class that doesn't require ordering (e.g.,
OverwriteWithLatestAvroPayload).")
+ }
+ // Return default ordering value for payloads that don't require
ordering
+ OrderingValues.getDefault.asInstanceOf[Comparable[_]]
+ } else {
+ fieldVal.asInstanceOf[Comparable[_]]
+ }
+ }))
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieCreateRecordUtils.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieCreateRecordUtils.scala
new file mode 100644
index 000000000000..74e800534c43
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieCreateRecordUtils.scala
@@ -0,0 +1,291 @@
+/*
+ * 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
+
+import org.apache.hudi.common.config.RecordMergeMode
+import org.apache.hudi.common.model.WriteOperationType
+import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.keygen.constant.KeyGeneratorOptions
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.types._
+import org.junit.jupiter.api.{AfterAll, BeforeAll, Test}
+import org.junit.jupiter.api.Assertions.{assertNotNull, assertTrue}
+
+/**
+ * Test cases for {@link HoodieCreateRecordUtils}.
+ */
+class TestHoodieCreateRecordUtils {
+
+ private val SPARK_SCHEMA = StructType(Seq(
+ StructField("uuid", StringType, nullable = false),
+ StructField("name", StringType, nullable = false),
+ StructField("age", IntegerType, nullable = false),
+ StructField("ts", LongType, nullable = true),
+ StructField("partition", StringType, nullable = false)
+ ))
+
+ // Common test constants
+ private val TEST_TABLE_NAME = "test_table"
+ private val RECORD_NAME = "TestRecord"
+ private val RECORD_NAMESPACE = "org.apache.hudi.test"
+ private val INSTANT_TIME = "20231031000000"
+ private val RECORD_KEY_FIELD = "uuid"
+ private val PARTITION_FIELD = "partition"
+ private val PRECOMBINE_FIELD = "ts"
+
+ /**
+ * Helper method to create DataFrame from Row data
+ */
+ private def createTestDataFrame(rows: Row*): org.apache.spark.sql.DataFrame
= {
+ val spark = TestHoodieCreateRecordUtils.spark
+ spark.createDataFrame(spark.sparkContext.parallelize(rows), SPARK_SCHEMA)
+ }
+
+ /**
+ * Helper method to get the root cause of an exception.
+ * Iterative implementation to avoid stack overflow and handle circular
references.
+ *
+ * @param t The throwable to extract root cause from
+ * @return The root cause throwable
+ */
+ private def getRootCause(t: Throwable): Throwable = {
+ var current = t
+ val visited = scala.collection.mutable.Set[Throwable]()
+
+ while (current.getCause != null && !visited.contains(current)) {
+ visited += current
+ current = current.getCause
+ }
+
+ current
+ }
+
+ /**
+ * Helper method to create base parameters common to all tests.
+ * These are the mandatory properties required by SimpleKeyGenerator.
+ */
+ private def createBaseParameters(): Map[String, String] = {
+ Map(
+ // KeyGeneratorOptions (used by some parts of the pipeline)
+ KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key() -> RECORD_KEY_FIELD,
+ KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key() -> PARTITION_FIELD,
+ // DataSourceWriteOptions (required by SimpleKeyGenerator)
+ DataSourceWriteOptions.RECORDKEY_FIELD.key() -> RECORD_KEY_FIELD,
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key() -> PARTITION_FIELD
+ )
+ }
+
+ /**
+ * Helper method to create common parameters for tests with precombine
+ */
+ private def createParametersWithPrecombine(payloadClass: String =
"org.apache.hudi.common.model.DefaultHoodieRecordPayload"): Map[String, String]
= {
+ createBaseParameters() ++ Map(
+ DataSourceWriteOptions.PRECOMBINE_FIELD.key() -> PRECOMBINE_FIELD,
+ DataSourceWriteOptions.PAYLOAD_CLASS_NAME.key() -> payloadClass,
+ HoodieWriteConfig.COMBINE_BEFORE_UPSERT.key() -> "true",
+ DataSourceWriteOptions.INSERT_DROP_DUPS.key() -> "false"
+ )
+ }
+
+ /**
+ * Helper method to create parameters for tests without precombine
+ */
+ private def createParametersWithoutPrecombine(): Map[String, String] = {
+ createBaseParameters() ++ Map(
+ DataSourceWriteOptions.PAYLOAD_CLASS_NAME.key() ->
"org.apache.hudi.common.model.OverwriteWithLatestAvroPayload",
+ HoodieWriteConfig.COMBINE_BEFORE_INSERT.key() -> "false",
+ DataSourceWriteOptions.INSERT_DROP_DUPS.key() -> "false"
+ )
+ }
+
+ @Test
+ def testNullPrecombineFieldThrowsClearError(): Unit = {
+ val df = createTestDataFrame(Row("id1", "Alice", 25, null, "par1"))
+ val parameters = createParametersWithPrecombine()
+
+ val exception = try {
+ // Attempt to write which will trigger HoodieCreateRecordUtils
+ df.write
+ .format("hudi")
+ .options(parameters)
+ .option(DataSourceWriteOptions.TABLE_NAME.key(), TEST_TABLE_NAME)
+ .option("hoodie.table.name", TEST_TABLE_NAME)
+ .option("path", TestHoodieCreateRecordUtils.tempDir +
"/test_null_precombine")
+ .mode("overwrite")
+ .save()
+ null
+ } catch {
+ case e: SparkException =>
+ getRootCause(e) match {
+ case iae: IllegalArgumentException => iae
+ case other => other
+ }
+ case e: IllegalArgumentException => e
+ case e: Exception =>
+ getRootCause(e) match {
+ case iae: IllegalArgumentException => iae
+ case _ => throw e
+ }
+ }
+
+ assertNotNull(exception, "Expected IllegalArgumentException for null
precombine field")
+ assertTrue(exception.isInstanceOf[IllegalArgumentException],
+ s"Expected IllegalArgumentException but got
${exception.getClass.getName}")
+ assertTrue(exception.getMessage.contains("has null value for record key"),
+ s"Exception message should mention null value for record key. Actual:
${exception.getMessage}")
+ assertTrue(exception.getMessage.contains("Please ensure all records have
non-null values for the ordering field"),
+ s"Exception message should provide guidance. Actual:
${exception.getMessage}")
+ assertTrue(exception.getMessage.contains("OverwriteWithLatestAvroPayload"),
+ s"Exception message should suggest alternative payload class. Actual:
${exception.getMessage}")
+ }
+
+ @Test
+ def testValidPrecombineFieldSucceeds(): Unit = {
+ val df = createTestDataFrame(Row("id1", "Alice", 25, 1000L, "par1"))
+ val parameters = createParametersWithPrecombine()
+
+ // Should not throw exception
+ df.write
+ .format("hudi")
+ .options(parameters)
+ .option(DataSourceWriteOptions.TABLE_NAME.key(), TEST_TABLE_NAME)
+ .option("hoodie.table.name", TEST_TABLE_NAME)
+ .option("path", TestHoodieCreateRecordUtils.tempDir +
"/test_valid_precombine")
+ .mode("overwrite")
+ .save()
+
+ // Verify data was written
+ val result = TestHoodieCreateRecordUtils.spark.read
+ .format("hudi")
+ .load(TestHoodieCreateRecordUtils.tempDir + "/test_valid_precombine")
+ assertTrue(result.count() > 0, "Data should have been written
successfully")
+ }
+
+ @Test
+ def testNullPrecombineFieldErrorContainsRecordKey(): Unit = {
+ val testRecordKey = "test_key_123"
+ val df = createTestDataFrame(Row(testRecordKey, "Bob", 30, null, "par2"))
+ val parameters = createParametersWithPrecombine()
+
+ val exception = try {
+ df.write
+ .format("hudi")
+ .options(parameters)
+ .option(DataSourceWriteOptions.TABLE_NAME.key(), TEST_TABLE_NAME)
+ .option("hoodie.table.name", TEST_TABLE_NAME)
+ .option("path", TestHoodieCreateRecordUtils.tempDir +
"/test_null_precombine_key")
+ .mode("overwrite")
+ .save()
+ null
+ } catch {
+ case e: Exception =>
+ getRootCause(e) match {
+ case iae: IllegalArgumentException => iae
+ case _ => throw e
+ }
+ }
+
+ assertNotNull(exception)
+ assertTrue(exception.getMessage.contains(testRecordKey),
+ s"Exception message should contain the record key '$testRecordKey' to
help identify the problematic record. Actual: ${exception.getMessage}")
+ }
+
+ @Test
+ def testNullPrecombineFieldWithOverwritePayloadSucceeds(): Unit = {
+ // OverwriteWithLatestAvroPayload should allow null precombine values
+ val df = createTestDataFrame(Row("id1", "Alice", 25, null, "par1"))
+ val parameters = createParametersWithPrecombine(
+ payloadClass =
"org.apache.hudi.common.model.OverwriteWithLatestAvroPayload")
+
+ // Should not throw exception - OverwriteWithLatestAvroPayload doesn't
require ordering values
+ df.write
+ .format("hudi")
+ .options(parameters)
+ .option(DataSourceWriteOptions.TABLE_NAME.key(), TEST_TABLE_NAME)
+ .option("hoodie.table.name", TEST_TABLE_NAME)
+ .option("path", TestHoodieCreateRecordUtils.tempDir +
"/test_null_precombine_overwrite")
+ .mode("overwrite")
+ .save()
+
+ // Verify data was written
+ val result = TestHoodieCreateRecordUtils.spark.read
+ .format("hudi")
+ .load(TestHoodieCreateRecordUtils.tempDir +
"/test_null_precombine_overwrite")
+ assertTrue(result.count() > 0, "Data should have been written successfully
with null precombine using OverwriteWithLatestAvroPayload")
+ }
+
+ @Test
+ def testNullPrecombineFieldWithCommitTimeOrderingSucceeds(): Unit = {
+ // COMMIT_TIME_ORDERING merge mode should allow null precombine values
+ val df = createTestDataFrame(Row("id1", "Alice", 25, null, "par1"))
+ val parameters = createBaseParameters() ++ Map(
+ DataSourceWriteOptions.PRECOMBINE_FIELD.key() -> PRECOMBINE_FIELD,
+ HoodieWriteConfig.COMBINE_BEFORE_UPSERT.key() -> "true",
+ DataSourceWriteOptions.INSERT_DROP_DUPS.key() -> "false",
+ DataSourceWriteOptions.RECORD_MERGE_MODE.key() ->
RecordMergeMode.COMMIT_TIME_ORDERING.name()
+ )
+
+ // Should not throw exception - COMMIT_TIME_ORDERING doesn't require
ordering values
+ df.write
+ .format("hudi")
+ .options(parameters)
+ .option(DataSourceWriteOptions.TABLE_NAME.key(), TEST_TABLE_NAME)
+ .option("hoodie.table.name", TEST_TABLE_NAME)
+ .option("path", TestHoodieCreateRecordUtils.tempDir +
"/test_null_precombine_commit_time")
+ .mode("overwrite")
+ .save()
+
+ // Verify data was written
+ val result = TestHoodieCreateRecordUtils.spark.read
+ .format("hudi")
+ .load(TestHoodieCreateRecordUtils.tempDir +
"/test_null_precombine_commit_time")
+ assertTrue(result.count() > 0, "Data should have been written successfully
with null precombine using COMMIT_TIME_ORDERING")
+ }
+}
+
+object TestHoodieCreateRecordUtils {
+ var spark: SparkSession = _
+ var tempDir: String = _
+
+ @BeforeAll
+ def setupSpark(): Unit = {
+ tempDir =
java.nio.file.Files.createTempDirectory("hudi_test_").toFile.getAbsolutePath
+ spark = SparkSession.builder()
+ .appName("TestHoodieCreateRecordUtils")
+ .master("local[2]")
+ .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
+ .config("spark.sql.shuffle.partitions", "1")
+ .config("spark.sql.extensions",
"org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
+ .config("spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.hudi.catalog.HoodieCatalog")
+ .getOrCreate()
+ }
+
+ @AfterAll
+ def teardownSpark(): Unit = {
+ if (spark != null) {
+ spark.stop()
+ }
+ // Clean up temp directory
+ if (tempDir != null) {
+ org.apache.commons.io.FileUtils.deleteQuietly(new java.io.File(tempDir))
+ }
+ }
+}
