Fix parallel-safety code for parallel aggregation.

has_parallel_hazard() was ignoring the proparallel markings for
aggregates, which is no good.  Fix that.  There was no way to mark
an aggregate as actually being parallel-safe, either, so add a
PARALLEL option to CREATE AGGREGATE.

Patch by me, reviewed by David Rowley.

4 files changed, 32 insertions, 6 deletions

diff --git a/src/backend/catalog/pg_aggregate.c b/src/backend/catalog/pg_aggregate.c
index b420349835b..bcc941104f5 100644
--- a/src/backend/catalog/pg_aggregate.c
+++ b/src/backend/catalog/pg_aggregate.c
@@ -72,7 +72,8 @@ AggregateCreate(const char *aggName,
 				Oid aggmTransType,
 				int32 aggmTransSpace,
 				const char *agginitval,
-				const char *aggminitval)
+				const char *aggminitval,
+				char proparallel)
 {
 	Relation	aggdesc;
 	HeapTuple	tup;
@@ -622,7 +623,7 @@ AggregateCreate(const char *aggName,
 							 false,		/* isStrict (not needed for agg) */
 							 PROVOLATILE_IMMUTABLE,		/* volatility (not
 														 * needed for agg) */
-							 PROPARALLEL_UNSAFE,
+							 proparallel,
 							 parameterTypes,	/* paramTypes */
 							 allParameterTypes, /* allParamTypes */
 							 parameterModes,	/* parameterModes */
diff --git a/src/backend/commands/aggregatecmds.c b/src/backend/commands/aggregatecmds.c
index 3424f842b9c..5c4d576b866 100644
--- a/src/backend/commands/aggregatecmds.c
+++ b/src/backend/commands/aggregatecmds.c
@@ -78,6 +78,7 @@ DefineAggregate(List *name, List *args, bool oldstyle, List *parameters,
 	int32		mtransSpace = 0;
 	char	   *initval = NULL;
 	char	   *minitval = NULL;
+	char	   *parallel = NULL;
 	int			numArgs;
 	int			numDirectArgs = 0;
 	oidvector  *parameterTypes;
@@ -91,6 +92,7 @@ DefineAggregate(List *name, List *args, bool oldstyle, List *parameters,
 	Oid			mtransTypeId = InvalidOid;
 	char		transTypeType;
 	char		mtransTypeType = 0;
+	char		proparallel = PROPARALLEL_UNSAFE;
 	ListCell   *pl;
 
 	/* Convert list of names to a name and namespace */
@@ -178,6 +180,8 @@ DefineAggregate(List *name, List *args, bool oldstyle, List *parameters,
 			initval = defGetString(defel);
 		else if (pg_strcasecmp(defel->defname, "minitcond") == 0)
 			minitval = defGetString(defel);
+		else if (pg_strcasecmp(defel->defname, "parallel") == 0)
+			parallel = defGetString(defel);
 		else
 			ereport(WARNING,
 					(errcode(ERRCODE_SYNTAX_ERROR),
@@ -449,6 +453,20 @@ DefineAggregate(List *name, List *args, bool oldstyle, List *parameters,
 		(void) OidInputFunctionCall(typinput, minitval, typioparam, -1);
 	}
 
+	if (parallel)
+	{
+		if (pg_strcasecmp(parallel, "safe") == 0)
+			proparallel = PROPARALLEL_SAFE;
+		else if (pg_strcasecmp(parallel, "restricted") == 0)
+			proparallel = PROPARALLEL_RESTRICTED;
+		else if (pg_strcasecmp(parallel, "unsafe") == 0)
+			proparallel = PROPARALLEL_UNSAFE;
+		else
+			ereport(ERROR,
+					(errcode(ERRCODE_SYNTAX_ERROR),
+					 errmsg("parameter \"parallel\" must be SAFE, RESTRICTED, or UNSAFE")));
+	}
+
 	/*
 	 * Most of the argument-checking is done inside of AggregateCreate
 	 */
@@ -480,5 +498,6 @@ DefineAggregate(List *name, List *args, bool oldstyle, List *parameters,
 						   mtransTypeId,		/* transition data type */
 						   mtransSpace, /* transition space */
 						   initval,		/* initial condition */
-						   minitval);	/* initial condition */
+						   minitval,	/* initial condition */
+						   proparallel);		/* parallel safe? */
 }
diff --git a/src/backend/commands/functioncmds.c b/src/backend/commands/functioncmds.c
index a745d73c7a5..748c8f75d48 100644
--- a/src/backend/commands/functioncmds.c
+++ b/src/backend/commands/functioncmds.c
@@ -566,9 +566,8 @@ interpret_func_parallel(DefElem *defel)
 	else
 	{
 		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-				 errmsg("parallel option \"%s\" not recognized",
-						str)));
+				(errcode(ERRCODE_SYNTAX_ERROR),
+				 errmsg("parameter \"parallel\" must be SAFE, RESTRICTED, or UNSAFE")));
 		return PROPARALLEL_UNSAFE;		/* keep compiler quiet */
 	}
 }
diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index c615717dea3..5674a73dfe0 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -1419,6 +1419,13 @@ has_parallel_hazard_walker(Node *node, has_parallel_hazard_arg *context)
 		if (parallel_too_dangerous(func_parallel(expr->funcid), context))
 			return true;
 	}
+	else if (IsA(node, Aggref))
+	{
+		Aggref	   *aggref = (Aggref *) node;
+
+		if (parallel_too_dangerous(func_parallel(aggref->aggfnoid), context))
+			return true;
+	}
 	else if (IsA(node, OpExpr))
 	{
 		OpExpr	   *expr = (OpExpr *) node;