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 2cf4f28 [SPARK-58339] Send `RemoveCachedRemoteRelationCommand` when
checkpointed `DataFrame` is deallocated
2cf4f28 is described below
commit 2cf4f28b53cdeb38f39a2beed74b40773de45721
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Fri Jul 24 17:16:48 2026 -0700
[SPARK-58339] Send `RemoveCachedRemoteRelationCommand` when checkpointed
`DataFrame` is deallocated
### What changes were proposed in this pull request?
This PR aims to send `RemoveCachedRemoteRelationCommand` when a
checkpointed `DataFrame` is deallocated. Since `deinit` cannot be `async`, the
command is sent best-effort in a detached task, like the Scala and Python
clients.
### Why are the changes needed?
`checkpoint`/`localCheckpoint` creates a server-side `CachedRemoteRelation`
which was never removed, leaking on the server for the lifetime of the session.
### Does this PR introduce _any_ user-facing change?
No.
### 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 #444 from dongjoon-hyun/SPARK-58339.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
Sources/SparkConnect/DataFrame.swift | 11 +++++++++++
Sources/SparkConnect/SparkConnectClient.swift | 9 +++++++++
Tests/SparkConnectTests/DataFrameInternalTests.swift | 16 ++++++++++++++++
3 files changed, 36 insertions(+)
diff --git a/Sources/SparkConnect/DataFrame.swift
b/Sources/SparkConnect/DataFrame.swift
index a377e36..15fdad6 100644
--- a/Sources/SparkConnect/DataFrame.swift
+++ b/Sources/SparkConnect/DataFrame.swift
@@ -233,6 +233,17 @@ public actor DataFrame: Sendable {
self.plan = sqlText.toSparkConnectPlan(args)
}
+ deinit {
+ // `deinit` cannot be `async`, so send `RemoveCachedRemoteRelationCommand`
best-effort
+ // in a detached task like Scala and Python Spark Connect clients.
+ if case .cachedRemoteRelation(let cachedRemoteRelation) =
plan.root.relType {
+ let client = spark.client
+ Task {
+ try? await client.removeCachedRemoteRelation(cachedRemoteRelation)
+ }
+ }
+ }
+
public func getPlan() -> Sendable {
return self.plan
}
diff --git a/Sources/SparkConnect/SparkConnectClient.swift
b/Sources/SparkConnect/SparkConnectClient.swift
index 893a21b..93ef80f 100644
--- a/Sources/SparkConnect/SparkConnectClient.swift
+++ b/Sources/SparkConnect/SparkConnectClient.swift
@@ -1202,6 +1202,15 @@ public actor SparkConnectClient {
return Self.createPlan { $0.cachedRemoteRelation = cachedRemoteRelation }
}
+ func removeCachedRemoteRelation(_ relation:
Spark_Connect_CachedRemoteRelation) async throws {
+ var removeCommand = Spark_Connect_RemoveCachedRemoteRelationCommand()
+ removeCommand.relation = relation
+
+ var command = Spark_Connect_Command()
+ command.removeCachedRemoteRelationCommand = removeCommand
+ _ = try await execute(self.sessionID!, command)
+ }
+
static func getWithWatermark(
_ child: Relation,
_ eventTime: String,
diff --git a/Tests/SparkConnectTests/DataFrameInternalTests.swift
b/Tests/SparkConnectTests/DataFrameInternalTests.swift
index 1b79419..feed245 100644
--- a/Tests/SparkConnectTests/DataFrameInternalTests.swift
+++ b/Tests/SparkConnectTests/DataFrameInternalTests.swift
@@ -82,4 +82,20 @@ struct DataFrameInternalTests {
""")
await spark.stop()
}
+
+ @Test
+ func removeCachedRemoteRelation() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ if await spark.version >= "4.0.0" {
+ let df = try await spark.range(10).localCheckpoint()
+ #expect(try await df.count() == 10)
+ let cachedRemoteRelation = await df.plan.root.cachedRemoteRelation
+ try await spark.client.removeCachedRemoteRelation(cachedRemoteRelation)
+ // The `DataFrame` is no longer queryable after the server-side cached
relation is removed.
+ try await #require(throws: Error.self) {
+ try await df.count()
+ }
+ }
+ await spark.stop()
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
