Always set a sql state in spark connect client

Author: garlandz-db

common/utils/src/main/resources/error/error-conditions.json

@@ -870,6 +870,18 @@
     },
     "sqlState" : "56K00"
   },
+  "CONNECT_CLIENT_INTERNAL_ERROR" : {
+  "message" : [
+    "<message>"
+  ],
+  "sqlState" : "XXKCI"
+  },
+  "CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE" : {
+  "message" : [
+  "<message>"
+  ],
+  "sqlState" : "XXKCM"
+  },
   "CONNECT_ML" : {
     "message" : [
       "Generic Spark Connect ML error."

sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/ClientE2ETestSuite.scala

@@ -1792,6 +1792,46 @@ class ClientE2ETestSuite
     assert(result.length === 1)
     assert(result(0).getAs[Array[Integer]]("arr_col") === Array(1, null))
   }
+
+  test("SPARK-53883: GrpcExceptionConverter ensures all exceptions have errorClass and sqlState") {
+    // Test that normal Spark errors have proper errorClass and sqlState
+    val analysisEx = intercept[AnalysisException] {
+      spark.sql("select nonexistent_column").collect()
+    }
+    assert(analysisEx.getErrorClass != null,
+      "AnalysisException should have non-null errorClass")
+    assert(analysisEx.getSqlState != null,
+      "AnalysisException should have non-null sqlState")
+    assert(analysisEx.getErrorClass == "UNRESOLVED_COLUMN.WITHOUT_SUGGESTION",
+      s"AnalysisException should have correct errorClass, got: ${analysisEx.getErrorClass}")
+
+    val parseEx = intercept[ParseException] {
+      spark.sql("SELECT !!!").collect()
+    }
+    assert(parseEx.getErrorClass != null,
+      "ParseException should have non-null errorClass")
+    assert(parseEx.getSqlState != null,
+      "ParseException should have non-null sqlState")
+    assert(parseEx.getErrorClass.startsWith("PARSE_SYNTAX_ERROR"),
+      s"ParseException should have correct errorClass, got: ${parseEx.getErrorClass}")
+
+    // Test that runtime errors have proper errorClass and sqlState
+    val runtimeEx = intercept[SparkException] {
+      spark.udf.register("badUdf", () => { throw new RuntimeException("test error") })
+      spark.sql("SELECT badUdf()").collect()
+    }
+    assert(runtimeEx.getErrorClass != null,
+      "Runtime SparkException should have non-null errorClass")
+    assert(runtimeEx.getSqlState != null,
+      "Runtime SparkException should have non-null sqlState")
+
+    // Verify that the error classes we added exist in error-conditions.json
+    // by checking they would be valid if used
+    assert(analysisEx.getSqlState != "XXKCI",
+      "Normal errors should not use CONNECT_CLIENT_INTERNAL_ERROR sqlState")
+    assert(analysisEx.getSqlState != "XXKCM",
+      "Normal errors should not use CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE sqlState")
+  }
 }
 
 private[sql] case class ClassData(a: String, b: Int)

sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/GrpcExceptionConverter.scala

@@ -20,6 +20,7 @@ import java.time.DateTimeException
 
 import scala.jdk.CollectionConverters._
 import scala.reflect.ClassTag
+import scala.util.control.NonFatal
 
 import com.google.rpc.ErrorInfo
 import io.grpc.{ManagedChannel, StatusRuntimeException}
@@ -59,6 +60,11 @@ private[client] class GrpcExceptionConverter(channel: ManagedChannel) extends Lo
     } catch {
       case e: StatusRuntimeException =>
         throw toThrowable(e, sessionId, userContext, clientType)
+      case NonFatal(e) =>
+        throw new SparkException(
+          errorClass = "CONNECT_CLIENT_INTERNAL_ERROR",
+          messageParameters = Map("message" -> e.toString),
+          cause = e)
     }
   }
 
@@ -139,6 +145,22 @@ private[client] class GrpcExceptionConverter(channel: ManagedChannel) extends Lo
       clientType: String): Throwable = {
     val status = StatusProto.fromThrowable(ex)
 
+    if (status == null) {
+      val statusCode = ex.getStatus.getCode
+      val (errorClass, sqlState) = statusCode match {
+        case io.grpc.Status.Code.PERMISSION_DENIED =>
+          ("INSUFFICIENT_PERMISSIONS", "42501")
+        case io.grpc.Status.Code.UNAUTHENTICATED =>
+          ("UNAUTHENTICATED", "08000")
+        case _ =>
+          ("CONNECT_CLIENT_INTERNAL_ERROR", "XXKCI")
+      }
+      return new SparkException(
+        errorClass = errorClass,
+        messageParameters = Map("message" -> ex.toString),
+        cause = ex)
+    }
+
     // Extract the ErrorInfo from the StatusProto, if present.
     val errorInfoOpt = status.getDetailsList.asScala
       .find(_.is(classOf[ErrorInfo]))
@@ -159,7 +181,10 @@ private[client] class GrpcExceptionConverter(channel: ManagedChannel) extends Lo
     }
 
     // If no ErrorInfo is found, create a SparkException based on the StatusRuntimeException.
-    new SparkException(ex.toString, ex.getCause)
+    new SparkException(
+      errorClass = "CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE",
+      messageParameters = Map("message" -> ex.toString),
+      cause = ex.getCause)
   }
 }
 
@@ -168,7 +193,6 @@ private[client] object GrpcExceptionConverter {
   private[client] case class ErrorParams(
       message: String,
       cause: Option[Throwable],
-      // errorClass will only be set if the error is SparkThrowable.
       errorClass: Option[String],
       // messageParameters will only be set if the error is both enriched and SparkThrowable.
       messageParameters: Map[String, String],
@@ -186,13 +210,13 @@ private[client] object GrpcExceptionConverter {
       new StreamingQueryException(
         params.message,
         params.cause.orNull,
-        params.errorClass.orNull,
+        params.errorClass.getOrElse("CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE"),
         params.messageParameters)),
     errorConstructor(params =>
       new ParseException(
         None,
         Origin(),
-        errorClass = params.errorClass.orNull,
+        errorClass = params.errorClass.getOrElse("CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE"),
         messageParameters = params.messageParameters,
         queryContext = params.queryContext)),
     errorConstructor(params =>
@@ -202,26 +226,29 @@ private[client] object GrpcExceptionConverter {
         cause = params.cause,
         context = params.queryContext)),
     errorConstructor(params =>
-      new NamespaceAlreadyExistsException(params.errorClass.orNull, params.messageParameters)),
+      new NamespaceAlreadyExistsException(params.errorClass.getOrElse(
+        "CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE"), params.messageParameters)),
     errorConstructor(params =>
       new TableAlreadyExistsException(
-        params.errorClass.orNull,
+        params.errorClass.getOrElse("CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE"),
         params.messageParameters,
         params.cause)),
     errorConstructor(params =>
       new TempTableAlreadyExistsException(
-        params.errorClass.orNull,
+        params.errorClass.getOrElse("CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE"),
         params.messageParameters,
         params.cause)),
     errorConstructor(params =>
       new NoSuchDatabaseException(
-        params.errorClass.orNull,
+        params.errorClass.getOrElse("CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE"),
         params.messageParameters,
         params.cause)),
     errorConstructor(params =>
-      new NoSuchNamespaceException(params.errorClass.orNull, params.messageParameters)),
+      new NoSuchNamespaceException(params.errorClass.getOrElse(
+        "CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE"), params.messageParameters)),
     errorConstructor(params =>
-      new NoSuchTableException(params.errorClass.orNull, params.messageParameters, params.cause)),
+      new NoSuchTableException(params.errorClass.getOrElse(
+        "CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE"), params.messageParameters, params.cause)),
     errorConstructor[NumberFormatException](params =>
       new SparkNumberFormatException(
         errorClass = params.errorClass.getOrElse("_LEGACY_ERROR_TEMP_3104"),
@@ -255,20 +282,21 @@ private[client] object GrpcExceptionConverter {
         params.queryContext)),
     errorConstructor(params =>
       new SparkRuntimeException(
-        params.errorClass.orNull,
+        params.errorClass.getOrElse("CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE"),
         params.messageParameters,
         params.cause.orNull,
         params.queryContext)),
     errorConstructor(params =>
       new SparkUpgradeException(
-        params.errorClass.orNull,
+        params.errorClass.getOrElse("CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE"),
         params.messageParameters,
         params.cause.orNull)),
     errorConstructor(params =>
       new SparkException(
         message = params.message,
         cause = params.cause.orNull,
-        errorClass = params.errorClass,
+        errorClass = params.errorClass.orElse(
+          Option("CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE")),
         messageParameters = params.messageParameters,
         context = params.queryContext)))