You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pt...@apache.org on 2026/05/05 15:33:04 UTC
(spark) branch branch-4.2 updated: [SPARK-56588][CORE] Fix `PathOutputCommitProtocol` dynamic partition overwrite
This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new a579aa9d4fed [SPARK-56588][CORE] Fix `PathOutputCommitProtocol` dynamic partition overwrite
a579aa9d4fed is described below
commit a579aa9d4feddf33a350923ce77d4955be583eb9
Author: Peter Toth <pe...@gmail.com>
AuthorDate: Tue May 5 17:28:23 2026 +0200
[SPARK-56588][CORE] Fix `PathOutputCommitProtocol` dynamic partition overwrite
### What changes were proposed in this pull request?
Fix `PathOutputCommitProtocol` to correctly handle `dynamicPartitionOverwrite=true` when the underlying committer is a `FileOutputCommitter`. Two bugs were introduced by #37468 ([SPARK-40034](https://issues.apache.org/jira/browse/SPARK-40034)):
1. `FileOutputCommitter` was not redirected to the staging directory (`stagingDir`) in `setupCommitter`. As a result, the committer wrote task output directly to the final destination path rather than staging it, so the `commitJob` rename-and-overwrite logic never ran and `INSERT OVERWRITE` would append instead of replacing existing partition directories.
2. `newTaskTempFile` did not populate `partitionPaths`. With no recorded partitions, `commitJob` skipped the delete-before-rename step entirely, again leaving old partition data in place.
The fix mirrors the approach already used in `SQLHadoopMapReduceCommitProtocol`:
- In `setupCommitter`, reinitialize `FileOutputCommitter` with `stagingDir` when `dynamicPartitionOverwrite=true` (matching `SQLHadoopMapReduceCommitProtocol`).
- In `newTaskTempFile`, track `partitionPaths += dir.get` when the committer is a `FileOutputCommitter` (matching the parent class `HadoopMapReduceCommitProtocol`; the guard is intentionally absent for cloud committers that handle dynamic partition overwrite natively via `StreamCapabilities`).
- Widen `partitionPaths` in `HadoopMapReduceCommitProtocol` from `private` to `protected` so the subclass can write to it.
### Why are the changes needed?
`INSERT OVERWRITE` on a partitioned table with `partitionOverwriteMode=dynamic` silently appends data instead of replacing the written partitions when `PathOutputCommitProtocol` is in use.
The bug was introduced by #37468 ([SPARK-40034](https://issues.apache.org/jira/browse/SPARK-40034)), which enabled `FileOutputCommitter`-backed dynamic partition overwrite in `PathOutputCommitProtocol` without wiring up the staging mechanism that `HadoopMapReduceCommitProtocol` and `SQLHadoopMapReduceCommitProtocol` rely on to delete old partition directories and atomically move staged output into place.
The problem became more visible after #32518 ([SPARK-35383](https://issues.apache.org/jira/browse/SPARK-35383)) changed `SparkContext` to activate `PathOutputCommitProtocol` whenever hadoop-cloud is on the classpath (via `fillMissingMagicCommitterConfsIfNeeded()`), not only when a magic-committer bucket is explicitly configured.
### Does this PR introduce _any_ user-facing change?
Yes. With `spark.sql.sources.partitionOverwriteMode=dynamic` and `PathOutputCommitProtocol` active (the default when hadoop-cloud is available), `INSERT OVERWRITE` now correctly replaces the written partition directories instead of appending to them.
### How was this patch tested?
Three unit tests added to `CommitterBindingSuite`:
- `SPARK-56588: FileOutputCommitter dynamic partition overwrite stages output and tracks partitions` — verifies the temp file path goes through `.spark-staging-` and that `partitionPaths` is populated for `FileOutputCommitter`.
- `SPARK-56588: Cloud committer with dynamic partition support does not track partitions in partitionPaths` — verifies that cloud committers implementing `CAPABILITY_DYNAMIC_PARTITIONING` via `StreamCapabilities` do not have their partitions tracked in Spark's `partitionPaths` (they manage overwrite themselves).
- `SPARK-56588: FileOutputCommitter without dynamicPartitionOverwrite does not track partitions` — baseline regression guard.
Manually verified with a Spark shell session (hadoop-cloud on the classpath):
```
➜ bin/spark-shell
scala> // Write initial data to two partitions
scala> spark.range(6).selectExpr("id % 2 as p", "id as v").write.partitionBy("p").mode("overwrite").parquet("/tmp/repro")
scala> // p=0: rows 0,2,4 p=1: rows 1,3,5
scala> spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
scala> // Overwrite only p=0
scala> spark.createDataFrame(Seq((0L, 99L))).toDF("p", "v").write.partitionBy("p").mode("overwrite").parquet("/tmp/repro")
scala> spark.read.parquet("/tmp/repro").orderBy("p", "v").show()
// Before fix: p=0 still shows rows 0,2,4,99 (appended, not replaced)
+---+---+
| v| p|
+---+---+
| 0| 0|
| 2| 0|
| 4| 0|
| 99| 0|
| 1| 1|
| 3| 1|
| 5| 1|
+---+---+
// After fix:
+---+---+
| v| p|
+---+---+
| 99| 0| <- p=0 correctly replaced
| 1| 1| <- p=1 untouched
| 3| 1|
| 5| 1|
+---+---+
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 4.6
Closes #55622 from peter-toth/SPARK-56588-fix-pathoutputcommitprotocol-dynamic-partition-overwrite.
Authored-by: Peter Toth <pe...@gmail.com>
Signed-off-by: Peter Toth <pe...@gmail.com>
(cherry picked from commit 4b95f2ac08d3b428675dc73793a20ddfce6b1f92)
Signed-off-by: Peter Toth <pe...@gmail.com>
---
.../io/HadoopMapReduceCommitProtocol.scala | 2 +-
.../io/cloud/PathOutputCommitProtocol.scala | 16 +++
.../internal/io/cloud/CommitterBindingSuite.scala | 117 ++++++++++++++++++++-
3 files changed, 133 insertions(+), 2 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala b/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
index d7bda5bbe721..3c045fcd9541 100644
--- a/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
+++ b/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
@@ -99,7 +99,7 @@ class HadoopMapReduceCommitProtocol(
* e.g. a=1/b=2. Files under these partitions will be saved into staging directory and moved to
* destination directory at the end, if `dynamicPartitionOverwrite` is true.
*/
- @transient private var partitionPaths: mutable.Set[String] = null
+ @transient protected var partitionPaths: mutable.Set[String] = null
/**
* The staging directory of this write job. Spark uses it to deal with files with absolute output
diff --git a/hadoop-cloud/src/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala b/hadoop-cloud/src/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala
index 44a521bd636c..ae4f5660faca 100644
--- a/hadoop-cloud/src/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala
+++ b/hadoop-cloud/src/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala
@@ -115,6 +115,17 @@ class PathOutputCommitProtocol(
// failures. Warn
logTrace(s"Committer $committer may not be tolerant of task commit failures")
}
+
+ if (dynamicPartitionOverwrite) {
+ // FileOutputCommitter must be initialized with the staging directory so that task output
+ // lands under stagingDir/_temporary/... and commitJob can later delete the old partition
+ // directories and move staged files to final dest. Without this, the committer writes
+ // directly to the final path and the dynamic-overwrite cleanup in commitJob never sees any
+ // partitionPaths.
+ val ctor =
+ committer.getClass.getDeclaredConstructor(classOf[Path], classOf[TaskAttemptContext])
+ committer = ctor.newInstance(stagingDir, context)
+ }
} else {
// if required other committers need to be checked for dynamic partition
// compatibility through a StreamCapabilities probe.
@@ -161,6 +172,11 @@ class PathOutputCommitProtocol(
}.getOrElse(workDir)
val file = new Path(parent, getFilename(taskContext, spec))
logTrace(s"Creating task file $file for dir $dir and spec $spec")
+ if (dynamicPartitionOverwrite && committer.isInstanceOf[FileOutputCommitter]) {
+ assert(dir.isDefined,
+ "The dataset to be written must be partitioned when dynamicPartitionOverwrite is true.")
+ partitionPaths += dir.get
+ }
file.toString
}
diff --git a/hadoop-cloud/src/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala b/hadoop-cloud/src/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala
index ee89adab94fb..954d9011f3bb 100644
--- a/hadoop-cloud/src/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala
+++ b/hadoop-cloud/src/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala
@@ -30,6 +30,19 @@ import org.apache.spark.internal.io.{FileCommitProtocol, FileNameSpec}
import org.apache.spark.internal.io.cloud.PathOutputCommitProtocol.{CAPABILITY_DYNAMIC_PARTITIONING, OUTPUTCOMMITTER_FACTORY_SCHEME}
import org.apache.spark.network.util.JavaUtils
+/**
+ * Subclass that exposes the protected `partitionPaths` field so tests can
+ * assert on it without going through the full `commitTask` path (which
+ * requires `SparkEnv`).
+ */
+private class PathOutputCommitProtocolForTest(
+ jobId: String,
+ dest: String,
+ dynamicPartitionOverwrite: Boolean)
+ extends PathOutputCommitProtocol(jobId, dest, dynamicPartitionOverwrite) {
+ def capturedPartitionPaths: Set[String] = partitionPaths.toSet
+}
+
class CommitterBindingSuite extends SparkFunSuite {
private val jobId = "2007071202143_0101"
@@ -264,5 +277,107 @@ class CommitterBindingSuite extends SparkFunSuite {
"org.apache.hadoop.mapreduce.lib.output.FileOutputCommitterFactory")
}
-}
+ /**
+ * With dynamicPartitionOverwrite=true and a FileOutputCommitter, newTaskTempFile must route
+ * output through the staging directory (not the final output path) and must record the partition
+ * in partitionPaths so that commitJob can delete the old partition directory and rename the
+ * staged one into place.
+ */
+ test("SPARK-56588: FileOutputCommitter dynamic partition overwrite stages output and tracks " +
+ "partitions") {
+ val jobCommitDir = File.createTempFile("dyn-part-overwrite-staging", "")
+ try {
+ jobCommitDir.delete()
+ val jobUri = jobCommitDir.toURI
+ val path = new Path(jobUri)
+ val job = newJob(path)
+ val conf = job.getConfiguration
+ conf.set(MRJobConfig.TASK_ATTEMPT_ID, taskAttempt0)
+ conf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID, 1)
+ bindToFileOutputCommitterFactory(conf, "file")
+ val tContext = new TaskAttemptContextImpl(conf, taskAttemptId0)
+ val committer = new PathOutputCommitProtocolForTest(jobId, jobUri.toString, true)
+ committer.setupJob(tContext)
+ committer.setupTask(tContext)
+
+ val spec = FileNameSpec("", ".parquet")
+ val partition = "a=1/b=2"
+ val tempPath = committer.newTaskTempFile(tContext, Some(partition), spec)
+
+ // The temp file must be under the staging directory, not the final output path.
+ assert(tempPath.contains(".spark-staging-"),
+ s"Expected temp path under staging dir, got: $tempPath")
+ assert(!tempPath.startsWith(path.toUri.toString.stripSuffix("/") + "/" + partition),
+ s"Temp path must not point directly to the final output location: $tempPath")
+
+ // The partition must have been recorded so commitJob can overwrite it.
+ assert(committer.capturedPartitionPaths === Set(partition),
+ s"Expected partitionPaths = {$partition}, got: ${committer.capturedPartitionPaths}")
+ } finally {
+ jobCommitDir.delete()
+ }
+ }
+
+ /**
+ * A cloud committer that handles dynamic partitioning natively (via StreamCapabilities) must NOT
+ * have its partitions tracked in Spark's partitionPaths set: the committer takes care of
+ * overwriting itself, and the commitJob rename loop must not interfere.
+ */
+ test("SPARK-56588: Cloud committer with dynamic partition support does not track partitions in " +
+ "partitionPaths") {
+ val path = new Path("http://example/data")
+ val job = newJob(path)
+ val conf = job.getConfiguration
+ conf.set(MRJobConfig.TASK_ATTEMPT_ID, taskAttempt0)
+ conf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID, 1)
+ StubPathOutputCommitterBinding.bindWithDynamicPartitioning(conf, "http")
+ val tContext = new TaskAttemptContextImpl(conf, taskAttemptId0)
+ val committer = new PathOutputCommitProtocolForTest(jobId, path.toUri.toString, true)
+ committer.setupJob(tContext)
+ committer.setupTask(tContext)
+
+ val tempPath = committer.newTaskTempFile(tContext, Some("a=1"), FileNameSpec("", ".parquet"))
+
+ // The temp file must be under the committer's own work dir (path/_temporary),
+ // not written directly to the final output location.
+ val expectedWorkDir = path.toUri.toString.stripSuffix("/") + "/_temporary"
+ assert(tempPath.startsWith(expectedWorkDir),
+ s"Expected temp path under committer work dir ($expectedWorkDir), got: $tempPath")
+ assert(committer.capturedPartitionPaths.isEmpty,
+ s"partitionPaths must stay empty for cloud committers that handle " +
+ s"dynamic partition overwrite natively, " +
+ s"got: ${committer.capturedPartitionPaths}")
+ }
+
+ /**
+ * Without dynamicPartitionOverwrite, partitionPaths must remain empty even for
+ * FileOutputCommitter (baseline: existing behaviour must not regress).
+ */
+ test("SPARK-56588: FileOutputCommitter without dynamicPartitionOverwrite does not track " +
+ "partitions") {
+ val jobCommitDir = File.createTempFile("no-dyn-part-overwrite", "")
+ try {
+ jobCommitDir.delete()
+ val jobUri = jobCommitDir.toURI
+ val path = new Path(jobUri)
+ val job = newJob(path)
+ val conf = job.getConfiguration
+ conf.set(MRJobConfig.TASK_ATTEMPT_ID, taskAttempt0)
+ conf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID, 1)
+ bindToFileOutputCommitterFactory(conf, "file")
+ val tContext = new TaskAttemptContextImpl(conf, taskAttemptId0)
+ val committer = new PathOutputCommitProtocolForTest(jobId, jobUri.toString, false)
+ committer.setupJob(tContext)
+ committer.setupTask(tContext)
+
+ committer.newTaskTempFile(tContext, Some("a=1"), FileNameSpec("", ".parquet"))
+
+ assert(committer.capturedPartitionPaths.isEmpty,
+ s"partitionPaths must be empty when dynamicPartitionOverwrite=false, " +
+ s"got: ${committer.capturedPartitionPaths}")
+ } finally {
+ jobCommitDir.delete()
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org