Repository: apache/iceberg (language: python)

Issue:
Improve Position Deletes in V3
### Proposed Change

This proposal aims to enhance the handling of position deletes in Iceberg. It builds on lessons learned from deploying the current approach at scale and addresses all unresolved questions from past community discussions and proposals. This effort primarily targets the V3 spec but some parts can be completed independently.

Shortcomings of position deletes we observe today:
- Choosing between fewer delete files on disk (partition granularity) or targeted deletes (file granularity).
- Dependence on external maintenance for consistent write and read performance.
- Writing and reading overhead as in-memory and on-disk representations differ.


### Proposal document

https://docs.google.com/document/d/18Bqhr-vnzFfQk1S4AgRISkA_5_m5m32Nnc2Cw0zn2XM

### Specifications

- [X] Table
- [ ] View
- [ ] REST
- [X] Puffin
- [ ] Encryption
- [ ] Other


Discussion / hints:

Fix patch:
diff --git a/core/src/main/java/org/apache/iceberg/deletes/BitmapPositionDeleteIndex.java b/core/src/main/java/org/apache/iceberg/deletes/BitmapPositionDeleteIndex.java
index cfb163e8379c..376b391d9c24 100644
--- a/core/src/main/java/org/apache/iceberg/deletes/BitmapPositionDeleteIndex.java
+++ b/core/src/main/java/org/apache/iceberg/deletes/BitmapPositionDeleteIndex.java
@@ -18,13 +18,23 @@
  */
 package org.apache.iceberg.deletes;
 
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.Collection;
 import java.util.List;
 import java.util.function.LongConsumer;
+import java.util.zip.CRC32;
 import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 
 class BitmapPositionDeleteIndex implements PositionDeleteIndex {
+  private static final int LENGTH_SIZE_BYTES = 4;
+  private static final int MAGIC_NUMBER_SIZE_BYTES = 4;
+  private static final int CRC_SIZE_BYTES = 4;
+  private static final int BITMAP_DATA_OFFSET = 4;
+  private static final int MAGIC_NUMBER = 1681511377;
+
   private final RoaringPositionBitmap bitmap;
   private final List<DeleteFile> deleteFiles;
 
@@ -43,6 +53,11 @@ class BitmapPositionDeleteIndex implements PositionDeleteIndex {
     this.deleteFiles = deleteFile != null ? Lists.newArrayList(deleteFile) : Lists.newArrayList();
   }
 
+  BitmapPositionDeleteIndex(RoaringPositionBitmap bitmap, DeleteFile deleteFile) {
+    this.bitmap = bitmap;
+    this.deleteFiles = deleteFile != null ? Lists.newArrayList(deleteFile) : Lists.newArrayList();
+  }
+
   void merge(BitmapPositionDeleteIndex that) {
     bitmap.setAll(that.bitmap);
     deleteFiles.addAll(that.deleteFiles);
@@ -92,4 +107,113 @@ public Collection<DeleteFile> deleteFiles() {
   public long cardinality() {
     return bitmap.cardinality();
   }
+
+  /**
+   * Serializes the index using the following format:
+   *
+   * <ul>
+   *   <li>The length of the magic bytes and bitmap stored as 4 bytes (big-endian).
+   *   <li>A 4-byte {@link #MAGIC_NUMBER} (little-endian).
+   *   <li>The bitmap serialized using the portable Roaring spec (little-endian).
+   *   <li>A CRC-32 checksum of the magic bytes and bitmap as 4-bytes (big-endian).
+   * </ul>
+   *
+   * Note that the length and the checksum are computed for the bitmap data, which includes the
+   * magic bytes and bitmap for compatibility with Delta.
+   */
+  @Override
+  public ByteBuffer serialize() {
+    bitmap.runLengthEncode(); // run-length encode the bitmap before serializing
+    int bitmapDataLength = computeBitmapDataLength(bitmap); // magic bytes + bitmap
+    byte[] bytes = new byte[LENGTH_SIZE_BYTES + bitmapDataLength + CRC_SIZE_BYTES];
+    ByteBuffer buffer = ByteBuffer.wrap(bytes);
+    buffer.putInt(bitmapDataLength);
+    serializeBitmapData(bytes, bitmapDataLength, bitmap);
+    int crcOffset = LENGTH_SIZE_BYTES + bitmapDataLength;
+    int crc = computeChecksum(bytes, bitmapDataLength);
+    buffer.putInt(crcOffset, crc);
+    buffer.rewind();
+    return buffer;
+  }
+
+  /**
+   * Deserializes the index from bytes, assuming the format described in {@link #serialize()}.
+   *
+   * @param bytes an array containing the serialized index
+   * @param deleteFile the DV file
+   * @return the deserialized index
+   */
+  public static PositionDeleteIndex deserialize(byte[] bytes, DeleteFile deleteFile) {
+    ByteBuffer buffer = ByteBuffer.wrap(bytes);
+    int bitmapDataLength = readBitmapDataLength(buffer, deleteFile);
+    RoaringPositionBitmap bitmap = deserializeBitmap(bytes, bitmapDataLength, deleteFile);
+    int crc = computeChecksum(bytes, bitmapDataLength);
+    int crcOffset = LENGTH_SIZE_BYTES + bitmapDataLength;
+    int expectedCrc = buffer.getInt(crcOffset);
+    Preconditions.checkArgument(crc == expectedCrc, "Invalid CRC");
+    return new BitmapPositionDeleteIndex(bitmap, deleteFile);
+  }
+
+  // computes and validates the length of the bitmap data (magic bytes + bitmap)
+  private static int computeBitmapDataLength(RoaringPositionBitmap bitmap) {
+    long length = MAGIC_NUMBER_SIZE_BYTES + bitmap.serializedSizeInBytes();
+    long bufferSize = LENGTH_SIZE_BYTES + length + CRC_SIZE_BYTES;
+    Preconditions.checkState(bufferSize <= Integer.MAX_VALUE, "Can't serialize index > 2GB");
+    return (int) length;
+  }
+
+  // serializes the bitmap data (magic bytes + bitmap) using the little-endian byte order
+  private static void serializeBitmapData(
+      byte[] bytes, int bitmapDataLength, RoaringPositionBitmap bitmap) {
+    ByteBuffer bitmapData = pointToBitmapData(bytes, bitmapDataLength);
+    bitmapData.putInt(MAGIC_NUMBER);
+    bitmap.serialize(bitmapData);
+  }
+
+  // points to the bitmap data in the blob
+  private static ByteBuffer pointToBitmapData(byte[] bytes, int bitmapDataLength) {
+    ByteBuffer bitmapData = ByteBuffer.wrap(bytes, BITMAP_DATA_OFFSET, bitmapDataLength);
+    bitmapData.order(ByteOrder.LITTLE_ENDIAN);
+    return bitmapData;
+  }
+
+  // checks the blob size is equal to the bitmap data length + extra bytes for length and CRC
+  private static int readBitmapDataLength(ByteBuffer buffer, DeleteFile deleteFile) {
+    int length = buffer.getInt();
+    long expectedLength = deleteFile.contentSizeInBytes() - LENGTH_SIZE_BYTES - CRC_SIZE_BYTES;
+    Preconditions.checkArgument(
+        length == expectedLength,
+        "Invalid bitmap data length: %s, expected %s",
+        length,
+        expectedLength);
+    return length;
+  }
+
+  // validates magic bytes and deserializes the bitmap
+  private static RoaringPositionBitmap deserializeBitmap(
+      byte[] bytes, int bitmapDataLength, DeleteFile deleteFile) {
+    ByteBuffer bitmapData = pointToBitmapData(bytes, bitmapDataLength);
+    int magicNumber = bitmapData.getInt();
+    Preconditions.checkArgument(
+        magicNumber == MAGIC_NUMBER,
+        "Invalid magic number: %s, expected %s",
+        magicNumber,
+        MAGIC_NUMBER);
+    RoaringPositionBitmap bitmap = RoaringPositionBitmap.deserialize(bitmapData);
+    long cardinality = bitmap.cardinality();
+    long expectedCardinality = deleteFile.recordCount();
+    Preconditions.checkArgument(
+        cardinality == expectedCardinality,
+        "Invalid cardinality: %s, expected %s",
+        cardinality,
+        expectedCardinality);
+    return bitmap;
+  }
+
+  // generates a 32-bit unsigned checksum for the magic bytes and serialized bitmap
+  private static int computeChecksum(byte[] bytes, int bitmapDataLength) {
+    CRC32 crc = new CRC32();
+    crc.update(bytes, BITMAP_DATA_OFFSET, bitmapDataLength);
+    return (int) crc.getValue();
+  }
 }
diff --git a/core/src/main/java/org/apache/iceberg/deletes/PositionDeleteIndex.java b/core/src/main/java/org/apache/iceberg/deletes/PositionDeleteIndex.java
index 8ccfc03d1a26..6f97b3a6ac87 100644
--- a/core/src/main/java/org/apache/iceberg/deletes/PositionDeleteIndex.java
+++ b/core/src/main/java/org/apache/iceberg/deletes/PositionDeleteIndex.java
@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.deletes;
 
+import java.nio.ByteBuffer;
 import java.util.Collection;
 import java.util.function.LongConsumer;
 import org.apache.iceberg.DeleteFile;
@@ -92,6 +93,26 @@ default long cardinality() {
     throw new UnsupportedOperationException(getClass().getName() + " does not support cardinality");
   }
 
+  /**
+   * Serializes this index.
+   *
+   * @return a buffer containing the serialized index
+   */
+  default ByteBuffer serialize() {
+    throw new UnsupportedOperationException(getClass().getName() + " does not support serialize");
+  }
+
+  /**
+   * Deserializes a position delete index.
+   *
+   * @param bytes an array containing the serialized index
+   * @param deleteFile the delete file that the index is created for
+   * @return the deserialized index
+   */
+  static PositionDeleteIndex deserialize(byte[] bytes, DeleteFile deleteFile) {
+    return BitmapPositionDeleteIndex.deserialize(bytes, deleteFile);
+  }
+
   /** Returns an empty immutable position delete index. */
   static PositionDeleteIndex empty() {
     return EmptyPositionDeleteIndex.get();


Test patch:
diff --git a/core/src/test/java/org/apache/iceberg/deletes/TestBitmapPositionDeleteIndex.java b/core/src/test/java/org/apache/iceberg/deletes/TestBitmapPositionDeleteIndex.java
index c8fc723deb9e..76b294f80611 100644
--- a/core/src/test/java/org/apache/iceberg/deletes/TestBitmapPositionDeleteIndex.java
+++ b/core/src/test/java/org/apache/iceberg/deletes/TestBitmapPositionDeleteIndex.java
@@ -20,12 +20,21 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.io.IOException;
+import java.net.URL;
+import java.nio.ByteBuffer;
 import java.util.List;
+import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.io.Resources;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 
 public class TestBitmapPositionDeleteIndex {
 
+  private static final long BITMAP_OFFSET = 0xFFFFFFFFL + 1L;
+  private static final long CONTAINER_OFFSET = Character.MAX_VALUE + 1L;
+
   @Test
   public void testForEach() {
     long pos1 = 10L; // Container 0 (high bits = 0)
@@ -105,6 +114,102 @@ public void testMergeBitmapIndexWithEmpty() {
     assertThat(positions).containsExactly(pos1, pos2, pos3, pos4);
   }
 
+  @Test
+  public void testEmptyIndexSerialization() throws Exception {
+    PositionDeleteIndex index = new BitmapPositionDeleteIndex();
+    validate(index, "empty-position-index.bin");
+  }
+
+  @Test
+  public void testSmallAlternatingValuesIndexSerialization() throws Exception {
+    PositionDeleteIndex index = new BitmapPositionDeleteIndex();
+    index.delete(1L);
+    index.delete(3L);
+    index.delete(5L);
+    index.delete(7L);
+    index.delete(9L);
+    validate(index, "small-alternating-values-position-index.bin");
+  }
+
+  @Test
+  public void testSmallAndLargeValuesIndexSerialization() throws Exception {
+    PositionDeleteIndex index = new BitmapPositionDeleteIndex();
+    index.delete(100L);
+    index.delete(101L);
+    index.delete(Integer.MAX_VALUE + 100L);
+    index.delete(Integer.MAX_VALUE + 101L);
+    validate(index, "small-and-large-values-position-index.bin");
+  }
+
+  @Test
+  public void testAllContainerTypesIndexSerialization() throws Exception {
+    PositionDeleteIndex index = new BitmapPositionDeleteIndex();
+
+    // bitmap 0, container 0 (array)
+    index.delete(position(0 /* bitmap */, 0 /* container */, 5L));
+    index.delete(position(0 /* bitmap */, 0 /* container */, 7L));
+
+    // bitmap 0, container 1 (array that can be compressed)
+    index.delete(
+        position(0 /* bitmap */, 1 /* container */, 1L),
+        position(0 /* bitmap */, 1 /* container */, 1000L));
+
+    // bitmap 1, container 2 (bitset)
+    index.delete(
+        position(0 /* bitmap */, 2 /* container */, 1L),
+        position(0 /* bitmap */, 2 /* container */, CONTAINER_OFFSET - 1L));
+
+    // bitmap 1, container 0 (array)
+    index.delete(position(1 /* bitmap */, 0 /* container */, 10L));
+    index.delete(position(1 /* bitmap */, 0 /* container */, 20L));
+
+    // bitmap 1, container 1 (array that can be compressed)
+    index.delete(
+        position(1 /* bitmap */, 1 /* container */, 10L),
+        position(1 /* bitmap */, 1 /* container */, 500L));
+
+    // bitmap 1, container 2 (bitset)
+    index.delete(
+        position(1 /* bitmap */, 2 /* container */, 1L),
+        position(1 /* bitmap */, 2 /* container */, CONTAINER_OFFSET - 1));
+
+    validate(index, "all-container-types-position-index.bin");
+  }
+
+  private static void validate(PositionDeleteIndex index, String goldenFile) throws Exception {
+    ByteBuffer buffer = index.serialize();
+    byte[] bytes = buffer.array();
+    DeleteFile dv = mockDV(bytes.length, index.cardinality());
+    PositionDeleteIndex indexCopy = PositionDeleteIndex.deserialize(bytes, dv);
+    assertEqual(index, indexCopy);
+    byte[] goldenBytes = readTestResource(goldenFile);
+    assertThat(bytes).isEqualTo(goldenBytes);
+    PositionDeleteIndex goldenIndex = PositionDeleteIndex.deserialize(goldenBytes, dv);
+    assertEqual(index, goldenIndex);
+  }
+
+  private static DeleteFile mockDV(long contentSize, long cardinality) {
+    DeleteFile mock = Mockito.mock(DeleteFile.class);
+    Mockito.when(mock.contentSizeInBytes()).thenReturn(contentSize);
+    Mockito.when(mock.recordCount()).thenReturn(cardinality);
+    return mock;
+  }
+
+  private static void assertEqual(PositionDeleteIndex index, PositionDeleteIndex thatIndex) {
+    assertThat(index.cardinality()).isEqualTo(thatIndex.cardinality());
+    index.forEach(position -> assertThat(thatIndex.isDeleted(position)).isTrue());
+    thatIndex.forEach(position -> assertThat(index.isDeleted(position)).isTrue());
+  }
+
+  private static long position(int bitmapIndex, int containerIndex, long value) {
+    return bitmapIndex * BITMAP_OFFSET + containerIndex * CONTAINER_OFFSET + value;
+  }
+
+  private static byte[] readTestResource(String resourceName) throws IOException {
+    URL resource = Resources.getResource(TestRoaringPositionBitmap.class, resourceName);
+    return Resources.toByteArray(resource);
+  }
+
   private List<Long> collect(PositionDeleteIndex index) {
     List<Long> positions = Lists.newArrayList();
     index.forEach(positions::add);
diff --git a/core/src/test/resources/org/apache/iceberg/deletes/all-container-types-position-index.bin b/core/src/test/resources/org/apache/iceberg/deletes/all-container-types-position-index.bin
new file mode 100644
index 000000000000..00d47303b11b
Binary files /dev/null and b/core/src/test/resources/org/apache/iceberg/deletes/all-container-types-position-index.bin differ
diff --git a/core/src/test/resources/org/apache/iceberg/deletes/empty-position-index.bin b/core/src/test/resources/org/apache/iceberg/deletes/empty-position-index.bin
new file mode 100644
index 000000000000..8bbc1265dc1d
Binary files /dev/null and b/core/src/test/resources/org/apache/iceberg/deletes/empty-position-index.bin differ
diff --git a/core/src/test/resources/org/apache/iceberg/deletes/small-alternating-values-position-index.bin b/core/src/test/resources/org/apache/iceberg/deletes/small-alternating-values-position-index.bin
new file mode 100644
index 000000000000..80829fae22c3
Binary files /dev/null and b/core/src/test/resources/org/apache/iceberg/deletes/small-alternating-values-position-index.bin differ
diff --git a/core/src/test/resources/org/apache/iceberg/deletes/small-and-large-values-position-index.bin b/core/src/test/resources/org/apache/iceberg/deletes/small-and-large-values-position-index.bin
new file mode 100644
index 000000000000..989dabf6ad53
Binary files /dev/null and b/core/src/test/resources/org/apache/iceberg/deletes/small-and-large-values-position-index.bin differ