Fix crash in count distinct with filters in repartition subqueries

now copies all column references in count distinct aggreagete
to worker target list and group by. Master target list is
also updated to reflect changes in attribute order.

Fixes 569

src/backend/distributed/planner/multi_logical_optimizer.c

@@ -1360,9 +1360,7 @@ MasterAggregateExpression(Aggref *originalAggregate,
 		walkerContext->repartitionSubquery)
 	{
 		Aggref *aggregate = (Aggref *) copyObject(originalAggregate);
-		List *aggTargetEntryList = aggregate->args;
-		TargetEntry *distinctTargetEntry = linitial(aggTargetEntryList);
-		List *varList = pull_var_clause_default((Node *) distinctTargetEntry->expr);
+		List *varList = pull_var_clause_default((Node *) aggregate);
 		ListCell *varCell = NULL;
 		List *uniqueVarList = NIL;
 		int startColumnCount = walkerContext->columnId;
@@ -1878,22 +1876,12 @@ WorkerAggregateExpressionList(Aggref *originalAggregate,
 		walkerContext->repartitionSubquery)
 	{
 		Aggref *aggregate = (Aggref *) copyObject(originalAggregate);
-		List *aggTargetEntryList = aggregate->args;
-		TargetEntry *distinctTargetEntry = (TargetEntry *) linitial(aggTargetEntryList);
-		List *columnList = pull_var_clause_default((Node *) distinctTargetEntry);
+		List *columnList = pull_var_clause_default((Node *) aggregate);
 		ListCell *columnCell = NULL;
-		List *processedColumnList = NIL;
-
 		foreach(columnCell, columnList)
 		{
 			Var *column = (Var *) lfirst(columnCell);
-			if (list_member(processedColumnList, column))
-			{
-				continue;
-			}
-
-			processedColumnList = lappend(processedColumnList, column);
-			workerAggregateList = lappend(workerAggregateList, copyObject(column));
+			workerAggregateList = list_append_unique(workerAggregateList, column);
 		}
 
 		walkerContext->createGroupByClause = true;

src/test/regress/input/multi_complex_count_distinct.source

@@ -76,6 +76,34 @@ SELECT *
 	ORDER BY 2 DESC, 1 DESC
 	LIMIT 10;
 
+-- there is a known issue with aggregates with filters in non-repartition queries (#395)
+SELECT
+	l_orderkey, count(DISTINCT l_partkey) FILTER (WHERE l_shipmode = 'AIR')
+	FROM lineitem_hash
+	GROUP BY l_orderkey
+	ORDER BY 2 DESC, 1 DESC
+	LIMIT 10;
+
+-- filter column already exists in target list
+SELECT *
+	FROM (
+		SELECT
+			l_orderkey, count(DISTINCT l_partkey) FILTER (WHERE l_orderkey > 100)
+			FROM lineitem_hash
+			GROUP BY l_orderkey) sub
+	ORDER BY 2 DESC, 1 DESC
+	LIMIT 10;
+
+-- filter column does not exist in target list
+SELECT *
+	FROM (
+		SELECT
+			l_orderkey, count(DISTINCT l_partkey) FILTER (WHERE l_shipmode = 'AIR')
+			FROM lineitem_hash
+			GROUP BY l_orderkey) sub
+	ORDER BY 2 DESC, 1 DESC
+	LIMIT 10;
+
 -- case expr in count distinct is supported.
 -- count orders partkeys if l_shipmode is air
 SELECT *

src/test/regress/output/multi_complex_count_distinct.source

@@ -122,6 +122,60 @@ SELECT *
       1927 |     3
 (10 rows)
 
+-- there is a known issue with aggregates with filters in non-repartition queries (#395)
+SELECT
+	l_orderkey, count(DISTINCT l_partkey) FILTER (WHERE l_shipmode = 'AIR')
+	FROM lineitem_hash
+	GROUP BY l_orderkey
+	ORDER BY 2 DESC, 1 DESC
+	LIMIT 10;
+ERROR:  attribute number 15 exceeds number of columns 2
+-- filter column already exists in target list
+SELECT *
+	FROM (
+		SELECT
+			l_orderkey, count(DISTINCT l_partkey) FILTER (WHERE l_orderkey > 100)
+			FROM lineitem_hash
+			GROUP BY l_orderkey) sub
+	ORDER BY 2 DESC, 1 DESC
+	LIMIT 10;
+ l_orderkey | count 
+------------+-------
+      14885 |     7
+      14884 |     7
+      14821 |     7
+      14790 |     7
+      14785 |     7
+      14755 |     7
+      14725 |     7
+      14694 |     7
+      14627 |     7
+      14624 |     7
+(10 rows)
+
+-- filter column does not exist in target list
+SELECT *
+	FROM (
+		SELECT
+			l_orderkey, count(DISTINCT l_partkey) FILTER (WHERE l_shipmode = 'AIR')
+			FROM lineitem_hash
+			GROUP BY l_orderkey) sub
+	ORDER BY 2 DESC, 1 DESC
+	LIMIT 10;
+ l_orderkey | count 
+------------+-------
+      12005 |     4
+       5409 |     4
+       4964 |     4
+      14848 |     3
+      14496 |     3
+      13473 |     3
+      13122 |     3
+      12929 |     3
+      12645 |     3
+      12417 |     3
+(10 rows)
+
 -- case expr in count distinct is supported.
 -- count orders partkeys if l_shipmode is air
 SELECT *