*/
Aggref *aggref;
+ /*
+ * Is this state value actually being shared by more than one Aggref?
+ */
+ bool aggshared;
+
/*
* Nominal number of arguments for aggregate function. For plain aggs,
* this excludes any ORDER BY expressions. For ordered-set aggs, this
{
/*
* Existing compatible trans found, so just point the 'peragg' to
- * the same per-trans struct.
+ * the same per-trans struct, and mark the trans state as shared.
*/
pertrans = &pertransstates[existing_transno];
+ pertrans->aggshared = true;
peragg->transno = existing_transno;
}
else
/* Begin filling in the pertrans data */
pertrans->aggref = aggref;
+ pertrans->aggshared = false;
pertrans->aggCollation = aggref->inputcollid;
pertrans->transfn_oid = aggtransfn;
pertrans->serialfn_oid = aggserialfn;
{
if (fcinfo->context && IsA(fcinfo->context, AggState))
{
+ AggState *aggstate = (AggState *) fcinfo->context;
AggStatePerAgg curperagg;
AggStatePerTrans curpertrans;
/* check curperagg (valid when in a final function) */
- curperagg = ((AggState *) fcinfo->context)->curperagg;
+ curperagg = aggstate->curperagg;
if (curperagg)
return curperagg->aggref;
/* check curpertrans (valid when in a transition function) */
- curpertrans = ((AggState *) fcinfo->context)->curpertrans;
+ curpertrans = aggstate->curpertrans;
if (curpertrans)
return curpertrans->aggref;
return NULL;
}
+/*
+ * AggStateIsShared - find out whether transition state is shared
+ *
+ * If the function is being called as an aggregate support function,
+ * return TRUE if the aggregate's transition state is shared across
+ * multiple aggregates, FALSE if it is not.
+ *
+ * Returns TRUE if not called as an aggregate support function.
+ * This is intended as a conservative answer, ie "no you'd better not
+ * scribble on your input". In particular, will return TRUE if the
+ * aggregate is being used as a window function, which is a scenario
+ * in which changing the transition state is a bad idea. We might
+ * want to refine the behavior for the window case in future.
+ */
+bool
+AggStateIsShared(FunctionCallInfo fcinfo)
+{
+ if (fcinfo->context && IsA(fcinfo->context, AggState))
+ {
+ AggState *aggstate = (AggState *) fcinfo->context;
+ AggStatePerAgg curperagg;
+ AggStatePerTrans curpertrans;
+
+ /* check curperagg (valid when in a final function) */
+ curperagg = aggstate->curperagg;
+
+ if (curperagg)
+ return aggstate->pertrans[curperagg->transno].aggshared;
+
+ /* check curpertrans (valid when in a transition function) */
+ curpertrans = aggstate->curpertrans;
+
+ if (curpertrans)
+ return curpertrans->aggshared;
+ }
+ return true;
+}
+
/*
* AggRegisterCallback - register a cleanup callback for an aggregate
*
* create just once per query because they will not change across groups.
* The per-query struct and subsidiary data live in the executor's per-query
* memory context, and go away implicitly at ExecutorEnd().
+ *
+ * These structs are set up during the first call of the transition function.
+ * Because we allow nodeAgg.c to merge ordered-set aggregates (but not
+ * hypothetical aggregates) with identical inputs and transition functions,
+ * this info must not depend on the particular aggregate (ie, particular
+ * final-function), nor on the direct argument(s) of the aggregate.
*/
typedef struct OSAPerQueryState
{
- /* Aggref for this aggregate: */
+ /* Representative Aggref for this aggregate: */
Aggref *aggref;
/* Memory context containing this struct and other per-query data: */
MemoryContext qcontext;
+ /* Do we expect multiple final-function calls within one group? */
+ bool rescan_needed;
/* These fields are used only when accumulating tuples: */
Tuplesortstate *sortstate;
/* Number of normal rows inserted into sortstate: */
int64 number_of_rows;
+ /* Have we already done tuplesort_performsort? */
+ bool sort_done;
} OSAPerGroupState;
static void ordered_set_shutdown(Datum arg);
qstate->aggref = aggref;
qstate->qcontext = qcontext;
+ /* We need to support rescans if the trans state is shared */
+ qstate->rescan_needed = AggStateIsShared(fcinfo);
+
/* Extract the sort information */
sortlist = aggref->aggorder;
numSortCols = list_length(sortlist);
qstate->sortOperators,
qstate->sortCollations,
qstate->sortNullsFirsts,
- work_mem, false);
+ work_mem,
+ qstate->rescan_needed);
else
osastate->sortstate = tuplesort_begin_datum(qstate->sortColType,
qstate->sortOperator,
qstate->sortCollation,
qstate->sortNullsFirst,
- work_mem, false);
+ work_mem,
+ qstate->rescan_needed);
osastate->number_of_rows = 0;
+ osastate->sort_done = false;
/* Now register a shutdown callback to clean things up at end of group */
AggRegisterCallback(fcinfo,
* group) by ExecutorEnd. But we must take care to release any potential
* non-memory resources.
*
- * This callback is arguably unnecessary, since we don't support use of
- * ordered-set aggs in AGG_HASHED mode and there is currently no non-error
- * code path in non-hashed modes wherein nodeAgg.c won't call the finalfn
- * after calling the transfn one or more times. So in principle we could rely
- * on the finalfn to delete the tuplestore etc. However, it's possible that
- * such a code path might exist in future, and in any case it'd be
- * notationally tedious and sometimes require extra data copying to ensure
- * we always delete the tuplestore in the finalfn.
+ * In the case where we're not expecting multiple finalfn calls, we could
+ * arguably rely on the finalfn to clean up; but it's easier and more testable
+ * if we just do it the same way in either case. Note that many of the
+ * finalfns could *not* free the tuplesort object, at least not without extra
+ * data copying, because what they return is a pointer to a datum inside the
+ * tuplesort object.
*/
static void
ordered_set_shutdown(Datum arg)
if (osastate->number_of_rows == 0)
PG_RETURN_NULL();
- /* Finish the sort */
- tuplesort_performsort(osastate->sortstate);
+ /* Finish the sort, or rescan if we already did */
+ if (!osastate->sort_done)
+ {
+ tuplesort_performsort(osastate->sortstate);
+ osastate->sort_done = true;
+ }
+ else
+ tuplesort_rescan(osastate->sortstate);
/*----------
* We need the smallest K such that (K/N) >= percentile.
if (!tuplesort_getdatum(osastate->sortstate, true, &val, &isnull, NULL))
elog(ERROR, "missing row in percentile_disc");
- /*
- * Note: we *cannot* clean up the tuplesort object here, because the value
- * to be returned is allocated inside its sortcontext. We could use
- * datumCopy to copy it out of there, but it doesn't seem worth the
- * trouble, since the cleanup callback will clear the tuplesort later.
- */
-
/* We shouldn't have stored any nulls, but do the right thing anyway */
if (isnull)
PG_RETURN_NULL();
Assert(expect_type == osastate->qstate->sortColType);
- /* Finish the sort */
- tuplesort_performsort(osastate->sortstate);
+ /* Finish the sort, or rescan if we already did */
+ if (!osastate->sort_done)
+ {
+ tuplesort_performsort(osastate->sortstate);
+ osastate->sort_done = true;
+ }
+ else
+ tuplesort_rescan(osastate->sortstate);
first_row = floor(percentile * (osastate->number_of_rows - 1));
second_row = ceil(percentile * (osastate->number_of_rows - 1));
val = lerpfunc(first_val, second_val, proportion);
}
- /*
- * Note: we *cannot* clean up the tuplesort object here, because the value
- * to be returned may be allocated inside its sortcontext. We could use
- * datumCopy to copy it out of there, but it doesn't seem worth the
- * trouble, since the cleanup callback will clear the tuplesort later.
- */
-
PG_RETURN_DATUM(val);
}
*/
if (i < num_percentiles)
{
- /* Finish the sort */
- tuplesort_performsort(osastate->sortstate);
+ /* Finish the sort, or rescan if we already did */
+ if (!osastate->sort_done)
+ {
+ tuplesort_performsort(osastate->sortstate);
+ osastate->sort_done = true;
+ }
+ else
+ tuplesort_rescan(osastate->sortstate);
for (; i < num_percentiles; i++)
{
}
}
- /*
- * We could clean up the tuplesort object after forming the array, but
- * probably not worth the trouble.
- */
-
/* We make the output array the same shape as the input */
PG_RETURN_POINTER(construct_md_array(result_datum, result_isnull,
ARR_NDIM(param),
*/
if (i < num_percentiles)
{
- /* Finish the sort */
- tuplesort_performsort(osastate->sortstate);
+ /* Finish the sort, or rescan if we already did */
+ if (!osastate->sort_done)
+ {
+ tuplesort_performsort(osastate->sortstate);
+ osastate->sort_done = true;
+ }
+ else
+ tuplesort_rescan(osastate->sortstate);
for (; i < num_percentiles; i++)
{
}
}
- /*
- * We could clean up the tuplesort object after forming the array, but
- * probably not worth the trouble.
- */
-
/* We make the output array the same shape as the input */
PG_RETURN_POINTER(construct_md_array(result_datum, result_isnull,
ARR_NDIM(param),
shouldfree = !(osastate->qstate->typByVal);
- /* Finish the sort */
- tuplesort_performsort(osastate->sortstate);
+ /* Finish the sort, or rescan if we already did */
+ if (!osastate->sort_done)
+ {
+ tuplesort_performsort(osastate->sortstate);
+ osastate->sort_done = true;
+ }
+ else
+ tuplesort_rescan(osastate->sortstate);
/* Scan tuples and count frequencies */
while (tuplesort_getdatum(osastate->sortstate, true, &val, &isnull, &abbrev_val))
if (shouldfree && !last_val_is_mode)
pfree(DatumGetPointer(last_val));
- /*
- * Note: we *cannot* clean up the tuplesort object here, because the value
- * to be returned is allocated inside its sortcontext. We could use
- * datumCopy to copy it out of there, but it doesn't seem worth the
- * trouble, since the cleanup callback will clear the tuplesort later.
- */
-
if (mode_freq)
PG_RETURN_DATUM(mode_val);
else
hypothetical_check_argtypes(fcinfo, nargs, osastate->qstate->tupdesc);
+ /* because we need a hypothetical row, we can't share transition state */
+ Assert(!osastate->sort_done);
+
/* insert the hypothetical row into the sort */
slot = osastate->qstate->tupslot;
ExecClearTuple(slot);
/* finish the sort */
tuplesort_performsort(osastate->sortstate);
+ osastate->sort_done = true;
/* iterate till we find the hypothetical row */
while (tuplesort_gettupleslot(osastate->sortstate, true, true, slot, NULL))
ExecClearTuple(slot);
- /* Might as well clean up the tuplesort object immediately */
- tuplesort_end(osastate->sortstate);
- osastate->sortstate = NULL;
-
return rank;
}
/* Get short-term context we can use for execTuplesMatch */
tmpcontext = AggGetTempMemoryContext(fcinfo);
+ /* because we need a hypothetical row, we can't share transition state */
+ Assert(!osastate->sort_done);
+
/* insert the hypothetical row into the sort */
slot = osastate->qstate->tupslot;
ExecClearTuple(slot);
/* finish the sort */
tuplesort_performsort(osastate->sortstate);
+ osastate->sort_done = true;
/*
* We alternate fetching into tupslot and extraslot so that we have the
ExecDropSingleTupleTableSlot(extraslot);
- /* Might as well clean up the tuplesort object immediately */
- tuplesort_end(osastate->sortstate);
- osastate->sortstate = NULL;
-
rank = rank - duplicate_count;
PG_RETURN_INT64(rank);
projects / postgresql.git / commitdiff