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 20:32:38 UTC
(spark) branch master updated: [SPARK-60387][SQL] Keep the existing table when `saveAsTable` overwrite fails before writing
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 5637184e54a2 [SPARK-60387][SQL] Keep the existing table when `saveAsTable` overwrite fails before writing
5637184e54a2 is described below
commit 5637184e54a25ab628f27fcea0f40ea253d63e56
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Tue Oct 20 13:32:28 2026 -0700
[SPARK-60387][SQL] Keep the existing table when `saveAsTable` overwrite fails before writing
### What changes were proposed in this pull request?
This PR makes `saveAsTable` with `SaveMode.Overwrite` drop the existing V1 table only after the checks that run before writing, such as the analysis, the data schema check, and the write options.
Previously, `SaveAsV1TableCommand` dropped the table before running the CTAS. Now the CTAS drops it:
- `CreateDataSourceTableAsSelectCommand` drops it right before the write job starts, in the `beforeSetupJob` callback of `FileFormatWriter.write` (SPARK-60360), or right before `CreatableRelationProvider.createRelation`. If an extension replaces `InsertIntoHadoopFsRelationCommand` in the physical plan (e.g. a native writer), it drops the table before the write, as before.
- `CreateHiveTableAsSelectCommand` drops it after the analysis, because it creates the new table before writing.
### Why are the changes needed?
When these checks fail, the table has already been dropped. A managed table loses its data too.
```scala
spark.range(3).write.saveAsTable("t")
spark.range(5).write.mode("overwrite").option("compression", "bogus").saveAsTable("t")
// SparkIllegalArgumentException: [CODEC_NOT_AVAILABLE.WITH_AVAILABLE_CODECS_SUGGESTION] ...
spark.table("t")
// AnalysisException: [TABLE_OR_VIEW_NOT_FOUND] The table or view `t` cannot be found. ...
```
The same happens with an unsupported data type (e.g. a map column in CSV) or a nonexistent partition column. This has been the behavior since Spark 1.3.0.
### Does this PR introduce _any_ user-facing change?
Yes. Such an overwrite keeps the existing table and its data. With `format("hive")`, only a failure in the analysis keeps the table, because `CreateHiveTableAsSelectCommand` creates the new table before writing. Valid writes are unchanged.
### How was this patch tested?
Added tests to `DataFrameReaderWriterSuite` and `MetastoreDataSourcesSuite`. The tests of failed overwrites fail without this fix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59643 from dongjoon-hyun/SPARK-60387.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../spark/sql/execution/QueryExecution.scala | 4 +-
.../execution/command/createDataSourceTables.scala | 27 +++++---
.../apache/spark/sql/execution/command/ddl.scala | 4 +-
.../sql/execution/datasources/DataSource.scala | 25 ++++++--
.../InsertIntoHadoopFsRelationCommand.scala | 18 +++++-
.../sql/test/DataFrameReaderWriterSuite.scala | 74 +++++++++++++++++++++-
.../execution/CreateHiveTableAsSelectCommand.scala | 11 ++--
.../spark/sql/hive/MetastoreDataSourcesSuite.scala | 22 +++++++
8 files changed, 161 insertions(+), 24 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
index dd5c7a22fbde..d0a0b0d150e8 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
@@ -1043,7 +1043,8 @@ object QueryExecution {
refreshPhaseEnabled: Boolean = true,
mode: CommandExecutionMode.Value = CommandExecutionMode.SKIP,
shuffleCleanupModeOpt: Option[ShuffleCleanupMode] = None,
- analyzerOpt: Option[Analyzer] = None)
+ analyzerOpt: Option[Analyzer] = None,
+ beforeExecution: SparkPlan => Unit = _ => ())
: (QueryExecution, Array[InternalRow]) = {
val qe = new QueryExecution(
sparkSession,
@@ -1054,6 +1055,7 @@ object QueryExecution {
analyzerOpt = analyzerOpt)
val result = QueryExecution.withInternalError(s"Executed $name failed.") {
SQLExecution.withNewExecutionId(qe, Some(name)) {
+ beforeExecution(qe.executedPlan)
qe.executedPlan.executeCollect()
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/createDataSourceTables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/createDataSourceTables.scala
index 7519216f1b36..dd6cd7dea4bf 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/createDataSourceTables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/createDataSourceTables.scala
@@ -158,11 +158,9 @@ case class CreateDataSourceTableAsSelectCommand(
val db = table.identifier.database.getOrElse(sessionState.catalog.getCurrentDatabase)
val tableIdentWithDB = table.identifier.copy(database = Some(db))
val tableName = tableIdentWithDB.unquotedString
+ val tableExists = sessionState.catalog.tableExists(tableIdentWithDB)
- if (sessionState.catalog.tableExists(tableIdentWithDB)) {
- assert(mode != SaveMode.Overwrite,
- s"Expect the table $tableName has been dropped when the save mode is Overwrite")
-
+ if (tableExists && mode != SaveMode.Overwrite) {
if (mode == SaveMode.ErrorIfExists) {
throw QueryCompilationErrors.tableAlreadyExistsError(tableName)
}
@@ -178,14 +176,26 @@ case class CreateDataSourceTableAsSelectCommand(
DataWritingCommand.assertEmptyRootPath(p, mode, sparkSession.sessionState.newHadoopConf())
}
assert(table.schema.isEmpty)
- sparkSession.sessionState.catalog.validateTableLocation(table)
+ val beforeWrite = if (tableExists) {
+ // `SaveAsV1TableCommand` runs this command with `SaveMode.Overwrite` to replace the
+ // table. Drop it right before writing, after the write passes its checks, so that an
+ // invalid write keeps it. Validate the location after the drop, which deletes the data
+ // of a managed table.
+ () => {
+ sessionState.catalog.dropTable(tableIdentWithDB, ignoreIfNotExists = true, purge = false)
+ sessionState.catalog.validateTableLocation(table)
+ }
+ } else {
+ sessionState.catalog.validateTableLocation(table)
+ () => ()
+ }
val tableLocation = if (table.tableType == CatalogTableType.MANAGED) {
Some(sessionState.catalog.defaultTablePath(table.identifier))
} else {
table.storage.locationUri
}
val result = saveDataIntoTable(
- sparkSession, table, tableLocation, SaveMode.Overwrite, tableExists = false)
+ sparkSession, table, tableLocation, SaveMode.Overwrite, tableExists = false, beforeWrite)
val tableSchema = CharVarcharUtils.getRawSchema(
removeInternalMetadata(result.schema), sessionState.conf)
val newTable = table.copy(
@@ -219,7 +229,8 @@ case class CreateDataSourceTableAsSelectCommand(
table: CatalogTable,
tableLocation: Option[URI],
mode: SaveMode,
- tableExists: Boolean): BaseRelation = {
+ tableExists: Boolean,
+ beforeWrite: () => Unit = () => ()): BaseRelation = {
// Create the relation based on the input logical plan: `query`.
val pathOption = tableLocation.map("path" -> CatalogUtils.URIToString(_))
val dataSource = DataSource(
@@ -231,7 +242,7 @@ case class CreateDataSourceTableAsSelectCommand(
catalogTable = if (tableExists) Some(table) else None)
try {
- dataSource.writeAndRead(mode, query, outputColumnNames)
+ dataSource.writeAndRead(mode, query, outputColumnNames, beforeWrite)
} catch {
case ex: AnalysisException =>
logError(log"Failed to write to table " +
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
index 01f74180f38e..a3f476b0afa4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
@@ -1060,8 +1060,8 @@ private[sql] case class SaveAsV1TableCommand(
case _ => // OK
}
- // Drop the existing table
- catalog.dropTable(qualifiedIdent, ignoreIfNotExists = true, purge = false)
+ // The CTAS replaces the existing table. It drops the table as late as it can, so that a
+ // write that fails its checks keeps the table.
runCommand(sparkSession, CreateTable(tableDescWithQualifiedIdent, mode, Some(query)))
// Refresh the cache of the table in the catalog.
catalog.refreshTable(qualifiedIdent)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
index 52b9e803a64e..e55fc1637c26 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
@@ -41,8 +41,8 @@ import org.apache.spark.sql.classic.ClassicConversions.castToImpl
import org.apache.spark.sql.classic.Dataset
import org.apache.spark.sql.connector.catalog.TableProvider
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
-import org.apache.spark.sql.execution.QueryExecution
-import org.apache.spark.sql.execution.command.DataWritingCommand
+import org.apache.spark.sql.execution.{QueryExecution, SparkPlan}
+import org.apache.spark.sql.execution.command.{DataWritingCommand, DataWritingCommandExec}
import org.apache.spark.sql.execution.datasources.csv.CSVFileFormat
import org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider
import org.apache.spark.sql.execution.datasources.json.JsonFileFormat
@@ -530,11 +530,15 @@ case class DataSource(
* @param outputColumnNames The original output column names of the input query plan. The
* optimizer may not preserve the output column's names' case, so we need
* this parameter instead of `data.output`.
+ * @param beforeWrite Called right before writing. For a [[FileFormat]], it runs after the driver
+ * side checks of the write. For a [[CreatableRelationProvider]], it runs
+ * before `createRelation`, which checks the write itself.
*/
def writeAndRead(
mode: SaveMode,
data: LogicalPlan,
- outputColumnNames: Seq[String]): BaseRelation = {
+ outputColumnNames: Seq[String],
+ beforeWrite: () => Unit = () => ()): BaseRelation = {
val outputColumns = DataWritingCommand.logicalPlanOutputWithNames(data, outputColumnNames)
providingInstance() match {
case dataSource: CreatableRelationProvider =>
@@ -544,13 +548,26 @@ case class DataSource(
dataSource.toString, StructField(attr.toString, attr.dataType))
}
}
+ beforeWrite()
dataSource.createRelation(
sparkSession.sqlContext, mode, caseInsensitiveOptions, Dataset.ofRows(sparkSession, data))
case format: FileFormat =>
disallowWritingIntervals(
outputColumns.toStructType.asNullable, format.toString, forbidAnsiIntervals = false)
val cmd = planForWritingFileFormat(format, mode, data)
- QueryExecution.runCommand(sparkSession, cmd, "file source write")
+ cmd.setTagValue(InsertIntoHadoopFsRelationCommand.BEFORE_WRITE_JOB, beforeWrite)
+ // An extension can replace the command in the physical plan (e.g. with a native writer),
+ // and then nothing calls `beforeWrite` during the write. Call it before such a write.
+ val callBeforeWriteIfReplaced = (plan: SparkPlan) => {
+ val runsCommand = plan.exists {
+ case DataWritingCommandExec(c, _) =>
+ c.getTagValue(InsertIntoHadoopFsRelationCommand.BEFORE_WRITE_JOB).isDefined
+ case _ => false
+ }
+ if (!runsCommand) beforeWrite()
+ }
+ QueryExecution.runCommand(
+ sparkSession, cmd, "file source write", beforeExecution = callBeforeWriteIfReplaced)
// Replace the schema with that of the DataFrame we just wrote out to avoid re-inferring
copy(userSpecifiedSchema = Some(outputColumns.toStructType.asNullable)).resolveRelation()
case _ => throw SparkException.internalError(
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 0ebab313135a..7f87ff5e7694 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
@@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.catalog.ExternalCatalogUtils._
import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.trees.TreeNodeTag
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
@@ -196,8 +197,11 @@ case class InsertIntoHadoopFsRelationCommand(
statsTrackers = Seq(basicWriteJobStatsTracker(hadoopConf)),
options = options,
numStaticPartitionCols = staticPartitions.size,
- beforeSetupJob = () => if (deleteMatchingOutput) {
- deleteMatchingPartitions(fs, qualifiedOutputPath, customPartitionLocations, committer)
+ beforeSetupJob = () => {
+ getTagValue(InsertIntoHadoopFsRelationCommand.BEFORE_WRITE_JOB).foreach(_())
+ if (deleteMatchingOutput) {
+ deleteMatchingPartitions(fs, qualifiedOutputPath, customPartitionLocations, committer)
+ }
})
@@ -296,3 +300,13 @@ case class InsertIntoHadoopFsRelationCommand(
override protected def withNewChildInternal(
newChild: LogicalPlan): InsertIntoHadoopFsRelationCommand = copy(query = newChild)
}
+
+object InsertIntoHadoopFsRelationCommand {
+
+ /**
+ * A callback that runs right before the write job starts, after the write passes the driver
+ * side checks of [[FileFormatWriter]]. A CTAS that replaces an existing table drops the table
+ * there, so that a write that fails these checks keeps the table.
+ */
+ val BEFORE_WRITE_JOB = TreeNodeTag[() => Unit]("before_write_job")
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/test/DataFrameReaderWriterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/test/DataFrameReaderWriterSuite.scala
index cf05be197e24..a1d6b148421f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/test/DataFrameReaderWriterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/test/DataFrameReaderWriterSuite.scala
@@ -32,7 +32,7 @@ import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import org.apache.parquet.schema.Type.Repetition
import org.scalatest.BeforeAndAfter
-import org.apache.spark.{SparkContext, SparkException, SparkIllegalArgumentException, TestUtils}
+import org.apache.spark.{SparkContext, SparkException, SparkIllegalArgumentException, SparkThrowable, TestUtils}
import org.apache.spark.internal.io.FileCommitProtocol.TaskCommitMessage
import org.apache.spark.internal.io.HadoopMapReduceCommitProtocol
import org.apache.spark.scheduler.{SparkListener, SparkListenerJobStart}
@@ -40,9 +40,9 @@ import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.plans.logical.{AppendData, LogicalPlan, OverwriteByExpression}
import org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPrecision
-import org.apache.spark.sql.execution.QueryExecution
+import org.apache.spark.sql.execution.{QueryExecution, SparkPlan, SparkStrategy}
import org.apache.spark.sql.execution.command.DataWritingCommandExec
-import org.apache.spark.sql.execution.datasources.{DataSourceUtils, HadoopFsRelation, LogicalRelation}
+import org.apache.spark.sql.execution.datasources.{DataSourceUtils, HadoopFsRelation, InsertIntoHadoopFsRelationCommand, LogicalRelation}
import org.apache.spark.sql.execution.datasources.noop.NoopDataSource
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
@@ -1018,6 +1018,74 @@ class DataFrameReaderWriterSuite extends SharedSparkSession with BeforeAndAfter
}
}
+ Seq(
+ // Rejected by the analysis of the CTAS.
+ ("a nonexistent partition column", "COLUMN_NOT_DEFINED_IN_TABLE",
+ () => spark.range(5).write.partitionBy("nonexistent")),
+ // Rejected by the data schema check of FileFormatWriter.
+ ("an unsupported data type", "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
+ () => Seq(1 -> Map("a" -> 1)).toDF("id", "m").write.format("csv")),
+ // Rejected by the write option check of ParquetFileFormat.prepareWrite.
+ ("an invalid write option", "CODEC_NOT_AVAILABLE.WITH_AVAILABLE_CODECS_SUGGESTION",
+ () => spark.range(5).write.format("parquet").option("compression", "bogus"))
+ ).foreach { case (reason, condition, newWriter) =>
+ test(s"SPARK-60387: saveAsTable with mode Overwrite keeps the existing table when it fails " +
+ s"on $reason") {
+ withTempPath { path =>
+ // A managed table and an external table
+ Seq(Map.empty[String, String], Map("path" -> path.getCanonicalPath)).foreach { options =>
+ withTable("t") {
+ Seq((1, 1), (2, 2)).toDF("i", "p").write.options(options).partitionBy("p")
+ .saveAsTable("t")
+ val e = intercept[SparkThrowable] {
+ newWriter().mode(SaveMode.Overwrite).options(options).saveAsTable("t")
+ }
+ assert(e.getCondition === condition)
+ checkAnswer(spark.table("t"), Seq(Row(1, 1), Row(2, 2)))
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-60387: saveAsTable with mode Overwrite keeps the existing table when " +
+ "a CreatableRelationProvider rejects the data type") {
+ val format = classOf[DefaultSource].getName
+ withTable("t") {
+ spark.range(3).write.format(format).saveAsTable("t")
+ val e = intercept[SparkThrowable] {
+ sql("SELECT INTERVAL '1' DAY AS a").write.mode(SaveMode.Overwrite).format(format)
+ .saveAsTable("t")
+ }
+ assert(e.getCondition === "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE")
+ assert(spark.catalog.tableExists("t"))
+ }
+ }
+
+ test("SPARK-60387: saveAsTable with mode Overwrite replaces the table when the planned " +
+ "write does not run the callback") {
+ // Plans the write command without its tags, like an extension that replaces the command.
+ object PlanWriteWithoutTags extends SparkStrategy {
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case c: InsertIntoHadoopFsRelationCommand
+ if c.getTagValue(InsertIntoHadoopFsRelationCommand.BEFORE_WRITE_JOB).isDefined =>
+ DataWritingCommandExec(c.copy(), planLater(c.query)) :: Nil
+ case _ => Nil
+ }
+ }
+ withTable("t") {
+ spark.range(3).write.saveAsTable("t")
+ val originalStrategies = spark.experimental.extraStrategies
+ try {
+ spark.experimental.extraStrategies = PlanWriteWithoutTags +: originalStrategies
+ spark.range(5, 7).write.mode(SaveMode.Overwrite).saveAsTable("t")
+ } finally {
+ spark.experimental.extraStrategies = originalStrategies
+ }
+ checkAnswer(spark.table("t"), spark.range(5, 7).toDF())
+ }
+ }
+
test("saveAsTable with mode Ignore should create the table if the table not exists " +
"but a same-name temp view exist") {
withTable("same_name") {
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/CreateHiveTableAsSelectCommand.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/CreateHiveTableAsSelectCommand.scala
index 154d07f80d89..c0f3bab3143b 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/CreateHiveTableAsSelectCommand.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/CreateHiveTableAsSelectCommand.scala
@@ -49,10 +49,7 @@ case class CreateHiveTableAsSelectCommand(
val catalog = sparkSession.sessionState.catalog
val tableExists = catalog.tableExists(tableIdentifier)
- if (tableExists) {
- assert(mode != SaveMode.Overwrite,
- s"Expect the table $tableIdentifier has been dropped when the save mode is Overwrite")
-
+ if (tableExists && mode != SaveMode.Overwrite) {
if (mode == SaveMode.ErrorIfExists) {
throw QueryCompilationErrors.tableIdentifierExistsError(tableIdentifier)
}
@@ -68,6 +65,12 @@ case class CreateHiveTableAsSelectCommand(
tableDesc.storage.locationUri.foreach { p =>
DataWritingCommand.assertEmptyRootPath(p, mode, sparkSession.sessionState.newHadoopConf())
}
+ if (tableExists) {
+ // `SaveAsV1TableCommand` runs this command with `SaveMode.Overwrite` to replace the
+ // table. Drop it here, after the analysis, because the new table is created before the
+ // write.
+ catalog.dropTable(tableIdentifier, ignoreIfNotExists = true, purge = false)
+ }
// TODO ideally, we should get the output data ready first and then
// add the relation into catalog, just in case of failure occurs while data
// processing.
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
index 405a14fc2afa..3073049799a2 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
@@ -24,6 +24,7 @@ import scala.collection.mutable.ArrayBuffer
import org.apache.hadoop.fs.Path
import org.apache.logging.log4j.Level
+import org.apache.spark.SparkThrowable
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
@@ -1301,6 +1302,27 @@ class MetastoreDataSourcesSuite extends QueryTest
}
}
+ Seq(
+ // Rejected by the write option check of ParquetFileFormat.prepareWrite.
+ ("parquet", "CODEC_NOT_AVAILABLE.WITH_AVAILABLE_CODECS_SUGGESTION",
+ () => spark.range(5).write.format("parquet").option("compression", "bogus")),
+ // Rejected by the analysis of the Hive CTAS.
+ ("hive", "COLUMN_NOT_DEFINED_IN_TABLE",
+ () => spark.range(5).write.format("hive").partitionBy("nonexistent"))
+ ).foreach { case (format, condition, newWriter) =>
+ test(s"SPARK-60387: saveAsTable with mode Overwrite keeps the existing $format table when " +
+ "it fails before writing") {
+ withTable("t") {
+ spark.range(3).write.format(format).saveAsTable("t")
+ val e = intercept[SparkThrowable] {
+ newWriter().mode(SaveMode.Overwrite).saveAsTable("t")
+ }
+ assert(e.getCondition === condition)
+ checkAnswer(table("t"), spark.range(3).toDF())
+ }
+ }
+ }
+
test("insertInto - source and target are the same table") {
val tableName = "tab1"
withTable(tableName) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org