voonhous commented on code in PR #19709: URL: https://github.com/apache/hudi/pull/19709#discussion_r3842550277
########## hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestSqlKeyGenerator.scala: ########## @@ -0,0 +1,213 @@ +/* + * 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.{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. + */ +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) + + /** 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] = None, Review Comment: `keyGenerator()` defaults the partition schema to `None`, but production sets `hoodie.sql.partition.schema` unconditionally -- `ProvidesHoodieConfig.scala:74,101,274,425` plus both `MergeIntoHoodieTableCommand` copies. Only line 139 passes one, so seven of the eight tests exit `convertPartitionPathToSqlType` at line 200 and never reach the HUDI-8315 default-partition guard, the size early-out, or hive-style handling -- including the two tests named after default-partition behaviour. Could we default this to `Some("dt string")`, keeping one explicit `None` case for the non-partitioned table? The existing assertions stay green under it. ```suggestion private def keyGenerator(partitionSchema: Option[String] = Some("dt string"), ``` ########## hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestSqlKeyGenerator.scala: ########## @@ -0,0 +1,213 @@ +/* + * 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.{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. + */ +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) + + /** 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] = None, + 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") + // Spark SQL always sets this (see MergeIntoHoodieTableCommand), and it is what drives + // convertPartitionPathToSqlType, so cover it rather than leaving it None. + 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] = None): 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 + } + + @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, which is the only arm that does any + * work: a `dt string` partition schema takes the identity case, so it pins nothing. The value is + * microseconds because that is what the GenericRecord path assumes when + * hoodie.datasource.write.keygenerator.consistent.logical.timestamp.enabled is off. Timezone is + * pinned because the output is formatted in the default zone. + */ + @Test + def testGetPartitionPathConvertsATimestampPartitionValue(): Unit = { + val previousZone = DateTimeZone.getDefault + DateTimeZone.setDefault(DateTimeZone.UTC) + try { + val record = new GenericData.Record(schema) + record.put("id", 1L) + // 2026-08-11 00:00:00 UTC expressed in microseconds. + record.put("dt", String.valueOf(1786406400000000L)) + assertEquals("2026-08-11 00%3A00%3A00", keyGenerator(Some("dt timestamp")).getPartitionPath(record)) + } finally { + DateTimeZone.setDefault(previousZone) + } + } + + /** + * 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. + */ + @Test + def testTimestampDelegateResolvesAnAbsentPartitionFieldToTheEpoch(): Unit = { + val previousZone = DateTimeZone.getDefault + DateTimeZone.setDefault(DateTimeZone.UTC) + try { + val record = new GenericData.Record(projected("id", "amount")) + record.put("id", 1L) + record.put("amount", 15.0d) + assertEquals("1970-01-01", timestampKeyGenerator().getPartitionPath(record)) Review Comment: This pins `1970-01-01` with no partition schema, but production always supplies one, and this delegate diverges under both spellings: with `dt timestamp` the epoch string reaches `_partitionValue.toLong` and throws `NumberFormatException`; with `dt string` it silently writes `1970-01-01`. So the test names the fragile branch and then avoids it. Could we add the `Some("dt timestamp")` and `Some("dt string")` variants here, so the crash and the silent wrong partition are both stated? ########## hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/command/TestSqlKeyGenerator.scala: ########## @@ -0,0 +1,213 @@ +/* + * 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.{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. + */ +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) + + /** 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] = None, + 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") + // Spark SQL always sets this (see MergeIntoHoodieTableCommand), and it is what drives + // convertPartitionPathToSqlType, so cover it rather than leaving it None. + 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] = None): 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 + } + + @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, which is the only arm that does any + * work: a `dt string` partition schema takes the identity case, so it pins nothing. The value is + * microseconds because that is what the GenericRecord path assumes when + * hoodie.datasource.write.keygenerator.consistent.logical.timestamp.enabled is off. Timezone is + * pinned because the output is formatted in the default zone. + */ + @Test + def testGetPartitionPathConvertsATimestampPartitionValue(): Unit = { + val previousZone = DateTimeZone.getDefault + DateTimeZone.setDefault(DateTimeZone.UTC) + try { + val record = new GenericData.Record(schema) + record.put("id", 1L) + // 2026-08-11 00:00:00 UTC expressed in microseconds. + record.put("dt", String.valueOf(1786406400000000L)) + assertEquals("2026-08-11 00%3A00%3A00", keyGenerator(Some("dt timestamp")).getPartitionPath(record)) + } finally { + DateTimeZone.setDefault(previousZone) + } + } + + /** + * 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. + */ + @Test + def testTimestampDelegateResolvesAnAbsentPartitionFieldToTheEpoch(): Unit = { + val previousZone = DateTimeZone.getDefault + DateTimeZone.setDefault(DateTimeZone.UTC) + try { + val record = new GenericData.Record(projected("id", "amount")) + record.put("id", 1L) + record.put("amount", 15.0d) + assertEquals("1970-01-01", timestampKeyGenerator().getPartitionPath(record)) + } finally { + DateTimeZone.setDefault(previousZone) + } + } + + /** + * 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. Uniqueness does not depend on the stride, but the change is + * pinned here rather than left as an unexercised side effect. + */ + @Test + def testAutoRecordKeyDelegateStillResolvesThePartitionPath(): 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)) Review Comment: This is red on `2c14ff2` for a harness reason rather than the production one. The base failure is `IllegalArgumentException: Property _hoodie.record.key.gen.partition.id not found`, but production always sets that prop (`HoodieCreateRecordUtils.scala:130,181`), so pre-fix production consumed a rowId silently instead of throwing. The stride change the scaladoc describes is therefore unpinned. Could we set `RECORD_KEY_GEN_INSTANT_TIME_CONFIG` and `RECORD_KEY_GEN_PARTITION_ID_CONFIG` in `keyGenerator()` (as `TestCreateKeyGeneratorByTypeWithFactory.java:114-115` does) and assert `getRecordKey` returns `100_1_0`? On base that yields `100_1_2`. ########## hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/SqlKeyGenerator.scala: ########## @@ -77,7 +77,13 @@ class SqlKeyGenerator(props: TypedProperties) extends BuiltinKeyGenerator(props) override def getRecordKey(record: GenericRecord): String = originalKeyGen.map { - _.getKey(record).getRecordKey + // Mirror of getPartitionPath below: resolve the record key alone where the generator exposes + // it, so neither accessor pays for the other half. Going through getKey would also compute the + // partition path and discard it, and would let a partition-side failure surface from a + // record-key lookup: TimestampBasedAvroKeyGenerator raises HoodieKeyGeneratorException + // independently of the key. + case baseKeyGen: BaseKeyGenerator => baseKeyGen.getRecordKey(record) Review Comment: Nothing discriminates this arm: `testGetRecordKeyStillRejectsAMissingRecordKey` passes identically on `2c14ff2` and here, so the behaviour change this comment describes is untested. A `TimestampBasedKeyGenerator` delegate with `dt = "not-a-date"` does discriminate it -- base throws `HoodieKeyGeneratorException: Unable to parse input partition field`, head returns `"1"`. Could we add that case, or fold it into the follow-up discussed above? -- 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]
