Use UnresolvedPartitionSpec

Author: imback82

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

@@ -587,6 +587,9 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog {
           case AlterTableDropPartition(ResolvedTable(_, _, table), parts, _, _) =>
             checkAlterTablePartition(table, parts)
 
+          case AlterTableRenamePartition(ResolvedTable(_, _, table), from, _) =>
+            checkAlterTablePartition(table, Seq(from))
+
           case showPartitions: ShowPartitions => checkShowPartitions(showPartitions)
 
           case _ => // Falls back to the following checks

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

@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.analysis
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
-import org.apache.spark.sql.catalyst.plans.logical.{AlterTableAddPartition, AlterTableDropPartition, LogicalPlan, ShowPartitions}
+import org.apache.spark.sql.catalyst.plans.logical.{AlterTableAddPartition, AlterTableDropPartition, AlterTableRenamePartition, LogicalPlan, ShowPartitions}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils
 import org.apache.spark.sql.connector.catalog.SupportsPartitionManagement
@@ -51,6 +51,15 @@ object ResolvePartitionSpec extends Rule[LogicalPlan] {
         partitionSchema,
         requireExactMatchedPartitionSpec(table.name, _, partitionSchema.fieldNames)))
 
+    case r @ AlterTableRenamePartition(
+        ResolvedTable(_, _, table: SupportsPartitionManagement), from, _) =>
+      val partitionSchema = table.partitionSchema()
+      r.copy(from = resolvePartitionSpecs(
+        table.name,
+        Seq(from),
+        partitionSchema,
+        requireExactMatchedPartitionSpec(table.name, _, partitionSchema.fieldNames)).head)
+
     case r @ ShowPartitions(ResolvedTable(_, _, table: SupportsPartitionManagement), partSpecs) =>
       r.copy(pattern = resolvePartitionSpecs(
         table.name,

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

@@ -3774,7 +3774,7 @@ class AstBuilder extends SqlBaseBaseVisitor[AnyRef] with SQLConfHelper with Logg
       UnresolvedTable(
         visitMultipartIdentifier(ctx.multipartIdentifier),
         "ALTER TABLE ... RENAME TO PARTITION"),
-      visitNonOptionalPartitionSpec(ctx.from),
+      UnresolvedPartitionSpec(visitNonOptionalPartitionSpec(ctx.from)),
       visitNonOptionalPartitionSpec(ctx.to))
   }
 

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

@@ -678,8 +678,11 @@ case class AlterTableDropPartition(
  */
 case class AlterTableRenamePartition(
     child: LogicalPlan,
-    from: TablePartitionSpec,
+    from: PartitionSpec,
     to: TablePartitionSpec) extends Command {
+  override lazy val resolved: Boolean =
+    childrenResolved && from.isInstanceOf[ResolvedPartitionSpec]
+
   override def children: Seq[LogicalPlan] = child :: Nil
 }
 

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, UnresolvedRelation, UnresolvedStar, UnresolvedTable, UnresolvedTableOrView, UnresolvedView}
+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.catalog.{ArchiveResource, BucketSpec, FileResource, FunctionResource, JarResource}
 import org.apache.spark.sql.catalyst.expressions.{EqualTo, Literal}
 import org.apache.spark.sql.catalyst.plans.logical._
@@ -2108,7 +2108,7 @@ class DDLParserSuite extends AnalysisTest {
     val parsed1 = parsePlan(sql1)
     val expected1 = AlterTableRenamePartition(
       UnresolvedTable(Seq("table_name"), "ALTER TABLE ... RENAME TO PARTITION"),
-      Map("dt" -> "2008-08-08", "country" -> "us"),
+      UnresolvedPartitionSpec(Map("dt" -> "2008-08-08", "country" -> "us")),
       Map("dt" -> "2008-09-09", "country" -> "uk"))
     comparePlans(parsed1, expected1)
 
@@ -2120,7 +2120,7 @@ class DDLParserSuite extends AnalysisTest {
     val parsed2 = parsePlan(sql2)
     val expected2 = AlterTableRenamePartition(
       UnresolvedTable(Seq("a", "b", "c"), "ALTER TABLE ... RENAME TO PARTITION"),
-      Map("ds" -> "2017-06-10"),
+      UnresolvedPartitionSpec(Map("ds" -> "2017-06-10")),
       Map("ds" -> "2018-06-10"))
     comparePlans(parsed2, expected2)
   }

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -445,7 +445,8 @@ class ResolveSessionCatalog(
         partSpecsAndLocs.asUnresolvedPartitionSpecs.map(spec => (spec.spec, spec.location)),
         ifNotExists)
 
-    case AlterTableRenamePartition(ResolvedV1TableIdentifier(ident), from, to) =>
+    case AlterTableRenamePartition(
+        ResolvedV1TableIdentifier(ident), UnresolvedPartitionSpec(from, _), to) =>
       AlterTableRenamePartitionCommand(
         ident.asTableIdentifier,
         from,

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

@@ -342,7 +342,7 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
       AlterTableDropPartitionExec(
         table, parts.asResolvedPartitionSpecs, ignoreIfNotExists) :: Nil
 
-    case AlterTableRenamePartition(_: ResolvedTable, _, _) =>
+    case AlterTableRenamePartition(_: ResolvedTable, _: ResolvedPartitionSpec, _) =>
       throw new AnalysisException(
         "ALTER TABLE ... RENAME TO PARTITION is not supported for v2 tables.")
 

sql/core/src/test/scala/org/apache/spark/sql/connector/AlterTablePartitionV2SQLSuite.scala

@@ -33,13 +33,20 @@ class AlterTablePartitionV2SQLSuite extends DatasourceV2SQLBase {
   }
 
   test("ALTER TABLE RENAME PARTITION") {
-    val t = "testcat.ns1.ns2.tbl"
-    withTable(t) {
-      spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
-      val e = intercept[AnalysisException] {
-        sql(s"ALTER TABLE $t PARTITION (id=1) RENAME TO PARTITION (id=2)")
+    val nonPartTbl = "testcat.ns1.ns2.tbl"
+    val partTbl = "testpart.ns1.ns2.tbl"
+    withTable(nonPartTbl, partTbl) {
+      spark.sql(s"CREATE TABLE $nonPartTbl (id bigint, data string) USING foo PARTITIONED BY (id)")
+      val e1 = intercept[AnalysisException] {
+        sql(s"ALTER TABLE $nonPartTbl PARTITION (id=1) RENAME TO PARTITION (id=2)")
       }
-      assert(e.message.contains(
+      assert(e1.message.contains(s"Table $nonPartTbl can not alter partitions"))
+
+      spark.sql(s"CREATE TABLE $partTbl (id bigint, data string) USING foo PARTITIONED BY (id)")
+      val e2 = intercept[AnalysisException] {
+        sql(s"ALTER TABLE $partTbl PARTITION (id=1) RENAME TO PARTITION (id=2)")
+      }
+      assert(e2.message.contains(
         "ALTER TABLE ... RENAME TO PARTITION is not supported for v2 tables."))
     }
   }