linliu-code commented on code in PR #19709: URL: https://github.com/apache/hudi/pull/19709#discussion_r3840211141
########## 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: Done, and your deletion check is right: I removed the `convertPartitionPathToSqlType(...)` call and all 6 passed, so that case was pinning an inert config. Replaced with a `dt timestamp` schema over micros `1786406400000000L`, pinning `2026-08-11 00%3A00%3A00`, which drives the `TimestampType` arm and `escapePathName`. The expected value is captured from a run rather than hand-derived (my first attempt at the arithmetic landed four days off, which is exactly the failure mode you flagged). `DateTimeZone` is pinned to UTC and restored in a `finally`. ########## 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: Done. Added `testTimestampDelegateResolvesAnAbsentPartitionFieldToTheEpoch`: a `TimestampBasedKeyGenerator` delegate (DATE_STRING, `yyyy-MM-dd`) against a record whose schema omits `dt`, pinning `1970-01-01`. That confirms your table, the HUDI-8315 guard does not fire for this generator, so there is no default-partition spelling to rely on. What I did not pin is the second half of your comment: the `TIMESTAMP` partition-schema case where `_partitionValue.toLong` then throws a bare `NumberFormatException`. Say if you want that pinned too and I will add it; I left it out because it asserts on an exception type that looks like it should itself be a wrapped error rather than settled behaviour. ########## 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: Done. Added `testAutoRecordKeyDelegateStillResolvesThePartitionPath`, which omits `RECORDKEY_FIELD_NAME` so `isAutoGeneratedRecordKeysEnabled` is true and the `AutoRecordGenWrapperKeyGenerator` is the delegate, taking the new arm. To be precise about what it pins: it resolves the partition path twice on the same record and asserts both, so it pins that the wrapper still resolves correctly and repeatedly through the direct arm. It does not observe the sequence-id stride itself, which is not exposed on this path. So the coverage is "the wrapper is exercised and its partition resolution is stable", not "the numbering is asserted". As you say, uniqueness only needs `(instantTime, partitionId)`, so I did not try to assert the stride. ########## 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: Taken. `getRecordKey(GenericRecord)` now takes the mirror arm, so each accessor does one lookup. Flagging the tradeoff since you marked this optional: it widens the diff past the reported defect, and it does change behaviour for a `TimestampBasedAvroKeyGenerator` delegate, where a partition-side `HoodieKeyGeneratorException` no longer surfaces from a record-key lookup. I think that is the correct direction and the asymmetry is real, but if you would rather this PR stay scoped to the one method, say so and I will split it into a follow-up. ########## 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: Kept it, with the pointer rather than the deletion. Added a NOTE naming `TestSimpleKeyGenerator.java:127` and its Row-path twin, and stating that the sibling case below is the one that discriminates the fix, so whoever addresses that TODO finds both expectations from either end. Dropping it is also fine by me if you would rather the file only carry the discriminating case; I kept it because the substitution predates this change and I wanted the pre-existing half stated separately from the half this change widens. ########## 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: Done. Dropped the three `s` prefixes, and added a `projected(fieldNames: String*)` helper that derives the partial schemas by filtering the base `schema` val, which removed both near-copies. -- 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]
