linliu-code commented on code in PR #13519:
URL: https://github.com/apache/hudi/pull/13519#discussion_r2223211702
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPayloadDeprecationFlow.scala:
##########
@@ -116,11 +142,137 @@ class TestPayloadDeprecationFlow extends
SparkClientFunctionalTestHarness {
assertTrue(
expectedDf.except(finalDf).isEmpty && finalDf.except(expectedDf).isEmpty)
}
+
+ /**
+ * Test if the payload based read produce the same result when upgrade
happens in between.
+ */
+ @ParameterizedTest
+ @MethodSource(Array("provideParamsForUpgradeBehavior"))
+ def testUpgradeUntilTableVersionNine(tableType: String,
+ payloadClazz: String): Unit = {
+ val opts: Map[String, String] = Map(
+ HoodieWriteConfig.WRITE_PAYLOAD_CLASS_NAME.key() -> payloadClazz,
+ HoodieTableConfig.MERGE_PROPERTIES.key() ->
+ "hoodie.payload.delete.field=Op,hoodie.payload.delete.marker=d")
+ val columns = Seq("ts", "key", "rider", "driver", "fare", "Op")
+
+ // 1. Add an insert.
+ val data = Seq(
+ (10, "1", "rider-A", "driver-A", 19.10, "i"),
+ (10, "2", "rider-B", "driver-B", 27.70, "i"),
+ (10, "3", "rider-C", "driver-C", 33.90, "i"),
+ (10, "4", "rider-D", "driver-D", 34.15, "i"),
+ (10, "5", "rider-E", "driver-E", 17.85, "i"))
+ val inserts = spark.createDataFrame(data).toDF(columns: _*)
+ inserts.write.format("hudi").
+ option(RECORDKEY_FIELD.key(), "key").
+ option(PRECOMBINE_FIELD.key(), "ts").
+ option(TABLE_TYPE.key(), tableType).
+ option(DataSourceWriteOptions.TABLE_NAME.key(), "test_table").
+ option(HoodieCompactionConfig.INLINE_COMPACT.key(), "false").
+ option(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), "6").
+ options(opts).
+ mode(SaveMode.Overwrite).
+ save(basePath)
+
+ // Validate table version.
+ var metaClient = getHoodieMetaClient(storageConf(), basePath())
+ assertEquals(HoodieTableVersion.SIX,
metaClient.getTableConfig.getTableVersion)
+
+ // 2. Add an update.
+ val firstUpdateData = Seq(
+ (11, "1", "rider-X", "driver-X", 19.10, "d"),
+ (11, "2", "rider-Y", "driver-Y", 27.70, "u"))
+ val firstUpdate = spark.createDataFrame(firstUpdateData).toDF(columns: _*)
+ firstUpdate.write.format("hudi").
+ option(OPERATION.key(), "upsert").
+ option(HoodieCompactionConfig.INLINE_COMPACT.key(), "false").
+ option(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), "8").
+ options(opts).
+ mode(SaveMode.Append).
+ save(basePath)
+
+ // Validate table version.
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ assertEquals(HoodieTableVersion.EIGHT,
metaClient.getTableConfig.getTableVersion)
+
+ // 3. Add an update.
+ val secondUpdateData = Seq(
+ (12, "3", "rider-CC", "driver-CC", 33.90, "i"),
+ (9, "4", "rider-DD", "driver-DD", 34.15, "i"),
+ (12, "5", "rider-EE", "driver-EE", 17.85, "i"))
+ val secondUpdate = spark.createDataFrame(secondUpdateData).toDF(columns:
_*)
+ secondUpdate.write.format("hudi").
+ option(OPERATION.key(), "upsert").
+ option(HoodieCompactionConfig.INLINE_COMPACT.key(), "true").
+ option(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(),
"1").
+ option(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), "9").
+ options(opts).
+ mode(SaveMode.Append).
+ save(basePath)
+
+ // Validate table version.
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ assertEquals(HoodieTableVersion.EIGHT,
metaClient.getTableConfig.getTableVersion)
+
+ // 4. Validate.
+ val df = spark.read.format("hudi").options(opts).load(basePath)
+ val finalDf = df.select("ts", "key", "rider", "driver", "fare",
"Op").sort("key")
+
+ val expectedData = if
(!payloadClazz.equals(classOf[AWSDmsAvroPayload].getName)) {
+ if (payloadClazz.equals(classOf[PartialUpdateAvroPayload].getName)
+ || payloadClazz.equals(classOf[EventTimeAvroPayload].getName)) {
+ Seq(
+ (11, "1", "rider-X", "driver-X", 19.10, "d"),
+ (11, "2", "rider-Y", "driver-Y", 27.70, "u"),
+ (12, "3", "rider-CC", "driver-CC", 33.90, "i"),
+ (10, "4", "rider-D", "driver-D", 34.15, "i"),
+ (12, "5", "rider-EE", "driver-EE", 17.85, "i"))
+ } else {
+ Seq(
+ (11, "1", "rider-X", "driver-X", 19.10, "d"),
+ (11, "2", "rider-Y", "driver-Y", 27.70, "u"),
+ (12, "3", "rider-CC", "driver-CC", 33.90, "i"),
+ (9, "4", "rider-DD", "driver-DD", 34.15, "i"),
+ (12, "5", "rider-EE", "driver-EE", 17.85, "i"))
+ }
+ } else {
+ Seq(
+ (11, "2", "rider-Y", "driver-Y", 27.70, "u"),
+ (12, "3", "rider-CC", "driver-CC", 33.90, "i"),
+ (9, "4", "rider-DD", "driver-DD", 34.15, "i"),
+ (12, "5", "rider-EE", "driver-EE", 17.85, "i"))
+ }
+ val expectedDf = spark.createDataFrame(
+ spark.sparkContext.parallelize(expectedData)).toDF(columns:
_*).sort("key")
+ assertTrue(
+ expectedDf.except(finalDf).isEmpty && finalDf.except(expectedDf).isEmpty)
+ }
}
-// TODO: Add COPY_ON_WRITE table type tests when write path is updated
accordingly.
+// TODO: Add COPY_ON_WRITE table type tests when write path is updated
accordingly.s
object TestPayloadDeprecationFlow {
- def provideParams(): java.util.List[Arguments] = {
+ def provideParamsForPayloadBehavior(): java.util.List[Arguments] = {
+ java.util.Arrays.asList(
+ Arguments.of("MERGE_ON_READ",
classOf[OverwriteWithLatestAvroPayload].getName, "6"),
Review Comment:
No, it will fail since the write path will have different behavior.
--
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]
