Spark: Async Spark Micro Batch Planner by RjLi13 · Pull Request #15059 · apache/iceberg

RjLi13 · 2026-01-15T19:12:08Z

Implements a new feature for Spark Structured Streaming and Iceberg users known as Async Spark Micro Batch Planner

Currently Microbatch planning in Iceberg is synchronous. Streaming queries plan out what batches to read and how many rows / files in each batch. Then it processes the data and repeats. By introducing an async planner, it improves streaming performance by pre-fetching table metadata and file scan tasks in a background thread, reducing micro-batch planning latency. This way planning can overlap with data processing and speed up dealing with large volumes.

This PR adds the option for users to set spark.sql.iceberg.async-micro-batch-planning-enabled if they want to use async planning. The code in SparkMicroBatchStream.java is moved to SyncSparkMicroBatchPlanner.java and SparkMicroBatchStream configures which planner to use. This option is defaulted to false, so existing behavior is unchanged.

This feature was originally authored by Drew Goya in our Netflix fork for Spark 3.3 & Iceberg 1.4. I built upon Drew's work by porting this to Spark ~~3.5~~ 4.1 and current Iceberg version.

Changes

New AsyncSparkMicroBatchPlanner that queues file scan tasks asynchronously
Refactored existing sync logic into SyncSparkMicroBatchPlanner
Created SparkMicroBatchPlanner interface for both implementations
SparkMicroBatchStream now selects planner based on configuration
Created new BaseSparkMicroBatchPlanner to dedupe code between Sync and Async Planners

RjLi13 · 2026-01-15T19:13:38Z

cc @bryanck

bryanck · 2026-01-15T19:21:12Z

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/StreamingOffset.java


-class StreamingOffset extends Offset {
-  static final StreamingOffset START_OFFSET = new StreamingOffset(-1L, -1, false);
+public class StreamingOffset extends Offset {


Does this need to be public?

bryanck · 2026-01-15T19:23:27Z

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/StreamingOffset.java

+   * @param snapshotTotalRows Total rows in the snapshot
   */
-  StreamingOffset(long snapshotId, long position, boolean scanAllFiles) {
+  public StreamingOffset(


Also here, curious why this needs to be public. Also below there are a few static methods that were made public.

Good catch, I originally had it to match the public interface SparkMicroBatchPlanner, only to realize all of them are implementation details. Reverting and making the planners package private as well

bryanck · 2026-01-15T19:26:33Z

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java

+    this.minQueuedRows = readConf.maxRecordsPerMicroBatch();
+    this.readConf = readConf;
+    this.lastOffsetForTriggerAvailableNow = lastOffsetForTriggerAvailableNow;
+    this.planFilesCache = Caffeine.newBuilder().maximumSize(10).build();


nit: make 10 a constant or configurable

bryanck · 2026-01-15T19:26:55Z

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java

+        this::refreshAndTrapException, pollingIntervalMs, pollingIntervalMs, TimeUnit.MILLISECONDS);
+    // Schedule queue fill to run frequently (use polling interval for tests, cap at 100ms for
+    // production)
+    long queueFillIntervalMs = Math.min(100L, pollingIntervalMs);


nit: make 100L a constant or configurable

bryanck · 2026-01-15T19:49:22Z

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/StreamingOffset.java

  private final long position;
  private final boolean scanAllFiles;
+  private final long snapshotTimestampMillis;
+  private final long snapshotTotalRows;


I wasn't clear why these need to be added, given we already have the snapshot ID?

Reverted all changes to StreamingOffset

bryanck · 2026-01-15T19:52:09Z

There is a fair amount of duplicated code between the existing microbatch planner and this, I'm wondering if there are opportunities for reuse.

RjLi13 · 2026-01-16T21:44:51Z

There is a fair amount of duplicated code between the existing microbatch planner and this, I'm wondering if there are opportunities for reuse.

Moved duplicated code out to BaseSparkMicroBatchPlanner abstract class that both Planners now extend.

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/BaseSparkMicroBatchPlanner.java

bryanck · 2026-01-19T19:46:56Z

This should be added to the latest Spark version. Once that is merged, we'll backport the changes.

bryanck · 2026-01-19T19:59:49Z

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchPlanner.java

+
+interface SparkMicroBatchPlanner {
+  List<FileScanTask> planFiles(StreamingOffset start, StreamingOffset end)
+      throws ExecutionException;


I feel ExecutionException should not be part of the method signature, the implementation can throw an unchecked exception.

bryanck · 2026-01-19T20:00:38Z

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java

+    List<FileScanTask> fileScanTasks;
+    try {
+      fileScanTasks = microBatchPlanner.planFiles(startOffset, endOffset);
+    } catch (ExecutionException e) {


We can move exception handling to implementation if we make the above change to the interface.

bryanck · 2026-01-19T20:04:57Z

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java

-          return ((ReadMaxFiles) limit).maxFiles();
-        }
+  @Override
+  public Map<String, String> metrics(Optional<Offset> latestConsumedOffset) {


It seems like this is only useful for the async planner, for the sync planner it will report the same thing.

Any recommendations on how to isolate it to Async Planner? Else I will remove it, and users can rely on logging instead, which will minimize implementing on ReportsSourceMetrics Interface

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java

bryanck · 2026-01-19T20:12:08Z

If you have any benchmarks or metrics that show the benefit of this, that would be helpful.

bryanck · 2026-01-19T20:19:13Z

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SyncSparkMicroBatchPlanner.java

+    List<FileScanTask> fileScanTasks = Lists.newArrayList();
+    StreamingOffset batchStartOffset =
+        StreamingOffset.START_OFFSET.equals(start)
+            ? SparkMicroBatchStream.determineStartingOffset(table(), fromTimestamp)


There is a circular dependency here, it would be better to have the planners not rely on SparkMicroBatchStream.

bryanck · 2026-01-19T20:21:17Z

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java

  }

-  private static StreamingOffset determineStartingOffset(Table table, Long fromTimestamp) {
+  static StreamingOffset determineStartingOffset(Table table, long fromTimestamp) {


IMO we should move this somewhere else so the planners don't need to reference SparkMicroBatchStream.

RjLi13 · 2026-01-20T17:59:24Z

This should be added to the latest Spark version. Once that is merged, we'll backport the changes.

@bryanck Would it be ok to move these files to 4.1 in this PR? Or would you prefer having a separate PR there and moving conversation and review to that PR?

RjLi13 · 2026-01-22T00:34:32Z

commit f5e7c6b Spark 3.5: Address circular dep, unneeded code: Addresses all the review comments about circular dependency, executionexception and possibly unnecessary metrics call

commit [e079fd3] Spark: Move feature to latest Spark: Addresses comment about making this feature for latest / current Spark, then separate PR for backporting

cc @bryanck

The only remaining comment I did not address for code review was #15059 (comment), to discuss.

RjLi13 · 2026-01-22T00:36:16Z

Sorry the branch name has 3-5 in it. But change PR to now target 4.1.

RjLi13 · 2026-01-22T00:40:46Z

...k/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java

    // Data appended after the timestamp should appear
    appendData(data);
+    // Allow async background thread to refresh, else test sometimes fails
+    Thread.sleep(50);


I'm open to other suggestions. This test sometimes fails, because background thread runs and isn't able to refresh before test finishes executing. Hence I added sleep here to wait before test finishes. In real use case, user should not have issues with this.

It failed for me locally once before this, but ran multiple times without it failing.

RjLi13 · 2026-01-27T16:29:35Z

If you have any benchmarks or metrics that show the benefit of this, that would be helpful.

I don't have the production metrics or benchmarks, but I took a stab at trying to create a JMH benchmark on latestOffset() calls with different snapshot numbers. Hopefully, this showcases sync planner scales linearly to number of snapshots whereas async planner is constant, but trading off detection latency and memory usage. It is great for tables with large number of snapshots and where processing time takes long enough for background thread to poll. Here is the link to it: RjLi13#2

bryanck · 2026-02-02T18:32:24Z

nit: I feel the refactor could have reused the existing code a little bit better, i.e. renamed the existing class to be the base class first, that would help reduce the size of the PR

This feature was originally built by Drew Goya <dgoya@netflix.com> for Spark 3.3 and Iceberg 1.4.

RjLi13 · 2026-02-11T19:53:40Z

Spoke with @bryanck offline. Decided that this PR should be split into two, one refactoring SparkMicroBatchStream logic into SyncSparkMicroBatchPlanner, and two actually introducing async. This PR will be closed in favor for those two new ones:
#15298
#15299

github-actions bot added the spark label Jan 15, 2026

bryanck reviewed Jan 15, 2026

View reviewed changes

RjLi13 force-pushed the async-micro-batch-planner-spark-3-5 branch 2 times, most recently from 97cfdbf to 0cfb405 Compare January 16, 2026 21:42

RjLi13 force-pushed the async-micro-batch-planner-spark-3-5 branch from d31b909 to d7f2883 Compare January 16, 2026 23:16

RjLi13 commented Jan 16, 2026

View reviewed changes

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/BaseSparkMicroBatchPlanner.java Outdated Show resolved Hide resolved

bryanck reviewed Jan 19, 2026

View reviewed changes

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java Show resolved Hide resolved

bryanck reviewed Jan 19, 2026

View reviewed changes

RjLi13 force-pushed the async-micro-batch-planner-spark-3-5 branch from d7f2883 to f5e7c6b Compare January 21, 2026 22:20

RjLi13 changed the title ~~Spark 3.5: Async Spark Micro Batch Planner~~ Spark: Async Spark Micro Batch Planner Jan 22, 2026

RjLi13 commented Jan 22, 2026

View reviewed changes

Spark 3.5: Asynchronous Spark Micro Batch Planner

c4280c1

Ruijing Li added 7 commits February 11, 2026 00:57

Spark: Credit original async micro-batch planner implementation

74b7b84

This feature was originally built by Drew Goya <dgoya@netflix.com> for Spark 3.3 and Iceberg 1.4.

Spark 3.5: Dedupe planners, reduce visibility, simplify offset

0c7026e

Revert unnecessary SparkMicroBatchStream changes

1be3104

Spark 3.5: Address circular dep, unneeded code

ac913ff

Fix formatting issues

6f544c9

Spark: Move feature to latest Spark

56b411b

Fix test failing sometimes when ran too fast

4a102ab

RjLi13 force-pushed the async-micro-batch-planner-spark-3-5 branch from 7292949 to 4a102ab Compare February 11, 2026 09:11

This was referenced Feb 11, 2026

Spark 4.1: Refactor SparkMicroBatchStream to SyncPlanner #15298

Merged

Spark 4.1: New Async Spark Micro Batch Planner #15299

Open

RjLi13 closed this Feb 11, 2026

RjLi13 deleted the async-micro-batch-planner-spark-3-5 branch February 11, 2026 19:53

Conversation

RjLi13 commented Jan 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Changes

Uh oh!

RjLi13 commented Jan 15, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bryanck Jan 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bryanck Jan 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bryanck commented Jan 15, 2026

Uh oh!

RjLi13 commented Jan 16, 2026

Uh oh!

Uh oh!

bryanck commented Jan 19, 2026

Uh oh!

bryanck Jan 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bryanck Jan 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

bryanck commented Jan 19, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RjLi13 commented Jan 20, 2026

Uh oh!

RjLi13 commented Jan 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

RjLi13 commented Jan 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RjLi13 commented Jan 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

bryanck commented Feb 2, 2026

Uh oh!

RjLi13 commented Feb 11, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

RjLi13 commented Jan 15, 2026 •

edited

Loading

bryanck Jan 15, 2026 •

edited

Loading

bryanck Jan 15, 2026 •

edited

Loading

bryanck Jan 19, 2026 •

edited

Loading

bryanck Jan 19, 2026 •

edited

Loading

RjLi13 commented Jan 22, 2026 •

edited

Loading

RjLi13 commented Jan 22, 2026 •

edited

Loading

RjLi13 commented Jan 27, 2026 •

edited

Loading