Spark: Add streaming-starting-offset read option to control stream start position

[sergiomartinswhg]

Context

This PR addresses a common streaming pattern in Iceberg by adding the ability to dictate exactly where a Spark stream starts reading when no previous checkpoint exists.

Related PRs:

• Please also see Spark: Add streaming-overwrite-mode option for handling OVERWRITE snapshots #15152 (currently open and awaiting review).
• Together, this PR (starting offsets) and (overwrite modes) provide users with strong and granular control over how Spark streams consume Iceberg tables. Reviews are highly appreciated!

Summary

This PR adds a new streaming-starting-offset option that provides flexibility for bootstrapping Spark Structured Streaming queries. Previously, Iceberg streams defaulted to starting from the oldest snapshot or required a specific stream-from-timestamp. This PR introduces four distinct starting modes, allowing users to easily skip historical data or perform a full table scan bootstrap before switching to incremental processing.

Motivation

When starting a new stream against an existing Iceberg table, users often need specific starting points. Currently, the default behavior incrementally processes from the oldest available (unexpired) snapshot. While useful for historical backfills, this lacks support for two very common streaming architectures:

• State Bootstrapping: Reading the entire current state of the table as a single initial batch to build downstream state, then seamlessly switching to streaming new changes incrementally.
• Processing New Data Only: Ignoring all existing historical data to strictly process data appended after the stream is initialized.

Changes

New option: streaming-starting-offset providing four distinct starting behaviors:

Value	Behavior
earliest (default)	Starts from the oldest available snapshot, streaming files added by each snapshot incrementally.
latest	Skips all existing data. Only snapshots appended after the stream starts are processed.
earliest-snapshot	Reads all files in the oldest available snapshot as the first micro-batch, then switches to incremental.
latest-snapshot	Reads all files in the current snapshot as the first micro-batch, then switches to incremental.

Compatibility & Precedence:

• streaming-starting-offset is entirely ignored if a checkpoint already exists.
• streaming-starting-offset and stream-from-timestamp are mutually exclusive in intent.
• If both are set, stream-from-timestamp takes precedence to maintain backward compatibility, and a warning is logged.

Usage

// Only process new data; skip everything already in the table
val df = spark.readStream
    .format("iceberg")
    .option("streaming-starting-offset", "latest")
    .load("catalog.db.table")

// Bootstrap from a full table scan of the current snapshot, then go incremental
val df = spark.readStream
    .format("iceberg")
    .option("streaming-starting-offset", "latest-snapshot")
    .load("catalog.db.table")

All feedback and reviews are highly welcome!

[github-actions]

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.

[github-actions]

This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.