[FLINK-37367][state/forst] Make ForSt inherit uploaded SST files after restorations (#26202)

Author: AlexYinHan

flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java

@@ -263,10 +263,16 @@ public ForStKeyedStateBackend<K> build() throws BackendBuildingException {
             defaultColumnFamilyHandle = restoreResult.getDefaultColumnFamilyHandle();
             nativeMetricMonitor = restoreResult.getNativeMetricMonitor();
 
-            // TODO: init materializedSstFiles and lastCompletedCheckpointId when implement restore
             SortedMap<Long, Collection<IncrementalKeyedStateHandle.HandleAndLocalPath>>
                     materializedSstFiles = new TreeMap<>();
             long lastCompletedCheckpointId = -1L;
+            if (restoreOperation instanceof ForStIncrementalRestoreOperation) {
+                backendUID = restoreResult.getBackendUID();
+                lastCompletedCheckpointId = restoreResult.getLastCompletedCheckpointId();
+                if (recoveryClaimMode != RecoveryClaimMode.NO_CLAIM) {
+                    materializedSstFiles = restoreResult.getRestoredSstFiles();
+                }
+            }
 
             snapshotStrategy =
                     initializeSnapshotStrategy(

flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/CopyDataTransferStrategy.java

@@ -31,7 +31,6 @@
 import org.apache.flink.runtime.state.StreamStateHandle;
 import org.apache.flink.runtime.state.filesystem.FileStateHandle;
 import org.apache.flink.state.forst.fs.ForStFlinkFileSystem;
-import org.apache.flink.state.forst.fs.filemapping.FileOwnership;
 import org.apache.flink.state.forst.fs.filemapping.MappingEntry;
 import org.apache.flink.util.IOUtils;
 import org.apache.flink.util.Preconditions;
@@ -115,10 +114,6 @@ private HandleAndLocalPath copyFileToCheckpoint(
                     ((ForStFlinkFileSystem) dbFileSystem).getMappingEntry(dbFilePath);
             Preconditions.checkNotNull(mappingEntry, "dbFile not found: " + dbFilePath);
             sourceStateHandle = mappingEntry.getSource().toStateHandle();
-            if (mappingEntry.getFileOwnership() == FileOwnership.NOT_OWNED) {
-                // The file is already owned by JM, simply return the state handle
-                return HandleAndLocalPath.of(sourceStateHandle, dbFilePath.getName());
-            }
         } else {
             // Construct a FileStateHandle base on the DB file
             FileSystem sourceFileSystem = dbFilePath.getFileSystem();

flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileMappingManager.java

@@ -165,6 +165,10 @@ public List<String> listByPrefix(String path) {
      * @return always return true except for IOException
      */
     public boolean renameFile(String src, String dst) throws IOException {
+        if (src.equals(dst)) {
+            return true;
+        }
+
         MappingEntry srcEntry = mappingTable.get(src);
         if (srcEntry != null) { // rename file
             if (mappingTable.containsKey(dst)) {