the-other-tim-brown commented on code in PR #13656:
URL: https://github.com/apache/hudi/pull/13656#discussion_r2282781148
##########
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:
We need to be able to support virtual keys with the user provided key
generator as well, right?
--
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]
