ORCA: preserve empty-input results in aggregate subqueries

contrib/pax_storage/src/test/regress/expected/bfv_planner_optimizer.out

@@ -454,18 +454,16 @@ explain (costs off) select * from t_hashdist where a > All (select random() from
 (10 rows)
 
 explain (costs off) select * from t_hashdist where a in (select random()::int from generate_series(1, 10));
-                        QUERY PLAN                        
-----------------------------------------------------------
- Gather Motion 3:1  (slice1; segments: 3)
-   ->  Hash Semi Join
-         Hash Cond: (t_hashdist.a = ((random())::integer))
+                   QUERY PLAN                   
+------------------------------------------------
+ Result
+   Filter: (hashed SubPlan 1)
+   ->  Gather Motion 3:1  (slice1; segments: 3)
          ->  Seq Scan on t_hashdist
-         ->  Hash
-               ->  Redistribute Motion 1:3  (slice2)
-                     Hash Key: ((random())::integer)
-                     ->  Function Scan on generate_series
- Optimizer: Pivotal Optimizer (GPORCA)
-(9 rows)
+   SubPlan 1
+     ->  Function Scan on generate_series
+ Optimizer: GPORCA
+(7 rows)
 
 -- subplan
 explain (costs off, verbose) select * from

contrib/pax_storage/src/test/regress/expected/rangefuncs_optimizer.out

@@ -261,8 +261,9 @@ HINT:  likely caused by a function that reads or modifies data in a distributed
 CONTEXT:  SQL function "rngfunct" statement 1
 -- function in subselect
 select * from rngfunc2 where f2 in (select f2 from rngfunct(1) z where z.rngfuncid = rngfunc2.rngfuncid) ORDER BY 1,2;
-ERROR:  function cannot execute on a QE slice because it accesses relation "public.rngfunc2"
-CONTEXT:  SQL function "rngfunct" during startup
+ERROR:  query plan with multiple segworker groups is not supported
+HINT:  likely caused by a function that reads or modifies data in a distributed table
+CONTEXT:  SQL function "rngfunct" statement 1
 -- function in subselect
 select * from rngfunc2 where f2 in (select f2 from rngfunct(rngfunc2.rngfuncid) z where z.rngfuncid = 1) ORDER BY 1,2;
 ERROR:  query plan with multiple segworker groups is not supported

contrib/pax_storage/src/test/regress/expected/rpt_optimizer.out

@@ -784,18 +784,15 @@ explain (costs off) select * from t_hashdist where a > All (select random() from
 (9 rows)
 
 explain (costs off) select * from t_hashdist where a in (select random()::int from t_replicate_volatile);
-                            QUERY PLAN                            
-------------------------------------------------------------------
+                   QUERY PLAN                   
+------------------------------------------------
  Gather Motion 3:1  (slice1; segments: 3)
-   ->  Hash Semi Join
-         Hash Cond: (t_hashdist.a = ((random())::integer))
-         ->  Seq Scan on t_hashdist
-         ->  Hash
-               ->  Redistribute Motion 1:3  (slice2; segments: 1)
-                     Hash Key: ((random())::integer)
-                     ->  Seq Scan on t_replicate_volatile
- Optimizer: Pivotal Optimizer (GPORCA)
-(9 rows)
+   ->  Seq Scan on t_hashdist
+         Filter: (hashed SubPlan 1)
+         SubPlan 1
+           ->  Seq Scan on t_replicate_volatile
+ Optimizer: GPORCA
+(6 rows)
 
 -- subplan
 explain (costs off, verbose) select * from t_hashdist left join t_replicate_volatile on t_hashdist.a > any (select random() from t_replicate_volatile);

contrib/pax_storage/src/test/regress/expected/subselect_gp_optimizer.out

@@ -890,24 +890,22 @@ select * from csq_pullup t0 where 1= (select count(*) from csq_pullup t1 where t
 
 -- subquery contains a HAVING clause
 explain select * from csq_pullup t0 where 1= (select count(*) from csq_pullup t1 where t0.t=t1.t HAVING count(*) < 10);
-                                               QUERY PLAN                                                
----------------------------------------------------------------------------------------------------------
+                                                                                  QUERY PLAN                                                                                  
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
  Gather Motion 3:1  (slice1; segments: 3)  (cost=0.00..862.00 rows=1 width=17)
    ->  Result  (cost=0.00..862.00 rows=1 width=17)
-         Filter: (1 = COALESCE((count(*)), '0'::bigint))
-         ->  Hash Left Join  (cost=0.00..862.00 rows=1 width=25)
+         Filter: (1 = CASE WHEN (true) THEN CASE WHEN ((count(*)) < 10) THEN (count(*)) ELSE NULL::bigint END WHEN ('0'::bigint < 10) THEN '0'::bigint ELSE NULL::bigint END)
+         ->  Hash Left Join  (cost=0.00..862.00 rows=1 width=34)
                Hash Cond: (t0.t = t1.t)
                ->  Seq Scan on csq_pullup t0  (cost=0.00..431.00 rows=1 width=17)
-               ->  Hash  (cost=431.00..431.00 rows=1 width=12)
-                     ->  Result  (cost=0.00..431.00 rows=1 width=12)
-                           Filter: ((count(*)) < 10)
-                           ->  GroupAggregate  (cost=0.00..431.00 rows=1 width=20)
-                                 Group Key: t1.t
-                                 ->  Sort  (cost=0.00..431.00 rows=1 width=4)
-                                       Sort Key: t1.t
-                                       ->  Seq Scan on csq_pullup t1  (cost=0.00..431.00 rows=1 width=4)
+               ->  Hash  (cost=431.00..431.00 rows=1 width=21)
+                     ->  GroupAggregate  (cost=0.00..431.00 rows=1 width=20)
+                           Group Key: t1.t
+                           ->  Sort  (cost=0.00..431.00 rows=1 width=4)
+                                 Sort Key: t1.t
+                                 ->  Seq Scan on csq_pullup t1  (cost=0.00..431.00 rows=1 width=4)
  Optimizer: GPORCA
-(15 rows)
+(13 rows)
 
 select * from csq_pullup t0 where 1= (select count(*) from csq_pullup t1 where t0.t=t1.t HAVING count(*) < 10);
   t  | n | i |  v  

contrib/pax_storage/src/test/regress/sql/qp_with_functional.sql

@@ -19,7 +19,7 @@
 -- m/.*ERROR:.*function cannot execute on a QE slice because it accesses relation.*/
 -- s/.*ERROR:.*/ERROR: error message might be different for CTE/
 --
--- m/.*ERROR:  query plan with multiple segworker groups is not supported.*/
+-- m/.*ERROR:.*query plan with multiple segworker groups is not supported.*/
 -- s/.*ERROR:.*/ERROR: error message might be different for CTE/
 --
 -- # This test file is included in the launcher script,

src/backend/gpopt/gpdbwrappers.cpp

@@ -415,6 +415,24 @@ gpdb::GetAggIntermediateResultType(Oid aggid)
 	return 0;
 }
 
+bool
+gpdb::IsAggFinalFuncVolatile(Oid aggid)
+{
+	GP_WRAP_START;
+	{
+		HeapTuple agg_tuple = SearchSysCache1(AGGFNOID, ObjectIdGetDatum(aggid));
+		if (!HeapTupleIsValid(agg_tuple))
+			elog(ERROR, "cache lookup failed for aggregate %u", aggid);
+
+		Oid final_func = ((Form_pg_aggregate) GETSTRUCT(agg_tuple))->aggfinalfn;
+		ReleaseSysCache(agg_tuple);
+		return OidIsValid(final_func) &&
+			   func_volatile(final_func) == PROVOLATILE_VOLATILE;
+	}
+	GP_WRAP_END;
+	return false;
+}
+
 int
 gpdb::GetAggregateArgTypes(Aggref *aggref, Oid *inputTypes)
 {

src/backend/gpopt/translate/CTranslatorRelcacheToDXL.cpp

@@ -1634,7 +1634,8 @@ CTranslatorRelcacheToDXL::RetrieveAgg(CMemoryPool *mp, IMDId *mdid)
 
 	CMDAggregateGPDB *pmdagg = GPOS_NEW(mp) CMDAggregateGPDB(
 		mp, mdid, mdname, result_type_mdid, intermediate_result_type_mdid,
-		is_ordered, is_splittable, is_hash_agg_capable, is_repsafe);
+		is_ordered, is_splittable, is_hash_agg_capable, is_repsafe,
+		gpdb::IsAggFinalFuncVolatile(agg_oid));
 	return pmdagg;
 }
 

src/backend/gporca/data/dxl/parse_tests/AggregateFinalFuncVolatile.xml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<dxl:DXLMessage xmlns:dxl="http://greenplum.com/dxl/2010/12/">
+  <dxl:Metadata>
+    <dxl:GPDBAgg Mdid="0.200001.1.0" Name="volatile_agg" FinalFuncVolatile="true" IsSplittable="false" HashAggCapable="false">
+      <dxl:ResultType Mdid="0.23.1.0"/>
+      <dxl:IntermediateResultType Mdid="0.23.1.0"/>
+    </dxl:GPDBAgg>
+    <dxl:GPDBAgg Mdid="0.200002.1.0" Name="ordinary_agg" IsSplittable="false" HashAggCapable="false">
+      <dxl:ResultType Mdid="0.23.1.0"/>
+      <dxl:IntermediateResultType Mdid="0.23.1.0"/>
+    </dxl:GPDBAgg>
+  </dxl:Metadata>
+</dxl:DXLMessage>

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

@@ -269,6 +269,9 @@ class CUtils
 	// check if given expression has a count(*)/count(Any) agg
 	static BOOL FHasCountAgg(CExpression *pexpr, CColRef **ppcrCount);
 
+	// check for an aggregate with an empty grouping set in a logical tree
+	static BOOL FHasScalarAgg(CExpression *pexpr);
+
 	// check if given expression has count matching the given column, returns the Logical GroupBy Agg above
 	static BOOL FHasCountAggMatchingColumn(const CExpression *pexpr,
 										   const CColRef *colref,

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

@@ -134,6 +134,9 @@ class CScalarAggFunc : public CScalar
 	// match function
 	BOOL Matches(COperator *pop) const override;
 
+	CFunctionProp *DeriveFunctionProperties(
+		CMemoryPool *mp, CExpressionHandle &exprhdl) const override;
+
 	// sensitivity to order of inputs
 	BOOL
 	FInputOrderSensitive() const override

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

@@ -80,7 +80,13 @@ class CXformApply2Join : public CXformExploration
 		TApply *popApply = TApply::PopConvert(pexprApply->Pop());
 		CColRefArray *colref_array = popApply->PdrgPcrInner();
 		GPOS_ASSERT(nullptr != colref_array);
-		GPOS_ASSERT(1 == colref_array->Size());
+		// Scalar aggregate decorrelation can expose its aggregate columns and
+		// a match flag. Scalar correlated joins already support these outputs.
+		GPOS_ASSERT(0 < colref_array->Size());
+		GPOS_ASSERT(
+			1 == colref_array->Size() ||
+			(COperator::EopLogicalLeftOuterApply == popApply->Eopid() &&
+			 COperator::EopScalarSubquery == popApply->EopidOriginSubq()));
 
 		colref_array->AddRef();
 		COperator::EOperatorId eopidSubq = popApply->EopidOriginSubq();

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

@@ -2162,6 +2162,30 @@ CUtils::FHasCountAgg(CExpression *pexpr,
 }
 
 
+BOOL
+CUtils::FHasScalarAgg(CExpression *pexpr)
+{
+	GPOS_CHECK_STACK_SIZE;
+	GPOS_ASSERT(pexpr->Pop()->FLogical());
+
+	if (COperator::EopLogicalGbAgg == pexpr->Pop()->Eopid() &&
+		0 == CLogicalGbAgg::PopConvert(pexpr->Pop())->Pdrgpcr()->Size())
+	{
+		return true;
+	}
+
+	for (ULONG ul = 0; ul < pexpr->Arity(); ul++)
+	{
+		CExpression *pexprChild = (*pexpr)[ul];
+		if (pexprChild->Pop()->FLogical() && FHasScalarAgg(pexprChild))
+		{
+			return true;
+		}
+	}
+	return false;
+}
+
+
 static BOOL
 FCountAggMatchingColumn(CExpression *pexprPrjElem, const CColRef *colref)
 {

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

@@ -81,8 +81,11 @@ CLogicalLeftOuterCorrelatedApply::CLogicalLeftOuterCorrelatedApply(
 	  // discard the tuple 2 output from generate_series. Instead we want to
 	  // preserve it for a NULL match in the LOJ so that we correctly evaulate
 	  // subplan ALL_SUBLINK.
-	  m_allow_predicate_pushdown(COperator::EopScalarSubqueryAll !=
-								 eopidOriginSubq)
+	  // ANY must also keep NULL comparison results. Moving its test into a
+	  // filter would turn ANY over a NULL row into ANY over no rows (false).
+	  m_allow_predicate_pushdown(
+		  COperator::EopScalarSubqueryAll != eopidOriginSubq &&
+		  COperator::EopScalarSubqueryAny != eopidOriginSubq)
 {
 }
 

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

@@ -24,6 +24,20 @@
 using namespace gpopt;
 using namespace gpmd;
 
+CFunctionProp *
+CScalarAggFunc::DeriveFunctionProperties(CMemoryPool *mp,
+										CExpressionHandle &exprhdl) const
+{
+	const IMDAggregate *mdagg =
+		COptCtxt::PoctxtFromTLS()->Pmda()->RetrieveAgg(m_pmdidAggFunc);
+	IMDFunction::EFuncStbl stability = mdagg->IsFinalFuncVolatile()
+										 ? IMDFunction::EfsVolatile
+										 : IMDFunction::EfsImmutable;
+	return PfpDeriveFromChildren(mp, exprhdl, stability,
+								false /*fHasVolatileFunctionScan*/,
+								false /*fScan*/);
+}
+
 //---------------------------------------------------------------------------
 //	@function:
 //		CScalarAggFunc::CScalarAggFunc

src/backend/gporca/libgpopt/src/xforms/CDecorrelator.cpp

@@ -170,6 +170,13 @@ CDecorrelator::FProcess(CMemoryPool *mp, CExpression *pexpr, BOOL fEqualityOnly,
 					CUtils::FCorrelatedApply(pexpr->Pop()) &&
 						"Apply expression is encountered by decorrelator");
 
+	// Decorrelation can share an inner result between outer rows. This also
+	// applies to aggregates whose final function is volatile.
+	if (IMDFunction::EfsVolatile == pexpr->DeriveFunctionProperties()->Efs())
+	{
+		return false;
+	}
+
 	// no outer references?
 	if (outerRefsToRemove->IsDisjoint(pexpr->DeriveOuterReferences()))
 	{