[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 called initPartitionedSplits() / initNonPartitionedSplits(), which internally invokes getSnapshotAndLogSplits(). This method may produce mixed split types — HybridSnapshotLogSplit for buckets with KV snapshots and LogSplit for buckets without — which the Flink connector does not support merging in batch mode.

This fix replaces the fallback path with initLogTablePartitionSplits() / getLogSplit() to generate uniform LogSplit for all buckets, avoiding the mixed split type issue.

Changes

• Partitioned tables: initPartitionedSplits() → initLogTablePartitionSplits()
• Non-partitioned tables: initNonPartitionedSplits() → getLogSplit(null, null)

[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..