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 231545998a12 fix(spark-sql): resolve a partition path without
validating the record key (#19709)
231545998a12 is described below
commit 231545998a12fd1943e2393065f76cbdc0602eac
Author: Lin Liu <[email protected]>
AuthorDate: Tue Aug 25 01:34:30 2026 -0700
fix(spark-sql): resolve a partition path without validating the record key
(#19709)
* fix(spark-sql): resolve a partition path without validating the record key
SqlKeyGenerator#getPartitionPath(GenericRecord) delegated to
BaseKeyGenerator#getKey,
which builds the whole HoodieKey and so computes AND validates the record
key. Asking
for a partition path therefore failed on a record whose record key was
unset, even with
the partition column fully populated.
That is reachable on MOR with a global bloom or global simple index. The
tagging stage
merges the incoming record with its existing version --
mayContainDuplicateLookup is
tableType == MERGE_ON_READ in HoodieGlobalBloomIndex and
HoodieGlobalSimpleIndex, and
requiresMergingWithOlderRecordVersion in
HoodieIndexUtils#tagGlobalLocationBackToRecords is true on MOR whereas CoW
defers the
merge to the file rewrite -- and then asks the key generator for the merged
record's
partition path via HoodieIndexUtils#inferPartitionPath. Under MOR partial
updates that
merged record is materialised against WRITE_PARTIAL_UPDATE_SCHEMA, carrying
only the
columns named in UPDATE SET, so a MERGE INTO that does not assign the
record key -- the
ordinary partial-update shape -- died with
HoodieKeyException: recordKey value: "null" for field: "id" cannot be
null or empty
Resolve the partition path directly where the delegate exposes it.
Narrowing on
BaseKeyGenerator rather than casting keeps behaviour identical for a key
generator that
implements only SparkKeyGeneratorInterface, which declares just the Row and
InternalRow
accessors; its only Avro-facing method, inherited from
KeyGeneratorInterface, is getKey.
getPartitionPathFields in this same class already uses that construct.
Scoped to the GenericRecord overload. getRecordKey(GenericRecord) still
routes through
getKey and still validates -- it wants the record key -- and getKey itself
is untouched,
so callers needing a fully validated HoodieKey keep getting one. The change
is strictly
more permissive: getKey is
new HoodieKey(getRecordKey(record), getPartitionPath(record)), so this
returns the same
expression for the partition half and merely stops computing the record-key
half.
TestSqlKeyGenerator pins all three edges: partition resolution with the
record key unset
now succeeds (this case fails on the parent commit with the exception
above), record-key
resolution still rejects a missing key, and a complete record is unaffected.
This unmasks a separate, previously unreachable defect on the same path,
where the
partial-update merged record is serialized against a mismatched schema and
raises
UnresolvedUnionException from BaseAvroPayload#getRecordBytes. That is a
distinct root
cause in the payload path and is left for a follow-up so this change stays
reviewable;
end-to-end coverage of the shape is held with it.
* Cover the sql partition schema and the default-partition substitution
Three cases added after review. One sets hoodie.sql.partition.schema, which
Spark SQL always
supplies, so convertPartitionPathToSqlType is actually exercised rather
than short-circuited.
The other two pin what happens when the partition field itself cannot be
resolved from the
record. KeyGenUtils.getPartitionPath substitutes the default partition for
a null or absent
value, and that predates this change: getKey called the same
getPartitionPath, so the
substitution already applied whenever the record key resolved. Confirmed on
the parent commit,
where the record-key-present case passes and only the both-missing case
fails. What this change
widens is therefore narrow, and now stated: with the record key also
unresolvable, the record-key
exception no longer pre-empts the substitution.
Resolving a partition path is not a validity check on the record, so a
caller that needs the
partition to be genuinely present has to assert that itself.
* Cover the timestamp arms and the auto-record-key delegate, and mirror
getRecordKey
Addresses review feedback.
The sql-partition-schema case pinned nothing: a `dt string` schema takes
the identity arm of
convertPartitionPathToSqlType, so deleting that call entirely left every
test green. Replaced
with a `dt timestamp` schema over a microsecond value, which drives the
TimestampType arm and
escapePathName. The expected value was captured from a run rather than
hand-derived, and the
default DateTimeZone is pinned and restored because the output is formatted
in it.
Added a TimestampBasedKeyGenerator delegate case. That generator resolves
an absent partition
field to the formatted epoch rather than HUDI_DEFAULT_PARTITION_PATH, so
the guard in
convertPartitionPathToSqlType never fires for it. This change makes that
reachable, where the
record-key exception previously pre-empted it, so the behaviour is pinned
rather than implied.
Added a case that omits the record-key config. That makes
isAutoGeneratedRecordKeysEnabled true
and exercises the AutoRecordGenWrapperKeyGenerator, which this change does
alter: as a
BaseKeyGenerator it now takes the direct arm, so getPartitionPath no longer
consumes a generated
sequence id as a side effect of building a HoodieKey. Uniqueness does not
depend on that stride,
but it is now a pinned decision rather than an unexercised side effect.
getRecordKey(GenericRecord) takes the mirror arm, so neither accessor pays
for the other half.
Previously it computed the partition path and discarded it on every Avro
record, and a
partition-side failure could surface from a record-key lookup:
TimestampBasedAvroKeyGenerator
raises independently of the key.
Annotated the default-partition case with a pointer to
TestSimpleKeyGenerator's "TODO this
should throw as well" on the Avro path, so both expectations are unwound
together. Dropped the
string-interpolation prefixes that interpolate nothing and derived the two
partial schemas from
the base schema instead of restating them.
* Set the partition schema in the fixtures, guard the MergeIntoKeyGenerator
ordinals, drop the getRecordKey mirror
Addresses review feedback. Three things, the first of which invalidated
most of the previous
round's coverage.
The fixtures defaulted the partition schema to None, but production sets
hoodie.sql.partition.schema at every construction site
(ProvidesHoodieConfig and both
MergeIntoHoodieTableCommand copies). With it unset,
convertPartitionPathToSqlType returns its
input immediately, so seven of the eight cases never reached the
default-partition guard, the
fragment-count early-out or hive-style handling, including the two named
after default-partition
behaviour. It now defaults to a real schema, with one explicit None case
for the non-partitioned
table, paired against the timestamp case on the same value so the
conversion is shown to be
skipped rather than merely absent. The auto-record-key case was likewise
red on the parent commit
only because the fixture omitted a property that HoodieCreateRecordUtils
always sets, so it failed
on the harness rather than on the behaviour it claimed to pin; it now sets
both properties and
asserts the sequence id, which reads 100_1_0 here and 100_1_2 on the
parent. The timestamp
delegate is pinned under all three partition-schema spellings, because they
diverge: a timestamp
column turns the epoch string into a bare NumberFormatException, a string
column silently accepts
the epoch partition.
MergeIntoKeyGenerator reads the record key and partition path by meta-field
ordinal, and a record
materialised against WRITE_PARTIAL_UPDATE_SCHEMA can be shorter than the
ordinal, so the read
raised ArrayIndexOutOfBoundsException from inside the key generator. Both
Avro arms now check the
length first and fall through to the existing fallback. This is a bounds
check, not a proof that
the field at the ordinal is a meta field: a long record whose schema is not
meta-prefixed still
reads data there, as before. The new test covers both short-record arms
plus two meta-prefixed
regression cases, without which the guard could disable meta-field
resolution outright and the
short-record cases would still pass.
getRecordKey(GenericRecord) goes back to the getKey form. The mirror is the
right end state but it
is a second behaviour change, unrelated to the reported issue, and this PR
is already gated on
merge order; it belongs in a follow-up with the case that discriminates it.
Of the 15 cases, 5 fail on the parent commit and 10 pass there. The 10 pin
behaviour rather than
guarding a regression, so they are coverage, not proof of the fix.
* Set the partition id property as a string so both lines compile
The Int literal boxes to Object on this base but not on the 1.x line, where
scalac rejects the
implicit conversion. Both TypedProperties variants render the value through
String.valueOf on read, so the string form is equivalent at runtime and
compiles on either.
---
.../sql/hudi/command/MergeIntoKeyGenerator.scala | 28 ++-
.../spark/sql/hudi/command/SqlKeyGenerator.scala | 10 +-
.../hudi/command/TestMergeIntoKeyGenerator.scala | 141 +++++++++++
.../sql/hudi/command/TestSqlKeyGenerator.scala | 257 +++++++++++++++++++++
4 files changed, 433 insertions(+), 3 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoKeyGenerator.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoKeyGenerator.scala
index 8ae16217f1aa..b6f92494b1b6 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoKeyGenerator.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoKeyGenerator.scala
@@ -39,7 +39,11 @@ import org.apache.spark.unsafe.types.UTF8String
class MergeIntoKeyGenerator(props: TypedProperties) extends
SqlKeyGenerator(props) {
override def getRecordKey(record: GenericRecord): String = {
- val recordKey = record.get(RECORD_KEY_META_FIELD_ORD)
+ val recordKey = if (carriesMetaField(record, RECORD_KEY_META_FIELD_ORD)) {
+ record.get(RECORD_KEY_META_FIELD_ORD)
+ } else {
+ null
+ }
if (recordKey != null) {
recordKey.toString
} else {
@@ -66,7 +70,11 @@ class MergeIntoKeyGenerator(props: TypedProperties) extends
SqlKeyGenerator(prop
}
override def getPartitionPath(record: GenericRecord): String = {
- val partitionPath = record.get(PARTITION_PATH_META_FIELD_ORD)
+ val partitionPath = if (carriesMetaField(record,
PARTITION_PATH_META_FIELD_ORD)) {
+ record.get(PARTITION_PATH_META_FIELD_ORD)
+ } else {
+ null
+ }
if (partitionPath != null) {
partitionPath.toString
} else {
@@ -92,4 +100,20 @@ class MergeIntoKeyGenerator(props: TypedProperties) extends
SqlKeyGenerator(prop
}
}
+ /**
+ * Whether the record is long enough for `ord` to address a meta field at
all.
+ *
+ * The meta fields are only prepended once a record has been through the
write path, so a record
+ * built against a projected schema can be shorter than the ordinal: a MOR
partial update
+ * materialises the merged record against `WRITE_PARTIAL_UPDATE_SCHEMA`,
which carries only the
+ * columns named in `UPDATE SET`. Reading the ordinal off such a record
raises a bare
+ * `ArrayIndexOutOfBoundsException` out of the key generator, which does not
name the statement
+ * that caused it. Falling back to the SQL key generator gives the same
answer the unpopulated
+ * meta field would have routed to.
+ *
+ * This is a bounds check, not a proof that the field at `ord` is a meta
field: a record of the
+ * right length whose schema is not meta-prefixed still reads data here,
exactly as before.
+ */
+ private def carriesMetaField(record: GenericRecord, ord: Int): Boolean =
+ record.getSchema.getFields.size > ord
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/SqlKeyGenerator.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/SqlKeyGenerator.scala
index 3df7c6673961..7062beb1440c 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/SqlKeyGenerator.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/SqlKeyGenerator.scala
@@ -99,7 +99,15 @@ class SqlKeyGenerator(props: TypedProperties) extends
BuiltinKeyGenerator(props)
override def getPartitionPath(record: GenericRecord): String = {
val partitionPath = originalKeyGen.map {
- _.getKey(record).getPartitionPath
+ // Resolve the partition path on its own where the key generator exposes
it. Going through
+ // BaseKeyGenerator#getKey would also compute and validate the record
key, which a MOR partial
+ // update legitimately leaves unset: the merged record is materialised
against
+ // WRITE_PARTIAL_UPDATE_SCHEMA and so carries only the columns named in
UPDATE SET. Callers
+ // that want the record key validated still ask for it, via getKey or
getRecordKey.
+ case baseKeyGen: BaseKeyGenerator => baseKeyGen.getPartitionPath(record)
+ // SparkKeyGeneratorInterface exposes no Avro-facing accessor beyond
getKey, so a generator
+ // that is not a BaseKeyGenerator retains the previous behaviour.
+ case keyGen => keyGen.getKey(record).getPartitionPath
} getOrElse {
complexKeyGen.getPartitionPath(record)
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestMergeIntoKeyGenerator.scala
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestMergeIntoKeyGenerator.scala
new file mode 100644
index 000000000000..e306cc7311bd
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestMergeIntoKeyGenerator.scala
@@ -0,0 +1,141 @@
+/*
+ * 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.spark.sql.hudi.command
+
+import org.apache.hudi.common.config.TypedProperties
+import org.apache.hudi.common.util.PartitionPathEncodeUtils
+import org.apache.hudi.exception.HoodieKeyException
+import org.apache.hudi.keygen.SimpleKeyGenerator
+import org.apache.hudi.keygen.constant.KeyGeneratorOptions
+
+import org.apache.avro.Schema
+import org.apache.avro.generic.GenericData
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
+import org.junit.jupiter.api.Test
+
+/**
+ * Tests that [[MergeIntoKeyGenerator]] reads the meta-field ordinals only
from records long enough
+ * to have them.
+ *
+ * It resolves both the record key and the partition path by ordinal (2 and 3)
off the meta-field
+ * prefix, falling back to [[SqlKeyGenerator]] when the meta field is
unpopulated. A MOR partial
+ * update materialises the merged record against
`WRITE_PARTIAL_UPDATE_SCHEMA`, so the record can be
+ * shorter than the ordinal, and an unguarded read raises
`ArrayIndexOutOfBoundsException` from
+ * inside the key generator, naming nothing about the statement that caused it.
+ */
+class TestMergeIntoKeyGenerator {
+
+ /** A record carrying only the columns an `UPDATE SET amount, ts` would
assign. */
+ private val partialSchema = new Schema.Parser().parse(
+ """
+ |{
+ | "type": "record",
+ | "name": "partial_record",
+ | "fields": [
+ | {"name": "amount", "type": ["null", "double"], "default": null},
+ | {"name": "ts", "type": ["null", "long"], "default": null}
+ | ]
+ |}
+ """.stripMargin)
+
+ /** The shape the write path produces: the five meta fields, then the data
columns. */
+ private val metaPrefixedSchema = new Schema.Parser().parse(
+ """
+ |{
+ | "type": "record",
+ | "name": "meta_prefixed_record",
+ | "fields": [
+ | {"name": "_hoodie_commit_time", "type": ["null", "string"],
"default": null},
+ | {"name": "_hoodie_commit_seqno", "type": ["null", "string"],
"default": null},
+ | {"name": "_hoodie_record_key", "type": ["null", "string"],
"default": null},
+ | {"name": "_hoodie_partition_path", "type": ["null", "string"],
"default": null},
+ | {"name": "_hoodie_file_name", "type": ["null", "string"],
"default": null},
+ | {"name": "id", "type": ["null", "long"], "default": null},
+ | {"name": "amount", "type": ["null", "double"], "default": null},
+ | {"name": "dt", "type": ["null", "string"], "default": null}
+ | ]
+ |}
+ """.stripMargin)
+
+ private def keyGenerator: MergeIntoKeyGenerator = {
+ val props = new TypedProperties()
+ props.put(SqlKeyGenerator.ORIGINAL_KEYGEN_CLASS_NAME,
classOf[SimpleKeyGenerator].getName)
+ props.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key, "id")
+ props.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key, "dt")
+ props.put(SqlKeyGenerator.PARTITION_SCHEMA, "dt string")
+ new MergeIntoKeyGenerator(props)
+ }
+
+ private def partialRecord: GenericData.Record = {
+ val record = new GenericData.Record(partialSchema)
+ record.put("amount", 15.0d)
+ record.put("ts", 200L)
+ record
+ }
+
+ /**
+ * Two fields against partition-path ordinal 3. Before the guard this raised
+ * ArrayIndexOutOfBoundsException; it now falls through to the SQL key
generator, which resolves
+ * the absent partition field to the default partition.
+ */
+ @Test
+ def testGetPartitionPathFallsBackOnARecordShorterThanTheOrdinal(): Unit = {
+ assertEquals(PartitionPathEncodeUtils.DEFAULT_PARTITION_PATH,
+ keyGenerator.getPartitionPath(partialRecord))
+ }
+
+ /**
+ * Same record against record-key ordinal 2. The fallback reaches record-key
validation, so the
+ * failure names the field rather than an array bound.
+ */
+ @Test
+ def testGetRecordKeyFallsBackOnARecordShorterThanTheOrdinal(): Unit = {
+ val thrown = assertThrows(classOf[HoodieKeyException], () =>
keyGenerator.getRecordKey(partialRecord))
+ assert(thrown.getMessage.contains("id"), s"expected the message to name
the field, got: ${thrown.getMessage}")
+ }
+
+ /**
+ * Regression guard for the normal path: a meta-prefixed record is long
enough, so both accessors
+ * still read the ordinal rather than falling back. Without this the guard
could disable meta-field
+ * resolution outright and the tests above would still pass.
+ */
+ @Test
+ def testMetaPrefixedRecordStillResolvesFromTheMetaFields(): Unit = {
+ val record = new GenericData.Record(metaPrefixedSchema)
+ record.put("_hoodie_record_key", "id:1")
+ record.put("_hoodie_partition_path", "dt=2026-08-11")
+ record.put("id", 1L)
+ record.put("dt", "2026-08-12") // deliberately disagrees, to show the meta
field is what is read
+ assertEquals("id:1", keyGenerator.getRecordKey(record))
+ assertEquals("dt=2026-08-11", keyGenerator.getPartitionPath(record))
+ }
+
+ /**
+ * A meta-prefixed record whose meta fields are unpopulated is still long
enough, so the ordinal is
+ * read, found null, and the existing fallback runs. Pins that the guard did
not change which of
+ * the two fallback reasons applies.
+ */
+ @Test
+ def testMetaPrefixedRecordWithNullMetaFieldsFallsBackToTheDataColumns():
Unit = {
+ val record = new GenericData.Record(metaPrefixedSchema)
+ record.put("id", 1L)
+ record.put("dt", "2026-08-11")
+ assertEquals("1", keyGenerator.getRecordKey(record))
+ assertEquals("2026-08-11", keyGenerator.getPartitionPath(record))
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestSqlKeyGenerator.scala
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestSqlKeyGenerator.scala
new file mode 100644
index 000000000000..cdb90b1a7e44
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestSqlKeyGenerator.scala
@@ -0,0 +1,257 @@
+/*
+ * 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.spark.sql.hudi.command
+
+import org.apache.hudi.common.config.TimestampKeyGeneratorConfig
+import org.apache.hudi.common.config.TypedProperties
+import org.apache.hudi.common.util.PartitionPathEncodeUtils
+import org.apache.hudi.exception.HoodieKeyException
+import org.apache.hudi.keygen.{KeyGenUtils, SimpleKeyGenerator,
TimestampBasedKeyGenerator}
+import org.apache.hudi.keygen.constant.KeyGeneratorOptions
+
+import org.apache.avro.Schema
+import org.apache.avro.generic.GenericData
+import org.joda.time.DateTimeZone
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
+import org.junit.jupiter.api.Test
+
+import scala.collection.JavaConverters._
+
+/**
+ * Tests that [[SqlKeyGenerator]] resolves a partition path without also
requiring the record key.
+ *
+ * MOR partial updates materialise the merged record against
`WRITE_PARTIAL_UPDATE_SCHEMA`, which
+ * carries only the fields named in `UPDATE SET`.
`HoodieIndexUtils#inferPartitionPath` then asks
+ * the key generator for that record's partition path, so a record key absent
from the assignments
+ * is legitimately unset at that point and must not fail partition resolution.
+ *
+ * The fixtures set `hoodie.sql.partition.schema` by default because
production always does, at
+ * every construction site (`ProvidesHoodieConfig` and both
`MergeIntoHoodieTableCommand` copies).
+ * Leaving it unset makes `convertPartitionPathToSqlType` return its input
immediately, which skips
+ * the default-partition guard, the fragment-count early-out and hive-style
handling, so the cases
+ * named after those behaviours would never reach them.
+ */
+class TestSqlKeyGenerator {
+
+ private val schema = new Schema.Parser().parse(
+ """
+ |{
+ | "type": "record",
+ | "name": "test_record",
+ | "fields": [
+ | {"name": "id", "type": ["null", "long"], "default": null},
+ | {"name": "amount", "type": ["null", "double"], "default": null},
+ | {"name": "dt", "type": ["null", "string"], "default": null}
+ | ]
+ |}
+ """.stripMargin)
+
+ /** 2026-08-11 00:00:00 UTC in microseconds, which is what the GenericRecord
path assumes. */
+ private val timestampMicros = String.valueOf(1786406400000000L)
+
+ /** The same record shape with only the named fields, as a partial update
produces. */
+ private def projected(fieldNames: String*): Schema = {
+ val fields = schema.getFields.asScala
+ .filter(f => fieldNames.contains(f.name))
+ .map(f => new Schema.Field(f.name, f.schema, null, f.defaultVal))
+ Schema.createRecord("test_record", null, null, false, fields.asJava)
+ }
+
+ private def keyGenerator(partitionSchema: Option[String] = Some("dt string"),
+ withRecordKey: Boolean = true): SqlKeyGenerator = {
+ val props = new TypedProperties()
+ props.put(SqlKeyGenerator.ORIGINAL_KEYGEN_CLASS_NAME,
classOf[SimpleKeyGenerator].getName)
+ if (withRecordKey) {
+ props.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key, "id")
+ }
+ props.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key, "dt")
+ // Production sets both whenever record keys are auto-generated
(HoodieCreateRecordUtils), so
+ // without them the auto-record-key delegate fails on the missing property
rather than on the
+ // behaviour under test.
+ props.put(KeyGenUtils.RECORD_KEY_GEN_INSTANT_TIME_CONFIG, "100")
+ props.put(KeyGenUtils.RECORD_KEY_GEN_PARTITION_ID_CONFIG, "1")
+ partitionSchema.foreach(ps => props.put(SqlKeyGenerator.PARTITION_SCHEMA,
ps))
+ new SqlKeyGenerator(props)
+ }
+
+ /** Delegates to a TimestampBasedKeyGenerator rather than a
SimpleKeyGenerator. */
+ private def timestampKeyGenerator(partitionSchema: Option[String]):
SqlKeyGenerator = {
+ val props = new TypedProperties()
+ props.put(SqlKeyGenerator.ORIGINAL_KEYGEN_CLASS_NAME,
classOf[TimestampBasedKeyGenerator].getName)
+ props.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key, "id")
+ props.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key, "dt")
+ props.put(TimestampKeyGeneratorConfig.TIMESTAMP_TYPE_FIELD.key,
"DATE_STRING")
+ props.put(TimestampKeyGeneratorConfig.TIMESTAMP_INPUT_DATE_FORMAT.key,
"yyyy-MM-dd")
+ props.put(TimestampKeyGeneratorConfig.TIMESTAMP_OUTPUT_DATE_FORMAT.key,
"yyyy-MM-dd")
+ partitionSchema.foreach(ps => props.put(SqlKeyGenerator.PARTITION_SCHEMA,
ps))
+ new SqlKeyGenerator(props)
+ }
+
+ /** The partition column is populated; only the record key is missing, as
under a partial update. */
+ private def recordWithoutRecordKey: GenericData.Record = {
+ val record = new GenericData.Record(schema)
+ record.put("amount", 15.0d)
+ record.put("dt", "2026-08-11")
+ record
+ }
+
+ /** A partial-update shape: the partition field is not in the schema at all.
*/
+ private def recordMissingThePartitionField: GenericData.Record = {
+ val record = new GenericData.Record(projected("id", "amount"))
+ record.put("id", 1L)
+ record.put("amount", 15.0d)
+ record
+ }
+
+ /** Runs `f` with the default timezone pinned, since the timestamp arms
format in it. */
+ private def inUtc[T](f: => T): T = {
+ val previousZone = DateTimeZone.getDefault
+ DateTimeZone.setDefault(DateTimeZone.UTC)
+ try f finally DateTimeZone.setDefault(previousZone)
+ }
+
+ @Test
+ def testGetPartitionPathDoesNotRequireRecordKey(): Unit = {
+ // Before the fix this threw HoodieKeyException, because getPartitionPath
delegated to
+ // BaseKeyGenerator#getKey, which builds the whole HoodieKey and so
validates the record key.
+ assertEquals("2026-08-11",
keyGenerator().getPartitionPath(recordWithoutRecordKey))
+ }
+
+ @Test
+ def testGetRecordKeyStillRejectsAMissingRecordKey(): Unit = {
+ // Scope guard: the fix must not weaken record-key validation, only stop
getPartitionPath from
+ // triggering it. A record key that is genuinely required and absent is
still an error.
+ assertThrows(classOf[HoodieKeyException], () =>
keyGenerator().getRecordKey(recordWithoutRecordKey))
+ }
+
+ @Test
+ def testGetPartitionPathAndRecordKeyOnACompleteRecord(): Unit = {
+ val record = recordWithoutRecordKey
+ record.put("id", 1L)
+ assertEquals("2026-08-11", keyGenerator().getPartitionPath(record))
+ assertEquals("1", keyGenerator().getRecordKey(record))
+ }
+
+ /**
+ * Drives convertPartitionPathToSqlType's TimestampType arm, the only arm
that rewrites the value.
+ * The expected string was captured from a run rather than derived.
+ */
+ @Test
+ def testGetPartitionPathConvertsATimestampPartitionValue(): Unit = inUtc {
+ val record = new GenericData.Record(schema)
+ record.put("id", 1L)
+ record.put("dt", timestampMicros)
+ assertEquals("2026-08-11 00%3A00%3A00",
+ keyGenerator(Some("dt timestamp")).getPartitionPath(record))
+ }
+
+ /**
+ * The one shape that legitimately supplies no partition schema is a
non-partitioned table, where
+ * convertPartitionPathToSqlType returns its input untouched. Paired with
the case above on the
+ * same value, so the conversion is shown to be skipped rather than merely
absent.
+ */
+ @Test
+ def testNonPartitionedTableLeavesThePartitionValueUnconverted(): Unit =
inUtc {
+ val record = new GenericData.Record(schema)
+ record.put("id", 1L)
+ record.put("dt", timestampMicros)
+ assertEquals(timestampMicros, keyGenerator(partitionSchema =
None).getPartitionPath(record))
+ }
+
+ /**
+ * A TimestampBasedKeyGenerator delegate does NOT substitute
HUDI_DEFAULT_PARTITION_PATH for an
+ * absent partition field: it formats the epoch instead, so the HUDI-8315
guard in
+ * convertPartitionPathToSqlType never fires for it. Pinned because this
change makes the case
+ * reachable, where previously the record-key exception pre-empted it.
+ *
+ * All three partition-schema spellings are pinned separately below because
they diverge, and the
+ * divergence is the point: a `timestamp` column turns the epoch string into
a bare
+ * NumberFormatException, while a `string` column silently accepts the epoch
partition.
+ */
+ @Test
+ def testTimestampDelegateResolvesAnAbsentPartitionFieldToTheEpoch(): Unit =
inUtc {
+ assertEquals("1970-01-01",
+
timestampKeyGenerator(None).getPartitionPath(recordMissingThePartitionField))
+ }
+
+ @Test
+ def
testTimestampDelegateSilentlyAcceptsTheEpochUnderAStringPartitionSchema(): Unit
= inUtc {
+ assertEquals("1970-01-01",
+ timestampKeyGenerator(Some("dt
string")).getPartitionPath(recordMissingThePartitionField))
+ }
+
+ @Test
+ def testTimestampDelegateThrowsUnderATimestampPartitionSchema(): Unit =
inUtc {
+ // The epoch string the delegate produced is not microseconds, so the
TimestampType arm cannot
+ // parse it. NumberFormatException rather than a Hudi exception is what
the code does today.
+ assertThrows(classOf[NumberFormatException],
+ () => timestampKeyGenerator(Some("dt
timestamp")).getPartitionPath(recordMissingThePartitionField))
+ }
+
+ /**
+ * Without a record-key config KeyGenUtils#isAutoGeneratedRecordKeysEnabled
is true, so the delegate
+ * is wrapped in an AutoRecordGenWrapperKeyGenerator. That wrapper is a
BaseKeyGenerator, so it now
+ * takes the direct arm and getPartitionPath no longer consumes a generated
sequence id as a side
+ * effect of building a HoodieKey.
+ *
+ * The record key is asserted after two partition lookups to pin that
stride: the sequence id is
+ * `instantTime_partitionId_rowId`, so it reads 100_1_0 here and 100_1_2
before the change.
+ * Uniqueness never depended on the stride, but it is now a pinned decision.
+ */
+ @Test
+ def testAutoRecordKeyDelegateDoesNotConsumeSequenceIdsOnPartitionLookups():
Unit = {
+ val record = new GenericData.Record(schema)
+ record.put("amount", 15.0d)
+ record.put("dt", "2026-08-11")
+ val keyGen = keyGenerator(withRecordKey = false)
+ assertEquals("2026-08-11", keyGen.getPartitionPath(record))
+ assertEquals("2026-08-11", keyGen.getPartitionPath(record))
+ assertEquals("100_1_0", keyGen.getRecordKey(record))
+ }
+
+ /**
+ * An unresolvable partition field yields the default partition rather than
an error. That is
+ * KeyGenUtils#getPartitionPath substituting HUDI_DEFAULT_PARTITION_PATH for
a null or absent
+ * value, and it predates this change: getKey called the very same
getPartitionPath, so the
+ * substitution already happened whenever the record key resolved. Pinned
here so the behaviour is
+ * explicit, because the fix does widen when it is observable, see the
sibling test below.
+ *
+ * NOTE this is the behaviour TestSimpleKeyGenerator marks with "TODO this
should throw as well" on
+ * the Avro path, its Row-path twin already asserting HoodieException. If
that TODO is addressed,
+ * this expectation changes with it; the sibling test below is the one that
discriminates the fix.
+ */
+ @Test
+ def testPartitionFieldMissingFromTheSchemaYieldsTheDefaultPartition(): Unit
= {
+ assertEquals(PartitionPathEncodeUtils.DEFAULT_PARTITION_PATH,
+ keyGenerator().getPartitionPath(recordMissingThePartitionField))
+ }
+
+ /**
+ * With BOTH the record key and the partition field unresolvable, the
record-key exception no
+ * longer pre-empts the default-partition substitution. This is the one
behaviour this change
+ * widens, and it is the shape a MOR partial update produces, so it is
stated rather than left to
+ * be discovered: resolving a partition path is not a validity check on the
record.
+ */
+ @Test
+ def testPartitionAndRecordKeyBothMissingYieldTheDefaultPartition(): Unit = {
+ val record = new GenericData.Record(projected("amount"))
+ record.put("amount", 15.0d)
+ assertEquals(PartitionPathEncodeUtils.DEFAULT_PARTITION_PATH,
+ keyGenerator().getPartitionPath(record))
+ }
+}
