[SPARK-28265][SQL] Add renameTable to TableCatalog API

sql/catalyst/src/main/java/org/apache/spark/sql/catalog/v2/TableCatalog.java

@@ -134,4 +134,24 @@ Table alterTable(
    * @return true if a table was deleted, false if no table exists for the identifier
    */
   boolean dropTable(Identifier ident);
+
+  /**
+   * Renames a table in the catalog.
+   * <p>
+   * If the catalog supports views and contains a view for the old identifier and not a table, this
+   * throws {@link NoSuchTableException}. Additionally, if the new identifier is a table or a view,
+   * this throws {@link TableAlreadyExistsException}.
+   * <p>
+   * If the catalog does not support table renames between namespaces, it throws
+   * {@link UnsupportedOperationException}.
+   *
+   * @param oldIdent the table identifier of the existing table to rename
+   * @param newIdent the new table identifier of the table
+   * @throws NoSuchTableException If the table to rename doesn't exist or is a view
+   * @throws TableAlreadyExistsException If the new table name already exists or is a view
+   * @throws UnsupportedOperationException If the namespaces of old and new identiers do not
+   *                                       match (optional)
+   */
+  void renameTable(Identifier oldIdent, Identifier newIdent)
+      throws NoSuchTableException, TableAlreadyExistsException;
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalog/v2/TableCatalogSuite.scala

@@ -45,6 +45,7 @@ class TableCatalogSuite extends SparkFunSuite {
 
   private val testNs = Array("`", ".")
   private val testIdent = Identifier.of(testNs, "test_table")
+  private val testIdentNew = Identifier.of(testNs, "test_table_new")
 
   test("Catalogs can load the catalog") {
     val catalog = newCatalog()
@@ -656,6 +657,52 @@ class TableCatalogSuite extends SparkFunSuite {
     assert(!catalog.tableExists(testIdent))
   }
 
+  test("renameTable") {
+    val catalog = newCatalog()
+
+    assert(!catalog.tableExists(testIdent))
+    assert(!catalog.tableExists(testIdentNew))
+
+    catalog.createTable(testIdent, schema, Array.empty, emptyProps)
+
+    assert(catalog.tableExists(testIdent))
+    catalog.renameTable(testIdent, testIdentNew)
+
+    assert(!catalog.tableExists(testIdent))
+    assert(catalog.tableExists(testIdentNew))
+  }
+
+  test("renameTable: fail if table does not exist") {
+    val catalog = newCatalog()
+
+    val exc = intercept[NoSuchTableException] {
+      catalog.renameTable(testIdent, testIdentNew)
+    }
+
+    assert(exc.message.contains(testIdent.quoted))
+    assert(exc.message.contains("not found"))
+  }
+
+  test("renameTable: fail if new table name already exists") {
+    val catalog = newCatalog()
+
+    assert(!catalog.tableExists(testIdent))
+    assert(!catalog.tableExists(testIdentNew))
+
+    catalog.createTable(testIdent, schema, Array.empty, emptyProps)
+    catalog.createTable(testIdentNew, schema, Array.empty, emptyProps)
+
+    assert(catalog.tableExists(testIdent))
+    assert(catalog.tableExists(testIdentNew))
+
+    val exc = intercept[TableAlreadyExistsException] {
+      catalog.renameTable(testIdent, testIdentNew)
+    }
+
+    assert(exc.message.contains(testIdentNew.quoted))
+    assert(exc.message.contains("already exists"))
+  }
+
   test("listNamespaces: list namespaces from metadata") {
     val catalog = newCatalog()
     catalog.createNamespace(Array("ns1"), Map("property" -> "value").asJava)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalog/v2/TestTableCatalog.scala

@@ -93,6 +93,19 @@ class TestTableCatalog extends TableCatalog with SupportsNamespaces {
 
   override def dropTable(ident: Identifier): Boolean = Option(tables.remove(ident)).isDefined
 
+  override def renameTable(oldIdent: Identifier, newIdent: Identifier): Unit = {
+    if (tables.containsKey(newIdent)) {
+      throw new TableAlreadyExistsException(newIdent)
+    }
+
+    Option(tables.remove(oldIdent)) match {
+      case Some(table) =>
+        tables.put(newIdent, InMemoryTable(table.name, table.schema, table.properties))
+      case _ =>
+        throw new NoSuchTableException(oldIdent)
+    }
+  }
+
   private def allNamespaces: Seq[Seq[String]] = {
     (tables.keySet.asScala.map(_.namespace.toSeq) ++ namespaces.keySet.asScala).toSeq.distinct
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalog.scala

@@ -153,6 +153,14 @@ class V2SessionCatalog(sessionState: SessionState) extends TableCatalog {
     }
   }
 
+  override def renameTable(oldIdent: Identifier, newIdent: Identifier): Unit = {
+    if (Option(loadTable(newIdent)).isDefined) throw new TableAlreadyExistsException(newIdent)
+
+    // Load table to make sure the table exists
+    loadTable(oldIdent)
+    catalog.renameTable(oldIdent.asTableIdentifier, newIdent.asTableIdentifier)
+  }
+
   implicit class TableIdentifierHelper(ident: Identifier) {
     def asTableIdentifier: TableIdentifier = {
       ident.namespace match {

sql/core/src/test/scala/org/apache/spark/sql/sources/v2/TestInMemoryTableCatalog.scala

@@ -104,6 +104,20 @@ class TestInMemoryTableCatalog extends TableCatalog {
     Option(tables.remove(ident)).isDefined
   }
 
+  override def renameTable(oldIdent: Identifier, newIdent: Identifier): Unit = {
+    if (tables.containsKey(newIdent)) {
+      throw new TableAlreadyExistsException(newIdent)
+    }
+
+    Option(tables.remove(oldIdent)) match {
+      case Some(table) =>
+        tables.put(newIdent,
+          new InMemoryTable(table.name, table.schema, table.partitioning, table.properties))
+      case _ =>
+        throw new NoSuchTableException(oldIdent)
+    }
+  }
+
   def clearTables(): Unit = {
     tables.clear()
   }