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 84a6cc5 [SPARK-58579] Support `TIME` type in
`SparkSession.createDataFrame`
84a6cc5 is described below
commit 84a6cc5de6759da44406e573de856c82d094f64a
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Aug 4 17:58:24 2026 -0700
[SPARK-58579] Support `TIME` type in `SparkSession.createDataFrame`
### What changes were proposed in this pull request?
This PR aims to support `TIME` type in `SparkSession.createDataFrame`.
```swift
let time = LocalTime(hour: 12, minute: 34, second: 56, nanosecond:
123_456_000)!
let df = try await spark.createDataFrame([[time], [nil]], "t TIME")
try await df.collect() // [Row(12:34:56.123456), Row(nil)]
```
- `ConvertToArrow` maps the `time` DDL type to an Arrow
`time64(NANOSECOND)` column and
writes `LocalTime.nanoOfDay` values as-is, symmetrically with the
`collect()` decoding
path added by SPARK-58571. `nil` is mapped to `NULL` like the other types.
- Fix the vendored `ArrowWriterHelper` to serialize the `bitWidth` field
(64) of the
FlatBuffers `Time` type for `time64`. Without it, the field defaults to
32 and the
server rejects the uploaded column with
`[UNSUPPORTED_ARROWTYPE] Unsupported arrow type Time(NANOSECOND, 32)`.
(The same omission exists in upstream `apache/arrow-swift`; `time32` is
unaffected
because 32 is the correct default there.)
### Why are the changes needed?
This is the last missing piece of `TIME` type support.
`DataType.simpleString`
(SPARK-58568), `collect()` (SPARK-58571), and `lit`/SQL parameter binding
(SPARK-58574)
already support `TIME`, but `createDataFrame` rejected `t TIME` schemas with
`SparkConnectError.InvalidType`.
### Does this PR introduce _any_ user-facing change?
Yes, `createDataFrame` now accepts `TIME` columns with `LocalTime` values.
Previously it threw `SparkConnectError.InvalidType`.
### How was this patch tested?
Pass the CIs with a newly added test case, `CreateDataFrameTests.timeType`,
which
verifies a `createDataFrame` → `collect` round trip including `LocalTime`
values and
`nil`, and the `time(6)`/`time(0)` schema mapping. The test is guarded by
`spark.version >= "4.2"` and `spark.sql.timeType.enabled`, and was verified
manually
against both Apache Spark 4.2.0 (runs) and 4.1.3 (skips) Spark Connect
servers.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #483 from dongjoon-hyun/dongjoon/silly-lederberg-905677.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
Sources/SparkConnect/ArrowWriterHelper.swift | 1 +
Sources/SparkConnect/ConvertToArrow.swift | 4 ++++
Sources/SparkConnect/SparkSession.swift | 4 ++--
Tests/SparkConnectTests/CreateDataFrameTests.swift | 15 +++++++++++++++
4 files changed, 22 insertions(+), 2 deletions(-)
diff --git a/Sources/SparkConnect/ArrowWriterHelper.swift
b/Sources/SparkConnect/ArrowWriterHelper.swift
index 9a05f29..2e20d27 100644
--- a/Sources/SparkConnect/ArrowWriterHelper.swift
+++ b/Sources/SparkConnect/ArrowWriterHelper.swift
@@ -116,6 +116,7 @@ func toFBType( // swiftlint:disable:this
cyclomatic_complexity function_body_le
if let timeType = arrowType as? ArrowTypeTime64 {
org_apache_arrow_flatbuf_Time.add(
unit: timeType.unit == .microseconds ? .microsecond : .nanosecond,
&fbb)
+ org_apache_arrow_flatbuf_Time.add(bitWidth: 64, &fbb)
return .success(org_apache_arrow_flatbuf_Time.endTime(&fbb, start:
startOffset))
}
diff --git a/Sources/SparkConnect/ConvertToArrow.swift
b/Sources/SparkConnect/ConvertToArrow.swift
index 6473282..c626e7e 100644
--- a/Sources/SparkConnect/ConvertToArrow.swift
+++ b/Sources/SparkConnect/ConvertToArrow.swift
@@ -91,6 +91,10 @@ enum ConvertToArrow {
return try fill(
ArrowArrayBuilders.loadTimestampArrayBuilder(.microseconds, timezone:
"UTC"), column
) { ($0 as? Date).map { Int64(($0.timeIntervalSince1970 *
1_000_000).rounded()) } }
+ case .time:
+ return try fill(ArrowArrayBuilders.loadTime64ArrayBuilder(.nanoseconds),
column) {
+ ($0 as? LocalTime)?.nanoOfDay
+ }
default:
throw SparkConnectError.InvalidType
}
diff --git a/Sources/SparkConnect/SparkSession.swift
b/Sources/SparkConnect/SparkSession.swift
index 4b99641..aeaf480 100644
--- a/Sources/SparkConnect/SparkSession.swift
+++ b/Sources/SparkConnect/SparkSession.swift
@@ -121,8 +121,8 @@ public actor SparkSession {
///
/// The data is serialized into an `Apache Arrow` IPC stream and embedded
into the plan as a
/// `LocalRelation`. The supported schema types are `BOOLEAN`, `TINYINT`,
`SMALLINT`, `INT`,
- /// `BIGINT`, `FLOAT`, `DOUBLE`, `STRING`, `BINARY`, `DATE`, and
`TIMESTAMP`. `nil` values are
- /// mapped to `NULL`. When the serialized data is equal to or larger than
+ /// `BIGINT`, `FLOAT`, `DOUBLE`, `STRING`, `BINARY`, `DATE`, `TIMESTAMP`,
and `TIME`.
+ /// `nil` values are mapped to `NULL`. When the serialized data is equal to
or larger than
/// `spark.sql.session.localRelationCacheThreshold` (1MiB by default), it is
uploaded to the
/// server as a `cache/` artifact and referenced by a `CachedLocalRelation`
plan instead, so
/// it is not limited by the gRPC message size limit.
diff --git a/Tests/SparkConnectTests/CreateDataFrameTests.swift
b/Tests/SparkConnectTests/CreateDataFrameTests.swift
index e2dd454..11947b8 100644
--- a/Tests/SparkConnectTests/CreateDataFrameTests.swift
+++ b/Tests/SparkConnectTests/CreateDataFrameTests.swift
@@ -60,6 +60,21 @@ struct CreateDataFrameTests {
await spark.stop()
}
+ @Test
+ func timeType() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ if await spark.version >= "4.2" {
+ try await spark.conf.set("spark.sql.timeType.enabled", "true")
+ let time = try #require(LocalTime(hour: 12, minute: 34, second: 56,
nanosecond: 123_456_000))
+ let df = try await spark.createDataFrame([[time], [nil]], "t TIME")
+ #expect(try await df.dtypes.map { $0.1 } == ["time(6)"])
+ #expect(try await df.collect() == [Row(time), Row(nil)])
+ #expect(
+ try await spark.createDataFrame([[time]], "t TIME(0)").dtypes.map {
$0.1 } == ["time(0)"])
+ }
+ await spark.stop()
+ }
+
@Test
func binaryType() async throws {
let spark = try await SparkSession.builder.getOrCreate()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
