[SPARK-51793] Support ddlParse and jsonToDdl in SparkConnectClient

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

This PR aims to support `ddlParse` and `jsonToDdl` in `SparkConnectClient`.

### Why are the changes needed?

For feature parity.

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

No, these will be used internally.

### How was this patch tested?

Pass the CIs.

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

No.

Closes #57 from dongjoon-hyun/SPARK-51793.

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

Author: dongjoon-hyun

Sources/SparkConnect/SparkConnectClient.swift

@@ -438,4 +438,46 @@ public actor SparkConnectClient {
   public func clearTags() {
     tags.removeAll()
   }
+
+  /// Parse a DDL string to ``Spark_Connect_DataType`` instance.
+  /// - Parameter ddlString: A string to parse.
+  /// - Returns: A ``Spark_Connect_DataType`` instance.
+  func ddlParse(_ ddlString: String) async throws -> Spark_Connect_DataType {
+    try await withGRPCClient(
+      transport: .http2NIOPosix(
+        target: .dns(host: self.host, port: self.port),
+        transportSecurity: .plaintext
+      )
+    ) { client in
+      let service = SparkConnectService.Client(wrapping: client)
+      let request = analyze(self.sessionID!, {
+        var ddlParse = AnalyzePlanRequest.DDLParse()
+        ddlParse.ddlString = ddlString
+        return OneOf_Analyze.ddlParse(ddlParse)
+      })
+      let response = try await service.analyzePlan(request)
+      return response.ddlParse.parsed
+    }
+  }
+
+  /// Convert an JSON string to a DDL string.
+  /// - Parameter jsonString: A JSON string.
+  /// - Returns: A DDL string.
+  func jsonToDdl(_ jsonString: String) async throws -> String {
+    try await withGRPCClient(
+      transport: .http2NIOPosix(
+        target: .dns(host: self.host, port: self.port),
+        transportSecurity: .plaintext
+      )
+    ) { client in
+      let service = SparkConnectService.Client(wrapping: client)
+      let request = analyze(self.sessionID!, {
+        var jsonToDDL = AnalyzePlanRequest.JsonToDDL()
+        jsonToDDL.jsonString = jsonString
+        return OneOf_Analyze.jsonToDdl(jsonToDDL)
+      })
+      let response = try await service.analyzePlan(request)
+      return response.jsonToDdl.ddlString
+    }
+  }
 }

Tests/SparkConnectTests/SparkConnectClientTests.swift

@@ -62,4 +62,24 @@ struct SparkConnectClientTests {
     #expect(await client.getExecutePlanRequest(plan).tags.isEmpty)
     await client.stop()
   }
+
+  @Test
+  func ddlParse() async throws {
+    let client = SparkConnectClient(remote: "sc://localhost", user: "test")
+    let _ = try await client.connect(UUID().uuidString)
+    #expect(try await client.ddlParse("a int").simpleString == "struct<a:int>")
+    await client.stop()
+  }
+
+#if !os(Linux) // TODO: Enable this with the offical Spark 4 docker image
+  @Test
+  func jsonToDdl() async throws {
+    let client = SparkConnectClient(remote: "sc://localhost", user: "test")
+    let _ = try await client.connect(UUID().uuidString)
+    let json =
+      #"{"type":"struct","fields":[{"name":"id","type":"long","nullable":false,"metadata":{}}]}"#
+    #expect(try await client.jsonToDdl(json) == "id BIGINT NOT NULL")
+    await client.stop()
+  }
+#endif
 }