Skip to content

Commit 41ea0c2

Browse files
robertmhaasrobertmhaas
committed
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.
1 parent 09adc9a commit 41ea0c2

File tree

8 files changed

+63
-11
lines changed

8 files changed

+63
-11
lines changed

‎doc/src/sgml/ref/create_aggregate.sgml

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ CREATE AGGREGATE <replaceable class="parameter">name</replaceable> ( [ <replacea
4040
[ , MFINALFUNC_EXTRA ]
4141
[ , MINITCOND = <replaceable class="PARAMETER">minitial_condition</replaceable> ]
4242
[ , SORTOP = <replaceable class="PARAMETER">sort_operator</replaceable> ]
43+
[ , PARALLEL = { SAFE | RESTRICTED | UNSAFE } ]
4344
)
4445

4546
CREATE AGGREGATE <replaceable class="parameter">name</replaceable> ( [ [ <replaceable class="parameter">argmode</replaceable> ] [ <replaceable class="parameter">argname</replaceable> ] <replaceable class="parameter">arg_data_type</replaceable> [ , ... ] ]
@@ -55,6 +56,8 @@ CREATE AGGREGATE <replaceable class="parameter">name</replaceable> ( [ [ <replac
5556
[ , SERIALTYPE = <replaceable class="PARAMETER">serialtype</replaceable> ]
5657
[ , INITCOND = <replaceable class="PARAMETER">initial_condition</replaceable> ]
5758
[ , HYPOTHETICAL ]
59+
[ , PARALLEL = { SAFE | RESTRICTED | UNSAFE } ]
60+
5861
)
5962

6063
<phrase>or the old syntax</phrase>
@@ -684,6 +687,12 @@ SELECT col FROM tab ORDER BY col USING sortop LIMIT 1;
684687
Currently, ordered-set aggregates do not need to support
685688
moving-aggregate mode, since they cannot be used as window functions.
686689
</para>
690+
691+
<para>
692+
The meaning of <literal>PARALLEL SAFE</>, <literal>PARALLEL RESTRICTED</>,
693+
and <literal>PARALLEL UNSAFE</> is the same as for
694+
<xref linkend="sql-createfunction">.
695+
</para>
687696
</refsect1>
688697

689698
<refsect1>

‎src/backend/catalog/pg_aggregate.c

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,8 @@ AggregateCreate(const char *aggName,
7272
Oid aggmTransType,
7373
int32 aggmTransSpace,
7474
const char *agginitval,
75-
const char *aggminitval)
75+
const char *aggminitval,
76+
char proparallel)
7677
{
7778
Relation aggdesc;
7879
HeapTuple tup;
@@ -622,7 +623,7 @@ AggregateCreate(const char *aggName,
622623
false, /* isStrict (not needed for agg) */
623624
PROVOLATILE_IMMUTABLE, /* volatility (not
624625
* needed for agg) */
625-
PROPARALLEL_UNSAFE,
626+
proparallel,
626627
parameterTypes, /* paramTypes */
627628
allParameterTypes, /* allParamTypes */
628629
parameterModes, /* parameterModes */

‎src/backend/commands/aggregatecmds.c

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,7 @@ DefineAggregate(List *name, List *args, bool oldstyle, List *parameters,
7878
int32 mtransSpace = 0;
7979
char *initval = NULL;
8080
char *minitval = NULL;
81+
char *parallel = NULL;
8182
int numArgs;
8283
int numDirectArgs = 0;
8384
oidvector *parameterTypes;
@@ -91,6 +92,7 @@ DefineAggregate(List *name, List *args, bool oldstyle, List *parameters,
9192
Oid mtransTypeId = InvalidOid;
9293
char transTypeType;
9394
char mtransTypeType = 0;
95+
char proparallel = PROPARALLEL_UNSAFE;
9496
ListCell *pl;
9597

9698
/* Convert list of names to a name and namespace */
@@ -178,6 +180,8 @@ DefineAggregate(List *name, List *args, bool oldstyle, List *parameters,
178180
initval = defGetString(defel);
179181
else if (pg_strcasecmp(defel->defname, "minitcond") == 0)
180182
minitval = defGetString(defel);
183+
else if (pg_strcasecmp(defel->defname, "parallel") == 0)
184+
parallel = defGetString(defel);
181185
else
182186
ereport(WARNING,
183187
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -449,6 +453,20 @@ DefineAggregate(List *name, List *args, bool oldstyle, List *parameters,
449453
(void) OidInputFunctionCall(typinput, minitval, typioparam, -1);
450454
}
451455

456+
if (parallel)
457+
{
458+
if (pg_strcasecmp(parallel, "safe") == 0)
459+
proparallel = PROPARALLEL_SAFE;
460+
else if (pg_strcasecmp(parallel, "restricted") == 0)
461+
proparallel = PROPARALLEL_RESTRICTED;
462+
else if (pg_strcasecmp(parallel, "unsafe") == 0)
463+
proparallel = PROPARALLEL_UNSAFE;
464+
else
465+
ereport(ERROR,
466+
(errcode(ERRCODE_SYNTAX_ERROR),
467+
errmsg("parameter \"parallel\" must be SAFE, RESTRICTED, or UNSAFE")));
468+
}
469+
452470
/*
453471
* Most of the argument-checking is done inside of AggregateCreate
454472
*/
@@ -480,5 +498,6 @@ DefineAggregate(List *name, List *args, bool oldstyle, List *parameters,
480498
mtransTypeId, /* transition data type */
481499
mtransSpace, /* transition space */
482500
initval, /* initial condition */
483-
minitval); /* initial condition */
501+
minitval, /* initial condition */
502+
proparallel); /* parallel safe? */
484503
}

‎src/backend/commands/functioncmds.c

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -566,9 +566,8 @@ interpret_func_parallel(DefElem *defel)
566566
else
567567
{
568568
ereport(ERROR,
569-
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
570-
errmsg("parallel option \"%s\" not recognized",
571-
str)));
569+
(errcode(ERRCODE_SYNTAX_ERROR),
570+
errmsg("parameter \"parallel\" must be SAFE, RESTRICTED, or UNSAFE")));
572571
return PROPARALLEL_UNSAFE; /* keep compiler quiet */
573572
}
574573
}

‎src/backend/optimizer/util/clauses.c

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1419,6 +1419,13 @@ has_parallel_hazard_walker(Node *node, has_parallel_hazard_arg *context)
14191419
if (parallel_too_dangerous(func_parallel(expr->funcid), context))
14201420
return true;
14211421
}
1422+
else if (IsA(node, Aggref))
1423+
{
1424+
Aggref *aggref = (Aggref *) node;
1425+
1426+
if (parallel_too_dangerous(func_parallel(aggref->aggfnoid), context))
1427+
return true;
1428+
}
14221429
else if (IsA(node, OpExpr))
14231430
{
14241431
OpExpr *expr = (OpExpr *) node;

‎src/include/catalog/pg_aggregate.h

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -349,6 +349,7 @@ extern ObjectAddress AggregateCreate(const char *aggName,
349349
Oid aggmTransType,
350350
int32 aggmTransSpace,
351351
const char *agginitval,
352-
const char *aggminitval);
352+
const char *aggminitval,
353+
char proparallel);
353354

354355
#endif /* PG_AGGREGATE_H */

‎src/test/regress/expected/create_aggregate.out

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,9 @@ CREATE AGGREGATE newsum (
2020
-- zero-argument aggregate
2121
CREATE AGGREGATE newcnt (*) (
2222
sfunc = int8inc, stype = int8,
23-
initcond = '0'
23+
initcond = '0', parallel = safe
2424
);
25-
-- old-style spelling of same
25+
-- old-style spelling of same (except without parallel-safe; that's too new)
2626
CREATE AGGREGATE oldcnt (
2727
sfunc = int8inc, basetype = 'ANY', stype = int8,
2828
initcond = '0'
@@ -188,6 +188,14 @@ WHERE aggfnoid = 'myavg'::REGPROC;
188188
(1 row)
189189

190190
DROP AGGREGATE myavg (numeric);
191+
-- invalid: bad parallel-safety marking
192+
CREATE AGGREGATE mysum (int)
193+
(
194+
stype = int,
195+
sfunc = int4pl,
196+
parallel = pear
197+
);
198+
ERROR: parameter "parallel" must be SAFE, RESTRICTED, or UNSAFE
191199
-- invalid: nonstrict inverse with strict forward function
192200
CREATE FUNCTION float8mi_n(float8, float8) RETURNS float8 AS
193201
$$ SELECT $1 - $2; $$

‎src/test/regress/sql/create_aggregate.sql

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,10 +23,10 @@ CREATE AGGREGATE newsum (
2323
-- zero-argument aggregate
2424
CREATE AGGREGATE newcnt (*) (
2525
sfunc = int8inc, stype = int8,
26-
initcond = '0'
26+
initcond = '0', parallel = safe
2727
);
2828

29-
-- old-style spelling of same
29+
-- old-style spelling of same (except without parallel-safe; that's too new)
3030
CREATE AGGREGATE oldcnt (
3131
sfunc = int8inc, basetype = 'ANY', stype = int8,
3232
initcond = '0'
@@ -201,6 +201,14 @@ WHERE aggfnoid = 'myavg'::REGPROC;
201201

202202
DROP AGGREGATE myavg (numeric);
203203

204+
-- invalid: bad parallel-safety marking
205+
CREATE AGGREGATE mysum (int)
206+
(
207+
stype = int,
208+
sfunc = int4pl,
209+
parallel = pear
210+
);
211+
204212
-- invalid: nonstrict inverse with strict forward function
205213

206214
CREATE FUNCTION float8mi_n(float8, float8) RETURNS float8 AS

0 commit comments

Comments
 (0)