[HADP-54510] Validate the folder for dynamic partition to avoid data quality issue (apache#476)

Author: colinmjj

core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala

@@ -293,6 +293,17 @@ class HadoopMapReduceCommitProtocol(
       if (dynamicPartitionOverwrite) {
         val partitionPaths = allPartitionPaths.foldLeft(Set[String]())(_ ++ _)
         logDebug(s"Clean up default partition directories for overwriting: $partitionPaths")
+        // to avoid partial commit because of missed expected partition,
+        // we need to check if all required partitions exist.
+        // but the process of validation and rename are not atomic,
+        // it is possible that partial commit will happen
+        for (part <- partitionPaths) {
+          if (!fs.exists(new Path(stagingDir, part))) {
+            throw new IOException(s"Failed to validate all required partitions exist when " +
+              s"committing files staged for overwriting dynamic partitions, " +
+              s"missed partition: $part")
+          }
+        }
         for (part <- partitionPaths) {
           val finalPartPath = new Path(path, part)
           if (!fs.delete(finalPartPath, true) && !fs.exists(finalPartPath.getParent)) {

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

@@ -2628,6 +2628,27 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
     }
   }
 
+  test("HADP-54510: Validate the folder for dynamic partition to avoid data quality issue") {
+    withSQLConf(
+      "fs.file.impl" -> classOf[ExistsFromSparkStagingToFinalDirAlwaysTurnsFalseFilesystem].getName,
+      "fs.file.impl.disable.cache" -> "true",
+      SQLConf.PARTITION_OVERWRITE_MODE.key -> PartitionOverwriteMode.DYNAMIC.toString) {
+
+      withTable("t") {
+        sql(
+          """
+            |create table t(i int, part1 int, part2 int) using parquet
+            |partitioned by (part1, part2)
+          """.stripMargin)
+
+        val e = intercept[IOException] {
+          sql(s"insert overwrite table t partition(part1, part2) values (1, 1, 1)")
+        }
+        assert(e.getMessage.contains("Failed to validate all required partitions"))
+      }
+    }
+  }
+
   test("SPARK-36980: Insert support query with CTE") {
     withTable("t") {
       sql("CREATE TABLE t(i int, part1 int, part2 int) using parquet")
@@ -2851,3 +2872,13 @@ class RenameFromSparkStagingToFinalDirAlwaysTurnsFalseFilesystem extends RawLoca
     path.toString.contains(".spark-staging-")
   }
 }
+
+class ExistsFromSparkStagingToFinalDirAlwaysTurnsFalseFilesystem extends RawLocalFileSystem {
+  override def exists(path: Path): Boolean = {
+    !isSparkStagingDir(path) && super.exists(path)
+  }
+
+  private def isSparkStagingDir(path: Path): Boolean = {
+    path.toString.contains(".spark-staging-")
+  }
+}