apache / apache/iceberg

Spark: Async micro-batch preload does not stop when either limit is reached

Open Beginner friendly
#17,951 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.