orca: fallback to Postgres optimizer on cross-slice replicated CTE Consumer.

Alena0704 · Alena0704 · commit 0fac2960d874 · 2026-05-06T10:18:58.000+03:00
Inspired by greengage 51fe92e: before Expr->DXL translation, walk the physical tree and track which slice each CTE Producer and Consumer lives on. If a Consumer is on a different slice than its Producer and the Producer's distribution is replicated, force a fallback to the Postgres optimizer. The replicated filter is essential: ordinary cross-slice CTE plans (non-replicated Producer with Gather/Redistribute Consumer) are a normal ORCA pattern and must not trigger fallback. 51fe92e doesn't trigger when a CTE over a replicated table is referenced from a scalar subquery, so the query hangs. This commit replaces the single-point check with a whole-tree walker that catches both cases. Tests: bring the original qp_orca_fallback case from 51fe92e and add a scalar-subquery reproducer in shared_scan guarded by statement_timeout. (cherry picked from commit open-gpdb/gpdb@3a9aebf)
diff --git a/src/backend/gporca/libgpopt/include/gpopt/base/CUtils.h b/src/backend/gporca/libgpopt/include/gpopt/base/CUtils.h
@@ -1026,6 +1026,21 @@ class CUtils
 	static CTableDescriptorHashSet *RemoveDuplicateMdids(
 		CMemoryPool *mp, CTableDescriptorHashSet *tabdescs);
 
+	// hash set from CTE ids
+	typedef CHashSet<ULONG, gpos::HashValue<ULONG>, gpos::Equals<ULONG>,
+					 CleanupDelete<ULONG> >
+		UlongCteIdHashSet;
+
+	static void CollectConsumersAndProducers(CMemoryPool *mp,
+											 CExpression *pexpr,
+											 ULongPtrArray *cteConsumers,
+											 UlongCteIdHashSet *cteProducerSet);
+
+	static BOOL hasUnpairedCTEConsumer(CMemoryPool *mp, CExpression *pexpr);
+
+	static BOOL FHasCrossSliceReplicatedCTEConsumer(CMemoryPool *mp,
+													CExpression *pexpr);
+
 	static CExpression *ReplaceColrefWithProjectExpr(CMemoryPool *mp,
 													 CExpression *pexpr,
 													 CColRef *pcolref,
diff --git a/src/backend/gporca/libgpopt/src/base/CUtils.cpp b/src/backend/gporca/libgpopt/src/base/CUtils.cpp
@@ -978,6 +978,180 @@ CUtils::FHasCTEAnchor(CExpression *pexpr)
 	return false;
 }
 
+// return CTEConsumers' and a set of CTEProducers' CTE ids in the given subtree
+void
+CUtils::CollectConsumersAndProducers(CMemoryPool *mp, CExpression *pexpr,
+									 ULongPtrArray *cteConsumers,
+									 UlongCteIdHashSet *cteProducerSet)
+{
+	COperator *pop = pexpr->Pop();
+
+	if (COperator::EopPhysicalCTEConsumer == pop->Eopid())
+	{
+		cteConsumers->Append(GPOS_NEW(mp) ULONG(
+			CPhysicalCTEConsumer::PopConvert(pop)->UlCTEId()));
+	}
+	else if (COperator::EopPhysicalCTEProducer == pop->Eopid())
+	{
+		cteProducerSet->Insert(GPOS_NEW(mp) ULONG(
+			CPhysicalCTEProducer::PopConvert(pop)->UlCTEId()));
+	}
+
+	for (ULONG ul = 0; ul < pexpr->Arity(); ul++)
+	{
+		CExpression *pexprChild = (*pexpr)[ul];
+
+		if (!pexprChild->Pop()->FScalar())
+		{
+			CollectConsumersAndProducers(mp, pexprChild, cteConsumers,
+										 cteProducerSet);
+		}
+	}
+}
+
+BOOL
+CUtils::hasUnpairedCTEConsumer(CMemoryPool *mp, CExpression *pexpr)
+{
+	BOOL hasUnpairedConsumer = false;
+
+	ULongPtrArray *cteConsumers = GPOS_NEW(mp) ULongPtrArray(mp);
+	UlongCteIdHashSet *cteProducerSet = GPOS_NEW(mp) UlongCteIdHashSet(mp);
+
+	CollectConsumersAndProducers(mp, pexpr, cteConsumers, cteProducerSet);
+
+	// check if every consumer's producer is in ProducerSet
+	for (ULONG ul = 0; ul < cteConsumers->Size(); ul++)
+	{
+		if (!cteProducerSet->Contains((*cteConsumers)[ul]))
+		{
+			hasUnpairedConsumer = true;
+			break;
+		}
+	}
+	cteConsumers->Release();
+	cteProducerSet->Release();
+
+	return hasUnpairedConsumer;
+}
+
+// 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];
+
+		if (pexprChild->Pop()->FScalar())
+		{
+			continue;
+		}
+
+		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
diff --git a/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXL.cpp b/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXL.cpp
@@ -265,6 +265,20 @@ CTranslatorExprToDXL::PdxlnTranslate(CExpression *pexpr,
 
 	GPOS_ASSERT(nullptr == 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();
 
@@ -4250,6 +4264,10 @@ CTranslatorExprToDXL::BuildScalarSubplans(
 {
 	const ULONG size = pdrgpcrInner->Size();
 
+	// Fallback to Postgres optimizer if the SubPlan's inner expression contains a
+	// CTE Consumer whose Producer lives outside this subtree. Such a Consumer
+	// would become a cross-slice Shared Scan reader without a local Producer,
+	// which can hang the query or fail at execution time.
 	CDXLNodeArray *pdrgpdxlnInner = GPOS_NEW(m_mp) CDXLNodeArray(m_mp);
 	for (ULONG ul = 0; ul < size; ul++)
 	{
diff --git a/src/test/regress/expected/qp_orca_fallback.out b/src/test/regress/expected/qp_orca_fallback.out
@@ -293,6 +293,52 @@ SELECT * FROM jsonb_array_elements('["b", "a"]'::jsonb) WITH ORDINALITY;
  "a"   |          2
 (2 rows)
 
+-- ORCA should fallback in case there are any CTE Consumers beneath a duplicate-hazard motion
+-- start_ignore
+DROP TABLE IF EXISTS tbl1, tbl2;
+-- end_ignore
+CREATE TABLE tbl2 (
+	id numeric NULL,
+	refrcode varchar(255) NULL,
+	referenceid numeric NULL
+)
+DISTRIBUTED REPLICATED;
+CREATE TABLE tbl1 (
+	id bigserial NOT NULL,
+	iscalctrg varchar(15) NOT NULL,
+	iscalcdetail varchar(15) NULL
+)
+DISTRIBUTED REPLICATED;
+EXPLAIN WITH
+t1 AS (SELECT * FROM tbl1),
+t2 AS (SELECT id, refrcode FROM tbl2 WHERE REFERENCEID = 101991)
+ SELECT p.*FROM t1 p
+  JOIN t2 r
+   ON p.isCalcTRG = r.RefrCode
+  JOIN t2 r1
+   ON p.isCalcDetail = r1.RefrCode
+ LIMIT 1;
+                                         QUERY PLAN                                          
+---------------------------------------------------------------------------------------------
+ Limit  (cost=0.34..83.24 rows=1 width=104)
+   ->  Gather Motion 1:1  (slice1; segments: 1)  (cost=0.34..581.83 rows=8 width=104)
+         ->  Hash Join  (cost=0.34..581.69 rows=8 width=104)
+               Hash Cond: ((tbl1.iscalcdetail)::text = (r1.refrcode)::text)
+               ->  Hash Join  (cost=0.17..580.97 rows=130 width=104)
+                     Hash Cond: ((tbl1.iscalctrg)::text = (r.refrcode)::text)
+                     ->  Seq Scan on tbl1  (cost=0.00..344.00 rows=24400 width=104)
+                     ->  Hash  (cost=0.11..0.11 rows=2 width=516)
+                           ->  Subquery Scan on r  (cost=0.00..0.11 rows=6 width=516)
+                                 ->  Seq Scan on tbl2  (cost=0.00..166.25 rows=6 width=548)
+                                       Filter: (referenceid = '101991'::numeric)
+               ->  Hash  (cost=0.11..0.11 rows=2 width=516)
+                     ->  Subquery Scan on r1  (cost=0.00..0.11 rows=6 width=516)
+                           ->  Seq Scan on tbl2 tbl2_1  (cost=0.00..166.25 rows=6 width=548)
+                                 Filter: (referenceid = '101991'::numeric)
+ Optimizer: Postgres query optimizer
+(16 rows)
+
+DROP TABLE tbl1, tbl2;
 -- start_ignore
 -- FIXME: gpcheckcat fails due to mismatching distribution policy if this table isn't dropped
 -- Keep this table around once this is fixed
diff --git a/src/test/regress/expected/qp_orca_fallback_optimizer.out b/src/test/regress/expected/qp_orca_fallback_optimizer.out
@@ -353,6 +353,54 @@ DETAIL:  Falling back to Postgres-based planner because GPORCA does not support
  "a"   |          2
 (2 rows)
 
+-- ORCA should fallback in case there are any CTE Consumers beneath a duplicate-hazard motion
+-- start_ignore
+DROP TABLE IF EXISTS tbl1, tbl2;
+-- end_ignore
+CREATE TABLE tbl2 (
+	id numeric NULL,
+	refrcode varchar(255) NULL,
+	referenceid numeric NULL
+)
+DISTRIBUTED REPLICATED;
+CREATE TABLE tbl1 (
+	id bigserial NOT NULL,
+	iscalctrg varchar(15) NOT NULL,
+	iscalcdetail varchar(15) NULL
+)
+DISTRIBUTED REPLICATED;
+EXPLAIN WITH
+t1 AS (SELECT * FROM tbl1),
+t2 AS (SELECT id, refrcode FROM tbl2 WHERE REFERENCEID = 101991)
+ SELECT p.*FROM t1 p
+  JOIN t2 r
+   ON p.isCalcTRG = r.RefrCode
+  JOIN t2 r1
+   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
+                                         QUERY PLAN                                          
+---------------------------------------------------------------------------------------------
+ Limit  (cost=0.34..83.24 rows=1 width=104)
+   ->  Gather Motion 1:1  (slice1; segments: 1)  (cost=0.34..581.83 rows=8 width=104)
+         ->  Hash Join  (cost=0.34..581.69 rows=8 width=104)
+               Hash Cond: ((tbl1.iscalcdetail)::text = (r1.refrcode)::text)
+               ->  Hash Join  (cost=0.17..580.97 rows=130 width=104)
+                     Hash Cond: ((tbl1.iscalctrg)::text = (r.refrcode)::text)
+                     ->  Seq Scan on tbl1  (cost=0.00..344.00 rows=24400 width=104)
+                     ->  Hash  (cost=0.11..0.11 rows=2 width=516)
+                           ->  Subquery Scan on r  (cost=0.00..0.11 rows=6 width=516)
+                                 ->  Seq Scan on tbl2  (cost=0.00..166.25 rows=6 width=548)
+                                       Filter: (referenceid = '101991'::numeric)
+               ->  Hash  (cost=0.11..0.11 rows=2 width=516)
+                     ->  Subquery Scan on r1  (cost=0.00..0.11 rows=6 width=516)
+                           ->  Seq Scan on tbl2 tbl2_1  (cost=0.00..166.25 rows=6 width=548)
+                                 Filter: (referenceid = '101991'::numeric)
+ Optimizer: Postgres query optimizer
+(16 rows)
+
+DROP TABLE tbl1, tbl2;
 -- start_ignore
 -- FIXME: gpcheckcat fails due to mismatching distribution policy if this table isn't dropped
 -- Keep this table around once this is fixed
diff --git a/src/test/regress/expected/shared_scan.out b/src/test/regress/expected/shared_scan.out
@@ -234,3 +234,31 @@ where
  Optimizer: Postgres query optimizer
 (37 rows)
 
+-- ORCA should fallback when a CTE over a replicated table is referenced
+-- from multiple scalar subqueries.
+-- ss_t1 needs enough rows (40000) to push ORCA to the cross-slice plan;
+-- with fewer rows the bug does not manifest and the test would silently
+-- pass even without the fix.
+CREATE TABLE ss_t1 AS
+  SELECT generate_series(1, 40000) id
+  DISTRIBUTED BY (id);
+CREATE TABLE ss_t2 AS
+  SELECT * FROM (VALUES (1, 10), (2, 20)) AS v(id, v)
+  DISTRIBUTED REPLICATED;
+ANALYZE ss_t1;
+ANALYZE ss_t2;
+SET statement_timeout = '15s';
+WITH
+    cte1 AS (SELECT v FROM ss_t2 WHERE id = 1),
+    cte2 AS (SELECT v FROM ss_t2 WHERE id = 2)
+  SELECT (SELECT v FROM cte1) + (SELECT v FROM cte2) +
+         (SELECT v FROM cte1) + (SELECT v FROM cte2) AS result
+  FROM ss_t1
+  LIMIT 1;
+ result 
+--------
+     60
+(1 row)
+
+RESET statement_timeout;
+DROP TABLE ss_t1, ss_t2;
diff --git a/src/test/regress/expected/shared_scan_optimizer.out b/src/test/regress/expected/shared_scan_optimizer.out
diff --git a/src/test/regress/sql/qp_orca_fallback.sql b/src/test/regress/sql/qp_orca_fallback.sql
diff --git a/src/test/regress/sql/shared_scan.sql b/src/test/regress/sql/shared_scan.sql
