Core: Reduce DeleteFileIndex memory for file-scoped position deletes

core/src/main/java/org/apache/iceberg/BaseFile.java

@@ -202,6 +202,23 @@ abstract class BaseFile<F> extends SupportsIndexProjection
    *     column stat is kept.
    */
   BaseFile(BaseFile<F> toCopy, boolean copyStats, Set<Integer> requestedColumnIds) {
+    this(toCopy, copyStats, requestedColumnIds, toCopy.referencedDataFile);
+  }
+
+  /**
+   * Copy constructor that overrides the referenced data file.
+   *
+   * @param toCopy a file to copy.
+   * @param copyStats whether to copy all fields or to drop column-level stats
+   * @param requestedColumnIds column ids for which to keep stats. If <code>null</code> then every
+   *     column stat is kept.
+   * @param referencedDataFile location of the data file that all deletes reference
+   */
+  BaseFile(
+      BaseFile<F> toCopy,
+      boolean copyStats,
+      Set<Integer> requestedColumnIds,
+      String referencedDataFile) {
     super(toCopy);
     this.fileOrdinal = toCopy.fileOrdinal;
     this.manifestLocation = toCopy.manifestLocation;
@@ -246,7 +263,7 @@ abstract class BaseFile<F> extends SupportsIndexProjection
     this.dataSequenceNumber = toCopy.dataSequenceNumber;
     this.fileSequenceNumber = toCopy.fileSequenceNumber;
     this.firstRowId = toCopy.firstRowId;
-    this.referencedDataFile = toCopy.referencedDataFile;
+    this.referencedDataFile = referencedDataFile;
     this.contentOffset = toCopy.contentOffset;
     this.contentSizeInBytes = toCopy.contentSizeInBytes;
   }

core/src/main/java/org/apache/iceberg/DeleteFileIndex.java

@@ -69,6 +69,8 @@
  */
 class DeleteFileIndex {
   private static final DeleteFile[] EMPTY_DELETES = new DeleteFile[0];
+  private static final Set<Integer> DELETE_FILE_PATH_COLUMN =
+      Set.of(MetadataColumns.DELETE_FILE_PATH.fieldId());
 
   private final EqualityDeletes globalDeletes;
   private final PartitionMap<EqualityDeletes> eqDeletesByPartition;
@@ -493,6 +495,8 @@ private Collection<DeleteFile> loadDeleteFiles() {
       // read all of the matching delete manifests in parallel and accumulate the matching files in
       // a queue
       Queue<DeleteFile> files = new ConcurrentLinkedQueue<>();
+      // share one location instance across deletes that reference the same data file
+      Map<String, String> referencedDataFiles = Maps.newConcurrentMap();
       Tasks.foreach(deleteManifestReaders())
           .stopOnFailure()
           .throwFailureWhenFinished()
@@ -502,14 +506,7 @@ private Collection<DeleteFile> loadDeleteFiles() {
                 try (CloseableIterable<ManifestEntry<DeleteFile>> reader = deleteFile) {
                   for (ManifestEntry<DeleteFile> entry : reader) {
                     if (entry.dataSequenceNumber() > minSequenceNumber) {
-                      DeleteFile file = entry.file();
-                      // keep minimum stats to avoid memory pressure
-                      Set<Integer> columns =
-                          file.content() == FileContent.POSITION_DELETES
-                              ? Set.of(MetadataColumns.DELETE_FILE_PATH.fieldId())
-                              : Set.copyOf(file.equalityFieldIds());
-                      // copy with stats for better filtering against data file stats
-                      files.add(ContentFileUtil.copy(file, true, columns));
+                      files.add(copyWithMinStats(entry.file(), referencedDataFiles));
                     }
                   }
                 } catch (IOException e) {
@@ -519,6 +516,26 @@ private Collection<DeleteFile> loadDeleteFiles() {
       return files;
     }
 
+    /** Copies a delete file with the minimum stats needed for indexing to limit memory use. */
+    private static DeleteFile copyWithMinStats(
+        DeleteFile file, Map<String, String> referencedDataFiles) {
+      if (file.content() == FileContent.EQUALITY_DELETES) {
+        // copy with stats for better filtering against data file stats
+        return file.copyWithStats(Set.copyOf(file.equalityFieldIds()));
+      }
+
+      String referencedDataFile = ContentFileUtil.referencedDataFileLocation(file);
+      if (referencedDataFile != null && file instanceof GenericDeleteFile) {
+        // a file-scoped position delete is matched by location and needs no stats
+        String location = referencedDataFiles.computeIfAbsent(referencedDataFile, key -> key);
+        return ((GenericDeleteFile) file).copyWithoutStats(location);
+      }
+
+      // keep the file_path bounds for position deletes that span multiple data files and for
+      // other DeleteFile implementations
+      return file.copyWithStats(DELETE_FILE_PATH_COLUMN);
+    }
+
     private Collection<Schema> schemas() {
       if (schemasById != null) {
         return schemasById.values();

core/src/main/java/org/apache/iceberg/GenericDeleteFile.java

@@ -90,6 +90,16 @@ private GenericDeleteFile(
     super(toCopy, copyStats, requestedColumnIds);
   }
 
+  /**
+   * Copy constructor that drops column-level stats and sets the referenced data file.
+   *
+   * @param toCopy a generic delete file to copy.
+   * @param referencedDataFile location of the data file that all deletes reference
+   */
+  private GenericDeleteFile(GenericDeleteFile toCopy, String referencedDataFile) {
+    super(toCopy, false /* drop stats */, null, referencedDataFile);
+  }
+
   /** Constructor for Java serialization. */
   GenericDeleteFile() {}
 
@@ -98,6 +108,19 @@ public DeleteFile copyWithoutStats() {
     return new GenericDeleteFile(this, false /* drop stats */, null);
   }
 
+  /**
+   * Copies this file without column-level stats and records the referenced data file.
+   *
+   * <p>A file-scoped position delete is matched to its data file by location, so the file_path
+   * bounds that identify the data file are not needed once the location is recorded.
+   *
+   * @param referencedDataFile location of the data file that all deletes reference
+   * @return a copy without stats that references the given data file
+   */
+  DeleteFile copyWithoutStats(String referencedDataFile) {
+    return new GenericDeleteFile(this, referencedDataFile);
+  }
+
   @Override
   public DeleteFile copyWithStats(Set<Integer> requestedColumnIds) {
     return new GenericDeleteFile(this, true, requestedColumnIds);

core/src/test/java/org/apache/iceberg/DeleteFileIndexTestBase.java

@@ -92,7 +92,8 @@ private static DeleteFile partitionedPosDeletes(PartitionSpec spec, StructLike p
         .build();
   }
 
-  private static DeleteFile posDeletesWithMetrics(PartitionSpec spec, String path) {
+  private static DeleteFile posDeletesWithMetrics(
+      PartitionSpec spec, String lowerPath, String upperPath) {
     return FileMetadata.deleteFileBuilder(spec)
         .ofPositionDeletes()
         .withPath(UUID.randomUUID() + "/path/to/data-partitioned-pos-deletes.parquet")
@@ -109,12 +110,12 @@ private static DeleteFile posDeletesWithMetrics(PartitionSpec spec, String path)
                     1,
                     ByteBuffer.wrap(new byte[10]),
                     MetadataColumns.DELETE_FILE_PATH.fieldId(),
-                    ByteBuffer.wrap(path.getBytes(StandardCharsets.UTF_8))),
+                    ByteBuffer.wrap(lowerPath.getBytes(StandardCharsets.UTF_8))),
                 ImmutableMap.of(
                     1,
                     ByteBuffer.wrap(new byte[10]),
                     MetadataColumns.DELETE_FILE_PATH.fieldId(),
-                    ByteBuffer.wrap(path.getBytes(StandardCharsets.UTF_8)))))
+                    ByteBuffer.wrap(upperPath.getBytes(StandardCharsets.UTF_8)))))
         .build();
   }
 
@@ -604,17 +605,20 @@ public void testPartitionedTableWithExistingDeleteFile() {
   }
 
   @TestTemplate
-  public void testPositionDeleteDiscardMetrics() {
+  public void testFileScopedPositionDeleteDiscardMetrics() {
     assumeThat(formatVersion).isEqualTo(2);
 
     Table table =
         TestTables.create(tableDir, "partitioned", SCHEMA, PartitionSpec.unpartitioned(), 2);
 
     table.newAppend().appendFile(FILE_A).commit();
 
-    // add a delete file
-    DeleteFile posDeletesWithMetrics = posDeletesWithMetrics(table.spec(), FILE_A.location());
-    table.newRowDelta().addDeletes(posDeletesWithMetrics).commit();
+    // add two delete files whose file_path bounds both equal FILE_A
+    table
+        .newRowDelta()
+        .addDeletes(posDeletesWithMetrics(table.spec(), FILE_A.location(), FILE_A.location()))
+        .addDeletes(posDeletesWithMetrics(table.spec(), FILE_A.location(), FILE_A.location()))
+        .commit();
 
     List<T> tasks = Lists.newArrayList(newScan(table).planFiles().iterator());
     assertThat(tasks).as("Should have one task").hasSize(1);
@@ -623,23 +627,101 @@ public void testPositionDeleteDiscardMetrics() {
     assertThat(task.file().location())
         .as("Should have the correct data file path")
         .isEqualTo(FILE_A.location());
+    assertThat(task.deletes()).as("Should have two associated delete files").hasSize(2);
+
+    // verify scanned delete files keep the referenced data file instead of metrics
+    for (DeleteFile deleteFile : task.deletes()) {
+      assertThat(deleteFile.referencedDataFile()).isEqualTo(FILE_A.location());
+      assertThat(ContentFileUtil.isFileScoped(deleteFile)).isTrue();
+      assertThat(deleteFile.dataSequenceNumber()).isEqualTo(2L);
+      assertThat(deleteFile.fileSequenceNumber()).isEqualTo(2L);
+      assertThat(deleteFile.recordCount()).isEqualTo(1L);
+      assertThat(deleteFile.columnSizes()).isNull();
+      assertThat(deleteFile.valueCounts()).isNull();
+      assertThat(deleteFile.nullValueCounts()).isNull();
+      assertThat(deleteFile.nanValueCounts()).isNull();
+      assertThat(deleteFile.lowerBounds()).isNull();
+      assertThat(deleteFile.upperBounds()).isNull();
+    }
+
+    // verify delete files that reference the same data file share one location instance
+    assertThat(task.deletes().get(0).referencedDataFile())
+        .isSameAs(task.deletes().get(1).referencedDataFile());
+  }
+
+  @TestTemplate
+  public void testPositionDeleteWithReferencedDataFileDiscardMetrics() {
+    assumeThat(formatVersion).isEqualTo(2);
+
+    Table table =
+        TestTables.create(tableDir, "partitioned", SCHEMA, PartitionSpec.unpartitioned(), 2);
+
+    table.newAppend().appendFile(FILE_A).commit();
+
+    // add a delete file with referenced_data_file set and no file_path bounds
+    DeleteFile posDeletes =
+        FileMetadata.deleteFileBuilder(table.spec())
+            .ofPositionDeletes()
+            .withPath(UUID.randomUUID() + "/path/to/data-unpartitioned-pos-deletes.parquet")
+            .withFileSizeInBytes(10)
+            .withRecordCount(1)
+            .withReferencedDataFile(FILE_A.location())
+            .build();
+    table.newRowDelta().addDeletes(posDeletes).commit();
+
+    List<T> tasks = Lists.newArrayList(newScan(table).planFiles().iterator());
+    assertThat(tasks).as("Should have one task").hasSize(1);
+
+    FileScanTask task = (FileScanTask) tasks.get(0);
     assertThat(task.deletes()).as("Should have one associated delete file").hasSize(1);
 
-    // verify scanned delete file only contains metrics for 'file_path' field
-    int fieldId = MetadataColumns.DELETE_FILE_PATH.fieldId();
     DeleteFile deleteFile = task.deletes().get(0);
-    assertThat(deleteFile.columnSizes()).containsExactly(Map.entry(fieldId, 2L));
-    assertThat(deleteFile.valueCounts()).containsExactly(Map.entry(fieldId, 2L));
-    assertThat(deleteFile.nullValueCounts()).containsExactly(Map.entry(fieldId, 2L));
-    assertThat(deleteFile.nanValueCounts()).containsExactly(Map.entry(fieldId, 2L));
-    assertThat(deleteFile.lowerBounds())
-        .containsExactly(
-            Map.entry(
-                fieldId, ByteBuffer.wrap(FILE_A.location().getBytes(StandardCharsets.UTF_8))));
-    assertThat(deleteFile.upperBounds())
-        .containsExactly(
-            Map.entry(
-                fieldId, ByteBuffer.wrap(FILE_A.location().getBytes(StandardCharsets.UTF_8))));
+    assertThat(deleteFile.location()).isEqualTo(posDeletes.location());
+    assertThat(deleteFile.referencedDataFile()).isEqualTo(FILE_A.location());
+    assertThat(ContentFileUtil.isFileScoped(deleteFile)).isTrue();
+    assertThat(deleteFile.lowerBounds()).isNull();
+    assertThat(deleteFile.upperBounds()).isNull();
+  }
+
+  @TestTemplate
+  public void testMultiFilePositionDeleteDiscardMetrics() {
+    assumeThat(formatVersion).isEqualTo(2);
+
+    Table table =
+        TestTables.create(tableDir, "partitioned", SCHEMA, PartitionSpec.unpartitioned(), 2);
+
+    table.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
+
+    // add a delete file whose file_path bounds span FILE_A and FILE_B
+    DeleteFile posDeletesWithMetrics =
+        posDeletesWithMetrics(table.spec(), FILE_A.location(), FILE_B.location());
+    table.newRowDelta().addDeletes(posDeletesWithMetrics).commit();
+
+    List<T> tasks = Lists.newArrayList(newScan(table).planFiles().iterator());
+    assertThat(tasks).as("Should have two tasks").hasSize(2);
+
+    for (T scanTask : tasks) {
+      FileScanTask task = (FileScanTask) scanTask;
+      assertThat(task.deletes()).as("Should have one associated delete file").hasSize(1);
+
+      // verify scanned delete file only contains metrics for 'file_path' field
+      int fieldId = MetadataColumns.DELETE_FILE_PATH.fieldId();
+      DeleteFile deleteFile = task.deletes().get(0);
+      assertThat(deleteFile.referencedDataFile()).isNull();
+      assertThat(ContentFileUtil.isFileScoped(deleteFile)).isFalse();
+      assertThat(deleteFile.columnSizes()).containsExactly(Map.entry(fieldId, 2L));
+      assertThat(deleteFile.valueCounts()).containsExactly(Map.entry(fieldId, 2L));
+      assertThat(deleteFile.nullValueCounts()).containsExactly(Map.entry(fieldId, 2L));
+      assertThat(deleteFile.nanValueCounts()).containsExactly(Map.entry(fieldId, 2L));
+      assertThat(deleteFile.lowerBounds())
+          .containsExactly(
+              Map.entry(
+                  fieldId, ByteBuffer.wrap(FILE_A.location().getBytes(StandardCharsets.UTF_8))));
+      assertThat(deleteFile.upperBounds())
+          .containsExactly(
+              Map.entry(
+                  fieldId, ByteBuffer.wrap(FILE_B.location().getBytes(StandardCharsets.UTF_8))));
+    }
   }
 
   @TestTemplate