[SPARK-52321] Add SessionClosed, SqlConfNotFound, ParseSyntaxError to SparkConnectError

Author: dongjoon-hyun

Sources/SparkConnect/DataFrame.swift

@@ -311,6 +311,8 @@ public actor DataFrame: Sendable {
           throw SparkConnectError.TableOrViewNotFound
         case let m where m.contains("UNRESOLVED_COLUMN.WITH_SUGGESTION"):
           throw SparkConnectError.ColumnNotFound
+        case let m where m.contains("PARSE_SYNTAX_ERROR"):
+          throw SparkConnectError.ParseSyntaxError
         default:
           throw error
         }

Sources/SparkConnect/SparkConnectClient.swift

@@ -131,6 +131,10 @@ public actor SparkConnectClient {
         return try await f(client)
       } catch let error as RPCError where error.code == .internalError {
         switch error.message {
+        case let m where m.contains("INVALID_HANDLE.SESSION_CLOSED"):
+          throw SparkConnectError.SessionClosed
+        case let m where m.contains("SQL_CONF_NOT_FOUND"):
+          throw SparkConnectError.SqlConfNotFound
         case let m where m.contains("TABLE_OR_VIEW_ALREADY_EXISTS"):
           throw SparkConnectError.TableOrViewAlreadyExists
         case let m where m.contains("TABLE_OR_VIEW_NOT_FOUND"):

Sources/SparkConnect/SparkConnectError.swift

@@ -25,7 +25,10 @@ public enum SparkConnectError: Error {
   case InvalidSessionID
   case InvalidType
   case InvalidViewName
+  case ParseSyntaxError
   case SchemaNotFound
+  case SessionClosed
+  case SqlConfNotFound
   case TableOrViewAlreadyExists
   case TableOrViewNotFound
   case UnsupportedOperation

Tests/SparkConnectTests/CatalogTests.swift

@@ -144,7 +144,7 @@ struct CatalogTests {
     })
     #expect(try await spark.catalog.tableExists(tableName) == false)
 
-    try await #require(throws: Error.self) {
+    try await #require(throws: SparkConnectError.ParseSyntaxError) {
       try await spark.catalog.tableExists("invalid table name")
     }
     await spark.stop()
@@ -190,7 +190,7 @@ struct CatalogTests {
     #expect(try await spark.catalog.functionExists("base64"))
     #expect(try await spark.catalog.functionExists("non_exist_function") == false)
 
-    try await #require(throws: Error.self) {
+    try await #require(throws: SparkConnectError.ParseSyntaxError) {
       try await spark.catalog.functionExists("invalid function name")
     }
     await spark.stop()

Tests/SparkConnectTests/RuntimeConfTests.swift

@@ -35,7 +35,7 @@ struct RuntimeConfTests {
 
     #expect(try await !conf.get("spark.app.name").isEmpty)
 
-    try await #require(throws: Error.self) {
+    try await #require(throws: SparkConnectError.SqlConfNotFound) {
       try await conf.get("spark.test.non-exist")
     }
 
@@ -86,7 +86,7 @@ struct RuntimeConfTests {
     #expect(try await conf.get("spark.test.key1") == "value1")
 
     try await conf.unset("spark.test.key1")
-    try await #require(throws: Error.self) {
+    try await #require(throws: SparkConnectError.SqlConfNotFound) {
       try await conf.get("spark.test.key1")
     }
 

Tests/SparkConnectTests/SparkSessionTests.swift

@@ -72,7 +72,7 @@ struct SparkSessionTests {
       let sessionID = spark1.sessionID
       await spark1.stop()
       let remote = ProcessInfo.processInfo.environment["SPARK_REMOTE"] ?? "sc://localhost"
-      try await #require(throws: Error.self) {
+      try await #require(throws: SparkConnectError.SessionClosed) {
         try await SparkSession.builder.remote("\(remote)/;session_id=\(sessionID)").getOrCreate()
       }
     }