orca: fall back on replicated CTE only when consumed in 2+ slices

src/backend/cdb/cdbmutate.c

@@ -2212,6 +2212,71 @@ shareinput_peekmot(ApplyShareInputContext *ctxt)
 	return linitial_int(ctxt->motStack);
 }
 
+/*
+ * Segment-coverage key of a slice, used to detect cross-slice shared scans
+ * whose producer and consumer gangs run on different segment sets (which
+ * deadlocks the writer/reader rendezvous in nodeShareInputScan.c).
+ *
+ * The key mirrors how FillSliceTable() sizes an ORCA plan's gangs: a slice
+ * runs on a single process when the flow at the top of the slice is
+ * FLOW_SINGLETON (segindex -1 = QD, >= 0 = one segment); otherwise ORCA
+ * dispatches it to all segments. Two slices share the same segment set iff
+ * their keys are equal.
+ */
+#define SLICE_COVERAGE_ALLSEG	(INT_MIN)		/* dispatched to all segments */
+#define SLICE_COVERAGE_UNSET	(INT_MIN + 1)	/* producer not seen yet */
+
+static int
+shareinput_slice_coverage(Flow *flow)
+{
+	if (flow != NULL && flow->flotype == FLOW_SINGLETON)
+		return flow->segindex;
+
+	return SLICE_COVERAGE_ALLSEG;
+}
+
+/*
+ * Coverage key of the top slice. A plain SELECT's top slice runs on the QD,
+ * but when the query is part of a CTAS/COPY/REFRESH, or is a DML on a
+ * distributed relation, FillSliceTable() turns the top slice into a writer
+ * gang dispatched to all segments instead.
+ */
+static int
+shareinput_topslice_coverage(PlannerInfo *root)
+{
+	Query	   *parse = root->parse;
+
+	if (parse->parentStmtType != PARENTSTMTTYPE_NONE)
+		return SLICE_COVERAGE_ALLSEG;
+
+	if (parse->commandType != CMD_SELECT && parse->resultRelation > 0)
+	{
+		RangeTblEntry *rte = rt_fetch(parse->resultRelation, parse->rtable);
+
+		if (GpPolicyFetch(rte->relid)->ptype != POLICYTYPE_ENTRY)
+			return SLICE_COVERAGE_ALLSEG;
+	}
+
+	return -1;					/* top slice runs on the QD */
+}
+
+/* Parallel stack carrying the coverage key of each enclosing slice. */
+static void
+shareinput_pushcov(ApplyShareInputContext *ctxt, int cov)
+{
+	ctxt->covStack = lcons_int(cov, ctxt->covStack);
+}
+static void
+shareinput_popcov(ApplyShareInputContext *ctxt)
+{
+	ctxt->covStack = list_delete_first(ctxt->covStack);
+}
+static int
+shareinput_peekcov(ApplyShareInputContext *ctxt)
+{
+	return linitial_int(ctxt->covStack);
+}
+
 
 /*
  * Replace the target list of ShareInputScan nodes, with references
@@ -2373,7 +2438,10 @@ shareinput_mutator_xslice_1(Node *node, PlannerInfo *root, bool fPop)
 	if (fPop)
 	{
 		if (IsA(plan, Motion))
+		{
 			shareinput_popmot(ctxt);
+			shareinput_popcov(ctxt);
+		}
 		return false;
 	}
 
@@ -2382,6 +2450,10 @@ shareinput_mutator_xslice_1(Node *node, PlannerInfo *root, bool fPop)
 		Motion	   *motion = (Motion *) plan;
 
 		shareinput_pushmot(ctxt, motion->motionID);
+		shareinput_pushcov(ctxt,
+						   shareinput_slice_coverage(motion->plan.lefttree ?
+													 motion->plan.lefttree->flow :
+													 NULL));
 		return true;
 	}
 
@@ -2411,6 +2483,7 @@ shareinput_mutator_xslice_1(Node *node, PlannerInfo *root, bool fPop)
 			 */
 			ctxt->producers[sisc->share_id] = sisc;
 			ctxt->sliceMarks[sisc->share_id] = motId;
+			ctxt->producerCoverage[sisc->share_id] = shareinput_peekcov(ctxt);
 		}
 	}
 
@@ -2431,7 +2504,10 @@ shareinput_mutator_xslice_2(Node *node, PlannerInfo *root, bool fPop)
 	if (fPop)
 	{
 		if (IsA(plan, Motion))
+		{
 			shareinput_popmot(ctxt);
+			shareinput_popcov(ctxt);
+		}
 		return false;
 	}
 
@@ -2440,6 +2516,10 @@ shareinput_mutator_xslice_2(Node *node, PlannerInfo *root, bool fPop)
 		Motion	   *motion = (Motion *) plan;
 
 		shareinput_pushmot(ctxt, motion->motionID);
+		shareinput_pushcov(ctxt,
+						   shareinput_slice_coverage(motion->plan.lefttree ?
+													 motion->plan.lefttree->flow :
+													 NULL));
 		return true;
 	}
 
@@ -2469,6 +2549,27 @@ shareinput_mutator_xslice_2(Node *node, PlannerInfo *root, bool fPop)
 
 				incr_plan_nsharer_xslice(plan_slicemark.plan);
 				sisc->driver_slice = motId;
+
+				/*
+				 * A cross-slice share only rendezvouses correctly when the
+				 * producer and this consumer run on the same set of segments:
+				 * the writer on each producing segment waits for an ack from a
+				 * reader on that same segment (nodeShareInputScan.c). If the
+				 * two slices cover different segment sets, a writer waits for
+				 * an ack that never comes, or a reader waits for a producer
+				 * that never runs -- the query hangs.
+				 *
+				 * QD shares are exempt: pass 4 relocates every slice touching
+				 * such a share onto the QD, so they end up consistent.
+				 */
+				if (!list_member_int(ctxt->qdShares, sisc->share_id))
+				{
+					int			prodCov = ctxt->producerCoverage[sisc->share_id];
+					int			consCov = shareinput_peekcov(ctxt);
+
+					if (prodCov != SLICE_COVERAGE_UNSET && prodCov != consCov)
+						ctxt->crossSliceCoverageHazard = true;
+				}
 			}
 		}
 	}
@@ -2592,15 +2693,22 @@ apply_shareinput_xslice(Plan *plan, PlannerInfo *root)
 	PlannerGlobal *glob = root->glob;
 	ApplyShareInputContext *ctxt = &glob->share;
 	ShareInputContext walker_ctxt;
+	int			i;
 
 	ctxt->motStack = NULL;
+	ctxt->covStack = NULL;
 	ctxt->qdShares = NULL;
 	ctxt->qdSlices = NULL;
 	ctxt->nextPlanId = 0;
+	ctxt->crossSliceCoverageHazard = false;
 
 	ctxt->sliceMarks = palloc0(ctxt->producer_count * sizeof(int));
+	ctxt->producerCoverage = palloc(ctxt->producer_count * sizeof(int));
+	for (i = 0; i < ctxt->producer_count; i++)
+		ctxt->producerCoverage[i] = SLICE_COVERAGE_UNSET;
 
 	shareinput_pushmot(ctxt, 0);
+	shareinput_pushcov(ctxt, shareinput_topslice_coverage(root));
 
 	walker_ctxt.base.node = (Node *) root;
 

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

@@ -1024,9 +1024,6 @@ class CUtils
 
 	static BOOL FScalarConstOrBinaryCoercible(CExpression *pexpr);
 
-	static BOOL FHasCrossSliceReplicatedCTEConsumer(CMemoryPool *mp,
-													CExpression *pexpr);
-
 	static CExpression *ReplaceColrefWithProjectExpr(CMemoryPool *mp,
 													 CExpression *pexpr,
 													 CColRef *pcolref,

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

@@ -901,128 +901,6 @@ CUtils::FHasCTEAnchor(CExpression *pexpr)
 	return false;
 }
 
-// True if the distribution is replicated-like.
-static BOOL
-FReplicatedLikeDistribution(CDistributionSpec::EDistributionType edt)
-{
-	return (CDistributionSpec::EdtStrictReplicated == edt ||
-			CDistributionSpec::EdtTaintedReplicated == edt ||
-			CDistributionSpec::EdtUniversal == edt);
-}
-
-struct SCTEInfo
-{
-	ULONG cteId;
-	ULONG sliceId;
-
-	SCTEInfo(ULONG cte_id, ULONG slice_id) : cteId(cte_id), sliceId(slice_id)
-	{
-	}
-};
-
-typedef CDynamicPtrArray<SCTEInfo, CleanupDelete<SCTEInfo> > CTEInfoArray;
-
-// Walk the physical tree, recording the slice id of every replicated
-// CTE Producer and every CTE Consumer. Slices are delimited by Motion
-// nodes: each non-scalar child of a Motion lives in a fresh slice --
-// same motId-stack idea as in apply_shareinput_xslice.
-static void
-CollectCTESlices(CMemoryPool *mp, CExpression *pexpr, ULONG curSlice,
-				 ULONG *pNextSlice, CTEInfoArray *prodInfos,
-				 CTEInfoArray *consInfos)
-{
-	COperator *pop = pexpr->Pop();
-
-	if (COperator::EopPhysicalCTEProducer == pop->Eopid())
-	{
-		// Producer's distribution comes from its only child -- inspect
-		// it there. Skip non-replicated Producers; they cannot trigger
-		// the cross-slice issue we are checking for.
-		GPOS_ASSERT(1 == pexpr->Arity());
-		CExpression *pexprChild = (*pexpr)[0];
-		CDrvdPropPlan *pdpplan =
-			CDrvdPropPlan::Pdpplan(pexprChild->PdpDerive());
-
-		if (FReplicatedLikeDistribution(pdpplan->Pds()->Edt()))
-		{
-			prodInfos->Append(GPOS_NEW(mp) SCTEInfo(
-				CPhysicalCTEProducer::PopConvert(pop)->UlCTEId(), curSlice));
-		}
-	}
-	else if (COperator::EopPhysicalCTEConsumer == pop->Eopid())
-	{
-		// Consumer is a leaf -- record (cteId, curSlice) and let the
-		// caller decide later, once the whole tree has been walked.
-		consInfos->Append(GPOS_NEW(mp) SCTEInfo(
-			CPhysicalCTEConsumer::PopConvert(pop)->UlCTEId(), curSlice));
-	}
-
-	BOOL isMotion = CUtils::FPhysicalMotion(pop);
-
-	for (ULONG ul = 0; ul < pexpr->Arity(); ul++)
-	{
-		CExpression *pexprChild = (*pexpr)[ul];
-
-		// Scalar subtrees (predicates, project elements) never run as
-		// separate executor groups, so they cannot host a slice.
-		if (pexprChild->Pop()->FScalar())
-		{
-			continue;
-		}
-
-		// Allocate a fresh slice id for each non-scalar child of a
-		// Motion; otherwise the child stays in the parent's slice.
-		ULONG childSlice = curSlice;
-		if (isMotion)
-		{
-			(*pNextSlice)++;
-			childSlice = *pNextSlice;
-		}
-
-		CollectCTESlices(mp, pexprChild, childSlice, pNextSlice, prodInfos,
-						 consInfos);
-	}
-}
-
-BOOL
-CUtils::FHasCrossSliceReplicatedCTEConsumer(CMemoryPool *mp, CExpression *pexpr)
-{
-	if (NULL == pexpr)
-	{
-		return false;
-	}
-
-	CTEInfoArray *prodInfos = GPOS_NEW(mp) CTEInfoArray(mp);
-	CTEInfoArray *consInfos = GPOS_NEW(mp) CTEInfoArray(mp);
-	ULONG nextSlice = 0;
-
-	CollectCTESlices(mp, pexpr, 0 /*curSlice*/, &nextSlice, prodInfos,
-					 consInfos);
-
-	BOOL cross = false;
-
-	for (ULONG ic = 0; ic < consInfos->Size(); ic++)
-	{
-		SCTEInfo *cons = (*consInfos)[ic];
-
-		for (ULONG ip = 0; ip < prodInfos->Size(); ip++)
-		{
-			SCTEInfo *prod = (*prodInfos)[ip];
-			if (prod->cteId == cons->cteId && prod->sliceId != cons->sliceId)
-			{
-				cross = true;
-				goto lExit;
-			}
-		}
-	}
-lExit:
-
-	prodInfos->Release();
-	consInfos->Release();
-
-	return cross;
-}
-
 //---------------------------------------------------------------------------
 //	@class:
 //		CUtils::FHasSubqueryOrApply

src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXL.cpp

@@ -351,20 +351,6 @@ CTranslatorExprToDXL::PdxlnTranslate(CExpression *pexpr,
 
 	GPOS_ASSERT(NULL == m_pdpplan);
 
-	// Walk the physical tree and detect a CTE Consumer placed on a
-	// different slice than its Producer when the Producer's output is
-	// replicated-like (StrictReplicated/TaintedReplicated/Universal).
-	// Fall back to the Postgres optimizer if it is detected because
-	// it breaks Producer-Consumer locality and can hang the
-	// query at execution.
-	if (CUtils::FHasCrossSliceReplicatedCTEConsumer(m_mp, pexpr))
-	{
-		GPOS_RAISE(
-			gpdxl::ExmaDXL, gpdxl::ExmiExpr2DXLUnsupportedFeature,
-			GPOS_WSZ_LIT(
-				"CTE Consumer placed on a different slice than its replicated Producer"));
-	}
-
 	m_pdpplan = CDrvdPropPlan::Pdpplan(pexpr->PdpDerive());
 	m_pdpplan->AddRef();
 

src/backend/optimizer/plan/orca.c

@@ -184,6 +184,22 @@ optimize_query(Query *parse, ParamListInfo boundParams)
 	/* Post-process ShareInputScan nodes */
 	(void) apply_shareinput_xslice(result->planTree, root);
 
+	/*
+	 * apply_shareinput_xslice() flags a cross-slice shared scan whose producer
+	 * and consumer gangs run on different segment sets (e.g. a replicated CTE
+	 * materialized on all segments but consumed in a single-segment slice).
+	 * Such a plan deadlocks at execution in the ShareInputScan writer/reader
+	 * rendezvous, so discard it and let the Postgres planner plan the query.
+	 */
+	if (glob->share.crossSliceCoverageHazard)
+	{
+		ereport(optimizer_trace_fallback ? INFO : DEBUG1,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("GPORCA failed to produce a plan, falling back to planner"),
+				 errdetail("GPORCA produced a cross-slice shared scan with mismatched segment coverage.")));
+		return NULL;
+	}
+
 	/*
 	 * Fix ShareInputScans for EXPLAIN, like in standard_planner(). For all
 	 * subplans first, and then for the main plan tree.

src/include/nodes/relation.h

@@ -88,14 +88,23 @@ typedef struct ApplyShareInputContext
 	int		   *share_refcounts;
 	int			share_refcounts_sz;		/* allocated sized of 'share_refcounts' */
 	List *motStack;
+	List *covStack;					/* parallel to motStack: segment-coverage
+									 * key of each enclosing slice */
 	List *qdShares;
 	List *qdSlices;
 	int nextPlanId;
 
 	ShareInputScan **producers;
 	int		   *sliceMarks;			/* one for each producer */
+	int		   *producerCoverage;	/* one per producer: coverage key of the
+									 * slice its producer runs in */
 	int			producer_count;
 
+	bool		crossSliceCoverageHazard;	/* a cross-slice shared scan whose
+											 * producer and consumer slices run
+											 * on different segment sets was
+											 * found; ORCA plan must fall back */
+
 } ApplyShareInputContext;
 
 

src/test/regress/expected/qp_orca_fallback_optimizer.out

@@ -410,7 +410,7 @@ t2 AS (SELECT id, refrcode FROM tbl2 WHERE REFERENCEID = 101991)
    ON p.isCalcDetail = r1.RefrCode
  LIMIT 1;
 INFO:  GPORCA failed to produce a plan, falling back to planner
-DETAIL:  Feature not supported: CTE Consumer placed on a different slice than its replicated Producer
+DETAIL:  GPORCA produced a cross-slice shared scan with mismatched segment coverage.
                                          QUERY PLAN                                          
 ---------------------------------------------------------------------------------------------
  Limit  (cost=0.34..83.24 rows=1 width=104)