From f14cfe985a524c417dca96ef499115185c14282e Mon Sep 17 00:00:00 2001 From: shisan Date: Wed, 26 Aug 2026 14:12:50 +0800 Subject: [PATCH 1/3] [spark] Advance streaming source past empty full snapshots --- .../sources/PaimonMicroBatchStream.scala | 6 +- .../spark/sources/PaimonSourceOffset.scala | 15 ++- .../paimon/spark/sources/StreamHelper.scala | 67 ++++++++-- .../sources/PaimonSourceOffsetTest.scala | 30 +++++ .../paimon/spark/PaimonSourceTest.scala | 82 ++++++++++++ .../PaimonMicroBatchStreamITCase.scala | 126 +++++++++++++++++- .../sources/PaimonMicroBatchStreamTest.scala | 19 +++ 7 files changed, 327 insertions(+), 18 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 c9133691c324..7fc7548c5eab 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 d0311a35d9c1..b05e1e46e7be 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 68272d32f0f8..c00673009f03 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 0e4e90194250..739443bb4058 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 624c2c9dc7c6..cd6aa554f783 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 c014a2547473..bd4af811dd8a 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) + 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("T") + 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 4f605d71cc50..d4c6965a2268 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, From 25165c6ffe46319f4879d796b12c074546e510f2 Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Wed, 26 Aug 2026 15:01:32 +0800 Subject: [PATCH 2/3] fix code format --- .../sources/PaimonMicroBatchStreamITCase.scala | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) 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 bd4af811dd8a..159def4225e8 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 @@ -401,7 +401,8 @@ class PaimonMicroBatchStreamITCase extends PaimonSparkTestBase { try { restartedQuery.processAllAvailable() checkAnswer(spark.sql("SELECT * FROM T2"), Seq(Row(20, "v_20"))) - } finally {≠≠ + } finally { + ≠≠ restartedQuery.stop() } } @@ -416,11 +417,11 @@ class PaimonMicroBatchStreamITCase extends PaimonSparkTestBase { 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) + |TBLPROPERTIES ( + | 'bucket' = '2', + | 'bucket-key' = 'a', + | 'file.format' = 'parquet' + |)""".stripMargin) loadTable(tableName) } From c28c6d729cfabce01be91aa9d45d2931f141a097 Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Wed, 26 Aug 2026 15:51:45 +0800 Subject: [PATCH 3/3] fix error --- .../paimon/spark/sources/PaimonMicroBatchStreamITCase.scala | 1 - 1 file changed, 1 deletion(-) 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 159def4225e8..ca3add928b4c 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 @@ -402,7 +402,6 @@ class PaimonMicroBatchStreamITCase extends PaimonSparkTestBase { restartedQuery.processAllAvailable() checkAnswer(spark.sql("SELECT * FROM T2"), Seq(Row(20, "v_20"))) } finally { - ≠≠ restartedQuery.stop() } }