modify the check logic

Author: cxzl25

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

@@ -407,10 +407,8 @@ case class PreprocessTableInsertion(conf: SQLConf) extends Rule[LogicalPlan] {
       catalogTable.isDefined &&
       catalogTable.get.partitionColumnNames.nonEmpty &&
       catalogTable.get.tracksPartitionsInCatalog
-    // check static partition
     if (partitionsTrackedByCatalog &&
-      normalizedPartSpec.nonEmpty &&
-      staticPartCols.size == partColNames.size) {
+      normalizedPartSpec.nonEmpty) {
       // empty partition column value
       if (normalizedPartSpec.filter(_._2.isDefined).exists(_._2.get.isEmpty)) {
         val spec = normalizedPartSpec.map(p => p._1 + "=" + p._2).mkString("[", ", ", "]")

sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala

@@ -877,10 +877,17 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
             """.stripMargin)
 
       withSQLConf(SQLConf.HIVE_MANAGE_FILESOURCE_PARTITIONS.key -> "true") {
-        val e = intercept[AnalysisException] {
+        val msg = "Partition spec is invalid"
+        assert(intercept[AnalysisException] {
           sql("INSERT INTO TABLE insertTable PARTITION(part1=1, part2='') SELECT 1")
-        }.getMessage
-        assert(e.contains("Partition spec is invalid"))
+        }.getMessage.contains(msg))
+        assert(intercept[AnalysisException] {
+          sql("INSERT INTO TABLE insertTable PARTITION(part1='', part2) SELECT 1 ,'' AS part2")
+        }.getMessage.contains(msg))
+
+        sql("INSERT INTO TABLE insertTable PARTITION(part1='1', part2='2') SELECT 1")
+        sql("INSERT INTO TABLE insertTable PARTITION(part1='1', part2) SELECT 1 ,'2' AS part2")
+        sql("INSERT INTO TABLE insertTable PARTITION(part1='1', part2) SELECT 1 ,'' AS part2")
       }
 
       withSQLConf(SQLConf.HIVE_MANAGE_FILESOURCE_PARTITIONS.key -> "false") {