[procedure] Merge migrate_file to migrate_table

Author: JingsongLi

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

@@ -30,59 +30,41 @@ Apache Hive supports ORC, Parquet file formats that could be migrated to Paimon.
 When migrating data to a paimon table, the origin table will be permanently disappeared. So please back up your data if you
 still need the original table. The migrated table will be [append table]({{< ref "append-table/overview" >}}).
 
-Now, we can use paimon hive catalog with Migrate Table Procedure and Migrate File Procedure to totally migrate a table from hive to paimon.
+Now, we can use paimon hive catalog with Migrate Table Procedure to totally migrate a table from hive to paimon.
 At the same time, you can use paimon hive catalog with Migrate Database Procedure to fully synchronize all tables in the database to paimon.
 
 * Migrate Table Procedure: Paimon table does not exist, use the procedure upgrade hive table to paimon table. Hive table will disappear after action done.
 * Migrate Database Procedure: Paimon table does not exist, use the procedure upgrade all hive tables in database to paimon table. All hive tables will disappear after action done.
-* Migrate File Procedure:  Paimon table already exists, use the procedure to migrate files from hive table to paimon table. **Notice that, Hive table will also disappear after action done.**
 
 These three actions now support file format of hive "orc" and "parquet" and "avro".
 
 <span style="color: red; "> **We highly recommend to back up hive table data before migrating, because migrating action is not atomic. If been interrupted while migrating, you may lose your data.** </span>
 
-## Example for Migration
-
-**Migrate Hive Table**
-
-Command: <br>
-
-***CALL <font color="green">sys.migrate_table</font>(&#39;hive&#39;, &#39;&lt;hive_database&gt;.&lt;hive_tablename&gt;&#39;, &#39;&lt;paimon_tableconf&gt;&#39;);***
-
-**Example**
+## Migrate Hive Table
 
+{{< tabs "migrate table" >}}
+{{< tab "Flink SQL" >}}
 ```sql
-CREATE CATALOG PAIMON WITH ('type'='paimon', 'metastore' = 'hive', 'uri' = 'thrift://localhost:9083', 'warehouse'='/path/to/warehouse/');
+CREATE CATALOG PAIMON WITH (
+   'type'='paimon',
+   'metastore' = 'hive',
+   'uri' = 'thrift://localhost:9083',
+   'warehouse'='/path/to/warehouse/');
 
 USE CATALOG PAIMON;
 
-CALL sys.migrate_table(connector => 'hive', source_table => 'default.hivetable', options => 'file.format=orc');
-```
-After invoke, "hivetable" will totally convert to paimon format. Writing and reading the table by old "hive way" will fail.
-We can add our table properties while importing by sys.migrate_table('<database>.<tablename>', '<tableproperties>').
-<tableproperties> here should be separated by ",".  For example:
-
-```sql
 CALL sys.migrate_table(
-    connector => 'hive', 
-    source_table => 'my_db.wait_to_upgrate', 
-    options => 'file.format=orc,read.batch-size=2096,write-only=true'
-);
+    connector => 'hive',
+    source_table => 'default.hivetable',
+    -- You can specify the target table, and if the target table already exists
+    -- the file will be migrated directly to it
+    -- target_table => 'default.paimontarget',
+    -- You can specify delete_origin is false, this won't delete hivetable
+    -- delete_origin => false,
+    options => 'file.format=orc');
 ```
-
-If your flink version is below 1.17, you can use flink action to achieve this:
-```bash
-<FLINK_HOME>/bin/flink run \
-/path/to/paimon-flink-action-{{< version >}}.jar \
-migrate_table \
---warehouse <warehouse-path> \
---source_type hive \
---table <database.table-name> \
-[--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \
-[--options <paimon-table-conf  [,paimon-table-conf ...]> ]
-```
-
-Example:
+{{< /tab >}}
+{{< tab "Flink Action" >}}
 ```bash
 <FLINK_HOME>/flink run ./paimon-flink-action-{{< version >}}.jar \
 migrate_table \
@@ -92,35 +74,31 @@ migrate_table \
 --source_type hive \
 --table default.hive_or_paimon
 ```
+{{< /tab >}}
+{{< /tabs >}}
 
-**Migrate Hive Database**
-
-Command: <br>
-
-***CALL <font color="green">sys.migrate_database</font>(&#39;hive&#39;, &#39;&lt;hive_database&gt;&#39;, &#39;&lt;paimon_tableconf&gt;&#39;);***
+After invoke, "hivetable" will totally convert to paimon format. Writing and reading the table by old "hive way" will fail.
 
-**Example**
+## Migrate Hive Database
 
+{{< tabs "migrate database" >}}
+{{< tab "Flink SQL" >}}
 ```sql
-CREATE CATALOG PAIMON WITH ('type'='paimon', 'metastore' = 'hive', 'uri' = 'thrift://localhost:9083', 'warehouse'='/path/to/warehouse/');
+CREATE CATALOG PAIMON WITH (
+   'type'='paimon', 
+   'metastore' = 'hive', 
+   'uri' = 'thrift://localhost:9083', 
+   'warehouse'='/path/to/warehouse/');
 
 USE CATALOG PAIMON;
 
-CALL sys.migrate_database(connector => 'hive', source_database => 'default', options => 'file.format=orc');
-```
-After invoke, all tables in "default" database will totally convert to paimon format. Writing and reading the table by old "hive way" will fail.
-We can add our table properties while importing by sys.migrate_database('<database>', '<tableproperties>').
-<tableproperties> here should be separated by ",".  For example:
-
-```sql
 CALL sys.migrate_database(
-    connector => 'hive', 
-    source_database => 'my_db', 
-    options => 'file.format=orc,read.batch-size=2096,write-only=true'
-);
+    connector => 'hive',
+    source_database => 'default',
+    options => 'file.format=orc');
 ```
-
-If your flink version is below 1.17, you can use flink action to achieve this:
+{{< /tab >}}
+{{< tab "Flink Action" >}}
 ```bash
 <FLINK_HOME>/bin/flink run \
 /path/to/paimon-flink-action-{{< version >}}.jar \
@@ -141,21 +119,7 @@ Example:
 --source_type hive \
 --database default
 ```
+{{< /tab >}}
+{{< /tabs >}}
 
-**Migrate Hive File**
-
-Command: <br>
-
-***CALL <font color="green">sys.migrate_file</font>(&#39;hive&#39;, &#39;&lt;hive_database&gt;.&lt;hive_table_name&gt;&#39;, &#39;&lt;paimon_database&gt;.&lt;paimon_tablename&gt;&#39;);***
-
-**Example**
-
-```sql
-CREATE CATALOG PAIMON WITH ('type'='paimon', 'metastore' = 'hive', 'uri' = 'thrift://localhost:9083', 'warehouse'='/path/to/warehouse/');
-
-USE CATALOG PAIMON;
-
-CALL sys.migrate_file(connector => 'hive', source_table => 'default.hivetable', target_table => 'default.paimontable');
-```
-After invoke, "hivetable" will disappear. And all files will be moved and renamed to paimon directory. "paimontable" here must have the same
-partition keys with "hivetable", and "paimontable" should be in unaware-bucket mode.
+After invoke, all tables in "default" database will totally convert to paimon format. Writing and reading the table by old "hive way" will fail.

paimon-core/src/main/java/org/apache/paimon/migrate/Migrator.java

@@ -25,5 +25,5 @@ public interface Migrator {
 
     void renameTable(boolean ignoreIfNotExists) throws Exception;
 
-    public void deleteOriginTable(boolean delete) throws Exception;
+    void deleteOriginTable(boolean delete) throws Exception;
 }

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

@@ -53,7 +53,9 @@ public void run() throws Exception {
                 new DefaultProcedureContext(env),
                 connector,
                 hiveTableFullName,
+                null,
                 tableProperties,
-                parallelism);
+                parallelism,
+                null);
     }
 }

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

@@ -21,7 +21,6 @@
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.flink.utils.TableMigrationUtils;
 import org.apache.paimon.migrate.Migrator;
-import org.apache.paimon.utils.ParameterUtils;
 
 import org.apache.flink.table.annotation.ArgumentHint;
 import org.apache.flink.table.annotation.DataTypeHint;
@@ -30,6 +29,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.paimon.utils.ParameterUtils.parseCommaSeparatedKeyValues;
+
 /** Migrate procedure to migrate hive table to paimon table. */
 public class MigrateTableProcedure extends ProcedureBase {
 
@@ -46,25 +47,34 @@ public String identifier() {
             argument = {
                 @ArgumentHint(name = "connector", type = @DataTypeHint("STRING")),
                 @ArgumentHint(name = "source_table", type = @DataTypeHint("STRING")),
+                @ArgumentHint(name = "target_table", type = @DataTypeHint("STRING")),
                 @ArgumentHint(name = "options", type = @DataTypeHint("STRING"), isOptional = true),
                 @ArgumentHint(
                         name = "parallelism",
                         type = @DataTypeHint("Integer"),
+                        isOptional = true),
+                @ArgumentHint(
+                        name = "delete_origin",
+                        type = @DataTypeHint("BOOLEAN"),
                         isOptional = true)
             })
     public String[] call(
             ProcedureContext procedureContext,
             String connector,
-            String sourceTablePath,
+            String sourceTable,
+            String targetTable,
             String properties,
-            Integer parallelism)
+            Integer parallelism,
+            Boolean deleteOrigin)
             throws Exception {
-        properties = notnull(properties);
-
-        String targetPaimonTablePath = sourceTablePath + PAIMON_SUFFIX;
+        if (targetTable == null && !deleteOrigin) {
+            throw new IllegalArgumentException("delete_origin is false but targetTable is null.");
+        }
 
-        Identifier sourceTableId = Identifier.fromString(sourceTablePath);
-        Identifier targetTableId = Identifier.fromString(targetPaimonTablePath);
+        Identifier sourceTableId = Identifier.fromString(sourceTable);
+        Identifier targetTableId =
+                Identifier.fromString(
+                        targetTable == null ? sourceTable + PAIMON_SUFFIX : targetTable);
 
         Integer p = parallelism == null ? Runtime.getRuntime().availableProcessors() : parallelism;
 
@@ -77,11 +87,14 @@ public String[] call(
                         targetTableId.getDatabaseName(),
                         targetTableId.getObjectName(),
                         p,
-                        ParameterUtils.parseCommaSeparatedKeyValues(properties));
+                        parseCommaSeparatedKeyValues(notnull(properties)));
         LOG.info("create migrator success.");
+        migrator.deleteOriginTable(deleteOrigin);
         migrator.executeMigrate();
 
-        migrator.renameTable(false);
+        if (targetTable == null) {
+            migrator.renameTable(false);
+        }
         return new String[] {"Success"};
     }
 }

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

@@ -103,11 +103,11 @@ public void test(String format, boolean isNamedArgument) throws Exception {
                 "CREATE TABLE paimontable (id STRING, id2 INT, id3 INT) PARTITIONED BY (id2, id3) with ('bucket' = '-1');");
         if (isNamedArgument) {
             tEnv.executeSql(
-                            "CALL sys.migrate_file(connector => 'hive', source_table => 'default.hivetable', target_table => 'default.paimontable')")
+                            "CALL sys.migrate_table(connector => 'hive', source_table => 'default.hivetable', target_table => 'default.paimontable')")
                     .await();
         } else {
             tEnv.executeSql(
-                            "CALL sys.migrate_file('hive', 'default.hivetable', 'default.paimontable')")
+                            "CALL sys.migrate_table('hive', 'default.hivetable', 'default.paimontable')")
                     .await();
         }
         List<Row> r2 = ImmutableList.copyOf(tEnv.executeSql("SELECT * FROM paimontable").collect());
@@ -149,11 +149,11 @@ public void testMigrateFileAction(String format, boolean isNamedArgument) throws
 
         if (isNamedArgument) {
             tEnv.executeSql(
-                            "CALL sys.migrate_file(connector => 'hive', source_table => 'default.hivetable01', target_table => 'default.paimontable01', delete_origin => false)")
+                            "CALL sys.migrate_table(connector => 'hive', source_table => 'default.hivetable01', target_table => 'default.paimontable01', delete_origin => false)")
                     .await();
         } else {
             tEnv.executeSql(
-                            "CALL sys.migrate_file('hive', 'default.hivetable01', 'default.paimontable01', false)")
+                            "CALL sys.migrate_table('hive', 'default.hivetable01', 'default.paimontable01', false)")
                     .await();
         }
 

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java

@@ -96,6 +96,9 @@ public InternalRow[] call(InternalRow args) {
         int parallelism =
                 args.isNullAt(6) ? Runtime.getRuntime().availableProcessors() : args.getInt(6);
 
+        if (targetTable == null && !deleteNeed) {
+            throw new IllegalArgumentException("delete_origin is false but targetTable is null.");
+        }
         Identifier sourceTableId = Identifier.fromString(sourceTable);
         Identifier tmpTableId =
                 StringUtils.isEmpty(targetTable)