Spark 3.5: Fix Migrate procedure renaming issue for custom catalog

api/src/main/java/org/apache/iceberg/actions/MigrateTable.java

@@ -60,6 +60,16 @@ default MigrateTable backupTableName(String tableName) {
     throw new UnsupportedOperationException("Backup table name cannot be specified");
   }
 
+  /**
+   * Sets a destination catalog name to use the catalog for renaming a backup table
+   *
+   * @param catalogName - the destination catalog name for backup table rename
+   * @return this for method chaining
+   */
+  default MigrateTable destCatalogName(String catalogName) {
+    throw new UnsupportedOperationException("Destination catalog name cannot be specified");
+  }
+
   /** The action result that contains a summary of the execution. */
   interface Result {
     /** Returns the number of migrated data files. */

spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMigrateTableProcedure.java

@@ -142,6 +142,49 @@ public void testMigrateWithBackupTableName() throws IOException {
     Assertions.assertThat(spark.catalog().tableExists(dbName + "." + backupTableName)).isTrue();
   }
 
+  @Test
+  public void testMigrateWithDestCatalogName() throws IOException {
+    Assume.assumeTrue(catalogName.equals("spark_catalog"));
+
+    spark
+        .conf()
+        .set("spark.sql.catalog.spark_catalog", "org.apache.iceberg.spark.SparkSessionCatalog");
+
+    String location = temp.newFolder().toString();
+    sql(
+        "CREATE TABLE %s (id bigint NOT NULL, data string) USING parquet LOCATION '%s'",
+        tableName, location);
+    sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
+
+    Object result =
+        scalarSql(
+            "CALL %s.system.migrate(table => '%s', drop_backup => false, dest_catalog_name => '%s')",
+            catalogName, tableName, catalogName);
+    Assertions.assertThat(result).isEqualTo(1L);
+    Assertions.assertThat(spark.catalog().tableExists(tableName + "_BACKUP_")).isTrue();
+  }
+
+  @Test
+  public void testMigrateWithDestCatalogNameWithNonExistingCatalog() throws IOException {
+    Assume.assumeTrue(catalogName.equals("spark_catalog"));
+
+    String destCatalogName = "non_existing_catalog";
+
+    String location = temp.newFolder().toString();
+    sql(
+        "CREATE TABLE %s (id bigint NOT NULL, data string) USING parquet LOCATION '%s'",
+        tableName, location);
+
+    Assertions.assertThatThrownBy(
+            () -> {
+              sql(
+                  "CALL %s.system.migrate(table => '%s', dest_catalog_name => '%s')",
+                  catalogName, tableName, destCatalogName);
+            })
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("The destination catalog %s doesn't exist in SparkSession.", destCatalogName);
+  }
+
   @Test
   public void testMigrateWithInvalidMetricsConfig() throws IOException {
     Assume.assumeTrue(catalogName.equals("spark_catalog"));

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/MigrateTableSparkAction.java

@@ -34,6 +34,7 @@
 import org.apache.iceberg.spark.source.StagedSparkTable;
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.catalyst.TableIdentifier;
+import org.apache.spark.sql.connector.catalog.CatalogManager;
 import org.apache.spark.sql.connector.catalog.CatalogPlugin;
 import org.apache.spark.sql.connector.catalog.Identifier;
 import org.apache.spark.sql.connector.catalog.StagingTableCatalog;
@@ -54,9 +55,9 @@ public class MigrateTableSparkAction extends BaseTableCreationSparkAction<Migrat
   private static final Logger LOG = LoggerFactory.getLogger(MigrateTableSparkAction.class);
   private static final String BACKUP_SUFFIX = "_BACKUP_";
 
-  private final StagingTableCatalog destCatalog;
   private final Identifier destTableIdent;
 
+  private StagingTableCatalog destCatalog;
   private Identifier backupIdent;
   private boolean dropBackup = false;
 
@@ -108,6 +109,21 @@ public MigrateTableSparkAction backupTableName(String tableName) {
     return this;
   }
 
+  @Override
+  public MigrateTableSparkAction destCatalogName(String catalogName) {
+    CatalogManager catalogManager = spark().sessionState().catalogManager();
+
+    CatalogPlugin catalogPlugin;
+    if (catalogManager.isCatalogRegistered(catalogName)) {
+      catalogPlugin = catalogManager.catalog(catalogName);
+    } else {
+      throw new IllegalArgumentException(
+          String.format("The destination catalog %s doesn't exist in SparkSession.", catalogName));
+    }
+    this.destCatalog = checkDestinationCatalog(catalogPlugin);
+    return this;
+  }
+
   @Override
   public MigrateTable.Result execute() {
     String desc = String.format("Migrating table %s", destTableIdent().toString());

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/MigrateTableProcedure.java

@@ -40,7 +40,8 @@ class MigrateTableProcedure extends BaseProcedure {
         ProcedureParameter.required("table", DataTypes.StringType),
         ProcedureParameter.optional("properties", STRING_MAP),
         ProcedureParameter.optional("drop_backup", DataTypes.BooleanType),
-        ProcedureParameter.optional("backup_table_name", DataTypes.StringType)
+        ProcedureParameter.optional("backup_table_name", DataTypes.StringType),
+        ProcedureParameter.optional("dest_catalog_name", DataTypes.StringType),
       };
 
   private static final StructType OUTPUT_TYPE =
@@ -93,6 +94,7 @@ public InternalRow[] call(InternalRow args) {
 
     boolean dropBackup = args.isNullAt(2) ? false : args.getBoolean(2);
     String backupTableName = args.isNullAt(3) ? null : args.getString(3);
+    String destCatalogName = args.isNullAt(4) ? null : args.getString(4);
 
     MigrateTableSparkAction migrateTableSparkAction =
         SparkActions.get().migrateTable(tableName).tableProperties(properties);
@@ -105,6 +107,10 @@ public InternalRow[] call(InternalRow args) {
       migrateTableSparkAction = migrateTableSparkAction.backupTableName(backupTableName);
     }
 
+    if (destCatalogName != null) {
+      migrateTableSparkAction = migrateTableSparkAction.destCatalogName(destCatalogName);
+    }
+
     MigrateTable.Result result = migrateTableSparkAction.execute();
     return new InternalRow[] {newInternalRow(result.migratedDataFilesCount())};
   }