Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * nodeAgg.c
4 : : * Routines to handle aggregate nodes.
5 : : *
6 : : * ExecAgg normally evaluates each aggregate in the following steps:
7 : : *
8 : : * transvalue = initcond
9 : : * foreach input_tuple do
10 : : * transvalue = transfunc(transvalue, input_value(s))
11 : : * result = finalfunc(transvalue, direct_argument(s))
12 : : *
13 : : * If a finalfunc is not supplied then the result is just the ending
14 : : * value of transvalue.
15 : : *
16 : : * Other behaviors can be selected by the "aggsplit" mode, which exists
17 : : * to support partial aggregation. It is possible to:
18 : : * * Skip running the finalfunc, so that the output is always the
19 : : * final transvalue state.
20 : : * * Substitute the combinefunc for the transfunc, so that transvalue
21 : : * states (propagated up from a child partial-aggregation step) are merged
22 : : * rather than processing raw input rows. (The statements below about
23 : : * the transfunc apply equally to the combinefunc, when it's selected.)
24 : : * * Apply the serializefunc to the output values (this only makes sense
25 : : * when skipping the finalfunc, since the serializefunc works on the
26 : : * transvalue data type).
27 : : * * Apply the deserializefunc to the input values (this only makes sense
28 : : * when using the combinefunc, for similar reasons).
29 : : * It is the planner's responsibility to connect up Agg nodes using these
30 : : * alternate behaviors in a way that makes sense, with partial aggregation
31 : : * results being fed to nodes that expect them.
32 : : *
33 : : * If a normal aggregate call specifies DISTINCT or ORDER BY, we sort the
34 : : * input tuples and eliminate duplicates (if required) before performing
35 : : * the above-depicted process. (However, we don't do that for ordered-set
36 : : * aggregates; their "ORDER BY" inputs are ordinary aggregate arguments
37 : : * so far as this module is concerned.) Note that partial aggregation
38 : : * is not supported in these cases, since we couldn't ensure global
39 : : * ordering or distinctness of the inputs.
40 : : *
41 : : * If transfunc is marked "strict" in pg_proc and initcond is NULL,
42 : : * then the first non-NULL input_value is assigned directly to transvalue,
43 : : * and transfunc isn't applied until the second non-NULL input_value.
44 : : * The agg's first input type and transtype must be the same in this case!
45 : : *
46 : : * If transfunc is marked "strict" then NULL input_values are skipped,
47 : : * keeping the previous transvalue. If transfunc is not strict then it
48 : : * is called for every input tuple and must deal with NULL initcond
49 : : * or NULL input_values for itself.
50 : : *
51 : : * If finalfunc is marked "strict" then it is not called when the
52 : : * ending transvalue is NULL, instead a NULL result is created
53 : : * automatically (this is just the usual handling of strict functions,
54 : : * of course). A non-strict finalfunc can make its own choice of
55 : : * what to return for a NULL ending transvalue.
56 : : *
57 : : * Ordered-set aggregates are treated specially in one other way: we
58 : : * evaluate any "direct" arguments and pass them to the finalfunc along
59 : : * with the transition value.
60 : : *
61 : : * A finalfunc can have additional arguments beyond the transvalue and
62 : : * any "direct" arguments, corresponding to the input arguments of the
63 : : * aggregate. These are always just passed as NULL. Such arguments may be
64 : : * needed to allow resolution of a polymorphic aggregate's result type.
65 : : *
66 : : * We compute aggregate input expressions and run the transition functions
67 : : * in a temporary econtext (aggstate->tmpcontext). This is reset at least
68 : : * once per input tuple, so when the transvalue datatype is
69 : : * pass-by-reference, we have to be careful to copy it into a longer-lived
70 : : * memory context, and free the prior value to avoid memory leakage. We
71 : : * store transvalues in another set of econtexts, aggstate->aggcontexts
72 : : * (one per grouping set, see below), which are also used for the hashtable
73 : : * structures in AGG_HASHED mode. These econtexts are rescanned, not just
74 : : * reset, at group boundaries so that aggregate transition functions can
75 : : * register shutdown callbacks via AggRegisterCallback.
76 : : *
77 : : * The node's regular econtext (aggstate->ss.ps.ps_ExprContext) is used to
78 : : * run finalize functions and compute the output tuple; this context can be
79 : : * reset once per output tuple.
80 : : *
81 : : * The executor's AggState node is passed as the fmgr "context" value in
82 : : * all transfunc and finalfunc calls. It is not recommended that the
83 : : * transition functions look at the AggState node directly, but they can
84 : : * use AggCheckCallContext() to verify that they are being called by
85 : : * nodeAgg.c (and not as ordinary SQL functions). The main reason a
86 : : * transition function might want to know this is so that it can avoid
87 : : * palloc'ing a fixed-size pass-by-ref transition value on every call:
88 : : * it can instead just scribble on and return its left input. Ordinarily
89 : : * it is completely forbidden for functions to modify pass-by-ref inputs,
90 : : * but in the aggregate case we know the left input is either the initial
91 : : * transition value or a previous function result, and in either case its
92 : : * value need not be preserved. See int8inc() for an example. Notice that
93 : : * the EEOP_AGG_PLAIN_TRANS step is coded to avoid a data copy step when
94 : : * the previous transition value pointer is returned. It is also possible
95 : : * to avoid repeated data copying when the transition value is an expanded
96 : : * object: to do that, the transition function must take care to return
97 : : * an expanded object that is in a child context of the memory context
98 : : * returned by AggCheckCallContext(). Also, some transition functions want
99 : : * to store working state in addition to the nominal transition value; they
100 : : * can use the memory context returned by AggCheckCallContext() to do that.
101 : : *
102 : : * Note: AggCheckCallContext() is available as of PostgreSQL 9.0. The
103 : : * AggState is available as context in earlier releases (back to 8.1),
104 : : * but direct examination of the node is needed to use it before 9.0.
105 : : *
106 : : * As of 9.4, aggregate transition functions can also use AggGetAggref()
107 : : * to get hold of the Aggref expression node for their aggregate call.
108 : : * This is mainly intended for ordered-set aggregates, which are not
109 : : * supported as window functions. (A regular aggregate function would
110 : : * need some fallback logic to use this, since there's no Aggref node
111 : : * for a window function.)
112 : : *
113 : : * Grouping sets:
114 : : *
115 : : * A list of grouping sets which is structurally equivalent to a ROLLUP
116 : : * clause (e.g. (a,b,c), (a,b), (a)) can be processed in a single pass over
117 : : * ordered data. We do this by keeping a separate set of transition values
118 : : * for each grouping set being concurrently processed; for each input tuple
119 : : * we update them all, and on group boundaries we reset those states
120 : : * (starting at the front of the list) whose grouping values have changed
121 : : * (the list of grouping sets is ordered from most specific to least
122 : : * specific).
123 : : *
124 : : * Where more complex grouping sets are used, we break them down into
125 : : * "phases", where each phase has a different sort order (except phase 0
126 : : * which is reserved for hashing). During each phase but the last, the
127 : : * input tuples are additionally stored in a tuplesort which is keyed to the
128 : : * next phase's sort order; during each phase but the first, the input
129 : : * tuples are drawn from the previously sorted data. (The sorting of the
130 : : * data for the first phase is handled by the planner, as it might be
131 : : * satisfied by underlying nodes.)
132 : : *
133 : : * Hashing can be mixed with sorted grouping. To do this, we have an
134 : : * AGG_MIXED strategy that populates the hashtables during the first sorted
135 : : * phase, and switches to reading them out after completing all sort phases.
136 : : * We can also support AGG_HASHED with multiple hash tables and no sorting
137 : : * at all.
138 : : *
139 : : * From the perspective of aggregate transition and final functions, the
140 : : * only issue regarding grouping sets is this: a single call site (flinfo)
141 : : * of an aggregate function may be used for updating several different
142 : : * transition values in turn. So the function must not cache in the flinfo
143 : : * anything which logically belongs as part of the transition value (most
144 : : * importantly, the memory context in which the transition value exists).
145 : : * The support API functions (AggCheckCallContext, AggRegisterCallback) are
146 : : * sensitive to the grouping set for which the aggregate function is
147 : : * currently being called.
148 : : *
149 : : * Plan structure:
150 : : *
151 : : * What we get from the planner is actually one "real" Agg node which is
152 : : * part of the plan tree proper, but which optionally has an additional list
153 : : * of Agg nodes hung off the side via the "chain" field. This is because an
154 : : * Agg node happens to be a convenient representation of all the data we
155 : : * need for grouping sets.
156 : : *
157 : : * For many purposes, we treat the "real" node as if it were just the first
158 : : * node in the chain. The chain must be ordered such that hashed entries
159 : : * come before sorted/plain entries; the real node is marked AGG_MIXED if
160 : : * there are both types present (in which case the real node describes one
161 : : * of the hashed groupings, other AGG_HASHED nodes may optionally follow in
162 : : * the chain, followed in turn by AGG_SORTED or (one) AGG_PLAIN node). If
163 : : * the real node is marked AGG_HASHED or AGG_SORTED, then all the chained
164 : : * nodes must be of the same type; if it is AGG_PLAIN, there can be no
165 : : * chained nodes.
166 : : *
167 : : * We collect all hashed nodes into a single "phase", numbered 0, and create
168 : : * a sorted phase (numbered 1..n) for each AGG_SORTED or AGG_PLAIN node.
169 : : * Phase 0 is allocated even if there are no hashes, but remains unused in
170 : : * that case.
171 : : *
172 : : * AGG_HASHED nodes actually refer to only a single grouping set each,
173 : : * because for each hashed grouping we need a separate grpColIdx and
174 : : * numGroups estimate. AGG_SORTED nodes represent a "rollup", a list of
175 : : * grouping sets that share a sort order. Each AGG_SORTED node other than
176 : : * the first one has an associated Sort node which describes the sort order
177 : : * to be used; the first sorted node takes its input from the outer subtree,
178 : : * which the planner has already arranged to provide ordered data.
179 : : *
180 : : * Memory and ExprContext usage:
181 : : *
182 : : * Because we're accumulating aggregate values across input rows, we need to
183 : : * use more memory contexts than just simple input/output tuple contexts.
184 : : * In fact, for a rollup, we need a separate context for each grouping set
185 : : * so that we can reset the inner (finer-grained) aggregates on their group
186 : : * boundaries while continuing to accumulate values for outer
187 : : * (coarser-grained) groupings. On top of this, we might be simultaneously
188 : : * populating hashtables; however, we only need one context for all the
189 : : * hashtables.
190 : : *
191 : : * So we create an array, aggcontexts, with an ExprContext for each grouping
192 : : * set in the largest rollup that we're going to process, and use the
193 : : * per-tuple memory context of those ExprContexts to store the aggregate
194 : : * transition values. hashcontext is the single context created to support
195 : : * all hash tables.
196 : : *
197 : : * Spilling To Disk
198 : : *
199 : : * When performing hash aggregation, if the hash table memory exceeds the
200 : : * limit (see hash_agg_check_limits()), we enter "spill mode". In spill
201 : : * mode, we advance the transition states only for groups already in the
202 : : * hash table. For tuples that would need to create a new hash table
203 : : * entries (and initialize new transition states), we instead spill them to
204 : : * disk to be processed later. The tuples are spilled in a partitioned
205 : : * manner, so that subsequent batches are smaller and less likely to exceed
206 : : * hash_mem (if a batch does exceed hash_mem, it must be spilled
207 : : * recursively).
208 : : *
209 : : * Spilled data is written to logical tapes. These provide better control
210 : : * over memory usage, disk space, and the number of files than if we were
211 : : * to use a BufFile for each spill. We don't know the number of tapes needed
212 : : * at the start of the algorithm (because it can recurse), so a tape set is
213 : : * allocated at the beginning, and individual tapes are created as needed.
214 : : * As a particular tape is read, logtape.c recycles its disk space. When a
215 : : * tape is read to completion, it is destroyed entirely.
216 : : *
217 : : * Tapes' buffers can take up substantial memory when many tapes are open at
218 : : * once. We only need one tape open at a time in read mode (using a buffer
219 : : * that's a multiple of BLCKSZ); but we need one tape open in write mode (each
220 : : * requiring a buffer of size BLCKSZ) for each partition.
221 : : *
222 : : * Note that it's possible for transition states to start small but then
223 : : * grow very large; for instance in the case of ARRAY_AGG. In such cases,
224 : : * it's still possible to significantly exceed hash_mem. We try to avoid
225 : : * this situation by estimating what will fit in the available memory, and
226 : : * imposing a limit on the number of groups separately from the amount of
227 : : * memory consumed.
228 : : *
229 : : * Transition / Combine function invocation:
230 : : *
231 : : * For performance reasons transition functions, including combine
232 : : * functions, aren't invoked one-by-one from nodeAgg.c after computing
233 : : * arguments using the expression evaluation engine. Instead
234 : : * ExecBuildAggTrans() builds one large expression that does both argument
235 : : * evaluation and transition function invocation. That avoids performance
236 : : * issues due to repeated uses of expression evaluation, complications due
237 : : * to filter expressions having to be evaluated early, and allows to JIT
238 : : * the entire expression into one native function.
239 : : *
240 : : * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
241 : : * Portions Copyright (c) 1994, Regents of the University of California
242 : : *
243 : : * IDENTIFICATION
244 : : * src/backend/executor/nodeAgg.c
245 : : *
246 : : *-------------------------------------------------------------------------
247 : : */
248 : :
249 : : #include "postgres.h"
250 : :
251 : : #include "access/htup_details.h"
252 : : #include "access/parallel.h"
253 : : #include "catalog/objectaccess.h"
254 : : #include "catalog/pg_aggregate.h"
255 : : #include "catalog/pg_proc.h"
256 : : #include "catalog/pg_type.h"
257 : : #include "common/hashfn.h"
258 : : #include "executor/execExpr.h"
259 : : #include "executor/executor.h"
260 : : #include "executor/nodeAgg.h"
261 : : #include "lib/hyperloglog.h"
262 : : #include "miscadmin.h"
263 : : #include "nodes/nodeFuncs.h"
264 : : #include "optimizer/optimizer.h"
265 : : #include "parser/parse_agg.h"
266 : : #include "parser/parse_coerce.h"
267 : : #include "utils/acl.h"
268 : : #include "utils/builtins.h"
269 : : #include "utils/datum.h"
270 : : #include "utils/expandeddatum.h"
271 : : #include "utils/injection_point.h"
272 : : #include "utils/logtape.h"
273 : : #include "utils/lsyscache.h"
274 : : #include "utils/memutils.h"
275 : : #include "utils/memutils_memorychunk.h"
276 : : #include "utils/syscache.h"
277 : : #include "utils/tuplesort.h"
278 : :
279 : : /*
280 : : * Control how many partitions are created when spilling HashAgg to
281 : : * disk.
282 : : *
283 : : * HASHAGG_PARTITION_FACTOR is multiplied by the estimated number of
284 : : * partitions needed such that each partition will fit in memory. The factor
285 : : * is set higher than one because there's not a high cost to having a few too
286 : : * many partitions, and it makes it less likely that a partition will need to
287 : : * be spilled recursively. Another benefit of having more, smaller partitions
288 : : * is that small hash tables may perform better than large ones due to memory
289 : : * caching effects.
290 : : *
291 : : * We also specify a min and max number of partitions per spill. Too few might
292 : : * mean a lot of wasted I/O from repeated spilling of the same tuples. Too
293 : : * many will result in lots of memory wasted buffering the spill files (which
294 : : * could instead be spent on a larger hash table).
295 : : */
296 : : #define HASHAGG_PARTITION_FACTOR 1.50
297 : : #define HASHAGG_MIN_PARTITIONS 4
298 : : #define HASHAGG_MAX_PARTITIONS 1024
299 : :
300 : : /*
301 : : * For reading from tapes, the buffer size must be a multiple of
302 : : * BLCKSZ. Larger values help when reading from multiple tapes concurrently,
303 : : * but that doesn't happen in HashAgg, so we simply use BLCKSZ. Writing to a
304 : : * tape always uses a buffer of size BLCKSZ.
305 : : */
306 : : #define HASHAGG_READ_BUFFER_SIZE BLCKSZ
307 : : #define HASHAGG_WRITE_BUFFER_SIZE BLCKSZ
308 : :
309 : : /*
310 : : * HyperLogLog is used for estimating the cardinality of the spilled tuples in
311 : : * a given partition. 5 bits corresponds to a size of about 32 bytes and a
312 : : * worst-case error of around 18%. That's effective enough to choose a
313 : : * reasonable number of partitions when recursing.
314 : : */
315 : : #define HASHAGG_HLL_BIT_WIDTH 5
316 : :
317 : : /*
318 : : * Assume the palloc overhead always uses sizeof(MemoryChunk) bytes.
319 : : */
320 : : #define CHUNKHDRSZ sizeof(MemoryChunk)
321 : :
322 : : /*
323 : : * Represents partitioned spill data for a single hashtable. Contains the
324 : : * necessary information to route tuples to the correct partition, and to
325 : : * transform the spilled data into new batches.
326 : : *
327 : : * The high bits are used for partition selection (when recursing, we ignore
328 : : * the bits that have already been used for partition selection at an earlier
329 : : * level).
330 : : */
331 : : typedef struct HashAggSpill
332 : : {
333 : : int npartitions; /* number of partitions */
334 : : LogicalTape **partitions; /* spill partition tapes */
335 : : int64 *ntuples; /* number of tuples in each partition */
336 : : uint32 mask; /* mask to find partition from hash value */
337 : : int shift; /* after masking, shift by this amount */
338 : : hyperLogLogState *hll_card; /* cardinality estimate for contents */
339 : : } HashAggSpill;
340 : :
341 : : /*
342 : : * Represents work to be done for one pass of hash aggregation (with only one
343 : : * grouping set).
344 : : *
345 : : * Also tracks the bits of the hash already used for partition selection by
346 : : * earlier iterations, so that this batch can use new bits. If all bits have
347 : : * already been used, no partitioning will be done (any spilled data will go
348 : : * to a single output tape).
349 : : */
350 : : typedef struct HashAggBatch
351 : : {
352 : : int setno; /* grouping set */
353 : : int used_bits; /* number of bits of hash already used */
354 : : LogicalTape *input_tape; /* input partition tape */
355 : : int64 input_tuples; /* number of tuples in this batch */
356 : : double input_card; /* estimated group cardinality */
357 : : } HashAggBatch;
358 : :
359 : : /* used to find referenced colnos */
360 : : typedef struct FindColsContext
361 : : {
362 : : bool is_aggref; /* is under an aggref */
363 : : Bitmapset *aggregated; /* column references under an aggref */
364 : : Bitmapset *unaggregated; /* other column references */
365 : : } FindColsContext;
366 : :
367 : : static void select_current_set(AggState *aggstate, int setno, bool is_hash);
368 : : static void initialize_phase(AggState *aggstate, int newphase);
369 : : static TupleTableSlot *fetch_input_tuple(AggState *aggstate);
370 : : static void initialize_aggregates(AggState *aggstate,
371 : : AggStatePerGroup *pergroups,
372 : : int numReset);
373 : : static void advance_transition_function(AggState *aggstate,
374 : : AggStatePerTrans pertrans,
375 : : AggStatePerGroup pergroupstate);
376 : : static void advance_aggregates(AggState *aggstate);
377 : : static void process_ordered_aggregate_single(AggState *aggstate,
378 : : AggStatePerTrans pertrans,
379 : : AggStatePerGroup pergroupstate);
380 : : static void process_ordered_aggregate_multi(AggState *aggstate,
381 : : AggStatePerTrans pertrans,
382 : : AggStatePerGroup pergroupstate);
383 : : static void finalize_aggregate(AggState *aggstate,
384 : : AggStatePerAgg peragg,
385 : : AggStatePerGroup pergroupstate,
386 : : Datum *resultVal, bool *resultIsNull);
387 : : static void finalize_partialaggregate(AggState *aggstate,
388 : : AggStatePerAgg peragg,
389 : : AggStatePerGroup pergroupstate,
390 : : Datum *resultVal, bool *resultIsNull);
391 : : static inline void prepare_hash_slot(AggStatePerHash perhash,
392 : : TupleTableSlot *inputslot,
393 : : TupleTableSlot *hashslot);
394 : : static void prepare_projection_slot(AggState *aggstate,
395 : : TupleTableSlot *slot,
396 : : int currentSet);
397 : : static void finalize_aggregates(AggState *aggstate,
398 : : AggStatePerAgg peraggs,
399 : : AggStatePerGroup pergroup);
400 : : static TupleTableSlot *project_aggregates(AggState *aggstate);
401 : : static void find_cols(AggState *aggstate, Bitmapset **aggregated,
402 : : Bitmapset **unaggregated);
403 : : static bool find_cols_walker(Node *node, FindColsContext *context);
404 : : static void build_hash_tables(AggState *aggstate);
405 : : static void build_hash_table(AggState *aggstate, int setno, long nbuckets);
406 : : static void hashagg_recompile_expressions(AggState *aggstate, bool minslot,
407 : : bool nullcheck);
408 : : static void hash_create_memory(AggState *aggstate);
409 : : static long hash_choose_num_buckets(double hashentrysize,
410 : : long ngroups, Size memory);
411 : : static int hash_choose_num_partitions(double input_groups,
412 : : double hashentrysize,
413 : : int used_bits,
414 : : int *log2_npartitions);
415 : : static void initialize_hash_entry(AggState *aggstate,
416 : : TupleHashTable hashtable,
417 : : TupleHashEntry entry);
418 : : static void lookup_hash_entries(AggState *aggstate);
419 : : static TupleTableSlot *agg_retrieve_direct(AggState *aggstate);
420 : : static void agg_fill_hash_table(AggState *aggstate);
421 : : static bool agg_refill_hash_table(AggState *aggstate);
422 : : static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate);
423 : : static TupleTableSlot *agg_retrieve_hash_table_in_memory(AggState *aggstate);
424 : : static void hash_agg_check_limits(AggState *aggstate);
425 : : static void hash_agg_enter_spill_mode(AggState *aggstate);
426 : : static void hash_agg_update_metrics(AggState *aggstate, bool from_tape,
427 : : int npartitions);
428 : : static void hashagg_finish_initial_spills(AggState *aggstate);
429 : : static void hashagg_reset_spill_state(AggState *aggstate);
430 : : static HashAggBatch *hashagg_batch_new(LogicalTape *input_tape, int setno,
431 : : int64 input_tuples, double input_card,
432 : : int used_bits);
433 : : static MinimalTuple hashagg_batch_read(HashAggBatch *batch, uint32 *hashp);
434 : : static void hashagg_spill_init(HashAggSpill *spill, LogicalTapeSet *tapeset,
435 : : int used_bits, double input_groups,
436 : : double hashentrysize);
437 : : static Size hashagg_spill_tuple(AggState *aggstate, HashAggSpill *spill,
438 : : TupleTableSlot *inputslot, uint32 hash);
439 : : static void hashagg_spill_finish(AggState *aggstate, HashAggSpill *spill,
440 : : int setno);
441 : : static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
442 : : static void build_pertrans_for_aggref(AggStatePerTrans pertrans,
443 : : AggState *aggstate, EState *estate,
444 : : Aggref *aggref, Oid transfn_oid,
445 : : Oid aggtranstype, Oid aggserialfn,
446 : : Oid aggdeserialfn, Datum initValue,
447 : : bool initValueIsNull, Oid *inputTypes,
448 : : int numArguments);
449 : :
450 : :
451 : : /*
452 : : * Select the current grouping set; affects current_set and
453 : : * curaggcontext.
454 : : */
455 : : static void
3136 rhodiumtoad@postgres 456 :CBC 3942537 : select_current_set(AggState *aggstate, int setno, bool is_hash)
457 : : {
458 : : /*
459 : : * When changing this, also adapt ExecAggPlainTransByVal() and
460 : : * ExecAggPlainTransByRef().
461 : : */
462 [ + + ]: 3942537 : if (is_hash)
463 : 3609292 : aggstate->curaggcontext = aggstate->hashcontext;
464 : : else
465 : 333245 : aggstate->curaggcontext = aggstate->aggcontexts[setno];
466 : :
467 : 3942537 : aggstate->current_set = setno;
468 : 3942537 : }
469 : :
470 : : /*
471 : : * Switch to phase "newphase", which must either be 0 or 1 (to reset) or
472 : : * current_phase + 1. Juggle the tuplesorts accordingly.
473 : : *
474 : : * Phase 0 is for hashing, which we currently handle last in the AGG_MIXED
475 : : * case, so when entering phase 0, all we need to do is drop open sorts.
476 : : */
477 : : static void
3817 andres@anarazel.de 478 : 43439 : initialize_phase(AggState *aggstate, int newphase)
479 : : {
3136 rhodiumtoad@postgres 480 [ + + - + ]: 43439 : Assert(newphase <= 1 || newphase == aggstate->current_phase + 1);
481 : :
482 : : /*
483 : : * Whatever the previous state, we're now done with whatever input
484 : : * tuplesort was in use.
485 : : */
3817 andres@anarazel.de 486 [ + + ]: 43439 : if (aggstate->sort_in)
487 : : {
488 : 21 : tuplesort_end(aggstate->sort_in);
489 : 21 : aggstate->sort_in = NULL;
490 : : }
491 : :
3136 rhodiumtoad@postgres 492 [ + + ]: 43439 : if (newphase <= 1)
493 : : {
494 : : /*
495 : : * Discard any existing output tuplesort.
496 : : */
3817 andres@anarazel.de 497 [ + + ]: 43337 : if (aggstate->sort_out)
498 : : {
499 : 3 : tuplesort_end(aggstate->sort_out);
500 : 3 : aggstate->sort_out = NULL;
501 : : }
502 : : }
503 : : else
504 : : {
505 : : /*
506 : : * The old output tuplesort becomes the new input one, and this is the
507 : : * right time to actually sort it.
508 : : */
509 : 102 : aggstate->sort_in = aggstate->sort_out;
510 : 102 : aggstate->sort_out = NULL;
511 [ - + ]: 102 : Assert(aggstate->sort_in);
512 : 102 : tuplesort_performsort(aggstate->sort_in);
513 : : }
514 : :
515 : : /*
516 : : * If this isn't the last phase, we need to sort appropriately for the
517 : : * next phase in sequence.
518 : : */
3136 rhodiumtoad@postgres 519 [ + + + + ]: 43439 : if (newphase > 0 && newphase < aggstate->numphases - 1)
520 : : {
3810 bruce@momjian.us 521 : 129 : Sort *sortnode = aggstate->phases[newphase + 1].sortnode;
3817 andres@anarazel.de 522 : 129 : PlanState *outerNode = outerPlanState(aggstate);
523 : 129 : TupleDesc tupDesc = ExecGetResultType(outerNode);
524 : :
525 : 129 : aggstate->sort_out = tuplesort_begin_heap(tupDesc,
526 : : sortnode->numCols,
527 : : sortnode->sortColIdx,
528 : : sortnode->sortOperators,
529 : : sortnode->collations,
530 : : sortnode->nullsFirst,
531 : : work_mem,
532 : : NULL, TUPLESORT_NONE);
533 : : }
534 : :
535 : 43439 : aggstate->current_phase = newphase;
536 : 43439 : aggstate->phase = &aggstate->phases[newphase];
537 : 43439 : }
538 : :
539 : : /*
540 : : * Fetch a tuple from either the outer plan (for phase 1) or from the sorter
541 : : * populated by the previous phase. Copy it to the sorter for the next phase
542 : : * if any.
543 : : *
544 : : * Callers cannot rely on memory for tuple in returned slot remaining valid
545 : : * past any subsequently fetched tuple.
546 : : */
547 : : static TupleTableSlot *
548 : 14439208 : fetch_input_tuple(AggState *aggstate)
549 : : {
550 : : TupleTableSlot *slot;
551 : :
552 [ + + ]: 14439208 : if (aggstate->sort_in)
553 : : {
554 : : /* make sure we check for interrupts in either path through here */
3016 555 [ - + ]: 147450 : CHECK_FOR_INTERRUPTS();
3126 556 [ + + ]: 147450 : if (!tuplesort_gettupleslot(aggstate->sort_in, true, false,
557 : : aggstate->sort_slot, NULL))
3817 558 : 102 : return NULL;
559 : 147348 : slot = aggstate->sort_slot;
560 : : }
561 : : else
562 : 14291758 : slot = ExecProcNode(outerPlanState(aggstate));
563 : :
564 [ + + + + : 14439066 : if (!TupIsNull(slot) && aggstate->sort_out)
+ + ]
565 : 147348 : tuplesort_puttupleslot(aggstate->sort_out, slot);
566 : :
567 : 14439066 : return slot;
568 : : }
569 : :
570 : : /*
571 : : * (Re)Initialize an individual aggregate.
572 : : *
573 : : * This function handles only one grouping set, already set in
574 : : * aggstate->current_set.
575 : : *
576 : : * When called, CurrentMemoryContext should be the per-query context.
577 : : */
578 : : static void
3737 heikki.linnakangas@i 579 : 565806 : initialize_aggregate(AggState *aggstate, AggStatePerTrans pertrans,
580 : : AggStatePerGroup pergroupstate)
581 : : {
582 : : /*
583 : : * Start a fresh sort operation for each DISTINCT/ORDER BY aggregate.
584 : : */
1182 drowley@postgresql.o 585 [ + + ]: 565806 : if (pertrans->aggsortrequired)
586 : : {
587 : : /*
588 : : * In case of rescan, maybe there could be an uncompleted sort
589 : : * operation? Clean it up if so.
590 : : */
3737 heikki.linnakangas@i 591 [ - + ]: 26921 : if (pertrans->sortstates[aggstate->current_set])
3737 heikki.linnakangas@i 592 :UBC 0 : tuplesort_end(pertrans->sortstates[aggstate->current_set]);
593 : :
594 : :
595 : : /*
596 : : * We use a plain Datum sorter when there's a single input column;
597 : : * otherwise sort the full tuple. (See comments for
598 : : * process_ordered_aggregate_single.)
599 : : */
3737 heikki.linnakangas@i 600 [ + + ]:CBC 26921 : if (pertrans->numInputs == 1)
601 : : {
2990 andres@anarazel.de 602 : 26879 : Form_pg_attribute attr = TupleDescAttr(pertrans->sortdesc, 0);
603 : :
3737 heikki.linnakangas@i 604 : 26879 : pertrans->sortstates[aggstate->current_set] =
2990 andres@anarazel.de 605 : 26879 : tuplesort_begin_datum(attr->atttypid,
3737 heikki.linnakangas@i 606 : 26879 : pertrans->sortOperators[0],
607 : 26879 : pertrans->sortCollations[0],
608 : 26879 : pertrans->sortNullsFirst[0],
609 : : work_mem, NULL, TUPLESORT_NONE);
610 : : }
611 : : else
612 : 42 : pertrans->sortstates[aggstate->current_set] =
3253 andres@anarazel.de 613 : 42 : tuplesort_begin_heap(pertrans->sortdesc,
614 : : pertrans->numSortCols,
615 : : pertrans->sortColIdx,
616 : : pertrans->sortOperators,
617 : : pertrans->sortCollations,
618 : : pertrans->sortNullsFirst,
619 : : work_mem, NULL, TUPLESORT_NONE);
620 : : }
621 : :
622 : : /*
623 : : * (Re)set transValue to the initial value.
624 : : *
625 : : * Note that when the initial value is pass-by-ref, we must copy it (into
626 : : * the aggcontext) since we will pfree the transValue later.
627 : : */
3737 heikki.linnakangas@i 628 [ + + ]: 565806 : if (pertrans->initValueIsNull)
629 : 297859 : pergroupstate->transValue = pertrans->initValue;
630 : : else
631 : : {
632 : : MemoryContext oldContext;
633 : :
2097 alvherre@alvh.no-ip. 634 : 267947 : oldContext = MemoryContextSwitchTo(aggstate->curaggcontext->ecxt_per_tuple_memory);
3737 heikki.linnakangas@i 635 : 535894 : pergroupstate->transValue = datumCopy(pertrans->initValue,
636 : 267947 : pertrans->transtypeByVal,
637 : 267947 : pertrans->transtypeLen);
3817 andres@anarazel.de 638 : 267947 : MemoryContextSwitchTo(oldContext);
639 : : }
3737 heikki.linnakangas@i 640 : 565806 : pergroupstate->transValueIsNull = pertrans->initValueIsNull;
641 : :
642 : : /*
643 : : * If the initial value for the transition state doesn't exist in the
644 : : * pg_aggregate table then we will let the first non-NULL value returned
645 : : * from the outer procNode become the initial value. (This is useful for
646 : : * aggregates like max() and min().) The noTransValue flag signals that we
647 : : * still need to do this.
648 : : */
649 : 565806 : pergroupstate->noTransValue = pertrans->initValueIsNull;
3817 andres@anarazel.de 650 : 565806 : }
651 : :
652 : : /*
653 : : * Initialize all aggregate transition states for a new group of input values.
654 : : *
655 : : * If there are multiple grouping sets, we initialize only the first numReset
656 : : * of them (the grouping sets are ordered so that the most specific one, which
657 : : * is reset most often, is first). As a convenience, if numReset is 0, we
658 : : * reinitialize all sets.
659 : : *
660 : : * NB: This cannot be used for hash aggregates, as for those the grouping set
661 : : * number has to be specified from further up.
662 : : *
663 : : * When called, CurrentMemoryContext should be the per-query context.
664 : : */
665 : : static void
666 : 149813 : initialize_aggregates(AggState *aggstate,
667 : : AggStatePerGroup *pergroups,
668 : : int numReset)
669 : : {
670 : : int transno;
3810 bruce@momjian.us 671 : 149813 : int numGroupingSets = Max(aggstate->phase->numsets, 1);
672 : 149813 : int setno = 0;
3136 rhodiumtoad@postgres 673 : 149813 : int numTrans = aggstate->numtrans;
3737 heikki.linnakangas@i 674 : 149813 : AggStatePerTrans transstates = aggstate->pertrans;
675 : :
3136 rhodiumtoad@postgres 676 [ - + ]: 149813 : if (numReset == 0)
3817 andres@anarazel.de 677 :UBC 0 : numReset = numGroupingSets;
678 : :
2855 andres@anarazel.de 679 [ + + ]:CBC 306713 : for (setno = 0; setno < numReset; setno++)
680 : : {
681 : 156900 : AggStatePerGroup pergroup = pergroups[setno];
682 : :
683 : 156900 : select_current_set(aggstate, setno, false);
684 : :
685 [ + + ]: 489827 : for (transno = 0; transno < numTrans; transno++)
686 : : {
687 : 332927 : AggStatePerTrans pertrans = &transstates[transno];
688 : 332927 : AggStatePerGroup pergroupstate = &pergroup[transno];
689 : :
690 : 332927 : initialize_aggregate(aggstate, pertrans, pergroupstate);
691 : : }
692 : : }
9450 tgl@sss.pgh.pa.us 693 : 149813 : }
694 : :
695 : : /*
696 : : * Given new input value(s), advance the transition function of one aggregate
697 : : * state within one grouping set only (already set in aggstate->current_set)
698 : : *
699 : : * The new values (and null flags) have been preloaded into argument positions
700 : : * 1 and up in pertrans->transfn_fcinfo, so that we needn't copy them again to
701 : : * pass to the transition function. We also expect that the static fields of
702 : : * the fcinfo are already initialized; that was done by ExecInitAgg().
703 : : *
704 : : * It doesn't matter which memory context this is called in.
705 : : */
706 : : static void
8391 707 : 362155 : advance_transition_function(AggState *aggstate,
708 : : AggStatePerTrans pertrans,
709 : : AggStatePerGroup pergroupstate)
710 : : {
2466 andres@anarazel.de 711 : 362155 : FunctionCallInfo fcinfo = pertrans->transfn_fcinfo;
712 : : MemoryContext oldContext;
713 : : Datum newVal;
714 : :
3737 heikki.linnakangas@i 715 [ + + ]: 362155 : if (pertrans->transfn.fn_strict)
716 : : {
717 : : /*
718 : : * For a strict transfn, nothing happens when there's a NULL input; we
719 : : * just keep the prior transValue.
720 : : */
721 : 112500 : int numTransInputs = pertrans->numTransInputs;
722 : : int i;
723 : :
4326 tgl@sss.pgh.pa.us 724 [ + + ]: 225000 : for (i = 1; i <= numTransInputs; i++)
725 : : {
2466 andres@anarazel.de 726 [ - + ]: 112500 : if (fcinfo->args[i].isnull)
7032 tgl@sss.pgh.pa.us 727 :UBC 0 : return;
728 : : }
8391 tgl@sss.pgh.pa.us 729 [ - + ]:CBC 112500 : if (pergroupstate->noTransValue)
730 : : {
731 : : /*
732 : : * transValue has not been initialized. This is the first non-NULL
733 : : * input value. We use it as the initial value for transValue. (We
734 : : * already checked that the agg's input type is binary-compatible
735 : : * with its transtype, so straight copy here is OK.)
736 : : *
737 : : * We must copy the datum into aggcontext if it is pass-by-ref. We
738 : : * do not need to pfree the old transValue, since it's NULL.
739 : : */
2097 alvherre@alvh.no-ip. 740 :UBC 0 : oldContext = MemoryContextSwitchTo(aggstate->curaggcontext->ecxt_per_tuple_memory);
2466 andres@anarazel.de 741 : 0 : pergroupstate->transValue = datumCopy(fcinfo->args[1].value,
3737 heikki.linnakangas@i 742 : 0 : pertrans->transtypeByVal,
743 : 0 : pertrans->transtypeLen);
8391 tgl@sss.pgh.pa.us 744 : 0 : pergroupstate->transValueIsNull = false;
745 : 0 : pergroupstate->noTransValue = false;
746 : 0 : MemoryContextSwitchTo(oldContext);
9233 747 : 0 : return;
748 : : }
8391 tgl@sss.pgh.pa.us 749 [ - + ]:CBC 112500 : if (pergroupstate->transValueIsNull)
750 : : {
751 : : /*
752 : : * Don't call a strict function with NULL inputs. Note it is
753 : : * possible to get here despite the above tests, if the transfn is
754 : : * strict *and* returned a NULL on a prior cycle. If that happens
755 : : * we will propagate the NULL all the way to the end.
756 : : */
9233 tgl@sss.pgh.pa.us 757 :UBC 0 : return;
758 : : }
759 : : }
760 : :
761 : : /* We run the transition functions in per-input-tuple memory context */
8391 tgl@sss.pgh.pa.us 762 :CBC 362155 : oldContext = MemoryContextSwitchTo(aggstate->tmpcontext->ecxt_per_tuple_memory);
763 : :
764 : : /* set up aggstate->curpertrans for AggGetAggref() */
3737 heikki.linnakangas@i 765 : 362155 : aggstate->curpertrans = pertrans;
766 : :
767 : : /*
768 : : * OK to call the transition function
769 : : */
2466 andres@anarazel.de 770 : 362155 : fcinfo->args[0].value = pergroupstate->transValue;
771 : 362155 : fcinfo->args[0].isnull = pergroupstate->transValueIsNull;
4310 tgl@sss.pgh.pa.us 772 : 362155 : fcinfo->isnull = false; /* just in case transfn doesn't set it */
773 : :
7032 774 : 362155 : newVal = FunctionCallInvoke(fcinfo);
775 : :
3737 heikki.linnakangas@i 776 : 362155 : aggstate->curpertrans = NULL;
777 : :
778 : : /*
779 : : * If pass-by-ref datatype, must copy the new value into aggcontext and
780 : : * free the prior transValue. But if transfn returned a pointer to its
781 : : * first input, we don't need to do anything.
782 : : *
783 : : * It's safe to compare newVal with pergroup->transValue without regard
784 : : * for either being NULL, because ExecAggCopyTransValue takes care to set
785 : : * transValue to 0 when NULL. Otherwise we could end up accidentally not
786 : : * reparenting, when the transValue has the same numerical value as
787 : : * newValue, despite being NULL. This is a somewhat hot path, making it
788 : : * undesirable to instead solve this with another branch for the common
789 : : * case of the transition function returning its (modified) input
790 : : * argument.
791 : : */
792 [ - + - - ]: 362155 : if (!pertrans->transtypeByVal &&
7317 bruce@momjian.us 793 :UBC 0 : DatumGetPointer(newVal) != DatumGetPointer(pergroupstate->transValue))
917 tgl@sss.pgh.pa.us 794 : 0 : newVal = ExecAggCopyTransValue(aggstate, pertrans,
795 : 0 : newVal, fcinfo->isnull,
796 : : pergroupstate->transValue,
797 : 0 : pergroupstate->transValueIsNull);
798 : :
8391 tgl@sss.pgh.pa.us 799 :CBC 362155 : pergroupstate->transValue = newVal;
7032 800 : 362155 : pergroupstate->transValueIsNull = fcinfo->isnull;
801 : :
8391 802 : 362155 : MemoryContextSwitchTo(oldContext);
803 : : }
804 : :
805 : : /*
806 : : * Advance each aggregate transition state for one input tuple. The input
807 : : * tuple has been stored in tmpcontext->ecxt_outertuple, so that it is
808 : : * accessible to ExecEvalExpr.
809 : : *
810 : : * We have two sets of transition states to handle: one for sorted aggregation
811 : : * and one for hashed; we do them both here, to avoid multiple evaluation of
812 : : * the inputs.
813 : : *
814 : : * When called, CurrentMemoryContext should be the per-query context.
815 : : */
816 : : static void
2848 andres@anarazel.de 817 : 14777566 : advance_aggregates(AggState *aggstate)
818 : : {
230 dgustafsson@postgres 819 : 14777566 : ExecEvalExprNoReturnSwitchContext(aggstate->phase->evaltrans,
820 : : aggstate->tmpcontext);
8391 tgl@sss.pgh.pa.us 821 : 14777527 : }
822 : :
823 : : /*
824 : : * Run the transition function for a DISTINCT or ORDER BY aggregate
825 : : * with only one input. This is called after we have completed
826 : : * entering all the input values into the sort object. We complete the
827 : : * sort, read out the values in sorted order, and run the transition
828 : : * function on each value (applying DISTINCT if appropriate).
829 : : *
830 : : * Note that the strictness of the transition function was checked when
831 : : * entering the values into the sort, so we don't check it again here;
832 : : * we just apply standard SQL DISTINCT logic.
833 : : *
834 : : * The one-input case is handled separately from the multi-input case
835 : : * for performance reasons: for single by-value inputs, such as the
836 : : * common case of count(distinct id), the tuplesort_getdatum code path
837 : : * is around 300% faster. (The speedup for by-reference types is less
838 : : * but still noticeable.)
839 : : *
840 : : * This function handles only one grouping set (already set in
841 : : * aggstate->current_set).
842 : : *
843 : : * When called, CurrentMemoryContext should be the per-query context.
844 : : */
845 : : static void
5795 846 : 26879 : process_ordered_aggregate_single(AggState *aggstate,
847 : : AggStatePerTrans pertrans,
848 : : AggStatePerGroup pergroupstate)
849 : : {
9238 850 : 26879 : Datum oldVal = (Datum) 0;
5722 bruce@momjian.us 851 : 26879 : bool oldIsNull = true;
9238 tgl@sss.pgh.pa.us 852 : 26879 : bool haveOldVal = false;
8391 853 : 26879 : MemoryContext workcontext = aggstate->tmpcontext->ecxt_per_tuple_memory;
854 : : MemoryContext oldContext;
3737 heikki.linnakangas@i 855 : 26879 : bool isDistinct = (pertrans->numDistinctCols > 0);
3540 rhaas@postgresql.org 856 : 26879 : Datum newAbbrevVal = (Datum) 0;
857 : 26879 : Datum oldAbbrevVal = (Datum) 0;
2466 andres@anarazel.de 858 : 26879 : FunctionCallInfo fcinfo = pertrans->transfn_fcinfo;
859 : : Datum *newVal;
860 : : bool *isNull;
861 : :
3737 heikki.linnakangas@i 862 [ - + ]: 26879 : Assert(pertrans->numDistinctCols < 2);
863 : :
864 : 26879 : tuplesort_performsort(pertrans->sortstates[aggstate->current_set]);
865 : :
866 : : /* Load the column into argument 1 (arg 0 will be transition value) */
2466 andres@anarazel.de 867 : 26879 : newVal = &fcinfo->args[1].value;
868 : 26879 : isNull = &fcinfo->args[1].isnull;
869 : :
870 : : /*
871 : : * Note: if input type is pass-by-ref, the datums returned by the sort are
872 : : * freshly palloc'd in the per-query context, so we must be careful to
873 : : * pfree them when they are no longer needed.
874 : : */
875 : :
3737 heikki.linnakangas@i 876 [ + + ]: 449138 : while (tuplesort_getdatum(pertrans->sortstates[aggstate->current_set],
877 : : true, false, newVal, isNull, &newAbbrevVal))
878 : : {
879 : : /*
880 : : * Clear and select the working context for evaluation of the equality
881 : : * function and transition function.
882 : : */
8391 tgl@sss.pgh.pa.us 883 : 422259 : MemoryContextReset(workcontext);
884 : 422259 : oldContext = MemoryContextSwitchTo(workcontext);
885 : :
886 : : /*
887 : : * If DISTINCT mode, and not distinct from prior, skip it.
888 : : */
5795 889 [ + + + + ]: 422259 : if (isDistinct &&
890 [ - + ]: 155227 : haveOldVal &&
5795 tgl@sss.pgh.pa.us 891 [ # # ]:UBC 0 : ((oldIsNull && *isNull) ||
5795 tgl@sss.pgh.pa.us 892 [ + - + - ]:CBC 155227 : (!oldIsNull && !*isNull &&
3540 rhaas@postgresql.org 893 [ + + + + ]: 302974 : oldAbbrevVal == newAbbrevVal &&
2411 peter@eisentraut.org 894 : 147747 : DatumGetBool(FunctionCall2Coll(&pertrans->equalfnOne,
895 : : pertrans->aggCollation,
896 : : oldVal, *newVal)))))
897 : : {
1095 drowley@postgresql.o 898 : 60212 : MemoryContextSwitchTo(oldContext);
899 : 60212 : continue;
900 : : }
901 : : else
902 : : {
3737 heikki.linnakangas@i 903 : 362047 : advance_transition_function(aggstate, pertrans, pergroupstate);
904 : :
1095 drowley@postgresql.o 905 : 362047 : MemoryContextSwitchTo(oldContext);
906 : :
907 : : /*
908 : : * Forget the old value, if any, and remember the new one for
909 : : * subsequent equality checks.
910 : : */
911 [ + + ]: 362047 : if (!pertrans->inputtypeByVal)
912 : : {
913 [ + + ]: 262644 : if (!oldIsNull)
914 : 262554 : pfree(DatumGetPointer(oldVal));
915 [ + + ]: 262644 : if (!*isNull)
916 : 262614 : oldVal = datumCopy(*newVal, pertrans->inputtypeByVal,
917 : 262614 : pertrans->inputtypeLen);
918 : : }
919 : : else
920 : 99403 : oldVal = *newVal;
3540 rhaas@postgresql.org 921 : 362047 : oldAbbrevVal = newAbbrevVal;
5795 tgl@sss.pgh.pa.us 922 : 362047 : oldIsNull = *isNull;
9450 923 : 362047 : haveOldVal = true;
924 : : }
925 : : }
926 : :
3737 heikki.linnakangas@i 927 [ + + + + ]: 26879 : if (!oldIsNull && !pertrans->inputtypeByVal)
9238 tgl@sss.pgh.pa.us 928 : 60 : pfree(DatumGetPointer(oldVal));
929 : :
3737 heikki.linnakangas@i 930 : 26879 : tuplesort_end(pertrans->sortstates[aggstate->current_set]);
931 : 26879 : pertrans->sortstates[aggstate->current_set] = NULL;
9238 tgl@sss.pgh.pa.us 932 : 26879 : }
933 : :
934 : : /*
935 : : * Run the transition function for a DISTINCT or ORDER BY aggregate
936 : : * with more than one input. This is called after we have completed
937 : : * entering all the input values into the sort object. We complete the
938 : : * sort, read out the values in sorted order, and run the transition
939 : : * function on each value (applying DISTINCT if appropriate).
940 : : *
941 : : * This function handles only one grouping set (already set in
942 : : * aggstate->current_set).
943 : : *
944 : : * When called, CurrentMemoryContext should be the per-query context.
945 : : */
946 : : static void
5795 947 : 42 : process_ordered_aggregate_multi(AggState *aggstate,
948 : : AggStatePerTrans pertrans,
949 : : AggStatePerGroup pergroupstate)
950 : : {
2811 andres@anarazel.de 951 : 42 : ExprContext *tmpcontext = aggstate->tmpcontext;
2466 952 : 42 : FunctionCallInfo fcinfo = pertrans->transfn_fcinfo;
3253 953 : 42 : TupleTableSlot *slot1 = pertrans->sortslot;
3737 heikki.linnakangas@i 954 : 42 : TupleTableSlot *slot2 = pertrans->uniqslot;
955 : 42 : int numTransInputs = pertrans->numTransInputs;
956 : 42 : int numDistinctCols = pertrans->numDistinctCols;
3540 rhaas@postgresql.org 957 : 42 : Datum newAbbrevVal = (Datum) 0;
958 : 42 : Datum oldAbbrevVal = (Datum) 0;
5722 bruce@momjian.us 959 : 42 : bool haveOldValue = false;
2811 andres@anarazel.de 960 : 42 : TupleTableSlot *save = aggstate->tmpcontext->ecxt_outertuple;
961 : : int i;
962 : :
3737 heikki.linnakangas@i 963 : 42 : tuplesort_performsort(pertrans->sortstates[aggstate->current_set]);
964 : :
5795 tgl@sss.pgh.pa.us 965 : 42 : ExecClearTuple(slot1);
966 [ - + ]: 42 : if (slot2)
5795 tgl@sss.pgh.pa.us 967 :UBC 0 : ExecClearTuple(slot2);
968 : :
3737 heikki.linnakangas@i 969 [ + + ]:CBC 150 : while (tuplesort_gettupleslot(pertrans->sortstates[aggstate->current_set],
970 : : true, true, slot1, &newAbbrevVal))
971 : : {
3016 andres@anarazel.de 972 [ - + ]: 108 : CHECK_FOR_INTERRUPTS();
973 : :
2811 974 : 108 : tmpcontext->ecxt_outertuple = slot1;
975 : 108 : tmpcontext->ecxt_innertuple = slot2;
976 : :
5795 tgl@sss.pgh.pa.us 977 [ - + ]: 108 : if (numDistinctCols == 0 ||
5795 tgl@sss.pgh.pa.us 978 [ # # ]:UBC 0 : !haveOldValue ||
3540 rhaas@postgresql.org 979 [ # # ]: 0 : newAbbrevVal != oldAbbrevVal ||
2811 andres@anarazel.de 980 [ # # ]: 0 : !ExecQual(pertrans->equalfnMulti, tmpcontext))
981 : : {
982 : : /*
983 : : * Extract the first numTransInputs columns as datums to pass to
984 : : * the transfn.
985 : : */
2811 andres@anarazel.de 986 :CBC 108 : slot_getsomeattrs(slot1, numTransInputs);
987 : :
988 : : /* Load values into fcinfo */
989 : : /* Start from 1, since the 0th arg will be the transition value */
4326 tgl@sss.pgh.pa.us 990 [ + + ]: 306 : for (i = 0; i < numTransInputs; i++)
991 : : {
2466 andres@anarazel.de 992 : 198 : fcinfo->args[i + 1].value = slot1->tts_values[i];
993 : 198 : fcinfo->args[i + 1].isnull = slot1->tts_isnull[i];
994 : : }
995 : :
3737 heikki.linnakangas@i 996 : 108 : advance_transition_function(aggstate, pertrans, pergroupstate);
997 : :
5795 tgl@sss.pgh.pa.us 998 [ - + ]: 108 : if (numDistinctCols > 0)
999 : : {
1000 : : /* swap the slot pointers to retain the current tuple */
5795 tgl@sss.pgh.pa.us 1001 :UBC 0 : TupleTableSlot *tmpslot = slot2;
1002 : :
1003 : 0 : slot2 = slot1;
1004 : 0 : slot1 = tmpslot;
1005 : : /* avoid ExecQual() calls by reusing abbreviated keys */
3540 rhaas@postgresql.org 1006 : 0 : oldAbbrevVal = newAbbrevVal;
5795 tgl@sss.pgh.pa.us 1007 : 0 : haveOldValue = true;
1008 : : }
1009 : : }
1010 : :
1011 : : /* Reset context each time */
2811 andres@anarazel.de 1012 :CBC 108 : ResetExprContext(tmpcontext);
1013 : :
5795 tgl@sss.pgh.pa.us 1014 : 108 : ExecClearTuple(slot1);
1015 : : }
1016 : :
1017 [ - + ]: 42 : if (slot2)
5795 tgl@sss.pgh.pa.us 1018 :UBC 0 : ExecClearTuple(slot2);
1019 : :
3737 heikki.linnakangas@i 1020 :CBC 42 : tuplesort_end(pertrans->sortstates[aggstate->current_set]);
1021 : 42 : pertrans->sortstates[aggstate->current_set] = NULL;
1022 : :
1023 : : /* restore previous slot, potentially in use for grouping sets */
2811 andres@anarazel.de 1024 : 42 : tmpcontext->ecxt_outertuple = save;
5795 tgl@sss.pgh.pa.us 1025 : 42 : }
1026 : :
1027 : : /*
1028 : : * Compute the final value of one aggregate.
1029 : : *
1030 : : * This function handles only one grouping set (already set in
1031 : : * aggstate->current_set).
1032 : : *
1033 : : * The finalfn will be run, and the result delivered, in the
1034 : : * output-tuple context; caller's CurrentMemoryContext does not matter.
1035 : : * (But note that in some cases, such as when there is no finalfn, the
1036 : : * result might be a pointer to or into the agg's transition value.)
1037 : : *
1038 : : * The finalfn uses the state as set in the transno. This also might be
1039 : : * being used by another aggregate function, so it's important that we do
1040 : : * nothing destructive here. Moreover, the aggregate's final value might
1041 : : * get used in multiple places, so we mustn't return a R/W expanded datum.
1042 : : */
1043 : : static void
8391 1044 : 558318 : finalize_aggregate(AggState *aggstate,
1045 : : AggStatePerAgg peragg,
1046 : : AggStatePerGroup pergroupstate,
1047 : : Datum *resultVal, bool *resultIsNull)
1048 : : {
2466 andres@anarazel.de 1049 : 558318 : LOCAL_FCINFO(fcinfo, FUNC_MAX_ARGS);
4326 tgl@sss.pgh.pa.us 1050 : 558318 : bool anynull = false;
1051 : : MemoryContext oldContext;
1052 : : int i;
1053 : : ListCell *lc;
3737 heikki.linnakangas@i 1054 : 558318 : AggStatePerTrans pertrans = &aggstate->pertrans[peragg->transno];
1055 : :
8362 tgl@sss.pgh.pa.us 1056 : 558318 : oldContext = MemoryContextSwitchTo(aggstate->ss.ps.ps_ExprContext->ecxt_per_tuple_memory);
1057 : :
1058 : : /*
1059 : : * Evaluate any direct arguments. We do this even if there's no finalfn
1060 : : * (which is unlikely anyway), so that side-effects happen as expected.
1061 : : * The direct arguments go into arg positions 1 and up, leaving position 0
1062 : : * for the transition state value.
1063 : : */
4326 1064 : 558318 : i = 1;
2933 1065 [ + + + + : 558805 : foreach(lc, peragg->aggdirectargs)
+ + ]
1066 : : {
4326 1067 : 487 : ExprState *expr = (ExprState *) lfirst(lc);
1068 : :
2466 andres@anarazel.de 1069 : 487 : fcinfo->args[i].value = ExecEvalExpr(expr,
1070 : : aggstate->ss.ps.ps_ExprContext,
1071 : : &fcinfo->args[i].isnull);
1072 : 487 : anynull |= fcinfo->args[i].isnull;
4326 tgl@sss.pgh.pa.us 1073 : 487 : i++;
1074 : : }
1075 : :
1076 : : /*
1077 : : * Apply the agg's finalfn if one is provided, else return transValue.
1078 : : */
3737 heikki.linnakangas@i 1079 [ + + ]: 558318 : if (OidIsValid(peragg->finalfn_oid))
1080 : : {
1081 : 168524 : int numFinalArgs = peragg->numFinalArgs;
1082 : :
1083 : : /* set up aggstate->curperagg for AggGetAggref() */
2937 tgl@sss.pgh.pa.us 1084 : 168524 : aggstate->curperagg = peragg;
1085 : :
2466 andres@anarazel.de 1086 : 168524 : InitFunctionCallInfoData(*fcinfo, &peragg->finalfn,
1087 : : numFinalArgs,
1088 : : pertrans->aggCollation,
1089 : : (Node *) aggstate, NULL);
1090 : :
1091 : : /* Fill in the transition state value */
1092 : 168524 : fcinfo->args[0].value =
1093 [ + + + + ]: 168524 : MakeExpandedObjectReadOnly(pergroupstate->transValue,
1094 : : pergroupstate->transValueIsNull,
1095 : : pertrans->transtypeLen);
1096 : 168524 : fcinfo->args[0].isnull = pergroupstate->transValueIsNull;
4326 tgl@sss.pgh.pa.us 1097 : 168524 : anynull |= pergroupstate->transValueIsNull;
1098 : :
1099 : : /* Fill any remaining argument positions with nulls */
4205 1100 [ + + ]: 243956 : for (; i < numFinalArgs; i++)
1101 : : {
2466 andres@anarazel.de 1102 : 75432 : fcinfo->args[i].value = (Datum) 0;
1103 : 75432 : fcinfo->args[i].isnull = true;
4326 tgl@sss.pgh.pa.us 1104 : 75432 : anynull = true;
1105 : : }
1106 : :
2466 andres@anarazel.de 1107 [ + + - + ]: 168524 : if (fcinfo->flinfo->fn_strict && anynull)
1108 : : {
1109 : : /* don't call a strict function with NULL inputs */
9283 tgl@sss.pgh.pa.us 1110 :UBC 0 : *resultVal = (Datum) 0;
1111 : 0 : *resultIsNull = true;
1112 : : }
1113 : : else
1114 : : {
1115 : : Datum result;
1116 : :
925 tgl@sss.pgh.pa.us 1117 :CBC 168524 : result = FunctionCallInvoke(fcinfo);
2466 andres@anarazel.de 1118 : 168518 : *resultIsNull = fcinfo->isnull;
925 tgl@sss.pgh.pa.us 1119 [ + + + + ]: 168518 : *resultVal = MakeExpandedObjectReadOnly(result,
1120 : : fcinfo->isnull,
1121 : : peragg->resulttypeLen);
1122 : : }
2937 1123 : 168518 : aggstate->curperagg = NULL;
1124 : : }
1125 : : else
1126 : : {
1117 1127 : 389794 : *resultVal =
1128 [ + + + + ]: 389794 : MakeExpandedObjectReadOnly(pergroupstate->transValue,
1129 : : pergroupstate->transValueIsNull,
1130 : : pertrans->transtypeLen);
8391 1131 : 389794 : *resultIsNull = pergroupstate->transValueIsNull;
1132 : : }
1133 : :
1134 : 558312 : MemoryContextSwitchTo(oldContext);
9450 1135 : 558312 : }
1136 : :
1137 : : /*
1138 : : * Compute the output value of one partial aggregate.
1139 : : *
1140 : : * The serialization function will be run, and the result delivered, in the
1141 : : * output-tuple context; caller's CurrentMemoryContext does not matter.
1142 : : */
1143 : : static void
3499 rhaas@postgresql.org 1144 : 9257 : finalize_partialaggregate(AggState *aggstate,
1145 : : AggStatePerAgg peragg,
1146 : : AggStatePerGroup pergroupstate,
1147 : : Datum *resultVal, bool *resultIsNull)
1148 : : {
3427 1149 : 9257 : AggStatePerTrans pertrans = &aggstate->pertrans[peragg->transno];
1150 : : MemoryContext oldContext;
1151 : :
3499 1152 : 9257 : oldContext = MemoryContextSwitchTo(aggstate->ss.ps.ps_ExprContext->ecxt_per_tuple_memory);
1153 : :
1154 : : /*
1155 : : * serialfn_oid will be set if we must serialize the transvalue before
1156 : : * returning it
1157 : : */
1158 [ + + ]: 9257 : if (OidIsValid(pertrans->serialfn_oid))
1159 : : {
1160 : : /* Don't call a strict serialization function with NULL input. */
1161 [ + - + + ]: 333 : if (pertrans->serialfn.fn_strict && pergroupstate->transValueIsNull)
1162 : : {
1163 : 35 : *resultVal = (Datum) 0;
1164 : 35 : *resultIsNull = true;
1165 : : }
1166 : : else
1167 : : {
2466 andres@anarazel.de 1168 : 298 : FunctionCallInfo fcinfo = pertrans->serialfn_fcinfo;
1169 : : Datum result;
1170 : :
1171 : 298 : fcinfo->args[0].value =
1172 [ + - + - ]: 298 : MakeExpandedObjectReadOnly(pergroupstate->transValue,
1173 : : pergroupstate->transValueIsNull,
1174 : : pertrans->transtypeLen);
1175 : 298 : fcinfo->args[0].isnull = pergroupstate->transValueIsNull;
2015 tgl@sss.pgh.pa.us 1176 : 298 : fcinfo->isnull = false;
1177 : :
925 1178 : 298 : result = FunctionCallInvoke(fcinfo);
3499 rhaas@postgresql.org 1179 : 298 : *resultIsNull = fcinfo->isnull;
925 tgl@sss.pgh.pa.us 1180 [ + - + - ]: 298 : *resultVal = MakeExpandedObjectReadOnly(result,
1181 : : fcinfo->isnull,
1182 : : peragg->resulttypeLen);
1183 : : }
1184 : : }
1185 : : else
1186 : : {
1117 1187 : 8924 : *resultVal =
1188 [ + + + + ]: 8924 : MakeExpandedObjectReadOnly(pergroupstate->transValue,
1189 : : pergroupstate->transValueIsNull,
1190 : : pertrans->transtypeLen);
3499 rhaas@postgresql.org 1191 : 8924 : *resultIsNull = pergroupstate->transValueIsNull;
1192 : : }
1193 : :
1194 : 9257 : MemoryContextSwitchTo(oldContext);
1195 : 9257 : }
1196 : :
1197 : : /*
1198 : : * Extract the attributes that make up the grouping key into the
1199 : : * hashslot. This is necessary to compute the hash or perform a lookup.
1200 : : */
1201 : : static inline void
1919 jdavis@postgresql.or 1202 : 4112545 : prepare_hash_slot(AggStatePerHash perhash,
1203 : : TupleTableSlot *inputslot,
1204 : : TupleTableSlot *hashslot)
1205 : : {
1206 : : int i;
1207 : :
1208 : : /* transfer just the needed columns into hashslot */
2077 1209 : 4112545 : slot_getsomeattrs(inputslot, perhash->largestGrpColIdx);
1210 : 4112545 : ExecClearTuple(hashslot);
1211 : :
1212 [ + + ]: 10199782 : for (i = 0; i < perhash->numhashGrpCols; i++)
1213 : : {
1214 : 6087237 : int varNumber = perhash->hashGrpColIdxInput[i] - 1;
1215 : :
1216 : 6087237 : hashslot->tts_values[i] = inputslot->tts_values[varNumber];
1217 : 6087237 : hashslot->tts_isnull[i] = inputslot->tts_isnull[varNumber];
1218 : : }
1219 : 4112545 : ExecStoreVirtualTuple(hashslot);
1220 : 4112545 : }
1221 : :
1222 : : /*
1223 : : * Prepare to finalize and project based on the specified representative tuple
1224 : : * slot and grouping set.
1225 : : *
1226 : : * In the specified tuple slot, force to null all attributes that should be
1227 : : * read as null in the context of the current grouping set. Also stash the
1228 : : * current group bitmap where GroupingExpr can get at it.
1229 : : *
1230 : : * This relies on three conditions:
1231 : : *
1232 : : * 1) Nothing is ever going to try and extract the whole tuple from this slot,
1233 : : * only reference it in evaluations, which will only access individual
1234 : : * attributes.
1235 : : *
1236 : : * 2) No system columns are going to need to be nulled. (If a system column is
1237 : : * referenced in a group clause, it is actually projected in the outer plan
1238 : : * tlist.)
1239 : : *
1240 : : * 3) Within a given phase, we never need to recover the value of an attribute
1241 : : * once it has been set to null.
1242 : : *
1243 : : * Poking into the slot this way is a bit ugly, but the consensus is that the
1244 : : * alternative was worse.
1245 : : */
1246 : : static void
3817 andres@anarazel.de 1247 : 416314 : prepare_projection_slot(AggState *aggstate, TupleTableSlot *slot, int currentSet)
1248 : : {
1249 [ + + ]: 416314 : if (aggstate->phase->grouped_cols)
1250 : : {
3810 bruce@momjian.us 1251 : 274199 : Bitmapset *grouped_cols = aggstate->phase->grouped_cols[currentSet];
1252 : :
3817 andres@anarazel.de 1253 : 274199 : aggstate->grouped_cols = grouped_cols;
1254 : :
2569 1255 [ + + ]: 274199 : if (TTS_EMPTY(slot))
1256 : : {
1257 : : /*
1258 : : * Force all values to be NULL if working on an empty input tuple
1259 : : * (i.e. an empty grouping set for which no input rows were
1260 : : * supplied).
1261 : : */
3817 1262 : 30 : ExecStoreAllNullTuple(slot);
1263 : : }
1264 [ + + ]: 274169 : else if (aggstate->all_grouped_cols)
1265 : : {
1266 : : ListCell *lc;
1267 : :
1268 : : /* all_grouped_cols is arranged in desc order */
1269 : 274145 : slot_getsomeattrs(slot, linitial_int(aggstate->all_grouped_cols));
1270 : :
1271 [ + - + + : 752335 : foreach(lc, aggstate->all_grouped_cols)
+ + ]
1272 : : {
3810 bruce@momjian.us 1273 : 478190 : int attnum = lfirst_int(lc);
1274 : :
3817 andres@anarazel.de 1275 [ + + ]: 478190 : if (!bms_is_member(attnum, grouped_cols))
1276 : 28916 : slot->tts_isnull[attnum - 1] = true;
1277 : : }
1278 : : }
1279 : : }
1280 : 416314 : }
1281 : :
1282 : : /*
1283 : : * Compute the final value of all aggregates for one group.
1284 : : *
1285 : : * This function handles only one grouping set at a time, which the caller must
1286 : : * have selected. It's also the caller's responsibility to adjust the supplied
1287 : : * pergroup parameter to point to the current set's transvalues.
1288 : : *
1289 : : * Results are stored in the output econtext aggvalues/aggnulls.
1290 : : */
1291 : : static void
1292 : 416314 : finalize_aggregates(AggState *aggstate,
1293 : : AggStatePerAgg peraggs,
1294 : : AggStatePerGroup pergroup)
1295 : : {
1296 : 416314 : ExprContext *econtext = aggstate->ss.ps.ps_ExprContext;
1297 : 416314 : Datum *aggvalues = econtext->ecxt_aggvalues;
1298 : 416314 : bool *aggnulls = econtext->ecxt_aggnulls;
1299 : : int aggno;
1300 : :
1301 : : /*
1302 : : * If there were any DISTINCT and/or ORDER BY aggregates, sort their
1303 : : * inputs and run the transition functions.
1304 : : */
1160 drowley@postgresql.o 1305 [ + + ]: 983760 : for (int transno = 0; transno < aggstate->numtrans; transno++)
1306 : : {
3737 heikki.linnakangas@i 1307 : 567446 : AggStatePerTrans pertrans = &aggstate->pertrans[transno];
1308 : : AggStatePerGroup pergroupstate;
1309 : :
3136 rhodiumtoad@postgres 1310 : 567446 : pergroupstate = &pergroup[transno];
1311 : :
1182 drowley@postgresql.o 1312 [ + + ]: 567446 : if (pertrans->aggsortrequired)
1313 : : {
3136 rhodiumtoad@postgres 1314 [ + - - + ]: 26921 : Assert(aggstate->aggstrategy != AGG_HASHED &&
1315 : : aggstate->aggstrategy != AGG_MIXED);
1316 : :
3737 heikki.linnakangas@i 1317 [ + + ]: 26921 : if (pertrans->numInputs == 1)
3817 andres@anarazel.de 1318 : 26879 : process_ordered_aggregate_single(aggstate,
1319 : : pertrans,
1320 : : pergroupstate);
1321 : : else
1322 : 42 : process_ordered_aggregate_multi(aggstate,
1323 : : pertrans,
1324 : : pergroupstate);
1325 : : }
1182 drowley@postgresql.o 1326 [ + + + + ]: 540525 : else if (pertrans->numDistinctCols > 0 && pertrans->haslast)
1327 : : {
1328 : 9179 : pertrans->haslast = false;
1329 : :
1330 [ + + ]: 9179 : if (pertrans->numDistinctCols == 1)
1331 : : {
1332 [ + + + + ]: 9131 : if (!pertrans->inputtypeByVal && !pertrans->lastisnull)
1333 : 131 : pfree(DatumGetPointer(pertrans->lastdatum));
1334 : :
1335 : 9131 : pertrans->lastisnull = false;
1336 : 9131 : pertrans->lastdatum = (Datum) 0;
1337 : : }
1338 : : else
1339 : 48 : ExecClearTuple(pertrans->uniqslot);
1340 : : }
1341 : : }
1342 : :
1343 : : /*
1344 : : * Run the final functions.
1345 : : */
3233 heikki.linnakangas@i 1346 [ + + ]: 983883 : for (aggno = 0; aggno < aggstate->numaggs; aggno++)
1347 : : {
1348 : 567575 : AggStatePerAgg peragg = &peraggs[aggno];
1349 : 567575 : int transno = peragg->transno;
1350 : : AggStatePerGroup pergroupstate;
1351 : :
3136 rhodiumtoad@postgres 1352 : 567575 : pergroupstate = &pergroup[transno];
1353 : :
3410 tgl@sss.pgh.pa.us 1354 [ + + ]: 567575 : if (DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit))
3499 rhaas@postgresql.org 1355 : 9257 : finalize_partialaggregate(aggstate, peragg, pergroupstate,
1356 : 9257 : &aggvalues[aggno], &aggnulls[aggno]);
1357 : : else
3410 tgl@sss.pgh.pa.us 1358 : 558318 : finalize_aggregate(aggstate, peragg, pergroupstate,
1359 : 558318 : &aggvalues[aggno], &aggnulls[aggno]);
1360 : : }
3817 andres@anarazel.de 1361 : 416308 : }
1362 : :
1363 : : /*
1364 : : * Project the result of a group (whose aggs have already been calculated by
1365 : : * finalize_aggregates). Returns the result slot, or NULL if no row is
1366 : : * projected (suppressed by qual).
1367 : : */
1368 : : static TupleTableSlot *
1369 : 416308 : project_aggregates(AggState *aggstate)
1370 : : {
1371 : 416308 : ExprContext *econtext = aggstate->ss.ps.ps_ExprContext;
1372 : :
1373 : : /*
1374 : : * Check the qual (HAVING clause); if the group does not match, ignore it.
1375 : : */
3149 1376 [ + + ]: 416308 : if (ExecQual(aggstate->ss.ps.qual, econtext))
1377 : : {
1378 : : /*
1379 : : * Form and return projection tuple using the aggregate results and
1380 : : * the representative input tuple.
1381 : : */
3203 1382 : 363070 : return ExecProject(aggstate->ss.ps.ps_ProjInfo);
1383 : : }
1384 : : else
3817 1385 [ - + ]: 53238 : InstrCountFiltered1(aggstate, 1);
1386 : :
1387 : 53238 : return NULL;
1388 : : }
1389 : :
1390 : : /*
1391 : : * Find input-tuple columns that are needed, dividing them into
1392 : : * aggregated and unaggregated sets.
1393 : : */
1394 : : static void
1933 jdavis@postgresql.or 1395 : 3499 : find_cols(AggState *aggstate, Bitmapset **aggregated, Bitmapset **unaggregated)
1396 : : {
1629 tgl@sss.pgh.pa.us 1397 : 3499 : Agg *agg = (Agg *) aggstate->ss.ps.plan;
1398 : : FindColsContext context;
1399 : :
1933 jdavis@postgresql.or 1400 : 3499 : context.is_aggref = false;
1401 : 3499 : context.aggregated = NULL;
1402 : 3499 : context.unaggregated = NULL;
1403 : :
1404 : : /* Examine tlist and quals */
1405 : 3499 : (void) find_cols_walker((Node *) agg->plan.targetlist, &context);
1406 : 3499 : (void) find_cols_walker((Node *) agg->plan.qual, &context);
1407 : :
1408 : : /* In some cases, grouping columns will not appear in the tlist */
1726 tgl@sss.pgh.pa.us 1409 [ + + ]: 8902 : for (int i = 0; i < agg->numCols; i++)
1410 : 5403 : context.unaggregated = bms_add_member(context.unaggregated,
1411 : 5403 : agg->grpColIdx[i]);
1412 : :
1933 jdavis@postgresql.or 1413 : 3499 : *aggregated = context.aggregated;
1414 : 3499 : *unaggregated = context.unaggregated;
7061 tgl@sss.pgh.pa.us 1415 : 3499 : }
1416 : :
1417 : : static bool
1933 jdavis@postgresql.or 1418 : 41861 : find_cols_walker(Node *node, FindColsContext *context)
1419 : : {
7061 tgl@sss.pgh.pa.us 1420 [ + + ]: 41861 : if (node == NULL)
1421 : 7540 : return false;
1422 [ + + ]: 34321 : if (IsA(node, Var))
1423 : : {
1424 : 9440 : Var *var = (Var *) node;
1425 : :
1426 : : /* setrefs.c should have set the varno to OUTER_VAR */
5130 1427 [ - + ]: 9440 : Assert(var->varno == OUTER_VAR);
7061 1428 [ - + ]: 9440 : Assert(var->varlevelsup == 0);
1933 jdavis@postgresql.or 1429 [ + + ]: 9440 : if (context->is_aggref)
1430 : 3049 : context->aggregated = bms_add_member(context->aggregated,
1431 : 3049 : var->varattno);
1432 : : else
1433 : 6391 : context->unaggregated = bms_add_member(context->unaggregated,
1434 : 6391 : var->varattno);
7061 tgl@sss.pgh.pa.us 1435 : 9440 : return false;
1436 : : }
1933 jdavis@postgresql.or 1437 [ + + ]: 24881 : if (IsA(node, Aggref))
1438 : : {
1439 [ - + ]: 4291 : Assert(!context->is_aggref);
1440 : 4291 : context->is_aggref = true;
333 peter@eisentraut.org 1441 : 4291 : expression_tree_walker(node, find_cols_walker, context);
1933 jdavis@postgresql.or 1442 : 4291 : context->is_aggref = false;
7061 tgl@sss.pgh.pa.us 1443 : 4291 : return false;
1444 : : }
333 peter@eisentraut.org 1445 : 20590 : return expression_tree_walker(node, find_cols_walker, context);
1446 : : }
1447 : :
1448 : : /*
1449 : : * (Re-)initialize the hash table(s) to empty.
1450 : : *
1451 : : * To implement hashed aggregation, we need a hashtable that stores a
1452 : : * representative tuple and an array of AggStatePerGroup structs for each
1453 : : * distinct set of GROUP BY column values. We compute the hash key from the
1454 : : * GROUP BY columns. The per-group data is allocated in initialize_hash_entry(),
1455 : : * for each entry.
1456 : : *
1457 : : * We have a separate hashtable and associated perhash data structure for each
1458 : : * grouping set for which we're doing hashing.
1459 : : *
1460 : : * The contents of the hash tables always live in the hashcontext's per-tuple
1461 : : * memory context (there is only one of these for all tables together, since
1462 : : * they are all reset at the same time).
1463 : : */
1464 : : static void
2077 jdavis@postgresql.or 1465 : 8777 : build_hash_tables(AggState *aggstate)
1466 : : {
1467 : : int setno;
1468 : :
1469 [ + + ]: 17726 : for (setno = 0; setno < aggstate->num_hashes; ++setno)
1470 : : {
1471 : 8949 : AggStatePerHash perhash = &aggstate->perhash[setno];
1472 : : long nbuckets;
1473 : : Size memory;
1474 : :
2049 1475 [ + + ]: 8949 : if (perhash->hashtable != NULL)
1476 : : {
1477 : 6231 : ResetTupleHashTable(perhash->hashtable);
1478 : 6231 : continue;
1479 : : }
1480 : :
3136 rhodiumtoad@postgres 1481 [ - + ]: 2718 : Assert(perhash->aggnode->numGroups > 0);
1482 : :
2049 jdavis@postgresql.or 1483 : 2718 : memory = aggstate->hash_mem_limit / aggstate->num_hashes;
1484 : :
1485 : : /* choose reasonable number of buckets per hashtable */
1992 tgl@sss.pgh.pa.us 1486 : 2718 : nbuckets = hash_choose_num_buckets(aggstate->hashentrysize,
1487 : 2718 : perhash->aggnode->numGroups,
1488 : : memory);
1489 : :
1490 : : #ifdef USE_INJECTION_POINTS
1491 : : if (IS_INJECTION_POINT_ATTACHED("hash-aggregate-oversize-table"))
1492 : : {
1493 : : nbuckets = memory / TupleHashEntrySize();
1494 : : INJECTION_POINT_CACHED("hash-aggregate-oversize-table", NULL);
1495 : : }
1496 : : #endif
1497 : :
2049 jdavis@postgresql.or 1498 : 2718 : build_hash_table(aggstate, setno, nbuckets);
1499 : : }
1500 : :
1501 : 8777 : aggstate->hash_ngroups_current = 0;
6220 neilc@samurai.com 1502 : 8777 : }
1503 : :
1504 : : /*
1505 : : * Build a single hashtable for this grouping set.
1506 : : */
1507 : : static void
2077 jdavis@postgresql.or 1508 : 2718 : build_hash_table(AggState *aggstate, int setno, long nbuckets)
1509 : : {
1510 : 2718 : AggStatePerHash perhash = &aggstate->perhash[setno];
1992 tgl@sss.pgh.pa.us 1511 : 2718 : MemoryContext metacxt = aggstate->hash_metacxt;
217 jdavis@postgresql.or 1512 : 2718 : MemoryContext tablecxt = aggstate->hash_tablecxt;
1992 tgl@sss.pgh.pa.us 1513 : 2718 : MemoryContext tmpcxt = aggstate->tmpcontext->ecxt_per_tuple_memory;
1514 : : Size additionalsize;
1515 : :
2077 jdavis@postgresql.or 1516 [ + + - + ]: 2718 : Assert(aggstate->aggstrategy == AGG_HASHED ||
1517 : : aggstate->aggstrategy == AGG_MIXED);
1518 : :
1519 : : /*
1520 : : * Used to make sure initial hash table allocation does not exceed
1521 : : * hash_mem. Note that the estimate does not include space for
1522 : : * pass-by-reference transition data values, nor for the representative
1523 : : * tuple of each group.
1524 : : */
1525 : 2718 : additionalsize = aggstate->numtrans * sizeof(AggStatePerGroupData);
1526 : :
312 tgl@sss.pgh.pa.us 1527 : 5436 : perhash->hashtable = BuildTupleHashTable(&aggstate->ss.ps,
1528 : 2718 : perhash->hashslot->tts_tupleDescriptor,
1529 : 2718 : perhash->hashslot->tts_ops,
1530 : : perhash->numCols,
1531 : : perhash->hashGrpColIdxHash,
1532 : 2718 : perhash->eqfuncoids,
1533 : : perhash->hashfunctions,
1534 : 2718 : perhash->aggnode->grpCollations,
1535 : : nbuckets,
1536 : : additionalsize,
1537 : : metacxt,
1538 : : tablecxt,
1539 : : tmpcxt,
1540 : 2718 : DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit));
2077 jdavis@postgresql.or 1541 : 2718 : }
1542 : :
1543 : : /*
1544 : : * Compute columns that actually need to be stored in hashtable entries. The
1545 : : * incoming tuples from the child plan node will contain grouping columns,
1546 : : * other columns referenced in our targetlist and qual, columns used to
1547 : : * compute the aggregate functions, and perhaps just junk columns we don't use
1548 : : * at all. Only columns of the first two types need to be stored in the
1549 : : * hashtable, and getting rid of the others can make the table entries
1550 : : * significantly smaller. The hashtable only contains the relevant columns,
1551 : : * and is packed/unpacked in lookup_hash_entries() / agg_retrieve_hash_table()
1552 : : * into the format of the normal input descriptor.
1553 : : *
1554 : : * Additional columns, in addition to the columns grouped by, come from two
1555 : : * sources: Firstly functionally dependent columns that we don't need to group
1556 : : * by themselves, and secondly ctids for row-marks.
1557 : : *
1558 : : * To eliminate duplicates, we build a bitmapset of the needed columns, and
1559 : : * then build an array of the columns included in the hashtable. We might
1560 : : * still have duplicates if the passed-in grpColIdx has them, which can happen
1561 : : * in edge cases from semijoins/distinct; these can't always be removed,
1562 : : * because it's not certain that the duplicate cols will be using the same
1563 : : * hash function.
1564 : : *
1565 : : * Note that the array is preserved over ExecReScanAgg, so we allocate it in
1566 : : * the per-query context (unlike the hash table itself).
1567 : : */
1568 : : static void
6220 neilc@samurai.com 1569 : 3499 : find_hash_columns(AggState *aggstate)
1570 : : {
1571 : : Bitmapset *base_colnos;
1572 : : Bitmapset *aggregated_colnos;
1933 jdavis@postgresql.or 1573 : 3499 : TupleDesc scanDesc = aggstate->ss.ss_ScanTupleSlot->tts_tupleDescriptor;
3253 andres@anarazel.de 1574 : 3499 : List *outerTlist = outerPlanState(aggstate)->plan->targetlist;
3136 rhodiumtoad@postgres 1575 : 3499 : int numHashes = aggstate->num_hashes;
2811 andres@anarazel.de 1576 : 3499 : EState *estate = aggstate->ss.ps.state;
1577 : : int j;
1578 : :
1579 : : /* Find Vars that will be needed in tlist and qual */
1933 jdavis@postgresql.or 1580 : 3499 : find_cols(aggstate, &aggregated_colnos, &base_colnos);
1581 : 3499 : aggstate->colnos_needed = bms_union(base_colnos, aggregated_colnos);
1582 : 3499 : aggstate->max_colno_needed = 0;
1583 : 3499 : aggstate->all_cols_needed = true;
1584 : :
1585 [ + + ]: 14639 : for (int i = 0; i < scanDesc->natts; i++)
1586 : : {
1629 tgl@sss.pgh.pa.us 1587 : 11140 : int colno = i + 1;
1588 : :
1933 jdavis@postgresql.or 1589 [ + + ]: 11140 : if (bms_is_member(colno, aggstate->colnos_needed))
1590 : 8099 : aggstate->max_colno_needed = colno;
1591 : : else
1592 : 3041 : aggstate->all_cols_needed = false;
1593 : : }
1594 : :
3136 rhodiumtoad@postgres 1595 [ + + ]: 7255 : for (j = 0; j < numHashes; ++j)
1596 : : {
1597 : 3756 : AggStatePerHash perhash = &aggstate->perhash[j];
1598 : 3756 : Bitmapset *colnos = bms_copy(base_colnos);
1599 : 3756 : AttrNumber *grpColIdx = perhash->aggnode->grpColIdx;
1600 : 3756 : List *hashTlist = NIL;
1601 : : TupleDesc hashDesc;
1602 : : int maxCols;
1603 : : int i;
1604 : :
1605 : 3756 : perhash->largestGrpColIdx = 0;
1606 : :
1607 : : /*
1608 : : * If we're doing grouping sets, then some Vars might be referenced in
1609 : : * tlist/qual for the benefit of other grouping sets, but not needed
1610 : : * when hashing; i.e. prepare_projection_slot will null them out, so
1611 : : * there'd be no point storing them. Use prepare_projection_slot's
1612 : : * logic to determine which.
1613 : : */
1614 [ + - ]: 3756 : if (aggstate->phases[0].grouped_cols)
1615 : : {
1616 : 3756 : Bitmapset *grouped_cols = aggstate->phases[0].grouped_cols[j];
1617 : : ListCell *lc;
1618 : :
1619 [ + - + + : 10142 : foreach(lc, aggstate->all_grouped_cols)
+ + ]
1620 : : {
1621 : 6386 : int attnum = lfirst_int(lc);
1622 : :
1623 [ + + ]: 6386 : if (!bms_is_member(attnum, grouped_cols))
1624 : 672 : colnos = bms_del_member(colnos, attnum);
1625 : : }
1626 : : }
1627 : :
1628 : : /*
1629 : : * Compute maximum number of input columns accounting for possible
1630 : : * duplications in the grpColIdx array, which can happen in some edge
1631 : : * cases where HashAggregate was generated as part of a semijoin or a
1632 : : * DISTINCT.
1633 : : */
2349 1634 : 3756 : maxCols = bms_num_members(colnos) + perhash->numCols;
1635 : :
3136 1636 : 3756 : perhash->hashGrpColIdxInput =
2349 1637 : 3756 : palloc(maxCols * sizeof(AttrNumber));
3136 1638 : 3756 : perhash->hashGrpColIdxHash =
1639 : 3756 : palloc(perhash->numCols * sizeof(AttrNumber));
1640 : :
1641 : : /* Add all the grouping columns to colnos */
2349 1642 [ + + ]: 9470 : for (i = 0; i < perhash->numCols; i++)
1643 : 5714 : colnos = bms_add_member(colnos, grpColIdx[i]);
1644 : :
1645 : : /*
1646 : : * First build mapping for columns directly hashed. These are the
1647 : : * first, because they'll be accessed when computing hash values and
1648 : : * comparing tuples for exact matches. We also build simple mapping
1649 : : * for execGrouping, so it knows where to find the to-be-hashed /
1650 : : * compared columns in the input.
1651 : : */
3136 1652 [ + + ]: 9470 : for (i = 0; i < perhash->numCols; i++)
1653 : : {
1654 : 5714 : perhash->hashGrpColIdxInput[i] = grpColIdx[i];
1655 : 5714 : perhash->hashGrpColIdxHash[i] = i + 1;
1656 : 5714 : perhash->numhashGrpCols++;
1657 : : /* delete already mapped columns */
970 tgl@sss.pgh.pa.us 1658 : 5714 : colnos = bms_del_member(colnos, grpColIdx[i]);
1659 : : }
1660 : :
1661 : : /* and add the remaining columns */
1662 : 3756 : i = -1;
1663 [ + + ]: 4383 : while ((i = bms_next_member(colnos, i)) >= 0)
1664 : : {
3136 rhodiumtoad@postgres 1665 : 627 : perhash->hashGrpColIdxInput[perhash->numhashGrpCols] = i;
1666 : 627 : perhash->numhashGrpCols++;
1667 : : }
1668 : :
1669 : : /* and build a tuple descriptor for the hashtable */
1670 [ + + ]: 10097 : for (i = 0; i < perhash->numhashGrpCols; i++)
1671 : : {
1672 : 6341 : int varNumber = perhash->hashGrpColIdxInput[i] - 1;
1673 : :
1674 : 6341 : hashTlist = lappend(hashTlist, list_nth(outerTlist, varNumber));
1675 : 6341 : perhash->largestGrpColIdx =
1676 : 6341 : Max(varNumber + 1, perhash->largestGrpColIdx);
1677 : : }
1678 : :
2533 andres@anarazel.de 1679 : 3756 : hashDesc = ExecTypeFromTL(hashTlist);
1680 : :
2811 1681 : 3756 : execTuplesHashPrepare(perhash->numCols,
1682 : 3756 : perhash->aggnode->grpOperators,
1683 : : &perhash->eqfuncoids,
1684 : : &perhash->hashfunctions);
2810 1685 : 3756 : perhash->hashslot =
2538 1686 : 3756 : ExecAllocTableSlot(&estate->es_tupleTable, hashDesc,
1687 : : &TTSOpsMinimalTuple);
1688 : :
3136 rhodiumtoad@postgres 1689 : 3756 : list_free(hashTlist);
1690 : 3756 : bms_free(colnos);
1691 : : }
1692 : :
1693 : 3499 : bms_free(base_colnos);
8391 tgl@sss.pgh.pa.us 1694 : 3499 : }
1695 : :
1696 : : /*
1697 : : * Estimate per-hash-table-entry overhead.
1698 : : */
1699 : : Size
2033 jdavis@postgresql.or 1700 : 20735 : hash_agg_entry_size(int numTrans, Size tupleWidth, Size transitionSpace)
1701 : : {
1702 : : Size tupleChunkSize;
1703 : : Size pergroupChunkSize;
1704 : : Size transitionChunkSize;
1992 tgl@sss.pgh.pa.us 1705 : 20735 : Size tupleSize = (MAXALIGN(SizeofMinimalTupleHeader) +
1706 : : tupleWidth);
1707 : 20735 : Size pergroupSize = numTrans * sizeof(AggStatePerGroupData);
1708 : :
1709 : : /*
1710 : : * Entries use the Bump allocator, so the chunk sizes are the same as the
1711 : : * requested sizes.
1712 : : */
217 jdavis@postgresql.or 1713 : 20735 : tupleChunkSize = MAXALIGN(tupleSize);
1714 : 20735 : pergroupChunkSize = pergroupSize;
1715 : :
1716 : : /*
1717 : : * Transition values use AllocSet, which has a chunk header and also uses
1718 : : * power-of-two allocations.
1719 : : */
2033 1720 [ + + ]: 20735 : if (transitionSpace > 0)
217 1721 : 2699 : transitionChunkSize = CHUNKHDRSZ + pg_nextpower2_size_t(transitionSpace);
1722 : : else
2033 1723 : 18036 : transitionChunkSize = 0;
1724 : :
1725 : : return
217 1726 : 20735 : TupleHashEntrySize() +
2033 1727 : 20735 : tupleChunkSize +
1728 : 20735 : pergroupChunkSize +
1729 : : transitionChunkSize;
1730 : : }
1731 : :
1732 : : /*
1733 : : * hashagg_recompile_expressions()
1734 : : *
1735 : : * Identifies the right phase, compiles the right expression given the
1736 : : * arguments, and then sets phase->evalfunc to that expression.
1737 : : *
1738 : : * Different versions of the compiled expression are needed depending on
1739 : : * whether hash aggregation has spilled or not, and whether it's reading from
1740 : : * the outer plan or a tape. Before spilling to disk, the expression reads
1741 : : * from the outer plan and does not need to perform a NULL check. After
1742 : : * HashAgg begins to spill, new groups will not be created in the hash table,
1743 : : * and the AggStatePerGroup array may be NULL; therefore we need to add a null
1744 : : * pointer check to the expression. Then, when reading spilled data from a
1745 : : * tape, we change the outer slot type to be a fixed minimal tuple slot.
1746 : : *
1747 : : * It would be wasteful to recompile every time, so cache the compiled
1748 : : * expressions in the AggStatePerPhase, and reuse when appropriate.
1749 : : */
1750 : : static void
2049 1751 : 32892 : hashagg_recompile_expressions(AggState *aggstate, bool minslot, bool nullcheck)
1752 : : {
1753 : : AggStatePerPhase phase;
1992 tgl@sss.pgh.pa.us 1754 : 32892 : int i = minslot ? 1 : 0;
1755 : 32892 : int j = nullcheck ? 1 : 0;
1756 : :
2049 jdavis@postgresql.or 1757 [ + + - + ]: 32892 : Assert(aggstate->aggstrategy == AGG_HASHED ||
1758 : : aggstate->aggstrategy == AGG_MIXED);
1759 : :
1760 [ + + ]: 32892 : if (aggstate->aggstrategy == AGG_HASHED)
1761 : 6606 : phase = &aggstate->phases[0];
1762 : : else /* AGG_MIXED */
1763 : 26286 : phase = &aggstate->phases[1];
1764 : :
1765 [ + + ]: 32892 : if (phase->evaltrans_cache[i][j] == NULL)
1766 : : {
1992 tgl@sss.pgh.pa.us 1767 : 42 : const TupleTableSlotOps *outerops = aggstate->ss.ps.outerops;
1768 : 42 : bool outerfixed = aggstate->ss.ps.outeropsfixed;
1769 : 42 : bool dohash = true;
1766 jdavis@postgresql.or 1770 : 42 : bool dosort = false;
1771 : :
1772 : : /*
1773 : : * If minslot is true, that means we are processing a spilled batch
1774 : : * (inside agg_refill_hash_table()), and we must not advance the
1775 : : * sorted grouping sets.
1776 : : */
1777 [ + + + + ]: 42 : if (aggstate->aggstrategy == AGG_MIXED && !minslot)
1778 : 6 : dosort = true;
1779 : :
1780 : : /* temporarily change the outerops while compiling the expression */
2049 1781 [ + + ]: 42 : if (minslot)
1782 : : {
1783 : 21 : aggstate->ss.ps.outerops = &TTSOpsMinimalTuple;
1784 : 21 : aggstate->ss.ps.outeropsfixed = true;
1785 : : }
1786 : :
1992 tgl@sss.pgh.pa.us 1787 : 42 : phase->evaltrans_cache[i][j] = ExecBuildAggTrans(aggstate, phase,
1788 : : dosort, dohash,
1789 : : nullcheck);
1790 : :
1791 : : /* change back */
2049 jdavis@postgresql.or 1792 : 42 : aggstate->ss.ps.outerops = outerops;
1793 : 42 : aggstate->ss.ps.outeropsfixed = outerfixed;
1794 : : }
1795 : :
1796 : 32892 : phase->evaltrans = phase->evaltrans_cache[i][j];
1797 : 32892 : }
1798 : :
1799 : : /*
1800 : : * Set limits that trigger spilling to avoid exceeding hash_mem. Consider the
1801 : : * number of partitions we expect to create (if we do spill).
1802 : : *
1803 : : * There are two limits: a memory limit, and also an ngroups limit. The
1804 : : * ngroups limit becomes important when we expect transition values to grow
1805 : : * substantially larger than the initial value.
1806 : : */
1807 : : void
1917 1808 : 32985 : hash_agg_set_limits(double hashentrysize, double input_groups, int used_bits,
1809 : : Size *mem_limit, uint64 *ngroups_limit,
1810 : : int *num_partitions)
1811 : : {
1812 : : int npartitions;
1813 : : Size partition_mem;
1555 tgl@sss.pgh.pa.us 1814 : 32985 : Size hash_mem_limit = get_hash_memory_limit();
1815 : :
1816 : : /* if not expected to spill, use all of hash_mem */
1817 [ + + ]: 32985 : if (input_groups * hashentrysize <= hash_mem_limit)
1818 : : {
2039 jdavis@postgresql.or 1819 [ + + ]: 31770 : if (num_partitions != NULL)
1820 : 19470 : *num_partitions = 0;
1555 tgl@sss.pgh.pa.us 1821 : 31770 : *mem_limit = hash_mem_limit;
1822 : 31770 : *ngroups_limit = hash_mem_limit / hashentrysize;
2049 jdavis@postgresql.or 1823 : 31770 : return;
1824 : : }
1825 : :
1826 : : /*
1827 : : * Calculate expected memory requirements for spilling, which is the size
1828 : : * of the buffers needed for all the tapes that need to be open at once.
1829 : : * Then, subtract that from the memory available for holding hash tables.
1830 : : */
1831 : 1215 : npartitions = hash_choose_num_partitions(input_groups,
1832 : : hashentrysize,
1833 : : used_bits,
1834 : : NULL);
1835 [ + + ]: 1215 : if (num_partitions != NULL)
1836 : 48 : *num_partitions = npartitions;
1837 : :
1838 : 1215 : partition_mem =
1839 : 1215 : HASHAGG_READ_BUFFER_SIZE +
1840 : : HASHAGG_WRITE_BUFFER_SIZE * npartitions;
1841 : :
1842 : : /*
1843 : : * Don't set the limit below 3/4 of hash_mem. In that case, we are at the
1844 : : * minimum number of partitions, so we aren't going to dramatically exceed
1845 : : * work mem anyway.
1846 : : */
1555 tgl@sss.pgh.pa.us 1847 [ - + ]: 1215 : if (hash_mem_limit > 4 * partition_mem)
1555 tgl@sss.pgh.pa.us 1848 :UBC 0 : *mem_limit = hash_mem_limit - partition_mem;
1849 : : else
1555 tgl@sss.pgh.pa.us 1850 :CBC 1215 : *mem_limit = hash_mem_limit * 0.75;
1851 : :
2049 jdavis@postgresql.or 1852 [ + - ]: 1215 : if (*mem_limit > hashentrysize)
1853 : 1215 : *ngroups_limit = *mem_limit / hashentrysize;
1854 : : else
2049 jdavis@postgresql.or 1855 :UBC 0 : *ngroups_limit = 1;
1856 : : }
1857 : :
1858 : : /*
1859 : : * hash_agg_check_limits
1860 : : *
1861 : : * After adding a new group to the hash table, check whether we need to enter
1862 : : * spill mode. Allocations may happen without adding new groups (for instance,
1863 : : * if the transition state size grows), so this check is imperfect.
1864 : : */
1865 : : static void
2049 jdavis@postgresql.or 1866 :CBC 258544 : hash_agg_check_limits(AggState *aggstate)
1867 : : {
1992 tgl@sss.pgh.pa.us 1868 : 258544 : uint64 ngroups = aggstate->hash_ngroups_current;
1869 : 258544 : Size meta_mem = MemoryContextMemAllocated(aggstate->hash_metacxt,
1870 : : true);
217 jdavis@postgresql.or 1871 : 258544 : Size entry_mem = MemoryContextMemAllocated(aggstate->hash_tablecxt,
1872 : : true);
1873 : 258544 : Size tval_mem = MemoryContextMemAllocated(aggstate->hashcontext->ecxt_per_tuple_memory,
1874 : : true);
1875 : 258544 : Size total_mem = meta_mem + entry_mem + tval_mem;
258 1876 : 258544 : bool do_spill = false;
1877 : :
1878 : : #ifdef USE_INJECTION_POINTS
1879 : : if (ngroups >= 1000)
1880 : : {
1881 : : if (IS_INJECTION_POINT_ATTACHED("hash-aggregate-spill-1000"))
1882 : : {
1883 : : do_spill = true;
1884 : : INJECTION_POINT_CACHED("hash-aggregate-spill-1000", NULL);
1885 : : }
1886 : : }
1887 : : #endif
1888 : :
1889 : : /*
1890 : : * Don't spill unless there's at least one group in the hash table so we
1891 : : * can be sure to make progress even in edge cases.
1892 : : */
2049 1893 [ + - ]: 258544 : if (aggstate->hash_ngroups_current > 0 &&
217 1894 [ + + ]: 258544 : (total_mem > aggstate->hash_mem_limit ||
2049 1895 [ + + ]: 245347 : ngroups > aggstate->hash_ngroups_limit))
1896 : : {
258 1897 : 13224 : do_spill = true;
1898 : : }
1899 : :
1900 [ + + ]: 258544 : if (do_spill)
1901 : 13224 : hash_agg_enter_spill_mode(aggstate);
2049 1902 : 258544 : }
1903 : :
1904 : : /*
1905 : : * Enter "spill mode", meaning that no new groups are added to any of the hash
1906 : : * tables. Tuples that would create a new group are instead spilled, and
1907 : : * processed later.
1908 : : */
1909 : : static void
1910 : 13224 : hash_agg_enter_spill_mode(AggState *aggstate)
1911 : : {
1912 : : INJECTION_POINT("hash-aggregate-enter-spill-mode", NULL);
1913 : 13224 : aggstate->hash_spill_mode = true;
1914 : 13224 : hashagg_recompile_expressions(aggstate, aggstate->table_filled, true);
1915 : :
1916 [ + + ]: 13224 : if (!aggstate->hash_ever_spilled)
1917 : : {
1470 heikki.linnakangas@i 1918 [ - + ]: 30 : Assert(aggstate->hash_tapeset == NULL);
2049 jdavis@postgresql.or 1919 [ - + ]: 30 : Assert(aggstate->hash_spills == NULL);
1920 : :
1921 : 30 : aggstate->hash_ever_spilled = true;
1922 : :
1470 heikki.linnakangas@i 1923 : 30 : aggstate->hash_tapeset = LogicalTapeSetCreate(true, NULL, -1);
1924 : :
1992 tgl@sss.pgh.pa.us 1925 : 30 : aggstate->hash_spills = palloc(sizeof(HashAggSpill) * aggstate->num_hashes);
1926 : :
2049 jdavis@postgresql.or 1927 [ + + ]: 90 : for (int setno = 0; setno < aggstate->num_hashes; setno++)
1928 : : {
1992 tgl@sss.pgh.pa.us 1929 : 60 : AggStatePerHash perhash = &aggstate->perhash[setno];
1930 : 60 : HashAggSpill *spill = &aggstate->hash_spills[setno];
1931 : :
1470 heikki.linnakangas@i 1932 : 60 : hashagg_spill_init(spill, aggstate->hash_tapeset, 0,
2049 jdavis@postgresql.or 1933 : 60 : perhash->aggnode->numGroups,
1934 : : aggstate->hashentrysize);
1935 : : }
1936 : : }
1937 : 13224 : }
1938 : :
1939 : : /*
1940 : : * Update metrics after filling the hash table.
1941 : : *
1942 : : * If reading from the outer plan, from_tape should be false; if reading from
1943 : : * another tape, from_tape should be true.
1944 : : */
1945 : : static void
1946 : 22120 : hash_agg_update_metrics(AggState *aggstate, bool from_tape, int npartitions)
1947 : : {
1948 : : Size meta_mem;
1949 : : Size entry_mem;
1950 : : Size hashkey_mem;
1951 : : Size buffer_mem;
1952 : : Size total_mem;
1953 : :
1954 [ + + ]: 22120 : if (aggstate->aggstrategy != AGG_MIXED &&
1955 [ - + ]: 8917 : aggstate->aggstrategy != AGG_HASHED)
2049 jdavis@postgresql.or 1956 :UBC 0 : return;
1957 : :
1958 : : /* memory for the hash table itself */
2049 jdavis@postgresql.or 1959 :CBC 22120 : meta_mem = MemoryContextMemAllocated(aggstate->hash_metacxt, true);
1960 : :
1961 : : /* memory for hash entries */
217 1962 : 22120 : entry_mem = MemoryContextMemAllocated(aggstate->hash_tablecxt, true);
1963 : :
1964 : : /* memory for byref transition states */
1928 pg@bowt.ie 1965 : 22120 : hashkey_mem = MemoryContextMemAllocated(aggstate->hashcontext->ecxt_per_tuple_memory, true);
1966 : :
1967 : : /* memory for read/write tape buffers, if spilled */
2049 jdavis@postgresql.or 1968 : 22120 : buffer_mem = npartitions * HASHAGG_WRITE_BUFFER_SIZE;
1969 [ + + ]: 22120 : if (from_tape)
1970 : 13467 : buffer_mem += HASHAGG_READ_BUFFER_SIZE;
1971 : :
1972 : : /* update peak mem */
217 1973 : 22120 : total_mem = meta_mem + entry_mem + hashkey_mem + buffer_mem;
2049 1974 [ + + ]: 22120 : if (total_mem > aggstate->hash_mem_peak)
1975 : 2476 : aggstate->hash_mem_peak = total_mem;
1976 : :
1977 : : /* update disk usage */
1470 heikki.linnakangas@i 1978 [ + + ]: 22120 : if (aggstate->hash_tapeset != NULL)
1979 : : {
1980 : 13497 : uint64 disk_used = LogicalTapeSetBlocks(aggstate->hash_tapeset) * (BLCKSZ / 1024);
1981 : :
2049 jdavis@postgresql.or 1982 [ + + ]: 13497 : if (aggstate->hash_disk_used < disk_used)
1983 : 24 : aggstate->hash_disk_used = disk_used;
1984 : : }
1985 : :
1986 : : /* update hashentrysize estimate based on contents */
1987 [ + + ]: 22120 : if (aggstate->hash_ngroups_current > 0)
1988 : : {
1989 : 21904 : aggstate->hashentrysize =
217 1990 : 21904 : TupleHashEntrySize() +
1928 pg@bowt.ie 1991 : 21904 : (hashkey_mem / (double) aggstate->hash_ngroups_current);
1992 : : }
1993 : : }
1994 : :
1995 : : /*
1996 : : * Create memory contexts used for hash aggregation.
1997 : : */
1998 : : static void
217 jdavis@postgresql.or 1999 : 3499 : hash_create_memory(AggState *aggstate)
2000 : : {
2001 : 3499 : Size maxBlockSize = ALLOCSET_DEFAULT_MAXSIZE;
2002 : :
2003 : : /*
2004 : : * The hashcontext's per-tuple memory will be used for byref transition
2005 : : * values and returned by AggCheckCallContext().
2006 : : */
2007 : 3499 : aggstate->hashcontext = CreateWorkExprContext(aggstate->ss.ps.state);
2008 : :
2009 : : /*
2010 : : * The meta context will be used for the bucket array of
2011 : : * TupleHashEntryData (or arrays, in the case of grouping sets). As the
2012 : : * hash table grows, the bucket array will double in size and the old one
2013 : : * will be freed, so an AllocSet is appropriate. For large bucket arrays,
2014 : : * the large allocation path will be used, so it's not worth worrying
2015 : : * about wasting space due to power-of-two allocations.
2016 : : */
2017 : 3499 : aggstate->hash_metacxt = AllocSetContextCreate(aggstate->ss.ps.state->es_query_cxt,
2018 : : "HashAgg meta context",
2019 : : ALLOCSET_DEFAULT_SIZES);
2020 : :
2021 : : /*
2022 : : * The hash entries themselves, which include the grouping key
2023 : : * (firstTuple) and pergroup data, are stored in the table context. The
2024 : : * bump allocator can be used because the entries are not freed until the
2025 : : * entire hash table is reset. The bump allocator is faster for
2026 : : * allocations and avoids wasting space on the chunk header or
2027 : : * power-of-two allocations.
2028 : : *
2029 : : * Like CreateWorkExprContext(), use smaller sizings for smaller work_mem,
2030 : : * to avoid large jumps in memory usage.
2031 : : */
2032 : :
2033 : : /*
2034 : : * Like CreateWorkExprContext(), use smaller sizings for smaller work_mem,
2035 : : * to avoid large jumps in memory usage.
2036 : : */
2037 : 3499 : maxBlockSize = pg_prevpower2_size_t(work_mem * (Size) 1024 / 16);
2038 : :
2039 : : /* But no bigger than ALLOCSET_DEFAULT_MAXSIZE */
2040 : 3499 : maxBlockSize = Min(maxBlockSize, ALLOCSET_DEFAULT_MAXSIZE);
2041 : :
2042 : : /* and no smaller than ALLOCSET_DEFAULT_INITSIZE */
2043 : 3499 : maxBlockSize = Max(maxBlockSize, ALLOCSET_DEFAULT_INITSIZE);
2044 : :
2045 : 3499 : aggstate->hash_tablecxt = BumpContextCreate(aggstate->ss.ps.state->es_query_cxt,
2046 : : "HashAgg table context",
2047 : : ALLOCSET_DEFAULT_MINSIZE,
2048 : : ALLOCSET_DEFAULT_INITSIZE,
2049 : : maxBlockSize);
2050 : :
2051 : 3499 : }
2052 : :
2053 : : /*
2054 : : * Choose a reasonable number of buckets for the initial hash table size.
2055 : : */
2056 : : static long
2049 2057 : 2718 : hash_choose_num_buckets(double hashentrysize, long ngroups, Size memory)
2058 : : {
2059 : : long max_nbuckets;
1992 tgl@sss.pgh.pa.us 2060 : 2718 : long nbuckets = ngroups;
2061 : :
2049 jdavis@postgresql.or 2062 : 2718 : max_nbuckets = memory / hashentrysize;
2063 : :
2064 : : /*
2065 : : * Underestimating is better than overestimating. Too many buckets crowd
2066 : : * out space for group keys and transition state values.
2067 : : */
2044 2068 : 2718 : max_nbuckets >>= 1;
2069 : :
2049 2070 [ + + ]: 2718 : if (nbuckets > max_nbuckets)
2071 : 36 : nbuckets = max_nbuckets;
2072 : :
1967 2073 : 2718 : return Max(nbuckets, 1);
2074 : : }
2075 : :
2076 : : /*
2077 : : * Determine the number of partitions to create when spilling, which will
2078 : : * always be a power of two. If log2_npartitions is non-NULL, set
2079 : : * *log2_npartitions to the log2() of the number of partitions.
2080 : : */
2081 : : static int
1917 2082 : 7521 : hash_choose_num_partitions(double input_groups, double hashentrysize,
2083 : : int used_bits, int *log2_npartitions)
2084 : : {
1555 tgl@sss.pgh.pa.us 2085 : 7521 : Size hash_mem_limit = get_hash_memory_limit();
2086 : : double partition_limit;
2087 : : double mem_wanted;
2088 : : double dpartitions;
2089 : : int npartitions;
2090 : : int partition_bits;
2091 : :
2092 : : /*
2093 : : * Avoid creating so many partitions that the memory requirements of the
2094 : : * open partition files are greater than 1/4 of hash_mem.
2095 : : */
2049 jdavis@postgresql.or 2096 : 7521 : partition_limit =
1555 tgl@sss.pgh.pa.us 2097 : 7521 : (hash_mem_limit * 0.25 - HASHAGG_READ_BUFFER_SIZE) /
2098 : : HASHAGG_WRITE_BUFFER_SIZE;
2099 : :
2049 jdavis@postgresql.or 2100 : 7521 : mem_wanted = HASHAGG_PARTITION_FACTOR * input_groups * hashentrysize;
2101 : :
2102 : : /* make enough partitions so that each one is likely to fit in memory */
1555 tgl@sss.pgh.pa.us 2103 : 7521 : dpartitions = 1 + (mem_wanted / hash_mem_limit);
2104 : :
2105 [ + + ]: 7521 : if (dpartitions > partition_limit)
2106 : 7488 : dpartitions = partition_limit;
2107 : :
2108 [ + - ]: 7521 : if (dpartitions < HASHAGG_MIN_PARTITIONS)
2109 : 7521 : dpartitions = HASHAGG_MIN_PARTITIONS;
2110 [ - + ]: 7521 : if (dpartitions > HASHAGG_MAX_PARTITIONS)
1555 tgl@sss.pgh.pa.us 2111 :UBC 0 : dpartitions = HASHAGG_MAX_PARTITIONS;
2112 : :
2113 : : /* HASHAGG_MAX_PARTITIONS limit makes this safe */
1555 tgl@sss.pgh.pa.us 2114 :CBC 7521 : npartitions = (int) dpartitions;
2115 : :
2116 : : /* ceil(log2(npartitions)) */
47 michael@paquier.xyz 2117 :GNC 7521 : partition_bits = pg_ceil_log2_32(npartitions);
2118 : :
2119 : : /* make sure that we don't exhaust the hash bits */
2049 jdavis@postgresql.or 2120 [ - + ]:CBC 7521 : if (partition_bits + used_bits >= 32)
2049 jdavis@postgresql.or 2121 :UBC 0 : partition_bits = 32 - used_bits;
2122 : :
2049 jdavis@postgresql.or 2123 [ + + ]:CBC 7521 : if (log2_npartitions != NULL)
2124 : 6306 : *log2_npartitions = partition_bits;
2125 : :
2126 : : /* number of partitions will be a power of two */
1555 tgl@sss.pgh.pa.us 2127 : 7521 : npartitions = 1 << partition_bits;
2128 : :
2049 jdavis@postgresql.or 2129 : 7521 : return npartitions;
2130 : : }
2131 : :
2132 : : /*
2133 : : * Initialize a freshly-created TupleHashEntry.
2134 : : */
2135 : : static void
1919 2136 : 258544 : initialize_hash_entry(AggState *aggstate, TupleHashTable hashtable,
2137 : : TupleHashEntry entry)
2138 : : {
2139 : : AggStatePerGroup pergroup;
2140 : : int transno;
2141 : :
2142 : 258544 : aggstate->hash_ngroups_current++;
2143 : 258544 : hash_agg_check_limits(aggstate);
2144 : :
2145 : : /* no need to allocate or initialize per-group state */
2146 [ + + ]: 258544 : if (aggstate->numtrans == 0)
2147 : 100648 : return;
2148 : :
217 2149 : 157896 : pergroup = (AggStatePerGroup) TupleHashEntryGetAdditional(hashtable, entry);
2150 : :
2151 : : /*
2152 : : * Initialize aggregates for new tuple group, lookup_hash_entries()
2153 : : * already has selected the relevant grouping set.
2154 : : */
1919 2155 [ + + ]: 390775 : for (transno = 0; transno < aggstate->numtrans; transno++)
2156 : : {
2157 : 232879 : AggStatePerTrans pertrans = &aggstate->pertrans[transno];
2158 : 232879 : AggStatePerGroup pergroupstate = &pergroup[transno];
2159 : :
2160 : 232879 : initialize_aggregate(aggstate, pertrans, pergroupstate);
2161 : : }
2162 : : }
2163 : :
2164 : : /*
2165 : : * Look up hash entries for the current tuple in all hashed grouping sets.
2166 : : *
2167 : : * Some entries may be left NULL if we are in "spill mode". The same tuple
2168 : : * will belong to different groups for each grouping set, so may match a group
2169 : : * already in memory for one set and match a group not in memory for another
2170 : : * set. When in "spill mode", the tuple will be spilled for each grouping set
2171 : : * where it doesn't match a group in memory.
2172 : : *
2173 : : * NB: It's possible to spill the same tuple for several different grouping
2174 : : * sets. This may seem wasteful, but it's actually a trade-off: if we spill
2175 : : * the tuple multiple times for multiple grouping sets, it can be partitioned
2176 : : * for each grouping set, making the refilling of the hash table very
2177 : : * efficient.
2178 : : */
2179 : : static void
3136 rhodiumtoad@postgres 2180 : 3452507 : lookup_hash_entries(AggState *aggstate)
2181 : : {
2182 : 3452507 : AggStatePerGroup *pergroup = aggstate->hash_pergroup;
1919 jdavis@postgresql.or 2183 : 3452507 : TupleTableSlot *outerslot = aggstate->tmpcontext->ecxt_outertuple;
2184 : : int setno;
2185 : :
2049 2186 [ + + ]: 6972240 : for (setno = 0; setno < aggstate->num_hashes; setno++)
2187 : : {
1992 tgl@sss.pgh.pa.us 2188 : 3519733 : AggStatePerHash perhash = &aggstate->perhash[setno];
1919 jdavis@postgresql.or 2189 : 3519733 : TupleHashTable hashtable = perhash->hashtable;
2190 : 3519733 : TupleTableSlot *hashslot = perhash->hashslot;
2191 : : TupleHashEntry entry;
2192 : : uint32 hash;
2193 : 3519733 : bool isnew = false;
2194 : : bool *p_isnew;
2195 : :
2196 : : /* if hash table already spilled, don't create new entries */
2197 [ + + ]: 3519733 : p_isnew = aggstate->hash_spill_mode ? NULL : &isnew;
2198 : :
3136 rhodiumtoad@postgres 2199 : 3519733 : select_current_set(aggstate, setno, true);
1919 jdavis@postgresql.or 2200 : 3519733 : prepare_hash_slot(perhash,
2201 : : outerslot,
2202 : : hashslot);
2203 : :
2204 : 3519733 : entry = LookupTupleHashEntry(hashtable, hashslot,
2205 : : p_isnew, &hash);
2206 : :
2207 [ + + ]: 3519733 : if (entry != NULL)
2208 : : {
2209 [ + + ]: 3139267 : if (isnew)
2210 : 184342 : initialize_hash_entry(aggstate, hashtable, entry);
217 2211 : 3139267 : pergroup[setno] = TupleHashEntryGetAdditional(hashtable, entry);
2212 : : }
2213 : : else
2214 : : {
1992 tgl@sss.pgh.pa.us 2215 : 380466 : HashAggSpill *spill = &aggstate->hash_spills[setno];
2216 : 380466 : TupleTableSlot *slot = aggstate->tmpcontext->ecxt_outertuple;
2217 : :
2049 jdavis@postgresql.or 2218 [ - + ]: 380466 : if (spill->partitions == NULL)
1470 heikki.linnakangas@i 2219 :UBC 0 : hashagg_spill_init(spill, aggstate->hash_tapeset, 0,
2049 jdavis@postgresql.or 2220 : 0 : perhash->aggnode->numGroups,
2221 : : aggstate->hashentrysize);
2222 : :
1933 jdavis@postgresql.or 2223 :CBC 380466 : hashagg_spill_tuple(aggstate, spill, slot, hash);
1919 2224 : 380466 : pergroup[setno] = NULL;
2225 : : }
2226 : : }
3136 rhodiumtoad@postgres 2227 : 3452507 : }
2228 : :
2229 : : /*
2230 : : * ExecAgg -
2231 : : *
2232 : : * ExecAgg receives tuples from its outer subplan and aggregates over
2233 : : * the appropriate attribute for each aggregate function use (Aggref
2234 : : * node) appearing in the targetlist or qual of the node. The number
2235 : : * of tuples to aggregate over depends on whether grouped or plain
2236 : : * aggregation is selected. In grouped aggregation, we produce a result
2237 : : * row for each group; in plain aggregation there's a single result row
2238 : : * for the whole query. In either case, the value of each aggregate is
2239 : : * stored in the expression context to be used when ExecProject evaluates
2240 : : * the result tuple.
2241 : : */
2242 : : static TupleTableSlot *
3024 andres@anarazel.de 2243 : 404080 : ExecAgg(PlanState *pstate)
2244 : : {
2245 : 404080 : AggState *node = castNode(AggState, pstate);
3136 rhodiumtoad@postgres 2246 : 404080 : TupleTableSlot *result = NULL;
2247 : :
3016 andres@anarazel.de 2248 [ + + ]: 404080 : CHECK_FOR_INTERRUPTS();
2249 : :
3817 2250 [ + + ]: 404080 : if (!node->agg_done)
2251 : : {
2252 : : /* Dispatch based on strategy */
3136 rhodiumtoad@postgres 2253 [ + + + - ]: 372571 : switch (node->phase->aggstrategy)
2254 : : {
3817 andres@anarazel.de 2255 : 236939 : case AGG_HASHED:
2256 [ + + ]: 236939 : if (!node->table_filled)
2257 : 8581 : agg_fill_hash_table(node);
2258 : : /* FALLTHROUGH */
2259 : : case AGG_MIXED:
2260 : 250620 : result = agg_retrieve_hash_table(node);
2261 : 250620 : break;
3136 rhodiumtoad@postgres 2262 : 121951 : case AGG_PLAIN:
2263 : : case AGG_SORTED:
3817 andres@anarazel.de 2264 : 121951 : result = agg_retrieve_direct(node);
2265 : 121860 : break;
2266 : : }
2267 : :
2268 [ + + + - ]: 372480 : if (!TupIsNull(result))
2269 : 363064 : return result;
2270 : : }
2271 : :
2272 : 40925 : return NULL;
2273 : : }
2274 : :
2275 : : /*
2276 : : * ExecAgg for non-hashed case
2277 : : */
2278 : : static TupleTableSlot *
8362 tgl@sss.pgh.pa.us 2279 : 121951 : agg_retrieve_direct(AggState *aggstate)
2280 : : {
3817 andres@anarazel.de 2281 : 121951 : Agg *node = aggstate->phase->aggnode;
2282 : : ExprContext *econtext;
2283 : : ExprContext *tmpcontext;
2284 : : AggStatePerAgg peragg;
2285 : : AggStatePerGroup *pergroups;
2286 : : TupleTableSlot *outerslot;
2287 : : TupleTableSlot *firstSlot;
2288 : : TupleTableSlot *result;
2289 : 121951 : bool hasGroupingSets = aggstate->phase->numsets > 0;
2290 : 121951 : int numGroupingSets = Max(aggstate->phase->numsets, 1);
2291 : : int currentSet;
2292 : : int nextSetSize;
2293 : : int numReset;
2294 : : int i;
2295 : :
2296 : : /*
2297 : : * get state info from node
2298 : : *
2299 : : * econtext is the per-output-tuple expression context
2300 : : *
2301 : : * tmpcontext is the per-input-tuple expression context
2302 : : */
8362 tgl@sss.pgh.pa.us 2303 : 121951 : econtext = aggstate->ss.ps.ps_ExprContext;
8391 2304 : 121951 : tmpcontext = aggstate->tmpcontext;
2305 : :
9528 2306 : 121951 : peragg = aggstate->peragg;
2855 andres@anarazel.de 2307 : 121951 : pergroups = aggstate->pergroups;
8362 tgl@sss.pgh.pa.us 2308 : 121951 : firstSlot = aggstate->ss.ss_ScanTupleSlot;
2309 : :
2310 : : /*
2311 : : * We loop retrieving groups until we find one matching
2312 : : * aggstate->ss.ps.qual
2313 : : *
2314 : : * For grouping sets, we have the invariant that aggstate->projected_set
2315 : : * is either -1 (initial call) or the index (starting from 0) in
2316 : : * gset_lengths for the group we just completed (either by projecting a
2317 : : * row or by discarding it in the qual).
2318 : : */
7779 2319 [ + + ]: 157367 : while (!aggstate->agg_done)
2320 : : {
2321 : : /*
2322 : : * Clear the per-output-tuple context for each group, as well as
2323 : : * aggcontext (which contains any pass-by-ref transvalues of the old
2324 : : * group). Some aggregate functions store working state in child
2325 : : * contexts; those now get reset automatically without us needing to
2326 : : * do anything special.
2327 : : *
2328 : : * We use ReScanExprContext not just ResetExprContext because we want
2329 : : * any registered shutdown callbacks to be called. That allows
2330 : : * aggregate functions to ensure they've cleaned up any non-memory
2331 : : * resources.
2332 : : */
3817 andres@anarazel.de 2333 : 157262 : ReScanExprContext(econtext);
2334 : :
2335 : : /*
2336 : : * Determine how many grouping sets need to be reset at this boundary.
2337 : : */
2338 [ + + ]: 157262 : if (aggstate->projected_set >= 0 &&
2339 [ + + ]: 123798 : aggstate->projected_set < numGroupingSets)
2340 : 123789 : numReset = aggstate->projected_set + 1;
2341 : : else
2342 : 33473 : numReset = numGroupingSets;
2343 : :
2344 : : /*
2345 : : * numReset can change on a phase boundary, but that's OK; we want to
2346 : : * reset the contexts used in _this_ phase, and later, after possibly
2347 : : * changing phase, initialize the right number of aggregates for the
2348 : : * _new_ phase.
2349 : : */
2350 : :
2351 [ + + ]: 325663 : for (i = 0; i < numReset; i++)
2352 : : {
2353 : 168401 : ReScanExprContext(aggstate->aggcontexts[i]);
2354 : : }
2355 : :
2356 : : /*
2357 : : * Check if input is complete and there are no more groups to project
2358 : : * in this phase; move to next phase or mark as done.
2359 : : */
2360 [ + + ]: 157262 : if (aggstate->input_done == true &&
2361 [ + + ]: 807 : aggstate->projected_set >= (numGroupingSets - 1))
2362 : : {
2363 [ + + ]: 399 : if (aggstate->current_phase < aggstate->numphases - 1)
2364 : : {
2365 : 102 : initialize_phase(aggstate, aggstate->current_phase + 1);
2366 : 102 : aggstate->input_done = false;
2367 : 102 : aggstate->projected_set = -1;
2368 : 102 : numGroupingSets = Max(aggstate->phase->numsets, 1);
2369 : 102 : node = aggstate->phase->aggnode;
2370 : 102 : numReset = numGroupingSets;
2371 : : }
3136 rhodiumtoad@postgres 2372 [ + + ]: 297 : else if (aggstate->aggstrategy == AGG_MIXED)
2373 : : {
2374 : : /*
2375 : : * Mixed mode; we've output all the grouped stuff and have
2376 : : * full hashtables, so switch to outputting those.
2377 : : */
2378 : 78 : initialize_phase(aggstate, 0);
2379 : 78 : aggstate->table_filled = true;
2380 : 78 : ResetTupleHashIterator(aggstate->perhash[0].hashtable,
2381 : : &aggstate->perhash[0].hashiter);
2382 : 78 : select_current_set(aggstate, 0, true);
2383 : 78 : return agg_retrieve_hash_table(aggstate);
2384 : : }
2385 : : else
2386 : : {
8391 tgl@sss.pgh.pa.us 2387 : 219 : aggstate->agg_done = true;
3817 andres@anarazel.de 2388 : 219 : break;
2389 : : }
2390 : : }
2391 : :
2392 : : /*
2393 : : * Get the number of columns in the next grouping set after the last
2394 : : * projected one (if any). This is the number of columns to compare to
2395 : : * see if we reached the boundary of that set too.
2396 : : */
2397 [ + + ]: 156965 : if (aggstate->projected_set >= 0 &&
2398 [ + + ]: 123399 : aggstate->projected_set < (numGroupingSets - 1))
2399 : 13647 : nextSetSize = aggstate->phase->gset_lengths[aggstate->projected_set + 1];
2400 : : else
2401 : 143318 : nextSetSize = 0;
2402 : :
2403 : : /*----------
2404 : : * If a subgroup for the current grouping set is present, project it.
2405 : : *
2406 : : * We have a new group if:
2407 : : * - we're out of input but haven't projected all grouping sets
2408 : : * (checked above)
2409 : : * OR
2410 : : * - we already projected a row that wasn't from the last grouping
2411 : : * set
2412 : : * AND
2413 : : * - the next grouping set has at least one grouping column (since
2414 : : * empty grouping sets project only once input is exhausted)
2415 : : * AND
2416 : : * - the previous and pending rows differ on the grouping columns
2417 : : * of the next grouping set
2418 : : *----------
2419 : : */
2811 2420 : 156965 : tmpcontext->ecxt_innertuple = econtext->ecxt_outertuple;
3817 2421 [ + + ]: 156965 : if (aggstate->input_done ||
3136 rhodiumtoad@postgres 2422 [ + + ]: 156557 : (node->aggstrategy != AGG_PLAIN &&
3817 andres@anarazel.de 2423 [ + + ]: 123913 : aggstate->projected_set != -1 &&
2424 [ + + + + ]: 122991 : aggstate->projected_set < (numGroupingSets - 1) &&
2425 : 9973 : nextSetSize > 0 &&
2811 2426 [ + + ]: 9973 : !ExecQualAndReset(aggstate->phase->eqfunctions[nextSetSize - 1],
2427 : : tmpcontext)))
2428 : : {
3817 2429 : 7075 : aggstate->projected_set += 1;
2430 : :
2431 [ - + ]: 7075 : Assert(aggstate->projected_set < numGroupingSets);
2432 [ + + - + ]: 7075 : Assert(nextSetSize > 0 || aggstate->input_done);
2433 : : }
2434 : : else
2435 : : {
2436 : : /*
2437 : : * We no longer care what group we just projected, the next
2438 : : * projection will always be the first (or only) grouping set
2439 : : * (unless the input proves to be empty).
2440 : : */
2441 : 149890 : aggstate->projected_set = 0;
2442 : :
2443 : : /*
2444 : : * If we don't already have the first tuple of the new group,
2445 : : * fetch it from the outer plan.
2446 : : */
2447 [ + + ]: 149890 : if (aggstate->grp_firstTuple == NULL)
2448 : : {
2449 : 33566 : outerslot = fetch_input_tuple(aggstate);
2450 [ + + + + ]: 33536 : if (!TupIsNull(outerslot))
2451 : : {
2452 : : /*
2453 : : * Make a copy of the first input tuple; we will use this
2454 : : * for comparisons (in group mode) and for projection.
2455 : : */
2537 2456 : 27117 : aggstate->grp_firstTuple = ExecCopySlotHeapTuple(outerslot);
2457 : : }
2458 : : else
2459 : : {
2460 : : /* outer plan produced no tuples at all */
3817 2461 [ + + ]: 6419 : if (hasGroupingSets)
2462 : : {
2463 : : /*
2464 : : * If there was no input at all, we need to project
2465 : : * rows only if there are grouping sets of size 0.
2466 : : * Note that this implies that there can't be any
2467 : : * references to ungrouped Vars, which would otherwise
2468 : : * cause issues with the empty output slot.
2469 : : *
2470 : : * XXX: This is no longer true, we currently deal with
2471 : : * this in finalize_aggregates().
2472 : : */
2473 : 39 : aggstate->input_done = true;
2474 : :
2475 [ + + ]: 54 : while (aggstate->phase->gset_lengths[aggstate->projected_set] > 0)
2476 : : {
2477 : 24 : aggstate->projected_set += 1;
2478 [ + + ]: 24 : if (aggstate->projected_set >= numGroupingSets)
2479 : : {
2480 : : /*
2481 : : * We can't set agg_done here because we might
2482 : : * have more phases to do, even though the
2483 : : * input is empty. So we need to restart the
2484 : : * whole outer loop.
2485 : : */
2486 : 9 : break;
2487 : : }
2488 : : }
2489 : :
2490 [ + + ]: 39 : if (aggstate->projected_set >= numGroupingSets)
2491 : 9 : continue;
2492 : : }
2493 : : else
2494 : : {
2495 : 6380 : aggstate->agg_done = true;
2496 : : /* If we are grouping, we should produce no tuples too */
2497 [ + + ]: 6380 : if (node->aggstrategy != AGG_PLAIN)
2498 : 38 : return NULL;
2499 : : }
2500 : : }
2501 : : }
2502 : :
2503 : : /*
2504 : : * Initialize working state for a new input tuple group.
2505 : : */
2855 2506 : 149813 : initialize_aggregates(aggstate, pergroups, numReset);
2507 : :
3817 2508 [ + + ]: 149813 : if (aggstate->grp_firstTuple != NULL)
2509 : : {
2510 : : /*
2511 : : * Store the copied first input tuple in the tuple table slot
2512 : : * reserved for it. The tuple will be deleted when it is
2513 : : * cleared from the slot.
2514 : : */
2537 2515 : 143441 : ExecForceStoreHeapTuple(aggstate->grp_firstTuple,
2516 : : firstSlot, true);
3050 tgl@sss.pgh.pa.us 2517 : 143441 : aggstate->grp_firstTuple = NULL; /* don't keep two pointers */
2518 : :
2519 : : /* set up for first advance_aggregates call */
3817 andres@anarazel.de 2520 : 143441 : tmpcontext->ecxt_outertuple = firstSlot;
2521 : :
2522 : : /*
2523 : : * Process each outer-plan tuple, and then fetch the next one,
2524 : : * until we exhaust the outer plan or cross a group boundary.
2525 : : */
2526 : : for (;;)
2527 : : {
2528 : : /*
2529 : : * During phase 1 only of a mixed agg, we need to update
2530 : : * hashtables as well in advance_aggregates.
2531 : : */
3136 rhodiumtoad@postgres 2532 [ + + ]: 10963624 : if (aggstate->aggstrategy == AGG_MIXED &&
2533 [ + - ]: 19031 : aggstate->current_phase == 1)
2534 : : {
2848 andres@anarazel.de 2535 : 19031 : lookup_hash_entries(aggstate);
2536 : : }
2537 : :
2538 : : /* Advance the aggregates (or combine functions) */
2539 : 10963624 : advance_aggregates(aggstate);
2540 : :
2541 : : /* Reset per-input-tuple context after each tuple */
3817 2542 : 10963585 : ResetExprContext(tmpcontext);
2543 : :
2544 : 10963585 : outerslot = fetch_input_tuple(aggstate);
2545 [ + + + + ]: 10963575 : if (TupIsNull(outerslot))
2546 : : {
2547 : : /* no more outer-plan tuples available */
2548 : :
2549 : : /* if we built hash tables, finalize any spills */
2049 jdavis@postgresql.or 2550 [ + + ]: 27065 : if (aggstate->aggstrategy == AGG_MIXED &&
2551 [ + - ]: 72 : aggstate->current_phase == 1)
2552 : 72 : hashagg_finish_initial_spills(aggstate);
2553 : :
3817 andres@anarazel.de 2554 [ + + ]: 27065 : if (hasGroupingSets)
2555 : : {
2556 : 360 : aggstate->input_done = true;
2557 : 360 : break;
2558 : : }
2559 : : else
2560 : : {
2561 : 26705 : aggstate->agg_done = true;
2562 : 26705 : break;
2563 : : }
2564 : : }
2565 : : /* set up for next advance_aggregates call */
2566 : 10936510 : tmpcontext->ecxt_outertuple = outerslot;
2567 : :
2568 : : /*
2569 : : * If we are grouping, check whether we've crossed a group
2570 : : * boundary.
2571 : : */
1013 tgl@sss.pgh.pa.us 2572 [ + + + + ]: 10936510 : if (node->aggstrategy != AGG_PLAIN && node->numCols > 0)
2573 : : {
2811 andres@anarazel.de 2574 : 1233944 : tmpcontext->ecxt_innertuple = firstSlot;
2575 [ + + ]: 1233944 : if (!ExecQual(aggstate->phase->eqfunctions[node->numCols - 1],
2576 : : tmpcontext))
2577 : : {
2537 2578 : 116327 : aggstate->grp_firstTuple = ExecCopySlotHeapTuple(outerslot);
3817 2579 : 116327 : break;
2580 : : }
2581 : : }
2582 : : }
2583 : : }
2584 : :
2585 : : /*
2586 : : * Use the representative input tuple for any references to
2587 : : * non-aggregated input columns in aggregate direct args, the node
2588 : : * qual, and the tlist. (If we are not grouping, and there are no
2589 : : * input rows at all, we will come here with an empty firstSlot
2590 : : * ... but if not grouping, there can't be any references to
2591 : : * non-aggregated input columns, so no problem.)
2592 : : */
2593 : 149764 : econtext->ecxt_outertuple = firstSlot;
2594 : : }
2595 : :
2596 [ - + ]: 156839 : Assert(aggstate->projected_set >= 0);
2597 : :
2598 : 156839 : currentSet = aggstate->projected_set;
2599 : :
2600 : 156839 : prepare_projection_slot(aggstate, econtext->ecxt_outertuple, currentSet);
2601 : :
3136 rhodiumtoad@postgres 2602 : 156839 : select_current_set(aggstate, currentSet, false);
2603 : :
2604 : 156839 : finalize_aggregates(aggstate,
2605 : : peragg,
2855 andres@anarazel.de 2606 : 156839 : pergroups[currentSet]);
2607 : :
2608 : : /*
2609 : : * If there's no row to project right now, we must continue rather
2610 : : * than returning a null since there might be more groups.
2611 : : */
3817 2612 : 156833 : result = project_aggregates(aggstate);
2613 [ + + ]: 156827 : if (result)
2614 : 121420 : return result;
2615 : : }
2616 : :
2617 : : /* No more groups */
7779 tgl@sss.pgh.pa.us 2618 : 324 : return NULL;
2619 : : }
2620 : :
2621 : : /*
2622 : : * ExecAgg for hashed case: read input and build hash table
2623 : : */
2624 : : static void
8362 2625 : 8581 : agg_fill_hash_table(AggState *aggstate)
2626 : : {
2627 : : TupleTableSlot *outerslot;
3136 rhodiumtoad@postgres 2628 : 8581 : ExprContext *tmpcontext = aggstate->tmpcontext;
2629 : :
2630 : : /*
2631 : : * Process each outer-plan tuple, and then fetch the next one, until we
2632 : : * exhaust the outer plan.
2633 : : */
2634 : : for (;;)
2635 : : {
3817 andres@anarazel.de 2636 : 3442057 : outerslot = fetch_input_tuple(aggstate);
8391 tgl@sss.pgh.pa.us 2637 [ + + + + ]: 3442057 : if (TupIsNull(outerslot))
2638 : : break;
2639 : :
2640 : : /* set up for lookup_hash_entries and advance_aggregates */
6822 2641 : 3433476 : tmpcontext->ecxt_outertuple = outerslot;
2642 : :
2643 : : /* Find or build hashtable entries */
2848 andres@anarazel.de 2644 : 3433476 : lookup_hash_entries(aggstate);
2645 : :
2646 : : /* Advance the aggregates (or combine functions) */
2647 : 3433476 : advance_aggregates(aggstate);
2648 : :
2649 : : /*
2650 : : * Reset per-input-tuple context after each tuple, but note that the
2651 : : * hash lookups do this too
2652 : : */
3136 rhodiumtoad@postgres 2653 : 3433476 : ResetExprContext(aggstate->tmpcontext);
2654 : : }
2655 : :
2656 : : /* finalize spills, if any */
2049 jdavis@postgresql.or 2657 : 8581 : hashagg_finish_initial_spills(aggstate);
2658 : :
8391 tgl@sss.pgh.pa.us 2659 : 8581 : aggstate->table_filled = true;
2660 : : /* Initialize to walk the first hash table */
3136 rhodiumtoad@postgres 2661 : 8581 : select_current_set(aggstate, 0, true);
2662 : 8581 : ResetTupleHashIterator(aggstate->perhash[0].hashtable,
2663 : : &aggstate->perhash[0].hashiter);
8391 tgl@sss.pgh.pa.us 2664 : 8581 : }
2665 : :
2666 : : /*
2667 : : * If any data was spilled during hash aggregation, reset the hash table and
2668 : : * reprocess one batch of spilled data. After reprocessing a batch, the hash
2669 : : * table will again contain data, ready to be consumed by
2670 : : * agg_retrieve_hash_table_in_memory().
2671 : : *
2672 : : * Should only be called after all in memory hash table entries have been
2673 : : * finalized and emitted.
2674 : : *
2675 : : * Return false when input is exhausted and there's no more work to be done;
2676 : : * otherwise return true.
2677 : : */
2678 : : static bool
2049 jdavis@postgresql.or 2679 : 22521 : agg_refill_hash_table(AggState *aggstate)
2680 : : {
2681 : : HashAggBatch *batch;
2682 : : AggStatePerHash perhash;
2683 : : HashAggSpill spill;
1470 heikki.linnakangas@i 2684 : 22521 : LogicalTapeSet *tapeset = aggstate->hash_tapeset;
1992 tgl@sss.pgh.pa.us 2685 : 22521 : bool spill_initialized = false;
2686 : :
2049 jdavis@postgresql.or 2687 [ + + ]: 22521 : if (aggstate->hash_batches == NIL)
2688 : 9054 : return false;
2689 : :
2690 : : /* hash_batches is a stack, with the top item at the end of the list */
1456 tgl@sss.pgh.pa.us 2691 : 13467 : batch = llast(aggstate->hash_batches);
2692 : 13467 : aggstate->hash_batches = list_delete_last(aggstate->hash_batches);
2693 : :
1917 jdavis@postgresql.or 2694 : 13467 : hash_agg_set_limits(aggstate->hashentrysize, batch->input_card,
2695 : : batch->used_bits, &aggstate->hash_mem_limit,
2696 : : &aggstate->hash_ngroups_limit, NULL);
2697 : :
2698 : : /*
2699 : : * Each batch only processes one grouping set; set the rest to NULL so
2700 : : * that advance_aggregates() knows to ignore them. We don't touch
2701 : : * pergroups for sorted grouping sets here, because they will be needed if
2702 : : * we rescan later. The expressions for sorted grouping sets will not be
2703 : : * evaluated after we recompile anyway.
2704 : : */
1766 2705 [ + - + - : 103704 : MemSet(aggstate->hash_pergroup, 0,
+ - + - +
+ ]
2706 : : sizeof(AggStatePerGroup) * aggstate->num_hashes);
2707 : :
2708 : : /* free memory and reset hash tables */
2049 2709 : 13467 : ReScanExprContext(aggstate->hashcontext);
217 2710 : 13467 : MemoryContextReset(aggstate->hash_tablecxt);
2049 2711 [ + + ]: 103704 : for (int setno = 0; setno < aggstate->num_hashes; setno++)
2712 : 90237 : ResetTupleHashTable(aggstate->perhash[setno].hashtable);
2713 : :
2714 : 13467 : aggstate->hash_ngroups_current = 0;
2715 : :
2716 : : /*
2717 : : * In AGG_MIXED mode, hash aggregation happens in phase 1 and the output
2718 : : * happens in phase 0. So, we switch to phase 1 when processing a batch,
2719 : : * and back to phase 0 after the batch is done.
2720 : : */
2721 [ - + ]: 13467 : Assert(aggstate->current_phase == 0);
2722 [ + + ]: 13467 : if (aggstate->phase->aggstrategy == AGG_MIXED)
2723 : : {
2724 : 13131 : aggstate->current_phase = 1;
2725 : 13131 : aggstate->phase = &aggstate->phases[aggstate->current_phase];
2726 : : }
2727 : :
2728 : 13467 : select_current_set(aggstate, batch->setno, true);
2729 : :
1919 2730 : 13467 : perhash = &aggstate->perhash[aggstate->current_set];
2731 : :
2732 : : /*
2733 : : * Spilled tuples are always read back as MinimalTuples, which may be
2734 : : * different from the outer plan, so recompile the aggregate expressions.
2735 : : *
2736 : : * We still need the NULL check, because we are only processing one
2737 : : * grouping set at a time and the rest will be NULL.
2738 : : */
2049 2739 : 13467 : hashagg_recompile_expressions(aggstate, true, true);
2740 : :
2741 : : INJECTION_POINT("hash-aggregate-process-batch", NULL);
2742 : : for (;;)
1992 tgl@sss.pgh.pa.us 2743 : 592812 : {
1919 jdavis@postgresql.or 2744 : 606279 : TupleTableSlot *spillslot = aggstate->hash_spill_rslot;
2745 : 606279 : TupleTableSlot *hashslot = perhash->hashslot;
217 2746 : 606279 : TupleHashTable hashtable = perhash->hashtable;
2747 : : TupleHashEntry entry;
2748 : : MinimalTuple tuple;
2749 : : uint32 hash;
1919 2750 : 606279 : bool isnew = false;
2751 [ + + ]: 606279 : bool *p_isnew = aggstate->hash_spill_mode ? NULL : &isnew;
2752 : :
2049 2753 [ - + ]: 606279 : CHECK_FOR_INTERRUPTS();
2754 : :
2755 : 606279 : tuple = hashagg_batch_read(batch, &hash);
2756 [ + + ]: 606279 : if (tuple == NULL)
2757 : 13467 : break;
2758 : :
1919 2759 : 592812 : ExecStoreMinimalTuple(tuple, spillslot, true);
2760 : 592812 : aggstate->tmpcontext->ecxt_outertuple = spillslot;
2761 : :
2762 : 592812 : prepare_hash_slot(perhash,
2763 : 592812 : aggstate->tmpcontext->ecxt_outertuple,
2764 : : hashslot);
217 2765 : 592812 : entry = LookupTupleHashEntryHash(hashtable, hashslot,
2766 : : p_isnew, hash);
2767 : :
1919 2768 [ + + ]: 592812 : if (entry != NULL)
2769 : : {
2770 [ + + ]: 380466 : if (isnew)
217 2771 : 74202 : initialize_hash_entry(aggstate, hashtable, entry);
2772 : 380466 : aggstate->hash_pergroup[batch->setno] = TupleHashEntryGetAdditional(hashtable, entry);
2049 2773 : 380466 : advance_aggregates(aggstate);
2774 : : }
2775 : : else
2776 : : {
2777 [ + + ]: 212346 : if (!spill_initialized)
2778 : : {
2779 : : /*
2780 : : * Avoid initializing the spill until we actually need it so
2781 : : * that we don't assign tapes that will never be used.
2782 : : */
2783 : 6246 : spill_initialized = true;
1470 heikki.linnakangas@i 2784 : 6246 : hashagg_spill_init(&spill, tapeset, batch->used_bits,
2785 : : batch->input_card, aggstate->hashentrysize);
2786 : : }
2787 : : /* no memory for a new group, spill */
1919 jdavis@postgresql.or 2788 : 212346 : hashagg_spill_tuple(aggstate, &spill, spillslot, hash);
2789 : :
2790 : 212346 : aggstate->hash_pergroup[batch->setno] = NULL;
2791 : : }
2792 : :
2793 : : /*
2794 : : * Reset per-input-tuple context after each tuple, but note that the
2795 : : * hash lookups do this too
2796 : : */
2049 2797 : 592812 : ResetExprContext(aggstate->tmpcontext);
2798 : : }
2799 : :
1470 heikki.linnakangas@i 2800 : 13467 : LogicalTapeClose(batch->input_tape);
2801 : :
2802 : : /* change back to phase 0 */
2049 jdavis@postgresql.or 2803 : 13467 : aggstate->current_phase = 0;
2804 : 13467 : aggstate->phase = &aggstate->phases[aggstate->current_phase];
2805 : :
2806 [ + + ]: 13467 : if (spill_initialized)
2807 : : {
2808 : 6246 : hashagg_spill_finish(aggstate, &spill, batch->setno);
1868 2809 : 6246 : hash_agg_update_metrics(aggstate, true, spill.npartitions);
2810 : : }
2811 : : else
2049 2812 : 7221 : hash_agg_update_metrics(aggstate, true, 0);
2813 : :
2814 : 13467 : aggstate->hash_spill_mode = false;
2815 : :
2816 : : /* prepare to walk the first hash table */
2817 : 13467 : select_current_set(aggstate, batch->setno, true);
2818 : 13467 : ResetTupleHashIterator(aggstate->perhash[batch->setno].hashtable,
2819 : : &aggstate->perhash[batch->setno].hashiter);
2820 : :
2821 : 13467 : pfree(batch);
2822 : :
2823 : 13467 : return true;
2824 : : }
2825 : :
2826 : : /*
2827 : : * ExecAgg for hashed case: retrieving groups from hash table
2828 : : *
2829 : : * After exhausting in-memory tuples, also try refilling the hash table using
2830 : : * previously-spilled tuples. Only returns NULL after all in-memory and
2831 : : * spilled tuples are exhausted.
2832 : : */
2833 : : static TupleTableSlot *
8362 tgl@sss.pgh.pa.us 2834 : 250698 : agg_retrieve_hash_table(AggState *aggstate)
2835 : : {
2049 jdavis@postgresql.or 2836 : 250698 : TupleTableSlot *result = NULL;
2837 : :
2838 [ + + ]: 505809 : while (result == NULL)
2839 : : {
2840 : 264165 : result = agg_retrieve_hash_table_in_memory(aggstate);
2841 [ + + ]: 264165 : if (result == NULL)
2842 : : {
2843 [ + + ]: 22521 : if (!agg_refill_hash_table(aggstate))
2844 : : {
2845 : 9054 : aggstate->agg_done = true;
2846 : 9054 : break;
2847 : : }
2848 : : }
2849 : : }
2850 : :
2851 : 250698 : return result;
2852 : : }
2853 : :
2854 : : /*
2855 : : * Retrieve the groups from the in-memory hash tables without considering any
2856 : : * spilled tuples.
2857 : : */
2858 : : static TupleTableSlot *
2859 : 264165 : agg_retrieve_hash_table_in_memory(AggState *aggstate)
2860 : : {
2861 : : ExprContext *econtext;
2862 : : AggStatePerAgg peragg;
2863 : : AggStatePerGroup pergroup;
2864 : : TupleHashEntry entry;
2865 : : TupleTableSlot *firstSlot;
2866 : : TupleTableSlot *result;
2867 : : AggStatePerHash perhash;
2868 : :
2869 : : /*
2870 : : * get state info from node.
2871 : : *
2872 : : * econtext is the per-output-tuple expression context.
2873 : : */
8362 tgl@sss.pgh.pa.us 2874 : 264165 : econtext = aggstate->ss.ps.ps_ExprContext;
8391 2875 : 264165 : peragg = aggstate->peragg;
8362 2876 : 264165 : firstSlot = aggstate->ss.ss_ScanTupleSlot;
2877 : :
2878 : : /*
2879 : : * Note that perhash (and therefore anything accessed through it) can
2880 : : * change inside the loop, as we change between grouping sets.
2881 : : */
3136 rhodiumtoad@postgres 2882 : 264165 : perhash = &aggstate->perhash[aggstate->current_set];
2883 : :
2884 : : /*
2885 : : * We loop retrieving groups until we find one satisfying
2886 : : * aggstate->ss.ps.qual
2887 : : */
2888 : : for (;;)
8391 tgl@sss.pgh.pa.us 2889 : 67977 : {
3136 rhodiumtoad@postgres 2890 : 332142 : TupleTableSlot *hashslot = perhash->hashslot;
217 jdavis@postgresql.or 2891 : 332142 : TupleHashTable hashtable = perhash->hashtable;
2892 : : int i;
2893 : :
3016 andres@anarazel.de 2894 [ + + ]: 332142 : CHECK_FOR_INTERRUPTS();
2895 : :
2896 : : /*
2897 : : * Find the next entry in the hash table
2898 : : */
217 jdavis@postgresql.or 2899 : 332142 : entry = ScanTupleHashTable(hashtable, &perhash->hashiter);
8326 tgl@sss.pgh.pa.us 2900 [ + + ]: 332142 : if (entry == NULL)
2901 : : {
3136 rhodiumtoad@postgres 2902 : 72667 : int nextset = aggstate->current_set + 1;
2903 : :
2904 [ + + ]: 72667 : if (nextset < aggstate->num_hashes)
2905 : : {
2906 : : /*
2907 : : * Switch to next grouping set, reinitialize, and restart the
2908 : : * loop.
2909 : : */
2910 : 50146 : select_current_set(aggstate, nextset, true);
2911 : :
2912 : 50146 : perhash = &aggstate->perhash[aggstate->current_set];
2913 : :
9 drowley@postgresql.o 2914 : 50146 : ResetTupleHashIterator(perhash->hashtable, &perhash->hashiter);
2915 : :
3136 rhodiumtoad@postgres 2916 : 50146 : continue;
2917 : : }
2918 : : else
2919 : : {
2920 : 22521 : return NULL;
2921 : : }
2922 : : }
2923 : :
2924 : : /*
2925 : : * Clear the per-output-tuple context for each group
2926 : : *
2927 : : * We intentionally don't use ReScanExprContext here; if any aggs have
2928 : : * registered shutdown callbacks, they mustn't be called yet, since we
2929 : : * might not be done with that agg.
2930 : : */
8391 tgl@sss.pgh.pa.us 2931 : 259475 : ResetExprContext(econtext);
2932 : :
2933 : : /*
2934 : : * Transform representative tuple back into one with the right
2935 : : * columns.
2936 : : */
217 jdavis@postgresql.or 2937 : 259475 : ExecStoreMinimalTuple(TupleHashEntryGetTuple(entry), hashslot, false);
3253 andres@anarazel.de 2938 : 259475 : slot_getallattrs(hashslot);
2939 : :
2940 : 259475 : ExecClearTuple(firstSlot);
2941 : 259475 : memset(firstSlot->tts_isnull, true,
2942 : 259475 : firstSlot->tts_tupleDescriptor->natts * sizeof(bool));
2943 : :
3136 rhodiumtoad@postgres 2944 [ + + ]: 685386 : for (i = 0; i < perhash->numhashGrpCols; i++)
2945 : : {
2946 : 425911 : int varNumber = perhash->hashGrpColIdxInput[i] - 1;
2947 : :
3253 andres@anarazel.de 2948 : 425911 : firstSlot->tts_values[varNumber] = hashslot->tts_values[i];
2949 : 425911 : firstSlot->tts_isnull[varNumber] = hashslot->tts_isnull[i];
2950 : : }
2951 : 259475 : ExecStoreVirtualTuple(firstSlot);
2952 : :
217 jdavis@postgresql.or 2953 : 259475 : pergroup = (AggStatePerGroup) TupleHashEntryGetAdditional(hashtable, entry);
2954 : :
2955 : : /*
2956 : : * Use the representative input tuple for any references to
2957 : : * non-aggregated input columns in the qual and tlist.
2958 : : */
6822 tgl@sss.pgh.pa.us 2959 : 259475 : econtext->ecxt_outertuple = firstSlot;
2960 : :
3136 rhodiumtoad@postgres 2961 : 259475 : prepare_projection_slot(aggstate,
2962 : : econtext->ecxt_outertuple,
2963 : : aggstate->current_set);
2964 : :
2965 : 259475 : finalize_aggregates(aggstate, peragg, pergroup);
2966 : :
3817 andres@anarazel.de 2967 : 259475 : result = project_aggregates(aggstate);
2968 [ + + ]: 259475 : if (result)
2969 : 241644 : return result;
2970 : : }
2971 : :
2972 : : /* No more groups */
2973 : : return NULL;
2974 : : }
2975 : :
2976 : : /*
2977 : : * hashagg_spill_init
2978 : : *
2979 : : * Called after we determined that spilling is necessary. Chooses the number
2980 : : * of partitions to create, and initializes them.
2981 : : */
2982 : : static void
1470 heikki.linnakangas@i 2983 : 6306 : hashagg_spill_init(HashAggSpill *spill, LogicalTapeSet *tapeset, int used_bits,
2984 : : double input_groups, double hashentrysize)
2985 : : {
2986 : : int npartitions;
2987 : : int partition_bits;
2988 : :
1992 tgl@sss.pgh.pa.us 2989 : 6306 : npartitions = hash_choose_num_partitions(input_groups, hashentrysize,
2990 : : used_bits, &partition_bits);
2991 : :
2992 : : #ifdef USE_INJECTION_POINTS
2993 : : if (IS_INJECTION_POINT_ATTACHED("hash-aggregate-single-partition"))
2994 : : {
2995 : : npartitions = 1;
2996 : : partition_bits = 0;
2997 : : INJECTION_POINT_CACHED("hash-aggregate-single-partition", NULL);
2998 : : }
2999 : : #endif
3000 : :
1470 heikki.linnakangas@i 3001 : 6306 : spill->partitions = palloc0(sizeof(LogicalTape *) * npartitions);
2049 jdavis@postgresql.or 3002 : 6306 : spill->ntuples = palloc0(sizeof(int64) * npartitions);
1917 3003 : 6306 : spill->hll_card = palloc0(sizeof(hyperLogLogState) * npartitions);
3004 : :
1470 heikki.linnakangas@i 3005 [ + + ]: 31530 : for (int i = 0; i < npartitions; i++)
3006 : 25224 : spill->partitions[i] = LogicalTapeCreate(tapeset);
3007 : :
2049 jdavis@postgresql.or 3008 : 6306 : spill->shift = 32 - used_bits - partition_bits;
258 3009 [ + - ]: 6306 : if (spill->shift < 32)
3010 : 6306 : spill->mask = (npartitions - 1) << spill->shift;
3011 : : else
258 jdavis@postgresql.or 3012 :UBC 0 : spill->mask = 0;
2049 jdavis@postgresql.or 3013 :CBC 6306 : spill->npartitions = npartitions;
3014 : :
1917 3015 [ + + ]: 31530 : for (int i = 0; i < npartitions; i++)
3016 : 25224 : initHyperLogLog(&spill->hll_card[i], HASHAGG_HLL_BIT_WIDTH);
2049 3017 : 6306 : }
3018 : :
3019 : : /*
3020 : : * hashagg_spill_tuple
3021 : : *
3022 : : * No room for new groups in the hash table. Save for later in the appropriate
3023 : : * partition.
3024 : : */
3025 : : static Size
1933 3026 : 592812 : hashagg_spill_tuple(AggState *aggstate, HashAggSpill *spill,
3027 : : TupleTableSlot *inputslot, uint32 hash)
3028 : : {
3029 : : TupleTableSlot *spillslot;
3030 : : int partition;
3031 : : MinimalTuple tuple;
3032 : : LogicalTape *tape;
1992 tgl@sss.pgh.pa.us 3033 : 592812 : int total_written = 0;
3034 : : bool shouldFree;
3035 : :
2049 jdavis@postgresql.or 3036 [ - + ]: 592812 : Assert(spill->partitions != NULL);
3037 : :
3038 : : /* spill only attributes that we actually need */
1933 3039 [ + + ]: 592812 : if (!aggstate->all_cols_needed)
3040 : : {
3041 : 1938 : spillslot = aggstate->hash_spill_wslot;
3042 : 1938 : slot_getsomeattrs(inputslot, aggstate->max_colno_needed);
3043 : 1938 : ExecClearTuple(spillslot);
3044 [ + + ]: 5814 : for (int i = 0; i < spillslot->tts_tupleDescriptor->natts; i++)
3045 : : {
3046 [ + + ]: 3876 : if (bms_is_member(i + 1, aggstate->colnos_needed))
3047 : : {
3048 : 1938 : spillslot->tts_values[i] = inputslot->tts_values[i];
3049 : 1938 : spillslot->tts_isnull[i] = inputslot->tts_isnull[i];
3050 : : }
3051 : : else
3052 : 1938 : spillslot->tts_isnull[i] = true;
3053 : : }
3054 : 1938 : ExecStoreVirtualTuple(spillslot);
3055 : : }
3056 : : else
3057 : 590874 : spillslot = inputslot;
3058 : :
3059 : 592812 : tuple = ExecFetchSlotMinimalTuple(spillslot, &shouldFree);
3060 : :
258 3061 [ + - ]: 592812 : if (spill->shift < 32)
3062 : 592812 : partition = (hash & spill->mask) >> spill->shift;
3063 : : else
258 jdavis@postgresql.or 3064 :UBC 0 : partition = 0;
3065 : :
2049 jdavis@postgresql.or 3066 :CBC 592812 : spill->ntuples[partition]++;
3067 : :
3068 : : /*
3069 : : * All hash values destined for a given partition have some bits in
3070 : : * common, which causes bad HLL cardinality estimates. Hash the hash to
3071 : : * get a more uniform distribution.
3072 : : */
1917 3073 : 592812 : addHyperLogLog(&spill->hll_card[partition], hash_bytes_uint32(hash));
3074 : :
1470 heikki.linnakangas@i 3075 : 592812 : tape = spill->partitions[partition];
3076 : :
1032 peter@eisentraut.org 3077 : 592812 : LogicalTapeWrite(tape, &hash, sizeof(uint32));
2049 jdavis@postgresql.or 3078 : 592812 : total_written += sizeof(uint32);
3079 : :
1032 peter@eisentraut.org 3080 : 592812 : LogicalTapeWrite(tape, tuple, tuple->t_len);
2049 jdavis@postgresql.or 3081 : 592812 : total_written += tuple->t_len;
3082 : :
3083 [ + + ]: 592812 : if (shouldFree)
3084 : 380466 : pfree(tuple);
3085 : :
3086 : 592812 : return total_written;
3087 : : }
3088 : :
3089 : : /*
3090 : : * hashagg_batch_new
3091 : : *
3092 : : * Construct a HashAggBatch item, which represents one iteration of HashAgg to
3093 : : * be done.
3094 : : */
3095 : : static HashAggBatch *
1470 heikki.linnakangas@i 3096 : 13467 : hashagg_batch_new(LogicalTape *input_tape, int setno,
3097 : : int64 input_tuples, double input_card, int used_bits)
3098 : : {
2049 jdavis@postgresql.or 3099 : 13467 : HashAggBatch *batch = palloc0(sizeof(HashAggBatch));
3100 : :
3101 : 13467 : batch->setno = setno;
3102 : 13467 : batch->used_bits = used_bits;
1470 heikki.linnakangas@i 3103 : 13467 : batch->input_tape = input_tape;
2049 jdavis@postgresql.or 3104 : 13467 : batch->input_tuples = input_tuples;
1917 3105 : 13467 : batch->input_card = input_card;
3106 : :
2049 3107 : 13467 : return batch;
3108 : : }
3109 : :
3110 : : /*
3111 : : * hashagg_batch_read
3112 : : * read the next tuple from a batch's tape. Return NULL if no more.
3113 : : */
3114 : : static MinimalTuple
3115 : 606279 : hashagg_batch_read(HashAggBatch *batch, uint32 *hashp)
3116 : : {
1470 heikki.linnakangas@i 3117 : 606279 : LogicalTape *tape = batch->input_tape;
3118 : : MinimalTuple tuple;
3119 : : uint32 t_len;
3120 : : size_t nread;
3121 : : uint32 hash;
3122 : :
3123 : 606279 : nread = LogicalTapeRead(tape, &hash, sizeof(uint32));
2049 jdavis@postgresql.or 3124 [ + + ]: 606279 : if (nread == 0)
3125 : 13467 : return NULL;
3126 [ - + ]: 592812 : if (nread != sizeof(uint32))
2049 jdavis@postgresql.or 3127 [ # # ]:UBC 0 : ereport(ERROR,
3128 : : (errcode_for_file_access(),
3129 : : errmsg_internal("unexpected EOF for tape %p: requested %zu bytes, read %zu bytes",
3130 : : tape, sizeof(uint32), nread)));
2049 jdavis@postgresql.or 3131 [ + - ]:CBC 592812 : if (hashp != NULL)
3132 : 592812 : *hashp = hash;
3133 : :
1470 heikki.linnakangas@i 3134 : 592812 : nread = LogicalTapeRead(tape, &t_len, sizeof(t_len));
2049 jdavis@postgresql.or 3135 [ - + ]: 592812 : if (nread != sizeof(uint32))
2049 jdavis@postgresql.or 3136 [ # # ]:UBC 0 : ereport(ERROR,
3137 : : (errcode_for_file_access(),
3138 : : errmsg_internal("unexpected EOF for tape %p: requested %zu bytes, read %zu bytes",
3139 : : tape, sizeof(uint32), nread)));
3140 : :
2049 jdavis@postgresql.or 3141 :CBC 592812 : tuple = (MinimalTuple) palloc(t_len);
3142 : 592812 : tuple->t_len = t_len;
3143 : :
1470 heikki.linnakangas@i 3144 : 592812 : nread = LogicalTapeRead(tape,
3145 : : (char *) tuple + sizeof(uint32),
3146 : : t_len - sizeof(uint32));
2049 jdavis@postgresql.or 3147 [ - + ]: 592812 : if (nread != t_len - sizeof(uint32))
2049 jdavis@postgresql.or 3148 [ # # ]:UBC 0 : ereport(ERROR,
3149 : : (errcode_for_file_access(),
3150 : : errmsg_internal("unexpected EOF for tape %p: requested %zu bytes, read %zu bytes",
3151 : : tape, t_len - sizeof(uint32), nread)));
3152 : :
2049 jdavis@postgresql.or 3153 :CBC 592812 : return tuple;
3154 : : }
3155 : :
3156 : : /*
3157 : : * hashagg_finish_initial_spills
3158 : : *
3159 : : * After a HashAggBatch has been processed, it may have spilled tuples to
3160 : : * disk. If so, turn the spilled partitions into new batches that must later
3161 : : * be executed.
3162 : : */
3163 : : static void
3164 : 8653 : hashagg_finish_initial_spills(AggState *aggstate)
3165 : : {
3166 : : int setno;
1992 tgl@sss.pgh.pa.us 3167 : 8653 : int total_npartitions = 0;
3168 : :
2049 jdavis@postgresql.or 3169 [ + + ]: 8653 : if (aggstate->hash_spills != NULL)
3170 : : {
3171 [ + + ]: 90 : for (setno = 0; setno < aggstate->num_hashes; setno++)
3172 : : {
3173 : 60 : HashAggSpill *spill = &aggstate->hash_spills[setno];
3174 : :
3175 : 60 : total_npartitions += spill->npartitions;
3176 : 60 : hashagg_spill_finish(aggstate, spill, setno);
3177 : : }
3178 : :
3179 : : /*
3180 : : * We're not processing tuples from outer plan any more; only
3181 : : * processing batches of spilled tuples. The initial spill structures
3182 : : * are no longer needed.
3183 : : */
3184 : 30 : pfree(aggstate->hash_spills);
3185 : 30 : aggstate->hash_spills = NULL;
3186 : : }
3187 : :
3188 : 8653 : hash_agg_update_metrics(aggstate, false, total_npartitions);
3189 : 8653 : aggstate->hash_spill_mode = false;
3190 : 8653 : }
3191 : :
3192 : : /*
3193 : : * hashagg_spill_finish
3194 : : *
3195 : : * Transform spill partitions into new batches.
3196 : : */
3197 : : static void
3198 : 6306 : hashagg_spill_finish(AggState *aggstate, HashAggSpill *spill, int setno)
3199 : : {
3200 : : int i;
1992 tgl@sss.pgh.pa.us 3201 : 6306 : int used_bits = 32 - spill->shift;
3202 : :
2049 jdavis@postgresql.or 3203 [ - + ]: 6306 : if (spill->npartitions == 0)
1992 tgl@sss.pgh.pa.us 3204 :UBC 0 : return; /* didn't spill */
3205 : :
2049 jdavis@postgresql.or 3206 [ + + ]:CBC 31530 : for (i = 0; i < spill->npartitions; i++)
3207 : : {
1470 heikki.linnakangas@i 3208 : 25224 : LogicalTape *tape = spill->partitions[i];
3209 : : HashAggBatch *new_batch;
3210 : : double cardinality;
3211 : :
3212 : : /* if the partition is empty, don't create a new batch of work */
2049 jdavis@postgresql.or 3213 [ + + ]: 25224 : if (spill->ntuples[i] == 0)
3214 : 11757 : continue;
3215 : :
1917 3216 : 13467 : cardinality = estimateHyperLogLog(&spill->hll_card[i]);
3217 : 13467 : freeHyperLogLog(&spill->hll_card[i]);
3218 : :
3219 : : /* rewinding frees the buffer while not in use */
1470 heikki.linnakangas@i 3220 : 13467 : LogicalTapeRewindForRead(tape, HASHAGG_READ_BUFFER_SIZE);
3221 : :
3222 : 13467 : new_batch = hashagg_batch_new(tape, setno,
1868 jdavis@postgresql.or 3223 : 13467 : spill->ntuples[i], cardinality,
3224 : : used_bits);
1456 tgl@sss.pgh.pa.us 3225 : 13467 : aggstate->hash_batches = lappend(aggstate->hash_batches, new_batch);
2049 jdavis@postgresql.or 3226 : 13467 : aggstate->hash_batches_used++;
3227 : : }
3228 : :
3229 : 6306 : pfree(spill->ntuples);
1917 3230 : 6306 : pfree(spill->hll_card);
2049 3231 : 6306 : pfree(spill->partitions);
3232 : : }
3233 : :
3234 : : /*
3235 : : * Free resources related to a spilled HashAgg.
3236 : : */
3237 : : static void
3238 : 28991 : hashagg_reset_spill_state(AggState *aggstate)
3239 : : {
3240 : : /* free spills from initial pass */
3241 [ - + ]: 28991 : if (aggstate->hash_spills != NULL)
3242 : : {
3243 : : int setno;
3244 : :
2049 jdavis@postgresql.or 3245 [ # # ]:UBC 0 : for (setno = 0; setno < aggstate->num_hashes; setno++)
3246 : : {
3247 : 0 : HashAggSpill *spill = &aggstate->hash_spills[setno];
3248 : :
3249 : 0 : pfree(spill->ntuples);
3250 : 0 : pfree(spill->partitions);
3251 : : }
3252 : 0 : pfree(aggstate->hash_spills);
3253 : 0 : aggstate->hash_spills = NULL;
3254 : : }
3255 : :
3256 : : /* free batches */
1456 tgl@sss.pgh.pa.us 3257 :CBC 28991 : list_free_deep(aggstate->hash_batches);
2049 jdavis@postgresql.or 3258 : 28991 : aggstate->hash_batches = NIL;
3259 : :
3260 : : /* close tape set */
1470 heikki.linnakangas@i 3261 [ + + ]: 28991 : if (aggstate->hash_tapeset != NULL)
3262 : : {
3263 : 30 : LogicalTapeSetClose(aggstate->hash_tapeset);
3264 : 30 : aggstate->hash_tapeset = NULL;
3265 : : }
2049 jdavis@postgresql.or 3266 : 28991 : }
3267 : :
3268 : :
3269 : : /* -----------------
3270 : : * ExecInitAgg
3271 : : *
3272 : : * Creates the run-time information for the agg node produced by the
3273 : : * planner and initializes its outer subtree.
3274 : : *
3275 : : * -----------------
3276 : : */
3277 : : AggState *
7181 tgl@sss.pgh.pa.us 3278 : 22889 : ExecInitAgg(Agg *node, EState *estate, int eflags)
3279 : : {
3280 : : AggState *aggstate;
3281 : : AggStatePerAgg peraggs;
3282 : : AggStatePerTrans pertransstates;
3283 : : AggStatePerGroup *pergroups;
3284 : : Plan *outerPlan;
3285 : : ExprContext *econtext;
3286 : : TupleDesc scanDesc;
3287 : : int max_aggno;
3288 : : int max_transno;
3289 : : int numaggrefs;
3290 : : int numaggs;
3291 : : int numtrans;
3292 : : int phase;
3293 : : int phaseidx;
3294 : : ListCell *l;
3817 andres@anarazel.de 3295 : 22889 : Bitmapset *all_grouped_cols = NULL;
3296 : 22889 : int numGroupingSets = 1;
3297 : : int numPhases;
3298 : : int numHashes;
3299 : 22889 : int i = 0;
3300 : 22889 : int j = 0;
3136 rhodiumtoad@postgres 3301 [ + + ]: 42395 : bool use_hashing = (node->aggstrategy == AGG_HASHED ||
3302 [ + + ]: 19506 : node->aggstrategy == AGG_MIXED);
3303 : :
3304 : : /* check for unsupported flags */
7181 tgl@sss.pgh.pa.us 3305 [ - + ]: 22889 : Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
3306 : :
3307 : : /*
3308 : : * create state structure
3309 : : */
10277 bruce@momjian.us 3310 : 22889 : aggstate = makeNode(AggState);
8362 tgl@sss.pgh.pa.us 3311 : 22889 : aggstate->ss.ps.plan = (Plan *) node;
3312 : 22889 : aggstate->ss.ps.state = estate;
3024 andres@anarazel.de 3313 : 22889 : aggstate->ss.ps.ExecProcNode = ExecAgg;
3314 : :
8362 tgl@sss.pgh.pa.us 3315 : 22889 : aggstate->aggs = NIL;
3316 : 22889 : aggstate->numaggs = 0;
3737 heikki.linnakangas@i 3317 : 22889 : aggstate->numtrans = 0;
3136 rhodiumtoad@postgres 3318 : 22889 : aggstate->aggstrategy = node->aggstrategy;
3410 tgl@sss.pgh.pa.us 3319 : 22889 : aggstate->aggsplit = node->aggsplit;
3817 andres@anarazel.de 3320 : 22889 : aggstate->maxsets = 0;
3321 : 22889 : aggstate->projected_set = -1;
3322 : 22889 : aggstate->current_set = 0;
8391 tgl@sss.pgh.pa.us 3323 : 22889 : aggstate->peragg = NULL;
3737 heikki.linnakangas@i 3324 : 22889 : aggstate->pertrans = NULL;
2937 tgl@sss.pgh.pa.us 3325 : 22889 : aggstate->curperagg = NULL;
3737 heikki.linnakangas@i 3326 : 22889 : aggstate->curpertrans = NULL;
3817 andres@anarazel.de 3327 : 22889 : aggstate->input_done = false;
3410 tgl@sss.pgh.pa.us 3328 : 22889 : aggstate->agg_done = false;
2855 andres@anarazel.de 3329 : 22889 : aggstate->pergroups = NULL;
8391 tgl@sss.pgh.pa.us 3330 : 22889 : aggstate->grp_firstTuple = NULL;
3817 andres@anarazel.de 3331 : 22889 : aggstate->sort_in = NULL;
3332 : 22889 : aggstate->sort_out = NULL;
3333 : :
3334 : : /*
3335 : : * phases[0] always exists, but is dummy in sorted/plain mode
3336 : : */
3136 rhodiumtoad@postgres 3337 [ + + ]: 22889 : numPhases = (use_hashing ? 1 : 2);
3338 : 22889 : numHashes = (use_hashing ? 1 : 0);
3339 : :
3340 : : /*
3341 : : * Calculate the maximum number of grouping sets in any phase; this
3342 : : * determines the size of some allocations. Also calculate the number of
3343 : : * phases, since all hashed/mixed nodes contribute to only a single phase.
3344 : : */
3817 andres@anarazel.de 3345 [ + + ]: 22889 : if (node->groupingSets)
3346 : : {
3347 : 457 : numGroupingSets = list_length(node->groupingSets);
3348 : :
3349 [ + + + + : 956 : foreach(l, node->chain)
+ + ]
3350 : : {
3810 bruce@momjian.us 3351 : 499 : Agg *agg = lfirst(l);
3352 : :
3817 andres@anarazel.de 3353 [ + + ]: 499 : numGroupingSets = Max(numGroupingSets,
3354 : : list_length(agg->groupingSets));
3355 : :
3356 : : /*
3357 : : * additional AGG_HASHED aggs become part of phase 0, but all
3358 : : * others add an extra phase.
3359 : : */
3136 rhodiumtoad@postgres 3360 [ + + ]: 499 : if (agg->aggstrategy != AGG_HASHED)
3361 : 242 : ++numPhases;
3362 : : else
3363 : 257 : ++numHashes;
3364 : : }
3365 : : }
3366 : :
3817 andres@anarazel.de 3367 : 22889 : aggstate->maxsets = numGroupingSets;
3136 rhodiumtoad@postgres 3368 : 22889 : aggstate->numphases = numPhases;
3369 : :
3817 andres@anarazel.de 3370 : 22889 : aggstate->aggcontexts = (ExprContext **)
3371 : 22889 : palloc0(sizeof(ExprContext *) * numGroupingSets);
3372 : :
3373 : : /*
3374 : : * Create expression contexts. We need three or more, one for
3375 : : * per-input-tuple processing, one for per-output-tuple processing, one
3376 : : * for all the hashtables, and one for each grouping set. The per-tuple
3377 : : * memory context of the per-grouping-set ExprContexts (aggcontexts)
3378 : : * replaces the standalone memory context formerly used to hold transition
3379 : : * values. We cheat a little by using ExecAssignExprContext() to build
3380 : : * all of them.
3381 : : *
3382 : : * NOTE: the details of what is stored in aggcontexts and what is stored
3383 : : * in the regular per-query memory context are driven by a simple
3384 : : * decision: we want to reset the aggcontext at group boundaries (if not
3385 : : * hashing) and in ExecReScanAgg to recover no-longer-wanted space.
3386 : : */
3387 : 22889 : ExecAssignExprContext(estate, &aggstate->ss.ps);
3388 : 22889 : aggstate->tmpcontext = aggstate->ss.ps.ps_ExprContext;
3389 : :
3390 [ + + ]: 46210 : for (i = 0; i < numGroupingSets; ++i)
3391 : : {
3392 : 23321 : ExecAssignExprContext(estate, &aggstate->ss.ps);
3393 : 23321 : aggstate->aggcontexts[i] = aggstate->ss.ps.ps_ExprContext;
3394 : : }
3395 : :
3136 rhodiumtoad@postgres 3396 [ + + ]: 22889 : if (use_hashing)
217 jdavis@postgresql.or 3397 : 3499 : hash_create_memory(aggstate);
3398 : :
3817 andres@anarazel.de 3399 : 22889 : ExecAssignExprContext(estate, &aggstate->ss.ps);
3400 : :
3401 : : /*
3402 : : * Initialize child nodes.
3403 : : *
3404 : : * If we are doing a hashed aggregation then the child plan does not need
3405 : : * to handle REWIND efficiently; see ExecReScanAgg.
3406 : : */
7181 tgl@sss.pgh.pa.us 3407 [ + + ]: 22889 : if (node->aggstrategy == AGG_HASHED)
3408 : 3383 : eflags &= ~EXEC_FLAG_REWIND;
8362 3409 : 22889 : outerPlan = outerPlan(node);
7181 3410 : 22889 : outerPlanState(aggstate) = ExecInitNode(outerPlan, estate, eflags);
3411 : :
3412 : : /*
3413 : : * initialize source tuple type.
3414 : : */
2286 andres@anarazel.de 3415 : 22889 : aggstate->ss.ps.outerops =
3416 : 22889 : ExecGetResultSlotOps(outerPlanState(&aggstate->ss),
3417 : : &aggstate->ss.ps.outeropsfixed);
3418 : 22889 : aggstate->ss.ps.outeropsset = true;
3419 : :
2538 3420 : 22889 : ExecCreateScanSlotFromOuterPlan(estate, &aggstate->ss,
3421 : : aggstate->ss.ps.outerops);
2811 3422 : 22889 : scanDesc = aggstate->ss.ss_ScanTupleSlot->tts_tupleDescriptor;
3423 : :
3424 : : /*
3425 : : * If there are more than two phases (including a potential dummy phase
3426 : : * 0), input will be resorted using tuplesort. Need a slot for that.
3427 : : */
2286 3428 [ + + ]: 22889 : if (numPhases > 2)
3429 : : {
2538 3430 : 105 : aggstate->sort_slot = ExecInitExtraTupleSlot(estate, scanDesc,
3431 : : &TTSOpsMinimalTuple);
3432 : :
3433 : : /*
3434 : : * The output of the tuplesort, and the output from the outer child
3435 : : * might not use the same type of slot. In most cases the child will
3436 : : * be a Sort, and thus return a TTSOpsMinimalTuple type slot - but the
3437 : : * input can also be presorted due an index, in which case it could be
3438 : : * a different type of slot.
3439 : : *
3440 : : * XXX: For efficiency it would be good to instead/additionally
3441 : : * generate expressions with corresponding settings of outerops* for
3442 : : * the individual phases - deforming is often a bottleneck for
3443 : : * aggregations with lots of rows per group. If there's multiple
3444 : : * sorts, we know that all but the first use TTSOpsMinimalTuple (via
3445 : : * the nodeAgg.c internal tuplesort).
3446 : : */
2286 3447 [ + - ]: 105 : if (aggstate->ss.ps.outeropsfixed &&
3448 [ + + ]: 105 : aggstate->ss.ps.outerops != &TTSOpsMinimalTuple)
3449 : 6 : aggstate->ss.ps.outeropsfixed = false;
3450 : : }
3451 : :
3452 : : /*
3453 : : * Initialize result type, slot and projection.
3454 : : */
2538 3455 : 22889 : ExecInitResultTupleSlotTL(&aggstate->ss.ps, &TTSOpsVirtual);
6842 tgl@sss.pgh.pa.us 3456 : 22889 : ExecAssignProjectionInfo(&aggstate->ss.ps, NULL);
3457 : :
3458 : : /*
3459 : : * initialize child expressions
3460 : : *
3461 : : * We expect the parser to have checked that no aggs contain other agg
3462 : : * calls in their arguments (and just to be sure, we verify it again while
3463 : : * initializing the plan node). This would make no sense under SQL
3464 : : * semantics, and it's forbidden by the spec. Because it is true, we
3465 : : * don't need to worry about evaluating the aggs in any particular order.
3466 : : *
3467 : : * Note: execExpr.c finds Aggrefs for us, and adds them to aggstate->aggs.
3468 : : * Aggrefs in the qual are found here; Aggrefs in the targetlist are found
3469 : : * during ExecAssignProjectionInfo, above.
3470 : : */
2810 andres@anarazel.de 3471 : 22889 : aggstate->ss.ps.qual =
3472 : 22889 : ExecInitQual(node->plan.qual, (PlanState *) aggstate);
3473 : :
3474 : : /*
3475 : : * We should now have found all Aggrefs in the targetlist and quals.
3476 : : */
1798 heikki.linnakangas@i 3477 : 22889 : numaggrefs = list_length(aggstate->aggs);
3478 : 22889 : max_aggno = -1;
3479 : 22889 : max_transno = -1;
3480 [ + + + + : 48838 : foreach(l, aggstate->aggs)
+ + ]
3481 : : {
3482 : 25949 : Aggref *aggref = (Aggref *) lfirst(l);
3483 : :
3484 : 25949 : max_aggno = Max(max_aggno, aggref->aggno);
3485 : 25949 : max_transno = Max(max_transno, aggref->aggtransno);
3486 : : }
293 jdavis@postgresql.or 3487 : 22889 : aggstate->numaggs = numaggs = max_aggno + 1;
3488 : 22889 : aggstate->numtrans = numtrans = max_transno + 1;
3489 : :
3490 : : /*
3491 : : * For each phase, prepare grouping set data and fmgr lookup data for
3492 : : * compare functions. Accumulate all_grouped_cols in passing.
3493 : : */
3817 andres@anarazel.de 3494 : 22889 : aggstate->phases = palloc0(numPhases * sizeof(AggStatePerPhaseData));
3495 : :
3136 rhodiumtoad@postgres 3496 : 22889 : aggstate->num_hashes = numHashes;
3497 [ + + ]: 22889 : if (numHashes)
3498 : : {
3499 : 3499 : aggstate->perhash = palloc0(sizeof(AggStatePerHashData) * numHashes);
3500 : 3499 : aggstate->phases[0].numsets = 0;
3501 : 3499 : aggstate->phases[0].gset_lengths = palloc(numHashes * sizeof(int));
3502 : 3499 : aggstate->phases[0].grouped_cols = palloc(numHashes * sizeof(Bitmapset *));
3503 : : }
3504 : :
3505 : 22889 : phase = 0;
3506 [ + + ]: 46277 : for (phaseidx = 0; phaseidx <= list_length(node->chain); ++phaseidx)
3507 : : {
3508 : : Agg *aggnode;
3509 : : Sort *sortnode;
3510 : :
3511 [ + + ]: 23388 : if (phaseidx > 0)
3512 : : {
3122 tgl@sss.pgh.pa.us 3513 : 499 : aggnode = list_nth_node(Agg, node->chain, phaseidx - 1);
1208 3514 : 499 : sortnode = castNode(Sort, outerPlan(aggnode));
3515 : : }
3516 : : else
3517 : : {
3817 andres@anarazel.de 3518 : 22889 : aggnode = node;
3519 : 22889 : sortnode = NULL;
3520 : : }
3521 : :
3136 rhodiumtoad@postgres 3522 [ + + - + ]: 23388 : Assert(phase <= 1 || sortnode);
3523 : :
3524 [ + + ]: 23388 : if (aggnode->aggstrategy == AGG_HASHED
3525 [ + + ]: 19748 : || aggnode->aggstrategy == AGG_MIXED)
3817 andres@anarazel.de 3526 : 3756 : {
3136 rhodiumtoad@postgres 3527 : 3756 : AggStatePerPhase phasedata = &aggstate->phases[0];
3528 : : AggStatePerHash perhash;
3529 : 3756 : Bitmapset *cols = NULL;
3530 : :
3531 [ - + ]: 3756 : Assert(phase == 0);
3532 : 3756 : i = phasedata->numsets++;
3533 : 3756 : perhash = &aggstate->perhash[i];
3534 : :
3535 : : /* phase 0 always points to the "real" Agg in the hash case */
3536 : 3756 : phasedata->aggnode = node;
3537 : 3756 : phasedata->aggstrategy = node->aggstrategy;
3538 : :
3539 : : /* but the actual Agg node representing this hash is saved here */
3540 : 3756 : perhash->aggnode = aggnode;
3541 : :
3542 : 3756 : phasedata->gset_lengths[i] = perhash->numCols = aggnode->numCols;
3543 : :
3544 [ + + ]: 9470 : for (j = 0; j < aggnode->numCols; ++j)
3545 : 5714 : cols = bms_add_member(cols, aggnode->grpColIdx[j]);
3546 : :
3547 : 3756 : phasedata->grouped_cols[i] = cols;
3548 : :
3549 : 3756 : all_grouped_cols = bms_add_members(all_grouped_cols, cols);
3550 : 3756 : continue;
3551 : : }
3552 : : else
3553 : : {
3554 : 19632 : AggStatePerPhase phasedata = &aggstate->phases[++phase];
3555 : : int num_sets;
3556 : :
3557 : 19632 : phasedata->numsets = num_sets = list_length(aggnode->groupingSets);
3558 : :
3559 [ + + ]: 19632 : if (num_sets)
3560 : : {
3561 : 500 : phasedata->gset_lengths = palloc(num_sets * sizeof(int));
3562 : 500 : phasedata->grouped_cols = palloc(num_sets * sizeof(Bitmapset *));
3563 : :
3564 : 500 : i = 0;
3565 [ + - + + : 1468 : foreach(l, aggnode->groupingSets)
+ + ]
3566 : : {
3567 : 968 : int current_length = list_length(lfirst(l));
3568 : 968 : Bitmapset *cols = NULL;
3569 : :
3570 : : /* planner forces this to be correct */
3571 [ + + ]: 1902 : for (j = 0; j < current_length; ++j)
3572 : 934 : cols = bms_add_member(cols, aggnode->grpColIdx[j]);
3573 : :
3574 : 968 : phasedata->grouped_cols[i] = cols;
3575 : 968 : phasedata->gset_lengths[i] = current_length;
3576 : :
3577 : 968 : ++i;
3578 : : }
3579 : :
3580 : 500 : all_grouped_cols = bms_add_members(all_grouped_cols,
3050 tgl@sss.pgh.pa.us 3581 : 500 : phasedata->grouped_cols[0]);
3582 : : }
3583 : : else
3584 : : {
3136 rhodiumtoad@postgres 3585 [ - + ]: 19132 : Assert(phaseidx == 0);
3586 : :
3587 : 19132 : phasedata->gset_lengths = NULL;
3588 : 19132 : phasedata->grouped_cols = NULL;
3589 : : }
3590 : :
3591 : : /*
3592 : : * If we are grouping, precompute fmgr lookup data for inner loop.
3593 : : */
3594 [ + + ]: 19632 : if (aggnode->aggstrategy == AGG_SORTED)
3595 : : {
3596 : : /*
3597 : : * Build a separate function for each subset of columns that
3598 : : * need to be compared.
3599 : : */
3600 : 1259 : phasedata->eqfunctions =
2811 andres@anarazel.de 3601 : 1259 : (ExprState **) palloc0(aggnode->numCols * sizeof(ExprState *));
3602 : :
3603 : : /* for each grouping set */
1118 drowley@postgresql.o 3604 [ + + ]: 2079 : for (int k = 0; k < phasedata->numsets; k++)
3605 : : {
3606 : 820 : int length = phasedata->gset_lengths[k];
3607 : :
3608 : : /* nothing to do for empty grouping set */
1029 tgl@sss.pgh.pa.us 3609 [ + + ]: 820 : if (length == 0)
3610 : 169 : continue;
3611 : :
3612 : : /* if we already had one of this length, it'll do */
2811 andres@anarazel.de 3613 [ + + ]: 651 : if (phasedata->eqfunctions[length - 1] != NULL)
3614 : 69 : continue;
3615 : :
3616 : 582 : phasedata->eqfunctions[length - 1] =
3617 : 582 : execTuplesMatchPrepare(scanDesc,
3618 : : length,
3619 : 582 : aggnode->grpColIdx,
3620 : 582 : aggnode->grpOperators,
2411 peter@eisentraut.org 3621 : 582 : aggnode->grpCollations,
3622 : : (PlanState *) aggstate);
3623 : : }
3624 : :
3625 : : /* and for all grouped columns, unless already computed */
1013 tgl@sss.pgh.pa.us 3626 [ + + ]: 1259 : if (aggnode->numCols > 0 &&
3627 [ + + ]: 1212 : phasedata->eqfunctions[aggnode->numCols - 1] == NULL)
3628 : : {
2811 andres@anarazel.de 3629 : 824 : phasedata->eqfunctions[aggnode->numCols - 1] =
3630 : 824 : execTuplesMatchPrepare(scanDesc,
3631 : : aggnode->numCols,
3632 : 824 : aggnode->grpColIdx,
3633 : 824 : aggnode->grpOperators,
2411 peter@eisentraut.org 3634 : 824 : aggnode->grpCollations,
3635 : : (PlanState *) aggstate);
3636 : : }
3637 : : }
3638 : :
3136 rhodiumtoad@postgres 3639 : 19632 : phasedata->aggnode = aggnode;
3640 : 19632 : phasedata->aggstrategy = aggnode->aggstrategy;
3641 : 19632 : phasedata->sortnode = sortnode;
3642 : : }
3643 : : }
3644 : :
3645 : : /*
3646 : : * Convert all_grouped_cols to a descending-order list.
3647 : : */
3817 andres@anarazel.de 3648 : 22889 : i = -1;
3649 [ + + ]: 28913 : while ((i = bms_next_member(all_grouped_cols, i)) >= 0)
3650 : 6024 : aggstate->all_grouped_cols = lcons_int(i, aggstate->all_grouped_cols);
3651 : :
3652 : : /*
3653 : : * Set up aggregate-result storage in the output expr context, and also
3654 : : * allocate my private per-agg working storage
3655 : : */
8362 tgl@sss.pgh.pa.us 3656 : 22889 : econtext = aggstate->ss.ps.ps_ExprContext;
8384 bruce@momjian.us 3657 : 22889 : econtext->ecxt_aggvalues = (Datum *) palloc0(sizeof(Datum) * numaggs);
3658 : 22889 : econtext->ecxt_aggnulls = (bool *) palloc0(sizeof(bool) * numaggs);
3659 : :
3737 heikki.linnakangas@i 3660 : 22889 : peraggs = (AggStatePerAgg) palloc0(sizeof(AggStatePerAggData) * numaggs);
1798 3661 : 22889 : pertransstates = (AggStatePerTrans) palloc0(sizeof(AggStatePerTransData) * numtrans);
3662 : :
3737 3663 : 22889 : aggstate->peragg = peraggs;
3664 : 22889 : aggstate->pertrans = pertransstates;
3665 : :
3666 : :
2848 andres@anarazel.de 3667 : 22889 : aggstate->all_pergroups =
3668 : 22889 : (AggStatePerGroup *) palloc0(sizeof(AggStatePerGroup)
3669 : 22889 : * (numGroupingSets + numHashes));
3670 : 22889 : pergroups = aggstate->all_pergroups;
3671 : :
3672 [ + + ]: 22889 : if (node->aggstrategy != AGG_HASHED)
3673 : : {
3674 [ + + ]: 39444 : for (i = 0; i < numGroupingSets; i++)
3675 : : {
3676 : 19938 : pergroups[i] = (AggStatePerGroup) palloc0(sizeof(AggStatePerGroupData)
3677 : : * numaggs);
3678 : : }
3679 : :
3680 : 19506 : aggstate->pergroups = pergroups;
3681 : 19506 : pergroups += numGroupingSets;
3682 : : }
3683 : :
3684 : : /*
3685 : : * Hashing can only appear in the initial phase.
3686 : : */
3136 rhodiumtoad@postgres 3687 [ + + ]: 22889 : if (use_hashing)
3688 : : {
1992 tgl@sss.pgh.pa.us 3689 : 3499 : Plan *outerplan = outerPlan(node);
3690 : 3499 : uint64 totalGroups = 0;
3691 : :
1933 jdavis@postgresql.or 3692 : 3499 : aggstate->hash_spill_rslot = ExecInitExtraTupleSlot(estate, scanDesc,
3693 : : &TTSOpsMinimalTuple);
3694 : 3499 : aggstate->hash_spill_wslot = ExecInitExtraTupleSlot(estate, scanDesc,
3695 : : &TTSOpsVirtual);
3696 : :
3697 : : /* this is an array of pointers, not structures */
2848 andres@anarazel.de 3698 : 3499 : aggstate->hash_pergroup = pergroups;
3699 : :
1992 tgl@sss.pgh.pa.us 3700 : 6998 : aggstate->hashentrysize = hash_agg_entry_size(aggstate->numtrans,
3701 : 3499 : outerplan->plan_width,
3702 : : node->transitionSpace);
3703 : :
3704 : : /*
3705 : : * Consider all of the grouping sets together when setting the limits
3706 : : * and estimating the number of partitions. This can be inaccurate
3707 : : * when there is more than one grouping set, but should still be
3708 : : * reasonable.
3709 : : */
1118 drowley@postgresql.o 3710 [ + + ]: 7255 : for (int k = 0; k < aggstate->num_hashes; k++)
3711 : 3756 : totalGroups += aggstate->perhash[k].aggnode->numGroups;
3712 : :
2049 jdavis@postgresql.or 3713 : 3499 : hash_agg_set_limits(aggstate->hashentrysize, totalGroups, 0,
3714 : : &aggstate->hash_mem_limit,
3715 : : &aggstate->hash_ngroups_limit,
3716 : : &aggstate->hash_planned_partitions);
3136 rhodiumtoad@postgres 3717 : 3499 : find_hash_columns(aggstate);
3718 : :
3719 : : /* Skip massive memory allocation if we are just doing EXPLAIN */
1804 heikki.linnakangas@i 3720 [ + + ]: 3499 : if (!(eflags & EXEC_FLAG_EXPLAIN_ONLY))
3721 : 2576 : build_hash_tables(aggstate);
3722 : :
8391 tgl@sss.pgh.pa.us 3723 : 3499 : aggstate->table_filled = false;
3724 : :
3725 : : /* Initialize this to 1, meaning nothing spilled, yet */
1916 drowley@postgresql.o 3726 : 3499 : aggstate->hash_batches_used = 1;
3727 : : }
3728 : :
3729 : : /*
3730 : : * Initialize current phase-dependent values to initial phase. The initial
3731 : : * phase is 1 (first sort pass) for all strategies that use sorting (if
3732 : : * hashing is being done too, then phase 0 is processed last); but if only
3733 : : * hashing is being done, then phase 0 is all there is.
3734 : : */
3136 rhodiumtoad@postgres 3735 [ + + ]: 22889 : if (node->aggstrategy == AGG_HASHED)
3736 : : {
3737 : 3383 : aggstate->current_phase = 0;
3738 : 3383 : initialize_phase(aggstate, 0);
3739 : 3383 : select_current_set(aggstate, 0, true);
3740 : : }
3741 : : else
3742 : : {
3743 : 19506 : aggstate->current_phase = 1;
3744 : 19506 : initialize_phase(aggstate, 1);
3745 : 19506 : select_current_set(aggstate, 0, false);
3746 : : }
3747 : :
3748 : : /*
3749 : : * Perform lookups of aggregate function info, and initialize the
3750 : : * unchanging fields of the per-agg and per-trans data.
3751 : : */
7824 neilc@samurai.com 3752 [ + + + + : 48835 : foreach(l, aggstate->aggs)
+ + ]
3753 : : {
1798 heikki.linnakangas@i 3754 : 25949 : Aggref *aggref = lfirst(l);
3755 : : AggStatePerAgg peragg;
3756 : : AggStatePerTrans pertrans;
3757 : : Oid aggTransFnInputTypes[FUNC_MAX_ARGS];
3758 : : int numAggTransFnArgs;
3759 : : int numDirectArgs;
3760 : : HeapTuple aggTuple;
3761 : : Form_pg_aggregate aggform;
3762 : : AclResult aclresult;
3763 : : Oid finalfn_oid;
3764 : : Oid serialfn_oid,
3765 : : deserialfn_oid;
3766 : : Oid aggOwner;
3767 : : Expr *finalfnexpr;
3768 : : Oid aggtranstype;
3769 : :
3770 : : /* Planner should have assigned aggregate to correct level */
8179 tgl@sss.pgh.pa.us 3771 [ - + ]: 25949 : Assert(aggref->agglevelsup == 0);
3772 : : /* ... and the split mode should match */
3410 3773 [ - + ]: 25949 : Assert(aggref->aggsplit == aggstate->aggsplit);
3774 : :
1798 heikki.linnakangas@i 3775 : 25949 : peragg = &peraggs[aggref->aggno];
3776 : :
3777 : : /* Check if we initialized the state for this aggregate already. */
3778 [ + + ]: 25949 : if (peragg->aggref != NULL)
8301 tgl@sss.pgh.pa.us 3779 : 242 : continue;
3780 : :
3737 heikki.linnakangas@i 3781 : 25707 : peragg->aggref = aggref;
1798 3782 : 25707 : peragg->transno = aggref->aggtransno;
3783 : :
3784 : : /* Fetch the pg_aggregate row */
5734 rhaas@postgresql.org 3785 : 25707 : aggTuple = SearchSysCache1(AGGFNOID,
3786 : : ObjectIdGetDatum(aggref->aggfnoid));
9528 tgl@sss.pgh.pa.us 3787 [ - + ]: 25707 : if (!HeapTupleIsValid(aggTuple))
8134 tgl@sss.pgh.pa.us 3788 [ # # ]:UBC 0 : elog(ERROR, "cache lookup failed for aggregate %u",
3789 : : aggref->aggfnoid);
9528 tgl@sss.pgh.pa.us 3790 :CBC 25707 : aggform = (Form_pg_aggregate) GETSTRUCT(aggTuple);
3791 : :
3792 : : /* Check permission to call aggregate function */
1079 peter@eisentraut.org 3793 : 25707 : aclresult = object_aclcheck(ProcedureRelationId, aggref->aggfnoid, GetUserId(),
3794 : : ACL_EXECUTE);
8582 tgl@sss.pgh.pa.us 3795 [ + + ]: 25707 : if (aclresult != ACLCHECK_OK)
2886 peter_e@gmx.net 3796 : 3 : aclcheck_error(aclresult, OBJECT_AGGREGATE,
8123 tgl@sss.pgh.pa.us 3797 : 3 : get_func_name(aggref->aggfnoid));
4581 rhaas@postgresql.org 3798 [ - + ]: 25704 : InvokeFunctionExecuteHook(aggref->aggfnoid);
3799 : :
3800 : : /* planner recorded transition state type in the Aggref itself */
3419 tgl@sss.pgh.pa.us 3801 : 25704 : aggtranstype = aggref->aggtranstype;
3802 [ - + ]: 25704 : Assert(OidIsValid(aggtranstype));
3803 : :
3804 : : /* Final function only required if we're finalizing the aggregates */
3410 3805 [ + + ]: 25704 : if (DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit))
3568 rhaas@postgresql.org 3806 : 2697 : peragg->finalfn_oid = finalfn_oid = InvalidOid;
3807 : : else
3410 tgl@sss.pgh.pa.us 3808 : 23007 : peragg->finalfn_oid = finalfn_oid = aggform->aggfinalfn;
3809 : :
3499 rhaas@postgresql.org 3810 : 25704 : serialfn_oid = InvalidOid;
3811 : 25704 : deserialfn_oid = InvalidOid;
3812 : :
3813 : : /*
3814 : : * Check if serialization/deserialization is required. We only do it
3815 : : * for aggregates that have transtype INTERNAL.
3816 : : */
3410 tgl@sss.pgh.pa.us 3817 [ + + ]: 25704 : if (aggtranstype == INTERNALOID)
3818 : : {
3819 : : /*
3820 : : * The planner should only have generated a serialize agg node if
3821 : : * every aggregate with an INTERNAL state has a serialization
3822 : : * function. Verify that.
3823 : : */
3824 [ + + ]: 10398 : if (DO_AGGSPLIT_SERIALIZE(aggstate->aggsplit))
3825 : : {
3826 : : /* serialization only valid when not running finalfn */
3827 [ - + ]: 168 : Assert(DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit));
3828 : :
3829 [ - + ]: 168 : if (!OidIsValid(aggform->aggserialfn))
3410 tgl@sss.pgh.pa.us 3830 [ # # ]:UBC 0 : elog(ERROR, "serialfunc not provided for serialization aggregation");
3499 rhaas@postgresql.org 3831 :CBC 168 : serialfn_oid = aggform->aggserialfn;
3832 : : }
3833 : :
3834 : : /* Likewise for deserialization functions */
3410 tgl@sss.pgh.pa.us 3835 [ + + ]: 10398 : if (DO_AGGSPLIT_DESERIALIZE(aggstate->aggsplit))
3836 : : {
3837 : : /* deserialization only valid when combining states */
3838 [ - + ]: 60 : Assert(DO_AGGSPLIT_COMBINE(aggstate->aggsplit));
3839 : :
3840 [ - + ]: 60 : if (!OidIsValid(aggform->aggdeserialfn))
3410 tgl@sss.pgh.pa.us 3841 [ # # ]:UBC 0 : elog(ERROR, "deserialfunc not provided for deserialization aggregation");
3499 rhaas@postgresql.org 3842 :CBC 60 : deserialfn_oid = aggform->aggdeserialfn;
3843 : : }
3844 : : }
3845 : :
3846 : : /* Check that aggregate owner has permission to call component fns */
3847 : : {
3848 : : HeapTuple procTuple;
3849 : :
5734 3850 : 25704 : procTuple = SearchSysCache1(PROCOID,
3851 : : ObjectIdGetDatum(aggref->aggfnoid));
7578 tgl@sss.pgh.pa.us 3852 [ - + ]: 25704 : if (!HeapTupleIsValid(procTuple))
7578 tgl@sss.pgh.pa.us 3853 [ # # ]:UBC 0 : elog(ERROR, "cache lookup failed for function %u",
3854 : : aggref->aggfnoid);
7578 tgl@sss.pgh.pa.us 3855 :CBC 25704 : aggOwner = ((Form_pg_proc) GETSTRUCT(procTuple))->proowner;
3856 : 25704 : ReleaseSysCache(procTuple);
3857 : :
3858 [ + + ]: 25704 : if (OidIsValid(finalfn_oid))
3859 : : {
1079 peter@eisentraut.org 3860 : 11173 : aclresult = object_aclcheck(ProcedureRelationId, finalfn_oid, aggOwner,
3861 : : ACL_EXECUTE);
7578 tgl@sss.pgh.pa.us 3862 [ - + ]: 11173 : if (aclresult != ACLCHECK_OK)
2886 peter_e@gmx.net 3863 :UBC 0 : aclcheck_error(aclresult, OBJECT_FUNCTION,
7578 tgl@sss.pgh.pa.us 3864 : 0 : get_func_name(finalfn_oid));
4581 rhaas@postgresql.org 3865 [ - + ]:CBC 11173 : InvokeFunctionExecuteHook(finalfn_oid);
3866 : : }
3499 3867 [ + + ]: 25704 : if (OidIsValid(serialfn_oid))
3868 : : {
1079 peter@eisentraut.org 3869 : 168 : aclresult = object_aclcheck(ProcedureRelationId, serialfn_oid, aggOwner,
3870 : : ACL_EXECUTE);
3499 rhaas@postgresql.org 3871 [ - + ]: 168 : if (aclresult != ACLCHECK_OK)
2886 peter_e@gmx.net 3872 :UBC 0 : aclcheck_error(aclresult, OBJECT_FUNCTION,
3499 rhaas@postgresql.org 3873 : 0 : get_func_name(serialfn_oid));
3499 rhaas@postgresql.org 3874 [ - + ]:CBC 168 : InvokeFunctionExecuteHook(serialfn_oid);
3875 : : }
3876 [ + + ]: 25704 : if (OidIsValid(deserialfn_oid))
3877 : : {
1079 peter@eisentraut.org 3878 : 60 : aclresult = object_aclcheck(ProcedureRelationId, deserialfn_oid, aggOwner,
3879 : : ACL_EXECUTE);
3499 rhaas@postgresql.org 3880 [ - + ]: 60 : if (aclresult != ACLCHECK_OK)
2886 peter_e@gmx.net 3881 :UBC 0 : aclcheck_error(aclresult, OBJECT_FUNCTION,
3499 rhaas@postgresql.org 3882 : 0 : get_func_name(deserialfn_oid));
3499 rhaas@postgresql.org 3883 [ - + ]:CBC 60 : InvokeFunctionExecuteHook(deserialfn_oid);
3884 : : }
3885 : : }
3886 : :
3887 : : /*
3888 : : * Get actual datatypes of the (nominal) aggregate inputs. These
3889 : : * could be different from the agg's declared input types, when the
3890 : : * agg accepts ANY or a polymorphic type.
3891 : : */
1576 drowley@postgresql.o 3892 : 25704 : numAggTransFnArgs = get_aggregate_argtypes(aggref,
3893 : : aggTransFnInputTypes);
3894 : :
3895 : : /* Count the "direct" arguments, if any */
4326 tgl@sss.pgh.pa.us 3896 : 25704 : numDirectArgs = list_length(aggref->aggdirectargs);
3897 : :
3898 : : /* Detect how many arguments to pass to the finalfn */
3737 heikki.linnakangas@i 3899 [ + + ]: 25704 : if (aggform->aggfinalextra)
1576 drowley@postgresql.o 3900 : 7203 : peragg->numFinalArgs = numAggTransFnArgs + 1;
3901 : : else
3737 heikki.linnakangas@i 3902 : 18501 : peragg->numFinalArgs = numDirectArgs + 1;
3903 : :
3904 : : /* Initialize any direct-argument expressions */
2933 tgl@sss.pgh.pa.us 3905 : 25704 : peragg->aggdirectargs = ExecInitExprList(aggref->aggdirectargs,
3906 : : (PlanState *) aggstate);
3907 : :
3908 : : /*
3909 : : * build expression trees using actual argument & result types for the
3910 : : * finalfn, if it exists and is required.
3911 : : */
8154 3912 [ + + ]: 25704 : if (OidIsValid(finalfn_oid))
3913 : : {
1576 drowley@postgresql.o 3914 : 11173 : build_aggregate_finalfn_expr(aggTransFnInputTypes,
3915 : : peragg->numFinalArgs,
3916 : : aggtranstype,
3917 : : aggref->aggtype,
3918 : : aggref->inputcollid,
3919 : : finalfn_oid,
3920 : : &finalfnexpr);
3737 heikki.linnakangas@i 3921 : 11173 : fmgr_info(finalfn_oid, &peragg->finalfn);
3922 : 11173 : fmgr_info_set_expr((Node *) finalfnexpr, &peragg->finalfn);
3923 : : }
3924 : :
3925 : : /* get info about the output value's datatype */
3410 tgl@sss.pgh.pa.us 3926 : 25704 : get_typlenbyval(aggref->aggtype,
3927 : : &peragg->resulttypeLen,
3928 : : &peragg->resulttypeByVal);
3929 : :
3930 : : /*
3931 : : * Build working state for invoking the transition function, if we
3932 : : * haven't done it already.
3933 : : */
1798 heikki.linnakangas@i 3934 : 25704 : pertrans = &pertransstates[aggref->aggtransno];
3935 [ + + ]: 25704 : if (pertrans->aggref == NULL)
3936 : : {
3937 : : Datum textInitVal;
3938 : : Datum initValue;
3939 : : bool initValueIsNull;
3940 : : Oid transfn_oid;
3941 : :
3942 : : /*
3943 : : * If this aggregation is performing state combines, then instead
3944 : : * of using the transition function, we'll use the combine
3945 : : * function.
3946 : : */
3947 [ + + ]: 25575 : if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit))
3948 : : {
3949 : 1097 : transfn_oid = aggform->aggcombinefn;
3950 : :
3951 : : /* If not set then the planner messed up */
3952 [ - + ]: 1097 : if (!OidIsValid(transfn_oid))
1798 heikki.linnakangas@i 3953 [ # # ]:UBC 0 : elog(ERROR, "combinefn not set for aggregate function");
3954 : : }
3955 : : else
1798 heikki.linnakangas@i 3956 :CBC 24478 : transfn_oid = aggform->aggtransfn;
3957 : :
1079 peter@eisentraut.org 3958 : 25575 : aclresult = object_aclcheck(ProcedureRelationId, transfn_oid, aggOwner, ACL_EXECUTE);
1798 heikki.linnakangas@i 3959 [ - + ]: 25575 : if (aclresult != ACLCHECK_OK)
1798 heikki.linnakangas@i 3960 :UBC 0 : aclcheck_error(aclresult, OBJECT_FUNCTION,
3961 : 0 : get_func_name(transfn_oid));
1798 heikki.linnakangas@i 3962 [ - + ]:CBC 25575 : InvokeFunctionExecuteHook(transfn_oid);
3963 : :
3964 : : /*
3965 : : * initval is potentially null, so don't try to access it as a
3966 : : * struct field. Must do it the hard way with SysCacheGetAttr.
3967 : : */
3968 : 25575 : textInitVal = SysCacheGetAttr(AGGFNOID, aggTuple,
3969 : : Anum_pg_aggregate_agginitval,
3970 : : &initValueIsNull);
3971 [ + + ]: 25575 : if (initValueIsNull)
3972 : 14999 : initValue = (Datum) 0;
3973 : : else
3974 : 10576 : initValue = GetAggInitVal(textInitVal, aggtranstype);
3975 : :
1576 drowley@postgresql.o 3976 [ + + ]: 25575 : if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit))
3977 : : {
3978 : 1097 : Oid combineFnInputTypes[] = {aggtranstype,
3979 : : aggtranstype};
3980 : :
3981 : : /*
3982 : : * When combining there's only one input, the to-be-combined
3983 : : * transition value. The transition value is not counted
3984 : : * here.
3985 : : */
3986 : 1097 : pertrans->numTransInputs = 1;
3987 : :
3988 : : /* aggcombinefn always has two arguments of aggtranstype */
3989 : 1097 : build_pertrans_for_aggref(pertrans, aggstate, estate,
3990 : : aggref, transfn_oid, aggtranstype,
3991 : : serialfn_oid, deserialfn_oid,
3992 : : initValue, initValueIsNull,
3993 : : combineFnInputTypes, 2);
3994 : :
3995 : : /*
3996 : : * Ensure that a combine function to combine INTERNAL states
3997 : : * is not strict. This should have been checked during CREATE
3998 : : * AGGREGATE, but the strict property could have been changed
3999 : : * since then.
4000 : : */
4001 [ + + - + ]: 1097 : if (pertrans->transfn.fn_strict && aggtranstype == INTERNALOID)
1576 drowley@postgresql.o 4002 [ # # ]:UBC 0 : ereport(ERROR,
4003 : : (errcode(ERRCODE_INVALID_FUNCTION_DEFINITION),
4004 : : errmsg("combine function with transition type %s must not be declared STRICT",
4005 : : format_type_be(aggtranstype))));
4006 : : }
4007 : : else
4008 : : {
4009 : : /* Detect how many arguments to pass to the transfn */
1576 drowley@postgresql.o 4010 [ + + ]:CBC 24478 : if (AGGKIND_IS_ORDERED_SET(aggref->aggkind))
4011 : 126 : pertrans->numTransInputs = list_length(aggref->args);
4012 : : else
4013 : 24352 : pertrans->numTransInputs = numAggTransFnArgs;
4014 : :
4015 : 24478 : build_pertrans_for_aggref(pertrans, aggstate, estate,
4016 : : aggref, transfn_oid, aggtranstype,
4017 : : serialfn_oid, deserialfn_oid,
4018 : : initValue, initValueIsNull,
4019 : : aggTransFnInputTypes,
4020 : : numAggTransFnArgs);
4021 : :
4022 : : /*
4023 : : * If the transfn is strict and the initval is NULL, make sure
4024 : : * input type and transtype are the same (or at least
4025 : : * binary-compatible), so that it's OK to use the first
4026 : : * aggregated input value as the initial transValue. This
4027 : : * should have been checked at agg definition time, but we
4028 : : * must check again in case the transfn's strictness property
4029 : : * has been changed.
4030 : : */
4031 [ + + + + ]: 24478 : if (pertrans->transfn.fn_strict && pertrans->initValueIsNull)
4032 : : {
4033 [ + - ]: 2533 : if (numAggTransFnArgs <= numDirectArgs ||
4034 [ - + ]: 2533 : !IsBinaryCoercible(aggTransFnInputTypes[numDirectArgs],
4035 : : aggtranstype))
1576 drowley@postgresql.o 4036 [ # # ]:UBC 0 : ereport(ERROR,
4037 : : (errcode(ERRCODE_INVALID_FUNCTION_DEFINITION),
4038 : : errmsg("aggregate %u needs to have compatible input type and transition type",
4039 : : aggref->aggfnoid)));
4040 : : }
4041 : : }
4042 : : }
4043 : : else
1798 heikki.linnakangas@i 4044 :CBC 129 : pertrans->aggshared = true;
3737 4045 : 25704 : ReleaseSysCache(aggTuple);
4046 : : }
4047 : :
4048 : : /*
4049 : : * Last, check whether any more aggregates got added onto the node while
4050 : : * we processed the expressions for the aggregate arguments (including not
4051 : : * only the regular arguments and FILTER expressions handled immediately
4052 : : * above, but any direct arguments we might've handled earlier). If so,
4053 : : * we have nested aggregate functions, which is semantically nonsensical,
4054 : : * so complain. (This should have been caught by the parser, so we don't
4055 : : * need to work hard on a helpful error message; but we defend against it
4056 : : * here anyway, just to be sure.)
4057 : : */
1798 4058 [ - + ]: 22886 : if (numaggrefs != list_length(aggstate->aggs))
2848 andres@anarazel.de 4059 [ # # ]:UBC 0 : ereport(ERROR,
4060 : : (errcode(ERRCODE_GROUPING_ERROR),
4061 : : errmsg("aggregate function calls cannot be nested")));
4062 : :
4063 : : /*
4064 : : * Build expressions doing all the transition work at once. We build a
4065 : : * different one for each phase, as the number of transition function
4066 : : * invocation can differ between phases. Note this'll work both for
4067 : : * transition and combination functions (although there'll only be one
4068 : : * phase in the latter case).
4069 : : */
2848 andres@anarazel.de 4070 [ + + ]:CBC 65401 : for (phaseidx = 0; phaseidx < aggstate->numphases; phaseidx++)
4071 : : {
4072 : 42515 : AggStatePerPhase phase = &aggstate->phases[phaseidx];
4073 : 42515 : bool dohash = false;
4074 : 42515 : bool dosort = false;
4075 : :
4076 : : /* phase 0 doesn't necessarily exist */
4077 [ + + ]: 42515 : if (!phase->aggnode)
4078 : 19387 : continue;
4079 : :
4080 [ + + + + ]: 23128 : if (aggstate->aggstrategy == AGG_MIXED && phaseidx == 1)
4081 : : {
4082 : : /*
4083 : : * Phase one, and only phase one, in a mixed agg performs both
4084 : : * sorting and aggregation.
4085 : : */
4086 : 116 : dohash = true;
4087 : 116 : dosort = true;
4088 : : }
4089 [ + + + + ]: 23012 : else if (aggstate->aggstrategy == AGG_MIXED && phaseidx == 0)
4090 : : {
4091 : : /*
4092 : : * No need to compute a transition function for an AGG_MIXED phase
4093 : : * 0 - the contents of the hashtables will have been computed
4094 : : * during phase 1.
4095 : : */
4096 : 116 : continue;
4097 : : }
4098 [ + + ]: 22896 : else if (phase->aggstrategy == AGG_PLAIN ||
4099 [ + + ]: 4611 : phase->aggstrategy == AGG_SORTED)
4100 : : {
4101 : 19513 : dohash = false;
4102 : 19513 : dosort = true;
4103 : : }
4104 [ + - ]: 3383 : else if (phase->aggstrategy == AGG_HASHED)
4105 : : {
4106 : 3383 : dohash = true;
4107 : 3383 : dosort = false;
4108 : : }
4109 : : else
2848 andres@anarazel.de 4110 :UBC 0 : Assert(false);
4111 : :
2063 jdavis@postgresql.or 4112 :CBC 23012 : phase->evaltrans = ExecBuildAggTrans(aggstate, phase, dosort, dohash,
4113 : : false);
4114 : :
4115 : : /* cache compiled expression for outer slot without NULL check */
2049 4116 : 23012 : phase->evaltrans_cache[0][0] = phase->evaltrans;
4117 : : }
4118 : :
3737 heikki.linnakangas@i 4119 : 22886 : return aggstate;
4120 : : }
4121 : :
4122 : : /*
4123 : : * Build the state needed to calculate a state value for an aggregate.
4124 : : *
4125 : : * This initializes all the fields in 'pertrans'. 'aggref' is the aggregate
4126 : : * to initialize the state for. 'transfn_oid', 'aggtranstype', and the rest
4127 : : * of the arguments could be calculated from 'aggref', but the caller has
4128 : : * calculated them already, so might as well pass them.
4129 : : *
4130 : : * 'transfn_oid' may be either the Oid of the aggtransfn or the aggcombinefn.
4131 : : */
4132 : : static void
4133 : 25575 : build_pertrans_for_aggref(AggStatePerTrans pertrans,
4134 : : AggState *aggstate, EState *estate,
4135 : : Aggref *aggref,
4136 : : Oid transfn_oid, Oid aggtranstype,
4137 : : Oid aggserialfn, Oid aggdeserialfn,
4138 : : Datum initValue, bool initValueIsNull,
4139 : : Oid *inputTypes, int numArguments)
4140 : : {
4141 : 25575 : int numGroupingSets = Max(aggstate->maxsets, 1);
4142 : : Expr *transfnexpr;
4143 : : int numTransArgs;
3499 rhaas@postgresql.org 4144 : 25575 : Expr *serialfnexpr = NULL;
4145 : 25575 : Expr *deserialfnexpr = NULL;
4146 : : ListCell *lc;
4147 : : int numInputs;
4148 : : int numDirectArgs;
4149 : : List *sortlist;
4150 : : int numSortCols;
4151 : : int numDistinctCols;
4152 : : int i;
4153 : :
4154 : : /* Begin filling in the pertrans data */
3737 heikki.linnakangas@i 4155 : 25575 : pertrans->aggref = aggref;
2933 tgl@sss.pgh.pa.us 4156 : 25575 : pertrans->aggshared = false;
3737 heikki.linnakangas@i 4157 : 25575 : pertrans->aggCollation = aggref->inputcollid;
1576 drowley@postgresql.o 4158 : 25575 : pertrans->transfn_oid = transfn_oid;
3499 rhaas@postgresql.org 4159 : 25575 : pertrans->serialfn_oid = aggserialfn;
4160 : 25575 : pertrans->deserialfn_oid = aggdeserialfn;
3737 heikki.linnakangas@i 4161 : 25575 : pertrans->initValue = initValue;
4162 : 25575 : pertrans->initValueIsNull = initValueIsNull;
4163 : :
4164 : : /* Count the "direct" arguments, if any */
4165 : 25575 : numDirectArgs = list_length(aggref->aggdirectargs);
4166 : :
4167 : : /* Count the number of aggregated input columns */
4168 : 25575 : pertrans->numInputs = numInputs = list_length(aggref->args);
4169 : :
4170 : 25575 : pertrans->aggtranstype = aggtranstype;
4171 : :
4172 : : /* account for the current transition state */
1576 drowley@postgresql.o 4173 : 25575 : numTransArgs = pertrans->numTransInputs + 1;
4174 : :
4175 : : /*
4176 : : * Set up infrastructure for calling the transfn. Note that invtransfn is
4177 : : * not needed here.
4178 : : */
4179 : 25575 : build_aggregate_transfn_expr(inputTypes,
4180 : : numArguments,
4181 : : numDirectArgs,
4182 : 25575 : aggref->aggvariadic,
4183 : : aggtranstype,
4184 : : aggref->inputcollid,
4185 : : transfn_oid,
4186 : : InvalidOid,
4187 : : &transfnexpr,
4188 : : NULL);
4189 : :
4190 : 25575 : fmgr_info(transfn_oid, &pertrans->transfn);
4191 : 25575 : fmgr_info_set_expr((Node *) transfnexpr, &pertrans->transfn);
4192 : :
4193 : 25575 : pertrans->transfn_fcinfo =
4194 : 25575 : (FunctionCallInfo) palloc(SizeForFunctionCallInfo(numTransArgs));
4195 : 25575 : InitFunctionCallInfoData(*pertrans->transfn_fcinfo,
4196 : : &pertrans->transfn,
4197 : : numTransArgs,
4198 : : pertrans->aggCollation,
4199 : : (Node *) aggstate, NULL);
4200 : :
4201 : : /* get info about the state value's datatype */
3737 heikki.linnakangas@i 4202 : 25575 : get_typlenbyval(aggtranstype,
4203 : : &pertrans->transtypeLen,
4204 : : &pertrans->transtypeByVal);
4205 : :
3499 rhaas@postgresql.org 4206 [ + + ]: 25575 : if (OidIsValid(aggserialfn))
4207 : : {
3414 tgl@sss.pgh.pa.us 4208 : 168 : build_aggregate_serialfn_expr(aggserialfn,
4209 : : &serialfnexpr);
3499 rhaas@postgresql.org 4210 : 168 : fmgr_info(aggserialfn, &pertrans->serialfn);
4211 : 168 : fmgr_info_set_expr((Node *) serialfnexpr, &pertrans->serialfn);
4212 : :
2466 andres@anarazel.de 4213 : 168 : pertrans->serialfn_fcinfo =
4214 : 168 : (FunctionCallInfo) palloc(SizeForFunctionCallInfo(1));
4215 : 168 : InitFunctionCallInfoData(*pertrans->serialfn_fcinfo,
4216 : : &pertrans->serialfn,
4217 : : 1,
4218 : : InvalidOid,
4219 : : (Node *) aggstate, NULL);
4220 : : }
4221 : :
3499 rhaas@postgresql.org 4222 [ + + ]: 25575 : if (OidIsValid(aggdeserialfn))
4223 : : {
3414 tgl@sss.pgh.pa.us 4224 : 60 : build_aggregate_deserialfn_expr(aggdeserialfn,
4225 : : &deserialfnexpr);
3499 rhaas@postgresql.org 4226 : 60 : fmgr_info(aggdeserialfn, &pertrans->deserialfn);
4227 : 60 : fmgr_info_set_expr((Node *) deserialfnexpr, &pertrans->deserialfn);
4228 : :
2466 andres@anarazel.de 4229 : 60 : pertrans->deserialfn_fcinfo =
4230 : 60 : (FunctionCallInfo) palloc(SizeForFunctionCallInfo(2));
4231 : 60 : InitFunctionCallInfoData(*pertrans->deserialfn_fcinfo,
4232 : : &pertrans->deserialfn,
4233 : : 2,
4234 : : InvalidOid,
4235 : : (Node *) aggstate, NULL);
4236 : : }
4237 : :
4238 : : /*
4239 : : * If we're doing either DISTINCT or ORDER BY for a plain agg, then we
4240 : : * have a list of SortGroupClause nodes; fish out the data in them and
4241 : : * stick them into arrays. We ignore ORDER BY for an ordered-set agg,
4242 : : * however; the agg's transfn and finalfn are responsible for that.
4243 : : *
4244 : : * When the planner has set the aggpresorted flag, the input to the
4245 : : * aggregate is already correctly sorted. For ORDER BY aggregates we can
4246 : : * simply treat these as normal aggregates. For presorted DISTINCT
4247 : : * aggregates an extra step must be added to remove duplicate consecutive
4248 : : * inputs.
4249 : : *
4250 : : * Note that by construction, if there is a DISTINCT clause then the ORDER
4251 : : * BY clause is a prefix of it (see transformDistinctClause).
4252 : : */
3737 heikki.linnakangas@i 4253 [ + + ]: 25575 : if (AGGKIND_IS_ORDERED_SET(aggref->aggkind))
4254 : : {
4255 : 126 : sortlist = NIL;
4256 : 126 : numSortCols = numDistinctCols = 0;
1182 drowley@postgresql.o 4257 : 126 : pertrans->aggsortrequired = false;
4258 : : }
4259 [ + + + + ]: 25449 : else if (aggref->aggpresorted && aggref->aggdistinct == NIL)
4260 : : {
4261 : 1056 : sortlist = NIL;
4262 : 1056 : numSortCols = numDistinctCols = 0;
4263 : 1056 : pertrans->aggsortrequired = false;
4264 : : }
3737 heikki.linnakangas@i 4265 [ + + ]: 24393 : else if (aggref->aggdistinct)
4266 : : {
4267 : 291 : sortlist = aggref->aggdistinct;
4268 : 291 : numSortCols = numDistinctCols = list_length(sortlist);
4269 [ - + ]: 291 : Assert(numSortCols >= list_length(aggref->aggorder));
1182 drowley@postgresql.o 4270 : 291 : pertrans->aggsortrequired = !aggref->aggpresorted;
4271 : : }
4272 : : else
4273 : : {
3737 heikki.linnakangas@i 4274 : 24102 : sortlist = aggref->aggorder;
4275 : 24102 : numSortCols = list_length(sortlist);
4276 : 24102 : numDistinctCols = 0;
1182 drowley@postgresql.o 4277 : 24102 : pertrans->aggsortrequired = (numSortCols > 0);
4278 : : }
4279 : :
3737 heikki.linnakangas@i 4280 : 25575 : pertrans->numSortCols = numSortCols;
4281 : 25575 : pertrans->numDistinctCols = numDistinctCols;
4282 : :
4283 : : /*
4284 : : * If we have either sorting or filtering to do, create a tupledesc and
4285 : : * slot corresponding to the aggregated inputs (including sort
4286 : : * expressions) of the agg.
4287 : : */
2933 tgl@sss.pgh.pa.us 4288 [ + + + + ]: 25575 : if (numSortCols > 0 || aggref->aggfilter)
4289 : : {
2533 andres@anarazel.de 4290 : 715 : pertrans->sortdesc = ExecTypeFromTL(aggref->args);
2810 4291 : 715 : pertrans->sortslot =
2538 4292 : 715 : ExecInitExtraTupleSlot(estate, pertrans->sortdesc,
4293 : : &TTSOpsMinimalTuple);
4294 : : }
4295 : :
2933 tgl@sss.pgh.pa.us 4296 [ + + ]: 25575 : if (numSortCols > 0)
4297 : : {
4298 : : /*
4299 : : * We don't implement DISTINCT or ORDER BY aggs in the HASHED case
4300 : : * (yet)
4301 : : */
3136 rhodiumtoad@postgres 4302 [ + - - + ]: 360 : Assert(aggstate->aggstrategy != AGG_HASHED && aggstate->aggstrategy != AGG_MIXED);
4303 : :
4304 : : /* ORDER BY aggregates are not supported with partial aggregation */
1576 drowley@postgresql.o 4305 [ - + ]: 360 : Assert(!DO_AGGSPLIT_COMBINE(aggstate->aggsplit));
4306 : :
4307 : : /* If we have only one input, we need its len/byval info. */
3737 heikki.linnakangas@i 4308 [ + + ]: 360 : if (numInputs == 1)
4309 : : {
4310 : 285 : get_typlenbyval(inputTypes[numDirectArgs],
4311 : : &pertrans->inputtypeLen,
4312 : : &pertrans->inputtypeByVal);
4313 : : }
4314 [ + + ]: 75 : else if (numDistinctCols > 0)
4315 : : {
4316 : : /* we will need an extra slot to store prior values */
2810 andres@anarazel.de 4317 : 54 : pertrans->uniqslot =
2538 4318 : 54 : ExecInitExtraTupleSlot(estate, pertrans->sortdesc,
4319 : : &TTSOpsMinimalTuple);
4320 : : }
4321 : :
4322 : : /* Extract the sort information for use later */
3737 heikki.linnakangas@i 4323 : 360 : pertrans->sortColIdx =
4324 : 360 : (AttrNumber *) palloc(numSortCols * sizeof(AttrNumber));
4325 : 360 : pertrans->sortOperators =
4326 : 360 : (Oid *) palloc(numSortCols * sizeof(Oid));
4327 : 360 : pertrans->sortCollations =
4328 : 360 : (Oid *) palloc(numSortCols * sizeof(Oid));
4329 : 360 : pertrans->sortNullsFirst =
4330 : 360 : (bool *) palloc(numSortCols * sizeof(bool));
4331 : :
4332 : 360 : i = 0;
4333 [ + - + + : 819 : foreach(lc, sortlist)
+ + ]
4334 : : {
4335 : 459 : SortGroupClause *sortcl = (SortGroupClause *) lfirst(lc);
4336 : 459 : TargetEntry *tle = get_sortgroupclause_tle(sortcl, aggref->args);
4337 : :
4338 : : /* the parser should have made sure of this */
4339 [ - + ]: 459 : Assert(OidIsValid(sortcl->sortop));
4340 : :
4341 : 459 : pertrans->sortColIdx[i] = tle->resno;
4342 : 459 : pertrans->sortOperators[i] = sortcl->sortop;
4343 : 459 : pertrans->sortCollations[i] = exprCollation((Node *) tle->expr);
4344 : 459 : pertrans->sortNullsFirst[i] = sortcl->nulls_first;
4345 : 459 : i++;
4346 : : }
4347 [ - + ]: 360 : Assert(i == numSortCols);
4348 : : }
4349 : :
4350 [ + + ]: 25575 : if (aggref->aggdistinct)
4351 : : {
4352 : : Oid *ops;
4353 : :
4354 [ - + ]: 291 : Assert(numArguments > 0);
2811 andres@anarazel.de 4355 [ - + ]: 291 : Assert(list_length(aggref->aggdistinct) == numDistinctCols);
4356 : :
4357 : 291 : ops = palloc(numDistinctCols * sizeof(Oid));
4358 : :
3737 heikki.linnakangas@i 4359 : 291 : i = 0;
4360 [ + - + + : 672 : foreach(lc, aggref->aggdistinct)
+ + ]
2811 andres@anarazel.de 4361 : 381 : ops[i++] = ((SortGroupClause *) lfirst(lc))->eqop;
4362 : :
4363 : : /* lookup / build the necessary comparators */
4364 [ + + ]: 291 : if (numDistinctCols == 1)
4365 : 237 : fmgr_info(get_opcode(ops[0]), &pertrans->equalfnOne);
4366 : : else
4367 : 54 : pertrans->equalfnMulti =
4368 : 54 : execTuplesMatchPrepare(pertrans->sortdesc,
4369 : : numDistinctCols,
4370 : 54 : pertrans->sortColIdx,
4371 : : ops,
2411 peter@eisentraut.org 4372 : 54 : pertrans->sortCollations,
4373 : : &aggstate->ss.ps);
2811 andres@anarazel.de 4374 : 291 : pfree(ops);
4375 : : }
4376 : :
3737 heikki.linnakangas@i 4377 : 25575 : pertrans->sortstates = (Tuplesortstate **)
4378 : 25575 : palloc0(sizeof(Tuplesortstate *) * numGroupingSets);
10702 scrappy@hub.org 4379 : 25575 : }
4380 : :
4381 : :
4382 : : static Datum
8600 tgl@sss.pgh.pa.us 4383 : 10576 : GetAggInitVal(Datum textInitVal, Oid transtype)
4384 : : {
4385 : : Oid typinput,
4386 : : typioparam;
4387 : : char *strInitVal;
4388 : : Datum initVal;
4389 : :
7813 4390 : 10576 : getTypeInputInfo(transtype, &typinput, &typioparam);
6425 4391 : 10576 : strInitVal = TextDatumGetCString(textInitVal);
7146 4392 : 10576 : initVal = OidInputFunctionCall(typinput, strInitVal,
4393 : : typioparam, -1);
8600 4394 : 10576 : pfree(strInitVal);
4395 : 10576 : return initVal;
4396 : : }
4397 : :
4398 : : void
8362 4399 : 22790 : ExecEndAgg(AggState *node)
4400 : : {
4401 : : PlanState *outerPlan;
4402 : : int transno;
3817 andres@anarazel.de 4403 : 22790 : int numGroupingSets = Max(node->maxsets, 1);
4404 : : int setno;
4405 : :
4406 : : /*
4407 : : * When ending a parallel worker, copy the statistics gathered by the
4408 : : * worker back into shared memory so that it can be picked up by the main
4409 : : * process to report in EXPLAIN ANALYZE.
4410 : : */
1956 drowley@postgresql.o 4411 [ + + + + ]: 22790 : if (node->shared_info && IsParallelWorker())
4412 : : {
4413 : : AggregateInstrumentation *si;
4414 : :
4415 [ - + ]: 84 : Assert(ParallelWorkerNumber <= node->shared_info->num_workers);
4416 : 84 : si = &node->shared_info->sinstrument[ParallelWorkerNumber];
4417 : 84 : si->hash_batches_used = node->hash_batches_used;
4418 : 84 : si->hash_disk_used = node->hash_disk_used;
4419 : 84 : si->hash_mem_peak = node->hash_mem_peak;
4420 : : }
4421 : :
4422 : : /* Make sure we have closed any open tuplesorts */
4423 : :
3817 andres@anarazel.de 4424 [ + + ]: 22790 : if (node->sort_in)
4425 : 81 : tuplesort_end(node->sort_in);
4426 [ + + ]: 22790 : if (node->sort_out)
4427 : 24 : tuplesort_end(node->sort_out);
4428 : :
2049 jdavis@postgresql.or 4429 : 22790 : hashagg_reset_spill_state(node);
4430 : :
4431 [ + + ]: 22790 : if (node->hash_metacxt != NULL)
4432 : : {
4433 : 3495 : MemoryContextDelete(node->hash_metacxt);
4434 : 3495 : node->hash_metacxt = NULL;
4435 : : }
217 4436 [ + + ]: 22790 : if (node->hash_tablecxt != NULL)
4437 : : {
4438 : 3495 : MemoryContextDelete(node->hash_tablecxt);
4439 : 3495 : node->hash_tablecxt = NULL;
4440 : : }
4441 : :
4442 : :
3737 heikki.linnakangas@i 4443 [ + + ]: 48267 : for (transno = 0; transno < node->numtrans; transno++)
4444 : : {
4445 : 25477 : AggStatePerTrans pertrans = &node->pertrans[transno];
4446 : :
3817 andres@anarazel.de 4447 [ + + ]: 51473 : for (setno = 0; setno < numGroupingSets; setno++)
4448 : : {
3737 heikki.linnakangas@i 4449 [ - + ]: 25996 : if (pertrans->sortstates[setno])
3737 heikki.linnakangas@i 4450 :UBC 0 : tuplesort_end(pertrans->sortstates[setno]);
4451 : : }
4452 : : }
4453 : :
4454 : : /* And ensure any agg shutdown callbacks have been called */
3817 andres@anarazel.de 4455 [ + + ]:CBC 46012 : for (setno = 0; setno < numGroupingSets; setno++)
4456 : 23222 : ReScanExprContext(node->aggcontexts[setno]);
3136 rhodiumtoad@postgres 4457 [ + + ]: 22790 : if (node->hashcontext)
4458 : 3495 : ReScanExprContext(node->hashcontext);
4459 : :
8362 tgl@sss.pgh.pa.us 4460 : 22790 : outerPlan = outerPlanState(node);
4461 : 22790 : ExecEndNode(outerPlan);
10702 scrappy@hub.org 4462 : 22790 : }
4463 : :
4464 : : void
5586 tgl@sss.pgh.pa.us 4465 : 27068 : ExecReScanAgg(AggState *node)
4466 : : {
8362 4467 : 27068 : ExprContext *econtext = node->ss.ps.ps_ExprContext;
3810 bruce@momjian.us 4468 : 27068 : PlanState *outerPlan = outerPlanState(node);
3817 andres@anarazel.de 4469 : 27068 : Agg *aggnode = (Agg *) node->ss.ps.plan;
4470 : : int transno;
3810 bruce@momjian.us 4471 : 27068 : int numGroupingSets = Max(node->maxsets, 1);
4472 : : int setno;
4473 : :
8186 tgl@sss.pgh.pa.us 4474 : 27068 : node->agg_done = false;
4475 : :
3136 rhodiumtoad@postgres 4476 [ + + ]: 27068 : if (node->aggstrategy == AGG_HASHED)
4477 : : {
4478 : : /*
4479 : : * In the hashed case, if we haven't yet built the hash table then we
4480 : : * can just return; nothing done yet, so nothing to undo. If subnode's
4481 : : * chgParam is not NULL then it will be re-scanned by ExecProcNode,
4482 : : * else no reason to re-scan it at all.
4483 : : */
8186 tgl@sss.pgh.pa.us 4484 [ + + ]: 6698 : if (!node->table_filled)
4485 : 75 : return;
4486 : :
4487 : : /*
4488 : : * If we do have the hash table, and it never spilled, and the subplan
4489 : : * does not have any parameter changes, and none of our own parameter
4490 : : * changes affect input expressions of the aggregated functions, then
4491 : : * we can just rescan the existing hash table; no need to build it
4492 : : * again.
4493 : : */
2049 jdavis@postgresql.or 4494 [ + + + - ]: 6623 : if (outerPlan->chgParam == NULL && !node->hash_ever_spilled &&
3351 tgl@sss.pgh.pa.us 4495 [ + + ]: 449 : !bms_overlap(node->ss.ps.chgParam, aggnode->aggParams))
4496 : : {
3136 rhodiumtoad@postgres 4497 : 437 : ResetTupleHashIterator(node->perhash[0].hashtable,
4498 : : &node->perhash[0].hashiter);
4499 : 437 : select_current_set(node, 0, true);
8186 tgl@sss.pgh.pa.us 4500 : 437 : return;
4501 : : }
4502 : : }
4503 : :
4504 : : /* Make sure we have closed any open tuplesorts */
3737 heikki.linnakangas@i 4505 [ + + ]: 60480 : for (transno = 0; transno < node->numtrans; transno++)
4506 : : {
3817 andres@anarazel.de 4507 [ + + ]: 67866 : for (setno = 0; setno < numGroupingSets; setno++)
4508 : : {
3737 heikki.linnakangas@i 4509 : 33942 : AggStatePerTrans pertrans = &node->pertrans[transno];
4510 : :
4511 [ - + ]: 33942 : if (pertrans->sortstates[setno])
4512 : : {
3737 heikki.linnakangas@i 4513 :UBC 0 : tuplesort_end(pertrans->sortstates[setno]);
4514 : 0 : pertrans->sortstates[setno] = NULL;
4515 : : }
4516 : : }
4517 : : }
4518 : :
4519 : : /*
4520 : : * We don't need to ReScanExprContext the output tuple context here;
4521 : : * ExecReScan already did it. But we do need to reset our per-grouping-set
4522 : : * contexts, which may have transvalues stored in them. (We use rescan
4523 : : * rather than just reset because transfns may have registered callbacks
4524 : : * that need to be run now.) For the AGG_HASHED case, see below.
4525 : : */
4526 : :
3817 andres@anarazel.de 4527 [ + + ]:CBC 53130 : for (setno = 0; setno < numGroupingSets; setno++)
4528 : : {
4529 : 26574 : ReScanExprContext(node->aggcontexts[setno]);
4530 : : }
4531 : :
4532 : : /* Release first tuple of group, if we have made a copy */
8362 tgl@sss.pgh.pa.us 4533 [ - + ]: 26556 : if (node->grp_firstTuple != NULL)
4534 : : {
8362 tgl@sss.pgh.pa.us 4535 :UBC 0 : heap_freetuple(node->grp_firstTuple);
4536 : 0 : node->grp_firstTuple = NULL;
4537 : : }
3817 andres@anarazel.de 4538 :CBC 26556 : ExecClearTuple(node->ss.ss_ScanTupleSlot);
4539 : :
4540 : : /* Forget current agg values */
8362 tgl@sss.pgh.pa.us 4541 [ + - + - : 60480 : MemSet(econtext->ecxt_aggvalues, 0, sizeof(Datum) * node->numaggs);
+ - + - +
+ ]
4542 [ + - + + : 26556 : MemSet(econtext->ecxt_aggnulls, 0, sizeof(bool) * node->numaggs);
+ - + - -
+ ]
4543 : :
4544 : : /*
4545 : : * With AGG_HASHED/MIXED, the hash table is allocated in a sub-context of
4546 : : * the hashcontext. This used to be an issue, but now, resetting a context
4547 : : * automatically deletes sub-contexts too.
4548 : : */
3136 rhodiumtoad@postgres 4549 [ + + + + ]: 26556 : if (node->aggstrategy == AGG_HASHED || node->aggstrategy == AGG_MIXED)
4550 : : {
2049 jdavis@postgresql.or 4551 : 6201 : hashagg_reset_spill_state(node);
4552 : :
4553 : 6201 : node->hash_ever_spilled = false;
4554 : 6201 : node->hash_spill_mode = false;
4555 : 6201 : node->hash_ngroups_current = 0;
4556 : :
3136 rhodiumtoad@postgres 4557 : 6201 : ReScanExprContext(node->hashcontext);
217 jdavis@postgresql.or 4558 : 6201 : MemoryContextReset(node->hash_tablecxt);
4559 : : /* Rebuild an empty hash table */
2077 4560 : 6201 : build_hash_tables(node);
8362 tgl@sss.pgh.pa.us 4561 : 6201 : node->table_filled = false;
4562 : : /* iterator will be reset when the table is filled */
4563 : :
2049 jdavis@postgresql.or 4564 : 6201 : hashagg_recompile_expressions(node, false, false);
4565 : : }
4566 : :
3136 rhodiumtoad@postgres 4567 [ + + ]: 26556 : if (node->aggstrategy != AGG_HASHED)
4568 : : {
4569 : : /*
4570 : : * Reset the per-group state (in particular, mark transvalues null)
4571 : : */
2855 andres@anarazel.de 4572 [ + + ]: 40758 : for (setno = 0; setno < numGroupingSets; setno++)
4573 : : {
4574 [ + - + - : 88224 : MemSet(node->pergroups[setno], 0,
+ - + - +
+ ]
4575 : : sizeof(AggStatePerGroupData) * node->numaggs);
4576 : : }
4577 : :
4578 : : /* reset to phase 1 */
3136 rhodiumtoad@postgres 4579 : 20370 : initialize_phase(node, 1);
4580 : :
3817 andres@anarazel.de 4581 : 20370 : node->input_done = false;
4582 : 20370 : node->projected_set = -1;
4583 : : }
4584 : :
3829 rhaas@postgresql.org 4585 [ + + ]: 26556 : if (outerPlan->chgParam == NULL)
4586 : 109 : ExecReScan(outerPlan);
4587 : : }
4588 : :
4589 : :
4590 : : /***********************************************************************
4591 : : * API exposed to aggregate functions
4592 : : ***********************************************************************/
4593 : :
4594 : :
4595 : : /*
4596 : : * AggCheckCallContext - test if a SQL function is being called as an aggregate
4597 : : *
4598 : : * The transition and/or final functions of an aggregate may want to verify
4599 : : * that they are being called as aggregates, rather than as plain SQL
4600 : : * functions. They should use this function to do so. The return value
4601 : : * is nonzero if being called as an aggregate, or zero if not. (Specific
4602 : : * nonzero values are AGG_CONTEXT_AGGREGATE or AGG_CONTEXT_WINDOW, but more
4603 : : * values could conceivably appear in future.)
4604 : : *
4605 : : * If aggcontext isn't NULL, the function also stores at *aggcontext the
4606 : : * identity of the memory context that aggregate transition values are being
4607 : : * stored in. Note that the same aggregate call site (flinfo) may be called
4608 : : * interleaved on different transition values in different contexts, so it's
4609 : : * not kosher to cache aggcontext under fn_extra. It is, however, kosher to
4610 : : * cache it in the transvalue itself (for internal-type transvalues).
4611 : : */
4612 : : int
5740 tgl@sss.pgh.pa.us 4613 : 2705311 : AggCheckCallContext(FunctionCallInfo fcinfo, MemoryContext *aggcontext)
4614 : : {
4615 [ + + + + ]: 2705311 : if (fcinfo->context && IsA(fcinfo->context, AggState))
4616 : : {
4617 [ + + ]: 2699544 : if (aggcontext)
4618 : : {
3810 bruce@momjian.us 4619 : 1284233 : AggState *aggstate = ((AggState *) fcinfo->context);
3136 rhodiumtoad@postgres 4620 : 1284233 : ExprContext *cxt = aggstate->curaggcontext;
4621 : :
3817 andres@anarazel.de 4622 : 1284233 : *aggcontext = cxt->ecxt_per_tuple_memory;
4623 : : }
5740 tgl@sss.pgh.pa.us 4624 : 2699544 : return AGG_CONTEXT_AGGREGATE;
4625 : : }
4626 [ + + + - ]: 5767 : if (fcinfo->context && IsA(fcinfo->context, WindowAggState))
4627 : : {
4628 [ + + ]: 4830 : if (aggcontext)
4216 4629 : 355 : *aggcontext = ((WindowAggState *) fcinfo->context)->curaggcontext;
5740 4630 : 4830 : return AGG_CONTEXT_WINDOW;
4631 : : }
4632 : :
4633 : : /* this is just to prevent "uninitialized variable" warnings */
4634 [ + + ]: 937 : if (aggcontext)
4635 : 913 : *aggcontext = NULL;
4636 : 937 : return 0;
4637 : : }
4638 : :
4639 : : /*
4640 : : * AggGetAggref - allow an aggregate support function to get its Aggref
4641 : : *
4642 : : * If the function is being called as an aggregate support function,
4643 : : * return the Aggref node for the aggregate call. Otherwise, return NULL.
4644 : : *
4645 : : * Aggregates sharing the same inputs and transition functions can get
4646 : : * merged into a single transition calculation. If the transition function
4647 : : * calls AggGetAggref, it will get some one of the Aggrefs for which it is
4648 : : * executing. It must therefore not pay attention to the Aggref fields that
4649 : : * relate to the final function, as those are indeterminate. But if a final
4650 : : * function calls AggGetAggref, it will get a precise result.
4651 : : *
4652 : : * Note that if an aggregate is being used as a window function, this will
4653 : : * return NULL. We could provide a similar function to return the relevant
4654 : : * WindowFunc node in such cases, but it's not needed yet.
4655 : : */
4656 : : Aggref *
4326 4657 : 123 : AggGetAggref(FunctionCallInfo fcinfo)
4658 : : {
4659 [ + - + - ]: 123 : if (fcinfo->context && IsA(fcinfo->context, AggState))
4660 : : {
2933 4661 : 123 : AggState *aggstate = (AggState *) fcinfo->context;
4662 : : AggStatePerAgg curperagg;
4663 : : AggStatePerTrans curpertrans;
4664 : :
4665 : : /* check curperagg (valid when in a final function) */
4666 : 123 : curperagg = aggstate->curperagg;
4667 : :
2937 4668 [ - + ]: 123 : if (curperagg)
2937 tgl@sss.pgh.pa.us 4669 :UBC 0 : return curperagg->aggref;
4670 : :
4671 : : /* check curpertrans (valid when in a transition function) */
2933 tgl@sss.pgh.pa.us 4672 :CBC 123 : curpertrans = aggstate->curpertrans;
4673 : :
3737 heikki.linnakangas@i 4674 [ + - ]: 123 : if (curpertrans)
4675 : 123 : return curpertrans->aggref;
4676 : : }
4326 tgl@sss.pgh.pa.us 4677 :UBC 0 : return NULL;
4678 : : }
4679 : :
4680 : : /*
4681 : : * AggGetTempMemoryContext - fetch short-term memory context for aggregates
4682 : : *
4683 : : * This is useful in agg final functions; the context returned is one that
4684 : : * the final function can safely reset as desired. This isn't useful for
4685 : : * transition functions, since the context returned MAY (we don't promise)
4686 : : * be the same as the context those are called in.
4687 : : *
4688 : : * As above, this is currently not useful for aggs called as window functions.
4689 : : */
4690 : : MemoryContext
4134 4691 : 0 : AggGetTempMemoryContext(FunctionCallInfo fcinfo)
4692 : : {
4326 4693 [ # # # # ]: 0 : if (fcinfo->context && IsA(fcinfo->context, AggState))
4694 : : {
4695 : 0 : AggState *aggstate = (AggState *) fcinfo->context;
4696 : :
4134 4697 : 0 : return aggstate->tmpcontext->ecxt_per_tuple_memory;
4698 : : }
4326 4699 : 0 : return NULL;
4700 : : }
4701 : :
4702 : : /*
4703 : : * AggStateIsShared - find out whether transition state is shared
4704 : : *
4705 : : * If the function is being called as an aggregate support function,
4706 : : * return true if the aggregate's transition state is shared across
4707 : : * multiple aggregates, false if it is not.
4708 : : *
4709 : : * Returns true if not called as an aggregate support function.
4710 : : * This is intended as a conservative answer, ie "no you'd better not
4711 : : * scribble on your input". In particular, will return true if the
4712 : : * aggregate is being used as a window function, which is a scenario
4713 : : * in which changing the transition state is a bad idea. We might
4714 : : * want to refine the behavior for the window case in future.
4715 : : */
4716 : : bool
2933 tgl@sss.pgh.pa.us 4717 :CBC 123 : AggStateIsShared(FunctionCallInfo fcinfo)
4718 : : {
4719 [ + - + - ]: 123 : if (fcinfo->context && IsA(fcinfo->context, AggState))
4720 : : {
4721 : 123 : AggState *aggstate = (AggState *) fcinfo->context;
4722 : : AggStatePerAgg curperagg;
4723 : : AggStatePerTrans curpertrans;
4724 : :
4725 : : /* check curperagg (valid when in a final function) */
4726 : 123 : curperagg = aggstate->curperagg;
4727 : :
4728 [ - + ]: 123 : if (curperagg)
2933 tgl@sss.pgh.pa.us 4729 :UBC 0 : return aggstate->pertrans[curperagg->transno].aggshared;
4730 : :
4731 : : /* check curpertrans (valid when in a transition function) */
2933 tgl@sss.pgh.pa.us 4732 :CBC 123 : curpertrans = aggstate->curpertrans;
4733 : :
4734 [ + - ]: 123 : if (curpertrans)
4735 : 123 : return curpertrans->aggshared;
4736 : : }
2933 tgl@sss.pgh.pa.us 4737 :UBC 0 : return true;
4738 : : }
4739 : :
4740 : : /*
4741 : : * AggRegisterCallback - register a cleanup callback for an aggregate
4742 : : *
4743 : : * This is useful for aggs to register shutdown callbacks, which will ensure
4744 : : * that non-memory resources are freed. The callback will occur just before
4745 : : * the associated aggcontext (as returned by AggCheckCallContext) is reset,
4746 : : * either between groups or as a result of rescanning the query. The callback
4747 : : * will NOT be called on error paths. The typical use-case is for freeing of
4748 : : * tuplestores or tuplesorts maintained in aggcontext, or pins held by slots
4749 : : * created by the agg functions. (The callback will not be called until after
4750 : : * the result of the finalfn is no longer needed, so it's safe for the finalfn
4751 : : * to return data that will be freed by the callback.)
4752 : : *
4753 : : * As above, this is currently not useful for aggs called as window functions.
4754 : : */
4755 : : void
4134 tgl@sss.pgh.pa.us 4756 :CBC 330 : AggRegisterCallback(FunctionCallInfo fcinfo,
4757 : : ExprContextCallbackFunction func,
4758 : : Datum arg)
4759 : : {
4326 4760 [ + - + - ]: 330 : if (fcinfo->context && IsA(fcinfo->context, AggState))
4761 : : {
4762 : 330 : AggState *aggstate = (AggState *) fcinfo->context;
3136 rhodiumtoad@postgres 4763 : 330 : ExprContext *cxt = aggstate->curaggcontext;
4764 : :
3817 andres@anarazel.de 4765 : 330 : RegisterExprContextCallback(cxt, func, arg);
4766 : :
4134 tgl@sss.pgh.pa.us 4767 : 330 : return;
4768 : : }
4134 tgl@sss.pgh.pa.us 4769 [ # # ]:UBC 0 : elog(ERROR, "aggregate function cannot register a callback in this context");
4770 : : }
4771 : :
4772 : :
4773 : : /* ----------------------------------------------------------------
4774 : : * Parallel Query Support
4775 : : * ----------------------------------------------------------------
4776 : : */
4777 : :
4778 : : /* ----------------------------------------------------------------
4779 : : * ExecAggEstimate
4780 : : *
4781 : : * Estimate space required to propagate aggregate statistics.
4782 : : * ----------------------------------------------------------------
4783 : : */
4784 : : void
1956 drowley@postgresql.o 4785 :CBC 281 : ExecAggEstimate(AggState *node, ParallelContext *pcxt)
4786 : : {
4787 : : Size size;
4788 : :
4789 : : /* don't need this if not instrumenting or no workers */
4790 [ + + - + ]: 281 : if (!node->ss.ps.instrument || pcxt->nworkers == 0)
4791 : 230 : return;
4792 : :
4793 : 51 : size = mul_size(pcxt->nworkers, sizeof(AggregateInstrumentation));
4794 : 51 : size = add_size(size, offsetof(SharedAggInfo, sinstrument));
4795 : 51 : shm_toc_estimate_chunk(&pcxt->estimator, size);
4796 : 51 : shm_toc_estimate_keys(&pcxt->estimator, 1);
4797 : : }
4798 : :
4799 : : /* ----------------------------------------------------------------
4800 : : * ExecAggInitializeDSM
4801 : : *
4802 : : * Initialize DSM space for aggregate statistics.
4803 : : * ----------------------------------------------------------------
4804 : : */
4805 : : void
4806 : 281 : ExecAggInitializeDSM(AggState *node, ParallelContext *pcxt)
4807 : : {
4808 : : Size size;
4809 : :
4810 : : /* don't need this if not instrumenting or no workers */
4811 [ + + - + ]: 281 : if (!node->ss.ps.instrument || pcxt->nworkers == 0)
4812 : 230 : return;
4813 : :
4814 : 51 : size = offsetof(SharedAggInfo, sinstrument)
4815 : 51 : + pcxt->nworkers * sizeof(AggregateInstrumentation);
4816 : 51 : node->shared_info = shm_toc_allocate(pcxt->toc, size);
4817 : : /* ensure any unfilled slots will contain zeroes */
4818 : 51 : memset(node->shared_info, 0, size);
4819 : 51 : node->shared_info->num_workers = pcxt->nworkers;
4820 : 51 : shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id,
4821 : 51 : node->shared_info);
4822 : : }
4823 : :
4824 : : /* ----------------------------------------------------------------
4825 : : * ExecAggInitializeWorker
4826 : : *
4827 : : * Attach worker to DSM space for aggregate statistics.
4828 : : * ----------------------------------------------------------------
4829 : : */
4830 : : void
4831 : 782 : ExecAggInitializeWorker(AggState *node, ParallelWorkerContext *pwcxt)
4832 : : {
4833 : 782 : node->shared_info =
4834 : 782 : shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, true);
4835 : 782 : }
4836 : :
4837 : : /* ----------------------------------------------------------------
4838 : : * ExecAggRetrieveInstrumentation
4839 : : *
4840 : : * Transfer aggregate statistics from DSM to private memory.
4841 : : * ----------------------------------------------------------------
4842 : : */
4843 : : void
4844 : 51 : ExecAggRetrieveInstrumentation(AggState *node)
4845 : : {
4846 : : Size size;
4847 : : SharedAggInfo *si;
4848 : :
4849 [ - + ]: 51 : if (node->shared_info == NULL)
1956 drowley@postgresql.o 4850 :UBC 0 : return;
4851 : :
1956 drowley@postgresql.o 4852 :CBC 51 : size = offsetof(SharedAggInfo, sinstrument)
4853 : 51 : + node->shared_info->num_workers * sizeof(AggregateInstrumentation);
4854 : 51 : si = palloc(size);
4855 : 51 : memcpy(si, node->shared_info, size);
4856 : 51 : node->shared_info = si;
4857 : : }
|
