This is an automated email from the ASF dual-hosted git repository.
pan3793 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 ed16a7eef937 [SPARK-58352][SQL] Assign a name to the error condition
_LEGACY_ERROR_TEMP_2070
ed16a7eef937 is described below
commit ed16a7eef93798cbbdd77eb0e0d74c5840577fa7
Author: Xiao Li <[email protected]>
AuthorDate: Thu Aug 13 17:13:45 2026 +0800
[SPARK-58352][SQL] Assign a name to the error condition
_LEGACY_ERROR_TEMP_2070
### What changes were proposed in this pull request?
Assign the name `WRITING_JOB_FAILED` to the legacy error condition
`_LEGACY_ERROR_TEMP_2070`. This error is a `SparkException` raised by
`QueryExecutionErrors.writingJobFailedError`, thrown by
`WriteToDataSourceV2Exec`
when a DataSource V2 batch write fails and the subsequent `abort()` also
fails (the
original cause is chained onto the exception). The new condition is given
SQLSTATE
`58030` (matching the sibling `TASK_WRITE_FAILED`); it stays parameterless
and
preserves the chained `cause`.
### Why are the changes needed?
The error conditions
[README](https://github.com/apache/spark/blob/master/common/utils/src/main/resources/error/README.md)
disallows new `_LEGACY_ERROR_TEMP_*` entries and asks existing ones to be
assigned
proper names. This resolves one of them, under the umbrella
[SPARK-37935](https://issues.apache.org/jira/browse/SPARK-37935).
### Does this PR introduce _any_ user-facing change?
No. The `_LEGACY_ERROR_TEMP_*` names are not part of the public API. The
error
condition name changes to `WRITING_JOB_FAILED` and now reports SQLSTATE
`58030`; the
message text ("Writing job failed.") is unchanged.
### How was this patch tested?
Added a `checkError` test in `DataSourceV2Suite` with a new
`CommitAndAbortFailingDataSource` whose batch write fails to commit and
then fails to
abort, driving the exact `WriteToDataSourceV2Exec` branch that raises
`WRITING_JOB_FAILED`. `SparkThrowableSuite` validates SQLSTATE presence and
error-file formatting/sorting for the new entry.
### Was this patch authored or co-authored using generative AI tooling?
Yes
This patch had conflicts when merged, resolved by
Committer: Cheng Pan <[email protected]>
Closes #57540 from gatorsmile/SPARK-58352-writing-job-failed.
Authored-by: Xiao Li <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 11 ++--
.../spark/sql/errors/QueryExecutionErrors.scala | 2 +-
.../spark/sql/connector/DataSourceV2Suite.scala | 64 +++++++++++++++++++++-
3 files changed, 70 insertions(+), 7 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index dc5631bdbdb1..1e8be8d30cf3 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -9746,6 +9746,12 @@
],
"sqlState" : "42601"
},
+ "WRITING_JOB_FAILED" : {
+ "message" : [
+ "Writing job failed."
+ ],
+ "sqlState" : "58030"
+ },
"WRONG_COMMAND_FOR_OBJECT_TYPE" : {
"message" : [
"The operation <operation> requires a <requiredType>. But <objectName>
is a <foundType>. Use <alternative> instead."
@@ -10832,11 +10838,6 @@
"Missing database location."
]
},
- "_LEGACY_ERROR_TEMP_2070" : {
- "message" : [
- "Writing job failed."
- ]
- },
"_LEGACY_ERROR_TEMP_2071" : {
"message" : [
"Commit denied for partition <partId> (task <taskId>, attempt
<attemptId>, stage <stageId>.<stageAttempt>)."
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index 646d296c11c6..91afa6b4dbf6 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -974,7 +974,7 @@ private[sql] object QueryExecutionErrors extends
QueryErrorsBase with ExecutionE
def writingJobFailedError(cause: Throwable): Throwable = {
new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_2070",
+ errorClass = "WRITING_JOB_FAILED",
messageParameters = Map.empty,
cause = cause)
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
index 08afcf79f3e7..eb4e6b53dd23 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
@@ -34,7 +34,7 @@ import org.apache.spark.sql.catalyst.expressions.{
LessThan => CatalystLessThan, Literal => CatalystLiteral, ScalarSubquery}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter =>
LogicalFilter, Project}
import
org.apache.spark.sql.catalyst.plans.logical.statsEstimation.EstimationUtils
-import org.apache.spark.sql.connector.catalog.{PartitionInternalRow,
SupportsRead, Table, TableCapability, TableProvider}
+import org.apache.spark.sql.connector.catalog.{PartitionInternalRow,
SupportsRead, SupportsWrite, Table, TableCapability, TableProvider}
import org.apache.spark.sql.connector.catalog.TableCapability._
import org.apache.spark.sql.connector.expressions.{Expression, FieldReference,
Literal, NamedReference, NullOrdering, SortDirection, SortOrder, Transform}
import org.apache.spark.sql.connector.expressions.filter.Predicate
@@ -42,6 +42,7 @@ import org.apache.spark.sql.connector.read._
import org.apache.spark.sql.connector.read.Scan.ColumnarSupportMode
import org.apache.spark.sql.connector.read.colstats.ColumnStatistics
import
org.apache.spark.sql.connector.read.partitioning.{KeyGroupedPartitioning,
Partitioning, UnknownPartitioning}
+import org.apache.spark.sql.connector.write.{BatchWrite, DataWriter,
DataWriterFactory, LogicalWriteInfo, PhysicalWriteInfo, Write, WriteBuilder,
WriterCommitMessage}
import org.apache.spark.sql.execution.SortExec
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.datasources.v2.{
@@ -504,6 +505,17 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
}
}
+ test("SPARK-58352: WRITING_JOB_FAILED when batch write commit and abort both
fail") {
+ val cls = classOf[CommitAndAbortFailingDataSource]
+ checkError(
+ exception = intercept[SparkException] {
+ spark.range(1).select($"id" as Symbol("i"), -$"id" as Symbol("j"))
+ .write.format(cls.getName).mode("append").save()
+ },
+ condition = "WRITING_JOB_FAILED",
+ parameters = Map.empty[String, String])
+ }
+
test("simple counter in writer with onDataWriterCommit") {
Seq(classOf[SimpleWritableDataSource]).foreach { cls =>
withTempPath { file =>
@@ -2506,6 +2518,56 @@ object SpecificReaderFactory extends
PartitionReaderFactory {
class SchemaReadAttemptException(m: String) extends RuntimeException(m)
+/**
+ * A writable data source whose batch write both fails to commit and then
fails to abort. This
+ * drives the exact branch in `WriteToDataSourceV2Exec` that raises
`WRITING_JOB_FAILED`: the
+ * write's `commit` throws, and the follow-up `abort` also throws, so the
original failure is
+ * wrapped rather than re-thrown. The per-task `DataWriter` succeeds so the
failure is driver-side
+ * and deterministic (no dependence on task scheduling).
+ */
+class CommitAndAbortFailingDataSource extends TestingV2Source {
+
+ override def getTable(options: CaseInsensitiveStringMap): Table = new
SimpleBatchTable
+ with SupportsWrite {
+
+ override def capabilities(): java.util.Set[TableCapability] =
+ java.util.EnumSet.of(TableCapability.BATCH_READ,
TableCapability.BATCH_WRITE,
+ TableCapability.TRUNCATE)
+
+ override def newScanBuilder(options: CaseInsensitiveStringMap):
ScanBuilder =
+ new SimpleScanBuilder {
+ override def planInputPartitions(): Array[InputPartition] = Array.empty
+ }
+
+ override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = new
WriteBuilder {
+ override def build(): Write = new Write {
+ override def toBatch: BatchWrite = new BatchWrite {
+ override def createBatchWriterFactory(info: PhysicalWriteInfo):
DataWriterFactory =
+ CommitAndAbortFailingDataSource.WriterFactory
+
+ override def commit(messages: Array[WriterCommitMessage]): Unit =
+ throw new RuntimeException("commit failed")
+
+ override def abort(messages: Array[WriterCommitMessage]): Unit =
+ throw new RuntimeException("abort failed")
+ }
+ }
+ }
+ }
+}
+
+object CommitAndAbortFailingDataSource {
+ object WriterFactory extends DataWriterFactory {
+ override def createWriter(partitionId: Int, taskId: Long):
DataWriter[InternalRow] =
+ new DataWriter[InternalRow] {
+ override def write(record: InternalRow): Unit = {}
+ override def commit(): WriterCommitMessage = null
+ override def abort(): Unit = {}
+ override def close(): Unit = {}
+ }
+ }
+}
+
class SimpleWriteOnlyDataSource extends SimpleWritableDataSource {
override def getTable(options: CaseInsensitiveStringMap): Table = {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
