apache / apache/iceberg

Spark Trigger.AvailableNow fails when the initial streaming snapshot is expired

Open
#18,000 0 comments 0 reactions 0 assignees View on GitHub
bug
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

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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.