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 13:13:34 UTC
(spark) branch master updated: [SPARK-60360][SQL] Validate file source writes before deleting existing data in static overwrite
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 3b068a9c7aed [SPARK-60360][SQL] Validate file source writes before deleting existing data in static overwrite
3b068a9c7aed is described below
commit 3b068a9c7aedeef2a2edfd484cba8adeb5dbc274
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Tue Oct 20 06:13:25 2026 -0700
[SPARK-60360][SQL] Validate file source writes before deleting existing data in static overwrite
### What changes were proposed in this pull request?
This PR keeps the existing data when a static-mode overwrite fails the driver side checks of the write. The S3A magic committer on S3 benefits the most:
- The magic committer makes new files visible only at job commit, but a static-mode overwrite deletes the existing data in Spark itself (`FileCommitProtocol.deleteWithJob`, which `PathOutputCommitProtocol` does not override) before the job starts.
- Previously, Spark ran this delete before the driver side checks. An invalid write, such as a typo in the compression codec, bulk-deleted every object under the target S3 path and then failed.
- Now such a write fails before Spark deletes anything, so the existing data stays as is and no delete request is sent to S3.
`InsertIntoHadoopFsRelationCommand` used to delete the matching output before calling `FileFormatWriter.write`, which then checks the data schema (`verifySchema`, `checkFieldNames`) and the write options (`FileFormat.prepareWrite`). Now `FileFormatWriter.write` takes a `beforeSetupJob` callback (a no-op by default) that runs right before `committer.setupJob`, and `InsertIntoHadoopFsRelationCommand` deletes the matching output there. The delete still happens before `setupJob`, where th [...]
Out of scope: `saveAsTable` with `SaveMode.Overwrite` on an existing V1 table. `SaveAsV1TableCommand` drops the table before the CTAS reaches `FileFormatWriter`, so a write that fails these checks still drops the table (and the data of a managed table). This needs a separate fix.
### Why are the changes needed?
When these checks fail, the existing data has already been deleted.
```scala
spark.range(3).write.parquet(path)
spark.range(5).write.mode("overwrite").option("compression", "bogus").parquet(path)
// SparkIllegalArgumentException: [CODEC_NOT_AVAILABLE.WITH_AVAILABLE_CODECS_SUGGESTION] ...
spark.read.parquet(path)
// AnalysisException: [PATH_NOT_FOUND] Path does not exist: ...
```
The same happens with an unsupported data type (e.g. a map column in CSV), with `binaryFile`, which does not support writes, and with `INSERT OVERWRITE` into tables (including converted Hive tables) and directories. This has been the behavior since Spark 1.4.
### Does this PR introduce _any_ user-facing change?
Yes. A static-mode overwrite that fails these checks keeps the existing data instead of deleting it, with any committer. Valid writes are unchanged.
### How was this patch tested?
Added tests to `InsertSuite` of `sql/core` and `sql/hive`. They fail without this fix. The delete does not depend on the committer, so the tests use the default one.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59610 from dongjoon-hyun/SPARK-60360.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../execution/datasources/FileFormatWriter.scala | 10 ++++-
.../InsertIntoHadoopFsRelationCommand.scala | 10 ++++-
.../org/apache/spark/sql/sources/InsertSuite.scala | 48 +++++++++++++++++++++-
.../org/apache/spark/sql/hive/InsertSuite.scala | 18 +++++++-
4 files changed, 81 insertions(+), 5 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 5932c986ae3c..946c40d7f933 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
@@ -80,6 +80,9 @@ object FileFormatWriter extends Logging {
* thrown during job commitment, also aborts the job.
* 5. If the job is successfully committed, perform post-commit operations such as
* processing statistics.
+ * @param beforeSetupJob called right before the committer sets up the job, after the
+ * driver side checks of the write (e.g. data schema and write options)
+ * pass.
* @return The set of all partition paths that were updated during this write job.
*/
def write(
@@ -93,7 +96,8 @@ object FileFormatWriter extends Logging {
bucketSpec: Option[BucketSpec],
statsTrackers: Seq[WriteJobStatsTracker],
options: Map[String, String],
- numStaticPartitionCols: Int = 0)
+ numStaticPartitionCols: Int = 0,
+ beforeSetupJob: () => Unit = () => ())
: Set[String] = {
require(partitionColumns.size >= numStaticPartitionCols)
@@ -152,6 +156,10 @@ object FileFormatWriter extends Logging {
// get an ID guaranteed to be unique.
job.getConfiguration.set("spark.sql.sources.writeJobUUID", description.uuid)
+ // Called only after the checks above pass, so that a static-mode overwrite, which deletes
+ // the existing data here, keeps it when the write fails one of them.
+ beforeSetupJob()
+
// 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
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala
index 4c59ad2933bb..0ebab313135a 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala
@@ -119,6 +119,7 @@ case class InsertIntoHadoopFsRelationCommand(
outputPath = outputPath.toString,
dynamicPartitionOverwrite = dynamicPartitionOverwrite)
+ var deleteMatchingOutput = false
val doInsertion = if (mode == SaveMode.Append) {
true
} else {
@@ -133,7 +134,9 @@ case class InsertIntoHadoopFsRelationCommand(
// For dynamic partition overwrite, do not delete partition directories ahead.
true
} else {
- deleteMatchingPartitions(fs, qualifiedOutputPath, customPartitionLocations, committer)
+ // Deleted in `beforeSetupJob` below, so that a write that fails the driver side
+ // checks of FileFormatWriter (e.g. an invalid write option) keeps the existing data.
+ deleteMatchingOutput = true
true
}
case (SaveMode.Overwrite, _) | (SaveMode.ErrorIfExists, false) =>
@@ -192,7 +195,10 @@ case class InsertIntoHadoopFsRelationCommand(
bucketSpec = bucketSpec,
statsTrackers = Seq(basicWriteJobStatsTracker(hadoopConf)),
options = options,
- numStaticPartitionCols = staticPartitions.size)
+ numStaticPartitionCols = staticPartitions.size,
+ beforeSetupJob = () => if (deleteMatchingOutput) {
+ deleteMatchingPartitions(fs, qualifiedOutputPath, customPartitionLocations, committer)
+ })
// update metastore partition metadata
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 a25cc27fddc5..32d61626fb8a 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
@@ -23,7 +23,7 @@ import java.time.{Duration, Period}
import org.apache.hadoop.fs.{FileAlreadyExistsException, FSDataOutputStream, Path, RawLocalFileSystem}
-import org.apache.spark.{SparkArithmeticException, SparkException, SparkRuntimeException}
+import org.apache.spark.{SparkArithmeticException, SparkException, SparkRuntimeException, SparkThrowable}
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
@@ -3123,6 +3123,52 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
}
}
+ Seq(
+ // Rejected by the data schema check of FileFormatWriter.
+ ("an unsupported data type", "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
+ (dir: String) => Seq(1 -> Map("a" -> 1)).toDF("id", "m").write.mode("overwrite").csv(dir)),
+ // Rejected by the write option check of ParquetFileFormat.prepareWrite.
+ ("an invalid write option", "CODEC_NOT_AVAILABLE.WITH_AVAILABLE_CODECS_SUGGESTION",
+ (dir: String) =>
+ spark.range(5).write.mode("overwrite").option("compression", "bogus").parquet(dir)),
+ ("an invalid INSERT OVERWRITE DIRECTORY option",
+ "CODEC_NOT_AVAILABLE.WITH_AVAILABLE_CODECS_SUGGESTION",
+ (dir: String) => sql(s"INSERT OVERWRITE DIRECTORY '$dir' USING parquet " +
+ "OPTIONS (compression 'bogus') SELECT 5L AS id")),
+ // Rejected by BinaryFileFormat.prepareWrite, as the format does not support writes.
+ ("a format that does not support writes", "UNSUPPORTED_FEATURE.WRITE_FOR_BINARY_SOURCE",
+ (dir: String) => spark.range(5).write.mode("overwrite").format("binaryFile").save(dir))
+ ).foreach { case (reason, condition, overwrite) =>
+ test(s"SPARK-60360: overwrite keeps the existing data when it fails on $reason") {
+ withTempPath { path =>
+ val dir = path.getCanonicalPath
+ spark.range(3).write.parquet(dir)
+ assert(intercept[SparkThrowable](overwrite(dir)).getCondition === condition)
+ checkAnswer(spark.read.parquet(dir), spark.range(3).toDF())
+ }
+ }
+ }
+
+ test("SPARK-60360: INSERT OVERWRITE keeps the existing partitions when it fails on " +
+ "an invalid write option") {
+ withTable("t") {
+ sql("CREATE TABLE t (i INT, p INT) USING parquet PARTITIONED BY (p)")
+ sql("INSERT INTO t VALUES (1, 1), (2, 2)")
+ Seq(PartitionOverwriteMode.STATIC, PartitionOverwriteMode.DYNAMIC).foreach { mode =>
+ withSQLConf(SQLConf.PARTITION_OVERWRITE_MODE.key -> mode.toString) {
+ Seq(
+ "INSERT OVERWRITE t WITH ('compression' = 'bogus') PARTITION (p = 1) SELECT 3",
+ "INSERT OVERWRITE t WITH ('compression' = 'bogus') SELECT 3, 1"
+ ).foreach { insert =>
+ assert(intercept[SparkThrowable](sql(insert)).getCondition ===
+ "CODEC_NOT_AVAILABLE.WITH_AVAILABLE_CODECS_SUGGESTION")
+ checkAnswer(spark.table("t"), Seq(Row(1, 1), Row(2, 2)))
+ }
+ }
+ }
+ }
+ }
+
test("SPARK-51830: dynamic partition insert into a table with a non-numeric partition value") {
// `spark.sql.legacy.skipTypeValidationOnAlterPartition` lets us register a partition whose
// value does not match the numeric type of the partition column
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 db1743862e90..b12d4a4a7fa4 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
@@ -24,7 +24,7 @@ import java.util.Locale
import org.apache.hadoop.fs.Path
import org.scalatest.BeforeAndAfter
-import org.apache.spark.SparkException
+import org.apache.spark.{SparkException, SparkThrowable}
import org.apache.spark.sql.{QueryTest, _}
import org.apache.spark.sql.catalyst.expressions.Hex
import org.apache.spark.sql.catalyst.parser.ParseException
@@ -1038,4 +1038,20 @@ class InsertSuite extends QueryTest with TestHiveSingleton with BeforeAndAfter {
testDefaultColumn
}
}
+
+ test("SPARK-60360: INSERT OVERWRITE into a converted table keeps the existing data when " +
+ "it fails on an invalid write option") {
+ Seq("parquet" -> "parquet.compression", "orc" -> "orc.compress").foreach {
+ case (format, compressionProperty) =>
+ withTable("t") {
+ sql(s"CREATE TABLE t (id INT) STORED AS $format")
+ sql("INSERT INTO t VALUES (1)")
+ sql(s"ALTER TABLE t SET TBLPROPERTIES ('$compressionProperty' = 'bogus')")
+ assert(intercept[SparkThrowable](sql("INSERT OVERWRITE TABLE t SELECT 2"))
+ .getCondition === "CODEC_NOT_AVAILABLE.WITH_AVAILABLE_CODECS_SUGGESTION")
+ sql(s"ALTER TABLE t UNSET TBLPROPERTIES ('$compressionProperty')")
+ checkAnswer(spark.table("t"), Row(1))
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org