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 937684ef72ff [SPARK-58737][SQL] Support nanosecond-precision
timestamps in ColumnarRow/ColumnarBatchRow copy() and get()
937684ef72ff is described below
commit 937684ef72ff60df4ef26ddbbf9ec7fd84fd8773
Author: Stevo Mitric <[email protected]>
AuthorDate: Wed Aug 12 14:05:51 2026 +0000
[SPARK-58737][SQL] Support nanosecond-precision timestamps in
ColumnarRow/ColumnarBatchRow copy() and get()
### What changes were proposed in this pull request?
Extend the generic type-dispatch in `ColumnarRow` and `ColumnarBatchRow` to
cover the nanosecond-precision timestamp types (`TimestampNTZNanosType(p)` /
`TimestampLTZNanosType(p)`, p in [7, 9]; carried as
`org.apache.spark.unsafe.types.TimestampNanosVal`):
- `copy()` dispatches on `PhysicalDataType`; added
`PhysicalTimestampNTZNanosType` / `PhysicalTimestampLTZNanosType` branches.
`TimestampNanosVal` is immutable, so the value is shared without copying.
- `get(int, DataType)` dispatches on `DataType`; added
`TimestampNTZNanosType` / `TimestampLTZNanosType` branches next to
`TimestampNTZType`.
The per-cell leaf accessors (`getTimestampNTZNanos` /
`getTimestampLTZNanos`) already existed; only these two generic dispatch
methods were missing the branches.
### Why are the changes needed?
Any path that materializes a columnar (vectorized Parquet/ORC)
nanosecond-timestamp row via `row.copy()` (spills, debug output, some UDF
paths) or `row.get(ordinal, dataType)` (`Dataset.as[Row]`, `foreach`) threw at
runtime: `copy()` fell through to `RuntimeException("Not implemented. ...")`
and `get()` to `SparkUnsupportedOperationException` (`_LEGACY_ERROR_TEMP_3155`
/ `_LEGACY_ERROR_TEMP_3152`).
### Does this PR introduce any user-facing change?
No. Nanosecond timestamp types are behind the
`spark.sql.timestampNanosTypes.enabled` preview.
### How was this patch tested?
Added `ColumnarBatchSuite` cases covering
`TimestampNTZNanosType`/`TimestampLTZNanosType` for precisions 7/8/9,
exercising both `ColumnarBatchRow` and `ColumnarRow` (via `getStruct`). Each
case populates a two-child vector (child 0 = epochMicros Long, child 1 =
nanosWithinMicro Short) with a null, pre-epoch instants and the sub-microsecond
boundaries (0 and 999), and asserts the typed leaf accessor, that `get(ordinal,
dataType)` returns a `TimestampNanosVal` (not a boxed Long), and th [...]
### Was this patch authored or co-authored using generative AI tooling?
Co-Authored: Claude Code 4.8
Closes #57960 from stevomitric/stevomitric/columnar-nanos-dispatch.
Authored-by: Stevo Mitric <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
(cherry picked from commit bbd9025a5256f99cf46fcd813097a2de0223ab6c)
Signed-off-by: Uros Bojanic <[email protected]>
---
.../spark/sql/vectorized/ColumnarBatchRow.java | 9 +++
.../apache/spark/sql/vectorized/ColumnarRow.java | 9 +++
.../execution/vectorized/ColumnarBatchSuite.scala | 89 ++++++++++++++++++++++
3 files changed, 107 insertions(+)
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ColumnarBatchRow.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ColumnarBatchRow.java
index c689d9faf4b3..89b0b7c0a466 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ColumnarBatchRow.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ColumnarBatchRow.java
@@ -86,6 +86,11 @@ public final class ColumnarBatchRow extends InternalRow {
row.update(i, getMap(i).copy());
} else if (pdt instanceof PhysicalVariantType) {
row.update(i, getVariant(i));
+ } else if (pdt instanceof PhysicalTimestampNTZNanosType) {
+ // TimestampNanosVal is immutable, so it can be shared without
copying.
+ row.update(i, getTimestampNTZNanos(i));
+ } else if (pdt instanceof PhysicalTimestampLTZNanosType) {
+ row.update(i, getTimestampLTZNanos(i));
} else {
throw new RuntimeException("Not implemented. " + dt);
}
@@ -208,6 +213,10 @@ public final class ColumnarBatchRow extends InternalRow {
return getLong(ordinal);
} else if (dataType instanceof TimestampNTZType) {
return getLong(ordinal);
+ } else if (dataType instanceof TimestampNTZNanosType) {
+ return getTimestampNTZNanos(ordinal);
+ } else if (dataType instanceof TimestampLTZNanosType) {
+ return getTimestampLTZNanos(ordinal);
} else if (dataType instanceof ArrayType) {
return getArray(ordinal);
} else if (dataType instanceof StructType structType) {
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ColumnarRow.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ColumnarRow.java
index 9086726c5ee5..2db280c53de6 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ColumnarRow.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ColumnarRow.java
@@ -91,6 +91,11 @@ public final class ColumnarRow extends InternalRow {
row.update(i, getMap(i).copy());
} else if (pdt instanceof PhysicalVariantType) {
row.update(i, getVariant(i));
+ } else if (pdt instanceof PhysicalTimestampNTZNanosType) {
+ // TimestampNanosVal is immutable, so it can be shared without
copying.
+ row.update(i, getTimestampNTZNanos(i));
+ } else if (pdt instanceof PhysicalTimestampLTZNanosType) {
+ row.update(i, getTimestampLTZNanos(i));
} else {
throw new RuntimeException("Not implemented. " + dt);
}
@@ -212,6 +217,10 @@ public final class ColumnarRow extends InternalRow {
return getLong(ordinal);
} else if (dataType instanceof TimestampNTZType) {
return getLong(ordinal);
+ } else if (dataType instanceof TimestampNTZNanosType) {
+ return getTimestampNTZNanos(ordinal);
+ } else if (dataType instanceof TimestampLTZNanosType) {
+ return getTimestampLTZNanos(ordinal);
} else if (dataType instanceof ArrayType) {
return getArray(ordinal);
} else if (dataType instanceof StructType) {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala
index 6f0e39bb9c66..a29a4b2b2f26 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala
@@ -2103,6 +2103,95 @@ class ColumnarBatchSuite extends SparkFunSuite {
}
}
+ // Nanosecond-precision timestamp values used to exercise
ColumnarRow/ColumnarBatchRow
+ // dispatch. They cover a null (index 3), the sub-microsecond boundaries (0
and 999), and
+ // several pre-epoch instants; nanosWithinMicro must survive the round-trip
unchanged.
+ private val nanosTestValues: Seq[Option[TimestampNanosVal]] = Seq(
+ Some(TimestampNanosVal.fromParts(0L, 0.toShort)), //
epoch, no sub-micro
+ Some(TimestampNanosVal.fromParts(1L, 1.toShort)),
+ Some(TimestampNanosVal.fromParts(1000000L, 999.toShort)), // max
sub-micro
+ None, // null
slot
+ Some(TimestampNanosVal.fromParts(-1L, 500.toShort)), //
pre-epoch
+ Some(TimestampNanosVal.fromParts(-1000000000000L, 1.toShort)), //
pre-epoch, far
+ Some(TimestampNanosVal.fromParts(1234567890123456L, 789.toShort)),
+ Some(TimestampNanosVal.fromParts(-42L, 999.toShort)), //
pre-epoch, max sub-micro
+ Some(TimestampNanosVal.fromParts(Long.MaxValue, 0.toShort)),
+ Some(TimestampNanosVal.fromParts(Long.MinValue, 123.toShort)))
+
+ // Populates a nanos-typed vector (child 0 = epochMicros Long, child 1 =
nanosWithinMicro
+ // Short) from `nanosTestValues`, writing a null for the None slot.
+ private def putNanosTestValues(column: WritableColumnVector, isLtz:
Boolean): Unit = {
+ nanosTestValues.zipWithIndex.foreach {
+ case (Some(v), i) =>
+ if (isLtz) column.putTimestampLTZNanos(i, v) else
column.putTimestampNTZNanos(i, v)
+ case (None, i) =>
+ column.putNull(i)
+ }
+ }
+
+ // Verifies the typed leaf accessor, get(ordinal, dataType) and the copy()
round-trip for a
+ // single nanosecond-timestamp field at ordinal 0 of `row`.
+ private def assertNanosRow(
+ row: InternalRow,
+ dt: DataType,
+ isLtz: Boolean,
+ expected: Option[TimestampNanosVal]): Unit = {
+ def leaf(r: InternalRow): TimestampNanosVal =
+ if (isLtz) r.getTimestampLTZNanos(0) else r.getTimestampNTZNanos(0)
+ expected match {
+ case Some(v) =>
+ assert(!row.isNullAt(0))
+ assert(leaf(row) === v)
+ assert(leaf(row).nanosWithinMicro == v.nanosWithinMicro)
+ // get(ordinal, dataType) must return a TimestampNanosVal, not a boxed
Long.
+ val got = row.get(0, dt)
+ assert(got.isInstanceOf[TimestampNanosVal])
+ assert(got === v)
+ // copy() must yield a GenericInternalRow whose nanos field matches
the source.
+ val copied = row.copy()
+ assert(copied.isInstanceOf[GenericInternalRow])
+ assert(!copied.isNullAt(0))
+ assert(leaf(copied) === v)
+ assert(copied.get(0, dt) === v)
+ case None =>
+ assert(row.isNullAt(0))
+ assert(row.get(0, dt) == null)
+ assert(row.copy().isNullAt(0))
+ }
+ }
+
+ Seq(7, 8, 9).foreach { p =>
+ Seq(
+ (TimestampNTZNanosType(p): DataType, false),
+ (TimestampLTZNanosType(p): DataType, true)).foreach { case (dt, isLtz) =>
+ val family = if (isLtz) "LTZ" else "NTZ"
+
+ testVector(
+ s"ColumnarBatchRow nanos $family(precision=$p) round-trips
copy()/get()",
+ nanosTestValues.length,
+ dt) { column =>
+ putNanosTestValues(column, isLtz)
+ val batchRow = new ColumnarBatchRow(Array(column))
+ nanosTestValues.zipWithIndex.foreach { case (expected, i) =>
+ batchRow.rowId = i
+ assertNanosRow(batchRow, dt, isLtz, expected)
+ }
+ }
+
+ // ColumnarRow is produced by getStruct on a struct-typed vector, so
nest the nanos
+ // column inside a struct to exercise ColumnarRow's copy()/get()
dispatch.
+ testVector(
+ s"ColumnarRow nanos $family(precision=$p) round-trips copy()/get()",
+ nanosTestValues.length,
+ new StructType().add("ts", dt)) { column =>
+ putNanosTestValues(column.getChild(0), isLtz)
+ nanosTestValues.zipWithIndex.foreach { case (expected, i) =>
+ assertNanosRow(column.getStruct(i), dt, isLtz, expected)
+ }
+ }
+ }
+ }
+
testVector("WritableColumnVector.reserve(): requested capacity is negative",
1024, ByteType) {
column =>
val ex = intercept[RuntimeException] { column.reserve(-1) }
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
