[SPARK-14922][SPARK-17732][SPARK-23866][SQL] Support partition filter in ALTER TABLE DROP PARTITION and batch dropping PARTITIONS

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -161,8 +161,8 @@ statement
         partitionSpecLocation+                                         #addTablePartition
     | ALTER TABLE multipartIdentifier
         from=partitionSpec RENAME TO to=partitionSpec                  #renameTablePartition
-    | ALTER (TABLE | VIEW) multipartIdentifier
-        DROP (IF EXISTS)? partitionSpec (',' partitionSpec)* PURGE?    #dropTablePartitions
+    | ALTER (TABLE | VIEW) multipartIdentifier DROP (IF EXISTS)?
+        dropPartitionSpec (',' dropPartitionSpec)* PURGE?              #dropTablePartitions
     | ALTER TABLE multipartIdentifier
         (partitionSpec)? SET locationSpec                              #setTableLocation
     | ALTER TABLE multipartIdentifier RECOVER PARTITIONS               #recoverPartitions
@@ -328,6 +328,14 @@ database
     | SCHEMA
     ;
 
+dropPartitionSpec
+    : PARTITION '(' dropPartitionVal (',' dropPartitionVal)* ')'
+    ;
+
+dropPartitionVal
+    : identifier (comparisonOperator constant)?
+    ;
+
 describeFuncName
     : qualifiedName
     | STRING

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

@@ -188,7 +188,8 @@ trait ExternalCatalog {
       parts: Seq[TablePartitionSpec],
       ignoreIfNotExists: Boolean,
       purge: Boolean,
-      retainData: Boolean): Unit
+      retainData: Boolean,
+      supportBatch: Boolean): Unit
 
   /**
    * Override the specs of one or many existing table partitions, assuming they exist.

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

@@ -203,8 +203,10 @@ class ExternalCatalogWithListener(delegate: ExternalCatalog)
       partSpecs: Seq[TablePartitionSpec],
       ignoreIfNotExists: Boolean,
       purge: Boolean,
-      retainData: Boolean): Unit = {
-    delegate.dropPartitions(db, table, partSpecs, ignoreIfNotExists, purge, retainData)
+      retainData: Boolean,
+      supportBatch: Boolean): Unit = {
+    delegate.dropPartitions(db, table, partSpecs, ignoreIfNotExists, purge, retainData,
+      supportBatch)
   }
 
   override def renamePartitions(

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

@@ -425,7 +425,8 @@ class InMemoryCatalog(
       partSpecs: Seq[TablePartitionSpec],
       ignoreIfNotExists: Boolean,
       purge: Boolean,
-      retainData: Boolean): Unit = synchronized {
+      retainData: Boolean,
+      supportBatch: Boolean): Unit = synchronized {
     requireTableExists(db, table)
     val existingParts = catalog(db).tables(table).partitions
     if (!ignoreIfNotExists) {

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

@@ -918,14 +918,16 @@ class SessionCatalog(
       specs: Seq[TablePartitionSpec],
       ignoreIfNotExists: Boolean,
       purge: Boolean,
-      retainData: Boolean): Unit = {
+      retainData: Boolean,
+      supportBatch: Boolean): Unit = {
     val db = formatDatabaseName(tableName.database.getOrElse(getCurrentDatabase))
     val table = formatTableName(tableName.table)
     requireDbExists(db)
     requireTableExists(TableIdentifier(table, Option(db)))
     requirePartialMatchedPartitionSpec(specs, getTableMetadata(tableName))
     requireNonEmptyValueInPartitionSpec(specs)
-    externalCatalog.dropPartitions(db, table, specs, ignoreIfNotExists, purge, retainData)
+    externalCatalog.dropPartitions(db, table, specs, ignoreIfNotExists, purge, retainData,
+      supportBatch)
   }
 
   /**

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala

@@ -391,6 +391,35 @@ case class OuterReference(e: NamedExpression)
   override def newInstance(): NamedExpression = OuterReference(e.newInstance())
 }
 
+/**
+ * A place holder used to hold the name of the partition attributes specified when running commands
+ * involving partitions, eg. ALTER TABLE ... DROP PARTITIONS.
+ */
+case class PartitioningAttribute(
+    name: String,
+    override val exprId: ExprId = NamedExpression.newExprId)
+  extends Attribute with Unevaluable {
+  // We need a dataType to be used during analysis for resolving the expressions (see
+  // checkInputDataTypes). The String type is used because all the literals in PARTITION operations
+  // are parsed as strings and eventually casted later.
+  override def dataType: DataType = StringType
+  override def nullable: Boolean = false
+
+  override def qualifier: Seq[String] = throw new UnsupportedOperationException
+  override def withNullability(newNullability: Boolean): Attribute =
+    throw new UnsupportedOperationException
+  override def newInstance(): Attribute = throw new UnsupportedOperationException
+  override def withQualifier(newQualifier: Seq[String]): Attribute =
+    throw new UnsupportedOperationException
+  override def withName(newName: String): Attribute = throw new UnsupportedOperationException
+  override def withMetadata(newMetadata: Metadata): Attribute =
+    throw new UnsupportedOperationException
+
+  override lazy val canonicalized: Expression = this.copy(exprId = ExprId(0))
+
+  override def withExprId(newExprId: ExprId): Attribute = throw new UnsupportedOperationException
+}
+
 object VirtualColumn {
   // The attribute name used by Hive, which has different result than Spark, deprecated.
   val hiveGroupingIdName: String = "grouping__id"

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -499,6 +499,29 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
     }
   }
 
+  /**
+   * Create a partition specification map with filters.
+   */
+  override def visitDropPartitionSpec(
+      ctx: DropPartitionSpecContext): Seq[Expression] = withOrigin(ctx) {
+    ctx.dropPartitionVal().asScala.map { pFilter =>
+      if (pFilter.constant() == null || pFilter.comparisonOperator() == null) {
+        throw new ParseException(s"Invalid partition spec: ${pFilter.getText}", ctx)
+      }
+      // We cannot use UnresolvedAttribute because resolution is performed after Analysis, when
+      // running the command.
+      val partition = PartitioningAttribute(pFilter.identifier().getText)
+      val value = Literal(visitStringConstant(pFilter.constant()))
+      val operator = pFilter.comparisonOperator().getChild(0).asInstanceOf[TerminalNode]
+      val comparison = buildComparison(partition, value, operator)
+      if (comparison.isInstanceOf[EqualNullSafe]) {
+        throw new ParseException(
+          "'<=>' operator is not supported in ALTER TABLE ... DROP PARTITION.", ctx)
+      }
+      comparison
+    }
+  }
+
   /**
    * Convert a constant of any type into a string. This is typically used in DDL commands, and its
    * main purpose is to prevent slight differences due to back to back conversions i.e.:
@@ -1321,6 +1344,23 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
     val left = expression(ctx.left)
     val right = expression(ctx.right)
     val operator = ctx.comparisonOperator().getChild(0).asInstanceOf[TerminalNode]
+    buildComparison(left, right, operator)
+  }
+
+  /**
+   * Creates a comparison expression. The following comparison operators are supported:
+   * - Equal: '=' or '=='
+   * - Null-safe Equal: '<=>'
+   * - Not Equal: '<>' or '!='
+   * - Less than: '<'
+   * - Less then or Equal: '<='
+   * - Greater than: '>'
+   * - Greater then or Equal: '>='
+   */
+  private def buildComparison(
+      left: Expression,
+      right: Expression,
+      operator: TerminalNode): Expression = {
     operator.getSymbol.getType match {
       case SqlBaseParser.EQ =>
         EqualTo(left, right)
@@ -3153,7 +3193,7 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
     }
     AlterTableDropPartitionStatement(
       visitMultipartIdentifier(ctx.multipartIdentifier),
-      ctx.partitionSpec.asScala.map(visitNonOptionalPartitionSpec),
+      ctx.dropPartitionSpec.asScala.map(visitDropPartitionSpec),
       ifExists = ctx.EXISTS != null,
       purge = ctx.PURGE != null,
       retainData = false)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala

@@ -209,7 +209,7 @@ case class AlterTableRenamePartitionStatement(
  */
 case class AlterTableDropPartitionStatement(
     tableName: Seq[String],
-    specs: Seq[TablePartitionSpec],
+    specs: Seq[Seq[Expression]],
     ifExists: Boolean,
     purge: Boolean,
     retainData: Boolean) extends ParsedStatement