This is an automated email from the ASF dual-hosted git repository.
MaxGekk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 59fdb3e3ecff [SPARK-57456][SQL] Support nanosecond-precision timestamp
types in the JSON datasource (v1 and v2)
59fdb3e3ecff is described below
commit 59fdb3e3ecff20ac8df368a5c1181327332c3495
Author: Maxim Gekk <[email protected]>
AuthorDate: Tue Jun 30 16:44:16 2026 +0200
[SPARK-57456][SQL] Support nanosecond-precision timestamp types in the JSON
datasource (v1 and v2)
### What changes were proposed in this pull request?
Umbrella: [SPARK-56822](https://issues.apache.org/jira/browse/SPARK-56822)
(Timestamps with nanosecond precision).
This PR adds read and write support for the nanosecond-capable timestamp
types `TIMESTAMP_NTZ(p)` and `TIMESTAMP_LTZ(p)` (`p` in 7-9) to the JSON
datasource, for both the v1 (`JsonFileFormat`) and v2 (`JsonTable`) paths,
reaching parity with the microsecond `TimestampType` / `TimestampNTZType`, and
removes the [SPARK-57166](https://issues.apache.org/jira/browse/SPARK-57166)
rejection guardrail.
Specifically:
- `JacksonParser`: adds `TimestampLTZNanosType` / `TimestampNTZNanosType`
read cases that delegate to the existing `parseNanos` /
`parseWithoutTimeZoneNanos` formatter methods with the column precision.
- `JacksonGenerator`: adds the corresponding write cases that delegate to
`formatNanos` / `formatWithoutTimeZoneNanos`.
- `JsonFileFormat` (v1) and `JsonTable` (v2): drop the
`AnyTimestampNanoType` rejection in `supportDataType` / `supportsDataType`.
Notes:
- Schema inference (`JsonInferSchema`) keeps inferring microsecond
`TimestampType` / `TimestampNTZType` by default; nanosecond types are reached
only via an explicit user schema.
- No new options: the existing `timestampFormat` / `timestampNTZFormat`
options drive the nanos path. The column type carries the precision, and the
count of `S` letters in the pattern controls how many fractional-second digits
are emitted on write (text output needs up to 9 `S` for full precision; reads
with the default formatter parse the full fraction and truncate to the declared
precision).
- The legacy time parser policy rejects nanos: the legacy LTZ formatter
cannot represent sub-microsecond digits, so it raises
`UNSUPPORTED_FEATURE.TIMESTAMP_NANOS_WITH_LEGACY_TIME_PARSER` (the NTZ
formatter always uses the ISO-8601 path).
### Why are the changes needed?
JSON rejected nanos timestamp types in its datasource capability checks and
lacked the conversions to round-trip them, so these columns could not be
written or read through JSON. This extends nanosecond-precision timestamp
support (umbrella SPARK-56822) to the JSON datasource, matching the existing
microsecond timestamp behavior and the Parquet/ORC/Avro/CSV nanosecond support.
### Does this PR introduce _any_ user-facing change?
Yes. With `spark.sql.timestampNanosTypes.enabled=true`, columns of type
`TIMESTAMP_NTZ(7-9)` / `TIMESTAMP_LTZ(7-9)` can now be written to and read from
JSON files, and parsed/generated by `from_json` / `to_json`. Previously such
columns were rejected with `UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE`. This is a
change within the unreleased master/branch only.
### How was this patch tested?
- `JsonExpressionsSuite`: `JsonToStructs` nanosecond parsing at the
catalyst expression level.
- `JsonFunctionsSuite`: flipped the existing `from_json` nanosecond test to
assert successful parsing and the truncated value (instead of an
unsupported-type error); added `to_json` and `to_json` / `from_json` round-trip
tests.
- `FileBasedDataSourceSuite`: removed JSON from the SPARK-57166 rejection
list; added end-to-end round-trip (precisions 7-9, NTZ and LTZ, v1 and v2), a
nested struct/array/map round-trip, and a LEGACY time-parser-policy rejection
test (write and read).
- `JsonSuite`: `DataFrameReader.json(Dataset[String])` read, a
custom-schema file round-trip, and a mixed microsecond/nanosecond schema
round-trip; these run under the `JsonV1Suite`, `JsonV2Suite`,
`JsonLegacyTimeParserSuite`, and `JsonUnsafeRowSuite` variants.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor 2.1, Claude Opus 4.8
Closes #56865 from MaxGekk/nanos-json-ds.
Authored-by: Maxim Gekk <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
---
.../spark/sql/catalyst/json/JacksonGenerator.scala | 13 ++
.../spark/sql/catalyst/json/JacksonParser.scala | 17 +-
.../expressions/JsonExpressionsSuite.scala | 29 +++
.../datasources/json/JsonFileFormat.scala | 3 -
.../execution/datasources/v2/json/JsonTable.scala | 3 -
.../spark/sql/FileBasedDataSourceSuite.scala | 202 +++++++++++++++------
.../org/apache/spark/sql/JsonFunctionsSuite.scala | 85 ++++++++-
.../sql/execution/datasources/json/JsonSuite.scala | 97 +++++++++-
8 files changed, 371 insertions(+), 78 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala
index f34fab37d1de..7abf86850fc4 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala
@@ -159,6 +159,19 @@ class JacksonGenerator(
timestampNTZFormatter.format(DateTimeUtils.microsToLocalDateTime(row.getLong(ordinal)))
gen.writeString(timestampString)
+ case t: TimestampLTZNanosType =>
+ (row: SpecializedGetters, ordinal: Int) =>
+ val timestampString =
+ timestampFormatter.formatNanos(row.getTimestampLTZNanos(ordinal),
t.precision)
+ gen.writeString(timestampString)
+
+ case t: TimestampNTZNanosType =>
+ (row: SpecializedGetters, ordinal: Int) =>
+ val timestampString =
+ timestampNTZFormatter.formatWithoutTimeZoneNanos(
+ row.getTimestampNTZNanos(ordinal), t.precision)
+ gen.writeString(timestampString)
+
case DateType =>
(row: SpecializedGetters, ordinal: Int) =>
val dateString = dateFormatter.format(row.getInt(ordinal))
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
index fa03a9aea833..f5d90c5a09bf 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
@@ -39,7 +39,7 @@ import org.apache.spark.sql.internal.{LegacyBehaviorPolicy,
SQLConf}
import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types._
import org.apache.spark.types.variant._
-import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String, VariantVal}
+import org.apache.spark.unsafe.types.{CalendarInterval, TimestampNanosVal,
UTF8String, VariantVal}
import org.apache.spark.util.Utils
/**
@@ -380,6 +380,21 @@ class JacksonParser(
timestampNTZFormatter.parseWithoutTimeZone(parser.getText, false)
}
+ case t: TimestampLTZNanosType =>
+ (parser: JsonParser) => parseJsonToken[TimestampNanosVal](parser,
dataType) {
+ // Unlike the microsecond TimestampType, the nanosecond types accept
only string input.
+ // The numeric-epoch shorthand (a JSON integer read as epoch seconds)
is legacy
+ // TimestampType behavior and is intentionally not carried over to the
nanos types.
+ case VALUE_STRING if parser.getTextLength >= 1 =>
+ timestampFormatter.parseNanos(parser.getText, t.precision)
+ }
+
+ case t: TimestampNTZNanosType =>
+ (parser: JsonParser) => parseJsonToken[TimestampNanosVal](parser,
dataType) {
+ case VALUE_STRING if parser.getTextLength >= 1 =>
+ timestampNTZFormatter.parseWithoutTimeZoneNanos(parser.getText,
t.precision, false)
+ }
+
case DateType =>
(parser: JsonParser) => parseJsonToken[java.lang.Integer](parser,
dataType) {
case VALUE_STRING if parser.getTextLength >= 1 =>
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
index 7f6ae46be3a8..37916f5a93be 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.catalyst.expressions
import java.text.{DecimalFormat, DecimalFormatSymbols, SimpleDateFormat}
+import java.time.{LocalDateTime, ZoneOffset}
import java.util.{Calendar, Locale, TimeZone}
import org.scalatest.exceptions.TestFailedException
@@ -30,6 +31,7 @@ import org.apache.spark.sql.catalyst.expressions.Cast._
import
org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{PST, UTC, UTC_OPT}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
@@ -570,6 +572,33 @@ class JsonExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper {
}
}
+ test("SPARK-57456: from_json with nanos timestamp") {
+ val jsonData = """{"t": "2016-01-01T00:00:00.123456789"}"""
+ // No timestamp format option: the default formatter parses the full
sub-second fraction and
+ // truncates the sub-precision digits toward zero to the declared
precision.
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ TimestampNanosTestUtils.foreachNanosPrecision { p =>
+ val nano = TimestampNanosTestUtils.nanoOfSecTruncator(p)(123456789)
+ val ldt = LocalDateTime.of(2016, 1, 1, 0, 0, 0, nano)
+ checkEvaluation(
+ JsonToStructs(
+ StructType(StructField("t", TimestampNTZNanosType(p)) :: Nil),
+ Map.empty[String, String],
+ Literal(jsonData),
+ UTC_OPT),
+ InternalRow(TimestampNanosTestUtils.localDateTimeToNanosVal(ldt)))
+ // LTZ: the string has no zone, so it is interpreted in the given time
zone (UTC here).
+ checkEvaluation(
+ JsonToStructs(
+ StructType(StructField("t", TimestampLTZNanosType(p)) :: Nil),
+ Map.empty[String, String],
+ Literal(jsonData),
+ UTC_OPT),
+
InternalRow(TimestampNanosTestUtils.instantToNanosVal(ldt.toInstant(ZoneOffset.UTC))))
+ }
+ }
+ }
+
test("SPARK-19543: from_json empty input column") {
val schema = StructType(StructField("a", IntegerType) :: Nil)
checkEvaluation(
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
index 09883c0786fa..ee01d67d1f7b 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
@@ -127,9 +127,6 @@ case class JsonFileFormat() extends TextBasedFileFormat
with DataSourceRegister
case _: GeometryType | _: GeographyType => false
- // Nanosecond-capable timestamps are not yet supported by this datasource.
- case _: AnyTimestampNanoType => false
-
case _: AtomicType => true
case st: StructType => st.forall { f => supportDataType(f.dataType) }
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/json/JsonTable.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/json/JsonTable.scala
index 0612041c1a72..74095e85a0f6 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/json/JsonTable.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/json/JsonTable.scala
@@ -61,9 +61,6 @@ case class JsonTable(
override def supportsDataType(dataType: DataType): Boolean = dataType match {
case _: GeometryType | _: GeographyType => false
- // Nanosecond-capable timestamps are not yet supported by this datasource.
- case _: AnyTimestampNanoType => false
-
case _: AtomicType => true
case st: StructType => st.forall { f => supportsDataType(f.dataType) }
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
index 16d2bc0d3082..1251d78818b2 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
@@ -46,7 +46,6 @@ import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
-import org.apache.spark.unsafe.types.TimestampNanosVal
import org.apache.spark.util.HadoopFSUtils
class FileBasedDataSourceSuite extends SharedSparkSession
@@ -1339,65 +1338,6 @@ class FileBasedDataSourceSuite extends SharedSparkSession
}
}
- test("SPARK-57166: nanosecond timestamp types are not supported in selected
file data sources") {
- // Parquet and ORC support nanosecond-capable timestamps, while these
formats still reject them.
- val unsupportedDataSources = Seq("json")
- val nanosTypes = Seq(TimestampNTZNanosType(9), TimestampLTZNanosType(9))
- withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
- // Test both v1 and v2 data sources.
- Seq(true, false).foreach { useV1 =>
- val useV1List = if (useV1) {
- unsupportedDataSources.mkString(",")
- } else {
- ""
- }
- withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1List) {
- unsupportedDataSources.foreach { format =>
- nanosTypes.foreach { nanosType =>
- val expectedType = s""""${nanosType.sql}""""
- withTempDir { dir =>
- // Write path: a nanos-typed column cannot be written. The
nanos literal is built
- // directly from its internal value to avoid relying on
cast/parser support.
- val nanosLiteral =
- Literal.create(new TimestampNanosVal(0L, 0.toShort),
nanosType)
- val df = spark.range(1).select(Column(nanosLiteral).as("ts"))
- val writeDir = new File(dir, "write").getCanonicalPath
- checkError(
- exception = intercept[AnalysisException] {
- df.write.format(format).mode("overwrite").save(writeDir)
- },
- condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
- parameters = Map(
- "columnName" -> "`ts`",
- "columnType" -> expectedType,
- "format" -> formatMapping(format)))
-
- // Read path: a user-specified nanos schema is rejected. Write
a benign file first
- // so schema validation (not file listing) is what fails.
- val readDir = new File(dir, "read").getCanonicalPath
- // XML requires a `rowTag` option on both the read and write
paths.
- val extraOptions =
- if (format == "xml") Map("rowTag" -> "row") else
Map.empty[String, String]
- Seq("a").toDF("ts").write.format(format).options(extraOptions)
- .mode("overwrite").save(readDir)
- checkError(
- exception = intercept[AnalysisException] {
- spark.read.schema(new StructType().add("ts", nanosType))
-
.format(format).options(extraOptions).load(readDir).collect()
- },
- condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
- parameters = Map(
- "columnName" -> "`ts`",
- "columnType" -> expectedType,
- "format" -> formatMapping(format)))
- }
- }
- }
- }
- }
- }
- }
-
test("SPARK-57166: ORC supports nanosecond timestamp types in v1 and v2") {
withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
// Validate both v1 and v2 ORC paths.
@@ -1473,6 +1413,148 @@ class FileBasedDataSourceSuite extends
SharedSparkSession
}
}
+ test("SPARK-57456: JSON supports nanosecond timestamp types in v1 and v2") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ Seq(true, false).foreach { useV1 =>
+ val useV1List = if (useV1) "json" else ""
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1List) {
+ foreachNanosPrecision { precision =>
+ // JSON is text-based: the format string must carry enough
fractional-second digits to
+ // represent the full precision. Use exactly `precision`
S-characters.
+ val fracPat = "S" * precision
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ withTempDir { dir =>
+ val wallClock = LocalDateTime.of(1970, 1, 1, 0, 20, 34,
567890123)
+ val value: Any = nanosType match {
+ case _: TimestampNTZNanosType => wallClock
+ case _: TimestampLTZNanosType =>
wallClock.toInstant(ZoneOffset.UTC)
+ }
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(value))),
+ new StructType().add("ts", nanosType))
+ val path = new File(dir,
s"json_nanos_${nanosType.typeName}").getCanonicalPath
+ val (fmtKey, fmtVal) = nanosType match {
+ case _: TimestampNTZNanosType =>
+ ("timestampNTZFormat", s"yyyy-MM-dd'T'HH:mm:ss.$fracPat")
+ case _: TimestampLTZNanosType =>
+ ("timestampFormat",
s"yyyy-MM-dd'T'HH:mm:ss.${fracPat}XXX")
+ }
+ df.write.format("json").option(fmtKey,
fmtVal).mode("overwrite").save(path)
+ val readBack = spark.read
+ .schema(new StructType().add("ts", nanosType))
+ .option(fmtKey, fmtVal)
+ .format("json").load(path)
+ checkAnswer(readBack, df)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57456: JSON supports nested nanosecond timestamp types in v1 and
v2") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ Seq(true, false).foreach { useV1 =>
+ val useV1List = if (useV1) "json" else ""
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1List) {
+ foreachNanosPrecision { precision =>
+ val fracPat = "S" * precision
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ withTempDir { dir =>
+ val wallClock = LocalDateTime.of(1970, 1, 1, 0, 20, 34,
567890123)
+ val leaf: Any = nanosType match {
+ case _: TimestampNTZNanosType => wallClock
+ case _: TimestampLTZNanosType =>
wallClock.toInstant(ZoneOffset.UTC)
+ }
+ // Embed the nanos leaf inside a struct, an array, and a map
value. The guardrails
+ // and Jackson read/write paths recurse into all three.
+ val schema = new StructType()
+ .add("s", new StructType().add("ts", nanosType))
+ .add("a", ArrayType(nanosType))
+ .add("m", MapType(StringType, nanosType))
+ val row = Row(Row(leaf), Seq(leaf), Map("k" -> leaf))
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(row)), schema)
+ val (fmtKey, fmtVal) = nanosType match {
+ case _: TimestampNTZNanosType =>
+ ("timestampNTZFormat", s"yyyy-MM-dd'T'HH:mm:ss.$fracPat")
+ case _: TimestampLTZNanosType =>
+ ("timestampFormat",
s"yyyy-MM-dd'T'HH:mm:ss.${fracPat}XXX")
+ }
+ val path =
+ new File(dir,
s"json_nested_${nanosType.typeName}").getCanonicalPath
+ df.write.format("json").option(fmtKey,
fmtVal).mode("overwrite").save(path)
+ val readBack = spark.read.schema(schema).option(fmtKey,
fmtVal)
+ .format("json").load(path)
+ checkAnswer(readBack, df)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57456: JSON rejects nanosecond timestamps under the LEGACY time
parser policy") {
+ // The legacy timestamp formatter cannot represent sub-microsecond digits,
so the nanos
+ // formatter methods raise TIMESTAMP_NANOS_WITH_LEGACY_TIME_PARSER. Only
the LTZ formatter is
+ // legacy under this policy (the NTZ formatter always uses the ISO-8601
path), so this covers
+ // TimestampLTZNanosType.
+ def rootNanosError(e: Throwable): SparkUnsupportedOperationException = {
+ var cause: Throwable = e
+ while (cause != null &&
!cause.isInstanceOf[SparkUnsupportedOperationException]) {
+ cause = cause.getCause
+ }
+ assert(cause != null,
+ s"Expected TIMESTAMP_NANOS_WITH_LEGACY_TIME_PARSER, but got:
${e.getMessage}")
+ cause.asInstanceOf[SparkUnsupportedOperationException]
+ }
+
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.LEGACY_TIME_PARSER_POLICY.key -> "LEGACY") {
+ val nanosType = TimestampLTZNanosType(9)
+ val schema = new StructType().add("ts", nanosType)
+ val expectedParameters =
+ Map("config" -> ("\"" + SQLConf.LEGACY_TIME_PARSER_POLICY.key + "\""))
+ Seq(true, false).foreach { useV1 =>
+ val useV1List = if (useV1) "json" else ""
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1List) {
+ withTempDir { dir =>
+ // Write path.
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(
+ Seq(Row(LocalDateTime.of(2020, 1, 1, 0, 0, 0,
1).toInstant(ZoneOffset.UTC)))),
+ schema)
+ val writeDir = new File(dir, "write").getCanonicalPath
+ checkError(
+ exception = rootNanosError(intercept[SparkException] {
+ df.write.format("json").mode("overwrite").save(writeDir)
+ }),
+ condition =
"UNSUPPORTED_FEATURE.TIMESTAMP_NANOS_WITH_LEGACY_TIME_PARSER",
+ parameters = expectedParameters)
+
+ // Read path: write a benign file first so schema-driven parsing
is what fails. Use
+ // FAILFAST so the unsupported-feature error surfaces instead of
being turned into a
+ // null record by the permissive bad-record handling.
+ val readDir = new File(dir, "read").getCanonicalPath
+
Seq("a").toDF("ts").write.format("json").mode("overwrite").save(readDir)
+ checkError(
+ exception = rootNanosError(intercept[SparkException] {
+ spark.read.schema(schema).option("mode", "FAILFAST")
+ .format("json").load(readDir).collect()
+ }),
+ condition =
"UNSUPPORTED_FEATURE.TIMESTAMP_NANOS_WITH_LEGACY_TIME_PARSER",
+ parameters = expectedParameters)
+ }
+ }
+ }
+ }
+ }
+
// Asserts the ignoredPathSegmentRegex contract for `format`: the default
regex hides the
// '_'-prefixed file; a never-matching per-read option or session conf each
surface it; the
// option overrides the conf.
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
index 20deafb8b386..396b86144f4d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
@@ -18,19 +18,19 @@
package org.apache.spark.sql
import java.text.SimpleDateFormat
-import java.time.{Duration, LocalDateTime, Period}
+import java.time.{Duration, LocalDateTime, Period, ZoneOffset}
import java.util.Locale
import scala.jdk.CollectionConverters._
import com.fasterxml.jackson.core.StreamReadConstraints
-import org.apache.spark.{SparkException, SparkRuntimeException,
SparkUnsupportedOperationException}
+import org.apache.spark.{SparkException, SparkRuntimeException}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{JsonToStructs, Literal,
MultiGetJsonObject}
import org.apache.spark.sql.catalyst.expressions.Cast._
+import org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils
import
org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPrecision
-import org.apache.spark.sql.errors.DataTypeErrors.toSQLType
import org.apache.spark.sql.execution.{InputAdapter, SparkPlan,
WholeStageCodegenExec}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
@@ -609,8 +609,12 @@ class JsonFunctionsSuite extends SharedSparkSession {
test("SPARK-57164: from_json with a nanos timestamp DDL schema string") {
val df = Seq("""{"c": "2020-01-01T00:00:00.123456789"}""").toDF("value")
- withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ // Fix the session timezone so the TIMESTAMP_LTZ expected value is
deterministic.
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
foreachNanosPrecision { p =>
+ val nano = TimestampNanosTestUtils.nanoOfSecTruncator(p)(123456789)
Seq(
s"TIMESTAMP_NTZ($p)" -> TimestampNTZNanosType(p),
s"TIMESTAMP_LTZ($p)" -> TimestampLTZNanosType(p),
@@ -621,12 +625,73 @@ class JsonFunctionsSuite extends SharedSparkSession {
from_json($"value", s"c $spelling", Map.empty[String,
String]).as("v"))
// The schema string resolves to the nanos type ...
assert(parsed.schema("v").dataType.asInstanceOf[StructType]("c").dataType ===
expected)
- // ... but the JSON datasource does not support nanosecond
timestamps yet, so the
- // value converter rejects it at execution.
- checkError(
- exception =
intercept[SparkUnsupportedOperationException](parsed.collect()),
- condition = "UNSUPPORTED_DATATYPE",
- parameters = Map("typeName" -> toSQLType(expected)))
+ // ... and the JSON datasource correctly parses the nanosecond
timestamp, truncating
+ // sub-precision digits toward zero.
+ val expectedValue = expected match {
+ case _: TimestampNTZNanosType => LocalDateTime.of(2020, 1, 1, 0,
0, 0, nano)
+ case _: TimestampLTZNanosType =>
+ LocalDateTime.of(2020, 1, 1, 0, 0, 0,
nano).toInstant(ZoneOffset.UTC)
+ }
+ checkAnswer(parsed, Row(Row(expectedValue)))
+ }
+ }
+ }
+ }
+
+ test("SPARK-57456: to_json with nanos timestamp types") {
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ foreachNanosPrecision { p =>
+ // The pattern must carry `p` fractional digits to emit the full
declared precision; the
+ // floored value has exactly `p` significant digits, so the rendered
fraction is the first
+ // `p` digits of 123456789.
+ val fracPat = "S" * p
+ val frac = "123456789".take(p)
+ val ldt = LocalDateTime.of(2020, 1, 1, 0, 0, 0, 123456789)
+ Seq(
+ (TimestampNTZNanosType(p): DataType, "timestampNTZFormat",
+ s"yyyy-MM-dd'T'HH:mm:ss.$fracPat", ldt: Any,
+ s"""{"ts":"2020-01-01T00:00:00.$frac"}"""),
+ (TimestampLTZNanosType(p): DataType, "timestampFormat",
+ s"yyyy-MM-dd'T'HH:mm:ss.${fracPat}XXX",
ldt.toInstant(ZoneOffset.UTC): Any,
+ s"""{"ts":"2020-01-01T00:00:00.${frac}Z"}""")).foreach {
+ case (nanosType, optKey, fmt, value, expectedJson) =>
+ val schema = new StructType().add("ts", nanosType)
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(value))), schema)
+ checkAnswer(
+ df.select(to_json(struct($"ts"), Map(optKey -> fmt))),
+ Row(expectedJson))
+ }
+ }
+ }
+ }
+
+ test("SPARK-57456: roundtrip in to_json and from_json - nanos timestamps") {
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ foreachNanosPrecision { p =>
+ val fracPat = "S" * p
+ val ldt = LocalDateTime.of(2020, 1, 1, 0, 0, 0, 123456789)
+ Seq(
+ (TimestampNTZNanosType(p): DataType, "timestampNTZFormat",
+ s"yyyy-MM-dd'T'HH:mm:ss.$fracPat", ldt: Any),
+ (TimestampLTZNanosType(p): DataType, "timestampFormat",
+ s"yyyy-MM-dd'T'HH:mm:ss.${fracPat}XXX",
ldt.toInstant(ZoneOffset.UTC): Any)).foreach {
+ case (nanosType, optKey, fmt, value) =>
+ val schema = new StructType().add("ts", nanosType)
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(value))), schema)
+ val options = Map(optKey -> fmt)
+ // The input column already carries precision `p`, so the to_json
-> from_json
+ // round-trip with a `p`-digit pattern is loss-free.
+ val readBack = df
+ .select(to_json(struct($"ts"), options).as("json"))
+ .select(from_json($"json", schema, options).as("data"))
+ .select($"data.ts".as("ts"))
+ checkAnswer(readBack, df.select($"ts"))
}
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
index 8cdae049723a..99be72b08a94 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
@@ -21,7 +21,7 @@ import java.io._
import java.nio.charset.{Charset, StandardCharsets}
import java.nio.file.Files
import java.sql.{Date, Timestamp}
-import java.time.{Duration, Instant, LocalDate, LocalDateTime, Period, ZoneId}
+import java.time.{Duration, Instant, LocalDate, LocalDateTime, Period, ZoneId,
ZoneOffset}
import java.util.Locale
import java.util.concurrent.atomic.AtomicLong
@@ -39,6 +39,8 @@ import org.apache.spark.sql.{functions => F, _}
import org.apache.spark.sql.catalyst.json._
import org.apache.spark.sql.catalyst.util.{CharsetProvider, DateTimeTestUtils,
DateTimeUtils, HadoopCompressionCodec}
import org.apache.spark.sql.catalyst.util.HadoopCompressionCodec.GZIP
+import org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils
+import
org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPrecision
import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLType
import org.apache.spark.sql.errors.QueryExecutionErrors.toSQLId
import org.apache.spark.sql.execution.ExternalRDD
@@ -2979,6 +2981,99 @@ abstract class JsonSuite
}
}
+ test("SPARK-57456: read nanosecond timestamps via
DataFrameReader.json(Dataset[String])") {
+ val ds = Seq("""{"ts": "2020-01-01T00:00:00.123456789"}""").toDS()
+ // CORRECTED overrides the LEGACY policy set by the inherited
JsonLegacyTimeParserSuite (which
+ // would reject nanos). UTC fixes the LTZ expected value.
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.LEGACY_TIME_PARSER_POLICY.key -> "CORRECTED",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ foreachNanosPrecision { p =>
+ val nano = TimestampNanosTestUtils.nanoOfSecTruncator(p)(123456789)
+ Seq(TimestampNTZNanosType(p), TimestampLTZNanosType(p)).foreach {
nanosType =>
+ val readBack = spark.read.schema(new StructType().add("ts",
nanosType)).json(ds)
+ val expected = nanosType match {
+ case _: TimestampNTZNanosType =>
+ Row(LocalDateTime.of(2020, 1, 1, 0, 0, 0, nano))
+ case _: TimestampLTZNanosType =>
+ Row(LocalDateTime.of(2020, 1, 1, 0, 0, 0,
nano).toInstant(ZoneOffset.UTC))
+ }
+ checkAnswer(readBack, expected)
+ }
+ }
+ }
+ }
+
+ test("SPARK-57456: Roundtrip in reading and writing nanosecond timestamps
with custom schema") {
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.LEGACY_TIME_PARSER_POLICY.key -> "CORRECTED") {
+ foreachNanosPrecision { precision =>
+ // Text output needs a pattern carrying `precision` fractional-second
digits.
+ val fracPat = "S" * precision
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ withTempPath { path =>
+ val wallClock = LocalDateTime.of(1970, 1, 1, 0, 20, 34,
567890123)
+ val value: Any = nanosType match {
+ case _: TimestampNTZNanosType => wallClock
+ case _: TimestampLTZNanosType =>
wallClock.toInstant(ZoneOffset.UTC)
+ }
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(value))),
+ new StructType().add("ts", nanosType))
+ val (fmtKey, fmtVal) = nanosType match {
+ case _: TimestampNTZNanosType =>
+ ("timestampNTZFormat", s"yyyy-MM-dd'T'HH:mm:ss.$fracPat")
+ case _: TimestampLTZNanosType =>
+ ("timestampFormat", s"yyyy-MM-dd'T'HH:mm:ss.${fracPat}XXX")
+ }
+ df.write.option(fmtKey, fmtVal).json(path.getAbsolutePath)
+ val readBack = spark.read.schema(new StructType().add("ts",
nanosType))
+ .option(fmtKey, fmtVal).json(path.getAbsolutePath)
+ checkAnswer(readBack, df)
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57456: mixed microsecond and nanosecond timestamps in one
schema") {
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.LEGACY_TIME_PARSER_POLICY.key -> "CORRECTED",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ withTempPath { path =>
+ // One LTZ pair (micros + nanos) shares `timestampFormat`; one NTZ
pair shares
+ // `timestampNTZFormat`. A single 9-S pattern per kind serves both
precisions: the micros
+ // columns just get trailing-zero fractions.
+ val schema = new StructType()
+ .add("ltz_micros", TimestampType)
+ .add("ltz_nanos", TimestampLTZNanosType(9))
+ .add("ntz_micros", TimestampNTZType)
+ .add("ntz_nanos", TimestampNTZNanosType(9))
+ val row = Row(
+ LocalDateTime.of(2020, 1, 1, 0, 0, 0,
123456000).toInstant(ZoneOffset.UTC),
+ LocalDateTime.of(2020, 1, 1, 0, 0, 0,
123456789).toInstant(ZoneOffset.UTC),
+ LocalDateTime.of(2020, 1, 1, 0, 0, 0, 123456000),
+ LocalDateTime.of(2020, 1, 1, 0, 0, 0, 123456789))
+ val df =
spark.createDataFrame(spark.sparkContext.parallelize(Seq(row)), schema)
+ val ltzFmt = "yyyy-MM-dd'T'HH:mm:ss.SSSSSSSSSXXX"
+ val ntzFmt = "yyyy-MM-dd'T'HH:mm:ss.SSSSSSSSS"
+ df.write
+ .option("timestampFormat", ltzFmt)
+ .option("timestampNTZFormat", ntzFmt)
+ .json(path.getAbsolutePath)
+ val readBack = spark.read.schema(schema)
+ .option("timestampFormat", ltzFmt)
+ .option("timestampNTZFormat", ntzFmt)
+ .json(path.getAbsolutePath)
+ checkAnswer(readBack, df)
+ }
+ }
+ }
+
test("SPARK-37360: Timestamp type inference for a column with TIMESTAMP_NTZ
values") {
withTempPath { path =>
val exp = spark.sql("""
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
