[flink] Add restore_as_latest procedure

[zhuxiangyi]

Purpose

This PR adds a non-destructive restore procedure for Flink:

CALL sys.restore_as_latest(`table` => 'default.T', snapshot_id => 3);
CALL sys.restore_as_latest(`table` => 'default.T', tag => 'tag-1');

Unlike rollback_to, this procedure restores the table to the state of a target snapshot or tag by creating a new latest snapshot. Later snapshots and tags are preserved.

What is changed

• Add RestoreAsLatestProcedure and register it in the Flink procedure factory list.
• Add commit support to create a new latest snapshot from the complete data manifests of the target snapshot.
• Add IT coverage for restoring from snapshot and tag, preserving later snapshots, and writing after restore.
• Document the procedure in Flink procedures and snapshot/tag maintenance docs.

Tests

mvn -pl paimon-flink/paimon-flink-common -am -Pfast-build -DfailIfNoTests=false -Dtest=RestoreAsLatestProcedureITCase test
git diff --check

Notes

This PR is not associated with an issue yet. If the community prefers following the discussion-first flow strictly, I can open or join an issue/discussion and adjust the design accordingly.

[JingsongLi]

I think restoreAsLatest can be invisible to streaming readers that handle overwrite snapshots.

The new snapshot writes the target snapshot's files into the base manifest list, but writes an empty delta manifest list and marks the commit as CommitKind.OVERWRITE (FileStoreCommitImpl.java:1174-1200). DataTableStreamScan first handles overwrite snapshots via the overwrite-change path, and if the returned plan is empty it advances past the snapshot. Since the restore snapshot has no delta, a streaming reader with streaming-read-overwrite=true can skip the restore entirely, missing both files/rows that should be removed from the current latest snapshot and files/rows that should be restored from the target snapshot.

Could restoreAsLatest produce an overwrite delta relative to the previous latest snapshot (DELETE previous-only files and ADD target-only files), or introduce a dedicated commit kind/streaming-scan handling for restore snapshots?

[zhuxiangyi]

@JingsongLi
Thanks for pointing this out. I agree that the current restore snapshot is not sufficient for streaming readers with streaming-read-overwrite=true.

The new snapshot currently has the target snapshot's complete data manifests in baseManifestList, but its deltaManifestList is empty. This makes the final table state correct for batch/full scans, but the restore can be invisible to streaming overwrite readers.

I will update restoreAsLatest to generate an overwrite delta relative to the previous latest snapshot: DELETE files that exist only in the previous latest snapshot, and ADD files that exist only in the target snapshot. The baseManifestList will contain the previous latest snapshot's merged effective ADD files, while deltaManifestList will describe the previous-latest-to-target transition.

[zhuxiangyi]

@JingsongLi
Updated in the latest commit. restoreAsLatest now writes an overwrite delta from the previous latest snapshot to the target snapshot: DELETE files that exist only in the previous latest snapshot and ADD files that exist only in the target snapshot.

I also added IT coverage to verify both DELETE-only and ADD-only restore deltas.

[JingsongLi]

This restore commit bypasses the normal post-commit callbacks. Regular commits call commitCallbacks with the committed snapshot and delta files after commitSnapshotImpl succeeds, and those callbacks keep external state in sync (for example Iceberg compatibility metadata uses context.snapshot/context.deltaFiles, and chain-table overwrite handling reacts to CommitKind.OVERWRITE). restoreAsLatest also changes the table state with an overwrite delta, but it returns immediately after writing the Paimon snapshot, so those external views can remain at the pre-restore state. Could we trigger the same commit callback path after a successful restore, using the restored base/delta/index files and the new snapshot context?

[zhuxiangyi]

Good catch, thanks. restoreAsLatest committed via commitSnapshotImpl directly and skipped the commit callbacks, so external views (Iceberg metadata, chain-table overwrite) could stay at the pre-restore state.

Fixed by notifying the callbacks after a successful restore, like a regular commit. The context uses the restore delta (DELETE previous-only + ADD target-only) and an index delta derived the same way from the previous-latest and target index manifests. Both callbacks are idempotent, so retries stay correct.

Added an IT case testRestoreTriggersCommitCallback asserting Iceberg metadata is generated for the restore snapshot.