Fix CopyFileOperator

apache · Dec 24, 2024 · a8ebd60 · a8ebd60
1 parent 9b79d56
commit a8ebd60
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 5 deletions.
diff --git a/...-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneFileInfo.java b/...-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneFileInfo.java
@@ -21,17 +21,26 @@
 /** The information of copy file. */
 public class CloneFileInfo {
 
+    private final String sourceFilePath;
     private final String filePathExcludeTableRoot;
     private final String sourceIdentifier;
     private final String targetIdentifier;
 
     public CloneFileInfo(
-            String filePathExcludeTableRoot, String sourceIdentifier, String targetIdentifier) {
+            String sourceFilePath,
+            String filePathExcludeTableRoot,
+            String sourceIdentifier,
+            String targetIdentifier) {
+        this.sourceFilePath = sourceFilePath;
         this.filePathExcludeTableRoot = filePathExcludeTableRoot;
         this.sourceIdentifier = sourceIdentifier;
         this.targetIdentifier = targetIdentifier;
     }
 
+    public String getSourceFilePath() {
+        return sourceFilePath;
+    }
+
     public String getFilePathExcludeTableRoot() {
         return filePathExcludeTableRoot;
     }
@@ -47,7 +56,7 @@ public String getTargetIdentifier() {
     @Override
     public String toString() {
         return String.format(
-                "{ filePath: %s, sourceIdentifier: %s, targetIdentifier: %s }",
-                filePathExcludeTableRoot, sourceIdentifier, targetIdentifier);
+                "{ sourceFilePath: %s, filePathExcludeTableRoot: %s, sourceIdentifier: %s, targetIdentifier: %s }",
+                sourceFilePath, filePathExcludeTableRoot, sourceIdentifier, targetIdentifier);
     }
 }
diff --git a/...ink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CopyFileOperator.java b/...ink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CopyFileOperator.java
@@ -99,7 +99,7 @@ public void processElement(StreamRecord<CloneFileInfo> streamRecord) throws Exce
                         });
 
         String filePathExcludeTableRoot = cloneFileInfo.getFilePathExcludeTableRoot();
-        Path sourcePath = new Path(sourceTableRootPath + filePathExcludeTableRoot);
+        Path sourcePath = new Path(cloneFileInfo.getSourceFilePath());
         Path targetPath = new Path(targetTableRootPath + filePathExcludeTableRoot);
 
         if (targetTableFileIO.exists(targetPath)

diff --git a/...n-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesForCloneOperator.java b/...n-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesForCloneOperator.java
@@ -123,7 +123,11 @@ private List<CloneFileInfo> toCloneFileInfos(
         for (Path file : files) {
             Path relativePath = getPathExcludeTableRoot(file, sourceTableRoot);
             result.add(
-                    new CloneFileInfo(relativePath.toString(), sourceIdentifier, targetIdentifier));
+                    new CloneFileInfo(
+                            file.toUri().toString(),
+                            relativePath.toString(),
+                            sourceIdentifier,
+                            targetIdentifier));
         }
         return result;
     }