[spark] Support spark DropPartition with sync metastore

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala

@@ -69,10 +69,21 @@ trait PaimonPartitionManagement extends SupportsAtomicPartitionManagement {
       case fileStoreTable: FileStoreTable =>
         val partitions = toPaimonPartitions(rows).map(_.asInstanceOf[JMap[String, String]])
         val commit: FileStoreCommit = fileStoreTable.store.newCommit(UUID.randomUUID.toString)
+        var metastoreClient: MetastoreClient = null
+        val clientFactory = fileStoreTable.catalogEnvironment().metastoreClientFactory
         try {
           commit.dropPartitions(partitions.toSeq.asJava, BatchWriteBuilder.COMMIT_IDENTIFIER)
+          // syn to metastore with delete partitions
+          if (clientFactory != null) {
+            metastoreClient = clientFactory.create()
+            toPaimonPartitions(rows).foreach(metastoreClient.deletePartition)
+          }
+
         } finally {
           commit.close()
+          if (metastoreClient != null) {
+            metastoreClient.close()
+          }
         }
         true
 

paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala

@@ -59,6 +59,50 @@ abstract class DDLWithHiveCatalogTestBase extends PaimonHiveTestBase {
     }
   }
 
+  test("Paimon DDL with hive catalog: drop partition for paimon table sparkCatalogName") {
+    Seq(paimonHiveCatalogName).foreach {
+      catalogName =>
+        spark.sql(s"USE $catalogName")
+        withTempDir {
+          dBLocation =>
+            withDatabase("paimon_db") {
+              val comment = "this is a test comment"
+              spark.sql(
+                s"CREATE DATABASE paimon_db LOCATION '${dBLocation.getCanonicalPath}' COMMENT '$comment'")
+              Assertions.assertEquals(getDatabaseLocation("paimon_db"), dBLocation.getCanonicalPath)
+              Assertions.assertEquals(getDatabaseComment("paimon_db"), comment)
+
+              withTable("paimon_db.paimon_tbl") {
+                spark.sql(s"""
+                             |CREATE TABLE paimon_db.paimon_tbl (id STRING, name STRING, pt STRING)
+                             |USING PAIMON
+                             |PARTITIONED BY (name, pt)
+                             |TBLPROPERTIES('metastore.partitioned-table' = 'true')
+                             |""".stripMargin)
+                Assertions.assertEquals(
+                  getTableLocation("paimon_db.paimon_tbl"),
+                  s"${dBLocation.getCanonicalPath}/paimon_tbl")
+                spark.sql("insert into paimon_db.paimon_tbl select '1', 'n', 'cc'")
+                spark.sql("insert into paimon_db.paimon_tbl select '1', 'n1', 'aa'")
+                spark.sql("insert into paimon_db.paimon_tbl select '1', 'n2', 'bb'")
+
+                spark.sql("show partitions paimon_db.paimon_tbl")
+                checkAnswer(
+                  spark.sql("show partitions paimon_db.paimon_tbl"),
+                  Row("name=n/pt=cc") :: Row("name=n1/pt=aa") :: Row("name=n2/pt=bb") :: Nil)
+                spark.sql(
+                  "alter table paimon_db.paimon_tbl drop partition (name='n1', `pt`='aa'), partition (name='n2', `pt`='bb')")
+                spark.sql("show partitions paimon_db.paimon_tbl")
+                checkAnswer(
+                  spark.sql("show partitions paimon_db.paimon_tbl"),
+                  Row("name=n/pt=cc") :: Nil)
+
+              }
+            }
+        }
+    }
+  }
+
   test("Paimon DDL with hive catalog: create partition for paimon table sparkCatalogName") {
     Seq(paimonHiveCatalogName).foreach {
       catalogName =>