[SUPPORT] Incremental read query performance
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
**Describe the problem you faced**
I'm running an application that reads from 4 medium-sized (few hundred GB) Hudi MoR tables which are compacted weekly.
When running incremental queries to look over 3~5 commits with `hoodie.datasource.read.incr.fallback.fulltablescan.enable -> true`, I am seeing a 5 hour gap in my Spark History Server:

Incremental read config:
```
Scan MergeOnReadIncrementalRelation(
org.apache.spark.sql.SQLContext@3b457d8a,Map(
hoodie.datasource.read.incr.fallback.fulltablescan.enable -> true,
path -> s3://lake/names.all_hudi,
hoodie.write.lock.zookeeper.url -> ip-10-80-36-225.ec2.internal,
hoodie.write.lock.zookeeper.base_path -> /hudi,
hoodie.metadata.enable -> true,
hoodie.datasource.hive_sync.jdbcurl -> jdbc:hive2://ip-10-xx-xx-xxx.ec2.internal:10000,
hoodie.datasource.read.begin.instanttime -> 20231207170214000,
hoodie.datasource.query.type -> incremental,
hoodie.cleaner.policy.failed.writes -> EAGER,
hoodie.write.lock.zookeeper.port -> 2181,
hoodie.write.lock.provider -> org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider,
hoodie.write.concurrency.mode -> single_writer),HoodieTableMetaClient{basePath='s3://lake/names.all_hudi',
metaPath='s3://lake/names.all_hudi/.hoodie',
tableType=MERGE_ON_READ},None,None
)
Job time: 2023/12/12 00:47:50
=> 5 commits for incremental read
number of output rows: 19,926,780
```
During the gap, my driver kills all 40 of my executor instances and I see the following logs on driver: [log.txt](https://github.com/apache/hudi/files/13686484/log.txt)
When I use snapshot query + filter on _hoodie_commit_time, the gap reduces to ~20mins:

Snapshot read config:
```
Scan MergeOnReadSnapshotRelation(org.apache.spark.sql.SQLContext@24a4d185,
Map(
path -> s3://lake/names.all_hudi,
hoodie.write.lock.zookeeper.url -> ip-xx-xx-xxx.ec2.internal,
hoodie.write.lock.zookeeper.base_path -> /hudi,
hoodie.metadata.enable -> true,
hoodie.datasource.hive_sync.jdbcurl -> jdbc:hive2://ip-10-xx-xx-xxx.ec2.internal:10000,
hoodie.datasource.query.type -> snapshot,
hoodie.cleaner.policy.failed.writes -> EAGER,
hoodie.write.lock.zookeeper.port -> 2181,
hoodie.write.lock.provider -> org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider,
hoodie.write.concurrency.mode -> single_writer),
HoodieTableMetaClient{basePath='s3://lake/names.all_hudi',
metaPath='s3://lake/names.all_hudi/.hoodie',
tableType=MERGE_ON_READ},
List(),
None,
None
)
```
When I do incremental read with `hoodie.datasource.read.incr.fallback.fulltablescan.enable -> false`, I see the incremental query gap go from 5 hours -> 5 mins 😮 :

Incremental read config (no fulltable fallback):
```
Scan MergeOnReadIncrementalRelation(org.apache.spark.sql.SQLContext@dcb6efd,
Map(
path -> s3://lake/names.all_hudi,
hoodie.write.lock.zookeeper.url -> ip-10-xx-xx-xxx.ec2.internal,
hoodie.write.lock.zookeeper.base_path -> /hudi,
hoodie.metadata.enable -> true,
hoodie.datasource.hive_sync.jdbcurl -> jdbc:hive2://ip-10-xx-xx-xxx.ec2.internal:10000,
hoodie.datasource.read.begin.instanttime -> 20231212232646000,
hoodie.datasource.query.type -> incremental,
hoodie.cleaner.policy.failed.writes -> EAGER,
hoodie.write.lock.zookeeper.port -> 2181,
hoodie.write.lock.provider -> org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider,
hoodie.write.concurrency.mode -> single_writer),
HoodieTableMetaClient{basePath='s3://lake/names.all_hudi',
metaPath='s3://lake/names.all_hudi/.hoodie',
tableType=MERGE_ON_READ},
None,
None
)
number of output rows: 12,541,221
```
This leads me to believe the incremental query was degenerating into a full table scan when it shouldn't... (all my commits are active and uncleaned):
https://github.com/apache/hudi/blob/7a6543958368540d221ddc18e0c12b8d526b6859/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala#L142-L151
Wondering if someone can help me understand the following:
1. Why is the incremental load degenerating into a full table scan when I can run the same job with the flag disabled and do not run into File Not Found?
2. Is the full table scan expected to happen on only one instance (the driver?)
3. Can I safely leave this flag false, and expect to only run into FileNotFound if running cleaner?
**Expected behavior**
Only fall back to full table scan in incremental load when necessary (file not found issue) and not all the time.
**Environment Description**
* Hudi version : 0.13.1
* Spark version : 3.4.0
* Hive version : 3.1.3
* Hadoop version : 3.3.3
* Storage (HDFS/S3/GCS..) : S3
* Running on Docker? (yes/no) : no
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with MergeOnReadIncrementalRelation.scala at the referenced lines and compare the three incremental and snapshot configurations. Review log.txt alongside the full-table fallback setting and the reported executor shutdowns. Done means explaining when fallback is selected and whether the observed behavior matches the expected file-not-found-only fallback.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- aws, scala
- Domain
- data-engineering, distributed-systems, performance
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100