Spark Trigger.AvailableNow fails when the initial streaming snapshot is expired
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 11h
- Merged PRs (30d)
- 132
Description
### Apache Iceberg version
1.11.0 (latest release)
### Query engine
Spark
### Please describe the bug 🐞
When using an Iceberg table as a Spark Structured Streaming source with `Trigger.AvailableNow`, restarting the query can fail after `expire_snapshots` removes the snapshot referenced by the stream's initial offset.
The checkpoint may already contain newer committed offsets whose snapshots are still available, but `prepareForTriggerAvailableNow()` computes the target offset from `initialOffset`:
```java
lastOffsetForTriggerAvailableNow =
(StreamingOffset) latestOffset(initialOffset, ReadLimit.allAvailable());
```
`latestOffset()` eventually validates the snapshot referenced by `initialOffset`. If this old snapshot has been expired, the query fails with:
```text
Cannot load current offset at snapshot ,
the snapshot was expired or removed
```
Example:
```text
initial offset: S0 -> expired
latest committed offset: S30 -> available
current snapshot: S40 -> available
```
The stream still fails because `S0` is used by `prepareForTriggerAvailableNow()`.
### Expected behavior
Once the checkpoint has advanced, the original initial snapshot should not need to be retained indefinitely. `AvailableNow` should resume from the current checkpoint position without requiring the snapshot referenced only by the original `initialOffset`.
### Environment
* Spark 4.0
* Iceberg 1.11.0
* `Trigger.AvailableNow`
* Iceberg `readStream`
* Persistent checkpoint
* Regular `expire_snapshots`
### Willingness to contribute
- [x] I can contribute a fix for this bug independently
- [x] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time
Contributor guide
Research direction
Start at prepareForTriggerAvailableNow() and follow latestOffset(initialOffset, ReadLimit.allAvailable()) through the Spark Structured Streaming source offset validation. Reproduce the failure with Trigger.AvailableNow, a persistent checkpoint, and expire_snapshots removing the initial snapshot. Done means a restarted stream uses the advanced checkpoint position without requiring that expired initial snapshot.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, spark
- Domain
- data-engineering, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 52/100