From 11c8669c0cc7abc7a7702594cf13452c378d2517 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Tue, 5 Apr 2016 14:24:59 -0400
Subject: [PATCH] Add parallel query support functions for assorted aggregates.
This lets us use parallel aggregate for a variety of useful cases
that didn't work before, like sum(int8), sum(numeric), several
versions of avg(), and various other functions.
Add some regression tests, as well, testing the general sanity of
these and future catalog entries.
David Rowley, reviewed by Tomas Vondra, with a few further changes
by me.
---
src/backend/utils/adt/numeric.c | 860 ++++++++++++++++++
src/backend/utils/adt/timestamp.c | 49 +
src/include/catalog/catversion.h | 2 +-
src/include/catalog/pg_aggregate.h | 108 +--
src/include/catalog/pg_proc.h | 28 +
src/include/utils/builtins.h | 13 +
src/include/utils/timestamp.h | 1 +
.../regress/expected/create_aggregate.out | 91 +-
src/test/regress/expected/opr_sanity.out | 97 +-
src/test/regress/sql/create_aggregate.sql | 85 +-
src/test/regress/sql/opr_sanity.sql | 62 +-
11 files changed, 1317 insertions(+), 79 deletions(-)
diff --git a/src/backend/utils/adt/numeric.c b/src/backend/utils/adt/numeric.c
index 07b264572d9..3ba373a15b9 100644
--- a/src/backend/utils/adt/numeric.c
+++ b/src/backend/utils/adt/numeric.c
@@ -436,6 +436,7 @@ static int32 numericvar_to_int32(NumericVar *var);
static bool numericvar_to_int64(NumericVar *var, int64 *result);
static void int64_to_numericvar(int64 val, NumericVar *var);
#ifdef HAVE_INT128
+static bool numericvar_to_int128(NumericVar *var, int128 *result);
static void int128_to_numericvar(int128 val, NumericVar *var);
#endif
static double numeric_to_double_no_overflow(Numeric num);
@@ -3348,6 +3349,77 @@ numeric_accum(PG_FUNCTION_ARGS)
PG_RETURN_POINTER(state);
}
+/*
+ * Generic combine function for numeric aggregates which require sumX2
+ */
+Datum
+numeric_combine(PG_FUNCTION_ARGS)
+{
+ NumericAggState *state1;
+ NumericAggState *state2;
+ MemoryContext agg_context;
+ MemoryContext old_context;
+
+ if (!AggCheckCallContext(fcinfo, &agg_context))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ state1 = PG_ARGISNULL(0) ? NULL : (NumericAggState *) PG_GETARG_POINTER(0);
+ state2 = PG_ARGISNULL(1) ? NULL : (NumericAggState *) PG_GETARG_POINTER(1);
+
+ if (state2 == NULL)
+ PG_RETURN_POINTER(state1);
+
+ /* manually copy all fields from state2 to state1 */
+ if (state1 == NULL)
+ {
+ old_context = MemoryContextSwitchTo(agg_context);
+
+ state1 = makeNumericAggState(fcinfo, true);
+ state1->N = state2->N;
+ state1->NaNcount = state2->NaNcount;
+ state1->maxScale = state2->maxScale;
+ state1->maxScaleCount = state2->maxScaleCount;
+
+ init_var(&state1->sumX);
+ set_var_from_var(&state2->sumX, &state1->sumX);
+
+ init_var(&state1->sumX2);
+ set_var_from_var(&state2->sumX2, &state1->sumX2);
+
+ MemoryContextSwitchTo(old_context);
+
+ PG_RETURN_POINTER(state1);
+ }
+
+ if (state2->N > 0)
+ {
+ state1->N += state2->N;
+ state1->NaNcount += state2->NaNcount;
+
+ /*
+ * These are currently only needed for moving aggregates, but let's
+ * do the right thing anyway...
+ */
+ if (state2->maxScale > state1->maxScale)
+ {
+ state1->maxScale = state2->maxScale;
+ state1->maxScaleCount = state2->maxScaleCount;
+ }
+ else if (state2->maxScale == state1->maxScale)
+ state1->maxScaleCount += state2->maxScaleCount;
+
+ /* The rest of this needs to work in the aggregate context */
+ old_context = MemoryContextSwitchTo(agg_context);
+
+ /* Accumulate sums */
+ add_var(&(state1->sumX), &(state2->sumX), &(state1->sumX));
+ add_var(&(state1->sumX2), &(state2->sumX2), &(state1->sumX2));
+
+ MemoryContextSwitchTo(old_context);
+ }
+ PG_RETURN_POINTER(state1);
+}
+
/*
* Generic transition function for numeric aggregates that don't require sumX2.
*/
@@ -3368,6 +3440,307 @@ numeric_avg_accum(PG_FUNCTION_ARGS)
PG_RETURN_POINTER(state);
}
+/*
+ * Combine function for numeric aggregates which don't require sumX2
+ */
+Datum
+numeric_avg_combine(PG_FUNCTION_ARGS)
+{
+ NumericAggState *state1;
+ NumericAggState *state2;
+ MemoryContext agg_context;
+ MemoryContext old_context;
+
+ if (!AggCheckCallContext(fcinfo, &agg_context))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ state1 = PG_ARGISNULL(0) ? NULL : (NumericAggState *) PG_GETARG_POINTER(0);
+ state2 = PG_ARGISNULL(1) ? NULL : (NumericAggState *) PG_GETARG_POINTER(1);
+
+ if (state2 == NULL)
+ PG_RETURN_POINTER(state1);
+
+ /* manually copy all fields from state2 to state1 */
+ if (state1 == NULL)
+ {
+ old_context = MemoryContextSwitchTo(agg_context);
+
+ state1 = makeNumericAggState(fcinfo, false);
+ state1->N = state2->N;
+ state1->NaNcount = state2->NaNcount;
+ state1->maxScale = state2->maxScale;
+ state1->maxScaleCount = state2->maxScaleCount;
+
+ init_var(&state1->sumX);
+ set_var_from_var(&state2->sumX, &state1->sumX);
+
+ MemoryContextSwitchTo(old_context);
+
+ PG_RETURN_POINTER(state1);
+ }
+
+ if (state2->N > 0)
+ {
+ state1->N += state2->N;
+ state1->NaNcount += state2->NaNcount;
+
+ /*
+ * These are currently only needed for moving aggregates, but let's
+ * do the right thing anyway...
+ */
+ if (state2->maxScale > state1->maxScale)
+ {
+ state1->maxScale = state2->maxScale;
+ state1->maxScaleCount = state2->maxScaleCount;
+ }
+ else if (state2->maxScale == state1->maxScale)
+ state1->maxScaleCount += state2->maxScaleCount;
+
+ /* The rest of this needs to work in the aggregate context */
+ old_context = MemoryContextSwitchTo(agg_context);
+
+ /* Accumulate sums */
+ add_var(&(state1->sumX), &(state2->sumX), &(state1->sumX));
+
+ MemoryContextSwitchTo(old_context);
+ }
+ PG_RETURN_POINTER(state1);
+}
+
+/*
+ * numeric_avg_serialize
+ * Serialize NumericAggState for numeric aggregates that don't require
+ * sumX2. Serializes NumericAggState into bytea using the standard pq API.
+ *
+ * numeric_avg_deserialize(numeric_avg_serialize(state)) must result in a state
+ * which matches the original input state.
+ */
+Datum
+numeric_avg_serialize(PG_FUNCTION_ARGS)
+{
+ NumericAggState *state;
+ StringInfoData buf;
+ Datum temp;
+ bytea *sumX;
+ bytea *result;
+
+ /* Ensure we disallow calling when not in aggregate context */
+ if (!AggCheckCallContext(fcinfo, NULL))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ state = (NumericAggState *) PG_GETARG_POINTER(0);
+
+ /*
+ * This is a little wasteful since make_result converts the NumericVar
+ * into a Numeric and numeric_send converts it back again. Is it worth
+ * splitting the tasks in numeric_send into separate functions to stop
+ * this? Doing so would also remove the fmgr call overhead.
+ */
+ temp = DirectFunctionCall1(numeric_send,
+ NumericGetDatum(make_result(&state->sumX)));
+ sumX = DatumGetByteaP(temp);
+
+ pq_begintypsend(&buf);
+
+ /* N */
+ pq_sendint64(&buf, state->N);
+
+ /* sumX */
+ pq_sendbytes(&buf, VARDATA(sumX), VARSIZE(sumX) - VARHDRSZ);
+
+ /* maxScale */
+ pq_sendint(&buf, state->maxScale, 4);
+
+ /* maxScaleCount */
+ pq_sendint64(&buf, state->maxScaleCount);
+
+ /* NaNcount */
+ pq_sendint64(&buf, state->NaNcount);
+
+ result = pq_endtypsend(&buf);
+
+ PG_RETURN_BYTEA_P(result);
+}
+
+/*
+ * numeric_avg_deserialize
+ * Deserialize bytea into NumericAggState for numeric aggregates that
+ * don't require sumX2. Deserializes bytea into NumericAggState using the
+ * standard pq API.
+ *
+ * numeric_avg_serialize(numeric_avg_deserialize(bytea)) must result in a value
+ * which matches the original bytea value.
+ */
+Datum
+numeric_avg_deserialize(PG_FUNCTION_ARGS)
+{
+ bytea *sstate = PG_GETARG_BYTEA_P(0);
+ NumericAggState *result;
+ Datum temp;
+ StringInfoData buf;
+
+ if (!AggCheckCallContext(fcinfo, NULL))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ /*
+ * Copy the bytea into a StringInfo so that we can "receive" it using the
+ * standard pq API.
+ */
+ initStringInfo(&buf);
+ appendBinaryStringInfo(&buf, VARDATA(sstate), VARSIZE(sstate) - VARHDRSZ);
+
+ result = makeNumericAggState(fcinfo, false);
+
+ /* N */
+ result->N = pq_getmsgint64(&buf);
+
+ /* sumX */
+ temp = DirectFunctionCall3(numeric_recv,
+ PointerGetDatum(&buf),
+ InvalidOid,
+ -1);
+ set_var_from_num(DatumGetNumeric(temp), &result->sumX);
+
+ /* maxScale */
+ result->maxScale = pq_getmsgint(&buf, 4);
+
+ /* maxScaleCount */
+ result->maxScaleCount = pq_getmsgint64(&buf);
+
+ /* NaNcount */
+ result->NaNcount = pq_getmsgint64(&buf);
+
+ pq_getmsgend(&buf);
+ pfree(buf.data);
+
+ PG_RETURN_POINTER(result);
+}
+
+/*
+ * numeric_serialize
+ * Serialization function for NumericAggState for numeric aggregates that
+ * require sumX2. Serializes NumericAggState into bytea using the standard
+ * pq API.
+ *
+ * numeric_deserialize(numeric_serialize(state)) must result in a state which
+ * matches the original input state.
+ */
+Datum
+numeric_serialize(PG_FUNCTION_ARGS)
+{
+ NumericAggState *state;
+ StringInfoData buf;
+ Datum temp;
+ bytea *sumX;
+ bytea *sumX2;
+ bytea *result;
+
+ /* Ensure we disallow calling when not in aggregate context */
+ if (!AggCheckCallContext(fcinfo, NULL))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ state = (NumericAggState *) PG_GETARG_POINTER(0);
+
+ /*
+ * This is a little wasteful since make_result converts the NumericVar
+ * into a Numeric and numeric_send converts it back again. Is it worth
+ * splitting the tasks in numeric_send into separate functions to stop
+ * this? Doing so would also remove the fmgr call overhead.
+ */
+ temp = DirectFunctionCall1(numeric_send,
+ NumericGetDatum(make_result(&state->sumX)));
+ sumX = DatumGetByteaP(temp);
+
+ temp = DirectFunctionCall1(numeric_send,
+ NumericGetDatum(make_result(&state->sumX2)));
+ sumX2 = DatumGetByteaP(temp);
+
+ pq_begintypsend(&buf);
+
+ /* N */
+ pq_sendint64(&buf, state->N);
+
+ /* sumX */
+ pq_sendbytes(&buf, VARDATA(sumX), VARSIZE(sumX) - VARHDRSZ);
+
+ /* sumX2 */
+ pq_sendbytes(&buf, VARDATA(sumX2), VARSIZE(sumX2) - VARHDRSZ);
+
+ /* maxScale */
+ pq_sendint(&buf, state->maxScale, 4);
+
+ /* maxScaleCount */
+ pq_sendint64(&buf, state->maxScaleCount);
+
+ /* NaNcount */
+ pq_sendint64(&buf, state->NaNcount);
+
+ result = pq_endtypsend(&buf);
+
+ PG_RETURN_BYTEA_P(result);
+}
+
+/*
+ * numeric_deserialize
+ * Deserialization function for NumericAggState for numeric aggregates that
+ * require sumX2. Deserializes bytea into into NumericAggState using the
+ * standard pq API.
+ *
+ * numeric_serialize(numeric_deserialize(bytea)) must result in a value which
+ * matches the original bytea value.
+ */
+Datum
+numeric_deserialize(PG_FUNCTION_ARGS)
+{
+ bytea *sstate = PG_GETARG_BYTEA_P(0);
+ NumericAggState *result;
+ Datum temp;
+ StringInfoData buf;
+
+ if (!AggCheckCallContext(fcinfo, NULL))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ /*
+ * Copy the bytea into a StringInfo so that we can "receive" it using the
+ * standard pq API.
+ */
+ initStringInfo(&buf);
+ appendBinaryStringInfo(&buf, VARDATA(sstate), VARSIZE(sstate) - VARHDRSZ);
+
+ result = makeNumericAggState(fcinfo, false);
+
+ /* N */
+ result->N = pq_getmsgint64(&buf);
+
+ /* sumX */
+ temp = DirectFunctionCall3(numeric_recv,
+ PointerGetDatum(&buf),
+ InvalidOid,
+ -1);
+ set_var_from_num(DatumGetNumeric(temp), &result->sumX);
+
+ /* sumX2 */
+ temp = DirectFunctionCall3(numeric_recv,
+ PointerGetDatum(&buf),
+ InvalidOid,
+ -1);
+ set_var_from_num(DatumGetNumeric(temp), &result->sumX2);
+
+ /* maxScale */
+ result->maxScale = pq_getmsgint(&buf, 4);
+
+ /* maxScaleCount */
+ result->maxScaleCount = pq_getmsgint64(&buf);
+
+ /* NaNcount */
+ result->NaNcount = pq_getmsgint64(&buf);
+
+ pq_getmsgend(&buf);
+ pfree(buf.data);
+
+ PG_RETURN_POINTER(result);
+}
+
/*
* Generic inverse transition function for numeric aggregates
* (with or without requirement for X^2).
@@ -3551,6 +3924,215 @@ int8_accum(PG_FUNCTION_ARGS)
PG_RETURN_POINTER(state);
}
+/*
+ * Combine function for numeric aggregates which require sumX2
+ */
+Datum
+numeric_poly_combine(PG_FUNCTION_ARGS)
+{
+ PolyNumAggState *state1;
+ PolyNumAggState *state2;
+ MemoryContext agg_context;
+ MemoryContext old_context;
+
+ if (!AggCheckCallContext(fcinfo, &agg_context))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ state1 = PG_ARGISNULL(0) ? NULL : (PolyNumAggState *) PG_GETARG_POINTER(0);
+ state2 = PG_ARGISNULL(1) ? NULL : (PolyNumAggState *) PG_GETARG_POINTER(1);
+
+ if (state2 == NULL)
+ PG_RETURN_POINTER(state1);
+
+ /* manually copy all fields from state2 to state1 */
+ if (state1 == NULL)
+ {
+ old_context = MemoryContextSwitchTo(agg_context);
+
+ state1 = makePolyNumAggState(fcinfo, true);
+ state1->N = state2->N;
+
+#ifdef HAVE_INT128
+ state1->sumX = state2->sumX;
+ state1->sumX2 = state2->sumX2;
+#else
+ init_var(&(state1->sumX));
+ set_var_from_var(&(state2->sumX), &(state1->sumX));
+
+ init_var(&state1->sumX2);
+ set_var_from_var(&(state2->sumX2), &(state1->sumX2));
+#endif
+
+ MemoryContextSwitchTo(old_context);
+
+ PG_RETURN_POINTER(state1);
+ }
+
+ if (state2->N > 0)
+ {
+ state1->N += state2->N;
+
+#ifdef HAVE_INT128
+ state1->sumX += state2->sumX;
+ state1->sumX2 += state2->sumX2;
+#else
+ /* The rest of this needs to work in the aggregate context */
+ old_context = MemoryContextSwitchTo(agg_context);
+
+ /* Accumulate sums */
+ add_var(&(state1->sumX), &(state2->sumX), &(state1->sumX));
+ add_var(&(state1->sumX2), &(state2->sumX2), &(state1->sumX2));
+
+ MemoryContextSwitchTo(old_context);
+#endif
+
+ }
+ PG_RETURN_POINTER(state1);
+}
+
+/*
+ * numeric_poly_serialize
+ * Serialize PolyNumAggState into bytea using the standard pq API for
+ * aggregate functions which require sumX2.
+ *
+ * numeric_poly_deserialize(numeric_poly_serialize(state)) must result in a
+ * state which matches the original input state.
+ */
+Datum
+numeric_poly_serialize(PG_FUNCTION_ARGS)
+{
+ PolyNumAggState *state;
+ StringInfoData buf;
+ bytea *sumX;
+ bytea *sumX2;
+ bytea *result;
+
+ /* Ensure we disallow calling when not in aggregate context */
+ if (!AggCheckCallContext(fcinfo, NULL))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ state = (PolyNumAggState *) PG_GETARG_POINTER(0);
+
+ /*
+ * If the platform supports int128 then sumX and sumX2 will be a 128 bit
+ * integer type. Here we'll convert that into a numeric type so that the
+ * combine state is in the same format for both int128 enabled machines
+ * and machines which don't support that type. The logic here is that one
+ * day we might like to send these over to another server for further
+ * processing and we want a standard format to work with.
+ */
+ {
+ Datum temp;
+
+#ifdef HAVE_INT128
+ NumericVar num;
+
+ init_var(&num);
+ int128_to_numericvar(state->sumX, &num);
+ temp = DirectFunctionCall1(numeric_send,
+ NumericGetDatum(make_result(&num)));
+ sumX = DatumGetByteaP(temp);
+
+ int128_to_numericvar(state->sumX2, &num);
+ temp = DirectFunctionCall1(numeric_send,
+ NumericGetDatum(make_result(&num)));
+ sumX2 = DatumGetByteaP(temp);
+ free_var(&num);
+#else
+ temp = DirectFunctionCall1(numeric_send,
+ NumericGetDatum(make_result(&state->sumX)));
+ sumX = DatumGetByteaP(temp);
+
+ temp = DirectFunctionCall1(numeric_send,
+ NumericGetDatum(make_result(&state->sumX2)));
+ sumX2 = DatumGetByteaP(temp);
+#endif
+ }
+
+ pq_begintypsend(&buf);
+
+ /* N */
+ pq_sendint64(&buf, state->N);
+
+ /* sumX */
+ pq_sendbytes(&buf, VARDATA(sumX), VARSIZE(sumX) - VARHDRSZ);
+
+ /* sumX2 */
+ pq_sendbytes(&buf, VARDATA(sumX2), VARSIZE(sumX2) - VARHDRSZ);
+
+ result = pq_endtypsend(&buf);
+
+ PG_RETURN_BYTEA_P(result);
+}
+
+/*
+ * numeric_poly_deserialize
+ * Deserialize PolyNumAggState from bytea using the standard pq API for
+ * aggregate functions which require sumX2.
+ *
+ * numeric_poly_serialize(numeric_poly_deserialize(bytea)) must result in a
+ * state which matches the original input state.
+ */
+Datum
+numeric_poly_deserialize(PG_FUNCTION_ARGS)
+{
+ bytea *sstate = PG_GETARG_BYTEA_P(0);
+ PolyNumAggState *result;
+ Datum sumX;
+ Datum sumX2;
+ StringInfoData buf;
+
+ if (!AggCheckCallContext(fcinfo, NULL))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ /*
+ * Copy the bytea into a StringInfo so that we can "receive" it using the
+ * standard pq API.
+ */
+ initStringInfo(&buf);
+ appendBinaryStringInfo(&buf, VARDATA(sstate), VARSIZE(sstate) - VARHDRSZ);
+
+ result = makePolyNumAggState(fcinfo, false);
+
+ /* N */
+ result->N = pq_getmsgint64(&buf);
+
+ /* sumX */
+ sumX = DirectFunctionCall3(numeric_recv,
+ PointerGetDatum(&buf),
+ InvalidOid,
+ -1);
+
+ /* sumX2 */
+ sumX2 = DirectFunctionCall3(numeric_recv,
+ PointerGetDatum(&buf),
+ InvalidOid,
+ -1);
+
+#ifdef HAVE_INT128
+ {
+ NumericVar num;
+
+ init_var(&num);
+ set_var_from_num(DatumGetNumeric(sumX), &num);
+ numericvar_to_int128(&num, &result->sumX);
+
+ set_var_from_num(DatumGetNumeric(sumX2), &num);
+ numericvar_to_int128(&num, &result->sumX2);
+
+ free_var(&num);
+ }
+#else
+ set_var_from_num(DatumGetNumeric(sumX), &result->sumX);
+ set_var_from_num(DatumGetNumeric(sumX2), &result->sumX2);
+#endif
+
+ pq_getmsgend(&buf);
+ pfree(buf.data);
+
+ PG_RETURN_POINTER(result);
+}
+
/*
* Transition function for int8 input when we don't need sumX2.
*/
@@ -3581,6 +4163,180 @@ int8_avg_accum(PG_FUNCTION_ARGS)
PG_RETURN_POINTER(state);
}
+/*
+ * Combine function for PolyNumAggState for aggregates which don't require
+ * sumX2
+ */
+Datum
+int8_avg_combine(PG_FUNCTION_ARGS)
+{
+ PolyNumAggState *state1;
+ PolyNumAggState *state2;
+ MemoryContext agg_context;
+ MemoryContext old_context;
+
+ if (!AggCheckCallContext(fcinfo, &agg_context))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ state1 = PG_ARGISNULL(0) ? NULL : (PolyNumAggState *) PG_GETARG_POINTER(0);
+ state2 = PG_ARGISNULL(1) ? NULL : (PolyNumAggState *) PG_GETARG_POINTER(1);
+
+ if (state2 == NULL)
+ PG_RETURN_POINTER(state1);
+
+ /* manually copy all fields from state2 to state1 */
+ if (state1 == NULL)
+ {
+ old_context = MemoryContextSwitchTo(agg_context);
+
+ state1 = makePolyNumAggState(fcinfo, false);
+ state1->N = state2->N;
+
+#ifdef HAVE_INT128
+ state1->sumX = state2->sumX;
+#else
+ init_var(&state1->sumX);
+ set_var_from_var(&state2->sumX, &state1->sumX);
+#endif
+ MemoryContextSwitchTo(old_context);
+
+ PG_RETURN_POINTER(state1);
+ }
+
+ if (state2->N > 0)
+ {
+ state1->N += state2->N;
+
+#ifdef HAVE_INT128
+ state1->sumX += state2->sumX;
+#else
+ /* The rest of this needs to work in the aggregate context */
+ old_context = MemoryContextSwitchTo(agg_context);
+
+ /* Accumulate sums */
+ add_var(&(state1->sumX), &(state2->sumX), &(state1->sumX));
+
+ MemoryContextSwitchTo(old_context);
+#endif
+
+ }
+ PG_RETURN_POINTER(state1);
+}
+
+/*
+ * int8_avg_serialize
+ * Serialize PolyNumAggState into bytea using the standard pq API.
+ *
+ * int8_avg_deserialize(int8_avg_serialize(state)) must result in a state which
+ * matches the original input state.
+ */
+Datum
+int8_avg_serialize(PG_FUNCTION_ARGS)
+{
+ PolyNumAggState *state;
+ StringInfoData buf;
+ bytea *sumX;
+ bytea *result;
+
+ /* Ensure we disallow calling when not in aggregate context */
+ if (!AggCheckCallContext(fcinfo, NULL))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ state = (PolyNumAggState *) PG_GETARG_POINTER(0);
+
+ /*
+ * If the platform supports int128 then sumX will be a 128 integer type.
+ * Here we'll convert that into a numeric type so that the combine state
+ * is in the same format for both int128 enabled machines and machines
+ * which don't support that type. The logic here is that one day we might
+ * like to send these over to another server for further processing and we
+ * want a standard format to work with.
+ */
+ {
+ Datum temp;
+#ifdef HAVE_INT128
+ NumericVar num;
+
+ init_var(&num);
+ int128_to_numericvar(state->sumX, &num);
+ temp = DirectFunctionCall1(numeric_send,
+ NumericGetDatum(make_result(&num)));
+ free_var(&num);
+ sumX = DatumGetByteaP(temp);
+#else
+ temp = DirectFunctionCall1(numeric_send,
+ NumericGetDatum(make_result(&state->sumX)));
+ sumX = DatumGetByteaP(temp);
+#endif
+ }
+
+ pq_begintypsend(&buf);
+
+ /* N */
+ pq_sendint64(&buf, state->N);
+
+ /* sumX */
+ pq_sendbytes(&buf, VARDATA(sumX), VARSIZE(sumX) - VARHDRSZ);
+
+ result = pq_endtypsend(&buf);
+
+ PG_RETURN_BYTEA_P(result);
+}
+
+/*
+ * int8_avg_deserialize
+ * Deserialize bytea back into PolyNumAggState.
+ *
+ * int8_avg_serialize(int8_avg_deserialize(bytea)) must result in a value which
+ * matches the original bytea value.
+ */
+Datum
+int8_avg_deserialize(PG_FUNCTION_ARGS)
+{
+ bytea *sstate = PG_GETARG_BYTEA_P(0);
+ PolyNumAggState *result;
+ StringInfoData buf;
+ Datum temp;
+
+ if (!AggCheckCallContext(fcinfo, NULL))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ /*
+ * Copy the bytea into a StringInfo so that we can "receive" it using the
+ * standard pq API.
+ */
+ initStringInfo(&buf);
+ appendBinaryStringInfo(&buf, VARDATA(sstate), VARSIZE(sstate) - VARHDRSZ);
+
+ result = makePolyNumAggState(fcinfo, false);
+
+ /* N */
+ result->N = pq_getmsgint64(&buf);
+
+ /* sumX */
+ temp = DirectFunctionCall3(numeric_recv,
+ PointerGetDatum(&buf),
+ InvalidOid,
+ -1);
+
+#ifdef HAVE_INT128
+ {
+ NumericVar num;
+
+ init_var(&num);
+ set_var_from_num(DatumGetNumeric(temp), &num);
+ numericvar_to_int128(&num, &result->sumX);
+ free_var(&num);
+ }
+#else
+ set_var_from_num(DatumGetNumeric(temp), &result->sumX);
+#endif
+
+ pq_getmsgend(&buf);
+ pfree(buf.data);
+
+ PG_RETURN_POINTER(result);
+}
/*
* Inverse transition functions to go with the above.
@@ -4309,6 +5065,37 @@ int4_avg_accum(PG_FUNCTION_ARGS)
PG_RETURN_ARRAYTYPE_P(transarray);
}
+Datum
+int4_avg_combine(PG_FUNCTION_ARGS)
+{
+ ArrayType *transarray1;
+ ArrayType *transarray2;
+ Int8TransTypeData *state1;
+ Int8TransTypeData *state2;
+
+ if (!AggCheckCallContext(fcinfo, NULL))
+ elog(ERROR, "aggregate function called in non-aggregate context");
+
+ transarray1 = PG_GETARG_ARRAYTYPE_P(0);
+ transarray2 = PG_GETARG_ARRAYTYPE_P(1);
+
+ if (ARR_HASNULL(transarray1) ||
+ ARR_SIZE(transarray1) != ARR_OVERHEAD_NONULLS(1) + sizeof(Int8TransTypeData))
+ elog(ERROR, "expected 2-element int8 array");
+
+ if (ARR_HASNULL(transarray2) ||
+ ARR_SIZE(transarray2) != ARR_OVERHEAD_NONULLS(1) + sizeof(Int8TransTypeData))
+ elog(ERROR, "expected 2-element int8 array");
+
+ state1 = (Int8TransTypeData *) ARR_DATA_PTR(transarray1);
+ state2 = (Int8TransTypeData *) ARR_DATA_PTR(transarray2);
+
+ state1->count += state2->count;
+ state1->sum += state2->sum;
+
+ PG_RETURN_ARRAYTYPE_P(transarray1);
+}
+
Datum
int2_avg_accum_inv(PG_FUNCTION_ARGS)
{
@@ -5307,6 +6094,79 @@ int64_to_numericvar(int64 val, NumericVar *var)
}
#ifdef HAVE_INT128
+/*
+ * Convert numeric to int128, rounding if needed.
+ *
+ * If overflow, return FALSE (no error is raised). Return TRUE if okay.
+ */
+static bool
+numericvar_to_int128(NumericVar *var, int128 *result)
+{
+ NumericDigit *digits;
+ int ndigits;
+ int weight;
+ int i;
+ int128 val,
+ oldval;
+ bool neg;
+ NumericVar rounded;
+
+ /* Round to nearest integer */
+ init_var(&rounded);
+ set_var_from_var(var, &rounded);
+ round_var(&rounded, 0);
+
+ /* Check for zero input */
+ strip_var(&rounded);
+ ndigits = rounded.ndigits;
+ if (ndigits == 0)
+ {
+ *result = 0;
+ free_var(&rounded);
+ return true;
+ }
+
+ /*
+ * For input like 10000000000, we must treat stripped digits as real. So
+ * the loop assumes there are weight+1 digits before the decimal point.
+ */
+ weight = rounded.weight;
+ Assert(weight >= 0 && ndigits <= weight + 1);
+
+ /* Construct the result */
+ digits = rounded.digits;
+ neg = (rounded.sign == NUMERIC_NEG);
+ val = digits[0];
+ for (i = 1; i <= weight; i++)
+ {
+ oldval = val;
+ val *= NBASE;
+ if (i < ndigits)
+ val += digits[i];
+
+ /*
+ * The overflow check is a bit tricky because we want to accept
+ * INT128_MIN, which will overflow the positive accumulator. We can
+ * detect this case easily though because INT128_MIN is the only
+ * nonzero value for which -val == val (on a two's complement machine,
+ * anyway).
+ */
+ if ((val / NBASE) != oldval) /* possible overflow? */
+ {
+ if (!neg || (-val) != val || val == 0 || oldval < 0)
+ {
+ free_var(&rounded);
+ return false;
+ }
+ }
+ }
+
+ free_var(&rounded);
+
+ *result = neg ? -val : val;
+ return true;
+}
+
/*
* Convert 128 bit integer to numeric.
*/
diff --git a/src/backend/utils/adt/timestamp.c b/src/backend/utils/adt/timestamp.c
index 98a2ab5e58a..d7ee865cf75 100644
--- a/src/backend/utils/adt/timestamp.c
+++ b/src/backend/utils/adt/timestamp.c
@@ -3524,6 +3524,55 @@ interval_accum(PG_FUNCTION_ARGS)
PG_RETURN_ARRAYTYPE_P(result);
}
+Datum
+interval_combine(PG_FUNCTION_ARGS)
+{
+ ArrayType *transarray1 = PG_GETARG_ARRAYTYPE_P(0);
+ ArrayType *transarray2 = PG_GETARG_ARRAYTYPE_P(1);
+ Datum *transdatums1;
+ Datum *transdatums2;
+ int ndatums1;
+ int ndatums2;
+ Interval sum1,
+ N1;
+ Interval sum2,
+ N2;
+
+ Interval *newsum;
+ ArrayType *result;
+
+ deconstruct_array(transarray1,
+ INTERVALOID, sizeof(Interval), false, 'd',
+ &transdatums1, NULL, &ndatums1);
+ if (ndatums1 != 2)
+ elog(ERROR, "expected 2-element interval array");
+
+ sum1 = *(DatumGetIntervalP(transdatums1[0]));
+ N1 = *(DatumGetIntervalP(transdatums1[1]));
+
+ deconstruct_array(transarray2,
+ INTERVALOID, sizeof(Interval), false, 'd',
+ &transdatums2, NULL, &ndatums2);
+ if (ndatums2 != 2)
+ elog(ERROR, "expected 2-element interval array");
+
+ sum2 = *(DatumGetIntervalP(transdatums2[0]));
+ N2 = *(DatumGetIntervalP(transdatums2[1]));
+
+ newsum = DatumGetIntervalP(DirectFunctionCall2(interval_pl,
+ IntervalPGetDatum(&sum1),
+ IntervalPGetDatum(&sum2)));
+ N1.time += N2.time;
+
+ transdatums1[0] = IntervalPGetDatum(newsum);
+ transdatums1[1] = IntervalPGetDatum(&N1);
+
+ result = construct_array(transdatums1, 2,
+ INTERVALOID, sizeof(Interval), false, 'd');
+
+ PG_RETURN_ARRAYTYPE_P(result);
+}
+
Datum
interval_accum_inv(PG_FUNCTION_ARGS)
{
diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h
index 7e75f8ebb53..2be9aa9f49c 100644
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -53,6 +53,6 @@
*/
/* yyyymmddN */
-#define CATALOG_VERSION_NO 201604051
+#define CATALOG_VERSION_NO 201604052
#endif
diff --git a/src/include/catalog/pg_aggregate.h b/src/include/catalog/pg_aggregate.h
index 4205fabb576..101d073a04a 100644
--- a/src/include/catalog/pg_aggregate.h
+++ b/src/include/catalog/pg_aggregate.h
@@ -138,23 +138,23 @@ typedef FormData_pg_aggregate *Form_pg_aggregate;
*/
/* avg */
-DATA(insert ( 2100 n 0 int8_avg_accum numeric_poly_avg - - - int8_avg_accum int8_avg_accum_inv numeric_poly_avg f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2101 n 0 int4_avg_accum int8_avg - - - int4_avg_accum int4_avg_accum_inv int8_avg f f 0 1016 0 0 1016 0 "{0,0}" "{0,0}" ));
-DATA(insert ( 2102 n 0 int2_avg_accum int8_avg - - - int2_avg_accum int2_avg_accum_inv int8_avg f f 0 1016 0 0 1016 0 "{0,0}" "{0,0}" ));
-DATA(insert ( 2103 n 0 numeric_avg_accum numeric_avg - - - numeric_avg_accum numeric_accum_inv numeric_avg f f 0 2281 0 128 2281 128 _null_ _null_ ));
-DATA(insert ( 2104 n 0 float4_accum float8_avg - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2105 n 0 float8_accum float8_avg - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2106 n 0 interval_accum interval_avg - - - interval_accum interval_accum_inv interval_avg f f 0 1187 0 0 1187 0 "{0 second,0 second}" "{0 second,0 second}" ));
+DATA(insert ( 2100 n 0 int8_avg_accum numeric_poly_avg int8_avg_combine int8_avg_serialize int8_avg_deserialize int8_avg_accum int8_avg_accum_inv numeric_poly_avg f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2101 n 0 int4_avg_accum int8_avg int4_avg_combine - - int4_avg_accum int4_avg_accum_inv int8_avg f f 0 1016 0 0 1016 0 "{0,0}" "{0,0}" ));
+DATA(insert ( 2102 n 0 int2_avg_accum int8_avg int4_avg_combine - - int2_avg_accum int2_avg_accum_inv int8_avg f f 0 1016 0 0 1016 0 "{0,0}" "{0,0}" ));
+DATA(insert ( 2103 n 0 numeric_avg_accum numeric_avg numeric_avg_combine numeric_avg_serialize numeric_avg_deserialize numeric_avg_accum numeric_accum_inv numeric_avg f f 0 2281 17 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2104 n 0 float4_accum float8_avg - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2105 n 0 float8_accum float8_avg - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2106 n 0 interval_accum interval_avg interval_combine - - interval_accum interval_accum_inv interval_avg f f 0 1187 0 0 1187 0 "{0 second,0 second}" "{0 second,0 second}" ));
/* sum */
-DATA(insert ( 2107 n 0 int8_avg_accum numeric_poly_sum - - - int8_avg_accum int8_avg_accum_inv numeric_poly_sum f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2108 n 0 int4_sum - int8pl - - int4_avg_accum int4_avg_accum_inv int2int4_sum f f 0 20 0 0 1016 0 _null_ "{0,0}" ));
-DATA(insert ( 2109 n 0 int2_sum - int8pl - - int2_avg_accum int2_avg_accum_inv int2int4_sum f f 0 20 0 0 1016 0 _null_ "{0,0}" ));
-DATA(insert ( 2110 n 0 float4pl - float4pl - - - - - f f 0 700 0 0 0 0 _null_ _null_ ));
-DATA(insert ( 2111 n 0 float8pl - float8pl - - - - - f f 0 701 0 0 0 0 _null_ _null_ ));
-DATA(insert ( 2112 n 0 cash_pl - cash_pl - - cash_pl cash_mi - f f 0 790 0 0 790 0 _null_ _null_ ));
-DATA(insert ( 2113 n 0 interval_pl - interval_pl - - interval_pl interval_mi - f f 0 1186 0 0 1186 0 _null_ _null_ ));
-DATA(insert ( 2114 n 0 numeric_avg_accum numeric_sum - - - numeric_avg_accum numeric_accum_inv numeric_sum f f 0 2281 0 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2107 n 0 int8_avg_accum numeric_poly_sum int8_avg_combine int8_avg_serialize int8_avg_deserialize int8_avg_accum int8_avg_accum_inv numeric_poly_sum f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2108 n 0 int4_sum - int8pl - - int4_avg_accum int4_avg_accum_inv int2int4_sum f f 0 20 0 0 1016 0 _null_ "{0,0}" ));
+DATA(insert ( 2109 n 0 int2_sum - int8pl - - int2_avg_accum int2_avg_accum_inv int2int4_sum f f 0 20 0 0 1016 0 _null_ "{0,0}" ));
+DATA(insert ( 2110 n 0 float4pl - float4pl - - - - - f f 0 700 0 0 0 0 _null_ _null_ ));
+DATA(insert ( 2111 n 0 float8pl - float8pl - - - - - f f 0 701 0 0 0 0 _null_ _null_ ));
+DATA(insert ( 2112 n 0 cash_pl - cash_pl - - cash_pl cash_mi - f f 0 790 0 0 790 0 _null_ _null_ ));
+DATA(insert ( 2113 n 0 interval_pl - interval_pl - - interval_pl interval_mi - f f 0 1186 0 0 1186 0 _null_ _null_ ));
+DATA(insert ( 2114 n 0 numeric_avg_accum numeric_sum numeric_avg_combine numeric_avg_serialize numeric_avg_deserialize numeric_avg_accum numeric_accum_inv numeric_sum f f 0 2281 17 128 2281 128 _null_ _null_ ));
/* max */
DATA(insert ( 2115 n 0 int8larger - int8larger - - - - - f f 413 20 0 0 0 0 _null_ _null_ ));
@@ -207,52 +207,52 @@ DATA(insert ( 2147 n 0 int8inc_any - int8pl - - int8inc_any int8dec_any -
DATA(insert ( 2803 n 0 int8inc - int8pl - - int8inc int8dec - f f 0 20 0 0 20 0 "0" "0" ));
/* var_pop */
-DATA(insert ( 2718 n 0 int8_accum numeric_var_pop - - - int8_accum int8_accum_inv numeric_var_pop f f 0 2281 0 128 2281 128 _null_ _null_ ));
-DATA(insert ( 2719 n 0 int4_accum numeric_poly_var_pop - - - int4_accum int4_accum_inv numeric_poly_var_pop f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2720 n 0 int2_accum numeric_poly_var_pop - - - int2_accum int2_accum_inv numeric_poly_var_pop f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2721 n 0 float4_accum float8_var_pop - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2722 n 0 float8_accum float8_var_pop - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2723 n 0 numeric_accum numeric_var_pop - - - numeric_accum numeric_accum_inv numeric_var_pop f f 0 2281 0 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2718 n 0 int8_accum numeric_var_pop numeric_combine numeric_serialize numeric_deserialize int8_accum int8_accum_inv numeric_var_pop f f 0 2281 17 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2719 n 0 int4_accum numeric_poly_var_pop numeric_poly_combine numeric_poly_serialize numeric_poly_deserialize int4_accum int4_accum_inv numeric_poly_var_pop f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2720 n 0 int2_accum numeric_poly_var_pop numeric_poly_combine numeric_poly_serialize numeric_poly_deserialize int2_accum int2_accum_inv numeric_poly_var_pop f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2721 n 0 float4_accum float8_var_pop - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2722 n 0 float8_accum float8_var_pop - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2723 n 0 numeric_accum numeric_var_pop numeric_combine numeric_serialize numeric_deserialize numeric_accum numeric_accum_inv numeric_var_pop f f 0 2281 17 128 2281 128 _null_ _null_ ));
/* var_samp */
-DATA(insert ( 2641 n 0 int8_accum numeric_var_samp - - - int8_accum int8_accum_inv numeric_var_samp f f 0 2281 0 128 2281 128 _null_ _null_ ));
-DATA(insert ( 2642 n 0 int4_accum numeric_poly_var_samp - - - int4_accum int4_accum_inv numeric_poly_var_samp f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2643 n 0 int2_accum numeric_poly_var_samp - - - int2_accum int2_accum_inv numeric_poly_var_samp f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2644 n 0 float4_accum float8_var_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2645 n 0 float8_accum float8_var_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2646 n 0 numeric_accum numeric_var_samp - - - numeric_accum numeric_accum_inv numeric_var_samp f f 0 2281 0 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2641 n 0 int8_accum numeric_var_samp numeric_combine numeric_serialize numeric_deserialize int8_accum int8_accum_inv numeric_var_samp f f 0 2281 17 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2642 n 0 int4_accum numeric_poly_var_samp numeric_poly_combine numeric_poly_serialize numeric_poly_deserialize int4_accum int4_accum_inv numeric_poly_var_samp f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2643 n 0 int2_accum numeric_poly_var_samp numeric_poly_combine numeric_poly_serialize numeric_poly_deserialize int2_accum int2_accum_inv numeric_poly_var_samp f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2644 n 0 float4_accum float8_var_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2645 n 0 float8_accum float8_var_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2646 n 0 numeric_accum numeric_var_samp numeric_combine numeric_serialize numeric_deserialize numeric_accum numeric_accum_inv numeric_var_samp f f 0 2281 17 128 2281 128 _null_ _null_ ));
/* variance: historical Postgres syntax for var_samp */
-DATA(insert ( 2148 n 0 int8_accum numeric_var_samp - - - int8_accum int8_accum_inv numeric_var_samp f f 0 2281 0 128 2281 128 _null_ _null_ ));
-DATA(insert ( 2149 n 0 int4_accum numeric_poly_var_samp - - - int4_accum int4_accum_inv numeric_poly_var_samp f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2150 n 0 int2_accum numeric_poly_var_samp - - - int2_accum int2_accum_inv numeric_poly_var_samp f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2151 n 0 float4_accum float8_var_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2152 n 0 float8_accum float8_var_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2153 n 0 numeric_accum numeric_var_samp - - - numeric_accum numeric_accum_inv numeric_var_samp f f 0 2281 0 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2148 n 0 int8_accum numeric_var_samp numeric_combine numeric_serialize numeric_deserialize int8_accum int8_accum_inv numeric_var_samp f f 0 2281 17 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2149 n 0 int4_accum numeric_poly_var_samp numeric_poly_combine numeric_poly_serialize numeric_poly_deserialize int4_accum int4_accum_inv numeric_poly_var_samp f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2150 n 0 int2_accum numeric_poly_var_samp numeric_poly_combine numeric_poly_serialize numeric_poly_deserialize int2_accum int2_accum_inv numeric_poly_var_samp f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2151 n 0 float4_accum float8_var_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2152 n 0 float8_accum float8_var_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2153 n 0 numeric_accum numeric_var_samp numeric_combine numeric_serialize numeric_deserialize numeric_accum numeric_accum_inv numeric_var_samp f f 0 2281 17 128 2281 128 _null_ _null_ ));
/* stddev_pop */
-DATA(insert ( 2724 n 0 int8_accum numeric_stddev_pop - - - int8_accum int8_accum_inv numeric_stddev_pop f f 0 2281 0 128 2281 128 _null_ _null_ ));
-DATA(insert ( 2725 n 0 int4_accum numeric_poly_stddev_pop - - - int4_accum int4_accum_inv numeric_poly_stddev_pop f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2726 n 0 int2_accum numeric_poly_stddev_pop - - - int2_accum int2_accum_inv numeric_poly_stddev_pop f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2727 n 0 float4_accum float8_stddev_pop - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2728 n 0 float8_accum float8_stddev_pop - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2729 n 0 numeric_accum numeric_stddev_pop - - - numeric_accum numeric_accum_inv numeric_stddev_pop f f 0 2281 0 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2724 n 0 int8_accum numeric_stddev_pop numeric_combine numeric_serialize numeric_deserialize int8_accum int8_accum_inv numeric_stddev_pop f f 0 2281 17 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2725 n 0 int4_accum numeric_poly_stddev_pop numeric_poly_combine numeric_poly_serialize numeric_poly_deserialize int4_accum int4_accum_inv numeric_poly_stddev_pop f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2726 n 0 int2_accum numeric_poly_stddev_pop numeric_poly_combine numeric_poly_serialize numeric_poly_deserialize int2_accum int2_accum_inv numeric_poly_stddev_pop f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2727 n 0 float4_accum float8_stddev_pop - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2728 n 0 float8_accum float8_stddev_pop - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2729 n 0 numeric_accum numeric_stddev_pop numeric_combine numeric_serialize numeric_deserialize numeric_accum numeric_accum_inv numeric_stddev_pop f f 0 2281 17 128 2281 128 _null_ _null_ ));
/* stddev_samp */
-DATA(insert ( 2712 n 0 int8_accum numeric_stddev_samp - - - int8_accum int8_accum_inv numeric_stddev_samp f f 0 2281 0 128 2281 128 _null_ _null_ ));
-DATA(insert ( 2713 n 0 int4_accum numeric_poly_stddev_samp - - - int4_accum int4_accum_inv numeric_poly_stddev_samp f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2714 n 0 int2_accum numeric_poly_stddev_samp - - - int2_accum int2_accum_inv numeric_poly_stddev_samp f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2715 n 0 float4_accum float8_stddev_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2716 n 0 float8_accum float8_stddev_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2717 n 0 numeric_accum numeric_stddev_samp - - - numeric_accum numeric_accum_inv numeric_stddev_samp f f 0 2281 0 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2712 n 0 int8_accum numeric_stddev_samp numeric_combine numeric_serialize numeric_deserialize int8_accum int8_accum_inv numeric_stddev_samp f f 0 2281 17 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2713 n 0 int4_accum numeric_poly_stddev_samp numeric_poly_combine numeric_poly_serialize numeric_poly_deserialize int4_accum int4_accum_inv numeric_poly_stddev_samp f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2714 n 0 int2_accum numeric_poly_stddev_samp numeric_poly_combine numeric_poly_serialize numeric_poly_deserialize int2_accum int2_accum_inv numeric_poly_stddev_samp f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2715 n 0 float4_accum float8_stddev_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2716 n 0 float8_accum float8_stddev_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2717 n 0 numeric_accum numeric_stddev_samp numeric_combine numeric_serialize numeric_deserialize numeric_accum numeric_accum_inv numeric_stddev_samp f f 0 2281 17 128 2281 128 _null_ _null_ ));
/* stddev: historical Postgres syntax for stddev_samp */
-DATA(insert ( 2154 n 0 int8_accum numeric_stddev_samp - - - int8_accum int8_accum_inv numeric_stddev_samp f f 0 2281 0 128 2281 128 _null_ _null_ ));
-DATA(insert ( 2155 n 0 int4_accum numeric_poly_stddev_samp - - - int4_accum int4_accum_inv numeric_poly_stddev_samp f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2156 n 0 int2_accum numeric_poly_stddev_samp - - - int2_accum int2_accum_inv numeric_poly_stddev_samp f f 0 2281 0 48 2281 48 _null_ _null_ ));
-DATA(insert ( 2157 n 0 float4_accum float8_stddev_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2158 n 0 float8_accum float8_stddev_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
-DATA(insert ( 2159 n 0 numeric_accum numeric_stddev_samp - - - numeric_accum numeric_accum_inv numeric_stddev_samp f f 0 2281 0 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2154 n 0 int8_accum numeric_stddev_samp numeric_combine numeric_serialize numeric_deserialize int8_accum int8_accum_inv numeric_stddev_samp f f 0 2281 17 128 2281 128 _null_ _null_ ));
+DATA(insert ( 2155 n 0 int4_accum numeric_poly_stddev_samp numeric_poly_combine numeric_poly_serialize numeric_poly_deserialize int4_accum int4_accum_inv numeric_poly_stddev_samp f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2156 n 0 int2_accum numeric_poly_stddev_samp numeric_poly_combine numeric_poly_serialize numeric_poly_deserialize int2_accum int2_accum_inv numeric_poly_stddev_samp f f 0 2281 17 48 2281 48 _null_ _null_ ));
+DATA(insert ( 2157 n 0 float4_accum float8_stddev_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2158 n 0 float8_accum float8_stddev_samp - - - - - - f f 0 1022 0 0 0 0 "{0,0,0}" _null_ ));
+DATA(insert ( 2159 n 0 numeric_accum numeric_stddev_samp numeric_combine numeric_serialize numeric_deserialize numeric_accum numeric_accum_inv numeric_stddev_samp f f 0 2281 17 128 2281 128 _null_ _null_ ));
/* SQL2003 binary regression aggregates */
DATA(insert ( 2818 n 0 int8inc_float8_float8 - - - - - - - f f 0 20 0 0 0 0 "0" _null_ ));
@@ -269,9 +269,9 @@ DATA(insert ( 2828 n 0 float8_regr_accum float8_covar_samp - - - - - -
DATA(insert ( 2829 n 0 float8_regr_accum float8_corr - - - - - - f f 0 1022 0 0 0 0 "{0,0,0,0,0,0}" _null_ ));
/* boolean-and and boolean-or */
-DATA(insert ( 2517 n 0 booland_statefunc - - - - bool_accum bool_accum_inv bool_alltrue f f 58 16 0 0 2281 16 _null_ _null_ ));
-DATA(insert ( 2518 n 0 boolor_statefunc - - - - bool_accum bool_accum_inv bool_anytrue f f 59 16 0 0 2281 16 _null_ _null_ ));
-DATA(insert ( 2519 n 0 booland_statefunc - - - - bool_accum bool_accum_inv bool_alltrue f f 58 16 0 0 2281 16 _null_ _null_ ));
+DATA(insert ( 2517 n 0 booland_statefunc - booland_statefunc - - bool_accum bool_accum_inv bool_alltrue f f 58 16 0 0 2281 16 _null_ _null_ ));
+DATA(insert ( 2518 n 0 boolor_statefunc - boolor_statefunc - - bool_accum bool_accum_inv bool_anytrue f f 59 16 0 0 2281 16 _null_ _null_ ));
+DATA(insert ( 2519 n 0 booland_statefunc - booland_statefunc - - bool_accum bool_accum_inv bool_alltrue f f 58 16 0 0 2281 16 _null_ _null_ ));
/* bitwise integer */
DATA(insert ( 2236 n 0 int2and - int2and - - - - - f f 0 21 0 0 0 0 _null_ _null_ ));
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index c0d88ab2933..a0edf939b3a 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -2441,8 +2441,20 @@ DATA(insert OID = 1832 ( float8_stddev_samp PGNSP PGUID 12 1 0 0 0 f f f f t f
DESCR("aggregate final function");
DATA(insert OID = 1833 ( numeric_accum PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 1700" _null_ _null_ _null_ _null_ _null_ numeric_accum _null_ _null_ _null_ ));
DESCR("aggregate transition function");
+DATA(insert OID = 3341 ( numeric_combine PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 2281" _null_ _null_ _null_ _null_ _null_ numeric_combine _null_ _null_ _null_ ));
+DESCR("aggregate combine function");
DATA(insert OID = 2858 ( numeric_avg_accum PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 1700" _null_ _null_ _null_ _null_ _null_ numeric_avg_accum _null_ _null_ _null_ ));
DESCR("aggregate transition function");
+DATA(insert OID = 3337 ( numeric_avg_combine PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 2281" _null_ _null_ _null_ _null_ _null_ numeric_avg_combine _null_ _null_ _null_ ));
+DESCR("aggregate combine function");
+DATA(insert OID = 2740 ( numeric_avg_serialize PGNSP PGUID 12 1 0 0 0 f f f f t f i s 1 0 17 "2281" _null_ _null_ _null_ _null_ _null_ numeric_avg_serialize _null_ _null_ _null_ ));
+DESCR("aggregate serial function");
+DATA(insert OID = 2741 ( numeric_avg_deserialize PGNSP PGUID 12 1 0 0 0 f f f f t f i s 1 0 2281 "17" _null_ _null_ _null_ _null_ _null_ numeric_avg_deserialize _null_ _null_ _null_ ));
+DESCR("aggregate deserial function");
+DATA(insert OID = 3335 ( numeric_serialize PGNSP PGUID 12 1 0 0 0 f f f f t f i s 1 0 17 "2281" _null_ _null_ _null_ _null_ _null_ numeric_serialize _null_ _null_ _null_ ));
+DESCR("aggregate serial function");
+DATA(insert OID = 3336 ( numeric_deserialize PGNSP PGUID 12 1 0 0 0 f f f f t f i s 1 0 2281 "17" _null_ _null_ _null_ _null_ _null_ numeric_deserialize _null_ _null_ _null_ ));
+DESCR("aggregate deserial function");
DATA(insert OID = 3548 ( numeric_accum_inv PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 1700" _null_ _null_ _null_ _null_ _null_ numeric_accum_inv _null_ _null_ _null_ ));
DESCR("aggregate transition function");
DATA(insert OID = 1834 ( int2_accum PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 21" _null_ _null_ _null_ _null_ _null_ int2_accum _null_ _null_ _null_ ));
@@ -2451,6 +2463,12 @@ DATA(insert OID = 1835 ( int4_accum PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2
DESCR("aggregate transition function");
DATA(insert OID = 1836 ( int8_accum PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 20" _null_ _null_ _null_ _null_ _null_ int8_accum _null_ _null_ _null_ ));
DESCR("aggregate transition function");
+DATA(insert OID = 3338 ( numeric_poly_combine PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 2281" _null_ _null_ _null_ _null_ _null_ numeric_poly_combine _null_ _null_ _null_ ));
+DESCR("aggregate combine function");
+DATA(insert OID = 3339 ( numeric_poly_serialize PGNSP PGUID 12 1 0 0 0 f f f f t f i s 1 0 17 "2281" _null_ _null_ _null_ _null_ _null_ numeric_poly_serialize _null_ _null_ _null_ ));
+DESCR("aggregate serial function");
+DATA(insert OID = 3340 ( numeric_poly_deserialize PGNSP PGUID 12 1 0 0 0 f f f f t f i s 1 0 2281 "17" _null_ _null_ _null_ _null_ _null_ numeric_poly_deserialize _null_ _null_ _null_ ));
+DESCR("aggregate deserial function");
DATA(insert OID = 2746 ( int8_avg_accum PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 20" _null_ _null_ _null_ _null_ _null_ int8_avg_accum _null_ _null_ _null_ ));
DESCR("aggregate transition function");
DATA(insert OID = 3567 ( int2_accum_inv PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 21" _null_ _null_ _null_ _null_ _null_ int2_accum_inv _null_ _null_ _null_ ));
@@ -2461,6 +2479,14 @@ DATA(insert OID = 3569 ( int8_accum_inv PGNSP PGUID 12 1 0 0 0 f f f f f f i
DESCR("aggregate transition function");
DATA(insert OID = 3387 ( int8_avg_accum_inv PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 20" _null_ _null_ _null_ _null_ _null_ int8_avg_accum_inv _null_ _null_ _null_ ));
DESCR("aggregate transition function");
+DATA(insert OID = 2785 ( int8_avg_combine PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 2281" _null_ _null_ _null_ _null_ _null_ int8_avg_combine _null_ _null_ _null_ ));
+DESCR("aggregate combine function");
+DATA(insert OID = 2786 ( int8_avg_serialize PGNSP PGUID 12 1 0 0 0 f f f f t f i s 1 0 17 "2281" _null_ _null_ _null_ _null_ _null_ int8_avg_serialize _null_ _null_ _null_ ));
+DESCR("aggregate serial function");
+DATA(insert OID = 2787 ( int8_avg_deserialize PGNSP PGUID 12 1 0 0 0 f f f f t f i s 1 0 2281 "17" _null_ _null_ _null_ _null_ _null_ int8_avg_deserialize _null_ _null_ _null_ ));
+DESCR("aggregate deserial function");
+DATA(insert OID = 3324 ( int4_avg_combine PGNSP PGUID 12 1 0 0 0 f f f f f f i s 2 0 2281 "2281 2281" _null_ _null_ _null_ _null_ _null_ int4_avg_combine _null_ _null_ _null_ ));
+DESCR("aggregate combine function");
DATA(insert OID = 3178 ( numeric_sum PGNSP PGUID 12 1 0 0 0 f f f f f f i s 1 0 1700 "2281" _null_ _null_ _null_ _null_ _null_ numeric_sum _null_ _null_ _null_ ));
DESCR("aggregate final function");
DATA(insert OID = 1837 ( numeric_avg PGNSP PGUID 12 1 0 0 0 f f f f f f i s 1 0 1700 "2281" _null_ _null_ _null_ _null_ _null_ numeric_avg _null_ _null_ _null_ ));
@@ -2494,6 +2520,8 @@ DESCR("aggregate final function");
DATA(insert OID = 1843 ( interval_accum PGNSP PGUID 12 1 0 0 0 f f f f t f i s 2 0 1187 "1187 1186" _null_ _null_ _null_ _null_ _null_ interval_accum _null_ _null_ _null_ ));
DESCR("aggregate transition function");
+DATA(insert OID = 3325 ( interval_combine PGNSP PGUID 12 1 0 0 0 f f f f t f i s 2 0 1187 "1187 1187" _null_ _null_ _null_ _null_ _null_ interval_combine _null_ _null_ _null_ ));
+DESCR("aggregate combine function");
DATA(insert OID = 3549 ( interval_accum_inv PGNSP PGUID 12 1 0 0 0 f f f f t f i s 2 0 1187 "1187 1186" _null_ _null_ _null_ _null_ _null_ interval_accum_inv _null_ _null_ _null_ ));
DESCR("aggregate transition function");
DATA(insert OID = 1844 ( interval_avg PGNSP PGUID 12 1 0 0 0 f f f f t f i s 1 0 1186 "1187" _null_ _null_ _null_ _null_ _null_ interval_avg _null_ _null_ _null_ ));
diff --git a/src/include/utils/builtins.h b/src/include/utils/builtins.h
index d5c4b01ada4..41063720e5c 100644
--- a/src/include/utils/builtins.h
+++ b/src/include/utils/builtins.h
@@ -1064,15 +1064,27 @@ extern Datum numeric_float8_no_overflow(PG_FUNCTION_ARGS);
extern Datum float4_numeric(PG_FUNCTION_ARGS);
extern Datum numeric_float4(PG_FUNCTION_ARGS);
extern Datum numeric_accum(PG_FUNCTION_ARGS);
+extern Datum numeric_combine(PG_FUNCTION_ARGS);
extern Datum numeric_avg_accum(PG_FUNCTION_ARGS);
+extern Datum numeric_avg_combine(PG_FUNCTION_ARGS);
+extern Datum numeric_avg_serialize(PG_FUNCTION_ARGS);
+extern Datum numeric_avg_deserialize(PG_FUNCTION_ARGS);
+extern Datum numeric_serialize(PG_FUNCTION_ARGS);
+extern Datum numeric_deserialize(PG_FUNCTION_ARGS);
extern Datum numeric_accum_inv(PG_FUNCTION_ARGS);
extern Datum int2_accum(PG_FUNCTION_ARGS);
extern Datum int4_accum(PG_FUNCTION_ARGS);
extern Datum int8_accum(PG_FUNCTION_ARGS);
+extern Datum numeric_poly_combine(PG_FUNCTION_ARGS);
+extern Datum numeric_poly_serialize(PG_FUNCTION_ARGS);
+extern Datum numeric_poly_deserialize(PG_FUNCTION_ARGS);
extern Datum int2_accum_inv(PG_FUNCTION_ARGS);
extern Datum int4_accum_inv(PG_FUNCTION_ARGS);
extern Datum int8_accum_inv(PG_FUNCTION_ARGS);
extern Datum int8_avg_accum(PG_FUNCTION_ARGS);
+extern Datum int8_avg_combine(PG_FUNCTION_ARGS);
+extern Datum int8_avg_serialize(PG_FUNCTION_ARGS);
+extern Datum int8_avg_deserialize(PG_FUNCTION_ARGS);
extern Datum numeric_avg(PG_FUNCTION_ARGS);
extern Datum numeric_sum(PG_FUNCTION_ARGS);
extern Datum numeric_var_pop(PG_FUNCTION_ARGS);
@@ -1090,6 +1102,7 @@ extern Datum int4_sum(PG_FUNCTION_ARGS);
extern Datum int8_sum(PG_FUNCTION_ARGS);
extern Datum int2_avg_accum(PG_FUNCTION_ARGS);
extern Datum int4_avg_accum(PG_FUNCTION_ARGS);
+extern Datum int4_avg_combine(PG_FUNCTION_ARGS);
extern Datum int2_avg_accum_inv(PG_FUNCTION_ARGS);
extern Datum int4_avg_accum_inv(PG_FUNCTION_ARGS);
extern Datum int8_avg_accum_inv(PG_FUNCTION_ARGS);
diff --git a/src/include/utils/timestamp.h b/src/include/utils/timestamp.h
index 865e993e8ab..85cc7ce1fe8 100644
--- a/src/include/utils/timestamp.h
+++ b/src/include/utils/timestamp.h
@@ -187,6 +187,7 @@ extern Datum interval_mul(PG_FUNCTION_ARGS);
extern Datum mul_d_interval(PG_FUNCTION_ARGS);
extern Datum interval_div(PG_FUNCTION_ARGS);
extern Datum interval_accum(PG_FUNCTION_ARGS);
+extern Datum interval_combine(PG_FUNCTION_ARGS);
extern Datum interval_accum_inv(PG_FUNCTION_ARGS);
extern Datum interval_avg(PG_FUNCTION_ARGS);
diff --git a/src/test/regress/expected/create_aggregate.out b/src/test/regress/expected/create_aggregate.out
index 66e073d2b7b..dac26982bca 100644
--- a/src/test/regress/expected/create_aggregate.out
+++ b/src/test/regress/expected/create_aggregate.out
@@ -101,24 +101,93 @@ CREATE AGGREGATE sumdouble (float8)
msfunc = float8pl,
minvfunc = float8mi
);
--- Test aggregate combine function
+-- aggregate combine and serialization functions
+-- Ensure stype and serialtype can't be the same
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = internal
+);
+ERROR: aggregate serialization type cannot be "internal"
+-- if serialtype is specified we need a serialfunc and deserialfunc
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = bytea
+);
+ERROR: aggregate serialization function must be specified when serialization type is specified
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = bytea,
+ serialfunc = numeric_avg_serialize
+);
+ERROR: aggregate deserialization function must be specified when serialization type is specified
+-- serialfunc must have correct parameters
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = bytea,
+ serialfunc = numeric_avg_deserialize,
+ deserialfunc = numeric_avg_deserialize
+);
+ERROR: function numeric_avg_deserialize(internal) does not exist
+-- deserialfunc must have correct parameters
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = bytea,
+ serialfunc = numeric_avg_serialize,
+ deserialfunc = numeric_avg_serialize
+);
+ERROR: function numeric_avg_serialize(bytea) does not exist
+-- ensure return type of serialfunc is checked
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = text,
+ serialfunc = numeric_avg_serialize,
+ deserialfunc = numeric_avg_deserialize
+);
+ERROR: return type of serialization function numeric_avg_serialize is not text
+-- ensure combine function parameters are checked
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = bytea,
+ serialfunc = numeric_avg_serialize,
+ deserialfunc = numeric_avg_deserialize,
+ combinefunc = int4larger
+);
+ERROR: function int4larger(internal, internal) does not exist
-- ensure create aggregate works.
-CREATE AGGREGATE mysum (int)
+CREATE AGGREGATE myavg (numeric)
(
- stype = int,
- sfunc = int4pl,
- combinefunc = int4pl
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ finalfunc = numeric_avg,
+ serialtype = bytea,
+ serialfunc = numeric_avg_serialize,
+ deserialfunc = numeric_avg_deserialize,
+ combinefunc = numeric_avg_combine
);
-- Ensure all these functions made it into the catalog
-SELECT aggfnoid,aggtransfn,aggcombinefn,aggtranstype
+SELECT aggfnoid,aggtransfn,aggcombinefn,aggtranstype,aggserialfn,aggdeserialfn,aggserialtype
FROM pg_aggregate
-WHERE aggfnoid = 'mysum'::REGPROC;
- aggfnoid | aggtransfn | aggcombinefn | aggtranstype
-----------+------------+--------------+--------------
- mysum | int4pl | int4pl | 23
+WHERE aggfnoid = 'myavg'::REGPROC;
+ aggfnoid | aggtransfn | aggcombinefn | aggtranstype | aggserialfn | aggdeserialfn | aggserialtype
+----------+-------------------+---------------------+--------------+-----------------------+-------------------------+---------------
+ myavg | numeric_avg_accum | numeric_avg_combine | 2281 | numeric_avg_serialize | numeric_avg_deserialize | 17
(1 row)
-DROP AGGREGATE mysum (int);
+DROP AGGREGATE myavg (numeric);
-- invalid: nonstrict inverse with strict forward function
CREATE FUNCTION float8mi_n(float8, float8) RETURNS float8 AS
$$ SELECT $1 - $2; $$
diff --git a/src/test/regress/expected/opr_sanity.out b/src/test/regress/expected/opr_sanity.out
index a8dc0c150f5..b1596401306 100644
--- a/src/test/regress/expected/opr_sanity.out
+++ b/src/test/regress/expected/opr_sanity.out
@@ -279,15 +279,21 @@ ORDER BY 1, 2;
-- Look for functions that return type "internal" and do not have any
-- "internal" argument. Such a function would be a security hole since
-- it might be used to call an internal function from an SQL command.
--- As of 7.3 this query should find only internal_in.
+-- As of 7.3 this query should find internal_in, and as of 9.6 aggregate
+-- deserialization will be found too. These should contain a runtime check to
+-- ensure they can only be called in an aggregate context.
SELECT p1.oid, p1.proname
FROM pg_proc as p1
WHERE p1.prorettype = 'internal'::regtype AND NOT
'internal'::regtype = ANY (p1.proargtypes);
- oid | proname
-------+-------------
+ oid | proname
+------+--------------------------
+ 2741 | numeric_avg_deserialize
+ 3336 | numeric_deserialize
+ 3340 | numeric_poly_deserialize
+ 2787 | int8_avg_deserialize
2304 | internal_in
-(1 row)
+(5 rows)
-- Look for functions that return a polymorphic type and do not have any
-- polymorphic argument. Calls of such functions would be unresolvable
@@ -1528,6 +1534,89 @@ WHERE proisagg AND provariadic != 0 AND a.aggkind = 'n';
-----+---------
(0 rows)
+-- Check that all serial functions have a return type the same as the serial
+-- type.
+SELECT a.aggserialfn,a.aggserialtype,p.prorettype
+FROM pg_aggregate a
+INNER JOIN pg_proc p ON a.aggserialfn = p.oid
+WHERE a.aggserialtype <> p.prorettype;
+ aggserialfn | aggserialtype | prorettype
+-------------+---------------+------------
+(0 rows)
+
+-- Check that all the deserial functions have the same input type as the
+-- serialtype
+SELECT a.aggserialfn,a.aggserialtype,p.proargtypes[0]
+FROM pg_aggregate a
+INNER JOIN pg_proc p ON a.aggdeserialfn = p.oid
+WHERE p.proargtypes[0] <> a.aggserialtype;
+ aggserialfn | aggserialtype | proargtypes
+-------------+---------------+-------------
+(0 rows)
+
+-- An aggregate should either have a complete set of serialtype, serial func
+-- and deserial func, or none of them.
+SELECT aggserialtype,aggserialfn,aggdeserialfn
+FROM pg_aggregate
+WHERE (aggserialtype <> 0 OR aggserialfn <> 0 OR aggdeserialfn <> 0)
+ AND (aggserialtype = 0 OR aggserialfn = 0 OR aggdeserialfn = 0);
+ aggserialtype | aggserialfn | aggdeserialfn
+---------------+-------------+---------------
+(0 rows)
+
+-- Check that all aggregates with serialtypes have internal states.
+-- (There's no point in serializing anything apart from internal)
+SELECT aggfnoid,aggserialtype,aggtranstype
+FROM pg_aggregate
+WHERE aggserialtype <> 0 AND aggtranstype <> 'internal'::regtype;
+ aggfnoid | aggserialtype | aggtranstype
+----------+---------------+--------------
+(0 rows)
+
+-- Check that all serial functions are strict. It's wasteful for these to be
+-- called with NULL values.
+SELECT aggfnoid,aggserialfn
+FROM pg_aggregate a
+INNER JOIN pg_proc p ON a.aggserialfn = p.oid
+WHERE p.proisstrict = false;
+ aggfnoid | aggserialfn
+----------+-------------
+(0 rows)
+
+-- Check that all deserial functions are strict. It's wasteful for these to be
+-- called with NULL values.
+SELECT aggfnoid,aggdeserialfn
+FROM pg_aggregate a
+INNER JOIN pg_proc p ON a.aggdeserialfn = p.oid
+WHERE p.proisstrict = false;
+ aggfnoid | aggdeserialfn
+----------+---------------
+(0 rows)
+
+-- Check that no combine functions with an INTERNAL return type are strict.
+SELECT aggfnoid,aggcombinefn
+FROM pg_aggregate a
+INNER JOIN pg_proc p ON a.aggcombinefn = p.oid
+INNER JOIN pg_type t ON a.aggtranstype = t.oid
+WHERE t.typname = 'internal' AND p.proisstrict = true;
+ aggfnoid | aggcombinefn
+----------+--------------
+(0 rows)
+
+-- Check that aggregates which have the same transition function also have
+-- the same combine, serialization, and deserialization functions.
+SELECT a.aggfnoid, a.aggcombinefn, a.aggserialfn, a.aggdeserialfn,
+ b.aggfnoid, b.aggcombinefn, b.aggserialfn, b.aggdeserialfn
+FROM
+ pg_aggregate a, pg_aggregate b
+WHERE
+ a.aggfnoid < b.aggfnoid AND a.aggtransfn = b.aggtransfn AND
+ (a.aggcombinefn != b.aggcombinefn OR a.aggserialfn != b.aggserialfn
+ OR a.aggdeserialfn != b.aggdeserialfn);
+ aggfnoid | aggcombinefn | aggserialfn | aggdeserialfn | aggfnoid | aggcombinefn | aggserialfn | aggdeserialfn
+----------+--------------+-------------+---------------+----------+--------------+-------------+---------------
+(0 rows)
+
-- **************** pg_opfamily ****************
-- Look for illegal values in pg_opfamily fields
SELECT p1.oid
diff --git a/src/test/regress/sql/create_aggregate.sql b/src/test/regress/sql/create_aggregate.sql
index dfcbc5a0e8b..a7da31e5943 100644
--- a/src/test/regress/sql/create_aggregate.sql
+++ b/src/test/regress/sql/create_aggregate.sql
@@ -115,22 +115,91 @@ CREATE AGGREGATE sumdouble (float8)
minvfunc = float8mi
);
--- Test aggregate combine function
+-- aggregate combine and serialization functions
+
+-- Ensure stype and serialtype can't be the same
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = internal
+);
+
+-- if serialtype is specified we need a serialfunc and deserialfunc
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = bytea
+);
+
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = bytea,
+ serialfunc = numeric_avg_serialize
+);
+
+-- serialfunc must have correct parameters
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = bytea,
+ serialfunc = numeric_avg_deserialize,
+ deserialfunc = numeric_avg_deserialize
+);
+
+-- deserialfunc must have correct parameters
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = bytea,
+ serialfunc = numeric_avg_serialize,
+ deserialfunc = numeric_avg_serialize
+);
+
+-- ensure return type of serialfunc is checked
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = text,
+ serialfunc = numeric_avg_serialize,
+ deserialfunc = numeric_avg_deserialize
+);
+
+-- ensure combine function parameters are checked
+CREATE AGGREGATE myavg (numeric)
+(
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ serialtype = bytea,
+ serialfunc = numeric_avg_serialize,
+ deserialfunc = numeric_avg_deserialize,
+ combinefunc = int4larger
+);
-- ensure create aggregate works.
-CREATE AGGREGATE mysum (int)
+CREATE AGGREGATE myavg (numeric)
(
- stype = int,
- sfunc = int4pl,
- combinefunc = int4pl
+ stype = internal,
+ sfunc = numeric_avg_accum,
+ finalfunc = numeric_avg,
+ serialtype = bytea,
+ serialfunc = numeric_avg_serialize,
+ deserialfunc = numeric_avg_deserialize,
+ combinefunc = numeric_avg_combine
);
-- Ensure all these functions made it into the catalog
-SELECT aggfnoid,aggtransfn,aggcombinefn,aggtranstype
+SELECT aggfnoid,aggtransfn,aggcombinefn,aggtranstype,aggserialfn,aggdeserialfn,aggserialtype
FROM pg_aggregate
-WHERE aggfnoid = 'mysum'::REGPROC;
+WHERE aggfnoid = 'myavg'::REGPROC;
-DROP AGGREGATE mysum (int);
+DROP AGGREGATE myavg (numeric);
-- invalid: nonstrict inverse with strict forward function
diff --git a/src/test/regress/sql/opr_sanity.sql b/src/test/regress/sql/opr_sanity.sql
index 6c9784a3872..62c84d035d5 100644
--- a/src/test/regress/sql/opr_sanity.sql
+++ b/src/test/regress/sql/opr_sanity.sql
@@ -228,7 +228,9 @@ ORDER BY 1, 2;
-- Look for functions that return type "internal" and do not have any
-- "internal" argument. Such a function would be a security hole since
-- it might be used to call an internal function from an SQL command.
--- As of 7.3 this query should find only internal_in.
+-- As of 7.3 this query should find internal_in, and as of 9.6 aggregate
+-- deserialization will be found too. These should contain a runtime check to
+-- ensure they can only be called in an aggregate context.
SELECT p1.oid, p1.proname
FROM pg_proc as p1
@@ -1002,6 +1004,64 @@ SELECT p.oid, proname
FROM pg_proc AS p JOIN pg_aggregate AS a ON a.aggfnoid = p.oid
WHERE proisagg AND provariadic != 0 AND a.aggkind = 'n';
+-- Check that all serial functions have a return type the same as the serial
+-- type.
+SELECT a.aggserialfn,a.aggserialtype,p.prorettype
+FROM pg_aggregate a
+INNER JOIN pg_proc p ON a.aggserialfn = p.oid
+WHERE a.aggserialtype <> p.prorettype;
+
+-- Check that all the deserial functions have the same input type as the
+-- serialtype
+SELECT a.aggserialfn,a.aggserialtype,p.proargtypes[0]
+FROM pg_aggregate a
+INNER JOIN pg_proc p ON a.aggdeserialfn = p.oid
+WHERE p.proargtypes[0] <> a.aggserialtype;
+
+-- An aggregate should either have a complete set of serialtype, serial func
+-- and deserial func, or none of them.
+SELECT aggserialtype,aggserialfn,aggdeserialfn
+FROM pg_aggregate
+WHERE (aggserialtype <> 0 OR aggserialfn <> 0 OR aggdeserialfn <> 0)
+ AND (aggserialtype = 0 OR aggserialfn = 0 OR aggdeserialfn = 0);
+
+-- Check that all aggregates with serialtypes have internal states.
+-- (There's no point in serializing anything apart from internal)
+SELECT aggfnoid,aggserialtype,aggtranstype
+FROM pg_aggregate
+WHERE aggserialtype <> 0 AND aggtranstype <> 'internal'::regtype;
+
+-- Check that all serial functions are strict. It's wasteful for these to be
+-- called with NULL values.
+SELECT aggfnoid,aggserialfn
+FROM pg_aggregate a
+INNER JOIN pg_proc p ON a.aggserialfn = p.oid
+WHERE p.proisstrict = false;
+
+-- Check that all deserial functions are strict. It's wasteful for these to be
+-- called with NULL values.
+SELECT aggfnoid,aggdeserialfn
+FROM pg_aggregate a
+INNER JOIN pg_proc p ON a.aggdeserialfn = p.oid
+WHERE p.proisstrict = false;
+
+-- Check that no combine functions with an INTERNAL return type are strict.
+SELECT aggfnoid,aggcombinefn
+FROM pg_aggregate a
+INNER JOIN pg_proc p ON a.aggcombinefn = p.oid
+INNER JOIN pg_type t ON a.aggtranstype = t.oid
+WHERE t.typname = 'internal' AND p.proisstrict = true;
+
+-- Check that aggregates which have the same transition function also have
+-- the same combine, serialization, and deserialization functions.
+SELECT a.aggfnoid, a.aggcombinefn, a.aggserialfn, a.aggdeserialfn,
+ b.aggfnoid, b.aggcombinefn, b.aggserialfn, b.aggdeserialfn
+FROM
+ pg_aggregate a, pg_aggregate b
+WHERE
+ a.aggfnoid < b.aggfnoid AND a.aggtransfn = b.aggtransfn AND
+ (a.aggcombinefn != b.aggcombinefn OR a.aggserialfn != b.aggserialfn
+ OR a.aggdeserialfn != b.aggdeserialfn);
-- **************** pg_opfamily ****************
--
GitLab