feat: implement bidirectional LoadJob support

jlon · jlon · commit 0843f60cde3d · 2025-10-31T08:54:09.000+08:00
diff --git a/curvine-common/src/state/job.rs b/curvine-common/src/state/job.rs
@@ -77,11 +77,13 @@ pub struct JobStatus {
 #[derive(Default, Debug, Deserialize, Serialize)]
 pub struct LoadJobCommand {
     pub source_path: String,
+    pub target_path: Option<String>,
     pub replicas: Option<i32>,
     pub block_size: Option<i64>,
     pub storage_type: Option<StorageType>,
     pub ttl_ms: Option<i64>,
     pub ttl_action: Option<TtlAction>,
+    pub overwrite: Option<bool>,
 }
 
 impl LoadJobCommand {
@@ -93,11 +95,13 @@ impl LoadJobCommand {
 #[derive(Default)]
 pub struct LoadJobCommandBuilder {
     source_path: String,
+    target_path: Option<String>,
     replicas: Option<i32>,
     block_size: Option<i64>,
     storage_type: Option<StorageType>,
     ttl_ms: Option<i64>,
     ttl_action: Option<TtlAction>,
+    overwrite: Option<bool>,
 }
 
 impl LoadJobCommandBuilder {
@@ -108,6 +112,11 @@ impl LoadJobCommandBuilder {
         }
     }
 
+    pub fn target_path(mut self, target_path: impl Into<String>) -> Self {
+        let _ = self.target_path.insert(target_path.into());
+        self
+    }
+
     pub fn replicas(mut self, replicas: i32) -> Self {
         let _ = self.replicas.insert(replicas);
         self
@@ -133,14 +142,21 @@ impl LoadJobCommandBuilder {
         self
     }
 
+    pub fn overwrite(mut self, overwrite: bool) -> Self {
+        let _ = self.overwrite.insert(overwrite);
+        self
+    }
+
     pub fn build(self) -> LoadJobCommand {
         LoadJobCommand {
             source_path: self.source_path,
+            target_path: self.target_path,
             replicas: self.replicas,
             block_size: self.block_size,
             storage_type: self.storage_type,
             ttl_ms: self.ttl_ms,
             ttl_action: self.ttl_action,
+            overwrite: self.overwrite,
         }
     }
 }
@@ -157,6 +173,7 @@ pub struct LoadJobInfo {
     pub ttl_action: TtlAction,
     pub mount_info: MountInfo,
     pub create_time: i64,
+    pub overwrite: Option<bool>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
diff --git a/curvine-common/src/state/mount.rs b/curvine-common/src/state/mount.rs
@@ -347,4 +347,74 @@ mod tests {
             "s3://spark/a/b/c/dt=2025/1.csv"
         );
     }
+
+    #[test]
+    fn test_bidirectional_path_conversion() {
+        // Mount config: s3://flink/user → /mnt/s3
+        let info = MountInfo {
+            ufs_path: "s3://flink/user".to_string(),
+            cv_path: "/mnt/s3".to_string(),
+            ..Default::default()
+        };
+
+        // Test 1: UFS → CV (Import) - root level file
+        let ufs_path = Path::from_str("s3://flink/user/batch_add_path_migrate_task.py").unwrap();
+        let cv_result = info.get_cv_path(&ufs_path).unwrap();
+        assert_eq!(
+            cv_result.full_path(),
+            "/mnt/s3/batch_add_path_migrate_task.py"
+        );
+
+        // Test 2: CV → UFS (Export) - root level file
+        let cv_path = Path::from_str("/mnt/s3/batch_add_path_migrate_task.py").unwrap();
+        let ufs_result = info.get_ufs_path(&cv_path).unwrap();
+        assert_eq!(
+            ufs_result.full_path(),
+            "s3://flink/user/batch_add_path_migrate_task.py"
+        );
+
+        // Test 3: UFS → CV (Import) - nested directory
+        let ufs_nested = Path::from_str("s3://flink/user/dir1/dir2/file.txt").unwrap();
+        let cv_nested = info.get_cv_path(&ufs_nested).unwrap();
+        assert_eq!(cv_nested.full_path(), "/mnt/s3/dir1/dir2/file.txt");
+
+        // Test 4: CV → UFS (Export) - nested directory
+        let cv_nested = Path::from_str("/mnt/s3/dir1/dir2/file.txt").unwrap();
+        let ufs_nested = info.get_ufs_path(&cv_nested).unwrap();
+        assert_eq!(ufs_nested.full_path(), "s3://flink/user/dir1/dir2/file.txt");
+
+        // Test 5: UFS → CV (Import) - special characters in path
+        let ufs_special =
+            Path::from_str("s3://flink/user/test_data/dt=2025-01-30/part-00000.parquet").unwrap();
+        let cv_special = info.get_cv_path(&ufs_special).unwrap();
+        assert_eq!(
+            cv_special.full_path(),
+            "/mnt/s3/test_data/dt=2025-01-30/part-00000.parquet"
+        );
+
+        // Test 6: CV → UFS (Export) - special characters in path
+        let cv_special =
+            Path::from_str("/mnt/s3/test_data/dt=2025-01-30/part-00000.parquet").unwrap();
+        let ufs_special = info.get_ufs_path(&cv_special).unwrap();
+        assert_eq!(
+            ufs_special.full_path(),
+            "s3://flink/user/test_data/dt=2025-01-30/part-00000.parquet"
+        );
+
+        // Test 7: Verify is_cv() detection
+        assert!(cv_path.is_cv());
+        assert!(!ufs_path.is_cv());
+
+        // Test 8: Round-trip conversion (UFS → CV → UFS)
+        let original_ufs = Path::from_str("s3://flink/user/data/test.csv").unwrap();
+        let to_cv = info.get_cv_path(&original_ufs).unwrap();
+        let back_to_ufs = info.get_ufs_path(&to_cv).unwrap();
+        assert_eq!(original_ufs.full_path(), back_to_ufs.full_path());
+
+        // Test 9: Round-trip conversion (CV → UFS → CV)
+        let original_cv = Path::from_str("/mnt/s3/data/test.csv").unwrap();
+        let to_ufs = info.get_ufs_path(&original_cv).unwrap();
+        let back_to_cv = info.get_cv_path(&to_ufs).unwrap();
+        assert_eq!(original_cv.full_path(), back_to_cv.full_path());
+    }
 }
diff --git a/curvine-server/src/master/job/job_context.rs b/curvine-server/src/master/job/job_context.rs
@@ -82,6 +82,7 @@ impl JobContext {
             ttl_action,
             mount_info: mnt.clone(),
             create_time: LocalTime::mills() as i64,
+            overwrite: job_conf.overwrite,
         };
 
         JobContext {
diff --git a/curvine-server/src/master/job/job_manager.rs b/curvine-server/src/master/job/job_manager.rs
@@ -110,11 +110,10 @@ impl JobManager {
 
     pub fn submit_load_job(&self, command: LoadJobCommand) -> FsResult<LoadJobResult> {
         let source_path = Path::from_str(&command.source_path)?;
-        if source_path.is_cv() {
-            return err_box!("No need to load cv path");
-        }
 
-        // check mount
+        // Check mount info for both UFS and CV paths
+        // - For UFS path: Import (UFS → Curvine)
+        // - For CV path: Export (Curvine → UFS), requires mount info to determine target UFS
         let mnt = if let Some(mnt) = self.mount_manager.get_mount_info(&source_path)? {
             mnt
         } else {
diff --git a/curvine-server/src/master/job/job_runner.rs b/curvine-server/src/master/job/job_runner.rs
@@ -16,7 +16,6 @@ use crate::common::UfsFactory;
 use crate::master::fs::policy::ChooseContext;
 use crate::master::fs::MasterFilesystem;
 use crate::master::{JobContext, JobStore, TaskDetail};
-use curvine_client::unified::UfsFileSystem;
 use curvine_common::conf::ClientConf;
 use curvine_common::error::FsError;
 use curvine_common::fs::{FileSystem, Path};
@@ -110,18 +109,30 @@ impl LoadJobRunner {
         mnt: MountInfo,
     ) -> FsResult<LoadJobResult> {
         let source_path = Path::from_str(&command.source_path)?;
-        let target_path = mnt.get_cv_path(&source_path)?;
+
+        let target_path = if let Some(ref target) = command.target_path {
+            Path::from_str(target)?
+        } else {
+            if source_path.is_cv() {
+                mnt.get_ufs_path(&source_path)?
+            } else {
+                mnt.get_cv_path(&source_path)?
+            }
+        };
 
         let job_id = Self::create_job_id(source_path.full_path());
         let result = LoadJobResult {
             job_id: job_id.clone(),
             target_path: target_path.clone_path(),
         };
 
-        let ufs = self.factory.get_ufs(&mnt)?;
-        let source_status = ufs.get_status(&source_path).await?;
+        let source_status = if source_path.is_cv() {
+            self.master_fs.file_status(source_path.path())?
+        } else {
+            let ufs = self.factory.get_ufs(&mnt)?;
+            ufs.get_status(&source_path).await?
+        };
 
-        // check job status
         if self.check_job_exists(&job_id, &source_status, &target_path) {
             info!(
                 "job {}, source_path {} already exists",
@@ -136,13 +147,13 @@ impl LoadJobRunner {
             &command,
             job_id.clone(),
             source_path.clone_uri(),
-            target_path.clone_path(),
+            target_path.clone_uri(),
             &mnt,
             &ClientConf::default(),
         );
 
         let res = self
-            .create_all_tasks(&mut job_context, source_status, &ufs, &mnt)
+            .create_all_tasks(&mut job_context, source_status, &mnt)
             .await;
 
         match res {
@@ -189,7 +200,6 @@ impl LoadJobRunner {
         &self,
         job: &mut JobContext,
         source_status: FileStatus,
-        ufs: &UfsFileSystem,
         mnt: &MountInfo,
     ) -> FsResult<i64> {
         job.update_state(JobTaskState::Pending, "Assigning workers");
@@ -199,18 +209,46 @@ impl LoadJobRunner {
         let mut stack = LinkedList::new();
         let mut task_index = 0;
         stack.push_back(source_status);
+
+        // Get target base path for direction detection
+        let target_base = Path::from_str(&job.info.target_path)?;
+
         while let Some(status) = stack.pop_front() {
             if status.is_dir {
+                // List directory based on path type
                 let dir_path = Path::from_str(status.path)?;
-                let childs = ufs.list_status(&dir_path).await?;
+                let childs = if dir_path.is_cv() {
+                    // Traverse Curvine directory
+                    self.master_fs.list_status(dir_path.path())?
+                } else {
+                    // Traverse UFS directory
+                    let ufs = self.factory.get_ufs(mnt)?;
+                    ufs.list_status(&dir_path).await?
+                };
+
                 for child in childs {
                     stack.push_back(child);
                 }
             } else {
                 let worker = self.choose_worker(block_size)?;
 
                 let source_path = Path::from_str(status.path)?;
-                let target_path = mnt.get_cv_path(&source_path)?;
+
+                // Calculate target_path based on source and target types
+                let target_path = if source_path.is_cv() && !target_base.is_cv() {
+                    // Export: Curvine → UFS
+                    mnt.get_ufs_path(&source_path)?
+                } else if !source_path.is_cv() && target_base.is_cv() {
+                    // Import: UFS → Curvine
+                    mnt.get_cv_path(&source_path)?
+                } else {
+                    // Same type (Curvine→Curvine or UFS→UFS), not supported yet
+                    return err_box!(
+                        "Unsupported path combination: source={}, target={}",
+                        source_path.full_path(),
+                        target_base.full_path()
+                    );
+                };
 
                 let task_id = format!("{}_task_{}", job.info.job_id, task_index);
                 task_index += 1;
@@ -221,7 +259,7 @@ impl LoadJobRunner {
                     task_id: task_id.clone(),
                     worker: worker.clone(),
                     source_path: source_path.clone_uri(),
-                    target_path: target_path.clone_path(),
+                    target_path: target_path.clone_uri(),
                     create_time: LocalTime::mills() as i64,
                 };
                 job.add_task(task.clone());
diff --git a/curvine-server/src/worker/task/load_task_runner.rs b/curvine-server/src/worker/task/load_task_runner.rs
