throw exception when location is not consistent

Author: TJX2014

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

@@ -545,10 +545,13 @@ private[spark] class HiveExternalCatalog(conf: SparkConf, hadoopConf: Configurat
   }
 
   private def getLocationFromStorageProps(table: CatalogTable): Option[String] = {
-    if (conf.get(HiveUtils.FOLLOW_TABLE_LOCATION)) {
-      table.storage.locationUri.map(_.toString)
+    val storageLoc = table.storage.locationUri.map(_.toString)
+    val storageProp = CaseInsensitiveMap(table.storage.properties).get("path")
+    if (storageLoc.equals(storageProp)) {
+      storageProp
     } else {
-      CaseInsensitiveMap(table.storage.properties).get("path")
+      throw new AnalysisException(s"path in location ${storageLoc} " +
+        s"not equal to table prop path ${storageProp}, please use alter table in spark")
     }
   }
 

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

@@ -165,13 +165,6 @@ private[spark] object HiveUtils extends Logging {
     .booleanConf
     .createWithDefault(true)
 
-  val FOLLOW_TABLE_LOCATION =
-    buildStaticConf("spark.sql.hive.follow.table.location")
-      .doc("If prefer hive table location to compatible with legacy `path` in catalog")
-      .version("3.1.0")
-      .booleanConf
-      .createWithDefault(false)
-
   /**
    * The version of the hive client that will be used to communicate with the metastore.  Note that
    * this does not necessarily need to be the same version of Hive that is used internally by

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveExternalCatalogSuite.scala

@@ -22,6 +22,7 @@ import java.net.URI
 import org.apache.hadoop.conf.Configuration
 
 import org.apache.spark.SparkConf
+import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.execution.QueryExecutionException
@@ -228,19 +229,16 @@ class HiveExternalCatalogSuite extends ExternalCatalogSuite {
       schema = new StructType().add("col1", "int"),
       provider = Some("parquet"))
     catalog.createTable(hiveTable, ignoreIfExists = false)
+    val beforeAlterTable = externalCatalog.getTable("db1", "parq_alter")
+    assert(beforeAlterTable.storage.locationUri.toString.contains("parq_alter"))
+
     externalCatalog.client.runSqlHive(
       "alter table db1.parq_alter rename to db1.parq_alter2")
-    val noFollowTable = externalCatalog.getTable("db1", "parq_alter2")
-    assert(!noFollowTable.storage.locationUri.toString.contains("parq_alter2"))
-
-    val confField = classOf[HiveExternalCatalog].getDeclaredField("conf")
-    confField.setAccessible(true)
-    val sparkConf = confField.get(externalCatalog).asInstanceOf[SparkConf]
-    sparkConf.set("spark.sql.hive.follow.table.location", "true")
-    val followTable = externalCatalog.getTable("db1", "parq_alter2")
-    assert(followTable.storage.locationUri.toString.contains("parq_alter2"))
-
-    sparkConf.set("spark.sql.hive.follow.table.location", "false")
-    confField.setAccessible(false)
+
+    val e = intercept[AnalysisException](
+      externalCatalog.getTable("db1", "parq_alter2")
+    )
+    assert(e.getMessage.contains("not equal to table prop path")
+      && e.getMessage.contains("parq_alter2"))
   }
 }