From edaf474accf85539156b95b121b2e49bba75d4b1 Mon Sep 17 00:00:00 2001
From: Vivek Rai <43493515+Blazer-007@users.noreply.github.com>
Date: Thu, 31 Oct 2024 07:31:34 +0530
Subject: [PATCH] [GOBBLIN-2169]  avoid npe while overwrite if table has no
 previous snapshot (#4071)

---
 .../data/management/copy/iceberg/IcebergTable.java    | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java
index 5221007cdc..e3ec46aa1e 100644
--- a/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java
+++ b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java
@@ -234,8 +234,7 @@ protected void registerIcebergTable(TableMetadata srcMetadata, TableMetadata dst
    *
    * @param icebergPartitionFilterPredicate the predicate to filter partitions
    * @return a list of data files that match the partition filter predicate
-   * @throws TableNotFoundException if error occurred while accessing the table metadata
-   * @throws RuntimeException if error occurred while reading the manifest file
+   * @throws IOException if error occurred while accessing the table metadata or reading the manifest file
    */
   public List<DataFile> getPartitionSpecificDataFiles(Predicate<StructLike> icebergPartitionFilterPredicate)
       throws IOException {
@@ -286,7 +285,13 @@ protected void overwritePartition(List<DataFile> dataFiles, String partitionColN
     if (dataFiles.isEmpty()) {
       return;
     }
-    log.info("~{}~ SnapshotId before overwrite: {}", tableId, accessTableMetadata().currentSnapshot().snapshotId());
+    TableMetadata tableMetadata = accessTableMetadata();
+    Optional<Snapshot> currentSnapshot = Optional.ofNullable(tableMetadata.currentSnapshot());
+    if (currentSnapshot.isPresent()) {
+      log.info("~{}~ SnapshotId before overwrite: {}", tableId, currentSnapshot.get().snapshotId());
+    } else {
+      log.warn("~{}~ No current snapshot found before overwrite", tableId);
+    }
     OverwriteFiles overwriteFiles = this.table.newOverwrite();
     overwriteFiles.overwriteByRowFilter(Expressions.equal(partitionColName, partitionValue));
     dataFiles.forEach(overwriteFiles::addFile);