yihua commented on code in PR #11947:
URL: https://github.com/apache/hudi/pull/11947#discussion_r1797860501
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala:
##########
@@ -114,21 +114,24 @@ class TestMORDataSource extends HoodieSparkClientTestBase
with SparkDatasetMixin
.mode(SaveMode.Overwrite)
.save(basePath)
assertTrue(HoodieDataSourceHelpers.hasNewCommits(storage, basePath, "000"))
+ val df1CompletionTime =
HoodieDataSourceHelpers.latestCommitCompletionTime(storage, basePath)
Review Comment:
rename to `commit1CompletionTime`; same for other variables.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSparkDataSource.scala:
##########
@@ -138,7 +140,7 @@ class TestSparkDataSource extends
SparkClientFunctionalTestHarness {
val hoodieIncViewDf1 = spark.read.format("org.apache.hudi")
.option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
.option(DataSourceReadOptions.BEGIN_INSTANTTIME.key, "000")
- .option(DataSourceReadOptions.END_INSTANTTIME.key, firstCommit)
+ .option(DataSourceReadOptions.END_INSTANTTIME.key, commitCompletionTime1)
Review Comment:
Should the begin instant time be changed too?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala:
##########
@@ -205,13 +205,10 @@ class TestStreamingSource extends StreamTest {
addData(tablePath, Seq(("2", "a1", "11", "001")))
addData(tablePath, Seq(("3", "a1", "12", "002")))
- val timestamp = if (handlingMode == USE_TRANSITION_TIME) {
+ val timestamp =
Review Comment:
Similar here. Should we add a test case where the smallest completion time
in the active timeline does not have the smallest instant time to validate the
correctness of the incremental query logic?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala:
##########
@@ -46,7 +46,7 @@ class TestStreamingSource extends StreamTest {
)
private val columns = Seq("id", "name", "price", "ts")
- val handlingMode: HollowCommitHandling = BLOCK
+ val handlingMode: HollowCommitHandling = USE_TRANSITION_TIME
Review Comment:
Should this be removed now? Or do we want to keep it for incremental pulls
on 0.x tables?
--
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]
