anuragmantri commented on code in PR #18348:
URL: https://github.com/apache/iceberg/pull/18348#discussion_r4199245871
##########
spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java:
##########
@@ -572,6 +575,125 @@ public void
testPlanInputPartitionsIsIdempotentForSameOffsets() {
}
}
+ @TestTemplate
+ void latestOffsetIgnoresStreamFromTimestampWhenResuming() {
+ appendData(List.of(new SimpleRecord(1, "one")));
+ table.refresh();
+ Snapshot resumedSnapshot = table.currentSnapshot();
+ appendData(List.of(new SimpleRecord(2, "two")));
+ table.refresh();
+ long latestSnapshotId = table.currentSnapshot().snapshotId();
+
+ SparkMicroBatchStream stream =
+ newMicroBatchStream(
+ ImmutableMap.of(
+ SparkReadOptions.STREAM_FROM_TIMESTAMP,
+ Long.toString(timestampAfterCurrentSnapshot())),
+ "resumed-stream-checkpoint");
+
+ try {
+ StreamingOffset resumedOffset =
+ new StreamingOffset(
+ resumedSnapshot.snapshotId(),
+ MicroBatchUtils.addedFilesCount(table, resumedSnapshot),
+ false);
+ StreamingOffset endOffset =
+ (StreamingOffset) stream.latestOffset(resumedOffset,
stream.getDefaultReadLimit());
+
+ assertThat(endOffset.snapshotId()).isEqualTo(latestSnapshotId);
+ } finally {
+ stream.stop();
+ }
+ }
+
+ @TestTemplate
+ void availableNowResumeStopsAtSnapshotsAvailableAtStart() {
+ appendData(List.of(new SimpleRecord(1, "one")));
+ table.refresh();
+ Snapshot committed = table.currentSnapshot();
+ appendData(List.of(new SimpleRecord(2, "two")));
+ table.refresh();
+ Snapshot latest = table.currentSnapshot();
+ // A timestamp after the newest snapshot stores START_OFFSET as the
initial offset
+ SparkMicroBatchStream stream =
+ newMicroBatchStream(
+ ImmutableMap.of(
+ SparkReadOptions.STREAM_FROM_TIMESTAMP,
+ Long.toString(timestampAfterCurrentSnapshot())),
+ "available-now-resume-checkpoint");
+
+ try {
+ stream.prepareForTriggerAvailableNow();
+ appendData(List.of(new SimpleRecord(3, "three")));
+ StreamingOffset committedOffset =
+ new StreamingOffset(
+ committed.snapshotId(), MicroBatchUtils.addedFilesCount(table,
committed), false);
+
+ assertThat(stream.latestOffset(committedOffset,
stream.getDefaultReadLimit()))
+ .isEqualTo(
+ new StreamingOffset(
+ latest.snapshotId(), MicroBatchUtils.addedFilesCount(table,
latest), false));
+ } finally {
+ stream.stop();
+ }
+ }
+
+ @TestTemplate
+ void availableNowResumeStopsAtLatestAppendSnapshot() {
+ appendData(List.of(new SimpleRecord(1, "one")));
+ table.refresh();
+ Snapshot committed = table.currentSnapshot();
+ appendData(List.of(new SimpleRecord(2, "two")));
+ table.refresh();
+ Snapshot latestAppend = table.currentSnapshot();
+ table.rewriteManifests().clusterBy(file -> 1).commit();
+ table.refresh();
+
assertThat(table.currentSnapshot().operation()).isEqualTo(DataOperations.REPLACE);
+ SparkMicroBatchStream stream =
+ newMicroBatchStream(
+ ImmutableMap.of(
+ SparkReadOptions.STREAM_FROM_TIMESTAMP,
+ Long.toString(timestampAfterCurrentSnapshot())),
+ "available-now-resume-replace-checkpoint");
+
+ try {
+ stream.prepareForTriggerAvailableNow();
Review Comment:
Good call on compaction. Capping at the newest append makes sense. I tried
changing `latestAppendOffset()` to return the current snapshot, which here is
the `REPLACE` from `rewriteManifests()`. This test still passes with
`async=false`. The sync planner skips the `REPLACE` and stops at the current
snapshot anyway, because nothing is committed after prepare. Committing one
more snapshot after `prepareForTriggerAvailableNow` makes the test fail for
both planners under that change, and it passes on this branch.
```suggestion
stream.prepareForTriggerAvailableNow();
appendData(List.of(new SimpleRecord(3, "three")));
```
--
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]
