fix

JingsongLi · JingsongLi · commit daf37ab60190 · 2025-07-16T12:18:50.000+08:00
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java
@@ -44,6 +44,7 @@ public class CloneAction extends ActionBase {
     @Nullable private final String whereSql;
     @Nullable private final List<String> excludedTables;
     private final String cloneFrom;
+    @Nullable private final Integer bucket;
 
     public CloneAction(
             String sourceDatabase,
@@ -55,8 +56,10 @@ public CloneAction(
             @Nullable Integer parallelism,
             @Nullable String whereSql,
             @Nullable List<String> excludedTables,
-            String cloneFrom) {
+            String cloneFrom,
+            @Nullable Integer bucket) {
         super(sourceCatalogConfig);
+        this.bucket = bucket;
 
         if (cloneFrom.equalsIgnoreCase("hive")) {
             Catalog sourceCatalog = catalog;
@@ -113,7 +116,8 @@ public void build() throws Exception {
                         targetCatalogConfig,
                         parallelism,
                         whereSql,
-                        excludedTables);
+                        excludedTables,
+                        bucket);
                 break;
         }
     }
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneActionFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneActionFactory.java
@@ -38,6 +38,7 @@ public class CloneActionFactory implements ActionFactory {
     private static final String WHERE = "where";
     private static final String EXCLUDED_TABLES = "excluded_tables";
     private static final String CLONE_FROM = "clone_from";
+    private static final String BUCKET = "bucket";
 
     @Override
     public String identifier() {
@@ -68,6 +69,12 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
             cloneFrom = "hive";
         }
 
+        String bucketString = params.get(BUCKET);
+        Integer bucket = null;
+        if (bucketString != null) {
+            bucket = Integer.valueOf(bucketString);
+        }
+
         CloneAction cloneAction =
                 new CloneAction(
                         params.get(DATABASE),
@@ -79,7 +86,8 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
                         parallelism == null ? null : Integer.parseInt(parallelism),
                         params.get(WHERE),
                         excludedTables,
-                        cloneFrom);
+                        cloneFrom,
+                        bucket);
 
         return Optional.of(cloneAction);
     }
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneHiveTableUtils.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneHiveTableUtils.java
@@ -22,6 +22,12 @@
 import org.apache.paimon.catalog.DelegateCatalog;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.flink.action.CloneAction;
+import org.apache.paimon.flink.clone.files.CloneFileInfo;
+import org.apache.paimon.flink.clone.files.CloneFilesFunction;
+import org.apache.paimon.flink.clone.files.CommitTableOperator;
+import org.apache.paimon.flink.clone.files.DataFileInfo;
+import org.apache.paimon.flink.clone.files.ListCloneFilesFunction;
+import org.apache.paimon.flink.clone.files.ShuffleDataFileByTableComputer;
 import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
 import org.apache.paimon.hive.HiveCatalog;
 import org.apache.paimon.utils.StringUtils;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/ClonePaimonTableUtils.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/ClonePaimonTableUtils.java
@@ -21,6 +21,12 @@
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.flink.action.CloneAction;
+import org.apache.paimon.flink.clone.spits.CloneSplitInfo;
+import org.apache.paimon.flink.clone.spits.CloneSplitsFunction;
+import org.apache.paimon.flink.clone.spits.CommitMessageInfo;
+import org.apache.paimon.flink.clone.spits.CommitMessageTableOperator;
+import org.apache.paimon.flink.clone.spits.ListCloneSplitsFunction;
+import org.apache.paimon.flink.clone.spits.ShuffleCommitMessageByTableComputer;
 import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
 import org.apache.paimon.utils.StringUtils;
 
@@ -123,7 +129,8 @@ public static void build(
             Map<String, String> targetCatalogConfig,
             int parallelism,
             @Nullable String whereSql,
-            @Nullable List<String> excludedTables)
+            @Nullable List<String> excludedTables,
+            @Nullable Integer bucket)
             throws Exception {
         // list source tables
         DataStream<Tuple2<Identifier, Identifier>> source =
@@ -144,7 +151,7 @@ public static void build(
                 partitionedSource
                         .process(
                                 new ListCloneSplitsFunction(
-                                        sourceCatalogConfig, targetCatalogConfig, whereSql))
+                                        sourceCatalogConfig, targetCatalogConfig, whereSql, bucket))
                         .name("List Files")
                         .setParallelism(parallelism);
 
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneUtils.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneUtils.java
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/CloneFileInfo.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/CloneFileInfo.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.files;
 
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.BinaryRow;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/CloneFilesFunction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/CloneFilesFunction.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.files;
 
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.BinaryRow;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/CloneProcessFunction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/CloneProcessFunction.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.files;
 
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/CommitTableOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/CommitTableOperator.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.files;
 
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/DataFileInfo.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/DataFileInfo.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.files;
 
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.BinaryRow;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/ListCloneFilesFunction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/ListCloneFilesFunction.java
@@ -16,10 +16,9 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.files;
 
 import org.apache.paimon.CoreOptions;
-import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.flink.predicate.SimpleSqlPredicateConvertor;
@@ -169,9 +168,8 @@ private void checkCompatible(Schema sourceSchema, FileStoreTable existedTable) {
                 "source table partition keys is not compatible with existed paimon table partition keys.");
     }
 
-    @VisibleForTesting
     @Nullable
-    static PartitionPredicate getPartitionPredicate(
+    public static PartitionPredicate getPartitionPredicate(
             @Nullable String whereSql, RowType partitionType, Identifier tableId) throws Exception {
         if (whereSql == null) {
             return null;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/ShuffleDataFileByTableComputer.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/files/ShuffleDataFileByTableComputer.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.files;
 
 import org.apache.paimon.table.sink.ChannelComputer;
 
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/spits/CloneSplitInfo.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/spits/CloneSplitInfo.java
@@ -16,11 +16,13 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.spits;
 
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.table.source.Split;
+import org.apache.paimon.utils.InstantiationUtil;
 
+import java.io.IOException;
 import java.io.Serializable;
 
 /** Clone split with necessary information. */
@@ -30,12 +32,16 @@ public class CloneSplitInfo implements Serializable {
 
     private final Identifier sourceIdentifier;
     private final Identifier targetidentifier;
-    private final Split split;
+    private final byte[] split;
 
     public CloneSplitInfo(Identifier sourceIdentifier, Identifier targetidentifier, Split split) {
         this.sourceIdentifier = sourceIdentifier;
         this.targetidentifier = targetidentifier;
-        this.split = split;
+        try {
+            this.split = InstantiationUtil.serializeObject(split);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     public Identifier sourceIdentifier() {
@@ -47,6 +53,10 @@ public Identifier targetidentifier() {
     }
 
     public Split split() {
-        return split;
+        try {
+            return InstantiationUtil.deserializeObject(split, getClass().getClassLoader());
+        } catch (IOException | ClassNotFoundException e) {
+            throw new RuntimeException(e);
+        }
     }
 }
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/spits/CloneSplitsFunction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/spits/CloneSplitsFunction.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.spits;
 
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.options.Options;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/spits/CommitMessageInfo.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/spits/CommitMessageInfo.java
@@ -16,11 +16,13 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.spits;
 
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.utils.InstantiationUtil;
 
+import java.io.IOException;
 import java.io.Serializable;
 
 /** Commit message with necessary information. */
@@ -29,18 +31,26 @@ public class CommitMessageInfo implements Serializable {
     private static final long serialVersionUID = 1L;
 
     private final Identifier identifier;
-    private final CommitMessage commitMessage;
+    private final byte[] commitMessage;
 
     public CommitMessageInfo(Identifier identifier, CommitMessage commitMessage) {
         this.identifier = identifier;
-        this.commitMessage = commitMessage;
+        try {
+            this.commitMessage = InstantiationUtil.serializeObject(commitMessage);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     public Identifier identifier() {
         return identifier;
     }
 
     public CommitMessage commitMessage() {
-        return commitMessage;
+        try {
+            return InstantiationUtil.deserializeObject(commitMessage, getClass().getClassLoader());
+        } catch (IOException | ClassNotFoundException e) {
+            throw new RuntimeException(e);
+        }
     }
 }
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/spits/CommitMessageTableOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/spits/CommitMessageTableOperator.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.spits;
 
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/spits/ListCloneSplitsFunction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/spits/ListCloneSplitsFunction.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.spits;
 
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
@@ -39,8 +39,9 @@
 import java.util.Map;
 
 import static org.apache.paimon.CoreOptions.BUCKET;
+import static org.apache.paimon.CoreOptions.PATH;
 import static org.apache.paimon.flink.FlinkCatalogFactory.createPaimonCatalog;
-import static org.apache.paimon.flink.clone.ListCloneFilesFunction.getPartitionPredicate;
+import static org.apache.paimon.flink.clone.files.ListCloneFilesFunction.getPartitionPredicate;
 
 /** List splits for table. */
 public class ListCloneSplitsFunction
@@ -51,17 +52,20 @@ public class ListCloneSplitsFunction
     private final Map<String, String> sourceCatalogConfig;
     private final Map<String, String> targetCatalogConfig;
     @Nullable private final String whereSql;
+    @Nullable private final Integer bucket;
 
     private transient Catalog sourceCatalog;
     private transient Catalog targetCatalog;
 
     public ListCloneSplitsFunction(
             Map<String, String> sourceCatalogConfig,
             Map<String, String> targetCatalogConfig,
-            @Nullable String whereSql) {
+            @Nullable String whereSql,
+            @Nullable Integer bucket) {
         this.sourceCatalogConfig = sourceCatalogConfig;
         this.targetCatalogConfig = targetCatalogConfig;
         this.whereSql = whereSql;
+        this.bucket = bucket;
     }
 
     /**
@@ -104,13 +108,18 @@ public void processElement(
                 .options()
                 .forEach(
                         (k, v) -> {
-                            if (k.equalsIgnoreCase(BUCKET.key())) {
+                            if (k.equalsIgnoreCase(BUCKET.key())
+                                    || k.equalsIgnoreCase(PATH.key())) {
                                 // use default bucket from catalog
                                 return;
                             }
                             builder.option(k, v);
                         });
 
+        if (bucket != null) {
+            builder.option(BUCKET.key(), bucket.toString());
+        }
+
         targetCatalog.createTable(tuple.f1, builder.build(), true);
 
         PartitionPredicate predicate =
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/spits/ShuffleCommitMessageByTableComputer.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/spits/ShuffleCommitMessageByTableComputer.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone;
+package org.apache.paimon.flink.clone.spits;
 
 import org.apache.paimon.table.sink.ChannelComputer;
 
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CloneProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CloneProcedure.java
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/clone/ListCloneFilesFunctionTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/clone/ListCloneFilesFunctionTest.java
diff --git a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/CloneActionITCase.java b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/CloneActionITCase.java
