[client] Support bounded Log Table scanning #4544

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Several Fluss integrations need to read a bounded range of a Log Table, i.e. records within a per bucket offset range such as [startingOffset, stoppingOffset).

Today this bounded read behavior is implemented or needs to be implemented above the client. This leads Flink, Spark, and the upcoming Trino connector to maintain similar logic for stopping at an offset, trimming the final result, tracking completion, and handling edge cases such as empty ranges, filtering, and retention.

Keeping this logic in each compute engine has two drawbacks:

• duplicated implementation and testing across all connectors;
• subtle behavioral differences may appear over time as the copied implementations evolve independently, especially around offset boundaries, retention, filtering, and completion semantics.

This is also relevant to the native read-only Trino connector proposed in #4197, whose initial scope includes bounded reads of non-Lakehouse Log Tables.

It would be preferable for the Fluss client to provide the bounded-range semantics once and let compute engines build on the same primitive.

Solution

Extend LogScanner with an additive bounded subscription API for reading a bucket within an exclusive stopping offset:

subscribeBounded(bucket, startingOffset, stoppingOffset)
subscribeBounded(partitionId, bucket, startingOffset, stoppingOffset)

with the range semantics:

[startingOffset, stoppingOffset)

The exclusive stopping offset takes precedence over preserving UPDATE_BEFORE/UPDATE_AFTER pairs. A bounded range may therefore end between the two records of an update.

The client should:

• stop exposing records at the exclusive stopping offset;
• report when a bounded bucket has completed;
• support both explicit starting offsets and EARLIEST_OFFSET;
• handle empty ranges and retention correctly;
• provide consistent semantics for row and Arrow-based scans.
• expose offset progress even when filtering produces no records;
• report completion once per bounded subscription, including empty ranges.

The existing unbounded subscription APIs remain unchanged. Existing implementations of LogScanner should remain compatible, and this feature should not introduce public API behavioral changes for existing unbounded Log Table scans.
Correct handling of EARLIEST_OFFSET requires the fetch response to expose the starting offset resolved by the server. The initial implementation therefore includes minimal backward compatible protocol/server support. Explicit starting offsets remain usable with older servers; bounded scans requiring EARLIEST resolution should fail clearly when that support is unavailable.

Anything else?

This issue focuses on establishing a reusable bounded-scan primitive in the Fluss Java client, including the minimal protocol/server support needed for correct EARLIEST resolution.
Connector migrations and server-side enforcement of the stopping offset are separate follow-ups.

I am also actively working on the related follow-up work below.
Planned / in-progress follow-ups include:

1. Flink — migrate bounded Log Table reads to the client-side bounded scanner and remove duplicated offset-boundary/completion logic where applicable.
2. Spark — use the same bounded scanner for batch/bounded Log Table reads.
3. Trino — use this capability for bounded Log Table reads in the native read-only connector tracked by #4197.
4. Tiering/internal readers — evaluate and consolidate existing bounded-range handling where the same client semantics can be reused.
5. Server-side bounded reads — as a later optimization, consider pushing the stopping offset to the server to reduce unnecessary reads and data transfer beyond the requested range.

The server-side optimization is intentionally a follow-up rather than part of the initial PR. The first step is to provide a consistent client-side bounded-read contract while keeping the change isolated and preserving existing unbounded public API behavior.

Follow-ups

• Lake-only EARLIEST offset may prevent log fetch progress: [Bug] Log fetch may get stuck when EARLIEST resolves to a lake-only offset #4676
• FetchLog v1 and authoritative log_start_offset migration: [RPC] Make log_start_offset authoritative in FetchLog v1 #4677

Willingness to contribute

• I'm willing to submit a PR!