linliu-code commented on code in PR #13656:
URL: https://github.com/apache/hudi/pull/13656#discussion_r2271832033
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala:
##########
@@ -2065,6 +2065,66 @@ class TestCOWDataSource extends
HoodieSparkClientTestBase with ScalaAssertionSup
assertEquals(count, 0)
}
+ @ParameterizedTest
+ @MethodSource(Array("provideParamsForKeyGenTest"))
+ def testUserProvidedKeyGeneratorClass(keyGenClass: String,
+ keyGenType: String): Unit = {
+ val recordType = HoodieRecordType.AVRO
+ val (writeOpts, readOpts) = getWriterReaderOpts(recordType,
+ CommonOptionUtils.commonOpts ++ Map(
+ HoodieWriteConfig.KEYGENERATOR_CLASS_NAME.key -> keyGenClass,
+ HoodieWriteConfig.KEYGENERATOR_TYPE.key -> keyGenType,
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition"))
+ val expectedKeyGenType = if (StringUtils.isNullOrEmpty(keyGenClass)) {
+ KeyGeneratorType.USER_PROVIDED.name
+ } else {
+ KeyGeneratorType.fromClassName(keyGenClass).name
+ }
+
+ // Insert.
+ val records = recordsToStrings(dataGen.generateInserts("000",
10)).asScala.toList
+ val inputDF = spark.read.json(spark.sparkContext.parallelize(records, 2))
+
+ // Make sure the initial configurations set properly for USER_PROVIDED
type.
+ if (StringUtils.isNullOrEmpty(keyGenClass) &&
keyGenType.equals(KeyGeneratorType.USER_PROVIDED.name)) {
+ assertThrows(classOf[IllegalArgumentException])({
+ writeToHudi(writeOpts, inputDF)
+ })
+ // scalastyle:off return
+ return
+ // scalastyle:on return
+ } else { // Otherwise, the data is properly ingested.
+ writeToHudi(writeOpts, inputDF)
+ }
+ var actualDF = spark.read.format("hudi").options(readOpts).load(basePath)
+ val inputKeyDF = inputDF.select("_row_key").sort("_row_key")
+ var actualKeyDF = actualDF.select("_row_key").sort("_row_key")
+ assertTrue(inputKeyDF.except(actualKeyDF).isEmpty &&
actualKeyDF.except(inputKeyDF).isEmpty)
+ val metaClient = getHoodieMetaClient(storageConf, basePath)
+ val actualKeyGenType = metaClient.getTableConfig.getKeyGeneratorType
+ assertEquals(expectedKeyGenType, actualKeyGenType)
+ // For USER_PROVIDED type, the class should exist in table config.
+ if (KeyGeneratorType.USER_PROVIDED.name == actualKeyGenType) {
+ assertEquals(keyGenClass,
metaClient.getTableConfig.getKeyGeneratorClassName)
+ }
+
+ // First update.
+ val firstUpdate =
recordsToStrings(dataGen.generateUpdatesForAllRecords("001")).asScala.toList
+ val firstUpdateDF =
spark.read.json(spark.sparkContext.parallelize(firstUpdate, 2))
+ writeToHudi(writeOpts, firstUpdateDF)
+ val newReadOpts = readOpts ++
Map(HoodieTableConfig.POPULATE_META_FIELDS.key -> "false")
Review Comment:
Only certain types of key generator supports virtual keys during writes.
So I did not enable it for writes to avoid complex test cases, but only for
readers.
--
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]
