修复并行查询使用windows函数产生错误的bug

src/gausskernel/optimizer/plan/planner.cpp

@@ -170,6 +170,8 @@ static void deinit_optimizer_context(PlannerGlobal* glob);
 static void check_index_column();
 static bool check_sort_for_upsert(PlannerInfo* root);
 
+static bool IsSupportMakeSort(List* pathkeys);
+
 extern void PushDownFullPseudoTargetlist(PlannerInfo *root, Plan *topNode, Plan *botNode,
             List *fullEntryList);
 
@@ -8088,8 +8090,9 @@ static Plan* mark_windowagg_stream(
         Plan* gatherPlan = NULL;
         Sort* sortPlan = NULL;
         SimpleSort* streamSort = NULL;
+        bool isSupportMakeSort = IsSupportMakeSort(pathkeys);
 
-        if (pathkeys != NIL) {
+        if (isSupportMakeSort) {
             sortPlan = make_sort_from_pathkeys(root, bottomPlan, pathkeys, -1.0);
 
             streamSort = makeNode(SimpleSort);
@@ -8107,7 +8110,7 @@ static Plan* mark_windowagg_stream(
              * If have pathkeys, we can push down Sort to Datanode and then merge partial
              * sorted results in RemoteQuery.
              */
-            if (pathkeys != NIL) {
+            if (isSupportMakeSort) {
                 gatherPlan = make_simple_RemoteQuery((Plan*)sortPlan, root, false);
                 if (IsA(gatherPlan, RemoteQuery)) {
                     ((RemoteQuery*)gatherPlan)->sort = streamSort;
@@ -8121,7 +8124,7 @@ static Plan* mark_windowagg_stream(
             if (((unsigned int)u_sess->attr.attr_sql.cost_param & COST_ALTERNATIVE_MERGESORT) ||
                 root->is_under_recursive_cte) {
                 gatherPlan = make_stream_plan(root, bottomPlan, NIL, 1.0);
-                if (pathkeys != NIL)
+                if (isSupportMakeSort)
                     gatherPlan = (Plan*)make_sort_from_pathkeys(root, gatherPlan, pathkeys, -1.0);
             } else {
                 bool single_node =
@@ -8164,7 +8167,7 @@ static Plan* mark_windowagg_stream(
                 if (!single_node) {
                     gatherPlan = make_stream_plan(root, bottomPlan, NIL, 1.0);
                     pick_single_node_plan_for_replication(gatherPlan);
-                    if (pathkeys != NIL)
+                    if (isSupportMakeSort)
                         ((Stream*)gatherPlan)->sort = streamSort;
                 } else
                     gatherPlan = bottomPlan;
@@ -14676,3 +14679,15 @@ bool check_stream_for_loop_fetch(Portal portal)
     return has_stream;
 }
 
+static bool IsSupportMakeSort(List* pathkeys)
+{
+    ListCell* i = NULL;
+    foreach (i, pathkeys) {
+        PathKey* pathkey = (PathKey*)lfirst(i);
+        if (pathkey != NULL && pathkey->pk_eclass->ec_members != NIL) {
+            return true;
+        }
+    }
+    return false;
+}
+

src/test/regress/expected/window_agg_stream_test.out

@@ -0,0 +1,42 @@
+set query_dop=4;
+drop table if exists bmsql_item;
+NOTICE:  table "bmsql_item" does not exist, skipping
+CREATE TABLE bmsql_item (
+i_id int NoT NULL,
+i_name varchar(24),
+i_price numeric(5,2),
+i_data varchar( 50),
+i_im_id int
+);
+insert into bmsql_item values ('1','sqltest_varchar_1','0.01','sqltest_varchar_1','1') ;
+insert into bmsql_item values ('2','sqltest_varchar_2','0.02','sqltest_varchar_2','2') ;
+insert into bmsql_item values ('3','sqltest_varchar_3','0.03','sqltest_varchar_3','3') ;
+insert into bmsql_item values ('4','sqltest_varchar_4','0.04','sqltest_varchar_4','4') ;
+insert into bmsql_item(i_id) values ('5');
+select
+row_number() over(order by 2)
+from (select i_price as c1,'a' as c2 from bmsql_item order by 1 desc) tb1
+group by tb1.c2,cube(tb1.c1,tb1.c1)
+window window1 as (order by 3)
+order by 1;
+ row_number 
+------------
+          1
+          2
+          3
+          4
+          5
+          6
+          7
+          8
+          9
+         10
+         11
+         12
+         13
+         14
+         15
+         16
+(16 rows)
+
+drop table bmsql_item;

src/test/regress/parallel_schedule0

@@ -650,7 +650,7 @@ test: a_outerjoin_conversion
 #test: plan_table04
 
 test: setrefs
-test: agg
+test: agg window_agg_stream_test
 
 # test sql by pass
 test: bypass_simplequery_support

src/test/regress/sql/window_agg_stream_test.sql

@@ -0,0 +1,23 @@
+set query_dop=4;
+drop table if exists bmsql_item;
+CREATE TABLE bmsql_item (
+i_id int NoT NULL,
+i_name varchar(24),
+i_price numeric(5,2),
+i_data varchar( 50),
+i_im_id int
+);
+insert into bmsql_item values ('1','sqltest_varchar_1','0.01','sqltest_varchar_1','1') ;
+insert into bmsql_item values ('2','sqltest_varchar_2','0.02','sqltest_varchar_2','2') ;
+insert into bmsql_item values ('3','sqltest_varchar_3','0.03','sqltest_varchar_3','3') ;
+insert into bmsql_item values ('4','sqltest_varchar_4','0.04','sqltest_varchar_4','4') ;
+insert into bmsql_item(i_id) values ('5');
+
+select
+row_number() over(order by 2)
+from (select i_price as c1,'a' as c2 from bmsql_item order by 1 desc) tb1
+group by tb1.c2,cube(tb1.c1,tb1.c1)
+window window1 as (order by 3)
+order by 1;
+
+drop table bmsql_item;