[FLINK-36819] Support link() in ForStFlinkFileSystem

fredia · Nov 29, 2024 · a751599 · a751599
1 parent 7394878
commit a751599
Show file tree

Hide file tree

Showing 4 changed files with 77 additions and 0 deletions.
diff --git a/env/flink/env_flink.cc b/env/flink/env_flink.cc
@@ -878,6 +878,69 @@ IOStatus FlinkFileSystem::UnlockFile(FileLock* /*lock*/,
   return IOStatus::OK();
 }
 
+IOStatus FlinkFileSystem::LinkFile(const std::string& src,
+                                   const std::string& target,
+                                   const IOOptions& options,
+                                   IODebugContext* dbg) {
+  IOStatus status = FileExists(src, options, dbg);
+  if (!status.ok()) {
+    return status.IsNotFound()
+               ? IOStatus::PathNotFound(
+                     std::string(
+                         "Could not find src path when linkFile, path: ")
+                         .append(ConstructPath(src)))
+               : status;
+  }
+
+  JNIEnv* jniEnv = getJNIEnv();
+
+  std::string srcFilePath = ConstructPath(src);
+  // Construct src Path Instance
+  jobject srcPathInstance;
+  status = class_cache_->ConstructPathInstance(srcFilePath, &srcPathInstance);
+  if (!status.ok()) {
+    return status;
+  }
+
+  std::string targetFilePath = ConstructPath(target);
+  // Construct target Path Instance
+  jobject targetPathInstance;
+  status =
+      class_cache_->ConstructPathInstance(targetFilePath, &targetPathInstance);
+  if (!status.ok()) {
+    jniEnv->DeleteLocalRef(srcPathInstance);
+    return status;
+  }
+
+  JavaClassCache::JavaMethodContext renameMethod = class_cache_->GetJMethod(
+      JavaClassCache::JM_FLINK_FILE_SYSTEM_LINK_FILE);
+  jint linked =
+      jniEnv->CallIntMethod(file_system_instance_, renameMethod.javaMethod,
+                                srcPathInstance, targetPathInstance);
+  jniEnv->DeleteLocalRef(srcPathInstance);
+  jniEnv->DeleteLocalRef(targetPathInstance);
+
+  status = CurrentStatus([srcFilePath, targetFilePath]() {
+    return std::string("Exception when LinkFile, src: ")
+        .append(srcFilePath)
+        .append(", target: ")
+        .append(targetFilePath);
+  });
+  if (!status.ok()) {
+    return status;
+  }
+
+  if (linked == -1) {
+    return IOStatus::NotSupported();
+  } else  if (linked > 0 ){
+    return IOStatus::IOError(std::string("Exception when LinkFile, src: ")
+                          .append(srcFilePath)
+                          .append(", target: ")
+                          .append(targetFilePath));
+  }
+  return IOStatus::OK();
+}
+
 Status FlinkFileSystem::Create(const std::shared_ptr<FileSystem>& base,
                                const std::string& uri,
                                std::unique_ptr<FileSystem>* result,

diff --git a/env/flink/env_flink.h b/env/flink/env_flink.h
@@ -98,6 +98,11 @@ class FlinkFileSystem : public FileSystemWrapper {
                        const IOOptions& /*options*/, bool* /*is_dir*/,
                        IODebugContext* /*dbg*/) override;
 
+  IOStatus LinkFile(const std::string& /*src*/,
+                    const std::string& /*target*/,
+                    const IOOptions& /*options*/,
+                    IODebugContext* /*dbg*/) override;
+
  private:
   const std::string base_path_;
   JavaClassCache* class_cache_;

diff --git a/env/flink/jni_helper.cc b/env/flink/jni_helper.cc
@@ -151,6 +151,14 @@ IOStatus JavaClassCache::Init() {
       .signature =
       "(Lorg/apache/flink/core/fs/Path;Lorg/apache/flink/core/fs/Path;)Z";
 
+  cached_java_methods_[CachedJavaMethod::JM_FLINK_FILE_SYSTEM_LINK_FILE]
+      .javaClassAndName = cached_java_classes_[JC_FLINK_FILE_SYSTEM];
+  cached_java_methods_[CachedJavaMethod::JM_FLINK_FILE_SYSTEM_LINK_FILE]
+      .methodName = "link";
+  cached_java_methods_[CachedJavaMethod::JM_FLINK_FILE_SYSTEM_LINK_FILE]
+      .signature =
+      "(Lorg/apache/flink/core/fs/Path;Lorg/apache/flink/core/fs/Path;)I";
+
   cached_java_methods_[CachedJavaMethod::JM_FLINK_FILE_SYSTEM_OPEN]
       .javaClassAndName = cached_java_classes_[JC_FLINK_FILE_SYSTEM];
   cached_java_methods_[CachedJavaMethod::JM_FLINK_FILE_SYSTEM_OPEN].methodName =

diff --git a/env/flink/jni_helper.h b/env/flink/jni_helper.h
@@ -66,6 +66,7 @@ class JavaClassCache {
     JM_FLINK_FILE_STATUS_GET_LEN,
     JM_FLINK_FILE_STATUS_GET_MODIFICATION_TIME,
     JM_FLINK_FILE_STATUS_IS_DIR,
+    JM_FLINK_FILE_SYSTEM_LINK_FILE,
     NUM_CACHED_METHODS
   } CachedJavaMethod;