[spark] support batch read from fluss cluster

[YannByron]

Purpose

Linked issue: close #2376

Brief change log

Tests

API and Format

Documentation

[wuchong]

@YannByron thanks, I left some comments.

[wuchong]

Since this is a batch InputPartition, we should add an end offset to make the log split bounded. The latest end offset can be got from OffsetsInitializer.latest().getBucketOffsets(..) method.

We should:

1. fetch the latest kvSnapshots, it is a map<bucket, snapshot_id&log_start_offset>.
2. fetch the latest offset from OffsetsInitializer.latest, it is a map<bucket, log_end_offset>.
3. Join the kvSnapshots and OffsetsInitializer.latest, to generate a input partition list for each bucket.

[YannByron]

OK, then we should move OffsetsInitializer-related code to fluss-client first.

[wuchong]

logScanner.poll() is a best-effort API: it may return an empty result due to transient issues (e.g., network glitches) even when unread log records remain on the server. Therefore, we should poll in a loop until we reach the known end offset.

The end offset should be determined at job startup using OffsetsInitializer.latest().getBucketOffsets(...), which gives us the high-watermark for each bucket at the beginning of the batch job.

Since there’s no built-in API to read a bounded log split, we must manually:

• Skip any records with offsets beyond the precomputed end offset, and
• Signal there is no next once all buckets have reached their respective end offsets.

[YannByron]

Please make sure that BatchScanner.pollBatch() is also a best-effort API?

[wuchong]

@YannByron the differnece is that LogScanner is unbounded and BatchScanner is bounded. So when org.apache.fluss.client.table.scanner.log.LogScanner#poll returns an empty ScanRecords, it doesn't mean the source is finished. When the BatchScanner.pollBatch() returns null iterator, it means it reaches to the batch end.

[YannByron]

Understand. Thank you.

[wuchong]

The LogRecord is a changelog that contains -D (delete) and -U (update-before) records. To produce a consistent view, we need to merge these changes with the KV snapshot data in a union-read fashion—just like how we combine data lake snapshots with changelogs.

Fortunately, the KV snapshot scan is already sorted by primary key. We can leverage this by:

1. Materializing the delta changes into a temporary delta table;
2. Sorting the delta table by primary key using org.apache.fluss.row.encode.KeyEncoder#of(...);
3. Performing a sort-merge between the sorted KV snapshot reader and the sorted delta table reader.

This enables an efficient and correct merge without requiring random lookups or hash-based joins.

[YannByron]

So, I should fetch and keep all the changes between starting offset and stopping offset since the delta changes is no sorted, then sort then, and execute a sort-merge with kv snapshot, right?
And, if there are too many changes, spill is necessary in the future.

[wuchong]

@YannByron you are right. But Flink connector also didn't implement spill logic yet.