This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 599eec983 [spark] MigrateTableProcedure support keep origin table and
rename target paimon table (#3328)
599eec983 is described below
commit 599eec9839bbbf72fcab6680a298426d358f2771
Author: xuzifu666 <[email protected]>
AuthorDate: Mon May 13 20:14:25 2024 +0800
[spark] MigrateTableProcedure support keep origin table and rename target
paimon table (#3328)
---
docs/content/spark/procedures.md | 12 ++++++++++
.../java/org/apache/paimon/migrate/Migrator.java | 2 ++
.../apache/paimon/hive/migrate/HiveMigrator.java | 12 ++++++++--
.../spark/procedure/MigrateTableProcedure.java | 19 +++++++++++++---
.../procedure/MigrateTableProcedureTest.scala | 26 ++++++++++++++++++++++
5 files changed, 66 insertions(+), 5 deletions(-)
diff --git a/docs/content/spark/procedures.md b/docs/content/spark/procedures.md
index b23af90d2..be1551daf 100644
--- a/docs/content/spark/procedures.md
+++ b/docs/content/spark/procedures.md
@@ -101,6 +101,18 @@ This section introduce all available spark procedures
about paimon.
CALL sys.rollback(table => 'default.T', version => 10)
</td>
</tr>
+ <tr>
+ <td>migrate_table</td>
+ <td>
+ Migrate hive table to a paimon table. Arguments:
+ <li>source_type: the origin table's type to be migrated, such as
hive. Cannot be empty.</li>
+ <li>table: name of the origin table to be migrated. Cannot be
empty.</li>
+ <li>options: the table options of the paimon table to migrate.</li>
+ <li>target_table: name of the target paimon table to migrate. If
not set would keep the same name with origin table</li>
+ <li>delete_origin: If had set target_table, can set delete_origin
to decide whether delete the origin table metadata from hms after migrate.
Default is true</li>
+ </td>
+ <td>CALL sys.migrate_table(source_type => 'hive', table => 'default.T',
options => 'file.format=parquet')</td>
+ </tr>
<tr>
<td>remove_orphan_files</td>
<td>
diff --git a/paimon-core/src/main/java/org/apache/paimon/migrate/Migrator.java
b/paimon-core/src/main/java/org/apache/paimon/migrate/Migrator.java
index b09a9ba38..ad6de123d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/migrate/Migrator.java
+++ b/paimon-core/src/main/java/org/apache/paimon/migrate/Migrator.java
@@ -24,4 +24,6 @@ public interface Migrator {
void executeMigrate() throws Exception;
void renameTable(boolean ignoreIfNotExists) throws Exception;
+
+ public void deleteOriginTable(boolean delete) throws Exception;
}
diff --git
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
index f28f46f19..1af116cc1 100644
---
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
+++
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
@@ -78,6 +78,7 @@ public class HiveMigrator implements Migrator {
private final String targetDatabase;
private final String targetTable;
private final Map<String, String> options;
+ private Boolean delete = true;
public HiveMigrator(
HiveCatalog hiveCatalog,
@@ -116,6 +117,11 @@ public class HiveMigrator implements Migrator {
}
}
+ @Override
+ public void deleteOriginTable(boolean delete) {
+ this.delete = delete;
+ }
+
@Override
public void executeMigrate() throws Exception {
if (!client.tableExists(sourceDatabase, sourceTable)) {
@@ -202,8 +208,10 @@ public class HiveMigrator implements Migrator {
throw new RuntimeException("Migrating failed", e);
}
- // if all success, drop the origin table
- client.dropTable(sourceDatabase, sourceTable, true, true);
+ // if all success, drop the origin table according the delete field
+ if (delete) {
+ client.dropTable(sourceDatabase, sourceTable, true, true);
+ }
}
@Override
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java
index 5fac328d4..881e9d7d3 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java
@@ -24,6 +24,7 @@ import org.apache.paimon.migrate.Migrator;
import org.apache.paimon.spark.catalog.WithPaimonCatalog;
import org.apache.paimon.spark.utils.TableMigrationUtils;
import org.apache.paimon.utils.ParameterUtils;
+import org.apache.paimon.utils.StringUtils;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.connector.catalog.TableCatalog;
@@ -32,6 +33,7 @@ import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
+import static org.apache.spark.sql.types.DataTypes.BooleanType;
import static org.apache.spark.sql.types.DataTypes.StringType;
/**
@@ -49,7 +51,9 @@ public class MigrateTableProcedure extends BaseProcedure {
new ProcedureParameter[] {
ProcedureParameter.required("source_type", StringType),
ProcedureParameter.required("table", StringType),
- ProcedureParameter.optional("options", StringType)
+ ProcedureParameter.optional("options", StringType),
+ ProcedureParameter.optional("delete_origin", BooleanType),
+ ProcedureParameter.optional("target_table", StringType)
};
private static final StructType OUTPUT_TYPE =
@@ -77,9 +81,14 @@ public class MigrateTableProcedure extends BaseProcedure {
String format = args.getString(0);
String sourceTable = args.getString(1);
String properties = args.isNullAt(2) ? null : args.getString(2);
+ boolean deleteNeed = args.isNullAt(3) ? true : args.getBoolean(3);
+ String targetTable = args.isNullAt(4) ? null : args.getString(4);
Identifier sourceTableId = Identifier.fromString(sourceTable);
- Identifier tmpTableId = Identifier.fromString(sourceTable +
TMP_TBL_SUFFIX);
+ Identifier tmpTableId =
+ StringUtils.isEmpty(targetTable)
+ ? Identifier.fromString(sourceTable + TMP_TBL_SUFFIX)
+ : Identifier.fromString(targetTable);
Catalog paimonCatalog = ((WithPaimonCatalog)
tableCatalog()).paimonCatalog();
@@ -93,8 +102,12 @@ public class MigrateTableProcedure extends BaseProcedure {
tmpTableId.getDatabaseName(),
tmpTableId.getObjectName(),
ParameterUtils.parseCommaSeparatedKeyValues(properties));
+
+ migrator.deleteOriginTable(deleteNeed);
migrator.executeMigrate();
- paimonCatalog.renameTable(tmpTableId, sourceTableId, false);
+ if (StringUtils.isEmpty(targetTable)) {
+ paimonCatalog.renameTable(tmpTableId, sourceTableId, false);
+ }
} catch (Exception e) {
throw new RuntimeException("Call migrate_table error", e);
}
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala
index a9e658a1a..457b33e51 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala
@@ -45,6 +45,32 @@ class MigrateTableProcedureTest extends PaimonHiveTestBase {
}
})
+ Seq("parquet", "orc", "avro").foreach(
+ format => {
+ test(
+ s"Paimon migrate table procedure: migrate $format non-partitioned
table with set target table") {
+ withTable("hive_tbl_rn") {
+ // create hive table
+ spark.sql(s"""
+ |CREATE TABLE hive_tbl_$format (id STRING, name STRING,
pt STRING)
+ |USING $format
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO hive_tbl_$format VALUES ('1', 'a', 'p1'),
('2', 'b', 'p2')")
+
+ spark.sql(
+ s"CALL sys.migrate_table(source_type => 'hive', table =>
'$hiveDbName.hive_tbl_$format', options => 'file.format=$format', target_table
=> '$hiveDbName.target_$format', delete_origin => false)")
+
+ checkAnswer(
+ spark.sql(s"SELECT * FROM target_$format ORDER BY id"),
+ Row("1", "a", "p1") :: Row("2", "b", "p2") :: Nil)
+
+ checkAnswer(spark.sql(s"SELECT * FROM hive_tbl_$format ORDER BY
id"), Nil)
+
+ }
+ }
+ })
+
Seq("parquet", "orc", "avro").foreach(
format => {
test(s"Paimon migrate table procedure: migrate $format partitioned
table") {
