hudi-agent commented on code in PR #19657:
URL: https://github.com/apache/hudi/pull/19657#discussion_r3803922003
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala:
##########
@@ -1450,6 +1450,73 @@ 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"))
+ .withColumn("current_ts_seconds", (col("current_ts") /
1000).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)
+ }
+
+ // Test 1: EPOCHMILLISECONDS with timezone GMT+8:00
+ val tzMillisOutFmt = "yyyy-MM-dd HH"
+ val udfMillisTz = udf((millis: Long) =>
+ new
DateTime(millis).withZone(org.joda.time.DateTimeZone.forID("GMT+8:00"))
+
.toString(DateTimeFormat.forPattern(tzMillisOutFmt).withZone(org.joda.time.DateTimeZone.forID("GMT+8:00"))))
+ runTestCase(TestCase("current_ts", "EPOCHMILLISECONDS", tzMillisOutFmt,
+ Map(TIMESTAMP_TIMEZONE_FORMAT.key -> "GMT+8:00"), udfMillisTz))
+
+ // Test 2: EPOCHMICROSECONDS
+ val microsOutFmt = "yyyy-MM-dd HH"
+ val udfMicros = udf((micros: Long) =>
+ new DateTime(micros /
1000).toString(DateTimeFormat.forPattern(microsOutFmt)))
+ runTestCase(TestCase("current_ts_micros", "EPOCHMICROSECONDS",
microsOutFmt,
+ expectedPartitionUdf = udfMicros))
+
+ // Test 3: DATE_STRING with timezone
+ val dateStrOutFmt = "yyyy-MM-dd HH"
+ val dateStrInFmt = "yyyy-MM-dd HH:mm:ss"
Review Comment:
🤖 I think this UDF may be timezone-dependent and won't match the keygen.
Since `TIMESTAMP_TIMEZONE_FORMAT` is `GMT+8:00`, the keygen builds its input
formatter `withZone(GMT+8)`, so it parses the string *as* GMT+8 and formats in
GMT+8 — the wall-clock hour round-trips. Here `DateTime.parse(s,
DateTimeFormat.forPattern(dateStrInFmt))` has no zone, so it parses in the JVM
default zone before `.withZone(GMT+8)` shifts the hour. On a UTC CI box that's
an 8-hour skew (keygen `...01 12` vs UDF `...01 20`), and `"yyyy-MM-dd HH"`
includes the hour. Could you set the parse zone to GMT+8 (e.g.
`DateTimeFormat.forPattern(dateStrInFmt).withZone(GMT+8)`) so it matches the
keygen regardless of JVM timezone?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala:
##########
@@ -1450,6 +1450,73 @@ 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"))
+ .withColumn("current_ts_seconds", (col("current_ts") /
1000).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)
+ }
+
+ // Test 1: EPOCHMILLISECONDS with timezone GMT+8:00
+ val tzMillisOutFmt = "yyyy-MM-dd HH"
+ val udfMillisTz = udf((millis: Long) =>
+ new
DateTime(millis).withZone(org.joda.time.DateTimeZone.forID("GMT+8:00"))
+
.toString(DateTimeFormat.forPattern(tzMillisOutFmt).withZone(org.joda.time.DateTimeZone.forID("GMT+8:00"))))
+ runTestCase(TestCase("current_ts", "EPOCHMILLISECONDS", tzMillisOutFmt,
+ Map(TIMESTAMP_TIMEZONE_FORMAT.key -> "GMT+8:00"), udfMillisTz))
+
+ // Test 2: EPOCHMICROSECONDS
+ val microsOutFmt = "yyyy-MM-dd HH"
+ val udfMicros = udf((micros: Long) =>
+ new DateTime(micros /
1000).toString(DateTimeFormat.forPattern(microsOutFmt)))
+ runTestCase(TestCase("current_ts_micros", "EPOCHMICROSECONDS",
microsOutFmt,
+ expectedPartitionUdf = udfMicros))
+
+ // Test 3: DATE_STRING with timezone
+ val dateStrOutFmt = "yyyy-MM-dd HH"
+ val dateStrInFmt = "yyyy-MM-dd HH:mm:ss"
+ val udfDateStrTz = udf((s: String) =>
+ DateTime.parse(s, DateTimeFormat.forPattern(dateStrInFmt))
+ .withZone(org.joda.time.DateTimeZone.forID("GMT+8:00"))
+
.toString(DateTimeFormat.forPattern(dateStrOutFmt).withZone(org.joda.time.DateTimeZone.forID("GMT+8:00"))))
+ runTestCase(TestCase("current_date_string", "DATE_STRING", dateStrOutFmt,
+ Map("hoodie.keygen.timebased.input.dateformat" -> dateStrInFmt,
+ TIMESTAMP_TIMEZONE_FORMAT.key -> "GMT+8:00"), udfDateStrTz))
+
+ // Test 4: SCALAR with hours
+ val scalarHoursOutFmt = "yyyy-MM-dd HH"
+ val udfScalarHours = udf((hours: Long) =>
+ new DateTime(java.util.concurrent.TimeUnit.HOURS.toMillis(hours))
+ .toString(DateTimeFormat.forPattern(scalarHoursOutFmt)))
+ runTestCase(TestCase("current_ts_hours", "SCALAR", scalarHoursOutFmt,
+ Map("hoodie.keygen.timebased.timestamp.scalar.time.unit" -> "hours"),
udfScalarHours))
Review Comment:
🤖 nit: same as above —
`"hoodie.keygen.timebased.timestamp.scalar.time.unit"` could be replaced with
the named constant's `.key` accessor to match the pattern used for all the
other options.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala:
##########
@@ -1450,6 +1450,73 @@ 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"))
+ .withColumn("current_ts_seconds", (col("current_ts") /
1000).cast("long"))
Review Comment:
🤖 nit: `current_ts_seconds` is prepared here but isn't referenced in any of
the four test cases — could you either add a test that exercises it or drop the
column to avoid leaving future readers wondering if there's a missing scenario?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala:
##########
@@ -1450,6 +1450,73 @@ 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"))
+ .withColumn("current_ts_seconds", (col("current_ts") /
1000).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)
+ }
+
+ // Test 1: EPOCHMILLISECONDS with timezone GMT+8:00
+ val tzMillisOutFmt = "yyyy-MM-dd HH"
+ val udfMillisTz = udf((millis: Long) =>
+ new
DateTime(millis).withZone(org.joda.time.DateTimeZone.forID("GMT+8:00"))
+
.toString(DateTimeFormat.forPattern(tzMillisOutFmt).withZone(org.joda.time.DateTimeZone.forID("GMT+8:00"))))
+ runTestCase(TestCase("current_ts", "EPOCHMILLISECONDS", tzMillisOutFmt,
+ Map(TIMESTAMP_TIMEZONE_FORMAT.key -> "GMT+8:00"), udfMillisTz))
+
+ // Test 2: EPOCHMICROSECONDS
+ val microsOutFmt = "yyyy-MM-dd HH"
+ val udfMicros = udf((micros: Long) =>
+ new DateTime(micros /
1000).toString(DateTimeFormat.forPattern(microsOutFmt)))
+ runTestCase(TestCase("current_ts_micros", "EPOCHMICROSECONDS",
microsOutFmt,
+ expectedPartitionUdf = udfMicros))
+
+ // Test 3: DATE_STRING with timezone
+ val dateStrOutFmt = "yyyy-MM-dd HH"
+ val dateStrInFmt = "yyyy-MM-dd HH:mm:ss"
+ val udfDateStrTz = udf((s: String) =>
+ DateTime.parse(s, DateTimeFormat.forPattern(dateStrInFmt))
+ .withZone(org.joda.time.DateTimeZone.forID("GMT+8:00"))
+
.toString(DateTimeFormat.forPattern(dateStrOutFmt).withZone(org.joda.time.DateTimeZone.forID("GMT+8:00"))))
+ runTestCase(TestCase("current_date_string", "DATE_STRING", dateStrOutFmt,
+ Map("hoodie.keygen.timebased.input.dateformat" -> dateStrInFmt,
Review Comment:
🤖 nit: the rest of this test uses `.key` accessors (e.g.
`TIMESTAMP_TIMEZONE_FORMAT.key`) — could you replace
`"hoodie.keygen.timebased.input.dateformat"` with the corresponding constant so
all config keys here are refactor-safe?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
