linhongliu-db commented on code in PR #59962:
URL: https://github.com/apache/spark/pull/59962#discussion_r4458601778
##########
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:
Added a setup-failure grid in 09c7b5995e3 for setAutoCommit,
setTransactionIsolation, prepareStatement, and setQueryTimeout. Each case
asserts FAILED_JDBC.WRITE / HV000, URL/table parameters, original cause,
rollback/connection cleanup, and no batch execution or commit. All 102 focused
tests and SQL/Connect style checks pass.
--
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]
