fix(snowflake): transpile REDUCE to DuckDB LIST_REDUCE (#7505)

* fix(snowflake): transpile REDUCE to DuckDB LIST_REDUCE

* Add comment explaining finish parameter fallback

* refactor(duckdb): use @unsupported_args decorator for REDUCE finish parameter

Replace manual fallback check with @unsupported_args decorator as suggested in PR review.

* Sync w/ integration tests

* Fix tests

---------

Co-authored-by: George Sittas <giwrgos.sittas@gmail.com>

Author: fivetran-ashashankar

sqlglot-integration-tests

@@ -2332,6 +2332,17 @@ def randstr_sql(self, expression: exp.Randstr) -> str:
         replacements = {"seed": seed_value, "length": length}
         return f"({self.sql(exp.replace_placeholders(self.RANDSTR_TEMPLATE, **replacements))})"
 
+    @unsupported_args("finish")
+    def reduce_sql(self, expression: exp.Reduce) -> str:
+        array_arg = expression.this
+        initial_value = expression.args.get("initial")
+        merge_lambda = expression.args.get("merge")
+
+        if merge_lambda:
+            merge_lambda.set("colon", True)
+
+        return self.func("list_reduce", array_arg, merge_lambda, initial_value)
+
     def zipf_sql(self, expression: exp.Zipf) -> str:
         """
         Transpile Snowflake's ZIPF to DuckDB using CDF-based inverse sampling.

sqlglot/generators/duckdb.py

@@ -1399,7 +1399,6 @@ def test_array(self):
             "REDUCE(x, 0, (acc, x) -> acc + x, acc -> acc)",
             write={
                 "trino": "REDUCE(x, 0, (acc, x) -> acc + x, acc -> acc)",
-                "duckdb": "REDUCE(x, 0, (acc, x) -> acc + x, acc -> acc)",
                 "hive": "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)",

tests/dialects/test_dialect.py

@@ -786,7 +786,6 @@ def test_spark(self):
             "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)",