[spark] Advance streaming source past empty full snapshots

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala

@@ -127,7 +127,9 @@ class PaimonMicroBatchStream(
   private def normalizeStartOffset(start: Offset): PaimonSourceOffset = {
     val startOffset = PaimonSourceOffset(start)
     val snapshotCompleted = startOffset.snapshotCompleted
-    val resumeSnapshotId = if (snapshotCompleted) {
+    val resumeSnapshotId = if (startOffset.emptySnapshotCompleted) {
+      startOffset.snapshotId
+    } else if (snapshotCompleted) {
       startOffset.snapshotId + 1
     } else {
       startOffset.snapshotId
@@ -220,6 +222,8 @@ class PaimonMicroBatchStream(
     consumerId.foreach {
       id =>
         offset.totalSplits match {
+          case Some(0L) if offset.emptySnapshotCompleted =>
+            notifyConsumerCheckpointComplete(offset.snapshotId)
           case Some(totalSplits) if offset.index >= totalSplits =>
             throw new IllegalStateException(
               s"Invalid Paimon source offset $offset: split index must be smaller than " +

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSourceOffset.scala

@@ -49,7 +49,11 @@ case class PaimonSourceOffset(snapshotId: Long, index: Long, scanSnapshot: Boole
       index: Long = this.index,
       scanSnapshot: Boolean = this.scanSnapshot): PaimonSourceOffset = {
     val copied = PaimonSourceOffset(snapshotId, index, scanSnapshot)
-    if (snapshotId == this.snapshotId && scanSnapshot == this.scanSnapshot) {
+    if (
+      snapshotId == this.snapshotId &&
+      scanSnapshot == this.scanSnapshot &&
+      totalSplitsValue.forall(_ > 0 || index == PaimonSourceOffset.INIT_OFFSET_INDEX)
+    ) {
       copied.totalSplitsValue = totalSplitsValue
     }
     copied
@@ -59,6 +63,9 @@ case class PaimonSourceOffset(snapshotId: Long, index: Long, scanSnapshot: Boole
     totalSplits.exists(index == _ - 1)
   }
 
+  /** Whether this offset is the cursor immediately after an empty full snapshot. */
+  private[spark] def emptySnapshotCompleted: Boolean = totalSplits.contains(0L)
+
   override def json(): String = {
     val node = JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.createObjectNode()
     node.put(PaimonSourceOffset.FIELD_SNAPSHOT_ID, snapshotId)
@@ -103,7 +110,11 @@ object PaimonSourceOffset {
       index: Long,
       scanSnapshot: Boolean,
       totalSplits: Long): PaimonSourceOffset = {
-    require(totalSplits > 0, s"Total splits must be positive, but was $totalSplits.")
+    require(
+      totalSplits > 0 ||
+        (totalSplits == 0 && index == INIT_OFFSET_INDEX && !scanSnapshot),
+      s"Total splits must be positive except for an empty full snapshot cursor, but was $totalSplits."
+    )
     val offset = PaimonSourceOffset(snapshotId, index, scanSnapshot)
     offset.totalSplitsValue = Some(totalSplits)
     offset

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/StreamHelper.scala

@@ -22,7 +22,7 @@ import org.apache.paimon.CoreOptions
 import org.apache.paimon.data.BinaryRow
 import org.apache.paimon.spark.SparkTypeUtils
 import org.apache.paimon.table.DataTable
-import org.apache.paimon.table.source.{DataSplit, StreamDataTableScan}
+import org.apache.paimon.table.source.{DataSplit, SnapshotNotExistPlan, StreamDataTableScan}
 import org.apache.paimon.table.source.TableScan.Plan
 import org.apache.paimon.table.source.snapshot.StartingContext
 import org.apache.paimon.utils.{InternalRowPartitionComputer, TypeUtils}
@@ -49,6 +49,10 @@ case class IndexedDataSplit(snapshotId: Long, index: Long, entry: DataSplit) {
   }
 }
 
+private case class BatchResult(
+    indexedDataSplits: Array[IndexedDataSplit],
+    emptyFullSnapshotNextId: Option[Long])
+
 private[spark] trait StreamHelper {
 
   def table: DataTable
@@ -85,29 +89,54 @@ private[spark] trait StreamHelper {
       startOffset: PaimonSourceOffset,
       endOffset: Option[PaimonSourceOffset],
       limit: ReadLimit): Option[PaimonSourceOffset] = {
-    val indexedDataSplits = getBatch(startOffset, endOffset, Some(limit))
-    indexedDataSplits.lastOption
+    val batchResult = getBatchResult(startOffset, endOffset, Some(limit))
+    batchResult.indexedDataSplits.lastOption
       .map {
         ids =>
           val scanSnapshot =
             startOffset.scanSnapshot && ids.snapshotId.equals(startOffset.snapshotId)
           if (includeSnapshotCompletionInOffset) {
-            val totalSplits = ids.totalSplits.getOrElse(
-              throw new IllegalStateException(
-                s"Missing total splits for snapshot ${ids.snapshotId}."))
+            val totalSplits = ids.totalSplits.getOrElse(throw new IllegalStateException(
+              s"Missing total splits for snapshot ${ids.snapshotId}."))
             PaimonSourceOffset.withTotalSplits(ids.snapshotId, ids.index, scanSnapshot, totalSplits)
           } else {
             PaimonSourceOffset(ids.snapshotId, ids.index, scanSnapshot)
           }
       }
+      .orElse {
+        batchResult.emptyFullSnapshotNextId.map {
+          nextSnapshotId =>
+            if (includeSnapshotCompletionInOffset) {
+              PaimonSourceOffset.withTotalSplits(
+                nextSnapshotId,
+                PaimonSourceOffset.INIT_OFFSET_INDEX,
+                scanSnapshot = false,
+                totalSplits = 0L)
+            } else {
+              PaimonSourceOffset(
+                nextSnapshotId,
+                PaimonSourceOffset.INIT_OFFSET_INDEX,
+                scanSnapshot = false)
+            }
+        }
+      }
   }
 
   def getBatch(
       startOffset: PaimonSourceOffset,
       endOffset: Option[PaimonSourceOffset],
       limit: Option[ReadLimit]): Array[IndexedDataSplit] = {
+    getBatchResult(startOffset, endOffset, limit).indexedDataSplits
+  }
+
+  private def getBatchResult(
+      startOffset: PaimonSourceOffset,
+      endOffset: Option[PaimonSourceOffset],
+      limit: Option[ReadLimit]): BatchResult = {
     if (startOffset != null) {
-      if (startOffset.snapshotCompleted) {
+      if (startOffset.emptySnapshotCompleted) {
+        streamScan.restore(startOffset.snapshotId, false)
+      } else if (startOffset.snapshotCompleted) {
         streamScan.restore(startOffset.snapshotId + 1, false)
       } else {
         streamScan.restore(startOffset.snapshotId, startOffset.scanSnapshot)
@@ -116,6 +145,9 @@ private[spark] trait StreamHelper {
 
     val readLimitGuard = limit.flatMap(PaimonReadLimits(_, lastTriggerMillis))
     var hasSplits = true
+    var hasNonEmptyPlan = false
+    var firstPlan = true
+    var emptyFullSnapshotNextId: Option[Long] = None
     def continue: Boolean = {
       hasSplits && readLimitGuard.forall(_.hasCapacity) && endOffset.forall(
         streamScan.checkpoint() <= _.snapshotId)
@@ -125,22 +157,35 @@ private[spark] trait StreamHelper {
     while (continue) {
       val plan = streamScan.plan()
       if (plan.splits.isEmpty) {
-        hasSplits = false
+        val isCompletedEmptyFullSnapshot =
+          firstPlan &&
+            startOffset != null &&
+            startOffset.scanSnapshot &&
+            (plan ne SnapshotNotExistPlan.INSTANCE)
+        if (isCompletedEmptyFullSnapshot) {
+          Option(streamScan.checkpoint()).foreach {
+            nextSnapshotId => emptyFullSnapshotNextId = Some(nextSnapshotId)
+          }
+        } else {
+          hasSplits = false
+        }
       } else {
+        hasNonEmptyPlan = true
         indexedDataSplits ++= convertPlanToIndexedSplits(plan)
           // Filter by (start, end]
           .filter(ids => inRange(ids, startOffset, endOffset))
           // Filter splits by read limits other than ReadMinRows.
           .takeWhile(s => readLimitGuard.forall(_.admit(s)))
       }
+      firstPlan = false
     }
 
     // Filter splits by ReadMinRows read limit if exists.
     // If this batch doesn't meet the condition of ReadMinRows, then nothing will be returned.
-    if (readLimitGuard.exists(_.skipBatch)) {
-      Array.empty
+    if (readLimitGuard.exists(_.skipBatch) && hasNonEmptyPlan) {
+      BatchResult(Array.empty, None)
     } else {
-      indexedDataSplits.toArray
+      BatchResult(indexedDataSplits.toArray, emptyFullSnapshotNextId)
     }
   }
 

paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sources/PaimonSourceOffsetTest.scala

@@ -35,6 +35,36 @@ class PaimonSourceOffsetTest extends AnyFunSuite {
     assert(restored.totalSplits.contains(2L))
   }
 
+  test("round trip an empty snapshot cursor") {
+    val offset = PaimonSourceOffset.withTotalSplits(
+      snapshotId = 4L,
+      index = PaimonSourceOffset.INIT_OFFSET_INDEX,
+      scanSnapshot = false,
+      totalSplits = 0L)
+
+    val restored = PaimonSourceOffset(offset.json())
+
+    assert(restored.emptySnapshotCompleted)
+    assert(restored.snapshotId == 4L)
+    assert(restored.index == PaimonSourceOffset.INIT_OFFSET_INDEX)
+    assert(!restored.scanSnapshot)
+  }
+
+  test("copying an empty snapshot cursor with a different index clears its marker") {
+    val offset = PaimonSourceOffset.withTotalSplits(
+      snapshotId = 4L,
+      index = PaimonSourceOffset.INIT_OFFSET_INDEX,
+      scanSnapshot = false,
+      totalSplits = 0L)
+
+    val copied = offset.copy(index = 0L)
+    val restored = PaimonSourceOffset(copied.json())
+
+    assert(copied.index == 0L)
+    assert(copied.totalSplits.isEmpty)
+    assert(restored.totalSplits.isEmpty)
+  }
+
   test("copy and Java serialization preserve total splits") {
     val offset = offsetWithTotalSplits(scanSnapshot = true)
 

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSourceTest.scala

@@ -537,6 +537,88 @@ class PaimonSourceTest extends PaimonSparkTestBase with StreamTest {
     }
   }
 
+  test("Paimon Source: advance past an empty initial full snapshot") {
+    withTempDir {
+      checkpointDir =>
+        spark.sql("""
+                    |CREATE TABLE T (a INT, b STRING)
+                    |TBLPROPERTIES ('primary-key'='a', 'bucket'='2', 'file.format'='parquet')
+                    |""".stripMargin)
+        val location = loadTable("T").location().toString
+        spark.sql("INSERT INTO T VALUES (1, 'before')")
+        spark.sql("INSERT OVERWRITE T SELECT * FROM T WHERE false")
+
+        val query = spark.readStream
+          .format("paimon")
+          .load(location)
+          .writeStream
+          .format("memory")
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .queryName("empty_full_snapshot")
+          .outputMode("append")
+          .start()
+
+        try {
+          query.processAllAvailable()
+          checkAnswer(spark.sql("SELECT * FROM empty_full_snapshot"), Seq.empty)
+
+          spark.sql("INSERT INTO T VALUES (2, 'after')")
+          query.processAllAvailable()
+          checkAnswer(spark.sql("SELECT * FROM empty_full_snapshot"), Seq(Row(2, "after")))
+        } finally {
+          query.stop()
+        }
+    }
+  }
+
+  test("Paimon Source: resume past an empty initial full snapshot after restart") {
+    withTempDir {
+      checkpointDir =>
+        spark.sql("DROP TABLE IF EXISTS T")
+        spark.sql("""
+                    |CREATE TABLE T (a INT, b STRING)
+                    |TBLPROPERTIES ('primary-key'='a', 'bucket'='2', 'file.format'='parquet')
+                    |""".stripMargin)
+        val location = loadTable("T").location().toString
+        spark.sql("INSERT INTO T VALUES (1, 'before')")
+        spark.sql("INSERT OVERWRITE T SELECT * FROM T WHERE false")
+
+        val targetLocation = prepareTableAndGetLocation(0, false, tableName = "T2").location
+
+        val df = spark.readStream
+          .format("paimon")
+          .load(location)
+          .writeStream
+          .format("paimon")
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+
+        val emptySnapshotId = loadTable("T").snapshotManager().latestSnapshotId()
+        val firstQuery = df.start(targetLocation)
+        try {
+          firstQuery.processAllAvailable()
+          checkAnswer(spark.sql("SELECT * FROM T2"), Seq.empty)
+          val endOffset = PaimonSourceOffset(firstQuery.lastProgress.sources(0).endOffset)
+          assert(endOffset.snapshotId == emptySnapshotId + 1L)
+          assert(endOffset.index == PaimonSourceOffset.INIT_OFFSET_INDEX)
+          assert(!endOffset.scanSnapshot)
+          assert(endOffset.totalSplits.isEmpty)
+        } finally {
+          firstQuery.stop()
+        }
+
+        spark.sql("INSERT INTO T VALUES (2, 'after')")
+
+        val restartedQuery = df.start(targetLocation)
+
+        try {
+          restartedQuery.processAllAvailable()
+          checkAnswer(spark.sql("SELECT * FROM T2"), Seq(Row(2, "after")))
+        } finally {
+          restartedQuery.stop()
+        }
+    }
+  }
+
   test("Paimon Source: from-snapshot and from-snapshot-full scan mode") {
     withTempDirs {
       (checkpointDir1, checkpointDir2) =>