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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
