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

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,7 +37,7 @@ 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 #require(throws: SparkConnectError.CatalogNotFound) {
       try await spark.catalog.setCurrentCatalog("not_exist_catalog")
     }
     await spark.stop()
@@ -63,7 +63,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 +91,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 +313,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 +330,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 +351,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 +386,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()