ORCA: Support pushdown partial aggregate below join

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

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

@@ -82,10 +82,6 @@ class CXformUtils
 													CTableDescriptor *ptabdesc,
 													CColRefArray *colref_array);
 
-	// helper for extracting foreign key
-	static CColRefSet *PcrsFKey(CMemoryPool *mp, CExpressionArray *pdrgpexpr,
-								CColRefSet *prcsOutput, CColRefSet *pcrsKey);
-
 	// return the set of columns from the given array of columns which appear
 	// in the index included / key columns
 	static CColRefSet *PcrsIndexColumns(CMemoryPool *mp,
@@ -172,13 +168,6 @@ class CXformUtils
 													CColRefSet *prcsOutput,
 													CColRefSet *pcrsGrpCols);
 
-	// check if the preconditions for pushing down Group by through join are satisfied
-	static BOOL FCanPushGbAggBelowJoin(CColRefSet *pcrsGrpCols,
-									   CColRefSet *pcrsJoinOuterChildOutput,
-									   CColRefSet *pcrsJoinScalarUsedFromOuter,
-									   CColRefSet *pcrsGrpByOutput,
-									   CColRefSet *pcrsGrpByUsed,
-									   CColRefSet *pcrsFKey);
 	// create a dynamic operator for a btree index plan
 	template <class T>
 	static CLogical *
@@ -295,11 +284,23 @@ class CXformUtils
 	static CXform::EXformPromise ExfpExpandJoinOrder(CExpressionHandle &exprhdl,
 													 const CXform *xform);
 
+	// extract foreign key
+	static CColRefSet *PcrsFKey(CMemoryPool *mp, CExpressionArray *pdrgpexpr,
+								CColRefSet *prcsOutput, CColRefSet *pcrsKey);
+
 	// extract foreign key
 	static CColRefSet *PcrsFKey(CMemoryPool *mp, CExpression *pexprOuter,
 								CExpression *pexprInner,
 								CExpression *pexprScalar);
 
+	// check if the preconditions for pushing down Group by through join are satisfied
+	static BOOL FCanPushGbAggBelowJoin(CColRefSet *pcrsGrpCols,
+									   CColRefSet *pcrsJoinOuterChildOutput,
+									   CColRefSet *pcrsJoinScalarUsedFromOuter,
+									   CColRefSet *pcrsGrpByOutput,
+									   CColRefSet *pcrsGrpByUsed,
+									   CColRefSet *pcrsFKey);
+
 	// compute a swap of the two given joins
 	static CExpression *PexprSwapJoins(CMemoryPool *mp,
 									   CExpression *pexprTopJoin,

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

@@ -132,6 +132,7 @@
 #include "gpopt/xforms/CXformPushGbBelowJoin.h"
 #include "gpopt/xforms/CXformPushGbBelowUnion.h"
 #include "gpopt/xforms/CXformPushGbBelowUnionAll.h"
+#include "gpopt/xforms/CXformPushPartialAggBelowJoin.h"
 #include "gpopt/xforms/CXformPushGbDedupBelowJoin.h"
 #include "gpopt/xforms/CXformPushGbWithHavingBelowJoin.h"
 #include "gpopt/xforms/CXformPushJoinBelowLeftUnionAll.h"

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

@@ -40,7 +40,8 @@ CLogicalGbAgg::CLogicalGbAgg(CMemoryPool *mp)
 	  m_pdrgpcr(nullptr),
 	  m_pdrgpcrMinimal(nullptr),
 	  m_egbaggtype(COperator::EgbaggtypeSentinel),
-	  m_aggStage(EasOthers)
+	  m_aggStage(EasOthers),
+	  m_aggpushdown(false)
 {
 	m_fPattern = true;
 }
@@ -62,7 +63,8 @@ CLogicalGbAgg::CLogicalGbAgg(CMemoryPool *mp, CColRefArray *colref_array,
 	  m_pdrgpcr(colref_array),
 	  m_pdrgpcrMinimal(nullptr),
 	  m_egbaggtype(egbaggtype),
-	  m_aggStage(EasOthers)
+	  m_aggStage(EasOthers),
+	  m_aggpushdown(false)
 {
 	if (COperator::EgbaggtypeLocal == egbaggtype)
 	{
@@ -86,7 +88,8 @@ CLogicalGbAgg::CLogicalGbAgg(CMemoryPool *mp, CColRefArray *colref_array,
 	  m_pdrgpcr(colref_array),
 	  m_pdrgpcrMinimal(nullptr),
 	  m_egbaggtype(egbaggtype),
-	  m_aggStage(aggStage)
+	  m_aggStage(aggStage),
+	  m_aggpushdown(false)
 {
 	if (COperator::EgbaggtypeLocal == egbaggtype)
 	{
@@ -120,7 +123,8 @@ CLogicalGbAgg::CLogicalGbAgg(CMemoryPool *mp, CColRefArray *colref_array,
 	  m_pdrgpcr(colref_array),
 	  m_pdrgpcrMinimal(nullptr),
 	  m_egbaggtype(egbaggtype),
-	  m_aggStage(EasOthers)
+	  m_aggStage(EasOthers),
+	  m_aggpushdown(false)
 {
 	GPOS_ASSERT(nullptr != colref_array);
 	GPOS_ASSERT(COperator::EgbaggtypeSentinel > egbaggtype);
@@ -142,7 +146,8 @@ CLogicalGbAgg::CLogicalGbAgg(CMemoryPool *mp, CColRefArray *colref_array,
 	  m_pdrgpcr(colref_array),
 	  m_pdrgpcrMinimal(nullptr),
 	  m_egbaggtype(egbaggtype),
-	  m_aggStage(aggStage)
+	  m_aggStage(aggStage),
+	  m_aggpushdown(false)
 {
 	GPOS_ASSERT(nullptr != colref_array);
 	GPOS_ASSERT(COperator::EgbaggtypeSentinel > egbaggtype);
@@ -171,7 +176,8 @@ CLogicalGbAgg::CLogicalGbAgg(CMemoryPool *mp, CColRefArray *colref_array,
 	  m_pdrgpcr(colref_array),
 	  m_pdrgpcrMinimal(pdrgpcrMinimal),
 	  m_egbaggtype(egbaggtype),
-	  m_aggStage(EasOthers)
+	  m_aggStage(EasOthers),
+	  m_aggpushdown(false)
 {
 	GPOS_ASSERT(nullptr != colref_array);
 	GPOS_ASSERT(COperator::EgbaggtypeSentinel > egbaggtype);
@@ -209,7 +215,8 @@ CLogicalGbAgg::CLogicalGbAgg(CMemoryPool *mp, CColRefArray *colref_array,
 	  m_pdrgpcr(colref_array),
 	  m_pdrgpcrMinimal(pdrgpcrMinimal),
 	  m_egbaggtype(egbaggtype),
-	  m_aggStage(EasOthers)
+	  m_aggStage(EasOthers),
+	  m_aggpushdown(false)
 {
 	GPOS_ASSERT(nullptr != colref_array);
 	GPOS_ASSERT(COperator::EgbaggtypeSentinel > egbaggtype);
@@ -592,6 +599,7 @@ CLogicalGbAgg::PxfsCandidates(CMemoryPool *mp) const
 	(void) xform_set->ExchangeSet(CXform::ExfPushGbBelowUnionAll);
 	if (FGlobal())
 	{
+		(void) xform_set->ExchangeSet(CXform::ExfPushPartialAggBelowJoin);
 		(void) xform_set->ExchangeSet(CXform::ExfSplitGbAgg);
 	}
 	(void) xform_set->ExchangeSet(CXform::ExfSplitDQA);
@@ -698,7 +706,7 @@ CLogicalGbAgg::OsPrint(IOstream &os) const
 
 	os << SzId() << "( ";
 	OsPrintGbAggType(os, m_egbaggtype);
-	os << " )";
+	os << (m_aggpushdown ? ", pushdown )" : " )");
 	os << " Grp Cols: [";
 	CUtils::OsPrintDrgPcr(os, m_pdrgpcr);
 	os << "]"

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

@@ -39,12 +39,14 @@ CPhysicalAgg::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)
 	: CPhysical(mp),
 	  m_pdrgpcr(colref_array),
 	  m_egbaggtype(egbaggtype),
 	  m_isAggFromSplitDQA(isAggFromSplitDQA),
 	  m_aggStage(aggStage),
+	  m_aggpushdown(isAggPushdown),
 	  m_pdrgpcrMinimal(nullptr),
 	  m_fGeneratesDuplicates(fGeneratesDuplicates),
 	  m_pdrgpcrArgDQA(pdrgpcrArgDQA),
@@ -696,6 +698,10 @@ CPhysicalAgg::OsPrint(IOstream &os) const
 	{
 		os << ", multi-stage";
 	}
+	if (m_aggpushdown)
+	{
+		os << ", pushdown";
+	}
 	os << " )";
 
 

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

@@ -33,10 +33,12 @@ CPhysicalHashAgg::CPhysicalHashAgg(
 	CMemoryPool *mp, CColRefArray *colref_array, CColRefArray *pdrgpcrMinimal,
 	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)
 	: CPhysicalAgg(mp, colref_array, pdrgpcrMinimal, egbaggtype,
 				   fGeneratesDuplicates, pdrgpcrArgDQA, fMultiStage,
-				   isAggFromSplitDQA, aggStage, should_enforce_distribution)
+				   isAggFromSplitDQA, aggStage, isAggPushdown,
+				   should_enforce_distribution)
 {
 }