Skip to content

[SPARK-57100][SQL] Add columnar (ColumnVector) support for nanosecond timestamp types #56198

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
MaxGekk wants to merge 6 commits into from
Closed

[SPARK-57100][SQL] Add columnar (ColumnVector) support for nanosecond timestamp types #56198

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
2912125
[SPARK-57100][SQL] Add columnar (ColumnVector) support for nanosecond…
MaxGekk May 29, 2026
68fa774
[SPARK-57100][SQL] Fix review findings in nanosecond timestamp column…
MaxGekk May 29, 2026
e46c1f2
[SPARK-57100][SQL] Fix Scalastyle import order in ColumnarBatchSuite
MaxGekk May 29, 2026
7a10e70
Remove unused import
MaxGekk May 29, 2026
51e74ca
[SPARK-57100][SQL] Address review comments for nanosecond timestamp c…
MaxGekk May 31, 2026
37c8283
Merge remote-tracking branch 'origin/master' into nanos-column-vector
MaxGekk May 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 18 additions & 3 deletions sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ColumnVector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@

import scala.PartialFunction;

import org.apache.spark.SparkUnsupportedOperationException;
import org.apache.spark.annotation.Evolving;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.Decimal;
Expand Down Expand Up @@ -328,12 +327,28 @@ public CalendarInterval getInterval(int rowId) {
return new CalendarInterval(months, days, microseconds);
}

/**
* Returns the nanosecond NTZ timestamp value for {@code rowId}, or null if the slot is null.
* <p>
* To support this type, implementations must implement {@link #getChild(int)} and define 2 child
* vectors: child 0 is a long vector holding {@code epochMicros}; child 1 is a short vector
* holding {@code nanosWithinMicro} (values in [0, 999]).
*/
public TimestampNanosVal getTimestampNTZNanos(int rowId) {
throw SparkUnsupportedOperationException.apply();
if (isNullAt(rowId)) return null;
return TimestampNanosVal.fromTrustedRowBytes(
getChild(0).getLong(rowId), getChild(1).getShort(rowId));
}

/**
* Returns the nanosecond LTZ timestamp value for {@code rowId}, or null if the slot is null.
* <p>
* Storage layout is identical to {@link #getTimestampNTZNanos(int)}.
*/
public TimestampNanosVal getTimestampLTZNanos(int rowId) {
throw SparkUnsupportedOperationException.apply();
if (isNullAt(rowId)) return null;
return TimestampNanosVal.fromTrustedRowBytes(
getChild(0).getLong(rowId), getChild(1).getShort(rowId));
}

/**
Expand Down
12 changes: 11 additions & 1 deletion sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVectorUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
import org.apache.spark.sql.vectorized.ColumnarBatch;
import org.apache.spark.sql.vectorized.ColumnarMap;
import org.apache.spark.unsafe.types.CalendarInterval;
import org.apache.spark.unsafe.types.TimestampNanosVal;
import org.apache.spark.unsafe.types.UTF8String;
import org.apache.spark.unsafe.types.VariantVal;

Expand Down Expand Up @@ -106,6 +107,9 @@ public static void populate(
} else if (pdt instanceof PhysicalCalendarIntervalType) {
// The value of `numRows` is irrelevant.
col.setCalendarInterval((CalendarInterval) row.get(fieldIdx, t));
} else if (pdt instanceof PhysicalTimestampNTZNanosType ||
pdt instanceof PhysicalTimestampLTZNanosType) {
col.setTimestampNanosVal((TimestampNanosVal) row.get(fieldIdx, t));
} else if (pdt instanceof PhysicalVariantType) {
col.setVariant((VariantVal)row.get(fieldIdx, t));
} else if (pdt instanceof PhysicalStructType) {
Expand Down Expand Up @@ -171,7 +175,8 @@ public static Map<Integer, Integer> toJavaIntMap(ColumnarMap map) {

private static void appendValue(WritableColumnVector dst, DataType t, Object o) {
if (o == null) {
if (t instanceof CalendarIntervalType || t instanceof VariantType) {
if (t instanceof CalendarIntervalType || t instanceof VariantType ||
t instanceof TimestampNTZNanosType || t instanceof TimestampLTZNanosType) {
dst.appendStruct(true);
} else {
dst.appendNull();
Expand Down Expand Up @@ -219,6 +224,11 @@ private static void appendValue(WritableColumnVector dst, DataType t, Object o)
dst.appendStruct(false);
dst.getChild(0).appendByteArray(v.getValue(), 0, v.getValue().length);
dst.getChild(1).appendByteArray(v.getMetadata(), 0, v.getMetadata().length);
} else if (t instanceof TimestampNTZNanosType || t instanceof TimestampLTZNanosType) {
TimestampNanosVal v = (TimestampNanosVal) o;
dst.appendStruct(false);
dst.getChild(0).appendLong(v.epochMicros);
dst.getChild(1).appendShort(v.nanosWithinMicro);
} else if (t instanceof DateType) {
dst.appendInt(DateTimeUtils.fromJavaDate((Date) o));
} else if (t instanceof TimestampType) {
Expand Down
14 changes: 14 additions & 0 deletions sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ConstantColumnVector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.spark.sql.vectorized.ColumnarArray;
import org.apache.spark.sql.vectorized.ColumnarMap;
import org.apache.spark.unsafe.types.CalendarInterval;
import org.apache.spark.unsafe.types.TimestampNanosVal;
import org.apache.spark.unsafe.types.UTF8String;
import org.apache.spark.unsafe.types.VariantVal;

Expand Down Expand Up @@ -73,6 +74,11 @@ public ConstantColumnVector(int numRows, DataType type) {
this.childData[0] = new ConstantColumnVector(1, DataTypes.IntegerType);
this.childData[1] = new ConstantColumnVector(1, DataTypes.IntegerType);
this.childData[2] = new ConstantColumnVector(1, DataTypes.LongType);
} else if (type instanceof TimestampNTZNanosType || type instanceof TimestampLTZNanosType) {
// Two columns. EpochMicros as Long. NanosWithinMicro as Short.
this.childData = new ConstantColumnVector[2];
this.childData[0] = new ConstantColumnVector(1, DataTypes.LongType);
this.childData[1] = new ConstantColumnVector(1, DataTypes.ShortType);
} else if (type instanceof VariantType) {
this.childData = new ConstantColumnVector[2];
this.childData[0] = new ConstantColumnVector(1, DataTypes.BinaryType);
Expand Down Expand Up @@ -359,6 +365,14 @@ public void setCalendarInterval(CalendarInterval value) {
this.childData[2].setLong(value.microseconds);
}

/**
* Sets the nanosecond timestamp `value` for all rows
*/
public void setTimestampNanosVal(TimestampNanosVal value) {
this.childData[0].setLong(value.epochMicros);
this.childData[1].setShort(value.nanosWithinMicro);
}

/**
* Sets the Variant `value` for all rows
*/
Expand Down
22 changes: 22 additions & 0 deletions sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/MutableColumnarRow.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,10 @@ public InternalRow copy() {
row.update(i, getMap(i).copy());
} else if (dt instanceof VariantType) {
row.update(i, getVariant(i));
} else if (dt instanceof TimestampNTZNanosType) {
row.update(i, getTimestampNTZNanos(i));
} else if (dt instanceof TimestampLTZNanosType) {
row.update(i, getTimestampLTZNanos(i));
} else {
throw new RuntimeException("Not implemented. " + dt);
}
Expand Down Expand Up @@ -232,6 +236,10 @@ public Object get(int ordinal, DataType dataType) {
return getMap(ordinal);
} else if (dataType instanceof VariantType) {
return getVariant(ordinal);
} else if (dataType instanceof TimestampNTZNanosType) {
return getTimestampNTZNanos(ordinal);
} else if (dataType instanceof TimestampLTZNanosType) {
return getTimestampLTZNanos(ordinal);
} else {
throw new SparkUnsupportedOperationException(
"_LEGACY_ERROR_TEMP_3192", Map.of("dt", dataType.toString()));
Expand Down Expand Up @@ -261,6 +269,10 @@ public void update(int ordinal, Object value) {
setDecimal(ordinal, d, t.precision());
} else if (dt instanceof CalendarIntervalType) {
setInterval(ordinal, (CalendarInterval) value);
} else if (dt instanceof TimestampNTZNanosType) {
setTimestampNTZNanos(ordinal, (TimestampNanosVal) value);
} else if (dt instanceof TimestampLTZNanosType) {
setTimestampLTZNanos(ordinal, (TimestampNanosVal) value);
} else {
throw new SparkUnsupportedOperationException(
"_LEGACY_ERROR_TEMP_3192", Map.of("dt", dt.toString()));
Expand Down Expand Up @@ -326,4 +338,14 @@ public void setInterval(int ordinal, CalendarInterval value) {
columns[ordinal].putNotNull(rowId);
columns[ordinal].putInterval(rowId, value);
}

public void setTimestampNTZNanos(int ordinal, TimestampNanosVal value) {
columns[ordinal].putNotNull(rowId);
columns[ordinal].putTimestampNTZNanos(rowId, value);
}

public void setTimestampLTZNanos(int ordinal, TimestampNanosVal value) {
columns[ordinal].putNotNull(rowId);
columns[ordinal].putTimestampLTZNanos(rowId, value);
}
}
20 changes: 19 additions & 1 deletion sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.apache.spark.sql.vectorized.ColumnarMap;
import org.apache.spark.unsafe.array.ByteArrayMethods;
import org.apache.spark.unsafe.types.CalendarInterval;
import org.apache.spark.unsafe.types.TimestampNanosVal;
import org.apache.spark.unsafe.types.UTF8String;

/**
Expand Down Expand Up @@ -493,6 +494,16 @@ public void putInterval(int rowId, CalendarInterval value) {
getChild(2).putLong(rowId, value.microseconds);
}

public void putTimestampNTZNanos(int rowId, TimestampNanosVal value) {
getChild(0).putLong(rowId, value.epochMicros);
getChild(1).putShort(rowId, value.nanosWithinMicro);
}

public void putTimestampLTZNanos(int rowId, TimestampNanosVal value) {
getChild(0).putLong(rowId, value.epochMicros);
getChild(1).putShort(rowId, value.nanosWithinMicro);
}

@Override
public UTF8String getUTF8String(int rowId) {
if (isNullAt(rowId)) return null;
Expand Down Expand Up @@ -751,7 +762,9 @@ public final int appendStruct(boolean isNull) {
putNull(elementsAppended);
elementsAppended++;
for (WritableColumnVector c: childColumns) {
if (c.type instanceof StructType || c.type instanceof VariantType) {
if (c.type instanceof StructType || c.type instanceof VariantType
|| c.type instanceof TimestampNTZNanosType
|| c.type instanceof TimestampLTZNanosType) {
c.appendStruct(true);
} else {
c.appendNull();
Expand Down Expand Up @@ -1056,6 +1069,11 @@ protected WritableColumnVector(int capacity, DataType dataType) {
this.childColumns[0] = reserveNewColumn(capacity, DataTypes.IntegerType);
this.childColumns[1] = reserveNewColumn(capacity, DataTypes.IntegerType);
this.childColumns[2] = reserveNewColumn(capacity, DataTypes.LongType);
} else if (type instanceof TimestampNTZNanosType || type instanceof TimestampLTZNanosType) {
// Two columns. EpochMicros as Long. NanosWithinMicro as Short.
this.childColumns = new WritableColumnVector[2];
this.childColumns[0] = reserveNewColumn(capacity, DataTypes.LongType);
this.childColumns[1] = reserveNewColumn(capacity, DataTypes.ShortType);
} else if (type instanceof VariantType) {
this.childColumns = new WritableColumnVector[2];
this.childColumns[0] = reserveNewColumn(capacity, DataTypes.BinaryType);
Expand Down
22 changes: 22 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -272,6 +272,8 @@ private object RowToColumnConverter {
case _: GeometryType => GeometryConverter
case CalendarIntervalType => CalendarConverter
case VariantType => VariantConverter
case _: TimestampNTZNanosType => TimestampNTZNanosConverter
case _: TimestampLTZNanosType => TimestampLTZNanosConverter
case at: ArrayType => ArrayConverter(getConverterForType(at.elementType, at.containsNull))
case st: StructType => new StructConverter(st.fields.map(
(f) => getConverterForType(f.dataType, f.nullable)))
Expand All @@ -284,6 +286,8 @@ private object RowToColumnConverter {
if (nullable) {
dataType match {
case CalendarIntervalType | VariantType => new StructNullableTypeConverter(core)
case _: TimestampNTZNanosType | _: TimestampLTZNanosType =>
new StructNullableTypeConverter(core)
case st: StructType => new StructNullableTypeConverter(core)
case _ => new BasicNullableTypeConverter(core)
}
Expand Down Expand Up @@ -374,6 +378,24 @@ private object RowToColumnConverter {
}
}

private object TimestampNTZNanosConverter extends TypeConverter {
override def append(row: SpecializedGetters, column: Int, cv: WritableColumnVector): Unit = {
val v = row.getTimestampNTZNanos(column)
cv.appendStruct(false)
cv.getChild(0).appendLong(v.epochMicros)
cv.getChild(1).appendShort(v.nanosWithinMicro)
}
}

private object TimestampLTZNanosConverter extends TypeConverter {
override def append(row: SpecializedGetters, column: Int, cv: WritableColumnVector): Unit = {
val v = row.getTimestampLTZNanos(column)
cv.appendStruct(false)
cv.getChild(0).appendLong(v.epochMicros)
cv.getChild(1).appendShort(v.nanosWithinMicro)
}
}

private case class ArrayConverter(childConverter: TypeConverter) extends TypeConverter {
override def append(row: SpecializedGetters, column: Int, cv: WritableColumnVector): Unit = {
val values = row.getArray(column)
Expand Down
56 changes: 55 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/execution/RowToColumnConverterSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData, GenericArrayData}
import org.apache.spark.sql.execution.vectorized.{OnHeapColumnVector, WritableColumnVector}
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.unsafe.types.{TimestampNanosVal, UTF8String}
import org.apache.spark.util.ArrayImplicits._

class RowToColumnConverterSuite extends SparkFunSuite {
Expand Down Expand Up @@ -130,6 +130,60 @@ class RowToColumnConverterSuite extends SparkFunSuite {
}
}

test("TimestampNTZNanosType column roundtrip") {
val t = TimestampNTZNanosType(9)
val schema = StructType(Seq(StructField("ts", t)))
val values = Seq(
TimestampNanosVal.fromParts(0L, 0.toShort),
TimestampNanosVal.fromParts(1_000_000L, 999.toShort),
TimestampNanosVal.fromParts(-1L, 123.toShort))
val rows = values.map(v => InternalRow(v))
val vectors = convertRows(rows, schema)
values.zipWithIndex.foreach { case (v, i) =>
assert(vectors.head.getTimestampNTZNanos(i) === v)
}
}

test("TimestampNTZNanosType column with nulls") {
val t = TimestampNTZNanosType(9)
val schema = StructType(Seq(StructField("ts", t, nullable = true)))
val rows = Seq(
InternalRow(TimestampNanosVal.fromParts(100L, 42.toShort)),
InternalRow(null),
InternalRow(TimestampNanosVal.fromParts(200L, 1.toShort)))
val vectors = convertRows(rows, schema)
assert(vectors.head.getTimestampNTZNanos(0) === TimestampNanosVal.fromParts(100L, 42.toShort))
assert(vectors.head.isNullAt(1))
assert(vectors.head.getTimestampNTZNanos(2) === TimestampNanosVal.fromParts(200L, 1.toShort))
}

test("TimestampLTZNanosType column roundtrip") {
val t = TimestampLTZNanosType(9)
val schema = StructType(Seq(StructField("ts", t)))
val values = Seq(
TimestampNanosVal.fromParts(0L, 0.toShort),
TimestampNanosVal.fromParts(1_000_000L, 999.toShort),
TimestampNanosVal.fromParts(-1L, 123.toShort))
val rows = values.map(v => InternalRow(v))
val vectors = convertRows(rows, schema)
values.zipWithIndex.foreach { case (v, i) =>
assert(vectors.head.getTimestampLTZNanos(i) === v)
}
}

test("TimestampLTZNanosType column with nulls") {
val t = TimestampLTZNanosType(9)
val schema = StructType(Seq(StructField("ts", t, nullable = true)))
val rows = Seq(
InternalRow(TimestampNanosVal.fromParts(100L, 42.toShort)),
InternalRow(null),
InternalRow(TimestampNanosVal.fromParts(200L, 1.toShort)))
val vectors = convertRows(rows, schema)
assert(vectors.head.getTimestampLTZNanos(0) === TimestampNanosVal.fromParts(100L, 42.toShort))
assert(vectors.head.isNullAt(1))
assert(vectors.head.getTimestampLTZNanos(2) === TimestampNanosVal.fromParts(200L, 1.toShort))
}

test("multiple columns") {
val schema = StructType(
Seq(StructField("s", ShortType), StructField("i", IntegerType), StructField("l", LongType)))
Expand Down
62 changes: 61 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnVectorSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import org.apache.spark.sql.execution.columnar.compression.ColumnBuilderHelper
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
import org.apache.spark.sql.vectorized.ColumnarArray
import org.apache.spark.unsafe.types.{UTF8String, VariantVal}
import org.apache.spark.unsafe.types.{TimestampNanosVal, UTF8String, VariantVal}
import org.apache.spark.util.ArrayImplicits._

class ColumnVectorSuite extends SparkFunSuite with SQLHelper {
Expand Down Expand Up @@ -379,6 +379,66 @@ class ColumnVectorSuite extends SparkFunSuite with SQLHelper {
}
}

testVectors("timestamp_ntz_nanos", 10, TimestampNTZNanosType(9)) { testVector =>
val values = (0 until 10).map(i => TimestampNanosVal.fromParts(i * 1000L, i.toShort))
values.foreach { v =>
testVector.putNotNull(testVector.elementsAppended)
testVector.putTimestampNTZNanos(testVector.elementsAppended, v)
testVector.elementsAppended += 1
}
values.zipWithIndex.foreach { case (v, i) =>
assert(testVector.getTimestampNTZNanos(i) === v)
}
testVector.putNull(0)
assert(testVector.isNullAt(0))
}

testVectors("timestamp_ltz_nanos", 10, TimestampLTZNanosType(9)) { testVector =>
val values = (0 until 10).map(i => TimestampNanosVal.fromParts(i * 1000L, i.toShort))
values.foreach { v =>
testVector.putNotNull(testVector.elementsAppended)
testVector.putTimestampLTZNanos(testVector.elementsAppended, v)
testVector.elementsAppended += 1
}
values.zipWithIndex.foreach { case (v, i) =>
assert(testVector.getTimestampLTZNanos(i) === v)
}
testVector.putNull(0)
assert(testVector.isNullAt(0))
}

testVectors("mutable ColumnarRow with TimestampNTZNanosType", 5,
TimestampNTZNanosType(9)) { testVector =>
val mutableRow = new MutableColumnarRow(Array(testVector))
val values = (0 until 5).map(i => TimestampNanosVal.fromParts(i * 100L, i.toShort))
values.zipWithIndex.foreach { case (v, i) =>
mutableRow.rowId = i
mutableRow.setTimestampNTZNanos(0, v)
}
values.zipWithIndex.foreach { case (v, i) =>
mutableRow.rowId = i
assert(mutableRow.getTimestampNTZNanos(0) === v)
assert(mutableRow.get(0, TimestampNTZNanosType(9)) === v)
assert(mutableRow.copy().get(0, TimestampNTZNanosType(9)) === v)
}
}
Copy link
Copy Markdown
Contributor

@peter-toth peter-toth May 30, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The PR adds a MutableColumnarRow test for TimestampNTZNanosType but not for TimestampLTZNanosType. The LTZ paths (setTimestampLTZNanos at MutableColumnarRow.java:348, the update(TimestampLTZNanosType) and get(TimestampLTZNanosType) dispatches at MutableColumnarRow.java:240,272, and the copy() branch at MutableColumnarRow.java:104) aren't exercised. Adding a parallel mutable ColumnarRow with TimestampLTZNanosType block right after this one closes the gap.

Copy link
Copy Markdown
Member Author

@MaxGekk MaxGekk May 31, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done. Added a parallel mutable ColumnarRow with TimestampLTZNanosType test that exercises setTimestampLTZNanos/getTimestampLTZNanos, the update/get dispatches, and the copy() branch.


testVectors("mutable ColumnarRow with TimestampLTZNanosType", 5,
TimestampLTZNanosType(9)) { testVector =>
val mutableRow = new MutableColumnarRow(Array(testVector))
val values = (0 until 5).map(i => TimestampNanosVal.fromParts(i * 100L, i.toShort))
values.zipWithIndex.foreach { case (v, i) =>
mutableRow.rowId = i
mutableRow.setTimestampLTZNanos(0, v)
}
values.zipWithIndex.foreach { case (v, i) =>
mutableRow.rowId = i
assert(mutableRow.getTimestampLTZNanos(0) === v)
assert(mutableRow.get(0, TimestampLTZNanosType(9)) === v)
assert(mutableRow.copy().get(0, TimestampLTZNanosType(9)) === v)
}
}

testVectors("mutable ColumnarRow with TimestampNTZType", 10, TimestampNTZType) { testVector =>
val mutableRow = new MutableColumnarRow(Array(testVector))
(0 until 10).foreach { i =>
Expand Down
Loading