voonhous commented on code in PR #19709:
URL: https://github.com/apache/hudi/pull/19709#discussion_r3834908532
##########
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)
Review Comment:
This looks blocking on merge order rather than on the code. The body says
fixing this only "exposes" an `UnresolvedUnionException`, i.e. the intermediate
state is loud. That holds only when the partial-update column types disagree
with the tail of the write schema; where they agree the record serializes fine
and a wrong row is written.
Could we land this together with (or after) #19713, and say `Addresses
#19708` rather than `Closes #19708`, so merging does not auto-close an issue
whose repro still fails?
<details><summary>Repro, driving <code>JoinedGenericRecord</code> +
<code>HoodieAvroUtils.avroToBytes</code> directly</summary>
```
Table (id bigint, name string, city string, ts bigint, dt string), UPDATE
SET name
prependMetaFields infers metaFieldSize = 9 (correct: 5)
avroToBytes: OK, 98 bytes -- no exception
-> name: null (the assigned value shifted into dt)
-> _hoodie_partition_path: __HIVE_DEFAULT_PARTITION__
```
On master the same statement threw `HoodieKeyException` from this line.
</details>
##########
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.
Review Comment:
The "no longer pre-empted" note in the body covers the
`__HIVE_DEFAULT_PARTITION__` spelling, but a `TimestampBasedKeyGenerator`
delegate substitutes the epoch instead, so the HUDI-8315 guard at line 170
never fires:
```
TimestampBased (DATE_STRING) -> '1970-01-01' guard hit =
false
Simple -> '__HIVE_DEFAULT_PARTITION__' guard hit =
true
```
With a `TIMESTAMP` partition schema `_partitionValue.toLong` then throws a
bare `NumberFormatException`; with a non-timestamp one the row silently takes
the `1970-01-01` partition. Both were previously pre-empted by the
`HoodieKeyException`. Could we add a `TimestampBasedKeyGenerator` case to
`TestSqlKeyGenerator` pinning what a missing timestamp partition field now
resolves to?
##########
hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestSqlKeyGenerator.scala:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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 [[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.
+ */
+class TestSqlKeyGenerator {
+
+ private val schema = new Schema.Parser().parse(
+ s"""
+ |{
+ | "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)
+
+ private def keyGenerator(partitionSchema: Option[String] = None):
SqlKeyGenerator = {
+ 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")
+ // Spark SQL always sets this (see MergeIntoHoodieTableCommand), and it is
what makes
+ // convertPartitionPathToSqlType do any work, so cover it rather than
leaving it None.
+ partitionSchema.foreach(schema =>
props.put(SqlKeyGenerator.PARTITION_SCHEMA, schema))
+ 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
+ }
+
+ @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))
+ }
+
+ @Test
+ def testGetPartitionPathResolvesThroughTheSqlPartitionSchema(): Unit = {
+ val record = recordWithoutRecordKey
+ record.put("id", 1L)
+ assertEquals("2026-08-11", keyGenerator(Some("dt
string")).getPartitionPath(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.
+ */
+ @Test
+ def testPartitionFieldMissingFromTheSchemaYieldsTheDefaultPartition(): Unit
= {
+ val partialSchema = new Schema.Parser().parse(
+ s"""
+ |{
+ | "type": "record",
+ | "name": "test_record",
+ | "fields": [
+ | {"name": "id", "type": ["null", "long"], "default": null},
+ | {"name": "amount", "type": ["null", "double"], "default": null}
+ | ]
+ |}
+ """.stripMargin)
+ val record = new GenericData.Record(partialSchema)
+ record.put("id", 1L)
+ record.put("amount", 15.0d)
+ assertEquals(PartitionPathEncodeUtils.DEFAULT_PARTITION_PATH,
keyGenerator().getPartitionPath(record))
Review Comment:
nit, feel free to ignore: this pins the exact behaviour the keygen owners
have a TODO to change. `TestSimpleKeyGenerator.java:127` marks the Avro path
with `// TODO this should throw as well`, and its Row-path twin already asserts
`HoodieException` for a wrong partition field.
The sibling test below already covers the substitution and is the one that
discriminates the fix. Could we drop this case, or add a pointer to that TODO
so both do not have to be unwound when it is fixed?
##########
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 {
Review Comment:
nit, feel free to ignore: this leaves the two sides asymmetric.
`getRecordKey(GenericRecord)` at lines 78-83 still goes through `getKey`, so it
computes the partition path and discards it on every Avro record, and a
record-key lookup can still fail on a partition-side error
(`TimestampBasedAvroKeyGenerator` throws `HoodieKeyGeneratorException`
independently of the key).
Could `getRecordKey` take the mirror arm, `case baseKeyGen: BaseKeyGenerator
=> baseKeyGen.getRecordKey(record)`, so each method does one lookup?
##########
hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestSqlKeyGenerator.scala:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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 [[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.
+ */
+class TestSqlKeyGenerator {
+
+ private val schema = new Schema.Parser().parse(
+ s"""
+ |{
+ | "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)
+
+ private def keyGenerator(partitionSchema: Option[String] = None):
SqlKeyGenerator = {
+ val props = new TypedProperties()
+ props.put(SqlKeyGenerator.ORIGINAL_KEYGEN_CLASS_NAME,
classOf[SimpleKeyGenerator].getName)
+ props.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key, "id")
Review Comment:
Setting the record key here makes `autoRecordKeyGen` always false, so the
`AutoRecordGenWrapperKeyGenerator` path is never exercised -- and this change
does alter it. The wrapper is a `BaseKeyGenerator`, so it now takes the new arm
and `getPartitionPath` no longer consumes a sequence id (stride 2 becomes 1).
It is reachable: `KeyGenUtils.isAutoGeneratedRecordKeysEnabled` returns true on
an *empty-string* record key, commented "spark-sql sets record key config to
empty string for update".
The numbering itself looks harmless (uniqueness only needs `(instantTime,
partitionId)`). Could we add one case that omits `RECORDKEY_FIELD_NAME`, so it
is a pinned decision rather than a side effect?
##########
hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestSqlKeyGenerator.scala:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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 [[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.
+ */
+class TestSqlKeyGenerator {
+
+ private val schema = new Schema.Parser().parse(
+ s"""
+ |{
+ | "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)
+
+ private def keyGenerator(partitionSchema: Option[String] = None):
SqlKeyGenerator = {
+ 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")
+ // Spark SQL always sets this (see MergeIntoHoodieTableCommand), and it is
what makes
+ // convertPartitionPathToSqlType do any work, so cover it rather than
leaving it None.
+ partitionSchema.foreach(schema =>
props.put(SqlKeyGenerator.PARTITION_SCHEMA, schema))
+ 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
+ }
+
+ @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))
+ }
+
+ @Test
+ def testGetPartitionPathResolvesThroughTheSqlPartitionSchema(): Unit = {
+ val record = recordWithoutRecordKey
+ record.put("id", 1L)
+ assertEquals("2026-08-11", keyGenerator(Some("dt
string")).getPartitionPath(record))
Review Comment:
This case pins nothing about the SQL partition schema: `dt string` takes the
`case _ => partitionValue` identity arm, so it is
`testGetPartitionPathAndRecordKeyOnACompleteRecord`'s first assertion plus an
inert config. Deleting the entire `convertPartitionPathToSqlType(...)` call
from `getPartitionPath(GenericRecord)` leaves all 6 tests green, so the intent
stated at lines 59-60 is not met.
Could we switch this to a `TimestampType` partition schema (a micros-valued
`dt`, with a pinned timezone since the output is TZ-dependent), which would
also cover the arm the change newly feeds?
##########
hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestSqlKeyGenerator.scala:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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 [[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.
+ */
+class TestSqlKeyGenerator {
+
+ private val schema = new Schema.Parser().parse(
+ s"""
Review Comment:
nit, feel free to ignore: `s"""` here (and at lines 112 and 137) has no
interpolation, so the `s` prefix can go. The two partial schemas are also
near-copies of this one -- deriving them by filtering this `schema` val's
fields would drop roughly 20 lines. Purely cosmetic, nothing fires on it.
##########
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 = {
Review Comment:
`MergeIntoKeyGenerator` overrides this method and is the generator used for
a primary-key-less table (`MergeIntoHoodieTableCommand.scala:780-785`). It
reads `record.get(PARTITION_PATH_META_FIELD_ORD)` (ordinal 3) unconditionally,
which throws `ArrayIndexOutOfBoundsException` on the same 2-field partial
record before `super.getPartitionPath` ever runs, so the fix does not reach
that spelling.
I did not confirm that keyless + MOR + global index + partial update is
actually configurable. If it is, could `MergeIntoKeyGenerator.getPartitionPath`
check `record.getSchema.getFields.size` before reading the ordinal?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
