RyanLin5967 opened a new issue, #922:
URL: https://github.com/apache/hudi-rs/issues/922
### Is there an existing issue for this?
- [x] I have searched the existing issues
### Description of the bug
Currently, an incremental read picks each touched file group's slice at the
end timestamp by requested time, so a commit requested before the end but
completed after it supplies the slice, and the completion-time filter drops its
rows. Before that commit completes the same read is right.
### Steps To Reproduce
```python
import os, tempfile, threading, time
from pyspark.sql import SparkSession, functions as F
from hudi import HudiTableBuilder, HudiReadOptions, HudiQueryType
spark = (SparkSession.builder
.config("spark.jars.packages",
"org.apache.hudi:hudi-spark4.0-bundle_2.13:1.2.1")
.config("spark.serializer",
"org.apache.spark.serializer.KryoSerializer").getOrCreate())
path = tempfile.mkdtemp()
opts = {"hoodie.table.name": "t", "hoodie.datasource.write.recordkey.field":
"id",
"hoodie.datasource.write.partitionpath.field": "p",
"hoodie.write.concurrency.mode": "OPTIMISTIC_CONCURRENCY_CONTROL",
"hoodie.write.lock.provider":
"org.apache.hudi.client.transaction.lock.InProcessLockProvider"}
def write(rows, delay=0):
df = spark.createDataFrame(rows, "id STRING, v STRING, p STRING")
df.withColumn("v", F.udf(lambda v: time.sleep(delay) or
v)("v")).write.format("hudi") \
.options(**opts).mode("append").save(path)
def commits(): # [requested, completed] of each completed commit
return sorted(f.split(".")[0].split("_") for f in os.listdir(path +
"/.hoodie/timeline") if f.endswith(".commit"))
def incremental(start, end):
o =
HudiReadOptions().with_query_type(HudiQueryType.Incremental).with_start_timestamp(start).with_end_timestamp(end)
return sorted((r["id"], r["v"]) for b in
HudiTableBuilder.from_base_uri(path).build().read(o) for r in b.to_pylist())
write([("a", "1", "p1"), ("b", "1", "p2")])
start = commits()[0][0]
a = threading.Thread(target=write, args=([("a", "2", "p1")], 20));
a.start(); time.sleep(5)
write([("b", "2", "p2")])
end = commits()[-1][1]
print(incremental(start, end))
# [('a', '1'), ('b', '2')], while the a=2 write is still running
a.join()
print(incremental(start, end))
# [('b', '2')], should be [('a', '1'), ('b', '2')]
# the a=2 commit was requested before end and completed after it
print(sorted(tuple(r) for r in
spark.read.format("hudi").option("hoodie.datasource.query.type", "incremental")
.option("hoodie.datasource.read.begin.instanttime",
start).option("hoodie.datasource.read.end.instanttime", end)
.load(path).select("id", "v").collect()))
# [('a', '1'), ('b', '2')]
```
### Expected behavior
Both reads return `[('a', '1'), ('b', '2')]`, as Spark does, since Hudi 1.0
bounds incremental queries by completion time and the a=2 commit completed
after `end`.
### Screenshots / Logs
_No response_
### Software information
- Operating system: macOS 26.7.1, arm64 (Darwin 25.6.0)
- Rust version: none, the PyPI wheel on Python 3.12.16
- Project version: hudi 0.5.0, and main (its CI wheel, 0.6.0.dev0)
- Writer: Hudi 1.2.1 (`hudi-spark4.0-bundle_2.13`) on pyspark 4.0.5, Java
21.0.12.1
### Additional context
A `MERGE_ON_READ` table loses the row the same way.
--
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]
