This is an automated email from the ASF dual-hosted git repository.
uros-b 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 bb62cf4121a2 [SPARK-57102][SQL] Support nanosecond-precision
timestamps in the Parquet data source
bb62cf4121a2 is described below
commit bb62cf4121a2dace3d5d3baec304c21180d2d164
Author: Stevo Mitric <[email protected]>
AuthorDate: Wed Jun 17 23:10:19 2026 +0200
[SPARK-57102][SQL] Support nanosecond-precision timestamps in the Parquet
data source
### What changes were proposed in this pull request?
This PR adds read and write support for the nanosecond-capable timestamp
types `TimestampNTZNanosType(p)` / `TimestampLTZNanosType(p)` (precision `p` in
`[7, 9]`, from the SPIP [SPARK-56822]) in the built-in Parquet data source,
gated behind the existing preview flag `spark.sql.timestampNanosTypes.enabled`.
- Schema conversion (`ParquetSchemaConverter`, both directions):
- Write: `TimestampLTZNanosType` / `TimestampNTZNanosType` -> `INT64`
annotated `TIMESTAMP(NANOS, isAdjustedToUTC)` (`isAdjustedToUTC = true` for
LTZ, `false` for NTZ).
- Read: `INT64` + `TIMESTAMP(NANOS, ...)` -> `TimestampLTZNanosType(9)` /
`TimestampNTZNanosType(9)`. Parquet's `NANOS` unit carries no precision
parameter, so reads mint the canonical precision 9. The legacy
`spark.sql.legacy.parquet.nanosAsLong` path keeps precedence and is unchanged.
- Read values (non-vectorized / row-based reader, `ParquetRowConverter`):
an `INT64` epoch-nanoseconds value is split into `epochMicros = floorDiv(v,
1000)` and `nanosWithinMicro = floorMod(v, 1000)` and stored as
`TimestampNanosVal`. `TIMESTAMP(NANOS)` values are exempt from datetime
rebasing on both read and write: the `NANOS` unit postdates the legacy
hybrid-calendar writers, so such files are always proleptic Gregorian (the
`spark.sql.parquet.datetimeRebaseModeIn{Read,Write}` conf [...]
- Write values (`ParquetWriteSupport`): a `TimestampNanosVal` is written as
`INT64` epoch-nanoseconds using exact arithmetic
(`Math.addExact(Math.multiplyExact(epochMicros, 1000), nanosWithinMicro)`);
values outside the representable `INT64` epoch-nanosecond range (~1677-09-21 ..
2262-04-11) fail instead of silently wrapping.
- The Parquet `supportDataType` guards (V1 `ParquetFileFormat` and V2
`ParquetTable`) are relaxed to accept the nanos types, and the feature flag is
propagated to the read Hadoop configuration in both the V1 and V2 paths.
- The nanos types are excluded from `ParquetUtils.isBatchReadSupported`, so
columnar reads transparently fall back to the row-based reader.
Vectorized-reader support is a follow-up.
Spark-written files round-trip the exact type (including precision) via the
Spark schema stored in the Parquet key-value metadata; "foreign" files with no
Spark metadata (e.g. produced by Trino/DuckDB/pandas) derive the nanos type
from the Parquet annotation.
### Why are the changes needed?
Nanosecond-precision timestamps are common in data produced by
pandas/PyArrow, Trino, ClickHouse, DuckDB, and similar systems. Spark currently
rejects Parquet `INT64 TIMESTAMP(NANOS)` (`PARQUET_TYPE_ILLEGAL`), or, with
`spark.sql.legacy.parquet.nanosAsLong=true`, reads it as a raw `LongType` that
drops all timestamp and time-zone semantics. This PR lets Spark read and write
such data as first-class nanosecond timestamp types, as part of the SPIP
[SPARK-56822] "Timestamps with nanoseco [...]
### Does this PR introduce _any_ user-facing change?
Yes, behind the preview flag `spark.sql.timestampNanosTypes.enabled`
(default off in production). When the flag is enabled:
- Parquet files with `INT64 TIMESTAMP(NANOS, isAdjustedToUTC=true/false)`
are read as `TimestampLTZNanosType(9)` / `TimestampNTZNanosType(9)` instead of
being rejected.
- Columns of these types can be written to Parquet (as `INT64
TIMESTAMP(NANOS)`).
When the flag is off, behavior is unchanged, including the legacy
`spark.sql.legacy.parquet.nanosAsLong` escape hatch.
### How was this patch tested?
New `ParquetTimestampNanosSuite` covering: Spark write/read round-trip
preserving value and precision at `p` = 7, 8, 9 (vectorized reader on and off);
reading "foreign" `TIMESTAMP(NANOS)` files written directly via parquet-mr for
both NTZ and LTZ, including a pre-epoch (negative) instant that exercises floor
semantics and nulls; `nanosAsLong` precedence; the disabled-feature error
pinned via `checkError` (`PARQUET_TYPE_ILLEGAL`, both `isAdjustedToUTC`
values); an out-of-`INT64`-range [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic Claude Opus 4.8)
Closes #56407 from stevomitric/stevomitric/SPARK-57102-parquet-nanos.
Authored-by: Stevo Mitric <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
(cherry picked from commit 59b6031202e315585181f1c379465cd7eab32440)
Signed-off-by: Uros Bojanic <[email protected]>
---
.../sql/catalyst/util/SparkDateTimeUtils.scala | 4 +-
.../spark/sql/errors/QueryExecutionErrors.scala | 19 +-
.../datasources/parquet/ParquetFileFormat.scala | 9 +-
.../datasources/parquet/ParquetRowConverter.scala | 49 ++-
.../parquet/ParquetSchemaConverter.scala | 20 ++
.../datasources/parquet/ParquetUtils.scala | 4 +-
.../datasources/parquet/ParquetWriteSupport.scala | 27 +-
.../datasources/v2/parquet/ParquetScan.scala | 3 +
.../datasources/v2/parquet/ParquetTable.scala | 3 -
.../spark/sql/FileBasedDataSourceSuite.scala | 7 +-
.../datasources/parquet/ParquetSchemaSuite.scala | 77 ++++-
.../parquet/ParquetTimestampNanosSuite.scala | 330 +++++++++++++++++++++
12 files changed, 526 insertions(+), 26 deletions(-)
diff --git
a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala
b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala
index d85aa65e3454..ea20057278ce 100644
---
a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala
+++
b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala
@@ -220,7 +220,9 @@ trait SparkDateTimeUtils {
* user-reachable input here. An out-of-range value therefore indicates an
internal caller bug
* and raises an internal error rather than silently retaining all
sub-microsecond digits.
*/
- private def truncateNanosWithinMicroToPrecision(nanosWithinMicro: Int,
precision: Int): Int = {
+ private[sql] def truncateNanosWithinMicroToPrecision(
+ nanosWithinMicro: Int,
+ precision: Int): Int = {
precision match {
case 7 => (nanosWithinMicro / 100) * 100
case 8 => (nanosWithinMicro / 10) * 10
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index d9441d6d3206..08c9e0462b8b 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -54,7 +54,7 @@ import
org.apache.spark.sql.internal.StaticSQLConf.GLOBAL_TEMP_DATABASE
import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.array.ByteArrayMethods
-import org.apache.spark.unsafe.types.UTF8String
+import org.apache.spark.unsafe.types.{TimestampNanosVal, UTF8String}
import org.apache.spark.util.{CircularBuffer, Utils}
/**
@@ -2609,6 +2609,23 @@ private[sql] object QueryExecutionErrors extends
QueryErrorsBase with ExecutionE
summary = "")
}
+ def parquetTimestampNanosOverflowError(
+ value: TimestampNanosVal, isNtz: Boolean): SparkArithmeticException = {
+ // Render TIMESTAMP_NTZ values without a zone (LocalDateTime, no trailing
`Z`); TIMESTAMP_LTZ
+ // values are absolute instants and render as UTC with a trailing `Z`.
+ val rendered =
+ if (isNtz) DateTimeUtils.timestampNanosToLocalDateTime(value).toString
+ else DateTimeUtils.timestampNanosToInstant(value).toString
+ new SparkArithmeticException(
+ errorClass = "DATETIME_OVERFLOW",
+ messageParameters = Map(
+ "operation" -> (s"write the timestamp value $rendered as Parquet INT64
" +
+ "epoch-nanoseconds " +
+ "(supported range: 1677-09-21T00:12:43.145224192Z to
2262-04-11T23:47:16.854775807Z)")),
+ context = Array.empty,
+ summary = "")
+ }
+
def timeAddIntervalOverflowError(
time: Long,
timePrecision: Int,
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
index 2b32c4e0d056..cf2626c2d63e 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
@@ -149,6 +149,9 @@ class ParquetFileFormat
hadoopConf.setBoolean(
SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key,
sqlConf.legacyParquetNanosAsLong)
+ hadoopConf.setBoolean(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key,
+ sqlConf.timestampNanosTypesEnabled)
hadoopConf.setBoolean(
SQLConf.PARQUET_READER_RESPECT_UNKNOWN_TYPE_ANNOTATION.key,
sqlConf.parquetReaderRespectUnknownTypeAnnotation)
@@ -416,9 +419,6 @@ class ParquetFileFormat
case g: GeometryType => GeometryType.isSridSupported(g.srid)
case g: GeographyType => GeographyType.isSridSupported(g.srid)
- // Nanosecond-capable timestamps are not yet supported by this datasource.
- case _: AnyTimestampNanoType => false
-
case _: AtomicType | _: NullType => true
case st: StructType => st.forall { f => supportDataType(f.dataType) }
@@ -458,6 +458,7 @@ object ParquetFileFormat extends Logging {
sqlConf.isParquetINT96AsTimestamp,
inferTimestampNTZ = sqlConf.parquetInferTimestampNTZEnabled,
nanosAsLong = sqlConf.legacyParquetNanosAsLong,
+ timestampNanosTypesEnabled = sqlConf.timestampNanosTypesEnabled,
respectUnknownTypeAnnotation =
sqlConf.parquetReaderRespectUnknownTypeAnnotation)
@@ -565,6 +566,7 @@ object ParquetFileFormat extends Logging {
val assumeInt96IsTimestamp = sqlConf.isParquetINT96AsTimestamp
val inferTimestampNTZ = sqlConf.parquetInferTimestampNTZEnabled
val nanosAsLong = sqlConf.legacyParquetNanosAsLong
+ val timestampNanosTypesEnabled = sqlConf.timestampNanosTypesEnabled
val respectUnknownTypeAnnotation =
sqlConf.parquetReaderRespectUnknownTypeAnnotation
@@ -576,6 +578,7 @@ object ParquetFileFormat extends Logging {
assumeInt96IsTimestamp = assumeInt96IsTimestamp,
inferTimestampNTZ = inferTimestampNTZ,
nanosAsLong = nanosAsLong,
+ timestampNanosTypesEnabled = timestampNanosTypesEnabled,
respectUnknownTypeAnnotation = respectUnknownTypeAnnotation)
readParquetFootersInParallel(conf, files, ignoreCorruptFiles,
ignoreMissingFiles)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
index 85cc50457323..6c9485dc6fc8 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
@@ -35,7 +35,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.types.{PhysicalByteType,
PhysicalShortType}
-import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData,
CaseInsensitiveMap, DateTimeUtils, GenericArrayData, ResolveDefaultColumns,
STUtils}
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData,
CaseInsensitiveMap, DateTimeConstants, DateTimeUtils, GenericArrayData,
ResolveDefaultColumns, STUtils}
import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
import org.apache.spark.sql.errors.QueryCompilationErrors
@@ -43,7 +43,7 @@ import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.datasources.{DataSourceUtils,
VariantMetadata}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
-import org.apache.spark.unsafe.types.{BinaryView, UTF8String, VariantVal}
+import org.apache.spark.unsafe.types.{BinaryView, TimestampNanosVal,
UTF8String, VariantVal}
import org.apache.spark.util.collection.Utils
/**
@@ -484,6 +484,17 @@ private[parquet] class ParquetRowConverter(
}
}
+ // The TIMESTAMP(NANOS) parquet type postdates Spark's switch to the
proleptic Gregorian
+ // calendar, so no legacy hybrid-calendar writer could have produced it.
Nanos values are
+ // always proleptic Gregorian and are exempt from datetime rebasing
+ // (`spark.sql.parquet.datetimeRebaseModeInRead` only covers DATE,
TIMESTAMP_MILLIS and
+ // TIMESTAMP_MICROS).
+ case t: TimestampLTZNanosType if isNanosTimestamp(parquetType) =>
+ makeNanosTimestampConverter(updater, t.precision)
+
+ case t: TimestampNTZNanosType if isNanosTimestamp(parquetType) =>
+ makeNanosTimestampConverter(updater, t.precision)
+
// Allow upcasting INT32 date to timestampNTZ.
case TimestampNTZType if
parquetType.asPrimitiveType().getPrimitiveTypeName == INT32 &&
parquetType.getLogicalTypeAnnotation.isInstanceOf[DateLogicalTypeAnnotation] =>
@@ -587,6 +598,40 @@ private[parquet] class ParquetRowConverter(
private def canReadAsTimestampNTZ(parquetType: Type): Boolean =
parquetType.getLogicalTypeAnnotation.isInstanceOf[TimestampLogicalTypeAnnotation]
+ // A Parquet INT64 column annotated as TIMESTAMP(NANOS), read into one of the
+ // nanosecond-precision Spark timestamp types.
+ private def isNanosTimestamp(parquetType: Type): Boolean =
+ parquetType.getLogicalTypeAnnotation match {
+ case ts: TimestampLogicalTypeAnnotation => ts.getUnit == TimeUnit.NANOS
+ case _ => false
+ }
+
+ /**
+ * Builds a converter for a Parquet INT64 `TIMESTAMP(NANOS)` column read
into a
+ * nanosecond-precision Spark type ([[TimestampNTZNanosType]] /
[[TimestampLTZNanosType]]). The
+ * int64 epoch-nanoseconds value is split into the `(epochMicros,
nanosWithinMicro)` pair with
+ * floor semantics (so pre-epoch values keep `nanosWithinMicro` in `[0,
999]`), then the
+ * sub-microsecond digits are truncated to `precision`. The truncation
mirrors
+ * [[DateTimeUtils.instantToTimestampNanos]] /
[[DateTimeUtils.localDateTimeToTimestampNanos]];
+ * it matters when an explicit read schema (e.g. `TIMESTAMP_NTZ(7)`) is
applied to a foreign
+ * full-precision file - otherwise the stored value would carry digits below
`precision`,
+ * violating the invariant the rest of the stack maintains. NANOS is exempt
from datetime
+ * rebasing (see the call site).
+ */
+ private def makeNanosTimestampConverter(
+ updater: ParentContainerUpdater,
+ precision: Int): ParquetPrimitiveConverter =
+ 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,
precision)
+ this.updater.set(TimestampNanosVal.fromParts(epochMicros,
nanosWithinMicro.toShort))
+ }
+ }
+
/**
* Parquet converter for strings. A dictionary is used to minimize string
decoding cost.
*/
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala
index b8ff84325635..c479c37b89fd 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala
@@ -60,6 +60,7 @@ class ParquetToSparkSchemaConverter(
caseSensitive: Boolean = SQLConf.CASE_SENSITIVE.defaultValue.get,
inferTimestampNTZ: Boolean =
SQLConf.PARQUET_INFER_TIMESTAMP_NTZ_ENABLED.defaultValue.get,
nanosAsLong: Boolean =
SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.defaultValue.get,
+ timestampNanosTypesEnabled: Boolean = false,
useFieldId: Boolean =
SQLConf.PARQUET_FIELD_ID_READ_ENABLED.defaultValue.get,
val ignoreVariantAnnotation: Boolean =
SQLConf.PARQUET_IGNORE_VARIANT_ANNOTATION.defaultValue.get,
@@ -72,6 +73,7 @@ class ParquetToSparkSchemaConverter(
caseSensitive = conf.caseSensitiveAnalysis,
inferTimestampNTZ = conf.parquetInferTimestampNTZEnabled,
nanosAsLong = conf.legacyParquetNanosAsLong,
+ timestampNanosTypesEnabled = conf.timestampNanosTypesEnabled,
useFieldId = conf.parquetFieldIdReadEnabled,
ignoreVariantAnnotation = conf.parquetIgnoreVariantAnnotation,
respectUnknownTypeAnnotation =
@@ -83,6 +85,8 @@ class ParquetToSparkSchemaConverter(
caseSensitive = conf.get(SQLConf.CASE_SENSITIVE.key).toBoolean,
inferTimestampNTZ =
conf.get(SQLConf.PARQUET_INFER_TIMESTAMP_NTZ_ENABLED.key).toBoolean,
nanosAsLong = conf.get(SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key).toBoolean,
+ timestampNanosTypesEnabled =
+ conf.getBoolean(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key, false),
useFieldId = conf.getBoolean(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key,
SQLConf.PARQUET_FIELD_ID_READ_ENABLED.defaultValue.get),
ignoreVariantAnnotation =
conf.getBoolean(SQLConf.PARQUET_IGNORE_VARIANT_ANNOTATION.key,
@@ -327,6 +331,13 @@ class ParquetToSparkSchemaConverter(
case timestamp: TimestampLogicalTypeAnnotation
if timestamp.getUnit == TimeUnit.NANOS && nanosAsLong =>
LongType
+ case timestamp: TimestampLogicalTypeAnnotation
+ if timestamp.getUnit == TimeUnit.NANOS &&
timestampNanosTypesEnabled =>
+ if (timestamp.isAdjustedToUTC) {
+ TimestampLTZNanosType(TimestampLTZNanosType.NANOS_PRECISION)
+ } else {
+ TimestampNTZNanosType(TimestampNTZNanosType.NANOS_PRECISION)
+ }
case time: TimeLogicalTypeAnnotation
if time.getUnit == TimeUnit.MICROS && !time.isAdjustedToUTC =>
TimeType(TimeType.MICROS_PRECISION)
@@ -735,6 +746,15 @@ class SparkToParquetSchemaConverter(
case TimestampNTZType =>
Types.primitive(INT64, repetition)
.as(LogicalTypeAnnotation.timestampType(false,
TimeUnit.MICROS)).named(field.name)
+
+ case _: TimestampLTZNanosType =>
+ Types.primitive(INT64, repetition)
+ .as(LogicalTypeAnnotation.timestampType(true,
TimeUnit.NANOS)).named(field.name)
+
+ case _: TimestampNTZNanosType =>
+ Types.primitive(INT64, repetition)
+ .as(LogicalTypeAnnotation.timestampType(false,
TimeUnit.NANOS)).named(field.name)
+
case BinaryType =>
Types.primitive(BINARY, repetition).named(field.name)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala
index b4e0fe5fddb8..c60754813994 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala
@@ -45,7 +45,7 @@ import
org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils, DataS
import org.apache.spark.sql.execution.datasources.v2.V2ColumnUtils
import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.internal.SQLConf.PARQUET_AGGREGATE_PUSHDOWN_ENABLED
-import org.apache.spark.sql.types.{ArrayType, AtomicType, DataType, MapType,
NullType, StructField, StructType, UserDefinedType, VariantType}
+import org.apache.spark.sql.types.{ArrayType, AtomicType, DataType, MapType,
NullType, StructField, StructType, TimestampLTZNanosType,
TimestampNTZNanosType, UserDefinedType, VariantType}
import org.apache.spark.util.ArrayImplicits._
object ParquetUtils extends Logging {
@@ -207,6 +207,8 @@ object ParquetUtils extends Logging {
schema.forall(f => isBatchReadSupported(sqlConf, f.dataType))
def isBatchReadSupported(sqlConf: SQLConf, dt: DataType): Boolean = dt match
{
+ case _: TimestampNTZNanosType | _: TimestampLTZNanosType =>
+ false
case _: AtomicType =>
true
case _: NullType =>
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala
index d7c9a24245e5..d7fd5991c75f 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala
@@ -34,11 +34,13 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.{SPARK_LEGACY_DATETIME_METADATA_KEY,
SPARK_LEGACY_INT96_METADATA_KEY, SPARK_TIMEZONE_METADATA_KEY,
SPARK_VERSION_METADATA_KEY}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
-import org.apache.spark.sql.catalyst.util.{DateTimeUtils, STUtils}
+import org.apache.spark.sql.catalyst.util.{DateTimeConstants, DateTimeUtils,
STUtils}
+import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.datasources.DataSourceUtils
import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.types._
import org.apache.spark.types.variant.Variant
+import org.apache.spark.unsafe.types.TimestampNanosVal
/**
* A Parquet [[WriteSupport]] implementation that writes Catalyst
[[InternalRow]]s as Parquet
@@ -188,6 +190,17 @@ class ParquetWriteSupport extends
WriteSupport[InternalRow] with Logging {
}
}
+ private def timestampNanosToEpochNanos(value: TimestampNanosVal, isNtz:
Boolean): Long = {
+ try {
+ Math.addExact(
+ Math.multiplyExact(value.epochMicros,
DateTimeConstants.NANOS_PER_MICROS),
+ value.nanosWithinMicro.toLong)
+ } catch {
+ case _: ArithmeticException =>
+ throw QueryExecutionErrors.parquetTimestampNanosOverflowError(value,
isNtz)
+ }
+ }
+
// `inShredded` indicates whether the current traversal is nested within a
shredded Variant
// schema. This affects how timestamp values are written.
private def makeWriter(dataType: DataType, inShredded: Boolean): ValueWriter
= {
@@ -268,6 +281,18 @@ class ParquetWriteSupport extends
WriteSupport[InternalRow] with Logging {
// MICROS time unit.
(row: SpecializedGetters, ordinal: Int) =>
recordConsumer.addLong(row.getLong(ordinal))
+ // TIMESTAMP(NANOS) values are always proleptic Gregorian and are exempt
from datetime
+ // rebasing; see the TIMESTAMP(NANOS) converters in
`ParquetRowConverter` for details.
+ case _: TimestampLTZNanosType =>
+ (row: SpecializedGetters, ordinal: Int) =>
+ recordConsumer.addLong(
+ timestampNanosToEpochNanos(row.getTimestampLTZNanos(ordinal),
isNtz = false))
+
+ case _: TimestampNTZNanosType =>
+ (row: SpecializedGetters, ordinal: Int) =>
+ recordConsumer.addLong(
+ timestampNanosToEpochNanos(row.getTimestampNTZNanos(ordinal),
isNtz = true))
+
case _: TimeType =>
(row: SpecializedGetters, ordinal: Int) =>
recordConsumer.addLong(DateTimeUtils.nanosToMicros(row.getLong(ordinal)))
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
index d0c7859964e0..c029ac4d0d98 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
@@ -161,6 +161,9 @@ case class ParquetScan(
hadoopConf.setBoolean(
SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key,
conf.legacyParquetNanosAsLong)
+ hadoopConf.setBoolean(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key,
+ conf.timestampNanosTypesEnabled)
hadoopConf.setBoolean(
SQLConf.PARQUET_READER_RESPECT_UNKNOWN_TYPE_ANNOTATION.key,
conf.parquetReaderRespectUnknownTypeAnnotation)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala
index 19974f613cc0..67052c201a9d 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala
@@ -55,9 +55,6 @@ case class ParquetTable(
case g: GeometryType => GeometryType.isSridSupported(g.srid)
case g: GeographyType => GeographyType.isSridSupported(g.srid)
- // 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 e5f64545986f..dbf5dee4146c 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
@@ -1336,9 +1336,10 @@ class FileBasedDataSourceSuite extends SharedSparkSession
}
}
- test("SPARK-57166: nanosecond timestamp types are not supported in file data
sources") {
- // None of these built-in file formats support nanosecond-capable
timestamps yet.
- val unsupportedDataSources = Seq("parquet", "orc", "json", "csv", "xml")
+ test("SPARK-57166: nanosecond timestamp types are not supported in
non-Parquet file sources") {
+ // These built-in file formats do not support nanosecond-capable
timestamps. Parquet support is
+ // covered separately in ParquetTimestampNanosSuite.
+ val unsupportedDataSources = Seq("orc", "json", "csv", "xml")
val nanosTypes = Seq(TimestampNTZNanosType(9), TimestampLTZNanosType(9))
withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
// Test both v1 and v2 data sources.
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala
index c8d866d8b2d0..e468a1449783 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala
@@ -60,6 +60,7 @@ abstract class ParquetSchemaTest extends ParquetTest with
SharedSparkSession {
nanosAsLong = nanosAsLong)
}
+ // scalastyle:off argcount
protected def testParquetToCatalyst(
testName: String,
sqlSchema: StructType,
@@ -70,13 +71,16 @@ abstract class ParquetSchemaTest extends ParquetTest with
SharedSparkSession {
inferTimestampNTZ: Boolean = true,
sparkReadSchema: Option[StructType] = None,
expectedParquetColumn: Option[ParquetColumn] = None,
- nanosAsLong: Boolean = false): Unit = {
+ nanosAsLong: Boolean = false,
+ timestampNanosTypesEnabled: Boolean = false): Unit = {
+ // scalastyle:on argcount
val converter = new ParquetToSparkSchemaConverter(
assumeBinaryIsString = binaryAsString,
assumeInt96IsTimestamp = int96AsTimestamp,
caseSensitive = caseSensitive,
inferTimestampNTZ = inferTimestampNTZ,
- nanosAsLong = nanosAsLong)
+ nanosAsLong = nanosAsLong,
+ timestampNanosTypesEnabled = timestampNanosTypesEnabled)
test(s"sql <= parquet: $testName") {
val actualParquetColumn = converter.convertParquetColumn(
@@ -126,7 +130,8 @@ abstract class ParquetSchemaTest extends ParquetTest with
SharedSparkSession {
outputTimestampType: SQLConf.ParquetOutputTimestampType.Value =
SQLConf.ParquetOutputTimestampType.INT96,
expectedParquetColumn: Option[ParquetColumn] = None,
- nanosAsLong: Boolean = false): Unit = {
+ nanosAsLong: Boolean = false,
+ timestampNanosTypesEnabled: Boolean = false): Unit = {
testCatalystToParquet(
testName,
@@ -142,7 +147,8 @@ abstract class ParquetSchemaTest extends ParquetTest with
SharedSparkSession {
binaryAsString,
int96AsTimestamp,
expectedParquetColumn = expectedParquetColumn,
- nanosAsLong = nanosAsLong)
+ nanosAsLong = nanosAsLong,
+ timestampNanosTypesEnabled = timestampNanosTypesEnabled)
}
protected def compareParquetColumn(actual: ParquetColumn, expected:
ParquetColumn): Unit = {
@@ -1111,13 +1117,15 @@ class ParquetSchemaSuite extends ParquetSchemaTest {
test("SPARK-40819: parquet file with TIMESTAMP(NANOS, true) (with default
nanosAsLong=false)") {
val testDataPath = testFile("test-data/timestamp-nanos.parquet")
- checkError(
- exception = intercept[AnalysisException] {
- spark.read.parquet(testDataPath).collect()
- },
- condition = "PARQUET_TYPE_ILLEGAL",
- parameters = Map("parquetType" -> "INT64 (TIMESTAMP(NANOS,true))")
- )
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "false") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.read.parquet(testDataPath).collect()
+ },
+ condition = "PARQUET_TYPE_ILLEGAL",
+ parameters = Map("parquetType" -> "INT64 (TIMESTAMP(NANOS,true))")
+ )
+ }
}
test("SPARK-47261: parquet file with unsupported type") {
@@ -2630,6 +2638,53 @@ class ParquetSchemaSuite extends ParquetSchemaTest {
inferTimestampNTZ = true)
}
+ // The nanosecond timestamp types are written as INT64 with the
TIMESTAMP(NANOS) annotation
+ // and, with `spark.sql.timestampNanosTypes.enabled`, read back at the
canonical precision 9.
+ testSchema(
+ "SPARK-57102: TimestampNTZNanos written and read as INT64 with
TIMESTAMP(NANOS,false)",
+ StructType(Seq(StructField("f1", TimestampNTZNanosType(9)))),
+ """message root {
+ | optional INT64 f1 (TIMESTAMP(NANOS,false));
+ |}
+ """.stripMargin,
+ binaryAsString = true,
+ int96AsTimestamp = true,
+ writeLegacyParquetFormat = false,
+ timestampNanosTypesEnabled = true)
+
+ testSchema(
+ "SPARK-57102: TimestampLTZNanos written and read as INT64 with
TIMESTAMP(NANOS,true)",
+ StructType(Seq(StructField("f1", TimestampLTZNanosType(9)))),
+ """message root {
+ | optional INT64 f1 (TIMESTAMP(NANOS,true));
+ |}
+ """.stripMargin,
+ binaryAsString = true,
+ int96AsTimestamp = true,
+ writeLegacyParquetFormat = false,
+ timestampNanosTypesEnabled = true)
+
+ testCatalystToParquet(
+ "SPARK-57102: TimestampNTZNanos(7) is written as INT64 with
TIMESTAMP(NANOS,false)",
+ StructType(Seq(StructField("f1", TimestampNTZNanosType(7)))),
+ """message root {
+ | optional INT64 f1 (TIMESTAMP(NANOS,false));
+ |}
+ """.stripMargin,
+ writeLegacyParquetFormat = false)
+
+ testParquetToCatalyst(
+ "SPARK-57102: legacy nanosAsLong takes precedence over the nanos timestamp
types",
+ StructType(Seq(StructField("f1", LongType))),
+ """message root {
+ | optional INT64 f1 (TIMESTAMP(NANOS,true));
+ |}
+ """.stripMargin,
+ binaryAsString = true,
+ int96AsTimestamp = true,
+ nanosAsLong = true,
+ timestampNanosTypesEnabled = true)
+
testCatalystToParquet(
"TimestampNTZ Spark to Parquet conversion for complex types",
StructType(
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala
new file mode 100644
index 000000000000..b3e8ecdd5868
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala
@@ -0,0 +1,330 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.io.File
+
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.example.data.simple.SimpleGroupFactory
+import org.apache.parquet.hadoop.ParquetFileWriter.Mode
+import org.apache.parquet.hadoop.example.ExampleParquetWriter
+import org.apache.parquet.schema.{LogicalTypeAnnotation, MessageType, Types}
+import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64
+
+import org.apache.spark.{SparkArithmeticException, SparkException}
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+
+class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with
SharedSparkSession {
+
+ private def withNanosEnabled(f: => Unit): Unit = {
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true")(f)
+ }
+
+ private def writeForeignNanosParquet(
+ file: File,
+ isAdjustedToUTC: Boolean,
+ values: Seq[Option[Long]]): Unit = {
+ val schema: MessageType = Types.buildMessage()
+ .optional(INT64)
+ .as(LogicalTypeAnnotation.timestampType(isAdjustedToUTC, TimeUnit.NANOS))
+ .named("ts")
+ .named("spark_schema")
+ val conf = spark.sessionState.newHadoopConf()
+ val writer = ExampleParquetWriter.builder(new Path(file.toURI))
+ .withType(schema)
+ .withConf(conf)
+ .withWriteMode(Mode.OVERWRITE)
+ .build()
+ try {
+ val factory = new SimpleGroupFactory(schema)
+ values.foreach { v =>
+ val group = factory.newGroup()
+ v.foreach(x => group.add("ts", x))
+ writer.write(group)
+ }
+ } finally {
+ writer.close()
+ }
+ }
+
+ test("SPARK-57102: Spark write/read round-trips nanos value and precision") {
+ withNanosEnabled {
+ Seq("true", "false").foreach { vectorized =>
+ withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key ->
vectorized) {
+ Seq(7, 8, 9).foreach { p =>
+ withTempPath { dir =>
+ val frac = "123456789".take(p)
+ val df = spark.sql(
+ s"""SELECT * FROM VALUES
+ | (TIMESTAMP_NTZ '2020-01-01 12:34:56.$frac',
+ | TIMESTAMP_LTZ '2020-01-01 12:34:56.$frac'),
+ | (TIMESTAMP_NTZ '1969-12-31 23:59:59.$frac',
+ | TIMESTAMP_LTZ '1969-12-31 23:59:59.$frac'),
+ | (CAST(NULL AS TIMESTAMP_NTZ($p)), CAST(NULL AS
TIMESTAMP_LTZ($p)))
+ | AS t(ntz, ltz)""".stripMargin)
+ assert(df.schema("ntz").dataType === TimestampNTZNanosType(p))
+ assert(df.schema("ltz").dataType === TimestampLTZNanosType(p))
+
+ df.write.parquet(dir.getCanonicalPath)
+ val read = spark.read.parquet(dir.getCanonicalPath)
+
+ assert(read.schema("ntz").dataType === TimestampNTZNanosType(p))
+ assert(read.schema("ltz").dataType === TimestampLTZNanosType(p))
+ checkAnswer(read, df.collect().toSeq)
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57102: read a foreign TIMESTAMP(NANOS) file as nanosecond
timestamp types") {
+ withNanosEnabled {
+ withSQLConf(
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ val values = Seq(Some(123L), Some(1000000123L), Some(-1L), None)
+
+ withTempPath { dir =>
+ val file = new File(dir, "ntz.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = false, values)
+ val read = spark.read.parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === TimestampNTZNanosType(9))
+ checkAnswer(read, spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_NTZ '1970-01-01 00:00:00.000000123'),
+ | (TIMESTAMP_NTZ '1970-01-01 00:00:01.000000123'),
+ | (TIMESTAMP_NTZ '1969-12-31 23:59:59.999999999'),
+ | (CAST(NULL AS TIMESTAMP_NTZ(9)))
+ | AS t(ts)""".stripMargin).collect().toSeq)
+ }
+
+ withTempPath { dir =>
+ val file = new File(dir, "ltz.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = true, values)
+ val read = spark.read.parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === TimestampLTZNanosType(9))
+ checkAnswer(read, spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_LTZ '1970-01-01 00:00:00.000000123'),
+ | (TIMESTAMP_LTZ '1970-01-01 00:00:01.000000123'),
+ | (TIMESTAMP_LTZ '1969-12-31 23:59:59.999999999'),
+ | (CAST(NULL AS TIMESTAMP_LTZ(9)))
+ | AS t(ts)""".stripMargin).collect().toSeq)
+ }
+ }
+ }
+ }
+
+ test("SPARK-57102: explicit lower-precision read schema truncates
sub-precision nanos") {
+ withNanosEnabled {
+ withSQLConf(
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ // Foreign file with full 9-digit precision: .123456789 after the
epoch, and -1ns (which
+ // floors to epochMicros = -1, nanosWithinMicro = 999).
+ val values = Seq(Some(123456789L), Some(-1L))
+
+ withTempPath { dir =>
+ val file = new File(dir, "ntz.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = false, values)
+ // Reading with an explicit TIMESTAMP_NTZ(7) schema must floor the
sub-microsecond
+ // digits to precision 7, matching
DateTimeUtils.localDateTimeToTimestampNanos.
+ val read = spark.read.schema("ts
TIMESTAMP_NTZ(7)").parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === TimestampNTZNanosType(7))
+ checkAnswer(read, spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_NTZ '1970-01-01 00:00:00.123456700'),
+ | (TIMESTAMP_NTZ '1969-12-31 23:59:59.999999900')
+ | AS t(ts)""".stripMargin).collect().toSeq)
+ }
+
+ withTempPath { dir =>
+ val file = new File(dir, "ltz.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = true, values)
+ // precision 8 drops only the last digit.
+ val read = spark.read.schema("ts
TIMESTAMP_LTZ(8)").parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === TimestampLTZNanosType(8))
+ checkAnswer(read, spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_LTZ '1970-01-01 00:00:00.123456780'),
+ | (TIMESTAMP_LTZ '1969-12-31 23:59:59.999999990')
+ | AS t(ts)""".stripMargin).collect().toSeq)
+ }
+ }
+ }
+ }
+
+ test("SPARK-57102: requesting a nanos type over a non-NANOS Parquet column
fails clearly") {
+ withNanosEnabled {
+ withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ withTempPath { dir =>
+ // Write a microsecond column: the Parquet annotation is
TIMESTAMP(MICROS), not NANOS.
+ spark.sql("SELECT TIMESTAMP_NTZ '2020-01-01 12:34:56.123456' AS ts")
+ .write.parquet(dir.getCanonicalPath)
+ // Forcing a nanosecond read schema leaves no matching converter
case (the guard requires
+ // a NANOS annotation), so it falls through to the generic
PARQUET_CONVERSION_FAILURE
+ // error - the same path every other type uses, not a confusing one.
+ val e = intercept[SparkException] {
+ spark.read.schema("ts
TIMESTAMP_NTZ(7)").parquet(dir.getCanonicalPath).collect()
+ }
+ var cause: Throwable = e
+ while (cause != null && (cause.getMessage == null ||
+ !cause.getMessage.contains("PARQUET_CONVERSION_FAILURE"))) {
+ cause = cause.getCause
+ }
+ assert(cause != null,
+ s"Expected a PARQUET_CONVERSION_FAILURE error, but got:
${e.getMessage}")
+ }
+ }
+ }
+ }
+
+ test("SPARK-57102: legacy nanosAsLong reads a foreign TIMESTAMP(NANOS) file
as LongType") {
+ withTempPath { dir =>
+ val file = new File(dir, "foreign.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = false, Seq(Some(123L),
Some(-1L)))
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key -> "true",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ val read = spark.read.parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === LongType)
+ checkAnswer(read, Seq(Row(123L), Row(-1L)))
+ }
+ }
+ }
+
+ test("SPARK-57102: reading a foreign TIMESTAMP(NANOS) file fails when the
feature is disabled") {
+ Seq(true, false).foreach { isAdjustedToUTC =>
+ withTempPath { dir =>
+ val file = new File(dir, "foreign.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC, Seq(Some(123L)))
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "false",
+ SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key -> "false") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.read.parquet(file.getCanonicalPath).schema
+ },
+ condition = "PARQUET_TYPE_ILLEGAL",
+ parameters = Map("parquetType" -> s"INT64
(TIMESTAMP(NANOS,$isAdjustedToUTC))"))
+ }
+ }
+ }
+ }
+
+ test("SPARK-57102: writing a timestamp outside the INT64 epoch-nanos range
fails loudly") {
+ withNanosEnabled {
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ Seq("TIMESTAMP_NTZ", "TIMESTAMP_LTZ").foreach { typeName =>
+ withTempPath { dir =>
+ val df = spark.sql(s"SELECT $typeName '9999-12-31
23:59:59.999999999' AS ts")
+ val e = intercept[SparkException] {
+ df.write.parquet(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 = cause.asInstanceOf[SparkArithmeticException],
+ condition = "DATETIME_OVERFLOW",
+ parameters = Map("operation" ->
+ (s"write the timestamp value $renderedValue as Parquet INT64 "
+
+ "epoch-nanoseconds (supported range:
1677-09-21T00:12:43.145224192Z to " +
+ "2262-04-11T23:47:16.854775807Z)")))
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57102: datetime rebase configs do not affect TIMESTAMP(NANOS)
reads") {
+ withNanosEnabled {
+ withSQLConf(
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ withTempPath { dir =>
+ val file = new File(dir, "ltz.parquet")
+ // 1800-01-01 00:00:00 UTC predates the last Julian-to-Gregorian
switch instant of the
+ // rebase logic, so applying a timestamp rebase (or the
EXCEPTION-mode guard) on this
+ // value would change the result or fail the read.
+ writeForeignNanosParquet(
+ file, isAdjustedToUTC = true, Seq(Some(-5364662400000000000L)))
+ Seq("EXCEPTION", "CORRECTED", "LEGACY").foreach { mode =>
+ withSQLConf(SQLConf.PARQUET_REBASE_MODE_IN_READ.key -> mode) {
+ checkAnswer(
+ spark.read.parquet(file.getCanonicalPath),
+ spark.sql(
+ "SELECT TIMESTAMP_LTZ '1800-01-01
00:00:00.000000000'").collect().toSeq)
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57102: nanos timestamps round-trip inside a nested (array)
column") {
+ withNanosEnabled {
+ withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ withTempPath { dir =>
+ val df = spark.sql(
+ "SELECT array(TIMESTAMP_NTZ '2020-01-01 00:00:00.123456789', " +
+ "TIMESTAMP_NTZ '1969-12-31 23:59:59.000000001') AS arr")
+ df.write.parquet(dir.getCanonicalPath)
+ val read = spark.read.parquet(dir.getCanonicalPath)
+
assert(read.schema("arr").dataType.asInstanceOf[ArrayType].elementType ===
+ TimestampNTZNanosType(9))
+ checkAnswer(read, df.collect().toSeq)
+ }
+ }
+ }
+ }
+
+ test("SPARK-57102: nanos timestamps round-trip via the V2 file source") {
+ withNanosEnabled {
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> "",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ withTempPath { dir =>
+ val df = spark.sql(
+ "SELECT TIMESTAMP_NTZ '2020-01-01 12:34:56.123456789' AS ntz, " +
+ "TIMESTAMP_LTZ '2020-01-01 12:34:56.123456789' AS ltz")
+ df.write.parquet(dir.getCanonicalPath)
+ val read = spark.read.parquet(dir.getCanonicalPath)
+ assert(read.schema("ntz").dataType === TimestampNTZNanosType(9))
+ assert(read.schema("ltz").dataType === TimestampLTZNanosType(9))
+ checkAnswer(read, df.collect().toSeq)
+ }
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
