spark-test commented on code in PR #59962:
URL: https://github.com/apache/spark/pull/59962#discussion_r4429756810
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/utils/ErrorUtils.scala:
##########
@@ -46,6 +46,21 @@ import org.apache.spark.util.Utils
private[connect] object ErrorUtils extends Logging {
+ private def compatibleErrorClass(error: SparkThrowable): (String,
Map[String, String]) = {
+ val parameters = Option(error.getMessageParameters)
+ .map(_.asScala.toMap)
+ .getOrElse(Map.empty[String, String])
+ if (error.getCondition == "FAILED_JDBC.WRITE") {
+ // Connect clients have no negotiated error-catalog capability. Keep the
URL supplied
+ // by the WRITE error; it has already been redacted by JDBCOptions.
+ ("FAILED_JDBC.UNCLASSIFIED", Map(
Review Comment:
**Nit (P3):** This rewrites the error class and parameters but not the
message. The `FetchErrorDetails` message (`currentError.getMessage`, line 103)
and the gRPC status description (`errorDescription(st)`, line 296) are still
rendered from the original exception, so both read `[FAILED_JDBC.WRITE] Failed
JDBC <url> on the operation: Write rows to the table <t>. SQLSTATE: HV000`.
The Python Connect client uses that server message verbatim
(`python/pyspark/errors/exceptions/connect.py`, lines 187-196) but takes the
condition from the projected metadata. So for any write failure classified as
`FAILED_JDBC.WRITE`, a PySpark Connect user sees `[FAILED_JDBC.WRITE]` in
`str(e)` while `e.getCondition()` returns `FAILED_JDBC.UNCLASSIFIED`. A handler
written from the printed text never matches. The Scala Connect client
re-renders the message from the projected class and prints
`[FAILED_JDBC.UNCLASSIFIED]`, so the two clients disagree about the same
failure. Classic sessions are not affected.
As long as the projection exists, it could also return the message rendered
for `FAILED_JDBC.UNCLASSIFIED` from the projected parameters (for example with
`SparkThrowableHelper.getMessage`) and use it on both channels. The new
`ErrorUtilsSuite` case checks the class, SQLSTATE and parameters but neither
message, so it does not catch this.
**Verification:**
- **Behavior:** When the server converts a FAILED_JDBC.WRITE error, both the
gRPC status description and the root error-details message name
FAILED_JDBC.UNCLASSIFIED and contain the redacted URL and the table name,
matching the projected error class and parameters.
- **Compatibility:** An error with any other condition keeps the same status
description and error-details message as before the change, and its cause chain
is serialized unchanged.
- **Regression:** For a failed JDBC write, a Connect client that builds the
exception from the server-rendered message, as the Python client does, reports
the same condition in its printed message and in getCondition().
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcTaskInterruptSuite.scala:
##########
@@ -57,6 +57,252 @@ class JdbcTaskInterruptSuite extends SharedSparkSession {
JdbcDialects.registerDialect(dialect)
}
+ private def savePartitionWithSourceError(iterator: Iterator[Row]):
SQLException = {
+ val conn = mock(classOf[Connection])
+ val stmt = mock(classOf[PreparedStatement])
+ when(conn.prepareStatement(anyString())).thenReturn(stmt)
+
+ val dialect = new JdbcDialect with NoLegacyJDBCError {
+ override def canHandle(url: String): Boolean = url == testJdbcUrl
+ override def createConnectionFactory(options: JDBCOptions): Int =>
Connection = _ => conn
+ }
+ registerDialect(dialect)
+ TaskContext.setTaskContext(TaskContext.empty())
+ try {
+ intercept[SQLException] {
+ JdbcUtils.savePartition(
+ "t",
+ iterator,
+ StructType(Seq(StructField("a", IntegerType))),
+ "INSERT INTO t (a) VALUES (?)",
+ batchSize = 1,
+ dialect,
+ Connection.TRANSACTION_NONE,
+ newWriteOptions())
+ }
+ } finally {
+ TaskContext.unset()
+ JdbcDialects.unregisterDialect(dialect)
+ }
+ }
+
+ test("savePartition does not classify a later source hasNext failure as a
JDBC write") {
+ val sourceFailure = new SQLException("source read failed", "08006")
+ val nextFailure = new SQLException("source connection detail", "08001")
+ sourceFailure.setNextException(nextFailure)
+ val iterator = new Iterator[Row] {
+ private var readFirstRow = false
+ override def hasNext: Boolean = {
+ if (readFirstRow) throw sourceFailure
+ true
+ }
+ override def next(): Row = {
+ readFirstRow = true
+ Row(1)
+ }
+ }
+
+ assert(savePartitionWithSourceError(iterator) eq sourceFailure)
+ assert(sourceFailure.getSQLState == "08006")
+ assert(sourceFailure.getNextException eq nextFailure)
+ assert(sourceFailure.getCause eq nextFailure)
+ }
+
+ test("savePartition preserves a classified later source next failure") {
Review Comment:
**Nit (P3):** This is the only test of a failure in the source iterator's
`next()`, and it throws a `SparkSQLException`. `classifyWriteException` would
rethrow that exception unchanged anyway, through its `SparkThrowable` bypass,
and it has no next exception to link. So this test still passes if `next()`
moves into `withWriteException` or the binding `try`, or if `sourceNext` stops
calling `linkNextException`. In the first case, a failing upstream read in
`next()` would be reported as `FAILED_JDBC.WRITE` on the destination table. A
plain `SQLException` with a chained next exception, like the one the `hasNext`
test above uses, would make this path regression-sensitive.
**Verification:**
- **Regression:** When the source iterator's next() throws a plain
SQLException with a chained next exception during a JDBC write, the same
exception propagates unclassified, keeps its original SQLSTATE, and has the
next exception as its cause. The check fails if next() moves into a classified
region or the linking is removed.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala:
##########
@@ -718,75 +718,127 @@ object JdbcUtils extends Logging with SQLConfHelper {
}
val supportsTransactions = finalIsolationLevel !=
Connection.TRANSACTION_NONE
var totalRowCount = 0L
+ def linkNextException(e: SQLException): Unit = {
+ val cause = e.getNextException
+ if (cause != null && e.getCause != cause) {
+ // If there is no cause already, set 'next exception' as cause. If
cause is null,
+ // it *may* be because no cause was set yet
+ if (e.getCause == null) {
+ try {
+ e.initCause(cause)
+ } catch {
+ // Or it may be null because the cause *was* explicitly
initialized, to *null*,
+ // in which case this fails. There is no other way to detect it.
+ // addSuppressed in this case as well.
+ case _: IllegalStateException => e.addSuppressed(cause)
+ }
+ } else {
+ e.addSuppressed(cause)
+ }
+ }
+ }
+
+ def sourceHasNext: Boolean = {
+ try {
+ iterator.hasNext
+ } catch {
+ case e: SQLException =>
+ linkNextException(e)
+ throw e
+ }
+ }
+
+ def sourceNext(): Row = {
+ try {
+ iterator.next()
+ } catch {
+ case e: SQLException =>
+ linkNextException(e)
+ throw e
+ }
+ }
+
+ def classifyWriteException(e: SQLException): Nothing = {
+ linkNextException(e)
+ // Closing the connection on task cancellation can cause a SQLException.
Do not
+ // report that cancellation side effect as a JDBC write failure.
+ if (TaskContext.get().isInterrupted() || e.isInstanceOf[SparkThrowable])
{
+ throw e
+ }
+ throw dialect.classifyException(
+ e,
+ condition = "FAILED_JDBC.WRITE",
+ messageParameters = Map("url" -> options.getRedactUrl(), "tableName"
-> table),
+ description = "Write rows to the table.",
Review Comment:
**Nit (P3):** Dialects that don't override the condition-based
`classifyException` fall back to the deprecated `classifyException(message,
e)`. On that path this description is the entire error message, and the URL is
fixed to `jdbc:`. That includes `NoopDialect`, which is used for any URL
without a registered dialect (SQLite, for example), and third-party dialects
that keep the default. Their write failures surface as
`[FAILED_JDBC.UNCLASSIFIED] Failed JDBC jdbc: on the operation: Write rows to
the table. SQLSTATE: HV000`, which names neither the database nor the table.
When a job writes several JDBC tables, the failing one can then only be found
through the cause or the stack trace. The `FAILED_JDBC.WRITE` rendering and the
Connect projection both name the table.
Built-in dialects ignore this description, so naming the table here only
changes the legacy path. The expected message in the legacy-dialect test would
change with it:
```suggestion
description = s"Write rows to the table $table.",
```
**Verification:**
- **Regression:** Classifying a write failure through a dialect that
inherits the default classifyException yields FAILED_JDBC.UNCLASSIFIED, and its
message parameter names the target table.
- **Compatibility:** For dialects with condition-based classification, a
write failure still yields FAILED_JDBC.WRITE whose only parameters are the
redacted URL and the table name.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcTaskInterruptSuite.scala:
##########
@@ -57,6 +57,252 @@ class JdbcTaskInterruptSuite extends SharedSparkSession {
JdbcDialects.registerDialect(dialect)
}
+ private def savePartitionWithSourceError(iterator: Iterator[Row]):
SQLException = {
+ val conn = mock(classOf[Connection])
+ val stmt = mock(classOf[PreparedStatement])
+ when(conn.prepareStatement(anyString())).thenReturn(stmt)
+
+ val dialect = new JdbcDialect with NoLegacyJDBCError {
+ override def canHandle(url: String): Boolean = url == testJdbcUrl
+ override def createConnectionFactory(options: JDBCOptions): Int =>
Connection = _ => conn
+ }
+ registerDialect(dialect)
+ TaskContext.setTaskContext(TaskContext.empty())
+ try {
+ intercept[SQLException] {
+ JdbcUtils.savePartition(
+ "t",
+ iterator,
+ StructType(Seq(StructField("a", IntegerType))),
+ "INSERT INTO t (a) VALUES (?)",
+ batchSize = 1,
+ dialect,
+ Connection.TRANSACTION_NONE,
+ newWriteOptions())
+ }
+ } finally {
+ TaskContext.unset()
+ JdbcDialects.unregisterDialect(dialect)
+ }
+ }
+
+ test("savePartition does not classify a later source hasNext failure as a
JDBC write") {
Review Comment:
**Nit (P3):** This test and the five after it cover `savePartition`
write-error classification and source-failure propagation, all under a
non-interrupted `TaskContext`. The suite's name and class ScalaDoc (lines
36-42) still scope it to JDBC task interrupt handling (SPARK-56019), and after
this PR six of its nine tests fall outside that scope. The ScalaDoc is now
wrong for most of the suite, and someone looking for the new
`FAILED_JDBC.WRITE` coverage is unlikely to look here. Moving these tests, with
the helpers they need, into a suite whose name and ScalaDoc describe
write-error classification would keep both suites accurate.
**Verification:**
- **Inspection:** Every test in the interrupt-handling suite exercises task
interruption, and every write-classification or source-propagation test lives
in a suite whose name and ScalaDoc describe JDBC write-error classification.
- **Behavior:** After the move, the relocated tests still run and keep all
of their original assertions on condition, SQLSTATE, parameters, cause identity
and rollback.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
