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 a468f669f7 [spark] Support consumer progress in streaming source
(#9273)
a468f669f7 is described below
commit a468f669f75171bb476d98fdaa5cf740ed569986
Author: LsomeYeah <[email protected]>
AuthorDate: Thu Aug 20 10:11:16 2026 +0800
[spark] Support consumer progress in streaming source (#9273)
---
docs/docs/spark/structured-streaming.md | 90 ++++++
.../spark/sources/PaimonMicroBatchStream.scala | 124 ++++++--
.../paimon/spark/sources/PaimonSourceOffset.scala | 47 ++-
.../apache/paimon/spark/sources/StreamHelper.scala | 52 +++-
.../spark/sources/PaimonSourceOffsetTest.scala | 135 ++++++++
.../sources/PaimonMicroBatchStreamITCase.scala | 339 +++++++++++++++++++++
.../spark/sources/PaimonMicroBatchStreamTest.scala | 89 ++++++
7 files changed, 837 insertions(+), 39 deletions(-)
diff --git a/docs/docs/spark/structured-streaming.md
b/docs/docs/spark/structured-streaming.md
index 41c0a19dbf..bee6d80aa3 100644
--- a/docs/docs/spark/structured-streaming.md
+++ b/docs/docs/spark/structured-streaming.md
@@ -115,6 +115,96 @@ val query = spark.readStream
.start()
```
+### Consumer progress
+
+You can assign a Paimon Consumer to a streaming query. The Consumer records a
+table-side, snapshot-level recovery position and acts as a fence during normal
+snapshot expiration. For tables with a decoupled changelog lifecycle, enabling
+`consumer.changelog-only` makes the Consumer protect long-lived changelogs
+instead of snapshots.
+
+Configure the Consumer lifetime as a table property before starting the query:
+
+```sql
+ALTER TABLE table_name SET TBLPROPERTIES (
+ 'consumer.expiration-time' = '1 d'
+);
+```
+
+```scala
+val query = spark.readStream
+ .format("paimon")
+ .option("consumer-id", "my-consumer")
+ .table("table_name")
+ .writeStream
+ .format("console")
+ .option("checkpointLocation", "/path/to/spark/checkpoint")
+ .start()
+```
+
+Spark checkpoint and Paimon Consumer progress have different granularities:
+
+- When the Spark checkpoint is available, Spark uses it for precise micro-batch
+ recovery.
+- A new query without that checkpoint starts from the Consumer position. This
+ recovery is conservative and may replay a completed snapshot if the query
+ failed after processing it but before updating the Consumer.
+
+Paimon updates the Consumer only from Spark's successful micro-batch commit
+callback. When a micro-batch ends partway through a delta snapshot, the
Consumer
+may be created or refreshed at that snapshot so that the whole snapshot remains
+protected and can be replayed. The Consumer advances past a snapshot only after
+the micro-batch containing its last split is committed. A Consumer update
failure
+is propagated, fails the running query, and leaves the Consumer at its previous
+conservative position.
+
+Spark never advances Consumer progress past an incomplete snapshot. An
incomplete
+initial full snapshot does not create a Consumer because Consumer recovery from
+that snapshot would use a delta scan. `consumer.mode` does not select a
different
+Spark source implementation.
+
+Offsets restored from an older Spark checkpoint do not contain snapshot
+completion metadata. Spark logs a warning and leaves the Consumer unchanged for
+those offsets rather than advancing it without proof that the snapshot
finished.
+
+If the Consumer does not exist, the configured startup options are used. For
+example, the default `latest-full` mode first reads the full snapshot and
creates
+the Consumer only after that snapshot is completely committed. If the Consumer
+already exists, its next snapshot takes precedence over startup options and is
+read incrementally, unless `consumer.ignore-progress` is enabled.
+
+:::note
+
+Spark can invoke the source commit callback while constructing a later
+micro-batch. Therefore the Consumer may temporarily lag behind the latest
+successful batch, especially while a query is idle or after its final available
+batch. This is safe and can only cause conservative replay.
+
+:::
+
+Consumer files are scoped to the branch being read. Use one active Spark query
+for each `(table, branch, consumer-id)` combination. Concurrent queries sharing
+the same Consumer are unsupported because a faster query could move the shared
+position past data still needed by a slower query.
+
+Configure `consumer.expiration-time` according to the longest expected snapshot
+processing time, failure recovery time, and idle interval. Eligible Spark
source
+commit callbacks refresh the Consumer file, but there is no separate
timer-based
+Spark heartbeat. Expired Consumer files are cleaned by non-write-only table
+commits. A write-only writer does not perform Consumer expiration, so stale
+Consumer IDs must be managed explicitly or by a separate non-write-only
+maintenance process.
+
+An initial full scan has no Consumer retention fence until it completes, so
+snapshot retention should be long enough for that scan. If its snapshot expires
+first, a restart without a Spark checkpoint performs a new initial full scan
+according to the startup options.
+
+When expired data would force recovery of a logged Spark batch to switch
between
+a full and an incremental scan, Paimon fails the query instead of combining the
+two plans. Start a new query with a new checkpoint to apply the startup options
+again.
+
Paimon Structured Streaming also supports a variety of streaming read modes,
it can support many triggers and many read limits.
These read limits are supported:
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 96883c04c6..c9133691c3 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
@@ -23,7 +23,7 @@ import org.apache.paimon.options.Options
import org.apache.paimon.schema.TableSchema
import org.apache.paimon.spark.{PaimonImplicits,
PaimonMicroBatchInputPartition, PaimonMicroBatchMetadata,
PaimonPartitionReaderFactory, SparkConnectorOptions}
import org.apache.paimon.table.DataTable
-import org.apache.paimon.table.source.{DataSplit, ReadBuilder}
+import org.apache.paimon.table.source.{DataSplit, OutOfRangeException,
ReadBuilder}
import org.apache.paimon.utils.DataEvolutionUtils
import org.apache.spark.internal.Logging
@@ -47,13 +47,18 @@ class PaimonMicroBatchStream(
with Logging {
private val options = Options.fromMap(table.options())
+ private val coreOptions = new CoreOptions(options)
+ private val consumerId = Option(coreOptions.consumerId())
+ private val warnedLegacyConsumerSnapshots = mutable.Set.empty[Long]
+
+ override protected def includeSnapshotCompletionInOffset: Boolean =
consumerId.isDefined
lazy val initOffset: PaimonSourceOffset = {
- val initSnapshotId = Math.max(
- table.snapshotManager().earliestSnapshotId(),
- streamScanStartingContext.getSnapshotId)
- val scanSnapshot = if (initSnapshotId ==
streamScanStartingContext.getSnapshotId) {
- streamScanStartingContext.getScanFullSnapshot.booleanValue()
+ val startingSnapshotId = streamScanStartingContext.getSnapshotId
+ val startingScanSnapshot =
streamScanStartingContext.getScanFullSnapshot.booleanValue()
+ val initSnapshotId = Math.max(earliestReadableId(startingScanSnapshot),
startingSnapshotId)
+ val scanSnapshot = if (initSnapshotId == startingSnapshotId) {
+ startingScanSnapshot
} else {
false
}
@@ -119,8 +124,35 @@ class PaimonMicroBatchStream(
"That latestOffset(Offset, ReadLimit) method should be called instead of
this method.")
}
- override def latestOffset(start: Offset, limit: ReadLimit): Offset = {
+ private def normalizeStartOffset(start: Offset): PaimonSourceOffset = {
val startOffset = PaimonSourceOffset(start)
+ val snapshotCompleted = startOffset.snapshotCompleted
+ val resumeSnapshotId = if (snapshotCompleted) {
+ startOffset.snapshotId + 1
+ } else {
+ startOffset.snapshotId
+ }
+ val resumeScanSnapshot = !snapshotCompleted && startOffset.scanSnapshot
+ val earliestReadable = earliestReadableId(resumeScanSnapshot)
+ // Fall back to initOffset only when the checkpointed resume position 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 (resumeSnapshotId < earliestReadable) {
+ logWarning(
+ s"Checkpointed start offset $startOffset is no longer available " +
+ s"(earliest readable snapshot or changelog: $earliestReadable), " +
+ s"attempting recovery from $initOffset.")
+ initOffset
+ } else {
+ startOffset
+ }
+ }
+
+ override def latestOffset(start: Offset, limit: ReadLimit): Offset = {
+ val startOffset = normalizeStartOffset(start)
getLatestOffset(startOffset, offsetForTriggerAvailableNow, limit).map {
offset =>
lastTriggerMillis = System.currentTimeMillis()
@@ -129,25 +161,23 @@ class PaimonMicroBatchStream(
}
override def planInputPartitions(start: Offset, end: Offset):
Array[InputPartition] = {
- val startOffset = {
- val startOffset0 = PaimonSourceOffset(start)
- // 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
- }
- }
+ val startOffset = normalizeStartOffset(start)
val endOffset = PaimonSourceOffset(end)
+ if (
+ startOffset.snapshotId == endOffset.snapshotId &&
+ startOffset.scanSnapshot != endOffset.scanSnapshot
+ ) {
+ throw new OutOfRangeException(
+ s"Cannot plan Paimon micro-batch because normalized start offset
$startOffset and " +
+ s"logged end offset $endOffset use different scan modes. The
checkpointed range " +
+ "is no longer readable without changing its data.")
+ }
+ if (startOffset.compareTo(endOffset) > 0) {
+ throw new OutOfRangeException(
+ s"Cannot plan Paimon micro-batch because normalized start offset
$startOffset is " +
+ s"newer than logged end offset $endOffset. The data needed to replay
the logged " +
+ "range is no longer readable.")
+ }
val admittedSplits = getBatch(startOffset, Some(endOffset), None)
val metadata = createMicroBatchMetadata(startOffset, endOffset,
admittedSplits)
@@ -186,7 +216,34 @@ class PaimonMicroBatchStream(
}
override def commit(end: Offset): Unit = {
- committedOffset = Some(PaimonSourceOffset(end))
+ val offset = PaimonSourceOffset(end)
+ consumerId.foreach {
+ id =>
+ offset.totalSplits match {
+ case Some(totalSplits) if offset.index >= totalSplits =>
+ throw new IllegalStateException(
+ s"Invalid Paimon source offset $offset: split index must be
smaller than " +
+ s"totalSplits ($totalSplits).")
+ case Some(_) if offset.snapshotCompleted =>
+ notifyConsumerCheckpointComplete(offset.snapshotId + 1)
+ case Some(_) if !offset.scanSnapshot =>
+ notifyConsumerCheckpointComplete(offset.snapshotId)
+ case Some(_) =>
+ // An incomplete full snapshot cannot be recovered from a
delta-only Consumer.
+ ()
+ case None =>
+ if (warnedLegacyConsumerSnapshots.add(offset.snapshotId)) {
+ logWarning(
+ s"Cannot advance Paimon consumer '$id' for snapshot " +
+ s"${offset.snapshotId} because the committed Spark offset
does not contain " +
+ "totalSplits. This can happen when recovering a checkpoint
written by an " +
+ "older Paimon version. The consumer remains unchanged and
the snapshot may " +
+ "be replayed.")
+ }
+ }
+ }
+
+ committedOffset = Some(offset)
logInfo(s"$committedOffset is committed.")
}
@@ -194,4 +251,19 @@ class PaimonMicroBatchStream(
override def table: DataTable = originTable
+ private def earliestReadableId(scanSnapshot: Boolean): Long = {
+ val earliestId: JLong =
+ if (!scanSnapshot && coreOptions.changelogLifecycleDecoupled()) {
+ val earliestChangelogId =
table.changelogManager().earliestLongLivedChangelogId()
+ if (earliestChangelogId == null) {
+ table.snapshotManager().earliestSnapshotId()
+ } else {
+ earliestChangelogId
+ }
+ } else {
+ table.snapshotManager().earliestSnapshotId()
+ }
+ if (earliestId == null) 0L else earliestId.longValue()
+ }
+
}
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 9b89729dd1..d0311a35d9 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
@@ -37,11 +37,34 @@ case class PaimonSourceOffset(snapshotId: Long, index:
Long, scanSnapshot: Boole
extends Offset
with Comparable[PaimonSourceOffset] {
+ // Keep this out of the case class constructor so the existing
three-argument constructor,
+ // Product3 API and pattern matching remain compatible. It is initialized
only by the companion
+ // factory.
+ private var totalSplitsValue: Option[Long] = None
+
+ private[spark] def totalSplits: Option[Long] = totalSplitsValue
+
+ def copy(
+ snapshotId: Long = this.snapshotId,
+ index: Long = this.index,
+ scanSnapshot: Boolean = this.scanSnapshot): PaimonSourceOffset = {
+ val copied = PaimonSourceOffset(snapshotId, index, scanSnapshot)
+ if (snapshotId == this.snapshotId && scanSnapshot == this.scanSnapshot) {
+ copied.totalSplitsValue = totalSplitsValue
+ }
+ copied
+ }
+
+ private[spark] def snapshotCompleted: Boolean = {
+ totalSplits.exists(index == _ - 1)
+ }
+
override def json(): String = {
val node = JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.createObjectNode()
node.put(PaimonSourceOffset.FIELD_SNAPSHOT_ID, snapshotId)
node.put(PaimonSourceOffset.FIELD_INDEX, index)
node.put(PaimonSourceOffset.FIELD_SCAN_SNAPSHOT, scanSnapshot)
+ totalSplits.foreach(node.put(PaimonSourceOffset.FIELD_TOTAL_SPLITS, _))
node.toString
}
@@ -65,6 +88,7 @@ object PaimonSourceOffset {
private val FIELD_SNAPSHOT_ID = "snapshotId"
private val FIELD_INDEX = "index"
private val FIELD_SCAN_SNAPSHOT = "scanSnapshot"
+ private val FIELD_TOTAL_SPLITS = "totalSplits"
def apply(version: Long, index: Long, scanSnapshot: Boolean):
PaimonSourceOffset = {
new PaimonSourceOffset(
@@ -74,15 +98,30 @@ object PaimonSourceOffset {
)
}
+ private[spark] def withTotalSplits(
+ snapshotId: Long,
+ index: Long,
+ scanSnapshot: Boolean,
+ totalSplits: Long): PaimonSourceOffset = {
+ require(totalSplits > 0, s"Total splits must be positive, but was
$totalSplits.")
+ val offset = PaimonSourceOffset(snapshotId, index, scanSnapshot)
+ offset.totalSplitsValue = Some(totalSplits)
+ offset
+ }
+
def apply(offset: Any): PaimonSourceOffset = {
offset match {
case o: PaimonSourceOffset => o
case json: String =>
val node = JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.readTree(json)
- PaimonSourceOffset(
- node.get(FIELD_SNAPSHOT_ID).asLong(),
- node.get(FIELD_INDEX).asLong(),
- node.get(FIELD_SCAN_SNAPSHOT).asBoolean())
+ val snapshotId = node.get(FIELD_SNAPSHOT_ID).asLong()
+ val index = node.get(FIELD_INDEX).asLong()
+ val scanSnapshot = node.get(FIELD_SCAN_SNAPSHOT).asBoolean()
+ Option(node.get(FIELD_TOTAL_SPLITS)) match {
+ case Some(totalSplits) =>
+ withTotalSplits(snapshotId, index, scanSnapshot,
totalSplits.asLong())
+ case None => PaimonSourceOffset(snapshotId, index, scanSnapshot)
+ }
case sc: StartingContext =>
PaimonSourceOffset(sc.getSnapshotId, INIT_OFFSET_INDEX,
sc.getScanFullSnapshot)
case _ => throw new IllegalArgumentException(s"Can't parse $offset to
PaimonSourceOffset.")
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 7e61d71ac1..68272d32f0 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
@@ -34,7 +34,20 @@ import org.apache.spark.sql.types.StructType
import scala.collection.JavaConverters._
import scala.collection.mutable
-case class IndexedDataSplit(snapshotId: Long, index: Long, entry: DataSplit)
+case class IndexedDataSplit(snapshotId: Long, index: Long, entry: DataSplit) {
+
+ // Keep the existing three-field case class API while carrying planning
metadata internally.
+ private var totalSplitsValue: Long = -1L
+
+ private[spark] def totalSplits: Option[Long] =
+ if (totalSplitsValue < 0) None else Some(totalSplitsValue)
+
+ private[spark] def withTotalSplits(totalSplits: Long): IndexedDataSplit = {
+ require(totalSplits > 0, s"Total splits must be positive, but was
$totalSplits.")
+ totalSplitsValue = totalSplits
+ this
+ }
+}
private[spark] trait StreamHelper {
@@ -44,6 +57,8 @@ private[spark] trait StreamHelper {
var lastTriggerMillis: Long
+ protected def includeSnapshotCompletionInOffset: Boolean = false
+
private lazy val streamScan: StreamDataTableScan =
table.newStreamScan().dropStats().asInstanceOf[StreamDataTableScan]
@@ -63,19 +78,28 @@ private[spark] trait StreamHelper {
// Used to get the initial offset.
lazy val streamScanStartingContext: StartingContext =
streamScan.startingContext()
+ protected def notifyConsumerCheckpointComplete(nextSnapshot: Long): Unit =
+ streamScan.notifyCheckpointComplete(nextSnapshot)
+
def getLatestOffset(
startOffset: PaimonSourceOffset,
endOffset: Option[PaimonSourceOffset],
limit: ReadLimit): Option[PaimonSourceOffset] = {
val indexedDataSplits = getBatch(startOffset, endOffset, Some(limit))
indexedDataSplits.lastOption
- .map(
+ .map {
ids =>
- PaimonSourceOffset(
- ids.snapshotId,
- ids.index,
- scanSnapshot =
- startOffset.scanSnapshot &&
ids.snapshotId.equals(startOffset.snapshotId)))
+ 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}."))
+ PaimonSourceOffset.withTotalSplits(ids.snapshotId, ids.index,
scanSnapshot, totalSplits)
+ } else {
+ PaimonSourceOffset(ids.snapshotId, ids.index, scanSnapshot)
+ }
+ }
}
def getBatch(
@@ -83,7 +107,11 @@ private[spark] trait StreamHelper {
endOffset: Option[PaimonSourceOffset],
limit: Option[ReadLimit]): Array[IndexedDataSplit] = {
if (startOffset != null) {
- streamScan.restore(startOffset.snapshotId, startOffset.scanSnapshot)
+ if (startOffset.snapshotCompleted) {
+ streamScan.restore(startOffset.snapshotId + 1, false)
+ } else {
+ streamScan.restore(startOffset.snapshotId, startOffset.scanSnapshot)
+ }
}
val readLimitGuard = limit.flatMap(PaimonReadLimits(_, lastTriggerMillis))
@@ -125,13 +153,19 @@ private[spark] trait StreamHelper {
val dataSplits =
plan.splits().asScala.collect { case dataSplit: DataSplit => dataSplit
}.toArray
val snapshotId = dataSplits.head.snapshotId()
+ val totalSplits = dataSplits.length.toLong
dataSplits
.sortWith((ds1, ds2) => compareByPartitionAndBucket(ds1, ds2) < 0)
.zipWithIndex
.map {
case (split, idx) =>
- IndexedDataSplit(snapshotId, idx, split)
+ val indexedSplit = IndexedDataSplit(snapshotId, idx, split)
+ if (includeSnapshotCompletionInOffset) {
+ indexedSplit.withTotalSplits(totalSplits)
+ } else {
+ indexedSplit
+ }
}
}
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
new file mode 100644
index 0000000000..0e4e901942
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sources/PaimonSourceOffsetTest.scala
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sources
+
+import org.apache.paimon.utils.JsonSerdeUtil
+
+import org.scalatest.funsuite.AnyFunSuite
+
+class PaimonSourceOffsetTest extends AnyFunSuite {
+
+ test("round trip total splits in offset JSON") {
+ val offset: PaimonSourceOffset = offsetWithTotalSplits(scanSnapshot = true)
+
+ val restored = PaimonSourceOffset(offset.json())
+
+ assert(restored.snapshotId == 3L)
+ assert(restored.index == 1L)
+ assert(restored.scanSnapshot)
+ assert(restored.totalSplits.contains(2L))
+ }
+
+ test("copy and Java serialization preserve total splits") {
+ val offset = offsetWithTotalSplits(scanSnapshot = true)
+
+ val copied = offset.copy(index = 0L)
+ val deserialized = org.apache.paimon.utils.InstantiationUtil.clone(offset)
+
+ assert(copied.index == 0L)
+ assert(copied.totalSplits.contains(2L))
+ assert(!copied.snapshotCompleted)
+ assert(deserialized.totalSplits.contains(2L))
+ assert(deserialized.snapshotCompleted)
+ }
+
+ test("copy clears total splits when snapshot identity changes") {
+ val offset = offsetWithTotalSplits(scanSnapshot = true)
+
+ assert(offset.copy(snapshotId = 4L).totalSplits.isEmpty)
+ assert(offset.copy(scanSnapshot = false).totalSplits.isEmpty)
+ }
+
+ test("read legacy offset JSON without total splits") {
+ val json = """{"snapshotId":3,"index":1,"scanSnapshot":false}"""
+
+ val restored = PaimonSourceOffset(json)
+
+ assert(restored.totalSplits.isEmpty)
+
assert(!JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.readTree(restored.json()).has("totalSplits"))
+ }
+
+ test("new offset JSON remains readable by the legacy decoder") {
+ val offset = offsetWithTotalSplits(scanSnapshot = false)
+
+ val restoredByLegacyDecoder = legacyRead(offset.json())
+
+ assert(restoredByLegacyDecoder.snapshotId == 3L)
+ assert(restoredByLegacyDecoder.index == 1L)
+ assert(!restoredByLegacyDecoder.scanSnapshot)
+ }
+
+ test("total splits does not change the three-field case class API") {
+ val offset: PaimonSourceOffset = offsetWithTotalSplits(scanSnapshot =
false)
+
+ assert(offset.productArity == 3)
+ val PaimonSourceOffset(snapshotId, index, scanSnapshot) = offset
+ assert(snapshotId == 3L)
+ assert(index == 1L)
+ assert(!scanSnapshot)
+
+ classOf[PaimonSourceOffset].getConstructor(
+ java.lang.Long.TYPE,
+ java.lang.Long.TYPE,
+ java.lang.Boolean.TYPE)
+ classOf[PaimonSourceOffset].getMethod(
+ "copy",
+ java.lang.Long.TYPE,
+ java.lang.Long.TYPE,
+ java.lang.Boolean.TYPE)
+ PaimonSourceOffset.getClass.getMethod(
+ "apply",
+ java.lang.Long.TYPE,
+ java.lang.Long.TYPE,
+ java.lang.Boolean.TYPE)
+ }
+
+ test("indexed data split keeps its three-field case class API") {
+ val split = IndexedDataSplit(3L, 1L, null)
+
+ assert(split.productArity == 3)
+ val IndexedDataSplit(snapshotId, index, entry) = split
+ assert(snapshotId == 3L)
+ assert(index == 1L)
+ assert(entry == null)
+
+ classOf[IndexedDataSplit].getConstructor(
+ java.lang.Long.TYPE,
+ java.lang.Long.TYPE,
+ classOf[org.apache.paimon.table.source.DataSplit])
+ assert(IndexedDataSplit.isInstanceOf[Function3[_, _, _, _]])
+ IndexedDataSplit.getClass.getMethod("tupled")
+ IndexedDataSplit.getClass.getMethod("curried")
+ }
+
+ private def offsetWithTotalSplits(scanSnapshot: Boolean): PaimonSourceOffset
= {
+ PaimonSourceOffset.withTotalSplits(
+ snapshotId = 3L,
+ index = 1L,
+ scanSnapshot = scanSnapshot,
+ totalSplits = 2L)
+ }
+
+ private def legacyRead(json: String): PaimonSourceOffset = {
+ val node = JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.readTree(json)
+ PaimonSourceOffset(
+ node.get("snapshotId").asLong(),
+ node.get("index").asLong(),
+ node.get("scanSnapshot").asBoolean())
+ }
+}
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
new file mode 100644
index 0000000000..c014a25474
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamITCase.scala
@@ -0,0 +1,339 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sources
+
+import org.apache.paimon.CoreOptions
+import org.apache.paimon.consumer.Consumer
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.OutOfRangeException
+
+import org.apache.spark.sql.connector.read.streaming.ReadLimit
+
+import java.util.{Collections, HashMap}
+
+class PaimonMicroBatchStreamITCase extends PaimonSparkTestBase {
+
+ private val consumerId = "spark-consumer"
+
+ test("initialize stream when table has no snapshots") {
+ val sourceTable = createTableWithoutSnapshot()
+ assert(sourceTable.snapshotManager().earliestSnapshotId() == null)
+
+ val initial =
createStream(sourceTable).initialOffset().asInstanceOf[PaimonSourceOffset]
+
+ assert(initial.snapshotId == 1L)
+ assert(initial.index == PaimonSourceOffset.INIT_OFFSET_INDEX)
+ assert(!initial.scanSnapshot)
+ }
+
+ test("keep legacy offset JSON when consumer is not configured") {
+ val sourceTable = createTableWithOneSnapshot()
+ val stream = createStream(sourceTable)
+ val initial = stream.initialOffset().asInstanceOf[PaimonSourceOffset]
+
+ val latest = latestOffset(stream, initial, ReadLimit.allAvailable())
+
+ assert(latest.totalSplits.isEmpty)
+ assert(!latest.json().contains("totalSplits"))
+ }
+
+ test("create consumer only after the initial full snapshot is completely
consumed") {
+ val sourceTable = withConsumer(createTableWithOneSnapshot())
+ val stream = createStream(sourceTable)
+ val initial = stream.initialOffset().asInstanceOf[PaimonSourceOffset]
+
+ assert(initial.scanSnapshot)
+ val partial = latestOffset(stream, initial, ReadLimit.maxFiles(1))
+ assert(partial.index == 0L)
+ assert(partial.totalSplits.contains(2L))
+
+ stream.commit(partial)
+ assert(!sourceTable.consumerManager().consumer(consumerId).isPresent)
+
+ val complete = latestOffset(stream, partial, ReadLimit.maxFiles(1))
+ assert(complete.index == 1L)
+ assert(complete.totalSplits.contains(2L))
+
+ stream.commit(complete)
+ assert(consumerNextSnapshot(sourceTable) == complete.snapshotId + 1)
+ }
+
+ test("protect partial delta snapshot after completing the initial full
snapshot") {
+ val sourceTable = withConsumer(createTableWithOneSnapshot())
+ val stream = createStream(sourceTable)
+ val initial = stream.initialOffset().asInstanceOf[PaimonSourceOffset]
+ assert(initial.scanSnapshot)
+
+ // Reuse the initial bucket keys so that both snapshots deterministically
contain two splits.
+ spark.sql("INSERT INTO T VALUES (10, 'v_10_2'), (11, 'v_11_2'), (12,
'v_12_2')")
+
+ val partialDelta = latestOffset(stream, initial, ReadLimit.maxFiles(3))
+ assert(partialDelta.snapshotId == initial.snapshotId + 1)
+ assert(partialDelta.index == 0L)
+ assert(partialDelta.totalSplits.contains(2L))
+ assert(!partialDelta.scanSnapshot)
+ assert(stream.planInputPartitions(initial, partialDelta).length == 3)
+
+ stream.commit(partialDelta)
+
+ assert(sourceTable.consumerManager().consumer(consumerId).isPresent)
+ assert(consumerNextSnapshot(sourceTable) == partialDelta.snapshotId)
+
+ val restartedInitial =
+
createStream(sourceTable).initialOffset().asInstanceOf[PaimonSourceOffset]
+ assert(restartedInitial.snapshotId == partialDelta.snapshotId)
+ assert(restartedInitial.index == PaimonSourceOffset.INIT_OFFSET_INDEX)
+ assert(!restartedInitial.scanSnapshot)
+ }
+
+ test("restart without Spark checkpoint from consumer progress") {
+ val sourceTable = withConsumer(createTableWithOneSnapshot())
+ val firstStream = createStream(sourceTable)
+ val firstInitial =
firstStream.initialOffset().asInstanceOf[PaimonSourceOffset]
+ val firstComplete = latestOffset(firstStream, firstInitial,
ReadLimit.allAvailable())
+ firstStream.commit(firstComplete)
+
+ spark.sql("INSERT INTO T VALUES (20, 'v_20'), (21, 'v_21'), (22, 'v_22')")
+
+ val restartedStream = createStream(sourceTable)
+ val restartedInitial =
restartedStream.initialOffset().asInstanceOf[PaimonSourceOffset]
+ assert(restartedInitial.snapshotId == firstComplete.snapshotId + 1)
+ assert(!restartedInitial.scanSnapshot)
+
+ val next = latestOffset(restartedStream, restartedInitial,
ReadLimit.allAvailable())
+ assert(next.snapshotId == restartedInitial.snapshotId)
+ assert(!next.scanSnapshot)
+ }
+
+ test("write consumer progress to the scanned branch") {
+ val mainTable = createTableWithOneSnapshot()
+ mainTable.createTag("branch-base",
mainTable.snapshotManager().latestSnapshotId())
+ mainTable.createBranch("dev", "branch-base")
+ val branchTable =
withConsumer(mainTable.switchToBranch("dev").asInstanceOf[FileStoreTable])
+ val stream = createStream(branchTable)
+ val initial = stream.initialOffset().asInstanceOf[PaimonSourceOffset]
+ val complete = latestOffset(stream, initial, ReadLimit.allAvailable())
+
+ stream.commit(complete)
+
+ assert(consumerNextSnapshot(branchTable) == complete.snapshotId + 1)
+ assert(!mainTable.consumerManager().consumer(consumerId).isPresent)
+ }
+
+ test("consumer progress protects the next snapshot from expiration") {
+ val mainTable = createTableWithOneSnapshot()
+ val sourceTable = withConsumer(mainTable)
+ val stream = createStream(sourceTable)
+ val initial = stream.initialOffset().asInstanceOf[PaimonSourceOffset]
+ val complete = latestOffset(stream, initial, ReadLimit.allAvailable())
+ stream.commit(complete)
+ val protectedSnapshot = complete.snapshotId + 1
+
+ spark.sql("INSERT INTO T VALUES (20, 'v_20')")
+ spark.sql("INSERT INTO T VALUES (30, 'v_30')")
+ spark.sql("INSERT INTO T VALUES (40, 'v_40')")
+
+ expireSnapshotsWithMinimalRetention(mainTable)
+
+ assert(!mainTable.snapshotManager().snapshotExists(complete.snapshotId))
+ assert(mainTable.snapshotManager().snapshotExists(protectedSnapshot))
+ assert(mainTable.snapshotManager().earliestSnapshotId() ==
protectedSnapshot)
+
+ val next = latestOffset(stream, complete, ReadLimit.allAvailable())
+ assert(next.snapshotId >= protectedSnapshot)
+ assert(!next.scanSnapshot)
+ assert(stream.planInputPartitions(complete, next).nonEmpty)
+ }
+
+ test("fail rather than mix expired checkpoint with a new full scan") {
+ val mainTable = createTableWithOneSnapshot()
+ val sourceTable = withConsumer(mainTable)
+ val initialStream = createStream(sourceTable)
+ val initial =
initialStream.initialOffset().asInstanceOf[PaimonSourceOffset]
+ val complete = latestOffset(initialStream, initial,
ReadLimit.allAvailable())
+ assert(complete.snapshotCompleted)
+
+ // Simulate a crash before the completed offset advances consumer progress.
+ spark.sql("INSERT INTO T VALUES (20, 'v_20')")
+ spark.sql("INSERT INTO T VALUES (30, 'v_30')")
+ spark.sql("INSERT INTO T VALUES (40, 'v_40')")
+ val loggedDeltaEnd = latestOffset(initialStream, complete,
ReadLimit.allAvailable())
+ assert(!loggedDeltaEnd.scanSnapshot)
+ assert(loggedDeltaEnd.snapshotId ==
mainTable.snapshotManager().latestSnapshotId())
+
+ expireSnapshotsWithMinimalRetention(mainTable)
+
+ assert(!mainTable.snapshotManager().snapshotExists(complete.snapshotId +
1))
+ assert(!mainTable.snapshotManager().snapshotExists(complete.snapshotId +
2))
+ assert(mainTable.snapshotManager().earliestSnapshotId() ==
loggedDeltaEnd.snapshotId)
+
+ val restartedStream = createStream(sourceTable)
+ val exception = intercept[OutOfRangeException] {
+ restartedStream.planInputPartitions(complete, loggedDeltaEnd)
+ }
+ assert(exception.getMessage.contains("no longer readable"))
+
+ val currentFullEnd = latestOffset(restartedStream, complete,
ReadLimit.allAvailable())
+ assert(currentFullEnd.scanSnapshot)
+ assert(restartedStream.planInputPartitions(complete,
currentFullEnd).nonEmpty)
+ }
+
+ test("fail when expired recovery start is newer than logged end") {
+ val mainTable = createTableWithOneSnapshot()
+ val sourceTable = withConsumer(mainTable)
+ val initialStream = createStream(sourceTable)
+ val initial =
initialStream.initialOffset().asInstanceOf[PaimonSourceOffset]
+ val complete = latestOffset(initialStream, initial,
ReadLimit.allAvailable())
+ assert(complete.snapshotCompleted)
+
+ spark.sql("INSERT INTO T VALUES (20, 'v_20')")
+ spark.sql("INSERT INTO T VALUES (30, 'v_30')")
+ spark.sql("INSERT INTO T VALUES (40, 'v_40')")
+ val loggedEnd = latestOffset(initialStream, complete,
ReadLimit.allAvailable())
+ assert(loggedEnd.snapshotId == complete.snapshotId + 3)
+
+ spark.sql("INSERT INTO T VALUES (50, 'v_50')")
+ val latestSnapshotId = mainTable.snapshotManager().latestSnapshotId()
+ assert(latestSnapshotId == loggedEnd.snapshotId + 1)
+
+ expireSnapshotsWithMinimalRetention(mainTable)
+
+ assert(!mainTable.snapshotManager().snapshotExists(loggedEnd.snapshotId))
+ assert(mainTable.snapshotManager().earliestSnapshotId() ==
latestSnapshotId)
+
+ val restartedStream = createStream(sourceTable)
+ val exception = intercept[OutOfRangeException] {
+ restartedStream.planInputPartitions(complete, loggedEnd)
+ }
+ assert(exception.getMessage.contains("newer than logged end offset"))
+ }
+
+ test("resume consumer from long-lived changelog after snapshot expiration") {
+ val mainTable = createTableWithOneSnapshot()
+ spark.sql("INSERT INTO T VALUES (20, 'v_20')")
+ spark.sql("INSERT INTO T VALUES (30, 'v_30')")
+
+ val lifecycleOptions = new HashMap[String, String]()
+ lifecycleOptions.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "1")
+ lifecycleOptions.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "1")
+ lifecycleOptions.put(CoreOptions.SNAPSHOT_TIME_RETAINED.key(), "0 ms")
+ lifecycleOptions.put(CoreOptions.CHANGELOG_NUM_RETAINED_MIN.key(), "1")
+ lifecycleOptions.put(CoreOptions.CHANGELOG_NUM_RETAINED_MAX.key(), "10")
+ lifecycleOptions.put(CoreOptions.CHANGELOG_TIME_RETAINED.key(), "1 d")
+ lifecycleOptions.put(CoreOptions.CONSUMER_CHANGELOG_ONLY.key(), "true")
+ val lifecycleTable = mainTable.copy(lifecycleOptions)
+ lifecycleTable.consumerManager().resetConsumer(consumerId, new
Consumer(1L))
+ lifecycleTable
+ .newExpireSnapshots()
+ .config(lifecycleTable.coreOptions().expireConfig())
+ .expire()
+
+ assert(lifecycleTable.snapshotManager().earliestSnapshotId() > 1L)
+ assert(lifecycleTable.changelogManager().earliestLongLivedChangelogId() ==
1L)
+
+ val stream = createStream(withConsumer(lifecycleTable))
+ val initial = stream.initialOffset().asInstanceOf[PaimonSourceOffset]
+ assert(initial.snapshotId == 1L)
+ assert(!initial.scanSnapshot)
+
+ val end = latestOffset(stream, initial, ReadLimit.maxFiles(1))
+ assert(end.snapshotId == 1L)
+ assert(stream.planInputPartitions(initial, end).nonEmpty)
+ }
+
+ test("Spark query commits consumer progress through the source callback") {
+ withTempDir {
+ checkpointDir =>
+ val mainTable = createTableWithOneSnapshot()
+ val query = spark.readStream
+ .format("paimon")
+ .option(CoreOptions.CONSUMER_ID.key(), consumerId)
+ .option("read.stream.maxFilesPerTrigger", "1")
+ .load(mainTable.location().toString)
+ .writeStream
+ .format("memory")
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ .queryName("spark_consumer_memory")
+ .outputMode("append")
+ .start()
+
+ try {
+ query.processAllAvailable()
+ // Spark reports a source commit while constructing a later
micro-batch.
+ spark.sql("INSERT INTO T VALUES (20, 'v_20'), (21, 'v_21'), (22,
'v_22')")
+ query.processAllAvailable()
+
+ val consumer = mainTable.consumerManager().consumer(consumerId)
+ assert(consumer.isPresent)
+ assert(consumer.get().nextSnapshot() >= 2L)
+ } finally {
+ query.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 withConsumer(table: FileStoreTable): FileStoreTable = {
+ table.copy(Collections.singletonMap(CoreOptions.CONSUMER_ID.key(),
consumerId))
+ }
+
+ private def createStream(table: FileStoreTable): PaimonMicroBatchStream = {
+ new PaimonMicroBatchStream(table, table.newReadBuilder(), "unused")
+ }
+
+ private def latestOffset(
+ stream: PaimonMicroBatchStream,
+ start: PaimonSourceOffset,
+ limit: ReadLimit): PaimonSourceOffset = {
+ stream.latestOffset(start, limit).asInstanceOf[PaimonSourceOffset]
+ }
+
+ private def expireSnapshotsWithMinimalRetention(table: FileStoreTable): Unit
= {
+ val expireOptions = new HashMap[String, String]()
+ expireOptions.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "1")
+ expireOptions.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "1")
+ expireOptions.put(CoreOptions.SNAPSHOT_TIME_RETAINED.key(), "0 ms")
+ val expireTable = table.copy(expireOptions)
+ expireTable
+ .newExpireSnapshots()
+ .config(expireTable.coreOptions().expireConfig())
+ .expire()
+ }
+
+ private def consumerNextSnapshot(table: FileStoreTable): Long = {
+ table.consumerManager().consumer(consumerId).get().nextSnapshot()
+ }
+}
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
new file mode 100644
index 0000000000..4f605d71cc
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamTest.scala
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sources
+
+import org.apache.paimon.CoreOptions
+import org.apache.paimon.table.DataTable
+import org.apache.paimon.table.source.{ReadBuilder, StreamDataTableScan}
+
+import org.mockito.ArgumentMatchers.anyLong
+import org.mockito.Mockito.{doNothing, doThrow, mock, never, times, verify,
when}
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.io.{IOException, UncheckedIOException}
+import java.util.Collections
+
+class PaimonMicroBatchStreamTest extends AnyFunSuite {
+
+ test("never advance consumer past an incomplete snapshot") {
+ val (stream, scan) = createStreamWithConsumer()
+ val partial = consumerOffset(index = 0L, totalSplits = 2L)
+ val complete = consumerOffset(index = 1L, totalSplits = 2L)
+
+ stream.commit(partial)
+ verify(scan).notifyCheckpointComplete(5L)
+ verify(scan, never()).notifyCheckpointComplete(6L)
+
+ stream.commit(complete)
+ verify(scan).notifyCheckpointComplete(6L)
+ }
+
+ test("do not advance consumer from a legacy offset without total splits") {
+ val (stream, scan) = createStreamWithConsumer()
+ val legacyOffset =
PaimonSourceOffset("""{"snapshotId":5,"index":1,"scanSnapshot":false}""")
+
+ stream.commit(legacyOffset)
+
+ verify(scan, never()).notifyCheckpointComplete(anyLong())
+ }
+
+ test("propagate consumer update failure and allow retry") {
+ val (stream, scan) = createStreamWithConsumer()
+ val complete = consumerOffset(index = 1L, totalSplits = 2L)
+ val failure = new UncheckedIOException(new IOException("expected failure"))
+ doThrow(failure).doNothing().when(scan).notifyCheckpointComplete(6L)
+
+ val thrown = intercept[UncheckedIOException] {
+ stream.commit(complete)
+ }
+ assert(thrown eq failure)
+
+ stream.commit(complete)
+ verify(scan, times(2)).notifyCheckpointComplete(6L)
+ }
+
+ private def consumerOffset(index: Long, totalSplits: Long):
PaimonSourceOffset = {
+ PaimonSourceOffset.withTotalSplits(
+ snapshotId = 5L,
+ index = index,
+ scanSnapshot = false,
+ totalSplits = totalSplits)
+ }
+
+ private def createStreamWithConsumer(): (PaimonMicroBatchStream,
StreamDataTableScan) = {
+ val table = mock(classOf[DataTable])
+ val scan = mock(classOf[StreamDataTableScan])
+ val readBuilder = mock(classOf[ReadBuilder])
+ when(table.options())
+ .thenReturn(Collections.singletonMap(CoreOptions.CONSUMER_ID.key(),
"spark-consumer"))
+ when(table.newStreamScan()).thenReturn(scan)
+ when(scan.dropStats()).thenReturn(scan)
+ (new PaimonMicroBatchStream(table, readBuilder, "unused"), scan)
+ }
+}
