Re: [PR] Spark: Aligned compaction commit filtering between fillQueue methods [iceberg]

["huan233usc (via GitHub)" <gi...@apache.org>]

huan233usc commented on code in PR #17176:
URL: https://github.com/apache/iceberg/pull/17176#discussion_r3619780455


##########
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)) {

Review Comment:
   Nice catch. This mirrors what `SyncSparkMicroBatchPlanner#planFiles` already does, so the two planners are now consistent.
   
   One thing worth calling out in the description: `shouldProcess` also throws for DELETE/OVERWRITE snapshots when the corresponding skip options aren't set. So beyond dropping REPLACE, a first/partial offset landing on a delete/overwrite snapshot will now surface that error instead of being silently queued. This matches the sync behavior, but it's a slightly broader change than just "drop the replacement commit".



##########
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 {

Review Comment:
   The parameterized coverage is great here — this exercises both the async (`async=true`) and sync (`async=false`) paths. Optional: a short comment noting that without the fix the assertion would also observe batch1's rewritten rows would help future readers see this is a regression test.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org