[SPARK-51992] Support interrupt(Tag|Operation|All) in SparkSession

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

This PR aims to support `interrupt(Tag|Operation|All)` API of `SparkSession`.

### Why are the changes needed?

For feature parity.

### 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 #108 from dongjoon-hyun/SPARK-51992.

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

Author: dongjoon-hyun

Sources/SparkConnect/Documentation.docc/SparkSession.md

@@ -32,15 +32,13 @@ let csvDf = spark.read.csv("path/to/file.csv")
 
 ### Creating Sessions
 
-- ``builder()``
-- ``active()``
+- ``builder``
 - ``stop()``
 
-### DataFrame Operations  
+### DataFrame Operations
 
-- ``range(_:_:)``
+- ``range(_:_:_:)``
 - ``sql(_:)``
-- ``createDataFrame(_:_:)``
 
 ### Data I/O
 
@@ -53,3 +51,13 @@ let csvDf = spark.read.csv("path/to/file.csv")
 ### Catalog Operations
 
 - ``catalog``
+
+### Managing Operations
+
+- ``addTag(_:)``
+- ``removeTag(_:)``
+- ``getTags()``
+- ``clearTags()``
+- ``interruptAll()``
+- ``interruptTag(_:)``
+- ``interruptOperation(_:)``

Sources/SparkConnect/SparkConnectClient.swift

@@ -559,6 +559,50 @@ public actor SparkConnectClient {
     tags.removeAll()
   }
 
+  public func interruptAll() async throws -> [String] {
+    var request = Spark_Connect_InterruptRequest()
+    request.sessionID = self.sessionID!
+    request.userContext = self.userContext
+    request.clientType = self.clientType
+    request.interruptType = .all
+
+    return try await withGPRC { client in
+      let service = Spark_Connect_SparkConnectService.Client(wrapping: client)
+      let response = try await service.interrupt(request)
+      return response.interruptedIds
+    }
+  }
+
+  public func interruptTag(_ tag: String) async throws -> [String] {
+    var request = Spark_Connect_InterruptRequest()
+    request.sessionID = self.sessionID!
+    request.userContext = self.userContext
+    request.clientType = self.clientType
+    request.interruptType = .tag
+    request.operationTag = tag
+
+    return try await withGPRC { client in
+      let service = Spark_Connect_SparkConnectService.Client(wrapping: client)
+      let response = try await service.interrupt(request)
+      return response.interruptedIds
+    }
+  }
+
+  public func interruptOperation(_ operationId: String) async throws -> [String] {
+    var request = Spark_Connect_InterruptRequest()
+    request.sessionID = self.sessionID!
+    request.userContext = self.userContext
+    request.clientType = self.clientType
+    request.interruptType = .operationID
+    request.operationID = operationId
+
+    return try await withGPRC { client in
+      let service = Spark_Connect_SparkConnectService.Client(wrapping: client)
+      let response = try await service.interrupt(request)
+      return response.interruptedIds
+    }
+  }
+
   /// Parse a DDL string to ``Spark_Connect_DataType`` instance.
   /// - Parameter ddlString: A string to parse.
   /// - Returns: A ``Spark_Connect_DataType`` instance.

Sources/SparkConnect/SparkSession.swift

@@ -314,6 +314,27 @@ public actor SparkSession {
     await client.clearTags()
   }
 
+  /// Request to interrupt all currently running operations of this session.
+  /// - Returns: Sequence of operation IDs requested to be interrupted.
+  @discardableResult
+  public func interruptAll() async throws -> [String] {
+    return try await client.interruptAll()
+  }
+
+  /// Request to interrupt all currently running operations of this session with the given job tag.
+  /// - Returns: Sequence of operation IDs requested to be interrupted.
+  @discardableResult
+  public func interruptTag(_ tag: String) async throws -> [String] {
+    return try await client.interruptTag(tag)
+  }
+
+  /// Request to interrupt an operation of this session, given its operation ID.
+  /// - Returns: Sequence of operation IDs requested to be interrupted.
+  @discardableResult
+  public func interruptOperation(_ operationId: String) async throws -> [String] {
+    return try await client.interruptOperation(operationId)
+  }
+
   func sameSemantics(_ plan: Plan, _ otherPlan: Plan) async throws -> Bool {
     return try await client.sameSemantics(plan, otherPlan)
   }

Tests/SparkConnectTests/SparkSessionTests.swift

@@ -136,4 +136,25 @@ struct SparkSessionTests {
     }
     await spark.stop()
   }
+
+  @Test
+  func interruptAll() async throws {
+    let spark = try await SparkSession.builder.getOrCreate()
+    #expect(try await spark.interruptAll() == [])
+    await spark.stop()
+  }
+
+  @Test
+  func interruptTag() async throws {
+    let spark = try await SparkSession.builder.getOrCreate()
+    #expect(try await spark.interruptTag("etl") == [])
+    await spark.stop()
+  }
+
+  @Test
+  func interruptOperation() async throws {
+    let spark = try await SparkSession.builder.getOrCreate()
+    #expect(try await spark.interruptOperation("id") == [])
+    await spark.stop()
+  }
 }