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 e7d95e0ca20b [SPARK-55444][SQL] Route nanosecond timestamp Parquet
support through the Types Framework
e7d95e0ca20b is described below
commit e7d95e0ca20b6f0306e1ed7737124ee077912234
Author: Stevo Mitric <[email protected]>
AuthorDate: Fri Jun 26 10:24:40 2026 +0200
[SPARK-55444][SQL] Route nanosecond timestamp Parquet support through the
Types Framework
### What changes were proposed in this pull request?
Migrate Parquet read/write for
`TimestampLTZNanosType`/`TimestampNTZNanosType` from inline `dt match` arms in
the `*Default` methods to the Phase-3a `ParquetTypeOps` framework, mirroring
how `TimeType` was integrated.
- Add `TimestampNanosParquetOps` (shared trait + LTZ/NTZ impls) covering
schema conversion, value write (INT64 epoch-nanos packing with overflow check),
and the row-based read converter.
- Register both nanos types in `ParquetTypeOps.apply`.
- Remove the now-dead nanos branches and private helpers from
`ParquetSchemaConverter`,
`ParquetWriteSupport`, `ParquetRowConverter`, and `ParquetUtils`
(framework dispatch runs first).
The Parquet→Spark read-schema inference stays inline since it keys on the
Parquet annotation, not a Spark type (same as `TimeType`).
### Why are the changes needed?
Two parallel Parquet code paths existed (the `ParquetTypeOps` registry for
`TimeType` and inline matches for nanos). This consolidates nanos onto the
single registration point, following the TimeType dead-branch cleanup in
SPARK-55444.
### Does this PR introduce _any_ user-facing change?
No. Behavior is unchanged; this is an internal refactor.
### How was this patch tested?
New tests in this PR.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56735 from stevomitric/stevomitric/route-parquet-ts-nanos.
Authored-by: Stevo Mitric <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
(cherry picked from commit e27a6046fe211f7a1cfbb263eb6efca5c462e41c)
Signed-off-by: Max Gekk <[email protected]>
---
.../datasources/parquet/ParquetRowConverter.scala | 49 +-----
.../parquet/ParquetSchemaConverter.scala | 8 -
.../datasources/parquet/ParquetUtils.scala | 4 +-
.../datasources/parquet/ParquetWriteSupport.scala | 24 ---
.../parquet/types/ops/ParquetTypeOps.scala | 4 +-
.../types/ops/TimestampNanosParquetOps.scala | 189 +++++++++++++++++++++
.../types/ops/TimestampNanosParquetOpsSuite.scala | 167 ++++++++++++++++++
7 files changed, 362 insertions(+), 83 deletions(-)
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 5c41f30255a4..1b633d63dc22 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, DateTimeConstants, DateTimeUtils, GenericArrayData,
ResolveDefaultColumns, STUtils}
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData,
CaseInsensitiveMap, 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
@@ -44,7 +44,7 @@ import
org.apache.spark.sql.execution.datasources.{DataSourceUtils, VariantMetad
import
org.apache.spark.sql.execution.datasources.parquet.types.ops.ParquetTypeOps
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
-import org.apache.spark.unsafe.types.{BinaryView, TimestampNanosVal,
UTF8String, VariantVal}
+import org.apache.spark.unsafe.types.{BinaryView, UTF8String, VariantVal}
import org.apache.spark.util.collection.Utils
/**
@@ -499,17 +499,6 @@ 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] =>
@@ -602,40 +591,6 @@ 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 df9db863f85f..3e33e33d6544 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
@@ -753,14 +753,6 @@ class SparkToParquetSchemaConverter(
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 f87e155d41ab..a884b4c5fe23 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
@@ -46,7 +46,7 @@ import
org.apache.spark.sql.execution.datasources.parquet.types.ops.ParquetTypeO
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, TimestampLTZNanosType,
TimestampNTZNanosType, UserDefinedType, VariantType}
+import org.apache.spark.sql.types.{ArrayType, AtomicType, DataType, MapType,
NullType, StructField, StructType, UserDefinedType, VariantType}
import org.apache.spark.util.ArrayImplicits._
object ParquetUtils extends Logging {
@@ -216,8 +216,6 @@ object ParquetUtils extends Logging {
.getOrElse(isBatchReadSupportedDefault(sqlConf, dt))
private def isBatchReadSupportedDefault(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 641a563cd7c1..5498611961d2 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
@@ -35,13 +35,11 @@ import
org.apache.spark.sql.{SPARK_LEGACY_DATETIME_METADATA_KEY, SPARK_LEGACY_IN
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.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.datasources.DataSourceUtils
import
org.apache.spark.sql.execution.datasources.parquet.types.ops.ParquetTypeOps
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
@@ -191,16 +189,6 @@ class ParquetWriteSupport extends
WriteSupport[InternalRow] with Logging {
}
}
- private def timestampNanosToEpochNanos(value: TimestampNanosVal, isNtz:
Boolean): Long = {
- try {
- DateTimeUtils.timestampNanosToEpochNanos(value)
- } catch {
- case _: ArithmeticException =>
- throw QueryExecutionErrors.timestampNanosEpochNanosOverflowError(
- value, isNtz, sink = "Parquet INT64")
- }
- }
-
// `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
= {
@@ -294,18 +282,6 @@ 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 BinaryType =>
(row: SpecializedGetters, ordinal: Int) =>
recordConsumer.addBinary(Binary.fromReusedByteArray(row.getBinary(ordinal)))
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetTypeOps.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetTypeOps.scala
index 76296500a792..598ec75fb45f 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetTypeOps.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetTypeOps.scala
@@ -27,7 +27,7 @@ import
org.apache.spark.sql.catalyst.expressions.SpecializedGetters
import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
import
org.apache.spark.sql.execution.datasources.parquet.{HasParentContainerUpdater,
ParentContainerUpdater, ParquetToSparkSchemaConverter}
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{DataType, StructType, TimeType}
+import org.apache.spark.sql.types.{DataType, StructType,
TimestampLTZNanosType, TimestampNTZNanosType, TimeType}
/**
* Optional trait for Parquet storage format integration in the Types
Framework.
@@ -220,6 +220,8 @@ private[parquet] object ParquetTypeOps {
def apply(dt: DataType): Option[ParquetTypeOps] = {
dt match {
case tt: TimeType => Some(TimeTypeParquetOps(tt))
+ case t: TimestampLTZNanosType => Some(TimestampLTZNanosParquetOps(t))
+ case t: TimestampNTZNanosType => Some(TimestampNTZNanosParquetOps(t))
// Add new types here - single registration point
case _ => None
}
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
new file mode 100644
index 000000000000..3359ba390809
--- /dev/null
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
@@ -0,0 +1,189 @@
+/*
+ * 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.types.ops
+
+import org.apache.parquet.io.api.{Converter, RecordConsumer}
+import org.apache.parquet.schema.{LogicalTypeAnnotation, Type, Types}
+import
org.apache.parquet.schema.LogicalTypeAnnotation.{TimestampLogicalTypeAnnotation,
TimeUnit}
+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.errors.QueryExecutionErrors
+import
org.apache.spark.sql.execution.datasources.parquet.{HasParentContainerUpdater,
ParentContainerUpdater, ParquetPrimitiveConverter}
+import org.apache.spark.sql.types.{DataType, TimestampLTZNanosType,
TimestampNTZNanosType}
+import org.apache.spark.unsafe.types.TimestampNanosVal
+
+/**
+ * Parquet operations shared by the nanosecond-precision timestamp types
+ * ([[TimestampLTZNanosType]] / [[TimestampNTZNanosType]]).
+ *
+ * Both are primitive types stored in Parquet as INT64 with a TIMESTAMP(NANOS)
annotation. The two
+ * differ only in the `isAdjustedToUTC` flag (LTZ = true, NTZ = false) and in
which row accessor /
+ * overflow-error flavor the write path uses; the schema annotation unit and
the entire read path
+ * are identical, so they share this trait and supply the differences via the
abstract members.
+ *
+ * IMPORTANT - internal vs Parquet representation:
+ * - Spark internal: [[TimestampNanosVal]] = (epochMicros: Long,
nanosWithinMicro: Short in
+ * [0, 999])
+ * - Parquet storage: INT64 epoch-nanoseconds (signed), so the on-disk range
is bounded to
+ * ~1677-09-21 .. 2262-04-11
+ * - Write path: (epochMicros, nanosWithinMicro) -> epochMicros * 1000 +
nanosWithinMicro, via
+ * `DateTimeUtils.timestampNanosToEpochNanos` (exact arithmetic);
out-of-range values throw
+ * `timestampNanosEpochNanosOverflowError`
+ * - Read path: epoch-nanos -> floorDiv / floorMod 1000 -> (epochMicros,
nanosWithinMicro) (floor
+ * semantics keep `nanosWithinMicro` in [0, 999] for pre-epoch values),
then the
+ * sub-microsecond digits are truncated to the requested precision
+ *
+ * TIMESTAMP(NANOS) postdates Spark's switch to the proleptic Gregorian
calendar, so the values are
+ * exempt from datetime rebasing (the rebase modes only cover DATE,
TIMESTAMP_MILLIS and
+ * TIMESTAMP_MICROS). Vectorized read is not supported: the value is a 16-byte
composite rather
+ * than a single long slot, so `isBatchReadSupported` stays false (the trait
default) and reads go
+ * through the row-based converter.
+ *
+ * @see ParquetTypeOps for the dispatch contract
+ * @since 4.3.0
+ */
+private[parquet] trait TimestampNanosParquetOps extends ParquetTypeOps {
+
+ /** The Spark type this ops handles, used for error messages. */
+ protected def sparkType: DataType
+
+ /** The requested fractional-second precision; sub-microsecond digits are
truncated to it. */
+ protected def precision: Int
+
+ /** True for [[TimestampNTZNanosType]] (no time zone), false for
[[TimestampLTZNanosType]]. */
+ protected def isNtz: Boolean
+
+ /** Reads the nanos value from the row using the type-specific accessor. */
+ protected def getNanos(row: SpecializedGetters, ordinal: Int):
TimestampNanosVal
+
+ // The Parquet TIMESTAMP `isAdjustedToUTC` flag: LTZ is UTC-adjusted, NTZ is
not.
+ private def isAdjustedToUTC: Boolean = !isNtz
+
+ // ==================== Schema Conversion ====================
+
+ override def convertToParquetType(
+ fieldName: String, repetition: Repetition, inShredded: Boolean): Type =
+ Types.primitive(INT64, repetition)
+ .as(LogicalTypeAnnotation.timestampType(isAdjustedToUTC, TimeUnit.NANOS))
+ .named(fieldName)
+
+ // ==================== Value Write ====================
+
+ override def makeWriter(
+ recordConsumer: () => RecordConsumer,
+ makeFieldWriter: DataType => (SpecializedGetters, Int) => Unit
+ ): (SpecializedGetters, Int) => Unit =
+ // TIMESTAMP(NANOS) values are always proleptic Gregorian and are exempt
from datetime
+ // rebasing. The supplier is evaluated at write time (not creation time)
because the
+ // RecordConsumer is null during init() and set later in prepareForWrite().
+ (row: SpecializedGetters, ordinal: Int) =>
+ recordConsumer().addLong(
+ TimestampNanosParquetOps.timestampNanosToEpochNanos(getNanos(row,
ordinal), isNtz))
+
+ // ==================== Row-Based Read ====================
+
+ override def newConverter(
+ parquetType: Type,
+ updater: ParentContainerUpdater): Converter with
HasParentContainerUpdater = {
+ // Framework-first dispatch in ParquetRowConverter routes here for any
nanos catalyst type,
+ // regardless of the actual Parquet encoding. Only an INT64
TIMESTAMP(NANOS) column can be
+ // decoded as a nanos timestamp; anything else (a non-NANOS timestamp, a
foreign annotation,
+ // etc.) must fail loudly, matching the legacy ParquetRowConverter
behavior where the guarded
+ // nanos arms fell through to the cannot-create-converter error.
+ if (!TimestampNanosParquetOps.isNanosTimestamp(parquetType)) {
+ throw QueryExecutionErrors.cannotCreateParquetConverterForDataTypeError(
+ sparkType, parquetType.toString)
+ }
+ 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))
+ }
+ }
+ }
+}
+
+/**
+ * Parquet operations for [[TimestampLTZNanosType]] (nanosecond precision,
with time zone).
+ * Stored as INT64 TIMESTAMP(NANOS, isAdjustedToUTC=true).
+ *
+ * @since 4.3.0
+ */
+case class TimestampLTZNanosParquetOps(t: TimestampLTZNanosType) extends
TimestampNanosParquetOps {
+ override protected def sparkType: DataType = t
+ override protected def precision: Int = t.precision
+ override protected def isNtz: Boolean = false
+ override protected def getNanos(row: SpecializedGetters, ordinal: Int):
TimestampNanosVal =
+ row.getTimestampLTZNanos(ordinal)
+}
+
+/**
+ * Parquet operations for [[TimestampNTZNanosType]] (nanosecond precision,
without time zone).
+ * Stored as INT64 TIMESTAMP(NANOS, isAdjustedToUTC=false).
+ *
+ * @since 4.3.0
+ */
+case class TimestampNTZNanosParquetOps(t: TimestampNTZNanosType) extends
TimestampNanosParquetOps {
+ override protected def sparkType: DataType = t
+ override protected def precision: Int = t.precision
+ override protected def isNtz: Boolean = true
+ override protected def getNanos(row: SpecializedGetters, ordinal: Int):
TimestampNanosVal =
+ row.getTimestampNTZNanos(ordinal)
+}
+
+private[ops] object TimestampNanosParquetOps {
+
+ /**
+ * Whether the Parquet field is an INT64 TIMESTAMP(NANOS) column. The
physical type is checked
+ * (isPrimitive && INT64) in addition to the logical annotation so a
malformed file that carries
+ * a TIMESTAMP(NANOS) annotation on a non-INT64 physical type is rejected by
the read guard with
+ * the clean cannotCreateParquetConverterForDataTypeError rather than
failing later in the
+ * primitive converter. Mirrors
TimeTypeParquetOps.requireCompatibleParquetType.
+ */
+ private[ops] def isNanosTimestamp(parquetType: Type): Boolean =
+ parquetType.isPrimitive &&
+ parquetType.asPrimitiveType.getPrimitiveTypeName == INT64 &&
+ (parquetType.getLogicalTypeAnnotation match {
+ 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
new file mode 100644
index 000000000000..4d08efb77c93
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOpsSuite.scala
@@ -0,0 +1,167 @@
+/*
+ * 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.types.ops
+
+import org.apache.parquet.io.api.PrimitiveConverter
+import org.apache.parquet.schema.{LogicalTypeAnnotation, Type, Types}
+import
org.apache.parquet.schema.LogicalTypeAnnotation.{TimestampLogicalTypeAnnotation,
TimeUnit}
+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.execution.datasources.parquet.ParentContainerUpdater
+import org.apache.spark.sql.types.{TimestampLTZNanosType,
TimestampNTZNanosType}
+import org.apache.spark.unsafe.types.TimestampNanosVal
+
+/**
+ * Unit tests for the nanosecond-timestamp Parquet ops
+ * ([[TimestampLTZNanosParquetOps]] / [[TimestampNTZNanosParquetOps]]), the
Types Framework
+ * integration for [[TimestampLTZNanosType]] / [[TimestampNTZNanosType]] in
Parquet.
+ *
+ * End-to-end read/write/round-trip behavior is covered by
`ParquetTimestampNanosSuite`; this
+ * suite pins the ops-level contracts the framework dispatch relies on:
+ * - the write schema annotation (INT64 TIMESTAMP(NANOS), isAdjustedToUTC =
true for LTZ /
+ * false for NTZ);
+ * - the read-path guard that fails loudly when a nanos type is requested
over a non-NANOS
+ * Parquet column (so framework-first dispatch never silently mis-decodes);
+ * - the (epochMicros, nanosWithinMicro) -> INT64 epoch-nanos packing and
its overflow error.
+ */
+class TimestampNanosParquetOpsSuite extends SparkFunSuite {
+
+ private val ltz = TimestampLTZNanosParquetOps(
+ TimestampLTZNanosType(TimestampLTZNanosType.NANOS_PRECISION))
+ private val ntz = TimestampNTZNanosParquetOps(
+ TimestampNTZNanosType(TimestampNTZNanosType.NANOS_PRECISION))
+
+ // ---------- schema conversion (write path) ----------
+
+ test("LTZ converts to INT64 TIMESTAMP(NANOS, isAdjustedToUTC=true)") {
+ assertNanosTimestampSchema(ltz.convertToParquetType("c", REQUIRED,
inShredded = false),
+ expectedAdjustedToUTC = true)
+ }
+
+ test("NTZ converts to INT64 TIMESTAMP(NANOS, isAdjustedToUTC=false)") {
+ assertNanosTimestampSchema(ntz.convertToParquetType("c", REQUIRED,
inShredded = false),
+ expectedAdjustedToUTC = false)
+ }
+
+ // ---------- read-path guard ----------
+
+ test("newConverter rejects a non-NANOS Parquet column (TIMESTAMP(MICROS))") {
+ val microsField = Types.primitive(INT64, REQUIRED)
+ .as(LogicalTypeAnnotation.timestampType(true, TimeUnit.MICROS))
+ .named("c")
+ val ex = intercept[SparkRuntimeException] {
+ ltz.newConverter(microsField, new ParentContainerUpdater {})
+ }
+ assert(ex.getCondition === "PARQUET_CONVERSION_FAILURE.UNSUPPORTED")
+ }
+
+ test("newConverter rejects a raw INT64 column with no annotation") {
+ val rawField = Types.primitive(INT64, REQUIRED).named("c")
+ val ex = intercept[SparkRuntimeException] {
+ ntz.newConverter(rawField, new ParentContainerUpdater {})
+ }
+ assert(ex.getCondition === "PARQUET_CONVERSION_FAILURE.UNSUPPORTED")
+ }
+
+ test("newConverter decodes INT64 epoch-nanos into the (epochMicros,
nanosWithinMicro) pair") {
+ // 1_000_000_500 ns = 1_000_000 micros + 500 ns; precision 9 keeps all
sub-micro digits.
+ assert(decode(ltz, nanosField(isAdjustedToUTC = true), 1000000500L) ===
+ TimestampNanosVal.fromParts(1000000L, 500.toShort))
+ // Floor semantics keep nanosWithinMicro in [0, 999] for pre-epoch values.
+ assert(decode(ltz, nanosField(isAdjustedToUTC = true), -1L) ===
+ TimestampNanosVal.fromParts(-1L, 999.toShort))
+ }
+
+ test("newConverter truncates sub-precision nanos to an explicit lower read
precision") {
+ val ntz7 = TimestampNTZNanosParquetOps(TimestampNTZNanosType(7))
+ // nanosWithinMicro 123 -> truncated to 100 at precision 7.
+ assert(decode(ntz7, nanosField(isAdjustedToUTC = false), 2000000123L) ===
+ TimestampNanosVal.fromParts(2000000L, 100.toShort))
+ }
+
+ // ---------- isNanosTimestamp helper ----------
+
+ test("isNanosTimestamp recognizes only INT64 TIMESTAMP(NANOS)") {
+ val nanos = Types.primitive(INT64, REQUIRED)
+ .as(LogicalTypeAnnotation.timestampType(false,
TimeUnit.NANOS)).named("c")
+ val micros = Types.primitive(INT64, REQUIRED)
+ .as(LogicalTypeAnnotation.timestampType(false,
TimeUnit.MICROS)).named("c")
+ val timeNanos = Types.primitive(INT64, REQUIRED)
+ .as(LogicalTypeAnnotation.timeType(false, TimeUnit.NANOS)).named("c")
+ val raw = Types.primitive(INT64, REQUIRED).named("c")
+
+ assert(TimestampNanosParquetOps.isNanosTimestamp(nanos))
+ assert(!TimestampNanosParquetOps.isNanosTimestamp(micros))
+ assert(!TimestampNanosParquetOps.isNanosTimestamp(timeNanos))
+ assert(!TimestampNanosParquetOps.isNanosTimestamp(raw))
+ }
+
+ // ---------- (epochMicros, nanosWithinMicro) -> INT64 epoch-nanos packing
----------
+
+ 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)
+ }
+
+ test("timestampNanosToEpochNanos throws DATETIME_OVERFLOW outside the INT64
epoch-nanos range") {
+ // Year ~5138; well past the int64 epoch-nanos cutoff (2262) but still
renderable, so the
+ // multiply overflows and the overflow error - not a rendering error - is
what surfaces.
+ val tooLarge = TimestampNanosVal.fromParts(100000000000000000L, 0.toShort)
+ Seq(true, false).foreach { isNtz =>
+ val ex = intercept[SparkArithmeticException] {
+ TimestampNanosParquetOps.timestampNanosToEpochNanos(tooLarge, isNtz)
+ }
+ assert(ex.getCondition === "DATETIME_OVERFLOW")
+ }
+ }
+
+ // ---------- helpers ----------
+
+ private def nanosField(isAdjustedToUTC: Boolean): Type =
+ Types.primitive(INT64, REQUIRED)
+ .as(LogicalTypeAnnotation.timestampType(isAdjustedToUTC, TimeUnit.NANOS))
+ .named("c")
+
+ // Builds the converter, feeds one INT64 epoch-nanos value through addLong,
and returns the
+ // value the converter set into its updater (the decoded TimestampNanosVal).
+ private def decode(ops: TimestampNanosParquetOps, field: Type, epochNanos:
Long): Any = {
+ var captured: Any = null
+ val updater = new ParentContainerUpdater {
+ override def set(value: Any): Unit = captured = value
+ }
+ ops.newConverter(field,
updater).asInstanceOf[PrimitiveConverter].addLong(epochNanos)
+ captured
+ }
+
+ private def assertNanosTimestampSchema(
+ parquetType: Type, expectedAdjustedToUTC: Boolean): Unit = {
+ assert(parquetType.isPrimitive, s"expected a primitive type, got
$parquetType")
+ assert(parquetType.asPrimitiveType.getPrimitiveTypeName === INT64)
+ parquetType.getLogicalTypeAnnotation match {
+ case ts: TimestampLogicalTypeAnnotation =>
+ assert(ts.getUnit === TimeUnit.NANOS)
+ assert(ts.isAdjustedToUTC === expectedAdjustedToUTC)
+ case other =>
+ fail(s"expected a TIMESTAMP logical type annotation, got $other")
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
