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 3df5a0a [SPARK-58337] Fix `StreamingQuery.exception()` to return
`nil` when no exception exists
3df5a0a is described below
commit 3df5a0a505544a1e9618de7225c8d4dd5e664f1b
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Fri Jul 24 17:15:08 2026 -0700
[SPARK-58337] Fix `StreamingQuery.exception()` to return `nil` when no
exception exists
### What changes were proposed in this pull request?
This PR aims to fix `StreamingQuery.exception()` to return `nil` when the
query has no exception, by checking the `hasExceptionMessage` accessor of the
optional proto field like the PySpark client.
### Why are the changes needed?
Previously, `exception()` always returned a `StreamingQueryException`
filled with empty strings even when there was no exception, although its return
type is optional. PySpark and Scala clients return `None`/`null` in that case.
### Does this PR introduce _any_ user-facing change?
Yes, `StreamingQuery.exception()` now returns `nil` instead of an empty
`StreamingQueryException` when the query has no exception.
### How was this patch tested?
Pass the CIs with a newly added test case.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #442 from dongjoon-hyun/SPARK-58337.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
Sources/SparkConnect/StreamingQuery.swift | 5 +++-
Tests/SparkConnectTests/StreamingQueryTests.swift | 34 +++++++++++++++++++++++
2 files changed, 38 insertions(+), 1 deletion(-)
diff --git a/Sources/SparkConnect/StreamingQuery.swift
b/Sources/SparkConnect/StreamingQuery.swift
index 2153cee..24cf1ed 100644
--- a/Sources/SparkConnect/StreamingQuery.swift
+++ b/Sources/SparkConnect/StreamingQuery.swift
@@ -83,10 +83,13 @@ public actor StreamingQuery: Sendable {
}
/// Returns the ``StreamingQueryException`` if the query was terminated by
an exception.
- /// - Returns: A ``StreamingQueryException``.
+ /// - Returns: A ``StreamingQueryException`` if the query was terminated by
an exception, `nil` otherwise.
public func exception() async throws -> StreamingQueryException? {
let response = try await
executeCommand(StreamingQueryCommand.OneOf_Command.exception(true))
let result = response.first!.streamingQueryCommandResult.exception
+ guard result.hasExceptionMessage else {
+ return nil
+ }
return StreamingQueryException(
exceptionMessage: result.exceptionMessage,
errorClass: result.errorClass,
diff --git a/Tests/SparkConnectTests/StreamingQueryTests.swift
b/Tests/SparkConnectTests/StreamingQueryTests.swift
index cde7802..ae2ad54 100644
--- a/Tests/SparkConnectTests/StreamingQueryTests.swift
+++ b/Tests/SparkConnectTests/StreamingQueryTests.swift
@@ -58,4 +58,38 @@ struct StreamingQueryTests {
}
await spark.stop()
}
+
+ @Test
+ func exception() 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)
+ #expect(try await query.isActive)
+
+ // A query without any exception returns nil.
+ #expect(try await query.exception() == nil)
+
+ try await query.stop()
+ #expect(try await query.isActive == false)
+ #expect(try await query.exception() == nil)
+
+ await spark.stop()
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
