voonhous commented on code in PR #19657:
URL: https://github.com/apache/hudi/pull/19657#discussion_r3941353472
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala:
##########
@@ -1450,6 +1451,74 @@ class TestCOWDataSource extends
HoodieSparkClientTestBase with ScalaAssertionSup
assertTrue(recordsReadDF.filter(col("_hoodie_partition_path") =!=
udf_date_format(col("current_ts"))).count() == 0)
}
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieRecordType], names = Array("AVRO",
"SPARK"))
+ def testTimestampBasedKeyGeneratorWithVariousConfigurations(recordType:
HoodieRecordType) {
+ val (writeOpts, readOpts) =
getWriterReaderOptsLessPartitionPath(recordType)
+
+ val records = recordsToStrings(dataGen.generateInserts("000",
100)).asScala.toList
+ val inputDF = spark.read.json(spark.sparkContext.parallelize(records, 2))
+ .withColumn("current_ts_micros", col("current_ts") * 1000)
+ .withColumn("current_date_string",
+ date_format((col("current_ts") / 1000).cast("timestamp"), "yyyy-MM-dd
HH:mm:ss"))
+ .withColumn("current_ts_hours", (col("current_ts") /
3600000).cast("long"))
+
+ case class TestCase(partitionCol: String, tsType: String, outFmt: String,
+ extraOpts: Map[String, String] = Map.empty,
+ expectedPartitionUdf:
org.apache.spark.sql.expressions.UserDefinedFunction)
+
+ def runTestCase(tc: TestCase): Unit = {
+ val writer = tc.extraOpts.foldLeft(
+ inputDF.write.format("hudi")
+ .options(writeOpts)
+ .option(KEYGENERATOR_CLASS_NAME.key(),
classOf[TimestampBasedKeyGenerator].getName)
+ .mode(SaveMode.Overwrite)
+ ) { case (w, (k, v)) => w.option(k, v) }
+ writer.partitionBy(tc.partitionCol)
+ .option(TIMESTAMP_TYPE_FIELD.key, tc.tsType)
+ .option(TIMESTAMP_OUTPUT_DATE_FORMAT.key, tc.outFmt)
+ .save(basePath)
+ val readDF =
spark.read.format("org.apache.hudi").options(readOpts).load(basePath)
+ assertTrue(readDF.filter(col("_hoodie_partition_path") =!=
tc.expectedPartitionUdf(col(tc.partitionCol))).count() == 0)
+ }
+
+ val outputDateFmt = "yyyy-MM-dd HH"
+ // Joda's DateTimeZone.forID does not recognise "GMT+08:00".
HoodieDateTimeParser resolves the
+ // configured id via java.util.TimeZone, so the expected values are built
the same way.
+ val tzId = "GMT+08:00"
+
+ // Test 1: EPOCHMILLISECONDS with timezone GMT+08:00
+ val udfMillisTz = udf((millis: Long) => {
+ val zone = DateTimeZone.forTimeZone(TimeZone.getTimeZone(tzId))
+ new DateTime(millis,
zone).toString(DateTimeFormat.forPattern(outputDateFmt).withZone(zone))
+ })
+ runTestCase(TestCase("current_ts", "EPOCHMILLISECONDS", outputDateFmt,
+ Map(TIMESTAMP_TIMEZONE_FORMAT.key -> tzId), udfMillisTz))
+
+ // Test 2: EPOCHMICROSECONDS (no timezone configured, so the key generator
uses the JVM default)
+ val udfMicros = udf((micros: Long) =>
+ new DateTime(micros /
1000).toString(DateTimeFormat.forPattern(outputDateFmt)))
+ runTestCase(TestCase("current_ts_micros", "EPOCHMICROSECONDS",
outputDateFmt,
+ expectedPartitionUdf = udfMicros))
+
+ // Test 3: DATE_STRING with timezone
+ val dateStrInFmt = "yyyy-MM-dd HH:mm:ss"
+ val udfDateStrTz = udf((s: String) => {
+ val zone = DateTimeZone.forTimeZone(TimeZone.getTimeZone(tzId))
+ DateTime.parse(s, DateTimeFormat.forPattern(dateStrInFmt).withZone(zone))
+ .toString(DateTimeFormat.forPattern(outputDateFmt).withZone(zone))
+ })
+ runTestCase(TestCase("current_date_string", "DATE_STRING", outputDateFmt,
+ Map(TIMESTAMP_INPUT_DATE_FORMAT.key -> dateStrInFmt,
+ TIMESTAMP_TIMEZONE_FORMAT.key -> tzId), udfDateStrTz))
Review Comment:
**major:** This case cannot detect a timezone bug. The output format is a
prefix of the input format, so parse-in-Z then format-in-Z is the identity for
any Z. Running the key generator on `2009-02-14 07:31:30` returns `2009-02-14
07` for `GMT+08:00`, for `UTC`, for `America/New_York`, and with
`hoodie.keygen.timebased.timezone` removed entirely.
Could we set `hoodie.keygen.timebased.input.timezone` and `.output.timezone`
to different values instead? in=UTC/out=GMT+08:00 gives `2009-02-14 15` and the
reverse gives `2009-02-13 23`, so the assertion would fail if the zones were
dropped. That pair is also the non-deprecated one - `TIMESTAMP_TIMEZONE_FORMAT`
is marked `@Deprecated` at `TimestampKeyGeneratorConfig.java:93`.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala:
##########
@@ -1450,6 +1451,74 @@ class TestCOWDataSource extends
HoodieSparkClientTestBase with ScalaAssertionSup
assertTrue(recordsReadDF.filter(col("_hoodie_partition_path") =!=
udf_date_format(col("current_ts"))).count() == 0)
}
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieRecordType], names = Array("AVRO",
"SPARK"))
+ def testTimestampBasedKeyGeneratorWithVariousConfigurations(recordType:
HoodieRecordType) {
+ val (writeOpts, readOpts) =
getWriterReaderOptsLessPartitionPath(recordType)
+
+ val records = recordsToStrings(dataGen.generateInserts("000",
100)).asScala.toList
+ val inputDF = spark.read.json(spark.sparkContext.parallelize(records, 2))
+ .withColumn("current_ts_micros", col("current_ts") * 1000)
+ .withColumn("current_date_string",
+ date_format((col("current_ts") / 1000).cast("timestamp"), "yyyy-MM-dd
HH:mm:ss"))
+ .withColumn("current_ts_hours", (col("current_ts") /
3600000).cast("long"))
+
+ case class TestCase(partitionCol: String, tsType: String, outFmt: String,
+ extraOpts: Map[String, String] = Map.empty,
+ expectedPartitionUdf:
org.apache.spark.sql.expressions.UserDefinedFunction)
+
+ def runTestCase(tc: TestCase): Unit = {
+ val writer = tc.extraOpts.foldLeft(
+ inputDF.write.format("hudi")
+ .options(writeOpts)
+ .option(KEYGENERATOR_CLASS_NAME.key(),
classOf[TimestampBasedKeyGenerator].getName)
+ .mode(SaveMode.Overwrite)
+ ) { case (w, (k, v)) => w.option(k, v) }
+ writer.partitionBy(tc.partitionCol)
+ .option(TIMESTAMP_TYPE_FIELD.key, tc.tsType)
+ .option(TIMESTAMP_OUTPUT_DATE_FORMAT.key, tc.outFmt)
+ .save(basePath)
+ val readDF =
spark.read.format("org.apache.hudi").options(readOpts).load(basePath)
+ assertTrue(readDF.filter(col("_hoodie_partition_path") =!=
tc.expectedPartitionUdf(col(tc.partitionCol))).count() == 0)
+ }
+
+ val outputDateFmt = "yyyy-MM-dd HH"
+ // Joda's DateTimeZone.forID does not recognise "GMT+08:00".
HoodieDateTimeParser resolves the
+ // configured id via java.util.TimeZone, so the expected values are built
the same way.
+ val tzId = "GMT+08:00"
+
+ // Test 1: EPOCHMILLISECONDS with timezone GMT+08:00
+ val udfMillisTz = udf((millis: Long) => {
+ val zone = DateTimeZone.forTimeZone(TimeZone.getTimeZone(tzId))
+ new DateTime(millis,
zone).toString(DateTimeFormat.forPattern(outputDateFmt).withZone(zone))
+ })
+ runTestCase(TestCase("current_ts", "EPOCHMILLISECONDS", outputDateFmt,
+ Map(TIMESTAMP_TIMEZONE_FORMAT.key -> tzId), udfMillisTz))
+
+ // Test 2: EPOCHMICROSECONDS (no timezone configured, so the key generator
uses the JVM default)
+ val udfMicros = udf((micros: Long) =>
+ new DateTime(micros /
1000).toString(DateTimeFormat.forPattern(outputDateFmt)))
+ runTestCase(TestCase("current_ts_micros", "EPOCHMICROSECONDS",
outputDateFmt,
+ expectedPartitionUdf = udfMicros))
+
+ // Test 3: DATE_STRING with timezone
+ val dateStrInFmt = "yyyy-MM-dd HH:mm:ss"
+ val udfDateStrTz = udf((s: String) => {
+ val zone = DateTimeZone.forTimeZone(TimeZone.getTimeZone(tzId))
+ DateTime.parse(s, DateTimeFormat.forPattern(dateStrInFmt).withZone(zone))
+ .toString(DateTimeFormat.forPattern(outputDateFmt).withZone(zone))
+ })
+ runTestCase(TestCase("current_date_string", "DATE_STRING", outputDateFmt,
+ Map(TIMESTAMP_INPUT_DATE_FORMAT.key -> dateStrInFmt,
+ TIMESTAMP_TIMEZONE_FORMAT.key -> tzId), udfDateStrTz))
+
+ // Test 4: SCALAR with hours (no timezone configured, so the key generator
uses the JVM default)
+ val udfScalarHours = udf((hours: Long) =>
+ new
DateTime(TimeUnit.HOURS.toMillis(hours)).toString(DateTimeFormat.forPattern(outputDateFmt)))
+ runTestCase(TestCase("current_ts_hours", "SCALAR", outputDateFmt,
+ Map(INPUT_TIME_UNIT.key -> "hours"), udfScalarHours))
Review Comment:
**major:** `hours` is one of the few time units that survives a Turkish
locale, so this case steps around a real bug rather than onto it.
`TimestampBasedAvroKeyGenerator.java:106` calls `timeUnitStr.toUpperCase()`
with no `Locale`, so under `-Duser.language=tr` a lowercase `minutes`,
`microseconds` or `milliseconds` fails `TimeUnit.valueOf` because `i`
uppercases to a dotted capital I. `hours`, `days` and `seconds` have no `i` and
pass.
This PR's own base commit a7deb61f7426 (#19835) fixed the same bug class
with `Locale.ROOT`. Could this case use `microseconds` and add
`toUpperCase(Locale.ROOT)` there, or would you rather keep the production fix
out of a test-only PR?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala:
##########
@@ -1450,6 +1451,74 @@ class TestCOWDataSource extends
HoodieSparkClientTestBase with ScalaAssertionSup
assertTrue(recordsReadDF.filter(col("_hoodie_partition_path") =!=
udf_date_format(col("current_ts"))).count() == 0)
}
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieRecordType], names = Array("AVRO",
"SPARK"))
+ def testTimestampBasedKeyGeneratorWithVariousConfigurations(recordType:
HoodieRecordType) {
+ val (writeOpts, readOpts) =
getWriterReaderOptsLessPartitionPath(recordType)
+
+ val records = recordsToStrings(dataGen.generateInserts("000",
100)).asScala.toList
+ val inputDF = spark.read.json(spark.sparkContext.parallelize(records, 2))
+ .withColumn("current_ts_micros", col("current_ts") * 1000)
+ .withColumn("current_date_string",
+ date_format((col("current_ts") / 1000).cast("timestamp"), "yyyy-MM-dd
HH:mm:ss"))
+ .withColumn("current_ts_hours", (col("current_ts") /
3600000).cast("long"))
+
+ case class TestCase(partitionCol: String, tsType: String, outFmt: String,
+ extraOpts: Map[String, String] = Map.empty,
+ expectedPartitionUdf:
org.apache.spark.sql.expressions.UserDefinedFunction)
+
+ def runTestCase(tc: TestCase): Unit = {
+ val writer = tc.extraOpts.foldLeft(
+ inputDF.write.format("hudi")
+ .options(writeOpts)
+ .option(KEYGENERATOR_CLASS_NAME.key(),
classOf[TimestampBasedKeyGenerator].getName)
+ .mode(SaveMode.Overwrite)
+ ) { case (w, (k, v)) => w.option(k, v) }
+ writer.partitionBy(tc.partitionCol)
+ .option(TIMESTAMP_TYPE_FIELD.key, tc.tsType)
+ .option(TIMESTAMP_OUTPUT_DATE_FORMAT.key, tc.outFmt)
+ .save(basePath)
+ val readDF =
spark.read.format("org.apache.hudi").options(readOpts).load(basePath)
+ assertTrue(readDF.filter(col("_hoodie_partition_path") =!=
tc.expectedPartitionUdf(col(tc.partitionCol))).count() == 0)
+ }
+
+ val outputDateFmt = "yyyy-MM-dd HH"
+ // Joda's DateTimeZone.forID does not recognise "GMT+08:00".
HoodieDateTimeParser resolves the
+ // configured id via java.util.TimeZone, so the expected values are built
the same way.
+ val tzId = "GMT+08:00"
+
+ // Test 1: EPOCHMILLISECONDS with timezone GMT+08:00
+ val udfMillisTz = udf((millis: Long) => {
+ val zone = DateTimeZone.forTimeZone(TimeZone.getTimeZone(tzId))
+ new DateTime(millis,
zone).toString(DateTimeFormat.forPattern(outputDateFmt).withZone(zone))
+ })
+ runTestCase(TestCase("current_ts", "EPOCHMILLISECONDS", outputDateFmt,
+ Map(TIMESTAMP_TIMEZONE_FORMAT.key -> tzId), udfMillisTz))
+
+ // Test 2: EPOCHMICROSECONDS (no timezone configured, so the key generator
uses the JVM default)
+ val udfMicros = udf((micros: Long) =>
+ new DateTime(micros /
1000).toString(DateTimeFormat.forPattern(outputDateFmt)))
+ runTestCase(TestCase("current_ts_micros", "EPOCHMICROSECONDS",
outputDateFmt,
+ expectedPartitionUdf = udfMicros))
+
+ // Test 3: DATE_STRING with timezone
Review Comment:
**minor:** not blocking. `testPartitionColumnsProperHandling` (line 1696 in
this file) already covers DATE_STRING with `TIMESTAMP_INPUT_DATE_FORMAT` and
`TIMESTAMP_TIMEZONE_FORMAT=GMT+8:00` end to end over AVRO and SPARK, asserting
exact `_hoodie_partition_path` values, and additionally exercises
`EXTRACT_PARTITION_VALUES_FROM_PARTITION_PATH`.
Could we drop this case, or repoint it at the input/output timezone split so
it covers something that one does not?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala:
##########
@@ -1450,6 +1451,74 @@ class TestCOWDataSource extends
HoodieSparkClientTestBase with ScalaAssertionSup
assertTrue(recordsReadDF.filter(col("_hoodie_partition_path") =!=
udf_date_format(col("current_ts"))).count() == 0)
}
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieRecordType], names = Array("AVRO",
"SPARK"))
+ def testTimestampBasedKeyGeneratorWithVariousConfigurations(recordType:
HoodieRecordType) {
+ val (writeOpts, readOpts) =
getWriterReaderOptsLessPartitionPath(recordType)
+
+ val records = recordsToStrings(dataGen.generateInserts("000",
100)).asScala.toList
+ val inputDF = spark.read.json(spark.sparkContext.parallelize(records, 2))
+ .withColumn("current_ts_micros", col("current_ts") * 1000)
+ .withColumn("current_date_string",
+ date_format((col("current_ts") / 1000).cast("timestamp"), "yyyy-MM-dd
HH:mm:ss"))
+ .withColumn("current_ts_hours", (col("current_ts") /
3600000).cast("long"))
+
+ case class TestCase(partitionCol: String, tsType: String, outFmt: String,
+ extraOpts: Map[String, String] = Map.empty,
+ expectedPartitionUdf:
org.apache.spark.sql.expressions.UserDefinedFunction)
+
+ def runTestCase(tc: TestCase): Unit = {
+ val writer = tc.extraOpts.foldLeft(
+ inputDF.write.format("hudi")
+ .options(writeOpts)
+ .option(KEYGENERATOR_CLASS_NAME.key(),
classOf[TimestampBasedKeyGenerator].getName)
+ .mode(SaveMode.Overwrite)
+ ) { case (w, (k, v)) => w.option(k, v) }
+ writer.partitionBy(tc.partitionCol)
+ .option(TIMESTAMP_TYPE_FIELD.key, tc.tsType)
+ .option(TIMESTAMP_OUTPUT_DATE_FORMAT.key, tc.outFmt)
+ .save(basePath)
+ val readDF =
spark.read.format("org.apache.hudi").options(readOpts).load(basePath)
+ assertTrue(readDF.filter(col("_hoodie_partition_path") =!=
tc.expectedPartitionUdf(col(tc.partitionCol))).count() == 0)
Review Comment:
**minor:** not blocking. This also passes when the read returns no rows, and
`=!=` yields NULL (which `filter` drops) if the UDF ever returns null, so the
check cannot fail on an empty result. It also names no sub-case: all four
report as the same method.
Could we assert the row count and label the case?
```suggestion
val mismatches = readDF.filter(col("_hoodie_partition_path") =!=
tc.expectedPartitionUdf(col(tc.partitionCol))).count()
assertEquals(100L, readDF.count(), s"unexpected row count for
${tc.tsType}")
assertEquals(0L, mismatches, s"partition path mismatch for
${tc.tsType}")
```
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala:
##########
@@ -1450,6 +1451,74 @@ class TestCOWDataSource extends
HoodieSparkClientTestBase with ScalaAssertionSup
assertTrue(recordsReadDF.filter(col("_hoodie_partition_path") =!=
udf_date_format(col("current_ts"))).count() == 0)
}
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieRecordType], names = Array("AVRO",
"SPARK"))
+ def testTimestampBasedKeyGeneratorWithVariousConfigurations(recordType:
HoodieRecordType) {
Review Comment:
**minor:** not blocking. This adds 8 full table delete, create, write and
read cycles (2 record types x 4 sub-cases); `SaveMode.Overwrite` deletes the
table path on each one (`HoodieSparkSqlWriter.scala:903-909`). `current_ts`
spans 3 days (`HoodieTestDataGenerator.java:1499-1505`), so at `yyyy-MM-dd HH`
each write lands 100 records across up to 72 partitions. `TestCOWDataSource`
already carries four `TimestampBasedKeyGenerator` tests (lines 715, 1355, 1439,
1696).
Could we keep only the sub-cases that survive the other threads?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala:
##########
@@ -1450,6 +1451,74 @@ class TestCOWDataSource extends
HoodieSparkClientTestBase with ScalaAssertionSup
assertTrue(recordsReadDF.filter(col("_hoodie_partition_path") =!=
udf_date_format(col("current_ts"))).count() == 0)
}
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieRecordType], names = Array("AVRO",
"SPARK"))
+ def testTimestampBasedKeyGeneratorWithVariousConfigurations(recordType:
HoodieRecordType) {
+ val (writeOpts, readOpts) =
getWriterReaderOptsLessPartitionPath(recordType)
+
+ val records = recordsToStrings(dataGen.generateInserts("000",
100)).asScala.toList
+ val inputDF = spark.read.json(spark.sparkContext.parallelize(records, 2))
+ .withColumn("current_ts_micros", col("current_ts") * 1000)
+ .withColumn("current_date_string",
+ date_format((col("current_ts") / 1000).cast("timestamp"), "yyyy-MM-dd
HH:mm:ss"))
+ .withColumn("current_ts_hours", (col("current_ts") /
3600000).cast("long"))
+
+ case class TestCase(partitionCol: String, tsType: String, outFmt: String,
+ extraOpts: Map[String, String] = Map.empty,
+ expectedPartitionUdf:
org.apache.spark.sql.expressions.UserDefinedFunction)
+
+ def runTestCase(tc: TestCase): Unit = {
+ val writer = tc.extraOpts.foldLeft(
+ inputDF.write.format("hudi")
+ .options(writeOpts)
+ .option(KEYGENERATOR_CLASS_NAME.key(),
classOf[TimestampBasedKeyGenerator].getName)
+ .mode(SaveMode.Overwrite)
+ ) { case (w, (k, v)) => w.option(k, v) }
+ writer.partitionBy(tc.partitionCol)
+ .option(TIMESTAMP_TYPE_FIELD.key, tc.tsType)
+ .option(TIMESTAMP_OUTPUT_DATE_FORMAT.key, tc.outFmt)
+ .save(basePath)
+ val readDF =
spark.read.format("org.apache.hudi").options(readOpts).load(basePath)
+ assertTrue(readDF.filter(col("_hoodie_partition_path") =!=
tc.expectedPartitionUdf(col(tc.partitionCol))).count() == 0)
+ }
+
+ val outputDateFmt = "yyyy-MM-dd HH"
+ // Joda's DateTimeZone.forID does not recognise "GMT+08:00".
HoodieDateTimeParser resolves the
+ // configured id via java.util.TimeZone, so the expected values are built
the same way.
+ val tzId = "GMT+08:00"
+
+ // Test 1: EPOCHMILLISECONDS with timezone GMT+08:00
+ val udfMillisTz = udf((millis: Long) => {
+ val zone = DateTimeZone.forTimeZone(TimeZone.getTimeZone(tzId))
+ new DateTime(millis,
zone).toString(DateTimeFormat.forPattern(outputDateFmt).withZone(zone))
+ })
+ runTestCase(TestCase("current_ts", "EPOCHMILLISECONDS", outputDateFmt,
+ Map(TIMESTAMP_TIMEZONE_FORMAT.key -> tzId), udfMillisTz))
+
+ // Test 2: EPOCHMICROSECONDS (no timezone configured, so the key generator
uses the JVM default)
+ val udfMicros = udf((micros: Long) =>
+ new DateTime(micros /
1000).toString(DateTimeFormat.forPattern(outputDateFmt)))
+ runTestCase(TestCase("current_ts_micros", "EPOCHMICROSECONDS",
outputDateFmt,
+ expectedPartitionUdf = udfMicros))
+
+ // Test 3: DATE_STRING with timezone
+ val dateStrInFmt = "yyyy-MM-dd HH:mm:ss"
+ val udfDateStrTz = udf((s: String) => {
+ val zone = DateTimeZone.forTimeZone(TimeZone.getTimeZone(tzId))
+ DateTime.parse(s, DateTimeFormat.forPattern(dateStrInFmt).withZone(zone))
+ .toString(DateTimeFormat.forPattern(outputDateFmt).withZone(zone))
+ })
+ runTestCase(TestCase("current_date_string", "DATE_STRING", outputDateFmt,
+ Map(TIMESTAMP_INPUT_DATE_FORMAT.key -> dateStrInFmt,
+ TIMESTAMP_TIMEZONE_FORMAT.key -> tzId), udfDateStrTz))
+
+ // Test 4: SCALAR with hours (no timezone configured, so the key generator
uses the JVM default)
Review Comment:
**minor:** not blocking. `hours` is not a separate code path:
`TimestampBasedAvroKeyGenerator.java:103-107` resolves every unit through a
single `TimeUnit.valueOf(...)` with no per-unit branch, and
`TestTimestampBasedKeyGenerator.testScalar` already covers lowercase `days`
(:243) and `MICROSECONDS` (:289).
Could we point this at `microseconds` so it reaches a unit nothing else does?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala:
##########
@@ -1450,6 +1451,74 @@ class TestCOWDataSource extends
HoodieSparkClientTestBase with ScalaAssertionSup
assertTrue(recordsReadDF.filter(col("_hoodie_partition_path") =!=
udf_date_format(col("current_ts"))).count() == 0)
}
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieRecordType], names = Array("AVRO",
"SPARK"))
+ def testTimestampBasedKeyGeneratorWithVariousConfigurations(recordType:
HoodieRecordType) {
+ val (writeOpts, readOpts) =
getWriterReaderOptsLessPartitionPath(recordType)
+
+ val records = recordsToStrings(dataGen.generateInserts("000",
100)).asScala.toList
+ val inputDF = spark.read.json(spark.sparkContext.parallelize(records, 2))
+ .withColumn("current_ts_micros", col("current_ts") * 1000)
+ .withColumn("current_date_string",
+ date_format((col("current_ts") / 1000).cast("timestamp"), "yyyy-MM-dd
HH:mm:ss"))
+ .withColumn("current_ts_hours", (col("current_ts") /
3600000).cast("long"))
+
+ case class TestCase(partitionCol: String, tsType: String, outFmt: String,
+ extraOpts: Map[String, String] = Map.empty,
+ expectedPartitionUdf:
org.apache.spark.sql.expressions.UserDefinedFunction)
+
+ def runTestCase(tc: TestCase): Unit = {
+ val writer = tc.extraOpts.foldLeft(
+ inputDF.write.format("hudi")
+ .options(writeOpts)
+ .option(KEYGENERATOR_CLASS_NAME.key(),
classOf[TimestampBasedKeyGenerator].getName)
+ .mode(SaveMode.Overwrite)
+ ) { case (w, (k, v)) => w.option(k, v) }
Review Comment:
**nit:** feel free to ignore. The `foldLeft` is exactly
`.options(tc.extraOpts)`, which the same expression already uses one line above.
```suggestion
val writer = inputDF.write.format("hudi")
.options(writeOpts)
.options(tc.extraOpts)
.option(KEYGENERATOR_CLASS_NAME.key(),
classOf[TimestampBasedKeyGenerator].getName)
.mode(SaveMode.Overwrite)
```
--
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]
