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