Repository: tobymao/sqlglot

Issue:
Spark And Duck DB Different Behavior for ARRAY_AGG
In spark sql, when using array_agg, it filters out null values. DuckDB keeps them which causes different results.
DuckDB recommends using filters to filter out null values.

**Fully reproducible code snippet**
spark code:
```python
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

# Initialize SparkSession
spark = SparkSession.builder.appName("ArrayAggExample").getOrCreate()

# Define schema
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True)
])

# Create a DataFrame with some null values
data = [
    (1, "Alice", 30),
    (2, "Bob", None),
    (3, "Charlie", 25),
    (4, None, 40),
    (5, "Eve", None)
]

df = spark.createDataFrame(data, schema)

# Show the DataFrame
df.show()

# Register the DataFrame as a temporary table
df.createOrReplaceTempView("people")

# Perform SQL aggregation using array_agg to collect names and ages
result = spark.sql("""
    SELECT 
        array_agg(name) as names_array,
        array_agg(age) as ages_array,
        array_agg(age) FILTER (WHERE age IS NOT NULL) as non_null_ages_array
    FROM people
""")

# Show the result
result.show(truncate=False)

# Stop the SparkSession
spark.stop()

```

duckdb code:
import duckdb

# Create a DuckDB in-memory connection
con = duckdb.connect()

# Create a table with some null values
con.execute("""
    CREATE TABLE people (
        id INTEGER,
        name VARCHAR,
        age INTEGER
    );
""")

# Insert data into the table, including some nulls
con.execute("""
    INSERT INTO people VALUES
        (1, 'Alice', 30),
        (2, 'Bob', NULL),
        (3, 'Charlie', 25),
        (4, NULL, 40),
        (5, 'Eve', NULL);
""")

# Perform SQL aggregation using array_agg to collect names and ages
result = con.execute("""
    SELECT 
        array_agg(name) AS names_array,
        array_agg(age) AS ages_array,
        array_agg(age) FILTER (WHERE age IS NOT NULL) AS non_null_ages_array
    FROM people;
""").fetchdf()

# Display the result
print(result)
 
**Official Documentation**
Please include links to official SQL documentation related to your issue.
1. duckdb docs - https://duckdb.org/docs/sql/data_types/nulls.html

Discussion:

Solution patch:
diff --git a/sqlglot/dialects/dialect.py b/sqlglot/dialects/dialect.py
index 669a4b7643..3c1ce35626 100644
--- a/sqlglot/dialects/dialect.py
+++ b/sqlglot/dialects/dialect.py
@@ -363,9 +363,11 @@ class Dialect(metaclass=_Dialect):
         HAVING
             my_id = 1
 
-    In most dialects "my_id" would refer to "data.my_id" (which is done in _qualify_columns()) across the query, except:
-        - BigQuery, which will forward the alias to GROUP BY + HAVING clauses i.e it resolves to "WHERE my_id = 1 GROUP BY id HAVING id = 1"
-        - Clickhouse, which will forward the alias across the query i.e it resolves to "WHERE id = 1 GROUP BY id HAVING id = 1"
+    In most dialects, "my_id" would refer to "data.my_id" across the query, except:
+        - BigQuery, which will forward the alias to GROUP BY + HAVING clauses i.e
+          it resolves to "WHERE my_id = 1 GROUP BY id HAVING id = 1"
+        - Clickhouse, which will forward the alias across the query i.e it resolves
+        to "WHERE id = 1 GROUP BY id HAVING id = 1"
     """
 
     EXPAND_ALIAS_REFS_EARLY_ONLY_IN_GROUP_BY = False
@@ -384,15 +386,19 @@ class Dialect(metaclass=_Dialect):
 
     SUPPORTS_FIXED_SIZE_ARRAYS = False
     """
-    Whether expressions such as x::INT[5] should be parsed as fixed-size array defs/casts e.g. in DuckDB. In
-    dialects which don't support fixed size arrays such as Snowflake, this should be interpreted as a subscript/index operator
+    Whether expressions such as x::INT[5] should be parsed as fixed-size array defs/casts e.g.
+    in DuckDB. In dialects which don't support fixed size arrays such as Snowflake, this should
+    be interpreted as a subscript/index operator.
     """
 
     STRICT_JSON_PATH_SYNTAX = True
     """Whether failing to parse a JSON path expression using the JSONPath dialect will log a warning."""
 
     ON_CONDITION_EMPTY_BEFORE_ERROR = True
-    """Whether "X ON EMPTY" should come before "X ON ERROR" (for dialects like T-SQL, MySQL, Oracle)"""
+    """Whether "X ON EMPTY" should come before "X ON ERROR" (for dialects like T-SQL, MySQL, Oracle)."""
+
+    ARRAY_AGG_INCLUDES_NULLS: t.Optional[bool] = True
+    """Whether ArrayAgg needs to filter NULL values."""
 
     SET_OP_DISTINCT_BY_DEFAULT: t.Dict[t.Type[exp.Expression], t.Optional[bool]] = {
         exp.Except: True,
diff --git a/sqlglot/dialects/hive.py b/sqlglot/dialects/hive.py
index a4d786f636..26cac7910e 100644
--- a/sqlglot/dialects/hive.py
+++ b/sqlglot/dialects/hive.py
@@ -196,6 +196,7 @@ class Hive(Dialect):
     IDENTIFIERS_CAN_START_WITH_DIGIT = True
     SUPPORTS_USER_DEFINED_TYPES = False
     SAFE_DIVISION = True
+    ARRAY_AGG_INCLUDES_NULLS = None
 
     # https://spark.apache.org/docs/latest/sql-ref-identifier.html#description
     NORMALIZATION_STRATEGY = NormalizationStrategy.CASE_INSENSITIVE
@@ -278,7 +279,7 @@ class Parser(parser.Parser):
         FUNCTIONS = {
             **parser.Parser.FUNCTIONS,
             "BASE64": exp.ToBase64.from_arg_list,
-            "COLLECT_LIST": exp.ArrayAgg.from_arg_list,
+            "COLLECT_LIST": lambda args: exp.ArrayAgg(this=seq_get(args, 0), nulls_excluded=True),
             "COLLECT_SET": exp.ArrayUniqueAgg.from_arg_list,
             "DATE_ADD": lambda args: exp.TsOrDsAdd(
                 this=seq_get(args, 0), expression=seq_get(args, 1), unit=exp.Literal.string("DAY")
diff --git a/sqlglot/dialects/snowflake.py b/sqlglot/dialects/snowflake.py
index 3264597808..734a0b2f0f 100644
--- a/sqlglot/dialects/snowflake.py
+++ b/sqlglot/dialects/snowflake.py
@@ -233,6 +233,7 @@ class Snowflake(Dialect):
     PREFER_CTE_ALIAS_COLUMN = True
     TABLESAMPLE_SIZE_IS_PERCENT = True
     COPY_PARAMS_ARE_CSV = False
+    ARRAY_AGG_INCLUDES_NULLS = None
 
     TIME_MAPPING = {
         "YYYY": "%Y",
@@ -291,7 +292,6 @@ class Parser(parser.Parser):
         FUNCTIONS = {
             **parser.Parser.FUNCTIONS,
             "APPROX_PERCENTILE": exp.ApproxQuantile.from_arg_list,
-            "ARRAYAGG": exp.ArrayAgg.from_arg_list,
             "ARRAY_CONSTRUCT": lambda args: exp.Array(expressions=args),
             "ARRAY_CONTAINS": lambda args: exp.ArrayContains(
                 this=seq_get(args, 1), expression=seq_get(args, 0)
diff --git a/sqlglot/expressions.py b/sqlglot/expressions.py
index cf0d6c2fab..414c57c9d0 100644
--- a/sqlglot/expressions.py
+++ b/sqlglot/expressions.py
@@ -4930,7 +4930,7 @@ class ExplodingGenerateSeries(GenerateSeries):
 
 
 class ArrayAgg(AggFunc):
-    pass
+    arg_types = {"this": True, "nulls_excluded": False}
 
 
 class ArrayUniqueAgg(AggFunc):
diff --git a/sqlglot/generator.py b/sqlglot/generator.py
index 7b5f9d267f..a19325d4c4 100644
--- a/sqlglot/generator.py
+++ b/sqlglot/generator.py
@@ -4250,3 +4250,16 @@ def jsonexists_sql(self, expression: exp.JSONExists) -> str:
         path = f"{path}{passing}{on_condition}"
 
         return self.func("JSON_EXISTS", this, path)
+
+    def arrayagg_sql(self, expression: exp.ArrayAgg) -> str:
+        array_agg = self.function_fallback_sql(expression)
+
+        if self.dialect.ARRAY_AGG_INCLUDES_NULLS and expression.args.get("nulls_excluded"):
+            parent = expression.parent
+            if isinstance(parent, exp.Filter):
+                parent_cond = parent.expression.this
+                parent_cond.replace(parent_cond.and_(expression.this.is_(exp.null()).not_()))
+            else:
+                array_agg = f"{array_agg} FILTER(WHERE {self.sql(expression, 'this')} IS NOT NULL)"
+
+        return array_agg
diff --git a/sqlglot/parser.py b/sqlglot/parser.py
index 68d6859146..b1892fedef 100644
--- a/sqlglot/parser.py
+++ b/sqlglot/parser.py
@@ -176,6 +176,12 @@ class Parser(metaclass=_Parser):
     FUNCTIONS: t.Dict[str, t.Callable] = {
         **{name: func.from_arg_list for name, func in exp.FUNCTION_BY_NAME.items()},
         "ARRAY": lambda args, dialect: exp.Array(expressions=args),
+        "ARRAYAGG": lambda args, dialect: exp.ArrayAgg(
+            this=seq_get(args, 0), nulls_excluded=dialect.ARRAY_AGG_INCLUDES_NULLS is None or None
+        ),
+        "ARRAY_AGG": lambda args, dialect: exp.ArrayAgg(
+            this=seq_get(args, 0), nulls_excluded=dialect.ARRAY_AGG_INCLUDES_NULLS is None or None
+        ),
         "COUNT": lambda args: exp.Count(this=seq_get(args, 0), expressions=args[1:], big_int=True),
         "CONCAT": lambda args, dialect: exp.Concat(
             expressions=args,


Test patch:
diff --git a/tests/dialects/test_hive.py b/tests/dialects/test_hive.py
index 8f422cd0de..2aca96db3e 100644
--- a/tests/dialects/test_hive.py
+++ b/tests/dialects/test_hive.py
@@ -716,8 +716,8 @@ def test_hive(self):
                 "presto": "ARRAY_AGG(x)",
             },
             write={
-                "duckdb": "ARRAY_AGG(x)",
-                "presto": "ARRAY_AGG(x)",
+                "duckdb": "ARRAY_AGG(x) FILTER(WHERE x IS NOT NULL)",
+                "presto": "ARRAY_AGG(x) FILTER(WHERE x IS NOT NULL)",
                 "hive": "COLLECT_LIST(x)",
                 "spark": "COLLECT_LIST(x)",
             },
diff --git a/tests/dialects/test_spark.py b/tests/dialects/test_spark.py
index cbaa169af9..4fed68caea 100644
--- a/tests/dialects/test_spark.py
+++ b/tests/dialects/test_spark.py
@@ -308,6 +308,13 @@ def test_spark(self):
                 },
             )
 
+        self.validate_all(
+            "SELECT ARRAY_AGG(x) FILTER (WHERE x = 5) FROM (SELECT 1 UNION ALL SELECT NULL) AS t(x)",
+            write={
+                "duckdb": "SELECT ARRAY_AGG(x) FILTER(WHERE x = 5 AND NOT x IS NULL) FROM (SELECT 1 UNION ALL SELECT NULL) AS t(x)",
+                "spark": "SELECT COLLECT_LIST(x) FILTER(WHERE x = 5) FROM (SELECT 1 UNION ALL SELECT NULL) AS t(x)",
+            },
+        )
         self.validate_all(
             "SELECT DATE_FORMAT(DATE '2020-01-01', 'EEEE') AS weekday",
             write={