From f74305ec96741589f2d3616143e1a5b962510c4a Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sat, 12 Dec 2020 17:50:31 +0300
Subject: [PATCH 01/15] Move some V2 tests to AlterTableDropPartitionSuite

---
 .../AlterTablePartitionV2SQLSuite.scala       | 30 -------
 .../AlterTableDropPartitionSuiteBase.scala    | 57 +++++++++++++
 .../v2/AlterTableDropPartitionSuite.scala     | 83 +++++++++++++++++++
 3 files changed, 140 insertions(+), 30 deletions(-)
 create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
 create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
index 570976965ec7c..16050ab26f224 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
@@ -52,36 +52,6 @@ class AlterTablePartitionV2SQLSuite extends DatasourceV2SQLBase {
     }
   }
 
-  test("ALTER TABLE DROP PARTITION") {
-    val t = "testpart.ns1.ns2.tbl"
-    withTable(t) {
-      spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
-      spark.sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
-      spark.sql(s"ALTER TABLE $t DROP PARTITION (id=1)")
-
-      val partTable =
-        catalog("testpart").asTableCatalog.loadTable(Identifier.of(Array("ns1", "ns2"), "tbl"))
-      assert(!partTable.asPartitionable.partitionExists(InternalRow.fromSeq(Seq(1))))
-    }
-  }
-
-  test("ALTER TABLE DROP PARTITIONS") {
-    val t = "testpart.ns1.ns2.tbl"
-    withTable(t) {
-      spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
-      spark.sql(s"ALTER TABLE $t ADD IF NOT EXISTS PARTITION (id=1) LOCATION 'loc'" +
-        " PARTITION (id=2) LOCATION 'loc1'")
-      spark.sql(s"ALTER TABLE $t DROP PARTITION (id=1), PARTITION (id=2)")
-
-      val partTable =
-        catalog("testpart").asTableCatalog.loadTable(Identifier.of(Array("ns1", "ns2"), "tbl"))
-      assert(!partTable.asPartitionable.partitionExists(InternalRow.fromSeq(Seq(1))))
-      assert(!partTable.asPartitionable.partitionExists(InternalRow.fromSeq(Seq(2))))
-      assert(
-        partTable.asPartitionable.listPartitionIdentifiers(Array.empty, InternalRow.empty).isEmpty)
-    }
-  }
-
   test("ALTER TABLE DROP PARTITIONS: partition not exists") {
     val t = "testpart.ns1.ns2.tbl"
     withTable(t) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
new file mode 100644
index 0000000000000..b4c5d9a106117
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
@@ -0,0 +1,57 @@
+/*
+ * 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.scalactic.source.Position
+import org.scalatest.Tag
+
+import org.apache.spark.sql.{QueryTest, Row}
+import org.apache.spark.sql.execution.datasources.PartitioningUtils
+import org.apache.spark.sql.test.SQLTestUtils
+
+trait AlterTableDropPartitionSuiteBase  extends QueryTest with SQLTestUtils {
+  protected def version: String
+  protected def catalog: String
+  protected def defaultUsing: String
+
+  override def test(testName: String, testTags: Tag*)(testFun: => Any)
+    (implicit pos: Position): Unit = {
+    super.test(s"ALTER TABLE .. ADD PARTITION $version: " + testName, testTags: _*)(testFun)
+  }
+
+  protected def checkPartitions(t: String, expected: Map[String, String]*): Unit = {
+    val partitions = sql(s"SHOW PARTITIONS $t")
+      .collect()
+      .toSet
+      .map((row: Row) => row.getString(0))
+      .map(PartitioningUtils.parsePathFragment)
+    assert(partitions === expected.toSet)
+  }
+
+  protected def withNsTable(ns: String, tableName: String, cat: String = catalog)
+    (f: String => Unit): Unit = {
+    val nsCat = s"$cat.$ns"
+    withNamespace(nsCat) {
+      sql(s"CREATE NAMESPACE $nsCat")
+      val t = s"$nsCat.$tableName"
+      withTable(t) {
+        f(t)
+      }
+    }
+  }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
new file mode 100644
index 0000000000000..862351d879388
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
@@ -0,0 +1,83 @@
+/*
+ * 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.SparkConf
+import org.apache.spark.sql.connector.InMemoryPartitionTableCatalog
+import org.apache.spark.sql.execution.command
+import org.apache.spark.sql.test.SharedSparkSession
+
+class AlterTableDropPartitionSuite
+  extends command.AlterTableDropPartitionSuiteBase
+  with SharedSparkSession {
+
+  override def version: String = "V2"
+  override def catalog: String = "test_catalog"
+  override def defaultUsing: String = "USING _"
+
+  override def sparkConf: SparkConf = super.sparkConf
+    .set(s"spark.sql.catalog.$catalog", classOf[InMemoryPartitionTableCatalog].getName)
+
+  protected def checkDropPartition(
+      t: String,
+      ifExists: String,
+      specs: Map[String, Any]*): Unit = {
+    checkPartitions(t, specs.map(_.mapValues(_.toString)): _*)
+    val specStr = specs.map(
+      _.map {
+        case (k, v: String) => s"$k = '$v'"
+        case (k, v) => s"$k = $v"
+      }.mkString("PARTITION (", ", ", ")"))
+      .mkString(", ")
+    sql(s"ALTER TABLE $t DROP $ifExists $specStr")
+    checkPartitions(t)
+  }
+
+  test("single partition") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
+      Seq("", "IF EXISTS").foreach { ifExists =>
+        sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+        checkDropPartition(t, ifExists, Map("id" -> 1))
+      }
+    }
+  }
+
+  test("multiple partitions") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
+      Seq("", "IF EXISTS").foreach { ifExists =>
+        sql(s"""
+          |ALTER TABLE $t ADD
+          |PARTITION (id=1) LOCATION 'loc'
+          |PARTITION (id=2) LOCATION 'loc1'""".stripMargin)
+        checkDropPartition(t, ifExists, Map("id" -> 1), Map("id" -> 2))
+      }
+    }
+  }
+
+  test("multi-part partition") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, a int, b string) $defaultUsing PARTITIONED BY (a, b)")
+      Seq("", "IF EXISTS").foreach { ifExists =>
+        sql(s"ALTER TABLE $t ADD PARTITION (a = 2, b = 'abc')")
+        checkDropPartition(t, ifExists, Map("a" -> 2, "b" -> "abc"))
+      }
+    }
+  }
+}

From 90d590735c4ef95f4ec682a652ba632d015680b2 Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sat, 12 Dec 2020 18:01:50 +0300
Subject: [PATCH 02/15] Move the test "partition not exists"

---
 .../AlterTablePartitionV2SQLSuite.scala       | 21 -------------------
 .../v2/AlterTableDropPartitionSuite.scala     | 17 +++++++++++++++
 2 files changed, 17 insertions(+), 21 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
index 16050ab26f224..4eef513a6dc5b 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
@@ -52,27 +52,6 @@ class AlterTablePartitionV2SQLSuite extends DatasourceV2SQLBase {
     }
   }
 
-  test("ALTER TABLE DROP PARTITIONS: partition not exists") {
-    val t = "testpart.ns1.ns2.tbl"
-    withTable(t) {
-      spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
-      spark.sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
-
-      assertThrows[NoSuchPartitionsException](
-        spark.sql(s"ALTER TABLE $t DROP PARTITION (id=1), PARTITION (id=2)"))
-
-      val partTable =
-        catalog("testpart").asTableCatalog.loadTable(Identifier.of(Array("ns1", "ns2"), "tbl"))
-      assert(partTable.asPartitionable.partitionExists(InternalRow.fromSeq(Seq(1))))
-
-      spark.sql(s"ALTER TABLE $t DROP IF EXISTS PARTITION (id=1), PARTITION (id=2)")
-      assert(!partTable.asPartitionable.partitionExists(InternalRow.fromSeq(Seq(1))))
-      assert(!partTable.asPartitionable.partitionExists(InternalRow.fromSeq(Seq(2))))
-      assert(
-        partTable.asPartitionable.listPartitionIdentifiers(Array.empty, InternalRow.empty).isEmpty)
-    }
-  }
-
   test("case sensitivity in resolving partition specs") {
     val t = "testpart.ns1.ns2.tbl"
     withTable(t) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
index 862351d879388..e252cba93d742 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
@@ -18,6 +18,7 @@
 package org.apache.spark.sql.execution.command.v2
 
 import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException
 import org.apache.spark.sql.connector.InMemoryPartitionTableCatalog
 import org.apache.spark.sql.execution.command
 import org.apache.spark.sql.test.SharedSparkSession
@@ -80,4 +81,20 @@ class AlterTableDropPartitionSuite
       }
     }
   }
+
+  test("partition not exists") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
+      sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+
+      val errMsg = intercept[NoSuchPartitionsException] {
+        sql(s"ALTER TABLE $t DROP PARTITION (id=1), PARTITION (id=2)")
+      }.getMessage
+      assert(errMsg.contains("The following partitions not found in table"))
+
+      checkPartitions(t, Map("id" -> "1"))
+      sql(s"ALTER TABLE $t DROP IF EXISTS PARTITION (id=1), PARTITION (id=2)")
+      checkPartitions(t)
+    }
+  }
 }

From 40ed7d8bedf1c24712fedd3182602ca7a1c56ff1 Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sat, 12 Dec 2020 18:06:55 +0300
Subject: [PATCH 03/15] Move the test "case sensitivity in resolving partition
 specs"

---
 .../AlterTablePartitionV2SQLSuite.scala       | 25 -------------------
 .../v2/AlterTableDropPartitionSuite.scala     | 21 ++++++++++++++++
 2 files changed, 21 insertions(+), 25 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
index 4eef513a6dc5b..cee309a23ceb8 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
@@ -52,31 +52,6 @@ class AlterTablePartitionV2SQLSuite extends DatasourceV2SQLBase {
     }
   }
 
-  test("case sensitivity in resolving partition specs") {
-    val t = "testpart.ns1.ns2.tbl"
-    withTable(t) {
-      spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
-      withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
-        val errMsg = intercept[AnalysisException] {
-          spark.sql(s"ALTER TABLE $t DROP PARTITION (ID=1)")
-        }.getMessage
-        assert(errMsg.contains(s"ID is not a valid partition column in table $t"))
-      }
-
-      val partTable = catalog("testpart").asTableCatalog
-        .loadTable(Identifier.of(Array("ns1", "ns2"), "tbl"))
-        .asPartitionable
-      assert(!partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
-
-      withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
-        spark.sql(s"ALTER TABLE $t ADD PARTITION (ID=1) LOCATION 'loc1'")
-        assert(partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
-        spark.sql(s"ALTER TABLE $t DROP PARTITION (Id=1)")
-        assert(!partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
-      }
-    }
-  }
-
   test("SPARK-33650: drop partition into a table which doesn't support partition management") {
     val t = "testcat.ns1.ns2.tbl"
     withTable(t) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
index e252cba93d742..d38a47e6a606e 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
@@ -18,9 +18,11 @@
 package org.apache.spark.sql.execution.command.v2
 
 import org.apache.spark.SparkConf
+import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException
 import org.apache.spark.sql.connector.InMemoryPartitionTableCatalog
 import org.apache.spark.sql.execution.command
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
 
 class AlterTableDropPartitionSuite
@@ -97,4 +99,23 @@ class AlterTableDropPartitionSuite
       checkPartitions(t)
     }
   }
+
+  test("case sensitivity in resolving partition specs") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
+      withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+        val errMsg = intercept[AnalysisException] {
+          sql(s"ALTER TABLE $t DROP PARTITION (ID=1)")
+        }.getMessage
+        assert(errMsg.contains(s"ID is not a valid partition column in table $t"))
+      }
+
+      withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+        Seq("", "IF EXISTS").foreach { ifExists =>
+          sql(s"ALTER TABLE $t ADD PARTITION (ID=1) LOCATION 'loc1'")
+          checkDropPartition(t, ifExists, Map("id" -> 1))
+        }
+      }
+    }
+  }
 }

From c2abbebcf39c99c8bb13780a674a9ffdd1d7cf2c Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sat, 12 Dec 2020 18:11:02 +0300
Subject: [PATCH 04/15] Port the test "table which doesn't support partition
 management"

---
 .../connector/AlterTablePartitionV2SQLSuite.scala | 11 -----------
 .../command/v2/AlterTableDropPartitionSuite.scala | 15 +++++++++++++--
 2 files changed, 13 insertions(+), 13 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
index cee309a23ceb8..9f19295da6109 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
@@ -52,17 +52,6 @@ class AlterTablePartitionV2SQLSuite extends DatasourceV2SQLBase {
     }
   }
 
-  test("SPARK-33650: drop partition into a table which doesn't support partition management") {
-    val t = "testcat.ns1.ns2.tbl"
-    withTable(t) {
-      spark.sql(s"CREATE TABLE $t (id bigint, data string) USING _")
-      val errMsg = intercept[AnalysisException] {
-        spark.sql(s"ALTER TABLE $t DROP PARTITION (id=1)")
-      }.getMessage
-      assert(errMsg.contains(s"Table $t can not alter partitions"))
-    }
-  }
-
   test("SPARK-33676: not fully specified partition spec") {
     val t = "testpart.ns1.ns2.tbl"
     withTable(t) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
index d38a47e6a606e..54e09ddb8ef61 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.command.v2
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException
-import org.apache.spark.sql.connector.InMemoryPartitionTableCatalog
+import org.apache.spark.sql.connector.{InMemoryPartitionTableCatalog, InMemoryTableCatalog}
 import org.apache.spark.sql.execution.command
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
@@ -35,6 +35,7 @@ class AlterTableDropPartitionSuite
 
   override def sparkConf: SparkConf = super.sparkConf
     .set(s"spark.sql.catalog.$catalog", classOf[InMemoryPartitionTableCatalog].getName)
+    .set(s"spark.sql.catalog.non_part_$catalog", classOf[InMemoryTableCatalog].getName)
 
   protected def checkDropPartition(
       t: String,
@@ -102,7 +103,7 @@ class AlterTableDropPartitionSuite
 
   test("case sensitivity in resolving partition specs") {
     withNsTable("ns", "tbl") { t =>
-      sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
+      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
       withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
         val errMsg = intercept[AnalysisException] {
           sql(s"ALTER TABLE $t DROP PARTITION (ID=1)")
@@ -118,4 +119,14 @@ class AlterTableDropPartitionSuite
       }
     }
   }
+
+  test("SPARK-33650: drop partition into a table which doesn't support partition management") {
+    withNsTable("ns", "tbl", s"non_part_$catalog") { t =>
+      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing")
+      val errMsg = intercept[AnalysisException] {
+        sql(s"ALTER TABLE $t DROP PARTITION (id=1)")
+      }.getMessage
+      assert(errMsg.contains("can not alter partitions"))
+    }
+  }
 }

From 9b439b5df6d0a93a09659e2137477f876b1e2f8b Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sat, 12 Dec 2020 18:13:48 +0300
Subject: [PATCH 05/15] Port the test "not fully specified partition spec"

---
 .../AlterTablePartitionV2SQLSuite.scala       | 25 -------------------
 .../v2/AlterTableDropPartitionSuite.scala     | 15 +++++++++++
 2 files changed, 15 insertions(+), 25 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
index 9f19295da6109..7beb34c06323c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala
@@ -18,18 +18,8 @@
 package org.apache.spark.sql.connector
 
 import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException
-import org.apache.spark.sql.connector.catalog.{CatalogV2Implicits, Identifier}
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits
-import org.apache.spark.sql.internal.SQLConf
 
 class AlterTablePartitionV2SQLSuite extends DatasourceV2SQLBase {
-
-  import CatalogV2Implicits._
-  import DataSourceV2Implicits._
-
-
   test("ALTER TABLE RECOVER PARTITIONS") {
     val t = "testcat.ns1.ns2.tbl"
     withTable(t) {
@@ -51,19 +41,4 @@ class AlterTablePartitionV2SQLSuite extends DatasourceV2SQLBase {
       assert(e.message.contains("ALTER TABLE RENAME PARTITION is only supported with v1 tables"))
     }
   }
-
-  test("SPARK-33676: not fully specified partition spec") {
-    val t = "testpart.ns1.ns2.tbl"
-    withTable(t) {
-      sql(s"""
-        |CREATE TABLE $t (id bigint, part0 int, part1 string)
-        |USING foo
-        |PARTITIONED BY (part0, part1)""".stripMargin)
-      val errMsg = intercept[AnalysisException] {
-        sql(s"ALTER TABLE $t DROP PARTITION (part0 = 1)")
-      }.getMessage
-      assert(errMsg.contains("Partition spec is invalid. " +
-        "The spec (part0) must match the partition spec (part0, part1)"))
-    }
-  }
 }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
index 54e09ddb8ef61..69988965f76d1 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
@@ -129,4 +129,19 @@ class AlterTableDropPartitionSuite
       assert(errMsg.contains("can not alter partitions"))
     }
   }
+
+
+  test("SPARK-33676: not fully specified partition spec") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"""
+        |CREATE TABLE $t (id bigint, part0 int, part1 string)
+        |$defaultUsing
+        |PARTITIONED BY (part0, part1)""".stripMargin)
+      val errMsg = intercept[AnalysisException] {
+        sql(s"ALTER TABLE $t DROP PARTITION (part0 = 1)")
+      }.getMessage
+      assert(errMsg.contains("Partition spec is invalid. " +
+        "The spec (part0) must match the partition spec (part0, part1)"))
+    }
+  }
 }

From 5dea29388e7439cfca2a5032adbccd7133cdf441 Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sat, 12 Dec 2020 18:28:17 +0300
Subject: [PATCH 06/15] Move test to the base trait

---
 .../AlterTableDropPartitionSuiteBase.scala    | 118 ++++++++++++++++--
 .../v1/AlterTableDropPartitionSuite.scala     |  32 +++++
 .../v2/AlterTableDropPartitionSuite.scala     | 102 +--------------
 3 files changed, 143 insertions(+), 109 deletions(-)
 create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableDropPartitionSuite.scala

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
index b4c5d9a106117..1a9f88152407e 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
@@ -20,8 +20,10 @@ package org.apache.spark.sql.execution.command
 import org.scalactic.source.Position
 import org.scalatest.Tag
 
-import org.apache.spark.sql.{QueryTest, Row}
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException
 import org.apache.spark.sql.execution.datasources.PartitioningUtils
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SQLTestUtils
 
 trait AlterTableDropPartitionSuiteBase  extends QueryTest with SQLTestUtils {
@@ -29,11 +31,25 @@ trait AlterTableDropPartitionSuiteBase  extends QueryTest with SQLTestUtils {
   protected def catalog: String
   protected def defaultUsing: String
 
+  protected def notFullPartitionSpecErr: String
+
   override def test(testName: String, testTags: Tag*)(testFun: => Any)
     (implicit pos: Position): Unit = {
     super.test(s"ALTER TABLE .. ADD PARTITION $version: " + testName, testTags: _*)(testFun)
   }
 
+  protected def withNsTable(ns: String, tableName: String, cat: String = catalog)
+    (f: String => Unit): Unit = {
+    val nsCat = s"$cat.$ns"
+    withNamespace(nsCat) {
+      sql(s"CREATE NAMESPACE $nsCat")
+      val t = s"$nsCat.$tableName"
+      withTable(t) {
+        f(t)
+      }
+    }
+  }
+
   protected def checkPartitions(t: String, expected: Map[String, String]*): Unit = {
     val partitions = sql(s"SHOW PARTITIONS $t")
       .collect()
@@ -43,15 +59,99 @@ trait AlterTableDropPartitionSuiteBase  extends QueryTest with SQLTestUtils {
     assert(partitions === expected.toSet)
   }
 
-  protected def withNsTable(ns: String, tableName: String, cat: String = catalog)
-    (f: String => Unit): Unit = {
-    val nsCat = s"$cat.$ns"
-    withNamespace(nsCat) {
-      sql(s"CREATE NAMESPACE $nsCat")
-      val t = s"$nsCat.$tableName"
-      withTable(t) {
-        f(t)
+  protected def checkDropPartition(
+      t: String,
+      ifExists: String,
+      specs: Map[String, Any]*): Unit = {
+    checkPartitions(t, specs.map(_.mapValues(_.toString)): _*)
+    val specStr = specs.map(
+      _.map {
+        case (k, v: String) => s"$k = '$v'"
+        case (k, v) => s"$k = $v"
+      }.mkString("PARTITION (", ", ", ")"))
+      .mkString(", ")
+    sql(s"ALTER TABLE $t DROP $ifExists $specStr")
+    checkPartitions(t)
+  }
+
+  test("single partition") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
+      Seq("", "IF EXISTS").foreach { ifExists =>
+        sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+        checkDropPartition(t, ifExists, Map("id" -> 1))
       }
     }
   }
+
+  test("multiple partitions") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
+      Seq("", "IF EXISTS").foreach { ifExists =>
+        sql(s"""
+          |ALTER TABLE $t ADD
+          |PARTITION (id=1) LOCATION 'loc'
+          |PARTITION (id=2) LOCATION 'loc1'""".stripMargin)
+        checkDropPartition(t, ifExists, Map("id" -> 1), Map("id" -> 2))
+      }
+    }
+  }
+
+  test("multi-part partition") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, a int, b string) $defaultUsing PARTITIONED BY (a, b)")
+      Seq("", "IF EXISTS").foreach { ifExists =>
+        sql(s"ALTER TABLE $t ADD PARTITION (a = 2, b = 'abc')")
+        checkDropPartition(t, ifExists, Map("a" -> 2, "b" -> "abc"))
+      }
+    }
+  }
+
+  test("partition not exists") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
+      sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+
+      val errMsg = intercept[NoSuchPartitionsException] {
+        sql(s"ALTER TABLE $t DROP PARTITION (id=1), PARTITION (id=2)")
+      }.getMessage
+      assert(errMsg.contains("The following partitions not found in table"))
+
+      checkPartitions(t, Map("id" -> "1"))
+      sql(s"ALTER TABLE $t DROP IF EXISTS PARTITION (id=1), PARTITION (id=2)")
+      checkPartitions(t)
+    }
+  }
+
+  ignore("case sensitivity in resolving partition specs") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
+      withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+        val errMsg = intercept[AnalysisException] {
+          sql(s"ALTER TABLE $t DROP PARTITION (ID=1)")
+        }.getMessage
+        assert(errMsg.contains(s"ID is not a valid partition column in table $t"))
+      }
+
+      withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+        Seq("", "IF EXISTS").foreach { ifExists =>
+          sql(s"ALTER TABLE $t ADD PARTITION (ID=1) LOCATION 'loc1'")
+          checkDropPartition(t, ifExists, Map("id" -> 1))
+        }
+      }
+    }
+  }
+
+  test("SPARK-33676: not fully specified partition spec") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"""
+        |CREATE TABLE $t (id bigint, part0 int, part1 string)
+        |$defaultUsing
+        |PARTITIONED BY (part0, part1)""".stripMargin)
+      val errMsg = intercept[AnalysisException] {
+        sql(s"ALTER TABLE $t DROP PARTITION (part0 = 1)")
+      }.getMessage
+      assert(errMsg.contains(notFullPartitionSpecErr))
+    }
+  }
 }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableDropPartitionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableDropPartitionSuite.scala
new file mode 100644
index 0000000000000..a0c62a93a17e0
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableDropPartitionSuite.scala
@@ -0,0 +1,32 @@
+/*
+ * 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.connector.catalog.CatalogManager
+import org.apache.spark.sql.execution.command
+import org.apache.spark.sql.test.SharedSparkSession
+
+trait AlterTableDropPartitionSuiteBase extends command.AlterTableDropPartitionSuiteBase {
+  override def version: String = "V1"
+  override def catalog: String = CatalogManager.SESSION_CATALOG_NAME
+  override def defaultUsing: String = "USING parquet"
+
+  override protected val notFullPartitionSpecErr = "The following partitions not found in table"
+}
+
+class AlterTableDropPartitionSuite extends AlterTableDropPartitionSuiteBase with SharedSparkSession
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
index 69988965f76d1..9dc1cad5a002d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
@@ -19,10 +19,8 @@ package org.apache.spark.sql.execution.command.v2
 
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException
 import org.apache.spark.sql.connector.{InMemoryPartitionTableCatalog, InMemoryTableCatalog}
 import org.apache.spark.sql.execution.command
-import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
 
 class AlterTableDropPartitionSuite
@@ -33,93 +31,12 @@ class AlterTableDropPartitionSuite
   override def catalog: String = "test_catalog"
   override def defaultUsing: String = "USING _"
 
+  override protected val notFullPartitionSpecErr = "Partition spec is invalid"
+
   override def sparkConf: SparkConf = super.sparkConf
     .set(s"spark.sql.catalog.$catalog", classOf[InMemoryPartitionTableCatalog].getName)
     .set(s"spark.sql.catalog.non_part_$catalog", classOf[InMemoryTableCatalog].getName)
 
-  protected def checkDropPartition(
-      t: String,
-      ifExists: String,
-      specs: Map[String, Any]*): Unit = {
-    checkPartitions(t, specs.map(_.mapValues(_.toString)): _*)
-    val specStr = specs.map(
-      _.map {
-        case (k, v: String) => s"$k = '$v'"
-        case (k, v) => s"$k = $v"
-      }.mkString("PARTITION (", ", ", ")"))
-      .mkString(", ")
-    sql(s"ALTER TABLE $t DROP $ifExists $specStr")
-    checkPartitions(t)
-  }
-
-  test("single partition") {
-    withNsTable("ns", "tbl") { t =>
-      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
-      Seq("", "IF EXISTS").foreach { ifExists =>
-        sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
-        checkDropPartition(t, ifExists, Map("id" -> 1))
-      }
-    }
-  }
-
-  test("multiple partitions") {
-    withNsTable("ns", "tbl") { t =>
-      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
-      Seq("", "IF EXISTS").foreach { ifExists =>
-        sql(s"""
-          |ALTER TABLE $t ADD
-          |PARTITION (id=1) LOCATION 'loc'
-          |PARTITION (id=2) LOCATION 'loc1'""".stripMargin)
-        checkDropPartition(t, ifExists, Map("id" -> 1), Map("id" -> 2))
-      }
-    }
-  }
-
-  test("multi-part partition") {
-    withNsTable("ns", "tbl") { t =>
-      sql(s"CREATE TABLE $t (id bigint, a int, b string) $defaultUsing PARTITIONED BY (a, b)")
-      Seq("", "IF EXISTS").foreach { ifExists =>
-        sql(s"ALTER TABLE $t ADD PARTITION (a = 2, b = 'abc')")
-        checkDropPartition(t, ifExists, Map("a" -> 2, "b" -> "abc"))
-      }
-    }
-  }
-
-  test("partition not exists") {
-    withNsTable("ns", "tbl") { t =>
-      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
-      sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
-
-      val errMsg = intercept[NoSuchPartitionsException] {
-        sql(s"ALTER TABLE $t DROP PARTITION (id=1), PARTITION (id=2)")
-      }.getMessage
-      assert(errMsg.contains("The following partitions not found in table"))
-
-      checkPartitions(t, Map("id" -> "1"))
-      sql(s"ALTER TABLE $t DROP IF EXISTS PARTITION (id=1), PARTITION (id=2)")
-      checkPartitions(t)
-    }
-  }
-
-  test("case sensitivity in resolving partition specs") {
-    withNsTable("ns", "tbl") { t =>
-      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
-      withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
-        val errMsg = intercept[AnalysisException] {
-          sql(s"ALTER TABLE $t DROP PARTITION (ID=1)")
-        }.getMessage
-        assert(errMsg.contains(s"ID is not a valid partition column in table $t"))
-      }
-
-      withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
-        Seq("", "IF EXISTS").foreach { ifExists =>
-          sql(s"ALTER TABLE $t ADD PARTITION (ID=1) LOCATION 'loc1'")
-          checkDropPartition(t, ifExists, Map("id" -> 1))
-        }
-      }
-    }
-  }
-
   test("SPARK-33650: drop partition into a table which doesn't support partition management") {
     withNsTable("ns", "tbl", s"non_part_$catalog") { t =>
       sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing")
@@ -129,19 +46,4 @@ class AlterTableDropPartitionSuite
       assert(errMsg.contains("can not alter partitions"))
     }
   }
-
-
-  test("SPARK-33676: not fully specified partition spec") {
-    withNsTable("ns", "tbl") { t =>
-      sql(s"""
-        |CREATE TABLE $t (id bigint, part0 int, part1 string)
-        |$defaultUsing
-        |PARTITIONED BY (part0, part1)""".stripMargin)
-      val errMsg = intercept[AnalysisException] {
-        sql(s"ALTER TABLE $t DROP PARTITION (part0 = 1)")
-      }.getMessage
-      assert(errMsg.contains("Partition spec is invalid. " +
-        "The spec (part0) must match the partition spec (part0, part1)"))
-    }
-  }
 }

From 7295763edd52b89ff30e41b079e123b3b7ae3cd5 Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sat, 12 Dec 2020 18:31:53 +0300
Subject: [PATCH 07/15] Fix case sens error msg

---
 .../execution/command/AlterTableDropPartitionSuiteBase.scala  | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
index 1a9f88152407e..65ade64d88679 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
@@ -123,14 +123,14 @@ trait AlterTableDropPartitionSuiteBase  extends QueryTest with SQLTestUtils {
     }
   }
 
-  ignore("case sensitivity in resolving partition specs") {
+  test("case sensitivity in resolving partition specs") {
     withNsTable("ns", "tbl") { t =>
       sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
       withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
         val errMsg = intercept[AnalysisException] {
           sql(s"ALTER TABLE $t DROP PARTITION (ID=1)")
         }.getMessage
-        assert(errMsg.contains(s"ID is not a valid partition column in table $t"))
+        assert(errMsg.contains("ID is not a valid partition column"))
       }
 
       withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {

From 8f2d5db35e243c202cedf908dcf9a9d29de9614e Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sat, 12 Dec 2020 18:40:45 +0300
Subject: [PATCH 08/15] Remove V1 tests

---
 .../AlterTableDropPartitionSuiteBase.scala    | 11 +++-
 .../sql/execution/command/DDLSuite.scala      | 57 -------------------
 .../sql/hive/execution/HiveDDLSuite.scala     |  4 --
 3 files changed, 10 insertions(+), 62 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
index 65ade64d88679..ba0d7b11ebd12 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
@@ -107,6 +107,15 @@ trait AlterTableDropPartitionSuiteBase  extends QueryTest with SQLTestUtils {
     }
   }
 
+  test("table to alter does not exist") {
+    withNsTable("ns", "does_not_exist") { t =>
+      val errMsg = intercept[AnalysisException] {
+        sql(s"ALTER TABLE $t DROP PARTITION (a='4', b='9')")
+      }.getMessage
+      assert(errMsg.contains("Table not found"))
+    }
+  }
+
   test("partition not exists") {
     withNsTable("ns", "tbl") { t =>
       sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
@@ -115,7 +124,7 @@ trait AlterTableDropPartitionSuiteBase  extends QueryTest with SQLTestUtils {
       val errMsg = intercept[NoSuchPartitionsException] {
         sql(s"ALTER TABLE $t DROP PARTITION (id=1), PARTITION (id=2)")
       }.getMessage
-      assert(errMsg.contains("The following partitions not found in table"))
+      assert(errMsg.contains("partitions not found in table"))
 
       checkPartitions(t, Map("id" -> "1"))
       sql(s"ALTER TABLE $t DROP IF EXISTS PARTITION (id=1), PARTITION (id=2)")
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 05e0f4f4a538c..d6474ae7d5f00 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
@@ -334,10 +334,6 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
     testChangeColumn(isDatasourceTable = true)
   }
 
-  test("alter table: drop partition (datasource table)") {
-    testDropPartitions(isDatasourceTable = true)
-  }
-
   test("alter table: rename partition (datasource table)") {
     testRenamePartitions(isDatasourceTable = true)
   }
@@ -1617,59 +1613,6 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
     }
   }
 
-  protected def testDropPartitions(isDatasourceTable: Boolean): Unit = {
-    if (!isUsingHiveMetastore) {
-      assert(isDatasourceTable, "InMemoryCatalog only supports data source tables")
-    }
-    val catalog = spark.sessionState.catalog
-    val tableIdent = TableIdentifier("tab1", Some("dbx"))
-    val part1 = Map("a" -> "1", "b" -> "5")
-    val part2 = Map("a" -> "2", "b" -> "6")
-    val part3 = Map("a" -> "3", "b" -> "7")
-    val part4 = Map("a" -> "4", "b" -> "8")
-    val part5 = Map("a" -> "9", "b" -> "9")
-    createDatabase(catalog, "dbx")
-    createTable(catalog, tableIdent, isDatasourceTable)
-    createTablePartition(catalog, part1, tableIdent)
-    createTablePartition(catalog, part2, tableIdent)
-    createTablePartition(catalog, part3, tableIdent)
-    createTablePartition(catalog, part4, tableIdent)
-    createTablePartition(catalog, part5, tableIdent)
-    assert(catalog.listPartitions(tableIdent).map(_.spec).toSet ==
-      Set(part1, part2, part3, part4, part5))
-
-    // basic drop partition
-    sql("ALTER TABLE dbx.tab1 DROP IF EXISTS PARTITION (a='4', b='8'), PARTITION (a='3', b='7')")
-    assert(catalog.listPartitions(tableIdent).map(_.spec).toSet == Set(part1, part2, part5))
-
-    // drop partitions without explicitly specifying database
-    catalog.setCurrentDatabase("dbx")
-    sql("ALTER TABLE tab1 DROP IF EXISTS PARTITION (a='2', b ='6')")
-    assert(catalog.listPartitions(tableIdent).map(_.spec).toSet == Set(part1, part5))
-
-    // table to alter does not exist
-    intercept[AnalysisException] {
-      sql("ALTER TABLE does_not_exist DROP IF EXISTS PARTITION (a='2')")
-    }
-
-    // partition to drop does not exist
-    intercept[AnalysisException] {
-      sql("ALTER TABLE tab1 DROP PARTITION (a='300')")
-    }
-
-    // partition to drop does not exist when using IF EXISTS
-    sql("ALTER TABLE tab1 DROP IF EXISTS PARTITION (a='300')")
-    assert(catalog.listPartitions(tableIdent).map(_.spec).toSet == Set(part1, part5))
-
-    // partition spec in DROP PARTITION should be case insensitive by default
-    sql("ALTER TABLE tab1 DROP PARTITION (A='1', B='5')")
-    assert(catalog.listPartitions(tableIdent).map(_.spec).toSet == Set(part5))
-
-    // use int literal as partition value for int type partition column
-    sql("ALTER TABLE tab1 DROP PARTITION (a=9, b=9)")
-    assert(catalog.listPartitions(tableIdent).isEmpty)
-  }
-
   protected def testRenamePartitions(isDatasourceTable: Boolean): Unit = {
     if (!isUsingHiveMetastore) {
       assert(isDatasourceTable, "InMemoryCatalog only supports data source tables")
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
index 9f75f8797fe37..1de1980911f6d 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
@@ -162,10 +162,6 @@ class HiveCatalogedDDLSuite extends DDLSuite with TestHiveSingleton with BeforeA
     testRenamePartitions(isDatasourceTable = false)
   }
 
-  test("alter table: drop partition") {
-    testDropPartitions(isDatasourceTable = false)
-  }
-
   test("drop table") {
     testDropTable(isDatasourceTable = false)
   }

From 06ac24e9f2fcaf0b2edf946e71d9693cc0f0534b Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sat, 12 Dec 2020 19:38:48 +0300
Subject: [PATCH 09/15] Add AlterTableDropPartitionParserSuite

---
 .../sql/catalyst/parser/DDLParserSuite.scala  | 52 +----------
 .../AlterTableDropPartitionParserSuite.scala  | 92 +++++++++++++++++++
 2 files changed, 93 insertions(+), 51 deletions(-)
 create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionParserSuite.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 b860571df0791..5892b1c197b62 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
@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.parser
 import java.util.Locale
 
 import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, GlobalTempView, LocalTempView, PersistedView, UnresolvedAttribute, UnresolvedFunc, UnresolvedNamespace, UnresolvedPartitionSpec, UnresolvedRelation, UnresolvedStar, UnresolvedTable, UnresolvedTableOrView, UnresolvedView}
+import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, GlobalTempView, LocalTempView, PersistedView, UnresolvedAttribute, UnresolvedFunc, UnresolvedNamespace, UnresolvedRelation, UnresolvedStar, UnresolvedTable, UnresolvedTableOrView, UnresolvedView}
 import org.apache.spark.sql.catalyst.catalog.{ArchiveResource, BucketSpec, FileResource, FunctionResource, JarResource}
 import org.apache.spark.sql.catalyst.expressions.{EqualTo, Literal}
 import org.apache.spark.sql.catalyst.plans.logical._
@@ -2114,56 +2114,6 @@ class DDLParserSuite extends AnalysisTest {
     comparePlans(parsed2, expected2)
   }
 
-  // ALTER TABLE table_name DROP [IF EXISTS] PARTITION spec1[, PARTITION spec2, ...]
-  // ALTER VIEW table_name DROP [IF EXISTS] PARTITION spec1[, PARTITION spec2, ...]
-  test("alter table: drop partition") {
-    val sql1_table =
-      """
-        |ALTER TABLE table_name DROP IF EXISTS PARTITION
-        |(dt='2008-08-08', country='us'), PARTITION (dt='2009-09-09', country='uk')
-      """.stripMargin
-    val sql2_table =
-      """
-        |ALTER TABLE table_name DROP PARTITION
-        |(dt='2008-08-08', country='us'), PARTITION (dt='2009-09-09', country='uk')
-      """.stripMargin
-    val sql1_view = sql1_table.replace("TABLE", "VIEW")
-    val sql2_view = sql2_table.replace("TABLE", "VIEW")
-
-    val parsed1_table = parsePlan(sql1_table)
-    val parsed2_table = parsePlan(sql2_table)
-    val parsed1_purge = parsePlan(sql1_table + " PURGE")
-
-    assertUnsupported(sql1_view)
-    assertUnsupported(sql2_view)
-
-    val expected1_table = AlterTableDropPartition(
-      UnresolvedTable(Seq("table_name"), "ALTER TABLE ... DROP PARTITION ..."),
-      Seq(
-        UnresolvedPartitionSpec(Map("dt" -> "2008-08-08", "country" -> "us")),
-        UnresolvedPartitionSpec(Map("dt" -> "2009-09-09", "country" -> "uk"))),
-      ifExists = true,
-      purge = false,
-      retainData = false)
-    val expected2_table = expected1_table.copy(ifExists = false)
-    val expected1_purge = expected1_table.copy(purge = true)
-
-    comparePlans(parsed1_table, expected1_table)
-    comparePlans(parsed2_table, expected2_table)
-    comparePlans(parsed1_purge, expected1_purge)
-
-    val sql3_table = "ALTER TABLE a.b.c DROP IF EXISTS PARTITION (ds='2017-06-10')"
-    val expected3_table = AlterTableDropPartition(
-      UnresolvedTable(Seq("a", "b", "c"), "ALTER TABLE ... DROP PARTITION ..."),
-      Seq(UnresolvedPartitionSpec(Map("ds" -> "2017-06-10"))),
-      ifExists = true,
-      purge = false,
-      retainData = false)
-
-    val parsed3_table = parsePlan(sql3_table)
-    comparePlans(parsed3_table, expected3_table)
-  }
-
   test("show current namespace") {
     comparePlans(
       parsePlan("SHOW CURRENT NAMESPACE"),
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionParserSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionParserSuite.scala
new file mode 100644
index 0000000000000..efb12066caf13
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionParserSuite.scala
@@ -0,0 +1,92 @@
+/*
+ * 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, UnresolvedPartitionSpec, UnresolvedTable}
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser.parsePlan
+import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.catalyst.plans.logical.AlterTableDropPartition
+import org.apache.spark.sql.test.SharedSparkSession
+
+class AlterTableDropPartitionParserSuite extends AnalysisTest with SharedSparkSession {
+  test("drop partition") {
+    val sql = """
+      |ALTER TABLE table_name DROP PARTITION
+      |(dt='2008-08-08', country='us'), PARTITION (dt='2009-09-09', country='uk')
+      """.stripMargin
+    val expected = AlterTableDropPartition(
+      UnresolvedTable(Seq("table_name"), "ALTER TABLE ... DROP PARTITION ..."),
+      Seq(
+        UnresolvedPartitionSpec(Map("dt" -> "2008-08-08", "country" -> "us")),
+        UnresolvedPartitionSpec(Map("dt" -> "2009-09-09", "country" -> "uk"))),
+      ifExists = false,
+      purge = false,
+      retainData = false)
+
+    comparePlans(parsePlan(sql), expected)
+  }
+
+  test("drop partition if exists") {
+    val sql = """
+      |ALTER TABLE table_name DROP IF EXISTS
+      |PARTITION (dt='2008-08-08', country='us'),
+      |PARTITION (dt='2009-09-09', country='uk')
+      """.stripMargin
+    val expected = AlterTableDropPartition(
+      UnresolvedTable(Seq("table_name"), "ALTER TABLE ... DROP PARTITION ..."),
+      Seq(
+        UnresolvedPartitionSpec(Map("dt" -> "2008-08-08", "country" -> "us")),
+        UnresolvedPartitionSpec(Map("dt" -> "2009-09-09", "country" -> "uk"))),
+      ifExists = true,
+      purge = false,
+      retainData = false)
+    comparePlans(parsePlan(sql), expected)
+  }
+
+  test("drop partition in a table with multi-part identifier") {
+    val sql = "ALTER TABLE a.b.c DROP IF EXISTS PARTITION (ds='2017-06-10')"
+    val expected = AlterTableDropPartition(
+      UnresolvedTable(Seq("a", "b", "c"), "ALTER TABLE ... DROP PARTITION ..."),
+      Seq(UnresolvedPartitionSpec(Map("ds" -> "2017-06-10"))),
+      ifExists = true,
+      purge = false,
+      retainData = false)
+
+    comparePlans(parsePlan(sql), expected)
+  }
+
+  test("drop partition with PURGE") {
+    val sql = "ALTER TABLE table_name DROP PARTITION (p=1) PURGE"
+    val expected = AlterTableDropPartition(
+      UnresolvedTable(Seq("table_name"), "ALTER TABLE ... DROP PARTITION ..."),
+      Seq(UnresolvedPartitionSpec(Map("p" -> "1"))),
+      ifExists = false,
+      purge = true,
+      retainData = false)
+
+    comparePlans(parsePlan(sql), expected)
+  }
+
+  test("drop partition from view") {
+    val sql = "ALTER VIEW table_name DROP PARTITION (p=1)"
+    val errMsg = intercept[ParseException] {
+      parsePlan(sql)
+    }.getMessage
+    assert(errMsg.contains("Operation not allowed"))
+  }
+}

From 506ac8618f0357497590ec8c41f635a7f0f966ae Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sat, 12 Dec 2020 19:56:31 +0300
Subject: [PATCH 10/15] Add AlterTableDropPartitionSuite

---
 .../AlterTableDropPartitionSuite.scala        | 30 +++++++++++++++++++
 1 file changed, 30 insertions(+)
 create mode 100644 sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableDropPartitionSuite.scala

diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableDropPartitionSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableDropPartitionSuite.scala
new file mode 100644
index 0000000000000..1ddf0d505268f
--- /dev/null
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableDropPartitionSuite.scala
@@ -0,0 +1,30 @@
+/*
+ * 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
+import org.apache.spark.sql.hive.test.TestHiveSingleton
+
+class AlterTableDropPartitionSuite
+    extends v1.AlterTableDropPartitionSuiteBase
+    with TestHiveSingleton {
+  override def version: String = "Hive V1"
+  override def defaultUsing: String = "USING HIVE"
+
+  override protected val notFullPartitionSpecErr = "No partition is dropped"
+}

From 90d167dcbb4347da9cc7c6c541f3ac78da46b6a8 Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Sat, 12 Dec 2020 20:03:16 +0300
Subject: [PATCH 11/15] Hive specific test "partition not exists"

---
 .../AlterTableDropPartitionSuiteBase.scala    | 17 --------------
 .../v1/AlterTableDropPartitionSuite.scala     | 22 ++++++++++++++++++-
 .../v2/AlterTableDropPartitionSuite.scala     | 17 ++++++++++++++
 .../AlterTableDropPartitionSuite.scala        | 17 ++++++++++++++
 4 files changed, 55 insertions(+), 18 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
index ba0d7b11ebd12..8b285b2b28ddb 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
@@ -21,7 +21,6 @@ import org.scalactic.source.Position
 import org.scalatest.Tag
 
 import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
-import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException
 import org.apache.spark.sql.execution.datasources.PartitioningUtils
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SQLTestUtils
@@ -116,22 +115,6 @@ trait AlterTableDropPartitionSuiteBase  extends QueryTest with SQLTestUtils {
     }
   }
 
-  test("partition not exists") {
-    withNsTable("ns", "tbl") { t =>
-      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
-      sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
-
-      val errMsg = intercept[NoSuchPartitionsException] {
-        sql(s"ALTER TABLE $t DROP PARTITION (id=1), PARTITION (id=2)")
-      }.getMessage
-      assert(errMsg.contains("partitions not found in table"))
-
-      checkPartitions(t, Map("id" -> "1"))
-      sql(s"ALTER TABLE $t DROP IF EXISTS PARTITION (id=1), PARTITION (id=2)")
-      checkPartitions(t)
-    }
-  }
-
   test("case sensitivity in resolving partition specs") {
     withNsTable("ns", "tbl") { t =>
       sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableDropPartitionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableDropPartitionSuite.scala
index a0c62a93a17e0..5ad182bc689b9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableDropPartitionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableDropPartitionSuite.scala
@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.execution.command.v1
 
+import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException
 import org.apache.spark.sql.connector.catalog.CatalogManager
 import org.apache.spark.sql.execution.command
 import org.apache.spark.sql.test.SharedSparkSession
@@ -29,4 +30,23 @@ trait AlterTableDropPartitionSuiteBase extends command.AlterTableDropPartitionSu
   override protected val notFullPartitionSpecErr = "The following partitions not found in table"
 }
 
-class AlterTableDropPartitionSuite extends AlterTableDropPartitionSuiteBase with SharedSparkSession
+class AlterTableDropPartitionSuite
+  extends AlterTableDropPartitionSuiteBase
+  with SharedSparkSession {
+
+  test("partition not exists") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
+      sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+
+      val errMsg = intercept[NoSuchPartitionsException] {
+        sql(s"ALTER TABLE $t DROP PARTITION (id=1), PARTITION (id=2)")
+      }.getMessage
+      assert(errMsg.contains("partitions not found in table"))
+
+      checkPartitions(t, Map("id" -> "1"))
+      sql(s"ALTER TABLE $t DROP IF EXISTS PARTITION (id=1), PARTITION (id=2)")
+      checkPartitions(t)
+    }
+  }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
index 9dc1cad5a002d..608e7d7c98f6f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableDropPartitionSuite.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.command.v2
 
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException
 import org.apache.spark.sql.connector.{InMemoryPartitionTableCatalog, InMemoryTableCatalog}
 import org.apache.spark.sql.execution.command
 import org.apache.spark.sql.test.SharedSparkSession
@@ -37,6 +38,22 @@ class AlterTableDropPartitionSuite
     .set(s"spark.sql.catalog.$catalog", classOf[InMemoryPartitionTableCatalog].getName)
     .set(s"spark.sql.catalog.non_part_$catalog", classOf[InMemoryTableCatalog].getName)
 
+  test("partition not exists") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
+      sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+
+      val errMsg = intercept[NoSuchPartitionsException] {
+        sql(s"ALTER TABLE $t DROP PARTITION (id=1), PARTITION (id=2)")
+      }.getMessage
+      assert(errMsg.contains("partitions not found in table"))
+
+      checkPartitions(t, Map("id" -> "1"))
+      sql(s"ALTER TABLE $t DROP IF EXISTS PARTITION (id=1), PARTITION (id=2)")
+      checkPartitions(t)
+    }
+  }
+
   test("SPARK-33650: drop partition into a table which doesn't support partition management") {
     withNsTable("ns", "tbl", s"non_part_$catalog") { t =>
       sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing")
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableDropPartitionSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableDropPartitionSuite.scala
index 1ddf0d505268f..dbffd2b6b9bae 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableDropPartitionSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableDropPartitionSuite.scala
@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.hive.execution.command
 
+import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.execution.command.v1
 import org.apache.spark.sql.hive.test.TestHiveSingleton
 
@@ -27,4 +28,20 @@ class AlterTableDropPartitionSuite
   override def defaultUsing: String = "USING HIVE"
 
   override protected val notFullPartitionSpecErr = "No partition is dropped"
+
+  test("partition not exists") {
+    withNsTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED BY (id)")
+      sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+
+      val errMsg = intercept[AnalysisException] {
+        sql(s"ALTER TABLE $t DROP PARTITION (id=1), PARTITION (id=2)")
+      }.getMessage
+      assert(errMsg.contains("No partition is dropped"))
+
+      checkPartitions(t, Map("id" -> "1"))
+      sql(s"ALTER TABLE $t DROP IF EXISTS PARTITION (id=1), PARTITION (id=2)")
+      checkPartitions(t)
+    }
+  }
 }

From 7ac23086bd9dc5b77ec3adcc2ffa075fc25d03a7 Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Mon, 14 Dec 2020 17:41:44 +0300
Subject: [PATCH 12/15] Fix test names

---
 .../execution/command/AlterTableDropPartitionSuiteBase.scala    | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
index 8b285b2b28ddb..a0e4fd6ff61f9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
@@ -34,7 +34,7 @@ trait AlterTableDropPartitionSuiteBase  extends QueryTest with SQLTestUtils {
 
   override def test(testName: String, testTags: Tag*)(testFun: => Any)
     (implicit pos: Position): Unit = {
-    super.test(s"ALTER TABLE .. ADD PARTITION $version: " + testName, testTags: _*)(testFun)
+    super.test(s"ALTER TABLE .. DROP PARTITION $version: " + testName, testTags: _*)(testFun)
   }
 
   protected def withNsTable(ns: String, tableName: String, cat: String = catalog)

From d4a1c9377e672d780d2b6cc8c6a29757d2a158dc Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Mon, 14 Dec 2020 17:46:43 +0300
Subject: [PATCH 13/15] Fix merge issues in AlterTableDropPartitionParserSuite

---
 .../command/AlterTableDropPartitionParserSuite.scala | 12 ++++--------
 1 file changed, 4 insertions(+), 8 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionParserSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionParserSuite.scala
index efb12066caf13..53edd5854f289 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionParserSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionParserSuite.scala
@@ -35,8 +35,7 @@ class AlterTableDropPartitionParserSuite extends AnalysisTest with SharedSparkSe
         UnresolvedPartitionSpec(Map("dt" -> "2008-08-08", "country" -> "us")),
         UnresolvedPartitionSpec(Map("dt" -> "2009-09-09", "country" -> "uk"))),
       ifExists = false,
-      purge = false,
-      retainData = false)
+      purge = false)
 
     comparePlans(parsePlan(sql), expected)
   }
@@ -53,8 +52,7 @@ class AlterTableDropPartitionParserSuite extends AnalysisTest with SharedSparkSe
         UnresolvedPartitionSpec(Map("dt" -> "2008-08-08", "country" -> "us")),
         UnresolvedPartitionSpec(Map("dt" -> "2009-09-09", "country" -> "uk"))),
       ifExists = true,
-      purge = false,
-      retainData = false)
+      purge = false)
     comparePlans(parsePlan(sql), expected)
   }
 
@@ -64,8 +62,7 @@ class AlterTableDropPartitionParserSuite extends AnalysisTest with SharedSparkSe
       UnresolvedTable(Seq("a", "b", "c"), "ALTER TABLE ... DROP PARTITION ..."),
       Seq(UnresolvedPartitionSpec(Map("ds" -> "2017-06-10"))),
       ifExists = true,
-      purge = false,
-      retainData = false)
+      purge = false)
 
     comparePlans(parsePlan(sql), expected)
   }
@@ -76,8 +73,7 @@ class AlterTableDropPartitionParserSuite extends AnalysisTest with SharedSparkSe
       UnresolvedTable(Seq("table_name"), "ALTER TABLE ... DROP PARTITION ..."),
       Seq(UnresolvedPartitionSpec(Map("p" -> "1"))),
       ifExists = false,
-      purge = true,
-      retainData = false)
+      purge = true)
 
     comparePlans(parsePlan(sql), expected)
   }

From 9126bc09b41074ba7891348208c1182519b3fa7e Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Mon, 14 Dec 2020 17:59:25 +0300
Subject: [PATCH 14/15] Fix indentation

---
 .../execution/command/AlterTableDropPartitionSuite.scala     | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableDropPartitionSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableDropPartitionSuite.scala
index dbffd2b6b9bae..fe26466cdad62 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableDropPartitionSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableDropPartitionSuite.scala
@@ -22,8 +22,9 @@ import org.apache.spark.sql.execution.command.v1
 import org.apache.spark.sql.hive.test.TestHiveSingleton
 
 class AlterTableDropPartitionSuite
-    extends v1.AlterTableDropPartitionSuiteBase
-    with TestHiveSingleton {
+  extends v1.AlterTableDropPartitionSuiteBase
+  with TestHiveSingleton {
+
   override def version: String = "Hive V1"
   override def defaultUsing: String = "USING HIVE"
 

From 62f539b1e5cc7f573e789fc2a1f5386c431f6ad3 Mon Sep 17 00:00:00 2001
From: Max Gekk <max.gekk@gmail.com>
Date: Mon, 14 Dec 2020 20:10:53 +0300
Subject: [PATCH 15/15] Fix for Scala 2.13

---
 .../execution/command/AlterTableDropPartitionSuiteBase.scala    | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
index a0e4fd6ff61f9..ed479e2824fb7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableDropPartitionSuiteBase.scala
@@ -62,7 +62,7 @@ trait AlterTableDropPartitionSuiteBase  extends QueryTest with SQLTestUtils {
       t: String,
       ifExists: String,
       specs: Map[String, Any]*): Unit = {
-    checkPartitions(t, specs.map(_.mapValues(_.toString)): _*)
+    checkPartitions(t, specs.map(_.mapValues(_.toString).toMap): _*)
     val specStr = specs.map(
       _.map {
         case (k, v: String) => s"$k = '$v'"