[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,86 @@
+/*
+ * 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 fractional precision
+ * ({@code p} in [7, 9]). Analogous to microsecond {@code TimestampType}, which stores a single
+ * {@code long} epoch micros value at UTC+00:00, but with sub-microsecond resolution.
+ *
+ * <p>Values are stored as two components:
+ * <ul>
+ *   <li>{@link #epochMicros} - microseconds since the Unix epoch at UTC+00:00 (same unit as
+ *   {@code TimestampType}),</li>
+ *   <li>{@link #nanosWithinMicro} - additional nanoseconds within that microsecond, in [0, 999].
+ *   The SQL fractional-second precision {@code p} applies when converting to and from external
+ *   representations; the physical value always retains up to three sub-micro digits.</li>
+ * </ul>
+ *
+ * <p>Logical row-size estimation uses 10 bytes (8 + 2). In {@code UnsafeRow}, values are stored in
+ * the variable-length region using a 16-byte payload (see
+ * {@link org.apache.spark.sql.catalyst.expressions.TimestampNanosRowValues}), the same pattern as
+ * {@link CalendarInterval}.
+ *
+ * <p>NTZ and LTZ nanosecond timestamps share this composite layout at the row layer; time-zone
+ * semantics are enforced in the SQL and conversion layers, not in this class.
+ *
+ * @since 4.2.0
+ */
+@Unstable
+public final class TimestampLTZNanos implements Serializable {
+  /** Size of the {@code UnsafeRow} variable-length payload for this type (two 8-byte words). */
+  public static final int SIZE_IN_BYTES = 16;
+
+  /** Microseconds since the Unix epoch at UTC+00:00. */
+  public final long epochMicros;
+  /** Nanoseconds within {@link #epochMicros}, in [0, 999]. */
+  public final short nanosWithinMicro;
+
+  /**
+   * @param epochMicros microseconds since the Unix epoch at UTC+00:00
+   * @param nanosWithinMicro nanoseconds within {@code epochMicros}, must be in [0, 999]
+   */
+  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,86 @@
+/*
+ * 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 fractional precision
+ * ({@code p} in [7, 9]). Analogous to microsecond {@code TimestampNTZType}, which stores a single
+ * {@code long} epoch micros value, but with sub-microsecond resolution.
+ *
+ * <p>Values are stored as two components:
+ * <ul>
+ *   <li>{@link #epochMicros} - microseconds since the Unix epoch in the proleptic Gregorian
+ *   calendar (same unit as {@code TimestampNTZType}),</li>
+ *   <li>{@link #nanosWithinMicro} - additional nanoseconds within that microsecond, in [0, 999].
+ *   The SQL fractional-second precision {@code p} applies when converting to and from external
+ *   representations; the physical value always retains up to three sub-micro digits.</li>
+ * </ul>
+ *
+ * <p>Logical row-size estimation uses 10 bytes (8 + 2). In {@code UnsafeRow}, values are stored in
+ * the variable-length region using a 16-byte payload (see
+ * {@link org.apache.spark.sql.catalyst.expressions.TimestampNanosRowValues}), the same pattern as
+ * {@link CalendarInterval}.
+ *
+ * <p>NTZ and LTZ nanosecond timestamps share this composite layout at the row layer; time-zone
+ * semantics are enforced in the SQL and conversion layers, not in this class.
+ *
+ * @since 4.2.0
+ */
+@Unstable
+public final class TimestampNTZNanos implements Serializable {
+  /** Size of the {@code UnsafeRow} variable-length payload for this type (two 8-byte words). */
+  public static final int SIZE_IN_BYTES = 16;
+
+  /** Microseconds since the Unix epoch. */
+  public final long epochMicros;
+  /** Nanoseconds within {@link #epochMicros}, in [0, 999]. */
+  public final short nanosWithinMicro;
+
+  /**
+   * @param epochMicros microseconds since the Unix epoch
+   * @param nanosWithinMicro nanoseconds within {@code epochMicros}, must be in [0, 999]
+   */
+  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,43 @@
+/*
+ * 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;
+
+/**
+ * Internal helpers shared by {@link TimestampNTZNanos} and {@link TimestampLTZNanos}.
+ */
+final class TimestampNanosUtils {
+  /** Maximum valid value for {@code nanosWithinMicro} (three sub-micro decimal digits). */
+  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/api/src/main/scala/org/apache/spark/sql/types/TimestampLTZNanosType.scala

@@ -44,6 +44,10 @@ case class TimestampLTZNanosType(precision: Int) extends DatetimeType {
   /**
    * Default size used by Spark for row-size estimation. Values are represented logically as epoch
    * microseconds (Long, 8 bytes) plus nanoseconds within that micro (Short, 2 bytes).
+   *
+   * In [[org.apache.spark.sql.catalyst.expressions.UnsafeRow]], the physical payload is 16 bytes
+   * in the variable-length region (two 8-byte words); see
+   * [[org.apache.spark.sql.catalyst.expressions.TimestampNanosRowValues]].
    */
   override def defaultSize: Int = 10
 

sql/api/src/main/scala/org/apache/spark/sql/types/TimestampNTZNanosType.scala

@@ -44,6 +44,10 @@ case class TimestampNTZNanosType(precision: Int) extends DatetimeType {
   /**
    * Default size used by Spark for row-size estimation. Values are represented logically as epoch
    * microseconds (Long, 8 bytes) plus nanoseconds within that micro (Short, 2 bytes).
+   *
+   * In [[org.apache.spark.sql.catalyst.expressions.UnsafeRow]], the physical payload is 16 bytes
+   * in the variable-length region (two 8-byte words); see
+   * [[org.apache.spark.sql.catalyst.expressions.TimestampNanosRowValues]].
    */
   override def defaultSize: Int = 10
 

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,12 @@ public interface SpecializedGetters {
 
   CalendarInterval getInterval(int ordinal);
 
+  /** Nanosecond NTZ timestamp; see {@link TimestampNanosRowValues} for UnsafeRow layout. */
+  TimestampNTZNanos getTimestampNTZNanos(int ordinal);
+
+  /** Nanosecond LTZ timestamp; see {@link TimestampNanosRowValues} for UnsafeRow layout. */
+  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);
     }