[SPARK-52361] Support executeCommand in SparkSession

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

This PR aims to support `executeCommand` API of `SparkSession`.

In addition, `DataSourceNotFound` is added to `SparkConnectError`.

### Why are the changes needed?

For feature parity.

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

No behavior change.

### How was this patch tested?

Pass the CIs.

Since `Apache Spark` distribution has no built-in `ExternalCommandRunner`. I crafted the following for manual testing. In this PR, the added test case only checks the API invocation part.

```bash
$ cat MyCommand.java
public class MyCommand implements org.apache.spark.sql.connector.ExternalCommandRunner {
  Override
  public String[] executeCommand(String command, org.apache.spark.sql.util.CaseInsensitiveStringMap options) {
    String[] result = {"Hello", command};
    return result;
  }
}
```

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

No.

Closes apache#187 from dongjoon-hyun/SPARK-52361.

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

Author: dongjoon-hyun

Sources/SparkConnect/SparkConnectClient.swift

@@ -141,6 +141,8 @@ public actor SparkConnectClient {
           throw SparkConnectError.TableOrViewNotFound
         case let m where m.contains("Invalid view name:"):
           throw SparkConnectError.InvalidViewName
+        case let m where m.contains("DATA_SOURCE_NOT_FOUND"):
+          throw SparkConnectError.DataSourceNotFound
         default:
           throw error
         }
@@ -715,6 +717,24 @@ public actor SparkConnectClient {
     return result
   }
 
+  func getExecuteExternalCommand(
+    _ runner: String,
+    _ command: String,
+    _ options: [String: String]
+  ) async throws -> Plan {
+    var executeExternalCommand = Spark_Connect_ExecuteExternalCommand()
+    executeExternalCommand.runner = runner
+    executeExternalCommand.command = command
+    executeExternalCommand.options = options
+    var command = Command()
+    command.commandType = .executeExternalCommand(executeExternalCommand)
+    let response = try await execute(self.sessionID!, command)
+    let relation = response.first!.sqlCommandResult.relation
+    var plan = Plan()
+    plan.opType = .root(relation)
+    return plan
+  }
+
   /// Add a tag to be assigned to all the operations started by this thread in this session.
   /// - Parameter tag: The tag to be added. Cannot contain ',' (comma) character or be an empty string.
   public func addTag(tag: String) throws {

Sources/SparkConnect/SparkConnectError.swift

@@ -21,6 +21,7 @@
 public enum SparkConnectError: Error {
   case CatalogNotFound
   case ColumnNotFound
+  case DataSourceNotFound
   case InvalidArgument
   case InvalidSessionID
   case InvalidType

Sources/SparkConnect/SparkSession.swift

@@ -267,6 +267,13 @@ public actor SparkSession {
     return await read.table(tableName)
   }
 
+  public func executeCommand(_ runner: String, _ command: String, _ options: [String: String])
+    async throws -> DataFrame
+  {
+    let plan = try await self.client.getExecuteExternalCommand(runner, command, options)
+    return DataFrame(spark: self, plan: plan)
+  }
+
   /// Executes a code block and prints the execution time.
   ///
   /// This utility method is useful for performance testing and optimization.

Tests/SparkConnectTests/SparkSessionTests.swift

@@ -141,6 +141,18 @@ struct SparkSessionTests {
     }
     await spark.stop()
   }
+
+  @Test
+  func executeCommand() async throws {
+    await SparkSession.builder.clear()
+    let spark = try await SparkSession.builder.getOrCreate()
+    if await spark.version.starts(with: "4.") {
+      await #expect(throws: SparkConnectError.DataSourceNotFound) {
+        try await spark.executeCommand("runner", "command", [:]).show()
+      }
+    }
+    await spark.stop()
+  }
 #endif
 
   @Test