diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java
index 3e442f9917d4..8fa5f45bc3d9 100644
--- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java
+++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java
@@ -428,7 +428,7 @@ private void fillQueue(StreamingOffset fromOffset, StreamingOffset toOffset) {
     LOG.debug("filling queue from {}, to: {}", fromOffset, toOffset);
     Snapshot currentSnapshot = table().snapshot(fromOffset.snapshotId());
     // this could be a partial snapshot so add it outside the loop
-    if (currentSnapshot != null) {
+    if (currentSnapshot != null && shouldProcess(currentSnapshot)) {
       addMicroBatchToQueue(
           currentSnapshot,
           fromOffset.position(),
diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
index 89947f73ea38..f283253c1fa5 100644
--- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
+++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
@@ -826,6 +826,25 @@ public void testReadStreamFromEmptyTable() throws Exception {
     assertThat(actual).isEmpty();
   }
 
+  @TestTemplate
+  public void testStreamingWithReplaceSnapshot() throws Exception {
+    List<SimpleRecord> batchAddedBeforeCompaction =
+        Lists.newArrayList(new SimpleRecord(1, "val1"), new SimpleRecord(2, "val2"));
+    appendData(batchAddedBeforeCompaction);
+    table.refresh();
+
+    long timeAfterFirstAppend = table.currentSnapshot().timestampMillis() + 1;
+    waitUntilAfter(timeAfterFirstAppend);
+    makeRewriteDataFiles();
+
+    List<SimpleRecord> batchAddedAfterCompaction =
+        Lists.newArrayList(new SimpleRecord(3, "val3"), new SimpleRecord(4, "val4"));
+    appendData(batchAddedAfterCompaction);
+    StreamingQuery query =
+        startStream(SparkReadOptions.STREAM_FROM_TIMESTAMP, Long.toString(timeAfterFirstAppend));
+    assertThat(rowsAvailable(query)).containsExactlyInAnyOrderElementsOf(batchAddedAfterCompaction);
+  }
+
   @TestTemplate
   public void testReadStreamWithSnapshotTypeOverwriteErrorsOut() throws Exception {
     // upgrade table to version 2 - to facilitate creation of Snapshot of type OVERWRITE.