linliu-code commented on code in PR #13519:
URL: https://github.com/apache/hudi/pull/13519#discussion_r2223955228
##########
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)
Review Comment:
This kinds of test does not work since it involves the write path changes.
Most likely the commit time based payload was replaced by
`defaulthoodierecordpayload`, which cause some issues. WE have to fix that
before fixing this tests. I will remove this tests for now.
--
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]
