feat(iceberg): add v3 deletion vector support

.github/workflows/integration-tests.yml

@@ -7,9 +7,13 @@ on:
     branches:
       - "master"
       - "staging"
+      - "feat/iceberg-deletion-vectors"
+      -  "feat/interoperability"
   pull_request:
     branches:
       - "*"
+      - "feat/iceberg-deletion-vectors"
+      - "feat/interoperability"
     paths:
       - '**/*.go'
       - '**/*.java'
@@ -215,7 +219,9 @@ jobs:
     # this too -- so gate on "not failed" rather than on success. go-cache is exempt entirely: its
     # cache warm and db2 clidriver download must not block seven drivers, and its lint and gosec
     # failures already surface as their own check.
-    if: ${{ !cancelled() && github.event_name == 'pull_request' && needs.preflight.result == 'success' && needs.preflight.outputs.drivers != '[]' && needs.build-jar.result != 'failure' && needs.apt-warm.result != 'failure' }}
+    # TEMP(hack): dropped `github.event_name == 'pull_request' &&` to run ITs from push while the
+    # PR has merge conflicts. REVERT BEFORE MERGE.
+    if: ${{ !cancelled() && needs.preflight.result == 'success' && needs.preflight.outputs.drivers != '[]' && needs.build-jar.result != 'failure' && needs.apt-warm.result != 'failure' }}
     runs-on: 16gb-runner
     # The whole workflow's approval, deliberately on the only job that spends a 16gb runner: one
     # prompt releases every driver, and "Re-run failed jobs" re-enters it so a replay costs another.

.gitignore

@@ -8,6 +8,7 @@ __debug_bin*
 go.sum
 go.work.sum
 */**/examples
+*/**/olake-table-index
 olake-data/
 olake-iceberg-java-writer-0.0.1-SNAPSHOT.jar
 destination/iceberg/olake-iceberg-java-writer/target/
@@ -20,3 +21,4 @@ destination/iceberg/olake-iceberg-java-writer/dependency-reduced-pom.xml
 tests/*/testdata/**/logs
 .cocoindex_code/
 CLAUDE.local.md
+

destination/iceberg/arrow-writer/utils.go

@@ -28,6 +28,8 @@ const (
 	fileTypePositionalDelete = "positionalDelete"
 	targetDataFileSize       = int64(512 * 1024 * 1024) // 512 MB
 	targetDeleteFileSize     = int64(64 * 1024 * 1024)  // 64 MB
+	// deletionVectorBatchSize is how many positions accumulate before being sent to the server for encoding as deletion vectors.
+	deletionVectorBatchSize = 100_000
 )
 
 func getDefaultWriterProps() []parquet.WriterProperty {

destination/iceberg/arrow-writer/writer.go

@@ -33,6 +33,17 @@ type ArrowWriter struct {
 	createdFiles   map[string]*PartitionFiles
 	upsertMode     bool
 	indexThread    *types.StreamIndexThread
+	// deleteMode decides how superseded rows are expressed: under DeletionVector, positions stream to the server as Puffin vectors instead of a delete file.
+	deleteMode types.UpdateType
+	// pendingVectors buffers positions per data file until a batch is worth sending. Only used under DeleteModeDeletionVector.
+	pendingVectors     map[string]*pendingVector
+	pendingVectorCount int
+}
+
+// pendingVector is the positions to delete from one data file, plus the partition it's stamped with (the one being written, not necessarily the referenced file's). First writer to touch a path wins - a data file carries only one vector.
+type pendingVector struct {
+	positions       []int64
+	partitionValues []any
 }
 
 type Writer struct {
@@ -79,6 +90,9 @@ func New(ctx context.Context, options *destination.Options, partitionInfo []inte
 		createdFiles:  make(map[string]*PartitionFiles),
 		upsertMode:    upsertMode,
 		indexThread:   types.NewStreamIndexThread(options.TableIndex),
+		deleteMode:    stream.GetUpdateType(),
+
+		pendingVectors: make(map[string]*pendingVector),
 	}
 
 	if err := writer.initialize(ctx); err != nil {
@@ -148,7 +162,9 @@ func (w *ArrowWriter) getOrCreateWriter(ctx context.Context, pKey string, values
 				return nil, err
 			}
 		}
-		if writer.positionalDeleteWriter == nil {
+		// Deletion vectors are encoded server-side from streamed positions, so this
+		// mode writes no delete file of its own - see sendPendingVectors.
+		if writer.positionalDeleteWriter == nil && w.deleteMode != types.UpdateTypeDeletionVector {
 			if writer.positionalDeleteWriter, err = w.createWriter(ctx, pKey, values, *w.arrowSchema[fileTypePositionalDelete], fileTypePositionalDelete); err != nil {
 				return nil, err
 			}
@@ -249,7 +265,11 @@ func (w *ArrowWriter) Write(ctx context.Context, records []types.RawRecord) erro
 		}
 
 		if w.upsertMode {
-			if len(writer.positionalDeletes) > 0 {
+			if w.deleteMode == types.UpdateTypeDeletionVector {
+				if err := w.queueVectorDeletes(ctx, writer.positionalDeletes, writer.dataWriter.partitionValues); err != nil {
+					return err
+				}
+			} else if len(writer.positionalDeletes) > 0 {
 				posRecord := createPositionalDeleteArrowRecord(writer.positionalDeletes, w.allocator, w.arrowSchema[fileTypePositionalDelete])
 				if err := writer.positionalDeleteWriter.currentWriter.WriteBuffered(posRecord); err != nil {
 					posRecord.Release()
@@ -373,6 +393,11 @@ func (w *ArrowWriter) Close(ctx context.Context, finalMetadataState any) (err er
 		return fmt.Errorf("failed to close arrow writers: %w", err)
 	}
 
+	// The trailing partial batch has to reach the server before the commit, which is where the vectors holding it are published.
+	if err := w.sendPendingVectors(ctx); err != nil {
+		return err
+	}
+
 	// Build ordered file list: equality deletes → data → positional deletes
 	var orderedFiles []*proto.ArrowPayload_FileMetadata
 	for _, pf := range w.createdFiles {
@@ -583,6 +608,66 @@ func (w *ArrowWriter) newRollingWriter(ctx context.Context, arrowSchema arrow.Sc
 	}, nil
 }
 
+// queueVectorDeletes buffers positions and ships them once enough have accumulated.
+func (w *ArrowWriter) queueVectorDeletes(ctx context.Context, deletes []PositionalDelete, partitionValues []any) error {
+	for _, d := range deletes {
+		pending, exists := w.pendingVectors[d.FilePath]
+		if !exists {
+			pending = &pendingVector{partitionValues: partitionValues}
+			w.pendingVectors[d.FilePath] = pending
+		}
+		pending.positions = append(pending.positions, d.Position)
+	}
+	w.pendingVectorCount += len(deletes)
+
+	if w.pendingVectorCount < deletionVectorBatchSize {
+		return nil
+	}
+	return w.sendPendingVectors(ctx)
+}
+
+// sendPendingVectors hands the buffered positions to the server, which folds them into the deletion vectors it publishes at commit.
+func (w *ArrowWriter) sendPendingVectors(ctx context.Context) error {
+	if w.pendingVectorCount == 0 {
+		return nil
+	}
+
+	entries := make([]*proto.ArrowPayload_DeletionVectorBatch_Entry, 0, len(w.pendingVectors))
+	for path, pending := range w.pendingVectors {
+		partitionValues, err := toProtoPartitionValues(pending.partitionValues)
+		if err != nil {
+			return fmt.Errorf("failed to convert partition values of %s: %s", path, err)
+		}
+
+		entries = append(entries, &proto.ArrowPayload_DeletionVectorBatch_Entry{
+			DataFilePath:    path,
+			Positions:       pending.positions,
+			PartitionValues: partitionValues,
+		})
+	}
+
+	request := &proto.ArrowPayload{
+		Type: proto.ArrowPayload_DELETION_VECTORS,
+		Metadata: &proto.ArrowPayload_Metadata{
+			ThreadId: w.options.ThreadID,
+			DeletionVectors: &proto.ArrowPayload_DeletionVectorBatch{
+				Entries: entries,
+			},
+		},
+	}
+
+	reqCtx, cancel := context.WithTimeout(ctx, constants.GRPCRequestTimeout)
+	defer cancel()
+
+	if _, err := w.server.SendClientRequest(reqCtx, request); err != nil {
+		return fmt.Errorf("failed to send deletion vector rows: %s", err)
+	}
+
+	w.pendingVectors = make(map[string]*pendingVector)
+	w.pendingVectorCount = 0
+	return nil
+}
+
 func (w *ArrowWriter) allocateFilePath(ctx context.Context, partitionKey string) (string, error) {
 	request := &proto.ArrowPayload{
 		Type: proto.ArrowPayload_FILEPATH,

destination/iceberg/iceberg.go

@@ -95,20 +95,24 @@ func (i *Iceberg) Setup(ctx context.Context, stream types.StreamInterface, _ any
 	}
 
 	upsertMode := isUpsertMode(stream, options.Backfill)
+	deleteMode := stream.GetUpdateType()
 
-	identifierField := utils.Ternary(i.config.NoIdentifierFields, "", constants.OlakeID).(string)
+	// Row-identity column, sent even where the catalog forbids declaring it.
+	identifierField := constants.OlakeID
 	iceSchema := stream.Schema().ToIceberg(!stream.NormalizationEnabled(), i.stream, partitionFields...)
 	requestPayload := proto.IcebergPayload{
 		Type: proto.IcebergPayload_GET_OR_CREATE_TABLE,
 		Metadata: &proto.IcebergPayload_Metadata{
-			Schema:               iceSchema,
-			DestTableName:        stream.GetDestinationTable(),
-			ThreadId:             options.ThreadID,
-			IdentifierField:      &identifierField,
-			Namespace:            stream.GetDestinationDatabase(&i.config.IcebergDatabase),
-			Upsert:               upsertMode,
-			UsePositionalDeletes: options.TableIndex != nil,
-			PartitionFields:      icebergPartFields,
+			Schema:          iceSchema,
+			DestTableName:   stream.GetDestinationTable(),
+			ThreadId:        options.ThreadID,
+			IdentifierField: &identifierField,
+			// Unity Catalog rejects identifier columns; only equality deletes need them declared.
+			DeclareIdentifierFields: !i.config.NoIdentifierFields,
+			Namespace:               stream.GetDestinationDatabase(&i.config.IcebergDatabase),
+			Upsert:                  upsertMode,
+			DeleteMode:              protoDeleteMode(deleteMode),
+			PartitionFields:         icebergPartFields,
 		},
 	}
 
@@ -196,7 +200,8 @@ func (i *Iceberg) Check(ctx context.Context) error {
 	ctx, cancel := context.WithTimeout(ctx, 300*time.Second)
 	defer cancel()
 
-	identifierField := utils.Ternary(i.config.NoIdentifierFields, "", constants.OlakeID).(string)
+	// Row-identity column, sent even where the catalog forbids declaring it.
+	identifierField := constants.OlakeID
 	request := &proto.IcebergPayload{
 		Type: proto.IcebergPayload_GET_OR_CREATE_TABLE,
 		Metadata: &proto.IcebergPayload_Metadata{
@@ -206,6 +211,12 @@ func (i *Iceberg) Check(ctx context.Context) error {
 			Namespace:       destinationDB,
 			Upsert:          false,
 			IdentifierField: &identifierField,
+			// Unity Catalog rejects identifier columns; only equality deletes need them declared.
+			DeclareIdentifierFields: !i.config.NoIdentifierFields,
+			// The check table is written once, appended, and never updated, so no
+			// delete mode is meaningful here. Equality is the one that constrains the
+			// table least: it creates at format version 2 and needs no table index.
+			DeleteMode: proto.IcebergPayload_DELETE_MODE_EQUALITY,
 		},
 	}
 
@@ -618,6 +629,22 @@ func isUpsertMode(stream types.StreamInterface, backfill bool) bool {
 	return utils.Ternary(stream.Self().StreamMetadata.AppendMode, false, !backfill).(bool)
 }
 
+// protoDeleteMode maps the config-facing delete mode onto the wire enum. An
+// unknown mode reaches the server as UNSPECIFIED, which it rejects; config is
+// validated by types.UpdateType.Validate() long before this point.
+func protoDeleteMode(mode types.UpdateType) proto.IcebergPayload_DeleteMode {
+	switch mode {
+	case types.UpdateTypeEquality:
+		return proto.IcebergPayload_DELETE_MODE_EQUALITY
+	case types.UpdateTypePosition:
+		return proto.IcebergPayload_DELETE_MODE_POSITION
+	case types.UpdateTypeDeletionVector:
+		return proto.IcebergPayload_DELETE_MODE_DELETION_VECTOR
+	default:
+		return proto.IcebergPayload_DELETE_MODE_UNSPECIFIED
+	}
+}
+
 func init() {
 	var server *serverInstance
 	var icebergConfig *Config

destination/iceberg/index.go

@@ -25,16 +25,17 @@ func (i *Iceberg) reconcileTableIndex(ctx context.Context, index types.StreamInd
 	table := fmt.Sprintf("%s.%s", i.stream.GetDestinationDatabase(&i.config.IcebergDatabase), i.stream.GetDestinationTable())
 
 	if hasEqualityDeletes {
-		// first check for equality deletes and migrate them to positional deletes
+		// Rewrite straight into the stream's target encoding, one commit, not eq -> pos -> dv.
 		migrated, err := i.server.tableIndexClient.MigrateEqualityDeletes(ctx, &proto.MigrateEqualityDeletesRequest{
-			ThreadId: i.options.ThreadID,
+			ThreadId:   i.options.ThreadID,
+			TargetMode: protoDeleteMode(i.stream.GetUpdateType()),
 		})
 		if err != nil {
 			return fmt.Errorf("failed to migrate equality deletes of table[%s]: %s", table, err)
 		}
 
-		logger.Infof("Table[%s]: rewrote %d equality delete file(s) as %d positional delete(s)",
-			table, migrated.GetRewrittenDeleteFiles(), migrated.GetPositionalDeletesWritten())
+		logger.Infof("Table[%s]: rewrote %d equality delete file(s) as %d %s delete(s)",
+			table, migrated.GetRewrittenDeleteFiles(), migrated.GetPositionalDeletesWritten(), i.stream.GetUpdateType())
 	}
 
 	indexedSnapshotID, err := index.LastCommittedSnapshot()

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

@@ -81,17 +81,29 @@ 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);
+  }
+
+  /**
+   * @param formatVersion table format version to create at. Deletion vectors need 3;
+   *                      every other delete mode works on 2. Callers derive this from
+   *                      {@code DeleteMode.minimumFormatVersion()} - see
+   *                      {@code OlakeRowsIngester.loadOrCreateTable}.
+   */
+  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());
+    LOGGER.warn("Creating table:'{}'\nschema:{}\nrowIdentifier:{}\nformatVersion:{}", tableIdentifier, schema,
+        schema.identifierFieldNames(), formatVersion);
 
     ensureNamespace(icebergCatalog, tableIdentifier);
 
     // If we have partition transforms, create a PartitionSpec
     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 +163,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))
@@ -206,6 +218,28 @@ public static GenericAppenderFactory getTableAppender(Table icebergTable) {
     }
   }
 
+  /**
+   * Raises an existing table to {@code formatVersion} when it sits below it, so a
+   * stream reconfigured to a delete mode needing a higher version (e.g. {@code eq} or
+   * {@code pos} -> {@code dv}) can keep writing to the table it already has instead of
+   * failing outright.
+   *
+   * <p>Iceberg only moves format versions forward, so this is one-way: a table raised
+   * to 3 for deletion vectors cannot be lowered back to 2 later, and every engine
+   * reading it afterward needs to understand v3. Never called implicitly for a
+   * mismatch in the other direction (table above what the mode needs) - that is
+   * rejected instead, see {@code OlakeRowsIngester.loadOrCreateTable}.
+   */
+  public static void ensureFormatVersion(Table table, int formatVersion) {
+    int current = ((org.apache.iceberg.HasTableOperations) 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();
+  }
+
   public static OutputFileFactory getTableOutputFileFactory(Table icebergTable, FileFormat format) {
     return OutputFileFactory.builderFor(icebergTable,
             IcebergUtil.partitionId(), 1L)

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

@@ -5,6 +5,8 @@
 import org.apache.iceberg.io.OutputFileFactory;
 
 import io.debezium.server.iceberg.IcebergUtil;
+import io.debezium.server.iceberg.tableoperator.ArrowDeletionVectorWriter;
+import io.debezium.server.iceberg.tableoperator.DeleteMode;
 import io.debezium.server.iceberg.tableoperator.IcebergTableOperator;
 
 public class IcebergSession {
@@ -13,20 +15,26 @@ public class IcebergSession {
     public final OutputFileFactory fileFactory;
     public final String identifierField;
     public final boolean upsert;
-    public final boolean usePositionalDeletes;
+    public final DeleteMode deleteMode;
+    /**
+     * Built lazily by the arrow path on the first DELETION_VECTORS batch and closed at
+     * REGISTER_AND_COMMIT. Null in every other mode, and until the first batch arrives.
+     */
+    public ArrowDeletionVectorWriter dvWriter;
 
-    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, identifierField);
         this.identifierField = identifierField;
         this.upsert = upsert;
-        this.usePositionalDeletes = usePositionalDeletes;
+        this.deleteMode = deleteMode;
 
         FileFormat fileFormat = IcebergUtil.getTableFileFormat(icebergTable);
         this.fileFactory = IcebergUtil.getTableOutputFileFactory(icebergTable, fileFormat);
     }
 
+    /** Whether the table declares identifier fields; read from the table, since a catalog may refuse the declaration. */
     public boolean createIdentifierFields() {
-        return identifierField != null && !identifierField.isEmpty();
+        return !icebergTable.schema().identifierFieldIds().isEmpty();
     }
 }