[spark] Support call migrate procedure in spark (apache#2414)

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>
 

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>

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java

@@ -25,10 +25,7 @@
 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.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 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 void initialize(String name, CaseInsensitiveStringMap options) {
         }
     }
 
+    @Override
+    public Catalog paimonCatalog() {
+        return catalog;
+    }
+
     @Override
     public String name() {
         return name;
@@ -311,18 +312,6 @@ public boolean dropTable(Identifier ident) {
         }
     }
 
-    @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;

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 @@
  * @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 Identifier[] listTables(String[] namespace) throws NoSuchNamespaceExcepti
     @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 Table loadTable(Identifier ident) throws NoSuchTableException {
     @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 Table loadTable(Identifier ident, String version) throws NoSuchTableExcep
     @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 Table loadTable(Identifier ident, long timestamp) throws NoSuchTableExcep
     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 Table createTable(
             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,28 +191,28 @@ public Table createTable(
 
     @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);
         }
     }
 
     @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 final void initialize(String name, CaseInsensitiveStringMap options) {
         }
 
         this.catalogName = name;
-        this.paimonCatalog = new SparkCatalog();
+        this.sparkCatalog = new SparkCatalog();
 
-        this.paimonCatalog.initialize(
+        this.sparkCatalog.initialize(
                 name, autoFillConfigurations(options, SparkSession.active().sessionState().conf()));
     }
 

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java

@@ -21,6 +21,8 @@
 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 @@ private static Map<String, Supplier<ProcedureBuilder>> initProcedureBuilders() {
         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();
     }
 }

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/ProcedureCatalog.java

@@ -21,14 +21,10 @@
 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}.

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SparkBaseCatalog.java

@@ -0,0 +1,44 @@
+/*
+ * 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.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.Identifier;
+import org.apache.spark.sql.connector.catalog.SupportsNamespaces;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+
+/** 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);
+    }
+}

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/WithPaimonCatalog.java

@@ -0,0 +1,26 @@
+/*
+ * 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.catalog;
+
+import org.apache.paimon.catalog.Catalog;
+
+/** With paimon catalog. */
+public interface WithPaimonCatalog {
+    Catalog paimonCatalog();
+}

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

@@ -123,6 +123,10 @@ protected SparkSession spark() {
         return spark;
     }
 
+    protected TableCatalog tableCatalog() {
+        return tableCatalog;
+    }
+
     protected abstract static class Builder<T extends BaseProcedure> implements ProcedureBuilder {
         private TableCatalog tableCatalog;