This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 972c4cbe4baa [SPARK-56919][SQL] Move setupJob before
materializeAdaptiveSparkPlan to prevent table path loss
972c4cbe4baa is described below
commit 972c4cbe4baa454e64f29484a78db3883af4d0a6
Author: Shrirang Mhalgi <[email protected]>
AuthorDate: Sun Jun 21 15:04:34 2026 -0700
[SPARK-56919][SQL] Move setupJob before materializeAdaptiveSparkPlan to
prevent table path loss
### What changes were proposed in this pull request?
1. Move `committer.setupJob(job)` from inside `writeAndCommit()` to before
`materializeAdaptiveSparkPlan()` in `FileFormatWriter.write()`, so the output
path is recreated before anything can throw.
2. Wrap the post-`setupJob` body in `try { ... } catch {
committer.abortJob(job); throw }` so the staging dir is cleaned up on any
failure (e.g., AQE shuffle stage failure in `materializeAdaptiveSparkPlan)`.
3. Remove `writeAndCommit`'s inner `try / catch + abortJob` since the outer
catch now handles it - avoiding double-calling of `abortJob` for write / commit
failures
### Why are the changes needed?
`INSERT OVERWRITE` deletes the output path before calling `write()`. When
`materializeAdaptiveSparkPlan` throws (AQE shuffle stage failure),
`writeAndCommit` is never reached, so `setupJob` never recreates the path. The
table path is permanently lost. The outer `try / catch` ensures `abortJob`
cleans up the `staging dir (_temporary / .spark-staging-*)` on any failure
after setupJob.
### Does this PR introduce _any_ user-facing change?
Yes. Previously, a failed `INSERT OVERWRITE` with AQE could permanently
delete the table path. Now the path survives the failure.
### How was this patch tested?
Added regression test in `InsertSuite` that uses a failing UDF in a shuffle
stage to trigger AQE failure during `materializeAdaptiveSparkPlan`. Verifies
the table path exists after the failed overwrite.
### Was this patch authored or co-authored using generative AI tooling?
Yes. Authored using Claude Opus 4.6.
Closes #56126 from shrirangmhalgi/SPARK-56919-setupJob-before-materialize.
Authored-by: Shrirang Mhalgi <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 802b0b0d7e8c6d0df77d6666b70e4c510fef177c)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../execution/datasources/FileFormatWriter.scala | 141 +++++++++++----------
.../org/apache/spark/sql/sources/InsertSuite.scala | 31 +++++
2 files changed, 104 insertions(+), 68 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 59c103577e13..c75c8f046214 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,52 +148,66 @@ object FileFormatWriter extends Logging {
writerBucketSpec.map(_.bucketIdExpression) ++ sortColumns
val writeFilesOpt = V1WritesUtils.getWriteFilesOpt(plan)
- // SPARK-40588: when planned writing is disabled and AQE is enabled,
- // plan contains an AdaptiveSparkPlanExec, which does not know
- // its final plan's ordering, so we have to materialize that plan first
- // it is fine to use plan further down as the final plan is cached in that
plan
- def materializeAdaptiveSparkPlan(plan: SparkPlan): SparkPlan = plan match {
- case a: AdaptiveSparkPlanExec => a.finalPhysicalPlan
- case p: SparkPlan =>
p.withNewChildren(p.children.map(materializeAdaptiveSparkPlan))
- }
+ // 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
+ // calls abortJob on any later failure (e.g. materialize throwing), which
cleans up the
+ // staging dir (_temporary / .spark-staging-*).
+ committer.setupJob(job)
- // the sort order doesn't matter
- val actualOrdering = writeFilesOpt.map(_.child)
- .getOrElse(materializeAdaptiveSparkPlan(plan))
- .outputOrdering
- val orderingMatched = V1WritesUtils.isOrderingMatched(requiredOrdering,
actualOrdering)
-
- 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
- // we set the variable to verify this behavior in tests.
- // There are two cases where FileFormatWriter still needs to add physical
sort:
- // 1) When the planned write config is disabled.
- // 2) When the concurrent writers are enabled (in this case the required
ordering of a
- // V1 write command will be empty).
- if (Utils.isTesting) outputOrderingMatched = orderingMatched
-
- if (writeFilesOpt.isDefined) {
- // build `WriteFilesSpec` for `WriteFiles`
- val concurrentOutputWriterSpecFunc = (plan: SparkPlan) => {
- val sortPlan = createSortPlan(plan, requiredOrdering, outputSpec)
- createConcurrentOutputWriterSpec(sparkSession, sortPlan, sortColumns)
+ try {
+ // SPARK-40588: when planned writing is disabled and AQE is enabled,
+ // plan contains an AdaptiveSparkPlanExec, which does not know
+ // its final plan's ordering, so we have to materialize that plan first
+ // it is fine to use plan further down as the final plan is cached in
that plan
+ def materializeAdaptiveSparkPlan(plan: SparkPlan): SparkPlan = plan
match {
+ case a: AdaptiveSparkPlanExec => a.finalPhysicalPlan
+ case p: SparkPlan =>
p.withNewChildren(p.children.map(materializeAdaptiveSparkPlan))
}
- val writeSpec = WriteFilesSpec(
- description = description,
- committer = committer,
- concurrentOutputWriterSpecFunc = concurrentOutputWriterSpecFunc
- )
- executeWrite(sparkSession, plan, writeSpec, job)
- } else {
- executeWrite(sparkSession, plan, job, description, committer, outputSpec,
- requiredOrdering, partitionColumns, sortColumns, orderingMatched)
+
+ // the sort order doesn't matter
+ val actualOrdering = writeFilesOpt.map(_.child)
+ .getOrElse(materializeAdaptiveSparkPlan(plan))
+ .outputOrdering
+ val orderingMatched = V1WritesUtils.isOrderingMatched(requiredOrdering,
actualOrdering)
+
+ 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
+ // we set the variable to verify this behavior in tests.
+ // There are two cases where FileFormatWriter still needs to add
physical sort:
+ // 1) When the planned write config is disabled.
+ // 2) When the concurrent writers are enabled (in this case the required
ordering of a
+ // V1 write command will be empty).
+ if (Utils.isTesting) outputOrderingMatched = orderingMatched
+
+ if (writeFilesOpt.isDefined) {
+ // build `WriteFilesSpec` for `WriteFiles`
+ val concurrentOutputWriterSpecFunc = (plan: SparkPlan) => {
+ val sortPlan = createSortPlan(plan, requiredOrdering, outputSpec)
+ createConcurrentOutputWriterSpec(sparkSession, sortPlan, sortColumns)
+ }
+ val writeSpec = WriteFilesSpec(
+ description = description,
+ committer = committer,
+ concurrentOutputWriterSpecFunc = concurrentOutputWriterSpecFunc
+ )
+ executeWrite(sparkSession, plan, writeSpec, job)
+ } else {
+ executeWrite(sparkSession, plan, job, description, committer,
outputSpec,
+ requiredOrdering, partitionColumns, sortColumns, orderingMatched)
+ }
+ } catch {
+ case cause: Throwable =>
+ logError(log"Aborting job ${MDC(WRITE_JOB_UUID, description.uuid)}.",
cause)
+ committer.abortJob(job)
+ throw cause
}
}
// scalastyle:on argcount
@@ -267,30 +281,21 @@ object FileFormatWriter extends Logging {
job: Job,
description: WriteJobDescription,
committer: FileCommitProtocol)(f: => Array[WriteTaskResult]):
Set[String] = {
- // This call shouldn't be put into the `try` block below because it only
initializes and
- // prepares the job, any exception thrown from here shouldn't cause
abortJob() to be called.
- committer.setupJob(job)
- try {
- val ret = f
- val commitMsgs = ret.map(_.commitMsg)
-
- logInfo(log"Start to commit write Job ${MDC(LogKeys.UUID,
description.uuid)}.")
- val (_, duration) = Utils
- .timeTakenMs { committer.commitJob(job,
commitMsgs.toImmutableArraySeq) }
- logInfo(log"Write Job ${MDC(LogKeys.UUID, description.uuid)} committed.
" +
- log"Elapsed time: ${MDC(LogKeys.ELAPSED_TIME, duration)} ms.")
-
- processStats(
- description.statsTrackers,
ret.map(_.summary.stats).toImmutableArraySeq, duration)
- logInfo(log"Finished processing stats for write job ${MDC(LogKeys.UUID,
description.uuid)}.")
-
- // return a set of all the partition paths that were updated during this
job
- ret.map(_.summary.updatedPartitions).reduceOption(_ ++
_).getOrElse(Set.empty)
- } catch { case cause: Throwable =>
- logError(log"Aborting job ${MDC(WRITE_JOB_UUID, description.uuid)}.",
cause)
- committer.abortJob(job)
- throw cause
- }
+ val ret = f
+ val commitMsgs = ret.map(_.commitMsg)
+
+ logInfo(log"Start to commit write Job ${MDC(LogKeys.UUID,
description.uuid)}.")
+ val (_, duration) = Utils
+ .timeTakenMs { committer.commitJob(job, commitMsgs.toImmutableArraySeq) }
+ logInfo(log"Write Job ${MDC(LogKeys.UUID, description.uuid)} committed. " +
+ log"Elapsed time: ${MDC(LogKeys.ELAPSED_TIME, duration)} ms.")
+
+ processStats(
+ description.statsTrackers, ret.map(_.summary.stats).toImmutableArraySeq,
duration)
+ logInfo(log"Finished processing stats for write job ${MDC(LogKeys.UUID,
description.uuid)}.")
+
+ // return a set of all the partition paths that were updated during this
job
+ ret.map(_.summary.updatedPartitions).reduceOption(_ ++
_).getOrElse(Set.empty)
}
/**
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
index c0786c738446..8d2418c5e052 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
@@ -3089,6 +3089,37 @@ class InsertSuite extends DataSourceTest with
SharedSparkSession {
}
}
}
+
+ test("SPARK-56919: INSERT OVERWRITE should not lose table path when AQE
fails") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_FORCE_APPLY.key -> "true",
+ SQLConf.PLANNED_WRITE_ENABLED.key -> "false") {
+ withTempPath { path =>
+ val tablePath = path.getAbsolutePath
+ spark.range(10).toDF("id")
+ .write.mode("overwrite").parquet(tablePath)
+ assert(new java.io.File(tablePath).exists())
+
+ spark.udf.register("fail_udf", (i: Long) => {
+ throw new RuntimeException("SPARK-56919")
+ i
+ })
+
+ // The repartition forces a shuffle stage. With planned write disabled,
+ // materializeAdaptiveSparkPlan runs the stage, which fails via the
UDF.
+ intercept[Exception] {
+ spark.sql(s"SELECT fail_udf(id) as id FROM parquet.`$tablePath`")
+ .repartition(2)
+ .write.mode("overwrite").parquet(tablePath)
+ }
+
+ // The table path must survive a failed overwrite.
+ assert(new java.io.File(tablePath).exists(),
+ "Table path should not be permanently lost after a failed INSERT
OVERWRITE")
+ }
+ }
+ }
}
class FileExistingTestFileSystem extends RawLocalFileSystem {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
