Repository: duckdb/duckdb (language: python)

Issue:
COALESCE with aggregate function in PIVOT query does not work
### What happens?

In pivot we cannot use COALESCE over aggregate function, for example something like 
```
PIVOT Cities
ON Year
USING COALESCE(sum(Population), 0);
```
fails with 
```
Binder Error: Pivot expression must be an aggregate
```
It would be great to support it as some combination of `Year` and `Cities` might not be present in the source data so just ` USING sum(COALESCE(Population, 0))` does not solve the issue. May be related issue - https://github.com/duckdb/duckdb/issues/14481

### To Reproduce

```
CREATE TABLE Cities (
    Country VARCHAR, Name VARCHAR, Year INTEGER, Population INTEGER
);
INSERT INTO Cities VALUES
    ('NL', 'Amsterdam', 2000, 1005),
    ('NL', 'Amsterdam', 2010, 1065),
    ('NL', 'Amsterdam', 2020, 1158),
    ('US', 'Seattle', 2000, 564),
    ('US', 'Seattle', 2010, 608),
    ('US', 'Seattle', 2020, 738),
    ('US', 'New York City', 2000, 8015);
```
Then run 
```
 PIVOT Cities
    ON Year
    USING COALESCE(sum(Population), 0);
```

### OS:

arm64

### DuckDB Version:

v1.1.3

### DuckDB Client:

go-duckdb

### Hardware:

_No response_

### Full Name:

Parag Jain

### Affiliation:

Rill Data

### What is the latest build you tested with? If possible, we recommend testing with the latest nightly build.

I have tested with a stable release

### Did you include all relevant data sets for reproducing the issue?

Yes

### Did you include all code required to reproduce the issue?

- [X] Yes, I have

### Did you include all relevant configuration (e.g., CPU architecture, Python version, Linux distribution) to reproduce the issue?

- [X] Yes, I have
COALESCE with aggregate function in PIVOT query does not work
### What happens?

In pivot we cannot use COALESCE over aggregate function, for example something like 
```
PIVOT Cities
ON Year
USING COALESCE(sum(Population), 0);
```
fails with 
```
Binder Error: Pivot expression must be an aggregate
```
It would be great to support it as some combination of `Year` and `Cities` might not be present in the source data so just ` USING sum(COALESCE(Population, 0))` does not solve the issue. May be related issue - https://github.com/duckdb/duckdb/issues/14481

### To Reproduce

```
CREATE TABLE Cities (
    Country VARCHAR, Name VARCHAR, Year INTEGER, Population INTEGER
);
INSERT INTO Cities VALUES
    ('NL', 'Amsterdam', 2000, 1005),
    ('NL', 'Amsterdam', 2010, 1065),
    ('NL', 'Amsterdam', 2020, 1158),
    ('US', 'Seattle', 2000, 564),
    ('US', 'Seattle', 2010, 608),
    ('US', 'Seattle', 2020, 738),
    ('US', 'New York City', 2000, 8015);
```
Then run 
```
 PIVOT Cities
    ON Year
    USING COALESCE(sum(Population), 0);
```

### OS:

arm64

### DuckDB Version:

v1.1.3

### DuckDB Client:

go-duckdb

### Hardware:

_No response_

### Full Name:

Parag Jain

### Affiliation:

Rill Data

### What is the latest build you tested with? If possible, we recommend testing with the latest nightly build.

I have tested with a stable release

### Did you include all relevant data sets for reproducing the issue?

Yes

### Did you include all code required to reproduce the issue?

- [X] Yes, I have

### Did you include all relevant configuration (e.g., CPU architecture, Python version, Linux distribution) to reproduce the issue?

- [X] Yes, I have


Discussion / hints:

Fix patch:
diff --git a/src/planner/binder/tableref/bind_pivot.cpp b/src/planner/binder/tableref/bind_pivot.cpp
index 17469855840c..e673c6011bbd 100644
--- a/src/planner/binder/tableref/bind_pivot.cpp
+++ b/src/planner/binder/tableref/bind_pivot.cpp
@@ -69,6 +69,27 @@ static void ExtractPivotExpressions(ParsedExpression &expr, case_insensitive_set
 	    expr, [&](ParsedExpression &child) { ExtractPivotExpressions(child, handled_columns); });
 }
 
+void ExtractPivotAggregateExpression(ClientContext &context, ParsedExpression &expr,
+                                     vector<reference<FunctionExpression>> &aggregates) {
+	if (expr.type == ExpressionType::FUNCTION) {
+		auto &aggr_function = expr.Cast<FunctionExpression>();
+
+		// check if this is an aggregate to ensure it is an aggregate and not a scalar function
+		auto &entry = Catalog::GetEntry(context, CatalogType::SCALAR_FUNCTION_ENTRY, aggr_function.catalog,
+		                                aggr_function.schema, aggr_function.function_name);
+		if (entry.type == CatalogType::AGGREGATE_FUNCTION_ENTRY) {
+			// aggregate
+			aggregates.push_back(aggr_function);
+			return;
+		}
+	}
+	if (expr.type == ExpressionType::COLUMN_REF) {
+		throw BinderException(expr, "Columns can only be referenced within the aggregate of a PIVOT expression");
+	}
+	ParsedExpressionIterator::EnumerateChildren(
+	    expr, [&](ParsedExpression &child) { ExtractPivotAggregateExpression(context, child, aggregates); });
+}
+
 static unique_ptr<SelectNode> ConstructInitialGrouping(PivotRef &ref, vector<unique_ptr<ParsedExpression>> all_columns,
                                                        const case_insensitive_set_t &handled_columns) {
 	auto subquery = make_uniq<SelectNode>();
@@ -98,7 +119,8 @@ static unique_ptr<SelectNode> ConstructInitialGrouping(PivotRef &ref, vector<uni
 	return subquery;
 }
 
-static unique_ptr<SelectNode> PivotFilteredAggregate(PivotRef &ref, vector<unique_ptr<ParsedExpression>> all_columns,
+static unique_ptr<SelectNode> PivotFilteredAggregate(ClientContext &context, PivotRef &ref,
+                                                     vector<unique_ptr<ParsedExpression>> all_columns,
                                                      const case_insensitive_set_t &handled_columns,
                                                      vector<PivotValueElement> pivot_values) {
 	auto subquery = ConstructInitialGrouping(ref, std::move(all_columns), handled_columns);
@@ -124,7 +146,12 @@ static unique_ptr<SelectNode> PivotFilteredAggregate(PivotRef &ref, vector<uniqu
 		}
 		for (auto &aggregate : ref.aggregates) {
 			auto copied_aggr = aggregate->Copy();
-			auto &aggr = copied_aggr->Cast<FunctionExpression>();
+
+			vector<reference<FunctionExpression>> aggregates;
+			ExtractPivotAggregateExpression(context, *copied_aggr, aggregates);
+			D_ASSERT(aggregates.size() == 1);
+
+			auto &aggr = aggregates[0].get().Cast<FunctionExpression>();
 			aggr.filter = filter->Copy();
 			auto &aggr_name = aggregate->alias;
 			auto name = pivot_value.name;
@@ -132,7 +159,7 @@ static unique_ptr<SelectNode> PivotFilteredAggregate(PivotRef &ref, vector<uniqu
 				// if there are multiple aggregates specified we add the name of the aggregate as well
 				name += "_" + (aggr_name.empty() ? aggregate->GetName() : aggr_name);
 			}
-			aggr.alias = name;
+			copied_aggr->alias = name;
 			subquery->select_list.push_back(std::move(copied_aggr));
 		}
 	}
@@ -360,22 +387,27 @@ unique_ptr<SelectNode> Binder::BindPivot(PivotRef &ref, vector<unique_ptr<Parsed
 	// keep track of the columns by which we pivot/aggregate
 	// any columns which are not pivoted/aggregated on are added to the GROUP BY clause
 	case_insensitive_set_t handled_columns;
+
+	vector<reference<FunctionExpression>> pivot_aggregates;
 	// parse the aggregate, and extract the referenced columns from the aggregate
 	for (auto &aggr : ref.aggregates) {
-		if (aggr->type != ExpressionType::FUNCTION) {
-			throw BinderException(*aggr, "Pivot expression must be an aggregate");
-		}
 		if (aggr->HasSubquery()) {
 			throw BinderException(*aggr, "Pivot expression cannot contain subqueries");
 		}
 		if (aggr->IsWindow()) {
 			throw BinderException(*aggr, "Pivot expression cannot contain window functions");
 		}
-		// bind the function as an aggregate to ensure it is an aggregate and not a scalar function
-		auto &aggr_function = aggr->Cast<FunctionExpression>();
-		(void)Catalog::GetEntry<AggregateFunctionCatalogEntry>(context, aggr_function.catalog, aggr_function.schema,
-		                                                       aggr_function.function_name);
-		ExtractPivotExpressions(*aggr, handled_columns);
+		idx_t aggregate_count = pivot_aggregates.size();
+		ExtractPivotAggregateExpression(context, *aggr, pivot_aggregates);
+		if (pivot_aggregates.size() != aggregate_count + 1) {
+			string error_str = pivot_aggregates.size() == aggregate_count
+			                       ? "but no aggregates were found"
+			                       : "but " + to_string(pivot_aggregates.size() - aggregate_count) + " were found";
+			throw BinderException(*aggr, "Pivot expression must contain exactly one aggregate, %s", error_str);
+		}
+	}
+	for (auto &aggr : pivot_aggregates) {
+		ExtractPivotExpressions(aggr.get(), handled_columns);
 	}
 
 	// first add all pivots to the set of handled columns, and check for duplicates
@@ -448,7 +480,8 @@ unique_ptr<SelectNode> Binder::BindPivot(PivotRef &ref, vector<unique_ptr<Parsed
 	// we switch dynamically based on the number of pivots to compute
 	if (pivot_values.size() <= client_config.pivot_filter_threshold) {
 		// use a set of filtered aggregates
-		pivot_node = PivotFilteredAggregate(ref, std::move(all_columns), handled_columns, std::move(pivot_values));
+		pivot_node =
+		    PivotFilteredAggregate(context, ref, std::move(all_columns), handled_columns, std::move(pivot_values));
 	} else {
 		// executing a pivot statement happens in three stages
 		// 1) execute the query "SELECT {groups}, {pivots}, {aggregates} FROM {from_clause} GROUP BY {groups}, {pivots}


Test patch:
diff --git a/test/sql/pivot/pivot_errors.test b/test/sql/pivot/pivot_errors.test
index da1395752387..04e3ecdae992 100644
--- a/test/sql/pivot/pivot_errors.test
+++ b/test/sql/pivot/pivot_errors.test
@@ -19,7 +19,7 @@ SET pivot_filter_threshold=0
 statement error
 PIVOT test ON j IN ('a', 'b') USING current_date();
 ----
-current_date is not an aggregate function
+no aggregates were found
 
 statement ok
 SET pivot_filter_threshold=100
@@ -27,12 +27,12 @@ SET pivot_filter_threshold=100
 statement error
 PIVOT test ON j IN ('a', 'b') USING current_date();
 ----
-current_date is not an aggregate function
+no aggregates were found
 
 statement error
 PIVOT test ON j IN ('a', 'b') USING sum(41) over ();
 ----
-must be an aggregate
+cannot contain window functions
 
 statement error
 PIVOT test ON j IN ('a', 'b') USING sum(sum(41) over ());
diff --git a/test/sql/pivot/pivot_expressions.test b/test/sql/pivot/pivot_expressions.test
index ab0ddf70a474..af5dd254dc9f 100644
--- a/test/sql/pivot/pivot_expressions.test
+++ b/test/sql/pivot/pivot_expressions.test
@@ -49,6 +49,42 @@ PIVOT Cities ON (CASE WHEN Country='NL' THEN NULL ELSE Country END) USING SUM(Po
 2010	8783
 2020	9510
 
+# we allow expressions in the aggregate as well
+query IIII rowsort
+PIVOT Cities ON Country || '_' || Name USING COALESCE(SUM(Population), 0) GROUP BY Year;
+----
+2000	1005	8015	564
+2010	1065	8175	608
+2020	1158	8772	738
+
+# casts
+query IIII rowsort
+PIVOT Cities ON Country || '_' || Name USING SUM(Population)::VARCHAR GROUP BY Year;
+----
+2000	1005	8015	564
+2010	1065	8175	608
+2020	1158	8772	738
+
+# functions
+query IIII rowsort
+PIVOT Cities ON Country || '_' || Name USING SUM(Population) + 42 GROUP BY Year;
+----
+2000	1047	8057	606
+2010	1107	8217	650
+2020	1200	8814	780
+
+# we cannot have multiple aggregates
+statement error
+PIVOT Cities ON Country || '_' || Name USING SUM(Population) + COUNT(*) GROUP BY Year;
+----
+Pivot expression must contain exactly one aggregate
+
+# columns can only be referenced within the aggregate
+statement error
+PIVOT Cities ON Country || '_' || Name USING SUM(Population) + Population GROUP BY Year;
+----
+Columns can only be referenced within the aggregate of a PIVOT expression
+
 statement error
 PIVOT Cities ON min(Country) over () USING SUM(Population) GROUP BY Year;
 ----