[SPARK-52460][SQL] Store internal TIME values as nanoseconds

### What changes were proposed in this pull request?
In the PR, I propose to store internal TIME values as nanoseconds in `Long` since the midnight instead of microseconds in `Long`.

### Why are the changes needed?
`Long` with nanoseconds precision can keep the full range of TIME values from 0 to 24 * 60 * 60 * 1000 * 1000 * 1000 - 1 which is less than maximum of Long. This will simplify support of `TIME(9)`, for example.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
By running the affected test suites:
```
$ build/sbt "sql/testOnly org.apache.spark.sql.SQLQueryTestSuite"
$ build/sbt "test:testOnly *TimeExpressionsSuite"
$ build/sbt "test:testOnly *TimeFormatterSuite"
```

### Was this patch authored or co-authored using generative AI tooling?
No.

Closes #51156 from MaxGekk/time-nanos.

Authored-by: Max Gekk <[email protected]>
Signed-off-by: Max Gekk <[email protected]>

Author: MaxGekk

sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala

@@ -19,7 +19,7 @@ package org.apache.spark.sql.catalyst.util
 import java.lang.invoke.{MethodHandles, MethodType}
 import java.sql.{Date, Timestamp}
 import java.time.{Instant, LocalDate, LocalDateTime, LocalTime, ZonedDateTime, ZoneId, ZoneOffset}
-import java.time.temporal.ChronoField.MICRO_OF_DAY
+import java.time.temporal.ChronoField.{MICRO_OF_DAY, NANO_OF_DAY}
 import java.util.TimeZone
 import java.util.concurrent.TimeUnit.{MICROSECONDS, NANOSECONDS}
 import java.util.regex.Pattern
@@ -83,6 +83,12 @@ trait SparkDateTimeUtils {
     case ldt: LocalDateTime => localDateTimeToMicros(ldt)
   }
 
+  /**
+   * Converts the time to microseconds since midnight. In Spark time values have nanoseconds
+   * precision, so this conversion is lossy.
+   */
+  def nanosToMicros(nanos: Long): Long = Math.floorDiv(nanos, MICROS_PER_MILLIS)
+
   /**
    * Converts the timestamp to milliseconds since epoch. In Spark timestamp values have
    * microseconds precision, so this conversion is lossy.
@@ -101,6 +107,11 @@ trait SparkDateTimeUtils {
     Math.multiplyExact(millis, MICROS_PER_MILLIS)
   }
 
+  /**
+   * Converts microseconds since the midnight to nanoseconds.
+   */
+  def microsToNanos(micros: Long): Long = Math.multiplyExact(micros, NANOS_PER_MICROS)
+
   // See issue SPARK-35679
   // min second cause overflow in instant to micro
   private val MIN_SECONDS = Math.floorDiv(Long.MinValue, MICROS_PER_SECOND)
@@ -225,17 +236,15 @@ trait SparkDateTimeUtils {
   }
 
   /**
-   * Converts the local time to the number of microseconds within the day, from 0 to (24 * 60 * 60
-   * * 1000000) - 1.
+   * Converts the local time to the number of nanoseconds within the day, from 0 to (24 * 60 * 60
+   * * 1000 * 1000 * 1000) - 1.
    */
-  def localTimeToMicros(localTime: LocalTime): Long = localTime.getLong(MICRO_OF_DAY)
+  def localTimeToNanos(localTime: LocalTime): Long = localTime.getLong(NANO_OF_DAY)
 
   /**
-   * Converts the number of microseconds within the day to the local time.
+   * Converts the number of nanoseconds within the day to the local time.
    */
-  def microsToLocalTime(micros: Long): LocalTime = {
-    LocalTime.ofNanoOfDay(Math.multiplyExact(micros, NANOS_PER_MICROS))
-  }
+  def nanosToLocalTime(nanos: Long): LocalTime = LocalTime.ofNanoOfDay(nanos)
 
   /**
    * Converts a local date at the default JVM time zone to the number of days since 1970-01-01 in
@@ -716,7 +725,7 @@ trait SparkDateTimeUtils {
       }
       val nanoseconds = MICROSECONDS.toNanos(segments(6))
       val localTime = LocalTime.of(segments(3), segments(4), segments(5), nanoseconds.toInt)
-      Some(localTimeToMicros(localTime))
+      Some(localTimeToNanos(localTime))
     } catch {
       case NonFatal(_) => None
     }

sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/TimeFormatter.scala

@@ -25,11 +25,11 @@ import org.apache.spark.sql.catalyst.util.SparkDateTimeUtils._
 import org.apache.spark.unsafe.types.UTF8String
 
 sealed trait TimeFormatter extends Serializable {
-  def parse(s: String): Long // returns microseconds since midnight
+  def parse(s: String): Long // returns nanoseconds since midnight
 
   def format(localTime: LocalTime): String
-  // Converts microseconds since the midnight to time string
-  def format(micros: Long): String
+  // Converts nanoseconds since the midnight to time string
+  def format(nanos: Long): String
 
   def validatePatternString(): Unit
 }
@@ -47,15 +47,15 @@ class Iso8601TimeFormatter(pattern: String, locale: Locale, isParsing: Boolean)
 
   override def parse(s: String): Long = {
     val localTime = toLocalTime(formatter.parse(s))
-    localTimeToMicros(localTime)
+    localTimeToNanos(localTime)
   }
 
   override def format(localTime: LocalTime): String = {
     localTime.format(formatter)
   }
 
-  override def format(micros: Long): String = {
-    format(microsToLocalTime(micros))
+  override def format(nanos: Long): String = {
+    format(nanosToLocalTime(nanos))
   }
 
   override def validatePatternString(): Unit = {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/CatalystTypeConverters.scala

@@ -375,14 +375,14 @@ object CatalystTypeConverters {
 
   private object TimeConverter extends CatalystTypeConverter[LocalTime, LocalTime, Any] {
     override def toCatalystImpl(scalaValue: LocalTime): Long = {
-      DateTimeUtils.localTimeToMicros(scalaValue)
+      DateTimeUtils.localTimeToNanos(scalaValue)
     }
     override def toScala(catalystValue: Any): LocalTime = {
       if (catalystValue == null) null
-      else DateTimeUtils.microsToLocalTime(catalystValue.asInstanceOf[Long])
+      else DateTimeUtils.nanosToLocalTime(catalystValue.asInstanceOf[Long])
     }
     override def toScalaImpl(row: InternalRow, column: Int): LocalTime =
-      DateTimeUtils.microsToLocalTime(row.getLong(column))
+      DateTimeUtils.nanosToLocalTime(row.getLong(column))
   }
 
   private object TimestampConverter extends CatalystTypeConverter[Any, Timestamp, Any] {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/DeserializerBuildHelper.scala

@@ -160,7 +160,7 @@ object DeserializerBuildHelper {
     StaticInvoke(
       DateTimeUtils.getClass,
       ObjectType(classOf[java.time.LocalTime]),
-      "microsToLocalTime",
+      "nanosToLocalTime",
       path :: Nil,
       returnNullable = false)
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala

@@ -103,7 +103,7 @@ object SerializerBuildHelper {
     StaticInvoke(
       DateTimeUtils.getClass,
       TimeType(),
-      "localTimeToMicros",
+      "localTimeToNanos",
       inputObject :: Nil,
       returnNullable = false)
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala

@@ -49,7 +49,7 @@ import org.apache.spark.sql.catalyst.trees.TreePattern
 import org.apache.spark.sql.catalyst.trees.TreePattern.{LITERAL, NULL_LITERAL, TRUE_OR_FALSE_LITERAL}
 import org.apache.spark.sql.catalyst.types._
 import org.apache.spark.sql.catalyst.util._
-import org.apache.spark.sql.catalyst.util.DateTimeUtils.{instantToMicros, localTimeToMicros}
+import org.apache.spark.sql.catalyst.util.DateTimeUtils.{instantToMicros, localTimeToNanos}
 import org.apache.spark.sql.catalyst.util.IntervalStringStyles.ANSI_STYLE
 import org.apache.spark.sql.catalyst.util.IntervalUtils.{durationToMicros, periodToMonths, toDayTimeIntervalString, toYearMonthIntervalString}
 import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
@@ -89,7 +89,7 @@ object Literal {
     case l: LocalDateTime => Literal(DateTimeUtils.localDateTimeToMicros(l), TimestampNTZType)
     case ld: LocalDate => Literal(ld.toEpochDay.toInt, DateType)
     case d: Date => Literal(DateTimeUtils.fromJavaDate(d), DateType)
-    case lt: LocalTime => Literal(localTimeToMicros(lt), TimeType())
+    case lt: LocalTime => Literal(localTimeToNanos(lt), TimeType())
     case d: Duration => Literal(durationToMicros(d), DayTimeIntervalType())
     case p: Period => Literal(periodToMonths(p), YearMonthIntervalType())
     case a: Array[Byte] => Literal(a, BinaryType)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala

@@ -109,7 +109,7 @@ object DateTimeUtils extends SparkDateTimeUtils {
    * Returns the hour value of a given TIME (TimeType) value.
    */
   def getHoursOfTime(micros: Long): Int = {
-    microsToLocalTime(micros).getHour
+    nanosToLocalTime(micros).getHour
   }
 
   /**
@@ -124,7 +124,7 @@ object DateTimeUtils extends SparkDateTimeUtils {
    * Returns the minute value of a given TIME (TimeType) value.
    */
   def getMinutesOfTime(micros: Long): Int = {
-    microsToLocalTime(micros).getMinute
+    nanosToLocalTime(micros).getMinute
   }
 
   /**
@@ -139,7 +139,7 @@ object DateTimeUtils extends SparkDateTimeUtils {
    * Returns the second value of a given TIME (TimeType) value.
    */
   def getSecondsOfTime(micros: Long): Int = {
-    microsToLocalTime(micros).getSecond
+    nanosToLocalTime(micros).getSecond
   }
   /**
    * Returns the seconds part and its fractional part with microseconds.
@@ -151,16 +151,16 @@ object DateTimeUtils extends SparkDateTimeUtils {
 
   /**
    * Returns the second value with fraction from a given TIME (TimeType) value.
-   * @param micros
-   *   The number of microseconds since the epoch.
+   * @param nanos
+   *   The number of nanoseconds since the epoch.
    * @param precision
    *   The time fractional seconds precision, which indicates the number of decimal digits
    *   maintained.
    */
-  def getSecondsOfTimeWithFraction(micros: Long, precision: Int): Decimal = {
-    val seconds = (micros / MICROS_PER_SECOND) % SECONDS_PER_MINUTE
+  def getSecondsOfTimeWithFraction(nanos: Long, precision: Int): Decimal = {
+    val seconds = (nanos / NANOS_PER_SECOND) % SECONDS_PER_MINUTE
     val scaleFactor = math.pow(10, precision).toLong
-    val scaledFraction = (micros % MICROS_PER_SECOND) * scaleFactor / MICROS_PER_SECOND
+    val scaledFraction = (nanos % NANOS_PER_SECOND) * scaleFactor / NANOS_PER_SECOND
     val fraction = scaledFraction.toDouble / scaleFactor
     Decimal(seconds + fraction, 8, 6)
   }
@@ -816,7 +816,7 @@ object DateTimeUtils extends SparkDateTimeUtils {
 
       val nanos = Math.floorMod(unscaledSecFrac, MICROS_PER_SECOND) * NANOS_PER_MICROS
       val lt = LocalTime.of(hours, minutes, fullSecs.toInt, nanos.toInt)
-      localTimeToMicros(lt)
+      localTimeToNanos(lt)
     } catch {
       case e @ (_: DateTimeException | _: ArithmeticException) =>
         throw QueryExecutionErrors.ansiDateTimeArgumentOutOfRangeWithoutSuggestion(e)

sql/catalyst/src/test/scala/org/apache/spark/sql/RandomDataGenerator.scala

@@ -292,7 +292,7 @@ object RandomDataGenerator {
         randomNumeric[LocalTime](
           rand,
           (rand: Random) => {
-            DateTimeUtils.microsToLocalTime(rand.between(0, 24 * 60 * 60 * 1000 * 1000L))
+            DateTimeUtils.nanosToLocalTime(rand.between(0, 24 * 60 * 60 * 1000 * 1000L))
           },
           specialTimes.map(LocalTime.parse)
         )

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/CatalystTypeConvertersSuite.scala

@@ -435,7 +435,7 @@ class CatalystTypeConvertersSuite extends SparkFunSuite with SQLHelper {
       "23:59:59.999999").foreach { time =>
       val input = LocalTime.parse(time)
       val result = CatalystTypeConverters.convertToCatalyst(input)
-      val expected = DateTimeUtils.localTimeToMicros(input)
+      val expected = DateTimeUtils.localTimeToNanos(input)
       assert(result === expected)
     }
   }
@@ -449,7 +449,7 @@ class CatalystTypeConvertersSuite extends SparkFunSuite with SQLHelper {
       43200999999L,
       86399000000L,
       86399999999L).foreach { us =>
-      val localTime = DateTimeUtils.microsToLocalTime(us)
+      val localTime = DateTimeUtils.nanosToLocalTime(us)
       assert(CatalystTypeConverters.createToScalaConverter(TimeType())(us) === localTime)
     }
   }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/RowEncoderSuite.scala

@@ -377,7 +377,7 @@ class RowEncoderSuite extends CodegenInterpretedPlanTest {
     val encoder = ExpressionEncoder(schema).resolveAndBind()
     val localTime = java.time.LocalTime.parse("20:38:45.123456")
     val row = toRow(encoder, Row(localTime))
-    assert(row.getLong(0) === DateTimeUtils.localTimeToMicros(localTime))
+    assert(row.getLong(0) === DateTimeUtils.localTimeToNanos(localTime))
     val readback = fromRow(encoder, row)
     assert(readback.get(0).equals(localTime))
   }