Flink: Fix deleted rows reappearing after a failed eq-delete conversion cycle

[vishnuprakaz]

If a conversion cycle fails after it has resolved an equality delete, the rows that delete was hiding come back and they stay back. The task reports success, so nothing looks wrong.

To reproduce it the table has two rows. I commit an equality delete for each of them in one snapshot, so both rows should be invisible, then remove row two's delete file from disk. The cycle resolves row one's delete and aborts when it cannot read row two's, so nothing is committed. I put the missing file back and trigger again. The cycle succeeds this time, but row one is visible again:

TestConvertEqualityDeletes > testDeleteResolvedBeforeFailureIsRetained() FAILED
    org.opentest4j.AssertionFailedError:
    expected: []
     but was: [Record(1, a)]

Row one's delete had already resolved during the failed cycle, and that is the one that gets lost. Row two's delete never got that far, and it converts correctly on the retry. That asymmetry is the bug. Resolving an equality delete consumes the index entries it matched, and a failed cycle leaves them consumed. Nothing puts them back, because the target branch never moved, and both the per-key clear and the planner's reindex only trigger when it does. So the retry resolves the same
delete against an index that no longer holds those rows, writes no deletion vector, and the committer then removes the equality delete file and marks the snapshot converted.

The planner already rebuilds the index when an external commit has moved the target branch. This is that situation without the move, so the planner now remembers which staging snapshot its last plan covered. If it picks the same one again, that cycle never committed, and the index is rebuilt before the delete phase.
The snapshot id is checkpointed so a restore taken mid-cycle is covered too, and it is recorded only once the staging inputs have validated.

Two planner tests ran several triggers without simulating the converter's own commit, so their no-re-emission assertions no longer hold once the index is rebuilton a replan. Both now simulate that commit, the way noMainReEmitAfterOwnCommit does, which made one of them identical to it, so I folded it in happy to restore it if you would rather keep both.

Also updates two statements in the maintenance docs that this change makes out of date. Needs a backport to 2.0 and 1.20.

The alternative I considered was making resolution non-destructive in the index, which changes what three of the PKIndex tests assert.

This is the smallest fix I found that leaves the index semantics as they are.

Happy to rework on this if there is a better way.

[vishnuprakaz]

cc: @mxm @pvary

[mxm]

Thanks for the PR @vishnuprakaz!

The core issue is that the planner does not get notified about downstream failures and that the index already gets modified before the cycle completes. While your fix looks correct, it also comes at the cost of a full re-index on failures. It would be better to buffer index modifications until the planner reads the committed snapshot. Only then, we can safely apply the index modifications without a full re-index.

How did you discover this issue? If the transient read failures are more common, it definitely makes sense to only modify the index after everything has been committed, to avoid frequent rebuilds of the index.

[mxm]

Let's rename sequenceNumber to generation and remove the Max.max logic, i.e. indexSequenceNumber = mainSnapshot.generation() + 1. The sequence number we used initially but the logic has diverted from it.

[mxm]

Suggested change

	private transient Long indexedSequenceNumber;
	private transient Long indexGeneration;

[mxm]

Should this be:

Suggested change

	private transient Long plannedStagingSnapshotId;
	private transient Long pendingStagingSnapshotId;

[mxm]

Should this be:

Suggested change

	private transient Long plannedStagingSnapshotId;
	private transient Long pendingStagingSnapshotId;

[vishnuprakaz]

Thanks for taking a look @mxm!

Actually no incident behind it,

I was following the conversion work and was reading through the operators to understand how a cycle works. I ended up wondering what happens when a cycle fails partway through. There's a retry test for that, but the failure comes before the deletes resolve, so I couldn't tell what happens when it comes after.

So I went back through the series to see whether it had come up before, and found #17038. Two cycles ended up running on the same uncommitted snapshot there, and one committed its marker without the DV. Same result as what I was worried about, just a different cause so I wanted to see whether one failing cycle could do it alone. That's what the test in the PR checks.

so I don't have numbers on how often cycles fail. My guess is occasional rather than frequent a transient read error, or a checkpoint landing mid cycle on a restart.

Agreed on the buffering,I think the planner needs the pending snapshot either way to tell the index whether to apply or discard.