Repository: tobymao/sqlglot

Issue:
Spark's AGGREGATE is not transpiled to Trino's REDUCE
```python
# sqlglot == 10.5.10
sqlglot.transpile("aggregate(my_arr, 0, (acc,x) -> acc + x, s -> s * 2)", read="spark", write="presto")[0]
```
Current output:
```
'AGGREGATE(my_arr, 0, (acc, x) -> acc + x, s -> s * 2)'
```
Expected output:
```
'REDUCE(my_arr, 0, (acc, x) -> acc + x, s -> s * 2)'
```


Discussion:

Patch:
diff --git a/sqlglot/dataframe/sql/functions.py b/sqlglot/dataframe/sql/functions.py
index a141fe46..01492893 100644
--- a/sqlglot/dataframe/sql/functions.py
+++ b/sqlglot/dataframe/sql/functions.py
@@ -1144,10 +1144,16 @@ def aggregate(
     merge_exp = _get_lambda_from_func(merge)
     if finish is not None:
         finish_exp = _get_lambda_from_func(finish)
-        return Column.invoke_anonymous_function(
-            col, "AGGREGATE", initialValue, Column(merge_exp), Column(finish_exp)
+        return Column.invoke_expression_over_column(
+            col,
+            glotexp.Reduce,
+            initial=initialValue,
+            merge=Column(merge_exp),
+            finish=Column(finish_exp),
         )
-    return Column.invoke_anonymous_function(col, "AGGREGATE", initialValue, Column(merge_exp))
+    return Column.invoke_expression_over_column(
+        col, glotexp.Reduce, initial=initialValue, merge=Column(merge_exp)
+    )
 
 
 def transform(
diff --git a/sqlglot/dialects/spark.py b/sqlglot/dialects/spark.py
index f20c4dc1..08ff775d 100644
--- a/sqlglot/dialects/spark.py
+++ b/sqlglot/dialects/spark.py
@@ -73,6 +73,7 @@ class Spark(Hive):
             ),
             "APPROX_PERCENTILE": exp.ApproxQuantile.from_arg_list,
             "IIF": exp.If.from_arg_list,
+            "AGGREGATE": exp.Reduce.from_arg_list,
         }
 
         FUNCTION_PARSERS = {
diff --git a/sqlglot/expressions.py b/sqlglot/expressions.py
index 135c49f2..1a1c6cbb 100644
--- a/sqlglot/expressions.py
+++ b/sqlglot/expressions.py
@@ -3248,7 +3248,7 @@ class ReadCSV(Func):
 
 
 class Reduce(Func):
-    arg_types = {"this": True, "initial": True, "merge": True, "finish": True}
+    arg_types = {"this": True, "initial": True, "merge": True, "finish": False}
 
 
 class RegexpLike(Func):


Test patch:
diff --git a/tests/dialects/test_dialect.py b/tests/dialects/test_dialect.py
index 446247c3..4267a4dc 100644
--- a/tests/dialects/test_dialect.py
+++ b/tests/dialects/test_dialect.py
@@ -715,6 +715,7 @@ class TestDialect(Validator):
                 "hive": "REDUCE(x, 0, (acc, x) -> acc + x, acc -> acc)",
                 "presto": "REDUCE(x, 0, (acc, x) -> acc + x, acc -> acc)",
                 "spark": "AGGREGATE(x, 0, (acc, x) -> acc + x, acc -> acc)",
+                "presto": "REDUCE(x, 0, (acc, x) -> acc + x, acc -> acc)",
             },
         )
 
diff --git a/tests/dialects/test_spark.py b/tests/dialects/test_spark.py
index 02d43aaa..be74a27f 100644
--- a/tests/dialects/test_spark.py
+++ b/tests/dialects/test_spark.py
@@ -212,6 +212,17 @@ TBLPROPERTIES (
         self.validate_identity("TRIM(BOTH 'SL' FROM 'SSparkSQLS')")
         self.validate_identity("TRIM(LEADING 'SL' FROM 'SSparkSQLS')")
         self.validate_identity("TRIM(TRAILING 'SL' FROM 'SSparkSQLS')")
+
+        self.validate_all(
+            "AGGREGATE(my_arr, 0, (acc, x) -> acc + x, s -> s * 2)",
+            write={
+                "trino": "REDUCE(my_arr, 0, (acc, x) -> acc + x, s -> s * 2)",
+                "duckdb": "REDUCE(my_arr, 0, (acc, x) -> acc + x, s -> s * 2)",
+                "hive": "REDUCE(my_arr, 0, (acc, x) -> acc + x, s -> s * 2)",
+                "presto": "REDUCE(my_arr, 0, (acc, x) -> acc + x, s -> s * 2)",
+                "spark": "AGGREGATE(my_arr, 0, (acc, x) -> acc + x, s -> s * 2)",
+            },
+        )
         self.validate_all(
             "TRIM('SL', 'SSparkSQLS')", write={"spark": "TRIM('SL' FROM 'SSparkSQLS')"}
         )