Zakelly commented on code in PR #25310:
URL: https://github.com/apache/flink/pull/25310#discussion_r1758296733
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java:
##########
@@ -125,30 +187,117 @@ public URI getUri() {
return delegateFS.getUri();
}
+ @Override
+ public boolean exists(final Path f) throws IOException {
+ Tuple2<Boolean, Path> localPathTuple = tryBuildLocalPath(f);
+ if (localPathTuple.f0) {
+ return localFS.exists(localPathTuple.f1);
+ }
+ return delegateFS.exists(f);
+ }
+
@Override
public FileStatus getFileStatus(Path path) throws IOException {
+ Tuple2<Boolean, Path> localPathTuple = tryBuildLocalPath(path);
+ if (localPathTuple.f0) {
+ return localFS.getFileStatus(localPathTuple.f1);
+ }
return delegateFS.getFileStatus(path);
}
@Override
public BlockLocation[] getFileBlockLocations(FileStatus file, long start,
long len)
throws IOException {
+ Path path = file.getPath();
+ Tuple2<Boolean, Path> localPathTuple = tryBuildLocalPath(path);
+ if (localPathTuple.f0) {
+ FileStatus localFile = localFS.getFileStatus(localPathTuple.f1);
+ return localFS.getFileBlockLocations(localFile, start, len);
+ }
return delegateFS.getFileBlockLocations(file, start, len);
}
@Override
public FileStatus[] listStatus(Path path) throws IOException {
- return delegateFS.listStatus(path);
+ FileStatus[] localFiles = new FileStatus[0];
+ Tuple2<Boolean, Path> localPathTuple = tryBuildLocalPath(path);
+ if (localPathTuple.f0) {
+ localFiles = localFS.listStatus(localPathTuple.f1);
+ }
+ int localFileNum = localFiles == null ? 0 : localFiles.length;
+ FileStatus[] remoteFiles = delegateFS.listStatus(path);
+ if (localFileNum == 0) {
+ return remoteFiles;
+ }
+ int remoteFileNum = remoteFiles == null ? 0 : remoteFiles.length;
+ FileStatus[] fileStatuses = new FileStatus[localFileNum +
remoteFileNum];
+ for (int index = 0; index < localFileNum; index++) {
+ final FileStatus localFile = localFiles[index];
+ fileStatuses[index] =
+ new FileStatus() {
+ @Override
+ public long getLen() {
+ return localFile.getLen();
+ }
+
+ @Override
+ public long getBlockSize() {
+ return localFile.getBlockSize();
+ }
+
+ @Override
+ public short getReplication() {
+ return localFile.getReplication();
+ }
+
+ @Override
+ public long getModificationTime() {
+ return localFile.getModificationTime();
+ }
+
+ @Override
+ public long getAccessTime() {
+ return localFile.getAccessTime();
+ }
+
+ @Override
+ public boolean isDir() {
+ return localFile.isDir();
+ }
+
+ @Override
+ public Path getPath() {
+ return new Path(
+ remoteBase,
path.toString().substring(localBase.length()));
+ }
+ };
+ }
+ if (remoteFiles != null) {
+ System.arraycopy(remoteFiles, 0, fileStatuses, localFileNum,
remoteFiles.length);
+ }
Review Comment:
```suggestion
if (remoteFileNum != 0) {
System.arraycopy(remoteFiles, 0, fileStatuses, localFileNum,
remoteFileNum);
}
```
--
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-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
