ORCA: Support pushdown partial aggregate below join

Author: jiaqizho

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

@@ -131,6 +131,20 @@ class CLogicalGbAgg : public CLogicalUnary
 		return m_fGeneratesDuplicates;
 	}
 
+	// dose this aggregate generated by pushdown
+	BOOL
+	FAggPushdown() const
+	{
+		return m_aggpushdown;
+	}
+
+	// this aggregate generated by pushdown
+	void
+	MarkAggPushdown()
+	{
+		m_aggpushdown = true;
+	}
+
 	// match function
 	BOOL Matches(COperator *pop) const override;
 
@@ -274,6 +288,8 @@ class CLogicalGbAgg : public CLogicalUnary
 	// which type of multi-stage agg it is
 	EAggStage m_aggStage;
 
+	// is current agg pushdown?
+	BOOL m_aggpushdown;
 };	// class CLogicalGbAgg
 
 }  // namespace gpopt

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

@@ -44,6 +44,9 @@ class CPhysicalAgg : public CPhysical
 
 	CLogicalGbAgg::EAggStage m_aggStage;
 
+	// is aggregate gernerated by pushdown
+	BOOL m_aggpushdown;
+
 	// compute required distribution of the n-th child of an intermediate aggregate
 	CDistributionSpec *PdsRequiredIntermediateAgg(CMemoryPool *mp,
 												  ULONG ulOptReq) const;
@@ -111,7 +114,7 @@ class CPhysicalAgg : public CPhysical
 				 COperator::EGbAggType egbaggtype, BOOL fGeneratesDuplicates,
 				 CColRefArray *pdrgpcrArgDQA, BOOL fMultiStage,
 				 BOOL isAggFromSplitDQA, CLogicalGbAgg::EAggStage aggStage,
-				 BOOL should_enforce_distribution);
+				 BOOL isAggPushdown, BOOL should_enforce_distribution);
 
 	// is this agg generated by CXformSplitDQA
 	BOOL IsAggFromSplitDQA() const;

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

@@ -41,7 +41,7 @@ class CPhysicalHashAgg : public CPhysicalAgg
 					 BOOL fGeneratesDuplicates, CColRefArray *pdrgpcrArgDQA,
 					 BOOL fMultiStage, BOOL isAggFromSplitDQA,
 					 CLogicalGbAgg::EAggStage aggStage,
-					 BOOL should_enforce_distribution = true
+					 BOOL isAggPushdown, BOOL should_enforce_distribution
 					 // should_enforce_distribution should be set to false if
 					 // 'local' and 'global' splits don't need to have different
 					 // distributions. This flag is set to false if the local

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

@@ -42,6 +42,7 @@ class CPhysicalHashAggDeduplicate : public CPhysicalHashAgg
 								BOOL fGeneratesDuplicates, BOOL fMultiStage,
 								BOOL isAggFromSplitDQA,
 								CLogicalGbAgg::EAggStage aggStage,
+								BOOL isAggPushdown,
 								BOOL should_enforce_distribution);
 
 	// dtor

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

@@ -40,7 +40,8 @@ class CPhysicalScalarAgg : public CPhysicalAgg
 		CColRefArray *pdrgpcrMinimal,  // minimal grouping columns based on FD's
 		COperator::EGbAggType egbaggtype, BOOL fGeneratesDuplicates,
 		CColRefArray *pdrgpcrArgDQA, BOOL fMultiStage, BOOL isAggFromSplitDQA,
-		CLogicalGbAgg::EAggStage aggStage, BOOL should_enforce_distribution);
+		CLogicalGbAgg::EAggStage aggStage, BOOL isAggPushdown,
+		BOOL should_enforce_distribution);
 
 	// dtor
 	~CPhysicalScalarAgg() override;

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

@@ -60,8 +60,8 @@ class CPhysicalStreamAgg : public CPhysicalAgg
 		CColRefArray *pdrgpcrMinimal,  // minimal grouping columns based on FD's
 		COperator::EGbAggType egbaggtype, BOOL fGeneratesDuplicates,
 		CColRefArray *pdrgpcrArgDQA, BOOL fMultiStage, BOOL isAggFromSplitDQA,
-		CLogicalGbAgg::EAggStage aggStage,
-		BOOL should_enforce_distribution = true
+		CLogicalGbAgg::EAggStage aggStage, BOOL isAggPushdown,
+		BOOL should_enforce_distribution
 		// should_enforce_distribution should be set to false if
 		// 'local' and 'global' splits don't need to have different
 		// distributions. This flag is set to false if the local

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

@@ -43,6 +43,7 @@ class CPhysicalStreamAggDeduplicate : public CPhysicalStreamAgg
 								  BOOL fGeneratesDuplicates, BOOL fMultiStage,
 								  BOOL isAggFromSplitDQA,
 								  CLogicalGbAgg::EAggStage aggStage,
+								  BOOL isAggPushdown,
 								  BOOL should_enforce_distribution);
 
 	// dtor

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

@@ -165,6 +165,7 @@ class CXform : public CRefCount, public DbgPrintMixin<CXform>
 		ExfPushGbWithHavingBelowJoin,
 		ExfPushGbBelowUnion,
 		ExfPushGbBelowUnionAll,
+		ExfPushPartialAggBelowJoin,
 		ExfSplitGbAgg,
 		ExfSplitGbAggDedup,
 		ExfSplitDQA,

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

@@ -0,0 +1,97 @@
+//---------------------------------------------------------------------------
+//	Greenplum Database
+//	Copyright (C) 2012 EMC Corp.
+//
+//	@filename:
+//		CXformPushPartialAggBelowJoin.h
+//
+//	@doc:
+//		Push group by below join transform
+//---------------------------------------------------------------------------
+#ifndef GPOPT_CXformPushPartialAggBelowJoin_H
+#define GPOPT_CXformPushPartialAggBelowJoin_H
+
+#include <tuple>
+
+#include "gpos/base.h"
+
+#include "gpopt/xforms/CXformExploration.h"
+namespace gpopt
+{
+using namespace gpos;
+
+//---------------------------------------------------------------------------
+//	@class:
+//		CXformPushPartialAggBelowJoin
+//
+//	@doc:
+//		Push group by below join transform
+//
+//---------------------------------------------------------------------------
+class CXformPushPartialAggBelowJoin : public CXformExploration
+{
+private:
+	static BOOL FLocalGbAggAlreadyPushed(CExpression *pexprGlobalGb);
+
+	static std::pair<BOOL, BOOL> FCanPushLocalGbAggBelowJoin(
+		CMemoryPool *mp, CExpression *pexpr);
+
+	static CExpression *ExchangeGbkeyFromJoinKey(CMemoryPool *mp,
+												 CExpression *pexpr);
+	static CColRefSet *PcrsJoinKey(CMemoryPool *mp, CExpression *pexprOuter,
+								   CExpression *pexprInner,
+								   CExpression *pexprScalar);
+
+	static CExpression *PushLocalGbAggBelowJoin(CMemoryPool *mp,
+												CExpression *pexprGlobalGb);
+
+	static CColRefSet *PexprGetGbAggkey(CMemoryPool *mp,
+										CExpression *pexprGbAgg);
+
+public:
+	CXformPushPartialAggBelowJoin(const CXformPushPartialAggBelowJoin &) =
+		delete;
+
+	// ctor
+	explicit CXformPushPartialAggBelowJoin(CMemoryPool *mp);
+
+	// ctor
+	explicit CXformPushPartialAggBelowJoin(CExpression *pexprPattern);
+
+	// dtor
+	~CXformPushPartialAggBelowJoin() override = default;
+
+	// ident accessors
+	EXformId
+	Exfid() const override
+	{
+		return ExfPushPartialAggBelowJoin;
+	}
+
+	const CHAR *
+	SzId() const override
+	{
+		return "CXformPushPartialAggBelowJoin";
+	}
+
+	// Compatibility function
+	BOOL
+	FCompatible(CXform::EXformId exfid) override
+	{
+		return (CXform::ExfPushPartialAggBelowJoin != 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 CXformPushPartialAggBelowJoin
+
+}  // namespace gpopt
+
+#endif	// !GPOPT_CXformPushPartialAggBelowJoin_H
+
+// EOF

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

@@ -117,7 +117,8 @@ class CXformSplitDQA : public CXformExploration
 	FCompatible(CXform::EXformId exfid) override
 	{
 		return (CXform::ExfSplitDQA != exfid) &&
-			   (CXform::ExfSplitGbAgg != exfid);
+			   (CXform::ExfSplitGbAgg != exfid) &&
+			   (CXform::ExfPushPartialAggBelowJoin != exfid);
 	}
 
 	// compute xform promise for a given expression handle