[flink] Make clone hive tables production-ready

Author: JingsongLi

docs/content/migration/clone-from-hive.md

@@ -0,0 +1,56 @@
+---
+title: "Clone From Hive"
+weight: 5
+type: docs
+aliases:
+- /migration/clone-from-hive.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Hive Table Clone
+
+Apache Hive supports Parquet, ORC file formats that could be cloned to Paimon. The cloned table will be
+[append table]({{< ref "append-table/overview" >}}). Supports file format of hive "orc" and "parquet" and "avro".
+
+## Clone Hive Table
+
+```bash
+<FLINK_HOME>/flink run ./paimon-flink-action-{{< version >}}.jar \
+clone_hive \
+--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
+```
+
+## Clone Hive Database
+
+```bash
+<FLINK_HOME>/flink run ./paimon-flink-action-{{< version >}}.jar \
+clone_hive \
+--database default
+--catalog_conf metastore=hive
+--catalog_conf uri=thrift://localhost:9088
+--target_database test
+--target_catalog_conf warehouse=my_warehouse
+```

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

@@ -19,17 +19,22 @@
 package org.apache.paimon.flink.action;
 
 import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.flink.clone.CloneHiveUtils;
+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.sink.FlinkStreamPartitioner;
 import org.apache.paimon.options.CatalogOptions;
 
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
 
 import javax.annotation.Nullable;
 
-import java.util.List;
 import java.util.Map;
 
 /** Clone source files managed by HiveMetaStore and commit metas to construct Paimon table. */
@@ -91,23 +96,33 @@ public void build() throws Exception {
                         source, new CloneHiveUtils.TableChannelComputer(), parallelism);
 
         // create target table, list files and group by <table, partition>
-        DataStream<List<CloneHiveUtils.CloneFilesInfo>> files =
+        DataStream<CloneFileInfo> files =
                 partitionedSource
                         .process(
-                                CloneHiveUtils.createTargetTableAndListFilesFunction(
+                                new ListHiveFilesFunction(
                                         sourceCatalogConfig, targetCatalogConfig, whereSql))
                         .name("List Files")
                         .setParallelism(parallelism);
 
         // copy files and commit
-        DataStream<Void> committed =
-                files.forward()
+        DataStream<DataFileInfo> dataFile =
+                files.rebalance()
                         .process(
-                                CloneHiveUtils.copyAndCommitFunction(
-                                        sourceCatalogConfig, targetCatalogConfig))
-                        .name("Copy and Commit")
+                                new CopyHiveFilesFunction(sourceCatalogConfig, targetCatalogConfig))
+                        .name("Copy Files")
                         .setParallelism(parallelism);
 
+        DataStream<DataFileInfo> partitionedDataFile =
+                FlinkStreamPartitioner.partition(
+                        dataFile, new CloneHiveUtils.DataFileChannelComputer(), parallelism);
+
+        DataStream<Long> committed =
+                partitionedDataFile
+                        .transform(
+                                "Commit table",
+                                BasicTypeInfo.LONG_TYPE_INFO,
+                                new CommitTableOperator(targetCatalogConfig))
+                        .setParallelism(parallelism);
         committed.sinkTo(new DiscardingSink<>()).name("end").setParallelism(1);
     }