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

Author: KnightChess

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 {
 
@@ -169,4 +168,77 @@ class TestAlterTable extends HoodieSparkSqlTestBase {
       }
     }
   }
+
+  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")
+        if (HoodieSparkUtils.isSpark3_2) {
+          assertResult(false)(
+            newLocation.equals(oldLocation)
+          )
+        } else {
+          assertResult(None) (oldLocation)
+          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")
+        if (HoodieSparkUtils.isSpark3_2) {
+          assertResult(true)(
+            newLocation2.equals(oldLocation2)
+          )
+        } else {
+          assertResult(None) (oldLocation2)
+          assertResult(None) (newLocation2)
+        }
+      }
+    }
+  }
 }