[SPARK-58741][SQL][TESTS] Add nanosecond-timestamp coverage for collect_list

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

@@ -740,6 +740,18 @@ Sort [c#x ASC NULLS FIRST], true
       +- LocalRelation [c#x]
 
 
+-- !query
+SELECT sort_array(collect_list(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'),
+  (CAST(NULL AS timestamp_ltz(9))) AS t(c)
+-- !query analysis
+Aggregate [sort_array(collect_list(c#x, 0, 0, true), true) AS sort_array(collect_list(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,18 @@ Sort [c#x ASC NULLS FIRST], true
       +- LocalRelation [c#x]
 
 
+-- !query
+SELECT sort_array(collect_list(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'),
+  (CAST(NULL AS timestamp_ntz(9))) AS t(c)
+-- !query analysis
+Aggregate [sort_array(collect_list(c#x, 0, 0, true), true) AS sort_array(collect_list(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,17 @@ 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_list over nanosecond-precision TIMESTAMP_LTZ. The buffer holds the full
+-- nanos value, so the sub-microsecond remainder survives and the result element type stays
+-- TIMESTAMP_LTZ(9); values render in the session time zone (America/Los_Angeles). collect_list
+-- order is non-deterministic, so the output is stabilized with sort_array; duplicates are kept and
+-- NULLs are dropped.
+SELECT sort_array(collect_list(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'),
+  (CAST(NULL AS timestamp_ltz(9))) 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,16 @@ 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_list over nanosecond-precision TIMESTAMP_NTZ. The buffer holds the full
+-- nanos value, so the sub-microsecond remainder survives and the result element type stays
+-- TIMESTAMP_NTZ(9). collect_list order is non-deterministic, so the output is stabilized with
+-- sort_array; duplicates are kept and NULLs are dropped.
+SELECT sort_array(collect_list(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'),
+  (CAST(NULL AS timestamp_ntz(9))) 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,18 @@ struct<c:timestamp_ltz(9),count(1):bigint>
 2019-12-31 16:00:00.000000999	1
 
 
+-- !query
+SELECT sort_array(collect_list(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'),
+  (CAST(NULL AS timestamp_ltz(9))) AS t(c)
+-- !query schema
+struct<sort_array(collect_list(c), true):array<timestamp_ltz(9)>>
+-- !query output
+[2019-12-31 16:00:00.000000001,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,18 @@ struct<c:timestamp_ntz(9),count(1):bigint>
 2020-01-01 00:00:00.000000999	1
 
 
+-- !query
+SELECT sort_array(collect_list(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'),
+  (CAST(NULL AS timestamp_ntz(9))) AS t(c)
+-- !query schema
+struct<sort_array(collect_list(c), true):array<timestamp_ntz(9)>>
+-- !query output
+[2020-01-01 00:00:00.000000001,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,53 @@ abstract class TimestampNanosFunctionsSuiteBase extends SharedSparkSession {
     }
   }
 
+  // collect_list over nanosecond-precision timestamps (SPARK-56822). `CollectList` is
+  // type-agnostic: its `ArrayBuffer` buffer holds the physical `TimestampNanosVal` and the
+  // result element type is exactly `child.dataType`, so the input precision, family (NTZ/LTZ)
+  // and sub-microsecond remainder survive with no truncation to micros. The collection order
+  // after aggregation is not deterministic, so the contents are compared as a set.
+
+  test("SPARK-56822: collect_list over nanosecond-precision timestamps preserves type and nanos " +
+    "remainder") {
+    Seq(7, 8, 9).foreach { p =>
+      val schema = new StructType()
+        .add("ntz", TimestampNTZNanosType(p))
+        .add("ltz", TimestampLTZNanosType(p))
+      // The sub-microsecond parts are multiples of 100ns, so they are exact at every p in [7, 9]
+      // (no flooring) yet non-zero -- a value truncated to micros would be visibly wrong. The NULL
+      // row is dropped (collect_list ignores nulls by default).
+      val data = Seq(
+        Row(LocalDateTime.parse("2020-01-01T12:34:56.000000100"),
+          Instant.parse("2020-01-01T12:34:56.000000100Z")),
+        Row(LocalDateTime.parse("2020-01-02T00:00:00.000000900"),
+          Instant.parse("2020-01-02T00:00:00.000000900Z")),
+        Row(null, null))
+      val df = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
+
+      val sqlRes = df.selectExpr("collect_list(ntz)", "collect_list(ltz)")
+      // The result keeps the family (NTZ/LTZ) and precision; nulls are dropped so
+      // containsNull=false.
+      assert(sqlRes.schema.map(_.dataType) === Seq(
+        ArrayType(TimestampNTZNanosType(p), containsNull = false),
+        ArrayType(TimestampLTZNanosType(p), containsNull = false)))
+
+      val expectedNtz = Set(
+        LocalDateTime.parse("2020-01-01T12:34:56.000000100"),
+        LocalDateTime.parse("2020-01-02T00:00:00.000000900"))
+      val expectedLtz = Set(
+        Instant.parse("2020-01-01T12:34:56.000000100Z"),
+        Instant.parse("2020-01-02T00:00:00.000000900Z"))
+      // The Scala Column API path agrees with the SQL path; both drop the null and keep
+      // both values.
+      val sqlRow = sqlRes.collect().head
+      val colRow = df.select(collect_list(col("ntz")), collect_list(col("ltz"))).collect().head
+      assert(sqlRow.getSeq[LocalDateTime](0).toSet === expectedNtz)
+      assert(sqlRow.getSeq[Instant](1).toSet === expectedLtz)
+      assert(colRow.getSeq[LocalDateTime](0).toSet === expectedNtz)
+      assert(colRow.getSeq[Instant](1).toSet === expectedLtz)
+    }
+  }
+
   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