change the warning for purge

Change-Id: I11a9dc214102854f533330a739d32a41500d9278

Author: stczwd

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala

@@ -249,13 +249,14 @@ class ResolveCatalogs(val catalogManager: CatalogManager)
 
     case c @ AlterTableDropPartitionStatement(
         NonSessionCatalogAndTable(catalog, tableName), _, _, _, _) =>
+      if (c.purge) {
+        logWarning("PURGE won't take effect here, please put it in table properties")
+      }
       val table = catalog.asTableCatalog.loadTable(tableName.asIdentifier).asPartitionable
       AlterTableDropPartition(
         table,
         c.specs.map(_.asPartitionIdentifier(table.partitionSchema())),
-        c.ifExists,
-        c.purge,
-        c.retainData)
+        c.ifExists)
   }
 
   object NonSessionCatalogAndTable {

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

@@ -573,12 +573,10 @@ case class AlterTableAddPartition(
  *
  * The syntax of this command is:
  * {{{
- *     ALTER TABLE table DROP [IF EXISTS] PARTITION spec1[, PARTITION spec2, ...] [PURGE];
+ *     ALTER TABLE table DROP [IF EXISTS] PARTITION spec1[, PARTITION spec2, ...];
  * }}}
  */
 case class AlterTableDropPartition(
     table: SupportsPartitions,
     partIdents: Seq[InternalRow],
-    ignoreIfNotExists: Boolean,
-    purge: Boolean,
-    retainData: Boolean) extends Command
+    ignoreIfNotExists: Boolean) extends Command

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

@@ -28,15 +28,11 @@ import org.apache.spark.sql.connector.catalog.SupportsPartitions
 case class AlterTableDropPartitionExec(
     table: SupportsPartitions,
     partIdents: Seq[InternalRow],
-    ignoreIfNotExists: Boolean,
-    purge: Boolean,
-    retainData: Boolean) extends V2CommandExec {
+    ignoreIfNotExists: Boolean) extends V2CommandExec {
 
   override def output: Seq[Attribute] = Seq.empty
 
   override protected def run(): Seq[InternalRow] = {
-    logWarning("PURGE won't take effect, please put it in table properties")
-
     partIdents.foreach { partIdent =>
       if (!table.dropPartition(partIdent) && !ignoreIfNotExists) {
         throw new NoSuchPartitionException(table.name(), partIdent, table.partitionSchema())

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

@@ -279,9 +279,9 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
     case AlterTableAddPartition(table, parts, ignoreIfExists) =>
       AlterTableAddPartitionExec(table, parts, ignoreIfExists) :: Nil
 
-    case AlterTableDropPartition(table, partIdents, ignoreIfNotExists, purge, retainData) =>
+    case AlterTableDropPartition(table, partIdents, ignoreIfNotExists) =>
       AlterTableDropPartitionExec(
-        table, partIdents, ignoreIfNotExists, purge, retainData) :: Nil
+        table, partIdents, ignoreIfNotExists) :: Nil
 
     case _ => Nil
   }