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 4cfb32f [SPARK-58338] Fix `defineFlow` to send `relation` via
`relation_flow_details`
4cfb32f is described below
commit 4cfb32fa614f65bcb3e3df00c24d6248125a5c1c
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Fri Jul 24 17:15:56 2026 -0700
[SPARK-58338] Fix `defineFlow` to send `relation` via
`relation_flow_details`
### What changes were proposed in this pull request?
This PR aims to fix `SparkConnectClient.defineFlow` to send the given
`Relation` to the server via the `relation_flow_details` field of
`PipelineCommand.DefineFlow`.
Since the `4.1.0-preview3` generated code update ([SPARK-54043]), the
`relation` assignment has been commented out because the upstream proto moved
the `relation` field into the `details` oneof (`WriteRelationFlowDetails`). As
a result, the caller-provided `Relation` was silently dropped and flows were
defined without a query.
### Why are the changes needed?
To fix a bug where `defineFlow` defines a flow without its defining query.
### Does this PR introduce _any_ user-facing change?
No, `defineFlow` is an internal API.
### 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 #443 from dongjoon-hyun/SPARK-58338.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
Sources/SparkConnect/SparkConnectClient.swift | 21 ++++++++++++++++-----
.../SparkConnectTests/SparkConnectClientTests.swift | 13 +++++++++++++
2 files changed, 29 insertions(+), 5 deletions(-)
diff --git a/Sources/SparkConnect/SparkConnectClient.swift
b/Sources/SparkConnect/SparkConnectClient.swift
index aecbc1b..893a21b 100644
--- a/Sources/SparkConnect/SparkConnectClient.swift
+++ b/Sources/SparkConnect/SparkConnectClient.swift
@@ -1342,6 +1342,20 @@ public actor SparkConnectClient {
}
}
+ static func getDefineFlow(
+ _ dataflowGraphID: String,
+ _ flowName: String,
+ _ targetDatasetName: String,
+ _ relation: Relation
+ ) -> Spark_Connect_PipelineCommand.DefineFlow {
+ var defineFlow = Spark_Connect_PipelineCommand.DefineFlow()
+ defineFlow.dataflowGraphID = dataflowGraphID
+ defineFlow.flowName = flowName
+ defineFlow.targetDatasetName = targetDatasetName
+ defineFlow.relationFlowDetails.relation = relation
+ return defineFlow
+ }
+
@discardableResult
func defineFlow(
_ dataflowGraphID: String,
@@ -1354,11 +1368,8 @@ public actor SparkConnectClient {
throw SparkConnectError.InvalidArgument
}
- var defineFlow = Spark_Connect_PipelineCommand.DefineFlow()
- defineFlow.dataflowGraphID = dataflowGraphID
- defineFlow.flowName = flowName
- defineFlow.targetDatasetName = targetDatasetName
- // defineFlow.relation = relation
+ let defineFlow = SparkConnectClient.getDefineFlow(
+ dataflowGraphID, flowName, targetDatasetName, relation)
var pipelineCommand = Spark_Connect_PipelineCommand()
pipelineCommand.commandType = .defineFlow(defineFlow)
diff --git a/Tests/SparkConnectTests/SparkConnectClientTests.swift
b/Tests/SparkConnectTests/SparkConnectClientTests.swift
index 4fac0e5..36fd5e9 100644
--- a/Tests/SparkConnectTests/SparkConnectClientTests.swift
+++ b/Tests/SparkConnectTests/SparkConnectClientTests.swift
@@ -178,6 +178,19 @@ struct SparkConnectClientTests {
await client.stop()
}
+ @Test
+ func getDefineFlow() async throws {
+ let dataflowGraphID = UUID().uuidString
+ var relation = Relation()
+ relation.range.end = 10
+ let defineFlow = SparkConnectClient.getDefineFlow(dataflowGraphID, "f1",
"ds1", relation)
+ #expect(defineFlow.dataflowGraphID == dataflowGraphID)
+ #expect(defineFlow.flowName == "f1")
+ #expect(defineFlow.targetDatasetName == "ds1")
+ #expect(defineFlow.relationFlowDetails.hasRelation)
+ #expect(defineFlow.relationFlowDetails.relation == relation)
+ }
+
// @Test
func defineFlow() async throws {
let client = try SparkConnectClient(remote: TEST_REMOTE)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
