This is an automated email from the ASF dual-hosted git repository.
MaxGekk pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 71b1aca729e2 [SPARK-57459][SQL] Support nanosecond-precision timestamp
types in the Avro datasource (v1 and v2)
71b1aca729e2 is described below
commit 71b1aca729e287cd83d94869a10c48b8d65d1acc
Author: Maxim Gekk <[email protected]>
AuthorDate: Sat Jun 27 21:56:47 2026 +0200
[SPARK-57459][SQL] Support nanosecond-precision timestamp types in the Avro
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 Avro
datasource (v1 `AvroFileFormat` and v2 `AvroTable`), reaching parity with the
microsecond `TimestampType` / `TimestampNTZType`, and removes the
[SPARK-57166](https://issues.apache.org/jira/browse/SPARK-57166) rejection
guardrail.
- `SchemaConverters`: map `TimestampLTZNanosType` / `TimestampNTZNanosType`
to the Avro `timestamp-nanos` / `local-timestamp-nanos` logical types
(available in the bundled Avro 1.12.1, on a `long` storing epoch-nanoseconds),
carrying the fractional-second precision via the `spark.sql.catalyst.type`
property. The reverse direction maps these logical types back, defaulting to
nanosecond precision (9) for files written by external tools that lack the
property.
- `AvroSerializer`: pack the internal `(epochMicros, nanosWithinMicro)`
value into a single epoch-nanoseconds `Long`
(`DateTimeUtils.timestampNanosToEpochNanos`), surfacing values outside the
signed-int64 epoch-nanos range (~1677-09-21 .. 2262-04-11) as a
`DATETIME_OVERFLOW` error.
- `AvroDeserializer`: unpack epoch-nanoseconds via `floorDiv` / `floorMod`
and truncate the sub-microsecond digits to the declared precision.
- `AvroUtils.supportsDataType`: drop the `AnyTimestampNanoType` rejection
so the types are accepted by both the v1 and v2 write/read paths.
Like the Parquet path, nanosecond timestamps are always proleptic Gregorian
and are therefore exempt from datetime rebasing.
### Why are the changes needed?
To extend nanosecond-precision timestamp support (umbrella SPARK-56822) to
the Avro datasource so it can read and write `TIMESTAMP_NTZ(p)` /
`TIMESTAMP_LTZ(p)` with `p` in 7-9, matching the existing microsecond timestamp
behavior and the Parquet/ORC 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
Avro files. 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?
Added tests in `AvroSuite`:
- round-trip for precisions 7-9 for both NTZ and LTZ, across the v1 and v2
sources, including nulls and inferred-schema precision preservation;
- external-reader unit-correctness: decode the written file with a plain
Avro `GenericDatumReader` and assert the stored epoch-nanoseconds and the
logical-type name;
- reading a plain Avro file produced without the `spark.sql.catalyst.type`
property (defaults to nanosecond precision);
- writing an out-of-range value fails loudly with `DATETIME_OVERFLOW`.
Ran `AvroV1Suite` / `AvroV2Suite` (new tests pass on both) plus
`AvroSerdeSuite`, `AvroV1/V2LogicalTypeSuite`, and
`AvroCatalystDataConversionSuite` (no regressions), and `sql` / `avro`
scalastyle.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor 2.1, Claude Opus 4.8
Closes #56825 from MaxGekk/nanos-avro.
Authored-by: Maxim Gekk <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
(cherry picked from commit 5798db67062c8aa1943273eee944efec63f3f396)
Signed-off-by: Max Gekk <[email protected]>
---
.../org/apache/spark/sql/avro/AvroSuite.scala | 362 +++++++++++++++++++--
docs/sql-data-sources-avro.md | 20 ++
.../spark/sql/catalyst/util/DateTimeUtils.scala | 30 ++
.../sql/catalyst/util/DateTimeUtilsSuite.scala | 46 +++
.../apache/spark/sql/avro/AvroDeserializer.scala | 24 +-
.../org/apache/spark/sql/avro/AvroSerializer.scala | 22 +-
.../org/apache/spark/sql/avro/AvroUtils.scala | 3 -
.../apache/spark/sql/avro/SchemaConverters.scala | 36 ++
.../types/ops/TimestampNanosParquetOps.scala | 28 +-
.../types/ops/TimestampNanosParquetOpsSuite.scala | 9 +-
10 files changed, 511 insertions(+), 69 deletions(-)
diff --git
a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
index 588d7e26206f..813401773834 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
+++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
@@ -33,12 +33,12 @@ import org.apache.avro.file.{DataFileReader, DataFileWriter}
import org.apache.avro.generic.{GenericData, GenericDatumReader,
GenericDatumWriter, GenericRecord}
import org.apache.avro.generic.GenericData.{EnumSymbol, Fixed}
-import org.apache.spark.{SPARK_VERSION_SHORT, SparkConf, SparkException,
SparkRuntimeException, SparkThrowable, SparkUpgradeException}
+import org.apache.spark.{SPARK_VERSION_SHORT, SparkArithmeticException,
SparkConf, SparkException, SparkRuntimeException, SparkThrowable,
SparkUpgradeException}
import org.apache.spark.TestUtils.assertExceptionMsg
import org.apache.spark.sql._
import org.apache.spark.sql.TestingUDT.IntervalData
import org.apache.spark.sql.avro.AvroCompressionCodec._
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Literal}
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
import org.apache.spark.sql.catalyst.plans.logical.Filter
import org.apache.spark.sql.catalyst.util.DateTimeTestUtils
import
org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone, LA,
UTC}
@@ -52,7 +52,6 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
import org.apache.spark.sql.v2.avro.AvroScan
-import org.apache.spark.unsafe.types.TimestampNanosVal
import org.apache.spark.util.Utils
abstract class AvroSuite
@@ -3192,46 +3191,337 @@ abstract class AvroSuite
}
}
- test("SPARK-57166: nanosecond timestamp types are not supported in Avro") {
- val nanosTypes = Seq(TimestampNTZNanosType(9), TimestampLTZNanosType(9))
+ test("SPARK-57459: nanosecond timestamp types round-trip through Avro (v1
and v2)") {
withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
- 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("avro").mode("overwrite").save(writeDir)
- },
- condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
- parameters = Map(
- "columnName" -> "`ts`",
- "columnType" -> expectedType,
- "format" -> "Avro"))
-
- // 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
-
Seq("a").toDF("ts").write.format("avro").mode("overwrite").save(readDir)
+ Seq(true, false).foreach { useV1 =>
+ val useV1List = if (useV1) "avro" else ""
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1List) {
+ Seq(7, 8, 9).foreach { precision =>
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ withTempDir { dir =>
+ // Build the row from an external java.time value; the
column schema carries the
+ // precision and truncates the sub-microsecond digits,
matching the ORC suites.
+ val wallClock = LocalDateTime.of(1970, 1, 1, 0, 20, 34,
567890123)
+ val value: Any = nanosType match {
+ case _: TimestampNTZNanosType => wallClock
+ case _: TimestampLTZNanosType =>
wallClock.toInstant(java.time.ZoneOffset.UTC)
+ }
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(value), Row(null))),
+ new StructType().add("ts", nanosType))
+ val path = new File(dir,
s"avro_nanos_${nanosType.typeName}").getCanonicalPath
+ df.write.format("avro").mode("overwrite").save(path)
+ // The inferred schema preserves the declared precision via
the catalyst prop.
+ val inferred = spark.read.format("avro").load(path)
+ assert(inferred.schema("ts").dataType == nanosType)
+ val readBack = spark.read.schema(new StructType().add("ts",
nanosType))
+ .format("avro").load(path)
+ checkAnswer(readBack, df)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57459: nanosecond timestamps are written as unit-correct
epoch-nanos") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ // LocalDateTime at the epoch plus 567890123 ns, truncated to each
precision.
+ val wallClock = LocalDateTime.of(1970, 1, 1, 0, 0, 0, 567890123)
+ val expectedNanos = Map(7 -> 567890100L, 8 -> 567890120L, 9 ->
567890123L)
+ Seq(
+ (false, "timestamp-nanos"),
+ (true, "local-timestamp-nanos")).foreach { case (isNtz, logicalName) =>
+ Seq(7, 8, 9).foreach { p =>
+ withTempPath { dir =>
+ val nanosType: DataType =
+ if (isNtz) TimestampNTZNanosType(p) else TimestampLTZNanosType(p)
+ val value: Any =
+ if (isNtz) wallClock else
wallClock.toInstant(java.time.ZoneOffset.UTC)
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(value)), numSlices = 1),
+ new StructType().add("t", nanosType))
+ df.write.format("avro").save(dir.toString)
+
+ val avroFile = dir.listFiles()
+ .filter(f => f.isFile && f.getName.endsWith("avro"))
+ .head
+ val reader = new DataFileReader[GenericRecord](
+ avroFile, new GenericDatumReader[GenericRecord]())
+ try {
+ val fieldSchema = reader.getSchema.getField("t").schema()
+ val tsSchema = if (fieldSchema.getType == Type.UNION) {
+ fieldSchema.getTypes.asScala.find(_.getType == Type.LONG).get
+ } else {
+ fieldSchema
+ }
+ assert(tsSchema.getLogicalType.getName == logicalName,
+ s"$nanosType should be written with the $logicalName logical
type")
+ assert(reader.hasNext)
+ val stored = reader.next().get("t").asInstanceOf[Long]
+ assert(stored == expectedNanos(p),
+ s"$nanosType should store epoch-nanos ${expectedNanos(p)}, but
was $stored")
+ } finally {
+ reader.close()
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57459: nanosecond timestamps read from a plain Avro file (no
catalyst prop)") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ // Build Avro files the way an external tool would: a nanos logical type
on a long with no
+ // `spark.sql.catalyst.type` property. Spark must default to nanosecond
precision and convert
+ // the stored epoch-nanoseconds back to its internal (epochMicros,
nanosWithinMicro) form.
+ val epochNanos = 567890123L
+ Seq(
+ ("timestamp-nanos", TimestampLTZNanosType(9),
+ java.time.Instant.ofEpochSecond(0L, epochNanos)),
+ ("local-timestamp-nanos", TimestampNTZNanosType(9),
+ LocalDateTime.of(1970, 1, 1, 0, 0, 0, epochNanos.toInt))).foreach {
+ case (logicalName, expectedType, expectedValue) =>
+ withTempDir { dir =>
+ val avroSchema = new Schema.Parser().parse(
+ s"""
+ |{
+ | "type": "record",
+ | "name": "top",
+ | "fields": [
+ | {"name": "t", "type": {"type": "long", "logicalType":
"$logicalName"}}
+ | ]
+ |}
+ """.stripMargin)
+ val avroFile = new File(dir, "external.avro")
+ val datumWriter = new GenericDatumWriter[GenericRecord](avroSchema)
+ val dataFileWriter = new DataFileWriter[GenericRecord](datumWriter)
+ dataFileWriter.create(avroSchema, avroFile)
+ try {
+ val record = new GenericData.Record(avroSchema)
+ record.put("t", epochNanos)
+ dataFileWriter.append(record)
+ } finally {
+ dataFileWriter.close()
+ }
+
+ val readDf = spark.read.format("avro").load(dir.toString)
+ assert(readDf.schema("t").dataType == expectedType)
+ checkAnswer(readDf, Row(expectedValue))
+ }
+ }
+ }
+ }
+
+ test("SPARK-57459: reading a foreign nanos Avro file with an explicit
lower-precision schema") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ // A full-precision (9-digit) foreign value read with an explicit
TIMESTAMP_*(7) schema must
+ // truncate the sub-microsecond digits (123 -> 100), mirroring
ParquetTimestampNanosSuite's
+ // "explicit lower-precision read schema" case.
+ val epochNanos = 567890123L
+ Seq(
+ ("timestamp-nanos", TimestampLTZNanosType(7),
+ java.time.Instant.ofEpochSecond(0L, 567890100L)),
+ ("local-timestamp-nanos", TimestampNTZNanosType(7),
+ LocalDateTime.of(1970, 1, 1, 0, 0, 0, 567890100))).foreach {
+ case (logicalName, readType, expectedValue) =>
+ withTempDir { dir =>
+ val avroSchema = new Schema.Parser().parse(
+ s"""
+ |{
+ | "type": "record",
+ | "name": "top",
+ | "fields": [
+ | {"name": "t", "type": {"type": "long", "logicalType":
"$logicalName"}}
+ | ]
+ |}
+ """.stripMargin)
+ val avroFile = new File(dir, "external.avro")
+ val datumWriter = new GenericDatumWriter[GenericRecord](avroSchema)
+ val dataFileWriter = new DataFileWriter[GenericRecord](datumWriter)
+ dataFileWriter.create(avroSchema, avroFile)
+ try {
+ val record = new GenericData.Record(avroSchema)
+ record.put("t", epochNanos)
+ dataFileWriter.append(record)
+ } finally {
+ dataFileWriter.close()
+ }
+
+ val readDf = spark.read.schema(new StructType().add("t", readType))
+ .format("avro").load(dir.toString)
+ assert(readDf.schema("t").dataType == readType)
+ checkAnswer(readDf, Row(expectedValue))
+ }
+ }
+ }
+ }
+
+ test("SPARK-57459: writing an out-of-range nanosecond timestamp fails
loudly") {
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ Seq("TIMESTAMP_NTZ", "TIMESTAMP_LTZ").foreach { typeName =>
+ withTempPath { dir =>
+ // Year 9999 is far outside the signed-int64 epoch-nanos range
(~2262).
+ val df = spark.sql(s"SELECT $typeName '9999-12-31
23:59:59.999999999' AS ts")
+ val e = intercept[SparkException] {
+ df.write.format("avro").save(dir.getCanonicalPath)
+ }
+ var cause: Throwable = e
+ while (cause != null &&
!cause.isInstanceOf[SparkArithmeticException]) {
+ cause = cause.getCause
+ }
+ assert(cause != null,
+ s"Expected a DATETIME_OVERFLOW error for $typeName, but got:
${e.getMessage}")
+ // NTZ renders without a zone; LTZ renders as a UTC instant with a
trailing `Z`.
+ val renderedValue =
+ if (typeName == "TIMESTAMP_NTZ") "9999-12-31T23:59:59.999999999"
+ else "9999-12-31T23:59:59.999999999Z"
checkError(
- exception = intercept[AnalysisException] {
- spark.read.schema(new StructType().add("ts", nanosType))
- .format("avro").load(readDir).collect()
- },
- condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
- parameters = Map(
- "columnName" -> "`ts`",
- "columnType" -> expectedType,
- "format" -> "Avro"))
+ exception = cause.asInstanceOf[SparkArithmeticException],
+ condition = "DATETIME_OVERFLOW",
+ parameters = Map("operation" ->
+ (s"write the timestamp value $renderedValue as Avro
epoch-nanoseconds " +
+ "(supported range: 1677-09-21T00:12:43.145224192Z to " +
+ "2262-04-11T23:47:16.854775807Z)")))
+ }
+ }
+ }
+ }
+
+ test("SPARK-57459: nanosecond timestamps round-trip at the maximum supported
instant") {
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ // Long.MaxValue epoch-nanoseconds = 2262-04-11T23:47:16.854775807Z, the
largest instant the
+ // INT64 epoch-nanos encoding can represent. The lower bound
Long.MinValue is intentionally
+ // not exercised: re-encoding its floored epoch-microseconds overflows
Long (see
+ // DateTimeUtilsSuite), so it is not writable.
+ val maxInstant = java.time.Instant.ofEpochSecond(9223372036L, 854775807L)
+ val maxLocal = LocalDateTime.ofEpochSecond(9223372036L, 854775807,
java.time.ZoneOffset.UTC)
+ Seq(
+ (TimestampLTZNanosType(9), maxInstant: Any),
+ (TimestampNTZNanosType(9), maxLocal: Any)).foreach { case (nanosType,
value) =>
+ withTempPath { dir =>
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(value)), numSlices = 1),
+ new StructType().add("ts", nanosType))
+ df.write.format("avro").mode("overwrite").save(dir.getCanonicalPath)
+ val readBack = spark.read.schema(new StructType().add("ts",
nanosType))
+ .format("avro").load(dir.getCanonicalPath)
+ checkAnswer(readBack, df)
}
}
}
}
+ test("SPARK-57459: pre-epoch and minimum-instant nanosecond timestamp
round-trips") {
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ // The smallest *writable* instant: Long.MinValue epoch-nanoseconds
floor to an epochMicros
+ // whose re-encode overflows Long, so the practical minimum is the next
whole microsecond up
+ // (~1677-09-21), the symmetric counterpart of the maximum-instant test
above.
+ val minEpochNanos = (Long.MinValue / 1000L) * 1000L
+ // Pre-epoch full-precision value: 1969-12-31T23:59:59.999999999Z (= -1
epoch-nanosecond),
+ // exercising the floor semantics end to end (mirrors the Parquet/ORC
suites).
+ Seq(-1L, minEpochNanos).foreach { epochNanos =>
+ val seconds = Math.floorDiv(epochNanos, 1000000000L)
+ val nanoOfSecond = Math.floorMod(epochNanos, 1000000000L).toInt
+ val instant = java.time.Instant.ofEpochSecond(seconds,
nanoOfSecond.toLong)
+ val localDateTime =
+ LocalDateTime.ofEpochSecond(seconds, nanoOfSecond,
java.time.ZoneOffset.UTC)
+ Seq(
+ (TimestampLTZNanosType(9), instant: Any),
+ (TimestampNTZNanosType(9), localDateTime: Any)).foreach { case
(nanosType, value) =>
+ withTempPath { dir =>
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(value)), numSlices = 1),
+ new StructType().add("ts", nanosType))
+
df.write.format("avro").mode("overwrite").save(dir.getCanonicalPath)
+ val readBack = spark.read.schema(new StructType().add("ts",
nanosType))
+ .format("avro").load(dir.getCanonicalPath)
+ checkAnswer(readBack, df)
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57459: TIMESTAMP_LTZ nanos on-disk value is independent of the
session time zone") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ // TIMESTAMP_LTZ is an absolute instant, so the stored epoch-nanoseconds
and the read-back
+ // value must not depend on the session time zone.
+ val instant = LocalDateTime.of(2024, 6, 15, 12, 34, 56, 789012345)
+ .toInstant(java.time.ZoneOffset.UTC)
+ val schema = new StructType().add("ts", TimestampLTZNanosType(9))
+
+ def writeAndReadStored(zone: String, dir: File): Long = {
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> zone) {
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(instant: Any)), numSlices =
1), schema)
+ df.write.format("avro").mode("overwrite").save(dir.getCanonicalPath)
+ // The instant reads back unchanged regardless of the session zone.
+ val readBack =
spark.read.schema(schema).format("avro").load(dir.getCanonicalPath)
+ checkAnswer(readBack, Row(instant))
+ }
+ val avroFile = dir.listFiles()
+ .filter(f => f.isFile && f.getName.endsWith("avro"))
+ .head
+ val reader = new DataFileReader[GenericRecord](
+ avroFile, new GenericDatumReader[GenericRecord]())
+ try {
+ val fieldSchema = reader.getSchema.getField("ts").schema()
+ val tsSchema = if (fieldSchema.getType == Type.UNION) {
+ fieldSchema.getTypes.asScala.find(_.getType == Type.LONG).get
+ } else {
+ fieldSchema
+ }
+ assert(tsSchema.getLogicalType.getName == "timestamp-nanos")
+ reader.next().get("ts").asInstanceOf[Long]
+ } finally {
+ reader.close()
+ }
+ }
+
+ withTempPath { utcDir =>
+ withTempPath { laDir =>
+ val storedUtc = writeAndReadStored("UTC", utcDir)
+ val storedLa = writeAndReadStored("America/Los_Angeles", laDir)
+ assert(storedUtc === storedLa,
+ "TIMESTAMP_LTZ epoch-nanoseconds must not depend on the session
time zone")
+ }
+ }
+ }
+ }
+
+ test("SPARK-57459: nanosecond timestamp types in nested and complex Avro
structures") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ val ntz = LocalDateTime.of(2024, 1, 2, 3, 4, 5, 123456789)
+ val ltz = ntz.toInstant(java.time.ZoneOffset.UTC)
+ val schema = new StructType()
+ .add("s", new StructType()
+ .add("a", TimestampNTZNanosType(9))
+ .add("b", TimestampLTZNanosType(8)))
+ .add("arr", ArrayType(TimestampNTZNanosType(7)))
+ .add("m", MapType(StringType, TimestampLTZNanosType(9)))
+ val row = Row(
+ Row(ntz, ltz),
+ Seq(ntz, null, ntz),
+ Map("k1" -> ltz, "k2" -> null))
+ withTempPath { dir =>
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(row), numSlices = 1), schema)
+ df.write.format("avro").mode("overwrite").save(dir.getCanonicalPath)
+ val readBack =
spark.read.schema(schema).format("avro").load(dir.getCanonicalPath)
+ checkAnswer(readBack, df)
+ }
+ }
+ }
+
test("TIME type read/write with Avro format") {
withTempPath { dir =>
// Test boundary values and NULL handling
diff --git a/docs/sql-data-sources-avro.md b/docs/sql-data-sources-avro.md
index a55cb00f9bd0..e786e7f2b9ef 100644
--- a/docs/sql-data-sources-avro.md
+++ b/docs/sql-data-sources-avro.md
@@ -565,6 +565,16 @@ It also supports reading the following Avro [logical
types](https://avro.apache.
<td>long</td>
<td>TimeType</td>
</tr>
+ <tr>
+ <td>timestamp-nanos</td>
+ <td>long</td>
+ <td>TimestampType(p) (with p in 7-9, requires
<code>spark.sql.timestampNanosTypes.enabled=true</code>)</td>
+ </tr>
+ <tr>
+ <td>local-timestamp-nanos</td>
+ <td>long</td>
+ <td>TimestampNTZType(p) (with p in 7-9, requires
<code>spark.sql.timestampNanosTypes.enabled=true</code>)</td>
+ </tr>
<tr>
<td>decimal</td>
<td>fixed</td>
@@ -613,6 +623,16 @@ Spark supports writing of all Spark SQL types into Avro.
For most types, the map
<td>long</td>
<td>time-micros</td>
</tr>
+ <tr>
+ <td>TimestampType(p) (with p in 7-9)</td>
+ <td>long</td>
+ <td>timestamp-nanos</td>
+ </tr>
+ <tr>
+ <td>TimestampNTZType(p) (with p in 7-9)</td>
+ <td>long</td>
+ <td>local-timestamp-nanos</td>
+ </tr>
<tr>
<td>DecimalType</td>
<td>fixed</td>
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
index 86cd01a0f3b5..6fcb6c3075ed 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
@@ -331,6 +331,36 @@ object DateTimeUtils extends SparkDateTimeUtils {
value.nanosWithinMicro.toLong)
}
+ /**
+ * Packs a [[TimestampNanosVal]] into a single int64 of epoch-nanoseconds
for a `sink` that uses
+ * that encoding (the Parquet INT64 and Avro `timestamp-nanos` /
`local-timestamp-nanos` physical
+ * types), translating the int64 overflow thrown by
[[timestampNanosToEpochNanos]] into a
+ * `DATETIME_OVERFLOW` error that names the `sink`. `isNtz` selects how the
offending value is
+ * rendered in that error (a zone-less local date-time vs. a UTC instant).
+ */
+ def timestampNanosToEpochNanos(value: TimestampNanosVal, isNtz: Boolean,
sink: String): Long = {
+ try {
+ timestampNanosToEpochNanos(value)
+ } catch {
+ case _: ArithmeticException =>
+ throw
QueryExecutionErrors.timestampNanosEpochNanosOverflowError(value, isNtz, sink)
+ }
+ }
+
+ /**
+ * Unpacks a single int64 of nanoseconds since the epoch (the representation
used by the Arrow
+ * nanosecond timestamp vectors and the Parquet / Avro INT64
epoch-nanoseconds encodings) back
+ * into a [[TimestampNanosVal]], truncating the sub-microsecond digits to
the given `precision`
+ * (in [7, 9]). This is the inverse of [[timestampNanosToEpochNanos]].
`floorDiv` / `floorMod`
+ * keep `nanosWithinMicro` in [0, 999] for pre-epoch (negative) values too.
+ */
+ def epochNanosToTimestampNanos(epochNanos: Long, precision: Int):
TimestampNanosVal = {
+ val epochMicros = Math.floorDiv(epochNanos, NANOS_PER_MICROS)
+ val rawNanosWithinMicro = Math.floorMod(epochNanos, NANOS_PER_MICROS).toInt
+ val nanosWithinMicro =
truncateNanosWithinMicroToPrecision(rawNanosWithinMicro, precision)
+ TimestampNanosVal.fromParts(epochMicros, nanosWithinMicro.toShort)
+ }
+
/**
* Adds a full interval (months, days, microseconds) to a timestamp
represented as the number of
* microseconds since 1970-01-01 00:00:00Z.
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateTimeUtilsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateTimeUtilsSuite.scala
index 34fc5286722c..d07265c26e97 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateTimeUtilsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateTimeUtilsSuite.scala
@@ -1232,6 +1232,52 @@ class DateTimeUtilsSuite extends SparkFunSuite with
Matchers with SQLHelper {
}
}
+ test("SPARK-57459: epochNanosToTimestampNanos unpacks int64
epoch-nanoseconds") {
+ def nanos(epochMicros: Long, nanosWithinMicro: Int): TimestampNanosVal =
+ TimestampNanosVal.fromParts(epochMicros, nanosWithinMicro.toShort)
+
+ // At full (nanosecond) precision it is the exact inverse of
timestampNanosToEpochNanos.
+ assert(epochNanosToTimestampNanos(0L, 9) === nanos(0L, 0))
+ assert(epochNanosToTimestampNanos(999L, 9) === nanos(0L, 999))
+ assert(epochNanosToTimestampNanos(NANOS_PER_MICROS, 9) === nanos(1L, 0))
+ assert(epochNanosToTimestampNanos(1234567L * NANOS_PER_MICROS + 7L, 9) ===
nanos(1234567L, 7))
+ // Pre-epoch values use floor semantics, keeping nanosWithinMicro in [0,
999].
+ assert(epochNanosToTimestampNanos(-1L, 9) === nanos(-1L, 999))
+ assert(epochNanosToTimestampNanos(-NANOS_PER_MICROS, 9) === nanos(-1L, 0))
+
+ // Lower precisions truncate the sub-microsecond digits.
+ assert(epochNanosToTimestampNanos(123456789L, 9) === nanos(123456L, 789))
+ assert(epochNanosToTimestampNanos(123456789L, 8) === nanos(123456L, 780))
+ assert(epochNanosToTimestampNanos(123456789L, 7) === nanos(123456L, 700))
+
+ // Truncation operates on the floored nanosWithinMicro, so it composes
with floor semantics for
+ // pre-epoch values too (-123456211 -> floor (-123457, 789) -> truncate
the 789).
+ assert(epochNanosToTimestampNanos(-123456211L, 9) === nanos(-123457L, 789))
+ assert(epochNanosToTimestampNanos(-123456211L, 8) === nanos(-123457L, 780))
+ assert(epochNanosToTimestampNanos(-123456211L, 7) === nanos(-123457L, 700))
+
+ // The int64 extremes decode without overflow: floor keeps
nanosWithinMicro in [0, 999].
+ assert(epochNanosToTimestampNanos(Long.MaxValue, 9) ===
nanos(9223372036854775L, 807))
+ assert(epochNanosToTimestampNanos(Long.MinValue, 9) ===
nanos(-9223372036854776L, 192))
+
+ // Round-trips with timestampNanosToEpochNanos at full precision.
Long.MinValue is excluded: its
+ // decode (-9223372036854776, 192) re-encodes through an intermediate
epochMicros * 1000 that
+ // overflows Long, an existing limitation of the multiplyExact-based pack
path.
+ Seq(
+ 0L, 999L, 1234567L * NANOS_PER_MICROS + 7L, -1234567L * NANOS_PER_MICROS
+ 13L,
+ Long.MaxValue).foreach { epochNanos =>
+ assert(timestampNanosToEpochNanos(epochNanosToTimestampNanos(epochNanos,
9)) === epochNanos)
+ }
+
+ // Precision outside [7, 9] is an internal-only contract violation and
fails loudly.
+ checkError(
+ exception = intercept[SparkException] {
+ epochNanosToTimestampNanos(123456789L, 6)
+ },
+ condition = "INTERNAL_ERROR",
+ parameters = Map("message" -> "Fractional second precision 6 is out of
range [7, 9]."))
+ }
+
test("SPARK-34903: subtract timestamps") {
DateTimeTestUtils.outstandingZoneIds.foreach { zid =>
Seq(
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
index ba574c091dae..ce16c4a2cc3a 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
@@ -25,7 +25,7 @@ import scala.jdk.CollectionConverters._
import org.apache.avro.{LogicalTypes, Schema, SchemaBuilder}
import org.apache.avro.Conversions.DecimalConversion
-import org.apache.avro.LogicalTypes.{LocalTimestampMicros,
LocalTimestampMillis, TimestampMicros, TimestampMillis}
+import org.apache.avro.LogicalTypes.{LocalTimestampMicros,
LocalTimestampMillis, LocalTimestampNanos, TimestampMicros, TimestampMillis,
TimestampNanos}
import org.apache.avro.Schema.Type._
import org.apache.avro.generic._
import org.apache.avro.util.Utf8
@@ -213,6 +213,28 @@ private[sql] class AvroDeserializer(
s"Avro logical type $other cannot be converted to SQL type
${TimeType().sql}.")
}
+ case (LONG, t: TimestampLTZNanosType) => avroType.getLogicalType match {
+ // The timestamp-nanos logical type stores epoch-nanoseconds (Long),
while the value is
+ // represented internally as (epochMicros, nanosWithinMicro). Floor
semantics keep
+ // nanosWithinMicro in [0, 999] for pre-epoch values. Nanos timestamps
are always proleptic
+ // Gregorian, so they are exempt from datetime rebasing.
+ case _: TimestampNanos => (updater, ordinal, value) =>
+ updater.set(ordinal,
+ DateTimeUtils.epochNanosToTimestampNanos(value.asInstanceOf[Long],
t.precision))
+ case other => throw new IncompatibleSchemaException(errorPrefix +
+ s"Avro logical type $other cannot be converted to SQL type " +
+ s"${TimestampLTZNanosType().sql}.")
+ }
+
+ case (LONG, t: TimestampNTZNanosType) => avroType.getLogicalType match {
+ case _: LocalTimestampNanos => (updater, ordinal, value) =>
+ updater.set(ordinal,
+ DateTimeUtils.epochNanosToTimestampNanos(value.asInstanceOf[Long],
t.precision))
+ case other => throw new IncompatibleSchemaException(errorPrefix +
+ s"Avro logical type $other cannot be converted to SQL type " +
+ s"${TimestampNTZNanosType().sql}.")
+ }
+
// Before we upgrade Avro to 1.8 for logical type support, spark-avro
converts Long to Date.
// For backward compatibility, we still keep this conversion.
case (LONG, DateType) => (updater, ordinal, value) =>
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
index aacf8dc9f347..d5405f69ff05 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
@@ -23,7 +23,7 @@ import scala.jdk.CollectionConverters._
import org.apache.avro.{LogicalTypes, Schema}
import org.apache.avro.Conversions.DecimalConversion
-import org.apache.avro.LogicalTypes.{LocalTimestampMicros,
LocalTimestampMillis, TimestampMicros, TimestampMillis}
+import org.apache.avro.LogicalTypes.{LocalTimestampMicros,
LocalTimestampMillis, LocalTimestampNanos, TimestampMicros, TimestampMillis,
TimestampNanos}
import org.apache.avro.Schema.Type
import org.apache.avro.Schema.Type._
import org.apache.avro.generic.GenericData.{EnumSymbol, Fixed, Record}
@@ -201,6 +201,26 @@ private[sql] class AvroSerializer(
s"SQL type ${TimeType().sql} cannot be converted to Avro logical
type $other")
}
+ case (_: TimestampLTZNanosType, LONG) => avroType.getLogicalType match {
+ // Nanosecond-precision timestamps are stored as epoch-nanoseconds
(Long). They are always
+ // proleptic Gregorian, so they are exempt from datetime rebasing.
+ case _: TimestampNanos => (getter, ordinal) =>
+ DateTimeUtils.timestampNanosToEpochNanos(
+ getter.getTimestampLTZNanos(ordinal), isNtz = false, sink = "Avro")
+ case other => throw new IncompatibleSchemaException(errorPrefix +
+ s"SQL type ${TimestampLTZNanosType().sql} cannot be converted to " +
+ s"Avro logical type $other")
+ }
+
+ case (_: TimestampNTZNanosType, LONG) => avroType.getLogicalType match {
+ case _: LocalTimestampNanos => (getter, ordinal) =>
+ DateTimeUtils.timestampNanosToEpochNanos(
+ getter.getTimestampNTZNanos(ordinal), isNtz = true, sink = "Avro")
+ case other => throw new IncompatibleSchemaException(errorPrefix +
+ s"SQL type ${TimestampNTZNanosType().sql} cannot be converted to " +
+ s"Avro logical type $other")
+ }
+
case (ArrayType(et, containsNull), ARRAY) =>
val elementConverter = newConverter(
et, resolveNullableType(avroType.getElementType, containsNull),
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
index b8d2c30b0838..9b6ed4d77a6d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
@@ -123,9 +123,6 @@ private[sql] object AvroUtils extends Logging {
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/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala
b/sql/core/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala
index 590eaeac6008..c6fba163309f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala
@@ -131,6 +131,26 @@ object SchemaConverters extends Logging {
case _: TimestampMillis | _: TimestampMicros =>
SchemaType(TimestampType, nullable = false)
case _: LocalTimestampMillis | _: LocalTimestampMicros =>
SchemaType(TimestampNTZType, nullable = false)
+ case _: TimestampNanos =>
+ // Avro stores nanoseconds-since-epoch in a long. The precision
(7-9) is carried via the
+ // spark.sql.catalyst.type property; external files without it
default to nanoseconds.
+ val catalystTypeAttrValue =
avroSchema.getProp(CATALYST_TYPE_PROP_NAME)
+ val nanosType = if (catalystTypeAttrValue == null) {
+ TimestampLTZNanosType()
+ } else {
+ CatalystSqlParser.parseDataType(catalystTypeAttrValue)
+ .asInstanceOf[TimestampLTZNanosType]
+ }
+ SchemaType(nanosType, nullable = false)
+ case _: LocalTimestampNanos =>
+ val catalystTypeAttrValue =
avroSchema.getProp(CATALYST_TYPE_PROP_NAME)
+ val nanosType = if (catalystTypeAttrValue == null) {
+ TimestampNTZNanosType()
+ } else {
+ CatalystSqlParser.parseDataType(catalystTypeAttrValue)
+ .asInstanceOf[TimestampNTZNanosType]
+ }
+ SchemaType(nanosType, nullable = false)
case _: LogicalTypes.TimeMicros =>
// Falls back to default precision for backward compatibility with
// Avro files written by external tools.
@@ -334,6 +354,14 @@ object SchemaConverters extends Logging {
LogicalTypes.timestampMicros().addToSchema(builder.longType())
case TimestampNTZType =>
LogicalTypes.localTimestampMicros().addToSchema(builder.longType())
+ case t: TimestampLTZNanosType =>
+ val tsSchema =
LogicalTypes.timestampNanos().addToSchema(builder.longType())
+ tsSchema.addProp(CATALYST_TYPE_PROP_NAME, t.typeName)
+ tsSchema
+ case t: TimestampNTZNanosType =>
+ val tsSchema =
LogicalTypes.localTimestampNanos().addToSchema(builder.longType())
+ tsSchema.addProp(CATALYST_TYPE_PROP_NAME, t.typeName)
+ tsSchema
case t: TimeType =>
val timeSchema =
LogicalTypes.timeMicros().addToSchema(builder.longType())
timeSchema.addProp(CATALYST_TYPE_PROP_NAME, t.typeName)
@@ -466,6 +494,14 @@ object SchemaConverters extends Logging {
case DateType => LogicalTypes.date().addToSchema(builder.intType())
case TimestampType =>
LogicalTypes.timestampMicros().addToSchema(builder.longType())
case TimestampNTZType =>
LogicalTypes.localTimestampMicros().addToSchema(builder.longType())
+ case t: TimestampLTZNanosType =>
+ val tsSchema =
LogicalTypes.timestampNanos().addToSchema(builder.longType())
+ tsSchema.addProp(CATALYST_TYPE_PROP_NAME, t.typeName)
+ tsSchema
+ case t: TimestampNTZNanosType =>
+ val tsSchema =
LogicalTypes.localTimestampNanos().addToSchema(builder.longType())
+ tsSchema.addProp(CATALYST_TYPE_PROP_NAME, t.typeName)
+ tsSchema
case d: DecimalType =>
val avroType = LogicalTypes.decimal(d.precision, d.scale)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
index 3359ba390809..4b5f01d17374 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
@@ -24,7 +24,7 @@ import
org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64
import org.apache.parquet.schema.Type.Repetition
import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
-import org.apache.spark.sql.catalyst.util.{DateTimeConstants, DateTimeUtils}
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.errors.QueryExecutionErrors
import
org.apache.spark.sql.execution.datasources.parquet.{HasParentContainerUpdater,
ParentContainerUpdater, ParquetPrimitiveConverter}
import org.apache.spark.sql.types.{DataType, TimestampLTZNanosType,
TimestampNTZNanosType}
@@ -96,7 +96,8 @@ private[parquet] trait TimestampNanosParquetOps extends
ParquetTypeOps {
// RecordConsumer is null during init() and set later in prepareForWrite().
(row: SpecializedGetters, ordinal: Int) =>
recordConsumer().addLong(
- TimestampNanosParquetOps.timestampNanosToEpochNanos(getNanos(row,
ordinal), isNtz))
+ DateTimeUtils.timestampNanosToEpochNanos(
+ getNanos(row, ordinal), isNtz, sink = "Parquet INT64"))
// ==================== Row-Based Read ====================
@@ -115,12 +116,7 @@ private[parquet] trait TimestampNanosParquetOps extends
ParquetTypeOps {
val p = precision
new ParquetPrimitiveConverter(updater) {
override def addLong(value: Long): Unit = {
- val epochMicros = Math.floorDiv(value,
DateTimeConstants.NANOS_PER_MICROS)
- val rawNanosWithinMicro =
- Math.floorMod(value, DateTimeConstants.NANOS_PER_MICROS).toInt
- val nanosWithinMicro =
-
DateTimeUtils.truncateNanosWithinMicroToPrecision(rawNanosWithinMicro, p)
- this.updater.set(TimestampNanosVal.fromParts(epochMicros,
nanosWithinMicro.toShort))
+ this.updater.set(DateTimeUtils.epochNanosToTimestampNanos(value, p))
}
}
}
@@ -170,20 +166,4 @@ private[ops] object TimestampNanosParquetOps {
case ts: TimestampLogicalTypeAnnotation => ts.getUnit == TimeUnit.NANOS
case _ => false
})
-
- /**
- * Combines the `(epochMicros, nanosWithinMicro)` pair into a single INT64
epoch-nanoseconds
- * value for Parquet storage. Delegates the exact-arithmetic packing to
- * [[DateTimeUtils.timestampNanosToEpochNanos]]; values outside the
signed-int64 epoch-nanos
- * range (~1677-09-21 .. 2262-04-11) throw
`timestampNanosEpochNanosOverflowError`.
- */
- private[ops] def timestampNanosToEpochNanos(value: TimestampNanosVal, isNtz:
Boolean): Long = {
- try {
- DateTimeUtils.timestampNanosToEpochNanos(value)
- } catch {
- case _: ArithmeticException =>
- throw QueryExecutionErrors.timestampNanosEpochNanosOverflowError(
- value, isNtz, sink = "Parquet INT64")
- }
- }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala
index 4d08efb77c93..deade76358fe 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala
@@ -24,7 +24,7 @@ import
org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64
import org.apache.parquet.schema.Type.Repetition.REQUIRED
import org.apache.spark.{SparkArithmeticException, SparkFunSuite,
SparkRuntimeException}
-import org.apache.spark.sql.catalyst.util.DateTimeConstants
+import org.apache.spark.sql.catalyst.util.{DateTimeConstants, DateTimeUtils}
import
org.apache.spark.sql.execution.datasources.parquet.ParentContainerUpdater
import org.apache.spark.sql.types.{TimestampLTZNanosType,
TimestampNTZNanosType}
import org.apache.spark.unsafe.types.TimestampNanosVal
@@ -118,8 +118,9 @@ class TimestampNanosParquetOpsSuite extends SparkFunSuite {
test("timestampNanosToEpochNanos combines micros and sub-micro nanos") {
val value = TimestampNanosVal.fromParts(1000000L, 500.toShort)
- assert(TimestampNanosParquetOps.timestampNanosToEpochNanos(value, isNtz =
false) ===
- 1000000L * DateTimeConstants.NANOS_PER_MICROS + 500L)
+ val packed = DateTimeUtils.timestampNanosToEpochNanos(
+ value, isNtz = false, sink = "Parquet INT64")
+ assert(packed === 1000000L * DateTimeConstants.NANOS_PER_MICROS + 500L)
}
test("timestampNanosToEpochNanos throws DATETIME_OVERFLOW outside the INT64
epoch-nanos range") {
@@ -128,7 +129,7 @@ class TimestampNanosParquetOpsSuite extends SparkFunSuite {
val tooLarge = TimestampNanosVal.fromParts(100000000000000000L, 0.toShort)
Seq(true, false).foreach { isNtz =>
val ex = intercept[SparkArithmeticException] {
- TimestampNanosParquetOps.timestampNanosToEpochNanos(tooLarge, isNtz)
+ DateTimeUtils.timestampNanosToEpochNanos(tooLarge, isNtz, sink =
"Parquet INT64")
}
assert(ex.getCondition === "DATETIME_OVERFLOW")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
