You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "AlexYinHan (via GitHub)" <gi...@apache.org> on 2025/01/08 12:35:57 UTC
[PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
AlexYinHan opened a new pull request, #25924:
URL: https://github.com/apache/flink/pull/25924
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
This PR implements fast snapshot/restore for ForSt. Specifically, it implements different strategies for ```ForStStateDataTransfer```, so that ForSt can reuse the checkpoint files as much as possible and thus reduce the cost of file copying.
## Brief change log
- Enhance the ```FileMappingManager``` so it can track the ownership of files
- Implement different strategies for ```ForStStateDataTransfer```
## Verifying this change
This change added tests and can be verified as follows:
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (not applicable)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "AlexYinHan (via GitHub)" <gi...@apache.org>.
AlexYinHan commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1917813829
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileOwnershipDecider.java:
##########
@@ -0,0 +1,70 @@
+package org.apache.flink.state.forst.fs.filemapping;
+
+import org.apache.flink.core.execution.RecoveryClaimMode;
+import org.apache.flink.core.fs.Path;
+
+public class FileOwnershipDecider {
+
+ public static final String SST_SUFFIX = ".sst";
+
+ private final RecoveryClaimMode recoveryClaimMode;
+ private final boolean isDbPathUnderCheckpointPath;
+
+ public static FileOwnershipDecider getDefault() {
+ return new FileOwnershipDecider(RecoveryClaimMode.DEFAULT, false);
+ }
+
+ public FileOwnershipDecider(
+ RecoveryClaimMode recoveryClaimMode, boolean isDbPathUnderCheckpointPath) {
+ this.recoveryClaimMode = recoveryClaimMode;
+ this.isDbPathUnderCheckpointPath = isDbPathUnderCheckpointPath;
+ }
+
+ public boolean isDbPathUnderCheckpointPath() {
+ return isDbPathUnderCheckpointPath;
+ }
+
+ public FileOwnership decideForNewFile(Path filePath) {
+ if (!isDbPathUnderCheckpointPath) {
+ return FileOwnership.PRIVATE_OWNED_BY_DB;
+ }
+
+ // only SST files are shareable
+ return isSstFile(filePath)
+ ? FileOwnership.SHAREABLE_OWNED_BY_DB
+ : FileOwnership.PRIVATE_OWNED_BY_DB;
+ }
+
+ public FileOwnership decideForRestoredFile(Path filePath) {
+ if (!isDbPathUnderCheckpointPath) {
+ return FileOwnership.PRIVATE_OWNED_BY_DB;
+ }
+
+ // only SST files are shareable
+ if (isSstFile(filePath)) {
+ return recoveryClaimMode == RecoveryClaimMode.CLAIM
+ ? FileOwnership.NOT_OWNED
+ : FileOwnership.SHAREABLE_OWNED_BY_DB;
+ } else {
+ return FileOwnership.PRIVATE_OWNED_BY_DB;
+ }
+ }
Review Comment:
Yeah this is a mistake. I'll fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #25924:
URL: https://github.com/apache/flink/pull/25924#issuecomment-2601472212
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "AlexYinHan (via GitHub)" <gi...@apache.org>.
AlexYinHan commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1917715662
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyBuilder.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst.datatransfer;
+
+import org.apache.flink.core.execution.RecoveryClaimMode;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Builder for {@link DataTransferStrategy}. */
+public class DataTransferStrategyBuilder {
+ private static final Logger LOG = LoggerFactory.getLogger(DataTransferStrategyBuilder.class);
+
+ private Path dbRemotePath;
+
+ private FileSystem dbDelegateFileSystem;
+
+ private boolean isDbPathUnderCheckpointPath;
+
+ private RecoveryClaimMode claimMode;
+
+ public void withoutDbRemotePath() {
+ this.dbRemotePath = null;
+ this.dbDelegateFileSystem = null;
+ this.claimMode = RecoveryClaimMode.DEFAULT;
+ }
+
+ public void withDbRemotePath(FileSystem dbDelegateFileSystem, Path dbRemotePath) {
+ this.dbRemotePath = dbRemotePath;
+ this.dbDelegateFileSystem = dbDelegateFileSystem;
+ }
+
+ public void withIsDbPathUnderCheckpointPath(Boolean isDbPathUnderCheckpointPath) {
+ this.isDbPathUnderCheckpointPath = isDbPathUnderCheckpointPath;
+ }
+
+ public void withClaimMode(RecoveryClaimMode claimMode) {
+ this.claimMode = claimMode;
+ }
+
+ public DataTransferStrategyBuilder() {
+ this.dbRemotePath = null;
+ this.dbDelegateFileSystem = null;
+ this.isDbPathUnderCheckpointPath = false;
+ }
+
+ public DataTransferStrategy build() {
+ DataTransferStrategy strategy;
+ if (dbDelegateFileSystem == null) {
+ strategy = new LocalFsDataTransferStrategy(claimMode);
+ LOG.info("Build DataTransferStrategy: {}", strategy);
+ return strategy;
+ }
+
+ RemoteFsDataTransferStrategy.FileTransferStyle snapshotStyle;
+ RemoteFsDataTransferStrategy.FileTransferStyle restoreStyle;
+ if (isDbPathUnderCheckpointPath) {
+ snapshotStyle = RemoteFsDataTransferStrategy.FileTransferStyle.REUSE;
+ restoreStyle =
+ claimMode == RecoveryClaimMode.CLAIM
+ ? RemoteFsDataTransferStrategy.FileTransferStyle.REUSE
+ : RemoteFsDataTransferStrategy.FileTransferStyle.COPY;
+ } else {
+ snapshotStyle = RemoteFsDataTransferStrategy.FileTransferStyle.COPY;
+ restoreStyle = RemoteFsDataTransferStrategy.FileTransferStyle.COPY;
+ }
+
+ strategy =
+ new RemoteFsDataTransferStrategy(
+ snapshotStyle,
+ restoreStyle,
+ claimMode,
+ dbDelegateFileSystem,
+ isDbPathUnderCheckpointPath);
+ LOG.info("Build DataTransferStrategy: {}", strategy);
+ return strategy;
Review Comment:
Local/Remote is not straightforward enough. I've modified them to ```CopyDataTransferStrategy``` and ```ReusableDataTransferStrategy``` , with the latter extending the former.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "AlexYinHan (via GitHub)" <gi...@apache.org>.
AlexYinHan commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1917698844
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileType.java:
##########
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst.fs.filemapping;
+
+/** Indicates the type of file in ForSt DB. */
+public enum FileType {
Review Comment:
That's right. I modified the values of ```FileOwnership``` and reorganized its semantics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1920000218
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java:
##########
@@ -167,6 +213,45 @@ public ForStResourceContainer(
this.metricGroup = metricGroup;
}
+ private static boolean isDbPathUnderCheckpointPath(
+ @Nullable CheckpointStorageAccess checkpointStorageAccess,
+ @Nullable Path dbRemotePath) {
+ if (dbRemotePath == null) {
+ return false;
+ }
+
+ // For checkpoint other that FsCheckpointStorageAccess, we treat it as 'DB path not under
+ // checkpoint path', since we cannot reuse checkpoint files in such case.
+ // todo: Support enabling 'cp file reuse' with FsMergingCheckpointStorageAccess
+ if (!(checkpointStorageAccess instanceof FsCheckpointStorageAccess)
+ || checkpointStorageAccess instanceof FsMergingCheckpointStorageAccess) {
+ return false;
+ }
+
+ FsCheckpointStorageAccess fsCheckpointStorageAccess =
+ (FsCheckpointStorageAccess) checkpointStorageAccess;
+ FileSystem checkpointFS = fsCheckpointStorageAccess.getFileSystem();
+ FileSystem dbRemoteFS;
+ try {
+ dbRemoteFS = dbRemotePath.getFileSystem();
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to get FileSystem from dbRemotePath: " + dbRemotePath, e);
+ }
+
+ if (!checkpointFS.getUri().equals(dbRemoteFS.getUri())) {
Review Comment:
```suggestion
if (!checkpointFS.equals(dbRemoteFS))) {
```
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java:
##########
@@ -167,6 +213,44 @@ public ForStResourceContainer(
this.metricGroup = metricGroup;
}
+ private static boolean isDbPathUnderCheckpointPath(
+ @Nullable CheckpointStorageAccess checkpointStorageAccess,
+ @Nullable Path dbRemotePath) {
+ if (dbRemotePath == null) {
+ return false;
+ }
+
+ // For checkpoint other that FsCheckpointStorageAccess, we treat it as 'DB path not under
+ // checkpoint path', since we cannot reuse checkpoint files in such case.
+ if (!(checkpointStorageAccess instanceof FsCheckpointStorageAccess)
+ || checkpointStorageAccess instanceof FsMergingCheckpointStorageAccess) {
Review Comment:
```suggestion
if (checkpointStorageAccess == null
|| checkpointStorageAccess.getClass() != FsCheckpointStorageAccess.class) {
```
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java:
##########
@@ -167,6 +213,44 @@ public ForStResourceContainer(
this.metricGroup = metricGroup;
}
+ private static boolean isDbPathUnderCheckpointPath(
+ @Nullable CheckpointStorageAccess checkpointStorageAccess,
+ @Nullable Path dbRemotePath) {
+ if (dbRemotePath == null) {
+ return false;
+ }
+
+ // For checkpoint other that FsCheckpointStorageAccess, we treat it as 'DB path not under
+ // checkpoint path', since we cannot reuse checkpoint files in such case.
+ if (!(checkpointStorageAccess instanceof FsCheckpointStorageAccess)
+ || checkpointStorageAccess instanceof FsMergingCheckpointStorageAccess) {
Review Comment:
And I suggest a TODO for file merging support
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly closed pull request #25924: [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend
URL: https://github.com/apache/flink/pull/25924
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #25924:
URL: https://github.com/apache/flink/pull/25924#issuecomment-2601286007
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "AlexYinHan (via GitHub)" <gi...@apache.org>.
AlexYinHan commented on PR #25924:
URL: https://github.com/apache/flink/pull/25924#issuecomment-2599463500
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "AlexYinHan (via GitHub)" <gi...@apache.org>.
AlexYinHan commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1920209858
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java:
##########
@@ -167,6 +213,44 @@ public ForStResourceContainer(
this.metricGroup = metricGroup;
}
+ private static boolean isDbPathUnderCheckpointPath(
+ @Nullable CheckpointStorageAccess checkpointStorageAccess,
+ @Nullable Path dbRemotePath) {
+ if (dbRemotePath == null) {
+ return false;
+ }
+
+ // For checkpoint other that FsCheckpointStorageAccess, we treat it as 'DB path not under
+ // checkpoint path', since we cannot reuse checkpoint files in such case.
+ if (!(checkpointStorageAccess instanceof FsCheckpointStorageAccess)
+ || checkpointStorageAccess instanceof FsMergingCheckpointStorageAccess) {
Review Comment:
Done
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java:
##########
@@ -167,6 +213,45 @@ public ForStResourceContainer(
this.metricGroup = metricGroup;
}
+ private static boolean isDbPathUnderCheckpointPath(
+ @Nullable CheckpointStorageAccess checkpointStorageAccess,
+ @Nullable Path dbRemotePath) {
+ if (dbRemotePath == null) {
+ return false;
+ }
+
+ // For checkpoint other that FsCheckpointStorageAccess, we treat it as 'DB path not under
+ // checkpoint path', since we cannot reuse checkpoint files in such case.
+ // todo: Support enabling 'cp file reuse' with FsMergingCheckpointStorageAccess
+ if (!(checkpointStorageAccess instanceof FsCheckpointStorageAccess)
+ || checkpointStorageAccess instanceof FsMergingCheckpointStorageAccess) {
+ return false;
+ }
+
+ FsCheckpointStorageAccess fsCheckpointStorageAccess =
+ (FsCheckpointStorageAccess) checkpointStorageAccess;
+ FileSystem checkpointFS = fsCheckpointStorageAccess.getFileSystem();
+ FileSystem dbRemoteFS;
+ try {
+ dbRemoteFS = dbRemotePath.getFileSystem();
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to get FileSystem from dbRemotePath: " + dbRemotePath, e);
+ }
+
+ if (!checkpointFS.getUri().equals(dbRemoteFS.getUri())) {
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "AlexYinHan (via GitHub)" <gi...@apache.org>.
AlexYinHan commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1919963613
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ReusableDataTransferStrategy.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst.datatransfer;
+
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.CheckpointedStateScope;
+import org.apache.flink.runtime.state.IncrementalKeyedStateHandle.HandleAndLocalPath;
+import org.apache.flink.runtime.state.StreamStateHandle;
+import org.apache.flink.runtime.state.filemerging.SegmentFileStateHandle;
+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.FileOwnershipDecider;
+import org.apache.flink.state.forst.fs.filemapping.MappingEntry;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * Data transfer strategy for ForSt DB with a remote DB path. When transferring data between
+ * Checkpoint and DB, this strategy MAY reuse the file from Checkpoint storage.
+ */
+public class ReusableDataTransferStrategy extends CopyDataTransferStrategy {
+
+ private final FileOwnershipDecider fileOwnershipDecider;
+
+ protected ReusableDataTransferStrategy(
+ @Nonnull FileOwnershipDecider fileOwnershipDecider, FileSystem dbFileSystem) {
+ super(dbFileSystem);
+
+ Preconditions.checkArgument(
+ dbFileSystem instanceof ForStFlinkFileSystem,
+ "Unexpected dbFileSystem type: "
+ + dbFileSystem.getClass()
+ + ", expected: "
+ + ForStFlinkFileSystem.class);
+
+ this.fileOwnershipDecider = fileOwnershipDecider;
+ }
+
+ private ForStFlinkFileSystem getForStFlinkFileSystem() {
+ // We do not check the type here, because it is checked in the constructor.
+ return (ForStFlinkFileSystem) dbFileSystem;
+ }
+
+ private @Nullable HandleAndLocalPath reuseFileToCheckpoint(
+ Path dbFilePath, ForStFlinkFileSystem forStFs) throws IOException {
+ LOG.trace("Reuse file to checkpoint: {}", dbFilePath);
+
+ // Find the real path of the file
+ Path sourcePath = forStFs.srcPath(dbFilePath);
+ if (sourcePath != null) {
+ dbFilePath = sourcePath;
+ }
+
+ // Create a StateHandle with the real path
+ FileStatus fileStatus = forStFs.getFileStatus(dbFilePath);
+ StreamStateHandle realSourceHandle = new FileStateHandle(sourcePath, fileStatus.getLen());
+
+ // Give file ownership to JM, i.e. DB will not delete it from now on
+ forStFs.giveUpOwnership(dbFilePath);
+
+ return HandleAndLocalPath.of(realSourceHandle, dbFilePath.getName());
+ }
+
+ @Override
+ public HandleAndLocalPath transferToCheckpoint(
+ Path dbFilePath,
+ long maxTransferBytes,
+ CheckpointStreamFactory checkpointStreamFactory,
+ CheckpointedStateScope stateScope,
+ CloseableRegistry closeableRegistry,
+ CloseableRegistry tmpResourcesRegistry)
+ throws IOException {
+ if (maxTransferBytes < 0) {
+ // Means transfer whole file to checkpoint storage.
+ maxTransferBytes = Long.MAX_VALUE;
+ }
+
+ ForStFlinkFileSystem forStFs = getForStFlinkFileSystem();
+ MappingEntry mappingEntry = forStFs.getMappingEntry(dbFilePath);
+ Preconditions.checkNotNull(mappingEntry);
+ FileOwnership fileOwnership = mappingEntry.getFileOwnership();
+ if (fileOwnership == FileOwnership.PRIVATE_OWNED_BY_DB) {
+ LOG.trace(
+ "Do not reuse file to checkpoint because the file is privately owned by DB: {}",
+ dbFilePath);
+ return super.transferToCheckpoint(
+ dbFilePath,
+ maxTransferBytes,
+ checkpointStreamFactory,
+ stateScope,
+ closeableRegistry,
+ tmpResourcesRegistry);
+ }
+ return reuseFileToCheckpoint(dbFilePath, forStFs);
+ }
+
+ @Override
+ public void transferFromCheckpoint(
+ StreamStateHandle sourceHandle, Path targetPath, CloseableRegistry closeableRegistry)
+ throws IOException {
+
+ FileOwnership fileOwnership = fileOwnershipDecider.decideForRestoredFile(targetPath);
+ if (fileOwnership != FileOwnership.PRIVATE_OWNED_BY_DB) {
+ if (tryReuseFileFromCheckpoint(sourceHandle, targetPath)) {
+ return;
+ }
+ } else {
+ LOG.trace(
+ "Do not reuse file from checkpoint because the file is privately owned by DB: {}",
+ targetPath);
+ }
+ super.transferFromCheckpoint(sourceHandle, targetPath, closeableRegistry);
+ }
+
+ private boolean tryReuseFileFromCheckpoint(StreamStateHandle sourceHandle, Path targetPath)
+ throws IOException {
+ Path sourcePath = prepareSourcePathFromStateHandle(sourceHandle);
+ if (sourcePath == null) {
+ LOG.trace(
+ "Do not reuse file from checkpoint because the StateHandle is not backed by a file: {}, {}",
+ sourceHandle,
+ targetPath);
+ return false;
+ } else {
+ reuseFileFromCheckpoint(sourcePath, targetPath);
+ return true;
+ }
+ }
+
+ private @Nullable Path prepareSourcePathFromStateHandle(StreamStateHandle sourceHandle) {
+ Path sourcePath = null;
+ if (sourceHandle instanceof FileStateHandle) {
+ sourcePath = ((FileStateHandle) sourceHandle).getFilePath();
+ } else if (sourceHandle instanceof SegmentFileStateHandle) {
+ sourcePath = ((SegmentFileStateHandle) sourceHandle).getFilePath();
+ }
+ return sourcePath;
+ }
Review Comment:
+1. I've added ```MappingEntrySource``` in ```MappingEntry```. Each entry can now be backed by either a file or a ```StreamStateHandle```.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1917823612
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileMappingManager.java:
##########
@@ -231,14 +270,23 @@ private boolean isParentDir(String path, String dir) {
}
}
- /** A wrapper of real path. */
- public static class RealPath {
- public final Path path;
- public final boolean isLocal;
+ public void giveUpOwnership(Path path) {
+ MappingEntry mappingEntry = mappingTable.getOrDefault(path.toString(), null);
+ Preconditions.checkArgument(
+ mappingEntry != null,
+ "Try to give up ownership of a file that is not in mapping table: " + path);
+ Preconditions.checkArgument(
+ mappingEntry.fileOwnership != FileOwnership.PRIVATE_OWNED_BY_DB,
+ "Try to give up ownership of a file that is not shareable: " + mappingEntry);
Review Comment:
Please use the string template version of `checkArgument` here, to avoid much string generation overhead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1917811473
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileOwnershipDecider.java:
##########
@@ -0,0 +1,70 @@
+package org.apache.flink.state.forst.fs.filemapping;
+
+import org.apache.flink.core.execution.RecoveryClaimMode;
+import org.apache.flink.core.fs.Path;
+
+public class FileOwnershipDecider {
+
+ public static final String SST_SUFFIX = ".sst";
+
+ private final RecoveryClaimMode recoveryClaimMode;
+ private final boolean isDbPathUnderCheckpointPath;
+
+ public static FileOwnershipDecider getDefault() {
+ return new FileOwnershipDecider(RecoveryClaimMode.DEFAULT, false);
+ }
+
+ public FileOwnershipDecider(
+ RecoveryClaimMode recoveryClaimMode, boolean isDbPathUnderCheckpointPath) {
+ this.recoveryClaimMode = recoveryClaimMode;
+ this.isDbPathUnderCheckpointPath = isDbPathUnderCheckpointPath;
+ }
+
+ public boolean isDbPathUnderCheckpointPath() {
+ return isDbPathUnderCheckpointPath;
+ }
+
+ public FileOwnership decideForNewFile(Path filePath) {
+ if (!isDbPathUnderCheckpointPath) {
+ return FileOwnership.PRIVATE_OWNED_BY_DB;
+ }
+
+ // only SST files are shareable
+ return isSstFile(filePath)
+ ? FileOwnership.SHAREABLE_OWNED_BY_DB
+ : FileOwnership.PRIVATE_OWNED_BY_DB;
+ }
+
+ public FileOwnership decideForRestoredFile(Path filePath) {
+ if (!isDbPathUnderCheckpointPath) {
+ return FileOwnership.PRIVATE_OWNED_BY_DB;
+ }
+
+ // only SST files are shareable
+ if (isSstFile(filePath)) {
+ return recoveryClaimMode == RecoveryClaimMode.CLAIM
+ ? FileOwnership.NOT_OWNED
+ : FileOwnership.SHAREABLE_OWNED_BY_DB;
+ } else {
+ return FileOwnership.PRIVATE_OWNED_BY_DB;
+ }
+ }
Review Comment:
IIUC, the restored file is always `NOT_OWNED` (own by JM's shared registry).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1914688767
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStorageAccess.java:
##########
@@ -124,10 +124,14 @@ public FsCheckpointStorageAccess(
// ------------------------------------------------------------------------
@VisibleForTesting
Review Comment:
If you make it public and use it in normal code other than test, please remove this annotation.
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileType.java:
##########
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst.fs.filemapping;
+
+/** Indicates the type of file in ForSt DB. */
+public enum FileType {
Review Comment:
Do we really need to introduce this? Is `FileOwnership.LOCAL_OWNED_BY_DB` is enough for `MISC_FILE`?
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java:
##########
@@ -364,11 +444,31 @@ public void prepareDirectories() throws Exception {
}
}
+ public void buildDataTransferStrategy() {
Review Comment:
I'd suggest:
- Build `DataTransferStrategy` in constructor.
- make `isDbPathUnderCheckpointPath` as a member of implementation of `DataTransferStrategy`, and calculate this in the builder.
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyBuilder.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst.datatransfer;
+
+import org.apache.flink.core.execution.RecoveryClaimMode;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Builder for {@link DataTransferStrategy}. */
+public class DataTransferStrategyBuilder {
+ private static final Logger LOG = LoggerFactory.getLogger(DataTransferStrategyBuilder.class);
+
+ private Path dbRemotePath;
+
+ private FileSystem dbDelegateFileSystem;
+
+ private boolean isDbPathUnderCheckpointPath;
+
+ private RecoveryClaimMode claimMode;
+
+ public void withoutDbRemotePath() {
+ this.dbRemotePath = null;
+ this.dbDelegateFileSystem = null;
+ this.claimMode = RecoveryClaimMode.DEFAULT;
+ }
+
+ public void withDbRemotePath(FileSystem dbDelegateFileSystem, Path dbRemotePath) {
+ this.dbRemotePath = dbRemotePath;
+ this.dbDelegateFileSystem = dbDelegateFileSystem;
+ }
+
+ public void withIsDbPathUnderCheckpointPath(Boolean isDbPathUnderCheckpointPath) {
+ this.isDbPathUnderCheckpointPath = isDbPathUnderCheckpointPath;
+ }
+
+ public void withClaimMode(RecoveryClaimMode claimMode) {
+ this.claimMode = claimMode;
+ }
+
+ public DataTransferStrategyBuilder() {
+ this.dbRemotePath = null;
+ this.dbDelegateFileSystem = null;
+ this.isDbPathUnderCheckpointPath = false;
+ }
+
+ public DataTransferStrategy build() {
+ DataTransferStrategy strategy;
+ if (dbDelegateFileSystem == null) {
+ strategy = new LocalFsDataTransferStrategy(claimMode);
+ LOG.info("Build DataTransferStrategy: {}", strategy);
+ return strategy;
+ }
+
+ RemoteFsDataTransferStrategy.FileTransferStyle snapshotStyle;
+ RemoteFsDataTransferStrategy.FileTransferStyle restoreStyle;
+ if (isDbPathUnderCheckpointPath) {
+ snapshotStyle = RemoteFsDataTransferStrategy.FileTransferStyle.REUSE;
+ restoreStyle =
+ claimMode == RecoveryClaimMode.CLAIM
+ ? RemoteFsDataTransferStrategy.FileTransferStyle.REUSE
+ : RemoteFsDataTransferStrategy.FileTransferStyle.COPY;
+ } else {
+ snapshotStyle = RemoteFsDataTransferStrategy.FileTransferStyle.COPY;
+ restoreStyle = RemoteFsDataTransferStrategy.FileTransferStyle.COPY;
+ }
+
+ strategy =
+ new RemoteFsDataTransferStrategy(
+ snapshotStyle,
+ restoreStyle,
+ claimMode,
+ dbDelegateFileSystem,
+ isDbPathUnderCheckpointPath);
+ LOG.info("Build DataTransferStrategy: {}", strategy);
+ return strategy;
Review Comment:
IIUC, there should be two types of strategy, one is for copying by stream while another is for file reuse (in best effort). But it seems we devide by Local/Remote, is there any reason for this?
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java:
##########
@@ -378,12 +478,21 @@ private static void prepareDirectories(Path basePath, Path dbPath) throws IOExce
throw new IOException(
String.format("Could not create ForSt directory at %s.", basePath));
}
- if (fileSystem.exists(dbPath)) {
- fileSystem.delete(dbPath, true);
- }
- if (!fileSystem.mkdirs(dbPath)) {
- throw new IOException(
- String.format("Could not create ForSt db directory at %s.", dbPath));
+ if (clearDbPath) {
Review Comment:
Why we need `clearDbPath` here?
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileMappingManager.java:
##########
@@ -231,14 +284,29 @@ private boolean isParentDir(String path, String dir) {
}
}
- /** A wrapper of real path. */
- public static class RealPath {
Review Comment:
Why removing this class?
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java:
##########
@@ -167,6 +210,43 @@ public ForStResourceContainer(
this.metricGroup = metricGroup;
}
+ private static boolean isDbPathUnderCheckpointPath(
+ @Nullable CheckpointStorageAccess checkpointStorageAccess,
+ @Nullable Path dbRemotePath) {
+ if (dbRemotePath == null) {
+ return false;
+ }
+
+ Preconditions.checkArgument(
+ checkpointStorageAccess instanceof FsCheckpointStorageAccess,
+ "Unexpected checkpointStorageAccess type: "
+ + (checkpointStorageAccess == null
+ ? "null"
+ : checkpointStorageAccess.getClass()));
Review Comment:
So we only support `FsCheckpointStorageAccess`, right?
Can we support `MemoryBackendCheckpointStorageAccess` by writing via stream? And what if it is `FsMergingCheckpointStorageAccess`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "AlexYinHan (via GitHub)" <gi...@apache.org>.
AlexYinHan commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1917688495
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java:
##########
@@ -364,11 +444,31 @@ public void prepareDirectories() throws Exception {
}
}
+ public void buildDataTransferStrategy() {
Review Comment:
The creation order of these StateBackend components can bring some troubles. When building the state backend, we:
- call ```prepareDirectories```, to create directories for the DB. By default we clear the target directory so we have a clean path for the DB. However, if we decide to reuse some files from the Checkpoint, we should keep the target directory as it is. That is decided base on ```isDbPathUnderCheckpointPath```
- also in ```prepareDirectories```, we create a ```ForStFlinkFileSystem```
- then we can construct the ```DataTransferStrategy``` based on a nullable```ForStFlinkFileSystem``` and ```isDbPathUnderCheckpointPath```
Therefore, it might be better to compute ```isDbPathUnderCheckpointPath``` before the above steps.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #25924:
URL: https://github.com/apache/flink/pull/25924#issuecomment-2601285910
CI is not triggering....
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #25924:
URL: https://github.com/apache/flink/pull/25924#issuecomment-2577579299
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "04e7de8c0f8047b37fa50bc0474f2c2f9338a3bb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "04e7de8c0f8047b37fa50bc0474f2c2f9338a3bb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 04e7de8c0f8047b37fa50bc0474f2c2f9338a3bb UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "AlexYinHan (via GitHub)" <gi...@apache.org>.
AlexYinHan commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1915975206
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileMappingManager.java:
##########
@@ -231,14 +284,29 @@ private boolean isParentDir(String path, String dir) {
}
}
- /** A wrapper of real path. */
- public static class RealPath {
Review Comment:
It seems we can simply replace ```RealPath``` with ```MappingEntry``` in all places.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "AlexYinHan (via GitHub)" <gi...@apache.org>.
AlexYinHan commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1917638328
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java:
##########
@@ -167,6 +210,43 @@ public ForStResourceContainer(
this.metricGroup = metricGroup;
}
+ private static boolean isDbPathUnderCheckpointPath(
+ @Nullable CheckpointStorageAccess checkpointStorageAccess,
+ @Nullable Path dbRemotePath) {
+ if (dbRemotePath == null) {
+ return false;
+ }
+
+ Preconditions.checkArgument(
+ checkpointStorageAccess instanceof FsCheckpointStorageAccess,
+ "Unexpected checkpointStorageAccess type: "
+ + (checkpointStorageAccess == null
+ ? "null"
+ : checkpointStorageAccess.getClass()));
Review Comment:
I've modified the condition here. For cases involving ```MemoryBackendCheckpointStorageAccess``` and ```FsMergingCheckpointStorageAccess```, this function returns null. Therefore we will follow the original logic of copying by writing via streams rather than reusing the files.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37021][state/forst] Implement fast cp/restore for ForSt StateBackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25924:
URL: https://github.com/apache/flink/pull/25924#discussion_r1917850722
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ReusableDataTransferStrategy.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst.datatransfer;
+
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.CheckpointedStateScope;
+import org.apache.flink.runtime.state.IncrementalKeyedStateHandle.HandleAndLocalPath;
+import org.apache.flink.runtime.state.StreamStateHandle;
+import org.apache.flink.runtime.state.filemerging.SegmentFileStateHandle;
+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.FileOwnershipDecider;
+import org.apache.flink.state.forst.fs.filemapping.MappingEntry;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * Data transfer strategy for ForSt DB with a remote DB path. When transferring data between
+ * Checkpoint and DB, this strategy MAY reuse the file from Checkpoint storage.
+ */
+public class ReusableDataTransferStrategy extends CopyDataTransferStrategy {
+
+ private final FileOwnershipDecider fileOwnershipDecider;
+
+ protected ReusableDataTransferStrategy(
+ @Nonnull FileOwnershipDecider fileOwnershipDecider, FileSystem dbFileSystem) {
+ super(dbFileSystem);
+
+ Preconditions.checkArgument(
+ dbFileSystem instanceof ForStFlinkFileSystem,
+ "Unexpected dbFileSystem type: "
+ + dbFileSystem.getClass()
+ + ", expected: "
+ + ForStFlinkFileSystem.class);
+
+ this.fileOwnershipDecider = fileOwnershipDecider;
+ }
+
+ private ForStFlinkFileSystem getForStFlinkFileSystem() {
+ // We do not check the type here, because it is checked in the constructor.
+ return (ForStFlinkFileSystem) dbFileSystem;
+ }
+
+ private @Nullable HandleAndLocalPath reuseFileToCheckpoint(
+ Path dbFilePath, ForStFlinkFileSystem forStFs) throws IOException {
+ LOG.trace("Reuse file to checkpoint: {}", dbFilePath);
+
+ // Find the real path of the file
+ Path sourcePath = forStFs.srcPath(dbFilePath);
+ if (sourcePath != null) {
+ dbFilePath = sourcePath;
+ }
+
+ // Create a StateHandle with the real path
+ FileStatus fileStatus = forStFs.getFileStatus(dbFilePath);
+ StreamStateHandle realSourceHandle = new FileStateHandle(sourcePath, fileStatus.getLen());
+
+ // Give file ownership to JM, i.e. DB will not delete it from now on
+ forStFs.giveUpOwnership(dbFilePath);
+
+ return HandleAndLocalPath.of(realSourceHandle, dbFilePath.getName());
+ }
+
+ @Override
+ public HandleAndLocalPath transferToCheckpoint(
+ Path dbFilePath,
+ long maxTransferBytes,
+ CheckpointStreamFactory checkpointStreamFactory,
+ CheckpointedStateScope stateScope,
+ CloseableRegistry closeableRegistry,
+ CloseableRegistry tmpResourcesRegistry)
+ throws IOException {
+ if (maxTransferBytes < 0) {
+ // Means transfer whole file to checkpoint storage.
+ maxTransferBytes = Long.MAX_VALUE;
+ }
+
+ ForStFlinkFileSystem forStFs = getForStFlinkFileSystem();
+ MappingEntry mappingEntry = forStFs.getMappingEntry(dbFilePath);
+ Preconditions.checkNotNull(mappingEntry);
+ FileOwnership fileOwnership = mappingEntry.getFileOwnership();
+ if (fileOwnership == FileOwnership.PRIVATE_OWNED_BY_DB) {
+ LOG.trace(
+ "Do not reuse file to checkpoint because the file is privately owned by DB: {}",
+ dbFilePath);
+ return super.transferToCheckpoint(
+ dbFilePath,
+ maxTransferBytes,
+ checkpointStreamFactory,
+ stateScope,
+ closeableRegistry,
+ tmpResourcesRegistry);
+ }
+ return reuseFileToCheckpoint(dbFilePath, forStFs);
+ }
+
+ @Override
+ public void transferFromCheckpoint(
+ StreamStateHandle sourceHandle, Path targetPath, CloseableRegistry closeableRegistry)
+ throws IOException {
+
+ FileOwnership fileOwnership = fileOwnershipDecider.decideForRestoredFile(targetPath);
+ if (fileOwnership != FileOwnership.PRIVATE_OWNED_BY_DB) {
+ if (tryReuseFileFromCheckpoint(sourceHandle, targetPath)) {
+ return;
+ }
+ } else {
+ LOG.trace(
+ "Do not reuse file from checkpoint because the file is privately owned by DB: {}",
+ targetPath);
+ }
+ super.transferFromCheckpoint(sourceHandle, targetPath, closeableRegistry);
+ }
+
+ private boolean tryReuseFileFromCheckpoint(StreamStateHandle sourceHandle, Path targetPath)
+ throws IOException {
+ Path sourcePath = prepareSourcePathFromStateHandle(sourceHandle);
+ if (sourcePath == null) {
+ LOG.trace(
+ "Do not reuse file from checkpoint because the StateHandle is not backed by a file: {}, {}",
+ sourceHandle,
+ targetPath);
+ return false;
+ } else {
+ reuseFileFromCheckpoint(sourcePath, targetPath);
+ return true;
+ }
+ }
+
+ private @Nullable Path prepareSourcePathFromStateHandle(StreamStateHandle sourceHandle) {
+ Path sourcePath = null;
+ if (sourceHandle instanceof FileStateHandle) {
+ sourcePath = ((FileStateHandle) sourceHandle).getFilePath();
+ } else if (sourceHandle instanceof SegmentFileStateHandle) {
+ sourcePath = ((SegmentFileStateHandle) sourceHandle).getFilePath();
+ }
+ return sourcePath;
+ }
Review Comment:
I'd suggest storing/reusing this `sourceHandle` with in our filemapping
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org