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:
This lines up with what sync already does in planFiles, good.
One heads up: shouldProcess also throws on delete/overwrite snapshots when
the skip options aren't set, so it's not only dropping replace. If the
starting/partial offset happens to land on a delete or overwrite snapshot it'll
now error out instead of being queued silently. Same as sync, just a bit
broader than what the title implies.
##########
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:
Nice that this covers both async and sync via the parameter. Might be worth
a one-liner saying that without the fix you'd also see batch1's rewritten rows
here, so it's obvious this is guarding against the regression.
--
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]
