Spark: Async micro-batch preload does not stop when either limit is reached
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 11h
- Merged PRs (30d)
- 132
Description
# Spark: Async micro-batch preload does not stop when either limit is reached
### Apache Iceberg version
`main` (`0107f19a4a`); present in Spark 3.5, 4.0, and 4.1.
### Query engine
Spark Structured Streaming with `async-micro-batch-planning-enabled=true`.
### Description
`AsyncSparkMicroBatchPlanner.fillQueueInitialBuffer` combines the initial row and file limits with
`||`:
```java
while ((queuedRowCount.get() < targetRows || queuedFileCount.get() < targetFiles)
&& current.snapshotId() != preloadEndSnapshot.snapshotId()) {
```
This continues preloading until both limits are reached. It should stop when either limit is reached,
which requires `&&` between the two below-limit checks.
This is also consistent with the background-refill path in the same class, which considers the
buffer full when either its row or file threshold is exceeded.
The default initial limits are 100 files and 100,000 rows. For example, if each pending snapshot adds
100 one-row files, the file limit is reached after the first snapshot, but current code can continue
through 1,000 snapshots and retain 100,000 file tasks before reaching the row limit. The inverse
happens when snapshots contain a small number of files with many rows.
This affects startup and catch-up with a snapshot backlog, such as after downtime, restart from an
old checkpoint, or `AvailableNow` processing. Extra preload causes unnecessary manifest planning,
driver allocations, startup latency, and possible memory pressure. It does not cause data loss.
### Proposed fix
Use `&&` in all three Spark versions:
```java
while ((queuedRowCount.get() < targetRows && queuedFileCount.get() < targetFiles)
&& current.snapshotId() != preloadEndSnapshot.snapshotId()) {
```
Add regression coverage for both asymmetric states:
```java
// File limit reached, row limit not reached
assertThat(shouldContinueInitialPreload(1L, 100L, 100_000L, 100L)).isFalse();
// Row limit reached, file limit not reached
assertThat(shouldContinueInitialPreload(100_000L, 1L, 100_000L, 100L)).isFalse();
```
A focused Spark 4.1 test failed with the current `||` condition and passed after changing it to
`&&`.
Snapshots are added atomically, so preload may still exceed a limit by one snapshot. Strict
within-snapshot enforcement is outside this fix.
### Willingness to contribute
- [x] I can contribute a fix independently
- [ ] I would be willing to contribute a fix with guidance
- [ ] I cannot contribute a fix at this time
Contributor guide
Research direction
Start with AsyncSparkMicroBatchPlanner.fillQueueInitialBuffer and compare its initial-preload condition with the background-refill path in the same class. Run or extend the focused Spark 4.1 regression test for both asymmetric row/file-limit states, then apply the matching fix across the three Spark versions and verify the tests pass.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, spark
- Domain
- data-engineering, stream-processing
- Issue type
- Bug
- Difficulty
- 2/5
- Estimated time
- 1-3 hours
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 90/100