This is an automated email from the ASF dual-hosted git repository.
findepi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 339be73d22 Core: Reduce DeleteFileIndex memory for file-scoped
position deletes (#18359)
339be73d22 is described below
commit 339be73d22dbb444b1ea3ad37ea5a9e38d08532d
Author: Raunaq Morarka <[email protected]>
AuthorDate: Thu Oct 29 01:01:41 2026 +0530
Core: Reduce DeleteFileIndex memory for file-scoped position deletes
(#18359)
* Core: Copy file-scoped position deletes without stats in DeleteFileIndex
A position delete whose file_path bounds name a single data file is
matched by location, so the copy records the data file in
referencedDataFile and drops column stats.
JOL retained bytes per delete, 100k v2 position deletes, 150-char paths:
| data files | before | after |
|------------|----------|---------|
| 100k (1:1) | 2122.6 B | 810.6 B |
| 37k (2.7:1)| 1908.1 B | 717.0 B |
* Core: Share referenced data file locations in DeleteFileIndex
Position deletes of the same data file hold one location String.
JOL retained bytes per delete, 100k v2 position deletes, 150-char paths,
2.7 deletes per data file: 717.0 B before, 596.1 B after.
---
.../src/main/java/org/apache/iceberg/BaseFile.java | 10 +-
.../java/org/apache/iceberg/DeleteFileIndex.java | 37 ++++++--
.../java/org/apache/iceberg/GenericDeleteFile.java | 8 ++
.../apache/iceberg/DeleteFileIndexTestBase.java | 45 +--------
.../apache/iceberg/TestLocalDeleteFileIndex.java | 104 +++++++++++++++++++++
5 files changed, 153 insertions(+), 51 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/BaseFile.java
b/core/src/main/java/org/apache/iceberg/BaseFile.java
index 8d94fd89be..cc6184e7aa 100644
--- a/core/src/main/java/org/apache/iceberg/BaseFile.java
+++ b/core/src/main/java/org/apache/iceberg/BaseFile.java
@@ -202,6 +202,14 @@ 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);
+ }
+
+ BaseFile(
+ BaseFile<F> toCopy,
+ boolean copyStats,
+ Set<Integer> requestedColumnIds,
+ String referencedDataFile) {
super(toCopy);
this.fileOrdinal = toCopy.fileOrdinal;
this.manifestLocation = toCopy.manifestLocation;
@@ -246,7 +254,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;
}
diff --git a/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
b/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
index b15114b1f0..8bf029985e 100644
--- a/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
+++ b/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
@@ -69,6 +69,8 @@ import org.apache.iceberg.util.Tasks;
*/
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 @@ class DeleteFileIndex {
// 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 @@ class DeleteFileIndex {
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,30 @@ class DeleteFileIndex {
return files;
}
+ 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()));
+ }
+
+ if (ContentFileUtil.isDV(file)) {
+ // a DV already records its referenced data file
+ return file.copyWithoutStats();
+ }
+
+ String referencedDataFile =
ContentFileUtil.referencedDataFileLocation(file);
+ if (referencedDataFile != null && file instanceof GenericDeleteFile
genericFile) {
+ // a file-scoped position delete is matched by location and needs no
stats
+ String location =
referencedDataFiles.computeIfAbsent(referencedDataFile, key -> key);
+ return genericFile.copyAsFileScoped(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();
diff --git a/core/src/main/java/org/apache/iceberg/GenericDeleteFile.java
b/core/src/main/java/org/apache/iceberg/GenericDeleteFile.java
index d3aafc2922..2a0995eaaf 100644
--- a/core/src/main/java/org/apache/iceberg/GenericDeleteFile.java
+++ b/core/src/main/java/org/apache/iceberg/GenericDeleteFile.java
@@ -90,6 +90,10 @@ class GenericDeleteFile extends BaseFile<DeleteFile>
implements DeleteFile {
super(toCopy, copyStats, requestedColumnIds);
}
+ private GenericDeleteFile(GenericDeleteFile toCopy, String
referencedDataFile) {
+ super(toCopy, false /* drop stats */, null, referencedDataFile);
+ }
+
/** Constructor for Java serialization. */
GenericDeleteFile() {}
@@ -98,6 +102,10 @@ class GenericDeleteFile extends BaseFile<DeleteFile>
implements DeleteFile {
return new GenericDeleteFile(this, false /* drop stats */, null);
}
+ DeleteFile copyAsFileScoped(String referencedDataFile) {
+ return new GenericDeleteFile(this, referencedDataFile);
+ }
+
@Override
public DeleteFile copyWithStats(Set<Integer> requestedColumnIds) {
return new GenericDeleteFile(this, true, requestedColumnIds);
diff --git a/core/src/test/java/org/apache/iceberg/DeleteFileIndexTestBase.java
b/core/src/test/java/org/apache/iceberg/DeleteFileIndexTestBase.java
index 08849fe351..9a66c98cc6 100644
--- a/core/src/test/java/org/apache/iceberg/DeleteFileIndexTestBase.java
+++ b/core/src/test/java/org/apache/iceberg/DeleteFileIndexTestBase.java
@@ -92,7 +92,7 @@ public abstract class DeleteFileIndexTestBase<
.build();
}
- private static DeleteFile posDeletesWithMetrics(PartitionSpec spec, String
path) {
+ 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 +109,12 @@ public abstract class DeleteFileIndexTestBase<
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();
}
@@ -603,45 +603,6 @@ public abstract class DeleteFileIndexTestBase<
.isEqualTo(Sets.newHashSet(FILE_A_EQ_1.location(),
fileADeletes().location()));
}
- @TestTemplate
- public void testPositionDeleteDiscardMetrics() {
- 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();
-
- 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.file().location())
- .as("Should have the correct data file path")
- .isEqualTo(FILE_A.location());
- 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))));
- }
-
@TestTemplate
public void testEqualityDeleteDiscardMetrics() {
Table table = TestTables.create(tableDir, "partitioned", SCHEMA, SPEC,
formatVersion);
diff --git
a/core/src/test/java/org/apache/iceberg/TestLocalDeleteFileIndex.java
b/core/src/test/java/org/apache/iceberg/TestLocalDeleteFileIndex.java
index d01e6253d4..adee7f2465 100644
--- a/core/src/test/java/org/apache/iceberg/TestLocalDeleteFileIndex.java
+++ b/core/src/test/java/org/apache/iceberg/TestLocalDeleteFileIndex.java
@@ -18,6 +18,17 @@
*/
package org.apache.iceberg;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.TestTemplate;
+
public class TestLocalDeleteFileIndex
extends DeleteFileIndexTestBase<TableScan, FileScanTask, CombinedScanTask>
{
@@ -25,4 +36,97 @@ public class TestLocalDeleteFileIndex
protected TableScan newScan(Table table) {
return table.newScan();
}
+
+ @TestTemplate
+ void fileScopedPositionDeleteDiscardMetrics() {
+ Table table =
+ TestTables.create(tableDir, "partitioned", SCHEMA,
PartitionSpec.unpartitioned(), 2);
+
+ table.newAppend().appendFile(FILE_A).commit();
+
+ // add a delete file whose file_path bounds both equal FILE_A
+ table
+ .newRowDelta()
+ .addDeletes(posDeletesWithMetrics(table.spec(), FILE_A.location(),
FILE_A.location()))
+ .commit();
+
+ // upgrade to the tested format version with the position deletes in place
+ table
+ .updateProperties()
+ .set(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))
+ .commit();
+
+ FileScanTask task = Iterables.getOnlyElement(table.newScan().planFiles());
+ DeleteFile deleteFile = Iterables.getOnlyElement(task.deletes());
+ assertThat(deleteFile.referencedDataFile()).isEqualTo(FILE_A.location());
+ assertThat(deleteFile.lowerBounds()).isNull();
+ assertThat(deleteFile.upperBounds()).isNull();
+ }
+
+ @TestTemplate
+ void positionDeleteWithReferencedDataFileDiscardMetrics() {
+ 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();
+
+ // upgrade to the tested format version with the position deletes in place
+ table
+ .updateProperties()
+ .set(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))
+ .commit();
+
+ FileScanTask task = Iterables.getOnlyElement(table.newScan().planFiles());
+ DeleteFile deleteFile = Iterables.getOnlyElement(task.deletes());
+ assertThat(deleteFile.referencedDataFile()).isEqualTo(FILE_A.location());
+ }
+
+ @TestTemplate
+ void multiFilePositionDeleteDiscardMetrics() {
+ 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
+ table
+ .newRowDelta()
+ .addDeletes(posDeletesWithMetrics(table.spec(), FILE_A.location(),
FILE_B.location()))
+ .commit();
+
+ // upgrade to the tested format version with the position deletes in place
+ table
+ .updateProperties()
+ .set(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))
+ .commit();
+
+ List<FileScanTask> tasks = Lists.newArrayList(table.newScan().planFiles());
+ assertThat(tasks).hasSize(2);
+
+ // verify scanned delete files only keep the file_path bounds
+ int fieldId = MetadataColumns.DELETE_FILE_PATH.fieldId();
+ for (FileScanTask task : tasks) {
+ DeleteFile deleteFile = Iterables.getOnlyElement(task.deletes());
+ assertThat(deleteFile.referencedDataFile()).isNull();
+ 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))));
+ }
+ }
}
