From 275df3dd446b9c885468d83061c9566156f226d5 Mon Sep 17 00:00:00 2001
From: dch nguyen <dgd_contributor@viettel.com.vn>
Date: Fri, 3 Dec 2021 15:17:24 +0700
Subject: [PATCH 1/4] [SPARK-37478] Unify v1 and v2 DROP NAMESPACE tests

---
 .../sql/catalyst/parser/DDLParserSuite.scala  | 27 ------
 .../sql/connector/DataSourceV2SQLSuite.scala  | 64 -------------
 .../sql/execution/command/DDLSuite.scala      | 81 +----------------
 .../command/DescribeNamespaceSuiteBase.scala  |  3 -
 .../command/DropNamespaceParserSuite.scala    | 51 +++++++++++
 .../command/DropNamespaceSuiteBase.scala      | 91 +++++++++++++++++++
 .../command/v1/DropNamespaceSuite.scala       | 71 +++++++++++++++
 .../command/v2/DropNamespaceSuite.scala       | 56 ++++++++++++
 .../command/DropNamespaceSuite.scala          | 28 ++++++
 9 files changed, 299 insertions(+), 173 deletions(-)
 create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceParserSuite.scala
 create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceSuiteBase.scala
 create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DropNamespaceSuite.scala
 create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropNamespaceSuite.scala
 create mode 100644 sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DropNamespaceSuite.scala

diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
index f4ab8076938dc..18a378f35e49a 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
@@ -1770,33 +1770,6 @@ class DDLParserSuite extends AnalysisTest {
           "location" -> "/home/user/db")))
   }
 
-  test("drop namespace") {
-    comparePlans(
-      parsePlan("DROP NAMESPACE a.b.c"),
-      DropNamespace(
-        UnresolvedNamespace(Seq("a", "b", "c")), ifExists = false, cascade = false))
-
-    comparePlans(
-      parsePlan("DROP NAMESPACE IF EXISTS a.b.c"),
-      DropNamespace(
-        UnresolvedNamespace(Seq("a", "b", "c")), ifExists = true, cascade = false))
-
-    comparePlans(
-      parsePlan("DROP NAMESPACE IF EXISTS a.b.c RESTRICT"),
-      DropNamespace(
-        UnresolvedNamespace(Seq("a", "b", "c")), ifExists = true, cascade = false))
-
-    comparePlans(
-      parsePlan("DROP NAMESPACE IF EXISTS a.b.c CASCADE"),
-      DropNamespace(
-        UnresolvedNamespace(Seq("a", "b", "c")), ifExists = true, cascade = true))
-
-    comparePlans(
-      parsePlan("DROP NAMESPACE a.b.c CASCADE"),
-      DropNamespace(
-        UnresolvedNamespace(Seq("a", "b", "c")), ifExists = false, cascade = true))
-  }
-
   test("set namespace properties") {
     comparePlans(
       parsePlan("ALTER DATABASE a.b.c SET PROPERTIES ('a'='a', 'b'='b', 'c'='c')"),
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index 58ed052fd3f64..bbe7b6f6a13c1 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -22,7 +22,6 @@ import java.time.{Duration, LocalDate, Period}
 
 import scala.collection.JavaConverters._
 
-import org.apache.spark.SparkException
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.{CannotReplaceMissingTableException, NamespaceAlreadyExistsException, NoSuchDatabaseException, NoSuchNamespaceException, TableAlreadyExistsException}
@@ -1192,69 +1191,6 @@ class DataSourceV2SQLSuite
     }
   }
 
-  test("DropNamespace: basic tests") {
-    // Session catalog is used.
-    sql("CREATE NAMESPACE ns")
-    testShowNamespaces("SHOW NAMESPACES", Seq("default", "ns"))
-    sql("DROP NAMESPACE ns")
-    testShowNamespaces("SHOW NAMESPACES", Seq("default"))
-
-    // V2 non-session catalog is used.
-    sql("CREATE NAMESPACE testcat.ns1")
-    testShowNamespaces("SHOW NAMESPACES IN testcat", Seq("ns1"))
-    sql("DROP NAMESPACE testcat.ns1")
-    testShowNamespaces("SHOW NAMESPACES IN testcat", Seq())
-  }
-
-  test("DropNamespace: drop non-empty namespace with a non-cascading mode") {
-    sql("CREATE TABLE testcat.ns1.table (id bigint) USING foo")
-    sql("CREATE TABLE testcat.ns1.ns2.table (id bigint) USING foo")
-    testShowNamespaces("SHOW NAMESPACES IN testcat", Seq("ns1"))
-    testShowNamespaces("SHOW NAMESPACES IN testcat.ns1", Seq("ns1.ns2"))
-
-    def assertDropFails(): Unit = {
-      val e = intercept[SparkException] {
-        sql("DROP NAMESPACE testcat.ns1")
-      }
-      assert(e.getMessage.contains("Cannot drop a non-empty namespace: ns1"))
-    }
-
-    // testcat.ns1.table is present, thus testcat.ns1 cannot be dropped.
-    assertDropFails()
-    sql("DROP TABLE testcat.ns1.table")
-
-    // testcat.ns1.ns2.table is present, thus testcat.ns1 cannot be dropped.
-    assertDropFails()
-    sql("DROP TABLE testcat.ns1.ns2.table")
-
-    // testcat.ns1.ns2 namespace is present, thus testcat.ns1 cannot be dropped.
-    assertDropFails()
-    sql("DROP NAMESPACE testcat.ns1.ns2")
-
-    // Now that testcat.ns1 is empty, it can be dropped.
-    sql("DROP NAMESPACE testcat.ns1")
-    testShowNamespaces("SHOW NAMESPACES IN testcat", Seq())
-  }
-
-  test("DropNamespace: drop non-empty namespace with a cascade mode") {
-    sql("CREATE TABLE testcat.ns1.table (id bigint) USING foo")
-    sql("CREATE TABLE testcat.ns1.ns2.table (id bigint) USING foo")
-    testShowNamespaces("SHOW NAMESPACES IN testcat", Seq("ns1"))
-    testShowNamespaces("SHOW NAMESPACES IN testcat.ns1", Seq("ns1.ns2"))
-
-    sql("DROP NAMESPACE testcat.ns1 CASCADE")
-    testShowNamespaces("SHOW NAMESPACES IN testcat", Seq())
-  }
-
-  test("DropNamespace: test handling of 'IF EXISTS'") {
-    sql("DROP NAMESPACE IF EXISTS testcat.unknown")
-
-    val exception = intercept[NoSuchNamespaceException] {
-      sql("DROP NAMESPACE testcat.ns1")
-    }
-    assert(exception.getMessage.contains("Namespace 'ns1' not found"))
-  }
-
   test("ALTER NAMESPACE .. SET PROPERTIES using v2 catalog") {
     withNamespace("testcat.ns1.ns2") {
       sql("CREATE NAMESPACE IF NOT EXISTS testcat.ns1.ns2 COMMENT " +
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
index 1cea49884b865..1cb7868a78197 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
@@ -787,64 +787,20 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
     }
   }
 
-  test("Drop/Alter Database - database does not exists") {
+  test("Alter Database - database does not exists") {
     val databaseNames = Seq("db1", "`database`")
 
     databaseNames.foreach { dbName =>
       val dbNameWithoutBackTicks = cleanIdentifier(dbName)
       assert(!spark.sessionState.catalog.databaseExists(dbNameWithoutBackTicks))
 
-      var message = intercept[AnalysisException] {
-        sql(s"DROP DATABASE $dbName")
-      }.getMessage
-      assert(message.contains(s"Database '$dbNameWithoutBackTicks' not found"))
-
-      message = intercept[AnalysisException] {
+      val message = intercept[AnalysisException] {
         sql(s"ALTER DATABASE $dbName SET DBPROPERTIES ('d'='d')")
       }.getMessage
       assert(message.contains(s"Database '$dbNameWithoutBackTicks' not found"))
     }
   }
 
-  test("drop non-empty database in restrict mode") {
-    val catalog = spark.sessionState.catalog
-    val dbName = "db1"
-    sql(s"CREATE DATABASE $dbName")
-
-    // create a table in database
-    val tableIdent1 = TableIdentifier("tab1", Some(dbName))
-    createTable(catalog, tableIdent1)
-
-    // drop a non-empty database in Restrict mode
-    val message = intercept[AnalysisException] {
-      sql(s"DROP DATABASE $dbName RESTRICT")
-    }.getMessage
-    assert(message.contains(s"Database $dbName is not empty. One or more tables exist"))
-
-
-    catalog.dropTable(tableIdent1, ignoreIfNotExists = false, purge = false)
-
-    assert(catalog.listDatabases().contains(dbName))
-    sql(s"DROP DATABASE $dbName RESTRICT")
-    assert(!catalog.listDatabases().contains(dbName))
-  }
-
-  test("drop non-empty database in cascade mode") {
-    val catalog = spark.sessionState.catalog
-    val dbName = "db1"
-    sql(s"CREATE DATABASE $dbName")
-
-    // create a table in database
-    val tableIdent1 = TableIdentifier("tab1", Some(dbName))
-    createTable(catalog, tableIdent1)
-
-    // drop a non-empty database in CASCADE mode
-    assert(catalog.listTables(dbName).contains(tableIdent1))
-    assert(catalog.listDatabases().contains(dbName))
-    sql(s"DROP DATABASE $dbName CASCADE")
-    assert(!catalog.listDatabases().contains(dbName))
-  }
-
   test("create table in default db") {
     val catalog = spark.sessionState.catalog
     val tableIdent1 = TableIdentifier("tab1", None)
@@ -1541,39 +1497,6 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
     }
   }
 
-  test("drop current database") {
-    withDatabase("temp") {
-      sql("CREATE DATABASE temp")
-      sql("USE temp")
-      sql("DROP DATABASE temp")
-      val e = intercept[AnalysisException] {
-        sql("CREATE TABLE t (a INT, b INT) USING parquet")
-      }.getMessage
-      assert(e.contains("Database 'temp' not found"))
-    }
-  }
-
-  test("drop default database") {
-    val caseSensitiveOptions = if (isUsingHiveMetastore) Seq("false") else Seq("true", "false")
-    caseSensitiveOptions.foreach { caseSensitive =>
-      withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive) {
-        var message = intercept[AnalysisException] {
-          sql("DROP DATABASE default")
-        }.getMessage
-        assert(message.contains("Can not drop default database"))
-
-        message = intercept[AnalysisException] {
-          sql("DROP DATABASE DeFault")
-        }.getMessage
-        if (caseSensitive == "true") {
-          assert(message.contains("Database 'DeFault' not found"))
-        } else {
-          assert(message.contains("Can not drop default database"))
-        }
-      }
-    }
-  }
-
   test("create temporary view with mismatched schema") {
     withTable("tab1") {
       spark.range(10).write.saveAsTable("tab1")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DescribeNamespaceSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DescribeNamespaceSuiteBase.scala
index d2e7d03b4d72e..e55f18007b23d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DescribeNamespaceSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DescribeNamespaceSuiteBase.scala
@@ -44,9 +44,6 @@ trait DescribeNamespaceSuiteBase extends QueryTest with DDLCommandTestUtils {
     }.getMessage
 
     assert(message.contains(s"$notFoundMsgPrefix '$ns' not found"))
-
-    // TODO: Move this to DropNamespaceSuite when the test suite is introduced.
-    sql(s"DROP NAMESPACE IF EXISTS $catalog.$ns")
   }
 
   test("Keep the legacy output schema") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceParserSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceParserSuite.scala
new file mode 100644
index 0000000000000..e6b4f41955919
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceParserSuite.scala
@@ -0,0 +1,51 @@
+/*
+ * 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.execution.command
+
+import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, UnresolvedNamespace}
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser.parsePlan
+import org.apache.spark.sql.catalyst.plans.logical.DropNamespace
+
+class DropNamespaceParserSuite extends AnalysisTest {
+  test("drop namespace") {
+    comparePlans(
+      parsePlan("DROP NAMESPACE a.b.c"),
+      DropNamespace(
+        UnresolvedNamespace(Seq("a", "b", "c")), ifExists = false, cascade = false))
+
+    comparePlans(
+      parsePlan("DROP NAMESPACE IF EXISTS a.b.c"),
+      DropNamespace(
+        UnresolvedNamespace(Seq("a", "b", "c")), ifExists = true, cascade = false))
+
+    comparePlans(
+      parsePlan("DROP NAMESPACE IF EXISTS a.b.c RESTRICT"),
+      DropNamespace(
+        UnresolvedNamespace(Seq("a", "b", "c")), ifExists = true, cascade = false))
+
+    comparePlans(
+      parsePlan("DROP NAMESPACE IF EXISTS a.b.c CASCADE"),
+      DropNamespace(
+        UnresolvedNamespace(Seq("a", "b", "c")), ifExists = true, cascade = true))
+
+    comparePlans(
+      parsePlan("DROP NAMESPACE a.b.c CASCADE"),
+      DropNamespace(
+        UnresolvedNamespace(Seq("a", "b", "c")), ifExists = false, cascade = true))
+  }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceSuiteBase.scala
new file mode 100644
index 0000000000000..46784e9d6c930
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceSuiteBase.scala
@@ -0,0 +1,91 @@
+/*
+ * 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.execution.command
+
+import org.apache.spark.sql.{QueryTest, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{StringType, StructType}
+
+/**
+ * This base suite contains unified tests for the `DROP NAMESPACE` command that check V1 and V2
+ * table catalogs. The tests that cannot run for all supported catalogs are located in more
+ * specific test suites:
+ *
+ *   - V2 table catalog tests: `org.apache.spark.sql.execution.command.v2.DropNamespaceSuite`
+ *   - V1 table catalog tests: `org.apache.spark.sql.execution.command.v1.DropNamespaceSuiteBase`
+ *     - V1 In-Memory catalog: `org.apache.spark.sql.execution.command.v1.DropNamespaceSuite`
+ *     - V1 Hive External catalog: `org.apache.spark.sql.hive.execution.command.DropNamespaceSuite`
+ */
+trait DropNamespaceSuiteBase extends QueryTest with DDLCommandTestUtils {
+  override val command = "DROP NAMESPACE"
+
+  protected def builtinTopNamespaces: Seq[String] = Seq.empty
+  protected def isCasePreserving: Boolean = true
+
+  protected def checkNamespace(sqlText: String, expected: Seq[String]) = {
+    val df = spark.sql(sqlText)
+    assert(df.schema === new StructType().add("namespace", StringType, false))
+    checkAnswer(df, expected.map(Row(_)))
+  }
+
+  test("basic tests") {
+    sql(s"CREATE NAMESPACE $catalog.ns")
+    checkNamespace(s"SHOW NAMESPACES IN $catalog", Seq("ns") ++ builtinTopNamespaces)
+
+    sql(s"DROP NAMESPACE $catalog.ns")
+    checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+  }
+
+  test("DropNamespace: test handling of 'IF EXISTS'") {
+    // It must not throw any exceptions
+    sql(s"DROP NAMESPACE IF EXISTS $catalog.unknown")
+    checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+  }
+
+  test("DropNamespace: drop non-empty namespace with a cascade mode") {
+    sql(s"CREATE NAMESPACE $catalog.ns")
+    sql(s"CREATE TABLE $catalog.ns.table (id bigint) $defaultUsing")
+    checkNamespace(s"SHOW NAMESPACES IN $catalog", Seq("ns") ++ builtinTopNamespaces)
+
+    sql(s"DROP NAMESPACE $catalog.ns CASCADE")
+    checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+  }
+
+  test("DropNamespace: drop current namespace") {
+    sql(s"CREATE NAMESPACE $catalog.ns")
+    sql(s"USE $catalog.ns")
+    sql(s"DROP NAMESPACE $catalog.ns")
+    checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+  }
+
+  test("DropNamespace: drop namespace with case sensitivity") {
+    Seq(true, false).foreach { caseSensitive =>
+      withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString) {
+        sql(s"CREATE NAMESPACE $catalog.AAA")
+        sql(s"CREATE NAMESPACE $catalog.bbb")
+        // TODO: The v1 in-memory catalog should be case preserving as well.
+        val casePreserving = isCasePreserving && (catalogVersion == "V2" || caseSensitive)
+        val expected = if (casePreserving) "AAA" else "aaa"
+
+        sql(s"DROP NAMESPACE $catalog.$expected")
+        sql(s"DROP NAMESPACE $catalog.bbb")
+        checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+      }
+    }
+  }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DropNamespaceSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DropNamespaceSuite.scala
new file mode 100644
index 0000000000000..e1407b0ec3f39
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DropNamespaceSuite.scala
@@ -0,0 +1,71 @@
+/*
+ * 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.execution.command.v1
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.execution.command
+
+/**
+ * This base suite contains unified tests for the `DROP NAMESPACE` commands that check V1 table
+ * catalogs. The tests that cannot run for all V1 catalogs are located in more specific test
+ * suites:
+ *
+ *   - V1 In-Memory catalog: `org.apache.spark.sql.execution.command.v1.DropNamespaceSuite`
+ *   - V1 Hive External catalog: `org.apache.spark.sql.hive.execution.command.DropNamespaceSuite`
+ */
+trait DropNamespaceSuiteBase extends command.DropNamespaceSuiteBase {
+  override protected def builtinTopNamespaces: Seq[String] = Seq("default")
+
+  test("DropNamespace: drop default namespace") {
+    val message = intercept[AnalysisException] {
+      sql(s"DROP NAMESPACE default")
+    }.getMessage
+    assert(message.contains("Can not drop default database"))
+  }
+
+  test("DropNamespace: drop non-empty namespace with a non-cascading mode") {
+    sql(s"CREATE NAMESPACE $catalog.ns")
+    sql(s"CREATE TABLE $catalog.ns.table (id bigint) $defaultUsing")
+    checkNamespace(s"SHOW NAMESPACES IN $catalog", Seq("ns") ++ builtinTopNamespaces)
+
+    def assertDropFails(): Unit = {
+      val e = intercept[AnalysisException] {
+        sql(s"DROP NAMESPACE $catalog.ns")
+      }
+      assert(e.getMessage.contains("Database ns is not empty. One or more tables exist"))
+    }
+
+    // $catalog.ns.table is present, thus $catalog.ns cannot be dropped.
+    assertDropFails()
+    sql(s"DROP TABLE $catalog.ns.table")
+
+    // Now that $catalog.ns is empty, it can be dropped.
+    sql(s"DROP NAMESPACE $catalog.ns")
+    checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+  }
+
+  test("DropNamespace: Namespace does not exist") {
+    // Namespace $catalog.unknown does not exist.
+    val message = intercept[AnalysisException] {
+      sql(s"DROP DATABASE $catalog.unknown")
+    }.getMessage
+    assert(message.contains(s"Database 'unknown' not found"))
+  }
+}
+
+class DropNamespaceSuite extends DropNamespaceSuiteBase with CommandSuiteBase
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropNamespaceSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropNamespaceSuite.scala
new file mode 100644
index 0000000000000..a8fb7b9d2d721
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropNamespaceSuite.scala
@@ -0,0 +1,56 @@
+/*
+ * 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.execution.command.v2
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.execution.command
+
+/**
+ * The class contains tests for the `DROP NAMESPACE` command to check V2 table catalogs.
+ */
+class DropNamespaceSuite extends command.DropNamespaceSuiteBase with CommandSuiteBase {
+  test("DropNamespace: drop non-empty namespace with a non-cascading mode") {
+    sql(s"CREATE NAMESPACE $catalog.ns")
+    sql(s"CREATE TABLE $catalog.ns.table (id bigint) $defaultUsing")
+    checkNamespace(s"SHOW NAMESPACES IN $catalog", Seq("ns") ++ builtinTopNamespaces)
+
+    def assertDropFails(): Unit = {
+      val e = intercept[SparkException] {
+        sql(s"DROP NAMESPACE $catalog.ns")
+      }
+      assert(e.getMessage.contains("Cannot drop a non-empty namespace: ns"))
+    }
+
+    // $catalog.ns.table is present, thus $catalog.ns cannot be dropped.
+    assertDropFails()
+    sql(s"DROP TABLE $catalog.ns.table")
+
+    // Now that $catalog.ns is empty, it can be dropped.
+    sql(s"DROP NAMESPACE $catalog.ns")
+    checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+  }
+
+  test("DropNamespace: Namespace does not exist") {
+    // Namespace $catalog.unknown does not exist.
+    val message = intercept[AnalysisException] {
+      sql(s"DROP DATABASE $catalog.unknown")
+    }.getMessage
+    assert(message.contains(s"Namespace 'unknown' not found"))
+  }
+}
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DropNamespaceSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DropNamespaceSuite.scala
new file mode 100644
index 0000000000000..f78087f9ec744
--- /dev/null
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DropNamespaceSuite.scala
@@ -0,0 +1,28 @@
+/*
+ * 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.hive.execution.command
+
+import org.apache.spark.sql.execution.command.v1
+
+/**
+ * The class contains tests for the `DROP NAMESPACE` command to check V1 Hive external
+ * table catalog.
+ */
+class DropNamespaceSuite extends v1.DropNamespaceSuiteBase with CommandSuiteBase {
+  override def isCasePreserving: Boolean = false
+}
\ No newline at end of file

From c60e31b0b2e0de88b268b4a547ffca1d37c25076 Mon Sep 17 00:00:00 2001
From: dch nguyen <dgd_contributor@viettel.com.vn>
Date: Mon, 6 Dec 2021 14:03:16 +0700
Subject: [PATCH 2/4] add new line to end of file

---
 .../spark/sql/hive/execution/command/DropNamespaceSuite.scala   | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DropNamespaceSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DropNamespaceSuite.scala
index f78087f9ec744..cabebb9e11510 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DropNamespaceSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DropNamespaceSuite.scala
@@ -25,4 +25,4 @@ import org.apache.spark.sql.execution.command.v1
  */
 class DropNamespaceSuite extends v1.DropNamespaceSuiteBase with CommandSuiteBase {
   override def isCasePreserving: Boolean = false
-}
\ No newline at end of file
+}

From 3970d4d548c128dba33b9d8e8ce0b563da56fc17 Mon Sep 17 00:00:00 2001
From: dch nguyen <dgd_contributor@viettel.com.vn>
Date: Tue, 7 Dec 2021 10:02:01 +0700
Subject: [PATCH 3/4] unify test

---
 .../command/DropNamespaceSuiteBase.scala      | 43 ++++++++++++++-----
 .../command/v1/DropNamespaceSuite.scala       | 36 +++-------------
 .../command/v2/DropNamespaceSuite.scala       | 31 ++-----------
 3 files changed, 44 insertions(+), 66 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceSuiteBase.scala
index 46784e9d6c930..915082fa34044 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceSuiteBase.scala
@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.execution.command
 
-import org.apache.spark.sql.{QueryTest, Row}
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{StringType, StructType}
 
@@ -36,41 +36,64 @@ trait DropNamespaceSuiteBase extends QueryTest with DDLCommandTestUtils {
 
   protected def builtinTopNamespaces: Seq[String] = Seq.empty
   protected def isCasePreserving: Boolean = true
+  protected def assertDropFails
 
-  protected def checkNamespace(sqlText: String, expected: Seq[String]) = {
-    val df = spark.sql(sqlText)
+  protected def checkNamespace(expected: Seq[String]) = {
+    val df = spark.sql(s"SHOW NAMESPACES IN $catalog")
     assert(df.schema === new StructType().add("namespace", StringType, false))
     checkAnswer(df, expected.map(Row(_)))
   }
 
   test("basic tests") {
     sql(s"CREATE NAMESPACE $catalog.ns")
-    checkNamespace(s"SHOW NAMESPACES IN $catalog", Seq("ns") ++ builtinTopNamespaces)
+    checkNamespace(Seq("ns") ++ builtinTopNamespaces)
 
     sql(s"DROP NAMESPACE $catalog.ns")
-    checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+    checkNamespace(builtinTopNamespaces)
   }
 
   test("DropNamespace: test handling of 'IF EXISTS'") {
     // It must not throw any exceptions
     sql(s"DROP NAMESPACE IF EXISTS $catalog.unknown")
-    checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+    checkNamespace(builtinTopNamespaces)
+  }
+
+  test("DropNamespace: Namespace does not exist") {
+    // Namespace $catalog.unknown does not exist.
+    val message = intercept[AnalysisException] {
+      sql(s"DROP DATABASE $catalog.unknown")
+    }.getMessage
+    assert(message.contains(s"'unknown' not found"))
+  }
+
+  test("DropNamespace: drop non-empty namespace with a non-cascading mode") {
+    sql(s"CREATE NAMESPACE $catalog.ns")
+    sql(s"CREATE TABLE $catalog.ns.table (id bigint) $defaultUsing")
+    checkNamespace(Seq("ns") ++ builtinTopNamespaces)
+
+    // $catalog.ns.table is present, thus $catalog.ns cannot be dropped.
+    assertDropFails
+    sql(s"DROP TABLE $catalog.ns.table")
+
+    // Now that $catalog.ns is empty, it can be dropped.
+    sql(s"DROP NAMESPACE $catalog.ns")
+    checkNamespace(builtinTopNamespaces)
   }
 
   test("DropNamespace: drop non-empty namespace with a cascade mode") {
     sql(s"CREATE NAMESPACE $catalog.ns")
     sql(s"CREATE TABLE $catalog.ns.table (id bigint) $defaultUsing")
-    checkNamespace(s"SHOW NAMESPACES IN $catalog", Seq("ns") ++ builtinTopNamespaces)
+    checkNamespace(Seq("ns") ++ builtinTopNamespaces)
 
     sql(s"DROP NAMESPACE $catalog.ns CASCADE")
-    checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+    checkNamespace(builtinTopNamespaces)
   }
 
   test("DropNamespace: drop current namespace") {
     sql(s"CREATE NAMESPACE $catalog.ns")
     sql(s"USE $catalog.ns")
     sql(s"DROP NAMESPACE $catalog.ns")
-    checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+    checkNamespace(builtinTopNamespaces)
   }
 
   test("DropNamespace: drop namespace with case sensitivity") {
@@ -84,7 +107,7 @@ trait DropNamespaceSuiteBase extends QueryTest with DDLCommandTestUtils {
 
         sql(s"DROP NAMESPACE $catalog.$expected")
         sql(s"DROP NAMESPACE $catalog.bbb")
-        checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+        checkNamespace(builtinTopNamespaces)
       }
     }
   }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DropNamespaceSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DropNamespaceSuite.scala
index e1407b0ec3f39..1e755472e7b75 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DropNamespaceSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DropNamespaceSuite.scala
@@ -31,40 +31,18 @@ import org.apache.spark.sql.execution.command
 trait DropNamespaceSuiteBase extends command.DropNamespaceSuiteBase {
   override protected def builtinTopNamespaces: Seq[String] = Seq("default")
 
-  test("DropNamespace: drop default namespace") {
-    val message = intercept[AnalysisException] {
-      sql(s"DROP NAMESPACE default")
-    }.getMessage
-    assert(message.contains("Can not drop default database"))
-  }
-
-  test("DropNamespace: drop non-empty namespace with a non-cascading mode") {
-    sql(s"CREATE NAMESPACE $catalog.ns")
-    sql(s"CREATE TABLE $catalog.ns.table (id bigint) $defaultUsing")
-    checkNamespace(s"SHOW NAMESPACES IN $catalog", Seq("ns") ++ builtinTopNamespaces)
-
-    def assertDropFails(): Unit = {
-      val e = intercept[AnalysisException] {
-        sql(s"DROP NAMESPACE $catalog.ns")
-      }
-      assert(e.getMessage.contains("Database ns is not empty. One or more tables exist"))
+  override protected def assertDropFails(): Unit = {
+    val e = intercept[AnalysisException] {
+      sql(s"DROP NAMESPACE $catalog.ns")
     }
-
-    // $catalog.ns.table is present, thus $catalog.ns cannot be dropped.
-    assertDropFails()
-    sql(s"DROP TABLE $catalog.ns.table")
-
-    // Now that $catalog.ns is empty, it can be dropped.
-    sql(s"DROP NAMESPACE $catalog.ns")
-    checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
+    assert(e.getMessage.contains("Database ns is not empty. One or more tables exist"))
   }
 
-  test("DropNamespace: Namespace does not exist") {
-    // Namespace $catalog.unknown does not exist.
+  test("DropNamespace: drop default namespace") {
     val message = intercept[AnalysisException] {
-      sql(s"DROP DATABASE $catalog.unknown")
+      sql(s"DROP NAMESPACE default")
     }.getMessage
-    assert(message.contains(s"Database 'unknown' not found"))
+    assert(message.contains("Can not drop default database"))
   }
 }
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropNamespaceSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropNamespaceSuite.scala
index a8fb7b9d2d721..6475c550d03ec 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropNamespaceSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropNamespaceSuite.scala
@@ -18,39 +18,16 @@
 package org.apache.spark.sql.execution.command.v2
 
 import org.apache.spark.SparkException
-import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.execution.command
 
 /**
  * The class contains tests for the `DROP NAMESPACE` command to check V2 table catalogs.
  */
 class DropNamespaceSuite extends command.DropNamespaceSuiteBase with CommandSuiteBase {
-  test("DropNamespace: drop non-empty namespace with a non-cascading mode") {
-    sql(s"CREATE NAMESPACE $catalog.ns")
-    sql(s"CREATE TABLE $catalog.ns.table (id bigint) $defaultUsing")
-    checkNamespace(s"SHOW NAMESPACES IN $catalog", Seq("ns") ++ builtinTopNamespaces)
-
-    def assertDropFails(): Unit = {
-      val e = intercept[SparkException] {
-        sql(s"DROP NAMESPACE $catalog.ns")
-      }
-      assert(e.getMessage.contains("Cannot drop a non-empty namespace: ns"))
+  override protected def assertDropFails(): Unit = {
+    val e = intercept[SparkException] {
+      sql(s"DROP NAMESPACE $catalog.ns")
     }
-
-    // $catalog.ns.table is present, thus $catalog.ns cannot be dropped.
-    assertDropFails()
-    sql(s"DROP TABLE $catalog.ns.table")
-
-    // Now that $catalog.ns is empty, it can be dropped.
-    sql(s"DROP NAMESPACE $catalog.ns")
-    checkNamespace(s"SHOW NAMESPACES IN $catalog", builtinTopNamespaces)
-  }
-
-  test("DropNamespace: Namespace does not exist") {
-    // Namespace $catalog.unknown does not exist.
-    val message = intercept[AnalysisException] {
-      sql(s"DROP DATABASE $catalog.unknown")
-    }.getMessage
-    assert(message.contains(s"Namespace 'unknown' not found"))
+    assert(e.getMessage.contains("Cannot drop a non-empty namespace: ns"))
   }
 }

From 980e2bef44448d94d990ff91ac8c8fa3db75024d Mon Sep 17 00:00:00 2001
From: dch nguyen <dgd_contributor@viettel.com.vn>
Date: Tue, 7 Dec 2021 14:33:40 +0700
Subject: [PATCH 4/4] resolve review

---
 .../command/DropNamespaceSuiteBase.scala       | 18 +++++++++---------
 .../command/v1/DropNamespaceSuite.scala        |  2 +-
 .../command/v2/DropNamespaceSuite.scala        |  1 +
 3 files changed, 11 insertions(+), 10 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceSuiteBase.scala
index 915082fa34044..1ff4e1bac049f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DropNamespaceSuiteBase.scala
@@ -36,7 +36,7 @@ trait DropNamespaceSuiteBase extends QueryTest with DDLCommandTestUtils {
 
   protected def builtinTopNamespaces: Seq[String] = Seq.empty
   protected def isCasePreserving: Boolean = true
-  protected def assertDropFails
+  protected def assertDropFails(): Unit
 
   protected def checkNamespace(expected: Seq[String]) = {
     val df = spark.sql(s"SHOW NAMESPACES IN $catalog")
@@ -52,27 +52,27 @@ trait DropNamespaceSuiteBase extends QueryTest with DDLCommandTestUtils {
     checkNamespace(builtinTopNamespaces)
   }
 
-  test("DropNamespace: test handling of 'IF EXISTS'") {
+  test("test handling of 'IF EXISTS'") {
     // It must not throw any exceptions
     sql(s"DROP NAMESPACE IF EXISTS $catalog.unknown")
     checkNamespace(builtinTopNamespaces)
   }
 
-  test("DropNamespace: Namespace does not exist") {
+  test("namespace does not exist") {
     // Namespace $catalog.unknown does not exist.
     val message = intercept[AnalysisException] {
-      sql(s"DROP DATABASE $catalog.unknown")
+      sql(s"DROP NAMESPACE $catalog.unknown")
     }.getMessage
     assert(message.contains(s"'unknown' not found"))
   }
 
-  test("DropNamespace: drop non-empty namespace with a non-cascading mode") {
+  test("drop non-empty namespace with a non-cascading mode") {
     sql(s"CREATE NAMESPACE $catalog.ns")
     sql(s"CREATE TABLE $catalog.ns.table (id bigint) $defaultUsing")
     checkNamespace(Seq("ns") ++ builtinTopNamespaces)
 
     // $catalog.ns.table is present, thus $catalog.ns cannot be dropped.
-    assertDropFails
+    assertDropFails()
     sql(s"DROP TABLE $catalog.ns.table")
 
     // Now that $catalog.ns is empty, it can be dropped.
@@ -80,7 +80,7 @@ trait DropNamespaceSuiteBase extends QueryTest with DDLCommandTestUtils {
     checkNamespace(builtinTopNamespaces)
   }
 
-  test("DropNamespace: drop non-empty namespace with a cascade mode") {
+  test("drop non-empty namespace with a cascade mode") {
     sql(s"CREATE NAMESPACE $catalog.ns")
     sql(s"CREATE TABLE $catalog.ns.table (id bigint) $defaultUsing")
     checkNamespace(Seq("ns") ++ builtinTopNamespaces)
@@ -89,14 +89,14 @@ trait DropNamespaceSuiteBase extends QueryTest with DDLCommandTestUtils {
     checkNamespace(builtinTopNamespaces)
   }
 
-  test("DropNamespace: drop current namespace") {
+  test("drop current namespace") {
     sql(s"CREATE NAMESPACE $catalog.ns")
     sql(s"USE $catalog.ns")
     sql(s"DROP NAMESPACE $catalog.ns")
     checkNamespace(builtinTopNamespaces)
   }
 
-  test("DropNamespace: drop namespace with case sensitivity") {
+  test("drop namespace with case sensitivity") {
     Seq(true, false).foreach { caseSensitive =>
       withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString) {
         sql(s"CREATE NAMESPACE $catalog.AAA")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DropNamespaceSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DropNamespaceSuite.scala
index 1e755472e7b75..2db4423f1be37 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DropNamespaceSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DropNamespaceSuite.scala
@@ -38,7 +38,7 @@ trait DropNamespaceSuiteBase extends command.DropNamespaceSuiteBase {
     assert(e.getMessage.contains("Database ns is not empty. One or more tables exist"))
   }
 
-  test("DropNamespace: drop default namespace") {
+  test("drop default namespace") {
     val message = intercept[AnalysisException] {
       sql(s"DROP NAMESPACE default")
     }.getMessage
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropNamespaceSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropNamespaceSuite.scala
index 6475c550d03ec..ddc913e0d80b9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropNamespaceSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropNamespaceSuite.scala
@@ -24,6 +24,7 @@ import org.apache.spark.sql.execution.command
  * The class contains tests for the `DROP NAMESPACE` command to check V2 table catalogs.
  */
 class DropNamespaceSuite extends command.DropNamespaceSuiteBase with CommandSuiteBase {
+  // TODO: Unify the error that throws from v1 and v2 test suite into `AnalysisException`
   override protected def assertDropFails(): Unit = {
     val e = intercept[SparkException] {
       sql(s"DROP NAMESPACE $catalog.ns")