[flink] Fix batch fallback generating mixed split types for primary-key tables

[matrixsparse]

Summary

Follow-up fix for #3208.

For primary-key tables in batch mode, when no lake snapshot exists, the previous
fallback logic had two issues:

1. Mixed split types: initPartitionedSplits() / initNonPartitionedSplits()
   may produce both HybridSnapshotLogSplit and LogSplit, which the Flink
   connector does not support merging.

2. Missing stopping offset: getLogSplit() used the 3-parameter LogSplit
   constructor without stopping offset, causing the reader to never stop in
   batch mode.

Changes

• Use initLogTablePartitionSplits() / getLogSplit() in the fallback path
  to generate uniform LogSplit for all buckets.
• Add stoppingOffsetsInitializer.getBucketOffsets() in getLogSplit() to
  provide stopping offsets for batch mode bounded reads.

[matrixsparse]

Hi @luoyuxia, this is a follow-up fix for the mixed split types issue you mentioned in #3208.

Could you PTAL? Thanks! cc @fresh-borzoni

[fresh-borzoni]

@matrixsparse Ty, LGTM in general, one comment:

It matches spark logic, but at the same time this scenario is a bit dangerous - we have a big table that was never tiered to lake, then we decide to tier it to lake and run batched query through this fallback, instead of using kv_snapshot and replaying log on top, we read from earliest which is potentially a lot of records. So it's not very efficient and potentially OOM prone.

Let's file an issue about this to address separately for Spark and Flink?
cc @luoyuxia WDYT about this plan?

[luoyuxia]

@matrixsparse Thanks for the quick fix. Left on comments. PTAL

[luoyuxia]

Note it'll just genereate log split without stopping offset which will then nerver stop..

[matrixsparse]

@matrixsparse Good catch! Updated getLogSplit() to use stoppingOffsetsInitializer.getBucketOffsets() and pass the stopping offset to the 4-parameter LogSplit constructor. PTAL.

[fresh-borzoni]

@matrixsparse @luoyuxia filed followups:
#3327
#3317 - resolved

[fresh-borzoni]

LGTM 👍