ORCA: Introduce hash windows aggregation when use the vectorization exector

In this PR, ORCA now supports generating `WindowHashAgg` plans which already
have implementation in the vectorization executor. However, the CBDB row executor
currently lacks implementation for the WindowHashAgg operator. To prevent ORCA
from generating this operator in the row executor, I've added an struct which named
`OptimizerOptions` to control the plan for row executor or vectorization executor.
(By the way, ORCA may later generate plans specifically for the vectorization executor).

The `WindowAgg` operator implemention in the vectorization execution is:

1. First, sorting the input rows by `ORDER BY` keys
2. Then do the `PARTITION` by `PARTITION BY` keys
3. Finally do the window function.

Since step1 must be globally sorted, it cannot be parallelized in the vectorization executor.
This results in poor performance of the `WindowAgg` operator.

By contrast, `WindowHashAgg` employs a more efficient approach:

1. First hashes input data into buckets based on `PARTITION BY` keys
2. Then sorts data `within each bucket` according to `ORDER BY` keys
3. Finally computes window functions on the sorted bucket data

For the row engine, `WindowHashAgg` operators will not be generated.

Also current commit introduces a new GUC named `optimizer_force_window_hash_agg`
to force generate plans with `WindowHashAgg` (Don't used this GUC expect debug ORCA).

Co-Author-By: zhangyue <zhangyue@hashdata.cn>

Author: jiaqizho

contrib/pg_stat_statements/pg_stat_statements.c

@@ -313,7 +313,8 @@ static void pgss_post_parse_analyze(ParseState *pstate, Query *query,
 static PlannedStmt *pgss_planner(Query *parse,
 								 const char *query_string,
 								 int cursorOptions,
-								 ParamListInfo boundParams);
+								 ParamListInfo boundParams,
+								 OptimizerOptions *optimizer_options);
 static void pgss_ExecutorStart(QueryDesc *queryDesc, int eflags);
 static void pgss_ExecutorRun(QueryDesc *queryDesc,
 							 ScanDirection direction,
@@ -866,7 +867,8 @@ static PlannedStmt *
 pgss_planner(Query *parse,
 			 const char *query_string,
 			 int cursorOptions,
-			 ParamListInfo boundParams)
+			 ParamListInfo boundParams,
+			 OptimizerOptions *optimizer_options)
 {
 	PlannedStmt *result;
 
@@ -908,10 +910,10 @@ pgss_planner(Query *parse,
 		{
 			if (prev_planner_hook)
 				result = prev_planner_hook(parse, query_string, cursorOptions,
-										   boundParams);
+										   boundParams, optimizer_options);
 			else
 				result = standard_planner(parse, query_string, cursorOptions,
-										  boundParams);
+										  boundParams, optimizer_options);
 		}
 		PG_FINALLY();
 		{
@@ -945,10 +947,10 @@ pgss_planner(Query *parse,
 	{
 		if (prev_planner_hook)
 			result = prev_planner_hook(parse, query_string, cursorOptions,
-									   boundParams);
+									   boundParams, optimizer_options);
 		else
 			result = standard_planner(parse, query_string, cursorOptions,
-									  boundParams);
+									  boundParams, optimizer_options);
 	}
 
 	return result;

gpcontrib/pg_hint_plan/pg_hint_plan.c

@@ -428,7 +428,7 @@ pg_hint_plan_add_paths_to_joinrel(PlannerInfo *root,
 static void *external_plan_hint_hook(Query *parse);
 #endif
 static PlannedStmt *pg_hint_plan_planner(Query *parse, const char *query_string, int cursorOptions,
-										 ParamListInfo boundParams);
+										 ParamListInfo boundParams, OptimizerOptions *optimizer_options);
 static RelOptInfo *pg_hint_plan_join_search(PlannerInfo *root,
 											int levels_needed,
 											List *initial_rels);
@@ -3118,7 +3118,7 @@ pg_hint_plan_ProcessUtility(PlannedStmt *pstmt, const char *queryString,
  */
 static PlannedStmt *
 pg_hint_plan_planner(Query *parse, const char *query_string,
-	int cursorOptions, ParamListInfo boundParams)
+	int cursorOptions, ParamListInfo boundParams, OptimizerOptions *optimizer_options)
 {
 	int				save_nestlevel;
 	PlannedStmt	   *result;
@@ -3236,9 +3236,9 @@ pg_hint_plan_planner(Query *parse, const char *query_string,
 		}
 
 		if (prev_planner)
-			result = (*prev_planner) (parse, query_string, cursorOptions, boundParams);
+			result = (*prev_planner) (parse, query_string, cursorOptions, boundParams, optimizer_options);
 		else
-			result = standard_planner(parse, query_string, cursorOptions, boundParams);
+			result = standard_planner(parse, query_string, cursorOptions, boundParams, optimizer_options);
 
 		current_hint_str = prev_hint_str;
 		recurse_level--;
@@ -3298,9 +3298,9 @@ pg_hint_plan_planner(Query *parse, const char *query_string,
 	}
 	current_hint_state = NULL;
 	if (prev_planner)
-		result =  (*prev_planner) (parse, query_string, cursorOptions, boundParams);
+		result =  (*prev_planner) (parse, query_string, cursorOptions, boundParams, optimizer_options);
 	else
-		result = standard_planner(parse, query_string, cursorOptions, boundParams);
+		result = standard_planner(parse, query_string, cursorOptions, boundParams, optimizer_options);
 
 	/* The upper-level planner still needs the current hint state */
 	if (HintStateStack != NIL)

src/backend/cdb/cdbllize.c

@@ -1506,6 +1506,7 @@ motion_sanity_walker(Node *node, sanity_result_t *result)
 	{
 		case T_Result:
 		case T_WindowAgg:
+		case T_WindowHashAgg:
 		case T_TableFunctionScan:
 		case T_ShareInputScan:
 		case T_Append:

src/backend/cdb/cdbplan.c

@@ -694,6 +694,25 @@ plan_tree_mutator(Node *node,
 				return (Node *) newwindow;
 			}
 			break;
+		case T_WindowHashAgg:
+			{
+				WindowHashAgg  *window = (WindowHashAgg *) node;
+				WindowHashAgg  *newwindow;
+
+				FLATCOPY(newwindow, window, WindowHashAgg);
+				PLANMUTATE(newwindow, window);
+
+				COPYARRAY(newwindow, window, partNumCols, partColIdx);
+				COPYARRAY(newwindow, window, partNumCols, partOperators);
+
+				COPYARRAY(newwindow, window, ordNumCols, ordColIdx);
+				COPYARRAY(newwindow, window, ordNumCols, ordOperators);
+				MUTATE(newwindow->startOffset, window->startOffset, Node *);
+				MUTATE(newwindow->endOffset, window->endOffset, Node *);
+
+				return (Node *) newwindow;
+			}
+			break;
 
 		case T_Unique:
 			{

src/backend/cdb/cdbtargeteddispatch.c

@@ -522,6 +522,7 @@ DirectDispatchUpdateContentIdsFromPlan(PlannerInfo *root, Plan *plan)
 			DisableTargetedDispatch(&dispatchInfo);
 			break;
 		case T_WindowAgg:
+		case T_WindowHashAgg:
 		case T_TableFunctionScan:
 		case T_RecursiveUnion:
 			/* no change to dispatchInfo */

src/backend/gpopt/CGPOptimizer.cpp

@@ -45,7 +45,8 @@ PlannedStmt *
 CGPOptimizer::GPOPTOptimizedPlan(
 	Query *query,
 	bool *
-		had_unexpected_failure	// output : set to true if optimizer unexpectedly failed to produce plan
+		had_unexpected_failure,	// output : set to true if optimizer unexpectedly failed to produce plan
+	OptimizerOptions *opts
 )
 {
 	SOptContext gpopt_context;
@@ -55,7 +56,7 @@ CGPOptimizer::GPOPTOptimizedPlan(
 
 	GPOS_TRY
 	{
-		plStmt = COptTasks::GPOPTOptimizedPlan(query, &gpopt_context);
+		plStmt = COptTasks::GPOPTOptimizedPlan(query, &gpopt_context, opts);
 		// clean up context
 		gpopt_context.Free(gpopt_context.epinQuery, gpopt_context.epinPlStmt);
 	}
@@ -199,9 +200,9 @@ CGPOptimizer::TerminateGPOPT()
 //---------------------------------------------------------------------------
 extern "C" {
 PlannedStmt *
-GPOPTOptimizedPlan(Query *query, bool *had_unexpected_failure)
+GPOPTOptimizedPlan(Query *query, bool *had_unexpected_failure, OptimizerOptions *opts)
 {
-	return CGPOptimizer::GPOPTOptimizedPlan(query, had_unexpected_failure);
+	return CGPOptimizer::GPOPTOptimizedPlan(query, had_unexpected_failure, opts);
 }
 }
 

src/backend/gpopt/config/CConfigParamMapping.cpp

@@ -326,6 +326,12 @@ CConfigParamMapping::SConfigMappingElem CConfigParamMapping::m_elements[] = {
 	 false,	 // m_negate_param
 	 GPOS_WSZ_LIT(
 		 "Disable the dynamic seq/bitmap/index scan in partition table")},
+
+	{EopttraceEnableWindowHashAgg, &optimizer_force_window_hash_agg,
+	 false,	 // m_negate_param
+	 GPOS_WSZ_LIT(
+		 "Enable create window hash agg")},
+	
 };
 
 //---------------------------------------------------------------------------
@@ -339,7 +345,8 @@ CConfigParamMapping::SConfigMappingElem CConfigParamMapping::m_elements[] = {
 CBitSet *
 CConfigParamMapping::PackConfigParamInBitset(
 	CMemoryPool *mp,
-	ULONG xform_id	// number of available xforms
+	ULONG xform_id,	// number of available xforms
+	BOOL create_vec_plan
 )
 {
 	CBitSet *traceflag_bitset = GPOS_NEW(mp) CBitSet(mp, EopttraceSentinel);
@@ -561,6 +568,10 @@ CConfigParamMapping::PackConfigParamInBitset(
 			GPOPT_DISABLE_XFORM_TF(CXform::ExfRightOuterJoin2HashJoin));
 	}
 
+	if (create_vec_plan) {
+		traceflag_bitset->ExchangeSet(EopttraceEnableWindowHashAgg);
+	}
+
 	return traceflag_bitset;
 }