cshuo commented on code in PR #19948:
URL: https://github.com/apache/hudi/pull/19948#discussion_r4036689007
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFileGroupReader.scala:
##########
@@ -123,6 +125,57 @@ class TestIncrementalReadWithFileGroupReader extends
SparkClientFunctionalTestHa
assertIncrementalRange(readVersion, instants, 6, 6, Set.empty)
}
+ @ParameterizedTest(name = "V2 MOR full-table scan applies instant range,
legacy RDD = {0}")
+ @ValueSource(booleans = Array(false, true))
+ def testV2MorFullTableScanAppliesInstantRange(useLegacyRdd: Boolean): Unit =
{
+ val archivalOptions = Map(
+ HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key -> "2",
+ HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key -> "5",
+ HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key -> "1",
+ HoodieCleanConfig.AUTO_CLEAN.key -> "false",
+ HoodieMetadataConfig.ENABLE.key -> "false",
+ HoodieWriteConfig.AUTO_UPGRADE_VERSION.key -> "false")
+
+ batches.zipWithIndex.foreach { case ((data, operation), i) =>
+ write(data, "MERGE_ON_READ", 6, operation,
Review Comment:
**Cover full-table-scan fallback on V8 tables**
Could we parameterize this regression test over table versions 6 and 8? The
range propagation applies to all V2 MOR incremental reads, while this fallback
test only exercises a V6 table. A V8 case using completion-time query
boundaries would also verify the conversion to the requested-time range passed
to the reader.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFileGroupReader.scala:
##########
@@ -123,6 +125,57 @@ class TestIncrementalReadWithFileGroupReader extends
SparkClientFunctionalTestHa
assertIncrementalRange(readVersion, instants, 6, 6, Set.empty)
}
+ @ParameterizedTest(name = "V2 MOR full-table scan applies instant range,
legacy RDD = {0}")
+ @ValueSource(booleans = Array(false, true))
+ def testV2MorFullTableScanAppliesInstantRange(useLegacyRdd: Boolean): Unit =
{
+ val archivalOptions = Map(
+ HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key -> "2",
+ HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key -> "5",
+ HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key -> "1",
+ HoodieCleanConfig.AUTO_CLEAN.key -> "false",
+ HoodieMetadataConfig.ENABLE.key -> "false",
+ HoodieWriteConfig.AUTO_UPGRADE_VERSION.key -> "false")
+
+ batches.zipWithIndex.foreach { case ((data, operation), i) =>
+ write(data, "MERGE_ON_READ", 6, operation,
+ if (i == 0) SaveMode.Overwrite else SaveMode.Append, archivalOptions)
+ }
+
+ val metaClient = HoodieTableMetaClient.builder()
+ .setConf(storageConf().newInstance()).setBasePath(basePath()).build()
+ assertEquals(6, metaClient.getTableConfig.getTableVersion.versionCode())
+ val archivedInstants =
metaClient.getArchivedTimeline.getCommitsTimeline.filterCompletedInstants
+ .getInstants.asScala.toList
+ val activeInstants =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
+ .getInstants.asScala.toList
+ val instants = (archivedInstants ++ activeInstants)
+ .map(instant => instant.requestedTime ->
instant).toMap.values.toList.sortBy(_.requestedTime)
+
+ assertTrue(archivedInstants.nonEmpty, "The query must include archived
instants to force a full-table scan")
+ assertEquals(6, instants.size)
+
assertTrue(instants.last.requestedTime.compareTo(instants(4).requestedTime) > 0,
+ "c6 must be outside the requested range")
+ val (_, logFiles) = listDataFiles()
+ assertEquals(3, logFiles.size, "The out-of-range c6 log must exist
physically")
+
+ val readOptions = Map(
+ DataSourceReadOptions.QUERY_TYPE.key ->
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL,
+ DataSourceReadOptions.START_COMMIT.key -> "000",
Review Comment:
**Exercise a bounded start and the `earliest` range variant**
This case only exercises an `EXACT_MATCH` range starting from `"000"`.
Consider adding `(c3, c5]` to verify exclusion of older base/log records, and
`earliest → c5` to exercise the `CLOSED_CLOSED` range with a null lower
boundary. The former should return only `k1=5`, `k2=4`, `k3=5`, and `k4=5`.
--
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]
