[migrate] Support migrate table with specific separator

paimon-common/src/main/java/org/apache/paimon/utils/ParameterUtils.java

@@ -35,9 +35,13 @@ public static List<Map<String, String>> getPartitions(String... partitionStrings
     }
 
     public static Map<String, String> parseCommaSeparatedKeyValues(String keyValues) {
+        return parseSeparatedKeyValues(keyValues, ",");
+    }
+
+    public static Map<String, String> parseSeparatedKeyValues(String keyValues, String separator) {
         Map<String, String> kvs = new HashMap<>();
         if (!StringUtils.isBlank(keyValues)) {
-            for (String kvString : keyValues.split(",")) {
+            for (String kvString : keyValues.split(separator)) {
                 parseKeyValueString(kvs, kvString);
             }
         }

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

@@ -31,23 +31,31 @@ public class MigrateTableAction extends ActionBase {
     private final String hiveTableFullName;
     private final String tableProperties;
 
+    private final String separator;
+
     public MigrateTableAction(
             String connector,
             String warehouse,
             String hiveTableFullName,
             Map<String, String> catalogConfig,
-            String tableProperties) {
+            String tableProperties,
+            String separator) {
         super(warehouse, catalogConfig);
         this.connector = connector;
         this.hiveTableFullName = hiveTableFullName;
         this.tableProperties = tableProperties;
+        this.separator = separator;
     }
 
     @Override
     public void run() throws Exception {
         MigrateTableProcedure migrateTableProcedure = new MigrateTableProcedure();
         migrateTableProcedure.withCatalog(catalog);
         migrateTableProcedure.call(
-                new DefaultProcedureContext(env), connector, hiveTableFullName, tableProperties);
+                new DefaultProcedureContext(env),
+                connector,
+                hiveTableFullName,
+                tableProperties,
+                separator);
     }
 }

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

@@ -28,6 +28,7 @@ public class MigrateTableActionFactory implements ActionFactory {
 
     private static final String SOURCE_TYPE = "source_type";
     private static final String OPTIONS = "options";
+    private static final String SEPARATOR = "separator";
 
     @Override
     public String identifier() {
@@ -41,10 +42,11 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
         String sourceHiveTable = params.get(TABLE);
         Map<String, String> catalogConfig = optionalConfigMap(params, CATALOG_CONF);
         String tableConf = params.get(OPTIONS);
+        String separator = params.get(SEPARATOR) == null ? "," : params.get(SEPARATOR);
 
         MigrateTableAction migrateTableAction =
                 new MigrateTableAction(
-                        connector, warehouse, sourceHiveTable, catalogConfig, tableConf);
+                        connector, warehouse, sourceHiveTable, catalogConfig, tableConf, separator);
         return Optional.of(migrateTableAction);
     }
 

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

@@ -50,6 +50,16 @@ public String[] call(
             String sourceTablePath,
             String properties)
             throws Exception {
+        return call(procedureContext, connector, sourceTablePath, properties, ",");
+    }
+
+    public String[] call(
+            ProcedureContext procedureContext,
+            String connector,
+            String sourceTablePath,
+            String properties,
+            String separator)
+            throws Exception {
         String targetPaimonTablePath = sourceTablePath + PAIMON_SUFFIX;
 
         Identifier sourceTableId = Identifier.fromString(sourceTablePath);
@@ -62,7 +72,7 @@ public String[] call(
                         sourceTableId.getObjectName(),
                         targetTableId.getDatabaseName(),
                         targetTableId.getObjectName(),
-                        ParameterUtils.parseCommaSeparatedKeyValues(properties))
+                        ParameterUtils.parseSeparatedKeyValues(properties, separator))
                 .executeMigrate();
 
         LOG.info("Last step: rename " + targetTableId + " to " + sourceTableId);

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

@@ -70,6 +70,8 @@ public void testAvro() throws Exception {
         testUpgradeNonPartitionTable("avro");
         resetMetastore();
         testUpgradePartitionTable("avro");
+        resetMetastore();
+        testUpgradePartitionTableWithSeparator("avro", ";");
     }
 
     @Test
@@ -118,6 +120,45 @@ public void testUpgradePartitionTable(String format) throws Exception {
         Assertions.assertThatList(r1).containsExactlyInAnyOrderElementsOf(r2);
     }
 
+    public void testUpgradePartitionTableWithSeparator(String format, String separator)
+            throws Exception {
+        TableEnvironment tEnv = tableEnvironmentBuilder().batchMode().build();
+        tEnv.executeSql("CREATE CATALOG HIVE WITH ('type'='hive')");
+        tEnv.useCatalog("HIVE");
+        tEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
+        tEnv.executeSql(
+                "CREATE TABLE hivetable (id string, name string) PARTITIONED BY (id2 int, id3 int) STORED AS "
+                        + format);
+        tEnv.executeSql("INSERT INTO hivetable VALUES ('a', 'bb', 2, 3)").await();
+        tEnv.executeSql("SHOW CREATE TABLE hivetable");
+
+        tEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
+        tEnv.executeSql("CREATE CATALOG PAIMON_GE WITH ('type'='paimon-generic')");
+        tEnv.useCatalog("PAIMON_GE");
+        List<Row> r1 = ImmutableList.copyOf(tEnv.executeSql("SELECT * FROM hivetable").collect());
+
+        tEnv.executeSql(
+                "CREATE CATALOG PAIMON WITH ('type'='paimon', 'metastore' = 'hive', 'uri' = 'thrift://localhost:"
+                        + PORT
+                        + "' , 'warehouse' = '"
+                        + System.getProperty(HiveConf.ConfVars.METASTOREWAREHOUSE.varname)
+                        + "')");
+        tEnv.useCatalog("PAIMON");
+        tEnv.executeSql(
+                        "CALL sys.migrate_table('hive', 'default.hivetable', 'file.format="
+                                + format
+                                + ";orc.encrypt=pii:id,name', '"
+                                + separator
+                                + "')")
+                .await();
+        List<Row> r2 = ImmutableList.copyOf(tEnv.executeSql("SELECT * FROM hivetable").collect());
+        Assertions.assertThatList(r1).containsExactlyInAnyOrderElementsOf(r2);
+
+        List<Row> r3 =
+                ImmutableList.copyOf(tEnv.executeSql("SHOW CREATE TABLE hivetable").collect());
+        assert (r3.get(0).toString().contains("'orc.encrypt' = 'pii:id,name',"));
+    }
+
     public void testUpgradeNonPartitionTable(String format) throws Exception {
         TableEnvironment tEnv = tableEnvironmentBuilder().batchMode().build();
         tEnv.executeSql("CREATE CATALOG HIVE WITH ('type'='hive')");
@@ -175,7 +216,8 @@ public void testMigrateAction(String format) throws Exception {
                         System.getProperty(HiveConf.ConfVars.METASTOREWAREHOUSE.varname),
                         "default.hivetable",
                         catalogConf,
-                        "");
+                        "",
+                        ",");
         migrateTableAction.run();
 
         tEnv.executeSql(

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

@@ -53,7 +53,8 @@ public class MigrateTableProcedure extends BaseProcedure {
                 ProcedureParameter.required("table", StringType),
                 ProcedureParameter.optional("options", StringType),
                 ProcedureParameter.optional("delete_origin", BooleanType),
-                ProcedureParameter.optional("target_table", StringType)
+                ProcedureParameter.optional("target_table", StringType),
+                ProcedureParameter.optional("separator", StringType)
             };
 
     private static final StructType OUTPUT_TYPE =
@@ -83,6 +84,7 @@ public InternalRow[] call(InternalRow args) {
         String properties = args.isNullAt(2) ? null : args.getString(2);
         boolean deleteNeed = args.isNullAt(3) || args.getBoolean(3);
         String targetTable = args.isNullAt(4) ? null : args.getString(4);
+        String separator = args.isNullAt(5) ? "," : args.getString(5);
 
         Identifier sourceTableId = Identifier.fromString(sourceTable);
         Identifier tmpTableId =
@@ -101,7 +103,7 @@ public InternalRow[] call(InternalRow args) {
                             sourceTableId.getObjectName(),
                             tmpTableId.getDatabaseName(),
                             tmpTableId.getObjectName(),
-                            ParameterUtils.parseCommaSeparatedKeyValues(properties));
+                            ParameterUtils.parseSeparatedKeyValues(properties, separator));
 
             migrator.deleteOriginTable(deleteNeed);
             migrator.executeMigrate();

paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala

@@ -46,6 +46,36 @@ class MigrateTableProcedureTest extends PaimonHiveTestBase {
       }
     })
 
+  Seq("parquet", "orc", "avro").foreach(
+    format => {
+      test(s"Paimon migrate table procedure: migrate $format table with separator") {
+        withTable("hive_tbl") {
+          // create hive table
+          spark.sql(s"""
+                       |CREATE TABLE hive_tbl (id STRING, name STRING, pt STRING)
+                       |USING $format
+                       |""".stripMargin)
+
+          spark.sql(s"INSERT INTO hive_tbl VALUES ('1', 'a', 'p1'), ('2', 'b', 'p2')")
+
+          spark.sql(
+            s"CALL sys.migrate_table(source_type => 'hive', table => '$hiveDbName.hive_tbl', options => 'file.format=$format;orc.encrypt=pii:id,name', separator => ';')")
+
+          checkAnswer(
+            spark.sql(s"SELECT * FROM hive_tbl ORDER BY id"),
+            Row("1", "a", "p1") :: Row("2", "b", "p2") :: Nil)
+
+          assert(
+            spark
+              .sql("SHOW CREATE TABLE hive_tbl")
+              .collect()
+              .apply(0)
+              .toString()
+              .contains("'orc.encrypt' = 'pii:id,name',"))
+        }
+      }
+    })
+
   Seq("parquet", "orc", "avro").foreach(
     format => {
       test(