This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-connect-swift.git
The following commit(s) were added to refs/heads/main by this push:
new dac41be [SPARK-60060] Make StreamingQueryStatus and
StreamingQueryException properties public
dac41be is described below
commit dac41bef8197b6869e46ed89b968b02040359eae
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Oct 7 16:13:27 2026 -0700
[SPARK-60060] Make StreamingQueryStatus and StreamingQueryException
properties public
### What changes were proposed in this pull request?
This PR aims to make the stored properties of `StreamingQueryStatus`
(`statusMessage`, `isDataAvailable`, `isTriggerActive`, `isActive`) and
`StreamingQueryException` (`exceptionMessage`, `errorClass`, `stackTrace`)
`public`, with brief doc comments.
### Why are the changes needed?
`StreamingQueryStatus` and `StreamingQueryException` are `public` structs
returned by the public APIs `StreamingQuery.status()` and
`StreamingQuery.exception()`, but their properties were `internal`. As a
result, applications outside the `SparkConnect` module could not read any of
the returned fields. For example, the following code failed to compile.
```swift
let status = try await query.status()
print(status.isActive) // error: 'isActive' is inaccessible due to
'internal' protection level
```
This aligns with PySpark, where `StreamingQuery.status` and
`StreamingQuery.exception()` expose this information to users.
### Does this PR introduce _any_ user-facing change?
Yes. Users can now read the fields of `StreamingQueryStatus` and
`StreamingQueryException`. This is an additive API change only.
### How was this patch tested?
Pass the CIs with the newly added `StreamingQueryTests.status` test case.
In addition, I manually verified that an out-of-module Swift package
depending on `SparkConnect` fails to compile before this PR and compiles after
this PR when it accesses these properties.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #587 from dongjoon-hyun/SPARK-60060.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
Sources/SparkConnect/StreamingQuery.swift | 21 ++++++++-----
Tests/SparkConnectTests/StreamingQueryTests.swift | 37 +++++++++++++++++++++++
2 files changed, 51 insertions(+), 7 deletions(-)
diff --git a/Sources/SparkConnect/StreamingQuery.swift
b/Sources/SparkConnect/StreamingQuery.swift
index f7cb5d8..dd833e1 100644
--- a/Sources/SparkConnect/StreamingQuery.swift
+++ b/Sources/SparkConnect/StreamingQuery.swift
@@ -23,16 +23,23 @@ import Foundation
#endif
public struct StreamingQueryException: Sendable {
- let exceptionMessage: String
- let errorClass: String
- let stackTrace: String
+ /// The exception message as a string.
+ public let exceptionMessage: String
+ /// The error class of the exception.
+ public let errorClass: String
+ /// The stack trace of the exception.
+ public let stackTrace: String
}
public struct StreamingQueryStatus: Sendable {
- let statusMessage: String
- let isDataAvailable: Bool
- let isTriggerActive: Bool
- let isActive: Bool
+ /// A human-readable description of what the query is currently doing.
+ public let statusMessage: String
+ /// Whether there is new data available to be processed.
+ public let isDataAvailable: Bool
+ /// Whether the trigger is actively firing.
+ public let isTriggerActive: Bool
+ /// Whether the query is actively running.
+ public let isActive: Bool
}
/// A handle to a query that is executing continuously in the background as
new data arrives.
diff --git a/Tests/SparkConnectTests/StreamingQueryTests.swift
b/Tests/SparkConnectTests/StreamingQueryTests.swift
index 69be862..0d37d4b 100644
--- a/Tests/SparkConnectTests/StreamingQueryTests.swift
+++ b/Tests/SparkConnectTests/StreamingQueryTests.swift
@@ -93,6 +93,43 @@ struct StreamingQueryTests {
await spark.stop()
}
+ @Test
+ func status() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+
+ // Prepare directories
+ let input = "/tmp/input-" + UUID().uuidString
+ let checkpoint = "/tmp/checkpoint-" + UUID().uuidString
+ let output = "/tmp/output-" + UUID().uuidString
+ try await spark.range(2025).write.orc(input)
+
+ // Start a streaming query
+ let query =
+ try await spark
+ .readStream
+ .schema("id LONG")
+ .orc(input)
+ .writeStream
+ .option("checkpointLocation", checkpoint)
+ .outputMode("append")
+ .format("orc")
+ .trigger(Trigger.ProcessingTime(1000))
+ .start(output)
+
+ let status = try await query.status()
+ #expect(status.isActive)
+ #expect(status.statusMessage.isEmpty == false)
+
+ try await query.stop()
+ let stoppedStatus = try await query.status()
+ #expect(stoppedStatus.isActive == false)
+ #expect(stoppedStatus.isDataAvailable == false)
+ #expect(stoppedStatus.isTriggerActive == false)
+ #expect(stoppedStatus.statusMessage == "Stopped")
+
+ await spark.stop()
+ }
+
@Test
func progress() async throws {
let spark = try await SparkSession.builder.getOrCreate()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
