[ORCA] Support two step window function

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/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

@@ -318,12 +318,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/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;
 	}

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

@@ -0,0 +1,112 @@
+/*-------------------------------------------------------------------------
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * 
+ *  @filename:
+ * 		CXformSplitWindowFunc.h
+ * 
+ *  @doc:
+ * 		Split a window function into pair of local and global window function
+ * 
+ *-------------------------------------------------------------------------
+ */
+#ifndef GPOPT_CXformSplitWindowFunc_H
+#define GPOPT_CXformSplitWindowFunc_H
+
+#include "gpos/base.h"
+
+#include "gpopt/xforms/CXformExploration.h"
+
+namespace gpopt
+{
+using namespace gpos;
+
+//---------------------------------------------------------------------------
+//	@class:
+//		CXformSplitWindowFunc
+//
+//	@doc:
+//		Split a window function into pair of local and global window function
+//
+//---------------------------------------------------------------------------
+class CXformSplitWindowFunc : public CXformExploration
+{
+private:
+	// generate a project lists for the local and global window function
+	// from the original window function
+	static CExpression *PexprWindowFunc(
+		CMemoryPool *mp,  // memory pool
+		CExpression *
+			pexprProjListOrig,	// project list of the original global window function
+		CExpression *
+			ppexprProjListLocal,  // project list of the new local window function
+		CExpression *
+			ppexprProjListGlobal  // project list of the new global window function
+	);
+
+	static void PopulateLocalGlobalProjectList(
+		CMemoryPool *mp, CExpression *pexprProjList,
+		CExpression **ppexprProjListLocal, CExpression **ppexprProjListGlobal);
+
+	static void PopulateSelect(
+			CMemoryPool *mp, CExpression *pexprSelect,
+			COperator **ppSelectCopy);
+public:
+	CXformSplitWindowFunc(const CXformSplitWindowFunc &) = delete;
+
+	// ctor
+	explicit CXformSplitWindowFunc(CMemoryPool *mp);
+
+	// dtor
+	~CXformSplitWindowFunc() override = default;
+
+	// ident accessors
+	EXformId
+	Exfid() const override
+	{
+		return ExfSplitWindowFunc;
+	}
+
+	// return a string for xform name
+	const CHAR *
+	SzId() const override
+	{
+		return "CXformSplitWindowFunc";
+	}
+
+	// Compatibility function for splitting limit
+	BOOL
+	FCompatible(CXform::EXformId exfid) override
+	{
+		return (CXform::ExfSplitWindowFunc != exfid);
+	}
+
+	// compute xform promise for a given expression handle
+	EXformPromise Exfp(CExpressionHandle &exprhdl) const override;
+
+	// actual transform
+	void Transform(CXformContext *pxfctxt, CXformResult *pxfres,
+				   CExpression *pexpr) const override;
+
+};	// class CXformSplitWindowFunc
+
+}  // namespace gpopt
+
+#endif	// !GPOPT_CXformSplitWindowFunc_H
+
+// EOF

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

@@ -160,6 +160,7 @@
 #include "gpopt/xforms/CXformSplitGbAgg.h"
 #include "gpopt/xforms/CXformSplitGbAggDedup.h"
 #include "gpopt/xforms/CXformSplitLimit.h"
+#include "gpopt/xforms/CXformSplitWindowFunc.h"
 #include "gpopt/xforms/CXformSubqJoin2Apply.h"
 #include "gpopt/xforms/CXformSubqNAryJoin2Apply.h"
 #include "gpopt/xforms/CXformUnion2UnionAll.h"

src/backend/gporca/libgpopt/src/base/CUtils.cpp

@@ -2267,7 +2267,8 @@ CUtils::PexprLogicalProject(CMemoryPool *mp, CExpression *pexpr,
 
 // generate a sequence project expression
 CExpression *
-CUtils::PexprLogicalSequenceProject(CMemoryPool *mp, CDistributionSpec *pds,
+CUtils::PexprLogicalSequenceProject(CMemoryPool *mp, COperator::ESPType sptype,
+									CDistributionSpec *pds,
 									COrderSpecArray *pdrgpos,
 									CWindowFrameArray *pdrgpwf,
 									CExpression *pexpr,
@@ -2283,7 +2284,8 @@ CUtils::PexprLogicalSequenceProject(CMemoryPool *mp, CDistributionSpec *pds,
 				pexprPrjList->Pop()->Eopid());
 
 	return GPOS_NEW(mp) CExpression(
-		mp, GPOS_NEW(mp) CLogicalSequenceProject(mp, pds, pdrgpos, pdrgpwf),
+		mp,
+		GPOS_NEW(mp) CLogicalSequenceProject(mp, sptype, pds, pdrgpos, pdrgpwf),
 		pexpr, pexprPrjList);
 }
 
@@ -3855,9 +3857,9 @@ CUtils::PexprFuncElemExpr(CMemoryPool *mp, CMDAccessor *md_accessor,
 		CWStringConst(mp, (cast_func->Mdname().GetMDName())->GetBuffer());
 	mdid_func->AddRef();
 	cast_func->GetResultTypeMdid()->AddRef();
-	CScalarFunc *popCastScalarFunc =
-		GPOS_NEW(mp) CScalarFunc(mp, mdid_func, cast_func->GetResultTypeMdid(),
-								 typmod, pstrFunc, 1 /* Explicit Cast */, false /* funcvariadic */);
+	CScalarFunc *popCastScalarFunc = GPOS_NEW(mp)
+		CScalarFunc(mp, mdid_func, cast_func->GetResultTypeMdid(), typmod,
+					pstrFunc, 1 /* Explicit Cast */, false /* funcvariadic */);
 	mdid_elem_type->AddRef();
 	CExpression *pexprCaseTest = GPOS_NEW(mp)
 		CExpression(mp, GPOS_NEW(mp) CScalarCaseTest(mp, mdid_elem_type));

src/backend/gporca/libgpopt/src/operators/CLogicalSelect.cpp

@@ -102,7 +102,8 @@ CLogicalSelect::PxfsCandidates(CMemoryPool *mp) const
 	(void) xform_set->ExchangeSet(CXform::ExfSelect2DynamicBitmapBoolOp);
 	(void) xform_set->ExchangeSet(CXform::ExfSimplifySelectWithSubquery);
 	(void) xform_set->ExchangeSet(CXform::ExfSelect2Filter);
-
+	if (GPOS_FTRACE(EopttraceForceSplitWindowFunc))
+		(void) xform_set->ExchangeSet(CXform::ExfSplitWindowFunc);
 	return xform_set;
 }