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 1650218 [SPARK-58653] Support typed `StreamingQueryProgress` in
`(last|recent)Progress`
1650218 is described below
commit 1650218bf4c2dff094f53ef4b618d08e9e2f29f7
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Fri Aug 7 13:28:21 2026 -0700
[SPARK-58653] Support typed `StreamingQueryProgress` in
`(last|recent)Progress`
### What changes were proposed in this pull request?
This PR aims to support typed `StreamingQueryProgress` in
`StreamingQuery.lastProgress` and `recentProgress`.
- `lastProgress` now returns `StreamingQueryProgress?` (previously
`String?`) and `recentProgress` returns `[StreamingQueryProgress]` (previously
`[String]`).
- The unused internal structs in `StreamingQueryManager.swift` are moved to
a new file `StreamingQueryProgress.swift` and promoted to public `Sendable` and
`Codable` types: `StreamingQueryProgress` (renamed from the typo
`StreamingQueryProcess`), `SourceProgress`, `SinkProgress`, and
`StateOperatorProgress`.
- JSON decoding is lenient like the Scala/PySpark clients: unknown fields
are ignored, missing fields fall back to defaults, and offset fields preserve
arbitrary JSON as strings. The original JSON remains accessible via the `json`
property.
### Why are the changes needed?
For feature parity with the Scala and PySpark clients which return typed
`StreamingQueryProgress` objects.
### Does this PR introduce _any_ user-facing change?
Yes, `lastProgress` and `recentProgress` now return typed values instead of
raw JSON strings. The raw JSON is still available via the `json` property.
### How was this patch tested?
Pass the CIs with the newly added unit tests.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #491 from dongjoon-hyun/SPARK-58653.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../Documentation.docc/SparkConnect.md | 4 +
Sources/SparkConnect/StreamingQuery.swift | 11 +-
Sources/SparkConnect/StreamingQueryManager.swift | 78 ------
Sources/SparkConnect/StreamingQueryProgress.swift | 292 +++++++++++++++++++++
.../StreamingQueryProgressTests.swift | 162 ++++++++++++
Tests/SparkConnectTests/StreamingQueryTests.swift | 47 ++++
6 files changed, 512 insertions(+), 82 deletions(-)
diff --git a/Sources/SparkConnect/Documentation.docc/SparkConnect.md
b/Sources/SparkConnect/Documentation.docc/SparkConnect.md
index cfd40fb..815bba8 100644
--- a/Sources/SparkConnect/Documentation.docc/SparkConnect.md
+++ b/Sources/SparkConnect/Documentation.docc/SparkConnect.md
@@ -59,3 +59,7 @@ SparkConnect is a modern Swift library that provides a native
interface to Apach
- ``DataStreamWriter``
- ``StreamingQuery``
- ``StreamingQueryManager``
+- ``StreamingQueryProgress``
+- ``SourceProgress``
+- ``SinkProgress``
+- ``StateOperatorProgress``
diff --git a/Sources/SparkConnect/StreamingQuery.swift
b/Sources/SparkConnect/StreamingQuery.swift
index 24cf1ed..5db54b3 100644
--- a/Sources/SparkConnect/StreamingQuery.swift
+++ b/Sources/SparkConnect/StreamingQuery.swift
@@ -113,22 +113,25 @@ public actor StreamingQuery: Sendable {
/// Returns an array of the most recent ``StreamingQueryProgress`` updates
for this query.
/// The number of progress updates retained for each stream is configured by
Spark session
/// configuration `spark.sql.streaming.numRecentProgressUpdates`.
- public var recentProgress: [String] {
+ public var recentProgress: [StreamingQueryProgress] {
get async throws {
let response = try await executeCommand(
StreamingQueryCommand.OneOf_Command.recentProgress(true))
let result = response.first!.streamingQueryCommandResult.recentProgress
- return result.recentProgressJson
+ return try result.recentProgressJson.map { try
StreamingQueryProgress.fromJson($0) }
}
}
/// Returns the most recent ``StreamingQueryProgress`` update of this
streaming query.
- public var lastProgress: String? {
+ public var lastProgress: StreamingQueryProgress? {
get async throws {
let response = try await executeCommand(
StreamingQueryCommand.OneOf_Command.lastProgress(true))
let result = response.first!.streamingQueryCommandResult.recentProgress
- return result.recentProgressJson.first
+ guard let json = result.recentProgressJson.first else {
+ return nil
+ }
+ return try StreamingQueryProgress.fromJson(json)
}
}
diff --git a/Sources/SparkConnect/StreamingQueryManager.swift
b/Sources/SparkConnect/StreamingQueryManager.swift
index 441cd19..bdac470 100644
--- a/Sources/SparkConnect/StreamingQueryManager.swift
+++ b/Sources/SparkConnect/StreamingQueryManager.swift
@@ -22,84 +22,6 @@ import FoundationEssentials
import Foundation
#endif
-/// Information about progress made for a source in the execution of a
``StreamingQuery``
-/// during a trigger. See ``StreamingQueryProgress`` for more information.
-public struct SourceProgress: Sendable {
- let description: String
- let startOffset: String
- let endOffset: String
- let latestOffset: String
- let numInputRows: Int64
- let inputRowsPerSecond: Double
- let processedRowsPerSecond: Double
- let metrics: [String: String] = [:]
-}
-
-/// Information about progress made for a sink in the execution of a
``StreamingQuery``
-/// during a trigger. See ``StreamingQueryProgress`` for more information.
-public struct SinkProgress: Sendable {
- let description: String
- let numOutputRows: Int64
- let metrics: [String: String] = [:]
-
- init(_ description: String) {
- self.description = description
- self.numOutputRows = -1
- }
-
- init(_ description: String, _ numOutputRows: Int64) {
- self.description = description
- self.numOutputRows = numOutputRows
- }
-}
-
-/// Information about updates made to stateful operators in a
``StreamingQuery``
-/// during a trigger. See ``StreamingQueryProgress`` for more information.
-public struct StateOperatorProgress: Sendable {
- let operatorName: String
- let numRowsTotal: Int64
- let numRowsUpdated: Int64
- let allUpdatesTimeMs: Int64
- let numRowsRemoved: Int64
- let allRemovalsTimeMs: Int64
- let commitTimeMs: Int64
- let memoryUsedBytes: Int64
- let numRowsDroppedByWatermark: Int64
- let numShufflePartitions: Int64
- let numStateStoreInstances: Int64
- let customMetrics: [String: Int64]
-}
-
-/// Information about progress made in the execution of a ``StreamingQuery``
-/// during a trigger. Each event relates to processing done for a single
trigger of
-/// the streaming query. Events are emitted even when no new data is available
to be processed.
-public struct StreamingQueryProcess {
- let id: UUID
- let runId: UUID
- let name: String
- let timestamp: String
- let batchId: Int64
- let batchDuration: Int64
- let durationMs: [String: Int64]
- let eventTime: [String: String]
- let stateOperators: [StateOperatorProgress]
- let sources: [SourceProgress]
- let sink: SinkProgress
- let observedMetrics: [String: Row]
-
- func numInputRows() -> Int64 {
- return sources.map { $0.numInputRows }.reduce(0, +)
- }
-
- func inputRowsPerSecond() -> Double {
- return sources.map { $0.inputRowsPerSecond }.reduce(0, +)
- }
-
- func processedRowsPerSecond() -> Double {
- return sources.map { $0.processedRowsPerSecond }.reduce(0, +)
- }
-}
-
/// A class to manage all the ``StreamingQuery``s active in a ``SparkSession``.
public actor StreamingQueryManager {
let sparkSession: SparkSession
diff --git a/Sources/SparkConnect/StreamingQueryProgress.swift
b/Sources/SparkConnect/StreamingQueryProgress.swift
new file mode 100644
index 0000000..ec1c4fa
--- /dev/null
+++ b/Sources/SparkConnect/StreamingQueryProgress.swift
@@ -0,0 +1,292 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+#if canImport(FoundationEssentials)
+import FoundationEssentials
+#else
+import Foundation
+#endif
+
+/// A JSON value used to preserve arbitrary JSON fragments (e.g. source
offsets) as strings.
+private enum JSONValue: Decodable, Sendable {
+ case null
+ case bool(Bool)
+ case int(Int64)
+ case double(Double)
+ case string(String)
+ case array([JSONValue])
+ case object([String: JSONValue])
+
+ init(from decoder: any Decoder) throws {
+ let container = try decoder.singleValueContainer()
+ if container.decodeNil() {
+ self = .null
+ } else if let value = try? container.decode(Bool.self) {
+ self = .bool(value)
+ } else if let value = try? container.decode(Int64.self) {
+ self = .int(value)
+ } else if let value = try? container.decode(Double.self) {
+ self = .double(value)
+ } else if let value = try? container.decode(String.self) {
+ self = .string(value)
+ } else if let value = try? container.decode([JSONValue].self) {
+ self = .array(value)
+ } else {
+ self = .object(try container.decode([String: JSONValue].self))
+ }
+ }
+
+ /// A compact JSON text of this value with alphabetically sorted object keys.
+ var compactString: String {
+ switch self {
+ case .null: return "null"
+ case .bool(let value): return value ? "true" : "false"
+ case .int(let value): return String(value)
+ case .double(let value): return String(value)
+ case .string(let value): return JSONValue.escape(value)
+ case .array(let values):
+ return "[" + values.map { $0.compactString }.joined(separator: ",") + "]"
+ case .object(let values):
+ let members = values.keys.sorted().map {
"\(JSONValue.escape($0)):\(values[$0]!.compactString)" }
+ return "{" + members.joined(separator: ",") + "}"
+ }
+ }
+
+ /// A string form following the Spark clients' offset handling; a JSON
string is returned as-is
+ /// and any other JSON value is converted to its compact JSON text.
+ var offsetString: String {
+ if case .string(let value) = self {
+ return value
+ }
+ return compactString
+ }
+
+ private static func escape(_ string: String) -> String {
+ var result = "\""
+ for scalar in string.unicodeScalars {
+ switch scalar {
+ case "\"": result += "\\\""
+ case "\\": result += "\\\\"
+ case "\n": result += "\\n"
+ case "\r": result += "\\r"
+ case "\t": result += "\\t"
+ default:
+ if scalar.value < 0x20 {
+ let hex = String(scalar.value, radix: 16)
+ result += "\\u" + String(repeating: "0", count: 4 - hex.count) + hex
+ } else {
+ result.unicodeScalars.append(scalar)
+ }
+ }
+ }
+ return result + "\""
+ }
+}
+
+/// Decodes a `Double` that servers may emit as a quoted string like `"NaN"`
or omit entirely.
+private func decodeLenientDouble<Key: CodingKey>(
+ _ container: KeyedDecodingContainer<Key>, _ key: Key
+) -> Double {
+ if let value = try? container.decode(Double.self, forKey: key) {
+ return value
+ }
+ if let value = try? container.decode(String.self, forKey: key), let double =
Double(value) {
+ return double
+ }
+ return .nan
+}
+
+/// Information about progress made for a source in the execution of a
``StreamingQuery``
+/// during a trigger. See ``StreamingQueryProgress`` for more information.
+public struct SourceProgress: Sendable, Codable {
+ /// Description of the source.
+ public let description: String
+ /// The starting offset for data being read, as a JSON string.
+ public let startOffset: String
+ /// The ending offset for data being read, as a JSON string.
+ public let endOffset: String
+ /// The latest offset from this source, as a JSON string.
+ public let latestOffset: String
+ /// The number of records read from this source.
+ public let numInputRows: Int64
+ /// The rate at which data is arriving from this source, `nan` if not
available.
+ public let inputRowsPerSecond: Double
+ /// The rate at which data from this source is being processed by Spark,
`nan` if not available.
+ public let processedRowsPerSecond: Double
+ /// Custom metrics of this source.
+ public let metrics: [String: String]
+
+ public init(from decoder: any Decoder) throws {
+ let container = try decoder.container(keyedBy: CodingKeys.self)
+ self.description = try container.decodeIfPresent(String.self, forKey:
.description) ?? ""
+ self.startOffset =
+ container.contains(.startOffset)
+ ? try container.decode(JSONValue.self, forKey:
.startOffset).offsetString : ""
+ self.endOffset =
+ container.contains(.endOffset)
+ ? try container.decode(JSONValue.self, forKey: .endOffset).offsetString
: ""
+ self.latestOffset =
+ container.contains(.latestOffset)
+ ? try container.decode(JSONValue.self, forKey:
.latestOffset).offsetString : ""
+ self.numInputRows = try container.decodeIfPresent(Int64.self, forKey:
.numInputRows) ?? 0
+ self.inputRowsPerSecond = decodeLenientDouble(container,
.inputRowsPerSecond)
+ self.processedRowsPerSecond = decodeLenientDouble(container,
.processedRowsPerSecond)
+ self.metrics = try container.decodeIfPresent([String: String].self,
forKey: .metrics) ?? [:]
+ }
+}
+
+/// Information about progress made for a sink in the execution of a
``StreamingQuery``
+/// during a trigger. See ``StreamingQueryProgress`` for more information.
+public struct SinkProgress: Sendable, Codable {
+ /// Description of the sink.
+ public let description: String
+ /// Number of rows written to the sink, `-1` if not available.
+ public let numOutputRows: Int64
+ /// Custom metrics of this sink.
+ public let metrics: [String: String]
+
+ public init(from decoder: any Decoder) throws {
+ let container = try decoder.container(keyedBy: CodingKeys.self)
+ self.description = try container.decodeIfPresent(String.self, forKey:
.description) ?? ""
+ self.numOutputRows = try container.decodeIfPresent(Int64.self, forKey:
.numOutputRows) ?? -1
+ self.metrics = try container.decodeIfPresent([String: String].self,
forKey: .metrics) ?? [:]
+ }
+}
+
+/// Information about updates made to stateful operators in a
``StreamingQuery``
+/// during a trigger. See ``StreamingQueryProgress`` for more information.
+public struct StateOperatorProgress: Sendable, Codable {
+ public let operatorName: String
+ public let numRowsTotal: Int64
+ public let numRowsUpdated: Int64
+ public let allUpdatesTimeMs: Int64
+ public let numRowsRemoved: Int64
+ public let allRemovalsTimeMs: Int64
+ public let commitTimeMs: Int64
+ public let memoryUsedBytes: Int64
+ public let numRowsDroppedByWatermark: Int64
+ public let numShufflePartitions: Int64
+ public let numStateStoreInstances: Int64
+ public let customMetrics: [String: Int64]
+
+ public init(from decoder: any Decoder) throws {
+ let container = try decoder.container(keyedBy: CodingKeys.self)
+ self.operatorName = try container.decodeIfPresent(String.self, forKey:
.operatorName) ?? ""
+ self.numRowsTotal = try container.decodeIfPresent(Int64.self, forKey:
.numRowsTotal) ?? 0
+ self.numRowsUpdated = try container.decodeIfPresent(Int64.self, forKey:
.numRowsUpdated) ?? 0
+ self.allUpdatesTimeMs =
+ try container.decodeIfPresent(Int64.self, forKey: .allUpdatesTimeMs) ?? 0
+ self.numRowsRemoved = try container.decodeIfPresent(Int64.self, forKey:
.numRowsRemoved) ?? 0
+ self.allRemovalsTimeMs =
+ try container.decodeIfPresent(Int64.self, forKey: .allRemovalsTimeMs) ?? 0
+ self.commitTimeMs = try container.decodeIfPresent(Int64.self, forKey:
.commitTimeMs) ?? 0
+ self.memoryUsedBytes = try container.decodeIfPresent(Int64.self, forKey:
.memoryUsedBytes) ?? 0
+ self.numRowsDroppedByWatermark =
+ try container.decodeIfPresent(Int64.self, forKey:
.numRowsDroppedByWatermark) ?? 0
+ self.numShufflePartitions =
+ try container.decodeIfPresent(Int64.self, forKey: .numShufflePartitions)
?? 0
+ self.numStateStoreInstances =
+ try container.decodeIfPresent(Int64.self, forKey:
.numStateStoreInstances) ?? 0
+ self.customMetrics =
+ try container.decodeIfPresent([String: Int64].self, forKey:
.customMetrics) ?? [:]
+ }
+}
+
+/// Information about progress made in the execution of a ``StreamingQuery``
+/// during a trigger. Each event relates to processing done for a single
trigger of
+/// the streaming query. Events are emitted even when no new data is available
to be processed.
+public struct StreamingQueryProgress: Sendable, Codable {
+ /// A unique query id that persists across restarts. See
``StreamingQuery/id``.
+ public let id: UUID
+ /// A query id that is unique for every start/restart. See
``StreamingQuery/runId``.
+ public let runId: UUID
+ /// User-specified name of the query, `nil` if not specified.
+ public let name: String?
+ /// Beginning time of the trigger in ISO8601 format, i.e. UTC timestamps.
+ public let timestamp: String
+ /// A unique id for the current batch of data being processed.
+ public let batchId: Int64
+ /// The process duration of each batch.
+ public let batchDuration: Int64
+ /// The amount of time taken to perform various operations in milliseconds.
+ public let durationMs: [String: Int64]
+ /// Statistics of event time seen in this batch, e.g.
`max`/`min`/`avg`/`watermark`.
+ public let eventTime: [String: String]
+ /// Information about operators in the query that store state.
+ public let stateOperators: [StateOperatorProgress]
+ /// Detailed statistics on data being read from each of the streaming
sources.
+ public let sources: [SourceProgress]
+ /// Information about progress made for the sink.
+ public let sink: SinkProgress
+ /// Observed metrics of this query, keyed by the observation name with the
metric row
+ /// preserved as a JSON string.
+ public let observedMetrics: [String: String]
+ /// The original JSON representation of this progress received from the
server.
+ public internal(set) var json: String = ""
+
+ /// The aggregate (across all sources) number of records processed in a
trigger.
+ public var numInputRows: Int64 {
+ return sources.map { $0.numInputRows }.reduce(0, +)
+ }
+
+ /// The aggregate (across all sources) rate of data arriving.
+ public var inputRowsPerSecond: Double {
+ return sources.map { $0.inputRowsPerSecond }.reduce(0, +)
+ }
+
+ /// The aggregate (across all sources) rate at which Spark is processing
data.
+ public var processedRowsPerSecond: Double {
+ return sources.map { $0.processedRowsPerSecond }.reduce(0, +)
+ }
+
+ enum CodingKeys: String, CodingKey {
+ case id, runId, name, timestamp, batchId, batchDuration, durationMs,
eventTime
+ case stateOperators, sources, sink, observedMetrics
+ }
+
+ public init(from decoder: any Decoder) throws {
+ let container = try decoder.container(keyedBy: CodingKeys.self)
+ self.id = try container.decode(UUID.self, forKey: .id)
+ self.runId = try container.decode(UUID.self, forKey: .runId)
+ self.name = try container.decodeIfPresent(String.self, forKey: .name)
+ self.timestamp = try container.decode(String.self, forKey: .timestamp)
+ self.batchId = try container.decode(Int64.self, forKey: .batchId)
+ self.batchDuration = try container.decodeIfPresent(Int64.self, forKey:
.batchDuration) ?? 0
+ self.durationMs =
+ try container.decodeIfPresent([String: Int64].self, forKey: .durationMs)
?? [:]
+ self.eventTime =
+ try container.decodeIfPresent([String: String].self, forKey: .eventTime)
?? [:]
+ self.stateOperators =
+ try container.decodeIfPresent([StateOperatorProgress].self, forKey:
.stateOperators) ?? []
+ self.sources = try container.decodeIfPresent([SourceProgress].self,
forKey: .sources) ?? []
+ self.sink = try container.decode(SinkProgress.self, forKey: .sink)
+ self.observedMetrics =
+ try container.decodeIfPresent([String: JSONValue].self, forKey:
.observedMetrics)?
+ .mapValues { $0.compactString } ?? [:]
+ }
+
+ /// Creates a ``StreamingQueryProgress`` from the given JSON string.
+ /// Unknown fields are ignored and most of the missing fields fall back to
default values.
+ /// - Parameter json: A JSON string emitted by a Spark Connect server.
+ /// - Returns: A ``StreamingQueryProgress``.
+ public static func fromJson(_ json: String) throws -> StreamingQueryProgress
{
+ var progress = try JSONDecoder().decode(StreamingQueryProgress.self, from:
Data(json.utf8))
+ progress.json = json
+ return progress
+ }
+}
diff --git a/Tests/SparkConnectTests/StreamingQueryProgressTests.swift
b/Tests/SparkConnectTests/StreamingQueryProgressTests.swift
new file mode 100644
index 0000000..8794658
--- /dev/null
+++ b/Tests/SparkConnectTests/StreamingQueryProgressTests.swift
@@ -0,0 +1,162 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+#if canImport(FoundationEssentials)
+import FoundationEssentials
+#else
+import Foundation
+#endif
+import Testing
+
+@testable import SparkConnect
+
+/// A test suite for ``StreamingQueryProgress`` JSON decoding
+struct StreamingQueryProgressTests {
+
+ @Test
+ func fromJson() throws {
+ let json = """
+ {
+ "id": "33bd4ba3-79b2-4f0e-8d1c-2ccb44e29d4e",
+ "runId": "68ea3dd5-8b8e-4290-8d78-4b0a3f0e4a52",
+ "name": "myQuery",
+ "timestamp": "2026-08-07T00:00:00.000Z",
+ "batchId": 2,
+ "batchDuration": 1000,
+ "durationMs": {"addBatch": 100, "triggerExecution": 245},
+ "eventTime": {"watermark": "2026-08-07T00:00:00.000Z"},
+ "stateOperators": [{
+ "operatorName": "stateStoreSave",
+ "numRowsTotal": 3,
+ "numRowsUpdated": 1,
+ "allUpdatesTimeMs": 10,
+ "numRowsRemoved": 0,
+ "allRemovalsTimeMs": 0,
+ "commitTimeMs": 5,
+ "memoryUsedBytes": 100,
+ "numRowsDroppedByWatermark": 0,
+ "numShufflePartitions": 5,
+ "numStateStoreInstances": 5,
+ "customMetrics": {"loadedMapCacheHitCount": 2}
+ }],
+ "sources": [{
+ "description": "FileStreamSource[file:/tmp/input]",
+ "startOffset": "0",
+ "endOffset": "1",
+ "latestOffset": "1",
+ "numInputRows": 10,
+ "inputRowsPerSecond": 12.5,
+ "processedRowsPerSecond": 25.0,
+ "metrics": {"avgOffsetsBehindLatest": "0.0"}
+ }],
+ "sink": {"description": "FileSink[file:/tmp/output]", "numOutputRows":
10},
+ "observedMetrics": {"event": {"c1": 1, "c2": 2}}
+ }
+ """
+ let progress = try StreamingQueryProgress.fromJson(json)
+ #expect(progress.id == UUID(uuidString:
"33bd4ba3-79b2-4f0e-8d1c-2ccb44e29d4e"))
+ #expect(progress.runId == UUID(uuidString:
"68ea3dd5-8b8e-4290-8d78-4b0a3f0e4a52"))
+ #expect(progress.name == "myQuery")
+ #expect(progress.timestamp == "2026-08-07T00:00:00.000Z")
+ #expect(progress.batchId == 2)
+ #expect(progress.batchDuration == 1000)
+ #expect(progress.durationMs == ["addBatch": 100, "triggerExecution": 245])
+ #expect(progress.eventTime == ["watermark": "2026-08-07T00:00:00.000Z"])
+ #expect(progress.stateOperators.count == 1)
+ #expect(progress.stateOperators[0].operatorName == "stateStoreSave")
+ #expect(progress.stateOperators[0].numRowsTotal == 3)
+ #expect(progress.stateOperators[0].customMetrics ==
["loadedMapCacheHitCount": 2])
+ #expect(progress.sources.count == 1)
+ #expect(progress.sources[0].description ==
"FileStreamSource[file:/tmp/input]")
+ #expect(progress.sources[0].startOffset == "0")
+ #expect(progress.sources[0].endOffset == "1")
+ #expect(progress.sources[0].numInputRows == 10)
+ #expect(progress.sources[0].inputRowsPerSecond == 12.5)
+ #expect(progress.sources[0].metrics == ["avgOffsetsBehindLatest": "0.0"])
+ #expect(progress.sink.description == "FileSink[file:/tmp/output]")
+ #expect(progress.sink.numOutputRows == 10)
+ #expect(progress.observedMetrics == ["event": "{\"c1\":1,\"c2\":2}"])
+ #expect(progress.numInputRows == 10)
+ #expect(progress.inputRowsPerSecond == 12.5)
+ #expect(progress.processedRowsPerSecond == 25.0)
+ #expect(progress.json == json)
+ }
+
+ @Test
+ func fromJsonWithMissingFields() throws {
+ let json = """
+ {
+ "id": "33bd4ba3-79b2-4f0e-8d1c-2ccb44e29d4e",
+ "runId": "68ea3dd5-8b8e-4290-8d78-4b0a3f0e4a52",
+ "name": null,
+ "timestamp": "2026-08-07T00:00:00.000Z",
+ "batchId": 0,
+ "unknownField": {"ignored": true},
+ "sink": {"description": "ForeachBatchSink"}
+ }
+ """
+ let progress = try StreamingQueryProgress.fromJson(json)
+ #expect(progress.name == nil)
+ #expect(progress.batchDuration == 0)
+ #expect(progress.durationMs.isEmpty)
+ #expect(progress.eventTime.isEmpty)
+ #expect(progress.stateOperators.isEmpty)
+ #expect(progress.sources.isEmpty)
+ #expect(progress.sink.description == "ForeachBatchSink")
+ #expect(progress.sink.numOutputRows == -1)
+ #expect(progress.sink.metrics.isEmpty)
+ #expect(progress.observedMetrics.isEmpty)
+ #expect(progress.numInputRows == 0)
+ }
+
+ @Test
+ func fromJsonWithObjectOffsetsAndNaN() throws {
+ let json = """
+ {
+ "id": "33bd4ba3-79b2-4f0e-8d1c-2ccb44e29d4e",
+ "runId": "68ea3dd5-8b8e-4290-8d78-4b0a3f0e4a52",
+ "timestamp": "2026-08-07T00:00:00.000Z",
+ "batchId": 0,
+ "sources": [{
+ "description": "KafkaV2[Subscribe[topic-0]]",
+ "startOffset": {"topic-0": {"0": 1}},
+ "endOffset": {"topic-0": {"0": 2}},
+ "latestOffset": null,
+ "numInputRows": 1,
+ "inputRowsPerSecond": "NaN"
+ }],
+ "sink": {"description": "MemorySink"}
+ }
+ """
+ let progress = try StreamingQueryProgress.fromJson(json)
+ #expect(progress.sources[0].startOffset == "{\"topic-0\":{\"0\":1}}")
+ #expect(progress.sources[0].endOffset == "{\"topic-0\":{\"0\":2}}")
+ #expect(progress.sources[0].latestOffset == "null")
+ #expect(progress.sources[0].inputRowsPerSecond.isNaN)
+ #expect(progress.sources[0].processedRowsPerSecond.isNaN)
+ #expect(progress.inputRowsPerSecond.isNaN)
+ }
+
+ @Test
+ func fromJsonWithInvalidJson() throws {
+ #expect(throws: Error.self) {
+ try StreamingQueryProgress.fromJson("invalid")
+ }
+ }
+}
diff --git a/Tests/SparkConnectTests/StreamingQueryTests.swift
b/Tests/SparkConnectTests/StreamingQueryTests.swift
index ae2ad54..69be862 100644
--- a/Tests/SparkConnectTests/StreamingQueryTests.swift
+++ b/Tests/SparkConnectTests/StreamingQueryTests.swift
@@ -92,4 +92,51 @@ struct StreamingQueryTests {
await spark.stop()
}
+
+ @Test
+ func progress() 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
+ .queryName("progress-test")
+ .option("checkpointLocation", checkpoint)
+ .outputMode("append")
+ .format("orc")
+ .trigger(Trigger.ProcessingTime(1000))
+ .start(output)
+ // Wait for processing
+ try await Task.sleep(nanoseconds: 2_000_000_000)
+
+ let progress = try #require(try await query.lastProgress)
+ #expect(await query.id == progress.id)
+ #expect(await query.runId == progress.runId)
+ #expect(progress.name == "progress-test")
+ #expect(progress.batchId >= 0)
+ #expect(progress.batchDuration >= 0)
+ #expect(progress.timestamp.isEmpty == false)
+ #expect(progress.durationMs.isEmpty == false)
+ #expect(progress.sources.count == 1)
+ #expect(progress.sources[0].description.contains("FileStreamSource"))
+ #expect(progress.sink.description.contains("FileSink"))
+ #expect(progress.json.contains("\"runId\""))
+
+ let recentProgress = try await query.recentProgress
+ #expect(recentProgress.isEmpty == false)
+ #expect(recentProgress.allSatisfy { $0.runId == progress.runId })
+
+ try await query.stop()
+ await spark.stop()
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
