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 6a1e37a [SPARK-60064] Fix DataFrameReader and DataStreamReader to not
send empty format when format is not specified
6a1e37a is described below
commit 6a1e37a9b19401413b4815aadfd02e6600e2f94b
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Oct 7 16:16:19 2026 -0700
[SPARK-60064] Fix DataFrameReader and DataStreamReader to not send empty
format when format is not specified
### What changes were proposed in this pull request?
This PR aims to fix `DataFrameReader.load` and `DataStreamReader.load` to
set `DataSource.format` only when a format is specified.
### Why are the changes needed?
`DataSource.format` is a proto3 `optional` field. Previously, `load` always
assigned `self.source` (initialized to `""`) to it, so the server saw
`hasFormat == true` and called `reader.format("")` instead of falling back to
`spark.sql.sources.default`. As a result, `spark.read.load(path)` and
`spark.readStream.schema(...).load(path)` without `format(...)` failed.
```
[DATA_SOURCE_NOT_FOUND] Failed to find the data source: . Make sure the
provider name is correct and the package is properly registered and compatible
with your Spark version. SQLSTATE: 42K02
```
This is consistent with the PySpark Connect client, which sets
`data_source.format` only when `format` is not `None`.
### Does this PR introduce _any_ user-facing change?
Yes. `load` without `format(...)` now uses the default data source
(`spark.sql.sources.default`) instead of failing.
### How was this patch tested?
Pass the CIs with the newly added test cases
(`DataFrameReaderTests.loadWithoutFormat` and
`DataStreamTests.loadWithoutFormat`). Without this fix, the newly added tests
fail with the `DATA_SOURCE_NOT_FOUND` error against Apache Spark 4.2.0 Connect
server.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #590 from dongjoon-hyun/SPARK-60064.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
Sources/SparkConnect/DataFrameReader.swift | 4 +++-
Sources/SparkConnect/DataStreamReader.swift | 4 +++-
Tests/SparkConnectTests/DataFrameReaderTests.swift | 8 ++++++++
Tests/SparkConnectTests/DataStreamTests.swift | 12 ++++++++++++
4 files changed, 26 insertions(+), 2 deletions(-)
diff --git a/Sources/SparkConnect/DataFrameReader.swift
b/Sources/SparkConnect/DataFrameReader.swift
index 245ae2e..1439513 100644
--- a/Sources/SparkConnect/DataFrameReader.swift
+++ b/Sources/SparkConnect/DataFrameReader.swift
@@ -165,7 +165,9 @@ public actor DataFrameReader: Sendable {
self.paths = paths
var dataSource = DataSource()
- dataSource.format = self.source
+ if !self.source.isEmpty {
+ dataSource.format = self.source
+ }
dataSource.paths = self.paths
dataSource.options = self.extraOptions.toStringDictionary()
if let userSpecifiedSchemaDDL = self.userSpecifiedSchemaDDL {
diff --git a/Sources/SparkConnect/DataStreamReader.swift
b/Sources/SparkConnect/DataStreamReader.swift
index 931ce99..c0a0405 100644
--- a/Sources/SparkConnect/DataStreamReader.swift
+++ b/Sources/SparkConnect/DataStreamReader.swift
@@ -144,7 +144,9 @@ public actor DataStreamReader: Sendable {
self.paths = paths
var dataSource = DataSource()
- dataSource.format = self.source
+ if !self.source.isEmpty {
+ dataSource.format = self.source
+ }
dataSource.paths = self.paths
dataSource.options = self.extraOptions.toStringDictionary()
if let userSpecifiedSchemaDDL = self.userSpecifiedSchemaDDL {
diff --git a/Tests/SparkConnectTests/DataFrameReaderTests.swift
b/Tests/SparkConnectTests/DataFrameReaderTests.swift
index 5843758..f9d76a7 100644
--- a/Tests/SparkConnectTests/DataFrameReaderTests.swift
+++ b/Tests/SparkConnectTests/DataFrameReaderTests.swift
@@ -123,6 +123,14 @@ struct DataFrameReaderTests {
await spark.stop()
}
+ @Test
+ func loadWithoutFormat() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let path = "../examples/src/main/resources/users.parquet"
+ #expect(try await spark.read.load(path).count() == 2)
+ await spark.stop()
+ }
+
@Test
func table() async throws {
let tableName = "TABLE_" + UUID().uuidString.replacingOccurrences(of: "-",
with: "")
diff --git a/Tests/SparkConnectTests/DataStreamTests.swift
b/Tests/SparkConnectTests/DataStreamTests.swift
index 7f1ad99..d844a04 100644
--- a/Tests/SparkConnectTests/DataStreamTests.swift
+++ b/Tests/SparkConnectTests/DataStreamTests.swift
@@ -85,6 +85,18 @@ struct DataStreamTests {
await spark.stop()
}
+ @Test
+ func loadWithoutFormat() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let input = "/tmp/input-" + UUID().uuidString
+ try await spark.range(1).write.parquet(input)
+
+ let df = try await spark.readStream.schema("id LONG").load(input)
+ #expect(try await df.isStreaming())
+ #expect(try await df.columns == ["id"])
+ await spark.stop()
+ }
+
@Test
func startConcurrently() async throws {
let spark = try await SparkSession.builder.getOrCreate()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
