[WIP][SPARK-28476][SQL] Support ALTER DATABASE SET LOCATION

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -89,6 +89,8 @@ statement
          (WITH DBPROPERTIES tablePropertyList))*                       #createDatabase
     | ALTER database db=errorCapturingIdentifier
         SET DBPROPERTIES tablePropertyList                             #setDatabaseProperties
+    | ALTER database db=errorCapturingIdentifier
+        SET locationSpec                                               #setDatabaseLocation
     | DROP database (IF EXISTS)? db=errorCapturingIdentifier
         (RESTRICT | CASCADE)?                                          #dropDatabase
     | SHOW DATABASES (LIKE? pattern=STRING)?                           #showDatabases

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala

@@ -532,6 +532,22 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder(conf) {
       visitPropertyKeyValues(ctx.tablePropertyList))
   }
 
+  /**
+   * Create an [[AlterDatabaseSetLocationCommand]] command.
+   *
+   * For example:
+   * {{{
+   *   ALTER (DATABASE|SCHEMA) database SET LOCATION path;
+   * }}}
+   */
+  override def visitSetDatabaseLocation(
+      ctx: SetDatabaseLocationContext): LogicalPlan = withOrigin(ctx) {
+    AlterDatabaseSetLocationCommand(
+      ctx.db.getText,
+      visitLocationSpec(ctx.locationSpec)
+    )
+  }
+
   /**
    * Create a [[DropDatabaseCommand]] command.
    *

sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala

@@ -132,6 +132,27 @@ case class AlterDatabasePropertiesCommand(
   }
 }
 
+/**
+ * A command for users to set new location path for a database
+ * If the database does not exist, an error message will be issued to indicate the database
+ * does not exist.
+ * The syntax of using this command in SQL is:
+ * {{{
+ *    ALTER (DATABASE|SCHEMA) database_name SET LOCATION path
+ * }}}
+ */
+case class AlterDatabaseSetLocationCommand(databaseName: String, location: String)
+  extends RunnableCommand {
+
+  override def run(sparkSession: SparkSession): Seq[Row] = {
+    val catalog = sparkSession.sessionState.catalog
+    val db: CatalogDatabase = catalog.getDatabaseMetadata(databaseName)
+    catalog.alterDatabase(db.copy(locationUri = CatalogUtils.stringToURI(location)))
+
+    Seq.empty[Row]
+  }
+}
+
 /**
  * A command for users to show the name of the database, its comment (if one has been set), and its
  * root location on the filesystem. When extended is true, it also shows the database's properties

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLParserSuite.scala

@@ -184,6 +184,15 @@ class DDLParserSuite extends AnalysisTest with SharedSQLContext {
       containsThesePhrases = Seq("key_without_value"))
   }
 
+  test("alter database set location") {
+    // ALTER (DATABASE|SCHEMA) database_name SET LOCATION
+    val sql1 = "ALTER DATABASE database_name SET LOCATION '/home/user/db'"
+    val parsed1 = parser.parsePlan(sql1)
+
+    val expected1 = AlterDatabaseSetLocationCommand("database_name", "/home/user/db")
+    comparePlans(parsed1, expected1)
+  }
+
   test("describe database") {
     // DESCRIBE DATABASE [EXTENDED] db_name;
     val sql1 = "DESCRIBE DATABASE EXTENDED db_name"

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala

@@ -730,6 +730,7 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
       try {
         val dbNameWithoutBackTicks = cleanIdentifier(dbName)
         val location = getDBPath(dbNameWithoutBackTicks)
+        val location2 = getDBPath(dbNameWithoutBackTicks + "2")
 
         sql(s"CREATE DATABASE $dbName")
 
@@ -757,6 +758,19 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
             Row("Description", "") ::
             Row("Location", CatalogUtils.URIToString(location)) ::
             Row("Properties", "((a,a), (b,b), (c,c), (d,d))") :: Nil)
+
+        withTempDir { tmpDir =>
+          val path = tmpDir.getCanonicalPath
+          val uri = tmpDir.toURI
+          logWarning(s"test change location: oldPath=${CatalogUtils.URIToString(location)}," +
+            s" newPath=${CatalogUtils.URIToString(uri)}")
+          sql(s"ALTER DATABASE $dbName SET LOCATION '$uri'")
+          val pathInCatalog = new Path(
+            catalog.getDatabaseMetadata(dbNameWithoutBackTicks).locationUri).toUri
+          assert("file" === pathInCatalog.getScheme)
+          val expectedPath = new Path(path).toUri
+          assert(expectedPath.getPath === pathInCatalog.getPath)
+        }
       } finally {
         catalog.reset()
       }

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala

@@ -205,12 +205,14 @@ private[spark] class HiveExternalCatalog(conf: SparkConf, hadoopConf: Configurat
    */
   override def alterDatabase(dbDefinition: CatalogDatabase): Unit = withClient {
     val existingDb = getDatabase(dbDefinition.name)
-    if (existingDb.properties == dbDefinition.properties) {
+    if (existingDb.properties == dbDefinition.properties &&
+      existingDb.locationUri == dbDefinition.locationUri) {
       logWarning(s"Request to alter database ${dbDefinition.name} is a no-op because " +
         s"the provided database properties are the same as the old ones. Hive does not " +
-        s"currently support altering other database fields.")
+        s"currently support altering other database fields and database location.")
     }
     client.alterDatabase(dbDefinition)
+    client
   }
 
   override def getDatabase(db: String): CatalogDatabase = withClient {

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala

@@ -361,13 +361,16 @@ private[hive] class HiveClientImpl(
   }
 
   override def alterDatabase(database: CatalogDatabase): Unit = withHiveState {
-    client.alterDatabase(
-      database.name,
-      new HiveDatabase(
-        database.name,
-        database.description,
-        CatalogUtils.URIToString(database.locationUri),
-        Option(database.properties).map(_.asJava).orNull))
+    val hiveDb = msClient.getDatabase(database.name)
+    val oldPath = hiveDb.getLocationUri
+    val newPath = CatalogUtils.URIToString(database.locationUri)
+    hiveDb.setDescription(database.description)
+    hiveDb.setLocationUri(CatalogUtils.URIToString(database.locationUri))
+    hiveDb.setParameters(Option(database.properties).map(_.asJava).orNull)
+    msClient.alterDatabase(database.name, hiveDb)
+    val newPathInHive = msClient.getDatabase(database.name).getLocationUri
+    logWarning(s"Change database location, " +
+      s"oldPath=${oldPath}, newPath=${newPath}, newPathInHive=${newPathInHive}")
   }
 
   override def getDatabase(dbName: String): CatalogDatabase = withHiveState {

sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.scala

@@ -21,6 +21,7 @@ import java.io.{ByteArrayOutputStream, File, PrintStream, PrintWriter}
 import java.net.URI
 
 import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
 import org.apache.hadoop.hive.common.StatsSetupConst
 import org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
 import org.apache.hadoop.hive.serde2.`lazy`.LazySimpleSerDe
@@ -156,6 +157,7 @@ class VersionsSuite extends SparkFunSuite with Logging {
     ///////////////////////////////////////////////////////////////////////////
 
     val tempDatabasePath = Utils.createTempDir().toURI
+    val tempDatabasePath2 = Utils.createTempDir().toURI
 
     test(s"$version: createDatabase") {
       val defaultDB = CatalogDatabase("default", "desc", new URI("loc"), Map())
@@ -197,6 +199,12 @@ class VersionsSuite extends SparkFunSuite with Logging {
       val database = client.getDatabase("temporary").copy(properties = Map("flag" -> "true"))
       client.alterDatabase(database)
       assert(client.getDatabase("temporary").properties.contains("flag"))
+
+      // test alter database location
+      client.alterDatabase(database.copy(locationUri = tempDatabasePath2))
+      val pathInCatalog = new Path(client.getDatabase("temporary").locationUri).toUri
+      val expectedPath = new Path(tempDatabasePath2).toUri
+      assert(expectedPath.getPath === pathInCatalog.getPath)
     }
 
     test(s"$version: dropDatabase") {