TLA Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * nodeWindowAgg.c
4 : * routines to handle WindowAgg nodes.
5 : *
6 : * A WindowAgg node evaluates "window functions" across suitable partitions
7 : * of the input tuple set. Any one WindowAgg works for just a single window
8 : * specification, though it can evaluate multiple window functions sharing
9 : * identical window specifications. The input tuples are required to be
10 : * delivered in sorted order, with the PARTITION BY columns (if any) as
11 : * major sort keys and the ORDER BY columns (if any) as minor sort keys.
12 : * (The planner generates a stack of WindowAggs with intervening Sort nodes
13 : * as needed, if a query involves more than one window specification.)
14 : *
15 : * Since window functions can require access to any or all of the rows in
16 : * the current partition, we accumulate rows of the partition into a
17 : * tuplestore. The window functions are called using the WindowObject API
18 : * so that they can access those rows as needed.
19 : *
20 : * We also support using plain aggregate functions as window functions.
21 : * For these, the regular Agg-node environment is emulated for each partition.
22 : * As required by the SQL spec, the output represents the value of the
23 : * aggregate function over all rows in the current row's window frame.
24 : *
25 : *
26 : * Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
27 : * Portions Copyright (c) 1994, Regents of the University of California
28 : *
29 : * IDENTIFICATION
30 : * src/backend/executor/nodeWindowAgg.c
31 : *
32 : *-------------------------------------------------------------------------
33 : */
34 : #include "postgres.h"
35 :
36 : #include "access/htup_details.h"
37 : #include "catalog/objectaccess.h"
38 : #include "catalog/pg_aggregate.h"
39 : #include "catalog/pg_proc.h"
40 : #include "executor/executor.h"
41 : #include "executor/nodeWindowAgg.h"
42 : #include "miscadmin.h"
43 : #include "nodes/nodeFuncs.h"
44 : #include "optimizer/clauses.h"
45 : #include "optimizer/optimizer.h"
46 : #include "parser/parse_agg.h"
47 : #include "parser/parse_coerce.h"
48 : #include "utils/acl.h"
49 : #include "utils/builtins.h"
50 : #include "utils/datum.h"
51 : #include "utils/expandeddatum.h"
52 : #include "utils/lsyscache.h"
53 : #include "utils/memutils.h"
54 : #include "utils/regproc.h"
55 : #include "utils/syscache.h"
56 : #include "windowapi.h"
57 :
58 : /*
59 : * All the window function APIs are called with this object, which is passed
60 : * to window functions as fcinfo->context.
61 : */
62 : typedef struct WindowObjectData
63 : {
64 : NodeTag type;
65 : WindowAggState *winstate; /* parent WindowAggState */
66 : List *argstates; /* ExprState trees for fn's arguments */
67 : void *localmem; /* WinGetPartitionLocalMemory's chunk */
68 : int markptr; /* tuplestore mark pointer for this fn */
69 : int readptr; /* tuplestore read pointer for this fn */
70 : int64 markpos; /* row that markptr is positioned on */
71 : int64 seekpos; /* row that readptr is positioned on */
72 : } WindowObjectData;
73 :
74 : /*
75 : * We have one WindowStatePerFunc struct for each window function and
76 : * window aggregate handled by this node.
77 : */
78 : typedef struct WindowStatePerFuncData
79 : {
80 : /* Links to WindowFunc expr and state nodes this working state is for */
81 : WindowFuncExprState *wfuncstate;
82 : WindowFunc *wfunc;
83 :
84 : int numArguments; /* number of arguments */
85 :
86 : FmgrInfo flinfo; /* fmgr lookup data for window function */
87 :
88 : Oid winCollation; /* collation derived for window function */
89 :
90 : /*
91 : * We need the len and byval info for the result of each function in order
92 : * to know how to copy/delete values.
93 : */
94 : int16 resulttypeLen;
95 : bool resulttypeByVal;
96 :
97 : bool plain_agg; /* is it just a plain aggregate function? */
98 : int aggno; /* if so, index of its WindowStatePerAggData */
99 :
100 : WindowObject winobj; /* object used in window function API */
101 : } WindowStatePerFuncData;
102 :
103 : /*
104 : * For plain aggregate window functions, we also have one of these.
105 : */
106 : typedef struct WindowStatePerAggData
107 : {
108 : /* Oids of transition functions */
109 : Oid transfn_oid;
110 : Oid invtransfn_oid; /* may be InvalidOid */
111 : Oid finalfn_oid; /* may be InvalidOid */
112 :
113 : /*
114 : * fmgr lookup data for transition functions --- only valid when
115 : * corresponding oid is not InvalidOid. Note in particular that fn_strict
116 : * flags are kept here.
117 : */
118 : FmgrInfo transfn;
119 : FmgrInfo invtransfn;
120 : FmgrInfo finalfn;
121 :
122 : int numFinalArgs; /* number of arguments to pass to finalfn */
123 :
124 : /*
125 : * initial value from pg_aggregate entry
126 : */
127 : Datum initValue;
128 : bool initValueIsNull;
129 :
130 : /*
131 : * cached value for current frame boundaries
132 : */
133 : Datum resultValue;
134 : bool resultValueIsNull;
135 :
136 : /*
137 : * We need the len and byval info for the agg's input, result, and
138 : * transition data types in order to know how to copy/delete values.
139 : */
140 : int16 inputtypeLen,
141 : resulttypeLen,
142 : transtypeLen;
143 : bool inputtypeByVal,
144 : resulttypeByVal,
145 : transtypeByVal;
146 :
147 : int wfuncno; /* index of associated WindowStatePerFuncData */
148 :
149 : /* Context holding transition value and possibly other subsidiary data */
150 : MemoryContext aggcontext; /* may be private, or winstate->aggcontext */
151 :
152 : /* Current transition value */
153 : Datum transValue; /* current transition value */
154 : bool transValueIsNull;
155 :
156 : int64 transValueCount; /* number of currently-aggregated rows */
157 :
158 : /* Data local to eval_windowaggregates() */
159 : bool restart; /* need to restart this agg in this cycle? */
160 : } WindowStatePerAggData;
161 :
162 : static void initialize_windowaggregate(WindowAggState *winstate,
163 : WindowStatePerFunc perfuncstate,
164 : WindowStatePerAgg peraggstate);
165 : static void advance_windowaggregate(WindowAggState *winstate,
166 : WindowStatePerFunc perfuncstate,
167 : WindowStatePerAgg peraggstate);
168 : static bool advance_windowaggregate_base(WindowAggState *winstate,
169 : WindowStatePerFunc perfuncstate,
170 : WindowStatePerAgg peraggstate);
171 : static void finalize_windowaggregate(WindowAggState *winstate,
172 : WindowStatePerFunc perfuncstate,
173 : WindowStatePerAgg peraggstate,
174 : Datum *result, bool *isnull);
175 :
176 : static void eval_windowaggregates(WindowAggState *winstate);
177 : static void eval_windowfunction(WindowAggState *winstate,
178 : WindowStatePerFunc perfuncstate,
179 : Datum *result, bool *isnull);
180 :
181 : static void begin_partition(WindowAggState *winstate);
182 : static void spool_tuples(WindowAggState *winstate, int64 pos);
183 : static void release_partition(WindowAggState *winstate);
184 :
185 : static int row_is_in_frame(WindowAggState *winstate, int64 pos,
186 : TupleTableSlot *slot);
187 : static void update_frameheadpos(WindowAggState *winstate);
188 : static void update_frametailpos(WindowAggState *winstate);
189 : static void update_grouptailpos(WindowAggState *winstate);
190 :
191 : static WindowStatePerAggData *initialize_peragg(WindowAggState *winstate,
192 : WindowFunc *wfunc,
193 : WindowStatePerAgg peraggstate);
194 : static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
195 :
196 : static bool are_peers(WindowAggState *winstate, TupleTableSlot *slot1,
197 : TupleTableSlot *slot2);
198 : static bool window_gettupleslot(WindowObject winobj, int64 pos,
199 : TupleTableSlot *slot);
200 :
201 :
202 : /*
203 : * initialize_windowaggregate
204 : * parallel to initialize_aggregates in nodeAgg.c
205 : */
206 : static void
207 CBC 1963 : initialize_windowaggregate(WindowAggState *winstate,
208 : WindowStatePerFunc perfuncstate,
209 : WindowStatePerAgg peraggstate)
210 : {
211 : MemoryContext oldContext;
212 :
213 : /*
214 : * If we're using a private aggcontext, we may reset it here. But if the
215 : * context is shared, we don't know which other aggregates may still need
216 : * it, so we must leave it to the caller to reset at an appropriate time.
217 : */
218 1963 : if (peraggstate->aggcontext != winstate->aggcontext)
219 1840 : MemoryContextResetAndDeleteChildren(peraggstate->aggcontext);
220 :
221 1963 : if (peraggstate->initValueIsNull)
222 473 : peraggstate->transValue = peraggstate->initValue;
223 : else
224 : {
225 1490 : oldContext = MemoryContextSwitchTo(peraggstate->aggcontext);
226 2980 : peraggstate->transValue = datumCopy(peraggstate->initValue,
227 1490 : peraggstate->transtypeByVal,
228 1490 : peraggstate->transtypeLen);
229 1490 : MemoryContextSwitchTo(oldContext);
230 : }
231 1963 : peraggstate->transValueIsNull = peraggstate->initValueIsNull;
232 1963 : peraggstate->transValueCount = 0;
233 1963 : peraggstate->resultValue = (Datum) 0;
234 1963 : peraggstate->resultValueIsNull = true;
235 1963 : }
236 :
237 : /*
238 : * advance_windowaggregate
239 : * parallel to advance_aggregates in nodeAgg.c
240 : */
241 : static void
242 73689 : advance_windowaggregate(WindowAggState *winstate,
243 : WindowStatePerFunc perfuncstate,
244 : WindowStatePerAgg peraggstate)
245 : {
246 73689 : LOCAL_FCINFO(fcinfo, FUNC_MAX_ARGS);
247 73689 : WindowFuncExprState *wfuncstate = perfuncstate->wfuncstate;
248 73689 : int numArguments = perfuncstate->numArguments;
249 : Datum newVal;
250 : ListCell *arg;
251 : int i;
252 : MemoryContext oldContext;
253 73689 : ExprContext *econtext = winstate->tmpcontext;
254 73689 : ExprState *filter = wfuncstate->aggfilter;
255 :
256 73689 : oldContext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
257 :
258 : /* Skip anything FILTERed out */
259 73689 : if (filter)
260 : {
261 : bool isnull;
262 171 : Datum res = ExecEvalExpr(filter, econtext, &isnull);
263 :
264 171 : if (isnull || !DatumGetBool(res))
265 : {
266 81 : MemoryContextSwitchTo(oldContext);
267 81 : return;
268 : }
269 : }
270 :
271 : /* We start from 1, since the 0th arg will be the transition value */
272 73608 : i = 1;
273 116955 : foreach(arg, wfuncstate->args)
274 : {
275 43347 : ExprState *argstate = (ExprState *) lfirst(arg);
276 :
277 43347 : fcinfo->args[i].value = ExecEvalExpr(argstate, econtext,
278 : &fcinfo->args[i].isnull);
279 43347 : i++;
280 : }
281 :
282 73608 : if (peraggstate->transfn.fn_strict)
283 : {
284 : /*
285 : * For a strict transfn, nothing happens when there's a NULL input; we
286 : * just keep the prior transValue. Note transValueCount doesn't
287 : * change either.
288 : */
289 114304 : for (i = 1; i <= numArguments; i++)
290 : {
291 42077 : if (fcinfo->args[i].isnull)
292 : {
293 111 : MemoryContextSwitchTo(oldContext);
294 111 : return;
295 : }
296 : }
297 :
298 : /*
299 : * For strict transition functions with initial value NULL we use the
300 : * first non-NULL input as the initial state. (We already checked
301 : * that the agg's input type is binary-compatible with its transtype,
302 : * so straight copy here is OK.)
303 : *
304 : * We must copy the datum into aggcontext if it is pass-by-ref. We do
305 : * not need to pfree the old transValue, since it's NULL.
306 : */
307 72227 : if (peraggstate->transValueCount == 0 && peraggstate->transValueIsNull)
308 : {
309 227 : MemoryContextSwitchTo(peraggstate->aggcontext);
310 454 : peraggstate->transValue = datumCopy(fcinfo->args[1].value,
311 227 : peraggstate->transtypeByVal,
312 227 : peraggstate->transtypeLen);
313 227 : peraggstate->transValueIsNull = false;
314 227 : peraggstate->transValueCount = 1;
315 227 : MemoryContextSwitchTo(oldContext);
316 227 : return;
317 : }
318 :
319 72000 : if (peraggstate->transValueIsNull)
320 : {
321 : /*
322 : * Don't call a strict function with NULL inputs. Note it is
323 : * possible to get here despite the above tests, if the transfn is
324 : * strict *and* returned a NULL on a prior cycle. If that happens
325 : * we will propagate the NULL all the way to the end. That can
326 : * only happen if there's no inverse transition function, though,
327 : * since we disallow transitions back to NULL when there is one.
328 : */
329 UBC 0 : MemoryContextSwitchTo(oldContext);
330 0 : Assert(!OidIsValid(peraggstate->invtransfn_oid));
331 0 : return;
332 : }
333 : }
334 :
335 : /*
336 : * OK to call the transition function. Set winstate->curaggcontext while
337 : * calling it, for possible use by AggCheckCallContext.
338 : */
339 CBC 73270 : InitFunctionCallInfoData(*fcinfo, &(peraggstate->transfn),
340 : numArguments + 1,
341 : perfuncstate->winCollation,
342 : (void *) winstate, NULL);
343 73270 : fcinfo->args[0].value = peraggstate->transValue;
344 73270 : fcinfo->args[0].isnull = peraggstate->transValueIsNull;
345 73270 : winstate->curaggcontext = peraggstate->aggcontext;
346 73270 : newVal = FunctionCallInvoke(fcinfo);
347 73270 : winstate->curaggcontext = NULL;
348 :
349 : /*
350 : * Moving-aggregate transition functions must not return null, see
351 : * advance_windowaggregate_base().
352 : */
353 73270 : if (fcinfo->isnull && OidIsValid(peraggstate->invtransfn_oid))
354 UBC 0 : ereport(ERROR,
355 : (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
356 : errmsg("moving-aggregate transition function must not return null")));
357 :
358 : /*
359 : * We must track the number of rows included in transValue, since to
360 : * remove the last input, advance_windowaggregate_base() mustn't call the
361 : * inverse transition function, but simply reset transValue back to its
362 : * initial value.
363 : */
364 CBC 73270 : peraggstate->transValueCount++;
365 :
366 : /*
367 : * If pass-by-ref datatype, must copy the new value into aggcontext and
368 : * free the prior transValue. But if transfn returned a pointer to its
369 : * first input, we don't need to do anything. Also, if transfn returned a
370 : * pointer to a R/W expanded object that is already a child of the
371 : * aggcontext, assume we can adopt that value without copying it.
372 : */
373 108829 : if (!peraggstate->transtypeByVal &&
374 35559 : DatumGetPointer(newVal) != DatumGetPointer(peraggstate->transValue))
375 : {
376 489 : if (!fcinfo->isnull)
377 : {
378 489 : MemoryContextSwitchTo(peraggstate->aggcontext);
379 492 : if (DatumIsReadWriteExpandedObject(newVal,
380 : false,
381 486 : peraggstate->transtypeLen) &&
382 3 : MemoryContextGetParent(DatumGetEOHP(newVal)->eoh_context) == CurrentMemoryContext)
383 : /* do nothing */ ;
384 : else
385 486 : newVal = datumCopy(newVal,
386 486 : peraggstate->transtypeByVal,
387 486 : peraggstate->transtypeLen);
388 : }
389 489 : if (!peraggstate->transValueIsNull)
390 : {
391 459 : if (DatumIsReadWriteExpandedObject(peraggstate->transValue,
392 : false,
393 : peraggstate->transtypeLen))
394 UBC 0 : DeleteExpandedObject(peraggstate->transValue);
395 : else
396 CBC 459 : pfree(DatumGetPointer(peraggstate->transValue));
397 : }
398 : }
399 :
400 73270 : MemoryContextSwitchTo(oldContext);
401 73270 : peraggstate->transValue = newVal;
402 73270 : peraggstate->transValueIsNull = fcinfo->isnull;
403 : }
404 :
405 : /*
406 : * advance_windowaggregate_base
407 : * Remove the oldest tuple from an aggregation.
408 : *
409 : * This is very much like advance_windowaggregate, except that we will call
410 : * the inverse transition function (which caller must have checked is
411 : * available).
412 : *
413 : * Returns true if we successfully removed the current row from this
414 : * aggregate, false if not (in the latter case, caller is responsible
415 : * for cleaning up by restarting the aggregation).
416 : */
417 : static bool
418 2199 : advance_windowaggregate_base(WindowAggState *winstate,
419 : WindowStatePerFunc perfuncstate,
420 : WindowStatePerAgg peraggstate)
421 : {
422 2199 : LOCAL_FCINFO(fcinfo, FUNC_MAX_ARGS);
423 2199 : WindowFuncExprState *wfuncstate = perfuncstate->wfuncstate;
424 2199 : int numArguments = perfuncstate->numArguments;
425 : Datum newVal;
426 : ListCell *arg;
427 : int i;
428 : MemoryContext oldContext;
429 2199 : ExprContext *econtext = winstate->tmpcontext;
430 2199 : ExprState *filter = wfuncstate->aggfilter;
431 :
432 2199 : oldContext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
433 :
434 : /* Skip anything FILTERed out */
435 2199 : if (filter)
436 : {
437 : bool isnull;
438 51 : Datum res = ExecEvalExpr(filter, econtext, &isnull);
439 :
440 51 : if (isnull || !DatumGetBool(res))
441 : {
442 24 : MemoryContextSwitchTo(oldContext);
443 24 : return true;
444 : }
445 : }
446 :
447 : /* We start from 1, since the 0th arg will be the transition value */
448 2175 : i = 1;
449 4341 : foreach(arg, wfuncstate->args)
450 : {
451 2166 : ExprState *argstate = (ExprState *) lfirst(arg);
452 :
453 2166 : fcinfo->args[i].value = ExecEvalExpr(argstate, econtext,
454 : &fcinfo->args[i].isnull);
455 2166 : i++;
456 : }
457 :
458 2175 : if (peraggstate->invtransfn.fn_strict)
459 : {
460 : /*
461 : * For a strict (inv)transfn, nothing happens when there's a NULL
462 : * input; we just keep the prior transValue. Note transValueCount
463 : * doesn't change either.
464 : */
465 2832 : for (i = 1; i <= numArguments; i++)
466 : {
467 1434 : if (fcinfo->args[i].isnull)
468 : {
469 45 : MemoryContextSwitchTo(oldContext);
470 45 : return true;
471 : }
472 : }
473 : }
474 :
475 : /* There should still be an added but not yet removed value */
476 2130 : Assert(peraggstate->transValueCount > 0);
477 :
478 : /*
479 : * In moving-aggregate mode, the state must never be NULL, except possibly
480 : * before any rows have been aggregated (which is surely not the case at
481 : * this point). This restriction allows us to interpret a NULL result
482 : * from the inverse function as meaning "sorry, can't do an inverse
483 : * transition in this case". We already checked this in
484 : * advance_windowaggregate, but just for safety, check again.
485 : */
486 2130 : if (peraggstate->transValueIsNull)
487 UBC 0 : elog(ERROR, "aggregate transition value is NULL before inverse transition");
488 :
489 : /*
490 : * We mustn't use the inverse transition function to remove the last
491 : * input. Doing so would yield a non-NULL state, whereas we should be in
492 : * the initial state afterwards which may very well be NULL. So instead,
493 : * we simply re-initialize the aggregate in this case.
494 : */
495 CBC 2130 : if (peraggstate->transValueCount == 1)
496 : {
497 51 : MemoryContextSwitchTo(oldContext);
498 51 : initialize_windowaggregate(winstate,
499 51 : &winstate->perfunc[peraggstate->wfuncno],
500 : peraggstate);
501 51 : return true;
502 : }
503 :
504 : /*
505 : * OK to call the inverse transition function. Set
506 : * winstate->curaggcontext while calling it, for possible use by
507 : * AggCheckCallContext.
508 : */
509 2079 : InitFunctionCallInfoData(*fcinfo, &(peraggstate->invtransfn),
510 : numArguments + 1,
511 : perfuncstate->winCollation,
512 : (void *) winstate, NULL);
513 2079 : fcinfo->args[0].value = peraggstate->transValue;
514 2079 : fcinfo->args[0].isnull = peraggstate->transValueIsNull;
515 2079 : winstate->curaggcontext = peraggstate->aggcontext;
516 2079 : newVal = FunctionCallInvoke(fcinfo);
517 2079 : winstate->curaggcontext = NULL;
518 :
519 : /*
520 : * If the function returns NULL, report failure, forcing a restart.
521 : */
522 2079 : if (fcinfo->isnull)
523 : {
524 122 : MemoryContextSwitchTo(oldContext);
525 122 : return false;
526 : }
527 :
528 : /* Update number of rows included in transValue */
529 1957 : peraggstate->transValueCount--;
530 :
531 : /*
532 : * If pass-by-ref datatype, must copy the new value into aggcontext and
533 : * free the prior transValue. But if invtransfn returned a pointer to its
534 : * first input, we don't need to do anything. Also, if invtransfn
535 : * returned a pointer to a R/W expanded object that is already a child of
536 : * the aggcontext, assume we can adopt that value without copying it.
537 : *
538 : * Note: the checks for null values here will never fire, but it seems
539 : * best to have this stanza look just like advance_windowaggregate.
540 : */
541 3028 : if (!peraggstate->transtypeByVal &&
542 1071 : DatumGetPointer(newVal) != DatumGetPointer(peraggstate->transValue))
543 : {
544 339 : if (!fcinfo->isnull)
545 : {
546 339 : MemoryContextSwitchTo(peraggstate->aggcontext);
547 339 : if (DatumIsReadWriteExpandedObject(newVal,
548 : false,
549 336 : peraggstate->transtypeLen) &&
550 UBC 0 : MemoryContextGetParent(DatumGetEOHP(newVal)->eoh_context) == CurrentMemoryContext)
551 : /* do nothing */ ;
552 : else
553 CBC 339 : newVal = datumCopy(newVal,
554 339 : peraggstate->transtypeByVal,
555 339 : peraggstate->transtypeLen);
556 : }
557 339 : if (!peraggstate->transValueIsNull)
558 : {
559 339 : if (DatumIsReadWriteExpandedObject(peraggstate->transValue,
560 : false,
561 : peraggstate->transtypeLen))
562 UBC 0 : DeleteExpandedObject(peraggstate->transValue);
563 : else
564 CBC 339 : pfree(DatumGetPointer(peraggstate->transValue));
565 : }
566 : }
567 :
568 1957 : MemoryContextSwitchTo(oldContext);
569 1957 : peraggstate->transValue = newVal;
570 1957 : peraggstate->transValueIsNull = fcinfo->isnull;
571 :
572 1957 : return true;
573 : }
574 :
575 : /*
576 : * finalize_windowaggregate
577 : * parallel to finalize_aggregate in nodeAgg.c
578 : */
579 : static void
580 5128 : finalize_windowaggregate(WindowAggState *winstate,
581 : WindowStatePerFunc perfuncstate,
582 : WindowStatePerAgg peraggstate,
583 : Datum *result, bool *isnull)
584 : {
585 : MemoryContext oldContext;
586 :
587 5128 : oldContext = MemoryContextSwitchTo(winstate->ss.ps.ps_ExprContext->ecxt_per_tuple_memory);
588 :
589 : /*
590 : * Apply the agg's finalfn if one is provided, else return transValue.
591 : */
592 5128 : if (OidIsValid(peraggstate->finalfn_oid))
593 : {
594 3626 : LOCAL_FCINFO(fcinfo, FUNC_MAX_ARGS);
595 3626 : int numFinalArgs = peraggstate->numFinalArgs;
596 : bool anynull;
597 : int i;
598 :
599 3626 : InitFunctionCallInfoData(fcinfodata.fcinfo, &(peraggstate->finalfn),
600 : numFinalArgs,
601 : perfuncstate->winCollation,
602 : (void *) winstate, NULL);
603 3626 : fcinfo->args[0].value =
604 3626 : MakeExpandedObjectReadOnly(peraggstate->transValue,
605 : peraggstate->transValueIsNull,
606 : peraggstate->transtypeLen);
607 3626 : fcinfo->args[0].isnull = peraggstate->transValueIsNull;
608 3626 : anynull = peraggstate->transValueIsNull;
609 :
610 : /* Fill any remaining argument positions with nulls */
611 3676 : for (i = 1; i < numFinalArgs; i++)
612 : {
613 50 : fcinfo->args[i].value = (Datum) 0;
614 50 : fcinfo->args[i].isnull = true;
615 50 : anynull = true;
616 : }
617 :
618 3626 : if (fcinfo->flinfo->fn_strict && anynull)
619 : {
620 : /* don't call a strict function with NULL inputs */
621 UBC 0 : *result = (Datum) 0;
622 0 : *isnull = true;
623 : }
624 : else
625 : {
626 CBC 3626 : winstate->curaggcontext = peraggstate->aggcontext;
627 3626 : *result = FunctionCallInvoke(fcinfo);
628 3626 : winstate->curaggcontext = NULL;
629 3626 : *isnull = fcinfo->isnull;
630 : }
631 : }
632 : else
633 : {
634 GNC 1502 : *result =
635 1502 : MakeExpandedObjectReadOnly(peraggstate->transValue,
636 : peraggstate->transValueIsNull,
637 : peraggstate->transtypeLen);
638 GIC 1502 : *isnull = peraggstate->transValueIsNull;
639 : }
640 ECB :
641 GIC 5128 : MemoryContextSwitchTo(oldContext);
642 5128 : }
643 :
644 : /*
645 : * eval_windowaggregates
646 : * evaluate plain aggregates being used as window functions
647 : *
648 : * This differs from nodeAgg.c in two ways. First, if the window's frame
649 ECB : * start position moves, we use the inverse transition function (if it exists)
650 : * to remove rows from the transition value. And second, we expect to be
651 : * able to call aggregate final functions repeatedly after aggregating more
652 : * data onto the same transition value. This is not a behavior required by
653 : * nodeAgg.c.
654 : */
655 : static void
656 GIC 64944 : eval_windowaggregates(WindowAggState *winstate)
657 : {
658 : WindowStatePerAgg peraggstate;
659 : int wfuncno,
660 : numaggs,
661 : numaggs_restart,
662 : i;
663 ECB : int64 aggregatedupto_nonrestarted;
664 : MemoryContext oldContext;
665 EUB : ExprContext *econtext;
666 : WindowObject agg_winobj;
667 : TupleTableSlot *agg_row_slot;
668 ECB : TupleTableSlot *temp_slot;
669 :
670 CBC 64944 : numaggs = winstate->numaggs;
671 64944 : if (numaggs == 0)
672 UIC 0 : return; /* nothing to do */
673 :
674 : /* final output execution is in ps_ExprContext */
675 GIC 64944 : econtext = winstate->ss.ps.ps_ExprContext;
676 64944 : agg_winobj = winstate->agg_winobj;
677 64944 : agg_row_slot = winstate->agg_row_slot;
678 64944 : temp_slot = winstate->temp_slot_1;
679 :
680 : /*
681 : * If the window's frame start clause is UNBOUNDED_PRECEDING and no
682 : * exclusion clause is specified, then the window frame consists of a
683 : * contiguous group of rows extending forward from the start of the
684 : * partition, and rows only enter the frame, never exit it, as the current
685 : * row advances forward. This makes it possible to use an incremental
686 : * strategy for evaluating aggregates: we run the transition function for
687 : * each row added to the frame, and run the final function whenever we
688 : * need the current aggregate value. This is considerably more efficient
689 : * than the naive approach of re-running the entire aggregate calculation
690 : * for each current row. It does assume that the final function doesn't
691 : * damage the running transition value, but we have the same assumption in
692 : * nodeAgg.c too (when it rescans an existing hash table).
693 : *
694 : * If the frame start does sometimes move, we can still optimize as above
695 : * whenever successive rows share the same frame head, but if the frame
696 : * head moves beyond the previous head we try to remove those rows using
697 : * the aggregate's inverse transition function. This function restores
698 : * the aggregate's current state to what it would be if the removed row
699 : * had never been aggregated in the first place. Inverse transition
700 : * functions may optionally return NULL, indicating that the function was
701 : * unable to remove the tuple from aggregation. If this happens, or if
702 : * the aggregate doesn't have an inverse transition function at all, we
703 : * must perform the aggregation all over again for all tuples within the
704 : * new frame boundaries.
705 : *
706 : * If there's any exclusion clause, then we may have to aggregate over a
707 : * non-contiguous set of rows, so we punt and recalculate for every row.
708 : * (For some frame end choices, it might be that the frame is always
709 : * contiguous anyway, but that's an optimization to investigate later.)
710 : *
711 : * In many common cases, multiple rows share the same frame and hence the
712 : * same aggregate value. (In particular, if there's no ORDER BY in a RANGE
713 : * window, then all rows are peers and so they all have window frame equal
714 : * to the whole partition.) We optimize such cases by calculating the
715 : * aggregate value once when we reach the first row of a peer group, and
716 : * then returning the saved value for all subsequent rows.
717 : *
718 : * 'aggregatedupto' keeps track of the first row that has not yet been
719 : * accumulated into the aggregate transition values. Whenever we start a
720 : * new peer group, we accumulate forward to the end of the peer group.
721 : */
722 ECB :
723 : /*
724 EUB : * First, update the frame head position.
725 : *
726 : * The frame head should never move backwards, and the code below wouldn't
727 : * cope if it did, so for safety we complain if it does.
728 : */
729 GIC 64944 : update_frameheadpos(winstate);
730 64941 : if (winstate->frameheadpos < winstate->aggregatedbase)
731 UIC 0 : elog(ERROR, "window frame head moved backward");
732 :
733 : /*
734 : * If the frame didn't change compared to the previous row, we can re-use
735 : * the result values that were previously saved at the bottom of this
736 : * function. Since we don't know the current frame's end yet, this is not
737 ECB : * possible to check for fully. But if the frame end mode is UNBOUNDED
738 : * FOLLOWING or CURRENT ROW, no exclusion clause is specified, and the
739 : * current row lies within the previous row's frame, then the two frames'
740 : * ends must coincide. Note that on the first row aggregatedbase ==
741 : * aggregatedupto, meaning this test must fail, so we don't need to check
742 : * the "there was no previous row" case explicitly here.
743 : */
744 CBC 64941 : if (winstate->aggregatedbase == winstate->frameheadpos &&
745 GIC 63115 : (winstate->frameOptions & (FRAMEOPTION_END_UNBOUNDED_FOLLOWING |
746 CBC 62164 : FRAMEOPTION_END_CURRENT_ROW)) &&
747 62164 : !(winstate->frameOptions & FRAMEOPTION_EXCLUSION) &&
748 62074 : winstate->aggregatedbase <= winstate->currentpos &&
749 62056 : winstate->aggregatedupto > winstate->currentpos)
750 : {
751 121258 : for (i = 0; i < numaggs; i++)
752 : {
753 GIC 60632 : peraggstate = &winstate->peragg[i];
754 60632 : wfuncno = peraggstate->wfuncno;
755 60632 : econtext->ecxt_aggvalues[wfuncno] = peraggstate->resultValue;
756 60632 : econtext->ecxt_aggnulls[wfuncno] = peraggstate->resultValueIsNull;
757 : }
758 60626 : return;
759 : }
760 :
761 : /*----------
762 : * Initialize restart flags.
763 : *
764 : * We restart the aggregation:
765 : * - if we're processing the first row in the partition, or
766 : * - if the frame's head moved and we cannot use an inverse
767 : * transition function, or
768 : * - we have an EXCLUSION clause, or
769 ECB : * - if the new frame doesn't overlap the old one
770 : *
771 : * Note that we don't strictly need to restart in the last case, but if
772 : * we're going to remove all rows from the aggregation anyway, a restart
773 : * surely is faster.
774 : *----------
775 : */
776 CBC 4315 : numaggs_restart = 0;
777 9449 : for (i = 0; i < numaggs; i++)
778 : {
779 5134 : peraggstate = &winstate->peragg[i];
780 5134 : if (winstate->currentpos == 0 ||
781 GIC 4174 : (winstate->aggregatedbase != winstate->frameheadpos &&
782 2489 : !OidIsValid(peraggstate->invtransfn_oid)) ||
783 CBC 4136 : (winstate->frameOptions & FRAMEOPTION_EXCLUSION) ||
784 GIC 3578 : winstate->aggregatedupto <= winstate->frameheadpos)
785 : {
786 1790 : peraggstate->restart = true;
787 1790 : numaggs_restart++;
788 : }
789 : else
790 3344 : peraggstate->restart = false;
791 : }
792 :
793 ECB : /*
794 : * If we have any possibly-moving aggregates, attempt to advance
795 : * aggregatedbase to match the frame's head by removing input rows that
796 : * fell off the top of the frame from the aggregations. This can fail,
797 : * i.e. advance_windowaggregate_base() can return false, in which case
798 : * we'll restart that aggregate below.
799 : */
800 CBC 5836 : while (numaggs_restart < numaggs &&
801 GIC 4130 : winstate->aggregatedbase < winstate->frameheadpos)
802 EUB : {
803 : /*
804 : * Fetch the next tuple of those being removed. This should never fail
805 ECB : * as we should have been here before.
806 : */
807 GIC 1521 : if (!window_gettupleslot(agg_winobj, winstate->aggregatedbase,
808 : temp_slot))
809 UIC 0 : elog(ERROR, "could not re-fetch previously fetched frame row");
810 :
811 ECB : /* Set tuple context for evaluation of aggregate arguments */
812 GIC 1521 : winstate->tmpcontext->ecxt_outertuple = temp_slot;
813 :
814 : /*
815 ECB : * Perform the inverse transition for each aggregate function in the
816 : * window, unless it has already been marked as needing a restart.
817 : */
818 GIC 3726 : for (i = 0; i < numaggs; i++)
819 ECB : {
820 : bool ok;
821 :
822 GIC 2205 : peraggstate = &winstate->peragg[i];
823 CBC 2205 : if (peraggstate->restart)
824 GIC 6 : continue;
825 :
826 CBC 2199 : wfuncno = peraggstate->wfuncno;
827 2199 : ok = advance_windowaggregate_base(winstate,
828 GIC 2199 : &winstate->perfunc[wfuncno],
829 : peraggstate);
830 2199 : if (!ok)
831 : {
832 ECB : /* Inverse transition function has failed, must restart */
833 GIC 122 : peraggstate->restart = true;
834 122 : numaggs_restart++;
835 ECB : }
836 : }
837 :
838 : /* Reset per-input-tuple context after each tuple */
839 GIC 1521 : ResetExprContext(winstate->tmpcontext);
840 :
841 : /* And advance the aggregated-row state */
842 1521 : winstate->aggregatedbase++;
843 1521 : ExecClearTuple(temp_slot);
844 ECB : }
845 :
846 : /*
847 : * If we successfully advanced the base rows of all the aggregates,
848 : * aggregatedbase now equals frameheadpos; but if we failed for any, we
849 : * must forcibly update aggregatedbase.
850 : */
851 CBC 4315 : winstate->aggregatedbase = winstate->frameheadpos;
852 :
853 : /*
854 : * If we created a mark pointer for aggregates, keep it pushed up to frame
855 : * head, so that tuplestore can discard unnecessary rows.
856 : */
857 GIC 4315 : if (agg_winobj->markptr >= 0)
858 3034 : WinSetMarkPosition(agg_winobj, winstate->frameheadpos);
859 :
860 : /*
861 : * Now restart the aggregates that require it.
862 : *
863 ECB : * We assume that aggregates using the shared context always restart if
864 : * *any* aggregate restarts, and we may thus clean up the shared
865 : * aggcontext if that is the case. Private aggcontexts are reset by
866 : * initialize_windowaggregate() if their owning aggregate restarts. If we
867 : * aren't restarting an aggregate, we need to free any previously saved
868 : * result for it, else we'll leak memory.
869 : */
870 CBC 4315 : if (numaggs_restart > 0)
871 GIC 1820 : MemoryContextResetAndDeleteChildren(winstate->aggcontext);
872 9449 : for (i = 0; i < numaggs; i++)
873 : {
874 CBC 5134 : peraggstate = &winstate->peragg[i];
875 :
876 ECB : /* Aggregates using the shared ctx must restart if *any* agg does */
877 CBC 5134 : Assert(peraggstate->aggcontext != winstate->aggcontext ||
878 ECB : numaggs_restart == 0 ||
879 : peraggstate->restart);
880 :
881 CBC 5134 : if (peraggstate->restart)
882 : {
883 1912 : wfuncno = peraggstate->wfuncno;
884 1912 : initialize_windowaggregate(winstate,
885 1912 : &winstate->perfunc[wfuncno],
886 ECB : peraggstate);
887 : }
888 GIC 3222 : else if (!peraggstate->resultValueIsNull)
889 : {
890 3105 : if (!peraggstate->resulttypeByVal)
891 950 : pfree(DatumGetPointer(peraggstate->resultValue));
892 3105 : peraggstate->resultValue = (Datum) 0;
893 3105 : peraggstate->resultValueIsNull = true;
894 : }
895 : }
896 :
897 : /*
898 : * Non-restarted aggregates now contain the rows between aggregatedbase
899 : * (i.e., frameheadpos) and aggregatedupto, while restarted aggregates
900 ECB : * contain no rows. If there are any restarted aggregates, we must thus
901 : * begin aggregating anew at frameheadpos, otherwise we may simply
902 : * continue at aggregatedupto. We must remember the old value of
903 : * aggregatedupto to know how long to skip advancing non-restarted
904 : * aggregates. If we modify aggregatedupto, we must also clear
905 : * agg_row_slot, per the loop invariant below.
906 : */
907 GIC 4315 : aggregatedupto_nonrestarted = winstate->aggregatedupto;
908 4315 : if (numaggs_restart > 0 &&
909 1820 : winstate->aggregatedupto != winstate->frameheadpos)
910 : {
911 707 : winstate->aggregatedupto = winstate->frameheadpos;
912 707 : ExecClearTuple(agg_row_slot);
913 : }
914 :
915 : /*
916 ECB : * Advance until we reach a row not in frame (or end of partition).
917 : *
918 : * Note the loop invariant: agg_row_slot is either empty or holds the row
919 : * at position aggregatedupto. We advance aggregatedupto after processing
920 : * a row.
921 : */
922 : for (;;)
923 GIC 73300 : {
924 ECB : int ret;
925 :
926 : /* Fetch next row if we didn't already */
927 GIC 77615 : if (TupIsNull(agg_row_slot))
928 : {
929 75753 : if (!window_gettupleslot(agg_winobj, winstate->aggregatedupto,
930 : agg_row_slot))
931 CBC 2027 : break; /* must be end of partition */
932 ECB : }
933 :
934 : /*
935 : * Exit loop if no more rows can be in frame. Skip aggregation if
936 : * current row is not in frame but there might be more in the frame.
937 : */
938 CBC 75588 : ret = row_is_in_frame(winstate, winstate->aggregatedupto, agg_row_slot);
939 GIC 75582 : if (ret < 0)
940 2282 : break;
941 CBC 73300 : if (ret == 0)
942 GIC 948 : goto next_tuple;
943 ECB :
944 : /* Set tuple context for evaluation of aggregate arguments */
945 GIC 72352 : winstate->tmpcontext->ecxt_outertuple = agg_row_slot;
946 ECB :
947 : /* Accumulate row into the aggregates */
948 CBC 156411 : for (i = 0; i < numaggs; i++)
949 : {
950 84059 : peraggstate = &winstate->peragg[i];
951 ECB :
952 : /* Non-restarted aggs skip until aggregatedupto_nonrestarted */
953 GIC 84059 : if (!peraggstate->restart &&
954 60076 : winstate->aggregatedupto < aggregatedupto_nonrestarted)
955 10370 : continue;
956 ECB :
957 GIC 73689 : wfuncno = peraggstate->wfuncno;
958 CBC 73689 : advance_windowaggregate(winstate,
959 GIC 73689 : &winstate->perfunc[wfuncno],
960 : peraggstate);
961 ECB : }
962 :
963 GIC 72352 : next_tuple:
964 : /* Reset per-input-tuple context after each tuple */
965 73300 : ResetExprContext(winstate->tmpcontext);
966 ECB :
967 : /* And advance the aggregated-row state */
968 GIC 73300 : winstate->aggregatedupto++;
969 73300 : ExecClearTuple(agg_row_slot);
970 : }
971 ECB :
972 : /* The frame's end is not supposed to move backwards, ever */
973 GIC 4309 : Assert(aggregatedupto_nonrestarted <= winstate->aggregatedupto);
974 :
975 : /*
976 ECB : * finalize aggregates and fill result/isnull fields.
977 : */
978 CBC 9437 : for (i = 0; i < numaggs; i++)
979 ECB : {
980 : Datum *result;
981 : bool *isnull;
982 :
983 GIC 5128 : peraggstate = &winstate->peragg[i];
984 5128 : wfuncno = peraggstate->wfuncno;
985 5128 : result = &econtext->ecxt_aggvalues[wfuncno];
986 5128 : isnull = &econtext->ecxt_aggnulls[wfuncno];
987 5128 : finalize_windowaggregate(winstate,
988 5128 : &winstate->perfunc[wfuncno],
989 : peraggstate,
990 : result, isnull);
991 :
992 ECB : /*
993 : * save the result in case next row shares the same frame.
994 : *
995 : * XXX in some framing modes, eg ROWS/END_CURRENT_ROW, we can know in
996 : * advance that the next row can't possibly share the same frame. Is
997 : * it worth detecting that and skipping this code?
998 : */
999 CBC 5128 : if (!peraggstate->resulttypeByVal && !*isnull)
1000 : {
1001 GIC 1235 : oldContext = MemoryContextSwitchTo(peraggstate->aggcontext);
1002 1235 : peraggstate->resultValue =
1003 CBC 1235 : datumCopy(*result,
1004 GIC 1235 : peraggstate->resulttypeByVal,
1005 CBC 1235 : peraggstate->resulttypeLen);
1006 GIC 1235 : MemoryContextSwitchTo(oldContext);
1007 : }
1008 : else
1009 : {
1010 3893 : peraggstate->resultValue = *result;
1011 : }
1012 5128 : peraggstate->resultValueIsNull = *isnull;
1013 : }
1014 : }
1015 :
1016 : /*
1017 : * eval_windowfunction
1018 : *
1019 ECB : * Arguments of window functions are not evaluated here, because a window
1020 : * function can need random access to arbitrary rows in the partition.
1021 : * The window function uses the special WinGetFuncArgInPartition and
1022 : * WinGetFuncArgInFrame functions to evaluate the arguments for the rows
1023 : * it wants.
1024 : */
1025 : static void
1026 GIC 391197 : eval_windowfunction(WindowAggState *winstate, WindowStatePerFunc perfuncstate,
1027 : Datum *result, bool *isnull)
1028 : {
1029 391197 : LOCAL_FCINFO(fcinfo, FUNC_MAX_ARGS);
1030 : MemoryContext oldContext;
1031 :
1032 391197 : oldContext = MemoryContextSwitchTo(winstate->ss.ps.ps_ExprContext->ecxt_per_tuple_memory);
1033 ECB :
1034 : /*
1035 : * We don't pass any normal arguments to a window function, but we do pass
1036 : * it the number of arguments, in order to permit window function
1037 : * implementations to support varying numbers of arguments. The real info
1038 : * goes through the WindowObject, which is passed via fcinfo->context.
1039 : */
1040 GIC 391197 : InitFunctionCallInfoData(*fcinfo, &(perfuncstate->flinfo),
1041 ECB : perfuncstate->numArguments,
1042 : perfuncstate->winCollation,
1043 : (void *) perfuncstate->winobj, NULL);
1044 : /* Just in case, make all the regular argument slots be null */
1045 GIC 486360 : for (int argno = 0; argno < perfuncstate->numArguments; argno++)
1046 95163 : fcinfo->args[argno].isnull = true;
1047 : /* Window functions don't have a current aggregate context, either */
1048 391197 : winstate->curaggcontext = NULL;
1049 :
1050 391197 : *result = FunctionCallInvoke(fcinfo);
1051 391182 : *isnull = fcinfo->isnull;
1052 :
1053 ECB : /*
1054 : * The window function might have returned a pass-by-ref result that's
1055 : * just a pointer into one of the WindowObject's temporary slots. That's
1056 : * not a problem if it's the only window function using the WindowObject;
1057 : * but if there's more than one function, we'd better copy the result to
1058 : * ensure it's not clobbered by later window functions.
1059 : */
1060 GIC 391182 : if (!perfuncstate->resulttypeByVal && !fcinfo->isnull &&
1061 GNC 504 : winstate->numfuncs > 1)
1062 GIC 48 : *result = datumCopy(*result,
1063 48 : perfuncstate->resulttypeByVal,
1064 48 : perfuncstate->resulttypeLen);
1065 :
1066 391182 : MemoryContextSwitchTo(oldContext);
1067 391182 : }
1068 ECB :
1069 : /*
1070 : * begin_partition
1071 : * Start buffering rows of the next partition.
1072 : */
1073 : static void
1074 GIC 1551 : begin_partition(WindowAggState *winstate)
1075 : {
1076 CBC 1551 : WindowAgg *node = (WindowAgg *) winstate->ss.ps.plan;
1077 1551 : PlanState *outerPlan = outerPlanState(winstate);
1078 1551 : int frameOptions = winstate->frameOptions;
1079 1551 : int numfuncs = winstate->numfuncs;
1080 ECB : int i;
1081 :
1082 CBC 1551 : winstate->partition_spooled = false;
1083 1551 : winstate->framehead_valid = false;
1084 1551 : winstate->frametail_valid = false;
1085 1551 : winstate->grouptail_valid = false;
1086 1551 : winstate->spooled_rows = 0;
1087 1551 : winstate->currentpos = 0;
1088 1551 : winstate->frameheadpos = 0;
1089 1551 : winstate->frametailpos = 0;
1090 1551 : winstate->currentgroup = 0;
1091 1551 : winstate->frameheadgroup = 0;
1092 1551 : winstate->frametailgroup = 0;
1093 1551 : winstate->groupheadpos = 0;
1094 GIC 1551 : winstate->grouptailpos = -1; /* see update_grouptailpos */
1095 1551 : ExecClearTuple(winstate->agg_row_slot);
1096 1551 : if (winstate->framehead_slot)
1097 437 : ExecClearTuple(winstate->framehead_slot);
1098 1551 : if (winstate->frametail_slot)
1099 CBC 761 : ExecClearTuple(winstate->frametail_slot);
1100 :
1101 ECB : /*
1102 : * If this is the very first partition, we need to fetch the first input
1103 : * row to store in first_part_slot.
1104 : */
1105 GIC 1551 : if (TupIsNull(winstate->first_part_slot))
1106 : {
1107 948 : TupleTableSlot *outerslot = ExecProcNode(outerPlan);
1108 ECB :
1109 CBC 948 : if (!TupIsNull(outerslot))
1110 939 : ExecCopySlot(winstate->first_part_slot, outerslot);
1111 : else
1112 : {
1113 : /* outer plan is empty, so we have nothing to do */
1114 GIC 9 : winstate->partition_spooled = true;
1115 CBC 9 : winstate->more_partitions = false;
1116 GIC 9 : return;
1117 : }
1118 : }
1119 :
1120 : /* Create new tuplestore for this partition */
1121 1542 : winstate->buffer = tuplestore_begin_heap(false, false, work_mem);
1122 ECB :
1123 : /*
1124 : * Set up read pointers for the tuplestore. The current pointer doesn't
1125 : * need BACKWARD capability, but the per-window-function read pointers do,
1126 : * and the aggregate pointer does if we might need to restart aggregation.
1127 : */
1128 CBC 1542 : winstate->current_ptr = 0; /* read pointer 0 is pre-allocated */
1129 :
1130 ECB : /* reset default REWIND capability bit for current ptr */
1131 CBC 1542 : tuplestore_set_eflags(winstate->buffer, 0);
1132 :
1133 : /* create read pointers for aggregates, if needed */
1134 GIC 1542 : if (winstate->numaggs > 0)
1135 : {
1136 882 : WindowObject agg_winobj = winstate->agg_winobj;
1137 CBC 882 : int readptr_flags = 0;
1138 ECB :
1139 : /*
1140 : * If the frame head is potentially movable, or we have an EXCLUSION
1141 : * clause, we might need to restart aggregation ...
1142 : */
1143 CBC 882 : if (!(frameOptions & FRAMEOPTION_START_UNBOUNDED_PRECEDING) ||
1144 GIC 364 : (frameOptions & FRAMEOPTION_EXCLUSION))
1145 : {
1146 ECB : /* ... so create a mark pointer to track the frame head */
1147 GIC 527 : agg_winobj->markptr = tuplestore_alloc_read_pointer(winstate->buffer, 0);
1148 ECB : /* and the read pointer will need BACKWARD capability */
1149 CBC 527 : readptr_flags |= EXEC_FLAG_BACKWARD;
1150 : }
1151 :
1152 882 : agg_winobj->readptr = tuplestore_alloc_read_pointer(winstate->buffer,
1153 ECB : readptr_flags);
1154 GIC 882 : agg_winobj->markpos = -1;
1155 882 : agg_winobj->seekpos = -1;
1156 :
1157 ECB : /* Also reset the row counters for aggregates */
1158 GIC 882 : winstate->aggregatedbase = 0;
1159 CBC 882 : winstate->aggregatedupto = 0;
1160 : }
1161 ECB :
1162 : /* create mark and read pointers for each real window function */
1163 CBC 3423 : for (i = 0; i < numfuncs; i++)
1164 : {
1165 1881 : WindowStatePerFunc perfuncstate = &(winstate->perfunc[i]);
1166 :
1167 1881 : if (!perfuncstate->plain_agg)
1168 : {
1169 918 : WindowObject winobj = perfuncstate->winobj;
1170 ECB :
1171 GIC 918 : winobj->markptr = tuplestore_alloc_read_pointer(winstate->buffer,
1172 : 0);
1173 918 : winobj->readptr = tuplestore_alloc_read_pointer(winstate->buffer,
1174 : EXEC_FLAG_BACKWARD);
1175 918 : winobj->markpos = -1;
1176 918 : winobj->seekpos = -1;
1177 : }
1178 : }
1179 :
1180 : /*
1181 : * If we are in RANGE or GROUPS mode, then determining frame boundaries
1182 ECB : * requires physical access to the frame endpoint rows, except in certain
1183 : * degenerate cases. We create read pointers to point to those rows, to
1184 : * simplify access and ensure that the tuplestore doesn't discard the
1185 : * endpoint rows prematurely. (Must create pointers in exactly the same
1186 : * cases that update_frameheadpos and update_frametailpos need them.)
1187 : */
1188 CBC 1542 : winstate->framehead_ptr = winstate->frametail_ptr = -1; /* if not used */
1189 ECB :
1190 CBC 1542 : if (frameOptions & (FRAMEOPTION_RANGE | FRAMEOPTION_GROUPS))
1191 ECB : {
1192 CBC 981 : if (((frameOptions & FRAMEOPTION_START_CURRENT_ROW) &&
1193 41 : node->ordNumCols != 0) ||
1194 940 : (frameOptions & FRAMEOPTION_START_OFFSET))
1195 437 : winstate->framehead_ptr =
1196 GIC 437 : tuplestore_alloc_read_pointer(winstate->buffer, 0);
1197 981 : if (((frameOptions & FRAMEOPTION_END_CURRENT_ROW) &&
1198 520 : node->ordNumCols != 0) ||
1199 646 : (frameOptions & FRAMEOPTION_END_OFFSET))
1200 761 : winstate->frametail_ptr =
1201 761 : tuplestore_alloc_read_pointer(winstate->buffer, 0);
1202 : }
1203 :
1204 : /*
1205 ECB : * If we have an exclusion clause that requires knowing the boundaries of
1206 : * the current row's peer group, we create a read pointer to track the
1207 : * tail position of the peer group (i.e., first row of the next peer
1208 : * group). The head position does not require its own pointer because we
1209 : * maintain that as a side effect of advancing the current row.
1210 : */
1211 CBC 1542 : winstate->grouptail_ptr = -1;
1212 ECB :
1213 GIC 1542 : if ((frameOptions & (FRAMEOPTION_EXCLUDE_GROUP |
1214 144 : FRAMEOPTION_EXCLUDE_TIES)) &&
1215 144 : node->ordNumCols != 0)
1216 : {
1217 138 : winstate->grouptail_ptr =
1218 138 : tuplestore_alloc_read_pointer(winstate->buffer, 0);
1219 ECB : }
1220 :
1221 : /*
1222 : * Store the first tuple into the tuplestore (it's always available now;
1223 : * we either read it above, or saved it at the end of previous partition)
1224 : */
1225 GIC 1542 : tuplestore_puttupleslot(winstate->buffer, winstate->first_part_slot);
1226 1542 : winstate->spooled_rows++;
1227 : }
1228 ECB :
1229 : /*
1230 : * Read tuples from the outer node, up to and including position 'pos', and
1231 : * store them into the tuplestore. If pos is -1, reads the whole partition.
1232 : */
1233 : static void
1234 GIC 799551 : spool_tuples(WindowAggState *winstate, int64 pos)
1235 ECB : {
1236 CBC 799551 : WindowAgg *node = (WindowAgg *) winstate->ss.ps.plan;
1237 ECB : PlanState *outerPlan;
1238 : TupleTableSlot *outerslot;
1239 : MemoryContext oldcontext;
1240 :
1241 GIC 799551 : if (!winstate->buffer)
1242 9 : return; /* just a safety check */
1243 799542 : if (winstate->partition_spooled)
1244 39650 : return; /* whole partition done already */
1245 :
1246 ECB : /*
1247 : * When in pass-through mode we can just exhaust all tuples in the current
1248 : * partition. We don't need these tuples for any further window function
1249 : * evaluation, however, we do need to keep them around if we're not the
1250 : * top-level window as another WindowAgg node above must see these.
1251 : */
1252 GIC 759892 : if (winstate->status != WINDOWAGG_RUN)
1253 : {
1254 9 : Assert(winstate->status == WINDOWAGG_PASSTHROUGH ||
1255 : winstate->status == WINDOWAGG_PASSTHROUGH_STRICT);
1256 :
1257 9 : pos = -1;
1258 : }
1259 :
1260 : /*
1261 : * If the tuplestore has spilled to disk, alternate reading and writing
1262 ECB : * becomes quite expensive due to frequent buffer flushes. It's cheaper
1263 EUB : * to force the entire partition to get spooled in one go.
1264 : *
1265 ECB : * XXX this is a horrid kluge --- it'd be better to fix the performance
1266 : * problem inside tuplestore. FIXME
1267 : */
1268 CBC 759883 : else if (!tuplestore_in_memory(winstate->buffer))
1269 UIC 0 : pos = -1;
1270 ECB :
1271 GIC 759892 : outerPlan = outerPlanState(winstate);
1272 ECB :
1273 : /* Must be in query context to call outerplan */
1274 GIC 759892 : oldcontext = MemoryContextSwitchTo(winstate->ss.ps.ps_ExprContext->ecxt_per_query_memory);
1275 :
1276 CBC 1972658 : while (winstate->spooled_rows <= pos || pos == -1)
1277 ECB : {
1278 CBC 454374 : outerslot = ExecProcNode(outerPlan);
1279 GIC 454374 : if (TupIsNull(outerslot))
1280 : {
1281 ECB : /* reached the end of the last partition */
1282 GIC 897 : winstate->partition_spooled = true;
1283 CBC 897 : winstate->more_partitions = false;
1284 GIC 897 : break;
1285 ECB : }
1286 :
1287 GIC 453477 : if (node->partNumCols > 0)
1288 : {
1289 CBC 61812 : ExprContext *econtext = winstate->tmpcontext;
1290 :
1291 GIC 61812 : econtext->ecxt_innertuple = winstate->first_part_slot;
1292 61812 : econtext->ecxt_outertuple = outerslot;
1293 :
1294 ECB : /* Check if this tuple still belongs to the current partition */
1295 CBC 61812 : if (!ExecQualAndReset(winstate->partEqfunction, econtext))
1296 ECB : {
1297 : /*
1298 : * end of partition; copy the tuple for the next cycle.
1299 : */
1300 GIC 603 : ExecCopySlot(winstate->first_part_slot, outerslot);
1301 603 : winstate->partition_spooled = true;
1302 603 : winstate->more_partitions = true;
1303 603 : break;
1304 : }
1305 ECB : }
1306 :
1307 : /*
1308 : * Remember the tuple unless we're the top-level window and we're in
1309 : * pass-through mode.
1310 : */
1311 GIC 452874 : if (winstate->status != WINDOWAGG_PASSTHROUGH_STRICT)
1312 : {
1313 ECB : /* Still in partition, so save it into the tuplestore */
1314 GIC 452868 : tuplestore_puttupleslot(winstate->buffer, outerslot);
1315 452868 : winstate->spooled_rows++;
1316 : }
1317 : }
1318 :
1319 759892 : MemoryContextSwitchTo(oldcontext);
1320 : }
1321 :
1322 ECB : /*
1323 : * release_partition
1324 : * clear information kept within a partition, including
1325 : * tuplestore and aggregate results.
1326 : */
1327 : static void
1328 CBC 2607 : release_partition(WindowAggState *winstate)
1329 : {
1330 : int i;
1331 ECB :
1332 CBC 5787 : for (i = 0; i < winstate->numfuncs; i++)
1333 : {
1334 GIC 3180 : WindowStatePerFunc perfuncstate = &(winstate->perfunc[i]);
1335 :
1336 : /* Release any partition-local state of this window function */
1337 3180 : if (perfuncstate->winobj)
1338 1527 : perfuncstate->winobj->localmem = NULL;
1339 : }
1340 :
1341 ECB : /*
1342 : * Release all partition-local memory (in particular, any partition-local
1343 : * state that we might have trashed our pointers to in the above loop, and
1344 : * any aggregate temp data). We don't rely on retail pfree because some
1345 : * aggregates might have allocated data we don't have direct pointers to.
1346 : */
1347 GIC 2607 : MemoryContextResetAndDeleteChildren(winstate->partcontext);
1348 2607 : MemoryContextResetAndDeleteChildren(winstate->aggcontext);
1349 CBC 4260 : for (i = 0; i < winstate->numaggs; i++)
1350 ECB : {
1351 CBC 1653 : if (winstate->peragg[i].aggcontext != winstate->aggcontext)
1352 1456 : MemoryContextResetAndDeleteChildren(winstate->peragg[i].aggcontext);
1353 ECB : }
1354 :
1355 GIC 2607 : if (winstate->buffer)
1356 1518 : tuplestore_end(winstate->buffer);
1357 2607 : winstate->buffer = NULL;
1358 2607 : winstate->partition_spooled = false;
1359 2607 : }
1360 :
1361 : /*
1362 : * row_is_in_frame
1363 : * Determine whether a row is in the current row's window frame according
1364 : * to our window framing rule
1365 : *
1366 : * The caller must have already determined that the row is in the partition
1367 : * and fetched it into a slot. This function just encapsulates the framing
1368 : * rules.
1369 : *
1370 : * Returns:
1371 : * -1, if the row is out of frame and no succeeding rows can be in frame
1372 ECB : * 0, if the row is out of frame but succeeding rows might be in frame
1373 : * 1, if the row is in frame
1374 : *
1375 : * May clobber winstate->temp_slot_2.
1376 : */
1377 : static int
1378 GIC 78588 : row_is_in_frame(WindowAggState *winstate, int64 pos, TupleTableSlot *slot)
1379 : {
1380 78588 : int frameOptions = winstate->frameOptions;
1381 :
1382 CBC 78588 : Assert(pos >= 0); /* else caller error */
1383 ECB :
1384 EUB : /*
1385 : * First, check frame starting conditions. We might as well delegate this
1386 : * to update_frameheadpos always; it doesn't add any notable cost.
1387 : */
1388 GIC 78588 : update_frameheadpos(winstate);
1389 78588 : if (pos < winstate->frameheadpos)
1390 UIC 0 : return 0;
1391 ECB :
1392 : /*
1393 : * Okay so far, now check frame ending conditions. Here, we avoid calling
1394 : * update_frametailpos in simple cases, so as not to spool tuples further
1395 : * ahead than necessary.
1396 : */
1397 CBC 78588 : if (frameOptions & FRAMEOPTION_END_CURRENT_ROW)
1398 : {
1399 63901 : if (frameOptions & FRAMEOPTION_ROWS)
1400 : {
1401 : /* rows after current row are out of frame */
1402 1047 : if (pos > winstate->currentpos)
1403 459 : return -1;
1404 ECB : }
1405 GIC 62854 : else if (frameOptions & (FRAMEOPTION_RANGE | FRAMEOPTION_GROUPS))
1406 : {
1407 EUB : /* following row that is not peer is out of frame */
1408 GIC 62854 : if (pos > winstate->currentpos &&
1409 CBC 61210 : !are_peers(winstate, slot, winstate->ss.ss_ScanTupleSlot))
1410 GIC 626 : return -1;
1411 ECB : }
1412 : else
1413 LBC 0 : Assert(false);
1414 : }
1415 GIC 14687 : else if (frameOptions & FRAMEOPTION_END_OFFSET)
1416 ECB : {
1417 CBC 7881 : if (frameOptions & FRAMEOPTION_ROWS)
1418 : {
1419 1878 : int64 offset = DatumGetInt64(winstate->endOffsetValue);
1420 ECB :
1421 : /* rows after current row + offset are out of frame */
1422 CBC 1878 : if (frameOptions & FRAMEOPTION_END_OFFSET_PRECEDING)
1423 GIC 57 : offset = -offset;
1424 :
1425 CBC 1878 : if (pos > winstate->currentpos + offset)
1426 540 : return -1;
1427 ECB : }
1428 GIC 6003 : else if (frameOptions & (FRAMEOPTION_RANGE | FRAMEOPTION_GROUPS))
1429 : {
1430 EUB : /* hard cases, so delegate to update_frametailpos */
1431 GIC 6003 : update_frametailpos(winstate);
1432 5988 : if (pos >= winstate->frametailpos)
1433 663 : return -1;
1434 ECB : }
1435 : else
1436 LBC 0 : Assert(false);
1437 ECB : }
1438 :
1439 : /* Check exclusion clause */
1440 CBC 76285 : if (frameOptions & FRAMEOPTION_EXCLUDE_CURRENT_ROW)
1441 ECB : {
1442 GIC 1233 : if (pos == winstate->currentpos)
1443 CBC 210 : return 0;
1444 : }
1445 GIC 75052 : else if ((frameOptions & FRAMEOPTION_EXCLUDE_GROUP) ||
1446 CBC 73621 : ((frameOptions & FRAMEOPTION_EXCLUDE_TIES) &&
1447 1485 : pos != winstate->currentpos))
1448 : {
1449 2646 : WindowAgg *node = (WindowAgg *) winstate->ss.ps.plan;
1450 :
1451 ECB : /* If no ORDER BY, all rows are peers with each other */
1452 CBC 2646 : if (node->ordNumCols == 0)
1453 234 : return 0;
1454 : /* Otherwise, check the group boundaries */
1455 GIC 2412 : if (pos >= winstate->groupheadpos)
1456 : {
1457 1296 : update_grouptailpos(winstate);
1458 CBC 1296 : if (pos < winstate->grouptailpos)
1459 GIC 504 : return 0;
1460 : }
1461 : }
1462 :
1463 : /* If we get here, it's in frame */
1464 75337 : return 1;
1465 : }
1466 :
1467 : /*
1468 : * update_frameheadpos
1469 : * make frameheadpos valid for the current row
1470 : *
1471 : * Note that frameheadpos is computed without regard for any window exclusion
1472 ECB : * clause; the current row and/or its peers are considered part of the frame
1473 : * for this purpose even if they must be excluded later.
1474 : *
1475 : * May clobber winstate->temp_slot_2.
1476 : */
1477 : static void
1478 CBC 147763 : update_frameheadpos(WindowAggState *winstate)
1479 ECB : {
1480 GIC 147763 : WindowAgg *node = (WindowAgg *) winstate->ss.ps.plan;
1481 147763 : int frameOptions = winstate->frameOptions;
1482 ECB : MemoryContext oldcontext;
1483 :
1484 CBC 147763 : if (winstate->framehead_valid)
1485 GIC 81082 : return; /* already known for current row */
1486 :
1487 ECB : /* We may be called in a short-lived context */
1488 CBC 66681 : oldcontext = MemoryContextSwitchTo(winstate->ss.ps.ps_ExprContext->ecxt_per_query_memory);
1489 :
1490 66681 : if (frameOptions & FRAMEOPTION_START_UNBOUNDED_PRECEDING)
1491 : {
1492 ECB : /* In UNBOUNDED PRECEDING mode, frame head is always row 0 */
1493 GIC 62249 : winstate->frameheadpos = 0;
1494 62249 : winstate->framehead_valid = true;
1495 ECB : }
1496 CBC 4432 : else if (frameOptions & FRAMEOPTION_START_CURRENT_ROW)
1497 : {
1498 1354 : if (frameOptions & FRAMEOPTION_ROWS)
1499 : {
1500 : /* In ROWS mode, frame head is the same as current */
1501 1140 : winstate->frameheadpos = winstate->currentpos;
1502 GIC 1140 : winstate->framehead_valid = true;
1503 EUB : }
1504 GBC 214 : else if (frameOptions & (FRAMEOPTION_RANGE | FRAMEOPTION_GROUPS))
1505 EUB : {
1506 : /* If no ORDER BY, all rows are peers with each other */
1507 GIC 214 : if (node->ordNumCols == 0)
1508 : {
1509 UIC 0 : winstate->frameheadpos = 0;
1510 0 : winstate->framehead_valid = true;
1511 0 : MemoryContextSwitchTo(oldcontext);
1512 0 : return;
1513 : }
1514 :
1515 : /*
1516 ECB : * In RANGE or GROUPS START_CURRENT_ROW mode, frame head is the
1517 : * first row that is a peer of current row. We keep a copy of the
1518 : * last-known frame head row in framehead_slot, and advance as
1519 : * necessary. Note that if we reach end of partition, we will
1520 : * leave frameheadpos = end+1 and framehead_slot empty.
1521 : */
1522 CBC 214 : tuplestore_select_read_pointer(winstate->buffer,
1523 : winstate->framehead_ptr);
1524 GBC 214 : if (winstate->frameheadpos == 0 &&
1525 GIC 106 : TupIsNull(winstate->framehead_slot))
1526 : {
1527 ECB : /* fetch first row into framehead_slot, if we didn't already */
1528 GIC 41 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
1529 ECB : winstate->framehead_slot))
1530 UIC 0 : elog(ERROR, "unexpected end of tuplestore");
1531 ECB : }
1532 :
1533 CBC 372 : while (!TupIsNull(winstate->framehead_slot))
1534 ECB : {
1535 CBC 372 : if (are_peers(winstate, winstate->framehead_slot,
1536 : winstate->ss.ss_ScanTupleSlot))
1537 GBC 214 : break; /* this row is the correct frame head */
1538 : /* Note we advance frameheadpos even if the fetch fails */
1539 CBC 158 : winstate->frameheadpos++;
1540 GIC 158 : spool_tuples(winstate, winstate->frameheadpos);
1541 158 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
1542 EUB : winstate->framehead_slot))
1543 UIC 0 : break; /* end of partition */
1544 ECB : }
1545 GIC 214 : winstate->framehead_valid = true;
1546 ECB : }
1547 : else
1548 UIC 0 : Assert(false);
1549 ECB : }
1550 GIC 3078 : else if (frameOptions & FRAMEOPTION_START_OFFSET)
1551 ECB : {
1552 CBC 3078 : if (frameOptions & FRAMEOPTION_ROWS)
1553 : {
1554 ECB : /* In ROWS mode, bound is physically n before/after current */
1555 GIC 756 : int64 offset = DatumGetInt64(winstate->startOffsetValue);
1556 ECB :
1557 CBC 756 : if (frameOptions & FRAMEOPTION_START_OFFSET_PRECEDING)
1558 726 : offset = -offset;
1559 :
1560 GIC 756 : winstate->frameheadpos = winstate->currentpos + offset;
1561 EUB : /* frame head can't go before first row */
1562 GBC 756 : if (winstate->frameheadpos < 0)
1563 111 : winstate->frameheadpos = 0;
1564 GIC 645 : else if (winstate->frameheadpos > winstate->currentpos + 1)
1565 ECB : {
1566 : /* make sure frameheadpos is not past end of partition */
1567 LBC 0 : spool_tuples(winstate, winstate->frameheadpos - 1);
1568 UIC 0 : if (winstate->frameheadpos > winstate->spooled_rows)
1569 0 : winstate->frameheadpos = winstate->spooled_rows;
1570 : }
1571 GIC 756 : winstate->framehead_valid = true;
1572 : }
1573 2322 : else if (frameOptions & FRAMEOPTION_RANGE)
1574 : {
1575 : /*
1576 : * In RANGE START_OFFSET mode, frame head is the first row that
1577 ECB : * satisfies the in_range constraint relative to the current row.
1578 : * We keep a copy of the last-known frame head row in
1579 : * framehead_slot, and advance as necessary. Note that if we
1580 : * reach end of partition, we will leave frameheadpos = end+1 and
1581 : * framehead_slot empty.
1582 : */
1583 GIC 1632 : int sortCol = node->ordColIdx[0];
1584 : bool sub,
1585 ECB : less;
1586 :
1587 : /* We must have an ordering column */
1588 CBC 1632 : Assert(node->ordNumCols == 1);
1589 ECB :
1590 : /* Precompute flags for in_range checks */
1591 CBC 1632 : if (frameOptions & FRAMEOPTION_START_OFFSET_PRECEDING)
1592 GIC 1416 : sub = true; /* subtract startOffset from current row */
1593 ECB : else
1594 CBC 216 : sub = false; /* add it */
1595 GIC 1632 : less = false; /* normally, we want frame head >= sum */
1596 : /* If sort order is descending, flip both flags */
1597 CBC 1632 : if (!winstate->inRangeAsc)
1598 : {
1599 282 : sub = !sub;
1600 282 : less = true;
1601 : }
1602 :
1603 1632 : tuplestore_select_read_pointer(winstate->buffer,
1604 : winstate->framehead_ptr);
1605 GBC 1632 : if (winstate->frameheadpos == 0 &&
1606 GIC 861 : TupIsNull(winstate->framehead_slot))
1607 : {
1608 ECB : /* fetch first row into framehead_slot, if we didn't already */
1609 GIC 210 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
1610 : winstate->framehead_slot))
1611 UIC 0 : elog(ERROR, "unexpected end of tuplestore");
1612 : }
1613 :
1614 GIC 2721 : while (!TupIsNull(winstate->framehead_slot))
1615 ECB : {
1616 : Datum headval,
1617 : currval;
1618 : bool headisnull,
1619 : currisnull;
1620 :
1621 GIC 2685 : headval = slot_getattr(winstate->framehead_slot, sortCol,
1622 ECB : &headisnull);
1623 GIC 2685 : currval = slot_getattr(winstate->ss.ss_ScanTupleSlot, sortCol,
1624 : &currisnull);
1625 CBC 2685 : if (headisnull || currisnull)
1626 : {
1627 : /* order of the rows depends only on nulls_first */
1628 GIC 54 : if (winstate->inRangeNullsFirst)
1629 : {
1630 : /* advance head if head is null and curr is not */
1631 CBC 24 : if (!headisnull || currisnull)
1632 : break;
1633 : }
1634 : else
1635 : {
1636 : /* advance head if head is not null and curr is null */
1637 30 : if (headisnull || !currisnull)
1638 : break;
1639 : }
1640 : }
1641 : else
1642 : {
1643 GIC 2631 : if (DatumGetBool(FunctionCall5Coll(&winstate->startInRangeFunc,
1644 ECB : winstate->inRangeColl,
1645 : headval,
1646 : currval,
1647 : winstate->startOffsetValue,
1648 : BoolGetDatum(sub),
1649 : BoolGetDatum(less))))
1650 GIC 1551 : break; /* this row is the correct frame head */
1651 ECB : }
1652 : /* Note we advance frameheadpos even if the fetch fails */
1653 CBC 1107 : winstate->frameheadpos++;
1654 GIC 1107 : spool_tuples(winstate, winstate->frameheadpos);
1655 CBC 1107 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
1656 : winstate->framehead_slot))
1657 GIC 18 : break; /* end of partition */
1658 : }
1659 1629 : winstate->framehead_valid = true;
1660 : }
1661 690 : else if (frameOptions & FRAMEOPTION_GROUPS)
1662 : {
1663 : /*
1664 : * In GROUPS START_OFFSET mode, frame head is the first row of the
1665 ECB : * first peer group whose number satisfies the offset constraint.
1666 : * We keep a copy of the last-known frame head row in
1667 : * framehead_slot, and advance as necessary. Note that if we
1668 : * reach end of partition, we will leave frameheadpos = end+1 and
1669 : * framehead_slot empty.
1670 : */
1671 CBC 690 : int64 offset = DatumGetInt64(winstate->startOffsetValue);
1672 : int64 minheadgroup;
1673 ECB :
1674 GIC 690 : if (frameOptions & FRAMEOPTION_START_OFFSET_PRECEDING)
1675 CBC 564 : minheadgroup = winstate->currentgroup - offset;
1676 ECB : else
1677 GIC 126 : minheadgroup = winstate->currentgroup + offset;
1678 :
1679 CBC 690 : tuplestore_select_read_pointer(winstate->buffer,
1680 : winstate->framehead_ptr);
1681 GBC 690 : if (winstate->frameheadpos == 0 &&
1682 GIC 375 : TupIsNull(winstate->framehead_slot))
1683 : {
1684 ECB : /* fetch first row into framehead_slot, if we didn't already */
1685 GIC 186 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
1686 ECB : winstate->framehead_slot))
1687 LBC 0 : elog(ERROR, "unexpected end of tuplestore");
1688 ECB : }
1689 :
1690 CBC 1761 : while (!TupIsNull(winstate->framehead_slot))
1691 ECB : {
1692 CBC 1059 : if (winstate->frameheadgroup >= minheadgroup)
1693 GIC 660 : break; /* this row is the correct frame head */
1694 CBC 399 : ExecCopySlot(winstate->temp_slot_2, winstate->framehead_slot);
1695 ECB : /* Note we advance frameheadpos even if the fetch fails */
1696 GIC 399 : winstate->frameheadpos++;
1697 CBC 399 : spool_tuples(winstate, winstate->frameheadpos);
1698 GIC 399 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
1699 ECB : winstate->framehead_slot))
1700 CBC 18 : break; /* end of partition */
1701 GIC 381 : if (!are_peers(winstate, winstate->temp_slot_2,
1702 : winstate->framehead_slot))
1703 GBC 261 : winstate->frameheadgroup++;
1704 : }
1705 GIC 690 : ExecClearTuple(winstate->temp_slot_2);
1706 GBC 690 : winstate->framehead_valid = true;
1707 : }
1708 ECB : else
1709 UIC 0 : Assert(false);
1710 : }
1711 : else
1712 0 : Assert(false);
1713 :
1714 GIC 66678 : MemoryContextSwitchTo(oldcontext);
1715 : }
1716 :
1717 : /*
1718 : * update_frametailpos
1719 : * make frametailpos valid for the current row
1720 : *
1721 : * Note that frametailpos is computed without regard for any window exclusion
1722 ECB : * clause; the current row and/or its peers are considered part of the frame
1723 : * for this purpose even if they must be excluded later.
1724 : *
1725 : * May clobber winstate->temp_slot_2.
1726 : */
1727 : static void
1728 CBC 72136 : update_frametailpos(WindowAggState *winstate)
1729 ECB : {
1730 GIC 72136 : WindowAgg *node = (WindowAgg *) winstate->ss.ps.plan;
1731 72136 : int frameOptions = winstate->frameOptions;
1732 ECB : MemoryContext oldcontext;
1733 :
1734 CBC 72136 : if (winstate->frametail_valid)
1735 GIC 7326 : return; /* already known for current row */
1736 :
1737 ECB : /* We may be called in a short-lived context */
1738 CBC 64810 : oldcontext = MemoryContextSwitchTo(winstate->ss.ps.ps_ExprContext->ecxt_per_query_memory);
1739 ECB :
1740 GIC 64810 : if (frameOptions & FRAMEOPTION_END_UNBOUNDED_FOLLOWING)
1741 ECB : {
1742 : /* In UNBOUNDED FOLLOWING mode, all partition rows are in frame */
1743 CBC 90 : spool_tuples(winstate, -1);
1744 GIC 90 : winstate->frametailpos = winstate->spooled_rows;
1745 90 : winstate->frametail_valid = true;
1746 ECB : }
1747 CBC 64720 : else if (frameOptions & FRAMEOPTION_END_CURRENT_ROW)
1748 : {
1749 62137 : if (frameOptions & FRAMEOPTION_ROWS)
1750 : {
1751 : /* In ROWS mode, exactly the rows up to current are in frame */
1752 60 : winstate->frametailpos = winstate->currentpos + 1;
1753 GIC 60 : winstate->frametail_valid = true;
1754 ECB : }
1755 CBC 62077 : else if (frameOptions & (FRAMEOPTION_RANGE | FRAMEOPTION_GROUPS))
1756 ECB : {
1757 : /* If no ORDER BY, all rows are peers with each other */
1758 CBC 62077 : if (node->ordNumCols == 0)
1759 : {
1760 GIC 30 : spool_tuples(winstate, -1);
1761 30 : winstate->frametailpos = winstate->spooled_rows;
1762 30 : winstate->frametail_valid = true;
1763 30 : MemoryContextSwitchTo(oldcontext);
1764 30 : return;
1765 : }
1766 :
1767 : /*
1768 : * In RANGE or GROUPS END_CURRENT_ROW mode, frame end is the last
1769 ECB : * row that is a peer of current row, frame tail is the row after
1770 : * that (if any). We keep a copy of the last-known frame tail row
1771 : * in frametail_slot, and advance as necessary. Note that if we
1772 : * reach end of partition, we will leave frametailpos = end+1 and
1773 : * frametail_slot empty.
1774 : */
1775 CBC 62047 : tuplestore_select_read_pointer(winstate->buffer,
1776 : winstate->frametail_ptr);
1777 GBC 62047 : if (winstate->frametailpos == 0 &&
1778 GIC 332 : TupIsNull(winstate->frametail_slot))
1779 : {
1780 ECB : /* fetch first row into frametail_slot, if we didn't already */
1781 GIC 332 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
1782 ECB : winstate->frametail_slot))
1783 LBC 0 : elog(ERROR, "unexpected end of tuplestore");
1784 : }
1785 ECB :
1786 GIC 123768 : while (!TupIsNull(winstate->frametail_slot))
1787 ECB : {
1788 CBC 111708 : if (winstate->frametailpos > winstate->currentpos &&
1789 109838 : !are_peers(winstate, winstate->frametail_slot,
1790 : winstate->ss.ss_ScanTupleSlot))
1791 49661 : break; /* this row is the frame tail */
1792 : /* Note we advance frametailpos even if the fetch fails */
1793 62047 : winstate->frametailpos++;
1794 GIC 62047 : spool_tuples(winstate, winstate->frametailpos);
1795 62047 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
1796 EUB : winstate->frametail_slot))
1797 GIC 326 : break; /* end of partition */
1798 ECB : }
1799 GIC 62047 : winstate->frametail_valid = true;
1800 ECB : }
1801 : else
1802 UIC 0 : Assert(false);
1803 ECB : }
1804 GIC 2583 : else if (frameOptions & FRAMEOPTION_END_OFFSET)
1805 ECB : {
1806 GBC 2583 : if (frameOptions & FRAMEOPTION_ROWS)
1807 : {
1808 ECB : /* In ROWS mode, bound is physically n before/after current */
1809 GIC 90 : int64 offset = DatumGetInt64(winstate->endOffsetValue);
1810 ECB :
1811 GBC 90 : if (frameOptions & FRAMEOPTION_END_OFFSET_PRECEDING)
1812 LBC 0 : offset = -offset;
1813 :
1814 GIC 90 : winstate->frametailpos = winstate->currentpos + offset + 1;
1815 ECB : /* smallest allowable value of frametailpos is 0 */
1816 CBC 90 : if (winstate->frametailpos < 0)
1817 LBC 0 : winstate->frametailpos = 0;
1818 GIC 90 : else if (winstate->frametailpos > winstate->currentpos + 1)
1819 ECB : {
1820 : /* make sure frametailpos is not past end of partition */
1821 CBC 90 : spool_tuples(winstate, winstate->frametailpos - 1);
1822 GIC 90 : if (winstate->frametailpos > winstate->spooled_rows)
1823 18 : winstate->frametailpos = winstate->spooled_rows;
1824 : }
1825 90 : winstate->frametail_valid = true;
1826 : }
1827 2493 : else if (frameOptions & FRAMEOPTION_RANGE)
1828 : {
1829 : /*
1830 : * In RANGE END_OFFSET mode, frame end is the last row that
1831 ECB : * satisfies the in_range constraint relative to the current row,
1832 : * frame tail is the row after that (if any). We keep a copy of
1833 : * the last-known frame tail row in frametail_slot, and advance as
1834 : * necessary. Note that if we reach end of partition, we will
1835 : * leave frametailpos = end+1 and frametail_slot empty.
1836 : */
1837 GIC 1833 : int sortCol = node->ordColIdx[0];
1838 : bool sub,
1839 ECB : less;
1840 :
1841 : /* We must have an ordering column */
1842 CBC 1833 : Assert(node->ordNumCols == 1);
1843 ECB :
1844 : /* Precompute flags for in_range checks */
1845 CBC 1833 : if (frameOptions & FRAMEOPTION_END_OFFSET_PRECEDING)
1846 GIC 276 : sub = true; /* subtract endOffset from current row */
1847 ECB : else
1848 CBC 1557 : sub = false; /* add it */
1849 GIC 1833 : less = true; /* normally, we want frame tail <= sum */
1850 : /* If sort order is descending, flip both flags */
1851 CBC 1833 : if (!winstate->inRangeAsc)
1852 : {
1853 309 : sub = !sub;
1854 309 : less = false;
1855 : }
1856 :
1857 1833 : tuplestore_select_read_pointer(winstate->buffer,
1858 : winstate->frametail_ptr);
1859 GBC 1833 : if (winstate->frametailpos == 0 &&
1860 GIC 291 : TupIsNull(winstate->frametail_slot))
1861 : {
1862 ECB : /* fetch first row into frametail_slot, if we didn't already */
1863 GIC 240 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
1864 : winstate->frametail_slot))
1865 UIC 0 : elog(ERROR, "unexpected end of tuplestore");
1866 : }
1867 :
1868 GIC 3405 : while (!TupIsNull(winstate->frametail_slot))
1869 ECB : {
1870 : Datum tailval,
1871 : currval;
1872 : bool tailisnull,
1873 : currisnull;
1874 :
1875 GIC 2928 : tailval = slot_getattr(winstate->frametail_slot, sortCol,
1876 ECB : &tailisnull);
1877 GIC 2928 : currval = slot_getattr(winstate->ss.ss_ScanTupleSlot, sortCol,
1878 : &currisnull);
1879 CBC 2928 : if (tailisnull || currisnull)
1880 ECB : {
1881 : /* order of the rows depends only on nulls_first */
1882 GIC 54 : if (winstate->inRangeNullsFirst)
1883 : {
1884 : /* advance tail if tail is null or curr is not */
1885 CBC 24 : if (!tailisnull)
1886 1341 : break;
1887 : }
1888 : else
1889 : {
1890 : /* advance tail if tail is not null or curr is null */
1891 30 : if (!currisnull)
1892 GIC 18 : break;
1893 : }
1894 : }
1895 : else
1896 : {
1897 2874 : if (!DatumGetBool(FunctionCall5Coll(&winstate->endInRangeFunc,
1898 ECB : winstate->inRangeColl,
1899 : tailval,
1900 : currval,
1901 : winstate->endOffsetValue,
1902 : BoolGetDatum(sub),
1903 : BoolGetDatum(less))))
1904 GIC 1110 : break; /* this row is the correct frame tail */
1905 ECB : }
1906 : /* Note we advance frametailpos even if the fetch fails */
1907 CBC 1773 : winstate->frametailpos++;
1908 GIC 1773 : spool_tuples(winstate, winstate->frametailpos);
1909 CBC 1773 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
1910 : winstate->frametail_slot))
1911 GIC 201 : break; /* end of partition */
1912 : }
1913 1818 : winstate->frametail_valid = true;
1914 : }
1915 660 : else if (frameOptions & FRAMEOPTION_GROUPS)
1916 : {
1917 : /*
1918 : * In GROUPS END_OFFSET mode, frame end is the last row of the
1919 ECB : * last peer group whose number satisfies the offset constraint,
1920 : * and frame tail is the row after that (if any). We keep a copy
1921 : * of the last-known frame tail row in frametail_slot, and advance
1922 : * as necessary. Note that if we reach end of partition, we will
1923 : * leave frametailpos = end+1 and frametail_slot empty.
1924 : */
1925 CBC 660 : int64 offset = DatumGetInt64(winstate->endOffsetValue);
1926 : int64 maxtailgroup;
1927 ECB :
1928 GIC 660 : if (frameOptions & FRAMEOPTION_END_OFFSET_PRECEDING)
1929 CBC 36 : maxtailgroup = winstate->currentgroup - offset;
1930 ECB : else
1931 GIC 624 : maxtailgroup = winstate->currentgroup + offset;
1932 :
1933 CBC 660 : tuplestore_select_read_pointer(winstate->buffer,
1934 : winstate->frametail_ptr);
1935 GBC 660 : if (winstate->frametailpos == 0 &&
1936 GIC 192 : TupIsNull(winstate->frametail_slot))
1937 : {
1938 ECB : /* fetch first row into frametail_slot, if we didn't already */
1939 GIC 183 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
1940 ECB : winstate->frametail_slot))
1941 LBC 0 : elog(ERROR, "unexpected end of tuplestore");
1942 ECB : }
1943 :
1944 CBC 1794 : while (!TupIsNull(winstate->frametail_slot))
1945 ECB : {
1946 CBC 1020 : if (winstate->frametailgroup > maxtailgroup)
1947 GIC 372 : break; /* this row is the correct frame tail */
1948 CBC 648 : ExecCopySlot(winstate->temp_slot_2, winstate->frametail_slot);
1949 ECB : /* Note we advance frametailpos even if the fetch fails */
1950 GIC 648 : winstate->frametailpos++;
1951 CBC 648 : spool_tuples(winstate, winstate->frametailpos);
1952 GIC 648 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
1953 ECB : winstate->frametail_slot))
1954 CBC 174 : break; /* end of partition */
1955 GIC 474 : if (!are_peers(winstate, winstate->temp_slot_2,
1956 : winstate->frametail_slot))
1957 GBC 300 : winstate->frametailgroup++;
1958 : }
1959 GIC 660 : ExecClearTuple(winstate->temp_slot_2);
1960 GBC 660 : winstate->frametail_valid = true;
1961 : }
1962 ECB : else
1963 UIC 0 : Assert(false);
1964 : }
1965 : else
1966 0 : Assert(false);
1967 :
1968 GIC 64765 : MemoryContextSwitchTo(oldcontext);
1969 : }
1970 :
1971 : /*
1972 ECB : * update_grouptailpos
1973 : * make grouptailpos valid for the current row
1974 : *
1975 : * May clobber winstate->temp_slot_2.
1976 : */
1977 : static void
1978 CBC 2436 : update_grouptailpos(WindowAggState *winstate)
1979 : {
1980 GIC 2436 : WindowAgg *node = (WindowAgg *) winstate->ss.ps.plan;
1981 ECB : MemoryContext oldcontext;
1982 :
1983 GIC 2436 : if (winstate->grouptail_valid)
1984 CBC 1977 : return; /* already known for current row */
1985 :
1986 EUB : /* We may be called in a short-lived context */
1987 GBC 459 : oldcontext = MemoryContextSwitchTo(winstate->ss.ps.ps_ExprContext->ecxt_per_query_memory);
1988 EUB :
1989 : /* If no ORDER BY, all rows are peers with each other */
1990 GBC 459 : if (node->ordNumCols == 0)
1991 : {
1992 UIC 0 : spool_tuples(winstate, -1);
1993 0 : winstate->grouptailpos = winstate->spooled_rows;
1994 0 : winstate->grouptail_valid = true;
1995 0 : MemoryContextSwitchTo(oldcontext);
1996 0 : return;
1997 : }
1998 :
1999 : /*
2000 ECB : * Because grouptail_valid is reset only when current row advances into a
2001 : * new peer group, we always reach here knowing that grouptailpos needs to
2002 : * be advanced by at least one row. Hence, unlike the otherwise similar
2003 : * case for frame tail tracking, we do not need persistent storage of the
2004 : * group tail row.
2005 : */
2006 CBC 459 : Assert(winstate->grouptailpos <= winstate->currentpos);
2007 459 : tuplestore_select_read_pointer(winstate->buffer,
2008 ECB : winstate->grouptail_ptr);
2009 : for (;;)
2010 : {
2011 : /* Note we advance grouptailpos even if the fetch fails */
2012 CBC 879 : winstate->grouptailpos++;
2013 GIC 879 : spool_tuples(winstate, winstate->grouptailpos);
2014 CBC 879 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
2015 : winstate->temp_slot_2))
2016 129 : break; /* end of partition */
2017 750 : if (winstate->grouptailpos > winstate->currentpos &&
2018 GIC 621 : !are_peers(winstate, winstate->temp_slot_2,
2019 ECB : winstate->ss.ss_ScanTupleSlot))
2020 GIC 330 : break; /* this row is the group tail */
2021 : }
2022 459 : ExecClearTuple(winstate->temp_slot_2);
2023 459 : winstate->grouptail_valid = true;
2024 :
2025 459 : MemoryContextSwitchTo(oldcontext);
2026 : }
2027 :
2028 :
2029 : /* -----------------
2030 : * ExecWindowAgg
2031 : *
2032 : * ExecWindowAgg receives tuples from its outer subplan and
2033 ECB : * stores them into a tuplestore, then processes window functions.
2034 : * This node doesn't reduce nor qualify any row so the number of
2035 : * returned rows is exactly the same as its outer subplan's result.
2036 : * -----------------
2037 : */
2038 : static TupleTableSlot *
2039 GIC 395277 : ExecWindowAgg(PlanState *pstate)
2040 : {
2041 CBC 395277 : WindowAggState *winstate = castNode(WindowAggState, pstate);
2042 : TupleTableSlot *slot;
2043 ECB : ExprContext *econtext;
2044 EUB : int i;
2045 : int numfuncs;
2046 :
2047 GIC 395277 : CHECK_FOR_INTERRUPTS();
2048 :
2049 395277 : if (winstate->status == WINDOWAGG_DONE)
2050 UIC 0 : return NULL;
2051 ECB :
2052 : /*
2053 : * Compute frame offset values, if any, during first call (or after a
2054 : * rescan). These are assumed to hold constant throughout the scan; if
2055 : * user gives us a volatile expression, we'll only use its initial value.
2056 : */
2057 GIC 395277 : if (winstate->all_first)
2058 : {
2059 CBC 948 : int frameOptions = winstate->frameOptions;
2060 ECB : Datum value;
2061 : bool isnull;
2062 : int16 len;
2063 : bool byval;
2064 :
2065 GNC 948 : econtext = winstate->ss.ps.ps_ExprContext;
2066 :
2067 GIC 948 : if (frameOptions & FRAMEOPTION_START_OFFSET)
2068 ECB : {
2069 GBC 330 : Assert(winstate->startOffset != NULL);
2070 GIC 330 : value = ExecEvalExprSwitchContext(winstate->startOffset,
2071 : econtext,
2072 : &isnull);
2073 CBC 330 : if (isnull)
2074 UIC 0 : ereport(ERROR,
2075 ECB : (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
2076 : errmsg("frame starting offset must not be null")));
2077 : /* copy value into query-lifespan context */
2078 GIC 330 : get_typlenbyval(exprType((Node *) winstate->startOffset->expr),
2079 ECB : &len, &byval);
2080 GIC 330 : winstate->startOffsetValue = datumCopy(value, byval, len);
2081 CBC 330 : if (frameOptions & (FRAMEOPTION_ROWS | FRAMEOPTION_GROUPS))
2082 EUB : {
2083 : /* value is known to be int8 */
2084 GIC 147 : int64 offset = DatumGetInt64(value);
2085 :
2086 147 : if (offset < 0)
2087 LBC 0 : ereport(ERROR,
2088 : (errcode(ERRCODE_INVALID_PRECEDING_OR_FOLLOWING_SIZE),
2089 ECB : errmsg("frame starting offset must not be negative")));
2090 : }
2091 : }
2092 GIC 948 : if (frameOptions & FRAMEOPTION_END_OFFSET)
2093 ECB : {
2094 GBC 372 : Assert(winstate->endOffset != NULL);
2095 GIC 372 : value = ExecEvalExprSwitchContext(winstate->endOffset,
2096 : econtext,
2097 : &isnull);
2098 CBC 372 : if (isnull)
2099 UIC 0 : ereport(ERROR,
2100 ECB : (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
2101 : errmsg("frame ending offset must not be null")));
2102 : /* copy value into query-lifespan context */
2103 GIC 372 : get_typlenbyval(exprType((Node *) winstate->endOffset->expr),
2104 ECB : &len, &byval);
2105 GIC 372 : winstate->endOffsetValue = datumCopy(value, byval, len);
2106 CBC 372 : if (frameOptions & (FRAMEOPTION_ROWS | FRAMEOPTION_GROUPS))
2107 EUB : {
2108 : /* value is known to be int8 */
2109 GIC 156 : int64 offset = DatumGetInt64(value);
2110 :
2111 156 : if (offset < 0)
2112 LBC 0 : ereport(ERROR,
2113 : (errcode(ERRCODE_INVALID_PRECEDING_OR_FOLLOWING_SIZE),
2114 : errmsg("frame ending offset must not be negative")));
2115 : }
2116 : }
2117 GIC 948 : winstate->all_first = false;
2118 ECB : }
2119 :
2120 : /* We need to loop as the runCondition or qual may filter out tuples */
2121 : for (;;)
2122 : {
2123 GIC 395313 : if (winstate->buffer == NULL)
2124 : {
2125 : /* Initialize for first partition and set current row = 0 */
2126 948 : begin_partition(winstate);
2127 ECB : /* If there are no input rows, we'll detect that and exit below */
2128 : }
2129 : else
2130 : {
2131 : /* Advance current row within partition */
2132 GIC 394365 : winstate->currentpos++;
2133 : /* This might mean that the frame moves, too */
2134 394365 : winstate->framehead_valid = false;
2135 394365 : winstate->frametail_valid = false;
2136 : /* we don't need to invalidate grouptail here; see below */
2137 : }
2138 ECB :
2139 : /*
2140 : * Spool all tuples up to and including the current row, if we haven't
2141 : * already
2142 : */
2143 GIC 395313 : spool_tuples(winstate, winstate->currentpos);
2144 ECB :
2145 : /* Move to the next partition if we reached the end of this partition */
2146 CBC 395313 : if (winstate->partition_spooled &&
2147 GIC 15835 : winstate->currentpos >= winstate->spooled_rows)
2148 ECB : {
2149 CBC 1503 : release_partition(winstate);
2150 :
2151 GIC 1503 : if (winstate->more_partitions)
2152 ECB : {
2153 GIC 603 : begin_partition(winstate);
2154 603 : Assert(winstate->spooled_rows > 0);
2155 :
2156 : /* Come out of pass-through mode when changing partition */
2157 CBC 603 : winstate->status = WINDOWAGG_RUN;
2158 ECB : }
2159 : else
2160 : {
2161 : /* No further partitions? We're done */
2162 GIC 900 : winstate->status = WINDOWAGG_DONE;
2163 CBC 900 : return NULL;
2164 : }
2165 : }
2166 ECB :
2167 : /* final output execution is in ps_ExprContext */
2168 GIC 394413 : econtext = winstate->ss.ps.ps_ExprContext;
2169 :
2170 : /* Clear the per-output-tuple context for current row */
2171 394413 : ResetExprContext(econtext);
2172 :
2173 : /*
2174 : * Read the current row from the tuplestore, and save in
2175 : * ScanTupleSlot. (We can't rely on the outerplan's output slot
2176 : * because we may have to read beyond the current row. Also, we have
2177 : * to actually copy the row out of the tuplestore, since window
2178 : * function evaluation might cause the tuplestore to dump its state to
2179 : * disk.)
2180 : *
2181 : * In GROUPS mode, or when tracking a group-oriented exclusion clause,
2182 : * we must also detect entering a new peer group and update associated
2183 ECB : * state when that happens. We use temp_slot_2 to temporarily hold
2184 : * the previous row for this purpose.
2185 : *
2186 : * Current row must be in the tuplestore, since we spooled it above.
2187 : */
2188 GIC 394413 : tuplestore_select_read_pointer(winstate->buffer, winstate->current_ptr);
2189 CBC 394413 : if ((winstate->frameOptions & (FRAMEOPTION_GROUPS |
2190 ECB : FRAMEOPTION_EXCLUDE_GROUP |
2191 GIC 1449 : FRAMEOPTION_EXCLUDE_TIES)) &&
2192 GBC 1449 : winstate->currentpos > 0)
2193 ECB : {
2194 GIC 1179 : ExecCopySlot(winstate->temp_slot_2, winstate->ss.ss_ScanTupleSlot);
2195 1179 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
2196 ECB : winstate->ss.ss_ScanTupleSlot))
2197 LBC 0 : elog(ERROR, "unexpected end of tuplestore");
2198 CBC 1179 : if (!are_peers(winstate, winstate->temp_slot_2,
2199 : winstate->ss.ss_ScanTupleSlot))
2200 ECB : {
2201 GIC 621 : winstate->currentgroup++;
2202 621 : winstate->groupheadpos = winstate->currentpos;
2203 621 : winstate->grouptail_valid = false;
2204 ECB : }
2205 GIC 1179 : ExecClearTuple(winstate->temp_slot_2);
2206 EUB : }
2207 : else
2208 : {
2209 GIC 393234 : if (!tuplestore_gettupleslot(winstate->buffer, true, true,
2210 ECB : winstate->ss.ss_ScanTupleSlot))
2211 UIC 0 : elog(ERROR, "unexpected end of tuplestore");
2212 : }
2213 :
2214 : /* don't evaluate the window functions when we're in pass-through mode */
2215 CBC 394413 : if (winstate->status == WINDOWAGG_RUN)
2216 ECB : {
2217 : /*
2218 : * Evaluate true window functions
2219 : */
2220 CBC 394386 : numfuncs = winstate->numfuncs;
2221 851337 : for (i = 0; i < numfuncs; i++)
2222 ECB : {
2223 CBC 456966 : WindowStatePerFunc perfuncstate = &(winstate->perfunc[i]);
2224 ECB :
2225 GIC 456966 : if (perfuncstate->plain_agg)
2226 65769 : continue;
2227 391197 : eval_windowfunction(winstate, perfuncstate,
2228 391197 : &(econtext->ecxt_aggvalues[perfuncstate->wfuncstate->wfuncno]),
2229 391197 : &(econtext->ecxt_aggnulls[perfuncstate->wfuncstate->wfuncno]));
2230 ECB : }
2231 :
2232 : /*
2233 : * Evaluate aggregates
2234 : */
2235 GIC 394371 : if (winstate->numaggs > 0)
2236 64944 : eval_windowaggregates(winstate);
2237 : }
2238 :
2239 : /*
2240 : * If we have created auxiliary read pointers for the frame or group
2241 : * boundaries, force them to be kept up-to-date, because we don't know
2242 : * whether the window function(s) will do anything that requires that.
2243 : * Failing to advance the pointers would result in being unable to
2244 ECB : * trim data from the tuplestore, which is bad. (If we could know in
2245 : * advance whether the window functions will use frame boundary info,
2246 : * we could skip creating these pointers in the first place ... but
2247 : * unfortunately the window function API doesn't require that.)
2248 : */
2249 CBC 394389 : if (winstate->framehead_ptr >= 0)
2250 GIC 2518 : update_frameheadpos(winstate);
2251 394389 : if (winstate->frametail_ptr >= 0)
2252 64525 : update_frametailpos(winstate);
2253 394389 : if (winstate->grouptail_ptr >= 0)
2254 CBC 750 : update_grouptailpos(winstate);
2255 :
2256 : /*
2257 : * Truncate any no-longer-needed rows from the tuplestore.
2258 : */
2259 GIC 394389 : tuplestore_trim(winstate->buffer);
2260 :
2261 ECB : /*
2262 : * Form and return a projection tuple using the windowfunc results and
2263 : * the current row. Setting ecxt_outertuple arranges that any Vars
2264 : * will be evaluated with respect to that row.
2265 : */
2266 GIC 394389 : econtext->ecxt_outertuple = winstate->ss.ss_ScanTupleSlot;
2267 ECB :
2268 GIC 394389 : slot = ExecProject(winstate->ss.ps.ps_ProjInfo);
2269 :
2270 394389 : if (winstate->status == WINDOWAGG_RUN)
2271 : {
2272 394362 : econtext->ecxt_scantuple = slot;
2273 ECB :
2274 : /*
2275 : * Now evaluate the run condition to see if we need to go into
2276 : * pass-through mode, or maybe stop completely.
2277 : */
2278 GIC 394362 : if (!ExecQual(winstate->runcondition, econtext))
2279 : {
2280 : /*
2281 : * Determine which mode to move into. If there is no
2282 : * PARTITION BY clause and we're the top-level WindowAgg then
2283 : * we're done. This tuple and any future tuples cannot
2284 : * possibly match the runcondition. However, when there is a
2285 ECB : * PARTITION BY clause or we're not the top-level window we
2286 : * can't just stop as we need to either process other
2287 : * partitions or ensure WindowAgg nodes above us receive all
2288 : * of the tuples they need to process their WindowFuncs.
2289 : */
2290 GIC 42 : if (winstate->use_pass_through)
2291 : {
2292 : /*
2293 ECB : * STRICT pass-through mode is required for the top window
2294 : * when there is a PARTITION BY clause. Otherwise we must
2295 : * ensure we store tuples that don't match the
2296 : * runcondition so they're available to WindowAggs above.
2297 : */
2298 GIC 21 : if (winstate->top_window)
2299 : {
2300 CBC 12 : winstate->status = WINDOWAGG_PASSTHROUGH_STRICT;
2301 GIC 12 : continue;
2302 : }
2303 : else
2304 : {
2305 9 : winstate->status = WINDOWAGG_PASSTHROUGH;
2306 :
2307 : /*
2308 : * If we're not the top-window, we'd better NULLify
2309 : * the aggregate results. In pass-through mode we no
2310 : * longer update these and this avoids the old stale
2311 : * results lingering. Some of these might be byref
2312 : * types so we can't have them pointing to free'd
2313 ECB : * memory. The planner insisted that quals used in
2314 : * the runcondition are strict, so the top-level
2315 : * WindowAgg will filter these NULLs out in the filter
2316 : * clause.
2317 : */
2318 GIC 9 : numfuncs = winstate->numfuncs;
2319 36 : for (i = 0; i < numfuncs; i++)
2320 : {
2321 27 : econtext->ecxt_aggvalues[i] = (Datum) 0;
2322 27 : econtext->ecxt_aggnulls[i] = true;
2323 : }
2324 : }
2325 : }
2326 : else
2327 ECB : {
2328 : /*
2329 : * Pass-through not required. We can just return NULL.
2330 : * Nothing else will match the runcondition.
2331 : */
2332 GIC 21 : winstate->status = WINDOWAGG_DONE;
2333 21 : return NULL;
2334 : }
2335 ECB : }
2336 :
2337 : /*
2338 : * Filter out any tuples we don't need in the top-level WindowAgg.
2339 : */
2340 GIC 394329 : if (!ExecQual(winstate->ss.ps.qual, econtext))
2341 ECB : {
2342 GIC 9 : InstrCountFiltered1(winstate, 1);
2343 9 : continue;
2344 : }
2345 :
2346 394320 : break;
2347 : }
2348 ECB :
2349 : /*
2350 : * When not in WINDOWAGG_RUN mode, we must still return this tuple if
2351 : * we're anything apart from the top window.
2352 : */
2353 GIC 27 : else if (!winstate->top_window)
2354 12 : break;
2355 : }
2356 :
2357 394332 : return slot;
2358 : }
2359 :
2360 : /* -----------------
2361 : * ExecInitWindowAgg
2362 : *
2363 ECB : * Creates the run-time information for the WindowAgg node produced by the
2364 : * planner and initializes its outer subtree
2365 : * -----------------
2366 : */
2367 : WindowAggState *
2368 GIC 1089 : ExecInitWindowAgg(WindowAgg *node, EState *estate, int eflags)
2369 : {
2370 : WindowAggState *winstate;
2371 ECB : Plan *outerPlan;
2372 : ExprContext *econtext;
2373 : ExprContext *tmpcontext;
2374 : WindowStatePerFunc perfunc;
2375 : WindowStatePerAgg peragg;
2376 GIC 1089 : int frameOptions = node->frameOptions;
2377 : int numfuncs,
2378 : wfuncno,
2379 : numaggs,
2380 ECB : aggno;
2381 : TupleDesc scanDesc;
2382 : ListCell *l;
2383 :
2384 : /* check for unsupported flags */
2385 CBC 1089 : Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
2386 ECB :
2387 : /*
2388 : * create state structure
2389 : */
2390 GIC 1089 : winstate = makeNode(WindowAggState);
2391 1089 : winstate->ss.ps.plan = (Plan *) node;
2392 1089 : winstate->ss.ps.state = estate;
2393 1089 : winstate->ss.ps.ExecProcNode = ExecWindowAgg;
2394 :
2395 ECB : /*
2396 : * Create expression contexts. We need two, one for per-input-tuple
2397 : * processing and one for per-output-tuple processing. We cheat a little
2398 : * by using ExecAssignExprContext() to build both.
2399 : */
2400 GIC 1089 : ExecAssignExprContext(estate, &winstate->ss.ps);
2401 CBC 1089 : tmpcontext = winstate->ss.ps.ps_ExprContext;
2402 1089 : winstate->tmpcontext = tmpcontext;
2403 GIC 1089 : ExecAssignExprContext(estate, &winstate->ss.ps);
2404 :
2405 : /* Create long-lived context for storage of partition-local memory etc */
2406 1089 : winstate->partcontext =
2407 1089 : AllocSetContextCreate(CurrentMemoryContext,
2408 : "WindowAgg Partition",
2409 : ALLOCSET_DEFAULT_SIZES);
2410 :
2411 : /*
2412 ECB : * Create mid-lived context for aggregate trans values etc.
2413 : *
2414 : * Note that moving aggregates each use their own private context, not
2415 : * this one.
2416 : */
2417 GIC 1089 : winstate->aggcontext =
2418 CBC 1089 : AllocSetContextCreate(CurrentMemoryContext,
2419 : "WindowAgg Aggregates",
2420 : ALLOCSET_DEFAULT_SIZES);
2421 ECB :
2422 : /* Only the top-level WindowAgg may have a qual */
2423 GIC 1089 : Assert(node->plan.qual == NIL || node->topWindow);
2424 :
2425 : /* Initialize the qual */
2426 1089 : winstate->ss.ps.qual = ExecInitQual(node->plan.qual,
2427 : (PlanState *) winstate);
2428 :
2429 : /*
2430 : * Setup the run condition, if we received one from the query planner.
2431 ECB : * When set, this may allow us to move into pass-through mode so that we
2432 : * don't have to perform any further evaluation of WindowFuncs in the
2433 : * current partition or possibly stop returning tuples altogether when all
2434 : * tuples are in the same partition.
2435 : */
2436 GIC 1089 : winstate->runcondition = ExecInitQual(node->runCondition,
2437 : (PlanState *) winstate);
2438 :
2439 ECB : /*
2440 : * When we're not the top-level WindowAgg node or we are but have a
2441 : * PARTITION BY clause we must move into one of the WINDOWAGG_PASSTHROUGH*
2442 : * modes when the runCondition becomes false.
2443 : */
2444 GIC 1089 : winstate->use_pass_through = !node->topWindow || node->partNumCols > 0;
2445 :
2446 : /* remember if we're the top-window or we are below the top-window */
2447 CBC 1089 : winstate->top_window = node->topWindow;
2448 ECB :
2449 : /*
2450 : * initialize child nodes
2451 : */
2452 GIC 1089 : outerPlan = outerPlan(node);
2453 1089 : outerPlanState(winstate) = ExecInitNode(outerPlan, estate, eflags);
2454 ECB :
2455 : /*
2456 : * initialize source tuple type (which is also the tuple type that we'll
2457 : * store in the tuplestore and use in all our working slots).
2458 : */
2459 CBC 1089 : ExecCreateScanSlotFromOuterPlan(estate, &winstate->ss, &TTSOpsMinimalTuple);
2460 1089 : scanDesc = winstate->ss.ss_ScanTupleSlot->tts_tupleDescriptor;
2461 :
2462 : /* the outer tuple isn't the child's tuple, but always a minimal tuple */
2463 GIC 1089 : winstate->ss.ps.outeropsset = true;
2464 1089 : winstate->ss.ps.outerops = &TTSOpsMinimalTuple;
2465 CBC 1089 : winstate->ss.ps.outeropsfixed = true;
2466 :
2467 ECB : /*
2468 : * tuple table initialization
2469 : */
2470 GIC 1089 : winstate->first_part_slot = ExecInitExtraTupleSlot(estate, scanDesc,
2471 ECB : &TTSOpsMinimalTuple);
2472 GIC 1089 : winstate->agg_row_slot = ExecInitExtraTupleSlot(estate, scanDesc,
2473 : &TTSOpsMinimalTuple);
2474 1089 : winstate->temp_slot_1 = ExecInitExtraTupleSlot(estate, scanDesc,
2475 : &TTSOpsMinimalTuple);
2476 1089 : winstate->temp_slot_2 = ExecInitExtraTupleSlot(estate, scanDesc,
2477 : &TTSOpsMinimalTuple);
2478 :
2479 ECB : /*
2480 : * create frame head and tail slots only if needed (must create slots in
2481 : * exactly the same cases that update_frameheadpos and update_frametailpos
2482 : * need them)
2483 : */
2484 CBC 1089 : winstate->framehead_slot = winstate->frametail_slot = NULL;
2485 ECB :
2486 CBC 1089 : if (frameOptions & (FRAMEOPTION_RANGE | FRAMEOPTION_GROUPS))
2487 : {
2488 627 : if (((frameOptions & FRAMEOPTION_START_CURRENT_ROW) &&
2489 38 : node->ordNumCols != 0) ||
2490 589 : (frameOptions & FRAMEOPTION_START_OFFSET))
2491 296 : winstate->framehead_slot = ExecInitExtraTupleSlot(estate, scanDesc,
2492 : &TTSOpsMinimalTuple);
2493 GIC 627 : if (((frameOptions & FRAMEOPTION_END_CURRENT_ROW) &&
2494 313 : node->ordNumCols != 0) ||
2495 421 : (frameOptions & FRAMEOPTION_END_OFFSET))
2496 494 : winstate->frametail_slot = ExecInitExtraTupleSlot(estate, scanDesc,
2497 : &TTSOpsMinimalTuple);
2498 ECB : }
2499 :
2500 : /*
2501 : * Initialize result slot, type and projection.
2502 : */
2503 CBC 1089 : ExecInitResultTupleSlotTL(&winstate->ss.ps, &TTSOpsVirtual);
2504 1089 : ExecAssignProjectionInfo(&winstate->ss.ps, NULL);
2505 :
2506 ECB : /* Set up data for comparing tuples */
2507 CBC 1089 : if (node->partNumCols > 0)
2508 306 : winstate->partEqfunction =
2509 GIC 306 : execTuplesMatchPrepare(scanDesc,
2510 : node->partNumCols,
2511 CBC 306 : node->partColIdx,
2512 306 : node->partOperators,
2513 306 : node->partCollations,
2514 : &winstate->ss.ps);
2515 ECB :
2516 CBC 1089 : if (node->ordNumCols > 0)
2517 928 : winstate->ordEqfunction =
2518 GIC 928 : execTuplesMatchPrepare(scanDesc,
2519 : node->ordNumCols,
2520 928 : node->ordColIdx,
2521 928 : node->ordOperators,
2522 928 : node->ordCollations,
2523 ECB : &winstate->ss.ps);
2524 :
2525 : /*
2526 : * WindowAgg nodes use aggvalues and aggnulls as well as Agg nodes.
2527 : */
2528 GIC 1089 : numfuncs = winstate->numfuncs;
2529 1089 : numaggs = winstate->numaggs;
2530 1089 : econtext = winstate->ss.ps.ps_ExprContext;
2531 1089 : econtext->ecxt_aggvalues = (Datum *) palloc0(sizeof(Datum) * numfuncs);
2532 CBC 1089 : econtext->ecxt_aggnulls = (bool *) palloc0(sizeof(bool) * numfuncs);
2533 ECB :
2534 : /*
2535 : * allocate per-wfunc/per-agg state information.
2536 : */
2537 CBC 1089 : perfunc = (WindowStatePerFunc) palloc0(sizeof(WindowStatePerFuncData) * numfuncs);
2538 1089 : peragg = (WindowStatePerAgg) palloc0(sizeof(WindowStatePerAggData) * numaggs);
2539 1089 : winstate->perfunc = perfunc;
2540 GIC 1089 : winstate->peragg = peragg;
2541 ECB :
2542 CBC 1089 : wfuncno = -1;
2543 GIC 1089 : aggno = -1;
2544 2433 : foreach(l, winstate->funcs)
2545 : {
2546 1344 : WindowFuncExprState *wfuncstate = (WindowFuncExprState *) lfirst(l);
2547 CBC 1344 : WindowFunc *wfunc = wfuncstate->wfunc;
2548 EUB : WindowStatePerFunc perfuncstate;
2549 : AclResult aclresult;
2550 : int i;
2551 :
2552 CBC 1344 : if (wfunc->winref != node->winref) /* planner screwed up? */
2553 UIC 0 : elog(ERROR, "WindowFunc with winref %u assigned to WindowAgg with winref %u",
2554 ECB : wfunc->winref, node->winref);
2555 :
2556 : /* Look for a previous duplicate window function */
2557 GIC 1665 : for (i = 0; i <= wfuncno; i++)
2558 ECB : {
2559 GIC 324 : if (equal(wfunc, perfunc[i].wfunc) &&
2560 3 : !contain_volatile_functions((Node *) wfunc))
2561 CBC 3 : break;
2562 ECB : }
2563 GIC 1344 : if (i <= wfuncno)
2564 : {
2565 : /* Found a match to an existing entry, so just mark it */
2566 CBC 3 : wfuncstate->wfuncno = i;
2567 GIC 3 : continue;
2568 : }
2569 ECB :
2570 : /* Nope, so assign a new PerAgg record */
2571 GIC 1341 : perfuncstate = &perfunc[++wfuncno];
2572 ECB :
2573 : /* Mark WindowFunc state node with assigned index in the result array */
2574 CBC 1341 : wfuncstate->wfuncno = wfuncno;
2575 EUB :
2576 : /* Check permission to call window function */
2577 GNC 1341 : aclresult = object_aclcheck(ProcedureRelationId, wfunc->winfnoid, GetUserId(),
2578 : ACL_EXECUTE);
2579 GIC 1341 : if (aclresult != ACLCHECK_OK)
2580 LBC 0 : aclcheck_error(aclresult, OBJECT_FUNCTION,
2581 0 : get_func_name(wfunc->winfnoid));
2582 CBC 1341 : InvokeFunctionExecuteHook(wfunc->winfnoid);
2583 ECB :
2584 : /* Fill in the perfuncstate data */
2585 CBC 1341 : perfuncstate->wfuncstate = wfuncstate;
2586 GIC 1341 : perfuncstate->wfunc = wfunc;
2587 1341 : perfuncstate->numArguments = list_length(wfuncstate->args);
2588 1341 : perfuncstate->winCollation = wfunc->inputcollid;
2589 :
2590 1341 : get_typlenbyval(wfunc->wintype,
2591 : &perfuncstate->resulttypeLen,
2592 : &perfuncstate->resulttypeByVal);
2593 ECB :
2594 : /*
2595 : * If it's really just a plain aggregate function, we'll emulate the
2596 : * Agg environment for it.
2597 : */
2598 CBC 1341 : perfuncstate->plain_agg = wfunc->winagg;
2599 1341 : if (wfunc->winagg)
2600 ECB : {
2601 : WindowStatePerAgg peraggstate;
2602 :
2603 GIC 669 : perfuncstate->aggno = ++aggno;
2604 669 : peraggstate = &winstate->peragg[aggno];
2605 CBC 669 : initialize_peragg(winstate, wfunc, peraggstate);
2606 GIC 669 : peraggstate->wfuncno = wfuncno;
2607 ECB : }
2608 : else
2609 : {
2610 CBC 672 : WindowObject winobj = makeNode(WindowObjectData);
2611 :
2612 GIC 672 : winobj->winstate = winstate;
2613 CBC 672 : winobj->argstates = wfuncstate->args;
2614 GIC 672 : winobj->localmem = NULL;
2615 CBC 672 : perfuncstate->winobj = winobj;
2616 :
2617 : /* It's a real window function, so set up to call it. */
2618 GIC 672 : fmgr_info_cxt(wfunc->winfnoid, &perfuncstate->flinfo,
2619 : econtext->ecxt_per_query_memory);
2620 CBC 672 : fmgr_info_set_expr((Node *) wfunc, &perfuncstate->flinfo);
2621 ECB : }
2622 : }
2623 :
2624 : /* Update numfuncs, numaggs to match number of unique functions found */
2625 GIC 1089 : winstate->numfuncs = wfuncno + 1;
2626 CBC 1089 : winstate->numaggs = aggno + 1;
2627 :
2628 ECB : /* Set up WindowObject for aggregates, if needed */
2629 CBC 1089 : if (winstate->numaggs > 0)
2630 ECB : {
2631 GIC 633 : WindowObject agg_winobj = makeNode(WindowObjectData);
2632 ECB :
2633 CBC 633 : agg_winobj->winstate = winstate;
2634 633 : agg_winobj->argstates = NIL;
2635 GIC 633 : agg_winobj->localmem = NULL;
2636 : /* make sure markptr = -1 to invalidate. It may not get used */
2637 633 : agg_winobj->markptr = -1;
2638 CBC 633 : agg_winobj->readptr = -1;
2639 GIC 633 : winstate->agg_winobj = agg_winobj;
2640 : }
2641 ECB :
2642 : /* Set the status to running */
2643 GIC 1089 : winstate->status = WINDOWAGG_RUN;
2644 ECB :
2645 : /* copy frame options to state node for easy access */
2646 CBC 1089 : winstate->frameOptions = frameOptions;
2647 :
2648 : /* initialize frame bound offset expressions */
2649 GIC 1089 : winstate->startOffset = ExecInitExpr((Expr *) node->startOffset,
2650 ECB : (PlanState *) winstate);
2651 CBC 1089 : winstate->endOffset = ExecInitExpr((Expr *) node->endOffset,
2652 ECB : (PlanState *) winstate);
2653 :
2654 : /* Lookup in_range support functions if needed */
2655 CBC 1089 : if (OidIsValid(node->startInRangeFunc))
2656 186 : fmgr_info(node->startInRangeFunc, &winstate->startInRangeFunc);
2657 GIC 1089 : if (OidIsValid(node->endInRangeFunc))
2658 CBC 219 : fmgr_info(node->endInRangeFunc, &winstate->endInRangeFunc);
2659 1089 : winstate->inRangeColl = node->inRangeColl;
2660 1089 : winstate->inRangeAsc = node->inRangeAsc;
2661 GIC 1089 : winstate->inRangeNullsFirst = node->inRangeNullsFirst;
2662 ECB :
2663 GIC 1089 : winstate->all_first = true;
2664 1089 : winstate->partition_spooled = false;
2665 1089 : winstate->more_partitions = false;
2666 :
2667 1089 : return winstate;
2668 : }
2669 :
2670 ECB : /* -----------------
2671 : * ExecEndWindowAgg
2672 : * -----------------
2673 : */
2674 : void
2675 CBC 1065 : ExecEndWindowAgg(WindowAggState *node)
2676 : {
2677 ECB : PlanState *outerPlan;
2678 : int i;
2679 :
2680 CBC 1065 : release_partition(node);
2681 ECB :
2682 CBC 1065 : ExecClearTuple(node->ss.ss_ScanTupleSlot);
2683 1065 : ExecClearTuple(node->first_part_slot);
2684 1065 : ExecClearTuple(node->agg_row_slot);
2685 1065 : ExecClearTuple(node->temp_slot_1);
2686 GIC 1065 : ExecClearTuple(node->temp_slot_2);
2687 1065 : if (node->framehead_slot)
2688 278 : ExecClearTuple(node->framehead_slot);
2689 1065 : if (node->frametail_slot)
2690 CBC 473 : ExecClearTuple(node->frametail_slot);
2691 ECB :
2692 : /*
2693 : * Free both the expr contexts.
2694 : */
2695 GIC 1065 : ExecFreeExprContext(&node->ss.ps);
2696 CBC 1065 : node->ss.ps.ps_ExprContext = node->tmpcontext;
2697 1065 : ExecFreeExprContext(&node->ss.ps);
2698 :
2699 1725 : for (i = 0; i < node->numaggs; i++)
2700 ECB : {
2701 GIC 660 : if (node->peragg[i].aggcontext != node->aggcontext)
2702 CBC 584 : MemoryContextDelete(node->peragg[i].aggcontext);
2703 ECB : }
2704 GIC 1065 : MemoryContextDelete(node->partcontext);
2705 CBC 1065 : MemoryContextDelete(node->aggcontext);
2706 ECB :
2707 CBC 1065 : pfree(node->perfunc);
2708 GIC 1065 : pfree(node->peragg);
2709 :
2710 1065 : outerPlan = outerPlanState(node);
2711 1065 : ExecEndNode(outerPlan);
2712 1065 : }
2713 :
2714 ECB : /* -----------------
2715 : * ExecReScanWindowAgg
2716 : * -----------------
2717 : */
2718 : void
2719 CBC 39 : ExecReScanWindowAgg(WindowAggState *node)
2720 ECB : {
2721 GIC 39 : PlanState *outerPlan = outerPlanState(node);
2722 39 : ExprContext *econtext = node->ss.ps.ps_ExprContext;
2723 ECB :
2724 GIC 39 : node->status = WINDOWAGG_RUN;
2725 39 : node->all_first = true;
2726 ECB :
2727 : /* release tuplestore et al */
2728 CBC 39 : release_partition(node);
2729 ECB :
2730 : /* release all temp tuples, but especially first_part_slot */
2731 CBC 39 : ExecClearTuple(node->ss.ss_ScanTupleSlot);
2732 GBC 39 : ExecClearTuple(node->first_part_slot);
2733 CBC 39 : ExecClearTuple(node->agg_row_slot);
2734 39 : ExecClearTuple(node->temp_slot_1);
2735 GIC 39 : ExecClearTuple(node->temp_slot_2);
2736 39 : if (node->framehead_slot)
2737 LBC 0 : ExecClearTuple(node->framehead_slot);
2738 CBC 39 : if (node->frametail_slot)
2739 GIC 3 : ExecClearTuple(node->frametail_slot);
2740 :
2741 : /* Forget current wfunc values */
2742 78 : MemSet(econtext->ecxt_aggvalues, 0, sizeof(Datum) * node->numfuncs);
2743 39 : MemSet(econtext->ecxt_aggnulls, 0, sizeof(bool) * node->numfuncs);
2744 ECB :
2745 : /*
2746 : * if chgParam of subnode is not null then plan will be re-scanned by
2747 : * first ExecProcNode.
2748 : */
2749 GIC 39 : if (outerPlan->chgParam == NULL)
2750 3 : ExecReScan(outerPlan);
2751 39 : }
2752 :
2753 : /*
2754 ECB : * initialize_peragg
2755 : *
2756 : * Almost same as in nodeAgg.c, except we don't support DISTINCT currently.
2757 : */
2758 : static WindowStatePerAggData *
2759 GIC 669 : initialize_peragg(WindowAggState *winstate, WindowFunc *wfunc,
2760 : WindowStatePerAgg peraggstate)
2761 : {
2762 : Oid inputTypes[FUNC_MAX_ARGS];
2763 : int numArguments;
2764 : HeapTuple aggTuple;
2765 : Form_pg_aggregate aggform;
2766 : Oid aggtranstype;
2767 : AttrNumber initvalAttNo;
2768 : AclResult aclresult;
2769 : bool use_ma_code;
2770 : Oid transfn_oid,
2771 : invtransfn_oid,
2772 : finalfn_oid;
2773 : bool finalextra;
2774 : char finalmodify;
2775 : Expr *transfnexpr,
2776 : *invtransfnexpr,
2777 ECB : *finalfnexpr;
2778 : Datum textInitVal;
2779 : int i;
2780 : ListCell *lc;
2781 :
2782 CBC 669 : numArguments = list_length(wfunc->args);
2783 :
2784 GIC 669 : i = 0;
2785 CBC 1272 : foreach(lc, wfunc->args)
2786 ECB : {
2787 GBC 603 : inputTypes[i++] = exprType((Node *) lfirst(lc));
2788 : }
2789 ECB :
2790 GIC 669 : aggTuple = SearchSysCache1(AGGFNOID, ObjectIdGetDatum(wfunc->winfnoid));
2791 669 : if (!HeapTupleIsValid(aggTuple))
2792 UIC 0 : elog(ERROR, "cache lookup failed for aggregate %u",
2793 : wfunc->winfnoid);
2794 GIC 669 : aggform = (Form_pg_aggregate) GETSTRUCT(aggTuple);
2795 :
2796 : /*
2797 : * Figure out whether we want to use the moving-aggregate implementation,
2798 : * and collect the right set of fields from the pg_attribute entry.
2799 : *
2800 : * It's possible that an aggregate would supply a safe moving-aggregate
2801 : * implementation and an unsafe normal one, in which case our hand is
2802 : * forced. Otherwise, if the frame head can't move, we don't need
2803 : * moving-aggregate code. Even if we'd like to use it, don't do so if the
2804 : * aggregate's arguments (and FILTER clause if any) contain any calls to
2805 : * volatile functions. Otherwise, the difference between restarting and
2806 : * not restarting the aggregation would be user-visible.
2807 : *
2808 : * We also don't risk using moving aggregates when there are subplans in
2809 ECB : * the arguments or FILTER clause. This is partly because
2810 : * contain_volatile_functions() doesn't look inside subplans; but there
2811 : * are other reasons why a subplan's output might be volatile. For
2812 : * example, syncscan mode can render the results nonrepeatable.
2813 EUB : */
2814 CBC 669 : if (!OidIsValid(aggform->aggminvtransfn))
2815 GBC 76 : use_ma_code = false; /* sine qua non */
2816 CBC 593 : else if (aggform->aggmfinalmodify == AGGMODIFY_READ_ONLY &&
2817 593 : aggform->aggfinalmodify != AGGMODIFY_READ_ONLY)
2818 LBC 0 : use_ma_code = true; /* decision forced by safety */
2819 GBC 593 : else if (winstate->frameOptions & FRAMEOPTION_START_UNBOUNDED_PRECEDING)
2820 UIC 0 : use_ma_code = false; /* non-moving frame head */
2821 CBC 593 : else if (contain_volatile_functions((Node *) wfunc))
2822 9 : use_ma_code = false; /* avoid possible behavioral change */
2823 GIC 584 : else if (contain_subplans((Node *) wfunc))
2824 LBC 0 : use_ma_code = false; /* subplans might contain volatile functions */
2825 ECB : else
2826 CBC 584 : use_ma_code = true; /* yes, let's use it */
2827 669 : if (use_ma_code)
2828 ECB : {
2829 CBC 584 : peraggstate->transfn_oid = transfn_oid = aggform->aggmtransfn;
2830 584 : peraggstate->invtransfn_oid = invtransfn_oid = aggform->aggminvtransfn;
2831 GIC 584 : peraggstate->finalfn_oid = finalfn_oid = aggform->aggmfinalfn;
2832 584 : finalextra = aggform->aggmfinalextra;
2833 584 : finalmodify = aggform->aggmfinalmodify;
2834 CBC 584 : aggtranstype = aggform->aggmtranstype;
2835 584 : initvalAttNo = Anum_pg_aggregate_aggminitval;
2836 ECB : }
2837 : else
2838 : {
2839 CBC 85 : peraggstate->transfn_oid = transfn_oid = aggform->aggtransfn;
2840 85 : peraggstate->invtransfn_oid = invtransfn_oid = InvalidOid;
2841 GIC 85 : peraggstate->finalfn_oid = finalfn_oid = aggform->aggfinalfn;
2842 85 : finalextra = aggform->aggfinalextra;
2843 85 : finalmodify = aggform->aggfinalmodify;
2844 85 : aggtranstype = aggform->aggtranstype;
2845 85 : initvalAttNo = Anum_pg_aggregate_agginitval;
2846 : }
2847 :
2848 : /*
2849 : * ExecInitWindowAgg already checked permission to call aggregate function
2850 : * ... but we still need to check the component functions
2851 : */
2852 :
2853 ECB : /* Check that aggregate owner has permission to call component fns */
2854 : {
2855 : HeapTuple procTuple;
2856 EUB : Oid aggOwner;
2857 :
2858 CBC 669 : procTuple = SearchSysCache1(PROCOID,
2859 ECB : ObjectIdGetDatum(wfunc->winfnoid));
2860 GIC 669 : if (!HeapTupleIsValid(procTuple))
2861 LBC 0 : elog(ERROR, "cache lookup failed for function %u",
2862 : wfunc->winfnoid);
2863 CBC 669 : aggOwner = ((Form_pg_proc) GETSTRUCT(procTuple))->proowner;
2864 GBC 669 : ReleaseSysCache(procTuple);
2865 EUB :
2866 GNC 669 : aclresult = object_aclcheck(ProcedureRelationId, transfn_oid, aggOwner,
2867 : ACL_EXECUTE);
2868 CBC 669 : if (aclresult != ACLCHECK_OK)
2869 UIC 0 : aclcheck_error(aclresult, OBJECT_FUNCTION,
2870 LBC 0 : get_func_name(transfn_oid));
2871 GIC 669 : InvokeFunctionExecuteHook(transfn_oid);
2872 ECB :
2873 GBC 669 : if (OidIsValid(invtransfn_oid))
2874 EUB : {
2875 GNC 584 : aclresult = object_aclcheck(ProcedureRelationId, invtransfn_oid, aggOwner,
2876 : ACL_EXECUTE);
2877 GIC 584 : if (aclresult != ACLCHECK_OK)
2878 LBC 0 : aclcheck_error(aclresult, OBJECT_FUNCTION,
2879 UIC 0 : get_func_name(invtransfn_oid));
2880 CBC 584 : InvokeFunctionExecuteHook(invtransfn_oid);
2881 : }
2882 ECB :
2883 GBC 669 : if (OidIsValid(finalfn_oid))
2884 EUB : {
2885 GNC 469 : aclresult = object_aclcheck(ProcedureRelationId, finalfn_oid, aggOwner,
2886 : ACL_EXECUTE);
2887 GIC 469 : if (aclresult != ACLCHECK_OK)
2888 UIC 0 : aclcheck_error(aclresult, OBJECT_FUNCTION,
2889 0 : get_func_name(finalfn_oid));
2890 GIC 469 : InvokeFunctionExecuteHook(finalfn_oid);
2891 : }
2892 : }
2893 :
2894 ECB : /*
2895 EUB : * If the selected finalfn isn't read-only, we can't run this aggregate as
2896 : * a window function. This is a user-facing error, so we take a bit more
2897 : * care with the error message than elsewhere in this function.
2898 : */
2899 GIC 669 : if (finalmodify != AGGMODIFY_READ_ONLY)
2900 UIC 0 : ereport(ERROR,
2901 ECB : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2902 : errmsg("aggregate function %s does not support use as a window function",
2903 : format_procedure(wfunc->winfnoid))));
2904 :
2905 : /* Detect how many arguments to pass to the finalfn */
2906 GIC 669 : if (finalextra)
2907 CBC 13 : peraggstate->numFinalArgs = numArguments + 1;
2908 : else
2909 GIC 656 : peraggstate->numFinalArgs = 1;
2910 :
2911 : /* resolve actual type of transition state, if polymorphic */
2912 669 : aggtranstype = resolve_aggregate_transtype(wfunc->winfnoid,
2913 ECB : aggtranstype,
2914 : inputTypes,
2915 : numArguments);
2916 :
2917 : /* build expression trees using actual argument & result types */
2918 GIC 669 : build_aggregate_transfn_expr(inputTypes,
2919 : numArguments,
2920 : 0, /* no ordered-set window functions yet */
2921 : false, /* no variadic window functions yet */
2922 : aggtranstype,
2923 : wfunc->inputcollid,
2924 : transfn_oid,
2925 ECB : invtransfn_oid,
2926 : &transfnexpr,
2927 : &invtransfnexpr);
2928 :
2929 : /* set up infrastructure for calling the transfn(s) and finalfn */
2930 CBC 669 : fmgr_info(transfn_oid, &peraggstate->transfn);
2931 669 : fmgr_info_set_expr((Node *) transfnexpr, &peraggstate->transfn);
2932 :
2933 GIC 669 : if (OidIsValid(invtransfn_oid))
2934 ECB : {
2935 GIC 584 : fmgr_info(invtransfn_oid, &peraggstate->invtransfn);
2936 CBC 584 : fmgr_info_set_expr((Node *) invtransfnexpr, &peraggstate->invtransfn);
2937 : }
2938 :
2939 GIC 669 : if (OidIsValid(finalfn_oid))
2940 : {
2941 469 : build_aggregate_finalfn_expr(inputTypes,
2942 : peraggstate->numFinalArgs,
2943 ECB : aggtranstype,
2944 : wfunc->wintype,
2945 : wfunc->inputcollid,
2946 : finalfn_oid,
2947 : &finalfnexpr);
2948 CBC 469 : fmgr_info(finalfn_oid, &peraggstate->finalfn);
2949 GIC 469 : fmgr_info_set_expr((Node *) finalfnexpr, &peraggstate->finalfn);
2950 : }
2951 ECB :
2952 : /* get info about relevant datatypes */
2953 GIC 669 : get_typlenbyval(wfunc->wintype,
2954 : &peraggstate->resulttypeLen,
2955 : &peraggstate->resulttypeByVal);
2956 669 : get_typlenbyval(aggtranstype,
2957 : &peraggstate->transtypeLen,
2958 : &peraggstate->transtypeByVal);
2959 ECB :
2960 : /*
2961 : * initval is potentially null, so don't try to access it as a struct
2962 : * field. Must do it the hard way with SysCacheGetAttr.
2963 : */
2964 GIC 669 : textInitVal = SysCacheGetAttr(AGGFNOID, aggTuple, initvalAttNo,
2965 ECB : &peraggstate->initValueIsNull);
2966 :
2967 GIC 669 : if (peraggstate->initValueIsNull)
2968 244 : peraggstate->initValue = (Datum) 0;
2969 : else
2970 425 : peraggstate->initValue = GetAggInitVal(textInitVal,
2971 : aggtranstype);
2972 :
2973 : /*
2974 : * If the transfn is strict and the initval is NULL, make sure input type
2975 ECB : * and transtype are the same (or at least binary-compatible), so that
2976 : * it's OK to use the first input value as the initial transValue. This
2977 : * should have been checked at agg definition time, but we must check
2978 : * again in case the transfn's strictness property has been changed.
2979 EUB : */
2980 GIC 669 : if (peraggstate->transfn.fn_strict && peraggstate->initValueIsNull)
2981 : {
2982 78 : if (numArguments < 1 ||
2983 78 : !IsBinaryCoercible(inputTypes[0], aggtranstype))
2984 UIC 0 : ereport(ERROR,
2985 : (errcode(ERRCODE_INVALID_FUNCTION_DEFINITION),
2986 : errmsg("aggregate %u needs to have compatible input type and transition type",
2987 : wfunc->winfnoid)));
2988 : }
2989 :
2990 : /*
2991 : * Insist that forward and inverse transition functions have the same
2992 : * strictness setting. Allowing them to differ would require handling
2993 ECB : * more special cases in advance_windowaggregate and
2994 : * advance_windowaggregate_base, for no discernible benefit. This should
2995 EUB : * have been checked at agg definition time, but we must check again in
2996 : * case either function's strictness property has been changed.
2997 : */
2998 GIC 669 : if (OidIsValid(invtransfn_oid) &&
2999 584 : peraggstate->transfn.fn_strict != peraggstate->invtransfn.fn_strict)
3000 UIC 0 : ereport(ERROR,
3001 : (errcode(ERRCODE_INVALID_FUNCTION_DEFINITION),
3002 : errmsg("strictness of aggregate's forward and inverse transition functions must match")));
3003 :
3004 : /*
3005 : * Moving aggregates use their own aggcontext.
3006 : *
3007 : * This is necessary because they might restart at different times, so we
3008 : * might never be able to reset the shared context otherwise. We can't
3009 : * make it the aggregates' responsibility to clean up after themselves,
3010 : * because strict aggregates must be restarted whenever we remove their
3011 : * last non-NULL input, which the aggregate won't be aware is happening.
3012 : * Also, just pfree()ing the transValue upon restarting wouldn't help,
3013 ECB : * since we'd miss any indirectly referenced data. We could, in theory,
3014 : * make the memory allocation rules for moving aggregates different than
3015 : * they have historically been for plain aggregates, but that seems grotty
3016 : * and likely to lead to memory leaks.
3017 : */
3018 GIC 669 : if (OidIsValid(invtransfn_oid))
3019 CBC 584 : peraggstate->aggcontext =
3020 GIC 584 : AllocSetContextCreate(CurrentMemoryContext,
3021 ECB : "WindowAgg Per Aggregate",
3022 : ALLOCSET_DEFAULT_SIZES);
3023 : else
3024 GIC 85 : peraggstate->aggcontext = winstate->aggcontext;
3025 :
3026 669 : ReleaseSysCache(aggTuple);
3027 ECB :
3028 GIC 669 : return peraggstate;
3029 : }
3030 :
3031 : static Datum
3032 425 : GetAggInitVal(Datum textInitVal, Oid transtype)
3033 : {
3034 ECB : Oid typinput,
3035 : typioparam;
3036 : char *strInitVal;
3037 : Datum initVal;
3038 :
3039 CBC 425 : getTypeInputInfo(transtype, &typinput, &typioparam);
3040 GIC 425 : strInitVal = TextDatumGetCString(textInitVal);
3041 425 : initVal = OidInputFunctionCall(typinput, strInitVal,
3042 : typioparam, -1);
3043 425 : pfree(strInitVal);
3044 425 : return initVal;
3045 : }
3046 :
3047 : /*
3048 : * are_peers
3049 ECB : * compare two rows to see if they are equal according to the ORDER BY clause
3050 : *
3051 : * NB: this does not consider the window frame mode.
3052 : */
3053 : static bool
3054 GIC 256728 : are_peers(WindowAggState *winstate, TupleTableSlot *slot1,
3055 : TupleTableSlot *slot2)
3056 ECB : {
3057 CBC 256728 : WindowAgg *node = (WindowAgg *) winstate->ss.ps.plan;
3058 GIC 256728 : ExprContext *econtext = winstate->tmpcontext;
3059 ECB :
3060 : /* If no ORDER BY, all rows are peers with each other */
3061 CBC 256728 : if (node->ordNumCols == 0)
3062 GIC 479 : return true;
3063 :
3064 256249 : econtext->ecxt_outertuple = slot1;
3065 256249 : econtext->ecxt_innertuple = slot2;
3066 256249 : return ExecQualAndReset(winstate->ordEqfunction, econtext);
3067 : }
3068 :
3069 : /*
3070 : * window_gettupleslot
3071 : * Fetch the pos'th tuple of the current partition into the slot,
3072 ECB : * using the winobj's read pointer
3073 : *
3074 : * Returns true if successful, false if no such row
3075 : */
3076 : static bool
3077 GIC 337020 : window_gettupleslot(WindowObject winobj, int64 pos, TupleTableSlot *slot)
3078 ECB : {
3079 GIC 337020 : WindowAggState *winstate = winobj->winstate;
3080 : MemoryContext oldcontext;
3081 ECB :
3082 : /* often called repeatedly in a row */
3083 GIC 337020 : CHECK_FOR_INTERRUPTS();
3084 :
3085 ECB : /* Don't allow passing -1 to spool_tuples here */
3086 GIC 337020 : if (pos < 0)
3087 CBC 84 : return false;
3088 ECB :
3089 : /* If necessary, fetch the tuple into the spool */
3090 CBC 336936 : spool_tuples(winstate, pos);
3091 EUB :
3092 GIC 336936 : if (pos >= winstate->spooled_rows)
3093 CBC 2153 : return false;
3094 :
3095 334783 : if (pos < winobj->markpos)
3096 UIC 0 : elog(ERROR, "cannot fetch row before WindowObject's mark position");
3097 :
3098 GIC 334783 : oldcontext = MemoryContextSwitchTo(winstate->ss.ps.ps_ExprContext->ecxt_per_query_memory);
3099 :
3100 CBC 334783 : tuplestore_select_read_pointer(winstate->buffer, winobj->readptr);
3101 :
3102 ECB : /*
3103 : * Advance or rewind until we are within one tuple of the one we want.
3104 : */
3105 GBC 334783 : if (winobj->seekpos < pos - 1)
3106 ECB : {
3107 GIC 1008 : if (!tuplestore_skiptuples(winstate->buffer,
3108 CBC 1008 : pos - 1 - winobj->seekpos,
3109 : true))
3110 LBC 0 : elog(ERROR, "unexpected end of tuplestore");
3111 CBC 1008 : winobj->seekpos = pos - 1;
3112 : }
3113 GBC 333775 : else if (winobj->seekpos > pos + 1)
3114 ECB : {
3115 GIC 1306 : if (!tuplestore_skiptuples(winstate->buffer,
3116 CBC 1306 : winobj->seekpos - (pos + 1),
3117 : false))
3118 UIC 0 : elog(ERROR, "unexpected end of tuplestore");
3119 GIC 1306 : winobj->seekpos = pos + 1;
3120 : }
3121 332469 : else if (winobj->seekpos == pos)
3122 : {
3123 : /*
3124 : * There's no API to refetch the tuple at the current position. We
3125 ECB : * have to move one tuple forward, and then one backward. (We don't
3126 : * do it the other way because we might try to fetch the row before
3127 : * our mark, which isn't allowed.) XXX this case could stand to be
3128 : * optimized.
3129 : */
3130 GIC 85271 : tuplestore_advance(winstate->buffer, true);
3131 85271 : winobj->seekpos++;
3132 : }
3133 :
3134 : /*
3135 : * Now we should be on the tuple immediately before or after the one we
3136 : * want, so just fetch forwards or backwards as appropriate.
3137 : *
3138 : * Notice that we tell tuplestore_gettupleslot to make a physical copy of
3139 : * the fetched tuple. This ensures that the slot's contents remain valid
3140 : * through manipulations of the tuplestore, which some callers depend on.
3141 ECB : */
3142 GIC 334783 : if (winobj->seekpos > pos)
3143 ECB : {
3144 GBC 86644 : if (!tuplestore_gettupleslot(winstate->buffer, false, true, slot))
3145 LBC 0 : elog(ERROR, "unexpected end of tuplestore");
3146 GIC 86644 : winobj->seekpos--;
3147 : }
3148 : else
3149 ECB : {
3150 GBC 248139 : if (!tuplestore_gettupleslot(winstate->buffer, true, true, slot))
3151 LBC 0 : elog(ERROR, "unexpected end of tuplestore");
3152 GIC 248139 : winobj->seekpos++;
3153 : }
3154 ECB :
3155 GIC 334783 : Assert(winobj->seekpos == pos);
3156 ECB :
3157 GIC 334783 : MemoryContextSwitchTo(oldcontext);
3158 ECB :
3159 GIC 334783 : return true;
3160 : }
3161 :
3162 :
3163 : /***********************************************************************
3164 : * API exposed to window functions
3165 : ***********************************************************************/
3166 :
3167 :
3168 : /*
3169 : * WinGetPartitionLocalMemory
3170 : * Get working memory that lives till end of partition processing
3171 : *
3172 : * On first call within a given partition, this allocates and zeroes the
3173 : * requested amount of space. Subsequent calls just return the same chunk.
3174 : *
3175 : * Memory obtained this way is normally used to hold state that should be
3176 : * automatically reset for each new partition. If a window function wants
3177 : * to hold state across the whole query, fcinfo->fn_extra can be used in the
3178 : * usual way for that.
3179 : */
3180 ECB : void *
3181 GIC 165735 : WinGetPartitionLocalMemory(WindowObject winobj, Size sz)
3182 ECB : {
3183 CBC 165735 : Assert(WindowObjectIsValid(winobj));
3184 165735 : if (winobj->localmem == NULL)
3185 222 : winobj->localmem =
3186 222 : MemoryContextAllocZero(winobj->winstate->partcontext, sz);
3187 GIC 165735 : return winobj->localmem;
3188 : }
3189 :
3190 : /*
3191 : * WinGetCurrentPosition
3192 : * Return the current row's position (counting from 0) within the current
3193 : * partition.
3194 : */
3195 ECB : int64
3196 GIC 363789 : WinGetCurrentPosition(WindowObject winobj)
3197 ECB : {
3198 CBC 363789 : Assert(WindowObjectIsValid(winobj));
3199 GIC 363789 : return winobj->winstate->currentpos;
3200 : }
3201 :
3202 : /*
3203 : * WinGetPartitionRowCount
3204 : * Return total number of rows contained in the current partition.
3205 : *
3206 : * Note: this is a relatively expensive operation because it forces the
3207 : * whole partition to be "spooled" into the tuplestore at once. Once
3208 : * executed, however, additional calls within the same partition are cheap.
3209 : */
3210 ECB : int64
3211 GIC 81 : WinGetPartitionRowCount(WindowObject winobj)
3212 ECB : {
3213 CBC 81 : Assert(WindowObjectIsValid(winobj));
3214 81 : spool_tuples(winobj->winstate, -1);
3215 GIC 81 : return winobj->winstate->spooled_rows;
3216 : }
3217 :
3218 : /*
3219 : * WinSetMarkPosition
3220 : * Set the "mark" position for the window object, which is the oldest row
3221 : * number (counting from 0) it is allowed to fetch during all subsequent
3222 : * operations within the current partition.
3223 : *
3224 : * Window functions do not have to call this, but are encouraged to move the
3225 : * mark forward when possible to keep the tuplestore size down and prevent
3226 : * having to spill rows to disk.
3227 : */
3228 ECB : void
3229 GIC 393817 : WinSetMarkPosition(WindowObject winobj, int64 markpos)
3230 : {
3231 : WindowAggState *winstate;
3232 ECB :
3233 CBC 393817 : Assert(WindowObjectIsValid(winobj));
3234 GIC 393817 : winstate = winobj->winstate;
3235 ECB :
3236 GBC 393817 : if (markpos < winobj->markpos)
3237 LBC 0 : elog(ERROR, "cannot move WindowObject's mark position backward");
3238 CBC 393817 : tuplestore_select_read_pointer(winstate->buffer, winobj->markptr);
3239 GIC 393817 : if (markpos > winobj->markpos)
3240 ECB : {
3241 CBC 391759 : tuplestore_skiptuples(winstate->buffer,
3242 GIC 391759 : markpos - winobj->markpos,
3243 ECB : true);
3244 GIC 391759 : winobj->markpos = markpos;
3245 ECB : }
3246 CBC 393817 : tuplestore_select_read_pointer(winstate->buffer, winobj->readptr);
3247 GIC 393817 : if (markpos > winobj->seekpos)
3248 ECB : {
3249 CBC 215984 : tuplestore_skiptuples(winstate->buffer,
3250 GIC 215984 : markpos - winobj->seekpos,
3251 ECB : true);
3252 GIC 215984 : winobj->seekpos = markpos;
3253 ECB : }
3254 GIC 393817 : }
3255 :
3256 : /*
3257 : * WinRowsArePeers
3258 : * Compare two rows (specified by absolute position in partition) to see
3259 : * if they are equal according to the ORDER BY clause.
3260 : *
3261 : * NB: this does not consider the window frame mode.
3262 : */
3263 ECB : bool
3264 GIC 82653 : WinRowsArePeers(WindowObject winobj, int64 pos1, int64 pos2)
3265 : {
3266 : WindowAggState *winstate;
3267 : WindowAgg *node;
3268 : TupleTableSlot *slot1;
3269 : TupleTableSlot *slot2;
3270 : bool res;
3271 ECB :
3272 CBC 82653 : Assert(WindowObjectIsValid(winobj));
3273 82653 : winstate = winobj->winstate;
3274 GIC 82653 : node = (WindowAgg *) winstate->ss.ps.plan;
3275 :
3276 ECB : /* If no ORDER BY, all rows are peers; don't bother to fetch them */
3277 GBC 82653 : if (node->ordNumCols == 0)
3278 UIC 0 : return true;
3279 :
3280 : /*
3281 : * Note: OK to use temp_slot_2 here because we aren't calling any
3282 : * frame-related functions (those tend to clobber temp_slot_2).
3283 ECB : */
3284 CBC 82653 : slot1 = winstate->temp_slot_1;
3285 GIC 82653 : slot2 = winstate->temp_slot_2;
3286 ECB :
3287 GBC 82653 : if (!window_gettupleslot(winobj, pos1, slot1))
3288 UIC 0 : elog(ERROR, "specified position is out of window: " INT64_FORMAT,
3289 ECB : pos1);
3290 GBC 82653 : if (!window_gettupleslot(winobj, pos2, slot2))
3291 UIC 0 : elog(ERROR, "specified position is out of window: " INT64_FORMAT,
3292 : pos2);
3293 ECB :
3294 GIC 82653 : res = are_peers(winstate, slot1, slot2);
3295 ECB :
3296 CBC 82653 : ExecClearTuple(slot1);
3297 GIC 82653 : ExecClearTuple(slot2);
3298 ECB :
3299 GIC 82653 : return res;
3300 : }
3301 :
3302 : /*
3303 : * WinGetFuncArgInPartition
3304 : * Evaluate a window function's argument expression on a specified
3305 : * row of the partition. The row is identified in lseek(2) style,
3306 : * i.e. relative to the current, first, or last row.
3307 : *
3308 : * argno: argument number to evaluate (counted from 0)
3309 : * relpos: signed rowcount offset from the seek position
3310 : * seektype: WINDOW_SEEK_CURRENT, WINDOW_SEEK_HEAD, or WINDOW_SEEK_TAIL
3311 : * set_mark: If the row is found and set_mark is true, the mark is moved to
3312 : * the row as a side-effect.
3313 : * isnull: output argument, receives isnull status of result
3314 : * isout: output argument, set to indicate whether target row position
3315 : * is out of partition (can pass NULL if caller doesn't care about this)
3316 : *
3317 : * Specifying a nonexistent row is not an error, it just causes a null result
3318 : * (plus setting *isout true, if isout isn't NULL).
3319 : */
3320 ECB : Datum
3321 GIC 91389 : WinGetFuncArgInPartition(WindowObject winobj, int argno,
3322 : int relpos, int seektype, bool set_mark,
3323 : bool *isnull, bool *isout)
3324 : {
3325 : WindowAggState *winstate;
3326 : ExprContext *econtext;
3327 : TupleTableSlot *slot;
3328 : bool gottuple;
3329 : int64 abs_pos;
3330 ECB :
3331 CBC 91389 : Assert(WindowObjectIsValid(winobj));
3332 91389 : winstate = winobj->winstate;
3333 91389 : econtext = winstate->ss.ps.ps_ExprContext;
3334 GIC 91389 : slot = winstate->temp_slot_1;
3335 ECB :
3336 GIC 91389 : switch (seektype)
3337 ECB : {
3338 CBC 91389 : case WINDOW_SEEK_CURRENT:
3339 91389 : abs_pos = winstate->currentpos + relpos;
3340 GBC 91389 : break;
3341 UBC 0 : case WINDOW_SEEK_HEAD:
3342 0 : abs_pos = relpos;
3343 0 : break;
3344 0 : case WINDOW_SEEK_TAIL:
3345 0 : spool_tuples(winstate, -1);
3346 0 : abs_pos = winstate->spooled_rows - 1 + relpos;
3347 0 : break;
3348 0 : default:
3349 UIC 0 : elog(ERROR, "unrecognized window seek type: %d", seektype);
3350 : abs_pos = 0; /* keep compiler quiet */
3351 : break;
3352 : }
3353 ECB :
3354 GIC 91389 : gottuple = window_gettupleslot(winobj, abs_pos, slot);
3355 ECB :
3356 GIC 91389 : if (!gottuple)
3357 ECB : {
3358 CBC 159 : if (isout)
3359 159 : *isout = true;
3360 159 : *isnull = true;
3361 GIC 159 : return (Datum) 0;
3362 : }
3363 : else
3364 ECB : {
3365 CBC 91230 : if (isout)
3366 91230 : *isout = false;
3367 91230 : if (set_mark)
3368 91152 : WinSetMarkPosition(winobj, abs_pos);
3369 91230 : econtext->ecxt_outertuple = slot;
3370 GIC 91230 : return ExecEvalExpr((ExprState *) list_nth(winobj->argstates, argno),
3371 : econtext, isnull);
3372 : }
3373 : }
3374 :
3375 : /*
3376 : * WinGetFuncArgInFrame
3377 : * Evaluate a window function's argument expression on a specified
3378 : * row of the window frame. The row is identified in lseek(2) style,
3379 : * i.e. relative to the first or last row of the frame. (We do not
3380 : * support WINDOW_SEEK_CURRENT here, because it's not very clear what
3381 : * that should mean if the current row isn't part of the frame.)
3382 : *
3383 : * argno: argument number to evaluate (counted from 0)
3384 : * relpos: signed rowcount offset from the seek position
3385 : * seektype: WINDOW_SEEK_HEAD or WINDOW_SEEK_TAIL
3386 : * set_mark: If the row is found/in frame and set_mark is true, the mark is
3387 : * moved to the row as a side-effect.
3388 : * isnull: output argument, receives isnull status of result
3389 : * isout: output argument, set to indicate whether target row position
3390 : * is out of frame (can pass NULL if caller doesn't care about this)
3391 : *
3392 : * Specifying a nonexistent or not-in-frame row is not an error, it just
3393 : * causes a null result (plus setting *isout true, if isout isn't NULL).
3394 : *
3395 : * Note that some exclusion-clause options lead to situations where the
3396 : * rows that are in-frame are not consecutive in the partition. But we
3397 : * count only in-frame rows when measuring relpos.
3398 : *
3399 : * The set_mark flag is interpreted as meaning that the caller will specify
3400 : * a constant (or, perhaps, monotonically increasing) relpos in successive
3401 : * calls, so that *if there is no exclusion clause* there will be no need
3402 : * to fetch a row before the previously fetched row. But we do not expect
3403 : * the caller to know how to account for exclusion clauses. Therefore,
3404 : * if there is an exclusion clause we take responsibility for adjusting the
3405 : * mark request to something that will be safe given the above assumption
3406 : * about relpos.
3407 : */
3408 ECB : Datum
3409 GIC 3081 : WinGetFuncArgInFrame(WindowObject winobj, int argno,
3410 : int relpos, int seektype, bool set_mark,
3411 : bool *isnull, bool *isout)
3412 : {
3413 : WindowAggState *winstate;
3414 : ExprContext *econtext;
3415 : TupleTableSlot *slot;
3416 : int64 abs_pos;
3417 : int64 mark_pos;
3418 ECB :
3419 CBC 3081 : Assert(WindowObjectIsValid(winobj));
3420 3081 : winstate = winobj->winstate;
3421 3081 : econtext = winstate->ss.ps.ps_ExprContext;
3422 GIC 3081 : slot = winstate->temp_slot_1;
3423 ECB :
3424 GIC 3081 : switch (seektype)
3425 EUB : {
3426 UBC 0 : case WINDOW_SEEK_CURRENT:
3427 UIC 0 : elog(ERROR, "WINDOW_SEEK_CURRENT is not supported for WinGetFuncArgInFrame");
3428 : abs_pos = mark_pos = 0; /* keep compiler quiet */
3429 ECB : break;
3430 GIC 1473 : case WINDOW_SEEK_HEAD:
3431 ECB : /* rejecting relpos < 0 is easy and simplifies code below */
3432 GBC 1473 : if (relpos < 0)
3433 LBC 0 : goto out_of_frame;
3434 CBC 1473 : update_frameheadpos(winstate);
3435 1473 : abs_pos = winstate->frameheadpos + relpos;
3436 GIC 1473 : mark_pos = abs_pos;
3437 :
3438 : /*
3439 : * Account for exclusion option if one is active, but advance only
3440 : * abs_pos not mark_pos. This prevents changes of the current
3441 : * row's peer group from resulting in trying to fetch a row before
3442 : * some previous mark position.
3443 : *
3444 : * Note that in some corner cases such as current row being
3445 : * outside frame, these calculations are theoretically too simple,
3446 : * but it doesn't matter because we'll end up deciding the row is
3447 : * out of frame. We do not attempt to avoid fetching rows past
3448 : * end of frame; that would happen in some cases anyway.
3449 ECB : */
3450 GIC 1473 : switch (winstate->frameOptions & FRAMEOPTION_EXCLUSION)
3451 ECB : {
3452 GIC 1143 : case 0:
3453 ECB : /* no adjustment needed */
3454 CBC 1143 : break;
3455 120 : case FRAMEOPTION_EXCLUDE_CURRENT_ROW:
3456 120 : if (abs_pos >= winstate->currentpos &&
3457 93 : winstate->currentpos >= winstate->frameheadpos)
3458 33 : abs_pos++;
3459 120 : break;
3460 60 : case FRAMEOPTION_EXCLUDE_GROUP:
3461 60 : update_grouptailpos(winstate);
3462 60 : if (abs_pos >= winstate->groupheadpos &&
3463 GIC 36 : winstate->grouptailpos > winstate->frameheadpos)
3464 ECB : {
3465 GIC 36 : int64 overlapstart = Max(winstate->groupheadpos,
3466 : winstate->frameheadpos);
3467 ECB :
3468 GIC 36 : abs_pos += winstate->grouptailpos - overlapstart;
3469 ECB : }
3470 CBC 60 : break;
3471 150 : case FRAMEOPTION_EXCLUDE_TIES:
3472 150 : update_grouptailpos(winstate);
3473 150 : if (abs_pos >= winstate->groupheadpos &&
3474 GIC 102 : winstate->grouptailpos > winstate->frameheadpos)
3475 ECB : {
3476 GIC 42 : int64 overlapstart = Max(winstate->groupheadpos,
3477 : winstate->frameheadpos);
3478 ECB :
3479 CBC 42 : if (abs_pos == overlapstart)
3480 GIC 42 : abs_pos = winstate->currentpos;
3481 EUB : else
3482 UIC 0 : abs_pos += winstate->grouptailpos - overlapstart - 1;
3483 ECB : }
3484 GBC 150 : break;
3485 UBC 0 : default:
3486 UIC 0 : elog(ERROR, "unrecognized frame option state: 0x%x",
3487 : winstate->frameOptions);
3488 : break;
3489 ECB : }
3490 CBC 1473 : break;
3491 GIC 1608 : case WINDOW_SEEK_TAIL:
3492 ECB : /* rejecting relpos > 0 is easy and simplifies code below */
3493 GBC 1608 : if (relpos > 0)
3494 LBC 0 : goto out_of_frame;
3495 CBC 1608 : update_frametailpos(winstate);
3496 GIC 1608 : abs_pos = winstate->frametailpos - 1 + relpos;
3497 :
3498 : /*
3499 : * Account for exclusion option if one is active. If there is no
3500 : * exclusion, we can safely set the mark at the accessed row. But
3501 : * if there is, we can only mark the frame start, because we can't
3502 : * be sure how far back in the frame the exclusion might cause us
3503 : * to fetch in future. Furthermore, we have to actually check
3504 : * against frameheadpos here, since it's unsafe to try to fetch a
3505 : * row before frame start if the mark might be there already.
3506 ECB : */
3507 GIC 1608 : switch (winstate->frameOptions & FRAMEOPTION_EXCLUSION)
3508 ECB : {
3509 GIC 1368 : case 0:
3510 ECB : /* no adjustment needed */
3511 CBC 1368 : mark_pos = abs_pos;
3512 1368 : break;
3513 60 : case FRAMEOPTION_EXCLUDE_CURRENT_ROW:
3514 60 : if (abs_pos <= winstate->currentpos &&
3515 6 : winstate->currentpos < winstate->frametailpos)
3516 6 : abs_pos--;
3517 60 : update_frameheadpos(winstate);
3518 60 : if (abs_pos < winstate->frameheadpos)
3519 3 : goto out_of_frame;
3520 57 : mark_pos = winstate->frameheadpos;
3521 57 : break;
3522 120 : case FRAMEOPTION_EXCLUDE_GROUP:
3523 120 : update_grouptailpos(winstate);
3524 120 : if (abs_pos < winstate->grouptailpos &&
3525 GIC 27 : winstate->groupheadpos < winstate->frametailpos)
3526 ECB : {
3527 GIC 27 : int64 overlapend = Min(winstate->grouptailpos,
3528 : winstate->frametailpos);
3529 ECB :
3530 GIC 27 : abs_pos -= overlapend - winstate->groupheadpos;
3531 ECB : }
3532 CBC 120 : update_frameheadpos(winstate);
3533 120 : if (abs_pos < winstate->frameheadpos)
3534 27 : goto out_of_frame;
3535 93 : mark_pos = winstate->frameheadpos;
3536 93 : break;
3537 60 : case FRAMEOPTION_EXCLUDE_TIES:
3538 60 : update_grouptailpos(winstate);
3539 60 : if (abs_pos < winstate->grouptailpos &&
3540 GIC 18 : winstate->groupheadpos < winstate->frametailpos)
3541 ECB : {
3542 GIC 18 : int64 overlapend = Min(winstate->grouptailpos,
3543 : winstate->frametailpos);
3544 ECB :
3545 CBC 18 : if (abs_pos == overlapend - 1)
3546 GIC 18 : abs_pos = winstate->currentpos;
3547 EUB : else
3548 UIC 0 : abs_pos -= overlapend - 1 - winstate->groupheadpos;
3549 ECB : }
3550 CBC 60 : update_frameheadpos(winstate);
3551 GBC 60 : if (abs_pos < winstate->frameheadpos)
3552 LBC 0 : goto out_of_frame;
3553 CBC 60 : mark_pos = winstate->frameheadpos;
3554 GBC 60 : break;
3555 UBC 0 : default:
3556 UIC 0 : elog(ERROR, "unrecognized frame option state: 0x%x",
3557 : winstate->frameOptions);
3558 : mark_pos = 0; /* keep compiler quiet */
3559 : break;
3560 ECB : }
3561 GBC 1578 : break;
3562 UBC 0 : default:
3563 UIC 0 : elog(ERROR, "unrecognized window seek type: %d", seektype);
3564 : abs_pos = mark_pos = 0; /* keep compiler quiet */
3565 : break;
3566 : }
3567 ECB :
3568 CBC 3051 : if (!window_gettupleslot(winobj, abs_pos, slot))
3569 GIC 51 : goto out_of_frame;
3570 :
3571 ECB : /* The code above does not detect all out-of-frame cases, so check */
3572 CBC 3000 : if (row_is_in_frame(winstate, abs_pos, slot) <= 0)
3573 GIC 6 : goto out_of_frame;
3574 ECB :
3575 GBC 2985 : if (isout)
3576 LBC 0 : *isout = false;
3577 CBC 2985 : if (set_mark)
3578 2964 : WinSetMarkPosition(winobj, mark_pos);
3579 2985 : econtext->ecxt_outertuple = slot;
3580 GIC 2985 : return ExecEvalExpr((ExprState *) list_nth(winobj->argstates, argno),
3581 : econtext, isnull);
3582 ECB :
3583 CBC 87 : out_of_frame:
3584 GBC 87 : if (isout)
3585 LBC 0 : *isout = true;
3586 CBC 87 : *isnull = true;
3587 GIC 87 : return (Datum) 0;
3588 : }
3589 :
3590 : /*
3591 : * WinGetFuncArgCurrent
3592 : * Evaluate a window function's argument expression on the current row.
3593 : *
3594 : * argno: argument number to evaluate (counted from 0)
3595 : * isnull: output argument, receives isnull status of result
3596 : *
3597 : * Note: this isn't quite equivalent to WinGetFuncArgInPartition or
3598 : * WinGetFuncArgInFrame targeting the current row, because it will succeed
3599 : * even if the WindowObject's mark has been set beyond the current row.
3600 : * This should generally be used for "ordinary" arguments of a window
3601 : * function, such as the offset argument of lead() or lag().
3602 : */
3603 ECB : Datum
3604 GIC 582 : WinGetFuncArgCurrent(WindowObject winobj, int argno, bool *isnull)
3605 : {
3606 : WindowAggState *winstate;
3607 : ExprContext *econtext;
3608 ECB :
3609 CBC 582 : Assert(WindowObjectIsValid(winobj));
3610 GIC 582 : winstate = winobj->winstate;
3611 ECB :
3612 GIC 582 : econtext = winstate->ss.ps.ps_ExprContext;
3613 ECB :
3614 CBC 582 : econtext->ecxt_outertuple = winstate->ss.ss_ScanTupleSlot;
3615 GIC 582 : return ExecEvalExpr((ExprState *) list_nth(winobj->argstates, argno),
3616 : econtext, isnull);
3617 : }
|
