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 483c96854a [spark] Advance streaming source past empty full snapshots
(#9401)
483c96854a is described below
commit 483c96854a71a816d8b12013c7e0f958cff171b9
Author: sanshi <[email protected]>
AuthorDate: Thu Aug 27 15:25:43 2026 +0800
[spark] Advance streaming source past empty full snapshots (#9401)
---
.../spark/sources/PaimonMicroBatchStream.scala | 6 +-
.../paimon/spark/sources/PaimonSourceOffset.scala | 15 ++-
.../apache/paimon/spark/sources/StreamHelper.scala | 67 ++++++++--
.../spark/sources/PaimonSourceOffsetTest.scala | 30 +++++
.../org/apache/paimon/spark/PaimonSourceTest.scala | 82 +++++++++++++
.../sources/PaimonMicroBatchStreamITCase.scala | 136 +++++++++++++++++++--
.../spark/sources/PaimonMicroBatchStreamTest.scala | 19 +++
7 files changed, 332 insertions(+), 23 deletions(-)
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 c9133691c3..7fc7548c5e 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
@@ -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 " +
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSourceOffset.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSourceOffset.scala
index d0311a35d9..b05e1e46e7 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSourceOffset.scala
+++
b/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
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/StreamHelper.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/StreamHelper.scala
index 68272d32f0..c00673009f 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/StreamHelper.scala
+++
b/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)
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sources/PaimonSourceOffsetTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sources/PaimonSourceOffsetTest.scala
index 0e4e901942..739443bb40 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sources/PaimonSourceOffsetTest.scala
+++
b/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)
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSourceTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSourceTest.scala
index 624c2c9dc7..cd6aa554f7 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSourceTest.scala
+++
b/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) =>
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamITCase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamITCase.scala
index c014a25474..ca3add928b 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamITCase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamITCase.scala
@@ -24,6 +24,7 @@ import org.apache.paimon.spark.PaimonSparkTestBase
import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.table.source.OutOfRangeException
+import org.apache.spark.sql.Row
import org.apache.spark.sql.connector.read.streaming.ReadLimit
import java.util.{Collections, HashMap}
@@ -54,6 +55,79 @@ class PaimonMicroBatchStreamITCase extends
PaimonSparkTestBase {
assert(!latest.json().contains("totalSplits"))
}
+ test("advance past an empty initial full snapshot without a consumer") {
+ val sourceTable = createTableWithOneSnapshot()
+ spark.sql("INSERT OVERWRITE T SELECT * FROM T WHERE false")
+
+ val stream = createStream(sourceTable)
+ val initial = stream.initialOffset().asInstanceOf[PaimonSourceOffset]
+ val emptySnapshotId = sourceTable.snapshotManager().latestSnapshotId()
+ assert(initial.snapshotId == emptySnapshotId)
+ assert(initial.scanSnapshot)
+
+ stream.lastTriggerMillis = System.currentTimeMillis()
+ val emptyEnd = latestOffset(stream, initial, ReadLimit.minRows(100L,
60000L))
+ assert(emptyEnd.snapshotId == emptySnapshotId + 1L)
+ assert(emptyEnd.index == PaimonSourceOffset.INIT_OFFSET_INDEX)
+ assert(!emptyEnd.scanSnapshot)
+
+ spark.sql("INSERT INTO T VALUES (20, 'v_20')")
+ val nextEnd = latestOffset(stream, emptyEnd, ReadLimit.allAvailable())
+
+ assert(nextEnd.snapshotId == emptyEnd.snapshotId)
+ assert(!nextEnd.scanSnapshot)
+ assert(stream.planInputPartitions(emptyEnd, nextEnd).nonEmpty)
+ }
+
+ test("advance consumer past an empty initial full snapshot") {
+ val sourceTable = createTableWithOneSnapshot()
+ spark.sql("INSERT OVERWRITE T SELECT * FROM T WHERE false")
+ val sourceTableWithConsumer = withConsumer(sourceTable)
+
+ val stream = createStream(sourceTableWithConsumer)
+ val initial = stream.initialOffset().asInstanceOf[PaimonSourceOffset]
+ val emptySnapshotId = sourceTable.snapshotManager().latestSnapshotId()
+ val emptyEnd = latestOffset(stream, initial, ReadLimit.allAvailable())
+
+ assert(emptyEnd.snapshotId == emptySnapshotId + 1L)
+ assert(emptyEnd.index == PaimonSourceOffset.INIT_OFFSET_INDEX)
+ assert(!emptyEnd.scanSnapshot)
+ assert(emptyEnd.emptySnapshotCompleted)
+
+ stream.commit(emptyEnd)
+ assert(consumerNextSnapshot(sourceTableWithConsumer) == emptySnapshotId +
1L)
+ }
+
+ test("do not advance an empty full snapshot past a delta deferred by
ReadMinRows") {
+ val sourceTable = createTableWithOneSnapshot()
+ spark.sql("INSERT OVERWRITE T SELECT * FROM T WHERE false")
+ val stream = createStream(sourceTable)
+ val initial = stream.initialOffset().asInstanceOf[PaimonSourceOffset]
+ spark.sql("INSERT INTO T VALUES (20, 'v_20')")
+ stream.lastTriggerMillis = System.currentTimeMillis()
+
+ val deferred = stream.latestOffset(initial, ReadLimit.minRows(100L,
60000L))
+
+ assert(deferred == null)
+ }
+
+ test("resume an empty full snapshot after ReadMinRows delay") {
+ val sourceTable = createTableWithOneSnapshot()
+ spark.sql("INSERT OVERWRITE T SELECT * FROM T WHERE false")
+ val stream = createStream(sourceTable)
+ val initial = stream.initialOffset().asInstanceOf[PaimonSourceOffset]
+ spark.sql("INSERT INTO T VALUES (20, 'v_20')")
+
+ stream.lastTriggerMillis = System.currentTimeMillis()
+ assert(stream.latestOffset(initial, ReadLimit.minRows(100L, 60000L)) ==
null)
+
+ stream.lastTriggerMillis = System.currentTimeMillis() - 60001L
+ val resumed = latestOffset(stream, initial, ReadLimit.minRows(100L,
60000L))
+ assert(resumed.snapshotId ==
sourceTable.snapshotManager().latestSnapshotId())
+ assert(!resumed.scanSnapshot)
+ assert(stream.planInputPartitions(initial, resumed).nonEmpty)
+ }
+
test("create consumer only after the initial full snapshot is completely
consumed") {
val sourceTable = withConsumer(createTableWithOneSnapshot())
val stream = createStream(sourceTable)
@@ -289,21 +363,65 @@ class PaimonMicroBatchStreamITCase extends
PaimonSparkTestBase {
}
}
+ test("Spark query restarts after an empty initial full snapshot with a
consumer") {
+ withTempDir {
+ checkpointDir =>
+ val sourceTable = createTableWithOneSnapshot()
+ spark.sql("INSERT OVERWRITE T SELECT * FROM T WHERE false")
+ val location = sourceTable.location().toString
+ val emptySnapshotId = sourceTable.snapshotManager().latestSnapshotId()
+ val targetTable = createTableWithoutSnapshot("T2")
+ val targetLocation = targetTable.location().toString
+
+ val df = spark.readStream
+ .format("paimon")
+ .option(CoreOptions.CONSUMER_ID.key(), consumerId)
+ .load(location)
+ .writeStream
+ .format("paimon")
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+
+ 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.emptySnapshotCompleted)
+ } finally {
+ firstQuery.stop()
+ }
+
+ spark.sql("INSERT INTO T VALUES (20, 'v_20')")
+
+ val restartedQuery = df.start(targetLocation)
+
+ try {
+ restartedQuery.processAllAvailable()
+ checkAnswer(spark.sql("SELECT * FROM T2"), Seq(Row(20, "v_20")))
+ } finally {
+ restartedQuery.stop()
+ }
+ }
+ }
+
private def createTableWithOneSnapshot(): FileStoreTable = {
createTableWithoutSnapshot()
spark.sql("INSERT INTO T VALUES (10, 'v_10'), (11, 'v_11'), (12, 'v_12')")
loadTable("T")
}
- private def createTableWithoutSnapshot(): FileStoreTable = {
- spark.sql("DROP TABLE IF EXISTS T")
- spark.sql("""CREATE TABLE T (a INT, b STRING)
- |TBLPROPERTIES (
- | 'bucket' = '2',
- | 'bucket-key' = 'a',
- | 'file.format' = 'parquet'
- |)""".stripMargin)
- loadTable("T")
+ private def createTableWithoutSnapshot(tableName: String = "T"):
FileStoreTable = {
+ spark.sql(s"DROP TABLE IF EXISTS $tableName")
+ spark.sql(s"""CREATE TABLE $tableName (a INT, b STRING)
+ |TBLPROPERTIES (
+ | 'bucket' = '2',
+ | 'bucket-key' = 'a',
+ | 'file.format' = 'parquet'
+ |)""".stripMargin)
+ loadTable(tableName)
}
private def withConsumer(table: FileStoreTable): FileStoreTable = {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamTest.scala
index 4f605d71cc..d4c6965a22 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamTest.scala
@@ -68,6 +68,25 @@ class PaimonMicroBatchStreamTest extends AnyFunSuite {
verify(scan, times(2)).notifyCheckpointComplete(6L)
}
+ test("propagate empty snapshot consumer update failure and allow retry") {
+ val (stream, scan) = createStreamWithConsumer()
+ val empty = PaimonSourceOffset.withTotalSplits(
+ snapshotId = 6L,
+ index = PaimonSourceOffset.INIT_OFFSET_INDEX,
+ scanSnapshot = false,
+ totalSplits = 0L)
+ val failure = new UncheckedIOException(new IOException("expected failure"))
+ doThrow(failure).doNothing().when(scan).notifyCheckpointComplete(6L)
+
+ val thrown = intercept[UncheckedIOException] {
+ stream.commit(empty)
+ }
+ assert(thrown eq failure)
+
+ stream.commit(empty)
+ verify(scan, times(2)).notifyCheckpointComplete(6L)
+ }
+
private def consumerOffset(index: Long, totalSplits: Long):
PaimonSourceOffset = {
PaimonSourceOffset.withTotalSplits(
snapshotId = 5L,
