[SPARK-52319] Add (Catalog|Schema|TableOrView)NotFound to SparkConnectError

### What changes were proposed in this pull request?

This PR aims to add `(Catalog|Schema|TableOrView)NotFound` to `SparkConnectError`.

### Why are the changes needed?

To provide a user can catch these exceptions easily instead of matching `internalError` with string patterns.

### Does this PR introduce _any_ user-facing change?

Yes, but these are more specific exceptions than before.

### How was this patch tested?

Pass the CIs.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #178 from dongjoon-hyun/SPARK-52319.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: dongjoon-hyun

Sources/SparkConnect/DataFrame.swift

@@ -299,7 +299,20 @@ public actor DataFrame: Sendable {
       ),
       interceptors: spark.client.getIntercepters()
     ) { client in
-      return try await f(client)
+      do {
+        return try await f(client)
+      } catch let error as RPCError where error.code == .internalError {
+        switch error.message {
+        case let m where m.contains("CATALOG_NOT_FOUND"):
+          throw SparkConnectError.CatalogNotFound
+        case let m where m.contains("SCHEMA_NOT_FOUND"):
+          throw SparkConnectError.SchemaNotFound
+        case let m where m.contains("TABLE_OR_VIEW_NOT_FOUND"):
+          throw SparkConnectError.TableOrViewNotFound
+        default:
+          throw error
+        }
+      }
     }
   }
 

Sources/SparkConnect/SparkConnectError.swift

@@ -19,8 +19,11 @@
 
 /// A enum for ``SparkConnect`` package errors
 public enum SparkConnectError: Error {
+  case CatalogNotFound
   case InvalidArgument
   case InvalidSessionID
   case InvalidType
+  case SchemaNotFound
+  case TableOrViewNotFound
   case UnsupportedOperation
 }

Tests/SparkConnectTests/CatalogTests.swift

@@ -37,8 +37,14 @@ struct CatalogTests {
   func setCurrentCatalog() async throws {
     let spark = try await SparkSession.builder.getOrCreate()
     try await spark.catalog.setCurrentCatalog("spark_catalog")
-    try await #require(throws: Error.self) {
-      try await spark.catalog.setCurrentCatalog("not_exist_catalog")
+    if await spark.version >= "4.0.0" {
+      try await #require(throws: SparkConnectError.CatalogNotFound) {
+        try await spark.catalog.setCurrentCatalog("not_exist_catalog")
+      }
+    } else {
+      try await #require(throws: Error.self) {
+        try await spark.catalog.setCurrentCatalog("not_exist_catalog")
+      }
     }
     await spark.stop()
   }
@@ -63,7 +69,7 @@ struct CatalogTests {
   func setCurrentDatabase() async throws {
     let spark = try await SparkSession.builder.getOrCreate()
     try await spark.catalog.setCurrentDatabase("default")
-    try await #require(throws: Error.self) {
+    try await #require(throws: SparkConnectError.SchemaNotFound) {
       try await spark.catalog.setCurrentDatabase("not_exist_database")
     }
     await spark.stop()
@@ -91,7 +97,7 @@ struct CatalogTests {
     #expect(db.catalog == "spark_catalog")
     #expect(db.description == "default database")
     #expect(db.locationUri.hasSuffix("spark-warehouse"))
-    try await #require(throws: Error.self) {
+    try await #require(throws: SparkConnectError.SchemaNotFound) {
       try await spark.catalog.getDatabase("not_exist_database")
     }
     await spark.stop()
@@ -313,7 +319,7 @@ struct CatalogTests {
       try await spark.catalog.cacheTable(tableName, StorageLevel.MEMORY_ONLY)
     })
 
-    try await #require(throws: Error.self) {
+    try await #require(throws: SparkConnectError.TableOrViewNotFound) {
       try await spark.catalog.cacheTable("not_exist_table")
     }
     await spark.stop()
@@ -330,7 +336,7 @@ struct CatalogTests {
       #expect(try await spark.catalog.isCached(tableName))
     })
 
-    try await #require(throws: Error.self) {
+    try await #require(throws: SparkConnectError.TableOrViewNotFound) {
       try await spark.catalog.isCached("not_exist_table")
     }
     await spark.stop()
@@ -351,7 +357,7 @@ struct CatalogTests {
       #expect(try await spark.catalog.isCached(tableName))
     })
 
-    try await #require(throws: Error.self) {
+    try await #require(throws: SparkConnectError.TableOrViewNotFound) {
       try await spark.catalog.refreshTable("not_exist_table")
     }
     await spark.stop()
@@ -386,7 +392,7 @@ struct CatalogTests {
       #expect(try await spark.catalog.isCached(tableName) == false)
     })
 
-    try await #require(throws: Error.self) {
+    try await #require(throws: SparkConnectError.TableOrViewNotFound) {
       try await spark.catalog.uncacheTable("not_exist_table")
     }
     await spark.stop()