This is an automated email from the ASF dual-hosted git repository.
viirya 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 5220ac3ceebc [SPARK-57975][SQL] Add an opt-in lossless Arrow struct
representation for nanosecond timestamps
5220ac3ceebc is described below
commit 5220ac3ceebc1d0318bf329eb91160edd6951df8
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Tue Jul 7 09:47:57 2026 -0700
[SPARK-57975][SQL] Add an opt-in lossless Arrow struct representation for
nanosecond timestamps
### What changes were proposed in this pull request?
This adds an opt-in Arrow mapping for the nanosecond timestamp types
(`TimestampNTZNanosType` / `TimestampLTZNanosType`), selected by a new
`losslessTimestampNanos` parameter on `ArrowUtils.toArrowSchema` /
`toArrowField` (default `false`). When enabled, a nanosecond timestamp column
maps to an Arrow struct of `(epochMicros: int64, nanosWithinMicro: int16)` --
`TimestampNanosVal`'s own layout -- instead of the default single int64 of
epoch-nanoseconds:
- **Schema (`ArrowUtils`)**: the struct's `epochMicros` child is tagged
through field metadata with the NTZ/LTZ kind and the column precision
(following the geometry/variant struct tag pattern), so `fromArrowField`
recovers the exact Spark type on read with no out-of-band information. Nested
occurrences (array/struct/map/UDT sqlType) are covered by threading the flag
through the recursive schema construction.
- **Write (`ArrowWriter`)**: new `TimestampNTZNanosStructWriter` /
`TimestampLTZNanosStructWriter` store the two components as-is -- no unit
conversion, hence no overflow. `TimestampNanosTypeOps.createArrowFieldWriter`
now dispatches on the vector shape instead of unconditionally casting to the
native nanos vectors.
- **Read (`ArrowColumnVector`)**: a dedicated
`TimestampNanosStructAccessor` recognizes the tagged struct and serves
`getTimestampNTZNanos` / `getTimestampLTZNanos` from the child vectors,
including nested inside arrays, structs, and maps.
The default `Timestamp(NANOSECOND)` mapping and every existing caller are
unchanged.
### Why are the changes needed?
Spark defines the nanosecond timestamp types over years 0001-9999, and
stores values losslessly as `(epochMicros, nanosWithinMicro)`. The standard
Arrow mapping packs the value into a single int64 of epoch-nanoseconds, which
only covers roughly years 1677-2262: a common sentinel value like `9999-12-31
23:59:59.999999999` fails with `DATETIME_OVERFLOW`. Internal Arrow-based
storage -- specifically the Arrow-backed Dataset cache proposed in #56334,
where the default in-memory cache hand [...]
**Why an opt-in parameter instead of changing the default mapping?** The
mismatch is structural, so the two representations serve two permanently
distinct needs:
- **Interchange paths must keep the standard int64 encoding.**
`toPandas()`, Arrow UDFs, and Connect result sets hand the produced bytes
directly to external consumers (pandas, PyArrow, arrow-rs clients) that only
understand the standard `Timestamp(NANOSECOND)` encoding -- SPARK-57159 added
that mapping precisely so pandas receives real timestamps. Moreover, those
consumers' own timestamp domains are equally int64-bound (pandas
`datetime64[ns]` is itself int64 epoch-nanos), so the red [...]
- **Internal storage is a closed write-then-read-back loop** with no
external consumer, where the only requirement is fidelity to Spark semantics --
hence the lossless struct.
Since Arrow's timestamp physical type is fixed at int64 by the Arrow format
spec and Spark's type domain will not shrink, this is not a transitional state
to be unified later. The per-call-site boolean follows the existing
`largeVarTypes` pattern (one Spark type, two Arrow encodings, chosen by the
consumer's needs), and only schema construction needs the flag: the struct is
self-describing through its child-field metadata tag, so `fromArrowField`,
`ArrowWriter`, and `ArrowColumnVector [...]
### Does this PR introduce _any_ user-facing change?
No. The new mapping is opt-in via an internal API parameter that defaults
to off; no existing behavior changes.
### How was this patch tested?
New tests:
- `ArrowUtilsSuite` "timestamp nanos lossless struct": schema shape (struct
of int64 + int16, non-null children), type/precision round-trip for NTZ/LTZ at
p=7/8/9, LTZ requiring no time zone, nested array/struct/map coverage,
user-metadata preservation, precision fallback for a missing/invalid tag, no
misfire on an untagged struct with the same child names, and the default
mapping staying unchanged.
- `ArrowWriterSuite` "timestamp nanos lossless struct round-trip covers the
full value domain": write-and-read-back through `ArrowWriter` +
`ArrowColumnVector` for values including `9999-12-31T23:59:59.999999999` and
`0001-01-01T00:00:00.000000001` (both far outside the int64 epoch-nanos range)
plus nulls, for NTZ/LTZ at p=9 and p=7.
- `ArrowWriterSuite` "timestamp nanos lossless struct round-trip inside
nested types": the same extreme values inside `array<...>`, `struct<...>`, and
`map<int, ...>`.
Existing regression suites pass: `ArrowUtilsSuite`, `ArrowWriterSuite`,
`ArrowConvertersSuite`, `ColumnVectorSuite`, `ColumnarBatchSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
This pull request and its description were written by Claude Code.
Closes #57053 from viirya/nanos-arrow-lossless.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
(cherry picked from commit 7dc70c1a0db68a1f7ebfadf65763fd8726d07982)
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../org/apache/spark/sql/util/ArrowUtils.scala | 139 +++++++++++++++++++--
.../spark/sql/vectorized/ArrowColumnVector.java | 40 +++++-
.../catalyst/types/ops/TimestampNanosTypeOps.scala | 15 ++-
.../spark/sql/execution/arrow/ArrowWriter.scala | 58 ++++++++-
.../apache/spark/sql/util/ArrowUtilsSuite.scala | 82 ++++++++++++
.../sql/execution/arrow/ArrowWriterSuite.scala | 104 +++++++++++++++
6 files changed, 423 insertions(+), 15 deletions(-)
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala
b/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala
index a06a77d9d113..a0a14099d619 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala
@@ -120,6 +120,12 @@ private[sql] object ArrowUtils {
// (namespaced like `metadataKey`, separate from the user metadata blob so
user metadata is
// untouched) and recovered on read in `fromArrowField`.
private val timePrecisionKey = "SPARK::time::precision"
+ // Marks the epochMicros child of the lossless struct representation of a
nanosecond timestamp
+ // (see `toArrowField` with `losslessTimestampNanos = true`). The value is
"ntz" or "ltz" and
+ // distinguishes TimestampNTZNanosType from TimestampLTZNanosType on read;
the precision is
+ // stored alongside under `timestampNanosPrecisionKey`. The tag lives on a
child field (like the
+ // geometry/variant struct tags) so it cannot collide with user metadata on
the struct itself.
+ private val timestampNanosStructKey = "SPARK::timestampNanos::struct"
private def toArrowMetaData(metadata: Metadata) = {
if (metadata != null && !metadata.isEmpty) {
Map(metadataKey -> metadata.json).asJava
@@ -156,14 +162,75 @@ private[sql] object ArrowUtils {
new Field(name, fieldType, Seq.empty[Field].asJava)
}
- /** Maps field from Spark to Arrow. NOTE: timeZoneId required for
TimestampType */
+ /**
+ * Builds the lossless Arrow struct representation of a nanosecond
timestamp: a struct of
+ * (epochMicros: int64, nanosWithinMicro: int16), mirroring
TimestampNanosVal's own layout with
+ * no unit conversion. Unlike the default Timestamp(NANOSECOND) mapping,
which packs the value
+ * into a single int64 of epoch-nanoseconds and therefore only covers
roughly years 1677-2262,
+ * this representation covers the full domain of the Spark types (years
0001-9999).
+ *
+ * Why two representations exist, permanently: the mismatch is structural.
Arrow's timestamp
+ * physical type is fixed at int64 by the Arrow format spec, while the Spark
types are defined
+ * over years 0001-9999, so no single Arrow timestamp encoding can serve
both goals.
+ * - Interchange paths (pandas conversion, Arrow UDFs, Connect result
sets) must keep the
+ * standard Timestamp(NANOSECOND) encoding: their consumers only
understand that encoding,
+ * and those consumers' own timestamp domains are equally int64-bound
(e.g. pandas
+ * datetime64[ns]), so the reduced domain is inherent to the destination
-- failing loudly
+ * at write (DATETIME_OVERFLOW) is the correct behavior there, not a
limitation of the
+ * mapping.
+ * - Internal storage (e.g. the Arrow-based Dataset cache) is a closed
write-then-read-back
+ * loop with no external consumer, where the only requirement is
fidelity to Spark
+ * semantics, hence this struct.
+ * The choice is made per call site via `losslessTimestampNanos` on
`toArrowSchema` /
+ * `toArrowField` (the same pattern as `largeVarTypes`: one Spark type, two
Arrow encodings,
+ * selected by the consumer's needs). Only schema construction needs the
flag: the struct is
+ * self-describing through its child-field tag, so `fromArrowField`,
`ArrowWriter`, and
+ * `ArrowColumnVector` recognize both shapes unconditionally and no mode
mismatch is possible.
+ */
+ private def toTimestampNanosStructField(
+ name: String,
+ isNtz: Boolean,
+ precision: Int,
+ nullable: Boolean,
+ metadata: Metadata): Field = {
+ val fieldType =
+ new FieldType(nullable, ArrowType.Struct.INSTANCE, null,
toArrowMetaData(metadata))
+ // Tag the epochMicros child so `fromArrowField` (and ArrowColumnVector)
can recognize that
+ // this struct represents a nanosecond timestamp, following the
geometry/variant tag pattern.
+ val microsFieldType = new FieldType(
+ false,
+ new ArrowType.Int(8 * 8, true),
+ null,
+ Map(
+ timestampNanosStructKey -> (if (isNtz) "ntz" else "ltz"),
+ timestampNanosPrecisionKey -> precision.toString).asJava)
+ val nanosFieldType = new FieldType(false, new ArrowType.Int(8 * 2, true),
null, null)
+ new Field(
+ name,
+ fieldType,
+ Seq(
+ new Field("epochMicros", microsFieldType, Seq.empty[Field].asJava),
+ new Field("nanosWithinMicro", nanosFieldType,
Seq.empty[Field].asJava)).asJava)
+ }
+
+ /**
+ * Maps field from Spark to Arrow. NOTE: timeZoneId required for
TimestampType
+ *
+ * @param losslessTimestampNanos
+ * when true, nanosecond timestamps map to the lossless struct
representation covering their
+ * full value domain instead of the standard int64 Timestamp(NANOSECOND)
encoding. Only
+ * internal-storage callers with no external Arrow consumer (e.g. the
Arrow-based Dataset
+ * cache) should pass true; interchange paths must keep the default. See
+ * `toTimestampNanosStructField` for the full rationale.
+ */
def toArrowField(
name: String,
dt: DataType,
nullable: Boolean,
timeZoneId: String,
largeVarTypes: Boolean = false,
- metadata: Metadata = Metadata.empty): Field = {
+ metadata: Metadata = Metadata.empty,
+ losslessTimestampNanos: Boolean = false): Field = {
dt match {
case ArrayType(elementType, containsNull) =>
val fieldType =
@@ -172,7 +239,14 @@ private[sql] object ArrowUtils {
name,
fieldType,
Seq(
- toArrowField("element", elementType, containsNull, timeZoneId,
largeVarTypes)).asJava)
+ toArrowField(
+ "element",
+ elementType,
+ containsNull,
+ timeZoneId,
+ largeVarTypes,
+ Metadata.empty,
+ losslessTimestampNanos)).asJava)
case StructType(fields) =>
val fieldType =
new FieldType(nullable, ArrowType.Struct.INSTANCE, null,
toArrowMetaData(metadata))
@@ -187,7 +261,8 @@ private[sql] object ArrowUtils {
field.nullable,
timeZoneId,
largeVarTypes,
- field.metadata)
+ field.metadata,
+ losslessTimestampNanos)
}
.toImmutableArraySeq
.asJava)
@@ -206,9 +281,18 @@ private[sql] object ArrowUtils {
.add(MapVector.VALUE_NAME, valueType, nullable =
valueContainsNull),
nullable = false,
timeZoneId,
- largeVarTypes)).asJava)
+ largeVarTypes,
+ Metadata.empty,
+ losslessTimestampNanos)).asJava)
case udt: UserDefinedType[_] =>
- toArrowField(name, udt.sqlType, nullable, timeZoneId, largeVarTypes,
metadata)
+ toArrowField(
+ name,
+ udt.sqlType,
+ nullable,
+ timeZoneId,
+ largeVarTypes,
+ metadata,
+ losslessTimestampNanos)
case g: GeometryType =>
val fieldType =
new FieldType(nullable, ArrowType.Struct.INSTANCE, null,
toArrowMetaData(metadata))
@@ -262,6 +346,10 @@ private[sql] object ArrowUtils {
Seq(
toArrowField("value", BinaryType, false, timeZoneId,
largeVarTypes),
new Field("metadata", metadataFieldType,
Seq.empty[Field].asJava)).asJava)
+ case t: TimestampNTZNanosType if losslessTimestampNanos =>
+ toTimestampNanosStructField(name, isNtz = true, t.precision, nullable,
metadata)
+ case t: TimestampLTZNanosType if losslessTimestampNanos =>
+ toTimestampNanosStructField(name, isNtz = false, t.precision,
nullable, metadata)
case t: TimestampNTZNanosType =>
toPrecisionTaggedArrowField(
name,
@@ -332,6 +420,23 @@ private[sql] object ArrowUtils {
}
}
+ /**
+ * Whether the Arrow struct field is the lossless representation of a
nanosecond timestamp built
+ * by `toArrowField` with `losslessTimestampNanos = true`. Also callable
from Java
+ * (ArrowColumnVector) to select the timestamp accessor for such structs.
+ */
+ def isTimestampNanosStructField(field: Field): Boolean = {
+ field.getType.isInstanceOf[ArrowType.Struct] &&
+ field.getChildren.asScala
+ .map(_.getName)
+ .asJava
+ .containsAll(Seq("epochMicros", "nanosWithinMicro").asJava) &&
+ field.getChildren.asScala.exists { child =>
+ child.getName == "epochMicros" &&
+ Set("ntz",
"ltz").contains(child.getMetadata.getOrDefault(timestampNanosStructKey, ""))
+ }
+ }
+
def fromArrowField(field: Field): DataType = {
field.getType match {
case _: ArrowType.Map =>
@@ -343,6 +448,18 @@ private[sql] object ArrowUtils {
val elementField = field.getChildren().get(0)
val elementType = fromArrowField(elementField)
ArrayType(elementType, containsNull = elementField.isNullable)
+ case ArrowType.Struct.INSTANCE if isTimestampNanosStructField(field) =>
+ val microsChild = field.getChildren.asScala.find(_.getName ==
"epochMicros").get
+ val isNtz = microsChild.getMetadata.get(timestampNanosStructKey) ==
"ntz"
+ // Recover the precision like the Timestamp(NANOSECOND) case below: a
missing or invalid
+ // precision key falls back to the canonical maximum precision.
+ val precision =
Option(microsChild.getMetadata.get(timestampNanosPrecisionKey))
+ .flatMap(s => scala.util.Try(s.toInt).toOption)
+ .filter { p =>
+ p >= TimestampNTZNanosType.MIN_PRECISION && p <=
TimestampNTZNanosType.MAX_PRECISION
+ }
+ .getOrElse(TimestampNTZNanosType.MAX_PRECISION)
+ if (isNtz) TimestampNTZNanosType(precision) else
TimestampLTZNanosType(precision)
case ArrowType.Struct.INSTANCE if isVariantField(field) =>
VariantType
case ArrowType.Struct.INSTANCE if isGeometryField(field) =>
@@ -401,12 +518,17 @@ private[sql] object ArrowUtils {
/**
* Maps schema from Spark to Arrow. NOTE: timeZoneId required for
TimestampType in StructType
+ *
+ * @param losslessTimestampNanos
+ * see `toArrowField`: opt-in full-domain struct encoding of nanosecond
timestamps for
+ * internal storage; interchange paths must keep the default.
*/
def toArrowSchema(
schema: StructType,
timeZoneId: String,
errorOnDuplicatedFieldNames: Boolean,
- largeVarTypes: Boolean): Schema = {
+ largeVarTypes: Boolean,
+ losslessTimestampNanos: Boolean = false): Schema = {
new Schema(schema.map { field =>
toArrowField(
field.name,
@@ -414,7 +536,8 @@ private[sql] object ArrowUtils {
field.nullable,
timeZoneId,
largeVarTypes,
- field.metadata)
+ field.metadata,
+ losslessTimestampNanos)
}.asJava)
}
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ArrowColumnVector.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ArrowColumnVector.java
index 3267daea6bcc..1ae653d6b725 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ArrowColumnVector.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ArrowColumnVector.java
@@ -227,11 +227,17 @@ public class ArrowColumnVector extends ColumnVector {
} else if (vector instanceof ListVector listVector) {
accessor = new ArrayAccessor(listVector);
} else if (vector instanceof StructVector structVector) {
- accessor = new StructAccessor(structVector);
+ if (ArrowUtils.isTimestampNanosStructField(structVector.getField())) {
+ // Lossless struct representation of a nanosecond timestamp
(ArrowUtils.toArrowField with
+ // losslessTimestampNanos = true): logically a scalar, so no child
columns are exposed.
+ accessor = new TimestampNanosStructAccessor(structVector);
+ } else {
+ accessor = new StructAccessor(structVector);
- childColumns = new ArrowColumnVector[structVector.size()];
- for (int i = 0; i < childColumns.length; ++i) {
- childColumns[i] = new ArrowColumnVector(structVector.getVectorById(i));
+ childColumns = new ArrowColumnVector[structVector.size()];
+ for (int i = 0; i < childColumns.length; ++i) {
+ childColumns[i] = new
ArrowColumnVector(structVector.getVectorById(i));
+ }
}
} else if (vector instanceof NullVector nullVector) {
accessor = new NullAccessor(nullVector);
@@ -619,6 +625,32 @@ public class ArrowColumnVector extends ColumnVector {
}
}
+ /**
+ * Reads the lossless struct representation of a nanosecond timestamp
(epochMicros: int64,
+ * nanosWithinMicro: int16), built by ArrowUtils.toArrowField with
losslessTimestampNanos = true.
+ * The components are stored as-is (TimestampNanosVal's own layout), so
unlike the int64
+ * epoch-nanoseconds accessors above there is no decoding arithmetic and no
reduced value domain.
+ */
+ static class TimestampNanosStructAccessor extends ArrowVectorAccessor {
+
+ private final BigIntVector epochMicros;
+ private final SmallIntVector nanosWithinMicro;
+
+ TimestampNanosStructAccessor(StructVector vector) {
+ super(vector);
+ this.epochMicros = (BigIntVector) vector.getChild("epochMicros");
+ this.nanosWithinMicro = (SmallIntVector)
vector.getChild("nanosWithinMicro");
+ }
+
+ @Override
+ final TimestampNanosVal getTimestampNanos(int rowId) {
+ // fromParts validates nanosWithinMicro is in [0, 999]; the write side
always stores a valid
+ // TimestampNanosVal, but this format may be deserialized from stored
bytes, so validate
+ // rather than trusting blindly.
+ return TimestampNanosVal.fromParts(epochMicros.get(rowId),
nanosWithinMicro.get(rowId));
+ }
+ }
+
static class ArrayAccessor extends ArrowVectorAccessor {
private final ListVector accessor;
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/ops/TimestampNanosTypeOps.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/ops/TimestampNanosTypeOps.scala
index 6ecebf9a3fe0..4697c85c660f 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/ops/TimestampNanosTypeOps.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/ops/TimestampNanosTypeOps.scala
@@ -126,7 +126,13 @@ case class TimestampNTZNanosTypeOps(override val t:
TimestampNTZNanosType)
returnNullable = false))
override def createArrowFieldWriter(vector: ValueVector):
Option[ArrowFieldWriter] =
- Some(new TimestampNTZNanosWriter(vector.asInstanceOf[TimeStampNanoVector]))
+ vector match {
+ case v: TimeStampNanoVector => Some(new TimestampNTZNanosWriter(v))
+ // The lossless struct representation (ArrowUtils.toArrowField with
+ // losslessTimestampNanos = true) is backed by a StructVector; its
writer needs the
+ // recursively-built child writers, so defer to ArrowWriter's default
matching.
+ case _ => None
+ }
}
/**
@@ -179,5 +185,10 @@ case class TimestampLTZNanosTypeOps(override val t:
TimestampLTZNanosType)
returnNullable = false))
override def createArrowFieldWriter(vector: ValueVector):
Option[ArrowFieldWriter] =
- Some(new
TimestampLTZNanosWriter(vector.asInstanceOf[TimeStampNanoTZVector]))
+ vector match {
+ case v: TimeStampNanoTZVector => Some(new TimestampLTZNanosWriter(v))
+ // See the NTZ counterpart above: the lossless StructVector shape is
handled by
+ // ArrowWriter's default matching.
+ case _ => None
+ }
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowWriter.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowWriter.scala
index 6030eee94a6c..47e5d10926c5 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowWriter.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowWriter.scala
@@ -23,12 +23,13 @@ import org.apache.arrow.vector._
import org.apache.arrow.vector.complex._
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
+import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow,
SpecializedGetters}
import org.apache.spark.sql.catalyst.types.ops.TypeOps
import org.apache.spark.sql.catalyst.util.{DateTimeUtils, STUtils}
import org.apache.spark.sql.errors.{ExecutionErrors, QueryExecutionErrors}
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.ArrowUtils
+import org.apache.spark.unsafe.types.TimestampNanosVal
object ArrowWriter {
@@ -95,6 +96,19 @@ object ArrowWriter {
case (_: DayTimeIntervalType, vector: DurationVector) => new
DurationWriter(vector)
case (CalendarIntervalType, vector: IntervalMonthDayNanoVector) =>
new IntervalMonthDayNanoWriter(vector)
+ // Lossless struct representation of nanosecond timestamps
(ArrowUtils.toArrowField with
+ // losslessTimestampNanos = true). The native TimeStampNano(TZ)Vector
writers are created by
+ // the TypeOps hook; only the struct-backed shape reaches this default
matching.
+ case (_: TimestampNTZNanosType, vector: StructVector) =>
+ val children = (0 until vector.size()).map { ordinal =>
+ createFieldWriter(vector.getChildByOrdinal(ordinal))
+ }
+ new TimestampNTZNanosStructWriter(vector, children.toArray)
+ case (_: TimestampLTZNanosType, vector: StructVector) =>
+ val children = (0 until vector.size()).map { ordinal =>
+ createFieldWriter(vector.getChildByOrdinal(ordinal))
+ }
+ new TimestampLTZNanosStructWriter(vector, children.toArray)
case (VariantType, vector: StructVector) =>
val children = (0 until vector.size()).map { ordinal =>
createFieldWriter(vector.getChildByOrdinal(ordinal))
@@ -540,6 +554,48 @@ private[arrow] class GeometryWriter(
}
}
+/**
+ * Writes a nanosecond timestamp into its lossless Arrow struct representation
+ * (epochMicros: int64, nanosWithinMicro: int16), built by
`ArrowUtils.toArrowField` with
+ * `losslessTimestampNanos = true`. The two components of TimestampNanosVal
are stored as-is with
+ * no unit conversion, so unlike the Timestamp(NANOSECOND) writers there is no
overflow: the full
+ * domain of the Spark types (years 0001-9999) round-trips.
+ */
+private[arrow] abstract class TimestampNanosStructWriter(
+ valueVector: StructVector,
+ children: Array[ArrowFieldWriter]) extends StructWriter(valueVector,
children) {
+
+ protected def getTimestampNanos(input: SpecializedGetters, ordinal: Int):
TimestampNanosVal
+
+ // Reused across rows; this writer is single-threaded like the vector it
wraps.
+ private val row = new GenericInternalRow(2)
+
+ override def setValue(input: SpecializedGetters, ordinal: Int): Unit = {
+ valueVector.setIndexDefined(count)
+ val v = getTimestampNanos(input, ordinal)
+ row.update(0, v.epochMicros)
+ row.update(1, v.nanosWithinMicro)
+ children(0).write(row, 0)
+ children(1).write(row, 1)
+ }
+}
+
+private[arrow] class TimestampNTZNanosStructWriter(
+ valueVector: StructVector,
+ children: Array[ArrowFieldWriter]) extends
TimestampNanosStructWriter(valueVector, children) {
+ override protected def getTimestampNanos(
+ input: SpecializedGetters, ordinal: Int): TimestampNanosVal =
+ input.getTimestampNTZNanos(ordinal)
+}
+
+private[arrow] class TimestampLTZNanosStructWriter(
+ valueVector: StructVector,
+ children: Array[ArrowFieldWriter]) extends
TimestampNanosStructWriter(valueVector, children) {
+ override protected def getTimestampNanos(
+ input: SpecializedGetters, ordinal: Int): TimestampNanosVal =
+ input.getTimestampLTZNanos(ordinal)
+}
+
private[arrow] class MapWriter(
val valueVector: MapVector,
val structVector: StructVector,
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/util/ArrowUtilsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/util/ArrowUtilsSuite.scala
index 2d2186aed85c..22611341cd37 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/util/ArrowUtilsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/util/ArrowUtilsSuite.scala
@@ -154,6 +154,88 @@ class ArrowUtilsSuite extends SparkFunSuite {
ArrowUtils.toArrowSchema(schemaWithMeta, null, true, false)) ===
schemaWithMeta)
}
+ test("timestamp nanos lossless struct") {
+ def losslessRoundtrip(schema: StructType, timeZoneId: String = null): Unit
= {
+ val arrowSchema =
+ ArrowUtils.toArrowSchema(schema, timeZoneId, true, false,
losslessTimestampNanos = true)
+ assert(ArrowUtils.fromArrowSchema(arrowSchema) === schema)
+ }
+
+ // Top-level: the lossless mapping is a struct of (epochMicros: int64,
nanosWithinMicro:
+ // int16); the NTZ/LTZ kind and the precision round-trip through the child
field metadata.
+ Seq(7, 8, 9).foreach { p =>
+ Seq[DataType](TimestampNTZNanosType(p),
TimestampLTZNanosType(p)).foreach { dt =>
+ val schema = new StructType().add("value", dt)
+ val arrowSchema =
+ ArrowUtils.toArrowSchema(schema, null, true, false,
losslessTimestampNanos = true)
+ val field = arrowSchema.findField("value")
+ assert(field.getType === ArrowType.Struct.INSTANCE)
+ val children = field.getChildren
+ assert(children.size() === 2)
+ assert(children.get(0).getName === "epochMicros")
+ assert(children.get(0).getType === new ArrowType.Int(64, true))
+ assert(!children.get(0).isNullable)
+ assert(children.get(1).getName === "nanosWithinMicro")
+ assert(children.get(1).getType === new ArrowType.Int(16, true))
+ assert(!children.get(1).isNullable)
+ assert(ArrowUtils.isTimestampNanosStructField(field))
+ assert(ArrowUtils.fromArrowSchema(arrowSchema) === schema)
+ }
+ }
+
+ // Unlike the default int64 mapping, LTZ needs no session time zone: the
struct stores the
+ // raw value components, which are zone-independent.
+ losslessRoundtrip(new StructType().add("value", TimestampLTZNanosType(9)))
+
+ // Nested: the flag must reach nanosecond timestamps inside arrays,
structs, and maps.
+ losslessRoundtrip(new StructType()
+ .add("arr", ArrayType(TimestampNTZNanosType(9)))
+ .add("struct", new StructType().add("ts", TimestampLTZNanosType(7)))
+ .add("map", MapType(IntegerType, TimestampNTZNanosType(8))))
+
+ // User metadata on the column is preserved alongside the struct tag.
+ val md = new MetadataBuilder().putString("city", "beijing").build()
+ losslessRoundtrip(new StructType().add("value", TimestampNTZNanosType(7),
true, md))
+
+ // An invalid or missing precision on the tagged child falls back to the
canonical maximum
+ // precision, mirroring the default int64 mapping's fallback.
+ def taggedStructField(precision: Option[String]): Field = {
+ val microsMd = new java.util.HashMap[String, String]()
+ microsMd.put("SPARK::timestampNanos::struct", "ntz")
+ precision.foreach(p => microsMd.put("SPARK::timestampNanos::precision",
p))
+ new Field(
+ "value",
+ new FieldType(true, ArrowType.Struct.INSTANCE, null, null),
+ java.util.Arrays.asList(
+ new Field(
+ "epochMicros",
+ new FieldType(false, new ArrowType.Int(64, true), null, microsMd),
+ java.util.Collections.emptyList[Field]()),
+ new Field(
+ "nanosWithinMicro",
+ new FieldType(false, new ArrowType.Int(16, true), null, null),
+ java.util.Collections.emptyList[Field]())))
+ }
+ assert(ArrowUtils.fromArrowField(taggedStructField(Some("5"))) ===
TimestampNTZNanosType(9))
+ assert(ArrowUtils.fromArrowField(taggedStructField(None)) ===
TimestampNTZNanosType(9))
+
+ // A plain struct that merely uses the same child names, but carries no
tag, stays a struct.
+ val untagged = new StructType().add(
+ "value",
+ new StructType()
+ .add("epochMicros", LongType, nullable = false)
+ .add("nanosWithinMicro", ShortType, nullable = false))
+ losslessRoundtrip(untagged)
+ assert(
+ ArrowUtils.fromArrowSchema(ArrowUtils.toArrowSchema(untagged, null,
true, false)) ===
+ untagged)
+
+ // The default mapping is untouched when the flag is off: still a single
nanosecond timestamp.
+ val defaultSchema = ArrowUtils.toArrowSchema(
+ new StructType().add("value", TimestampNTZNanosType(9)), null, true,
false)
+
assert(defaultSchema.findField("value").getType.isInstanceOf[ArrowType.Timestamp])
+ }
+
test("time") {
// Arrow's Time type has no precision field, so TIME(p) precision is
preserved via field
// metadata; the Arrow type itself stays Time(NANOSECOND, 64).
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/arrow/ArrowWriterSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/arrow/ArrowWriterSuite.scala
index e4a22ff18846..89caaa00e922 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/arrow/ArrowWriterSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/arrow/ArrowWriterSuite.scala
@@ -219,6 +219,110 @@ class ArrowWriterSuite extends SparkFunSuite {
check(TimestampLTZNanosType(9), "UTC")
}
+ test("timestamp nanos lossless struct round-trip covers the full value
domain") {
+ // The default int64 epoch-nanoseconds mapping only covers roughly years
1677-2262 (see the
+ // DATETIME_OVERFLOW test above). The lossless struct representation
stores the raw
+ // (epochMicros, nanosWithinMicro) pair, so the full domain of the Spark
types (years
+ // 0001-9999) must round-trip -- including values that overflow the
default mapping.
+ def losslessWriter(schema: StructType): ArrowWriter = {
+ val arrowSchema =
+ ArrowUtils.toArrowSchema(schema, null, true, false,
losslessTimestampNanos = true)
+ val root = VectorSchemaRoot.create(arrowSchema, ArrowUtils.rootAllocator)
+ ArrowWriter.create(root)
+ }
+
+ val values = Seq(
+ TimestampNanosVal.fromParts(0L, 0.toShort),
+ TimestampNanosVal.fromParts(1234567L, 789.toShort),
+ // pre-epoch instant with a sub-microsecond remainder
+ TimestampNanosVal.fromParts(-1234567L, 13.toShort),
+ // 9999-12-31T23:59:59.999999999: epochNanos would be ~2.5e20, far past
Long.MaxValue
+ TimestampNanosVal.fromParts(253402300799999999L, 999.toShort),
+ // 0001-01-01T00:00:00.000000001: epochNanos would be ~-6.2e19, far
below Long.MinValue
+ TimestampNanosVal.fromParts(-62135596800000000L, 1.toShort))
+
+ def check(dt: DataType): Unit = {
+ val schema = new StructType().add("value", dt, nullable = true)
+ val writer = losslessWriter(schema)
+ (values.map(Option(_)) :+ None).foreach(v =>
writer.write(InternalRow(v.orNull)))
+ writer.finish()
+
+ val reader = new ArrowColumnVector(writer.root.getFieldVectors().get(0))
+ assert(reader.dataType() === dt)
+ values.zipWithIndex.foreach { case (v, rowId) =>
+ val got = dt match {
+ case _: TimestampNTZNanosType => reader.getTimestampNTZNanos(rowId)
+ case _: TimestampLTZNanosType => reader.getTimestampLTZNanos(rowId)
+ }
+ assert(got === v)
+ }
+ assert(reader.isNullAt(values.length))
+ writer.root.close()
+ }
+
+ check(TimestampNTZNanosType(9))
+ check(TimestampLTZNanosType(9))
+ // Precision is schema metadata, not part of the value encoding.
+ check(TimestampNTZNanosType(7))
+ check(TimestampLTZNanosType(7))
+ }
+
+ test("timestamp nanos lossless struct round-trip inside nested types") {
+ val v1 = TimestampNanosVal.fromParts(253402300799999999L, 999.toShort)
+ val v2 = TimestampNanosVal.fromParts(-62135596800000000L, 1.toShort)
+
+ def losslessWriter(schema: StructType): ArrowWriter = {
+ val arrowSchema =
+ ArrowUtils.toArrowSchema(schema, null, true, false,
losslessTimestampNanos = true)
+ val root = VectorSchemaRoot.create(arrowSchema, ArrowUtils.rootAllocator)
+ ArrowWriter.create(root)
+ }
+
+ // array<timestamp_ntz(9)>
+ {
+ val schema = new StructType().add("arr",
ArrayType(TimestampNTZNanosType(9)))
+ val writer = losslessWriter(schema)
+ writer.write(InternalRow(new GenericArrayData(Array[Any](v1, null, v2))))
+ writer.finish()
+ val reader = new ArrowColumnVector(writer.root.getFieldVectors().get(0))
+ val arr = reader.getArray(0)
+ assert(arr.numElements() === 3)
+ assert(arr.getTimestampNTZNanos(0) === v1)
+ assert(arr.isNullAt(1))
+ assert(arr.getTimestampNTZNanos(2) === v2)
+ writer.root.close()
+ }
+
+ // struct<ts: timestamp_ltz(9)>
+ {
+ val schema = new StructType()
+ .add("struct", new StructType().add("ts", TimestampLTZNanosType(9)))
+ val writer = losslessWriter(schema)
+ writer.write(InternalRow(InternalRow(v1)))
+ writer.finish()
+ val reader = new ArrowColumnVector(writer.root.getFieldVectors().get(0))
+ assert(reader.getStruct(0).getTimestampLTZNanos(0) === v1)
+ writer.root.close()
+ }
+
+ // map<int, timestamp_ntz(9)>
+ {
+ val schema = new StructType().add("map", MapType(IntegerType,
TimestampNTZNanosType(9)))
+ val writer = losslessWriter(schema)
+ writer.write(InternalRow(
+ new ArrayBasedMapData(
+ new GenericArrayData(Array[Any](1, 2)),
+ new GenericArrayData(Array[Any](v1, v2)))))
+ writer.finish()
+ val reader = new ArrowColumnVector(writer.root.getFieldVectors().get(0))
+ val map = reader.getMap(0)
+ assert(map.numElements() === 2)
+ assert(map.valueArray().getTimestampNTZNanos(0) === v1)
+ assert(map.valueArray().getTimestampNTZNanos(1) === v2)
+ writer.root.close()
+ }
+ }
+
test("nested geographies") {
def check(
dt: StructType,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
