[SPARK-52317] Identify InvalidTypeException in SparkConnectClient

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

This PR aims to identify `InvalidTypeException` in `SparkConnectClient`.

### Why are the changes needed?

To centralize the `InvalidTypeException` handling into a single place. Previously, three actors do the same error handling logic redundantly.
- `DataFrame`
- `DataFrameReader`
- `DataStreamReader`

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

No.

### How was this patch tested?

Pass the CIs.

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

No.

Closes #176 from dongjoon-hyun/SPARK-52317.

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

Author: dongjoon-hyun

Sources/SparkConnect/DataFrame.swift

@@ -489,13 +489,8 @@ public actor DataFrame: Sendable {
   /// - Parameter schema: The given schema.
   /// - Returns: A ``DataFrame`` with the given schema.
   public func to(_ schema: String) async throws -> DataFrame {
-    // Validate by parsing.
-    do {
-      let dataType = try await sparkSession.client.ddlParse(schema)
-      return DataFrame(spark: self.spark, plan: SparkConnectClient.getToSchema(self.plan.root, dataType))
-    } catch {
-      throw SparkConnectError.InvalidTypeException
-    }
+    let dataType = try await sparkSession.client.ddlParse(schema)
+    return DataFrame(spark: self.spark, plan: SparkConnectClient.getToSchema(self.plan.root, dataType))
   }
 
   /// Returns the content of the Dataset as a Dataset of JSON strings.

Sources/SparkConnect/DataFrameReader.swift

@@ -123,12 +123,7 @@ public actor DataFrameReader: Sendable {
   /// - Returns: A ``DataFrameReader``.
   @discardableResult
   public func schema(_ schema: String) async throws -> DataFrameReader {
-    // Validate by parsing.
-    do {
-      try await sparkSession.client.ddlParse(schema)
-    } catch {
-      throw SparkConnectError.InvalidTypeException
-    }
+    try await sparkSession.client.ddlParse(schema)
     self.userSpecifiedSchemaDDL = schema
     return self
   }

Sources/SparkConnect/DataStreamReader.swift

@@ -50,12 +50,7 @@ public actor DataStreamReader: Sendable {
   /// - Returns: A ``DataStreamReader``.
   @discardableResult
   public func schema(_ schema: String) async throws -> DataStreamReader {
-    // Validate by parsing.
-    do {
-      try await sparkSession.client.ddlParse(schema)
-    } catch {
-      throw SparkConnectError.InvalidTypeException
-    }
+    try await sparkSession.client.ddlParse(schema)
     self.userSpecifiedSchemaDDL = schema
     return self
   }

Sources/SparkConnect/SparkConnectClient.swift

@@ -781,8 +781,17 @@ public actor SparkConnectClient {
         ddlParse.ddlString = ddlString
         return OneOf_Analyze.ddlParse(ddlParse)
       })
-      let response = try await service.analyzePlan(request)
-      return response.ddlParse.parsed
+      do {
+        let response = try await service.analyzePlan(request)
+        return response.ddlParse.parsed
+      } catch let error as RPCError where error.code == .internalError {
+        switch error.message {
+        case let m where m.contains("UNSUPPORTED_DATATYPE") || m.contains("INVALID_IDENTIFIER"):
+          throw SparkConnectError.InvalidTypeException
+        default:
+          throw error
+        }
+      }
     }
   }