Skip to content

Spark: Short-circuit SyncSparkMicroBatchPlanner.latestOffset for unbounded limits #18494

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
binarybrain-009 wants to merge 3 commits into
base: main
Choose a base branch
from
+500 −0
Open

Spark: Short-circuit SyncSparkMicroBatchPlanner.latestOffset for unbounded limits #18494

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
99b9966
Spark: Short-circuit SyncSparkMicroBatchPlanner.latestOffset for unbo…
binarybrain-009 Oct 14, 2026
4752d75
Spark: Fix end position of unbounded SyncSparkMicroBatchPlanner.lates…
binarybrain-009 Oct 28, 2026
2e8acc3
Merge branch 'main' into sync-planner-short-circuit-unbounded
binarybrain-009 Oct 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 42 additions & 0 deletions .../v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SyncSparkMicroBatchPlanner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.apache.iceberg.spark.SparkReadConf;
import org.apache.iceberg.util.Pair;
import org.apache.iceberg.util.SnapshotUtil;
import org.apache.spark.sql.connector.read.streaming.ReadAllAvailable;
import org.apache.spark.sql.connector.read.streaming.ReadLimit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -132,6 +133,12 @@ public StreamingOffset latestOffset(StreamingOffset startOffset, ReadLimit limit
Snapshot curSnapshot = table().snapshot(startingOffset.snapshotId());
validateCurrentSnapshotExists(curSnapshot, startingOffset);

// An unbounded limit has no per-file limit to enforce, so only the latest valid snapshot's
// manifests are indexed instead of reading every manifest between the start and the head.
if (limit instanceof ReadAllAvailable) {
return latestOffsetForUnbounded(startingOffset, curSnapshot);
}

// Use the pre-computed snapshotId when Trigger.AvailableNow is enabled.
long latestSnapshotId =
lastOffsetForTriggerAvailableNow != null
Expand Down Expand Up @@ -234,6 +241,41 @@ public StreamingOffset latestOffset(StreamingOffset startOffset, ReadLimit limit
return latestStreamingOffset.equals(startingOffset) ? null : latestStreamingOffset;
}

private StreamingOffset latestOffsetForUnbounded(
StreamingOffset startingOffset, Snapshot curSnapshot) {
Snapshot lastValidSnapshot;
if (lastOffsetForTriggerAvailableNow != null) {
lastValidSnapshot = table().snapshot(lastOffsetForTriggerAvailableNow.snapshotId());
validateCurrentSnapshotExists(lastValidSnapshot, lastOffsetForTriggerAvailableNow);
} else {
lastValidSnapshot = curSnapshot;
Snapshot next;
do {
next = nextValidSnapshot(lastValidSnapshot);
if (next != null) {
lastValidSnapshot = next;
}
} while (next != null);
}

StreamingOffset latestStreamingOffset =
new StreamingOffset(lastValidSnapshot.snapshotId(), endPosition(lastValidSnapshot), false);
return latestStreamingOffset.equals(startingOffset) ? null : latestStreamingOffset;
}

// Existing entries in merged manifests also take positions, so the end position can be larger
// than the snapshot's added-file count.
private long endPosition(Snapshot snapshot) {
List<Pair<ManifestFile, Integer>> indexedManifests =
MicroBatches.skippedManifestIndexesFromSnapshot(table().io(), snapshot, 0, false);
if (indexedManifests.isEmpty()) {
return 0;
}

Pair<ManifestFile, Integer> lastManifest = indexedManifests.get(indexedManifests.size() - 1);
return lastManifest.second() + lastManifest.first().addedFilesCount();
}

@Override
public void stop() {}

Expand Down
83 changes: 83 additions & 0 deletions ...3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,15 @@
import static org.apache.iceberg.expressions.Expressions.ref;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assumptions.assumeThat;

import java.io.File;
import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
Expand All @@ -40,9 +42,11 @@
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.Files;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.Parameter;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.RewriteFiles;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
Expand Down Expand Up @@ -71,6 +75,7 @@
import org.apache.spark.sql.Row;
import org.apache.spark.sql.connector.read.InputPartition;
import org.apache.spark.sql.connector.read.streaming.Offset;
import org.apache.spark.sql.connector.read.streaming.ReadLimit;
import org.apache.spark.sql.internal.SQLConf;
import org.apache.spark.sql.streaming.DataStreamWriter;
import org.apache.spark.sql.streaming.OutputMode;
Expand Down Expand Up @@ -483,6 +488,75 @@ public void testTriggerAvailableNowCapsAsyncPreloadAfterPrepare() {
}
}

@TestTemplate
void latestOffsetForUnboundedLimitAdvancesToHeadSnapshot() {
appendDataAsMultipleSnapshots(TEST_DATA_MULTIPLE_SNAPSHOTS);

table.refresh();
long headSnapshotId = table.currentSnapshot().snapshotId();

SparkMicroBatchStream stream =
newMicroBatchStream(ImmutableMap.of(), "all-available-latest-offset-checkpoint");

try {
Offset startOffset = stream.initialOffset();
StreamingOffset endOffset =
(StreamingOffset) stream.latestOffset(startOffset, ReadLimit.allAvailable());

assertThat(endOffset).isNotNull();
assertThat(endOffset.snapshotId()).isEqualTo(headSnapshotId);
} finally {
stream.stop();
}
}

@TestTemplate
void latestOffsetForUnboundedLimitIncludesFilesAfterMergedManifest() {
assumeThat(async)
.as("Async planning does not handle gaps left by existing files yet")
.isFalse();

PartitionSpec oldSpec = table.spec();
table.updateSpec().addField("data").commit();

DataFile first = newDataFile(table.spec());
table.newAppend().appendFile(first).commit();
DataFile second = newDataFile(table.spec());
table.newAppend().appendFile(second).commit();

// the new spec's manifests are merged and listed before the old spec's new manifest
DataFile third = newDataFile(table.spec());
DataFile fourth = newDataFile(oldSpec);
table.newAppend().appendFile(third).appendFile(fourth).commit();

List<ManifestFile> headManifests = table.currentSnapshot().dataManifests(table.io());
assertThat(headManifests).hasSize(2);
assertThat(headManifests.get(0).existingFilesCount()).isPositive();
assertThat(headManifests.get(1).addedFilesCount()).isPositive();

SparkMicroBatchStream stream =
newMicroBatchStream(ImmutableMap.of(), "all-available-merged-manifest-checkpoint");

try {
Offset startOffset = stream.initialOffset();
Offset endOffset = stream.latestOffset(startOffset, ReadLimit.allAvailable());

List<String> plannedFiles = Lists.newArrayList();
for (InputPartition partition : stream.planInputPartitions(startOffset, endOffset)) {
SparkInputPartition sparkInputPartition = (SparkInputPartition) partition;
for (FileScanTask task : sparkInputPartition.<FileScanTask>taskGroup().tasks()) {
plannedFiles.add(task.file().location());
}
}

assertThat(plannedFiles)
.containsExactlyInAnyOrder(
first.location(), second.location(), third.location(), fourth.location());
} finally {
stream.stop();
}
}

@TestTemplate
public void testLatestOffsetReturnsNullAfterFinalBatchIsConsumed() throws Exception {
appendDataAsMultipleSnapshots(TEST_DATA_MULTIPLE_SNAPSHOTS);
Expand Down Expand Up @@ -1400,6 +1474,15 @@ private long timestampAfterCurrentSnapshot() {
return waitUntilAfter(table.currentSnapshot().timestampMillis());
}

private DataFile newDataFile(PartitionSpec spec) {
return DataFiles.builder(spec)
.withPath(temp.resolve(UUID.randomUUID() + ".parquet").toString())
.withFileSizeInBytes(10)
.withRecordCount(1)
.withFormat(FileFormat.PARQUET)
.build();
}

private SparkMicroBatchStream newMicroBatchStream(
Map<String, String> options, String checkpointDirName) {
Map<String, String> allOptions = Maps.newHashMap(options);
Expand Down
42 changes: 42 additions & 0 deletions .../v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SyncSparkMicroBatchPlanner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.apache.iceberg.spark.SparkReadConf;
import org.apache.iceberg.util.Pair;
import org.apache.iceberg.util.SnapshotUtil;
import org.apache.spark.sql.connector.read.streaming.ReadAllAvailable;
import org.apache.spark.sql.connector.read.streaming.ReadLimit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -132,6 +133,12 @@ public StreamingOffset latestOffset(StreamingOffset startOffset, ReadLimit limit
Snapshot curSnapshot = table().snapshot(startingOffset.snapshotId());
validateCurrentSnapshotExists(curSnapshot, startingOffset);

// An unbounded limit has no per-file limit to enforce, so only the latest valid snapshot's
// manifests are indexed instead of reading every manifest between the start and the head.
if (limit instanceof ReadAllAvailable) {
return latestOffsetForUnbounded(startingOffset, curSnapshot);
}

// Use the pre-computed snapshotId when Trigger.AvailableNow is enabled.
long latestSnapshotId =
lastOffsetForTriggerAvailableNow != null
Expand Down Expand Up @@ -234,6 +241,41 @@ public StreamingOffset latestOffset(StreamingOffset startOffset, ReadLimit limit
return latestStreamingOffset.equals(startingOffset) ? null : latestStreamingOffset;
}

private StreamingOffset latestOffsetForUnbounded(
StreamingOffset startingOffset, Snapshot curSnapshot) {
Snapshot lastValidSnapshot;
if (lastOffsetForTriggerAvailableNow != null) {
lastValidSnapshot = table().snapshot(lastOffsetForTriggerAvailableNow.snapshotId());
validateCurrentSnapshotExists(lastValidSnapshot, lastOffsetForTriggerAvailableNow);
} else {
lastValidSnapshot = curSnapshot;
Snapshot next;
do {
next = nextValidSnapshot(lastValidSnapshot);
if (next != null) {
lastValidSnapshot = next;
}
} while (next != null);
}

StreamingOffset latestStreamingOffset =
new StreamingOffset(lastValidSnapshot.snapshotId(), endPosition(lastValidSnapshot), false);
return latestStreamingOffset.equals(startingOffset) ? null : latestStreamingOffset;
}

// Existing entries in merged manifests also take positions, so the end position can be larger
// than the snapshot's added-file count.
private long endPosition(Snapshot snapshot) {
List<Pair<ManifestFile, Integer>> indexedManifests =
MicroBatches.skippedManifestIndexesFromSnapshot(table().io(), snapshot, 0, false);
if (indexedManifests.isEmpty()) {
return 0;
}

Pair<ManifestFile, Integer> lastManifest = indexedManifests.get(indexedManifests.size() - 1);
return lastManifest.second() + lastManifest.first().addedFilesCount();
}

@Override
public void stop() {}

Expand Down
83 changes: 83 additions & 0 deletions ...4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,15 @@
import static org.apache.iceberg.expressions.Expressions.ref;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assumptions.assumeThat;

import java.io.File;
import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
Expand All @@ -40,9 +42,11 @@
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.Files;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.Parameter;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.RewriteFiles;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
Expand Down Expand Up @@ -71,6 +75,7 @@
import org.apache.spark.sql.Row;
import org.apache.spark.sql.connector.read.InputPartition;
import org.apache.spark.sql.connector.read.streaming.Offset;
import org.apache.spark.sql.connector.read.streaming.ReadLimit;
import org.apache.spark.sql.internal.SQLConf;
import org.apache.spark.sql.streaming.DataStreamWriter;
import org.apache.spark.sql.streaming.OutputMode;
Expand Down Expand Up @@ -490,6 +495,75 @@ public void testTriggerAvailableNowCapsAsyncPreloadAfterPrepare() {
}
}

@TestTemplate
void latestOffsetForUnboundedLimitAdvancesToHeadSnapshot() {
appendDataAsMultipleSnapshots(TEST_DATA_MULTIPLE_SNAPSHOTS);

table.refresh();
long headSnapshotId = table.currentSnapshot().snapshotId();

SparkMicroBatchStream stream =
newMicroBatchStream(ImmutableMap.of(), "all-available-latest-offset-checkpoint");

try {
Offset startOffset = stream.initialOffset();
StreamingOffset endOffset =
(StreamingOffset) stream.latestOffset(startOffset, ReadLimit.allAvailable());

assertThat(endOffset).isNotNull();
assertThat(endOffset.snapshotId()).isEqualTo(headSnapshotId);
} finally {
stream.stop();
}
}

@TestTemplate
void latestOffsetForUnboundedLimitIncludesFilesAfterMergedManifest() {
assumeThat(async)
.as("Async planning does not handle gaps left by existing files yet")
.isFalse();

PartitionSpec oldSpec = table.spec();
table.updateSpec().addField("data").commit();

DataFile first = newDataFile(table.spec());
table.newAppend().appendFile(first).commit();
DataFile second = newDataFile(table.spec());
table.newAppend().appendFile(second).commit();

// the new spec's manifests are merged and listed before the old spec's new manifest
DataFile third = newDataFile(table.spec());
DataFile fourth = newDataFile(oldSpec);
table.newAppend().appendFile(third).appendFile(fourth).commit();

List<ManifestFile> headManifests = table.currentSnapshot().dataManifests(table.io());
assertThat(headManifests).hasSize(2);
assertThat(headManifests.get(0).existingFilesCount()).isPositive();
assertThat(headManifests.get(1).addedFilesCount()).isPositive();

SparkMicroBatchStream stream =
newMicroBatchStream(ImmutableMap.of(), "all-available-merged-manifest-checkpoint");

try {
Offset startOffset = stream.initialOffset();
Offset endOffset = stream.latestOffset(startOffset, ReadLimit.allAvailable());

List<String> plannedFiles = Lists.newArrayList();
for (InputPartition partition : stream.planInputPartitions(startOffset, endOffset)) {
SparkInputPartition sparkInputPartition = (SparkInputPartition) partition;
for (FileScanTask task : sparkInputPartition.<FileScanTask>taskGroup().tasks()) {
plannedFiles.add(task.file().location());
}
}

assertThat(plannedFiles)
.containsExactlyInAnyOrder(
first.location(), second.location(), third.location(), fourth.location());
} finally {
stream.stop();
}
}

@TestTemplate
public void testLatestOffsetReturnsNullAfterFinalBatchIsConsumed() throws Exception {
appendDataAsMultipleSnapshots(TEST_DATA_MULTIPLE_SNAPSHOTS);
Expand Down Expand Up @@ -1411,6 +1485,15 @@ private long timestampAfterCurrentSnapshot() {
return waitUntilAfter(table.currentSnapshot().timestampMillis());
}

private DataFile newDataFile(PartitionSpec spec) {
return DataFiles.builder(spec)
.withPath(temp.resolve(UUID.randomUUID() + ".parquet").toString())
.withFileSizeInBytes(10)
.withRecordCount(1)
.withFormat(FileFormat.PARQUET)
.build();
}

private SparkMicroBatchStream newMicroBatchStream(
Map<String, String> options, String checkpointDirName) {
Map<String, String> allOptions = Maps.newHashMap(options);
Expand Down
Loading
Loading