[SPARK-43203][SQL] Move all Drop Table case to DataSource V2

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Identifier.java

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.connector.catalog;
 
 import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.internal.connector.IdentifierImpl;
 
 /**
  * Identifies an object in a catalog.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -849,6 +849,7 @@ class SessionCatalog(
         tempViews.remove(table)
       }
     }
+    invalidateCachedTable(name)
   }
 
   /**

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/IdentifierImpl.scala

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.internal.connector
+
+import java.util.{Objects, StringJoiner}
+
+import org.apache.spark.annotation.Evolving
+import org.apache.spark.sql.catalyst.util.quoteIfNeeded
+import org.apache.spark.sql.connector.catalog.Identifier
+
+/**
+ * An {@link Identifier} implementation.
+ */
+@Evolving
+class IdentifierImpl(namespace: Array[String], name: String)
+  extends Identifier {
+
+  assert(namespace != null, "Identifier namespace cannot be null")
+  assert(name != null, "Identifier name cannot be null")
+
+  override def namespace(): Array[String] = namespace
+
+  override def name(): String = name
+
+  override def toString: String = {
+    val joiner = new StringJoiner(".")
+    for (p <- namespace) {
+      joiner.add(quoteIfNeeded(p))
+    }
+    joiner.add(quoteIfNeeded(name))
+    joiner.toString
+  }
+
+  override def equals(o: Any): Boolean = {
+    o match {
+      case other: IdentifierImpl => (namespace sameElements other.namespace) &&
+        name.equals(other.name)
+      case _ => false
+    }
+  }
+
+  override def hashCode: Int = {
+    var result = Objects.hash(name)
+    result = 31 * result + namespace.toSeq.hashCode()
+    result
+  }
+
+}

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -210,7 +210,7 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
         c
       }
 
-    case DropTable(ResolvedV1Identifier(ident), ifExists, purge) =>
+    case DropTable(ResolvedV1Identifier(ident), ifExists, purge) if conf.useV1Command =>
       DropTableCommand(ident, ifExists, isView = false, purge = purge)
 
     // v1 DROP TABLE supports temp view.

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

@@ -24,6 +24,7 @@ import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.ConfigEntry
 import org.apache.spark.sql.{Dataset, SparkSession}
+import org.apache.spark.sql.catalyst.catalog.HiveTableRelation
 import org.apache.spark.sql.catalyst.expressions.{Attribute, SubqueryExpression}
 import org.apache.spark.sql.catalyst.optimizer.EliminateResolvedHint
 import org.apache.spark.sql.catalyst.plans.logical.{IgnoreCachedData, LogicalPlan, ResolvedHint, SubqueryAlias, View}
@@ -190,6 +191,11 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
         isSameName(ident.qualifier :+ ident.name) &&
           isSameName(v1Ident.catalog.toSeq ++ v1Ident.database :+ v1Ident.table)
 
+      case SubqueryAlias(ident, HiveTableRelation(catalogTable, _, _, _, _)) =>
+        val v1Ident = catalogTable.identifier
+        isSameName(ident.qualifier :+ ident.name) &&
+          isSameName(v1Ident.catalog.toSeq ++ v1Ident.database :+ v1Ident.table)
+
       case _ => false
     }
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalog.scala

@@ -71,7 +71,12 @@ class V2SessionCatalog(catalog: SessionCatalog)
   }
 
   override def loadTable(ident: Identifier): Table = {
-    V1Table(catalog.getTableMetadata(ident.asTableIdentifier))
+    try {
+      V1Table(catalog.getTableMetadata(ident.asTableIdentifier))
+    } catch {
+      case _: NoSuchDatabaseException =>
+        throw QueryCompilationErrors.noSuchTableError(ident)
+    }
   }
 
   override def loadTable(ident: Identifier, timestamp: Long): Table = {
@@ -110,6 +115,10 @@ class V2SessionCatalog(catalog: SessionCatalog)
     createTable(ident, CatalogV2Util.v2ColumnsToStructType(columns), partitions, properties)
   }
 
+  override def purgeTable(ident: Identifier): Boolean = {
+    dropTableInternal(ident, purge = true)
+  }
+
   // TODO: remove it when no tests calling this deprecated method.
   override def createTable(
       ident: Identifier,
@@ -194,12 +203,27 @@ class V2SessionCatalog(catalog: SessionCatalog)
   }
 
   override def dropTable(ident: Identifier): Boolean = {
+    dropTableInternal(ident)
+  }
+
+  private def dropTableInternal(ident: Identifier, purge: Boolean = false): Boolean = {
     try {
-      if (loadTable(ident) != null) {
+      val table = loadTable(ident)
+      if (table != null) {
+        val v1Table = table.asInstanceOf[V1Table].v1Table
+        if (v1Table.tableType == CatalogTableType.VIEW) {
+          throw QueryCompilationErrors.wrongCommandForObjectTypeError(
+            operation = "DROP TABLE",
+            requiredType = s"${CatalogTableType.EXTERNAL.name} or ${CatalogTableType.MANAGED.name}",
+            objectName = v1Table.qualifiedName,
+            foundType = v1Table.tableType.name,
+            alternative = "DROP VIEW"
+          )
+        }
         catalog.dropTable(
           ident.asTableIdentifier,
           ignoreIfNotExists = true,
-          purge = true /* skip HDFS trash */)
+          purge = purge)
         true
       } else {
         false

sql/core/src/test/resources/sql-tests/analyzer-results/ansi/conditional-functions.sql.out

@@ -101,4 +101,5 @@ Project [CASE WHEN (1 > 2) THEN (cast(1 as double) / cast(0 as double)) ELSE cas
 -- !query
 DROP TABLE conditional_t
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`conditional_t`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.conditional_t

sql/core/src/test/resources/sql-tests/analyzer-results/ansi/decimalArithmeticOperations.sql.out

@@ -166,4 +166,5 @@ Project [(10123456789012345678901234567890.123456 / 10.0) AS (101234567890123456
 -- !query
 drop table decimals_test
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`decimals_test`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.decimals_test

sql/core/src/test/resources/sql-tests/analyzer-results/change-column.sql.out

@@ -255,7 +255,8 @@ org.apache.spark.sql.AnalysisException
 -- !query
 DROP TABLE test_change
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`test_change`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.test_change
 
 
 -- !query

sql/core/src/test/resources/sql-tests/analyzer-results/charvarchar.sql.out

@@ -659,25 +659,29 @@ Project [cast(c7#x as string) AS c7#x, cast(c8#x as string) AS c8#x, cast(v#x as
 -- !query
 drop table char_tbl1
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`char_tbl1`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.char_tbl1
 
 
 -- !query
 drop table char_tbl2
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`char_tbl2`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.char_tbl2
 
 
 -- !query
 drop table char_tbl3
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`char_tbl3`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.char_tbl3
 
 
 -- !query
 drop table char_tbl4
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`char_tbl4`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.char_tbl4
 
 
 -- !query

sql/core/src/test/resources/sql-tests/analyzer-results/columnresolution-negative.sql.out

@@ -395,7 +395,8 @@ CreateViewCommand `spark_catalog`.`mydb1`.`v1`, SELECT * FROM t1, false, false,
 -- !query
 DROP TABLE t1
 -- !query analysis
-DropTableCommand `spark_catalog`.`mydb1`.`t1`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), mydb1.t1
 
 
 -- !query
@@ -441,7 +442,8 @@ CreateViewCommand `v2`, SELECT * FROM t1, false, false, LocalTempView, true
 -- !query
 DROP TABLE t1
 -- !query analysis
-DropTableCommand `spark_catalog`.`mydb2`.`t1`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), mydb2.t1
 
 
 -- !query

sql/core/src/test/resources/sql-tests/analyzer-results/decimalArithmeticOperations.sql.out

@@ -442,4 +442,5 @@ Project [(10123456789012345678901234567890.123456 / 10.0) AS (101234567890123456
 -- !query
 drop table decimals_test
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`decimals_test`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.decimals_test

sql/core/src/test/resources/sql-tests/analyzer-results/describe-part-after-analyze.sql.out

@@ -99,4 +99,5 @@ DescribeTableCommand `spark_catalog`.`default`.`t`, [ds=2017-09-01, hr=5], true,
 -- !query
 DROP TABLE t
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`t`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.t

sql/core/src/test/resources/sql-tests/analyzer-results/describe-query.sql.out

@@ -131,10 +131,12 @@ ExplainCommand DescribeQueryCommand WITH s AS (SELECT 'hello' as col1) SELECT *
 -- !query
 DROP TABLE desc_temp1
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`desc_temp1`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.desc_temp1
 
 
 -- !query
 DROP TABLE desc_temp2
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`desc_temp2`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.desc_temp2

sql/core/src/test/resources/sql-tests/analyzer-results/describe-table-after-alter-table.sql.out

@@ -26,7 +26,8 @@ DescribeTableCommand `spark_catalog`.`default`.`table_with_comment`, true, [col_
 -- !query
 DROP TABLE table_with_comment
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`table_with_comment`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.table_with_comment
 
 
 -- !query
@@ -68,4 +69,5 @@ DescribeTableCommand `spark_catalog`.`default`.`table_comment`, true, [col_name#
 -- !query
 DROP TABLE table_comment
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`table_comment`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.table_comment

sql/core/src/test/resources/sql-tests/analyzer-results/describe.sql.out

@@ -293,7 +293,8 @@ ExplainCommand 'DescribeRelation [c=Us, d=2], false, [col_name#x, data_type#x, c
 -- !query
 DROP TABLE t
 -- !query analysis
-DropTableCommand `spark_catalog`.`default`.`t`, false, false, false
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.t
 
 
 -- !query