[HUDI-3861] update tblp 'path' when rename table

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/AlterHoodieTableRenameCommand.scala

@@ -18,11 +18,10 @@
 package org.apache.spark.sql.hudi.command
 
 import org.apache.hudi.common.table.HoodieTableMetaClient
-
 import org.apache.spark.sql.{Row, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog.HoodieCatalogTable
-import org.apache.spark.sql.execution.command.AlterTableRenameCommand
+import org.apache.spark.sql.execution.command.{AlterTableRenameCommand, AlterTableSetPropertiesCommand}
 
 /**
  * Command for alter hudi table's table name.
@@ -46,6 +45,14 @@ case class AlterHoodieTableRenameCommand(
 
       // Call AlterTableRenameCommand#run to rename table in meta.
       AlterTableRenameCommand(oldName, newName, isView).run(sparkSession)
+
+      // update table properties path in every op
+      if (hoodieCatalogTable.table.properties.contains("path")) {
+        val catalogTable = sparkSession.sessionState.catalog.getTableMetadata(newName)
+        val path = catalogTable.storage.locationUri.get.getPath
+        AlterTableSetPropertiesCommand(newName, Map("path" -> path), isView).run(sparkSession)
+      }
+
     }
     Seq.empty[Row]
   }

hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestAlterTable.scala

@@ -17,10 +17,9 @@
 
 package org.apache.spark.sql.hudi
 
+import org.apache.hudi.HoodieSparkUtils
 import org.apache.hudi.common.table.HoodieTableMetaClient
-
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.types.{LongType, StructField, StructType}
 
 class TestAlterTable extends HoodieSparkSqlTestBase {
 
@@ -161,12 +160,85 @@ class TestAlterTable extends HoodieSparkSqlTestBase {
           Seq(1, "a1", 10.0, 1000, "2021-07-25", null)
         )
 
-        spark.sql(s"insert into $partitionedTable values(2, 'a2', 10, 1000, 1, '2021-07-25')");
+        spark.sql(s"insert into $partitionedTable values(2, 'a2', 10, 1000, 1, '2021-07-25')")
         checkAnswer(s"select id, name, price, ts, dt, ext0 from $partitionedTable order by id")(
           Seq(1, "a1", 10.0, 1000, "2021-07-25", null),
           Seq(2, "a2", 10.0, 1000, "2021-07-25", 1.0)
         )
       }
     }
   }
+
+  test("Test Alter Rename Table") {
+    withTempDir { tmp =>
+      Seq("cow", "mor").foreach { tableType =>
+        val tableName = generateTableName
+        // Create table
+        spark.sql(
+          s"""
+             |create table $tableName (
+             |  id int,
+             |  name string,
+             |  price double,
+             |  ts long
+             |) using hudi
+             | tblproperties (
+             |  type = '$tableType',
+             |  primaryKey = 'id',
+             |  preCombineField = 'ts'
+             | )
+       """.stripMargin)
+
+        // alter table name.
+        val newTableName = s"${tableName}_1"
+        val oldLocation = spark.sessionState.catalog.getTableMetadata(new TableIdentifier(tableName)).properties.get("path")
+        spark.sql(s"alter table $tableName rename to $newTableName")
+        val newLocation = spark.sessionState.catalog.getTableMetadata(new TableIdentifier(newTableName)).properties.get("path")
+        // only hoodieCatalog will set path to tblp
+        if (oldLocation.nonEmpty) {
+          assertResult(false)(
+            newLocation.equals(oldLocation)
+          )
+        } else {
+          assertResult(None) (newLocation)
+        }
+
+
+        // Create table with location
+        val locTableName = s"${tableName}_loc"
+        val tablePath = s"${tmp.getCanonicalPath}/$locTableName"
+        spark.sql(
+          s"""
+             |create table $locTableName (
+             |  id int,
+             |  name string,
+             |  price double,
+             |  ts long
+             |) using hudi
+             | location '$tablePath'
+             | tblproperties (
+             |  type = '$tableType',
+             |  primaryKey = 'id',
+             |  preCombineField = 'ts'
+             | )
+       """.stripMargin)
+
+        // alter table name.
+        val newLocTableName = s"${locTableName}_1"
+        val oldLocation2 = spark.sessionState.catalog.getTableMetadata(new TableIdentifier(locTableName))
+          .properties.get("path")
+        spark.sql(s"alter table $locTableName rename to $newLocTableName")
+        val newLocation2 = spark.sessionState.catalog.getTableMetadata(new TableIdentifier(newLocTableName))
+          .properties.get("path")
+        // only hoodieCatalog will set path to tblp
+        if (oldLocation2.nonEmpty) {
+          assertResult(true)(
+            newLocation2.equals(oldLocation2)
+          )
+        } else {
+          assertResult(None) (newLocation2)
+        }
+      }
+    }
+  }
 }