Make count return 0 if all shards are pruned away

Before this change, count on a distributed returned NULL if all shards
were pruned away, because on the master we replace with count(..) call
with a sum(..) call to sum the counts from the shards. However, sum
returns NULL when there are no rows, whereas count is expected to return
0.

src/backend/distributed/planner/multi_logical_optimizer.c

@@ -1499,6 +1499,10 @@ MasterAggregateExpression(Aggref *originalAggregate,
 		 */
 		Var *column = NULL;
 		TargetEntry *columnTargetEntry = NULL;
+		CoerceViaIO *coerceExpr = NULL;
+		Const *zeroConst = NULL;
+		List *coalesceArgs = NULL;
+		CoalesceExpr *coalesceExpr = NULL;
 
 		/* worker aggregate and original aggregate have the same return type */
 		Oid workerReturnType = exprType((Node *) originalAggregate);
@@ -1523,7 +1527,25 @@ MasterAggregateExpression(Aggref *originalAggregate,
 		columnTargetEntry = makeTargetEntry((Expr *) column, argumentId, NULL, false);
 		newMasterAggregate->args = list_make1(columnTargetEntry);
 
-		newMasterExpression = (Expr *) newMasterAggregate;
+		/* cast numeric sum result to bigint (count's return type) */
+		coerceExpr = makeNode(CoerceViaIO);
+		coerceExpr->arg = (Expr *) newMasterAggregate;
+		coerceExpr->resulttype = INT8OID;
+		coerceExpr->resultcollid = InvalidOid;
+		coerceExpr->coerceformat = COERCE_IMPLICIT_CAST;
+		coerceExpr->location = -1;
+
+		/* convert NULL to 0 in case of no rows */
+		zeroConst = MakeIntegerConstInt64(0);
+		coalesceArgs = list_make2(coerceExpr, zeroConst);
+
+		coalesceExpr = makeNode(CoalesceExpr);
+		coalesceExpr->coalescetype = INT8OID;
+		coalesceExpr->coalescecollid = InvalidOid;
+		coalesceExpr->args = coalesceArgs;
+		coalesceExpr->location = -1;
+
+		newMasterExpression = (Expr *) coalesceExpr;
 	}
 	else if (aggregateType == AGGREGATE_ARRAY_AGG)
 	{

src/test/regress/expected/multi_complex_expressions.out

@@ -224,7 +224,7 @@ SELECT count(*) FROM lineitem
 	WHERE 0 != 0;
  count 
 -------
-      
+     0
 (1 row)
 
 -- distinct expressions can be pushed down

src/test/regress/expected/multi_explain.out

@@ -43,7 +43,7 @@ Distributed Query into pg_merge_job_570000
               ->  Seq Scan on lineitem_290000 lineitem
 Master Query
   ->  Sort
-        Sort Key: (sum(((sum(intermediate_column_570000_1))::bigint)))::bigint, intermediate_column_570000_0
+        Sort Key: COALESCE((sum((COALESCE((sum(intermediate_column_570000_1))::bigint, '0'::bigint))))::bigint, '0'::bigint), intermediate_column_570000_0
         ->  HashAggregate
               Group Key: intermediate_column_570000_0
               ->  Seq Scan on pg_merge_job_570000
@@ -87,7 +87,7 @@ EXPLAIN (COSTS FALSE, FORMAT JSON)
       {
         "Plan": {
           "Node Type": "Sort",
-          "Sort Key": ["(sum(((sum(intermediate_column_570001_1))::bigint)))::bigint", "intermediate_column_570001_0"],
+          "Sort Key": ["COALESCE((sum((COALESCE((sum(intermediate_column_570001_1))::bigint, '0'::bigint))))::bigint, '0'::bigint)", "intermediate_column_570001_0"],
           "Plans": [
             {
               "Node Type": "Aggregate",
@@ -156,7 +156,7 @@ EXPLAIN (COSTS FALSE, FORMAT XML)
         <Plan>
           <Node-Type>Sort</Node-Type>
           <Sort-Key>
-            <Item>(sum(((sum(intermediate_column_570003_1))::bigint)))::bigint</Item>
+            <Item>COALESCE((sum((COALESCE((sum(intermediate_column_570003_1))::bigint, '0'::bigint))))::bigint, '0'::bigint)</Item>
             <Item>intermediate_column_570003_0</Item>
           </Sort-Key>
           <Plans>
@@ -213,7 +213,7 @@ EXPLAIN (COSTS FALSE, FORMAT YAML)
     - Plan: 
         Node Type: "Sort"
         Sort Key: 
-          - "(sum(((sum(intermediate_column_570005_1))::bigint)))::bigint"
+          - "COALESCE((sum((COALESCE((sum(intermediate_column_570005_1))::bigint, '0'::bigint))))::bigint, '0'::bigint)"
           - "intermediate_column_570005_0"
         Plans: 
           - Node Type: "Aggregate"
@@ -241,7 +241,7 @@ Distributed Query into pg_merge_job_570006
               ->  Seq Scan on lineitem_290000 lineitem
 Master Query
   ->  Sort
-        Sort Key: (sum(((sum(intermediate_column_570006_1))::bigint)))::bigint, intermediate_column_570006_0
+        Sort Key: COALESCE((sum((COALESCE((sum(intermediate_column_570006_1))::bigint, '0'::bigint))))::bigint, '0'::bigint), intermediate_column_570006_0
         ->  HashAggregate
               Group Key: intermediate_column_570006_0
               ->  Seq Scan on pg_merge_job_570006

src/test/regress/expected/multi_large_table_pruning.out

@@ -46,7 +46,7 @@ DEBUG:  predicate pruning for shardId 290008
 DEBUG:  predicate pruning for shardId 290009
  count 
 -------
-      
+     0
 (1 row)
 
 -- Single range-repartition join with a selection clause on the base table to
@@ -63,7 +63,7 @@ DEBUG:  predicate pruning for shardId 280001
 DEBUG:  predicate pruning for shardId 280000
  count 
 -------
-      
+     0
 (1 row)
 
 -- Dual hash-repartition join test case. Note that this query doesn't produce
@@ -127,7 +127,7 @@ DEBUG:  predicate pruning for shardId 290006
 DEBUG:  predicate pruning for shardId 290007
  count 
 -------
-      
+     0
 (1 row)
 
 -- Test cases with false in the WHERE clause

src/test/regress/expected/multi_router_planner.out

@@ -1668,7 +1668,7 @@ SELECT count(*), count(*) FILTER (WHERE id < 3)
 	WHERE author_id = 1 or author_id = 2;
  count | count 
 -------+-------
-    10 |      
+    10 |     0
 (1 row)
 
 -- prepare queries can be router plannable

src/test/regress/expected/multi_truncate.out

@@ -59,7 +59,7 @@ TRUNCATE TABLE test_truncate_append;
 SELECT count(*) FROM test_truncate_append;
  count 
 -------
-      
+     0
 (1 row)
 
 -- verify no shard exists anymore

src/test/regress/output/multi_master_delete_protocol.source

@@ -66,7 +66,7 @@ SELECT master_apply_delete_command('DELETE FROM customer_delete_protocol');
 SELECT count(*) FROM customer_delete_protocol;
  count 
 -------
-      
+     0
 (1 row)
 
 -- Verify that empty shards are deleted if no condition is provided