Skip to content

Commit 3c812fb

Browse files
viiryaMaxGekk
authored andcommitted
[SPARK-57735][SQL] Support nanosecond-precision timestamp types in the in-memory columnar cache
### What changes were proposed in this pull request? The default in-memory columnar cache serializer (`DefaultCachedBatchSerializer`) did not support `TimestampNTZNanosType` / `TimestampLTZNanosType`. Caching a DataFrame with such a column failed at materialization with `not support type: TimestampNTZNanosType(9)`, because none of the cache's type-dispatch sites had a case for them. This adds full support, following the fixed-width multi-field pattern already used by `CalendarInterval`. The physical value `TimestampNanosVal` is a fixed 16-byte payload (an 8-byte `epochMicros` plus an 8-byte word holding `nanosWithinMicro`), so it maps cleanly onto that pattern: - **`ColumnType`**: a `TIMESTAMP_NANOS` column type (with `TIMESTAMP_NTZ_NANOS` / `TIMESTAMP_LTZ_NANOS` singletons) whose `append`/`extract` read and write the 16-byte payload, with a `MutableUnsafeRow` direct-copy fast path. - **`ColumnBuilder`, `ColumnAccessor`**: builder and accessor classes plus dispatch cases. - **`ColumnStats`**: a `TimestampNanosColumnStats` collector (fixed size, no min/max bounds). - **`GenerateColumnAccessor`**: the codegen accessor-class selection and initialization branch. `TIMESTAMP_NTZ` and `TIMESTAMP_LTZ` nanos types share the same storage and differ only by physical type and row getter/setter, so the encode/decode logic is shared between them. ### Why are the changes needed? Nanosecond-precision timestamp types are otherwise unsupported by the cache, so `df.cache()` on a column of these types throws. With this change such DataFrames cache and read back correctly, consistent with the microsecond `TIMESTAMP_NTZ` / `TIMESTAMP` types which the cache already supports. ### Does this PR introduce _any_ user-facing change? Yes. Previously, caching a DataFrame containing a `TIMESTAMP_NTZ(p)` / `TIMESTAMP_LTZ(p)` column with `p` in the nanosecond range threw `not support type`. Now it caches and reads back the values, including sub-microsecond precision. ### How was this patch tested? - `ColumnTypeSuite`: append/extract round-trip for `TIMESTAMP_NTZ_NANOS` and `TIMESTAMP_LTZ_NANOS` (random values), plus `defaultSize` checks. - `InMemoryColumnarQuerySuite`: an end-to-end cache roundtrip for both nanos types, with the vectorized reader both on and off, covering sub-microsecond precision and null values. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code Closes #56842 from viirya/nanos-timestamp-default-cache. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Max Gekk <max.gekk@gmail.com> (cherry picked from commit 90dbc53) Signed-off-by: Max Gekk <max.gekk@gmail.com>
1 parent 8809def commit 3c812fb

11 files changed

Lines changed: 253 additions & 6 deletions

File tree

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnAccessor.scala

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ import org.apache.spark.sql.errors.QueryExecutionErrors
2828
import org.apache.spark.sql.execution.columnar.compression.CompressibleColumnAccessor
2929
import org.apache.spark.sql.execution.vectorized.WritableColumnVector
3030
import org.apache.spark.sql.types._
31-
import org.apache.spark.unsafe.types.{CalendarInterval, VariantVal}
31+
import org.apache.spark.unsafe.types.{CalendarInterval, TimestampNanosVal, VariantVal}
3232

3333
/**
3434
* An `Iterator` like trait used to extract values from columnar byte buffer. When a value is
@@ -115,6 +115,14 @@ private[columnar] class VariantColumnAccessor(buffer: ByteBuffer)
115115
extends BasicColumnAccessor[VariantVal](buffer, VARIANT)
116116
with NullableColumnAccessor
117117

118+
private[columnar] class TimestampNTZNanosColumnAccessor(buffer: ByteBuffer)
119+
extends BasicColumnAccessor[TimestampNanosVal](buffer, TIMESTAMP_NTZ_NANOS)
120+
with NullableColumnAccessor
121+
122+
private[columnar] class TimestampLTZNanosColumnAccessor(buffer: ByteBuffer)
123+
extends BasicColumnAccessor[TimestampNanosVal](buffer, TIMESTAMP_LTZ_NANOS)
124+
with NullableColumnAccessor
125+
118126
private[columnar] class CompactDecimalColumnAccessor(buffer: ByteBuffer, dataType: DecimalType)
119127
extends NativeColumnAccessor(buffer, COMPACT_DECIMAL(dataType))
120128

@@ -153,6 +161,8 @@ private[sql] object ColumnAccessor {
153161
case DoubleType => new DoubleColumnAccessor(buf)
154162
case s: StringType => new StringColumnAccessor(buf, s)
155163
case BinaryType => new BinaryColumnAccessor(buf)
164+
case _: TimestampNTZNanosType => new TimestampNTZNanosColumnAccessor(buf)
165+
case _: TimestampLTZNanosType => new TimestampLTZNanosColumnAccessor(buf)
156166
case dt: DecimalType if dt.precision <= Decimal.MAX_LONG_DIGITS =>
157167
new CompactDecimalColumnAccessor(buf, dt)
158168
case dt: DecimalType => new DecimalColumnAccessor(buf, dt)

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnBuilder.scala

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,12 @@ class IntervalColumnBuilder extends ComplexColumnBuilder(new IntervalColumnStats
134134
private[columnar]
135135
class VariantColumnBuilder extends ComplexColumnBuilder(new VariantColumnStats, VARIANT)
136136

137+
private[columnar] class TimestampNTZNanosColumnBuilder
138+
extends ComplexColumnBuilder(new TimestampNanosColumnStats, TIMESTAMP_NTZ_NANOS)
139+
140+
private[columnar] class TimestampLTZNanosColumnBuilder
141+
extends ComplexColumnBuilder(new TimestampNanosColumnStats, TIMESTAMP_LTZ_NANOS)
142+
137143
private[columnar] class CompactDecimalColumnBuilder(dataType: DecimalType)
138144
extends NativeColumnBuilder(new DecimalColumnStats(dataType), COMPACT_DECIMAL(dataType))
139145

@@ -193,6 +199,8 @@ private[columnar] object ColumnBuilder {
193199
case BinaryType => new BinaryColumnBuilder
194200
case CalendarIntervalType => new IntervalColumnBuilder
195201
case VariantType => new VariantColumnBuilder
202+
case _: TimestampNTZNanosType => new TimestampNTZNanosColumnBuilder
203+
case _: TimestampLTZNanosType => new TimestampLTZNanosColumnBuilder
196204
case dt: DecimalType if dt.precision <= Decimal.MAX_LONG_DIGITS =>
197205
new CompactDecimalColumnBuilder(dt)
198206
case dt: DecimalType => new DecimalColumnBuilder(dt)

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.columnar
2020
import org.apache.spark.sql.catalyst.InternalRow
2121
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeReference}
2222
import org.apache.spark.sql.types._
23-
import org.apache.spark.unsafe.types.UTF8String
23+
import org.apache.spark.unsafe.types.{TimestampNanosVal, UTF8String}
2424

2525
class ColumnStatisticsSchema(a: Attribute) extends Serializable {
2626
val upperBound = AttributeReference(a.name + ".upperBound", a.dataType, nullable = true)()
@@ -326,6 +326,30 @@ private[columnar] final class IntervalColumnStats extends ColumnStats {
326326
Array[Any](null, null, nullCount, count, sizeInBytes)
327327
}
328328

329+
private[columnar] final class TimestampNanosColumnStats extends ColumnStats {
330+
protected var upper: TimestampNanosVal = null
331+
protected var lower: TimestampNanosVal = null
332+
333+
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
334+
if (!row.isNullAt(ordinal)) {
335+
// TimestampNanosVal has a total order matching calendar order, so collect min/max bounds
336+
// (like DecimalColumnStats, not IntervalColumnStats) to enable partition pruning, matching
337+
// the micro-precision timestamp path (LongColumnStats). NTZ and LTZ share the same physical
338+
// payload, so a single getter reads the value for both.
339+
val value = row.getTimestampNTZNanos(ordinal)
340+
if (upper == null || value.compareTo(upper) > 0) upper = value
341+
if (lower == null || value.compareTo(lower) < 0) lower = value
342+
sizeInBytes += TimestampNanosVal.SIZE_IN_BYTES
343+
count += 1
344+
} else {
345+
gatherNullStats()
346+
}
347+
}
348+
349+
override def collectedStatistics: Array[Any] =
350+
Array[Any](lower, upper, nullCount, count, sizeInBytes)
351+
}
352+
329353
private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) extends ColumnStats {
330354
def this(dt: DecimalType) = this(dt.precision, dt.scale)
331355

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnType.scala

Lines changed: 70 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.types._
2828
import org.apache.spark.sql.errors.ExecutionErrors
2929
import org.apache.spark.sql.types._
3030
import org.apache.spark.unsafe.Platform
31-
import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String, VariantVal}
31+
import org.apache.spark.unsafe.types.{CalendarInterval, TimestampNanosVal, UTF8String, VariantVal}
3232

3333

3434
/**
@@ -815,6 +815,73 @@ private[columnar] object CALENDAR_INTERVAL extends ColumnType[CalendarInterval]
815815
}
816816
}
817817

818+
/**
819+
* Used to append/extract [[TimestampNanosVal]] into/from the underlying [[ByteBuffer]] of a
820+
* column. The on-buffer layout mirrors the UnsafeRow variable-length payload: an 8-byte
821+
* epochMicros followed by an 8-byte word holding nanosWithinMicro (zero-extended), 16 bytes total
822+
* (see TimestampNanosRowValues). NTZ and LTZ share this storage and differ only by physical type,
823+
* so the two singletons below pass their own physicalType and row getter/setter.
824+
*/
825+
private[columnar] abstract class TIMESTAMP_NANOS(physicalType: PhysicalDataType)
826+
extends ColumnType[TimestampNanosVal] {
827+
828+
override def dataType: PhysicalDataType = physicalType
829+
830+
override def defaultSize: Int = TimestampNanosVal.SIZE_IN_BYTES
831+
832+
protected def getNanos(row: InternalRow, ordinal: Int): TimestampNanosVal
833+
protected def setNanos(row: InternalRow, ordinal: Int, value: TimestampNanosVal): Unit
834+
835+
override def getField(row: InternalRow, ordinal: Int): TimestampNanosVal = getNanos(row, ordinal)
836+
837+
override def setField(row: InternalRow, ordinal: Int, value: TimestampNanosVal): Unit =
838+
setNanos(row, ordinal, value)
839+
840+
override def extract(buffer: ByteBuffer): TimestampNanosVal = {
841+
val epochMicros = ByteBufferHelper.getLong(buffer)
842+
// The nanos field is stored in a full 8-byte word (matching the UnsafeRow payload), so read a
843+
// long and narrow it; the writer guarantees the value is in [0, 999].
844+
val nanosWithinMicro = ByteBufferHelper.getLong(buffer).toShort
845+
TimestampNanosVal.fromTrustedRowBytes(epochMicros, nanosWithinMicro)
846+
}
847+
848+
// Copy the fixed 16-byte payload straight into the UnsafeRow, like CALENDAR_INTERVAL.
849+
override def extract(buffer: ByteBuffer, row: InternalRow, ordinal: Int): Unit = {
850+
row match {
851+
case mutable: MutableUnsafeRow =>
852+
val cursor = buffer.position()
853+
buffer.position(cursor + defaultSize)
854+
mutable.writer.write(ordinal, buffer.array(),
855+
buffer.arrayOffset() + cursor, defaultSize)
856+
case _ =>
857+
setField(row, ordinal, extract(buffer))
858+
}
859+
}
860+
861+
override def append(v: TimestampNanosVal, buffer: ByteBuffer): Unit = {
862+
ByteBufferHelper.putLong(buffer, v.epochMicros)
863+
ByteBufferHelper.putLong(buffer, v.nanosWithinMicro.toLong)
864+
}
865+
}
866+
867+
private[columnar] object TIMESTAMP_NTZ_NANOS
868+
extends TIMESTAMP_NANOS(PhysicalTimestampNTZNanosType) {
869+
override protected def getNanos(row: InternalRow, ordinal: Int): TimestampNanosVal =
870+
row.getTimestampNTZNanos(ordinal)
871+
override protected def setNanos(
872+
row: InternalRow, ordinal: Int, value: TimestampNanosVal): Unit =
873+
row.setTimestampNTZNanos(ordinal, value)
874+
}
875+
876+
private[columnar] object TIMESTAMP_LTZ_NANOS
877+
extends TIMESTAMP_NANOS(PhysicalTimestampLTZNanosType) {
878+
override protected def getNanos(row: InternalRow, ordinal: Int): TimestampNanosVal =
879+
row.getTimestampLTZNanos(ordinal)
880+
override protected def setNanos(
881+
row: InternalRow, ordinal: Int, value: TimestampNanosVal): Unit =
882+
row.setTimestampLTZNanos(ordinal, value)
883+
}
884+
818885
/**
819886
* Used to append/extract Java VariantVals into/from the underlying [[ByteBuffer]] of a column.
820887
*
@@ -876,6 +943,8 @@ private[columnar] object ColumnType {
876943
case s: StringType => STRING(s)
877944
case BinaryType => BINARY
878945
case i: CalendarIntervalType => CALENDAR_INTERVAL
946+
case _: TimestampNTZNanosType => TIMESTAMP_NTZ_NANOS
947+
case _: TimestampLTZNanosType => TIMESTAMP_LTZ_NANOS
879948
case dt: DecimalType if dt.precision <= Decimal.MAX_LONG_DIGITS => COMPACT_DECIMAL(dt)
880949
case dt: DecimalType => LARGE_DECIMAL(dt)
881950
case arr: ArrayType => ARRAY(PhysicalArrayType(arr.elementType, arr.containsNull))

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/GenerateColumnAccessor.scala

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,8 @@ object GenerateColumnAccessor extends CodeGenerator[Seq[DataType], ColumnarItera
9090
case BinaryType => classOf[BinaryColumnAccessor].getName
9191
case CalendarIntervalType => classOf[IntervalColumnAccessor].getName
9292
case VariantType => classOf[VariantColumnAccessor].getName
93+
case _: TimestampNTZNanosType => classOf[TimestampNTZNanosColumnAccessor].getName
94+
case _: TimestampLTZNanosType => classOf[TimestampLTZNanosColumnAccessor].getName
9395
case dt: DecimalType if dt.precision <= Decimal.MAX_LONG_DIGITS =>
9496
classOf[CompactDecimalColumnAccessor].getName
9597
case dt: DecimalType => classOf[DecimalColumnAccessor].getName
@@ -102,7 +104,8 @@ object GenerateColumnAccessor extends CodeGenerator[Seq[DataType], ColumnarItera
102104
val createCode = dt match {
103105
case t if CodeGenerator.isPrimitiveType(dt) =>
104106
s"$accessorName = new $accessorCls(ByteBuffer.wrap(buffers[$index]).order(nativeOrder));"
105-
case NullType | BinaryType | CalendarIntervalType | VariantType =>
107+
case NullType | BinaryType | CalendarIntervalType | VariantType |
108+
_: TimestampNTZNanosType | _: TimestampLTZNanosType =>
106109
s"$accessorName = new $accessorCls(ByteBuffer.wrap(buffers[$index]).order(nativeOrder));"
107110
case other =>
108111
s"""$accessorName = new $accessorCls(ByteBuffer.wrap(buffers[$index]).order(nativeOrder),

sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnStatsSuite.scala

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.columnar
2020
import org.apache.spark.SparkFunSuite
2121
import org.apache.spark.sql.catalyst.types.PhysicalDataType
2222
import org.apache.spark.sql.types.StringType
23+
import org.apache.spark.unsafe.types.TimestampNanosVal
2324

2425
class ColumnStatsSuite extends SparkFunSuite {
2526
testColumnStats(classOf[BooleanColumnStats], BOOLEAN, Array(true, false, 0))
@@ -32,6 +33,8 @@ class ColumnStatsSuite extends SparkFunSuite {
3233
testDecimalColumnStats(Array(null, null, 0))
3334
testIntervalColumnStats(Array(null, null, 0))
3435
testStringColumnStats(Array(null, null, 0))
36+
testTimestampNanosColumnStats(TIMESTAMP_NTZ_NANOS, Array(null, null, 0))
37+
testTimestampNanosColumnStats(TIMESTAMP_LTZ_NANOS, Array(null, null, 0))
3538

3639
def testColumnStats[T <: PhysicalDataType, U <: ColumnStats](
3740
columnStatsClass: Class[U],
@@ -143,6 +146,40 @@ class ColumnStatsSuite extends SparkFunSuite {
143146
}
144147
}
145148

149+
def testTimestampNanosColumnStats(
150+
columnType: ColumnType[TimestampNanosVal],
151+
initialStatistics: Array[Any]): Unit = {
152+
153+
val columnStatsName = classOf[TimestampNanosColumnStats].getSimpleName
154+
155+
test(s"$columnStatsName ($columnType): empty") {
156+
val columnStats = new TimestampNanosColumnStats
157+
columnStats.collectedStatistics.zip(initialStatistics).foreach {
158+
case (actual, expected) => assert(actual === expected)
159+
}
160+
}
161+
162+
test(s"$columnStatsName ($columnType): non-empty collects min/max bounds") {
163+
import org.apache.spark.sql.execution.columnar.ColumnarTestUtils._
164+
165+
val columnStats = new TimestampNanosColumnStats
166+
val rows = Seq.fill(10)(makeRandomRow(columnType)) ++ Seq.fill(10)(makeNullRow(1))
167+
rows.foreach(columnStats.gatherStats(_, 0))
168+
169+
val values = rows.take(10).map(_.get(0,
170+
ColumnarDataTypeUtils.toLogicalDataType(columnType.dataType))
171+
.asInstanceOf[TimestampNanosVal])
172+
val ordering = Ordering.fromLessThan[TimestampNanosVal](_.compareTo(_) < 0)
173+
val stats = columnStats.collectedStatistics
174+
175+
assertResult(values.min(ordering), "Wrong lower bound")(stats(0))
176+
assertResult(values.max(ordering), "Wrong upper bound")(stats(1))
177+
assertResult(10, "Wrong null count")(stats(2))
178+
assertResult(20, "Wrong row count")(stats(3))
179+
assertResult(TimestampNanosVal.SIZE_IN_BYTES * 10 + 4 * 10, "Wrong size in bytes")(stats(4))
180+
}
181+
}
182+
146183
def testStringColumnStats[T <: PhysicalDataType, U <: ColumnStats](
147184
initialStatistics: Array[Any]): Unit = {
148185

sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnTypeSuite.scala

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ class ColumnTypeSuite extends SparkFunSuite {
4444
STRING(StringType) -> 8, STRING(StringType("UTF8_LCASE")) -> 8,
4545
STRING(StringType("UNICODE")) -> 8, STRING(StringType("UNICODE_CI")) -> 8,
4646
BINARY -> 16, STRUCT_TYPE -> 20, ARRAY_TYPE -> 28, MAP_TYPE -> 68,
47-
CALENDAR_INTERVAL -> 16)
47+
CALENDAR_INTERVAL -> 16, TIMESTAMP_NTZ_NANOS -> 16, TIMESTAMP_LTZ_NANOS -> 16)
4848

4949
checks.foreach { case (columnType, expectedSize) =>
5050
assertResult(expectedSize, s"Wrong defaultSize for $columnType") {
@@ -113,6 +113,8 @@ class ColumnTypeSuite extends SparkFunSuite {
113113
testColumnType(ARRAY_TYPE)
114114
testColumnType(MAP_TYPE)
115115
testColumnType(CALENDAR_INTERVAL)
116+
testColumnType(TIMESTAMP_NTZ_NANOS)
117+
testColumnType(TIMESTAMP_LTZ_NANOS)
116118

117119
def testNativeColumnType[T <: PhysicalDataType](columnType: NativeColumnType[T]): Unit = {
118120
val typeName = columnType match {

sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnarDataTypeUtils.scala

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,8 @@ object ColumnarDataTypeUtils {
3030
case PhysicalShortType => ShortType
3131
case PhysicalBinaryType => BinaryType
3232
case PhysicalCalendarIntervalType => CalendarIntervalType
33+
case PhysicalTimestampNTZNanosType => TimestampNTZNanosType()
34+
case PhysicalTimestampLTZNanosType => TimestampLTZNanosType()
3335
case PhysicalFloatType => FloatType
3436
case PhysicalDoubleType => DoubleType
3537
case PhysicalStringType(collationId) => StringType(collationId)

sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnarTestUtils.scala

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
2525
import org.apache.spark.sql.catalyst.types.PhysicalDataType
2626
import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
2727
import org.apache.spark.sql.types.Decimal
28-
import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
28+
import org.apache.spark.unsafe.types.{CalendarInterval, TimestampNanosVal, UTF8String}
2929

3030
object ColumnarTestUtils {
3131
def makeNullRow(length: Int): GenericInternalRow = {
@@ -54,6 +54,10 @@ object ColumnarTestUtils {
5454
case BINARY => randomBytes(Random.nextInt(32))
5555
case CALENDAR_INTERVAL =>
5656
new CalendarInterval(Random.nextInt(), Random.nextInt(), Random.nextLong())
57+
case _: TIMESTAMP_NANOS =>
58+
// nanosWithinMicro must be in [0, 999]; epochMicros can be any long.
59+
TimestampNanosVal.fromParts(
60+
Random.nextLong(), Random.nextInt(TimestampNanosVal.MAX_NANOS_WITHIN_MICRO + 1).toShort)
5761
case COMPACT_DECIMAL(precision, scale) => Decimal(Random.nextLong() % 100, precision, scale)
5862
case LARGE_DECIMAL(precision, scale) => Decimal(Random.nextLong(), precision, scale)
5963
case STRUCT(_) =>

sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -222,6 +222,36 @@ class InMemoryColumnarQuerySuite extends SharedSparkSession with AdaptiveSparkPl
222222
}
223223
}
224224

225+
test("cache nanosecond-precision timestamp types") {
226+
// Nanosecond timestamps are non-primitive for the default cache (DefaultCachedBatchSerializer
227+
// .supportsColumnarOutput is true only for primitive types), so they always read back through
228+
// the row path -- the vectorized reader is not exercised, the same as for CalendarInterval,
229+
// Variant, and Decimal.
230+
withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
231+
Seq("TIMESTAMP_NTZ(9)", "TIMESTAMP_LTZ(9)").foreach { typeName =>
232+
withTempView("nanos") {
233+
// Include sub-microsecond precision and a null to exercise the full payload and null
234+
// handling through the cache.
235+
val df = sql(
236+
s"""SELECT * FROM VALUES
237+
| (cast('2020-01-01 00:00:00.123456789' as $typeName)),
238+
| (cast('1999-12-31 23:59:59.987654321' as $typeName)),
239+
| (cast(null as $typeName))
240+
| as t(ts)""".stripMargin)
241+
df.createOrReplaceTempView("nanos")
242+
val expected = sql("SELECT ts FROM nanos").collect().toSeq
243+
244+
spark.catalog.cacheTable("nanos")
245+
try {
246+
checkAnswer(sql("SELECT ts FROM nanos"), expected)
247+
} finally {
248+
spark.catalog.uncacheTable("nanos")
249+
}
250+
}
251+
}
252+
}
253+
}
254+
225255
test("SPARK-3320 regression: batched column buffer building should work with empty partitions") {
226256
checkAnswer(
227257
sql("SELECT * FROM withEmptyParts"),

0 commit comments

Comments
 (0)