This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new aabed4659396 fix(spark): honor 'earliest' begin instant in DataFrame
CDC reads (#20304)
aabed4659396 is described below
commit aabed46593960ff3553cdc1ace1a491987983702
Author: Sagar Sumit <[email protected]>
AuthorDate: Wed Oct 14 09:24:30 2026 +0530
fix(spark): honor 'earliest' begin instant in DataFrame CDC reads (#20304)
A DataFrame CDC read with hoodie.datasource.read.begin.instanttime=earliest
returned no rows: HoodieCDCFileIndex passed the literal into InstantRange,
and the string comparison against instant times excluded every commit.
Map it to HoodieTimeline.INIT_INSTANT_TS, matching what hudi_table_changes
already does for 'earliest'.
Also correct the begin.instanttime doc: the start bound is exclusive
(completion_time > START_COMMIT), not inclusive.
Fixes #20300
Fixes #20301
---
.../scala/org/apache/hudi/DataSourceOptions.scala | 5 ++--
.../org/apache/hudi/cdc/HoodieCDCFileIndex.scala | 7 +++++-
.../functional/cdc/TestCDCDataFrameSuite.scala | 28 ++++++++++++++++++++++
3 files changed, 37 insertions(+), 3 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
index 899b93a386f9..1d9caf830631 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
@@ -116,8 +116,9 @@ object DataSourceReadOptions {
.sinceVersion("0.9.0")
.withDocumentation("Required when `" + QUERY_TYPE.key() + "` is set to `"
+ QUERY_TYPE_INCREMENTAL_OPT_VAL + "`. "
+ "Represents the completion time to start incrementally pulling data
from. The completion time here need not necessarily "
- + "correspond to an instant on the timeline. New data written with
completion_time >= START_COMMIT are fetched out. "
- + "For e.g: ‘20170901080000’ will get all new data written on or after
Sep 1, 2017 08:00AM.")
+ + "correspond to an instant on the timeline. New data written with
completion_time > START_COMMIT are fetched out, "
+ + "i.e. the start is exclusive while
`hoodie.datasource.read.end.instanttime` is inclusive. "
+ + "For e.g: ‘20170901080000’ will get all new data written after Sep 1,
2017 08:00AM.")
val END_COMMIT: ConfigProperty[String] = ConfigProperty
.key("hoodie.datasource.read.end.instanttime")
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/HoodieCDCFileIndex.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/HoodieCDCFileIndex.scala
index bf8f0f52c2bd..8c8d6047e2bc 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/HoodieCDCFileIndex.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/HoodieCDCFileIndex.scala
@@ -25,6 +25,8 @@ import org.apache.hudi.common.table.HoodieTableMetaClient
import org.apache.hudi.common.table.cdc.{HoodieCDCExtractor, HoodieCDCUtils}
import org.apache.hudi.common.table.log.InstantRange
import org.apache.hudi.common.table.log.InstantRange.RangeType
+import org.apache.hudi.common.table.read.IncrementalQueryAnalyzer
+import org.apache.hudi.common.table.timeline.HoodieTimeline
import org.apache.hudi.exception.HoodieException
import org.apache.hadoop.fs.{FileStatus, Path}
@@ -57,7 +59,10 @@ class HoodieCDCFileIndex(override val spark: SparkSession,
.startInstant(options.getOrElse(DataSourceReadOptions.START_COMMIT.key(),
throw new HoodieException(s"CDC Query should provide the valid start
completion time "
+ s"through the option ${DataSourceReadOptions.START_COMMIT.key()}")
- ))
+ ) match {
+ case start if
start.equalsIgnoreCase(IncrementalQueryAnalyzer.START_COMMIT_EARLIEST) =>
HoodieTimeline.INIT_INSTANT_TS
+ case start => start
+ })
.endInstant(options.getOrElse(DataSourceReadOptions.END_COMMIT.key(),
HoodieCDCFileIndex.getTimestampOfLatestInstant(metaClient)
))
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala
index 894b7365ba16..853d82313740 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala
@@ -1023,4 +1023,32 @@ class TestCDCDataFrameSuite extends HoodieCDCTestBase {
}
}
}
+
+ @ParameterizedTest
+ @CsvSource(Array("COPY_ON_WRITE,earliest", "MERGE_ON_READ,earliest",
"COPY_ON_WRITE,EARLIEST", "MERGE_ON_READ,EARLIEST"))
+ def testCDCReadFromEarliest(tableType: String, beginInstant: String): Unit =
{
+ val options = commonOpts ++ Map(DataSourceWriteOptions.TABLE_TYPE.key() ->
tableType)
+
+ val records1 = recordsToStrings(dataGen.generateInserts("000",
100)).asScala.toList
+ spark.read.json(spark.sparkContext.parallelize(records1, 2))
+ .write.format("org.apache.hudi")
+ .options(options)
+ .mode(SaveMode.Overwrite)
+ .save(basePath)
+ metaClient = createMetaClient(spark, basePath)
+ val commitTime1 =
metaClient.reloadActiveTimeline.lastInstant().get().requestedTime
+
+ val records2 = recordsToStrings(dataGen.generateUniqueUpdates("001",
30)).asScala.toList
+ spark.read.json(spark.sparkContext.parallelize(records2, 2))
+ .write.format("org.apache.hudi")
+ .options(options)
+ .mode(SaveMode.Append)
+ .save(basePath)
+
+ val cdcFromEarliest = cdcDataFrame(beginInstant)
+ assertCDCOpCnt(cdcFromEarliest, 100, 30, 0)
+ assertEquals(
+
cdcDataFrame(instantBefore(commitTime1)).collect().map(_.toString).sorted.toSeq,
+ cdcFromEarliest.collect().map(_.toString).sorted.toSeq)
+ }
}
