Initial commit

Author: uros-db

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -2138,6 +2138,13 @@ object SQLConf {
     .enumConf(classOf[Level])
     .createWithDefault(Level.TRACE)
 
+  val DROP_TABLE_VIEW_ENABLED =
+    buildConf("spark.sql.dropTableOnView.enabled")
+      .doc("When true, DROP TABLE command will work on VIEW as well.")
+      .version("4.2.0")
+      .booleanConf
+      .createWithDefault(true)
+
   val CROSS_JOINS_ENABLED = buildConf("spark.sql.crossJoin.enabled")
     .internal()
     .doc("When false, we will throw an error if a query contains a cartesian product without " +
@@ -7442,6 +7449,8 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
 
   def dataframeCacheLogLevel: Level = getConf(DATAFRAME_CACHE_LOG_LEVEL)
 
+  def dropTableOnView: Boolean = getConf(DROP_TABLE_VIEW_ENABLED)
+
   def crossJoinEnabled: Boolean = getConf(SQLConf.CROSS_JOINS_ENABLED)
 
   override def sessionLocalTimeZone: String = getConf(SQLConf.SESSION_LOCAL_TIMEZONE)

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

@@ -330,7 +330,8 @@ class V2SessionCatalog(catalog: SessionCatalog)
   private def dropTableInternal(ident: Identifier, purge: Boolean = false): Boolean = {
     try {
       loadTable(ident) match {
-        case V1Table(v1Table) if v1Table.tableType == CatalogTableType.VIEW =>
+        case V1Table(v1Table) if v1Table.tableType == CatalogTableType.VIEW &&
+            !SQLConf.get.getConf(SQLConf.DROP_TABLE_VIEW_ENABLED) =>
           throw QueryCompilationErrors.wrongCommandForObjectTypeError(
             operation = "DROP TABLE",
             requiredType = s"${CatalogTableType.EXTERNAL.name} or" +

sql/core/src/test/scala/org/apache/spark/sql/execution/SQLViewSuite.scala

@@ -50,6 +50,26 @@ abstract class SQLViewSuite extends QueryTest with SQLTestUtils {
     }
   }
 
+  test("view can be dropped using DROP TABLE") {
+    val viewName = "v"
+    withView(viewName) {
+      sql(s"DROP VIEW IF EXISTS $viewName")
+      // First, create a simple view.
+      sql(s"CREATE VIEW $viewName AS SELECT 1 AS a")
+      checkAnswer(sql(s"SELECT * FROM $viewName"), Row(1))
+      // Then, drop the view using DROP TABLE.
+      sql(s"DROP TABLE $viewName")
+      // Finally, verify that the view is dropped.
+      checkError(
+        exception = intercept[AnalysisException] {
+          sql(s"SELECT * FROM $viewName")
+        },
+        condition = "TABLE_OR_VIEW_NOT_FOUND",
+        parameters = Map("relationName" -> s"`$viewName`"),
+        ExpectedContext(s"$viewName", 14, 13 + viewName.length))
+    }
+  }
+
   test("create a permanent view on a permanent view") {
     withView("jtv1", "jtv2") {
       sql("CREATE VIEW jtv1 AS SELECT * FROM jt WHERE id > 3")

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala

@@ -1155,17 +1155,16 @@ class HiveDDLSuite
       spark.range(10).write.saveAsTable("tab1")
       withView("view1") {
         sql("CREATE VIEW view1 AS SELECT * FROM tab1")
-        assertAnalysisErrorCondition(
-          sqlText = "DROP TABLE view1",
-          condition = "WRONG_COMMAND_FOR_OBJECT_TYPE",
-          parameters = Map(
-            "alternative" -> "DROP VIEW",
-            "operation" -> "DROP TABLE",
-            "foundType" -> "VIEW",
-            "requiredType" -> "EXTERNAL or MANAGED",
-            "objectName" -> "spark_catalog.default.view1"
-          )
-        )
+        // Dropping a VIEW using DROP TABLE is allowed.
+        sql("DROP TABLE view1")
+        // Verify that the VIEW has been dropped.
+        checkError(
+          exception = intercept[AnalysisException] {
+            sql(s"SELECT * FROM view1")
+          },
+          condition = "TABLE_OR_VIEW_NOT_FOUND",
+          parameters = Map("relationName" -> s"`view1`"),
+          ExpectedContext("view1", 14, 18))
       }
     }
   }