[spark] Advance streaming source past empty full snapshots

[lilei1128]

Purpose

When a Spark streaming source started in latest-full or
from-snapshot-full mode from an empty snapshot, no end offset was
generated and the source could not continue with subsequent delta
snapshots. With consumer-id configured, the consumer checkpoint also
remained unchanged.

Generate a completion offset for empty full snapshots, advance the
source and consumer to the next snapshot, preserve correct
ReadMinRows behavior, and support checkpoint recovery while keeping
legacy offset compatibility.

• Mark an existing zero-split full snapshot as completed and generate an end offset.

• Store the next snapshot position in that offset and resume subsequent reads in delta mode.

• Advance the Paimon consumer when the empty full snapshot is committed.

• Keep ReadMinRows semantics for batches containing actual delta data.

• Preserve the empty-snapshot state across Spark checkpoint serialization and restart.

• Keep legacy offsets without totalSplits readable.

• Prevent inconsistent totalSplits metadata when offsets are copied.

  fix [spark] Streaming source cannot advance past an empty initial full snapshot #9308

Tests

Tests cover：

• Empty snapshot offset serialization and copy behavior;
• Source advancement with and without consumer-id;
• Consumer checkpoint updates and retry failures;
• ReadMinRows delay and recovery;
• Spark query restart from checkpoint;
• Continued consumption of data after the empty snapshot;
• Existing full-snapshot, delta, trigger, and read-limit behavior.

[JingsongLi]

+1