[SPARK-56981][SQL] Add physical representation and UnsafeRow support for nanosecond timestamps

common/unsafe/src/main/java/org/apache/spark/unsafe/types/TimestampLTZNanos.java

@@ -0,0 +1,61 @@
+/*
+ * 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.unsafe.types;
+
+import org.apache.spark.annotation.Unstable;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * Physical representation of {@code TIMESTAMP_LTZ(p)} with nanosecond-capable precision.
+ * Values are stored as epoch microseconds plus nanoseconds within that microsecond (0-999).
+ *
+ * @since 4.2.0
+ */
+@Unstable
+public final class TimestampLTZNanos implements Serializable {
+  public static final int SIZE_IN_BYTES = 16;
+
+  public final long epochMicros;
+  public final short nanosWithinMicro;
+
+  public TimestampLTZNanos(long epochMicros, short nanosWithinMicro) {
+    TimestampNanosUtils.validateNanosWithinMicro(nanosWithinMicro);
+    this.epochMicros = epochMicros;
+    this.nanosWithinMicro = nanosWithinMicro;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    TimestampLTZNanos that = (TimestampLTZNanos) o;
+    return epochMicros == that.epochMicros && nanosWithinMicro == that.nanosWithinMicro;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(epochMicros, nanosWithinMicro);
+  }
+
+  @Override
+  public String toString() {
+    return "TimestampLTZNanos(" + epochMicros + ", " + nanosWithinMicro + ")";
+  }
+}

common/unsafe/src/main/java/org/apache/spark/unsafe/types/TimestampNTZNanos.java

@@ -0,0 +1,61 @@
+/*
+ * 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.unsafe.types;
+
+import org.apache.spark.annotation.Unstable;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * Physical representation of {@code TIMESTAMP_NTZ(p)} with nanosecond-capable precision.
+ * Values are stored as epoch microseconds plus nanoseconds within that microsecond (0-999).
+ *
+ * @since 4.2.0
+ */
+@Unstable
+public final class TimestampNTZNanos implements Serializable {
+  public static final int SIZE_IN_BYTES = 16;
+
+  public final long epochMicros;
+  public final short nanosWithinMicro;
+
+  public TimestampNTZNanos(long epochMicros, short nanosWithinMicro) {
+    TimestampNanosUtils.validateNanosWithinMicro(nanosWithinMicro);
+    this.epochMicros = epochMicros;
+    this.nanosWithinMicro = nanosWithinMicro;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    TimestampNTZNanos that = (TimestampNTZNanos) o;
+    return epochMicros == that.epochMicros && nanosWithinMicro == that.nanosWithinMicro;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(epochMicros, nanosWithinMicro);
+  }
+
+  @Override
+  public String toString() {
+    return "TimestampNTZNanos(" + epochMicros + ", " + nanosWithinMicro + ")";
+  }
+}

common/unsafe/src/main/java/org/apache/spark/unsafe/types/TimestampNanosUtils.java

@@ -0,0 +1,39 @@
+/*
+ * 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.unsafe.types;
+
+import java.util.Map;
+
+import org.apache.spark.SparkIllegalArgumentException;
+
+final class TimestampNanosUtils {
+  static final int MAX_NANOS_WITHIN_MICRO = 999;
+
+  private TimestampNanosUtils() {}
+
+  static void validateNanosWithinMicro(short nanosWithinMicro) {
+    if (nanosWithinMicro < 0 || nanosWithinMicro > MAX_NANOS_WITHIN_MICRO) {
+      throw new SparkIllegalArgumentException(
+        "INTERNAL_ERROR",
+        Map.of(
+          "message",
+          "nanosWithinMicro must be in [0, " + MAX_NANOS_WITHIN_MICRO + "], got: "
+            + nanosWithinMicro));
+    }
+  }
+}

common/unsafe/src/test/java/org/apache/spark/unsafe/types/TimestampNanosSuite.java

@@ -0,0 +1,65 @@
+/*
+ * 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.unsafe.types;
+
+import org.apache.spark.SparkIllegalArgumentException;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class TimestampNanosSuite {
+
+  @Test
+  public void timestampNTZNanosEqualsAndHashCode() {
+    TimestampNTZNanos t1 = new TimestampNTZNanos(1000L, (short) 100);
+    TimestampNTZNanos t2 = new TimestampNTZNanos(1001L, (short) 100);
+    TimestampNTZNanos t3 = new TimestampNTZNanos(1000L, (short) 101);
+    TimestampNTZNanos t4 = new TimestampNTZNanos(1000L, (short) 100);
+
+    assertNotEquals(t1, t2);
+    assertNotEquals(t1, t3);
+    assertEquals(t1, t4);
+    assertEquals(t1.hashCode(), t4.hashCode());
+  }
+
+  @Test
+  public void timestampLTZNanosEqualsAndHashCode() {
+    TimestampLTZNanos t1 = new TimestampLTZNanos(2000L, (short) 0);
+    TimestampLTZNanos t2 = new TimestampLTZNanos(2000L, (short) 1);
+    TimestampLTZNanos t3 = new TimestampLTZNanos(2000L, (short) 0);
+
+    assertNotEquals(t1, t2);
+    assertEquals(t1, t3);
+  }
+
+  @Test
+  public void invalidNanosWithinMicroNTZ() {
+    assertThrows(SparkIllegalArgumentException.class,
+      () -> new TimestampNTZNanos(0L, (short) -1));
+    assertThrows(SparkIllegalArgumentException.class,
+      () -> new TimestampNTZNanos(0L, (short) 1000));
+  }
+
+  @Test
+  public void invalidNanosWithinMicroLTZ() {
+    assertThrows(SparkIllegalArgumentException.class,
+      () -> new TimestampLTZNanos(0L, (short) -1));
+    assertThrows(SparkIllegalArgumentException.class,
+      () -> new TimestampLTZNanos(0L, (short) 1000));
+  }
+}

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/SpecializedGetters.java

@@ -23,6 +23,8 @@
 import org.apache.spark.sql.types.Decimal;
 import org.apache.spark.sql.catalyst.util.MapData;
 import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.TimestampLTZNanos;
+import org.apache.spark.unsafe.types.TimestampNTZNanos;
 import org.apache.spark.unsafe.types.UTF8String;
 import org.apache.spark.unsafe.types.VariantVal;
 import org.apache.spark.unsafe.types.GeographyVal;
@@ -58,6 +60,10 @@ public interface SpecializedGetters {
 
   CalendarInterval getInterval(int ordinal);
 
+  TimestampNTZNanos getTimestampNTZNanos(int ordinal);
+
+  TimestampLTZNanos getTimestampLTZNanos(int ordinal);
+
   VariantVal getVariant(int ordinal);
 
   InternalRow getStruct(int ordinal, int numFields);

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/SpecializedGettersReader.java

@@ -72,6 +72,12 @@ public static Object read(
     if (physicalDataType instanceof PhysicalCalendarIntervalType) {
       return obj.getInterval(ordinal);
     }
+    if (physicalDataType instanceof PhysicalTimestampNTZNanosType) {
+      return obj.getTimestampNTZNanos(ordinal);
+    }
+    if (physicalDataType instanceof PhysicalTimestampLTZNanosType) {
+      return obj.getTimestampLTZNanos(ordinal);
+    }
     if (physicalDataType instanceof PhysicalBinaryType) {
       return obj.getBinary(ordinal);
     }

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/TimestampNanosRowValues.java

@@ -0,0 +1,65 @@
+/*
+ * 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.catalyst.expressions;
+
+import org.apache.spark.unsafe.Platform;
+import org.apache.spark.unsafe.types.TimestampLTZNanos;
+import org.apache.spark.unsafe.types.TimestampNTZNanos;
+
+/**
+ * Shared read/write helpers for nanosecond timestamp values in UnsafeRow/UnsafeArrayData.
+ */
+public final class TimestampNanosRowValues {
+  public static final int SIZE_IN_BYTES = TimestampNTZNanos.SIZE_IN_BYTES;
+
+  private TimestampNanosRowValues() {
+  }
+
+  public static void writePayload(
+      Object baseObject, long baseOffset, int cursor, long epochMicros, short nanosWithinMicro) {
+    Platform.putLong(baseObject, baseOffset + cursor, epochMicros);
+    // Store nanos in the low 16 bits; upper 48 bits remain zero.
+    Platform.putLong(baseObject, baseOffset + cursor + 8, ((long) nanosWithinMicro) & 0xFFFFL);
+  }
+
+  public static void zeroPayload(Object baseObject, long baseOffset, int cursor) {
+    Platform.putLong(baseObject, baseOffset + cursor, 0L);
+    Platform.putLong(baseObject, baseOffset + cursor + 8, 0L);
+  }
+
+  public static long readEpochMicros(Object baseObject, long baseOffset, int offset) {
+    return Platform.getLong(baseObject, baseOffset + offset);
+  }
+
+  public static short readNanosWithinMicro(Object baseObject, long baseOffset, int offset) {
+    // Use getLong (not getShort) to match writePayload's putLong; endian-safe.
+    return (short) (Platform.getLong(baseObject, baseOffset + offset + 8) & 0xFFFFL);
+  }
+
+  public static TimestampNTZNanos readNTZ(Object baseObject, long baseOffset, int offset) {
+    return new TimestampNTZNanos(
+        readEpochMicros(baseObject, baseOffset, offset),
+        readNanosWithinMicro(baseObject, baseOffset, offset));
+  }
+
+  public static TimestampLTZNanos readLTZ(Object baseObject, long baseOffset, int offset) {
+    return new TimestampLTZNanos(
+        readEpochMicros(baseObject, baseOffset, offset),
+        readNanosWithinMicro(baseObject, baseOffset, offset));
+  }
+}

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeArrayData.java

@@ -38,6 +38,8 @@
 import org.apache.spark.unsafe.bitset.BitSetMethods;
 import org.apache.spark.unsafe.hash.Murmur3_x86_32;
 import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.TimestampLTZNanos;
+import org.apache.spark.unsafe.types.TimestampNTZNanos;
 import org.apache.spark.unsafe.types.UTF8String;
 import org.apache.spark.unsafe.types.VariantVal;
 import org.apache.spark.unsafe.types.GeographyVal;
@@ -248,6 +250,20 @@ public CalendarInterval getInterval(int ordinal) {
     return new CalendarInterval(months, days, microseconds);
   }
 
+  @Override
+  public TimestampNTZNanos getTimestampNTZNanos(int ordinal) {
+    if (isNullAt(ordinal)) return null;
+    final int offset = (int) (getLong(ordinal) >> 32);
+    return TimestampNanosRowValues.readNTZ(baseObject, baseOffset, offset);
+  }
+
+  @Override
+  public TimestampLTZNanos getTimestampLTZNanos(int ordinal) {
+    if (isNullAt(ordinal)) return null;
+    final int offset = (int) (getLong(ordinal) >> 32);
+    return TimestampNanosRowValues.readLTZ(baseObject, baseOffset, offset);
+  }
+
   @Override
   public VariantVal getVariant(int ordinal) {
     if (isNullAt(ordinal)) return null;