Introduce hash windows aggregation when enable vectorization

Author: jiaqizho

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_enable_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;
 }
 

src/backend/gpopt/translate/CTranslatorDXLToPlStmt.cpp

@@ -3336,6 +3336,8 @@ CTranslatorDXLToPlStmt::TranslateDXLWindow(
 		}
 	}
 
+	window->isWindowHashAgg = window_dxlop->IsWindowHashAgg();
+
 	SetParamIds(plan);
 
 	// cleanup

src/backend/gpopt/utils/COptTasks.cpp

@@ -913,7 +913,7 @@ COptTasks::OptimizeTask(void *ptr)
 	{
 		// set trace flags
 		trace_flags = CConfigParamMapping::PackConfigParamInBitset(
-			mp, CXform::ExfSentinel);
+			mp, CXform::ExfSentinel, opt_ctxt->m_create_vec_plan);
 		SetTraceflags(mp, trace_flags, &enabled_trace_flags,
 					  &disabled_trace_flags);
 
@@ -1156,13 +1156,15 @@ COptTasks::Optimize(Query *query)
 //
 //---------------------------------------------------------------------------
 PlannedStmt *
-COptTasks::GPOPTOptimizedPlan(Query *query, SOptContext *gpopt_context)
+COptTasks::GPOPTOptimizedPlan(Query *query, SOptContext *gpopt_context, OptimizerOptions *opts)
 {
 	Assert(query);
 	Assert(gpopt_context);
 
 	gpopt_context->m_query = query;
 	gpopt_context->m_should_generate_plan_stmt = true;
+	// Copy options in `OptimizerOptions` to `SOptContext`
+	gpopt_context->m_create_vec_plan = opts->create_vectorization_plan;
 	Execute(&OptimizeTask, gpopt_context);
 	return gpopt_context->m_plan_stmt;
 }

src/backend/gporca/libgpdbcost/include/gpdbcost/CCostModelGPDB.h

@@ -107,6 +107,12 @@ class CCostModelGPDB : public ICostModel
 									 const CCostModelGPDB *pcmgpdb,
 									 const SCostingInfo *pci);
 
+	// cost of hash sequence project
+	static CCost CostHashSequenceProject(CMemoryPool *mp, 
+									CExpressionHandle &exprhdl,
+									const CCostModelGPDB *pcmgpdb,
+									const SCostingInfo *pci);
+
 	// cost of CTE producer
 	static CCost CostCTEProducer(CMemoryPool *mp, CExpressionHandle &exprhdl,
 								 const CCostModelGPDB *pcmgpdb,

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

@@ -31,6 +31,7 @@
 #include "gpopt/operators/CPhysicalMotionBroadcast.h"
 #include "gpopt/operators/CPhysicalPartitionSelector.h"
 #include "gpopt/operators/CPhysicalSequenceProject.h"
+#include "gpopt/operators/CPhysicalHashSequenceProject.h"
 #include "gpopt/operators/CPhysicalStreamAgg.h"
 #include "gpopt/operators/CPhysicalUnionAll.h"
 #include "gpopt/operators/CPredicateUtils.h"
@@ -1594,11 +1595,17 @@ CCostModelGPDB::CostSequenceProject(CMemoryPool *mp, CExpressionHandle &exprhdl,
 	GPOS_ASSERT(COperator::EopPhysicalSequenceProject ==
 				exprhdl.Pop()->Eopid());
 
+	const CDouble dTupDefaultProcCostUnit =
+		pcmgpdb->GetCostModelParams()
+			->PcpLookup(CCostModelParamsGPDB::EcpTupDefaultProcCostUnit)
+			->Get();
+	GPOS_ASSERT(0 < dTupDefaultProcCostUnit);
+
 	CPhysicalSequenceProject *psp = CPhysicalSequenceProject::PopConvert(exprhdl.Pop());
 
-	if (GPOS_FTRACE(EopttraceForceSplitWindowFunc) && 
+	if (GPOS_FTRACE(EopttraceForceSplitWindowFunc) &&
 		psp->Pspt() == COperator::EsptypeGlobalTwoStep) {
-		return CCost(0);
+		return CCost(dTupDefaultProcCostUnit * 2);
 	}
 
 	const DOUBLE num_rows_outer = pci->PdRows()[0];
@@ -1614,12 +1621,59 @@ CCostModelGPDB::CostSequenceProject(CMemoryPool *mp, CExpressionHandle &exprhdl,
 		ulSortCols += pos->UlSortColumns();
 	}
 
+	// we process (sorted window of) input tuples to compute window function values
+	CCost costLocal =
+		CCost(pci->NumRebinds() * (ulSortCols * num_rows_outer * dWidthOuter *
+								   dTupDefaultProcCostUnit));
+	CCost costChild =
+		CostChildren(mp, exprhdl, pci, pcmgpdb->GetCostModelParams());
+	
+	return costLocal + costChild;
+}
+
+//---------------------------------------------------------------------------
+//	@function:
+//		CCostModelGPDB::CostSequenceProject
+//
+//	@doc:
+//		Cost of sequence project
+//
+//---------------------------------------------------------------------------
+CCost
+CCostModelGPDB::CostHashSequenceProject(CMemoryPool *mp, CExpressionHandle &exprhdl,
+									const CCostModelGPDB *pcmgpdb,
+									const SCostingInfo *pci)
+{
+	GPOS_ASSERT(nullptr != pcmgpdb);
+	GPOS_ASSERT(nullptr != pci);
+	GPOS_ASSERT(COperator::EopPhysicalHashSequenceProject ==
+				exprhdl.Pop()->Eopid());
 	const CDouble dTupDefaultProcCostUnit =
 		pcmgpdb->GetCostModelParams()
 			->PcpLookup(CCostModelParamsGPDB::EcpTupDefaultProcCostUnit)
 			->Get();
 	GPOS_ASSERT(0 < dTupDefaultProcCostUnit);
 
+	CPhysicalHashSequenceProject *psp = CPhysicalHashSequenceProject::PopConvert(exprhdl.Pop());
+
+	if (GPOS_FTRACE(EopttraceForceSplitWindowFunc) && 
+		psp->Pspt() == COperator::EsptypeGlobalTwoStep) {
+		return CCost(dTupDefaultProcCostUnit);
+	}
+
+	const DOUBLE num_rows_outer = pci->PdRows()[0];
+	const DOUBLE dWidthOuter = pci->GetWidth()[0];
+
+	ULONG ulSortCols = 0;
+	COrderSpecArray *pdrgpos =
+		CPhysicalHashSequenceProject::PopConvert(psp)->Pdrgpos();
+	const ULONG ulOrderSpecs = pdrgpos->Size();
+	for (ULONG ul = 0; ul < ulOrderSpecs; ul++)
+	{
+		COrderSpec *pos = (*pdrgpos)[ul];
+		ulSortCols += pos->UlSortColumns();
+	}
+
 	// we process (sorted window of) input tuples to compute window function values
 	CCost costLocal =
 		CCost(pci->NumRebinds() * (ulSortCols * num_rows_outer * dWidthOuter *
@@ -2462,18 +2516,26 @@ CCostModelGPDB::Cost(
 			return CostSequenceProject(m_mp, exprhdl, this, pci);
 		}
 
+		case COperator::EopPhysicalHashSequenceProject:
+		{
+			return CostHashSequenceProject(m_mp, exprhdl, this, pci);
+		}
+
 		case COperator::EopPhysicalCTEProducer:
 		{
 			return CostCTEProducer(m_mp, exprhdl, this, pci);
 		}
+
 		case COperator::EopPhysicalCTEConsumer:
 		{
 			return CostCTEConsumer(m_mp, exprhdl, this, pci);
 		}
+
 		case COperator::EopPhysicalConstTableGet:
 		{
 			return CostConstTableGet(m_mp, exprhdl, this, pci);
 		}
+
 		case COperator::EopPhysicalDML:
 		{
 			return CostDML(m_mp, exprhdl, this, pci);

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

@@ -179,6 +179,9 @@ class CDistributionSpec : public CPropSpec
 	}
 	// print
 	IOstream &OsPrint(IOstream &os) const override = 0;
+	virtual IOstream &OsPrintNoWrap(IOstream &os) const{
+		return OsPrint(os);
+	}
 
 	// return distribution partitioning type
 	virtual EDistributionPartitioningType Edpt() const = 0;

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

@@ -178,7 +178,8 @@ class CDistributionSpecHashed : public CDistributionSpecRandom
 
 	// print
 	IOstream &OsPrint(IOstream &os) const override;
-	IOstream &OsPrintWithPrefix(IOstream &os, const char *prefix) const;
+	IOstream &OsPrintNoWrap(IOstream &os) const override;
+	IOstream &OsPrintWithPrefix(IOstream &os, const char *prefix, BOOL wrap = true) const;
 
 	// return a hashed distribution on the maximal hashable subset of given columns
 	static CDistributionSpecHashed *PdshashedMaximal(CMemoryPool *mp,

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

@@ -213,6 +213,7 @@ class COperator : public CRefCount, public DbgPrintMixin<COperator>
 		EopPhysicalCTEProducer,
 		EopPhysicalCTEConsumer,
 		EopPhysicalSequenceProject,
+		EopPhysicalHashSequenceProject,
 		EopPhysicalDynamicIndexScan,
 
 		EopPhysicalInnerHashJoin,

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

@@ -0,0 +1,123 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ *
+ * CPhysicalHashSequenceProject.h
+ *
+ * IDENTIFICATION
+ *	  src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalHashSequenceProject.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef GPOPT_CPhysicalHashSequenceProject_H
+#define GPOPT_CPhysicalHashSequenceProject_H
+
+#include "gpos/base.h"
+
+#include "gpopt/base/CWindowFrame.h"
+#include "gpopt/operators/CPhysicalSequenceProject.h"
+
+namespace gpopt
+{
+// fwd declarations
+class CDistributionSpec;
+
+//---------------------------------------------------------------------------
+//	@class:
+//		CPhysicalHashSequenceProject
+//
+//	@doc:
+//		Physical Hash Sequence Project operator
+//
+//---------------------------------------------------------------------------
+class CPhysicalHashSequenceProject : public CPhysicalSequenceProject
+{
+public:
+	CPhysicalHashSequenceProject(const CPhysicalHashSequenceProject &) = delete;
+
+	// ctor
+	CPhysicalHashSequenceProject(CMemoryPool *mp, ESPType m_sptype,
+								 CDistributionSpec *pds,
+								 COrderSpecArray *pdrgpos,
+								 CWindowFrameArray *pdrgpwf);
+
+	// dtor
+	~CPhysicalHashSequenceProject() override;
+
+	// ident accessors
+	EOperatorId
+	Eopid() const override
+	{
+		return EopPhysicalHashSequenceProject;
+	}
+
+	// operator name
+	const CHAR *
+	SzId() const override
+	{
+		return "CPhysicalHashSequenceProject";
+	}
+
+	// match function
+	BOOL Matches(COperator *pop) const override;
+
+	// hashing function
+	ULONG HashValue() const override;
+
+	// sensitivity to order of inputs
+	BOOL FInputOrderSensitive() const override;
+
+	// compute required sort order of the n-th child
+	COrderSpec *PosRequired(CMemoryPool *mp, CExpressionHandle &exprhdl,
+							COrderSpec *posRequired, ULONG child_index,
+							CDrvdPropArray *pdrgpdpCtxt,
+							ULONG ulOptReq) const override;
+
+	// return order property enforcing type for this operator
+	CEnfdProp::EPropEnforcingType EpetOrder(
+		CExpressionHandle &exprhdl, const CEnfdOrder *peo) const override;
+
+	// return true if operator passes through stats obtained from children,
+	// this is used when computing stats during costing
+	BOOL
+	FPassThruStats() const override
+	{
+		return true;
+	}
+
+	// print
+	IOstream &OsPrint(IOstream &os) const override;
+
+	// conversion function
+	static CPhysicalHashSequenceProject *
+	PopConvert(COperator *pop)
+	{
+		GPOS_ASSERT(nullptr != pop);
+		GPOS_ASSERT(EopPhysicalHashSequenceProject == pop->Eopid());
+
+		return dynamic_cast<CPhysicalHashSequenceProject *>(pop);
+	}
+
+};	// class CPhysicalHashSequenceProject
+
+}  // namespace gpopt
+
+#endif	// !GPOPT_CPhysicalHashSequenceProject_H
+
+// EOF