codope opened a new issue, #20300:
URL: https://github.com/apache/hudi/issues/20300
### Bug Description
**What happened:**
A CDC read through the DataFrame API with `begin.instanttime = "earliest"`
succeeds and
returns zero rows. Nothing warns the user. The same table returns all
changes:
- through `hudi_table_changes(..., 'cdc', 'earliest')`;
- through the DataFrame API with begin `"000"`;
- through a `latest_state` incremental DataFrame read with begin
`"earliest"`.
Affects both COW and MOR.
**What you expected:**
`earliest` returns every change from the start of the active timeline, as
the SQL function and
the `latest_state` reader already do (4 rows here).
**Steps to reproduce:**
Using the two tables from steps to reproduce in #20299 (each has an insert
commit of 3 rows and an upsert of 1 row):
```scala
for (p <- Seq("/tmp/cdc_types_cow", "/tmp/cdc_types_mor")) {
def cdc(b: String) = spark.read.format("hudi")
.option("hoodie.datasource.query.type", "incremental")
.option("hoodie.datasource.query.incremental.format", "cdc")
.option("hoodie.datasource.read.begin.instanttime", b)
.load(p).count()
println(s"$p earliest=${cdc("earliest")} 000=${cdc("000")} " +
s"sql_earliest=${spark.sql(s"select count(*) from
hudi_table_changes('$p', 'cdc', 'earliest')").first.getLong(0)}")
}
```
Actual:
```
/tmp/cdc_types_cow earliest=0 000=4 sql_earliest=4
/tmp/cdc_types_mor earliest=0 000=4 sql_earliest=4
```
### Root cause
- `HoodieCDCFileIndex`
(`hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/HoodieCDCFileIndex.scala:57`)
passes the begin option straight into
`InstantRange.builder().startInstant(...)`.
- `HoodieCDCExtractor`
(`hudi-common/.../table/cdc/HoodieCDCExtractor.java:230`) then filters
with `instantRange.isInRange(instant.requestedTime())`. For an open-closed
range that is
`compareTimestamps(instant, GREATER_THAN, "earliest")`, which is a string
comparison, and
`"2026…" > "earliest"` is false. So every commit is excluded.
- The SQL path works because `HoodieTableChanges`
(`.../catalyst/plans/logical/HoodieTableChanges.scala:46`)
maps `"earliest"` to `"000"` before building the read options. The non-CDC
incremental path
works because `IncrementalQueryAnalyzer` recognizes
`START_COMMIT_EARLIEST`.
### Suggested fix
In `HoodieCDCFileIndex`, treat
`IncrementalQueryAnalyzer.START_COMMIT_EARLIEST`
(case-insensitive) as an open start: pass `null` (the range is built with
`nullableBoundary(true)`)
or `"000"`, mirroring `HoodieTableChanges`. Add a DataFrame CDC test with
`earliest` for COW and
MOR.
### Environment
**Hudi version:** 1.2.1 (same code in master)
**Query engine:** Spark 3.5.6, Scala 12, Java 11
### Logs and Stack Trace
_No response_
--
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]
