You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/12/01 07:03:17 UTC
(incubator-paimon) 43/46: [spark] Support call migrate procedure in spark (#2414)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.6
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit 5e4856e12907021fe0b47901b2766cb2fdcdb2c4
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Fri Dec 1 10:33:56 2023 +0800
[spark] Support call migrate procedure in spark (#2414)
---
docs/content/migration/migration-from-hive.md | 4 +-
paimon-spark/paimon-spark-common/pom.xml | 4 -
.../java/org/apache/paimon/spark/SparkCatalog.java | 25 ++---
.../apache/paimon/spark/SparkGenericCatalog.java | 37 ++++---
.../org/apache/paimon/spark/SparkProcedures.java | 4 +
.../paimon/spark/catalog/ProcedureCatalog.java | 8 +-
...ProcedureCatalog.java => SparkBaseCatalog.java} | 35 +++---
...rocedureCatalog.java => WithPaimonCatalog.java} | 23 +---
.../paimon/spark/procedure/BaseProcedure.java | 4 +
.../spark/procedure/MigrateFileProcedure.java | 121 +++++++++++++++++++++
.../spark/procedure/MigrateTableProcedure.java | 118 ++++++++++++++++++++
.../paimon/spark/utils/TableMigrationUtils.java | 53 +++++++++
.../apache/paimon/spark/PaimonHiveTestBase.scala | 62 +++++++++++
.../apache/paimon/spark/PaimonSparkTestBase.scala | 8 +-
.../spark/procedure/MigrateFileProcedureTest.scala | 94 ++++++++++++++++
.../procedure/MigrateTableProcedureTest.scala | 69 ++++++++++++
16 files changed, 587 insertions(+), 82 deletions(-)
diff --git a/docs/content/migration/migration-from-hive.md b/docs/content/migration/migration-from-hive.md
index 831d8102b..9d88d03a7 100644
--- a/docs/content/migration/migration-from-hive.md
+++ b/docs/content/migration/migration-from-hive.md
@@ -35,8 +35,8 @@ Now, we can use paimon hive catalog with Migrate Table Procedure and Migrate Fil
* 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 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 two actions now only support file format of hive "orc" and "parquet", if your table partition formatted by other format like avro, these procedures will fail.
-But we will support avro format in the future. Please make sure your table partition format is in "orc" and "parquet" now.
+These two actions now only support file format of hive "orc" and "parquet", if your table file is formatted in other format like avro, these procedures will fail.
+But we will support avro format in the future. Please make sure your table file format is in "orc" and "parquet" now.
<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>
diff --git a/paimon-spark/paimon-spark-common/pom.xml b/paimon-spark/paimon-spark-common/pom.xml
index 2cb2326cf..c7e6b9343 100644
--- a/paimon-spark/paimon-spark-common/pom.xml
+++ b/paimon-spark/paimon-spark-common/pom.xml
@@ -127,10 +127,6 @@ under the License.
<groupId>org.apache.orc</groupId>
<artifactId>orc-core</artifactId>
</exclusion>
- <exclusion>
- <groupId>org.apache.orc</groupId>
- <artifactId>orc-mapreduce</artifactId>
- </exclusion>
</exclusions>
</dependency>
<dependency>
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
index d691c2d6e..4436b7c8d 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
@@ -25,10 +25,7 @@ import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
-import org.apache.paimon.spark.analysis.NoSuchProcedureException;
-import org.apache.paimon.spark.catalog.ProcedureCatalog;
-import org.apache.paimon.spark.procedure.Procedure;
-import org.apache.paimon.spark.procedure.ProcedureBuilder;
+import org.apache.paimon.spark.catalog.SparkBaseCatalog;
import org.apache.paimon.table.Table;
import org.apache.paimon.utils.Preconditions;
@@ -39,7 +36,6 @@ import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.NamespaceChange;
-import org.apache.spark.sql.connector.catalog.SupportsNamespaces;
import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.apache.spark.sql.connector.catalog.TableChange;
import org.apache.spark.sql.connector.expressions.FieldReference;
@@ -61,7 +57,7 @@ import java.util.stream.Collectors;
import static org.apache.paimon.spark.SparkTypeUtils.toPaimonType;
/** Spark {@link TableCatalog} for paimon. */
-public class SparkCatalog implements TableCatalog, ProcedureCatalog, SupportsNamespaces {
+public class SparkCatalog extends SparkBaseCatalog {
private static final Logger LOG = LoggerFactory.getLogger(SparkCatalog.class);
@@ -84,6 +80,11 @@ public class SparkCatalog implements TableCatalog, ProcedureCatalog, SupportsNam
}
}
+ @Override
+ public Catalog paimonCatalog() {
+ return catalog;
+ }
+
@Override
public String name() {
return name;
@@ -311,18 +312,6 @@ public class SparkCatalog implements TableCatalog, ProcedureCatalog, SupportsNam
}
}
- @Override
- public Procedure loadProcedure(Identifier identifier) throws NoSuchProcedureException {
- if (isValidateNamespace(identifier.namespace())
- && Catalog.SYSTEM_DATABASE_NAME.equals(identifier.namespace()[0])) {
- ProcedureBuilder builder = SparkProcedures.newBuilder(identifier.name());
- if (builder != null) {
- return builder.withTableCatalog(this).build();
- }
- }
- throw new NoSuchProcedureException(identifier);
- }
-
private SchemaChange toSchemaChange(TableChange change) {
if (change instanceof TableChange.SetProperty) {
TableChange.SetProperty set = (TableChange.SetProperty) change;
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
index c626d96b7..2d82fa8bb 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
@@ -22,8 +22,10 @@
package org.apache.paimon.spark;
+import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.hive.HiveCatalogOptions;
import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.spark.catalog.SparkBaseCatalog;
import org.apache.paimon.utils.Preconditions;
import org.apache.hadoop.conf.Configuration;
@@ -69,16 +71,21 @@ import static org.apache.paimon.utils.Preconditions.checkNotNull;
* @param <T> CatalogPlugin class to avoid casting to TableCatalog and SupportsNamespaces.
*/
public class SparkGenericCatalog<T extends TableCatalog & SupportsNamespaces>
- implements CatalogExtension {
+ extends SparkBaseCatalog implements CatalogExtension {
private static final Logger LOG = LoggerFactory.getLogger(SparkGenericCatalog.class);
private static final String[] DEFAULT_NAMESPACE = new String[] {"default"};
private String catalogName = null;
- private SparkCatalog paimonCatalog = null;
+ private SparkCatalog sparkCatalog = null;
private T sessionCatalog = null;
+ @Override
+ public Catalog paimonCatalog() {
+ return this.sparkCatalog.paimonCatalog();
+ }
+
@Override
public String[] defaultNamespace() {
return DEFAULT_NAMESPACE;
@@ -132,7 +139,7 @@ public class SparkGenericCatalog<T extends TableCatalog & SupportsNamespaces>
@Override
public Table loadTable(Identifier ident) throws NoSuchTableException {
try {
- return paimonCatalog.loadTable(ident);
+ return sparkCatalog.loadTable(ident);
} catch (NoSuchTableException e) {
return throwsOldIfExceptionHappens(() -> getSessionCatalog().loadTable(ident), e);
}
@@ -141,7 +148,7 @@ public class SparkGenericCatalog<T extends TableCatalog & SupportsNamespaces>
@Override
public Table loadTable(Identifier ident, String version) throws NoSuchTableException {
try {
- return paimonCatalog.loadTable(ident, version);
+ return sparkCatalog.loadTable(ident, version);
} catch (NoSuchTableException e) {
return throwsOldIfExceptionHappens(
() -> getSessionCatalog().loadTable(ident, version), e);
@@ -151,7 +158,7 @@ public class SparkGenericCatalog<T extends TableCatalog & SupportsNamespaces>
@Override
public Table loadTable(Identifier ident, long timestamp) throws NoSuchTableException {
try {
- return paimonCatalog.loadTable(ident, timestamp);
+ return sparkCatalog.loadTable(ident, timestamp);
} catch (NoSuchTableException e) {
return throwsOldIfExceptionHappens(
() -> getSessionCatalog().loadTable(ident, timestamp), e);
@@ -162,7 +169,7 @@ public class SparkGenericCatalog<T extends TableCatalog & SupportsNamespaces>
public void invalidateTable(Identifier ident) {
// We do not need to check whether the table exists and whether
// it is an Paimon table to reduce remote service requests.
- paimonCatalog.invalidateTable(ident);
+ sparkCatalog.invalidateTable(ident);
getSessionCatalog().invalidateTable(ident);
}
@@ -175,7 +182,7 @@ public class SparkGenericCatalog<T extends TableCatalog & SupportsNamespaces>
throws TableAlreadyExistsException, NoSuchNamespaceException {
String provider = properties.get("provider");
if (usePaimon(provider)) {
- return paimonCatalog.createTable(ident, schema, partitions, properties);
+ return sparkCatalog.createTable(ident, schema, partitions, properties);
} else {
// delegate to the session catalog
return getSessionCatalog().createTable(ident, schema, partitions, properties);
@@ -184,8 +191,8 @@ public class SparkGenericCatalog<T extends TableCatalog & SupportsNamespaces>
@Override
public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException {
- if (paimonCatalog.tableExists(ident)) {
- return paimonCatalog.alterTable(ident, changes);
+ if (sparkCatalog.tableExists(ident)) {
+ return sparkCatalog.alterTable(ident, changes);
} else {
return getSessionCatalog().alterTable(ident, changes);
}
@@ -193,19 +200,19 @@ public class SparkGenericCatalog<T extends TableCatalog & SupportsNamespaces>
@Override
public boolean dropTable(Identifier ident) {
- return paimonCatalog.dropTable(ident) || getSessionCatalog().dropTable(ident);
+ return sparkCatalog.dropTable(ident) || getSessionCatalog().dropTable(ident);
}
@Override
public boolean purgeTable(Identifier ident) {
- return paimonCatalog.purgeTable(ident) || getSessionCatalog().purgeTable(ident);
+ return sparkCatalog.purgeTable(ident) || getSessionCatalog().purgeTable(ident);
}
@Override
public void renameTable(Identifier from, Identifier to)
throws NoSuchTableException, TableAlreadyExistsException {
- if (paimonCatalog.tableExists(from)) {
- paimonCatalog.renameTable(from, to);
+ if (sparkCatalog.tableExists(from)) {
+ sparkCatalog.renameTable(from, to);
} else {
getSessionCatalog().renameTable(from, to);
}
@@ -234,9 +241,9 @@ public class SparkGenericCatalog<T extends TableCatalog & SupportsNamespaces>
}
this.catalogName = name;
- this.paimonCatalog = new SparkCatalog();
+ this.sparkCatalog = new SparkCatalog();
- this.paimonCatalog.initialize(
+ this.sparkCatalog.initialize(
name, autoFillConfigurations(options, SparkSession.active().sessionState().conf()));
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
index fa25fd4a6..632bd416f 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
@@ -21,6 +21,8 @@ package org.apache.paimon.spark;
import org.apache.paimon.spark.procedure.CompactProcedure;
import org.apache.paimon.spark.procedure.CreateTagProcedure;
import org.apache.paimon.spark.procedure.DeleteTagProcedure;
+import org.apache.paimon.spark.procedure.MigrateFileProcedure;
+import org.apache.paimon.spark.procedure.MigrateTableProcedure;
import org.apache.paimon.spark.procedure.Procedure;
import org.apache.paimon.spark.procedure.ProcedureBuilder;
import org.apache.paimon.spark.procedure.RollbackProcedure;
@@ -50,6 +52,8 @@ public class SparkProcedures {
procedureBuilders.put("create_tag", CreateTagProcedure::builder);
procedureBuilders.put("delete_tag", DeleteTagProcedure::builder);
procedureBuilders.put("compact", CompactProcedure::builder);
+ procedureBuilders.put("migrate_table", MigrateTableProcedure::builder);
+ procedureBuilders.put("migrate_file", MigrateFileProcedure::builder);
return procedureBuilders.build();
}
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java
index b982c1896..d6431d2ac 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java
@@ -21,14 +21,10 @@ package org.apache.paimon.spark.catalog;
import org.apache.paimon.spark.analysis.NoSuchProcedureException;
import org.apache.paimon.spark.procedure.Procedure;
-import org.apache.spark.sql.connector.catalog.CatalogPlugin;
import org.apache.spark.sql.connector.catalog.Identifier;
-/**
- * A {@link CatalogPlugin catalog} interface that loads stored procedures called via CALL
- * statements.
- */
-public interface ProcedureCatalog extends CatalogPlugin {
+/** An interface that loads stored procedures called via CALL statements. */
+public interface ProcedureCatalog {
/**
* Loads a {@link Procedure stored procedure} by {@link Identifier identifier}.
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SparkBaseCatalog.java
similarity index 51%
copy from paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java
copy to paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SparkBaseCatalog.java
index b982c1896..3a18277de 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SparkBaseCatalog.java
@@ -7,7 +7,7 @@
* "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
+ * 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,
@@ -18,24 +18,27 @@
package org.apache.paimon.spark.catalog;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.spark.SparkProcedures;
import org.apache.paimon.spark.analysis.NoSuchProcedureException;
import org.apache.paimon.spark.procedure.Procedure;
+import org.apache.paimon.spark.procedure.ProcedureBuilder;
-import org.apache.spark.sql.connector.catalog.CatalogPlugin;
import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.SupportsNamespaces;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
-/**
- * A {@link CatalogPlugin catalog} interface that loads stored procedures called via CALL
- * statements.
- */
-public interface ProcedureCatalog extends CatalogPlugin {
-
- /**
- * Loads a {@link Procedure stored procedure} by {@link Identifier identifier}.
- *
- * @param identifier A stored procedure identifier.
- * @return The procedure's metadata of given identifier.
- * @throws NoSuchProcedureException Thrown, if there is no matching procedure stored.
- */
- Procedure loadProcedure(Identifier identifier) throws NoSuchProcedureException;
+/** Spark base catalog. */
+public abstract class SparkBaseCatalog
+ implements TableCatalog, SupportsNamespaces, ProcedureCatalog, WithPaimonCatalog {
+ @Override
+ public Procedure loadProcedure(Identifier identifier) throws NoSuchProcedureException {
+ if (Catalog.SYSTEM_DATABASE_NAME.equals(identifier.namespace()[0])) {
+ ProcedureBuilder builder = SparkProcedures.newBuilder(identifier.name());
+ if (builder != null) {
+ return builder.withTableCatalog(this).build();
+ }
+ }
+ throw new NoSuchProcedureException(identifier);
+ }
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/WithPaimonCatalog.java
similarity index 51%
copy from paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java
copy to paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/WithPaimonCatalog.java
index b982c1896..c09d1660d 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/WithPaimonCatalog.java
@@ -18,24 +18,9 @@
package org.apache.paimon.spark.catalog;
-import org.apache.paimon.spark.analysis.NoSuchProcedureException;
-import org.apache.paimon.spark.procedure.Procedure;
+import org.apache.paimon.catalog.Catalog;
-import org.apache.spark.sql.connector.catalog.CatalogPlugin;
-import org.apache.spark.sql.connector.catalog.Identifier;
-
-/**
- * A {@link CatalogPlugin catalog} interface that loads stored procedures called via CALL
- * statements.
- */
-public interface ProcedureCatalog extends CatalogPlugin {
-
- /**
- * Loads a {@link Procedure stored procedure} by {@link Identifier identifier}.
- *
- * @param identifier A stored procedure identifier.
- * @return The procedure's metadata of given identifier.
- * @throws NoSuchProcedureException Thrown, if there is no matching procedure stored.
- */
- Procedure loadProcedure(Identifier identifier) throws NoSuchProcedureException;
+/** With paimon catalog. */
+public interface WithPaimonCatalog {
+ Catalog paimonCatalog();
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BaseProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BaseProcedure.java
index 88903e463..377134a30 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BaseProcedure.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BaseProcedure.java
@@ -123,6 +123,10 @@ abstract class BaseProcedure implements Procedure {
return spark;
}
+ protected TableCatalog tableCatalog() {
+ return tableCatalog;
+ }
+
protected abstract static class Builder<T extends BaseProcedure> implements ProcedureBuilder {
private TableCatalog tableCatalog;
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateFileProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateFileProcedure.java
new file mode 100644
index 000000000..5978767ae
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateFileProcedure.java
@@ -0,0 +1,121 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.spark.procedure;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.migrate.Migrator;
+import org.apache.paimon.spark.catalog.WithPaimonCatalog;
+import org.apache.paimon.spark.utils.TableMigrationUtils;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import java.util.Collections;
+
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/**
+ * Migrate file procedure. Usage:
+ *
+ * <pre><code>
+ * CALL sys.migrate_file(format => 'hive', source_table => 'db.source_tbl', target_table => 'db.target_tbl')
+ * </code></pre>
+ */
+public class MigrateFileProcedure extends BaseProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("format", StringType),
+ ProcedureParameter.required("source_table", StringType),
+ ProcedureParameter.required("target_table", StringType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("result", DataTypes.BooleanType, true, Metadata.empty())
+ });
+
+ protected MigrateFileProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String format = args.getString(0);
+ String sourceTable = args.getString(1);
+ String targetTable = args.getString(2);
+
+ Identifier sourceTableId = Identifier.fromString(sourceTable);
+ Identifier targetTableId = Identifier.fromString(targetTable);
+
+ Catalog paimonCatalog = ((WithPaimonCatalog) tableCatalog()).paimonCatalog();
+
+ if (!(paimonCatalog.tableExists(targetTableId))) {
+ throw new IllegalArgumentException(
+ "Target paimon table does not exist: " + targetTable);
+ }
+
+ try {
+ Migrator migrator =
+ TableMigrationUtils.getImporter(
+ format,
+ paimonCatalog,
+ sourceTableId.getDatabaseName(),
+ sourceTableId.getObjectName(),
+ targetTableId.getDatabaseName(),
+ targetTableId.getObjectName(),
+ Collections.emptyMap());
+ migrator.executeMigrate();
+ } catch (Exception e) {
+ throw new RuntimeException("Call migrate_file error", e);
+ }
+
+ return new InternalRow[] {newInternalRow(true)};
+ }
+
+ public static ProcedureBuilder builder() {
+ return new BaseProcedure.Builder<MigrateFileProcedure>() {
+ @Override
+ public MigrateFileProcedure doBuild() {
+ return new MigrateFileProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "MigrateFileProcedure";
+ }
+}
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
new file mode 100644
index 000000000..bc15be752
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java
@@ -0,0 +1,118 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.spark.procedure;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+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.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.DataTypes;
+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.StringType;
+
+/**
+ * Migrate table procedure. Usage:
+ *
+ * <pre><code>
+ * CALL sys.migrate_table(format => 'hive', table => 'db.tbl', tblproperties => 'x1=y1,x2=y2')
+ * </code></pre>
+ */
+public class MigrateTableProcedure extends BaseProcedure {
+
+ private static final String TMP_TBL_SUFFIX = "_paimon_";
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("format", StringType),
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.optional("tblproperties", StringType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("result", DataTypes.BooleanType, true, Metadata.empty())
+ });
+
+ protected MigrateTableProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String format = args.getString(0);
+ String sourceTable = args.getString(1);
+ String properties = args.isNullAt(2) ? null : args.getString(2);
+
+ Identifier sourceTableId = Identifier.fromString(sourceTable);
+ Identifier tmpTableId = Identifier.fromString(sourceTable + TMP_TBL_SUFFIX);
+
+ Catalog paimonCatalog = ((WithPaimonCatalog) tableCatalog()).paimonCatalog();
+
+ try {
+ Migrator migrator =
+ TableMigrationUtils.getImporter(
+ format,
+ paimonCatalog,
+ sourceTableId.getDatabaseName(),
+ sourceTableId.getObjectName(),
+ tmpTableId.getDatabaseName(),
+ tmpTableId.getObjectName(),
+ ParameterUtils.parseCommaSeparatedKeyValues(properties));
+ migrator.executeMigrate();
+ paimonCatalog.renameTable(tmpTableId, sourceTableId, false);
+ } catch (Exception e) {
+ throw new RuntimeException("Call migrate_table error", e);
+ }
+
+ return new InternalRow[] {newInternalRow(true)};
+ }
+
+ public static ProcedureBuilder builder() {
+ return new BaseProcedure.Builder<MigrateTableProcedure>() {
+ @Override
+ public MigrateTableProcedure doBuild() {
+ return new MigrateTableProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "MigrateTableProcedure";
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/TableMigrationUtils.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/TableMigrationUtils.java
new file mode 100644
index 000000000..4a45b365a
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/TableMigrationUtils.java
@@ -0,0 +1,53 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.spark.utils;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.hive.HiveCatalog;
+import org.apache.paimon.hive.migrate.HiveMigrator;
+import org.apache.paimon.migrate.Migrator;
+
+import java.util.Map;
+
+/** Migration util to choose importer according to connector. */
+public class TableMigrationUtils {
+
+ public static Migrator getImporter(
+ String connector,
+ Catalog paimonCatalog,
+ String sourceDatabase,
+ String sourceTableName,
+ String targetDatabase,
+ String targetTableName,
+ Map<String, String> options) {
+ switch (connector) {
+ case "hive":
+ assert paimonCatalog instanceof HiveCatalog;
+ return new HiveMigrator(
+ (HiveCatalog) paimonCatalog,
+ sourceDatabase,
+ sourceTableName,
+ targetDatabase,
+ targetTableName,
+ options);
+ default:
+ throw new UnsupportedOperationException("Unsupported connector " + connector);
+ }
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonHiveTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonHiveTestBase.scala
new file mode 100644
index 000000000..9d5c51754
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonHiveTestBase.scala
@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+package org.apache.paimon.spark
+
+import org.apache.paimon.hive.TestHiveMetastore
+import org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
+
+import org.apache.spark.SparkConf
+
+class PaimonHiveTestBase extends PaimonSparkTestBase {
+
+ protected lazy val testHiveMetastore: TestHiveMetastore = new TestHiveMetastore
+
+ protected val hiveDbName: String = "test_hive"
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set("spark.sql.warehouse.dir", tempDBDir.getCanonicalPath)
+ .set("spark.sql.catalogImplementation", "hive")
+ .set("spark.sql.catalog.spark_catalog", classOf[SparkGenericCatalog[_]].getName)
+ .set("spark.sql.extensions", classOf[PaimonSparkSessionExtensions].getName)
+ }
+
+ override protected def beforeAll(): Unit = {
+ testHiveMetastore.start()
+ super.beforeAll()
+ spark.sql(s"USE spark_catalog")
+ spark.sql(s"CREATE DATABASE IF NOT EXISTS $hiveDbName")
+ spark.sql(s"USE $hiveDbName")
+ }
+
+ override protected def afterAll(): Unit = {
+ try {
+ spark.sql(s"USE spark_catalog")
+ spark.sql("USE default")
+ spark.sql(s"DROP DATABASE $hiveDbName CASCADE")
+ } finally {
+ super.afterAll()
+ testHiveMetastore.stop()
+ }
+ }
+
+ override protected def beforeEach(): Unit = {
+ spark.sql(s"USE spark_catalog")
+ }
+
+}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
index 5ba073120..abf4d54d3 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
@@ -24,6 +24,7 @@ import org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
import org.apache.paimon.spark.sql.WithTableOptions
import org.apache.paimon.table.AbstractFileStoreTable
+import org.apache.spark.SparkConf
import org.apache.spark.paimon.Utils
import org.apache.spark.sql.QueryTest
import org.apache.spark.sql.test.SharedSparkSession
@@ -42,7 +43,7 @@ class PaimonSparkTestBase extends QueryTest with SharedSparkSession with WithTab
protected val tableName0: String = "T"
- override protected def sparkConf = {
+ override protected def sparkConf: SparkConf = {
super.sparkConf
.set("spark.sql.catalog.paimon", classOf[SparkCatalog].getName)
.set("spark.sql.catalog.paimon.warehouse", tempDBDir.getCanonicalPath)
@@ -51,12 +52,14 @@ class PaimonSparkTestBase extends QueryTest with SharedSparkSession with WithTab
override protected def beforeAll(): Unit = {
super.beforeAll()
- spark.sql(s"CREATE DATABASE paimon.$dbName0")
+ spark.sql(s"USE paimon")
+ spark.sql(s"CREATE DATABASE IF NOT EXISTS paimon.$dbName0")
spark.sql(s"USE paimon.$dbName0")
}
override protected def afterAll(): Unit = {
try {
+ spark.sql(s"USE paimon")
spark.sql("USE default")
spark.sql(s"DROP DATABASE paimon.$dbName0 CASCADE")
} finally {
@@ -66,6 +69,7 @@ class PaimonSparkTestBase extends QueryTest with SharedSparkSession with WithTab
override protected def beforeEach(): Unit = {
super.beforeAll()
+ spark.sql(s"USE paimon")
spark.sql(s"DROP TABLE IF EXISTS $tableName0")
}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateFileProcedureTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateFileProcedureTest.scala
new file mode 100644
index 000000000..86d96274c
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateFileProcedureTest.scala
@@ -0,0 +1,94 @@
+/*
+ * 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.
+ */
+package org.apache.paimon.spark.procedure
+
+import org.apache.paimon.spark.PaimonHiveTestBase
+
+import org.apache.spark.sql.Row
+
+class MigrateFileProcedureTest extends PaimonHiveTestBase {
+ Seq("parquet", "orc").foreach(
+ format => {
+ test(s"Paimon migrate file procedure: migrate $format non-partitioned table") {
+ withTable("hive_tbl", "paimon_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')")
+
+ // create paimon table
+ spark.sql(s"""
+ |CREATE TABLE paimon_tbl (id STRING, name STRING, pt STRING)
+ |USING PAIMON
+ |TBLPROPERTIES ('file.format'='$format', 'bucket'='-1')
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO paimon_tbl VALUES ('3', 'c', 'p1'), ('4', 'd', 'p2')")
+
+ spark.sql(
+ s"CALL sys.migrate_file(format => 'hive', source_table => '$hiveDbName.hive_tbl', target_table => '$hiveDbName.paimon_tbl')")
+
+ checkAnswer(
+ spark.sql("SELECT * FROM paimon_tbl ORDER BY id"),
+ Row("1", "a", "p1") :: Row("2", "b", "p2") :: Row("3", "c", "p1") :: Row(
+ "4",
+ "d",
+ "p2") :: Nil)
+ }
+ }
+ })
+
+ Seq("parquet", "orc").foreach(
+ format => {
+ test(s"Paimon migrate file procedure: migrate $format partitioned table") {
+ withTable("hive_tbl", "paimon_tbl") {
+ // create hive table
+ spark.sql(s"""
+ |CREATE TABLE hive_tbl (id STRING, name STRING, pt STRING)
+ |USING $format
+ |PARTITIONED BY (pt)
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO hive_tbl VALUES ('1', 'a', 'p1'), ('2', 'b', 'p2')")
+
+ // create paimon table
+ spark.sql(s"""
+ |CREATE TABLE paimon_tbl (id STRING, name STRING, pt STRING)
+ |USING PAIMON
+ |TBLPROPERTIES ('file.format'='$format', 'bucket'='-1')
+ |PARTITIONED BY (pt)
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO paimon_tbl VALUES ('3', 'c', 'p1'), ('4', 'd', 'p2')")
+
+ spark.sql(
+ s"CALL sys.migrate_file(format => 'hive', source_table => '$hiveDbName.hive_tbl', target_table => '$hiveDbName.paimon_tbl')")
+
+ checkAnswer(
+ spark.sql("SELECT * FROM paimon_tbl ORDER BY id"),
+ Row("1", "a", "p1") :: Row("2", "b", "p2") :: Row("3", "c", "p1") :: Row(
+ "4",
+ "d",
+ "p2") :: Nil)
+ }
+ }
+ })
+}
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
new file mode 100644
index 000000000..179543a48
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala
@@ -0,0 +1,69 @@
+/*
+ * 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.
+ */
+package org.apache.paimon.spark.procedure
+
+import org.apache.paimon.spark.PaimonHiveTestBase
+
+import org.apache.spark.sql.Row
+
+class MigrateTableProcedureTest extends PaimonHiveTestBase {
+ Seq("parquet", "orc").foreach(
+ format => {
+ test(s"Paimon migrate table procedure: migrate $format non-partitioned table") {
+ 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(format => 'hive', table => '$hiveDbName.hive_tbl', tblproperties => 'file.format=$format')")
+
+ checkAnswer(
+ spark.sql(s"SELECT * FROM hive_tbl ORDER BY id"),
+ Row("1", "a", "p1") :: Row("2", "b", "p2") :: Nil)
+ }
+ }
+ })
+
+ Seq("parquet", "orc").foreach(
+ format => {
+ test(s"Paimon migrate table procedure: migrate $format partitioned table") {
+ withTable("hive_tbl") {
+ // create hive table
+ spark.sql(s"""
+ |CREATE TABLE hive_tbl (id STRING, name STRING, pt STRING)
+ |USING $format
+ |PARTITIONED BY (pt)
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO hive_tbl VALUES ('1', 'a', 'p1'), ('2', 'b', 'p2')")
+
+ spark.sql(
+ s"CALL sys.migrate_table(format => 'hive', table => '$hiveDbName.hive_tbl', tblproperties => 'file.format=$format')")
+
+ checkAnswer(
+ spark.sql(s"SELECT * FROM hive_tbl ORDER BY id"),
+ Row("1", "a", "p1") :: Row("2", "b", "p2") :: Nil)
+ }
+ }
+ })
+}