spark-test commented on code in PR #59962:
URL: https://github.com/apache/spark/pull/59962#discussion_r4454979205
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala:
##########
@@ -722,75 +722,132 @@ 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 = s"Write rows to the table $table.",
+ isRuntime = true)
+ }
+
+ def withWriteException[T](f: => T): T = {
+ try {
+ f
+ } catch {
+ case e: SQLException => classifyWriteException(e)
+ }
+ }
+
try {
- if (supportsTransactions) {
- conn.setAutoCommit(false) // Everything in the same db transaction.
- conn.setTransactionIsolation(finalIsolationLevel)
+ withWriteException {
Review Comment:
**Non-blocking (P2):** The connection that these classified calls use is
opened earlier, at line 673 (`dialect.createConnectionFactory(options)(-1)`),
outside every `withWriteException`. A `SQLException` from opening it on an
executor therefore still propagates raw. It gets no next-exception linking and
no classification, and the caller sees a job-aborted `SparkException` with no
condition or SQLSTATE.
This is a common write-task failure. For example, the driver-side connection
in `JdbcRelationProvider.createRelation` can succeed while some partitions hit
PostgreSQL's `too_many_connections` (53300) or a network rule that only
executors see. A failure one call later, in `setAutoCommit` or
`prepareStatement`, surfaces as `FAILED_JDBC.WRITE`. So handlers rewritten per
the migration guide get a condition for every destination failure except this
one. The guide's list of unclassified failures (source iterator errors and task
interruption) is also incomplete. The driver-side `JdbcUtils.withConnection`
classifies the same step as `FAILED_JDBC.CONNECTION`.
The classification helpers use only `dialect`, `options`, `table` and the
task context, so the connection-factory call can go through the same helper.
That keeps the interrupt and `SparkThrowable` bypasses.
**Verification:**
- **Regression:** A write task whose connection factory throws a
SQLException fails with the classified write error, carrying the redacted URL,
the table name and the original exception as its cause, instead of an
unclassified job-aborted error.
- **Behavior:** When the task is already marked interrupted, a
connection-acquisition SQLException is rethrown unchanged.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala:
##########
@@ -722,75 +722,132 @@ 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 = s"Write rows to the table $table.",
+ isRuntime = true)
+ }
+
+ def withWriteException[T](f: => T): T = {
+ try {
+ f
+ } catch {
+ case e: SQLException => classifyWriteException(e)
+ }
+ }
+
try {
- if (supportsTransactions) {
- conn.setAutoCommit(false) // Everything in the same db transaction.
- conn.setTransactionIsolation(finalIsolationLevel)
+ withWriteException {
+ if (supportsTransactions) {
+ conn.setAutoCommit(false) // Everything in the same db transaction.
+ conn.setTransactionIsolation(finalIsolationLevel)
+ }
}
- val stmt = conn.prepareStatement(insertStmt)
- val setters =
+ val stmt = withWriteException { conn.prepareStatement(insertStmt) }
Review Comment:
**Nit (P3):** A `SQLException` from `prepareStatement` here, as from the
transaction setup above it and `setQueryTimeout` below it, is now classified as
`FAILED_JDBC.WRITE`, but no test makes any of these calls throw.
`JdbcWriteExceptionSuite` injects failures only at `setInt`, `executeBatch`,
`commit`, the source iterator and `stmt.close()`. `JdbcTaskInterruptSuite`
stubs the setup calls with `doNothing`, and the H2 case in `JDBCWriteSuite`
fails inside `executeBatch`. Dropping any of these wrappers would leave every
assertion green. Prepare time is still a realistic failure point:
eager-preparing drivers such as H2 and Derby reject an INSERT there, for
example for a missing column or privilege. The existing Mockito fixtures could
make each call throw in one small grid test.
**Verification:**
- **Regression:** For each setup-time destination call (transaction setup,
statement preparation and query-timeout setup), a test that makes the call
throw a SQLException asserts the classified write condition, SQLSTATE,
parameters and original cause.
--
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]
