This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 3869163581 Spark: Don't reapply stream-from-timestamp when a stream
resumes (#18348)
3869163581 is described below
commit 38691635818adb63cf6e425fe8299af63593ada8
Author: Maximilian Michels <[email protected]>
AuthorDate: Mon Oct 12 11:06:32 2026 +0200
Spark: Don't reapply stream-from-timestamp when a stream resumes (#18348)
---
docs/docs/spark-configuration.md | 2 +-
.../spark/source/AsyncSparkMicroBatchPlanner.java | 4 +-
.../spark/source/BaseSparkMicroBatchPlanner.java | 12 +
.../spark/source/SparkMicroBatchPlanner.java | 9 +
.../spark/source/SparkMicroBatchStream.java | 19 +-
.../spark/source/SyncSparkMicroBatchPlanner.java | 8 +-
.../spark/source/TestStructuredStreamingRead3.java | 249 +++++++++++++++++++++
7 files changed, 294 insertions(+), 9 deletions(-)
diff --git a/docs/docs/spark-configuration.md b/docs/docs/spark-configuration.md
index 857f7c40f3..f45f89125a 100644
--- a/docs/docs/spark-configuration.md
+++ b/docs/docs/spark-configuration.md
@@ -233,7 +233,7 @@ spark.read
| file-open-cost | As per table property | Overrides this table's
read.split.open-file-cost
|
| vectorization-enabled | As per table property | Overrides this table's
read.parquet.vectorization.enabled
|
| batch-size | As per table property | Overrides this table's
read.parquet.vectorization.batch-size
|
-| stream-from-timestamp | (none) | A timestamp in milliseconds to stream from;
if before the oldest known ancestor snapshot, the oldest will be used
|
+| stream-from-timestamp | (none) | A timestamp in milliseconds to stream from;
if before the oldest known ancestor snapshot, the oldest will be used. Only
applies until the query has a position. A query restarted from a checkpoint
continues from its saved position |
| streaming-max-files-per-micro-batch | INT_MAX | Maximum number of files per
microbatch
|
| streaming-max-rows-per-micro-batch | INT_MAX | "Soft maximum" number of
rows per microbatch; always includes all rows in next unprocessed file,
excludes additional files if their inclusion would exceed the soft max limit |
| async-micro-batch-planning-enabled | false |
Enables asynchronous microbatch planning to reduce planning latency by
pre-fetching file scan tasks
|
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java
index fcbee2a87e..851421917a 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java
@@ -252,7 +252,9 @@ class AsyncSparkMicroBatchPlanner extends
BaseSparkMicroBatchPlanner implements
return StreamingOffset.START_OFFSET;
}
- if (table().currentSnapshot().timestampMillis() <
readConf().streamFromTimestamp()) {
+ // Only a new stream starts from the timestamp. A resumed stream continues
from its offset.
+ if (startOffset.equals(StreamingOffset.START_OFFSET)
+ && table().currentSnapshot().timestampMillis() <
readConf().streamFromTimestamp()) {
LOG.info("latestOffset returning START_OFFSET, currentSnapshot() <
fromTimestamp");
return StreamingOffset.START_OFFSET;
}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BaseSparkMicroBatchPlanner.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BaseSparkMicroBatchPlanner.java
index 9298c2bbdf..b80d04ee13 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BaseSparkMicroBatchPlanner.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BaseSparkMicroBatchPlanner.java
@@ -117,6 +117,18 @@ abstract class BaseSparkMicroBatchPlanner implements
SparkMicroBatchPlanner {
return nextSnapshot;
}
+ @Override
+ public StreamingOffset latestValidSnapshotOffset() {
+ for (Snapshot snapshot : SnapshotUtil.currentAncestors(table)) {
+ if (shouldProcess(snapshot)) {
+ return new StreamingOffset(
+ snapshot.snapshotId(), MicroBatchUtils.addedFilesCount(table,
snapshot), false);
+ }
+ }
+
+ return null;
+ }
+
static class UnpackedLimits {
private long maxRows = Integer.MAX_VALUE;
private long maxFiles = Integer.MAX_VALUE;
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchPlanner.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchPlanner.java
index 1986ddac5d..014de230b0 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchPlanner.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchPlanner.java
@@ -42,6 +42,15 @@ interface SparkMicroBatchPlanner {
*/
StreamingOffset latestOffset(StreamingOffset startOffset, ReadLimit limit);
+ /**
+ * Return the offset at the end of the latest snapshot in the table's
current history that the
+ * stream reads.
+ *
+ * @return the offset, or {@code null} if the stream reads no snapshot in
the current history
+ * @throws IllegalStateException if a later snapshot can't be skipped
+ */
+ StreamingOffset latestValidSnapshotOffset();
+
/** Stop the planner and release any resources. */
void stop();
}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
index 2d510e50ad..0370c43e6b 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
@@ -66,6 +66,7 @@ public class SparkMicroBatchStream implements
MicroBatchStream, SupportsTriggerA
private final boolean cacheDeleteFilesOnExecutors;
private SparkMicroBatchPlanner planner;
private StreamingOffset lastOffsetForTriggerAvailableNow;
+ private boolean noSnapshotMatchedForTriggerAvailableNow;
SparkMicroBatchStream(
JavaSparkContext sparkContext,
@@ -215,6 +216,12 @@ public class SparkMicroBatchStream implements
MicroBatchStream, SupportsTriggerA
"Invalid start offset: %s is not a StreamingOffset",
startOffset);
+ // Snapshots committed after Trigger.AvailableNow started are left for the
next run
+ if (noSnapshotMatchedForTriggerAvailableNow
+ && StreamingOffset.START_OFFSET.equals(startOffset)) {
+ return StreamingOffset.START_OFFSET;
+ }
+
// Initialize planner if not already done
if (planner == null) {
initializePlanner((StreamingOffset) startOffset, null);
@@ -244,10 +251,16 @@ public class SparkMicroBatchStream implements
MicroBatchStream, SupportsTriggerA
public void prepareForTriggerAvailableNow() {
LOG.info("The streaming query reports to use Trigger.AvailableNow");
- lastOffsetForTriggerAvailableNow =
+ StreamingOffset lastOffset =
(StreamingOffset) latestOffset(initialOffset,
ReadLimit.allAvailable());
-
- LOG.info("lastOffset for Trigger.AvailableNow is {}",
lastOffsetForTriggerAvailableNow.json());
+ // START_OFFSET means that no snapshot matched stream-from-timestamp. A
new stream has nothing
+ // to read in this run. A resumed stream continues from its offset and is
capped at the latest
+ // snapshot the planner reads, since a cap at a skipped snapshot is never
reached.
+ this.noSnapshotMatchedForTriggerAvailableNow =
StreamingOffset.START_OFFSET.equals(lastOffset);
+ this.lastOffsetForTriggerAvailableNow =
+ noSnapshotMatchedForTriggerAvailableNow ?
planner.latestValidSnapshotOffset() : lastOffset;
+
+ LOG.info("lastOffset for Trigger.AvailableNow is {}",
lastOffsetForTriggerAvailableNow);
// Reset planner so it gets recreated with the cap on next call
if (planner != null) {
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SyncSparkMicroBatchPlanner.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SyncSparkMicroBatchPlanner.java
index f1b0029c54..b80e2e94a4 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SyncSparkMicroBatchPlanner.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SyncSparkMicroBatchPlanner.java
@@ -118,15 +118,15 @@ class SyncSparkMicroBatchPlanner extends
BaseSparkMicroBatchPlanner {
return StreamingOffset.START_OFFSET;
}
- if (table().currentSnapshot().timestampMillis() < fromTimestamp) {
- return StreamingOffset.START_OFFSET;
- }
-
// end offset can expand to multiple snapshots
StreamingOffset startingOffset = startOffset;
+ // Only a new stream starts from the timestamp. A resumed stream continues
from its offset.
if (startOffset.equals(StreamingOffset.START_OFFSET)) {
startingOffset = MicroBatchUtils.determineStartingOffset(table(),
fromTimestamp);
+ if (startingOffset.equals(StreamingOffset.START_OFFSET)) {
+ return StreamingOffset.START_OFFSET;
+ }
}
Snapshot curSnapshot = table().snapshot(startingOffset.snapshotId());
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
index 8003497bd1..cb6ea12326 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
@@ -112,6 +112,9 @@ public final class TestStructuredStreamingRead3 extends
CatalogTestBase {
private final AtomicInteger microBatches = new AtomicInteger();
+ private final List<SimpleRecord> recordsReadByStream =
+ Collections.synchronizedList(Lists.newArrayList());
+
@Parameter(index = 3)
private Boolean async;
@@ -572,6 +575,163 @@ public final class TestStructuredStreamingRead3 extends
CatalogTestBase {
}
}
+ @TestTemplate
+ void latestOffsetIgnoresStreamFromTimestampWhenResuming() {
+ appendData(List.of(new SimpleRecord(1, "one")));
+ table.refresh();
+ Snapshot resumedSnapshot = table.currentSnapshot();
+ appendData(List.of(new SimpleRecord(2, "two")));
+ table.refresh();
+ long latestSnapshotId = table.currentSnapshot().snapshotId();
+
+ SparkMicroBatchStream stream =
+ newMicroBatchStream(
+ ImmutableMap.of(
+ SparkReadOptions.STREAM_FROM_TIMESTAMP,
+ Long.toString(timestampAfterCurrentSnapshot())),
+ "resumed-stream-checkpoint");
+
+ try {
+ StreamingOffset resumedOffset =
+ new StreamingOffset(
+ resumedSnapshot.snapshotId(),
+ MicroBatchUtils.addedFilesCount(table, resumedSnapshot),
+ false);
+ StreamingOffset endOffset =
+ (StreamingOffset) stream.latestOffset(resumedOffset,
stream.getDefaultReadLimit());
+
+ assertThat(endOffset.snapshotId()).isEqualTo(latestSnapshotId);
+ } finally {
+ stream.stop();
+ }
+ }
+
+ @TestTemplate
+ void availableNowResumeStopsAtSnapshotsAvailableAtStart() {
+ appendData(List.of(new SimpleRecord(1, "one")));
+ table.refresh();
+ Snapshot committed = table.currentSnapshot();
+ appendData(List.of(new SimpleRecord(2, "two")));
+ table.refresh();
+ Snapshot latest = table.currentSnapshot();
+ // A timestamp after the newest snapshot stores START_OFFSET as the
initial offset
+ SparkMicroBatchStream stream =
+ newMicroBatchStream(
+ ImmutableMap.of(
+ SparkReadOptions.STREAM_FROM_TIMESTAMP,
+ Long.toString(timestampAfterCurrentSnapshot())),
+ "available-now-resume-checkpoint");
+
+ try {
+ stream.prepareForTriggerAvailableNow();
+ appendData(List.of(new SimpleRecord(3, "three")));
+ StreamingOffset committedOffset =
+ new StreamingOffset(
+ committed.snapshotId(), MicroBatchUtils.addedFilesCount(table,
committed), false);
+
+ assertThat(stream.latestOffset(committedOffset,
stream.getDefaultReadLimit()))
+ .isEqualTo(
+ new StreamingOffset(
+ latest.snapshotId(), MicroBatchUtils.addedFilesCount(table,
latest), false));
+ } finally {
+ stream.stop();
+ }
+ }
+
+ @TestTemplate
+ void availableNowResumeStopsAtLatestAppendSnapshot() {
+ appendData(List.of(new SimpleRecord(1, "one")));
+ table.refresh();
+ Snapshot committed = table.currentSnapshot();
+ appendData(List.of(new SimpleRecord(2, "two")));
+ table.refresh();
+ Snapshot latestAppend = table.currentSnapshot();
+ table.rewriteManifests().clusterBy(file -> 1).commit();
+ table.refresh();
+
assertThat(table.currentSnapshot().operation()).isEqualTo(DataOperations.REPLACE);
+ SparkMicroBatchStream stream =
+ newMicroBatchStream(
+ ImmutableMap.of(
+ SparkReadOptions.STREAM_FROM_TIMESTAMP,
+ Long.toString(timestampAfterCurrentSnapshot())),
+ "available-now-resume-replace-checkpoint");
+
+ try {
+ stream.prepareForTriggerAvailableNow();
+ appendData(List.of(new SimpleRecord(3, "three")));
+ StreamingOffset committedOffset =
+ new StreamingOffset(
+ committed.snapshotId(), MicroBatchUtils.addedFilesCount(table,
committed), false);
+
+ assertThat(stream.latestOffset(committedOffset,
stream.getDefaultReadLimit()))
+ .isEqualTo(
+ new StreamingOffset(
+ latestAppend.snapshotId(),
+ MicroBatchUtils.addedFilesCount(table, latestAppend),
+ false));
+ } finally {
+ stream.stop();
+ }
+ }
+
+ @TestTemplate
+ void availableNowIgnoresSnapshotsCommittedAfterStart() {
+ appendData(List.of(new SimpleRecord(1, "one")));
+ SparkMicroBatchStream stream =
+ newMicroBatchStream(
+ ImmutableMap.of(
+ SparkReadOptions.STREAM_FROM_TIMESTAMP,
+ Long.toString(timestampAfterCurrentSnapshot())),
+ "available-now-new-stream-checkpoint");
+
+ try {
+ stream.prepareForTriggerAvailableNow();
+ appendData(List.of(new SimpleRecord(2, "two")));
+
+ assertThat(stream.latestOffset(stream.initialOffset(),
stream.getDefaultReadLimit()))
+ .isEqualTo(StreamingOffset.START_OFFSET);
+ } finally {
+ stream.stop();
+ }
+ }
+
+ @TestTemplate
+ void availableNowReadsNothingFromEmptyTable() {
+ SparkMicroBatchStream stream =
+ newMicroBatchStream(ImmutableMap.of(),
"available-now-empty-table-checkpoint");
+
+ try {
+ stream.prepareForTriggerAvailableNow();
+
+ assertThat(stream.latestOffset(stream.initialOffset(),
stream.getDefaultReadLimit()))
+ .isEqualTo(StreamingOffset.START_OFFSET);
+ } finally {
+ stream.stop();
+ }
+ }
+
+ @TestTemplate
+ void availableNowCapRejectsDeleteSnapshotWithoutSkipOption() {
+ table.updateSpec().removeField("id_bucket").addField(ref("id")).commit();
+ appendData(List.of(new SimpleRecord(1, "one")));
+ table.newDelete().deleteFromRowFilter(Expressions.equal("id", 1)).commit();
+
assertThat(table.currentSnapshot().operation()).isEqualTo(DataOperations.DELETE);
+ SparkMicroBatchStream stream =
+ newMicroBatchStream(
+ ImmutableMap.of(
+ SparkReadOptions.STREAM_FROM_TIMESTAMP,
+ Long.toString(timestampAfterCurrentSnapshot())),
+ "available-now-delete-checkpoint");
+
+ try {
+ assertThatThrownBy(stream::prepareForTriggerAvailableNow)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageStartingWith("Cannot process delete snapshot");
+ } finally {
+ stream.stop();
+ }
+ }
+
@TestTemplate
public void testReadStreamOnIcebergThenAddData() throws Exception {
List<List<SimpleRecord>> expected = TEST_DATA_MULTIPLE_SNAPSHOTS;
@@ -721,6 +881,48 @@ public final class TestStructuredStreamingRead3 extends
CatalogTestBase {
assertThat(actual).containsExactlyInAnyOrderElementsOf(expectedRecordList);
}
+ @TestTemplate
+ void resumeIgnoresLaterStreamFromTimestamp() throws Exception {
+ File checkpoint = temp.resolve("checkpoint").toFile();
+ appendData(List.of(new SimpleRecord(1, "one")));
+ StreamingQuery query = startStreamFromTimestamp(checkpoint,
timestampAfterCurrentSnapshot());
+ query.processAllAvailable();
+ List<SimpleRecord> expected = Lists.newArrayList(new SimpleRecord(2,
"two"));
+ appendData(expected);
+ awaitRecordsReadByStream(query, expected);
+ query.stop();
+
+ List<SimpleRecord> appendedWhileStopped = List.of(new SimpleRecord(3,
"three"));
+ appendData(appendedWhileStopped);
+ expected.addAll(appendedWhileStopped);
+
+ StreamingQuery restarted =
+ startStreamFromTimestamp(checkpoint, timestampAfterCurrentSnapshot());
+ awaitRecordsReadByStream(restarted, expected);
+ }
+
+ @TestTemplate
+ void availableNowResumeIgnoresLaterStreamFromTimestamp() throws Exception {
+ File checkpoint = temp.resolve("checkpoint").toFile();
+ appendData(List.of(new SimpleRecord(1, "one")));
+ StreamingQuery query = startStreamFromTimestamp(checkpoint,
timestampAfterCurrentSnapshot());
+ query.processAllAvailable();
+ List<SimpleRecord> expected = Lists.newArrayList(new SimpleRecord(2,
"two"));
+ appendData(expected);
+ awaitRecordsReadByStream(query, expected);
+ query.stop();
+
+ List<SimpleRecord> appendedWhileStopped = List.of(new SimpleRecord(3,
"three"));
+ appendData(appendedWhileStopped);
+ expected.addAll(appendedWhileStopped);
+
+ StreamingQuery rerun =
+ startStreamFromTimestamp(
+ checkpoint, timestampAfterCurrentSnapshot(),
Trigger.AvailableNow());
+
assertThat(rerun.awaitTermination(Duration.ofMinutes(1).toMillis())).isTrue();
+
assertThat(recordsReadByStream).containsExactlyInAnyOrderElementsOf(expected);
+ }
+
@TestTemplate
public void testResumingStreamReadFromCheckpoint() throws Exception {
File writerCheckpointFolder =
temp.resolve("writer-checkpoint-folder").toFile();
@@ -1163,6 +1365,53 @@ public final class TestStructuredStreamingRead3 extends
CatalogTestBase {
.collectAsList();
}
+ private StreamingQuery startStreamFromTimestamp(File checkpoint, long
fromTimestamp)
+ throws TimeoutException {
+ return startStreamFromTimestamp(checkpoint, fromTimestamp,
Trigger.ProcessingTime(0L));
+ }
+
+ private StreamingQuery startStreamFromTimestamp(
+ File checkpoint, long fromTimestamp, Trigger trigger) throws
TimeoutException {
+ Map<String, String> options = Maps.newHashMap();
+ options.put(SparkReadOptions.STREAM_FROM_TIMESTAMP,
Long.toString(fromTimestamp));
+ options.put(SparkReadOptions.ASYNC_MICRO_BATCH_PLANNING_ENABLED,
async.toString());
+ if (async) {
+ options.put(SparkReadOptions.STREAMING_SNAPSHOT_POLLING_INTERVAL_MS,
"1");
+ }
+
+ return spark
+ .readStream()
+ .options(options)
+ .format("iceberg")
+ .load(tableName)
+ .writeStream()
+ .option("checkpointLocation", checkpoint.toString())
+ .trigger(trigger)
+ .foreachBatch(
+ (VoidFunction2<Dataset<Row>, Long>)
+ (batch, batchId) -> {
+ List<SimpleRecord> records =
+
batch.as(Encoders.bean(SimpleRecord.class)).collectAsList();
+ recordsReadByStream.addAll(records);
+ })
+ .start();
+ }
+
+ private void awaitRecordsReadByStream(StreamingQuery query,
List<SimpleRecord> expected) {
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(30))
+ .untilAsserted(
+ () -> {
+ query.processAllAvailable();
+
assertThat(recordsReadByStream).containsExactlyInAnyOrderElementsOf(expected);
+ });
+ }
+
+ private long timestampAfterCurrentSnapshot() {
+ table.refresh();
+ return waitUntilAfter(table.currentSnapshot().timestampMillis());
+ }
+
private SparkMicroBatchStream newMicroBatchStream(
Map<String, String> options, String checkpointDirName) {
Map<String, String> allOptions = Maps.newHashMap(options);
