iamrajatrana opened a new issue, #17951:
URL: https://github.com/apache/iceberg/issues/17951
# 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
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
