This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 031fb354ff [spark] Fix restarted streaming query re-scanning full
snapshot instead of resuming from checkpoint (#8207)
031fb354ff is described below
commit 031fb354ffcf04f4f32601882bc469107365655f
Author: Taeseong Yu <[email protected]>
AuthorDate: Tue Jun 23 16:36:20 2026 +0900
[spark] Fix restarted streaming query re-scanning full snapshot instead of
resuming from checkpoint (#8207)
---
.../apache/paimon/spark/sources/PaimonMicroBatchStream.scala | 12 +++++++++++-
1 file changed, 11 insertions(+), 1 deletion(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala
index e8fbea0e69..c3d2dfc881 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala
@@ -116,7 +116,17 @@ class PaimonMicroBatchStream(
override def planInputPartitions(start: Offset, end: Offset):
Array[InputPartition] = {
val startOffset = {
val startOffset0 = PaimonSourceOffset(start)
- if (startOffset0.compareTo(initOffset) < 0) {
+ // Fall back to initOffset only when the checkpointed snapshot has
expired.
+ // initOffset is recomputed from the current table state on every
(re)start,
+ // so with scan modes like latest-full it points at the current snapshot
with
+ // scanSnapshot=true. Clamping a still-valid checkpointed offset up to
it made
+ // a restarted query silently skip the changelog gap and re-scan the
whole
+ // snapshot, re-emitting every row as +I.
+ if (startOffset0.snapshotId <
table.snapshotManager().earliestSnapshotId()) {
+ logWarning(
+ s"Checkpointed start offset $startOffset0 is no longer available " +
+ s"(earliest snapshot:
${table.snapshotManager().earliestSnapshotId()}), " +
+ s"falling back to $initOffset.")
initOffset
} else {
startOffset0
