Skip to content
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

[GOBBLIN-2169] Fix NPE when table has no snapshot while accessing current snapshot #4073

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[GOBBLIN-2169] Fix NPE when table has no snapshot while accessing current snapshot #4073

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4879219
throw exception incase no snapshot found for copying
Blazer-007 Nov 4, 2024
eb3ef7d
added nosnapshotfound exception
Blazer-007 Dec 7, 2024
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
31 changes: 23 additions & 8 deletions ...anagement/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,12 @@ public TableNotFoundException(TableIdentifier tableId) {
}
}

public static class NoSnapshotFoundException extends IOException {
public NoSnapshotFoundException(TableIdentifier tableId) {
super("No Snapshot found: '" + tableId + "'");
}
}

@Getter
private final TableIdentifier tableId;
/** allow the {@link IcebergCatalog} creating this table to qualify its {@link DatasetDescriptor#getName()} used for lineage, etc. */
Expand All @@ -97,19 +103,22 @@ public TableNotFoundException(TableIdentifier tableId) {
/** @return metadata info limited to the most recent (current) snapshot */
public IcebergSnapshotInfo getCurrentSnapshotInfo() throws IOException {
TableMetadata current = accessTableMetadata();
return createSnapshotInfo(current.currentSnapshot(), Optional.of(current.metadataFileLocation()), Optional.of(current));
Snapshot currentSnapshot = accessCurrentSnapshot(current);
return createSnapshotInfo(currentSnapshot, Optional.of(current.metadataFileLocation()), Optional.of(current));
}

/** @return metadata info for most recent snapshot, wherein manifests and their child data files ARE NOT listed */
public IcebergSnapshotInfo getCurrentSnapshotInfoOverviewOnly() throws IOException {
TableMetadata current = accessTableMetadata();
return createSnapshotInfo(current.currentSnapshot(), Optional.of(current.metadataFileLocation()), Optional.of(current), true);
Snapshot currentSnapshot = accessCurrentSnapshot(current);
return createSnapshotInfo(currentSnapshot, Optional.of(current.metadataFileLocation()), Optional.of(current), true);
}

/** @return metadata info for all known snapshots, ordered historically, with *most recent last* */
public Iterator<IcebergSnapshotInfo> getAllSnapshotInfosIterator() throws IOException {
TableMetadata current = accessTableMetadata();
long currentSnapshotId = current.currentSnapshot().snapshotId();
Snapshot currentSnapshot = accessCurrentSnapshot(current);
long currentSnapshotId = currentSnapshot.snapshotId();
List<Snapshot> snapshots = current.snapshots();
return Iterators.transform(snapshots.iterator(), snapshot -> {
try {
Expand Down Expand Up @@ -172,6 +181,12 @@ protected TableMetadata accessTableMetadata() throws TableNotFoundException {
return Optional.ofNullable(current).orElseThrow(() -> new TableNotFoundException(this.tableId));
}

/** @throws {@link IcebergTable.NoSnapshotFoundException} when table is empty i.e. table has zero snapshot */
protected Snapshot accessCurrentSnapshot(TableMetadata tableMetadata) throws NoSnapshotFoundException {
Snapshot currentSnapshot = tableMetadata.currentSnapshot();
return Optional.ofNullable(currentSnapshot).orElseThrow(() -> new NoSnapshotFoundException(this.tableId));
}

protected IcebergSnapshotInfo createSnapshotInfo(Snapshot snapshot, Optional<String> metadataFileLocation, Optional<TableMetadata> currentTableMetadata)
throws IOException {
return createSnapshotInfo(snapshot, metadataFileLocation, currentTableMetadata, false);
Expand Down Expand Up @@ -239,7 +254,7 @@ protected void registerIcebergTable(TableMetadata srcMetadata, TableMetadata dst
public List<DataFile> getPartitionSpecificDataFiles(Predicate<StructLike> icebergPartitionFilterPredicate)
throws IOException {
TableMetadata tableMetadata = accessTableMetadata();
Snapshot currentSnapshot = tableMetadata.currentSnapshot();
Snapshot currentSnapshot = accessCurrentSnapshot(tableMetadata);
long currentSnapshotId = currentSnapshot.snapshotId();
List<DataFile> knownDataFiles = new ArrayList<>();
GrowthMilestoneTracker growthMilestoneTracker = new GrowthMilestoneTracker();
Expand Down Expand Up @@ -286,10 +301,10 @@ protected void overwritePartition(List<DataFile> dataFiles, String partitionColN
return;
}
TableMetadata tableMetadata = accessTableMetadata();
Optional<Snapshot> currentSnapshot = Optional.ofNullable(tableMetadata.currentSnapshot());
if (currentSnapshot.isPresent()) {
log.info("~{}~ SnapshotId before overwrite: {}", tableId, currentSnapshot.get().snapshotId());
} else {
try {
Snapshot currentSnapshot = accessCurrentSnapshot(tableMetadata);
log.info("~{}~ SnapshotId before overwrite: {}", tableId, currentSnapshot.snapshotId());
} catch (NoSnapshotFoundException e) {
log.warn("~{}~ No current snapshot found before overwrite", tableId);
}
OverwriteFiles overwriteFiles = this.table.newOverwrite();
Expand Down
8 changes: 8 additions & 0 deletions ...ement/src/test/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTableTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,14 @@ public void testGetCurrentSnapshotInfoOnBogusTable() throws IOException {
Assert.fail("expected an exception when using table ID '" + bogusTableId + "'");
}

/** Verify failure when attempting to get current snapshot info for an empty table */
@Test(expectedExceptions = IcebergTable.NoSnapshotFoundException.class)
public void testGetCurrentSnapshotInfoOnEmptyTable() throws IOException {
IcebergSnapshotInfo snapshotInfo = new IcebergTable(tableId, catalog.newTableOps(tableId), catalogUri,
catalog.loadTable(tableId)).getCurrentSnapshotInfo();
Assert.fail("expected an exception when using table ID '" + tableId + "'");
}

/** Verify info about all (full) snapshots */
@Test
public void testGetAllSnapshotInfosIterator() throws IOException {
Expand Down