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

apache · May 14, 2022 · b0064e9 · b0064e9
1 parent 52e63b3
commit b0064e9
Show file tree

Hide file tree

Showing 2 changed files with 73 additions and 3 deletions.
diff --git a/...mmon/src/main/scala/org/apache/spark/sql/hudi/command/AlterHoodieTableRenameCommand.scala b/...mmon/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,15 @@ 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.catalogProperties.contains("path")) {
+        val catalogTable = sparkSession.sessionState.catalog.getTableMetadata(newName)
+        val path = catalogTable.storage.locationUri.get.getPath
+        logInfo(s"alter ${oldName} name to ${newName}, update tblp 'path' to ${path}")
+        AlterTableSetPropertiesCommand(newName, Map("path" -> path), isView).run(sparkSession)
+      }
+
     }
     Seq.empty[Row]
   }

diff --git a/...spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestAlterTable.scala b/...spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestAlterTable.scala
@@ -20,7 +20,6 @@ package org.apache.spark.sql.hudi
 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,67 @@ 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")
+        assertResult(false)(
+          newLocation.equals(oldLocation)
+        )
+
+
+        // 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")
+        assertResult(true)(
+          newLocation2.equals(oldLocation2)
+        )
+      }
+    }
+  }
 }