[#146190079] Falls back due to CTE prod-cons inconsistency

This commit introduces a check to detect if the CTE producer and
matching consumer is executed on the right place. So if CTE producer is
executed on master/segments/segment, then matching consumer also has to
execute on master/segments/segment. In rare cases, orca generates plans
that violate this assumption. This commit detects plans of that kind and
falls back.
Signed-off-by: NVenkatesh Raghavan <vraghavan@pivotal.io>
Signed-off-by: NEkta Khanna <ekhanna@pivotal.io>

libgpopt/include/gpopt/base/CUtils.h

@@ -27,6 +27,7 @@
 #include "gpopt/operators/CScalarSubquery.h"
 #include "gpopt/operators/CScalarAggFunc.h"
 #include "naucrates/md/CMDTypeInt4GPDB.h"
+#include "naucrates/statistics/IStatistics.h"
 
 // fwd declarations
 namespace gpmd
@@ -59,6 +60,7 @@ namespace gpopt
 	{
 
 		private:
+
 			// check if the expression is a scalar boolean const
 			static
 			BOOL FScalarConstBool(CExpression *pexpr, BOOL fVal);
@@ -81,6 +83,13 @@ namespace gpopt
 
 		public:
 
+			enum EExecLocalityType
+			{
+				EeltMaster,
+				EeltSegments,
+				EeltSingleton
+			};
+
 #ifdef GPOS_DEBUG
 
 			// print given expression to debug trace
@@ -99,6 +108,11 @@ namespace gpopt
 			// Helpers for generating expressions
 			//-------------------------------------------------------------------
 
+			// recursively check for a plan with CTE, if both CTEProducer and CTEConsumer are executed on the same locality.
+			// raises an exception if CTE Producer and CTE Consumer does not have the same locality
+			static
+			void ValidateCTEProducerConsumerLocality(IMemoryPool *pmp, CExpression *pexpr, EExecLocalityType edt, HMUlUl *phmulul);
+
 			// Check is a comparison between given types or a comparison after casting
 			// one side to an another exists
 			static
@@ -1057,6 +1071,10 @@ namespace gpopt
 			// check if the equivalance classes are same
 			static
 			BOOL FEquivalanceClassesEqual(IMemoryPool *pmp, DrgPcrs *pdrgpcrsFst, DrgPcrs *pdrgpcrsSnd);
+
+			// get execution locality
+			static
+			EExecLocalityType ExecLocalityType(CDistributionSpec *pds);
 	}; // class CUtils
 
 	// hash set from expressions

libgpopt/include/gpopt/exception.h

@@ -36,6 +36,7 @@ namespace gpopt
 		ExmiUnsupportedNonDeterministicUpdate,
 		ExmiUnsatisfiedRequiredProperties,
 		ExmiEvalUnsupportedScalarExpr,
+		ExmiCTEProducerConsumerMisAligned,
 
 		ExmiSentinel
 	};

libgpopt/include/gpopt/optimizer/COptimizer.h

@@ -90,6 +90,9 @@ namespace gpopt
 			static
 			void PrintQueryOrPlan(IMemoryPool *pmp, CExpression *pexpr, CQueryContext *pqc = NULL);
 
+			// Check for a plan with CTE, if both CTEProducer and CTEConsumer are executed on the same locality.
+			static
+			void CheckCTEConsistency(IMemoryPool *pmp, CExpression *pexpr);
 		public:
 			
 			// main optimizer function 

libgpopt/src/base/CUtils.cpp

@@ -6338,4 +6338,111 @@ CUtils::FEquivalanceClassesEqual
 	phmcscrs->Release();
 	return true;
 }
+
+// This function provides a check for a plan with CTE, if both
+// CTEProducer and CTEConsumer are executed on the same locality.
+// If it is not the case, the plan is bogus and cannot be executed
+// by the executor, therefore it throws an exception causing fallback
+// to planner.
+//
+// The overall algorithm for detecting CTE producer and consumer
+// inconsistency employs a HashMap while preorder traversing the tree.
+// Preorder traversal will guarantee that we visit the producer before
+// we visit the consumer. In this regard, when we see a CTE producer,
+// we add its CTE id as a key and its execution locality as a value to
+// the HashMap.
+// And when we encounter the matching CTE consumer while we traverse the
+// tree, we check if the locality matches by looking up the CTE id from
+// the HashMap. If we see a non-matching locality, we report the
+// anamoly.
+//
+// We change the locality and push it down the tree whenever we detect
+// a motion and the motion type enforces a locality change. We pass the
+// locality type by value instead of referance to avoid locality changes
+// affect parent and sibling localities.
+void
+CUtils::ValidateCTEProducerConsumerLocality
+	(
+	IMemoryPool *pmp,
+	CExpression *pexpr,
+	EExecLocalityType eelt,
+	HMUlUl *phmulul // Hash Map containing the CTE Producer id and its execution locality
+	)
+{
+	COperator *pop = pexpr->Pop();
+	if (COperator::EopPhysicalCTEProducer == pop->Eopid())
+	{
+		// record the location (either master or segment or singleton)
+		// where the CTE producer is being executed
+		ULONG ulCTEID = CPhysicalCTEProducer::PopConvert(pop)->UlCTEId();
+		phmulul->FInsert(GPOS_NEW(pmp) ULONG(ulCTEID), GPOS_NEW(pmp) ULONG(eelt));
+	}
+	else if (COperator::EopPhysicalCTEConsumer == pop->Eopid())
+	{
+		ULONG ulCTEID = CPhysicalCTEConsumer::PopConvert(pop)->UlCTEId();
+		ULONG *pulLocProducer = phmulul->PtLookup(&ulCTEID);
+
+		// check if the CTEConsumer is being executed in the same location
+		// as the CTE Producer
+		if (NULL == pulLocProducer || *pulLocProducer != (ULONG) eelt)
+		{
+			phmulul->Release();
+			GPOS_RAISE(gpopt::ExmaGPOPT, gpopt::ExmiCTEProducerConsumerMisAligned, ulCTEID);
+		}
+	}
+	// In case of a Gather motion, the execution locality is set to segments
+	// since the child of Gather motion executes on segments
+	else if (COperator::EopPhysicalMotionGather == pop->Eopid())
+	{
+		eelt = EeltSegments;
+	}
+	else if (COperator::EopPhysicalMotionHashDistribute == pop->Eopid() || COperator::EopPhysicalMotionRandom == pop->Eopid() || COperator::EopPhysicalMotionBroadcast == pop->Eopid())
+	{
+		// For any of these physical motions, the outer child's execution needs to be
+		// tracked for depending upon the distribution spec
+		CDrvdPropPlan *pdpplanChild = CDrvdPropPlan::Pdpplan((*pexpr)[0]->PdpDerive());
+		CDistributionSpec *pdsChild = pdpplanChild->Pds();
+
+		eelt = CUtils::ExecLocalityType(pdsChild);
+	}
+
+	const ULONG ulLen = pexpr->UlArity();
+	for (ULONG ul = 0; ul < ulLen; ul++)
+	{
+		CExpression *pexprChild = (*pexpr)[ul];
+
+		if (!pexprChild->Pop()->FScalar())
+		{
+			ValidateCTEProducerConsumerLocality(pmp, pexprChild, eelt, phmulul);
+		}
+	}
+}
+
+// get execution locality type
+CUtils::EExecLocalityType
+CUtils::ExecLocalityType
+	(
+	CDistributionSpec *pds
+	)
+{
+	EExecLocalityType eelt;
+	if (CDistributionSpec::EdtSingleton == pds->Edt() || CDistributionSpec::EdtStrictSingleton == pds->Edt())
+	{
+		CDistributionSpecSingleton *pdss = CDistributionSpecSingleton::PdssConvert(pds);
+		if (pdss->FOnMaster())
+		{
+			eelt = EeltMaster;
+		}
+		else
+		{
+			eelt = EeltSingleton;
+		}
+	}
+	else
+	{
+		eelt = EeltSegments;
+	}
+	return eelt;
+}
+
 // EOF

libgpopt/src/exception.cpp

@@ -88,6 +88,12 @@ gpopt::EresExceptionInit
 					 GPOS_WSZ_WSZLEN("Expecting a scalar expression without a subquery, received a %s"),
 					 1,
 					 GPOS_WSZ_WSZLEN("Non scalar expression or scalar expression with a subquery")),
+
+			CMessage(CException(gpopt::ExmaGPOPT, gpopt::ExmiCTEProducerConsumerMisAligned),
+					 CException::ExsevError,
+					 GPOS_WSZ_WSZLEN("CTE Producer-Consumer execution locality mismatch for CTE id %lld"),
+					 1,
+					 GPOS_WSZ_WSZLEN("CTE Producer-Consumer execution locality mismatch")),
 	};
 
 	GPOS_RESULT eres = GPOS_FAILED;

libgpopt/src/optimizer/COptimizer.cpp

@@ -22,6 +22,7 @@
 #include "gpopt/engine/CEngine.h"
 #include "gpopt/engine/CEnumeratorConfig.h"
 #include "gpopt/engine/CStatisticsConfig.h"
+#include "gpopt/exception.h"
 #include "gpopt/minidump/CMiniDumperDXL.h"
 #include "gpopt/minidump/CMinidumperUtils.h"
 #include "gpopt/minidump/CSerializableStackTrace.h"
@@ -340,6 +341,28 @@ COptimizer::HandleExceptionAfterFinalizingMinidump
 	GPOS_RETHROW(ex);
 }
 
+// This function provides an entry point to check for a plan with CTE,
+// if both CTEProducer and CTEConsumer are executed on the same locality.
+// If it is not the case, the plan is bogus and cannot be executed
+// by the executor and an exception is raised.
+//
+// To be able to enter the recursive logic, the execution locality of root
+// is determined before the recursive call.
+void
+COptimizer::CheckCTEConsistency
+	(
+	IMemoryPool *pmp,
+	CExpression *pexpr
+	)
+{
+	HMUlUl *phmulul = GPOS_NEW(pmp) HMUlUl(pmp);
+	CDrvdPropPlan *pdpplanChild = CDrvdPropPlan::Pdpplan(pexpr->PdpDerive());
+	CDistributionSpec *pdsChild = pdpplanChild->Pds();
+
+	CUtils::EExecLocalityType eelt = CUtils::ExecLocalityType(pdsChild);
+	CUtils::ValidateCTEProducerConsumerLocality(pmp, pexpr, eelt, phmulul);
+	phmulul->Release();
+}
 
 //---------------------------------------------------------------------------
 //	@function:
@@ -366,6 +389,8 @@ COptimizer::PexprOptimize
 	CExpression *pexprPlan = eng.PexprExtractPlan();
 	(void) pexprPlan->PrppCompute(pmp, pqc->Prpp());
 
+	CheckCTEConsistency(pmp, pexprPlan);
+
 	GPOS_CHECK_ABORT;
 
 	return pexprPlan;

server/src/unittest/gpopt/minidump/CICGTest.cpp

@@ -248,7 +248,8 @@ const struct UnSupportedTestCase unSupportedTestCases[] =
 	{
 		{"../data/dxl/minidump/OneSegmentGather.mdp", gpdxl::ExmaDXL, gpdxl::ExmiExpr2DXLUnsupportedFeature},
 		{"../data/dxl/minidump/CTEWithOuterReferences.mdp", gpopt::ExmaGPOPT, gpopt::ExmiUnsupportedOp},
-		{"../data/dxl/minidump/BitmapIndexUnsupportedOperator.mdp", gpopt::ExmaGPOPT, gpopt::ExmiNoPlanFound}
+		{"../data/dxl/minidump/BitmapIndexUnsupportedOperator.mdp", gpopt::ExmaGPOPT, gpopt::ExmiNoPlanFound},
+		{"../data/dxl/minidump/CTEMisAlignedProducerConsumer.mdp",gpopt::ExmaGPOPT, gpopt::ExmiCTEProducerConsumerMisAligned}
 	};
 
 // negative index apply tests