[SPARK-57528][SQL] Support nanosecond-precision timestamps in the unix_timestamp function

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

@@ -1429,6 +1429,8 @@ abstract class ToTimestamp
           daysToMicros(t.asInstanceOf[Int], zoneId) / downScaleFactor
         case TimestampType | TimestampNTZType =>
           t.asInstanceOf[Long] / downScaleFactor
+        case _: AnyTimestampNanoType =>
+          t.asInstanceOf[TimestampNanosVal].epochMicros / downScaleFactor
         case _: StringType =>
           val fmt = right.eval(input)
           if (fmt == null) {
@@ -1521,6 +1523,15 @@ abstract class ToTimestamp
           if (!${ev.isNull}) {
             ${ev.value} = ${eval1.value} / $downScaleFactor;
           }""")
+      case _: AnyTimestampNanoType =>
+        val eval1 = left.genCode(ctx)
+        ev.copy(code = code"""
+          ${eval1.code}
+          boolean ${ev.isNull} = ${eval1.isNull};
+          $javaType ${ev.value} = ${CodeGenerator.defaultValue(dataType)};
+          if (!${ev.isNull}) {
+            ${ev.value} = ${eval1.value}.epochMicros / $downScaleFactor;
+          }""")
       case DateType =>
         val zid = ctx.addReferenceObj("zoneId", zoneId, classOf[ZoneId].getName)
         val dtu = DateTimeUtils.getClass.getName.stripSuffix("$")
@@ -1538,6 +1549,19 @@ abstract class ToTimestamp
 
 abstract class UnixTime extends ToTimestamp {
   override val downScaleFactor: Long = MICROS_PER_SECOND
+
+  // In addition to the input types accepted by the base `ToTimestamp`, the timestamp-argument
+  // form of `unix_timestamp` / `to_unix_timestamp` also accepts the nanosecond-precision
+  // timestamp types. The result stays whole-second BIGINT, so the sub-second digits are dropped.
+  override def inputTypes: Seq[AbstractDataType] =
+    Seq(
+      TypeCollection(
+        StringTypeWithCollation(supportsTrimCollation = true),
+        DateType,
+        TimestampType,
+        TimestampNTZType,
+        AnyTimestampNanoType),
+      StringTypeWithCollation(supportsTrimCollation = true))
 }
 
 /**

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

@@ -1067,6 +1067,51 @@ class DateExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
       ToUnixTimestamp(Literal("2015-07-24"), Literal("\""), UTC_OPT) :: Nil)
   }
 
+  test("SPARK-57528: unix_timestamp / to_unix_timestamp over nanosecond-precision timestamps") {
+    import org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils._
+    val fmt = Literal("yyyy-MM-dd HH:mm:ss")
+
+    // A post-epoch value with non-zero sub-second digits: 2008-12-25 15:30:00.123456789 ->
+    // 1230219000. unix_timestamp does not apply a zone shift, so the NTZ wall-clock value and the
+    // LTZ instant at the same UTC reading produce the same whole-second result.
+    val ntz = localDateTimeToNanosVal(timestampNTZ(2008, 12, 25, 15, 30, 0, 123456789))
+    val ltz = instantToNanosVal(Instant.parse("2008-12-25T15:30:00.123456789Z"))
+    foreachNanosPrecision { p =>
+      checkEvaluation(
+        UnixTimestamp(Literal.create(ntz, TimestampNTZNanosType(p)), fmt, UTC_OPT), 1230219000L)
+      checkEvaluation(
+        ToUnixTimestamp(Literal.create(ntz, TimestampNTZNanosType(p)), fmt, UTC_OPT), 1230219000L)
+      checkEvaluation(
+        UnixTimestamp(Literal.create(ltz, TimestampLTZNanosType(p)), fmt, UTC_OPT), 1230219000L)
+      checkEvaluation(
+        ToUnixTimestamp(Literal.create(ltz, TimestampLTZNanosType(p)), fmt, UTC_OPT), 1230219000L)
+    }
+
+    // The format is ignored for the timestamp-argument form: a deliberately invalid format still
+    // yields the same whole-second result, since the timestamp branch never consults the format.
+    val badFmt = Literal("not-a-format")
+    foreachNanosPrecision { p =>
+      checkEvaluation(
+        UnixTimestamp(Literal.create(ntz, TimestampNTZNanosType(p)), badFmt, UTC_OPT),
+        1230219000L)
+      checkEvaluation(
+        ToUnixTimestamp(Literal.create(ltz, TimestampLTZNanosType(p)), badFmt, UTC_OPT),
+        1230219000L)
+    }
+
+    // Pre-epoch sub-second value: 1969-12-31 23:59:59.5 has epochMicros -500000, which divides to
+    // 0 (truncation toward zero), matching the existing microsecond-timestamp behavior.
+    val preEpoch = localDateTimeToNanosVal(timestampNTZ(1969, 12, 31, 23, 59, 59, 500000000))
+    checkEvaluation(
+      UnixTimestamp(Literal.create(preEpoch, TimestampNTZNanosType(9)), fmt, UTC_OPT), 0L)
+
+    // NULL input.
+    checkEvaluation(
+      UnixTimestamp(Literal.create(null, TimestampNTZNanosType(9)), fmt, UTC_OPT), null)
+    checkEvaluation(
+      ToUnixTimestamp(Literal.create(null, TimestampLTZNanosType(9)), fmt, UTC_OPT), null)
+  }
+
   test("datediff") {
     checkEvaluation(
       DateDiff(Literal(Date.valueOf("2015-07-24")), Literal(Date.valueOf("2015-07-21"))), 3)

sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ltz-nanos.sql.out

@@ -647,3 +647,52 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException
     "fragment" : "TIMESTAMP_LTZ '2020-01-02 03:04:05.123456789 UTC' + INTERVAL '1' MONTH"
   } ]
 }
+
+
+-- !query
+SELECT unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789')
+-- !query analysis
+Project [unix_timestamp(2020-01-01 13:24:35.123456789, yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789', yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation
+
+
+-- !query
+SELECT to_unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789')
+-- !query analysis
+Project [to_unix_timestamp(2020-01-01 13:24:35.123456789, yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS to_unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789', yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation
+
+
+-- !query
+SELECT unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789 UTC')
+-- !query analysis
+Project [unix_timestamp(2020-01-01 05:24:35.123456789, yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS unix_timestamp(TIMESTAMP_LTZ '2020-01-01 05:24:35.123456789', yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation
+
+
+-- !query
+SELECT to_unix_timestamp('2020-01-01 13:24:35.000000001 UTC' :: timestamp_ltz(9))
+-- !query analysis
+Project [to_unix_timestamp(cast(2020-01-01 13:24:35.000000001 UTC as timestamp_ltz(9)), yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS to_unix_timestamp(CAST(2020-01-01 13:24:35.000000001 UTC AS TIMESTAMP_LTZ(9)), yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation
+
+
+-- !query
+SELECT unix_timestamp('2020-01-01 13:24:35.999999999' :: timestamp_ltz(7))
+-- !query analysis
+Project [unix_timestamp(cast(2020-01-01 13:24:35.999999999 as timestamp_ltz(7)), yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS unix_timestamp(CAST(2020-01-01 13:24:35.999999999 AS TIMESTAMP_LTZ(7)), yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation
+
+
+-- !query
+SELECT unix_timestamp(TIMESTAMP_LTZ '1969-12-31 23:59:59.500000000 UTC')
+-- !query analysis
+Project [unix_timestamp(1969-12-31 15:59:59.5, yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS unix_timestamp(TIMESTAMP_LTZ '1969-12-31 15:59:59.500000000', yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation
+
+
+-- !query
+SELECT unix_timestamp(NULL :: timestamp_ltz(9)), to_unix_timestamp(NULL :: timestamp_ltz(9))
+-- !query analysis
+Project [unix_timestamp(cast(null as timestamp_ltz(9)), yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS unix_timestamp(CAST(NULL AS TIMESTAMP_LTZ(9)), yyyy-MM-dd HH:mm:ss)#xL, to_unix_timestamp(cast(null as timestamp_ltz(9)), yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS to_unix_timestamp(CAST(NULL AS TIMESTAMP_LTZ(9)), yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation

sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ntz-nanos.sql.out

@@ -574,3 +574,45 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException
     "fragment" : "TIMESTAMP_NTZ '2020-01-02 03:04:05.123456789' + INTERVAL '1' MONTH"
   } ]
 }
+
+
+-- !query
+SELECT unix_timestamp(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789')
+-- !query analysis
+Project [unix_timestamp(2020-01-01 13:24:35.123456789, yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS unix_timestamp(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789', yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation
+
+
+-- !query
+SELECT to_unix_timestamp(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789')
+-- !query analysis
+Project [to_unix_timestamp(2020-01-01 13:24:35.123456789, yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS to_unix_timestamp(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789', yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation
+
+
+-- !query
+SELECT unix_timestamp('2020-01-01 13:24:35.999999999' :: timestamp_ntz(7))
+-- !query analysis
+Project [unix_timestamp(cast(2020-01-01 13:24:35.999999999 as timestamp_ntz(7)), yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS unix_timestamp(CAST(2020-01-01 13:24:35.999999999 AS TIMESTAMP_NTZ(7)), yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation
+
+
+-- !query
+SELECT to_unix_timestamp('2020-01-01 13:24:35.000000001' :: timestamp_ntz(9))
+-- !query analysis
+Project [to_unix_timestamp(cast(2020-01-01 13:24:35.000000001 as timestamp_ntz(9)), yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS to_unix_timestamp(CAST(2020-01-01 13:24:35.000000001 AS TIMESTAMP_NTZ(9)), yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation
+
+
+-- !query
+SELECT unix_timestamp(TIMESTAMP_NTZ '1969-12-31 23:59:59.500000000')
+-- !query analysis
+Project [unix_timestamp(1969-12-31 23:59:59.5, yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS unix_timestamp(TIMESTAMP_NTZ '1969-12-31 23:59:59.500000000', yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation
+
+
+-- !query
+SELECT unix_timestamp(NULL :: timestamp_ntz(9)), to_unix_timestamp(NULL :: timestamp_ntz(9))
+-- !query analysis
+Project [unix_timestamp(cast(null as timestamp_ntz(9)), yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS unix_timestamp(CAST(NULL AS TIMESTAMP_NTZ(9)), yyyy-MM-dd HH:mm:ss)#xL, to_unix_timestamp(cast(null as timestamp_ntz(9)), yyyy-MM-dd HH:mm:ss, Some(America/Los_Angeles), true) AS to_unix_timestamp(CAST(NULL AS TIMESTAMP_NTZ(9)), yyyy-MM-dd HH:mm:ss)#xL]
++- OneRowRelation

sql/core/src/test/resources/sql-tests/inputs/timestamp-ltz-nanos.sql

@@ -173,3 +173,17 @@ SELECT TIMESTAMP_LTZ '1960-01-02 03:04:05.123456789 UTC' +
 -- operator overload.
 SELECT TIMESTAMP_LTZ '2020-01-02 03:04:05.123456789 UTC' + make_interval(0, 1, 0, 2, 0, 0, 0);
 SELECT TIMESTAMP_LTZ '2020-01-02 03:04:05.123456789 UTC' + INTERVAL '1' MONTH;
+
+-- SPARK-57528: unix_timestamp / to_unix_timestamp over nanosecond-precision values. The result is
+-- whole-second BIGINT; the sub-second digits are dropped. A literal without an explicit zone is
+-- read in the session time zone (America/Los_Angeles, UTC-08:00); an explicit-zone literal fixes
+-- the instant directly.
+SELECT unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789');
+SELECT to_unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789');
+SELECT unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789 UTC');
+SELECT to_unix_timestamp('2020-01-01 13:24:35.000000001 UTC' :: timestamp_ltz(9));
+SELECT unix_timestamp('2020-01-01 13:24:35.999999999' :: timestamp_ltz(7));
+-- Pre-epoch value exercises the negative-epoch path (truncation toward zero).
+SELECT unix_timestamp(TIMESTAMP_LTZ '1969-12-31 23:59:59.500000000 UTC');
+-- NULL nanosecond timestamp.
+SELECT unix_timestamp(NULL :: timestamp_ltz(9)), to_unix_timestamp(NULL :: timestamp_ltz(9));

sql/core/src/test/resources/sql-tests/inputs/timestamp-ntz-nanos.sql

@@ -149,3 +149,15 @@ SELECT TIMESTAMP_NTZ '1960-01-02 03:04:05.123456789' + INTERVAL '0 00:00:00.0000
 -- operator overload.
 SELECT TIMESTAMP_NTZ '2020-01-02 03:04:05.123456789' + make_interval(0, 1, 0, 2, 0, 0, 0);
 SELECT TIMESTAMP_NTZ '2020-01-02 03:04:05.123456789' + INTERVAL '1' MONTH;
+
+-- SPARK-57528: unix_timestamp / to_unix_timestamp over nanosecond-precision values. The result is
+-- whole-second BIGINT; the sub-second digits are dropped and NTZ applies no zone shift, so the
+-- wall-clock value is read as the epoch instant.
+SELECT unix_timestamp(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789');
+SELECT to_unix_timestamp(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789');
+SELECT unix_timestamp('2020-01-01 13:24:35.999999999' :: timestamp_ntz(7));
+SELECT to_unix_timestamp('2020-01-01 13:24:35.000000001' :: timestamp_ntz(9));
+-- Pre-epoch value exercises the negative-epoch path (truncation toward zero).
+SELECT unix_timestamp(TIMESTAMP_NTZ '1969-12-31 23:59:59.500000000');
+-- NULL nanosecond timestamp.
+SELECT unix_timestamp(NULL :: timestamp_ntz(9)), to_unix_timestamp(NULL :: timestamp_ntz(9));

sql/core/src/test/resources/sql-tests/results/timestamp-ltz-nanos.sql.out

@@ -726,3 +726,59 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException
     "fragment" : "TIMESTAMP_LTZ '2020-01-02 03:04:05.123456789 UTC' + INTERVAL '1' MONTH"
   } ]
 }
+
+
+-- !query
+SELECT unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789')
+-- !query schema
+struct<unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789', yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+1577913875
+
+
+-- !query
+SELECT to_unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789')
+-- !query schema
+struct<to_unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789', yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+1577913875
+
+
+-- !query
+SELECT unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789 UTC')
+-- !query schema
+struct<unix_timestamp(TIMESTAMP_LTZ '2020-01-01 05:24:35.123456789', yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+1577885075
+
+
+-- !query
+SELECT to_unix_timestamp('2020-01-01 13:24:35.000000001 UTC' :: timestamp_ltz(9))
+-- !query schema
+struct<to_unix_timestamp(CAST(2020-01-01 13:24:35.000000001 UTC AS TIMESTAMP_LTZ(9)), yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+1577885075
+
+
+-- !query
+SELECT unix_timestamp('2020-01-01 13:24:35.999999999' :: timestamp_ltz(7))
+-- !query schema
+struct<unix_timestamp(CAST(2020-01-01 13:24:35.999999999 AS TIMESTAMP_LTZ(7)), yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+1577913875
+
+
+-- !query
+SELECT unix_timestamp(TIMESTAMP_LTZ '1969-12-31 23:59:59.500000000 UTC')
+-- !query schema
+struct<unix_timestamp(TIMESTAMP_LTZ '1969-12-31 15:59:59.500000000', yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+0
+
+
+-- !query
+SELECT unix_timestamp(NULL :: timestamp_ltz(9)), to_unix_timestamp(NULL :: timestamp_ltz(9))
+-- !query schema
+struct<unix_timestamp(CAST(NULL AS TIMESTAMP_LTZ(9)), yyyy-MM-dd HH:mm:ss):bigint,to_unix_timestamp(CAST(NULL AS TIMESTAMP_LTZ(9)), yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+NULL	NULL

sql/core/src/test/resources/sql-tests/results/timestamp-ntz-nanos.sql.out

@@ -644,3 +644,51 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException
     "fragment" : "TIMESTAMP_NTZ '2020-01-02 03:04:05.123456789' + INTERVAL '1' MONTH"
   } ]
 }
+
+
+-- !query
+SELECT unix_timestamp(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789')
+-- !query schema
+struct<unix_timestamp(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789', yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+1577885075
+
+
+-- !query
+SELECT to_unix_timestamp(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789')
+-- !query schema
+struct<to_unix_timestamp(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789', yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+1577885075
+
+
+-- !query
+SELECT unix_timestamp('2020-01-01 13:24:35.999999999' :: timestamp_ntz(7))
+-- !query schema
+struct<unix_timestamp(CAST(2020-01-01 13:24:35.999999999 AS TIMESTAMP_NTZ(7)), yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+1577885075
+
+
+-- !query
+SELECT to_unix_timestamp('2020-01-01 13:24:35.000000001' :: timestamp_ntz(9))
+-- !query schema
+struct<to_unix_timestamp(CAST(2020-01-01 13:24:35.000000001 AS TIMESTAMP_NTZ(9)), yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+1577885075
+
+
+-- !query
+SELECT unix_timestamp(TIMESTAMP_NTZ '1969-12-31 23:59:59.500000000')
+-- !query schema
+struct<unix_timestamp(TIMESTAMP_NTZ '1969-12-31 23:59:59.500000000', yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+0
+
+
+-- !query
+SELECT unix_timestamp(NULL :: timestamp_ntz(9)), to_unix_timestamp(NULL :: timestamp_ntz(9))
+-- !query schema
+struct<unix_timestamp(CAST(NULL AS TIMESTAMP_NTZ(9)), yyyy-MM-dd HH:mm:ss):bigint,to_unix_timestamp(CAST(NULL AS TIMESTAMP_NTZ(9)), yyyy-MM-dd HH:mm:ss):bigint>
+-- !query output
+NULL	NULL

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

@@ -239,6 +239,54 @@ abstract class TimestampNanosFunctionsSuiteBase extends SharedSparkSession {
         Row(null, null, null, null))
     }
   }
+
+  test("SPARK-57528: unix_timestamp / to_unix_timestamp over nanosecond-precision timestamps") {
+    // unix_timestamp returns whole-second BIGINT and applies no zone shift to a timestamp
+    // argument, so the sub-second digits are dropped and the nanos result equals the
+    // microsecond-timestamp result.
+    val ntzStr = "2020-01-01T13:24:35.123456789"
+    val ltzStr = "2020-01-01T21:24:35.987654321Z"
+    Seq(7, 8, 9).foreach { p =>
+      val ntzNano = ntzNanos(ntzStr, p)
+      val ntzMicro = spark.createDataFrame(
+        spark.sparkContext.parallelize(Seq(Row(LocalDateTime.parse(ntzStr)))),
+        new StructType().add("c", TimestampNTZType))
+      checkAnswer(
+        ntzNano.select(unix_timestamp(col("c")), to_unix_timestamp(col("c"))),
+        ntzMicro.select(unix_timestamp(col("c")), to_unix_timestamp(col("c"))))
+      // 2020-01-01 13:24:35 read as the wall-clock instant -> 1577885075 epoch seconds.
+      checkAnswer(
+        ntzNano.select(unix_timestamp(col("c")), to_unix_timestamp(col("c"))),
+        Row(1577885075L, 1577885075L))
+
+      val ltzNano = ltzNanos(ltzStr, p)
+      val ltzMicro = spark.createDataFrame(
+        spark.sparkContext.parallelize(Seq(Row(Instant.parse(ltzStr)))),
+        new StructType().add("c", TimestampType))
+      checkAnswer(
+        ltzNano.select(unix_timestamp(col("c")), to_unix_timestamp(col("c"))),
+        ltzMicro.select(unix_timestamp(col("c")), to_unix_timestamp(col("c"))))
+      // 2020-01-01 21:24:35 UTC -> 1577913875 epoch seconds.
+      checkAnswer(
+        ltzNano.select(unix_timestamp(col("c")), to_unix_timestamp(col("c"))),
+        Row(1577913875L, 1577913875L))
+    }
+  }
+
+  test("SPARK-57528: unix_timestamp / to_unix_timestamp over NULL nanosecond timestamps") {
+    Seq(7, 8, 9).foreach { p =>
+      val ntz = spark.createDataFrame(
+        spark.sparkContext.parallelize(Seq(Row(null))),
+        new StructType().add("c", TimestampNTZNanosType(p)))
+      val ltz = spark.createDataFrame(
+        spark.sparkContext.parallelize(Seq(Row(null))),
+        new StructType().add("c", TimestampLTZNanosType(p)))
+      checkAnswer(
+        ntz.select(unix_timestamp(col("c")), to_unix_timestamp(col("c"))), Row(null, null))
+      checkAnswer(
+        ltz.select(unix_timestamp(col("c")), to_unix_timestamp(col("c"))), Row(null, null))
+    }
+  }
 }
 
 // Runs the nanosecond timestamp function tests with ANSI mode enabled explicitly.