[SPARK-51858] Support SPARK_REMOTE

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

This PR aims to support `SPARK_REMOTE` environment variable.

### Why are the changes needed?

For feature parity.

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

No behavior change. This is an additional way to set Spark connect remote string.

### How was this patch tested?

Pass the CIs with the revised test pipeline.

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

No.

Closes #79 from dongjoon-hyun/SPARK-51858.

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

Author: dongjoon-hyun

.github/workflows/build_and_test.yml

@@ -67,13 +67,15 @@ jobs:
 
   integration-test-linux:
     runs-on: ubuntu-latest
+    env:
+      SPARK_REMOTE: "sc://localhost:15003"
     services:
       spark:
         image: apache/spark:4.0.0-preview2
         env:
           SPARK_NO_DAEMONIZE: 1
         ports:
-          - 15002:15002
+          - 15003:15002
         options: --entrypoint /opt/spark/sbin/start-connect-server.sh
     steps:
     - uses: actions/checkout@v4

Sources/SparkConnect/SparkSession.swift

@@ -221,7 +221,8 @@ public actor SparkSession {
     /// Create a new ``SparkSession``. If `spark.remote` is not given, `sc://localhost:15002` is used.
     /// - Returns: A newly created `SparkSession`.
     func create() async throws -> SparkSession {
-      let session = SparkSession(sparkConf["spark.remote"] ?? "sc://localhost:15002")
+      let remote = ProcessInfo.processInfo.environment["SPARK_REMOTE"] ?? "sc://localhost:15002"
+      let session = SparkSession(sparkConf["spark.remote"] ?? remote)
       let response = try await session.client.connect(session.sessionID)
       await session.setVersion(response.sparkVersion.version)
       let isSuccess = try await session.client.setConf(map: sparkConf)

Tests/SparkConnectTests/BuilderTests.swift

@@ -17,19 +17,23 @@
 // under the License.
 //
 
+import Foundation
 import Testing
 
 @testable import SparkConnect
 
 /// A test suite for `SparkSession.Builder`
 @Suite(.serialized)
 struct BuilderTests {
+  let TEST_REMOTE = ProcessInfo.processInfo.environment["SPARK_REMOTE"] ?? "sc://localhost:15002"
+
   @Test
   func builderDefault() async throws {
+    let url = URL(string: self.TEST_REMOTE)!
     let spark = try await SparkSession.builder.getOrCreate()
     #expect(await spark.client.clientType == "swift")
-    #expect(await spark.client.url.host() == "localhost")
-    #expect(await spark.client.url.port == 15002)
+    #expect(await spark.client.url.host() == url.host())
+    #expect(await spark.client.url.port == url.port)
     await spark.stop()
   }
 

Tests/SparkConnectTests/RuntimeConfTests.swift

@@ -25,9 +25,11 @@ import Testing
 /// A test suite for `RuntimeConf`
 @Suite(.serialized)
 struct RuntimeConfTests {
+  let TEST_REMOTE = ProcessInfo.processInfo.environment["SPARK_REMOTE"] ?? "sc://localhost"
+
   @Test
   func get() async throws {
-    let client = SparkConnectClient(remote: "sc://localhost")
+    let client = SparkConnectClient(remote: TEST_REMOTE)
     _ = try await client.connect(UUID().uuidString)
     let conf = RuntimeConf(client)
 
@@ -42,7 +44,7 @@ struct RuntimeConfTests {
 
   @Test
   func set() async throws {
-    let client = SparkConnectClient(remote: "sc://localhost")
+    let client = SparkConnectClient(remote: TEST_REMOTE)
     _ = try await client.connect(UUID().uuidString)
     let conf = RuntimeConf(client)
     try await conf.set("spark.test.key1", "value1")
@@ -52,7 +54,7 @@ struct RuntimeConfTests {
 
   @Test
   func reset() async throws {
-    let client = SparkConnectClient(remote: "sc://localhost")
+    let client = SparkConnectClient(remote: TEST_REMOTE)
     _ = try await client.connect(UUID().uuidString)
     let conf = RuntimeConf(client)
 
@@ -73,7 +75,7 @@ struct RuntimeConfTests {
 
   @Test
   func getAll() async throws {
-    let client = SparkConnectClient(remote: "sc://localhost")
+    let client = SparkConnectClient(remote: TEST_REMOTE)
     _ = try await client.connect(UUID().uuidString)
     let conf = RuntimeConf(client)
     let map = try await conf.getAll()

Tests/SparkConnectTests/SparkConnectClientTests.swift

@@ -25,9 +25,11 @@ import Testing
 /// A test suite for `SparkConnectClient`
 @Suite(.serialized)
 struct SparkConnectClientTests {
+  let TEST_REMOTE = ProcessInfo.processInfo.environment["SPARK_REMOTE"] ?? "sc://localhost"
+
   @Test
   func createAndStop() async throws {
-    let client = SparkConnectClient(remote: "sc://localhost")
+    let client = SparkConnectClient(remote: TEST_REMOTE)
     await client.stop()
   }
 
@@ -44,7 +46,7 @@ struct SparkConnectClientTests {
 
   @Test
   func connectWithInvalidUUID() async throws {
-    let client = SparkConnectClient(remote: "sc://localhost")
+    let client = SparkConnectClient(remote: TEST_REMOTE)
     try await #require(throws: SparkConnectError.InvalidSessionIDException) {
       let _ = try await client.connect("not-a-uuid-format")
     }
@@ -53,14 +55,14 @@ struct SparkConnectClientTests {
 
   @Test
   func connect() async throws {
-    let client = SparkConnectClient(remote: "sc://localhost")
+    let client = SparkConnectClient(remote: TEST_REMOTE)
     let _ = try await client.connect(UUID().uuidString)
     await client.stop()
   }
 
   @Test
   func tags() async throws {
-    let client = SparkConnectClient(remote: "sc://localhost")
+    let client = SparkConnectClient(remote: TEST_REMOTE)
     let _ = try await client.connect(UUID().uuidString)
     let plan = await client.getPlanRange(0, 1, 1)
 
@@ -76,7 +78,7 @@ struct SparkConnectClientTests {
 
   @Test
   func ddlParse() async throws {
-    let client = SparkConnectClient(remote: "sc://localhost")
+    let client = SparkConnectClient(remote: TEST_REMOTE)
     let _ = try await client.connect(UUID().uuidString)
     #expect(try await client.ddlParse("a int").simpleString == "struct<a:int>")
     await client.stop()
@@ -85,7 +87,7 @@ struct SparkConnectClientTests {
 #if !os(Linux) // TODO: Enable this with the offical Spark 4 docker image
   @Test
   func jsonToDdl() async throws {
-    let client = SparkConnectClient(remote: "sc://localhost")
+    let client = SparkConnectClient(remote: TEST_REMOTE)
     let _ = try await client.connect(UUID().uuidString)
     let json =
       #"{"type":"struct","fields":[{"name":"id","type":"long","nullable":false,"metadata":{}}]}"#