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 1187c9b [SPARK-60055] Fix `dropDuplicatesWithinWatermark` to set
`withinWatermark` in `Deduplicate` relation
1187c9b is described below
commit 1187c9b6bacdd8db07fc7918d5c9e8d19576d017
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Oct 7 14:43:55 2026 -0700
[SPARK-60055] Fix `dropDuplicatesWithinWatermark` to set `withinWatermark`
in `Deduplicate` relation
### What changes were proposed in this pull request?
This PR aims to fix `SparkConnectClient.getDropDuplicates` to set the
`within_watermark` field of the `Deduplicate` relation by adding
`deduplicate.withinWatermark = withinWatermark`.
### Why are the changes needed?
Previously, `getDropDuplicates` accepted the `withinWatermark` parameter
but never propagated it to `Spark_Connect_Deduplicate`. As a result, the public
`DataFrame.dropDuplicatesWithinWatermark(...)` API silently behaved like
`dropDuplicates(...)`. For streaming queries, this gives the wrong semantics:
the server keeps unbounded deduplication state instead of the watermark-bounded
state of `DeduplicateWithinWatermark`.
### Does this PR introduce _any_ user-facing change?
Yes, this is a bug fix. `dropDuplicatesWithinWatermark` now sends
`within_watermark = true` to the Spark Connect server, consistent with the
Scala and PySpark Spark Connect clients. As a result, calling it on a batch
`DataFrame` now fails with `dropDuplicatesWithinWatermark is not supported with
batch DataFrames/DataSets`, the same as other Spark clients. Previously, it
silently behaved like `dropDuplicates`.
### How was this patch tested?
Pass the CIs with the newly added plan-level test case,
`DataFrameInternalTests.dropDuplicatesWithinWatermarkPlan`, which does not
require a Spark Connect server. It fails without this fix and passes with it.
```
$ swift test --no-parallel --filter
DataFrameInternalTests/dropDuplicatesWithinWatermarkPlan
```
In addition, the existing `DataFrameTests.dropDuplicatesWithinWatermark`
and `DataFrameTests.withWatermark` test cases are updated to expect an error
for batch `DataFrame`s.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #586 from dongjoon-hyun/SPARK-60055.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
Sources/SparkConnect/SparkConnectClient.swift | 1 +
Tests/SparkConnectTests/DataFrameInternalTests.swift | 13 +++++++++++++
Tests/SparkConnectTests/DataFrameTests.swift | 13 ++++++++++---
3 files changed, 24 insertions(+), 3 deletions(-)
diff --git a/Sources/SparkConnect/SparkConnectClient.swift
b/Sources/SparkConnect/SparkConnectClient.swift
index db28eb9..358f394 100644
--- a/Sources/SparkConnect/SparkConnectClient.swift
+++ b/Sources/SparkConnect/SparkConnectClient.swift
@@ -709,6 +709,7 @@ public actor SparkConnectClient {
} else {
deduplicate.columnNames = columnNames
}
+ deduplicate.withinWatermark = withinWatermark
return createPlan { $0.deduplicate = deduplicate }
}
diff --git a/Tests/SparkConnectTests/DataFrameInternalTests.swift
b/Tests/SparkConnectTests/DataFrameInternalTests.swift
index 0f84c48..c686c7c 100644
--- a/Tests/SparkConnectTests/DataFrameInternalTests.swift
+++ b/Tests/SparkConnectTests/DataFrameInternalTests.swift
@@ -165,4 +165,17 @@ struct DataFrameInternalTests {
}
await spark.stop()
}
+
+ @Test
+ func dropDuplicatesWithinWatermarkPlan() async throws {
+ // This test doesn't require a running server because it only inspects the
plan.
+ let spark = try SparkSession("sc://localhost")
+ var plan = Plan()
+ plan.root.range.end = 10
+ let df = DataFrame(spark: spark, plan: plan)
+ #expect(await df.dropDuplicates().plan.root.deduplicate.withinWatermark ==
false)
+ #expect(await
df.dropDuplicates("id").plan.root.deduplicate.withinWatermark == false)
+ #expect(await
df.dropDuplicatesWithinWatermark().plan.root.deduplicate.withinWatermark)
+ #expect(await
df.dropDuplicatesWithinWatermark("id").plan.root.deduplicate.withinWatermark)
+ }
}
diff --git a/Tests/SparkConnectTests/DataFrameTests.swift
b/Tests/SparkConnectTests/DataFrameTests.swift
index 5889470..f30fc8b 100644
--- a/Tests/SparkConnectTests/DataFrameTests.swift
+++ b/Tests/SparkConnectTests/DataFrameTests.swift
@@ -1312,8 +1312,13 @@ struct DataFrameTests {
func dropDuplicatesWithinWatermark() async throws {
let spark = try await SparkSession.builder.getOrCreate()
let df = try await spark.sql("SELECT * FROM VALUES (1), (2), (3), (1), (3)
T(a)")
- #expect(try await df.dropDuplicatesWithinWatermark().count() == 3)
- #expect(try await df.dropDuplicatesWithinWatermark("a").count() == 3)
+ // Spark doesn't support `dropDuplicatesWithinWatermark` with batch
DataFrames.
+ try await #require(throws: Error.self) {
+ try await df.dropDuplicatesWithinWatermark().count()
+ }
+ try await #require(throws: Error.self) {
+ try await df.dropDuplicatesWithinWatermark("a").count()
+ }
await spark.stop()
}
@@ -1332,7 +1337,9 @@ struct DataFrameTests {
"""
)
.withWatermark("eventTime", "1 minute") // This tests only API for now
- #expect(try await df.dropDuplicatesWithinWatermark("data").count() == 1)
+ try await #require(throws: Error.self) {
+ try await df.dropDuplicatesWithinWatermark("data").count()
+ }
await spark.stop()
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
