Fix the assert failure on pullup flow in within group

Flow in AggNode has wrong TargetList. AggNode has a different
TargetList from its child nodes, so copying flow directly from the
child node to AggNode is completely wrong. We need to use pullupflow to
generate this TargetList in creating the within group plan with single
QE.

src/backend/cdb/cdbgroup.c

@@ -5368,7 +5368,10 @@ make_parallel_or_sequential_agg(PlannerInfo *root, AggClauseCounts *agg_counts,
 										agg_counts->numAggs,
 										agg_counts->transitionSpace,
 										result_plan);
-		mark_passthru_locus(result_plan, true, current_pathkeys != NIL);
+		result_plan->flow = pull_up_Flow(result_plan,
+										 result_plan->lefttree,
+										 (current_pathkeys != NIL)
+										 && aggstrategy != AGG_HASHED );
 	}
 	else
 		current_pathkeys = *group_context->pcurrent_pathkeys;

src/test/regress/expected/qp_functions_idf.out

@@ -916,8 +916,34 @@ select distinct on (median(a)) median(a) ,b from perct group by b order by media
     100 | 10
 (11 rows)
 
+-- Test the with group with gather motion
+CREATE TABLE within_group_gather (
+	date1 timestamp without time zone,
+	date2 timestamp without time zone,
+	num bpchar
+			) DISTRIBUTED BY (date1);
+WITH detail_serials AS (select
+                                num,
+                                date1,
+                                extract(epoch FROM (date1::timestamp
+                                        -((CASE WHEN (num IS DISTINCT FROM lag(num) OVER (ORDER BY num,date1))
+												THEN date2 ELSE lag(date1,1) over (ORDER BY num, date1)
+														END)
+                                                  )))/3600 hours_at_op,
+                                date2
+                        FROM within_group_gather)
+                select  to_char(date1,'IYYYIW') fiscal_week,
+                        count(distinct num) sn_qty,
+                                percentile_disc(0.25) WITHIN GROUP (ORDER BY hours_at_op) p25_lt_hours
+                FROM detail_serials
+                GROUP BY 1;
+ fiscal_week | sn_qty | p25_lt_hours 
+-------------+--------+--------------
+(0 rows)
+
 -- start_ignore
 drop schema qp_idf cascade;
+NOTICE:  drop cascades to table within_group_gather
 NOTICE:  drop cascades to view idf_v4
 NOTICE:  drop cascades to rule _RETURN on view idf_v4
 NOTICE:  drop cascades to view idf_v1

src/test/regress/sql/qp_functions_idf.sql

@@ -258,6 +258,29 @@ select distinct(median(b)) from perct group by a;
 
 select distinct on (median(a)) median(a) ,b from perct group by b order by median(a),b;
 
+-- Test the with group with gather motion
+CREATE TABLE within_group_gather (
+	date1 timestamp without time zone,
+	date2 timestamp without time zone,
+	num bpchar
+			) DISTRIBUTED BY (date1);
+
+WITH detail_serials AS (select
+                                num,
+                                date1,
+                                extract(epoch FROM (date1::timestamp
+                                        -((CASE WHEN (num IS DISTINCT FROM lag(num) OVER (ORDER BY num,date1))
+												THEN date2 ELSE lag(date1,1) over (ORDER BY num, date1)
+														END)
+                                                  )))/3600 hours_at_op,
+                                date2
+                        FROM within_group_gather)
+                select  to_char(date1,'IYYYIW') fiscal_week,
+                        count(distinct num) sn_qty,
+                                percentile_disc(0.25) WITHIN GROUP (ORDER BY hours_at_op) p25_lt_hours
+                FROM detail_serials
+                GROUP BY 1;
+
 -- start_ignore
 drop schema qp_idf cascade;
--- end_ignore
\ No newline at end of file
+-- end_ignore