[ARCTIC-1730] Add a cache when building IcebergDeleteFile in IcebergT…

…ableFileScanHelper.scan to prevent too many generated IcebergDeleteFiles (#1731) create deleteFile by cache Co-authored-by: huyuanfeng <[email protected]>
apache · Jul 24, 2023 · 5581f4a · 5581f4a
1 parent c9bf12d
commit 5581f4a
Showing 1 changed file with 17 additions and 4 deletions.
diff --git a/...r/src/main/java/com/netease/arctic/server/optimizing/scan/IcebergTableFileScanHelper.java b/...r/src/main/java/com/netease/arctic/server/optimizing/scan/IcebergTableFileScanHelper.java
@@ -31,12 +31,14 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 public class IcebergTableFileScanHelper implements TableFileScanHelper {
@@ -62,7 +64,8 @@ public List<FileScanResult> scan() {
     long startTime = System.currentTimeMillis();
     PartitionSpec partitionSpec = table.spec();
     try (CloseableIterable<FileScanTask> filesIterable =
-             table.newScan().useSnapshot(snapshotId).planFiles()) {
+        table.newScan().useSnapshot(snapshotId).planFiles()) {
+      IcebergDeleteFileCacheGenerator deleteFilesGenerator = new IcebergDeleteFileCacheGenerator();
       for (FileScanTask task : filesIterable) {
         if (partitionFilter != null) {
           StructLike partition = task.file().partition();
@@ -73,7 +76,9 @@ public List<FileScanResult> scan() {
         }
         IcebergDataFile dataFile = createDataFile(task.file());
         List<IcebergContentFile<?>> deleteFiles =
-            task.deletes().stream().map(this::createDeleteFile).collect(Collectors.toList());
+            task.deletes().stream()
+                .map(deleteFilesGenerator::generate)
+                .collect(Collectors.toList());
         results.add(new FileScanResult(dataFile, deleteFiles));
       }
     } catch (IOException e) {
@@ -94,7 +99,15 @@ private IcebergDataFile createDataFile(DataFile dataFile) {
     return new IcebergDataFile(dataFile, sequenceNumberFetcher.sequenceNumberOf(dataFile.path().toString()));
   }
 
-  private IcebergDeleteFile createDeleteFile(DeleteFile deleteFile) {
-    return new IcebergDeleteFile(deleteFile, sequenceNumberFetcher.sequenceNumberOf(deleteFile.path().toString()));
+  private class IcebergDeleteFileCacheGenerator {
+    private final Map<DeleteFile, IcebergDeleteFile> deleteFilesCache = Maps.newHashMap();
+
+    IcebergDeleteFile generate(DeleteFile deleteFile) {
+      return deleteFilesCache.computeIfAbsent(
+          deleteFile,
+          file ->
+              new IcebergDeleteFile(
+                  file, sequenceNumberFetcher.sequenceNumberOf(file.path().toString())));
+    }
   }
 }