refactor(iceberg): replace reflection-based positional delta writer

[shubham19may]

Replaces the reflection-based positional delta writer, and fixes the regressions equality-mode users would otherwise have picked up from this branch.

Why

BaseDeltaTaskWriter reached into two private fields of Iceberg's BaseEqualityDeltaWriter:

Field dataField = BaseEqualityDeltaWriter.class.getDeclaredField("dataWriter");
Field posField  = BaseEqualityDeltaWriter.class.getDeclaredField("posDeleteWriter");

That fails at runtime, in a writer constructor, on the first record of a sync — so an Iceberg field rename breaks every sync with no static signal. The reflection was forced by the base class: BaseEqualityDeltaWriter exposes only write / delete / deleteKey / close, with no way to emit an arbitrary positional delete. In positional mode we were extending a class whose entire purpose — equality deletes and insertedRowMap — we had bypassed.

What changed

PositionalDeltaWriter (new) extends BaseTaskWriter and owns both halves directly: a RollingFileWriter per partition for data, and a SortingPositionOnlyDeleteWriter per partition for deletes. Everything it touches is public or inherited-protected. A partition key is computed for every row, so an unpartitioned table is one map entry on the empty struct — no partitioned/unpartitioned writer split.

BaseDeltaTaskWriter goes back to being the equality-only writer it was before this branch. It is now identical to staging apart from javadoc, so equality mode is provably untouched.

insertedRows. Dropping BaseEqualityDeltaWriter also drops its insertedRowMap, which the legacy path depends on. The row index only learns positions once a batch is answered, so within a batch every update of a key carries the same superseded location and the second would leave both rows live. Reimplemented keyed on the identifier values rather than a copied StructLike, and scoped to a single batch via a new batchCompleted() hook — holding it until commit would grow the map to every row written beforehand, most of which (a backfill writes each key once) can never be superseded.

Also fixed

• Write-run tracking is gated on positional mode. Equality-mode users were paying two extra partition-key evaluations per record and receiving a WriteRun response the caller discards — on an interleaved partitioned table that is one run per record, so a 10k batch shipped megabytes that were thrown away.
• The GET_OR_CREATE_TABLE equality-delete probe — a full planFiles() scan of the table, per writer thread, per sync — is gated the same way. It was unconditional.
• willWrite() added to PositionTrackableWriter: a row the writer skips no longer gets a run entry pointing at the position the next row takes. Unreachable today because keepDeletes is hardcoded true, but the run encoding could not represent "not written" and the corruption would have been silent.
• Cancelling mid-batch returns no runs, since the writer is discarded anyway.
• RowDelta now asks Iceberg to validate what positional deletes depend on: validateFromSnapshot, validateDeletedFiles, and validateDataFilesExist over the referenced data files. assertRowIndexCurrent only covers a rewrite landing before the pre-check; one landing between the refresh and the catalog commit previously produced deletes that resolved to nothing — silent duplicates rather than a failed commit. Equality mode is deliberately left alone: its deletes match on key, and the positional deletes it occasionally emits address rows written in the same commit, so nothing concurrent can invalidate them.

Testing

Unit — 19 tests

Against a real Iceberg table on a local Hadoop catalog. Rows are committed through RowDelta and read back with a normal scan, so Iceberg decides whether each delete applied rather than the test asserting on writer internals.

• unpartitioned inserts; positional delete superseding a committed row
• same key updated twice and three times within one batch
• same key updated across batches (chained: each supersedes the previous batch's row)
• per-batch supersede state released, driven through the real addToTablePerSchema loop over two batches on one uncommitted writer session
• partitioned fan-out with interleaved records — one file per partition, correct row counts
• partitioned delete carries the target's partition, and is file-scoped
• partitioned same-key updates in one batch, interleaved with another key
• 10k-row batch: 1000 keys × 10 rounds → exactly 1000 live rows, last write wins
• file rolling: 5000 rows at a 4 KB target, then superseding every row across the rolled files
• partition-changing update supersedes the old row
• the delta reports the data files its deletes depend on, across partitions
• validationRefusesDeletesWhoseDataFileWasRewritten — commits a rewrite, asserts ValidationException
• equality mode still produces EQUALITY_DELETES and does not use the new writer
• WriteRun shape: contiguous writes collapse to one run; interleaved partitions still map every record to exactly one (path, position)

End to end — Postgres → Iceberg, --delete-type pos

Scenario	Result
Unpartitioned CDC, 3 rounds	6 rows / 6 distinct ids, POSITION_DELETES only
Partitioned fan-out (/{region, identity})	3 partition dirs (2/1/2), interleaved records routed correctly
Positional deletes correct	read the delete Parquet directly — each targets the exact (path, pos) of the superseded row
Same key, same batch (2× and 3×)	one live row, newest wins
Same key, across batches	chained r → v2/v3 → v4, one live row
12k-record batch, 3000 keys × 4 rounds	50000 rows / 50000 distinct ids, 0 stale intermediates, 12000 delete records
Cursor-based incremental	5 rows / 5 distinct ids, updated row superseded
Chunked backfill — 4 chunks	20000 / 20000 across four concurrent chunk writers
Chunked table + CDC (13382 records)	20000 / 20000, 6666 final versions, 0 stale, 50 rows moved partitions cleanly
Soft delete	tombstone written, prior row superseded
Equality-mode regression	separate table, --delete-type eq: equality deletes still produced, no row-index directory created for the stream

Notes for reviewers

Two findings from my earlier review were wrong and are retracted here, both verified in the files:

• Partition-changing updates do not lose their delete. IcebergUtil sets write.metadata.metrics.column.file_path=full, which with DeleteGranularity.FILE makes deletes file-scoped, so Iceberg matches them to data files by path, not partition. Confirmed in e2e: a delete file physically in region=us targeting a region=in data file was applied.
• Consequently, switching DeleteGranularity to PARTITION is not the easy lever it looks like. A partition-granularity file spans many data files, its file_path bounds widen, it stops being file-scoped, and partition-changing updates silently break. FILE granularity is load-bearing for correctness. (The file-count cost turned out modest in practice — 12k updates over 3000 keys produced 2 delete files.)

Method references to RollingFileWriter do not link — close() is inherited from the package-private BaseRollingWriter, so RollingFileWriter::close dies at runtime with LambdaConversionException. close() uses plain loops with suppressed-exception accumulation.

Still open, both Go-side and out of scope here:

• Append writers ignore deleteFilePath, and Go's NeedsRowIndex does not check identifier fields or upsert — so no_identifier_fields + pos silently produces duplicates. Wants a config-validation refusal.
• opType != "r" skips the index lookup, so a backfill re-run into a non-empty table produces duplicates where equality mode produced none. One line in each writer.

Unrelated, found while testing: protocol/sync.go loads state only when --state is passed. Without it the file is written but never read back, so PreCDC calls AdvanceLSN on every run and CDC reads zero records. The e2e procedure in .cursor/rules/olake.mdc §6 omits --state, which makes CDC untestable as written.

Tests are not in this branch — .gitignore has a bare test pattern that matches any directory named test, including src/test, so the Java test tree is ignored repo-wide. Worth narrowing separately.