[flink] Rename clone_hive to clone and add procedure

Author: JingsongLi

docs/content/migration/clone-from-hive.md docs/content/migration/clone-to-paimon.mddocs/content/migration/clone-from-hive.md renamed to docs/content/migration/clone-to-paimon.md

@@ -1,9 +1,9 @@
 ---
-title: "Clone From Hive"
+title: "Clone To Paimon"
 weight: 5
 type: docs
 aliases:
-- /migration/clone-from-hive.html
+- /migration/clone.html
 ---
 <!--
 Licensed to the Apache Software Foundation (ASF) under one
@@ -24,27 +24,36 @@ specific language governing permissions and limitations
 under the License.
 -->
 
-# Clone Hive Table
+# Clone To Paimon
 
-Clone Hive Table supports cloning hive tables with parquet, orc and avro formats. The cloned table will be
-[append table]({{< ref "append-table/overview" >}}).
+Clone supports cloning tables to Paimon tables.
 
 1. Clone is `OVERWRITE` semantic that will overwrite the partitions of the target table according to the data.
 2. Clone is reentrant, but it requires existing tables to contain all fields from the source table and have the
    same partition fields.
 
+Currently, clone supports:
+
+1. Clone Hive tables in Hive Catalog to Paimon Catalog, supports Parquet, ORC, Avro formats, target table will
+   be append table.
+
+The source table below is currently under development:
+1. Clone Hudi tables in Hive Catalog to Paimon Catalog, target table will be append table.
+2. Clone Paimon tables to Paimon tables, target table can be primary table or append table.
+
 ## Clone Hive Table
 
 ```bash
 <FLINK_HOME>/flink run ./paimon-flink-action-{{< version >}}.jar \
-clone_hive \
+clone \
 --database default \
 --table hivetable \
 --catalog_conf metastore=hive \
 --catalog_conf uri=thrift://localhost:9088 \
 --target_database test \
 --target_table test_table \
 --target_catalog_conf warehouse=my_warehouse \
+--parallelism 10 \
 --where <filter_spec>
 ```
 
@@ -54,10 +63,11 @@ You can use filter spec to specify the filtering condition for the partition.
 
 ```bash
 <FLINK_HOME>/flink run ./paimon-flink-action-{{< version >}}.jar \
-clone_hive \
+clone \
 --database default \
 --catalog_conf metastore=hive \
 --catalog_conf uri=thrift://localhost:9088 \
 --target_database test \
+--parallelism 10 \
 --target_catalog_conf warehouse=my_warehouse
 ```

…paimon/flink/action/CloneHiveAction.java …che/paimon/flink/action/CloneAction.javapaimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneHiveAction.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneHiveAction.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java

@@ -21,12 +21,12 @@
 import org.apache.paimon.catalog.CachingCatalog;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.flink.clone.hive.CloneFileInfo;
-import org.apache.paimon.flink.clone.hive.CloneHiveUtils;
-import org.apache.paimon.flink.clone.hive.CommitTableOperator;
-import org.apache.paimon.flink.clone.hive.CopyHiveFilesFunction;
-import org.apache.paimon.flink.clone.hive.DataFileInfo;
-import org.apache.paimon.flink.clone.hive.ListHiveFilesFunction;
+import org.apache.paimon.flink.clone.CloneFileInfo;
+import org.apache.paimon.flink.clone.CloneFilesFunction;
+import org.apache.paimon.flink.clone.CloneUtils;
+import org.apache.paimon.flink.clone.CommitTableOperator;
+import org.apache.paimon.flink.clone.DataFileInfo;
+import org.apache.paimon.flink.clone.ListCloneFilesFunction;
 import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
 import org.apache.paimon.hive.HiveCatalog;
 
@@ -39,8 +39,8 @@
 
 import java.util.Map;
 
-/** Clone source files managed by HiveMetaStore and commit metas to construct Paimon table. */
-public class CloneHiveAction extends ActionBase {
+/** Clone source table to target table. */
+public class CloneAction extends ActionBase {
 
     private final Map<String, String> sourceCatalogConfig;
     private final String sourceDatabase;
@@ -53,7 +53,7 @@ public class CloneHiveAction extends ActionBase {
     private final int parallelism;
     @Nullable private final String whereSql;
 
-    public CloneHiveAction(
+    public CloneAction(
             String sourceDatabase,
             String sourceTableName,
             Map<String, String> sourceCatalogConfig,
@@ -90,7 +90,7 @@ public CloneHiveAction(
     public void build() throws Exception {
         // list source tables
         DataStream<Tuple2<Identifier, Identifier>> source =
-                CloneHiveUtils.buildSource(
+                CloneUtils.buildSource(
                         sourceDatabase,
                         sourceTableName,
                         targetDatabase,
@@ -100,28 +100,27 @@ public void build() throws Exception {
 
         DataStream<Tuple2<Identifier, Identifier>> partitionedSource =
                 FlinkStreamPartitioner.partition(
-                        source, new CloneHiveUtils.TableChannelComputer(), parallelism);
+                        source, new CloneUtils.TableChannelComputer(), parallelism);
 
         // create target table, list files and group by <table, partition>
         DataStream<CloneFileInfo> files =
                 partitionedSource
                         .process(
-                                new ListHiveFilesFunction(
+                                new ListCloneFilesFunction(
                                         sourceCatalogConfig, targetCatalogConfig, whereSql))
                         .name("List Files")
                         .setParallelism(parallelism);
 
         // copy files and commit
         DataStream<DataFileInfo> dataFile =
                 files.rebalance()
-                        .process(
-                                new CopyHiveFilesFunction(sourceCatalogConfig, targetCatalogConfig))
+                        .process(new CloneFilesFunction(sourceCatalogConfig, targetCatalogConfig))
                         .name("Copy Files")
                         .setParallelism(parallelism);
 
         DataStream<DataFileInfo> partitionedDataFile =
                 FlinkStreamPartitioner.partition(
-                        dataFile, new CloneHiveUtils.DataFileChannelComputer(), parallelism);
+                        dataFile, new CloneUtils.DataFileChannelComputer(), parallelism);
 
         DataStream<Long> committed =
                 partitionedDataFile

…flink/action/CloneHiveActionFactory.java …mon/flink/action/CloneActionFactory.javapaimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneHiveActionFactory.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneActionFactory.java paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneHiveActionFactory.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneActionFactory.java

@@ -22,10 +22,10 @@
 import java.util.Map;
 import java.util.Optional;
 
-/** Factory to create {@link CloneHiveAction}. */
-public class CloneHiveActionFactory implements ActionFactory {
+/** Factory to create {@link CloneAction}. */
+public class CloneActionFactory implements ActionFactory {
 
-    private static final String IDENTIFIER = "clone_hive";
+    private static final String IDENTIFIER = "clone";
     private static final String TARGET_WAREHOUSE = "target_warehouse";
     private static final String TARGET_DATABASE = "target_database";
     private static final String TARGET_TABLE = "target_table";
@@ -51,8 +51,8 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
 
         String parallelism = params.get(PARALLELISM);
 
-        CloneHiveAction cloneHiveAction =
-                new CloneHiveAction(
+        CloneAction cloneAction =
+                new CloneAction(
                         params.get(DATABASE),
                         params.get(TABLE),
                         catalogConfig,
@@ -62,13 +62,13 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
                         parallelism == null ? null : Integer.parseInt(parallelism),
                         params.get(WHERE));
 
-        return Optional.of(cloneHiveAction);
+        return Optional.of(cloneAction);
     }
 
     @Override
     public void printHelp() {
         System.out.println(
-                "Action \"clone_hive\" clones the source files and migrate them to paimon table.");
+                "Action \"clone\" clones the source files and migrate them to paimon table.");
         System.out.println();
     }
 }

…imon/flink/clone/hive/CloneFileInfo.java …he/paimon/flink/clone/CloneFileInfo.javapaimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/CloneFileInfo.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneFileInfo.java paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/CloneFileInfo.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneFileInfo.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone.hive;
+package org.apache.paimon.flink.clone;
 
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.BinaryRow;

…nk/clone/hive/CopyHiveFilesFunction.java …imon/flink/clone/CloneFilesFunction.javapaimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/CopyHiveFilesFunction.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneFilesFunction.java paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/CopyHiveFilesFunction.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneFilesFunction.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone.hive;
+package org.apache.paimon.flink.clone;
 
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.BinaryRow;
@@ -36,14 +36,14 @@
 import java.util.HashMap;
 import java.util.Map;
 
-/** Copy files for table. */
-public class CopyHiveFilesFunction extends CopyProcessFunction<CloneFileInfo, DataFileInfo> {
+/** Clone files for table. */
+public class CloneFilesFunction extends CloneProcessFunction<CloneFileInfo, DataFileInfo> {
 
     private static final long serialVersionUID = 1L;
 
     private transient Map<Identifier, Map<BinaryRow, DataFilePathFactory>> pathFactoryMap;
 
-    public CopyHiveFilesFunction(
+    public CloneFilesFunction(
             Map<String, String> sourceCatalogConfig, Map<String, String> targetCatalogConfig) {
         super(sourceCatalogConfig, targetCatalogConfig);
     }
@@ -92,7 +92,16 @@ public void processElement(
     }
 
     private DataFilePathFactory pathFactory(Identifier identifier, BinaryRow part) {
-        FileStoreTable targetTable = (FileStoreTable) getTable(identifier);
-        return targetTable.store().pathFactory().createDataFilePathFactory(part, 0);
+        return pathFactoryMap
+                .computeIfAbsent(identifier, k -> new HashMap<>())
+                .computeIfAbsent(
+                        part,
+                        k -> {
+                            FileStoreTable targetTable = (FileStoreTable) getTable(identifier);
+                            return targetTable
+                                    .store()
+                                    .pathFactory()
+                                    .createDataFilePathFactory(part, 0);
+                        });
     }
 }

…link/clone/hive/CopyProcessFunction.java …on/flink/clone/CloneProcessFunction.javapaimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/CopyProcessFunction.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneProcessFunction.java paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/CopyProcessFunction.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneProcessFunction.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone.hive;
+package org.apache.paimon.flink.clone;
 
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
@@ -35,12 +35,12 @@
 import java.util.Map;
 
 import static org.apache.paimon.flink.FlinkCatalogFactory.createPaimonCatalog;
-import static org.apache.paimon.flink.clone.hive.CloneHiveUtils.getRootHiveCatalog;
+import static org.apache.paimon.flink.clone.CloneUtils.getRootHiveCatalog;
 
 /** Abstract function for copying tables. */
-public abstract class CopyProcessFunction<I, O> extends ProcessFunction<I, O> {
+public abstract class CloneProcessFunction<I, O> extends ProcessFunction<I, O> {
 
-    protected static final Logger LOG = LoggerFactory.getLogger(CopyProcessFunction.class);
+    protected static final Logger LOG = LoggerFactory.getLogger(CloneProcessFunction.class);
 
     protected final Map<String, String> sourceCatalogConfig;
     protected final Map<String, String> targetCatalogConfig;
@@ -51,7 +51,7 @@ public abstract class CopyProcessFunction<I, O> extends ProcessFunction<I, O> {
     protected transient Map<Identifier, Table> tableCache;
     protected transient DataFileMetaSerializer dataFileSerializer;
 
-    public CopyProcessFunction(
+    public CloneProcessFunction(
             Map<String, String> sourceCatalogConfig, Map<String, String> targetCatalogConfig) {
         this.sourceCatalogConfig = sourceCatalogConfig;
         this.targetCatalogConfig = targetCatalogConfig;

…mon/flink/clone/hive/CloneHiveUtils.java …pache/paimon/flink/clone/CloneUtils.javapaimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/CloneHiveUtils.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneUtils.java paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/CloneHiveUtils.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneUtils.java

@@ -16,12 +16,12 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone.hive;
+package org.apache.paimon.flink.clone;
 
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.DelegateCatalog;
 import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.flink.action.CloneHiveAction;
+import org.apache.paimon.flink.action.CloneAction;
 import org.apache.paimon.hive.HiveCatalog;
 import org.apache.paimon.hive.migrate.HiveCloneUtils;
 import org.apache.paimon.table.sink.ChannelComputer;
@@ -40,10 +40,10 @@
 import static org.apache.paimon.utils.Preconditions.checkArgument;
 import static org.apache.paimon.utils.Preconditions.checkState;
 
-/** Utils for building {@link CloneHiveAction}. */
-public class CloneHiveUtils {
+/** Utils for building {@link CloneAction}. */
+public class CloneUtils {
 
-    private static final Logger LOG = LoggerFactory.getLogger(CloneHiveUtils.class);
+    private static final Logger LOG = LoggerFactory.getLogger(CloneUtils.class);
 
     public static DataStream<Tuple2<Identifier, Identifier>> buildSource(
             String sourceDatabase,

…link/clone/hive/CommitTableOperator.java …mon/flink/clone/CommitTableOperator.javapaimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/CommitTableOperator.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CommitTableOperator.java paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/CommitTableOperator.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CommitTableOperator.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone.hive;
+package org.apache.paimon.flink.clone;
 
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;

…aimon/flink/clone/hive/DataFileInfo.java …che/paimon/flink/clone/DataFileInfo.javapaimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/DataFileInfo.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/DataFileInfo.java paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/DataFileInfo.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/DataFileInfo.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone.hive;
+package org.apache.paimon.flink.clone;
 
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.BinaryRow;

…nk/clone/hive/ListHiveFilesFunction.java …/flink/clone/ListCloneFilesFunction.javapaimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/ListHiveFilesFunction.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/ListCloneFilesFunction.java paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/hive/ListHiveFilesFunction.java renamed to paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/ListCloneFilesFunction.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.clone.hive;
+package org.apache.paimon.flink.clone;
 
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.annotation.VisibleForTesting;
@@ -48,14 +48,14 @@
 import static org.apache.paimon.utils.Preconditions.checkState;
 
 /** List files for table. */
-public class ListHiveFilesFunction
-        extends CopyProcessFunction<Tuple2<Identifier, Identifier>, CloneFileInfo> {
+public class ListCloneFilesFunction
+        extends CloneProcessFunction<Tuple2<Identifier, Identifier>, CloneFileInfo> {
 
     private static final long serialVersionUID = 1L;
 
     @Nullable private final String whereSql;
 
-    public ListHiveFilesFunction(
+    public ListCloneFilesFunction(
             Map<String, String> sourceCatalogConfig,
             Map<String, String> targetCatalogConfig,
             @Nullable String whereSql) {