[SPARK-58743][SQL][TESTS] Add nanosecond-timestamp coverage for mode

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

@@ -740,6 +740,17 @@ Sort [c#x ASC NULLS FIRST], true
       +- LocalRelation [c#x]
 
 
+-- !query
+SELECT mode(c) FROM VALUES
+  (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC'),
+  (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000999 UTC'),
+  (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC') AS t(c)
+-- !query analysis
+Aggregate [mode(c#x, 0, 0, None) AS mode(c)#x]
++- SubqueryAlias t
+   +- LocalRelation [c#x]
+
+
 -- !query
 SELECT unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789')
 -- !query analysis

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

@@ -668,6 +668,17 @@ Sort [c#x ASC NULLS FIRST], true
       +- LocalRelation [c#x]
 
 
+-- !query
+SELECT mode(c) FROM VALUES
+  (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001'),
+  (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000999'),
+  (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001') AS t(c)
+-- !query analysis
+Aggregate [mode(c#x, 0, 0, None) AS mode(c)#x]
++- SubqueryAlias t
+   +- LocalRelation [c#x]
+
+
 -- !query
 SELECT unix_timestamp(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789')
 -- !query analysis

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

@@ -220,6 +220,15 @@ SELECT c, count(*) FROM VALUES
   (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC') AS t(c)
   GROUP BY c ORDER BY c;
 
+-- SPARK-56822: mode over nanosecond-precision TIMESTAMP_LTZ. Frequencies are counted on the full
+-- nanos value, so the most-frequent value is selected down to the sub-microsecond and the result
+-- type stays TIMESTAMP_LTZ(9); the value renders in the session time zone (America/Los_Angeles).
+-- .000000001 appears twice, .000000999 once.
+SELECT mode(c) FROM VALUES
+  (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC'),
+  (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000999 UTC'),
+  (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC') AS t(c);
+
 -- 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

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

@@ -193,6 +193,14 @@ SELECT c, count(*) FROM VALUES
   (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001') AS t(c)
   GROUP BY c ORDER BY c;
 
+-- SPARK-56822: mode over nanosecond-precision TIMESTAMP_NTZ. Frequencies are counted on the full
+-- nanos value, so the most-frequent value is selected down to the sub-microsecond and the result
+-- type stays TIMESTAMP_NTZ(9). .000000001 appears twice, .000000999 once.
+SELECT mode(c) FROM VALUES
+  (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001'),
+  (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000999'),
+  (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001') AS t(c);
+
 -- 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.

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

@@ -832,6 +832,17 @@ struct<c:timestamp_ltz(9),count(1):bigint>
 2019-12-31 16:00:00.000000999	1
 
 
+-- !query
+SELECT mode(c) FROM VALUES
+  (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC'),
+  (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000999 UTC'),
+  (TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC') AS t(c)
+-- !query schema
+struct<mode(c):timestamp_ltz(9)>
+-- !query output
+2019-12-31 16:00:00.000000001
+
+
 -- !query
 SELECT unix_timestamp(TIMESTAMP_LTZ '2020-01-01 13:24:35.123456789')
 -- !query schema

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

@@ -750,6 +750,17 @@ struct<c:timestamp_ntz(9),count(1):bigint>
 2020-01-01 00:00:00.000000999	1
 
 
+-- !query
+SELECT mode(c) FROM VALUES
+  (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001'),
+  (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000999'),
+  (TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001') AS t(c)
+-- !query schema
+struct<mode(c):timestamp_ntz(9)>
+-- !query output
+2020-01-01 00:00:00.000000001
+
+
 -- !query
 SELECT unix_timestamp(TIMESTAMP_NTZ '2020-01-01 13:24:35.123456789')
 -- !query schema

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

@@ -710,6 +710,36 @@ abstract class TimestampNanosFunctionsSuiteBase extends SharedSparkSession {
     }
   }
 
+  // mode over nanosecond-precision timestamps (SPARK-56822). `Mode` counts frequencies in an
+  // `OpenHashMap` keyed on the physical `TimestampNanosVal` (its `equals`/`hashCode` cover the full
+  // `(epochMicros, nanosWithinMicro)` pair) and returns `child.dataType`, so the most-frequent
+  // value is selected on the full nanos value and its precision and family (NTZ/LTZ) are preserved.
+
+  test("SPARK-56822: mode over nanosecond-precision timestamps returns the most frequent value") {
+    Seq(7, 8, 9).foreach { p =>
+      val schema = new StructType()
+        .add("ntz", TimestampNTZNanosType(p))
+        .add("ltz", TimestampLTZNanosType(p))
+      // The frequent value (3 rows) and the rare one (1 row) differ only within the microsecond, so
+      // frequency counting must key on the full nanos value; a NULL row is ignored. The fractions
+      // are multiples of 100ns, exact at every p in [7, 9]. There is a unique most-frequent value,
+      // so the result is deterministic without a WITHIN GROUP / deterministic argument.
+      val ldtHot = LocalDateTime.parse("2020-01-01T00:00:00.000000100")
+      val ldtCold = LocalDateTime.parse("2020-01-01T00:00:00.000000900")
+      val insHot = Instant.parse("2020-01-01T00:00:00.000000100Z")
+      val insCold = Instant.parse("2020-01-01T00:00:00.000000900Z")
+      val data = Seq(
+        Row(ldtHot, insHot), Row(ldtHot, insHot), Row(ldtHot, insHot),
+        Row(ldtCold, insCold), Row(null, null))
+      val df = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+
+      val res = df.selectExpr("mode(ntz)", "mode(ltz)")
+      // The result keeps the family (NTZ/LTZ) and precision of the input.
+      assert(res.schema.map(_.dataType) === Seq(TimestampNTZNanosType(p), TimestampLTZNanosType(p)))
+      checkAnswer(res, Row(ldtHot, insHot))
+    }
+  }
+
   test("SPARK-57816: date_format / to_char / to_varchar over nanosecond-precision timestamps") {
     // The 9-`S` pattern is a fixed-width fraction field, so it always emits 9 digits; truncating to
     // precision `p` zeros the low digits (floor); it does not drop them. The session zone is