fix style

Author: turboFei

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

@@ -149,21 +149,23 @@ object FileCommitProtocol extends Logging {
       outputPath: String,
       dynamicPartitionOverwrite: Boolean = false,
       isPartitionOverwrite: Boolean = false,
-      staticPartitionKVS: Seq[(String, String)] = Seq.empty[(String, String)]):
+      staticPartitionKVs: Seq[(String, String)] = Seq.empty[(String, String)]):
   FileCommitProtocol = {
 
     logDebug(s"Creating committer $className; job $jobId; output=$outputPath;" +
-      s" dynamic=$dynamicPartitionOverwrite")
+      s" dynamic=$dynamicPartitionOverwrite; isPartitionOverWrite=$isPartitionOverwrite;" +
+      s" staticPartitionKVS=$staticPartitionKVs")
     val clazz = Utils.classForName[FileCommitProtocol](className)
     // First try the constructor with arguments (jobId: String, outputPath: String,
-    // dynamicPartitionOverwrite: Boolean).
+    // dynamicPartitionOverwrite: Boolean, isPartitionOverwrite: Boolean,
+    // staticPartitionKVs: Seq[(String, String)]).
     // If that doesn't exist, try the one with (jobId: string, outputPath: String).
     try {
       val ctor = clazz.getDeclaredConstructor(classOf[String], classOf[String], classOf[Boolean],
         classOf[Boolean], classOf[Seq[(String, String)]])
       logDebug("Using (String, String, Boolean, Boolean, Seq[(String, String)]) constructor")
       ctor.newInstance(jobId, outputPath, dynamicPartitionOverwrite.asInstanceOf[java.lang.Boolean],
-        isPartitionOverwrite.asInstanceOf[java.lang.Boolean], staticPartitionKVS)
+        isPartitionOverwrite.asInstanceOf[java.lang.Boolean], staticPartitionKVs)
     } catch {
       case _: NoSuchMethodException =>
         logDebug("Falling back to (String, String) constructor")

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

@@ -55,7 +55,7 @@ class HadoopMapReduceCommitProtocol(
     path: String,
     dynamicPartitionOverwrite: Boolean = false,
     isPartitionOverwrite: Boolean = false,
-    staticPartitionKVS: Seq[(String, String)] = Seq.empty[(String, String)])
+    staticPartitionKVs: Seq[(String, String)] = Seq.empty[(String, String)])
   extends FileCommitProtocol with Serializable with Logging {
 
   import FileCommitProtocol._
@@ -112,7 +112,7 @@ class HadoopMapReduceCommitProtocol(
    * Get the determinable base path of results according to specified partition key-value pairs.
    */
   private def getStaticPartitionPath(): String = {
-    staticPartitionKVS.map{kv =>
+    staticPartitionKVs.map{kv =>
       escapePathName(kv._1) + "=" + escapePathName(kv._2)
     }.mkString(File.separator)
   }

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

@@ -117,7 +117,7 @@ case class InsertIntoHadoopFsRelationCommand(
       outputPath = outputPath.toString,
       dynamicPartitionOverwrite = dynamicPartitionOverwrite,
       isPartitionOverwrite = isPartitionOverwrite,
-      staticPartitionKVS = staticPartitionKVs)
+      staticPartitionKVs = staticPartitionKVs)
 
     val doInsertion = (mode, pathExists) match {
       case (SaveMode.ErrorIfExists, true) =>

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

@@ -34,9 +34,9 @@ class SQLHadoopMapReduceCommitProtocol(
     path: String,
     dynamicPartitionOverwrite: Boolean = false,
     isPartitionOverwrite: Boolean = false,
-    staticPartitionKVS: Seq[(String, String)] = Seq.empty[(String, String)])
+    staticPartitionKVs: Seq[(String, String)] = Seq.empty[(String, String)])
   extends HadoopMapReduceCommitProtocol(jobId, path, dynamicPartitionOverwrite,
-    isPartitionOverwrite, staticPartitionKVS)
+    isPartitionOverwrite, staticPartitionKVs)
     with Serializable with Logging {
 
   override protected def setupCommitter(context: TaskAttemptContext): OutputCommitter = {

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -3194,9 +3194,9 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession {
     }
   }
 
-  test("SPARK-29037: For non dynamic partition overwrite, set a unique staging dir") {
+  test("SPARK-29037: For static partition overwrite, set a unique staging dir") {
     withSQLConf(PARTITION_OVERWRITE_MODE.key -> PartitionOverwriteMode.STATIC.toString) {
-      withTable("ta", "tb") {
+      withTable("ta") {
         sql("create table ta(id int, p1 int, p2 int) using parquet partitioned by (p1, p2)")
         sql("insert overwrite table ta partition(p1=1,p2) select 1, 3")
         val df1 = sql("select * from ta order by p2")
@@ -3213,18 +3213,6 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession {
         sql("insert overwrite table ta select 9, 9, 9")
         val df5 = sql("select * from ta order by p2")
         checkAnswer(df5, Array(Row(9, 9, 9)))
-
-        // For non-partitioned table write.
-        sql("create table tb(id int, p1 int, p2 int) using parquet")
-        sql("insert overwrite table tb select 1, 2, 3")
-        val df6 = sql("select * from tb order by p2")
-        checkAnswer(df6, Array(Row(1, 2, 3)))
-        sql("insert overwrite table tb select 1, 2, 4")
-        val df7 = sql("select * from tb order by p2")
-        checkAnswer(df7, Array(Row(1, 2, 4)))
-        sql("insert into table tb select 1, 2, 5")
-        val df8 = sql("select * from tb order by p2")
-        checkAnswer(df8, Array(Row(1, 2, 4), Row(1, 2, 5)))
       }
     }
   }