(spark) branch branch-4.x updated: [SPARK-60023][SQL] Set writeJobUUID before setupJob in FileFormatWriter

[ch...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pan3793 pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.x by this push:
     new 19c0e29c89ad [SPARK-60023][SQL] Set writeJobUUID before setupJob in FileFormatWriter
19c0e29c89ad is described below

commit 19c0e29c89ad3d0a7e7581ecf3d7eb9bddde6c4c
Author: Yahya Kisana <yk...@protonmail.com>
AuthorDate: Mon Oct 19 11:33:36 2026 +0800

    [SPARK-60023][SQL] Set writeJobUUID before setupJob in FileFormatWriter
    
    ### What changes were proposed in this pull request?
    
    In `FileFormatWriter.write`, set `spark.sql.sources.writeJobUUID` on the job configuration before `committer.setupJob(job)` instead of after `materializeAdaptiveSparkPlan`, inside the `try` block.
    
    SPARK-56919 moved `committer.setupJob(job)` ahead of `materializeAdaptiveSparkPlan(plan)` so that an AQE failure calls `committer.abortJob(job)` instead of losing the output path. The `writeJobUUID` assignment stayed where it was, so committers now run `setupJob` (and `abortJob`, if materialization fails) before the UUID is set. This PR restores the pre-SPARK-56919 order: set the UUID, then call `setupJob`. `SparkHadoopWriter` uses the same order for the RDD write path. `description.u [...]
    
    ### Why are the changes needed?
    
    Committers that read `spark.sql.sources.writeJobUUID` in `setupJob` or `abortJob` fall back to `jobContext.getJobID()` when it is not set. That job ID is null for the `Job` that `FileFormatWriter` creates, because the job is never submitted. Hadoop's `ManifestCommitter`, the default committer for `gs://` and `abfs://` since Hadoop 3.4, fails every file source write with:
    
    ```
    java.lang.NullPointerException: Cannot invoke "org.apache.hadoop.mapreduce.JobID.toString()" because "jobId" is null
      at org.apache.hadoop.mapreduce.lib.output.committer.manifest.impl.ManifestCommitterSupport.buildJobUUID(ManifestCommitterSupport.java:118)
      at org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterConfig.<init>(ManifestCommitterConfig.java:190)
      at org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitter.enterCommitter(ManifestCommitter.java:181)
      at org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitter.setupJob(ManifestCommitter.java:196)
      at org.apache.spark.internal.io.HadoopMapReduceCommitProtocol.setupJob(HadoopMapReduceCommitProtocol.scala:198)
      at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:165)
    ```
    
    ### Does this PR introduce _any_ user-facing change?
    
    Yes, a bug fix. SPARK-56919 shipped in Spark 4.2.0, and since then file source writes (`DataFrameWriter`, `INSERT`, Hive `INSERT` and the file streaming sink) fail with the `NullPointerException` above when the output committer reads `spark.sql.sources.writeJobUUID` during job setup, e.g. `ManifestCommitter`. After this change, these writes succeed as they did before 4.2.0.
    
    ### How was this patch tested?
    
    Added a test, `SPARK-60023: writeJobUUID should be set before setupJob`, to `FileFormatWriterSuite`. It sets `spark.sql.sources.commitProtocolClass` to a test `SQLHadoopMapReduceCommitProtocol` subclass whose `setupJob` asserts that `spark.sql.sources.writeJobUUID` is set, then writes and reads back a parquet file.
    
    ```
    build/sbt 'sql/testOnly *FileFormatWriterSuite -- -z "SPARK-60023"'
    ```
    
    - With this change, both the codegen and interpreted variants pass.
    - Without the change in `FileFormatWriter.scala`, both fail with `assertion failed: spark.sql.sources.writeJobUUID should be set before setupJob`.
    
    `build/sbt sql/scalastyle sql/Test/scalastyle` also passes.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated by: Claude Opus 5.5
    Fix written by and all code manually reviewed by ykisana.
    Tests, PR description, and comments by Claude.
    
    Closes #59333 from ykisana/SPARK-60023.
    
    Authored-by: Yahya Kisana <yk...@protonmail.com>
    Signed-off-by: Cheng Pan <ch...@apache.org>
    (cherry picked from commit 0ac411160c03b761fabdc94893965f6225b87e72)
    Signed-off-by: Cheng Pan <ch...@apache.org>
---
 .../execution/datasources/FileFormatWriter.scala   |  8 +++---
 .../datasources/FileFormatWriterSuite.scala        | 30 ++++++++++++++++++++++
 2 files changed, 34 insertions(+), 4 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
index 50d5dd28ab74..5932c986ae3c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
@@ -148,6 +148,10 @@ object FileFormatWriter extends Logging {
         writerBucketSpec.map(_.bucketIdExpression) ++ sortColumns
     val writeFilesOpt = V1WritesUtils.getWriteFilesOpt(plan)
 
+    // propagate the description UUID into the jobs, so that committers
+    // get an ID guaranteed to be unique.
+    job.getConfiguration.set("spark.sql.sources.writeJobUUID", description.uuid)
+
     // SPARK-56919: setupJob must run before materializeAdaptiveSparkPlan, which can throw.
     // Otherwise INSERT OVERWRITE permanently loses the table path if AQE fails.
     // setupJob is outside the try below because it only initializes the job; the try/catch
@@ -173,10 +177,6 @@ object FileFormatWriter extends Logging {
 
       SQLExecution.checkSQLExecutionId(sparkSession)
 
-      // propagate the description UUID into the jobs, so that committers
-      // get an ID guaranteed to be unique.
-      job.getConfiguration.set("spark.sql.sources.writeJobUUID", description.uuid)
-
       // When `PLANNED_WRITE_ENABLED` is true, the optimizer rule V1Writes will add logical sort
       // operator based on the required ordering of the V1 write command. So the output
       // ordering of the physical plan should always match the required ordering. Here
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileFormatWriterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileFormatWriterSuite.scala
index 517c929cf193..5d0ef7c7e207 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileFormatWriterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileFormatWriterSuite.scala
@@ -17,8 +17,11 @@
 
 package org.apache.spark.sql.execution.datasources
 
+import org.apache.hadoop.mapreduce.JobContext
+
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.plans.CodegenInterpretedPlanTest
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
 
 class FileFormatWriterSuite
@@ -76,4 +79,31 @@ class FileFormatWriterSuite
       }
     }
   }
+
+  test("SPARK-60023: writeJobUUID should be set before setupJob") {
+    withSQLConf(SQLConf.FILE_COMMIT_PROTOCOL_CLASS.key ->
+        classOf[WriteJobUUIDCheckingCommitProtocol].getName) {
+      withTempPath { path =>
+        // WriteJobUUIDCheckingCommitProtocol fails setupJob if the UUID is not set yet.
+        spark.range(10).write.parquet(path.getCanonicalPath)
+        checkAnswer(spark.read.parquet(path.getCanonicalPath), spark.range(10).toDF())
+      }
+    }
+  }
+}
+
+/**
+ * A file commit protocol that fails in setupJob if "spark.sql.sources.writeJobUUID" is not set,
+ * like committers (e.g. Hadoop's ManifestCommitter) that read it to build the job UUID.
+ */
+private class WriteJobUUIDCheckingCommitProtocol(
+    jobId: String,
+    path: String,
+    dynamicPartitionOverwrite: Boolean)
+  extends SQLHadoopMapReduceCommitProtocol(jobId, path, dynamicPartitionOverwrite) {
+  override def setupJob(jobContext: JobContext): Unit = {
+    assert(jobContext.getConfiguration.get("spark.sql.sources.writeJobUUID") != null,
+      "spark.sql.sources.writeJobUUID should be set before setupJob")
+    super.setupJob(jobContext)
+  }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org