feat(iceberg): experimental deletion vector delete mode

destination/iceberg/iceberg.go

@@ -107,7 +107,10 @@ func (i *Iceberg) Setup(ctx context.Context, stream types.StreamInterface, exist
 			Namespace:            stream.GetDestinationDatabase(&i.config.IcebergDatabase),
 			Upsert:               upsertMode,
 			UsePositionalDeletes: options.RowIndex != nil,
-			PartitionFields:      icebergPartFields,
+			// Named mode drives the writer; the boolean above stays for servers
+			// built before deletion vectors existed.
+			DeleteMode:      string(options.DeleteMode),
+			PartitionFields: icebergPartFields,
 		},
 	}
 

destination/iceberg/olake-iceberg-java-writer/pom.xml

@@ -508,6 +508,12 @@
       <version>5.14.2</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter</artifactId>
+      <version>5.10.5</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>testcontainers</artifactId>

destination/iceberg/olake-iceberg-java-writer/src/main/java/io/debezium/server/iceberg/IcebergUtil.java

@@ -81,6 +81,17 @@ public static Table createIcebergTable(Catalog icebergCatalog, TableIdentifier t
 
   public static Table createIcebergTable(Catalog icebergCatalog, TableIdentifier tableIdentifier,
                                          Schema schema, String writeFormat, List<Map<String, String>> partitionTransforms) {
+    return createIcebergTable(icebergCatalog, tableIdentifier, schema, writeFormat, partitionTransforms, 2);
+  }
+
+  /**
+   * Creates the table at {@code formatVersion}. Deletion vectors are a v3 construct, so
+   * a table that will receive them has to be created as v3 up front; everything else
+   * stays on v2, which is what every reader understands.
+   */
+  public static Table createIcebergTable(Catalog icebergCatalog, TableIdentifier tableIdentifier,
+                                         Schema schema, String writeFormat, List<Map<String, String>> partitionTransforms,
+                                         int formatVersion) {
 
     LOGGER.warn("Creating table:'{}'\nschema:{}\nrowIdentifier:{}", tableIdentifier, schema,
         schema.identifierFieldNames());
@@ -91,7 +102,7 @@ public static Table createIcebergTable(Catalog icebergCatalog, TableIdentifier t
     if (partitionTransforms.isEmpty()) {
       // No partitioning - create a table as before
       return icebergCatalog.buildTable(tableIdentifier, schema)
-              .withProperty(FORMAT_VERSION, "2")
+              .withProperty(FORMAT_VERSION, String.valueOf(formatVersion))
               .withProperty(DEFAULT_FILE_FORMAT, writeFormat.toLowerCase(Locale.ENGLISH))
               .withSortOrder(IcebergUtil.getIdentifierFieldsAsSortOrder(schema))
               .create();
@@ -151,7 +162,7 @@ public static Table createIcebergTable(Catalog icebergCatalog, TableIdentifier t
 
       // Create the table with the partition spec
       return icebergCatalog.buildTable(tableIdentifier, schema)
-              .withProperty(FORMAT_VERSION, "2")
+              .withProperty(FORMAT_VERSION, String.valueOf(formatVersion))
               .withProperty(DEFAULT_FILE_FORMAT, writeFormat.toLowerCase(Locale.ENGLISH))
               .withPartitionSpec(specBuilder.build())
               .withSortOrder(IcebergUtil.getIdentifierFieldsAsSortOrder(schema))
@@ -168,6 +179,22 @@ private static SortOrder getIdentifierFieldsAsSortOrder(Schema schema) {
     return sob.build();
   }
 
+  /**
+   * Raises an existing table to {@code formatVersion} when it sits below it. Iceberg
+   * only moves format versions forward, so this is one-way: a table upgraded for
+   * deletion vectors cannot be read by anything that speaks only the older version.
+   */
+  public static void ensureFormatVersion(Table table, int formatVersion) {
+    int current = ((org.apache.iceberg.BaseTable) table).operations().current().formatVersion();
+    if (current >= formatVersion) {
+      return;
+    }
+    LOGGER.warn("Upgrading {} from format version {} to {}; this cannot be undone",
+        table.name(), current, formatVersion);
+    table.updateProperties().set(FORMAT_VERSION, String.valueOf(formatVersion)).commit();
+    table.refresh();
+  }
+
   public static Optional<Table> loadIcebergTable(Catalog icebergCatalog, TableIdentifier tableId) {
     try {
       Table table = icebergCatalog.loadTable(tableId);

destination/iceberg/olake-iceberg-java-writer/src/main/java/io/debezium/server/iceberg/rowindex/EqualityDeleteMigrator.java

@@ -25,6 +25,8 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.data.GenericAppenderFactory;
 import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
 import org.apache.iceberg.deletes.PositionDelete;
 import org.apache.iceberg.deletes.PositionDeleteWriter;
 import org.apache.iceberg.encryption.EncryptedOutputFile;
@@ -34,6 +36,7 @@
 import org.slf4j.LoggerFactory;
 
 import io.debezium.server.iceberg.IcebergUtil;
+import io.debezium.server.iceberg.tableoperator.DeleteMode;
 
 /**
  * Replaces a table's equality delete files with positional delete files that
@@ -68,6 +71,19 @@ private Result(long snapshotId, int rewrittenDeleteFiles, long positionalDeletes
   }
 
   public static Result migrate(Table table, String identifierField, OutputFileFactory fileFactory) throws Exception {
+    return migrate(table, identifierField, fileFactory, DeleteMode.POSITION);
+  }
+
+  /**
+   * Rewrites the table's equality deletes into {@code targetMode}.
+   *
+   * <p>The exchange stays a single {@link RewriteFiles} commit whichever form is
+   * produced, so a reader never observes the affected rows as undeleted. Only the
+   * representation differs: positional mode writes delete files sorted by path and
+   * offset, deletion vector mode writes one Puffin bitmap per data file.
+   */
+  public static Result migrate(Table table, String identifierField, OutputFileFactory fileFactory,
+      DeleteMode targetMode) throws Exception {
     table.refresh();
 
     Snapshot current = table.currentSnapshot();
@@ -102,7 +118,7 @@ public static Result migrate(Table table, String identifierField, OutputFileFact
       posConvCount += collectPositions(table, entry.dataFile, projection, identifierField, deletedKeys, group);
     }
 
-    List<DeleteFile> written = writePositionDeletes(table, fileFactory, groups);
+    List<DeleteFile> written = writeDeletes(table, fileFactory, groups, targetMode);
 
     RewriteFiles rewrite = table.newRewrite();
     for (DeleteFile deleteFile : replaced) {
@@ -193,6 +209,45 @@ private static long collectPositions(Table table, DataFile dataFile, Schema proj
     return matched;
   }
 
+  private static List<DeleteFile> writeDeletes(Table table, OutputFileFactory fileFactory,
+      Map<String, PartitionGroup> groups, DeleteMode targetMode) throws IOException {
+    if (targetMode == DeleteMode.DELETION_VECTOR) {
+      return writeDeletionVectors(table, fileFactory, groups);
+    }
+    return writePositionDeletes(table, fileFactory, groups);
+  }
+
+  /**
+   * One Puffin vector per data file. The migration replaces the table's equality
+   * deletes wholesale, so nothing has been deleted positionally yet and there is no
+   * previous vector to merge with; that is why a null loader is correct here, and why
+   * it would not be if a table could be migrated a second time.
+   */
+  private static List<DeleteFile> writeDeletionVectors(Table table, OutputFileFactory fileFactory,
+      Map<String, PartitionGroup> groups) throws IOException {
+    PartitionSpec spec = table.spec();
+    List<DeleteFile> written = new ArrayList<>();
+
+    DVFileWriter writer = new BaseDVFileWriter(fileFactory, path -> null);
+    try {
+      for (PartitionGroup group : groups.values()) {
+        if (group.positions.isEmpty()) {
+          LOGGER.info("No positions to write for partition {}", group.partition);
+          continue;
+        }
+        StructLike partition = spec.isUnpartitioned() ? null : group.partition;
+        for (RowPosition row : group.positions) {
+          writer.delete(row.path, row.position, spec, partition);
+        }
+      }
+    } finally {
+      writer.close();
+    }
+
+    written.addAll(writer.result().deleteFiles());
+    return written;
+  }
+
   private static List<DeleteFile> writePositionDeletes(Table table, OutputFileFactory fileFactory,
       Map<String, PartitionGroup> groups) throws IOException {
     GenericAppenderFactory appenderFactory = IcebergUtil.getTableAppender(table);

destination/iceberg/olake-iceberg-java-writer/src/main/java/io/debezium/server/iceberg/rowindex/TableRowIndexScanner.java

@@ -3,7 +3,6 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
-import java.util.BitSet;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
@@ -18,16 +17,16 @@
 import org.apache.avro.generic.GenericRecord;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.data.BaseDeleteLoader;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
 import org.apache.iceberg.FileContent;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.FileScanTask;
-import org.apache.iceberg.MetadataColumns;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.io.CloseableIterable;
-import org.apache.iceberg.io.DeleteSchemaUtil;
 import org.apache.iceberg.parquet.Parquet;
 import org.apache.iceberg.types.Types.NestedField;
 import org.slf4j.Logger;
@@ -47,7 +46,7 @@
  */
 public final class TableRowIndexScanner {
   private static final Logger LOGGER = LoggerFactory.getLogger(TableRowIndexScanner.class);
-  private static final BitSet EMPTY_POSITIONS = new BitSet(0);
+  private static final PositionDeleteIndex EMPTY_POSITIONS = PositionDeleteIndex.empty();
 
   private TableRowIndexScanner() {
   }
@@ -132,7 +131,7 @@ public static ScanResult scan(Table table, String identifierField, Long fromSnap
 
     consumer.begin(current.snapshotId());
     Schema projection = identifierProjection(table, identifierField);
-    Map<String, BitSet> deletedPositions = deletedPositions(table);
+    DeletedPositions deletedPositions = new DeletedPositions(table);
     long entries = 0L;
 
     // first remove index of removed data files (if exist)
@@ -143,7 +142,7 @@ public static ScanResult scan(Table table, String identifierField, Long fromSnap
       entries += emitFile(table, file, projection, identifierField, EMPTY_POSITIONS, true, consumer);
     }
     for (DataFile file : addedFiles) {
-      BitSet deleted = deletedPositions.getOrDefault(file.location(), EMPTY_POSITIONS);
+      PositionDeleteIndex deleted = deletedPositions.forFile(file.location());
       entries += emitFile(table, file, projection, identifierField, deleted, false, consumer);
     }
 
@@ -164,34 +163,41 @@ private static boolean isReadable(Table table, DataFile file) {
   }
 
   /**
-   * Positions already removed by positional delete files, keyed by data file path.
-   * Skipping these keeps the index proportional to the number of live rows rather
-   * than to everything the table has ever held.
-   */ 
-  private static Map<String, BitSet> deletedPositions(Table table) throws IOException {
-    Map<String, BitSet> byFile = new HashMap<>();
-    Schema pathPos = DeleteSchemaUtil.pathPosSchema();
-
-    for (DeleteFile delete : deleteFiles(table, FileContent.POSITION_DELETES)) {
-      try (CloseableIterable<Object> rows = openParquet(table, delete.location(), pathPos)) {
-        for (Object row : rows) {
-          Object path = getFieldValue(row, MetadataColumns.DELETE_FILE_PATH.name());
-          Object position = getFieldValue(row, MetadataColumns.DELETE_FILE_POS.name());
-          if (path == null || position == null) {
-            continue;
-          }
-          long ordinal = position instanceof Number n ? n.longValue() : Long.parseLong(position.toString());
-          if (ordinal > Integer.MAX_VALUE) {
-            // No realistic data file holds this many rows. Treating such a row as
-            // live only costs a redundant positional delete later on.
+   * Positions already removed from each data file, whether by positional delete files
+   * or by a deletion vector. Skipping them keeps the index proportional to the number
+   * of live rows rather than to everything the table has ever held.
+   *
+   * <p>Iceberg's loader reads both representations, so a table part way through a
+   * migration reports every deleted position without this having to know which form it
+   * is in. The table is planned once and each file's deletes are loaded on demand.
+   */
+  private static final class DeletedPositions {
+    private final Table table;
+    private final Map<String, List<DeleteFile>> byDataFile = new HashMap<>();
+    private final BaseDeleteLoader loader;
+
+    private DeletedPositions(Table table) throws IOException {
+      this.table = table;
+      this.loader = new BaseDeleteLoader(file -> table.io().newInputFile(file.location()));
+      try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
+        for (FileScanTask task : tasks) {
+          if (task.deletes().isEmpty()) {
             continue;
           }
-          byFile.computeIfAbsent(path.toString(), key -> new BitSet()).set((int) ordinal);
+          // planFiles can split one file across several tasks; merge their delete lists.
+          byDataFile.computeIfAbsent(task.file().location(), path -> new ArrayList<>())
+              .addAll(task.deletes());
         }
       }
     }
 
-    return byFile;
+    private PositionDeleteIndex forFile(String path) {
+      List<DeleteFile> deletes = byDataFile.get(path);
+      if (deletes == null || deletes.isEmpty()) {
+        return EMPTY_POSITIONS;
+      }
+      return loader.loadPositionDeletes(deletes, path);
+    }
   }
 
   /** Every data file visible in the table's current snapshot, oldest first. */
@@ -312,15 +318,15 @@ static Schema identifierProjection(Table table, String identifierField) {
    * Returns the number of entries emitted.
    */
   private static long emitFile(Table table, DataFile file, Schema projection, String identifierField,
-      BitSet deleted, boolean isDeletedFile, EntryConsumer consumer) throws Exception {
+      PositionDeleteIndex deleted, boolean isDeletedFile, EntryConsumer consumer) throws Exception {
     String path = file.location();
     long position = 0L;
     long emitted = 0L;
 
     try (CloseableIterable<Object> rows = openRows(table, file, projection)) {
       for (Object row : rows) {
         Object identifier = getFieldValue(row, identifierField);
-        if (identifier != null && !isDeleted(deleted, position)) {
+        if (identifier != null && !deleted.isDeleted(position)) {
           consumer.accept(identifier.toString(), path, position, isDeletedFile);
           emitted++;
         }
@@ -331,10 +337,6 @@ private static long emitFile(Table table, DataFile file, Schema projection, Stri
     return emitted;
   }
 
-  /** BitSet indexes by int, so ordinals beyond its range count as live. */
-  private static boolean isDeleted(BitSet deleted, long position) {
-    return position <= Integer.MAX_VALUE && deleted.get((int) position);
-  }
 
   /** Extracts a field value from either an Iceberg Record or an Avro GenericRecord. */
   public static Object getFieldValue(Object row, String fieldName) {

destination/iceberg/olake-iceberg-java-writer/src/main/java/io/debezium/server/iceberg/rpc/IcebergSession.java

@@ -1,6 +1,7 @@
 package io.debezium.server.iceberg.rpc;
 
 import org.apache.iceberg.FileFormat;
+import io.debezium.server.iceberg.tableoperator.DeleteMode;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.io.OutputFileFactory;
 
@@ -14,13 +15,15 @@ public class IcebergSession {
     public final String identifierField;
     public final boolean upsert;
     public final boolean usePositionalDeletes;
+    public final DeleteMode deleteMode;
 
-    public IcebergSession(Table icebergTable, boolean upsert, String identifierField, boolean usePositionalDeletes) {
+    public IcebergSession(Table icebergTable, boolean upsert, String identifierField, DeleteMode deleteMode) {
         this.icebergTable = icebergTable;
-        this.op = new IcebergTableOperator(upsert, usePositionalDeletes);
+        this.op = new IcebergTableOperator(upsert, deleteMode);
         this.identifierField = identifierField;
         this.upsert = upsert;
-        this.usePositionalDeletes = usePositionalDeletes;
+        this.deleteMode = deleteMode;
+        this.usePositionalDeletes = deleteMode.addressesPositions();
 
         FileFormat fileFormat = IcebergUtil.getTableFileFormat(icebergTable);
         this.fileFactory = IcebergUtil.getTableOutputFileFactory(icebergTable, fileFormat);

destination/iceberg/olake-iceberg-java-writer/src/main/java/io/debezium/server/iceberg/rpc/OlakeRowIndexer.java

@@ -7,6 +7,7 @@
 
 import io.debezium.server.iceberg.rowindex.EqualityDeleteMigrator;
 import io.debezium.server.iceberg.rowindex.TableRowIndexScanner;
+import io.debezium.server.iceberg.tableoperator.DeleteMode;
 import io.debezium.server.iceberg.rpc.RecordIngest.MigrateEqualityDeletesRequest;
 import io.debezium.server.iceberg.rpc.RecordIngest.MigrateEqualityDeletesResponse;
 import io.debezium.server.iceberg.rpc.RecordIngest.RowIndexScanBatch;
@@ -72,8 +73,10 @@ public void migrateEqualityDeletes(MigrateEqualityDeletesRequest request,
       StreamObserver<MigrateEqualityDeletesResponse> responseObserver) {
     try {
       IcebergSession session = requireSession(request.getThreadId());
+      // Empty target means positional, which is what callers predating vectors expect.
+      DeleteMode targetMode = DeleteMode.resolve(request.getTargetMode(), true);
       EqualityDeleteMigrator.Result result = EqualityDeleteMigrator.migrate(
-          session.icebergTable, session.identifierField, session.fileFactory);
+          session.icebergTable, session.identifierField, session.fileFactory, targetMode);
 
       responseObserver.onNext(MigrateEqualityDeletesResponse.newBuilder()
           .setSnapshotId(result.snapshotId)