Datasource partition table should load empty partitions.

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

@@ -166,7 +166,12 @@ case class InsertIntoHadoopFsRelationCommand(
 
 
       // update metastore partition metadata
-      refreshUpdatedPartitions(updatedPartitionPaths)
+      if (staticPartitions.nonEmpty) {
+        // Avoid empty partition can't loaded.
+        refreshUpdatedPartitions(Set(PartitioningUtils.getPathFragment(staticPartitions)))
+      } else {
+        refreshUpdatedPartitions(updatedPartitionPaths)
+      }
 
       // refresh cached files in FileIndex
       fileIndex.foreach(_.refresh())

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

@@ -284,6 +284,11 @@ object PartitioningUtils {
     }.mkString("/")
   }
 
+  def getPathFragment(partitions: TablePartitionSpec): String = partitions.map {
+    case (k, v) =>
+      escapePathName(k) + "=" + escapePathName(v)
+  }.mkString("/")
+
   /**
    * Normalize the column names in partition specification, w.r.t. the real partition column names
    * and case sensitivity. e.g., if the partition spec has a column named `monTh`, and there is a

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala

@@ -2249,6 +2249,27 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
     }
   }
 
+  test("Datasource partition table should load empty partitions") {
+    withTable("t", "t1") {
+      withTempPath { dir =>
+        spark.sql("CREATE TABLE t(a int) USING parquet")
+        spark.sql("CREATE TABLE t1(a int, b string, c string) " +
+          s"USING parquet PARTITIONED BY(b, c) LOCATION '${dir.toURI}'")
+
+        val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t1"))
+        assert(table.location == makeQualifiedPath(dir.getAbsolutePath))
+
+        spark.sql("INSERT INTO TABLE t1 PARTITION(b='b', c='c') SELECT * FROM t WHERE 1 = 0")
+
+        assert(spark.sql("SHOW PARTITIONS t1").count() == 1)
+
+        assert(new File(dir, "b=b/c=c").exists())
+
+        checkAnswer(spark.table("t1"), Nil)
+      }
+    }
+  }
+
   Seq(true, false).foreach { shouldDelete =>
     val tcName = if (shouldDelete) "non-existing" else "existed"
     test(s"CTAS for external data source table with a $tcName location") {