[spark] Support time-range incremental batch reads

[Yohahaha]

Summary

Support time-range incremental batch reads in the Spark connector, so pipelines can read only the data written within a [start, end) window (start inclusive, end exclusive).

New options (per-query read options, only read from scan options / TVF args — never from session configuration, so a window can't leak into later reads; batch-only, streaming ignores them):

Option	Default	Meaning
scan.incremental.start.timestamp	(none)	Enables the incremental read; inclusive lower bound. Epoch millis or yyyy-MM-dd HH:mm:ss in the Spark session time zone.
scan.incremental.end.timestamp	latest	Exclusive upper bound; latest stops at the latest committed data captured at planning time.
scan.incremental.timestamp.out-of-range	error	If the start predates the earliest data retained by Fluss (bounded by table.log.ttl): error fails fast (default), adjust clamps to the earliest retained offset.

New TVF fluss_incremental_between_timestamp(table, start[, end]) for pure SQL — sugar over the options above, registered via FlussSparkSessionExtensions; arguments accept string/integral/TIMESTAMP constant expressions (e.g. rolling past hour windows computed in SQL). Note: unlike Paimon's similarly named function, the window is start-inclusive/end-exclusive.

Read semantics per table type:

• Log table: raw records appended within the window.
• Primary key table: keys inserted/updated in the window, folded to their latest value as of the window end (reads only the changelog range, no kv snapshot; keys deleted in the window are excluded).
• Lake-enabled table: same as above, but always reads from Fluss only — an incremental read never unions the lake snapshot.

Also:

• scan.startup.mode is clarified to affect streaming reads only; plain batch reads remain full-table regardless, so default behavior is unchanged.
• Out-of-range end handling: an end timestamp in the future is rejected by the server; empty [start, stop) buckets emit no partitions.
• Docs: website/docs/engine-spark/reads.md (time-range batch read section) and options.md.

Fixes #3842

Test Plan

• New unit tests: FlussOffsetInitializersTest
• New integration tests: SparkTimeRangeTvfTest, SparkLakeTimeRangeReadTest
• mvn spotless:check passes on affected modules

🤖 AI-assisted changes - reviewed by human developer

[Yohahaha]

@YannByron @fresh-borzoni @luoyuxia @beryllw PTAL, this is an actual customer requirement.

[YannByron]

Thanks for the PR — the feature direction makes sense and the docs are unusually thorough. I traced the PK changelog-fold semantics (snapshotId = -1 -> reader skips the snapshot -> SortMergeReader drops delete rows) down through FlussUpsertPartitionReader and SortMergeReader, and it does behave as described.

I have three must-fix findings before this lands, all on the incremental upsert path plus the option parsing. Details are inline; summary:

1. read.optimized=true combined with an incremental PK read silently returns zero rows, with no error. This is the one I would call a genuine bug.
2. failOnTimestampOutOfRange uses .get and Enumeration.withName, and is evaluated eagerly for every batch read — so a bad value of an incremental-only option breaks plain full-table reads with a bare NoSuchElementException.
3. createIncrementalUpsertPartitions is missing the empty-range guard that the append path gained in this same PR, so empty buckets each still spin up a Spark task and a Fluss connection.

I also collected some non-blocking notes (short-circuiting the lake-snapshot probe in incremental mode, end-side handling versus the server's strict ts > now rejection in Replica#getOffsetByTimestamp, the scan.incremental.* naming versus Fluss's existing scan.startup.* vocabulary, and the Thread.sleep-based test timing). I left those out to keep this review focused — happy to post them separately if useful.

[YannByron]

Must fix: this can break plain (non-incremental) batch reads, and the failure modes are unfriendly.

Two problems in this method:

1. .get on the Option. incrementalOption filters blank values, so scan.incremental.timestamp.out-of-range="" yields None and this throws NoSuchElementException: None.get.
2. Enumeration.withName throws NoSuchElementException: No value found for 'WARN' on any invalid value — not an IllegalArgumentException, and it never tells the user which values are legal.

What amplifies both is the call site: failOnOutOfRange is a strict val in the planner body (SplitPlanner.scala:284 for append, :658 for upsert), so it is evaluated on every batch read, including non-incremental ones. A typo in this incremental-only option therefore fails an ordinary full-table read with a bare NoSuchElementException, which is quite hard to trace back to the option that caused it.

Suggestion: make the call sites lazy val so it is only evaluated in incremental mode, use getOrElse(option.defaultValue()) instead of .get, and throw an IllegalArgumentException naming the supported values — the same style startOffsetsInitializer already uses just below (L106-L113), which reads well.

[YannByron]

Must fix: missing empty-range guard here, asymmetric with the append path.

The append planner gained exactly this guard in the same PR (L365-L371):

if (startOffset >= stopOffset) {
  // Empty range (e.g. a time-range window with no data, or an empty bucket): emit no
  // partition so the append reader is not handed an invalid [start, start) range.
  Seq.empty[InputPartition]
}

but here every bucket emits a partition unconditionally. Unlike createUpsertPartitions, which starts from the EARLIEST_OFFSET sentinel and so can never hit start == stop, this path resolves a concrete start offset — so an empty window in a bucket/partition is routinely reachable.

To be clear, it does not throw: FlussUpsertPartitionReader's logScanFinished (logStartingOffset >= logStoppingOffset) short-circuits the poll loop, and snapshotId == -1 skips the snapshot, so the task cleanly yields 0 rows. But it is not free either — FlussPartitionReader.rowType is a strict val, so merely constructing the reader forces conn -> table -> getTableInfo. Every empty partition therefore costs a Spark task, a Fluss Connection and an RPC. On a partitioned PK table that is numPartitions x numBuckets empty tasks even when the window only touched one partition.

For reference, createLakeUpsertPartition also guards (if (!needLogSplit && !needLakeSplit) return None), so this is the only one of the four planner paths that emits unconditionally. Mirroring the append guard would make them consistent and remove the reliance on the reader's short-circuit as the sole safety net.

Note also that the existing "empty window returns no rows" test builds a log table, so it exercises the append path that already has the guard — the upsert empty-range case looks uncovered.

[YannByron]

Must fix: snapshotId = -1 combined with read.optimized=true silently returns zero rows.

FlussUpsertPartitionReader gates both of its inputs:

val logIterator = if (logScanFinished || readOptimized) {   // L194
  CloseableIterator.emptyIterator[KeyValueRow]()
} else {
  createLogChangesIterator()
}

val snapshotIterators = if (snapshotId == -1) { null } else { createSnapshotIterator() }  // L200

An incremental read always passes -1 here — correctly, that is the established "no kv snapshot" sentinel. But when read.optimized is on, the log side is emptied as well, so SortMergeReader receives two empty inputs and the query returns an empty result with no error and no warning.

readOptimized comes from flussConfig (FlussUpsertPartitionReader.scala:57), i.e. the session-level spark.sql.fluss.read.optimized. A deployment that enables it globally for performance would get silently empty results from every incremental TVF query — arguably the worst failure mode for an incremental pipeline, since it is indistinguishable from "no data was written in this window".

read.optimized has no meaning for an incremental read (there is no snapshot to read optimized). Suggest either ignoring it on this path (always read the log) or failing fast with an explicit message. Either way the combination deserves a test; it is currently uncovered.

Nit while here: -1 is now a magic number in three places (SplitPlanner.scala:738, this line, and FlussUpsertPartitionReader.scala:200). The Flink side has a named constant for it (HybridSnapshotLogSplit.NO_SNAPSHOT_ID); a NO_SNAPSHOT_ID on the FlussUpsertInputPartition companion object would be a cheap cleanup.