fix

Author: JingsongLi

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java

@@ -18,11 +18,8 @@
 
 package org.apache.paimon.flink.action;
 
-import org.apache.paimon.catalog.CachingCatalog;
-import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.flink.clone.CloneHiveTableUtils;
 import org.apache.paimon.flink.clone.ClonePaimonTableUtils;
-import org.apache.paimon.hive.HiveCatalog;
 
 import javax.annotation.Nullable;
 
@@ -43,7 +40,6 @@ public class CloneAction extends ActionBase {
     private final int parallelism;
     @Nullable private final String whereSql;
     @Nullable private final List<String> excludedTables;
-    private final String cloneFrom;
 
     public CloneAction(
             String sourceDatabase,
@@ -54,22 +50,9 @@ public CloneAction(
             Map<String, String> targetCatalogConfig,
             @Nullable Integer parallelism,
             @Nullable String whereSql,
-            @Nullable List<String> excludedTables,
-            String cloneFrom) {
+            @Nullable List<String> excludedTables) {
         super(sourceCatalogConfig);
 
-        if (cloneFrom.equalsIgnoreCase("hive")) {
-            Catalog sourceCatalog = catalog;
-            if (sourceCatalog instanceof CachingCatalog) {
-                sourceCatalog = ((CachingCatalog) sourceCatalog).wrapped();
-            }
-            if (!(sourceCatalog instanceof HiveCatalog)) {
-                throw new UnsupportedOperationException(
-                        "Only support clone hive tables using HiveCatalog, but current source catalog is "
-                                + sourceCatalog.getClass().getName());
-            }
-        }
-
         this.sourceDatabase = sourceDatabase;
         this.sourceTableName = sourceTableName;
         this.sourceCatalogConfig = sourceCatalogConfig;
@@ -81,40 +64,36 @@ public CloneAction(
         this.parallelism = parallelism == null ? env.getParallelism() : parallelism;
         this.whereSql = whereSql;
         this.excludedTables = excludedTables;
-        this.cloneFrom = cloneFrom;
     }
 
     @Override
     public void build() throws Exception {
-        switch (cloneFrom) {
-            case "hive":
-                CloneHiveTableUtils.build(
-                        env,
-                        catalog,
-                        sourceDatabase,
-                        sourceTableName,
-                        sourceCatalogConfig,
-                        targetDatabase,
-                        targetTableName,
-                        targetCatalogConfig,
-                        parallelism,
-                        whereSql,
-                        excludedTables);
-                break;
-            case "paimon":
-                ClonePaimonTableUtils.build(
-                        env,
-                        catalog,
-                        sourceDatabase,
-                        sourceTableName,
-                        sourceCatalogConfig,
-                        targetDatabase,
-                        targetTableName,
-                        targetCatalogConfig,
-                        parallelism,
-                        whereSql,
-                        excludedTables);
-                break;
+        if (CloneHiveTableUtils.isRootHiveCatalog(catalog)) {
+            CloneHiveTableUtils.build(
+                    env,
+                    catalog,
+                    sourceDatabase,
+                    sourceTableName,
+                    sourceCatalogConfig,
+                    targetDatabase,
+                    targetTableName,
+                    targetCatalogConfig,
+                    parallelism,
+                    whereSql,
+                    excludedTables);
+        } else {
+            ClonePaimonTableUtils.build(
+                    env,
+                    catalog,
+                    sourceDatabase,
+                    sourceTableName,
+                    sourceCatalogConfig,
+                    targetDatabase,
+                    targetTableName,
+                    targetCatalogConfig,
+                    parallelism,
+                    whereSql,
+                    excludedTables);
         }
     }
 

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneActionFactory.java

@@ -37,7 +37,6 @@ public class CloneActionFactory implements ActionFactory {
     private static final String PARALLELISM = "parallelism";
     private static final String WHERE = "where";
     private static final String EXCLUDED_TABLES = "excluded_tables";
-    private static final String CLONE_FROM = "clone_from";
 
     @Override
     public String identifier() {
@@ -63,11 +62,6 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
                         ? null
                         : Arrays.asList(StringUtils.split(excludedTablesStr, ","));
 
-        String cloneFrom = params.get(CLONE_FROM);
-        if (StringUtils.isNullOrWhitespaceOnly(cloneFrom)) {
-            cloneFrom = "hive";
-        }
-
         CloneAction cloneAction =
                 new CloneAction(
                         params.get(DATABASE),
@@ -78,8 +72,7 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
                         targetCatalogConfig,
                         parallelism == null ? null : Integer.parseInt(parallelism),
                         params.get(WHERE),
-                        excludedTables,
-                        cloneFrom);
+                        excludedTables);
 
         return Optional.of(cloneAction);
     }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneHiveTableUtils.java

@@ -122,6 +122,10 @@ public static DataStream<Tuple2<Identifier, Identifier>> buildSource(
         return env.fromCollection(result).forceNonParallel();
     }
 
+    public static boolean isRootHiveCatalog(Catalog catalog) {
+        return DelegateCatalog.rootCatalog(catalog) instanceof HiveCatalog;
+    }
+
     public static HiveCatalog getRootHiveCatalog(Catalog catalog) {
         Catalog rootCatalog = DelegateCatalog.rootCatalog(catalog);
         checkArgument(

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CloneProcedure.java

@@ -61,10 +61,6 @@ public class CloneProcedure extends ProcedureBase {
                 @ArgumentHint(
                         name = "excluded_tables",
                         type = @DataTypeHint("STRING"),
-                        isOptional = true),
-                @ArgumentHint(
-                        name = "clone_from",
-                        type = @DataTypeHint("STRING"),
                         isOptional = true)
             })
     public String[] call(
@@ -77,8 +73,7 @@ public String[] call(
             String targetCatalogConfigStr,
             Integer parallelism,
             String where,
-            String excludedTablesStr,
-            String cloneFrom)
+            String excludedTablesStr)
             throws Exception {
         Map<String, String> sourceCatalogConfig =
                 new HashMap<>(optionalConfigMap(sourceCatalogConfigStr));
@@ -101,8 +96,7 @@ public String[] call(
                         targetCatalogConfig,
                         parallelism,
                         where,
-                        excludedTables,
-                        cloneFrom);
+                        excludedTables);
         return execute(procedureContext, action, "Clone Job");
     }
 

paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/CloneActionITCase.java

@@ -143,9 +143,7 @@ public void testCloneBucketedAppendFromPaimon() throws Exception {
                         "--target_table",
                         "target",
                         "--target_catalog_conf",
-                        "warehouse=" + warehouse2,
-                        "--clone_from",
-                        "paimon")
+                        "warehouse=" + warehouse2)
                 .run();
 
         sql(tEnv, "CALL catalog2.sys.compact(`table` => 'default.target')");