You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ur...@apache.org on 2026/08/02 17:21:46 UTC
(spark) branch branch-4.x updated: [SPARK-57161][CONNECT][SQL] Convert nanosecond-capable timestamp types and literals between proto and Catalyst in Spark Connect
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 681f59598512 [SPARK-57161][CONNECT][SQL] Convert nanosecond-capable timestamp types and literals between proto and Catalyst in Spark Connect
681f59598512 is described below
commit 681f5959851296a7bbef94dbef97e8ce4f57b52a
Author: Stevo Mitric <st...@gmail.com>
AuthorDate: Sun Aug 2 17:21:25 2026 +0000
[SPARK-57161][CONNECT][SQL] Convert nanosecond-capable timestamp types and literals between proto and Catalyst in Spark Connect
### What changes were proposed in this pull request?
This PR adds bidirectional proto <-> Catalyst conversion for the nanosecond-capable timestamp types `TimestampNTZNanosType(p)` and `TimestampLTZNanosType(p)` (precision `p` in [7, 9]) in Spark Connect's shared connect-common converters, so the protocol messages added by SPARK-57160 become usable by both the JVM Connect client and server.
The Connect Types Framework delegates all proto DataType/Literal conversion through `ConnectTypeOps`, so the change is small and mirrors the existing `TimeType` integration:
- New `TimestampNanosTypeConnectOps` (a shared base plus `TimestampNTZNanosTypeConnectOps` / `TimestampLTZNanosTypeConnectOps`) implementing the proto DataType and Literal side of `ConnectTypeOps`. The external java.time value is `LocalDateTime` (NTZ) / `Instant` (LTZ), matching the server-side `TypeOps` and `RowEncoder`. The literal carries the physical value as `epoch_micros` + `nanos_within_micro` (mirroring the Catalyst `TimestampNanosVal`), because a single int64 of nanoseconds c [...]
- `ConnectTypeOps`: register both types at the three proto-dispatch points (`apply`, `opsForKindCase`, `literalCaseToKindCase`).
- `LiteralValueProtoConverter`: add the two `isCompatible` literal/type cases.
Arrow IPC serialization is out of scope for this sub-task, so the ops is not registered in the Arrow dispatch and its Arrow methods throw.
The `spark.sql.timestampNanosTypes.enabled` feature flag continues to be enforced on the server path via `TypeUtils.failUnsupportedDataType` (the same way `TimeType` is gated), so no extra guard is added in connect-common.
### Why are the changes needed?
The proto sub-task (SPARK-57160) added the wire surface only. Without these converters, schema responses, casts, UDF input/output types, and literal expressions that contain nanosecond timestamps fail in Connect with `CONNECT_INVALID_PLAN.DATA_TYPE_UNSUPPORTED_*`. These converters unblock both the client and the server. Sub-task of SPARK-56822.
### Does this PR introduce _any_ user-facing change?
No. The types remain gated behind `spark.sql.timestampNanosTypes.enabled` (default off in production).
### How was this patch tested?
New tests in `LiteralExpressionProtoConverterSuite` cover DataType round-trips across precisions 7-9, NTZ and LTZ literal round-trips (boundary, pre-epoch, and sub-microsecond values), the two-component literal encoding, and rejection when the feature flag is disabled.
- `connect/testOnly *LiteralExpressionProtoConverterSuite` -- 49 tests passed (5 new)
- `connect-client-jvm/testOnly *ColumnNodeToProtoConverterSuite` -- 18 tests passed
- scalastyle (main + test) -- 0 errors
### Was this patch authored or co-authored using generative AI tooling?
Co-authored-by: Claude Code (Claude Opus 4.8)
Closes #57699 from stevomitric/stevomitric/spark-57161-connect-nanos.
Authored-by: Stevo Mitric <st...@gmail.com>
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
(cherry picked from commit 83714e97647b1e8ab5b04a0275fead2486890f88)
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
---
.../types/ops/TimestampNanosTypeConnectOps.scala | 226 +++++++++++++++++++++
.../common/LiteralValueProtoConverter.scala | 8 +
.../connect/common/types/ops/ConnectTypeOps.scala | 14 +-
.../LiteralExpressionProtoConverterSuite.scala | 126 +++++++++++-
4 files changed, 371 insertions(+), 3 deletions(-)
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/types/ops/TimestampNanosTypeConnectOps.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/types/ops/TimestampNanosTypeConnectOps.scala
new file mode 100644
index 000000000000..b29e5da958fb
--- /dev/null
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/types/ops/TimestampNanosTypeConnectOps.scala
@@ -0,0 +1,226 @@
+/*
+ * 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.connect.client.arrow.types.ops
+
+import java.time.{Instant, LocalDateTime}
+
+import org.apache.arrow.vector.FieldVector
+
+import org.apache.spark.connect.proto
+import org.apache.spark.sql.catalyst.encoders.AgnosticEncoder
+import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{InstantNanosEncoder, LocalDateTimeNanosEncoder}
+import org.apache.spark.sql.catalyst.util.SparkDateTimeUtils
+import org.apache.spark.sql.connect.client.arrow.{ArrowDeserializers, ArrowSerializer, ArrowVectorReader}
+import org.apache.spark.sql.connect.common.InvalidPlanInput
+import org.apache.spark.sql.connect.common.types.ops.ConnectTypeOps
+import org.apache.spark.sql.types.{DataType, TimestampLTZNanosType, TimestampNTZNanosType}
+import org.apache.spark.unsafe.types.TimestampNanosVal
+
+/**
+ * Combined Connect operations shared by the nanosecond-capable timestamp types
+ * ([[TimestampNTZNanosType]] and [[TimestampLTZNanosType]], precision 7..9).
+ *
+ * Implements the proto DataType/Literal side of [[ConnectTypeOps]]. The physical value is
+ * [[org.apache.spark.unsafe.types.TimestampNanosVal]] (epoch micros + nanos within the micro),
+ * which the proto carries as `epoch_micros` + `nanos_within_micro` rather than a single int64 of
+ * nanoseconds, because nanoseconds-since-epoch cannot span the supported 0001..9999 year range.
+ * The two concrete subclasses differ only in their proto message arm, external java.time value
+ * ([[LocalDateTime]] for NTZ, [[Instant]] for LTZ) and the conversion helpers used.
+ *
+ * Arrow IPC serialization is out of scope for these types (SPARK-57161), so the ops is not
+ * registered in the Arrow dispatch of [[ConnectTypeOps]] and the Arrow methods below are never
+ * reached; they throw to make an accidental wiring obvious.
+ *
+ * Lives under the arrow.types.ops sub-package to co-locate with [[TimeTypeConnectOps]], the
+ * reference implementation this mirrors.
+ *
+ * @since 4.3.0
+ */
+private[connect] abstract class TimestampNanosTypeConnectOps extends ConnectTypeOps {
+
+ /**
+ * Rebuilds the physical value from the two proto components. `nanosWithinMicro` is an int32 on
+ * the wire, so its range is checked here before narrowing to `Short`: without the check
+ * `.toShort` would truncate an out-of-range value modulo 2^16 (e.g. 65536 -> 0) and slip it
+ * past the `[0, 999]` guard in `fromParts`, yielding a silently wrong value instead of a clear
+ * error.
+ */
+ protected def toTimestampNanosVal(
+ epochMicros: Long,
+ nanosWithinMicro: Int): TimestampNanosVal = {
+ if (nanosWithinMicro < 0 || nanosWithinMicro > TimestampNanosVal.MAX_NANOS_WITHIN_MICRO) {
+ throw InvalidPlanInput(
+ s"nanos_within_micro must be in [0, ${TimestampNanosVal.MAX_NANOS_WITHIN_MICRO}], got: " +
+ nanosWithinMicro)
+ }
+ TimestampNanosVal.fromParts(epochMicros, nanosWithinMicro.toShort)
+ }
+
+ // ==================== Arrow Serialization (unsupported) ====================
+
+ private def arrowUnsupported: Nothing =
+ throw new UnsupportedOperationException(
+ s"Arrow serialization is not supported for ${dataType.sql} over Spark Connect.")
+
+ override def createArrowSerializer(vector: AnyRef): ArrowSerializer.Serializer =
+ arrowUnsupported
+
+ override def createArrowDeserializer(
+ enc: AgnosticEncoder[_],
+ data: AnyRef,
+ timeZoneId: String): ArrowDeserializers.Deserializer[Any] = arrowUnsupported
+
+ override def createArrowVectorReader(vector: FieldVector): ArrowVectorReader = arrowUnsupported
+}
+
+/**
+ * Connect operations for [[TimestampNTZNanosType]]. The external java.time value is
+ * [[LocalDateTime]] (interpreted at UTC), matching the server-side TypeOps and RowEncoder.
+ *
+ * @param t
+ * The TimestampNTZNanosType with precision information
+ * @since 4.3.0
+ */
+private[connect] class TimestampNTZNanosTypeConnectOps(val t: TimestampNTZNanosType)
+ extends TimestampNanosTypeConnectOps {
+
+ override def dataType: DataType = t
+
+ override def encoder: AgnosticEncoder[_] = LocalDateTimeNanosEncoder(t.precision)
+
+ // ==================== Proto Conversions ====================
+
+ override def toCatalystTypeFromProto(t: proto.DataType): DataType = {
+ val nanos = t.getTimestampNtzNanos
+ if (nanos.hasPrecision) TimestampNTZNanosType(nanos.getPrecision) else TimestampNTZNanosType()
+ }
+
+ override def toConnectProtoType: proto.DataType = {
+ proto.DataType
+ .newBuilder()
+ .setTimestampNtzNanos(
+ proto.DataType.TimestampNTZNanos.newBuilder().setPrecision(t.precision).build())
+ .build()
+ }
+
+ override def toLiteralProto(
+ value: Any,
+ builder: proto.Expression.Literal.Builder): proto.Expression.Literal.Builder =
+ setLiteral(value, TimestampNTZNanosType.DEFAULT_PRECISION, builder)
+
+ override def toLiteralProtoWithType(
+ value: Any,
+ dt: DataType,
+ builder: proto.Expression.Literal.Builder): proto.Expression.Literal.Builder =
+ setLiteral(value, dt.asInstanceOf[TimestampNTZNanosType].precision, builder)
+
+ private def setLiteral(
+ value: Any,
+ precision: Int,
+ builder: proto.Expression.Literal.Builder): proto.Expression.Literal.Builder = {
+ val v = SparkDateTimeUtils
+ .localDateTimeToTimestampNanos(value.asInstanceOf[LocalDateTime], precision)
+ builder.setTimestampNtzNanos(
+ builder.getTimestampNtzNanosBuilder
+ .setEpochMicros(v.epochMicros)
+ .setNanosWithinMicro(v.nanosWithinMicro.toInt)
+ .setPrecision(precision))
+ }
+
+ override def getScalaConverter: proto.Expression.Literal => Any = { v =>
+ val nanos = v.getTimestampNtzNanos
+ SparkDateTimeUtils.timestampNanosToLocalDateTime(
+ toTimestampNanosVal(nanos.getEpochMicros, nanos.getNanosWithinMicro))
+ }
+
+ override def getProtoDataTypeFromLiteral(literal: proto.Expression.Literal): proto.DataType = {
+ val typeBuilder = proto.DataType.TimestampNTZNanos.newBuilder()
+ if (literal.getTimestampNtzNanos.hasPrecision) {
+ typeBuilder.setPrecision(literal.getTimestampNtzNanos.getPrecision)
+ }
+ proto.DataType.newBuilder().setTimestampNtzNanos(typeBuilder.build()).build()
+ }
+}
+
+/**
+ * Connect operations for [[TimestampLTZNanosType]]. The external java.time value is [[Instant]],
+ * matching the server-side TypeOps and RowEncoder.
+ *
+ * @param t
+ * The TimestampLTZNanosType with precision information
+ * @since 4.3.0
+ */
+private[connect] class TimestampLTZNanosTypeConnectOps(val t: TimestampLTZNanosType)
+ extends TimestampNanosTypeConnectOps {
+
+ override def dataType: DataType = t
+
+ override def encoder: AgnosticEncoder[_] = InstantNanosEncoder(t.precision)
+
+ // ==================== Proto Conversions ====================
+
+ override def toCatalystTypeFromProto(t: proto.DataType): DataType = {
+ val nanos = t.getTimestampLtzNanos
+ if (nanos.hasPrecision) TimestampLTZNanosType(nanos.getPrecision) else TimestampLTZNanosType()
+ }
+
+ override def toConnectProtoType: proto.DataType = {
+ proto.DataType
+ .newBuilder()
+ .setTimestampLtzNanos(
+ proto.DataType.TimestampLTZNanos.newBuilder().setPrecision(t.precision).build())
+ .build()
+ }
+
+ override def toLiteralProto(
+ value: Any,
+ builder: proto.Expression.Literal.Builder): proto.Expression.Literal.Builder =
+ setLiteral(value, TimestampLTZNanosType.DEFAULT_PRECISION, builder)
+
+ override def toLiteralProtoWithType(
+ value: Any,
+ dt: DataType,
+ builder: proto.Expression.Literal.Builder): proto.Expression.Literal.Builder =
+ setLiteral(value, dt.asInstanceOf[TimestampLTZNanosType].precision, builder)
+
+ private def setLiteral(
+ value: Any,
+ precision: Int,
+ builder: proto.Expression.Literal.Builder): proto.Expression.Literal.Builder = {
+ val v = SparkDateTimeUtils.instantToTimestampNanos(value.asInstanceOf[Instant], precision)
+ builder.setTimestampLtzNanos(
+ builder.getTimestampLtzNanosBuilder
+ .setEpochMicros(v.epochMicros)
+ .setNanosWithinMicro(v.nanosWithinMicro.toInt)
+ .setPrecision(precision))
+ }
+
+ override def getScalaConverter: proto.Expression.Literal => Any = { v =>
+ val nanos = v.getTimestampLtzNanos
+ SparkDateTimeUtils.timestampNanosToInstant(
+ toTimestampNanosVal(nanos.getEpochMicros, nanos.getNanosWithinMicro))
+ }
+
+ override def getProtoDataTypeFromLiteral(literal: proto.Expression.Literal): proto.DataType = {
+ val typeBuilder = proto.DataType.TimestampLTZNanos.newBuilder()
+ if (literal.getTimestampLtzNanos.hasPrecision) {
+ typeBuilder.setPrecision(literal.getTimestampLtzNanos.getPrecision)
+ }
+ proto.DataType.newBuilder().setTimestampLtzNanos(typeBuilder.build()).build()
+ }
+}
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/LiteralValueProtoConverter.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/LiteralValueProtoConverter.scala
index 33da07cc1b5b..43995df77d72 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/LiteralValueProtoConverter.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/LiteralValueProtoConverter.scala
@@ -497,6 +497,14 @@ object LiteralValueProtoConverter {
true
case (proto.Expression.Literal.LiteralTypeCase.TIME, proto.DataType.KindCase.TIME) =>
true
+ case (
+ proto.Expression.Literal.LiteralTypeCase.TIMESTAMP_NTZ_NANOS,
+ proto.DataType.KindCase.TIMESTAMP_NTZ_NANOS) =>
+ true
+ case (
+ proto.Expression.Literal.LiteralTypeCase.TIMESTAMP_LTZ_NANOS,
+ proto.DataType.KindCase.TIMESTAMP_LTZ_NANOS) =>
+ true
case (proto.Expression.Literal.LiteralTypeCase.ARRAY, proto.DataType.KindCase.ARRAY) =>
true
case (proto.Expression.Literal.LiteralTypeCase.MAP, proto.DataType.KindCase.MAP) =>
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/types/ops/ConnectTypeOps.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/types/ops/ConnectTypeOps.scala
index 7225bc05feef..84f0c5012b59 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/types/ops/ConnectTypeOps.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/types/ops/ConnectTypeOps.scala
@@ -23,8 +23,8 @@ import org.apache.spark.connect.proto
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoder
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.LocalTimeEncoder
import org.apache.spark.sql.connect.client.arrow.{ArrowDeserializers, ArrowSerializer, ArrowVectorReader}
-import org.apache.spark.sql.connect.client.arrow.types.ops.TimeTypeConnectOps
-import org.apache.spark.sql.types.{DataType, TimeType}
+import org.apache.spark.sql.connect.client.arrow.types.ops.{TimestampLTZNanosTypeConnectOps, TimestampNTZNanosTypeConnectOps, TimeTypeConnectOps}
+import org.apache.spark.sql.types.{DataType, TimestampLTZNanosType, TimestampNTZNanosType, TimeType}
/**
* Optional type operations for Spark Connect infrastructure.
@@ -97,6 +97,8 @@ object ConnectTypeOps {
/** DataType-keyed dispatch for proto conversions. */
def apply(dt: DataType): Option[ConnectTypeOps] = dt match {
case tt: TimeType => Some(new TimeTypeConnectOps(tt))
+ case t: TimestampNTZNanosType => Some(new TimestampNTZNanosTypeConnectOps(t))
+ case t: TimestampLTZNanosType => Some(new TimestampLTZNanosTypeConnectOps(t))
// Add new framework types here
case _ => None
}
@@ -119,6 +121,10 @@ object ConnectTypeOps {
private def opsForKindCase(kindCase: proto.DataType.KindCase): Option[ConnectTypeOps] =
kindCase match {
case proto.DataType.KindCase.TIME => Some(new TimeTypeConnectOps(TimeType()))
+ case proto.DataType.KindCase.TIMESTAMP_NTZ_NANOS =>
+ Some(new TimestampNTZNanosTypeConnectOps(TimestampNTZNanosType()))
+ case proto.DataType.KindCase.TIMESTAMP_LTZ_NANOS =>
+ Some(new TimestampLTZNanosTypeConnectOps(TimestampLTZNanosType()))
// Add new framework proto kinds here - single registration for all KindCase lookups
case _ => None
}
@@ -142,6 +148,10 @@ object ConnectTypeOps {
litCase: proto.Expression.Literal.LiteralTypeCase): proto.DataType.KindCase =
litCase match {
case proto.Expression.Literal.LiteralTypeCase.TIME => proto.DataType.KindCase.TIME
+ case proto.Expression.Literal.LiteralTypeCase.TIMESTAMP_NTZ_NANOS =>
+ proto.DataType.KindCase.TIMESTAMP_NTZ_NANOS
+ case proto.Expression.Literal.LiteralTypeCase.TIMESTAMP_LTZ_NANOS =>
+ proto.DataType.KindCase.TIMESTAMP_LTZ_NANOS
// Add new framework literal-to-kind mappings here
case _ => proto.DataType.KindCase.KIND_NOT_SET
}
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/LiteralExpressionProtoConverterSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/LiteralExpressionProtoConverterSuite.scala
index 9c176a7a54f2..242c241d2b7e 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/LiteralExpressionProtoConverterSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/LiteralExpressionProtoConverterSuite.scala
@@ -17,17 +17,20 @@
package org.apache.spark.sql.connect.planner
-import java.time.LocalTime
+import java.time.{Instant, LocalDateTime, LocalTime}
import org.scalatest.funsuite.AnyFunSuite // scalastyle:ignore funsuite
+import org.apache.spark.SparkException
import org.apache.spark.connect.proto
import org.apache.spark.sql.catalyst.{expressions, CatalystTypeConverters}
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
+import org.apache.spark.sql.connect.common.DataTypeProtoConverter
import org.apache.spark.sql.connect.common.InvalidPlanInput
import org.apache.spark.sql.connect.common.LiteralValueProtoConverter
import org.apache.spark.sql.connect.common.LiteralValueProtoConverter.ToLiteralProtoOptions
import org.apache.spark.sql.connect.planner.LiteralExpressionProtoConverter
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
class LiteralExpressionProtoConverterSuite extends AnyFunSuite { // scalastyle:ignore funsuite
@@ -77,6 +80,127 @@ class LiteralExpressionProtoConverterSuite extends AnyFunSuite { // scalastyle:i
assertResult(LocalTime.of(1, 2, 3))(LiteralValueProtoConverter.toScalaValue(literalProto))
}
+ test("SPARK-57161: nanosecond timestamp DataType proto round-trip across precisions") {
+ for (precision <-
+ TimestampNTZNanosType.MIN_PRECISION to TimestampNTZNanosType.MAX_PRECISION) {
+ for (dt <- Seq(TimestampNTZNanosType(precision), TimestampLTZNanosType(precision))) {
+ val protoType = DataTypeProtoConverter.toConnectProtoType(dt)
+ assertResult(dt)(DataTypeProtoConverter.toCatalystType(protoType))
+ }
+ }
+ }
+
+ test("SPARK-57161: TIMESTAMP_NTZ nanosecond literal proto and catalyst value round-trip") {
+ // Boundary and pre-epoch values, plus a sub-microsecond value that exercises the extra nanos.
+ val values = Seq(
+ LocalDateTime.of(1, 1, 1, 0, 0, 0, 0),
+ LocalDateTime.of(1969, 12, 31, 23, 59, 59, 999999999),
+ LocalDateTime.of(1970, 1, 1, 0, 0, 0, 123456789),
+ LocalDateTime.of(2023, 6, 15, 12, 34, 56, 987654321),
+ LocalDateTime.of(9999, 12, 31, 23, 59, 59, 999999999))
+ for (precision <- TimestampNTZNanosType.MIN_PRECISION to TimestampNTZNanosType.MAX_PRECISION;
+ v <- values) {
+ val t = TimestampNTZNanosType(precision)
+ val literalProto = toLiteralProto(v, t)
+ // The literal carries the nanos proto arm with the expected precision.
+ assert(literalProto.getTimestampNtzNanos.getPrecision == precision)
+ // Scala value -> proto -> Catalyst value equals converting the Scala value directly, so the
+ // same sub-microsecond truncation to `precision` is applied on both paths.
+ val convert = CatalystTypeConverters.createToCatalystConverter(t)
+ val expected = expressions.Literal(convert(v), t)
+ assertResult(expected)(LiteralExpressionProtoConverter.toCatalystExpression(literalProto))
+ }
+ }
+
+ test("SPARK-57161: TIMESTAMP_LTZ nanosecond literal proto and catalyst value round-trip") {
+ val values = Seq(
+ Instant.parse("0001-01-01T00:00:00Z"),
+ Instant.parse("1969-12-31T23:59:59.999999999Z"),
+ Instant.parse("1970-01-01T00:00:00.123456789Z"),
+ Instant.parse("2023-06-15T12:34:56.987654321Z"),
+ Instant.parse("9999-12-31T23:59:59.999999999Z"))
+ for (precision <- TimestampLTZNanosType.MIN_PRECISION to TimestampLTZNanosType.MAX_PRECISION;
+ v <- values) {
+ val t = TimestampLTZNanosType(precision)
+ val literalProto = toLiteralProto(v, t)
+ assert(literalProto.getTimestampLtzNanos.getPrecision == precision)
+ val convert = CatalystTypeConverters.createToCatalystConverter(t)
+ val expected = expressions.Literal(convert(v), t)
+ assertResult(expected)(LiteralExpressionProtoConverter.toCatalystExpression(literalProto))
+ }
+ }
+
+ test("SPARK-57161: nanosecond timestamp literal proto carries epoch micros + extra nanos") {
+ // 1970-01-01T00:00:00.000001500Z is 1 microsecond and 500 extra nanoseconds past the epoch.
+ val ntzProto = toLiteralProto(
+ LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1500),
+ TimestampNTZNanosType(TimestampNTZNanosType.NANOS_PRECISION))
+ assert(ntzProto.getTimestampNtzNanos.getEpochMicros == 1L)
+ assert(ntzProto.getTimestampNtzNanos.getNanosWithinMicro == 500)
+
+ val ltzProto = toLiteralProto(
+ Instant.parse("1970-01-01T00:00:00.000001500Z"),
+ TimestampLTZNanosType(TimestampLTZNanosType.NANOS_PRECISION))
+ assert(ltzProto.getTimestampLtzNanos.getEpochMicros == 1L)
+ assert(ltzProto.getTimestampLtzNanos.getNanosWithinMicro == 500)
+ }
+
+ test(
+ "SPARK-57161: nanosecond timestamp literal with out-of-range nanos_within_micro is " +
+ "rejected") {
+ // nanos_within_micro is an int32 on the wire but must be in [0, 999]. Build literals directly
+ // with out-of-range values, including 65536 which would truncate to 0 if narrowed to Short
+ // before validation, and confirm the read path rejects them instead of wrapping.
+ for (nanos <- Seq(1000, 65536, 65636, -1)) {
+ val ntzProto = proto.Expression.Literal
+ .newBuilder()
+ .setTimestampNtzNanos(
+ proto.Expression.Literal.TimestampNTZNanos
+ .newBuilder()
+ .setEpochMicros(0L)
+ .setNanosWithinMicro(nanos)
+ .setPrecision(TimestampNTZNanosType.NANOS_PRECISION))
+ .build()
+ val ltzProto = proto.Expression.Literal
+ .newBuilder()
+ .setTimestampLtzNanos(
+ proto.Expression.Literal.TimestampLTZNanos
+ .newBuilder()
+ .setEpochMicros(0L)
+ .setNanosWithinMicro(nanos)
+ .setPrecision(TimestampLTZNanosType.NANOS_PRECISION))
+ .build()
+ for (literalProto <- Seq(ntzProto, ltzProto)) {
+ val e = intercept[InvalidPlanInput] {
+ LiteralValueProtoConverter.toScalaValue(literalProto)
+ }
+ assert(e.getMessage.contains("nanos_within_micro"))
+ }
+ }
+ }
+
+ test("SPARK-57161: nanosecond timestamp literals are rejected when the feature is disabled") {
+ // Build the proto with the feature enabled (default in tests), mirroring a message arriving
+ // over the wire, then convert it on the server with the feature turned off.
+ val ntzProto = toLiteralProto(
+ LocalDateTime.of(2023, 1, 1, 0, 0, 0, 0),
+ TimestampNTZNanosType(TimestampNTZNanosType.NANOS_PRECISION))
+ val ltzProto = toLiteralProto(
+ Instant.parse("2023-01-01T00:00:00Z"),
+ TimestampLTZNanosType(TimestampLTZNanosType.NANOS_PRECISION))
+
+ val disabledConf = new SQLConf()
+ disabledConf.setConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED, false)
+ SQLConf.withExistingConf(disabledConf) {
+ for (literalProto <- Seq(ntzProto, ltzProto)) {
+ val e = intercept[SparkException] {
+ LiteralExpressionProtoConverter.toCatalystExpression(literalProto)
+ }
+ assert(e.getCondition == "FEATURE_NOT_ENABLED")
+ }
+ }
+ }
+
// The goal of this test is to check that converting a Scala value -> Proto -> Catalyst value
// is equivalent to converting a Scala value directly to a Catalyst value.
Seq[(Any, DataType)](
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org