[LAKE-6732] Spark 3.5: refuse to rewrite a table with duplicate data file registrations

[abrarsher23]

Problem

A file — data or delete — registered at two data sequence numbers cannot be compacted
safely. This is not something a healthy writer produces: it arises when a commit is retried after
its outcome became unknown (CommitStateUnknownException) and the retry re-registers a
WriteResult that had in fact already been applied server-side. Root cause of INC-7208 (SEV1).

File identity in the rewrite path is keyed on location alone, while a file's data sequence
number lives on the manifest entry rather than the ContentFile, so the two registrations
cannot be told apart:

• Data side removes 1-of-2. DataFileSet keys on location(). Two scan tasks for one path
  collapse to a single DataFile carrying a single manifestLocation(); the other manifest is
  never opened.
• Delete side removes 2-of-2. DeleteFileSet keys on (location, contentOffset, contentSizeInBytes) — identical across two registrations of one physical file. Removing one
  registration removes its sibling too.

Net effect: a surviving data registration left with no delete coverage, and rows that were
correctly suppressed become visible.

Observed in production: one physical file, 30 records, registered at seq 93949 and seq 93950.
rewrite_data_files removed only the 93949 registration; remove_dangling_deletes then removed
both registrations of both its delete files — including a position delete its own test
correctly judged not dangling. 23 rows across 6 keys surfaced.

Change

Two properties, both default true:

• validate-duplicate-file-registrations — detect any live file path (data or delete)
  registered at more than one distinct data sequence number.
• resolve-duplicate-file-registrations — repair it instead of failing: keep the registration
  at the LOWEST live data sequence number per duplicated path, drop every other, as one
  metadata-only commit, then continue.

Implemented as a new primitive on ManifestFilterManager (dropDuplicateRegistrations) plus a
small public commit class (DuplicateRegistrationRepair), because the public RewriteManifests
API cannot express this — its own contract forbids added/deleted files in a rewritten manifest.

Detection and repair live in a shared DuplicateFileRegistrationGuard, called from three
actions:

Action	Why it needs the guard
RewriteDataFilesSparkAction	the original path; the 1-of-2 data-side removal
RewritePositionDeleteFilesSparkAction	run standalone from oncall/backfill notebooks, with no dependency on rewrite_data_files having run
RemoveDanglingDeletesSparkAction	reproduces the delete-side 2-of-2 removal (see the A/B below)

RemoveDanglingDeletes takes a package-private skipDuplicateRegistrationCheck() so the
rewrite_data_files path, which already ran the guard, doesn't re-scan entries.

Keeping the lowest sequence number is not an arbitrary choice: any delete file committed between
two duplicate registrations has a sequence number greater than the lower one, so it continues to
cover a registration kept at the lower sequence number. Keeping the higher one instead would
silently drop that coverage and reproduce the exact defect this repairs.

Design notes

• Repairs by default, throws on opt-out. A guard that's off by default wouldn't have caught
  this. Repair defaults to on so an affected table self-heals instead of blocking every subsequent
  scheduled compaction run. Escape hatches retained: validate-duplicate-file-registrations=false
  skips the check entirely; resolve-duplicate-file-registrations=false keeps detection but
  reverts to the original fail-fast behavior.
• Rejects contradictory intent. A commit that both designates a survivor for a path AND
  separately asks for that same path to be deleted outright is rejected outright
  (validateNoContradictoryDuplicateRegistrationIntent), rather than letting the contradiction
  resolve silently in the destructive direction.
• Retry-safe. If a concurrent commit changes the intended survivor's path between the
  detection scan and this repair's commit, the repair aborts loudly
  (validateDuplicateRegistrationSurvivorsPresent) rather than risking a silent drop of every
  live registration of that path. A subsequent retry re-scans against the latest state.
• V3 deletion vectors are refused, not guessed at. A single Puffin file legitimately holds
  several distinct DVs sharing one on-disk path; this guard's matching is location-only, so it
  cannot always tell that apart from a genuine duplicate registration. Refuses to auto-repair the
  whole batch when a duplicated path might be a DV, regardless of the caller's resolve option.
  Defence-in-depth rather than a live path: Iceberg's own BaseRowDelta#validate already refuses
  to add a second DV for a data file that already has one.
• Known gap, tracked as a fast-follow: the only operational signal today is a single
  LOG.warn line — there is no metric/alert distinguishing "repaired N paths" from "found none,"
  and no way to notice a table repairing on every run (which would mean a writer bug is still
  actively producing duplicates, not a one-off). File a follow-up to add a counter/alert on repair
  frequency per table before this sees wide rollout.
• where/filter scope is ignored by the guard, so partition-scoped compaction pays a full-table
  entries scan whenever it runs. Cost is unquantified; no metric exists yet for how often the
  guard actually fires in prod.

Relationship to #4

Complementary, not overlapping. #4 (already merged on this base) makes remove_dangling_deletes
err safe by retaining any delete file attached to a live FileScanTask — see the A/B below for
exactly how much that holds back on its own. It does not prevent the initial doubling, which
happens in rewrite_data_files before the dangling action ever runs. This PR stops the rewrite
from starting at all, and repairs the underlying cause instead of only refusing to proceed.

A/B: does #4 fix the delete-side half on its own? — Partially

A duplicated delete-file registration (this PR's target) is not the same failure mode as a
duplicated data-file registration alone. On a fixture with a duplicated data registration but no
duplicated delete, #4's scan-based dangling-delete detection and the old min-sequence heuristic
come out identical — neither wrongly drops a needed delete, because a needed delete requires some
live X@x with x <= d, which forces min_data_seq <= d, while "dangling" requires d < min_data_seq; both cannot hold at once. The pre-#4 rule never wrongly removes a needed delete
on its own — the trigger requires a duplicated delete registration specifically, which is
exactly the "delete side removes 2-of-2" scenario this PR's guard targets in
RemoveDanglingDeletesSparkAction.

Testing

• TestDuplicateRegistrationRepair (new, core-level): 66 tests, 0 failures. Covers argument
  validation, the keep-lowest-sequence-number invariant (including a test pinning that the
  primitive honours a wrong caller choice verbatim rather than silently correcting it — that
  is the caller's contract by design), 3+ registrations collapsing to one, delete-file repair,
  that repairing a data file does not strip its own delete coverage, and retry safety against a
  concurrent commit changing the intended survivor between scan and commit.
• TestDuplicateFileRegistrationGuard (new): 5 tests, 0 failures. Direct unit coverage of the
  guard's V3-deletion-vector refusal decision function — this branch can't be driven end-to-end
  through any action, since Iceberg itself refuses to create the state it defends against (a
  second live DV for a data file that already has one).
• TestRemoveDanglingDeleteAction: 12 tests, 2 skipped, 0 failures. One pre-existing test's
  fixture incidentally re-registers two data files at a second sequence number; the guard now
  repairs that before the dangling-delete pass runs, so its post-repair assertions were updated
  accordingly. Added a new test (testDoesNotDropADuplicatedDeleteFileThatStillCoversLiveData)
  driving the delete-side defect deliberately rather than incidentally.
• TestRewriteDataFilesAction (full suite): 122 tests, 2 skipped, 0 failures. Includes
  testRejectsDuplicateDataFileRegistrations, testRepairsDuplicateDataFileRegistrations,
  testRepairsDuplicateDeleteFileRegistrations, testRepairsMultipleDuplicateFileRegistrationsInOneBatch,
  testRepairsMixedDataAndDeleteFileRegistrationsTogether, testRepairsMoreThanTwoRegistrationsOfSamePath,
  testRepairThenCompactEndToEnd, testDuplicateDataFileRegistrationCheckCanBeDisabled,
  testAcceptsTableWithoutDuplicateRegistrations.
• TestRewritePositionDeleteFilesAction (full suite): 19 tests, 0 failures. Includes standalone
  repair/reject tests (no prior rewrite_data_files call on the table) and a pinned test
  confirming Iceberg itself refuses to create a duplicate DV registration.

spotlessCheck and checkstyle clean on every file this PR touches.
iceberg-core:checkstyleTest currently fails on this branch, but only on
core/src/test/java/org/apache/iceberg/rest/* — unused imports and annotation style, pre-existing
since #1 merged and unrelated to this change.

Known limitations (tracked, not blocking)

• The guard runs before the unknown-option check, so a typo'd option commits a repair before
  throwing.
• Repair always targets main, ignoring toBranch(). Not reachable through
  table_compaction.py today (it never emits branch =>); live only if branch-scoped compaction
  is adopted later.
• The survivor check can pass with zero surviving registrations (robustness, not correctness).
• RemoveDanglingDeletes hardcodes the guard true,true with no opt-out reachable externally —
  RemoveDanglingDeleteFiles's API has no option-plumbing convention today, so exposing a toggle
  here means introducing that pattern for the first time, not reusing one.
• A table that trips the DV-duplicate refusal is permanently un-compactable via auto-repair; no
  stored procedure exists to clear that state. validate=false is not a safe rollback for such a
  table — it re-enables the exact pre-PR code path that caused the incident, on a table now
  known to have a live duplicate registration.

Upstream

Not fixed upstream for this defect. RepairManifests/RepairTable-shaped proposals have been
raised repeatedly since 2021 (apache/iceberg#2435, #2608, #10445, #10784), all closed
unmerged or still open. apache/iceberg#17399 ("API: Define RepairManifests action interface")
merged into the 1.12.0 milestone, but scoped to file-metric/stat repair (record_count,
file_size_in_bytes, bounds/null counts), not duplicate sequence-number registrations — a
same-named but functionally different problem. apache/iceberg#17198 added a "Content file
uniqueness" spec section stating duplicate live entries are undefined behavior and that writers
are not required to validate uniqueness at commit time — consistent with this being a
read/compaction-layer repair, not a write-time one.

Relationship to #8

This is the 1.8.1-affirm-patch / spark/v3.5 counterpart of #8, which ported this same guard
onto 1.11.0-affirm-patch / spark/v4.0 and evolved it further (the three-action wiring, the
shared DuplicateFileRegistrationGuard class, the core TestDuplicateRegistrationRepair suite,
and several self-review rounds) while this branch sat untouched. This PR ports that evolution back
here rather than leaving the two forks to drift, adapting for real differences between the two
bases:

• This branch's RewritePositionDeleteFilesSparkAction refuses to run against a V3 table at all
  (no V2→V3 DV-rewrite capability here), so TestRewritePositionDeleteFilesAction's
  DV-duplicate-refusal test generates the DV directly via FileGenerationUtil instead of via a
  rewrite, matching TestRemoveDanglingDeleteAction's existing pattern — same assertion, without
  depending on a feature this base doesn't have.
• Found and fixed a real bug in Spark 4.0: refuse-or-repair duplicate data file registrations before compaction #8's own code while porting it: DuplicateFileRegistrationGuard
  on the 1.11 branch has a comment stating validateNoContradictoryDuplicateRegistrationIntent
  must read callerDeletedFilePaths (not deleteFiles, which the filtering loop also writes to
  and never clears between commit attempts) — but the code there still reads deleteFiles. This
  port implements what the comment specifies, not what Spark 4.0: refuse-or-repair duplicate data file registrations before compaction #8's code currently does. Worth a look on
  Spark 4.0: refuse-or-repair duplicate data file registrations before compaction #8 too.

[abrarsher23]

The actual repair mechanism. Everything else in this diff is plumbing to get a path -> sequence number to keep map to this point and commit the result.

Why a new predicate was necessary rather than reusing delete(CharSequence)/delete(F file) on this same class: both existing predicates match by location only — for a path with two live registrations, either one would delete both, not the one you want to drop. This is the only predicate in the class that's sequence-number-aware.

Why this isn't built on the public RewriteManifests API instead: that interface's own contract says "the added manifest cannot contain new or deleted files" — it can reshuffle which manifest an entry lives in, but it structurally cannot change an entry's live/dead status. That ruled it out.

[abrarsher23]

Fixed during self-review: this was originally an if / else if chain. If deleteFiles were ever populated on the same ManifestFilterManager instance as duplicateRegistrationKeepSequence (not possible today — the only caller, DuplicateRegistrationRepair, calls nothing else — but this class is shared/reusable core internals), the deleteFiles partition-overlap check would short-circuit past a manifest that actually needed opening for a duplicate registration, silently under-scanning it. Made the checks independent instead of else-if so this can't happen regardless of what future caller combines with this predicate.

[abrarsher23]

This capture point matters and used to be wrong. startingSnapshotId is what planFileGroups() scans as-of a few lines down, and what the final compaction commit uses as its base. It used to be captured before the duplicate-registration check ran (top of execute()). If a repair commits a new snapshot, capturing this first would mean the scan and the final commit both silently plan against the pre-repair snapshot — the repair would run, but have zero effect on the compaction that immediately follows it. Moved to after the check for that reason.

[abrarsher23]

Routes each duplicated path to keepDataFile/keepDeleteFile based on content, then a single repair.commit() below covers the whole batch — multiple distinct duplicated paths found in one scan land in one snapshot, not one commit per path (see testRepairsMultipleDuplicateFileRegistrationsInOneBatch). table.refresh() right after is what makes the startingSnapshotId reordering fix (further up in execute()) actually take effect.

[abrarsher23]

Modeled directly on StreamingDelete (same three overrides: self()/operation()/validate(), nothing else). One deliberate difference: this constructor is public, where StreamingDelete's is protected — StreamingDelete is only ever constructed from inside org.apache.iceberg (e.g. BaseTable.newDelete()), but this needs to be instantiated directly from RewriteDataFilesSparkAction in the spark.actions package, via ((HasTableOperations) table).operations() — same pattern already used by RewriteManifestsSparkAction.

[abrarsher23]

The one invariant that matters for correctness here: sequenceNumber MUST be the lowest live one for path, not an arbitrary choice. Any delete file committed between two duplicate registrations has a sequence number greater than the lower one, so it still covers a registration kept at the lower sequence number. Keeping the higher one instead silently drops that delete coverage and reproduces the exact incident this PR fixes (resurrected rows that were correctly suppressed).

Not validated here — the only caller (RewriteDataFilesSparkAction) computes this via SQL min(sequence_number), so it's correct by construction today, but nothing in this class or ManifestFilterManager would catch a future caller passing a non-minimum value. Flagged in the javadoc as a caller-enforced-only contract.

[rayhondo]

Reviewed against the linked incident doc ("Anatomy of one duplicate row") and stress-tested the repair logic for concurrent-write and format-version edge cases. The fix correctly closes the mechanism the incident traced through rewrite_data_files on V2 tables. Left inline notes on gaps found during that pass -- scoped down after confirming actual call-site usage in all-the-things (V1 and standalone remove_dangling_deletes/COW-DML concerns don't apply to how these tables are actually used; the rewrite_position_delete_files and batch-repair-atomicity notes do).

[rayhondo]

This guard is scoped to RewriteDataFilesSparkAction only. RewritePositionDeleteFilesSparkAction hits the identical bug -- DeleteFileSet collapses duplicate registrations by location() the same way -- and it's not just theoretically reachable: we already run it standalone in production oncall/backfill notebooks, unconnected to any rewrite_data_files call (e.g. an executed CALL prod__lake__us.system.rewrite_position_delete_files(...) against chrono.ml_logs_model_run). We don't run COW DML against these tables and don't invoke remove_dangling_deletes standalone (it's always an option flag inside rewrite_data_files), so those aren't live concerns -- but oncall will keep reaching for other cleanup procs by hand, and none of them are protected today. Recommend pulling the detection+repair into a shared entry point (e.g. a helper both actions call, or a check inside ManifestFilterManager/MergingSnapshotProducer itself) rather than duplicating this guard per-action, or at minimum a fast-follow ticket to wire RewritePositionDeleteFilesSparkAction into the same check.

[rayhondo]

The repair folds every duplicated path from one scan into a single commit, and this method aborts the entire commit if even one path's intended survivor isn't observed live on a CAS retry (e.g. a legitimate concurrent write touched just that one path in the meantime). On a table with many duplicated paths, one unrelated concurrent write blocks repair of every other still-valid path in that batch, forcing rewrite_data_files to fail outright instead of repairing what it safely can. Worth considering whether the repair should drop the affected path from this attempt and retry, rather than failing the whole batch.

[rayhondo]

This method's "mark path as seen" side effect (duplicateRegistrationSurvivorsSeen.add(path)) only fires if it's actually evaluated -- callers use it inside a || chain in manifestHasDeletedFiles/filterManifestWithDeletedFiles, so if deletePaths/deleteFiles/dropPartitions already matches the same entry earlier in that chain, Java short-circuits and this side effect never runs. Not reachable via DuplicateRegistrationRepair today (it never populates those other fields on the same instance), but it's a landmine for the next caller that combines dropDuplicateRegistrations() with an ordinary delete()/dropPartition() targeting the same path -- that commit would wrongly throw ValidationException claiming the survivor was never observed. Worth a comment or an assertion guarding this invariant so it doesn't surprise the next person.

[rayhondo]

Minor: hand-rolls if (!missing.isEmpty()) { throw new ValidationException(...); } instead of this file's existing ValidationException.check(condition, msg, args) helper used elsewhere (e.g. validateRequiredDeletes). No functional difference, just a second idiom for the same check-and-throw pattern in a file that already has one convention.

[rayhondo]

Minor: reinvents entries-table grouping via a UUID-named temp view + raw SQL string. The sibling RemoveDanglingDeletesSparkAction (lines ~120-130) already does the identical "group by key, take min" shape via the Dataset API (groupBy(...).agg(min(...))) -- no temp view, compile-time column safety, no extra catalog round-trip.

[rayhondo]

Delete-file duplicate-registration repair is only tested for format version 2 (assumeThat(formatVersion).isEqualTo(2) here and at line 2401). DeleteFileSet keys equality on location() + contentOffset() + contentSizeInBytes() for v3 deletion vectors, but the repair's own matching (ManifestFilterManager.isDuplicateRegistrationToDrop) keys only on location(). We're V2-only today but expect to move up -- this needs either v3 coverage before that migration or an explicit tracking ticket, since right now it's genuinely unverified for v3, not just an untested-but-equivalent path.

[rayhondo]

Okay for now as all are on v2 but this needs to be addressed for v3