[SPARK-51421][SQL] Get seconds of TIME datatype

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala

@@ -648,7 +648,7 @@ object FunctionRegistry {
     expression[NextDay]("next_day"),
     expression[Now]("now"),
     expression[Quarter]("quarter"),
-    expression[Second]("second"),
+    expressionBuilder("second", SecondExpressionBuilder),
     expression[ParseToTimestamp]("to_timestamp"),
     expression[ParseToDate]("to_date"),
     expression[ToTime]("to_time"),

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

@@ -25,7 +25,7 @@ import org.apache.spark.sql.catalyst.util.DateTimeUtils
 import org.apache.spark.sql.catalyst.util.TimeFormatter
 import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
 import org.apache.spark.sql.internal.types.StringTypeWithCollation
-import org.apache.spark.sql.types.{AbstractDataType, IntegerType, ObjectType, TimeType}
+import org.apache.spark.sql.types.{AbstractDataType, IntegerType, ObjectType, TimeType, TypeCollection}
 import org.apache.spark.unsafe.types.UTF8String
 
 /**
@@ -290,3 +290,62 @@ object HourExpressionBuilder extends ExpressionBuilder {
     }
   }
 }
+
+case class SecondsOfTime(child: Expression)
+  extends RuntimeReplaceable
+    with ExpectsInputTypes {
+
+  override def replacement: Expression = StaticInvoke(
+    classOf[DateTimeUtils.type],
+    IntegerType,
+    "getSecondsOfTime",
+    Seq(child),
+    Seq(child.dataType)
+  )
+
+  override def inputTypes: Seq[AbstractDataType] =
+    Seq(TypeCollection(TimeType.MIN_PRECISION to TimeType.MAX_PRECISION map TimeType: _*))
+
+  override def children: Seq[Expression] = Seq(child)
+
+  override def prettyName: String = "second"
+
+  override protected def withNewChildrenInternal(
+    newChildren: IndexedSeq[Expression]): Expression = {
+      copy(child = newChildren.head)
+  }
+}
+
+@ExpressionDescription(
+  usage = """
+    _FUNC_(expr) - Returns the second component of the given expression.
+
+    If `expr` is a TIMESTAMP or a string that can be cast to timestamp,
+    it returns the second of that timestamp.
+    If `expr` is a TIME type (since 4.1.0), it returns the second of the time-of-day.
+  """,
+  examples = """
+    Examples:
+      > SELECT _FUNC_('2018-02-14 12:58:59');
+       59
+      > SELECT _FUNC_(TIME'13:25:59.999999');
+       59
+  """,
+  since = "1.5.0",
+  group = "datetime_funcs")
+object SecondExpressionBuilder extends ExpressionBuilder {
+  override def build(name: String, expressions: Seq[Expression]): Expression = {
+    if (expressions.isEmpty) {
+      throw QueryCompilationErrors.wrongNumArgsError(name, Seq("> 0"), expressions.length)
+    } else {
+      val child = expressions.head
+      child.dataType match {
+        case _: TimeType =>
+          SecondsOfTime(child)
+        case _ =>
+          Second(child)
+      }
+    }
+  }
+}
+

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

@@ -135,6 +135,12 @@ object DateTimeUtils extends SparkDateTimeUtils {
     getLocalDateTime(micros, zoneId).getSecond
   }
 
+  /**
+   * Returns the second value of a given TIME (TimeType) value.
+   */
+  def getSecondsOfTime(micros: Long): Int = {
+    microsToLocalTime(micros).getSecond
+  }
   /**
    * Returns the seconds part and its fractional part with microseconds.
    */

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/TimeExpressionsSuite.scala

@@ -168,4 +168,62 @@ class TimeExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
     checkConsistencyBetweenInterpretedAndCodegen(
       (child: Expression) => MinutesOfTime(child).replacement, TimeType())
   }
+
+  test("SecondExpressionBuilder") {
+    // Empty expressions list
+    checkError(
+      exception = intercept[AnalysisException] {
+        SecondExpressionBuilder.build("second", Seq.empty)
+      },
+      condition = "WRONG_NUM_ARGS.WITHOUT_SUGGESTION",
+      parameters = Map(
+        "functionName" -> "`second`",
+        "expectedNum" -> "> 0",
+        "actualNum" -> "0",
+        "docroot" -> SPARK_DOC_ROOT)
+    )
+
+    // test TIME-typed child should build SecondsOfTime
+    val timeExpr = Literal(localTime(12, 58, 59), TimeType())
+    val builtExprForTime = SecondExpressionBuilder.build("second", Seq(timeExpr))
+    assert(builtExprForTime.isInstanceOf[SecondsOfTime])
+    assert(builtExprForTime.asInstanceOf[SecondsOfTime].child eq timeExpr)
+
+    // test non TIME-typed child should build second
+    val tsExpr = Literal("2007-09-03 10:45:23")
+    val builtExprForTs = SecondExpressionBuilder.build("second", Seq(tsExpr))
+    assert(builtExprForTs.isInstanceOf[Second])
+    assert(builtExprForTs.asInstanceOf[Second].child eq tsExpr)
+  }
+
+  test("Second with TIME type") {
+    // A few test times in microseconds since midnight:
+    //   time in microseconds -> expected second
+    val testTimes = Seq(
+      localTime() -> 0,
+      localTime(1) -> 0,
+      localTime(0, 59) -> 0,
+      localTime(14, 30) -> 0,
+      localTime(12, 58, 59) -> 59,
+      localTime(23, 0, 1) -> 1,
+      localTime(23, 59, 59, 999999) -> 59
+    )
+
+    // Create a literal with TimeType() for each test microsecond value
+    // evaluate SecondsOfTime(...), and check that the result matches the expected second.
+    testTimes.foreach { case (micros, expectedSecond) =>
+      checkEvaluation(
+        SecondsOfTime(Literal(micros, TimeType())),
+        expectedSecond)
+    }
+
+    // Verify NULL handling
+    checkEvaluation(
+      SecondsOfTime(Literal.create(null, TimeType(TimeType.MICROS_PRECISION))),
+      null
+    )
+
+    checkConsistencyBetweenInterpretedAndCodegen(
+      (child: Expression) => SecondsOfTime(child).replacement, TimeType())
+  }
 }

sql/core/src/test/resources/sql-functions/sql-expression-schema.md

@@ -286,7 +286,7 @@
 | org.apache.spark.sql.catalyst.expressions.SchemaOfJson | schema_of_json | SELECT schema_of_json('[{"col":0}]') | struct<schema_of_json([{"col":0}]):string> |
 | org.apache.spark.sql.catalyst.expressions.SchemaOfXml | schema_of_xml | SELECT schema_of_xml('<p><a>1</a></p>') | struct<schema_of_xml(<p><a>1</a></p>):string> |
 | org.apache.spark.sql.catalyst.expressions.Sec | sec | SELECT sec(0) | struct<SEC(0):double> |
-| org.apache.spark.sql.catalyst.expressions.Second | second | SELECT second('2009-07-30 12:58:59') | struct<second(2009-07-30 12:58:59):int> |
+| org.apache.spark.sql.catalyst.expressions.SecondExpressionBuilder | second | SELECT second('2018-02-14 12:58:59') | struct<second(2018-02-14 12:58:59):int> |
 | org.apache.spark.sql.catalyst.expressions.SecondsToTimestamp | timestamp_seconds | SELECT timestamp_seconds(1230219000) | struct<timestamp_seconds(1230219000):timestamp> |
 | org.apache.spark.sql.catalyst.expressions.Sentences | sentences | SELECT sentences('Hi there! Good morning.') | struct<sentences(Hi there! Good morning., , ):array<array<string>>> |
 | org.apache.spark.sql.catalyst.expressions.Sequence | sequence | SELECT sequence(1, 5) | struct<sequence(1, 5):array<int>> |