You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/10/20 14:47:05 UTC
(spark) branch branch-4.2 updated: [SPARK-60366][CORE] Fix dynamic partition overwrite to replace partitions with non-`FileOutputCommitter` committers
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 ad92aad215dd [SPARK-60366][CORE] Fix dynamic partition overwrite to replace partitions with non-`FileOutputCommitter` committers
ad92aad215dd is described below
commit ad92aad215dda313f33efaecd3892920b36c519e
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Tue Oct 20 07:46:04 2026 -0700
[SPARK-60366][CORE] Fix dynamic partition overwrite to replace partitions with non-`FileOutputCommitter` committers
### What changes were proposed in this pull request?
This PR fixes dynamic partition overwrite when the Hadoop output committer is not a `FileOutputCommitter`. Since Hadoop 3.3.5, `mapred-default.xml` binds `abfs://` and `gs://` to the manifest committer (`ManifestCommitter`), which is not a `FileOutputCommitter`. Apache Spark 4.0.0+ bundles Hadoop 3.4.1+, so this is the default committer for these stores.
Dynamic partition overwrite is implemented in `HadoopMapReduceCommitProtocol.commitJob`: for each partition recorded in `partitionPaths`, it deletes the partition directory in the destination and renames the staged one from `.spark-staging-{jobId}` into its place. Two code paths did not record the partitions for committers other than `FileOutputCommitter`, so `commitJob` deleted nothing and the job appended the new files to the old ones.
1. `HadoopMapReduceCommitProtocol` (used by the default `SQLHadoopMapReduceCommitProtocol`): `newTaskTempFile` wrote the files of such committers directly to the final output path and recorded `partitionPaths` only for `FileOutputCommitter`. This PR records the partitions for every committer, and writes the files of committers other than `FileOutputCommitter` to the staging directory. For these committers, this restores the behavior before SPARK-27194/SPARK-29302 moved the partition t [...]
- #29000
2. `PathOutputCommitProtocol` (hadoop-cloud): the committer was bound to the final destination, and `newTaskTempFile` recorded the partitions only for `FileOutputCommitter` (SPARK-56588), assuming that committers declaring `mapreduce.job.committer.dynamic.partitioning` replace the partitions themselves. They do not. The [Spark Dynamic Partition overwriting](https://github.com/apache/hadoop/blob/rel/release-3.5.0/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client- [...]
- #55622
3. For committers other than `FileOutputCommitter`, `HadoopMapReduceCommitProtocol.commitJob` now checks that the staged partition directories exist before it deletes any destination partition. Such a committer may delete its output directory, which is the staging directory, on job commit. For example, the S3A directory committer does so in the `replace` conflict mode. Then the job fails without changing any destination partition. This adds one existence check per written partition fo [...]
### Why are the changes needed?
`INSERT OVERWRITE` with dynamic partitions silently appends instead of replacing the written partitions on `abfs://` and `gs://`. For example, with the default commit protocol and an ORC, JSON, CSV or text table, the following reproduces it on the local file system by binding the manifest committer to the `file` scheme, as the new `PartitionedWriteSuite` test does:
```sql
SET mapreduce.outputcommitter.factory.scheme.file=org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterFactory;
SET spark.sql.sources.partitionOverwriteMode=dynamic;
CREATE TABLE t(c1 INT, p1 INT) USING json PARTITIONED BY (p1);
INSERT INTO t VALUES (1, 1), (2, 2);
INSERT OVERWRITE TABLE t PARTITION(p1) SELECT 3, 1;
SELECT * FROM t; -- [1,1], [2,2], [3,1] instead of [2,2], [3,1]
```
On these stores, converted Hive ORC tables are affected even with the default configuration, because `HiveMetastoreCatalog` always uses the dynamic mode for them: `INSERT OVERWRITE TABLE t PARTITION (p1) SELECT ...` on a partitioned `STORED AS ORC` table keeps the old files of the written partitions.
Affected cases:
- Default commit protocol: ORC, JSON, CSV, text and other formats that do not set their own committer, on `abfs://` and `gs://`, in Spark 4.0.0 ([v4.0.0](https://github.com/apache/spark/releases/tag/v4.0.0), 2025-05-19) and later. Parquet is not affected because `ParquetOutputCommitter` is a `FileOutputCommitter`. `abfss://` is not bound to the manifest committer by default. Earlier releases since 3.1.1 ([v3.1.1](https://github.com/apache/spark/releases/tag/v3.1.1), 2021-02-22) are af [...]
- `PathOutputCommitProtocol`: all formats, including Parquet, since Spark 3.5.0 ([v3.5.0](https://github.com/apache/spark/releases/tag/v3.5.0), 2023-09-09, SPARK-40034) when it is configured and runs with Hadoop 3.3.5+. Since Spark 4.1.0 ([v4.1.0](https://github.com/apache/spark/releases/tag/v4.1.0), 2025-12-11, SPARK-47618), it is the default whenever `spark-hadoop-cloud` is on the classpath.
- In Spark 4.2.0 ([v4.2.0](https://github.com/apache/spark/releases/tag/v4.2.0), 2026-07-11), these writes fail with a `NullPointerException` in `ManifestCommitter.setupJob` (SPARK-56919). SPARK-60023 fixes that NPE in `master`, `branch-4.x`, `branch-4.3` and `branch-4.2`. Without this PR, the dynamic partition overwrites that fail in 4.2.0 silently append in 4.2.1 and 4.3.0.
- #59333
### Does this PR introduce _any_ user-facing change?
Yes. With a committer that is not a `FileOutputCommitter`, such as the manifest committer used by default for `abfs://` and `gs://`, dynamic partition overwrite now replaces the written partitions instead of appending to them, as it does with `FileOutputCommitter`.
As with `FileOutputCommitter`, the job commit now renames each written partition directory from the staging directory into the destination. This is a copy of the partition's files on stores without directory rename, such as GCS and S3. With `PathOutputCommitProtocol`, the committer's `_SUCCESS` job summary is written to the staging directory, which is deleted at job commit, so a dynamic partition overwrite no longer writes `_SUCCESS` to the table directory, and an existing `_SUCCESS` [...]
The S3A committers do not declare `mapreduce.job.committer.dynamic.partitioning`, so `PathOutputCommitProtocol` still rejects dynamic partition overwrite with them, for example with the magic committer that Spark uses for `s3a://` when `spark-hadoop-cloud` is on the classpath. `HadoopMapReduceCommitProtocol` does not check this capability: with `fs.s3a.committer.name=magic`, a dynamic partition overwrite of ORC, JSON, CSV or text data now replaces the partitions by renaming them on S3 [...]
### How was this patch tested?
New tests. The first three bind the real `ManifestCommitterFactory` of `hadoop-client-api` to the `file` scheme with `mapreduce.outputcommitter.factory.scheme.file`, and check that the manifest committer is used, from its non-empty `_SUCCESS` JSON summary.
- `PartitionedWriteSuite`: `SPARK-60366: Dynamic partition overwrite with a committer that is not a FileOutputCommitter`, a dynamic `INSERT OVERWRITE` into JSON and ORC tables with the default commit protocol.
- `CommitterBindingSuite`: `SPARK-60366: Dynamic partition overwrite through the manifest committer`, a dynamic partition overwrite of JSON and Parquet data with `PathOutputCommitProtocol`.
- Hive `InsertSuite`: `SPARK-60366: Dynamic partition overwrite of a converted Hive ORC table with a committer that is not a FileOutputCommitter`, a dynamic `INSERT OVERWRITE` into a partitioned `STORED AS ORC` table with the default settings.
- `CommitterBindingSuite`: `SPARK-60366: Cloud committer with dynamic partition support stages output and tracks partitions` replaces the SPARK-56588 test that asserted that the partitions of such committers are not tracked and that their task output is under the destination.
- `PartitionedWriteSuite`: `SPARK-60366: Dynamic partition overwrite keeps the existing partitions if the committer deletes the staged files`, with a committer that deletes its output directory on job commit, like the S3A directory committer in the `replace` conflict mode. It checks that the overwrite fails and keeps the existing rows.
Without the fix, the new tests fail, for example:
```
- SPARK-60366: Dynamic partition overwrite with a committer that is not a FileOutputCommitter *** FAILED ***
!== Correct Answer - 2 == == Spark Answer - 3 ==
![2,2] [1,1]
![3,1] [2,2]
! [3,1]
- SPARK-60366: Cloud committer with dynamic partition support stages output and tracks partitions *** FAILED ***
"http://example/data/_temporary/a=1/part-00000-2007071202143_0101.parquet" did not start with "http://example/data/.spark-staging-2007071202143_0101/_temporary"
- SPARK-60366: Dynamic partition overwrite through the manifest committer *** FAILED ***
ArraySeq((1, 1), (2, 2), (3, 1)) did not equal List((2, 2), (3, 1))
```
The Hive test fails in the same way as the first one. Without the existence check in `commitJob`, the last test fails because the overwrite deletes `p1=1`:
```
- SPARK-60366: Dynamic partition overwrite keeps the existing partitions if the committer deletes the staged files *** FAILED ***
!== Correct Answer - 2 == == Spark Answer - 1 ==
![1,1] [2,2]
![2,2]
```
In addition, temporary tests (not included in this PR) with the manifest committer on the local file system showed the following, before and after the fix:
| Case | Before | After |
|---|---|---|
| Default protocol, ORC/JSON/CSV: one dynamic column, a static prefix with a custom partition location, and two-level partitions through the DataFrame API | appends | replaces |
| Default protocol, Parquet | replaces | replaces |
| `PathOutputCommitProtocol`, JSON/Parquet/ORC/CSV, with and without a custom partition location | appends | replaces |
| `PathOutputCommitProtocol` with `FileOutputCommitter` | replaces | replaces |
| Converted Hive ORC table, default settings | appends | replaces |
| Hive ORC table with `spark.sql.hive.convertMetastoreOrc=false`, converted Hive Parquet table | replaces | replaces |
Existing tests also pass:
```
build/sbt "core/testOnly *FileCommitProtocolInstantiationSuite *PairRDDFunctionsSuite *OutputCommitCoordinatorSuite *OutputCommitCoordinatorIntegrationSuite"
build/sbt -Phadoop-cloud "hadoop-cloud/test"
build/sbt "sql/testOnly org.apache.spark.sql.sources.InsertSuite *PartitionedWriteSuite *FileFormatWriterSuite *DataFrameReaderWriterSuite *ParquetIOSuite *SQLMetricsSuite"
build/sbt -Phive "hive/testOnly org.apache.spark.sql.hive.InsertSuite *CommitFailureTestRelationSuite *JsonHadoopFsRelationSuite *ParquetHadoopFsRelationSuite *SimpleTextHadoopFsRelationSuite *OrcHadoopFsRelationSuite *HiveOrcQuerySuite *PartitionProviderCompatibilitySuite"
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59614 from dongjoon-hyun/SPARK-60366.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit ea33f465361ba7b0099e36ae553cbdc57a9d1abc)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../io/HadoopMapReduceCommitProtocol.scala | 32 ++++++++--
.../io/cloud/PathOutputCommitProtocol.scala | 27 ++++----
.../internal/io/cloud/CommitterBindingSuite.scala | 54 ++++++++++++----
.../spark/sql/sources/PartitionedWriteSuite.scala | 74 +++++++++++++++++++++-
.../org/apache/spark/sql/hive/InsertSuite.scala | 22 +++++++
5 files changed, 174 insertions(+), 35 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 3c045fcd9541..380e0ebc8578 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
@@ -59,6 +59,9 @@ import org.apache.spark.mapred.SparkHadoopMapRedUtil
* 2. When [[FileOutputCommitter]] algorithm version set to 2,
* committing tasks directly move task attempt output files to
* /path/to/outputPath/.spark-staging-{jobId}/a=1/b=1.
+ * 3. When the committer is not a [[FileOutputCommitter]], task
+ * attempts directly write files to
+ * /path/to/outputPath/.spark-staging-{jobId}/a=1/b=1.
*
* At the end of committing job, we move output files from
* intermediate path to final path, e.g., move files from
@@ -121,15 +124,18 @@ class HadoopMapReduceCommitProtocol(
taskContext: TaskAttemptContext, dir: Option[String], spec: FileNameSpec): String = {
val filename = getFilename(taskContext, spec)
+ if (dynamicPartitionOverwrite) {
+ assert(dir.isDefined,
+ "The dataset to be written must be partitioned when dynamicPartitionOverwrite is true.")
+ partitionPaths += dir.get
+ }
val stagingDir: Path = committer match {
// For FileOutputCommitter it has its own staging path called "work path".
case f: FileOutputCommitter =>
- if (dynamicPartitionOverwrite) {
- assert(dir.isDefined,
- "The dataset to be written must be partitioned when dynamicPartitionOverwrite is true.")
- partitionPaths += dir.get
- }
new Path(Option(f.getWorkPath).map(_.toString).getOrElse(path))
+ // Other committers, e.g. Hadoop's ManifestCommitter, do not commit the files written here,
+ // so write them to the staging directory, from which commitJob moves them into place.
+ case _ if dynamicPartitionOverwrite => this.stagingDir
case _ => new Path(path)
}
@@ -187,6 +193,21 @@ class HadoopMapReduceCommitProtocol(
val (allAbsPathFiles, allPartitionPaths) =
taskCommits.map(_.obj.asInstanceOf[(Map[String, String], Set[String])]).unzip
val fs = stagingDir.getFileSystem(jobContext.getConfiguration)
+ val partitionPaths = allPartitionPaths.foldLeft(Set[String]())(_ ++ _)
+
+ if (dynamicPartitionOverwrite && !committer.isInstanceOf[FileOutputCommitter]) {
+ // Other committers may delete their output directory, which is the staging directory
+ // here, on job commit, e.g. S3A's directory committer in the replace conflict mode.
+ // Check the staged partitions before deleting any destination partition.
+ for (part <- partitionPaths) {
+ val stagingPartPath = new Path(stagingDir, part)
+ if (!fs.exists(stagingPartPath)) {
+ throw new IOException(s"Failed to overwrite dynamic partitions because " +
+ s"$stagingPartPath does not exist after the job commit of " +
+ s"${committer.getClass.getName}. No destination partition was changed.")
+ }
+ }
+ }
val filesToMove = allAbsPathFiles.foldLeft(Map[String, String]())(_ ++ _)
logDebug(s"Committing files staged for absolute locations $filesToMove")
@@ -205,7 +226,6 @@ class HadoopMapReduceCommitProtocol(
}
if (dynamicPartitionOverwrite) {
- val partitionPaths = allPartitionPaths.foldLeft(Set[String]())(_ ++ _)
logDebug(s"Clean up default partition directories for overwriting: $partitionPaths")
for (part <- partitionPaths) {
val finalPartPath = new Path(path, part)
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 ae4f5660faca..8a2870517e0e 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
@@ -48,6 +48,8 @@ import org.apache.spark.internal.io.HadoopMapReduceCommitProtocol
* That feature is available on Hadoop releases with the Intermediate
* Manifest Committer for GCS and ABFS; it is not supported by the
* S3A committers.
+ * The committer is then bound to the staging directory, and the job commit
+ * replaces the written partitions of the destination with the staged ones.
* @constructor Instantiate.
* @param jobId job
* @param dest destination
@@ -86,8 +88,12 @@ class PathOutputCommitProtocol(
* `PathOutputCommitter`.
*/
override protected def setupCommitter(context: TaskAttemptContext): PathOutputCommitter = {
- logTrace(s"Setting up committer for path $destination")
- committer = PathOutputCommitterFactory.createCommitter(destPath, context)
+ // With dynamic partition overwrite, the committer must commit the task output to the staging
+ // directory, so that commitJob can delete the old partition directories and move the staged
+ // ones into place. Committers do not delete the old partition directories themselves.
+ val committerOutputPath = if (dynamicPartitionOverwrite) stagingDir else destPath
+ logTrace(s"Setting up committer for path $committerOutputPath")
+ committer = PathOutputCommitterFactory.createCommitter(committerOutputPath, context)
// Special feature to force out the FileOutputCommitter, so as to guarantee
// that the binding is working properly.
@@ -97,10 +103,10 @@ class PathOutputCommitProtocol(
// the output format returned a file output format committer, which
// is exactly what we do not want. So switch back to the factory.
val factory = PathOutputCommitterFactory.getCommitterFactory(
- destPath,
+ committerOutputPath,
context.getConfiguration)
logTrace(s"Using committer factory $factory")
- committer = factory.createOutputCommitter(destPath, context)
+ committer = factory.createOutputCommitter(committerOutputPath, context)
}
logTrace(s"Using committer ${committer.getClass}")
@@ -115,17 +121,6 @@ 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.
@@ -172,7 +167,7 @@ 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]) {
+ if (dynamicPartitionOverwrite) {
assert(dir.isDefined,
"The dataset to be written must be partitioned when dynamicPartitionOverwrite is true.")
partitionPaths += dir.get
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 954d9011f3bb..fa7ce49788e3 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
@@ -29,6 +29,8 @@ import org.apache.spark.SparkFunSuite
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
+import org.apache.spark.sql.{LocalSparkSession, SparkSession}
+import org.apache.spark.sql.internal.SQLConf
/**
* Subclass that exposes the protected `partitionPaths` field so tests can
@@ -319,12 +321,12 @@ class CommitterBindingSuite extends SparkFunSuite {
}
/**
- * 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.
+ * A cloud committer that declares dynamic partitioning support (via StreamCapabilities) does not
+ * delete the existing partitions itself. It must be bound to the staging directory, and the
+ * partitions must be tracked, so that commitJob can replace them in the destination.
*/
- test("SPARK-56588: Cloud committer with dynamic partition support does not track partitions in " +
- "partitionPaths") {
+ test("SPARK-60366: Cloud committer with dynamic partition support stages output and tracks " +
+ "partitions") {
val path = new Path("http://example/data")
val job = newJob(path)
val conf = job.getConfiguration
@@ -338,16 +340,44 @@ class CommitterBindingSuite extends SparkFunSuite {
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"
+ // The temp file must be under the work dir of the committer bound to the staging directory.
+ val expectedWorkDir = new Path(FileCommitProtocol.getStagingDir(path.toString, jobId),
+ StubPathOutputCommitterBinding.TEMP_DIR_NAME).toString
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}")
+ assert(committer.capturedPartitionPaths === Set("a=1"),
+ s"Expected partitionPaths = {a=1}, got: ${committer.capturedPartitionPaths}")
+ }
+
+ /**
+ * Dynamic partition overwrite through the manifest committer, which Hadoop 3.3.5+ uses for
+ * `abfs` and `gs`, must replace the written partitions instead of appending to them.
+ */
+ test("SPARK-60366: Dynamic partition overwrite through the manifest committer") {
+ withTempDir { dir =>
+ LocalSparkSession.withSparkSession(SparkSession.builder()
+ .master("local[2]")
+ .appName("manifest-committer-dynamic-partition-overwrite")
+ .config(SQLConf.FILE_COMMIT_PROTOCOL_CLASS.key, pathCommitProtocolClassname)
+ .config(SQLConf.PARTITION_OVERWRITE_MODE.key, "dynamic")
+ .config(s"spark.hadoop.$OUTPUTCOMMITTER_FACTORY_SCHEME.file",
+ "org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterFactory")
+ .getOrCreate()) { spark =>
+ import spark.implicits._
+ Seq("json", "parquet").foreach { format =>
+ val path = new File(dir, format).getCanonicalPath
+ Seq((1, 1), (2, 2)).toDF("c", "p").write.partitionBy("p").format(format).save(path)
+ // Unlike FileOutputCommitter, the manifest committer writes a JSON summary.
+ assert(new File(path, "_SUCCESS").length() > 0)
+
+ Seq((3, 1)).toDF("c", "p").write.partitionBy("p").mode("overwrite").format(format)
+ .save(path)
+ val rows = spark.read.schema("c INT, p INT").format(format).load(path).collect()
+ assert(rows.map(r => (r.getInt(0), r.getInt(1))).sorted.toSeq === Seq((2, 2), (3, 1)))
+ }
+ }
+ }
}
/**
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/sources/PartitionedWriteSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/sources/PartitionedWriteSuite.scala
index 5d9a7b0aafc7..2ca3c89208ff 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/sources/PartitionedWriteSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/sources/PartitionedWriteSuite.scala
@@ -17,11 +17,12 @@
package org.apache.spark.sql.sources
-import java.io.File
+import java.io.{File, IOException}
import java.sql.Timestamp
import org.apache.hadoop.fs.Path
import org.apache.hadoop.mapreduce.{JobContext, TaskAttemptContext}
+import org.apache.hadoop.mapreduce.lib.output.{PathOutputCommitter, PathOutputCommitterFactory}
import org.apache.spark.TestUtils
import org.apache.spark.internal.Logging
@@ -198,6 +199,54 @@ class PartitionedWriteSuite extends SharedSparkSession {
}
}
+ test("SPARK-60366: Dynamic partition overwrite with a committer that is not " +
+ "a FileOutputCommitter") {
+ // Hadoop 3.3.5+ binds `abfs` and `gs` to the manifest committer, which is not a
+ // FileOutputCommitter. Bind it to the local file system to test the same code path.
+ withSQLConf(SQLConf.PARTITION_OVERWRITE_MODE.key ->
+ SQLConf.PartitionOverwriteMode.DYNAMIC.toString,
+ "mapreduce.outputcommitter.factory.scheme.file" ->
+ "org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterFactory") {
+ Seq("json", "orc").foreach { format =>
+ withTempDir { d =>
+ withTable("t") {
+ sql(s"CREATE TABLE t(c1 INT, p1 INT) USING $format PARTITIONED BY (p1) " +
+ s"LOCATION '${d.getAbsolutePath}'")
+ sql("INSERT INTO t VALUES (1, 1), (2, 2)")
+ // Unlike FileOutputCommitter, the manifest committer writes a JSON summary.
+ assert(new File(d, "_SUCCESS").length() > 0)
+
+ sql("INSERT OVERWRITE TABLE t PARTITION(p1) SELECT 3, 1")
+ checkAnswer(spark.table("t"), Seq(Row(3, 1), Row(2, 2)))
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-60366: Dynamic partition overwrite keeps the existing partitions if the " +
+ "committer deletes the staged files") {
+ withSQLConf(SQLConf.PARTITION_OVERWRITE_MODE.key ->
+ SQLConf.PartitionOverwriteMode.DYNAMIC.toString) {
+ withTempDir { d =>
+ withTable("t") {
+ sql(s"CREATE TABLE t(c1 INT, p1 INT) USING json PARTITIONED BY (p1) " +
+ s"LOCATION '${d.getAbsolutePath}'")
+ sql("INSERT INTO t VALUES (1, 1), (2, 2)")
+
+ withSQLConf("mapreduce.outputcommitter.factory.scheme.file" ->
+ classOf[OutputDeletingCommitterFactory].getName) {
+ val e = intercept[IOException] {
+ sql("INSERT OVERWRITE TABLE t PARTITION(p1) SELECT 3, 1")
+ }
+ assert(e.getMessage.contains("No destination partition was changed"))
+ }
+ checkAnswer(spark.table("t"), Seq(Row(1, 1), Row(2, 2)))
+ }
+ }
+ }
+ }
+
test("SPARK-37231, SPARK-37240: Dynamic writes/reads of ANSI interval partitions") {
Seq("parquet", "json").foreach { format =>
Seq(
@@ -272,3 +321,26 @@ private class PartitionFileExistCommitProtocol(
conflictTaskFile.createNewFile()
}
}
+
+/**
+ * Creates committers that delete their output directory on job commit, like the S3A directory
+ * staging committer in the `replace` conflict mode.
+ */
+private class OutputDeletingCommitterFactory extends PathOutputCommitterFactory {
+ override def createOutputCommitter(
+ outputPath: Path,
+ context: TaskAttemptContext): PathOutputCommitter = {
+ new PathOutputCommitter(outputPath, context) {
+ override def getOutputPath: Path = outputPath
+ override def getWorkPath: Path = outputPath
+ override def setupJob(jobContext: JobContext): Unit = {}
+ override def setupTask(taskContext: TaskAttemptContext): Unit = {}
+ override def needsTaskCommit(taskContext: TaskAttemptContext): Boolean = false
+ override def commitTask(taskContext: TaskAttemptContext): Unit = {}
+ override def abortTask(taskContext: TaskAttemptContext): Unit = {}
+ override def commitJob(jobContext: JobContext): Unit = {
+ outputPath.getFileSystem(jobContext.getConfiguration).delete(outputPath, true)
+ }
+ }
+ }
+}
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala
index f8b9d982bfdc..e50d3533d5d8 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala
@@ -26,6 +26,7 @@ import org.scalatest.BeforeAndAfter
import org.apache.spark.{SparkException, SparkThrowable}
import org.apache.spark.sql.{QueryTest, _}
+import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.expressions.Hex
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.hive.execution.HiveTempPath
@@ -52,6 +53,27 @@ class InsertSuite extends QueryTest with TestHiveSingleton with BeforeAndAfter {
testData.createOrReplaceTempView("testData")
}
+ test("SPARK-60366: Dynamic partition overwrite of a converted Hive ORC table with a " +
+ "committer that is not a FileOutputCommitter") {
+ // Hadoop 3.3.5+ binds `abfs` and `gs` to the manifest committer. Converted Hive ORC tables
+ // always use dynamic partition overwrite.
+ withSQLConf(
+ "mapreduce.outputcommitter.factory.scheme.file" ->
+ "org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterFactory",
+ "hive.exec.dynamic.partition.mode" -> "nonstrict") {
+ withTable("t") {
+ sql("CREATE TABLE t (c1 INT) PARTITIONED BY (p1 INT) STORED AS ORC")
+ sql("INSERT INTO t PARTITION (p1) VALUES (1, 1), (2, 2)")
+ val location = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t")).location
+ // Unlike FileOutputCommitter, the manifest committer writes a JSON summary.
+ assert(new File(location.getPath, "_SUCCESS").length() > 0)
+
+ sql("INSERT OVERWRITE TABLE t PARTITION (p1) SELECT 3, 1")
+ checkAnswer(spark.table("t"), Seq(Row(3, 1), Row(2, 2)))
+ }
+ }
+ }
+
test("insertInto() HiveTable") {
withTable("createAndInsertTest") {
sql("CREATE TABLE createAndInsertTest (key int, value string)")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org