[SPARK-58742][SQL][TESTS] Add nanosecond-timestamp coverage for collect_set

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 sort_array(collect_set(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 [sort_array(collect_set(c#x, 0, 0, true), true) AS sort_array(collect_set(c), true)#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 sort_array(collect_set(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 [sort_array(collect_set(c#x, 0, 0, true), true) AS sort_array(collect_set(c), true)#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,16 @@ 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: collect_set over nanosecond-precision TIMESTAMP_LTZ. It deduplicates on the full
+-- sub-microsecond value: the two .000000001 rows collapse to one, the .000000999 row stays, so the
+-- sorted set has two distinct elements and the element type stays TIMESTAMP_LTZ(9); values render
+-- in the session time zone (America/Los_Angeles). collect_set order is non-deterministic, so the
+-- output is stabilized with sort_array.
+SELECT sort_array(collect_set(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,15 @@ 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: collect_set over nanosecond-precision TIMESTAMP_NTZ. It deduplicates on the full
+-- sub-microsecond value: the two .000000001 rows collapse to one, the .000000999 row stays, so the
+-- sorted set has two distinct elements and the element type stays TIMESTAMP_NTZ(9). collect_set
+-- order is non-deterministic, so the output is stabilized with sort_array.
+SELECT sort_array(collect_set(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 sort_array(collect_set(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<sort_array(collect_set(c), true):array<timestamp_ltz(9)>>
+-- !query output
+[2019-12-31 16:00:00.000000001,2019-12-31 16:00:00.000000999]
+
+
 -- !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 sort_array(collect_set(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<sort_array(collect_set(c), true):array<timestamp_ntz(9)>>
+-- !query output
+[2020-01-01 00:00:00.000000001,2020-01-01 00:00:00.000000999]
+
+
 -- !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,40 @@ abstract class TimestampNanosFunctionsSuiteBase extends SharedSparkSession {
     }
   }
 
+  // collect_set over nanosecond-precision timestamps (SPARK-56822). `CollectSet` deduplicates via a
+  // `HashSet` keyed on the physical `TimestampNanosVal`, whose `equals`/`hashCode` cover the full
+  // `(epochMicros, nanosWithinMicro)` pair, so sub-microsecond-distinct values are kept distinct;
+  // the result element type is exactly `child.dataType`.
+
+  test("SPARK-56822: collect_set over nanos deduplicates on the full sub-microsecond value") {
+    // Two values share the microsecond and differ only in the last nanosecond digit. Flooring the
+    // input to precision `p` collapses them when p < 9 (the distinguishing digit is below the
+    // grid), but keeps them apart at p = 9. collect_set must dedup on the full stored value,
+    // not on micros.
+    Seq(7, 8, 9).foreach { p =>
+      val schema = new StructType().add("ntz", TimestampNTZNanosType(p))
+      val data = Seq(
+        Row(LocalDateTime.parse("2020-01-01T12:34:56.123456780")),
+        Row(LocalDateTime.parse("2020-01-01T12:34:56.123456789")),
+        Row(LocalDateTime.parse("2020-01-01T12:34:56.123456780")))
+      val df = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+
+      val res = df.selectExpr("collect_set(ntz)")
+      assert(res.schema.head.dataType === ArrayType(TimestampNTZNanosType(p), containsNull = false))
+
+      // Values are floored to `p` on ingestion, so the distinct set depends on `p`.
+      val expected = p match {
+        case 7 => Set(LocalDateTime.parse("2020-01-01T12:34:56.123456700"))
+        case 8 => Set(LocalDateTime.parse("2020-01-01T12:34:56.123456780"))
+        case _ => Set(
+          LocalDateTime.parse("2020-01-01T12:34:56.123456780"),
+          LocalDateTime.parse("2020-01-01T12:34:56.123456789"))
+      }
+      val collected = res.collect().head.getSeq[LocalDateTime](0).toSet
+      assert(collected === expected, s"collect_set(p=$p) expected $expected, got $collected")
+    }
+  }
+
   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