Skip to content

Commit 6abed0f

Browse files
vinodkcMaxGekk
vinodkc
authored and
MaxGekk
committed
[SPARK-57457][SQL] Support nanosecond-precision timestamp types in the CSV datasource (v1 and v2)
### What changes were proposed in this pull request? This PR adds nanosecond-precision timestamp support (`TIMESTAMP_NTZ(p)` and `TIMESTAMP_LTZ(p)`) to the `CSV` datasource, for both the v1 (`CSVFileFormat`) and v2 (`CSVTable`) paths. Specifically: - Parser (`UnivocityParser`): adds `TimestampNTZNanosType` and `TimestampLTZNanosType` cases that delegate to the existing `parseWithoutTimeZoneNanos` / `parseNanos` formatter methods. - Generator (`UnivocityGenerator`): adds the corresponding write-path cases that delegate to `formatWithoutTimeZoneNanos` / f`ormatNanos`. ### Why are the changes needed? `CSV` 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 CSV. ### Does this PR introduce _any_ user-facing change? Yes. Users can write and read `TimestampNTZNanosType(p)` / `TimestampLTZNanosType(p)` (p in 7..9) with CSV ### How was this patch tested? - `CsvFunctionsSuite` — updated the existing from_csv nanosecond timestamp test: the test now asserts successful parsing and correct truncated value rather than expecting an UNSUPPORTED_DATATYPE exception. - `FileBasedDataSourceSuite` — new end-to-end round-trip test covering both v1 and v2 source paths, precisions (7–9), and both TimestampNTZNanosType and TimestampLTZNanosType, verifying that a DataFrame written to CSV and read back with a matching schema produces identical results. ### Was this patch authored or co-authored using generative AI tooling? Yes, Generated-by: Claude Code (Sonnet 4.6) was used to assist with this patch. Closes #56818 from vinodkc/spark-57457-nanosecond-csv. Authored-by: Vinod KC <vinod.kc.in@gmail.com> Signed-off-by: Max Gekk <max.gekk@gmail.com> (cherry picked from commit ab78cb5) Signed-off-by: Max Gekk <max.gekk@gmail.com>
1 parent 4da206c commit 6abed0f

6 files changed

Lines changed: 78 additions & 16 deletions

File tree

‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityGenerator.scala‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,15 @@ class UnivocityGenerator(
8585
(getter, ordinal) =>
8686
timestampNTZFormatter.format(DateTimeUtils.microsToLocalDateTime(getter.getLong(ordinal)))
8787

88+
case t: TimestampNTZNanosType =>
89+
(getter, ordinal) =>
90+
timestampNTZFormatter.formatWithoutTimeZoneNanos(
91+
getter.getTimestampNTZNanos(ordinal), t.precision)
92+
93+
case t: TimestampLTZNanosType =>
94+
(getter, ordinal) =>
95+
timestampFormatter.formatNanos(getter.getTimestampLTZNanos(ordinal), t.precision)
96+
8897
case _: TimeType => (getter, ordinal) => timeFormatter.format(getter.getLong(ordinal))
8998

9099
case YearMonthIntervalType(start, end) =>

‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -263,6 +263,16 @@ class UnivocityParser(
263263
timestampNTZFormatter.parseWithoutTimeZone(datum, false)
264264
}
265265

266+
case t: TimestampNTZNanosType => (d: String) =>
267+
nullSafeDatum(d, name, nullable, options) { datum =>
268+
timestampNTZFormatter.parseWithoutTimeZoneNanos(datum, t.precision, false)
269+
}
270+
271+
case t: TimestampLTZNanosType => (d: String) =>
272+
nullSafeDatum(d, name, nullable, options) { datum =>
273+
timestampFormatter.parseNanos(datum, t.precision)
274+
}
275+
266276
case _: TimeType => (d: String) =>
267277
nullSafeDatum(d, name, nullable, options) { datum =>
268278
timeFormatter.parse(datum)

‎sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -170,9 +170,6 @@ case class CSVFileFormat() extends TextBasedFileFormat with DataSourceRegister {
170170

171171
case _: GeometryType | _: GeographyType => false
172172

173-
// Nanosecond-capable timestamps are not yet supported by this datasource.
174-
case _: AnyTimestampNanoType => false
175-
176173
case _: AtomicType => true
177174

178175
case udt: UserDefinedType[_] => supportDataType(udt.sqlType)

‎sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVTable.scala‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ import org.apache.spark.sql.connector.write.{LogicalWriteInfo, Write, WriteBuild
2626
import org.apache.spark.sql.execution.datasources.FileFormat
2727
import org.apache.spark.sql.execution.datasources.csv.CSVDataSource
2828
import org.apache.spark.sql.execution.datasources.v2.FileTable
29-
import org.apache.spark.sql.types.{AnyTimestampNanoType, AtomicType, DataType, GeographyType, GeometryType, StructType, UserDefinedType}
29+
import org.apache.spark.sql.types.{AtomicType, DataType, GeographyType, GeometryType, StructType, UserDefinedType}
3030
import org.apache.spark.sql.util.CaseInsensitiveStringMap
3131

3232
case class CSVTable(
@@ -63,9 +63,6 @@ case class CSVTable(
6363
override def supportsDataType(dataType: DataType): Boolean = dataType match {
6464
case _: GeometryType | _: GeographyType => false
6565

66-
// Nanosecond-capable timestamps are not yet supported by this datasource.
67-
case _: AnyTimestampNanoType => false
68-
6966
case _: AtomicType => true
7067

7168
case udt: UserDefinedType[_] => supportsDataType(udt.sqlType)

‎sql/core/src/test/scala/org/apache/spark/sql/CsvFunctionsSuite.scala‎

Lines changed: 15 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -19,13 +19,14 @@ package org.apache.spark.sql
1919

2020
import java.nio.charset.StandardCharsets
2121
import java.text.SimpleDateFormat
22-
import java.time.{Duration, LocalDateTime, Period}
22+
import java.time.{Duration, LocalDateTime, Period, ZoneOffset}
2323
import java.util.Locale
2424

2525
import scala.jdk.CollectionConverters._
2626

2727
import org.apache.spark.{SparkException, SparkRuntimeException,
2828
SparkUnsupportedOperationException, SparkUpgradeException}
29+
import org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils
2930
import org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPrecision
3031
import org.apache.spark.sql.errors.DataTypeErrors.toSQLType
3132
import org.apache.spark.sql.functions._
@@ -50,8 +51,12 @@ class CsvFunctionsSuite extends SharedSparkSession {
5051

5152
test("SPARK-57164: from_csv with a nanos timestamp DDL schema string") {
5253
val df = Seq("2020-01-01T00:00:00.123456789").toDF("value")
53-
withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
54+
// Fix the session timezone so the TIMESTAMP_LTZ expected value is deterministic.
55+
withSQLConf(
56+
SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
57+
SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
5458
foreachNanosPrecision { p =>
59+
val nano = TimestampNanosTestUtils.nanoOfSecTruncator(p)(123456789)
5560
Seq(
5661
s"TIMESTAMP_NTZ($p)" -> TimestampNTZNanosType(p),
5762
s"TIMESTAMP_LTZ($p)" -> TimestampLTZNanosType(p),
@@ -62,12 +67,14 @@ class CsvFunctionsSuite extends SharedSparkSession {
6267
from_csv($"value", lit(s"c $spelling"), Map.empty[String, String].asJava).as("v"))
6368
// The schema string resolves to the nanos type ...
6469
assert(parsed.schema("v").dataType.asInstanceOf[StructType]("c").dataType === expected)
65-
// ... but the CSV datasource does not support nanosecond timestamps yet, so the
66-
// value converter rejects it at execution.
67-
checkError(
68-
exception = intercept[SparkUnsupportedOperationException](parsed.collect()),
69-
condition = "UNSUPPORTED_DATATYPE",
70-
parameters = Map("typeName" -> toSQLType(expected)))
70+
// ... and the CSV datasource correctly parses the nanosecond timestamp, truncating
71+
// sub-precision digits toward zero.
72+
val expectedValue = expected match {
73+
case _: TimestampNTZNanosType => LocalDateTime.of(2020, 1, 1, 0, 0, 0, nano)
74+
case _: TimestampLTZNanosType =>
75+
LocalDateTime.of(2020, 1, 1, 0, 0, 0, nano).toInstant(ZoneOffset.UTC)
76+
}
77+
checkAnswer(parsed, Row(Row(expectedValue)))
7178
}
7279
}
7380
}

‎sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala‎

Lines changed: 43 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1341,7 +1341,7 @@ class FileBasedDataSourceSuite extends SharedSparkSession
13411341

13421342
test("SPARK-57166: nanosecond timestamp types are not supported in selected file data sources") {
13431343
// Parquet and ORC support nanosecond-capable timestamps, while these formats still reject them.
1344-
val unsupportedDataSources = Seq("json", "csv", "xml")
1344+
val unsupportedDataSources = Seq("json", "xml")
13451345
val nanosTypes = Seq(TimestampNTZNanosType(9), TimestampLTZNanosType(9))
13461346
withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
13471347
// Test both v1 and v2 data sources.
@@ -1431,6 +1431,48 @@ class FileBasedDataSourceSuite extends SharedSparkSession
14311431
}
14321432
}
14331433

1434+
test("SPARK-57457: CSV supports nanosecond timestamp types in v1 and v2") {
1435+
withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
1436+
Seq(true, false).foreach { useV1 =>
1437+
val useV1List = if (useV1) "csv" else ""
1438+
withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1List) {
1439+
foreachNanosPrecision { precision =>
1440+
// CSV is text-based: the format string must carry enough fractional-second digits to
1441+
// represent the full precision. Use exactly `precision` S-characters so the emitted
1442+
// string is compact and unambiguously round-trips at the given precision.
1443+
val fracPat = "S" * precision
1444+
Seq(TimestampNTZNanosType(precision), TimestampLTZNanosType(precision)).foreach {
1445+
nanosType =>
1446+
withTempDir { dir =>
1447+
val wallClock = LocalDateTime.of(1970, 1, 1, 0, 20, 34, 567890123)
1448+
val value: Any = nanosType match {
1449+
case _: TimestampNTZNanosType => wallClock
1450+
case _: TimestampLTZNanosType => wallClock.toInstant(ZoneOffset.UTC)
1451+
}
1452+
val df = spark.createDataFrame(
1453+
spark.sparkContext.parallelize(Seq(Row(value))),
1454+
new StructType().add("ts", nanosType))
1455+
val path = new File(dir, s"csv_nanos_${nanosType.typeName}").getCanonicalPath
1456+
val (fmtKey, fmtVal) = nanosType match {
1457+
case _: TimestampNTZNanosType =>
1458+
("timestampNTZFormat", s"yyyy-MM-dd'T'HH:mm:ss.$fracPat")
1459+
case _: TimestampLTZNanosType =>
1460+
("timestampFormat", s"yyyy-MM-dd'T'HH:mm:ss.${fracPat}XXX")
1461+
}
1462+
df.write.format("csv").option(fmtKey, fmtVal).mode("overwrite").save(path)
1463+
val readBack = spark.read
1464+
.schema(new StructType().add("ts", nanosType))
1465+
.option(fmtKey, fmtVal)
1466+
.format("csv").load(path)
1467+
checkAnswer(readBack, df)
1468+
}
1469+
}
1470+
}
1471+
}
1472+
}
1473+
}
1474+
}
1475+
14341476
// Asserts the ignoredPathSegmentRegex contract for `format`: the default regex hides the
14351477
// '_'-prefixed file; a never-matching per-read option or session conf each surface it; the
14361478
// option overrides the conf.

0 commit comments

Comments
 (0)