ORCA: Support two phase window functions

For TPC-DS 67 queries, due to data skew, most of the data will be motioned to limited segments for processing.
If data is filtered before gathering, it can significantly reduce the amount of data that upstream nodes need to
process and improve query performance. So we introduce two-phase WindowAgg for the case:

explain SELECT * FROM (SELECT rank() OVER (PARTITION BY four ORDER BY ten) AS rank_1, ten, four FROM tenk1 WHERE unique2 < 10) t WHERE rank_1 < 3;
                                                        QUERY PLAN                                                         
---------------------------------------------------------------------------------------------------------------------------
 Gather Motion 3:1  (slice1; segments: 3)  (cost=0.00..0.00 rows=5 width=16)
   ->  Result  (cost=0.00..0.00 rows=2 width=16)
         Filter: ((rank() OVER (?)) < 3)
         ->  WindowAgg  (cost=0.00..0.00 rows=2 width=16)
               Partition By: four
               Order By: ten
               ->  Sort  (cost=0.00..6.00 rows=2 width=8)
                     Sort Key: four, ten
                     ->  Redistribute Motion 3:3  (slice2; segments: 3)  (cost=0.00..6.00 rows=2 width=8)
                           Hash Key: four
                           ->  Result  (cost=0.00..6.00 rows=2 width=8)
                                 Filter: ((rank() OVER (?)) < 3)
                                 ->  WindowAgg  (cost=0.00..6.00 rows=4 width=16)
                                       Partition By: four
                                       Order By: ten
                                       ->  Sort  (cost=0.00..6.00 rows=4 width=8)
                                             Sort Key: four, ten
                                             ->  Index Scan using tenk1_unique2 on tenk1  (cost=0.00..6.00 rows=4 width=8)
                                                   Index Cond: (unique2 < 10)
 Optimizer: GPORCA
(20 rows)

Author: jiaqizho

src/backend/gpopt/config/CConfigParamMapping.cpp

@@ -265,6 +265,11 @@ CConfigParamMapping::SConfigMappingElem CConfigParamMapping::m_elements[] = {
 	 false,	 // m_negate_param
 	 GPOS_WSZ_LIT(
 		 "Always pick a plan for aggregate distinct that minimizes skew.")},
+	
+	{EopttraceForceSplitWindowFunc, &optimizer_force_split_window_function,
+		false,	 // m_negate_param
+		GPOS_WSZ_LIT(
+				"Always split the window function.")},
 
 	{EopttraceEnableEagerAgg, &optimizer_enable_eageragg,
 	 false,	 // m_negate_param

src/backend/gpopt/utils/COptTasks.cpp

@@ -403,7 +403,7 @@ COptTasks::CreateOptimizerConfig(CMemoryPool *mp, ICostModel *cost_model,
 				  push_group_by_below_setop_threshold, xform_bind_threshold,
 				  skew_factor),
 		plan_hints,
-		GPOS_NEW(mp) CWindowOids(OID(F_ROW_NUMBER), OID(F_RANK_)));
+		GPOS_NEW(mp) CWindowOids(mp, OID(F_ROW_NUMBER), OID(F_RANK_), OID(F_DENSE_RANK_)));
 }
 
 //---------------------------------------------------------------------------

src/backend/gporca/libgpdbcost/src/CCostModelGPDB.cpp

@@ -1594,12 +1594,19 @@ CCostModelGPDB::CostSequenceProject(CMemoryPool *mp, CExpressionHandle &exprhdl,
 	GPOS_ASSERT(COperator::EopPhysicalSequenceProject ==
 				exprhdl.Pop()->Eopid());
 
+	CPhysicalSequenceProject *psp = CPhysicalSequenceProject::PopConvert(exprhdl.Pop());
+
+	if (GPOS_FTRACE(EopttraceForceSplitWindowFunc) && 
+		psp->Pspt() == COperator::EsptypeGlobalTwoStep) {
+		return CCost(0);
+	}
+
 	const DOUBLE num_rows_outer = pci->PdRows()[0];
 	const DOUBLE dWidthOuter = pci->GetWidth()[0];
 
 	ULONG ulSortCols = 0;
 	COrderSpecArray *pdrgpos =
-		CPhysicalSequenceProject::PopConvert(exprhdl.Pop())->Pdrgpos();
+		CPhysicalSequenceProject::PopConvert(psp)->Pdrgpos();
 	const ULONG ulOrderSpecs = pdrgpos->Size();
 	for (ULONG ul = 0; ul < ulOrderSpecs; ul++)
 	{
@@ -1619,7 +1626,7 @@ CCostModelGPDB::CostSequenceProject(CMemoryPool *mp, CExpressionHandle &exprhdl,
 								   dTupDefaultProcCostUnit));
 	CCost costChild =
 		CostChildren(mp, exprhdl, pci, pcmgpdb->GetCostModelParams());
-
+	
 	return costLocal + costChild;
 }
 

src/backend/gporca/libgpopt/include/gpopt/base/CUtils.h

@@ -320,12 +320,10 @@ class CUtils
 											BOOL fNewComputedCol);
 
 	// generate a sequence project expression
-	static CExpression *PexprLogicalSequenceProject(CMemoryPool *mp,
-													CDistributionSpec *pds,
-													COrderSpecArray *pdrgpos,
-													CWindowFrameArray *pdrgpwf,
-													CExpression *pexpr,
-													CExpression *pexprPrjList);
+	static CExpression *PexprLogicalSequenceProject(
+		CMemoryPool *mp, COperator::ESPType sptype, CDistributionSpec *pds,
+		COrderSpecArray *pdrgpos, CWindowFrameArray *pdrgpwf,
+		CExpression *pexpr, CExpression *pexprPrjList);
 
 	// generate a projection of NULL constants
 	// to the map 'colref_mapping', and add the mappings to the colref_mapping map if not NULL

src/backend/gporca/libgpopt/include/gpopt/base/CWindowOids.h

@@ -17,10 +17,16 @@
 
 #define DUMMY_ROW_NUMBER_OID OID(7000)
 #define DUMMY_WIN_RANK OID(7001)
+#define DUMMY_WIN_DENSE_RANK OID(7002)
 
+namespace gpmd
+{
+class IMDId;
+}
 
 namespace gpopt
 {
+using namespace gpmd;
 //---------------------------------------------------------------------------
 //	@class:
 //		CWindowOids
@@ -35,17 +41,36 @@ class CWindowOids : public CRefCount
 	// oid of window operation "row_number" function
 	OID m_oidRowNumber;
 
+	// metadata id of window operation "row_number" function
+	IMDId *m_MDIdRowNumber;
+
 	// oid of window operation "rank" function
 	OID m_oidRank;
 
+	// metadata id of window operation "rank" function
+	IMDId *m_MDIdRank;
+
+	// oid of window operation "dense_rank" function
+	OID m_oidDenseRank;
+
+	// metadata id of window operation "dense_rank" function
+	IMDId *m_MDDenseRank;
+
 public:
-	CWindowOids(OID row_number_oid, OID rank_oid);
+	CWindowOids(CMemoryPool *mp, OID row_number_oid, OID rank_oid,
+				OID dense_rank_oid);
 
 	// accessor of oid value of "row_number" function
 	OID OidRowNumber() const;
+	IMDId *MDIdRowNumber() const;
 
 	// accessor of oid value of "rank" function
 	OID OidRank() const;
+	IMDId *MDIdRank() const;
+
+	// accessor of oid value of "dense_rank" function
+	OID OidDenseRank() const;
+	IMDId *MDIdDenseRank() const;
 
 	// generate default window oids
 	static CWindowOids *GetWindowOids(CMemoryPool *mp);

src/backend/gporca/libgpopt/include/gpopt/operators/CExpressionHandle.h

@@ -173,6 +173,13 @@ class CExpressionHandle
 	// stats of attached expr/gexpr
 	IStatistics *Pstats();
 
+	// pop the memory pool
+	CMemoryPool *
+	Pmp() const
+	{
+		return m_mp;
+	}
+
 	// required properties of attached expr/gexpr
 	CReqdProp *
 	Prp() const

src/backend/gporca/libgpopt/include/gpopt/operators/CLogicalSequenceProject.h

@@ -33,6 +33,9 @@ class CDistributionSpec;
 class CLogicalSequenceProject : public CLogicalUnary
 {
 private:
+	// type of sequence project
+	COperator::ESPType m_sptype;
+
 	// partition by keys
 	CDistributionSpec *m_pds;
 
@@ -58,8 +61,8 @@ class CLogicalSequenceProject : public CLogicalUnary
 	CLogicalSequenceProject(const CLogicalSequenceProject &) = delete;
 
 	// ctor
-	CLogicalSequenceProject(CMemoryPool *mp, CDistributionSpec *pds,
-							COrderSpecArray *pdrgpos,
+	CLogicalSequenceProject(CMemoryPool *mp, COperator::ESPType sptype,
+							CDistributionSpec *pds, COrderSpecArray *pdrgpos,
 							CWindowFrameArray *pdrgpwf);
 
 	// ctor for pattern
@@ -82,6 +85,13 @@ class CLogicalSequenceProject : public CLogicalUnary
 		return "CLogicalSequenceProject";
 	}
 
+	// window type
+	COperator::ESPType
+	Pspt() const
+	{
+		return m_sptype;
+	}
+
 	// distribution spec
 	CDistributionSpec *
 	Pds() const
@@ -123,7 +133,8 @@ class CLogicalSequenceProject : public CLogicalUnary
 										  BOOL must_exist) override;
 
 	// return true if we can pull projections up past this operator from its given child
-	BOOL FCanPullProjectionsUp(ULONG  //child_index
+	BOOL
+	FCanPullProjectionsUp(ULONG	 //child_index
 	) const override
 	{
 		return false;
@@ -180,6 +191,9 @@ class CLogicalSequenceProject : public CLogicalUnary
 	// print
 	IOstream &OsPrint(IOstream &os) const override;
 
+	static IOstream &OsPrintWindowType(IOstream &os,
+									   COperator::ESPType wintype);
+
 	// remove outer references from Order By/ Partition By clauses, and return a new operator
 	CLogicalSequenceProject *PopRemoveLocalOuterRefs(
 		CMemoryPool *mp, CExpressionHandle &exprhdl);

src/backend/gporca/libgpopt/include/gpopt/operators/COperator.h

@@ -266,12 +266,25 @@ class COperator : public CRefCount, public DbgPrintMixin<COperator>
 
 		EopLogicalIndexOnlyGet,
 		EopLogicalDynamicIndexOnlyGet,
+
+		EopLogicalWindowFunc,
 		EopSentinel
 	};
 
+	//  sequence project type
+	enum ESPType
+	{
+		EsptypeGlobalTwoStep,	// global group by sequence project
+		EsptypeGlobalOneStep,	// global group by sequence project
+		EsptypeLocal,	// local group by sequence project
+
+		EsptypeSentinel
+	};
+
 	// aggregate type
 	enum EGbAggType
 	{
+		// todo(jiaqizho): change to onestep, twostep(global), twostep(local)
 		EgbaggtypeGlobal,		 // global group by aggregate
 		EgbaggtypeLocal,		 // local group by aggregate
 		EgbaggtypeIntermediate,	 // intermediate group by aggregate

src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalSequenceProject.h

@@ -32,6 +32,9 @@ class CDistributionSpec;
 class CPhysicalSequenceProject : public CPhysical
 {
 private:
+	// window type
+	ESPType m_sptype;
+
 	// partition by keys
 	CDistributionSpec *m_pds;
 
@@ -57,8 +60,8 @@ class CPhysicalSequenceProject : public CPhysical
 	CPhysicalSequenceProject(const CPhysicalSequenceProject &) = delete;
 
 	// ctor
-	CPhysicalSequenceProject(CMemoryPool *mp, CDistributionSpec *pds,
-							 COrderSpecArray *pdrgpos,
+	CPhysicalSequenceProject(CMemoryPool *mp, ESPType m_sptype,
+							 CDistributionSpec *pds, COrderSpecArray *pdrgpos,
 							 CWindowFrameArray *pdrgpwf);
 
 	// dtor
@@ -78,6 +81,13 @@ class CPhysicalSequenceProject : public CPhysical
 		return "CPhysicalSequenceProject";
 	}
 
+	// window type
+	COperator::ESPType
+	Pspt() const
+	{
+		return m_sptype;
+	}
+
 	// partition by keys
 	CDistributionSpec *
 	Pds() const

src/backend/gporca/libgpopt/include/gpopt/xforms/CXform.h

@@ -25,7 +25,8 @@
 #include "naucrates/traceflags/traceflags.h"
 
 // Macro for enabling and disabling xforms
-#define GPOPT_DISABLE_XFORM_TF(x) EopttraceDisableXformBase + static_cast<int>(x)
+#define GPOPT_DISABLE_XFORM_TF(x) \
+	EopttraceDisableXformBase + static_cast<int>(x)
 #define GPOPT_ENABLE_XFORM(x) GPOS_UNSET_TRACE(GPOPT_DISABLE_XFORM_TF(x))
 #define GPOPT_DISABLE_XFORM(x) GPOS_SET_TRACE(GPOPT_DISABLE_XFORM_TF(x))
 #define GPOPT_FENABLED_XFORM(x) !GPOS_FTRACE(GPOPT_DISABLE_XFORM_TF(x))
@@ -236,6 +237,7 @@ class CXform : public CRefCount, public DbgPrintMixin<CXform>
 		ExfLimit2IndexOnlyGet,
 		ExfFullOuterJoin2HashJoin,
 		ExfFullJoinCommutativity,
+		ExfSplitWindowFunc,
 		ExfInvalid,
 		ExfSentinel = ExfInvalid
 	};
@@ -297,7 +299,8 @@ class CXform : public CRefCount, public DbgPrintMixin<CXform>
 	}
 
 	// check compatibility with another xform
-	virtual BOOL FCompatible(CXform::EXformId)
+	virtual BOOL
+	FCompatible(CXform::EXformId)
 	{
 		return true;
 	}