fix: Implement AggregateUDFImpl::reverse_expr for StringAgg (apache#17165) (apache#17473)

* fix: Implement AggregateUDFImpl::reverse_expr for StringAgg

* Add a test with two invocations of aggregateion

---------

Co-authored-by: Nuno Faria <nunofpfaria@gmail.com>

Author: alamb

datafusion/functions-aggregate/src/string_agg.rs

@@ -178,6 +178,10 @@ impl AggregateUDFImpl for StringAgg {
         )))
     }
 
+    fn reverse_expr(&self) -> datafusion_expr::ReversedUDAF {
+        datafusion_expr::ReversedUDAF::Reversed(string_agg_udaf())
+    }
+
     fn documentation(&self) -> Option<&Documentation> {
         self.doc()
     }

datafusion/sqllogictest/test_files/aggregate.slt

@@ -6203,6 +6203,58 @@ from t;
 ----
 a,c,d,b
 
+# Test explain / reverse_expr for string_agg
+query TT
+explain select string_agg(k, ',' order by v) from t;
+----
+logical_plan
+01)Aggregate: groupBy=[[]], aggr=[[string_agg(t.k, Utf8(",")) ORDER BY [t.v ASC NULLS LAST]]]     
+02)--TableScan: t projection=[k, v]
+physical_plan
+01)AggregateExec: mode=Single, gby=[], aggr=[string_agg(t.k,Utf8(",")) ORDER BY [t.v ASC NULLS LAST]]
+02)--SortExec: expr=[v@1 ASC NULLS LAST], preserve_partitioning=[false]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+
+query T
+select string_agg(k, ',' order by v) from t;
+----
+c,a,b,d
+
+query TT
+explain select string_agg(k, ',' order by v desc) from t;
+----
+logical_plan
+01)Aggregate: groupBy=[[]], aggr=[[string_agg(t.k, Utf8(",")) ORDER BY [t.v DESC NULLS FIRST]]]   
+02)--TableScan: t projection=[k, v]
+physical_plan
+01)AggregateExec: mode=Single, gby=[], aggr=[string_agg(t.k,Utf8(",")) ORDER BY [t.v DESC NULLS FIRST]]
+02)--SortExec: expr=[v@1 DESC], preserve_partitioning=[false]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+
+query T
+select string_agg(k, ',' order by v desc) from t;
+----
+d,b,a,c
+
+# Call string_agg with both ASC and DESC orderings, and expect only one sort
+# (because the aggregate can handle reversed inputs)
+query TT
+explain select string_agg(k, ',' order by v asc), string_agg(k, ',' order by v desc) from t;
+----
+logical_plan
+01)Aggregate: groupBy=[[]], aggr=[[string_agg(t.k, Utf8(",")) ORDER BY [t.v ASC NULLS LAST], string_agg(t.k, Utf8(",")) ORDER BY [t.v DESC NULLS FIRST]]]
+02)--TableScan: t projection=[k, v]
+physical_plan
+01)AggregateExec: mode=Single, gby=[], aggr=[string_agg(t.k,Utf8(",")) ORDER BY [t.v ASC NULLS LAST], string_agg(t.k,Utf8(",")) ORDER BY [t.v DESC NULLS FIRST]]
+02)--SortExec: expr=[v@1 ASC NULLS LAST], preserve_partitioning=[false]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+
+query TT
+select string_agg(k, ',' order by v asc), string_agg(k, ',' order by v desc) from t;
+----
+c,a,b,d d,b,a,c
+
+
 statement ok
 drop table t;
 
@@ -7444,4 +7496,3 @@ NULL NULL
 
 statement ok
 drop table distinct_avg;
-