[taoran92]

Purpose

Fixes #11128.

With write-only=true and write.sequence-number-init-mode=snapshot, restarting a dynamic-bucket writer can replace an existing HASH index with only the current write's hashes. Later updates can then reach a different bucket and leave duplicate primary keys in query results.

Restore indexes independently of the data-file metadata fast path, retain data metadata required by primary-key indexes, and preserve explicit overwrite semantics. Pass the metadata-scan flag through the Flink write coordinator as well.

Tests

• Core regressions: 22 tests passed, covering HASH retention, bucket assignment, overwrite, sequence initialization, and file-backed deletion-vector/primary-key index restoration.
• TableWriteCoordinatorTest: 9 tests passed, including paginated index restoration with and without data-file metadata scanning.

[JingsongLi]

Requirement fit: SUPPORTED. Implementation: FINDINGS.

The HASH/index restoration fix addresses a real duplicate-primary-key failure. Normal verification passed 59 core tests and the 9 coordinator tests on both Flink 1 / JDK 8 and Flink 2 / JDK 17. Exact old/new coordination-request serialization probes also passed. One extension compatibility issue remains below.

[JingsongLi]

[P2] Preserve the existing WriteRestore extension method

Replacing the five-argument method with a new abstract six-argument method breaks existing WriteRestore implementations supplied through withWriteRestore, including thin wrappers around the built-in restore. When the new AbstractFileStoreWrite calls such a precompiled implementation, it throws AbstractMethodError before restoring the bucket. Existing binaries that invoke the former five-argument entry point instead throw NoSuchMethodError. I reproduced both directions on JDK 8, and confirmed an actual downstream compaction wrapper still implements/calls the former signature.

Please retain a compatible five-argument entry point and bridge the new overload (older implementations can safely keep scanning data metadata), or explicitly coordinate the consumer migration before rollout. The backward-compatible serialized request field does not cover this Java extension-method change.

[taoran92]

@JingsongLi Thanks for reviewing! agreed. I have restored the five-argument overload and added a default bridge for compatibility, with regression tests for legacy wrappers. PTAL.