[I] Spark: Async micro-batch preload does not stop when either limit is reached [iceberg]

["iamrajatrana (via GitHub)" <gi...@apache.org>]

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: issues-unsubscribe@iceberg.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org