You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/09/03 19:40:41 UTC
(spark-connect-swift) branch main updated: [SPARK-59216] Support JSON and CSV functions
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 fb9dceb [SPARK-59216] Support JSON and CSV functions
fb9dceb is described below
commit fb9dcebb7848a24b84386853749c85890bd59c3b
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Thu Sep 3 12:40:35 2026 -0700
[SPARK-59216] Support JSON and CSV functions
### What changes were proposed in this pull request?
This PR aims to support the following 10 JSON and CSV functions in two new files,
`JsonFunctions.swift` and `CsvFunctions.swift`.
**JSON functions (`JsonFunctions.swift`)**
| Function | Overloads added | Since |
| -------- | --------------- | ----- |
| `from_json` | `schema: String` (DDL), `schema: StructType` | 2.1.0 |
| `from_json` | `schema: Column` | 2.4.0 |
| `to_json` | | 2.1.0 |
| `schema_of_json` | `json: String`, `json: Column` | 2.4.0 (options since 3.0.0) |
| `get_json_object` | | 1.6.0 |
| `json_tuple` | | 1.6.0 |
| `json_array_length` | | 3.5.0 |
| `json_object_keys` | | 3.5.0 |
**CSV functions (`CsvFunctions.swift`)**
| Function | Overloads added | Since |
| -------- | --------------- | ----- |
| `from_csv` | `schema: String` (DDL), `schema: Column` | 3.0.0 |
| `to_csv` | | 3.0.0 |
| `schema_of_csv` | `csv: String`, `csv: Column` | 3.0.0 |
Unlike the previous function batches, these functions take an option map. Apache Spark
encodes it as a plain trailing argument rather than a dedicated proto field: both
`Column.fnWithOptions` in `sql/api/src/main/scala/org/apache/spark/sql/Column.scala` and
`_options_to_col` in `python/pyspark/sql/connect/functions/builtin.py` flatten the map into
`k1, v1, k2, v2, ...`, wrap it in a `map(...)` function call, and append it as the last
argument. When there is no option, no extra argument is appended at all. This PR mirrors
that exactly with a new internal `fn(_ name:options:_ args:)` helper in `Functions.swift`,
sorting the keys so that the generated expression is deterministic (Swift `Dictionary` is
unordered).
For the `schema` argument, this PR follows the existing
`DataFrameReader.schema(_ schema: String)` / `schema(_ schema: StructType)` precedent:
`from_json` accepts a DDL `String`, a `StructType` (converted via `StructType.toDDL`), or a
`Column` such as a `schema_of_json` result. `from_csv` accepts a DDL `String` or a `Column`,
matching the Spark Connect Python client. `ArrayType` and `MapType` schemas are covered by
the DDL string form, e.g. `ARRAY<STRUCT<a: INT>>`.
`json_tuple` is a generator function, and is implemented like the existing `explode` /
`posexplode` / `inline` functions.
### Why are the changes needed?
To improve the API coverage of the Swift Spark Connect client. Currently, no JSON or CSV
function is supported, so Swift users cannot parse or produce JSON and CSV strings inside a
`Column` expression without falling back to `selectExpr`.
All the added functions were introduced in Apache Spark 3.5.0 or earlier, so no server
version gate is needed.
### Does this PR introduce _any_ user-facing change?
No, this is a new feature which adds 10 new functions.
```swift
let df = try await spark.sql("SELECT 1 AS id, 'a' AS name")
let parsed = from_json(to_json(struct(col("id"), col("name"))), "id INT, name STRING")
try await df.select(parsed.alias("parsed")).selectExpr("parsed.id", "parsed.name").show()
try await spark.range(1).select(schema_of_json("{\"a\": 1, \"b\": \"x\"}")).show()
// STRUCT<a: BIGINT, b: STRING>
try await df.select(to_csv(struct(col("id"), col("name")), ["sep": ";"])).show()
// 1;a
```
### How was this patch tested?
Pass the CIs with the newly added test suites, `JsonFunctionsTests` and `CsvFunctionsTests`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
Closes #522 from dongjoon-hyun/SPARK-59216.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
Sources/SparkConnect/CsvFunctions.swift | 80 +++++++++++++
Sources/SparkConnect/Functions.swift | 12 ++
Sources/SparkConnect/JsonFunctions.swift | 128 ++++++++++++++++++++
Tests/SparkConnectTests/CsvFunctionsTests.swift | 104 ++++++++++++++++
Tests/SparkConnectTests/JsonFunctionsTests.swift | 146 +++++++++++++++++++++++
5 files changed, 470 insertions(+)
diff --git a/Sources/SparkConnect/CsvFunctions.swift b/Sources/SparkConnect/CsvFunctions.swift
new file mode 100644
index 0000000..f4c9a17
--- /dev/null
+++ b/Sources/SparkConnect/CsvFunctions.swift
@@ -0,0 +1,80 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+// MARK: - CSV functions
+
+/// Parses a column containing a CSV string into a struct with the given schema.
+/// Returns `NULL`, in the case of an unparseable string.
+/// - Parameters:
+/// - col: A ``Column`` that evaluates to a CSV string.
+/// - schema: A DDL schema string, e.g. `a INT, b STRING`.
+/// - options: Options to control how the CSV is parsed. It accepts the same options as
+/// the CSV data source.
+/// - Returns: A ``Column`` of the type given by the schema.
+public func from_csv(
+ _ col: Column, _ schema: String, _ options: [String: String] = [:]
+) -> Column {
+ return fn("from_csv", options: options, col, lit(schema))
+}
+
+/// Parses a column containing a CSV string into a struct with the given schema.
+/// Returns `NULL`, in the case of an unparseable string.
+/// - Parameters:
+/// - col: A ``Column`` that evaluates to a CSV string.
+/// - schema: A ``Column`` that evaluates to a constant schema string,
+/// e.g. a `schema_of_csv` result.
+/// - options: Options to control how the CSV is parsed. It accepts the same options as
+/// the CSV data source.
+/// - Returns: A ``Column`` of the type given by the schema.
+public func from_csv(
+ _ col: Column, _ schema: Column, _ options: [String: String] = [:]
+) -> Column {
+ return fn("from_csv", options: options, col, schema)
+}
+
+/// Parses a CSV string and infers its schema in DDL format.
+/// - Parameters:
+/// - csv: A CSV string.
+/// - options: Options to control how the CSV is parsed. It accepts the same options as
+/// the CSV data source.
+/// - Returns: A ``Column`` that evaluates to a schema string in DDL format.
+public func schema_of_csv(_ csv: String, _ options: [String: String] = [:]) -> Column {
+ return schema_of_csv(lit(csv), options)
+}
+
+/// Parses a CSV string and infers its schema in DDL format.
+/// - Parameters:
+/// - csv: A ``Column`` that evaluates to a constant CSV string.
+/// - options: Options to control how the CSV is parsed. It accepts the same options as
+/// the CSV data source.
+/// - Returns: A ``Column`` that evaluates to a schema string in DDL format.
+public func schema_of_csv(_ csv: Column, _ options: [String: String] = [:]) -> Column {
+ return fn("schema_of_csv", options: options, csv)
+}
+
+/// Converts a column containing a struct into a CSV string.
+/// Throws an exception, in the case of an unsupported type.
+/// - Parameters:
+/// - col: A ``Column`` that evaluates to a struct.
+/// - options: Options to control how the column is converted. It accepts the same options as
+/// the CSV data source.
+/// - Returns: A ``Column`` that evaluates to a string.
+public func to_csv(_ col: Column, _ options: [String: String] = [:]) -> Column {
+ return fn("to_csv", options: options, col)
+}
diff --git a/Sources/SparkConnect/Functions.swift b/Sources/SparkConnect/Functions.swift
index 9caf84d..e11f8b0 100644
--- a/Sources/SparkConnect/Functions.swift
+++ b/Sources/SparkConnect/Functions.swift
@@ -309,6 +309,18 @@ func fn(_ name: String, _ args: [Column], isDistinct: Bool = false) -> Column {
return Column(expr)
}
+/// Invokes a function with the given arguments followed by a map ``Column`` built from the given
+/// options like Spark SQL's `Column.fnWithOptions`. The map argument is omitted when there is no
+/// option. The keys are sorted to make the generated expression deterministic.
+func fn(_ name: String, options: [String: String], _ args: Column...) -> Column {
+ var arguments = args
+ if !options.isEmpty {
+ let pairs = options.sorted { $0.key < $1.key }.flatMap { [lit($0.key), lit($0.value)] }
+ arguments.append(fn("map", pairs))
+ }
+ return fn(name, arguments)
+}
+
private func litColumn(_ literal: ExpressionLiteral) -> Column {
var expr = Spark_Connect_Expression()
expr.literal = literal
diff --git a/Sources/SparkConnect/JsonFunctions.swift b/Sources/SparkConnect/JsonFunctions.swift
new file mode 100644
index 0000000..704a27b
--- /dev/null
+++ b/Sources/SparkConnect/JsonFunctions.swift
@@ -0,0 +1,128 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+// MARK: - JSON functions
+
+/// Parses a column containing a JSON string into a struct, an array, or a map with the given
+/// schema. Returns `NULL`, in the case of an unparseable string.
+/// - Parameters:
+/// - col: A ``Column`` that evaluates to a JSON string.
+/// - schema: A DDL schema string, e.g. `a INT, b STRING` or `ARRAY<STRUCT<a: INT>>`.
+/// - options: Options to control how the JSON is parsed. It accepts the same options as
+/// the JSON data source.
+/// - Returns: A ``Column`` of the type given by the schema.
+public func from_json(
+ _ col: Column, _ schema: String, _ options: [String: String] = [:]
+) -> Column {
+ return fn("from_json", options: options, col, lit(schema))
+}
+
+/// Parses a column containing a JSON string into a struct with the given schema.
+/// Returns `NULL`, in the case of an unparseable string.
+/// - Parameters:
+/// - col: A ``Column`` that evaluates to a JSON string.
+/// - schema: A ``StructType`` schema.
+/// - options: Options to control how the JSON is parsed. It accepts the same options as
+/// the JSON data source.
+/// - Returns: A ``Column`` of the type given by the schema.
+public func from_json(
+ _ col: Column, _ schema: StructType, _ options: [String: String] = [:]
+) -> Column {
+ return from_json(col, schema.toDDL, options)
+}
+
+/// Parses a column containing a JSON string into a struct, an array, or a map with the given
+/// schema. Returns `NULL`, in the case of an unparseable string.
+/// - Parameters:
+/// - col: A ``Column`` that evaluates to a JSON string.
+/// - schema: A ``Column`` that evaluates to a constant schema string,
+/// e.g. a `schema_of_json` result.
+/// - options: Options to control how the JSON is parsed. It accepts the same options as
+/// the JSON data source.
+/// - Returns: A ``Column`` of the type given by the schema.
+public func from_json(
+ _ col: Column, _ schema: Column, _ options: [String: String] = [:]
+) -> Column {
+ return fn("from_json", options: options, col, schema)
+}
+
+/// Extracts a JSON object from a JSON string based on the given JSON path.
+/// - Parameters:
+/// - col: A ``Column`` that evaluates to a JSON string.
+/// - path: A JSON path to extract, e.g. `$.a.b`.
+/// - Returns: A ``Column`` that evaluates to a string.
+public func get_json_object(_ col: Column, _ path: String) -> Column {
+ return fn("get_json_object", col, lit(path))
+}
+
+/// Returns the number of elements in the outermost JSON array. `NULL` is returned in case of
+/// any other valid JSON string, `NULL` or an invalid JSON.
+/// - Parameter col: A ``Column`` that evaluates to a JSON array string.
+/// - Returns: A ``Column``.
+public func json_array_length(_ col: Column) -> Column {
+ return fn("json_array_length", col)
+}
+
+/// Returns all the keys of the outermost JSON object as an array. `NULL` is returned in case of
+/// any other valid JSON string, `NULL` or an invalid JSON.
+/// - Parameter col: A ``Column`` that evaluates to a JSON object string.
+/// - Returns: A ``Column``.
+public func json_object_keys(_ col: Column) -> Column {
+ return fn("json_object_keys", col)
+}
+
+/// Creates a new row for a JSON column according to the given field names.
+/// - Parameters:
+/// - col: A ``Column`` that evaluates to a JSON string.
+/// - fields: The field names to extract.
+/// - Returns: A ``Column``.
+public func json_tuple(_ col: Column, _ fields: String...) -> Column {
+ return fn("json_tuple", [col] + fields.map { lit($0) })
+}
+
+/// Parses a JSON string and infers its schema in DDL format.
+/// - Parameters:
+/// - json: A JSON string.
+/// - options: Options to control how the JSON is parsed. It accepts the same options as
+/// the JSON data source.
+/// - Returns: A ``Column`` that evaluates to a schema string in DDL format.
+public func schema_of_json(_ json: String, _ options: [String: String] = [:]) -> Column {
+ return schema_of_json(lit(json), options)
+}
+
+/// Parses a JSON string and infers its schema in DDL format.
+/// - Parameters:
+/// - json: A ``Column`` that evaluates to a constant JSON string.
+/// - options: Options to control how the JSON is parsed. It accepts the same options as
+/// the JSON data source.
+/// - Returns: A ``Column`` that evaluates to a schema string in DDL format.
+public func schema_of_json(_ json: Column, _ options: [String: String] = [:]) -> Column {
+ return fn("schema_of_json", options: options, json)
+}
+
+/// Converts a column containing a struct, an array, a map, or a variant into a JSON string.
+/// Throws an exception, in the case of an unsupported type.
+/// - Parameters:
+/// - col: A ``Column`` that evaluates to a struct, an array, a map, or a variant.
+/// - options: Options to control how the column is converted. It accepts the same options as
+/// the JSON data source and additionally the `pretty` option.
+/// - Returns: A ``Column`` that evaluates to a string.
+public func to_json(_ col: Column, _ options: [String: String] = [:]) -> Column {
+ return fn("to_json", options: options, col)
+}
diff --git a/Tests/SparkConnectTests/CsvFunctionsTests.swift b/Tests/SparkConnectTests/CsvFunctionsTests.swift
new file mode 100644
index 0000000..ecf799f
--- /dev/null
+++ b/Tests/SparkConnectTests/CsvFunctionsTests.swift
@@ -0,0 +1,104 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+import Testing
+
+@testable import SparkConnect
+
+/// A test suite for `CsvFunctions`
+@Suite(.serialized)
+struct CsvFunctionsTests {
+
+ @Test
+ func csvFunctions() throws {
+ for (column, name) in [
+ (to_csv(col("a")), "to_csv"),
+ (schema_of_csv(col("a")), "schema_of_csv"),
+ ] {
+ let expr = column.expr
+ #expect(expr.unresolvedFunction.functionName == name)
+ #expect(expr.unresolvedFunction.arguments.count == 1)
+ #expect(expr.unresolvedFunction.arguments[0].unresolvedAttribute.unparsedIdentifier == "a")
+ }
+
+ let schemaOfCsv = schema_of_csv("1,a").expr
+ #expect(schemaOfCsv.unresolvedFunction.functionName == "schema_of_csv")
+ #expect(schemaOfCsv.unresolvedFunction.arguments[0].literal.string == "1,a")
+ }
+
+ @Test
+ func fromCsvSchema() throws {
+ let ddl = from_csv(col("a"), "b INT").expr
+ #expect(ddl.unresolvedFunction.functionName == "from_csv")
+ #expect(ddl.unresolvedFunction.arguments.count == 2)
+ #expect(ddl.unresolvedFunction.arguments[1].literal.string == "b INT")
+
+ let column = from_csv(col("a"), schema_of_csv("1")).expr
+ #expect(column.unresolvedFunction.functionName == "from_csv")
+ let schema = column.unresolvedFunction.arguments[1].unresolvedFunction
+ #expect(schema.functionName == "schema_of_csv")
+ }
+
+ /// Options are appended as a single `map` argument like Spark SQL's `Column.fnWithOptions`.
+ @Test
+ func options() throws {
+ #expect(to_csv(col("a"), [:]).expr == to_csv(col("a")).expr)
+
+ let expr = from_csv(col("a"), "b INT", ["sep": ";", "mode": "FAILFAST"]).expr
+ #expect(expr.unresolvedFunction.functionName == "from_csv")
+ #expect(expr.unresolvedFunction.arguments.count == 3)
+ let options = expr.unresolvedFunction.arguments[2].unresolvedFunction
+ #expect(options.functionName == "map")
+ #expect(options.arguments.map { $0.literal.string } == ["mode", "FAILFAST", "sep", ";"])
+ }
+
+ @Test
+ func roundTrip() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let df = try await spark.sql("SELECT 1 AS id, 'a' AS name")
+ let csv = to_csv(`struct`(col("id"), col("name")))
+ let parsed = from_csv(csv, "id INT, name STRING").alias("parsed")
+ let rows = try await df.select(parsed).selectExpr("parsed.id", "parsed.name").collect()
+ #expect(rows == [Row(Int32(1), "a")])
+ await spark.stop()
+ }
+
+ @Test
+ func schemaOfCsv() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let rows = try await spark.range(1).select(
+ schema_of_csv("1,abc"),
+ schema_of_csv(lit("1;abc"), ["sep": ";"])
+ ).collect()
+ #expect(rows == [Row("STRUCT<_c0: INT, _c1: STRING>", "STRUCT<_c0: INT, _c1: STRING>")])
+ await spark.stop()
+ }
+
+ @Test
+ func csvWithOptions() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let df = try await spark.sql("SELECT 1 AS id, 'a' AS name")
+ let rows = try await df.select(
+ to_csv(`struct`(col("id"), col("name")), ["sep": ";"]),
+ from_csv(lit("1;a"), "id INT, name STRING", ["sep": ";"])
+ ).collect()
+ #expect(try rows[0].get(0) as! String == "1;a")
+ await spark.stop()
+ }
+}
diff --git a/Tests/SparkConnectTests/JsonFunctionsTests.swift b/Tests/SparkConnectTests/JsonFunctionsTests.swift
new file mode 100644
index 0000000..a406fdb
--- /dev/null
+++ b/Tests/SparkConnectTests/JsonFunctionsTests.swift
@@ -0,0 +1,146 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+import Testing
+
+@testable import SparkConnect
+
+/// A test suite for `JsonFunctions`
+@Suite(.serialized)
+struct JsonFunctionsTests {
+
+ @Test
+ func jsonFunctions() throws {
+ for (column, name) in [
+ (json_array_length(col("a")), "json_array_length"),
+ (json_object_keys(col("a")), "json_object_keys"),
+ (to_json(col("a")), "to_json"),
+ (schema_of_json(col("a")), "schema_of_json"),
+ ] {
+ let expr = column.expr
+ #expect(expr.unresolvedFunction.functionName == name)
+ #expect(expr.unresolvedFunction.arguments.count == 1)
+ #expect(expr.unresolvedFunction.arguments[0].unresolvedAttribute.unparsedIdentifier == "a")
+ }
+ }
+
+ @Test
+ func jsonFunctionArguments() throws {
+ let getJsonObject = get_json_object(col("a"), "$.b").expr
+ #expect(getJsonObject.unresolvedFunction.functionName == "get_json_object")
+ #expect(getJsonObject.unresolvedFunction.arguments[1].literal.string == "$.b")
+
+ let jsonTuple = json_tuple(col("a"), "b", "c").expr
+ #expect(jsonTuple.unresolvedFunction.functionName == "json_tuple")
+ #expect(jsonTuple.unresolvedFunction.arguments.count == 3)
+ #expect(jsonTuple.unresolvedFunction.arguments[1].literal.string == "b")
+ #expect(jsonTuple.unresolvedFunction.arguments[2].literal.string == "c")
+
+ let schemaOfJson = schema_of_json("{\"a\": 1}").expr
+ #expect(schemaOfJson.unresolvedFunction.functionName == "schema_of_json")
+ #expect(schemaOfJson.unresolvedFunction.arguments[0].literal.string == "{\"a\": 1}")
+ }
+
+ @Test
+ func fromJsonSchema() throws {
+ let ddl = from_json(col("a"), "b INT").expr
+ #expect(ddl.unresolvedFunction.functionName == "from_json")
+ #expect(ddl.unresolvedFunction.arguments.count == 2)
+ #expect(ddl.unresolvedFunction.arguments[1].literal.string == "b INT")
+
+ let schema = StructType(fields: [StructField(name: "b", dataType: .integer)])
+ #expect(from_json(col("a"), schema).expr == ddl)
+
+ let column = from_json(col("a"), schema_of_json("{\"b\": 1}")).expr
+ #expect(column.unresolvedFunction.functionName == "from_json")
+ let inferred = column.unresolvedFunction.arguments[1].unresolvedFunction
+ #expect(inferred.functionName == "schema_of_json")
+ }
+
+ /// Options are appended as a single `map` argument like Spark SQL's `Column.fnWithOptions`.
+ @Test
+ func options() throws {
+ #expect(to_json(col("a"), [:]).expr == to_json(col("a")).expr)
+
+ let expr = to_json(col("a"), ["pretty": "true", "ignoreNullFields": "false"]).expr
+ #expect(expr.unresolvedFunction.functionName == "to_json")
+ #expect(expr.unresolvedFunction.arguments.count == 2)
+ let options = expr.unresolvedFunction.arguments[1].unresolvedFunction
+ #expect(options.functionName == "map")
+ let entries = options.arguments.map { $0.literal.string }
+ #expect(entries == ["ignoreNullFields", "false", "pretty", "true"])
+ }
+
+ @Test
+ func roundTrip() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let df = try await spark.sql("SELECT 1 AS id, 'a' AS name")
+ let schema = "id INT, name STRING"
+ let parsed = from_json(to_json(`struct`(col("id"), col("name"))), schema).alias("parsed")
+ let rows = try await df.select(parsed).selectExpr("parsed.id", "parsed.name").collect()
+ #expect(rows == [Row(Int32(1), "a")])
+ await spark.stop()
+ }
+
+ @Test
+ func schemaOfJson() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let rows = try await spark.range(1).select(
+ schema_of_json("{\"a\": 1, \"b\": \"x\"}"),
+ schema_of_json(lit("[1, 2]"))
+ ).collect()
+ #expect(rows == [Row("STRUCT<a: BIGINT, b: STRING>", "ARRAY<BIGINT>")])
+ await spark.stop()
+ }
+
+ @Test
+ func jsonAccessors() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let rows = try await spark.range(1).select(
+ get_json_object(lit("{\"a\": {\"b\": 1}}"), "$.a.b"),
+ json_array_length(lit("[1, 2, 3]")),
+ json_object_keys(lit("{\"a\": 1, \"b\": 2}"))
+ ).collect()
+ #expect(try rows[0].get(0) as! String == "1")
+ #expect(try rows[0].get(1) as! Int32 == 3)
+ #expect(try rows[0].get(2) as! [String] == ["a", "b"])
+ await spark.stop()
+ }
+
+ @Test
+ func jsonTuple() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let rows = try await spark.range(1)
+ .select(json_tuple(lit("{\"a\": 1, \"b\": 2}"), "a", "b")).collect()
+ #expect(rows == [Row("1", "2")])
+ await spark.stop()
+ }
+
+ @Test
+ func toJsonWithOptions() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let df = try await spark.sql("SELECT 1 AS id, CAST(NULL AS STRING) AS name")
+ let rows = try await df.select(
+ to_json(`struct`(col("id"), col("name"))),
+ to_json(`struct`(col("id"), col("name")), ["ignoreNullFields": "false"])
+ ).collect()
+ #expect(rows == [Row("{\"id\":1}", "{\"id\":1,\"name\":null}")])
+ await spark.stop()
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org