[SPARK-53109][SQL] Support TIME in the make_timestamp_ntz and try_make_timestamp_ntz functions in Scala

sql/api/src/main/scala/org/apache/spark/sql/functions.scala

@@ -8598,6 +8598,15 @@ object functions {
       secs: Column): Column =
     Column.fn("make_timestamp_ntz", years, months, days, hours, mins, secs)
 
+  /**
+   * Create a local date-time from date and time fields.
+   *
+   * @group datetime_funcs
+   * @since 4.1.0
+   */
+  def make_timestamp_ntz(date: Column, time: Column): Column =
+    Column.fn("make_timestamp_ntz", date, time)
+
   /**
    * Try to create a local date-time from years, months, days, hours, mins, secs fields. The
    * function returns NULL on invalid inputs.
@@ -8614,6 +8623,15 @@ object functions {
       secs: Column): Column =
     Column.fn("try_make_timestamp_ntz", years, months, days, hours, mins, secs)
 
+  /**
+   * Try to create a local date-time from date and time fields.
+   *
+   * @group datetime_funcs
+   * @since 4.1.0
+   */
+  def try_make_timestamp_ntz(date: Column, time: Column): Column =
+    Column.fn("try_make_timestamp_ntz", date, time)
+
   /**
    * Make year-month interval from years, months.
    *

sql/core/src/test/scala/org/apache/spark/sql/TimeFunctionsSuiteBase.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql
 
+import java.time.LocalDate
 import java.time.LocalTime
 import java.time.temporal.ChronoUnit
 
@@ -142,6 +143,54 @@ abstract class TimeFunctionsSuiteBase extends QueryTest with SharedSparkSession
     checkAnswer(result2, expected)
   }
 
+  test("SPARK-53109: make_timestamp_ntz function") {
+    // Input data for the function.
+    val schema = StructType(Seq(
+      StructField("date", DateType, nullable = false),
+      StructField("time", TimeType(), nullable = false)
+    ))
+    val data = Seq(
+      Row(LocalDate.parse("2020-01-01"), LocalTime.parse("00:00:00")),
+      Row(LocalDate.parse("2023-10-20"), LocalTime.parse("12:34:56")),
+      Row(LocalDate.parse("2023-12-31"), LocalTime.parse("23:59:59.999999"))
+    )
+    val df = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+
+    // Test the function using both `selectExpr` and `select`.
+    val result1 = df.selectExpr(
+      "make_timestamp_ntz(date, time)"
+    )
+    val result2 = df.select(
+      make_timestamp_ntz(col("date"), col("time"))
+    )
+    // Check that both methods produce the same result.
+    checkAnswer(result1, result2)
+
+    // Expected output of the function.
+    val expected = Seq(
+      "2020-01-01 00:00:00",
+      "2023-10-20 12:34:56",
+      "2023-12-31 23:59:59.999999"
+    ).toDF("timestamp_ntz").select(col("timestamp_ntz").cast("timestamp_ntz"))
+    // Check that the results match the expected output.
+    checkAnswer(result1, expected)
+    checkAnswer(result2, expected)
+
+    // NULL result is returned for any NULL input.
+    val nullInputDF = Seq(
+      (null, LocalTime.parse("00:00:00")),
+      (LocalDate.parse("2020-01-01"), null),
+      (null, null)
+    ).toDF("date", "time")
+    val nullResult = Seq[Integer](
+      null, null, null
+    ).toDF("ts").select(col("ts"))
+    checkAnswer(
+      nullInputDF.select(make_timestamp_ntz(col("date"), col("time"))),
+      nullResult
+    )
+  }
+
   test("SPARK-52885: hour function") {
     // Input data for the function.
     val schema = StructType(Seq(
@@ -328,6 +377,54 @@ abstract class TimeFunctionsSuiteBase extends QueryTest with SharedSparkSession
     )
   }
 
+  test("SPARK-53109: try_make_timestamp_ntz function") {
+    // Input data for the function.
+    val schema = StructType(Seq(
+      StructField("date", DateType, nullable = false),
+      StructField("time", TimeType(), nullable = false)
+    ))
+    val data = Seq(
+      Row(LocalDate.parse("2020-01-01"), LocalTime.parse("00:00:00")),
+      Row(LocalDate.parse("2023-10-20"), LocalTime.parse("12:34:56")),
+      Row(LocalDate.parse("2023-12-31"), LocalTime.parse("23:59:59.999999"))
+    )
+    val df = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+
+    // Test the function using both `selectExpr` and `select`.
+    val result1 = df.selectExpr(
+      "try_make_timestamp_ntz(date, time)"
+    )
+    val result2 = df.select(
+      try_make_timestamp_ntz(col("date"), col("time"))
+    )
+    // Check that both methods produce the same result.
+    checkAnswer(result1, result2)
+
+    // Expected output of the function.
+    val expected = Seq(
+      "2020-01-01 00:00:00",
+      "2023-10-20 12:34:56",
+      "2023-12-31 23:59:59.999999"
+    ).toDF("timestamp_ntz").select(col("timestamp_ntz").cast("timestamp_ntz"))
+    // Check that the results match the expected output.
+    checkAnswer(result1, expected)
+    checkAnswer(result2, expected)
+
+    // NULL result is returned for any NULL input.
+    val nullInputDF = Seq(
+      (null, LocalTime.parse("00:00:00")),
+      (LocalDate.parse("2020-01-01"), null),
+      (null, null)
+    ).toDF("date", "time")
+    val nullResult = Seq[Integer](
+      null, null, null
+    ).toDF("ts").select(col("ts"))
+    checkAnswer(
+      nullInputDF.select(try_make_timestamp_ntz(col("date"), col("time"))),
+      nullResult
+    )
+  }
+
   test("SPARK-52884: try_to_time function without format") {
     // Input data for the function.
     val schema = StructType(Seq(