feat(iceberg): add v3 deletion vector support by vikaxsh · Pull Request #1162 · datazip-inc/olake

vikaxsh · 2026-08-26T13:16:54Z

Description

Adds dv as a third per-stream delete mode alongside eq and pos, writing Iceberg v3 Puffin deletion vectors instead of Parquet positional deletes.

Introduces PositionalDeleteSink so PositionalDeltaWriter can target either representation, plus DeletionVectorConverter and PreviousDeleteLoader to merge new deletes with existing Parquet positional deletes or deletion vectors without resurrecting previously deleted rows.

Updates EqualityDeleteMigrator.migrate() to accept a targetMode, allowing tables to switch from equality deletes directly to pos or dv encoding in a single atomic RewriteFiles commit.

Fixes #1168

Type of change

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)
Breaking change (fix or feature that would cause existing functionality to not work as expected)
This change requires a documentation update

How Has This Been Tested?

Scenario A
Scenario B

Screenshots or Recordings

Documentation

Documentation Link: [link to README, olake.io/docs, or olake-docs]
N/A (bug fix, refactor, or test changes only)

Related PR's (If Any):

…e so for reference it is

…dex on the fly

…emitting pos deltes

…erg validation test pending

…c/olake into feat/iceberg-deletion-vectors

…on-vectors

…ion-vectors

hash-data · 2026-09-17T08:29:57Z

AI review


P0

1. CI workflow hacked, comment says so. .github/workflows/integration-tests.yml — feat/iceberg-deletion-vectors and feat/interoperability added to both push and pull_request triggers, and the github.event_name == 'pull_request' gate dropped under a TEMP(hack) ... REVERT BEFORE MERGE comment. Merges as-is, 16gb-runner ITs fire on every push to those branches.

2. Proto field numbers reused, no reserved. records_ingest.proto, IcebergPayload.Metadata:

┌─────┬─────────────────────────────────┬──────────────────────────────────────────┐
│ tag │               was               │                   now                    │
├─────┼─────────────────────────────────┼──────────────────────────────────────────┤
│ 9   │ bool use_positional_deletes     │ repeated PartitionField partition_fields │
├─────┼─────────────────────────────────┼──────────────────────────────────────────┤
│ 10  │ repeated PartitionField         │ optional int64 base_snapshot_id          │
├─────┼─────────────────────────────────┼──────────────────────────────────────────┤
│ 11  │ optional int64 base_snapshot_id │ DeleteMode delete_mode                   │
└─────┴─────────────────────────────────┴──────────────────────────────────────────┘

sink_jar_path is a user config knob (destination/iceberg/config.go:69), so Go-binary/jar skew is a reachable deployment shape, not hypothetical. Tags 9 and 10 change wire type and fail loudly; tag 11 stays varint — an old client's base_snapshot_id parses as a garbage delete_mode, silently. Fix costs nothing: keep 9/10/11 where they were, reserved 9;, put delete_mode/declare_identifier_fields at 12/13. If you decide lockstep shipping makes skew impossible, say so in the PR — but decide it explicitly.

P1

3. validateNoConflictingDeleteFiles() is unconditional and mostly redundant. IcebergTableOperator.java:242. Iceberg's BaseRowDelta.validate() already calls validateAddedDVs(base, startingSnapshotId, filter, parent) unconditionally (1.10.2, line 168) — that is the DV-staleness guard the comment describes. What the added call buys on top is validateNoNewDeleteFiles with the default conflictDetectionFilter = alwaysTrue(): any delete file added anywhere in the table since baseSnapshotId now fails the commit. It sits inside baseSnapshotId != null, so it hits pos mode too, which never needed it — and the code 60 lines below (staged.parentId() != baseSnapshotId → return 0L) exists precisely because concurrent commits are expected and were previously tolerated. Gate on deleteMode == DeleteMode.DELETION_VECTOR.

4. pos→dv leaves Parquet positional deletes in a v3 table. ensureFormatVersion upgrades unconditionally, nothing rewrites existing pos-delete files. PreviousDeleteLoader only merges deletes for data files this commit touches, and BaseDVFileWriter.close() only retires previous files where ContentFileUtil.isFileScoped holds. Untouched data files keep Parquet deletes forever. Iceberg-java still reads them; v3 spec says a table must not carry position delete files, and other engines may reject. eq→dv gets a migration RPC, pos→dv gets nothing — and iceberg_dv.go tests eq→dv but not pos→dv.

5. Migrator writes Puffin blobs named .parquet. OlakeTableIndexer passes session.fileFactory into EqualityDeleteMigrator.migrate(..., DELETION_VECTOR) → BaseDVFileWriter. That factory is built from getTableFileFormat(icebergTable) = parquet (IcebergSession.java:34-35). This is exactly what ArrowDeletionVectorWriter:53-56 and IcebergTableWriterFactory.deleteSink both build a separate PUFFIN factory to avoid. Metadata carries FileFormat.PUFFIN so reads work; anything inferring format from the extension does not. Build the PUFFIN factory inside writeDeletionVectors.

6. Wrong partition on the vector — consequence understated. TODO'd in ArrowDeletionVectorWriter.add and PositionalDeltaWriter.write; both say the delete still applies because DeleteFileIndex matches by path. True after planning — but scan planning prunes delete manifests by partition first, so a partition-filtered read of the moved-from partition can miss the vector and resurrect the row. The ITs deliberately avoid it (dv-part-update-id1 updates amount, not the partition column). Shipping TODO'd is defensible; "partitioned table + partition-changing update" needs to reach user-facing docs, not just a code comment.

P2

7. PreviousDeleteLoader.planDeletes runs table.newScan().planFiles() — full table plan, once per writer lifecycle, i.e. per commit. Pos mode pays nothing equivalent. Consider planning from delete manifests only.

8. Append-mode stream with update_type: dv still sends DELETE_MODE_DELETION_VECTOR — iceberg.go Setup now sends stream.GetUpdateType() unconditionally where it previously sent options.TableIndex != nil, and destination/writers.go:129 leaves TableIndex nil in append mode. The destination table gets irreversibly upgraded to v3 for a stream that will never write a delete, and the server builds position maps the client has no index to store.

9. ensureFormatVersion is a silent one-way table-wide upgrade driven by a per-stream setting, logged only as a Java-side WARN. Worth surfacing to the Go side.

result: Reviewed PR #1162 (iceberg DV v3) — 2 P0 (CI REVERT BEFORE MERGE hack left in; proto tags 9/10/11 reused without reserved, tag 11 misparses silently), 4 P1 (redundant validateNoConflictingDeleteFiles regresses pos-mode concurrency; pos→dv leaves Parquet pos-deletes in a v3 table, untested; migrator writes Puffin as .parquet; partition mis-stamp can resurrect rows on partition-filtered reads), 3 P2; both builds pass.

w.r.t. v3 spec


New P1: v3 row lineage is mandatory and this breaks it

No opt-out — row lineage is required for all v3 tables, no flag, no property. Spec, on a row moving to a different data file: "The row's existing non-null _row_id must be copied into the new data file."

Olake's upsert is "DV the old position, append a fresh row." The new row carries no _row_id, so it inherits a brand-new one from first_row_id. Every UPDATE mints a new row identity. Any consumer using row tracking — Databricks CDF-on-Iceberg, Dremio, incremental reads — sees delete+insert instead of update, on every table this PR upgrades.

Not cheaply fixable: carrying _row_id forward means the table index has to store it alongside RowLocation. Minimum bar is a documented limitation; it's a v3-only semantic that eq/pos at v2 never exposed.

Mechanics I checked and they're fine: row-id allocation advances by sum(recordCount) of added data files, and olake derives record count from ParquetUtil.fileMetrics on the real footer (IcebergTableOperator.java:483), so next-row-id accounting is correct. nextRowId is a primitive defaulting to INITIAL_ROW_ID = 0, and pre-upgrade files keep a null first_row_id — the v2→v3 upgrade neither NPEs nor corrupts.

New P1: the silent upgrade locks out most query engines, irreversibly

ensureFormatVersion flips a shared table to v3 from a per-stream setting, with a Java-side WARN as the only signal. Engine reality as of 2026:

┌──────────────────────────────────────────┬───────────────────────────────────────────────┐
│                  Engine                  │                 v3 / DV reads                 │
├──────────────────────────────────────────┼───────────────────────────────────────────────┤
│ Spark 4.0 + Iceberg 1.10.x               │ full                                          │
├──────────────────────────────────────────┼───────────────────────────────────────────────┤
│ EMR 7.12, Glue, S3 Tables, Databricks UC │ GA                                            │
├──────────────────────────────────────────┼───────────────────────────────────────────────┤
│ Athena                                   │ hard fail — Cannot read unsupported version 3 │
├──────────────────────────────────────────┼───────────────────────────────────────────────┤
│ Trino                                    │ experimental, no row-level DML                │
├──────────────────────────────────────────┼───────────────────────────────────────────────┤
│ ClickHouse                               │ DV reads still an open PR                     │
├──────────────────────────────────────────┼───────────────────────────────────────────────┤
│ Snowflake managed                        │ not public                                    │
└──────────────────────────────────────────┴───────────────────────────────────────────────┘

A user flipping update_type: dv on an existing table can lock their BI stack out of it, and format version only moves forward. This needs explicit opt-in — create-at-v3 is fine, upgrading someone's existing table shouldn't happen without them asking.

Softened: pos→dv leftovers (was P1 #4, now P2)

The spec is more forgiving than I assumed — files written before the upgrade stay valid; the obligation is to absorb lingering position deletes when that data file's deletes are next updated. This implementation does exactly that, and I traced the whole chain: BaseDeleteLoader.loadPositionDeletes tags the returned index with its source files, BitmapPositionDeleteIndex.merge preserves that list, BaseDVFileWriter.close() retires the file-scoped ones — and olake's granularity is file-scoped. Compliant for touched files, spec-permitted for untouched ones. Residual: orphan Parquet deletes linger on cold data files forever, and there's still no pos→dv IT.

Confirmed: rejecting eq on v3 is stricter than Iceberg

MergingSnapshotProducer.validateDeleteFileForVersion case 3 is content() == EQUALITY_DELETES || isDV(file) — equality deletes are legal in v3. The rejection in validateOrUpgradeFormatVersion is an olake writer limitation (BaseEqualityDeltaWriter emits a pos-delete for in-batch dedup), which the code comment states accurately. Worth surfacing as an operational consequence: once any stream upgrades a table to v3, every eq stream targeting that table fails permanently and the only recovery is dropping the table.

Also verified clean

- .parquet-named Puffin from the migrator still commits — ContentFileUtil.isDV tests format() == PUFFIN, not the extension. Stays a naming/ops issue (P2), not a commit failure.
- PreviousDeleteLoader's explicit FileContent.POSITION_DELETES filter is load-bearing, not defensive padding: a mid-migration table with eq deletes plus a DV on one data file would otherwise hand a Puffin blob to BaseDeleteLoader.openDeletes, whose format switch has no PUFFIN case.
- findDV asserts dv.dataSequenceNumber() >= data file seq — satisfied, both land in one commit.
- Old-DV retirement works end to end.

result: Deep v3 research on PR #1162 — partition mis-stamp escalated to P0 (verified DeleteFileIndex prunes delete manifests by partition before path matching, so partition-filtered reads resurrect superseded rows); two new P1s (mandatory v3 row lineage broken by delete+append upserts; silent one-way v3 upgrade locks out Athena/Trino/ClickHouse/Snowflake); pos→dv leftovers softened to P2 as spec-permitted.

hash-data

first review part 1

hash-data · 2026-09-17T12:48:33Z

+     * time after a whole sync's worth of work.
+     */
+    private void validateOrUpgradeFormatVersion(Table table, DeleteMode deleteMode) {
+        int current = ((org.apache.iceberg.HasTableOperations) table).operations().current().formatVersion();


Suggested change

int current = ((org.apache.iceberg.HasTableOperations) table).operations().current().formatVersion();

int curSpecVersion = ((org.apache.iceberg.HasTableOperations) table).operations().current().formatVersion();

hash-data · 2026-09-17T12:49:27Z

+        if (current < required) {
+            // e.g. a stream reconfigured from eq/pos to dv against a table that
+            // already exists at v2. One-way; see IcebergUtil.ensureFormatVersion.
+            IcebergUtil.ensureFormatVersion(table, required);


we can write logic here only, instead of creating a function

hash-data · 2026-09-18T08:35:39Z

+	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


this variable seems not required

it is required for checking delete batch size and flush

queueVectorDeletes runs once per batch, and each call adds len(deletes). The send happens only once the total reaches deletionVectorBatchSize, which can take many calls. local would reset to 0 on every call, so it would only ever see one batch's count.

hash-data · 2026-09-18T08:39:43Z

-                      if (filePath != null) {
-                        referencedDataFiles.add(filePath.toString());
-                      }
+                if (deleteFile.content() != FileContent.POSITION_DELETES) {


we need validation for deletion vector as well right? like deletion vector is committed for the file which not exist?

have we tested conflicting dv ?

POSITION_DELETES covers both dv and pos deletes files, rowDelta.validateDataFilesExist(referencedDataFiles) would handle that

in arrow writer as well?

yeah, Iceberg has only 3 content types (DATA, POSITION_DELETES, EQUALITY_DELETES)

added a comment

hash-data · 2026-09-18T08:54:15Z

+import io.debezium.server.iceberg.rpc.RecordIngest.IcebergPayload;
+
+/** How a writer represents the removal of a row that a later version supersedes. */
+public enum DeleteMode {


the resolver will be in common part right, it should not depend on arrow right? let us simplify if possible we can discuss

how is it depending on arrow mode? could you please explain more on this

hash-data · 2026-09-18T09:00:03Z

we need to run clear destination once someone change eq -> dv or pos -> dv

hash-data · 2026-09-21T07:06:43Z

Conflict guard is conditional. validateNoConflictingDeleteFiles() runs only when baseSnapshotId != null (IcebergTableOperator.java:236-245). On the null path a concurrent writer's DV can be silently swallowed by the merge.

hash-data · 2026-09-21T07:11:19Z

+    # 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' }}


hash-data · 2026-09-21T07:25:21Z

-		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 {


we can remove this check ? or let us also not create equality writer as well?

for eq there is already w.indexThread == nil check, it won't create eq writer

hash-data · 2026-09-21T11:59:46Z

-                      if (filePath != null) {
-                        referencedDataFiles.add(filePath.toString());
-                      }
+                if (deleteFile.content() != FileContent.POSITION_DELETES) {


in arrow writer as well?

hash-data · 2026-09-21T12:29:57Z

+   * including the destination check, which sends EQUALITY for its throwaway table.
+   * UNRECOGNIZED means the sender knows a mode this build does not.
+   */
+  public static DeleteMode resolve(IcebergPayload.DeleteMode deleteMode) {


let us do clear destination when user changes back from dv to anything

hash-data · 2026-09-21T18:43:50Z

+    }
+
+    DeleteWriteResult result = writer.result();
+    return new WrittenDeletes(result.deleteFiles(), result.rewrittenDeleteFiles());


we are sure that all older dv also merged? as well as pos got removed?

hash-data · 2026-09-21T19:42:25Z

+   */
+  static Map<String, List<DeleteFile>> planDeletes(Table table) {
+    Map<String, List<DeleteFile>> byPath = Maps.newHashMap();
+    try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {


this is heavy operation and we already do it eq migrator, see if we can have some different logic here

…tazip-inc/olake into feat/iceberg-deletion-vectors

…eat/iceberg-deletion-vectors

hash-data · 2026-09-22T18:52:05Z

+        // Deletion-vector replace semantics: a superseded vector must leave the table
+        // in the SAME commit the new one arrives in, or the table ends up with two
+        // vectors for one data file. Always empty outside DELETION_VECTOR mode.
+        rewrittenDeleteFiles.forEach(rowDelta::removeDeletes);


did not get this

hash-data · 2026-09-22T18:59:22Z

+      OutputFileFactory dvFileFactory = IcebergUtil.getTableOutputFileFactory(icebergTable, FileFormat.PUFFIN);
+      // A vector replaces the data file's previous one, so it has to be seeded with
+      // the positions already deleted or this commit would resurrect them.
+      return new PositionalDeleteSink.DeletionVectors(


do we need to have previous loader in delta writer as well?

hash-data · 2026-09-22T19:01:32Z

      // One writer for both layouts: an unpartitioned table is a single entry keyed
      // on the empty partition struct, so there is no partitioned/unpartitioned split.
+      // pos vs dv is entirely the sink's concern from here - the writer never branches.
      return new PositionalDeltaWriter(icebergTable.spec(), format, appenderFactory, fileFactory,


let us separate the names, call dv writer explicit

hash-data · 2026-09-22T19:01:51Z

+   * close. Keeping one delete file per referenced data file is what lets Iceberg treat
+   * them as file-scoped and match them to data files by path rather than by partition.
+   */
+  final class PositionalFiles implements PositionalDeleteSink {


this was not there in older version ?

hash-data and others added 30 commits July 15, 2026 15:43

wi[

b8d965b

chore: interoperability with pebble db, in next commit we might chang…

72a5db5

…e so for reference it is

resolving conflict with master and some changes in flow

2467688

chore: till now pushing all batch level things where we can create in…

f45ed95

…dex on the fly

chore: tested eq to pos migration and incremental snapshot upgrade + …

d17a330

…emitting pos deltes

Merge branch 'staging' into feat/interoperability

60f410c

Merge branch 'staging' into feat/interoperability

78a0f4c

chore: rm unnecessary package

5a98fc8

chore: integration test for interoperability

d7f551b

Merge branch 'staging' into feat/interoperability

f37565d

chore: fixing integration test

f781a0e

chore: add refresh cache in integration test

9ee5c55

chore: changing sync mode in integration tests

6247ccd

chore: refactoring in go side remains protobuf and java side

8d59899

fix: integration tests

2b5fa92

chore: no need for diff store and index

81da415

chore: remove deprecated function

b9f4714

chore: fixing case of external commit, and reviewed java side

67096c9

feat: finaly everything is fine

b34e99b

Merge branch 'staging' into feat/interoperability

fdc7a9d

chore: comment resolution and product discussion

fb3b430

chore: merge staging branch

1503cfc

chore: merge staging

8c8b476

chore: fixing lint

b2fdcd8

chore: fixing flow according to last product discussion and only iceb…

7851939

…erg validation test pending

chore: adding validations

aebe1ca

chore: merge staging

292fdef

chore: delete granularity is partition now

79d1405

chore: fix unit test and workflow

986fc84

fix: for conflict check and first snapshot

92e7348

vikaxsh requested a deployment to integration_tests August 28, 2026 06:34 — with GitHub Actions Waiting

vikaxsh and others added 9 commits August 28, 2026 15:23

fix: rm reserved filed from proto

24ffaa3

fix: revert to BasetaskWriter from TaskWriter

0fa2317

Merge branch 'feat/interoperability' of https://github.com/datazip-in…

37e8805

…c/olake into feat/iceberg-deletion-vectors

chore: merge staging

effc858

chore: rm stale comments

6eab205

fix: unity catalog

c792145

Merge remote-tracking branch 'origin/master' into feat/iceberg-deleti…

1cd31b6

…on-vectors

Merge remote-tracking branch 'origin/staging' into feat/iceberg-delet…

d5db11c

…ion-vectors

Merge branch 'staging' into feat/iceberg-deletion-vectors

d7bc17b

Merge branch 'staging' into feat/iceberg-deletion-vectors

c752dd3

hash-data reviewed Sep 18, 2026

View reviewed changes

Merge branch 'staging' into feat/iceberg-deletion-vectors

baa21e2

hash-data reviewed Sep 22, 2026

View reviewed changes

vikaxsh added 6 commits September 22, 2026 12:02

chore: revert it changes

597ab91

Merge branch 'feat/iceberg-deletion-vectors' of https://github.com/da…

a1109ed

…tazip-inc/olake into feat/iceberg-deletion-vectors

feat: reuse previousDeletes from eq migrator

9d0d03b

Merge branch 'staging' of https://github.com/datazip-inc/olake into f…

7e8f844

…eat/iceberg-deletion-vectors

chore: resolve review comments

93033a7

fix: block pos -> dv migration

fbf83b4

hash-data reviewed Sep 22, 2026

View reviewed changes

	int current = ((org.apache.iceberg.HasTableOperations) table).operations().current().formatVersion();
	int curSpecVersion = ((org.apache.iceberg.HasTableOperations) table).operations().current().formatVersion();

Conversation

vikaxsh commented Aug 26, 2026 • edited by hash-data Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Type of change

How Has This Been Tested?

Screenshots or Recordings

Documentation

Related PR's (If Any):

Uh oh!

hash-data commented Sep 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hash-data left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vikaxsh Sep 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hash-data commented Sep 18, 2026

Uh oh!

hash-data commented Sep 21, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch is waiting to be deployed

Reviewers

Assignees

vikaxsh commented Aug 26, 2026 •

edited by hash-data

Loading

hash-data commented Sep 17, 2026 •

edited

Loading

vikaxsh Sep 22, 2026 •

edited

Loading