Automatic staging committer conflict-mode for dynamic partition overwrite

As an attempt to support dynamic partition overwrite using S3A staging committers,
we disabled the previous hard-coded exception and dynamically set the value for
fs.s3a.committer.staging.conflict-mode
so to ensure PartitionedStagingCommitter behaves as expected in both
"INSERT INTO" and "INSERT OVERWRITE" scenarios.

The details are documented at:
https://docs.google.com/document/d/1fH4AtClYDiQt4fU9g-QzcoRu9SxMo8isGuLgvVxZgdc/edit?usp=sharing

(cherry picked from commit 902996c)

Author: dzhi-lyft

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

@@ -183,6 +183,18 @@ class HadoopMapReduceCommitProtocol(
     jobContext.getConfiguration.setBoolean("mapreduce.task.ismap", true)
     jobContext.getConfiguration.setInt("mapreduce.task.partition", 0)
 
+    // Automatically set conflict-mode based on value of dynamicPartitionOverwrite,
+    // unless configuration auto-staging-conflict-mode exists with value false.
+    val autoConflictMode = jobContext.getConfiguration.get(
+      "spark.internal.io.hmrcp.auto-staging-conflict-mode")
+    if (autoConflictMode == null || autoConflictMode != "false") {
+      if (dynamicPartitionOverwrite) {
+        jobContext.getConfiguration.set("fs.s3a.committer.staging.conflict-mode", "replace")
+      } else {
+        jobContext.getConfiguration.set("fs.s3a.committer.staging.conflict-mode", "append")
+      }
+    }
+
     val taskAttemptContext = new TaskAttemptContextImpl(jobContext.getConfiguration, taskAttemptId)
     committer = setupCommitter(taskAttemptContext)
     committer.setupJob(jobContext)

hadoop-cloud/src/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala

@@ -17,8 +17,6 @@
 
 package org.apache.spark.internal.io.cloud
 
-import java.io.IOException
-
 import org.apache.hadoop.fs.{Path, StreamCapabilities}
 import org.apache.hadoop.mapreduce.TaskAttemptContext
 import org.apache.hadoop.mapreduce.lib.output.{FileOutputCommitter, PathOutputCommitter, PathOutputCommitterFactory}
@@ -61,6 +59,15 @@ class PathOutputCommitProtocol(
   extends HadoopMapReduceCommitProtocol(jobId, dest, dynamicPartitionOverwrite)
     with Serializable {
 
+  if (dynamicPartitionOverwrite) {
+    // until there's explicit extensions to the PathOutputCommitProtocols
+    // to support the spark mechanism, it's left to the individual committer
+    // choice to handle partitioning.
+    // throw new IOException(PathOutputCommitProtocol.UNSUPPORTED)
+    // The above exception is disabled with automatic value of fs.s3a.committer.staging.conflict-mode
+    // in HadoopMapReduceCommitProtocol.
+  }
+
   /** The committer created. */
   @transient private var committer: PathOutputCommitter = _