Preserve input field nullability in ArrayAgg return field

datafusion/core/tests/dataframe/mod.rs

@@ -4801,7 +4801,7 @@ async fn unnest_with_redundant_columns() -> Result<()> {
         @r"
     Projection: shapes.shape_id [shape_id:UInt32]
       Unnest: lists[shape_id2|depth=1] structs[] [shape_id:UInt32, shape_id2:UInt32;N]
-        Aggregate: groupBy=[[shapes.shape_id]], aggr=[[array_agg(shapes.shape_id) AS shape_id2]] [shape_id:UInt32, shape_id2:List(Field { data_type: UInt32, nullable: true });N]
+        Aggregate: groupBy=[[shapes.shape_id]], aggr=[[array_agg(shapes.shape_id) AS shape_id2]] [shape_id:UInt32, shape_id2:List(Field { data_type: UInt32 });N]
           TableScan: shapes projection=[shape_id] [shape_id:UInt32]
     "
     );

datafusion/core/tests/sql/aggregates/basic.rs

@@ -35,11 +35,12 @@ async fn csv_query_array_agg_distinct() -> Result<()> {
     //    | [4, 2, 3, 5, 1]                          |
     //    +------------------------------------------+
     // Since ARRAY_AGG(DISTINCT) ordering is nondeterministic, check the schema and contents.
+    // The inner field nullability matches the input column c2 which is NOT NULL
     assert_eq!(
         *actual[0].schema(),
         Schema::new(vec![Field::new_list(
             "array_agg(DISTINCT aggregate_test_100.c2)",
-            Field::new_list_field(DataType::UInt32, true),
+            Field::new_list_field(DataType::UInt32, false),
             true
         ),])
     );

datafusion/functions-aggregate/benches/array_agg.rs

@@ -45,7 +45,7 @@ fn merge_batch_bench(c: &mut Criterion, name: &str, values: ArrayRef) {
         b.iter(|| {
             #[allow(clippy::unit_arg)]
             black_box(
-                ArrayAggAccumulator::try_new(&list_item_data_type, false)
+                ArrayAggAccumulator::try_new(&list_item_data_type, false, true)
                     .unwrap()
                     .merge_batch(std::slice::from_ref(&values))
                     .unwrap(),