fix

Author: JingsongLi

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

@@ -18,8 +18,11 @@
 
 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;
 
@@ -55,6 +58,18 @@ public CloneAction(
             String cloneFrom) {
         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;