Skip to content

Core: Replace SnapshotUtil firstSnapshotAfterTimestamp #3775

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rdblue merged 5 commits into from
Dec 21, 2021
Merged

Core: Replace SnapshotUtil firstSnapshotAfterTimestamp #3775

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
448238e
Core: Replace SnapshotUtil firstSnapshotAfterTimestamp.
rdblue Dec 20, 2021
f35a708
Fix boundary condition.
rdblue Dec 20, 2021
5d26e8a
Port changes to Spark 3.0.
rdblue Dec 20, 2021
e638b67
Spark: Refactor Spark Streaming Tests
RussellSpitzer Dec 20, 2021
d6dfa77
Spark: Refactor Spark streaming tests
rdblue Dec 20, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 16 additions & 24 deletions core/src/main/java/org/apache/iceberg/util/SnapshotUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,43 +102,35 @@ public static Iterable<Snapshot> ancestorsOf(long snapshotId, Function<Long, Sna
}

/**
* Traverses the history of the table's current snapshot and:
* 1. returns null, if no snapshot exists or target timestamp is more recent than the current snapshot.
* 2. else return the first snapshot which satisfies {@literal >=} targetTimestamp.
* <p>
* Given the snapshots (with timestamp): [S1 (10), S2 (11), S3 (12), S4 (14)]
* <p>
* firstSnapshotAfterTimestamp(table, x {@literal <=} 10) = S1
* firstSnapshotAfterTimestamp(table, 11) = S2
* firstSnapshotAfterTimestamp(table, 13) = S4
* firstSnapshotAfterTimestamp(table, 14) = S4
* firstSnapshotAfterTimestamp(table, x {@literal >} 14) = null
* <p>
* where x is the target timestamp in milliseconds and Si is the snapshot
* Traverses the history of the table's current snapshot and finds the first snapshot after the given timestamp.
Comment thread
rdblue marked this conversation as resolved.
Outdated
*
* @param table a table
* @param targetTimestampMillis a timestamp in milliseconds
* @return the first snapshot which satisfies {@literal >=} targetTimestamp, or null if the current snapshot is
* more recent than the target timestamp
* @param timestampMillis a timestamp in milliseconds
* @return the first snapshot after the given timestamp, or null if the current snapshot is older than the timestamp

@RussellSpitzer RussellSpitzer Dec 20, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would add the word "committed" again here and before all the "afters". I don't know why I can't handle both concepts in my head at the same time, but I keep visualizing a linked list fo snapshots.

@szehon-ho szehon-ho Dec 20, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm trying to understand the removal of ">=" from this and other comments, is it intended?

If so, it doens't seem to match the code below which still seems to return ">="

else if (snapshot.timestampMillis() == timestampMillis() 
   return snapshot;

@rdblue rdblue Dec 20, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I replaced the previous Javadoc with the original one that I wrote. I can be more clear here if needed.

@szehon-ho szehon-ho Dec 20, 2021 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK, got it that it's a new method rather than change. To me the previous javadoc is more clear, as "older" gives the impression of strictly greater than.

* @throws IllegalStateException if the first ancestor after the given time can't be determined
*/
public static Snapshot firstSnapshotAfterTimestamp(Table table, Long targetTimestampMillis) {
Snapshot currentSnapshot = table.currentSnapshot();
// Return null if no snapshot exists or target timestamp is more recent than the current snapshot
if (currentSnapshot == null || currentSnapshot.timestampMillis() < targetTimestampMillis) {
public static Snapshot oldestAncestorAfter(Table table, long timestampMillis) {
if (table.currentSnapshot() == null) {
// there are no snapshots or ancestors
return null;
}

// Return the oldest snapshot which satisfies >= targetTimestamp
Snapshot lastSnapshot = null;
for (Snapshot snapshot : currentAncestors(table)) {
if (snapshot.timestampMillis() < targetTimestampMillis) {
if (snapshot.timestampMillis() <= timestampMillis) {
return lastSnapshot;
}

lastSnapshot = snapshot;
}

// Return the oldest snapshot if the target timestamp is less than the oldest snapshot of the table
return lastSnapshot;
if (lastSnapshot != null && lastSnapshot.parentId() == null) {
// this is the first snapshot in the table, return it

@RussellSpitzer RussellSpitzer Dec 20, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am a little worried about having a function which works for a given input but only until the starting snapshot is expired. For example

oldestAncestorAfter(table,  Long.MinValue) // Returns first snapshot
expireSnapshots() // Expire first snapshot
oldestAncestorAfter(table,  Long.MinValue) // Throws exception

I think if we want to standardize this should probably also throw an exception

@rdblue rdblue Dec 20, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see your point here, but the result is based on the table state that gets passed in. If the table state is missing information, then we can't make it consistent.

Here's another way to think about it:

t1 = commitSnapshotOne()
t2 = commitSnapshotTwo()
oldestAncestorAfter(table, Long.MinValue) // returns snapshot one
expireSnapshots(t2 - 1)
oldestAncestorAfter(table, Long.MinValue) // returns snapshot two

I think that the behavior above is worse than throwing an exception based on the table state because it is silently inconsisent. At least throwing an exception tells you why it isn't returning the expected value.

@RussellSpitzer RussellSpitzer Dec 20, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree that we should change it, I just think we can just always throw the exception so

oldestAncestorAfter(table,  Long.MinValue) // Throw exception
expireSnapshots() // Expire first snapshot
oldestAncestorAfter(table,  Long.MinValue) // Throw exception

return lastSnapshot;
}

throw new IllegalStateException(
"Cannot find snapshot older than " + DateTimeUtil.formatTimestampMillis(timestampMillis));
}

/**
Expand Down
52 changes: 26 additions & 26 deletions spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,12 +99,12 @@ public class SparkMicroBatchStream implements MicroBatchStream {
@Override
public Offset latestOffset() {
table.refresh();
if (isStreamEmpty(table)) {
if (table.currentSnapshot() == null) {
return StreamingOffset.START_OFFSET;
}

if (isFutureStartTime(table, fromTimestamp)) {
return initialFutureStartOffset(table);
if (table.currentSnapshot().timestampMillis() < fromTimestamp) {
return StreamingOffset.START_OFFSET;
}

Snapshot latestSnapshot = table.currentSnapshot();
Expand Down Expand Up @@ -169,8 +169,7 @@ public void stop() {
private List<FileScanTask> planFiles(StreamingOffset startOffset, StreamingOffset endOffset) {
List<FileScanTask> fileScanTasks = Lists.newArrayList();
StreamingOffset batchStartOffset = StreamingOffset.START_OFFSET.equals(startOffset) ?
new StreamingOffset(SnapshotUtil.firstSnapshotAfterTimestamp(table, fromTimestamp).snapshotId(), 0, false) :
startOffset;
determineStartingOffset(table, fromTimestamp) : startOffset;

StreamingOffset currentOffset = null;

Expand Down Expand Up @@ -208,26 +207,31 @@ private boolean shouldProcess(Snapshot snapshot) {
return op.equals(DataOperations.APPEND);
}

private static boolean isStreamEmpty(Table table) {
return table.currentSnapshot() == null;
}

private static boolean isStreamNotEmpty(Table table) {
return table.currentSnapshot() != null;
}
private static StreamingOffset determineStartingOffset(Table table, Long fromTimestamp) {
if (table.currentSnapshot() == null) {
return StreamingOffset.START_OFFSET;
}

private static boolean isFutureStartTime(Table table, Long streamStartTimeStampMillis) {
if (streamStartTimeStampMillis == null) {
return false;
if (fromTimestamp == null) {
// match existing behavior and start from the oldest snapshot
return new StreamingOffset(SnapshotUtil.oldestAncestor(table).snapshotId(), 0, false);
}

return table.currentSnapshot().timestampMillis() < streamStartTimeStampMillis;
}
if (table.currentSnapshot().timestampMillis() < fromTimestamp) {
return StreamingOffset.START_OFFSET;
}

private static StreamingOffset initialFutureStartOffset(Table table) {
Preconditions.checkNotNull(table, "Cannot process future start offset with invalid table input.");
Snapshot latestSnapshot = table.currentSnapshot();
return new StreamingOffset(latestSnapshot.snapshotId(), Iterables.size(latestSnapshot.addedFiles()) + 1, false);
try {
Snapshot snapshot = SnapshotUtil.oldestAncestorAfter(table, fromTimestamp);
if (snapshot != null) {
return new StreamingOffset(snapshot.snapshotId(), 0, false);
} else {
return StreamingOffset.START_OFFSET;
}
} catch (IllegalStateException e) {
// could not determine the first snapshot after the timestamp. use the oldest ancestor instead
return new StreamingOffset(SnapshotUtil.oldestAncestor(table).snapshotId(), 0, false);
}
}

private static class InitialOffsetStore {
Expand All @@ -250,11 +254,7 @@ public StreamingOffset initialOffset() {
}

table.refresh();
StreamingOffset offset = StreamingOffset.START_OFFSET;
if (isStreamNotEmpty(table)) {
offset = isFutureStartTime(table, fromTimestamp) ? initialFutureStartOffset(table) :
new StreamingOffset(SnapshotUtil.firstSnapshotAfterTimestamp(table, fromTimestamp).snapshotId(), 0, false);
}
StreamingOffset offset = determineStartingOffset(table, fromTimestamp);

OutputFile outputFile = io.newOutputFile(initialOffsetLocation);
writeOffset(offset, outputFile);
Expand Down
52 changes: 26 additions & 26 deletions spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,12 +99,12 @@ public class SparkMicroBatchStream implements MicroBatchStream {
@Override
public Offset latestOffset() {
table.refresh();
if (isStreamEmpty(table)) {
if (table.currentSnapshot() == null) {
return StreamingOffset.START_OFFSET;
}

if (isFutureStartTime(table, fromTimestamp)) {
return initialFutureStartOffset(table);
if (table.currentSnapshot().timestampMillis() < fromTimestamp) {
return StreamingOffset.START_OFFSET;
}

Snapshot latestSnapshot = table.currentSnapshot();
Expand Down Expand Up @@ -169,8 +169,7 @@ public void stop() {
private List<FileScanTask> planFiles(StreamingOffset startOffset, StreamingOffset endOffset) {
List<FileScanTask> fileScanTasks = Lists.newArrayList();
StreamingOffset batchStartOffset = StreamingOffset.START_OFFSET.equals(startOffset) ?
new StreamingOffset(SnapshotUtil.firstSnapshotAfterTimestamp(table, fromTimestamp).snapshotId(), 0, false) :
startOffset;
determineStartingOffset(table, fromTimestamp) : startOffset;

StreamingOffset currentOffset = null;

Expand Down Expand Up @@ -208,26 +207,31 @@ private boolean shouldProcess(Snapshot snapshot) {
return op.equals(DataOperations.APPEND);
}

private static boolean isStreamEmpty(Table table) {
return table.currentSnapshot() == null;
}

private static boolean isStreamNotEmpty(Table table) {
return table.currentSnapshot() != null;
}
private static StreamingOffset determineStartingOffset(Table table, Long fromTimestamp) {
if (table.currentSnapshot() == null) {
return StreamingOffset.START_OFFSET;
}

private static boolean isFutureStartTime(Table table, Long streamStartTimeStampMillis) {
if (streamStartTimeStampMillis == null) {
return false;
if (fromTimestamp == null) {
// match existing behavior and start from the oldest snapshot
return new StreamingOffset(SnapshotUtil.oldestAncestor(table).snapshotId(), 0, false);
}

return table.currentSnapshot().timestampMillis() < streamStartTimeStampMillis;
}
if (table.currentSnapshot().timestampMillis() < fromTimestamp) {
return StreamingOffset.START_OFFSET;
}

private static StreamingOffset initialFutureStartOffset(Table table) {
Preconditions.checkNotNull(table, "Cannot process future start offset with invalid table input.");
Snapshot latestSnapshot = table.currentSnapshot();
return new StreamingOffset(latestSnapshot.snapshotId(), Iterables.size(latestSnapshot.addedFiles()) + 1, false);
try {
Snapshot snapshot = SnapshotUtil.oldestAncestorAfter(table, fromTimestamp);
if (snapshot != null) {
return new StreamingOffset(snapshot.snapshotId(), 0, false);
} else {
return StreamingOffset.START_OFFSET;
}
} catch (IllegalStateException e) {
// could not determine the first snapshot after the timestamp. use the oldest ancestor instead
return new StreamingOffset(SnapshotUtil.oldestAncestor(table).snapshotId(), 0, false);
}
}

private static class InitialOffsetStore {
Expand All @@ -250,11 +254,7 @@ public StreamingOffset initialOffset() {
}

table.refresh();
StreamingOffset offset = StreamingOffset.START_OFFSET;
if (isStreamNotEmpty(table)) {
offset = isFutureStartTime(table, fromTimestamp) ? initialFutureStartOffset(table) :
new StreamingOffset(SnapshotUtil.firstSnapshotAfterTimestamp(table, fromTimestamp).snapshotId(), 0, false);
}
StreamingOffset offset = determineStartingOffset(table, fromTimestamp);

OutputFile outputFile = io.newOutputFile(initialOffsetLocation);
writeOffset(offset, outputFile);
Expand Down