[AMORO-2208] Validate NULL data expiration configuration (#2209)

apache · Oct 31, 2023 · 8d50d9c · 8d50d9c
1 parent 2622dc4
commit 8d50d9c
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 6 deletions.
diff --git a/ams/server/src/main/java/com/netease/arctic/server/table/TableConfiguration.java b/ams/server/src/main/java/com/netease/arctic/server/table/TableConfiguration.java
@@ -7,6 +7,7 @@
 import com.netease.arctic.utils.CompatiblePropertyUtil;
 
 import java.util.Map;
+import java.util.Optional;
 
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class TableConfiguration {
@@ -86,7 +87,7 @@ public TableConfiguration setDeleteDanglingDeleteFilesEnabled(
   }
 
   public DataExpirationConfig getExpiringDataConfig() {
-    return expiringDataConfig;
+    return Optional.ofNullable(expiringDataConfig).orElse(new DataExpirationConfig());
   }
 
   public TableConfiguration setExpiringDataConfig(DataExpirationConfig expiringDataConfig) {

diff --git a/ams/server/src/main/java/com/netease/arctic/server/table/executor/DataExpiringExecutor.java b/ams/server/src/main/java/com/netease/arctic/server/table/executor/DataExpiringExecutor.java
@@ -187,8 +187,7 @@ private static CloseableIterable<IcebergFileEntry> fileScan(
     TableScan tableScan = table.newScan().filter(partitionFilter).includeColumnStats();
 
     CloseableIterable<FileScanTask> tasks;
-    long snapshotId =
-        null == snapshot ? ArcticServiceConstants.INVALID_SNAPSHOT_ID : snapshot.snapshotId();
+    long snapshotId = getSnapshotId(snapshot);
     if (snapshotId == ArcticServiceConstants.INVALID_SNAPSHOT_ID) {
       tasks = tableScan.planFiles();
     } else {
@@ -257,8 +256,8 @@ private static void purgeTableData(
       } catch (IOException e) {
         throw new RuntimeException(e);
       }
-      expireFiles(changeTable, changeSnapshot.snapshotId(), changeExpiredFiles, expireTimestamp);
-      expireFiles(baseTable, baseSnapshot.snapshotId(), baseExpiredFiles, expireTimestamp);
+      expireFiles(changeTable, getSnapshotId(changeSnapshot), changeExpiredFiles, expireTimestamp);
+      expireFiles(baseTable, getSnapshotId(baseSnapshot), baseExpiredFiles, expireTimestamp);
     } else {
       UnkeyedTable unkeyedTable = table.asUnkeyedTable();
       Snapshot snapshot = unkeyedTable.currentSnapshot();
@@ -279,7 +278,7 @@ private static void purgeTableData(
       } catch (IOException e) {
         throw new RuntimeException(e);
       }
-      expireFiles(unkeyedTable, snapshot.snapshotId(), expiredFiles, expireTimestamp);
+      expireFiles(unkeyedTable, getSnapshotId(snapshot), expiredFiles, expireTimestamp);
     }
   }
 
@@ -525,6 +524,10 @@ private static Literal<Long> getExpireTimestampLiteral(
     return literal;
   }
 
+  private static long getSnapshotId(Snapshot snapshot) {
+    return null == snapshot ? ArcticServiceConstants.INVALID_SNAPSHOT_ID : snapshot.snapshotId();
+  }
+
   protected static class FileEntry extends IcebergFileEntry {
 
     private final boolean isChange;