Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * nodeIncrementalSort.c
4 : : * Routines to handle incremental sorting of relations.
5 : : *
6 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : * IDENTIFICATION
10 : : * src/backend/executor/nodeIncrementalSort.c
11 : : *
12 : : * DESCRIPTION
13 : : *
14 : : * Incremental sort is an optimized variant of multikey sort for cases
15 : : * when the input is already sorted by a prefix of the sort keys. For
16 : : * example when a sort by (key1, key2 ... keyN) is requested, and the
17 : : * input is already sorted by (key1, key2 ... keyM), M < N, we can
18 : : * divide the input into groups where keys (key1, ... keyM) are equal,
19 : : * and only sort on the remaining columns.
20 : : *
21 : : * Consider the following example. We have input tuples consisting of
22 : : * two integers (X, Y) already presorted by X, while it's required to
23 : : * sort them by both X and Y. Let input tuples be following.
24 : : *
25 : : * (1, 5)
26 : : * (1, 2)
27 : : * (2, 9)
28 : : * (2, 1)
29 : : * (2, 5)
30 : : * (3, 3)
31 : : * (3, 7)
32 : : *
33 : : * An incremental sort algorithm would split the input into the following
34 : : * groups, which have equal X, and then sort them by Y individually:
35 : : *
36 : : * (1, 5) (1, 2)
37 : : * (2, 9) (2, 1) (2, 5)
38 : : * (3, 3) (3, 7)
39 : : *
40 : : * After sorting these groups and putting them altogether, we would get
41 : : * the following result which is sorted by X and Y, as requested:
42 : : *
43 : : * (1, 2)
44 : : * (1, 5)
45 : : * (2, 1)
46 : : * (2, 5)
47 : : * (2, 9)
48 : : * (3, 3)
49 : : * (3, 7)
50 : : *
51 : : * Incremental sort may be more efficient than plain sort, particularly
52 : : * on large datasets, as it reduces the amount of data to sort at once,
53 : : * making it more likely it fits into work_mem (eliminating the need to
54 : : * spill to disk). But the main advantage of incremental sort is that
55 : : * it can start producing rows early, before sorting the whole dataset,
56 : : * which is a significant benefit especially for queries with LIMIT.
57 : : *
58 : : * The algorithm we've implemented here is modified from the theoretical
59 : : * base described above by operating in two different modes:
60 : : * - Fetching a minimum number of tuples without checking prefix key
61 : : * group membership and sorting on all columns when safe.
62 : : * - Fetching all tuples for a single prefix key group and sorting on
63 : : * solely the unsorted columns.
64 : : * We always begin in the first mode, and employ a heuristic to switch
65 : : * into the second mode if we believe it's beneficial.
66 : : *
67 : : * Sorting incrementally can potentially use less memory, avoid fetching
68 : : * and sorting all tuples in the dataset, and begin returning tuples before
69 : : * the entire result set is available.
70 : : *
71 : : * The hybrid mode approach allows us to optimize for both very small
72 : : * groups (where the overhead of a new tuplesort is high) and very large
73 : : * groups (where we can lower cost by not having to sort on already sorted
74 : : * columns), albeit at some extra cost while switching between modes.
75 : : *
76 : : *-------------------------------------------------------------------------
77 : : */
78 : :
79 : : #include "postgres.h"
80 : :
81 : : #include "executor/execdebug.h"
82 : : #include "executor/nodeIncrementalSort.h"
83 : : #include "miscadmin.h"
84 : : #include "utils/lsyscache.h"
85 : : #include "utils/tuplesort.h"
86 : :
87 : : /*
88 : : * We need to store the instrumentation information in either local node's sort
89 : : * info or, for a parallel worker process, in the shared info (this avoids
90 : : * having to additionally memcpy the info from local memory to shared memory
91 : : * at each instrumentation call). This macro expands to choose the proper sort
92 : : * state and group info.
93 : : *
94 : : * Arguments:
95 : : * - node: type IncrementalSortState *
96 : : * - groupName: the token fullsort or prefixsort
97 : : */
98 : : #define INSTRUMENT_SORT_GROUP(node, groupName) \
99 : : do { \
100 : : if ((node)->ss.ps.instrument != NULL) \
101 : : { \
102 : : if ((node)->shared_info && (node)->am_worker) \
103 : : { \
104 : : Assert(IsParallelWorker()); \
105 : : Assert(ParallelWorkerNumber < (node)->shared_info->num_workers); \
106 : : instrumentSortedGroup(&(node)->shared_info->sinfo[ParallelWorkerNumber].groupName##GroupInfo, \
107 : : (node)->groupName##_state); \
108 : : } \
109 : : else \
110 : : { \
111 : : instrumentSortedGroup(&(node)->incsort_info.groupName##GroupInfo, \
112 : : (node)->groupName##_state); \
113 : : } \
114 : : } \
115 : : } while (0)
116 : :
117 : :
118 : : /* ----------------------------------------------------------------
119 : : * instrumentSortedGroup
120 : : *
121 : : * Because incremental sort processes (potentially many) sort batches, we need
122 : : * to capture tuplesort stats each time we finalize a sort state. This summary
123 : : * data is later used for EXPLAIN ANALYZE output.
124 : : * ----------------------------------------------------------------
125 : : */
126 : : static void
127 : 96 : instrumentSortedGroup(IncrementalSortGroupInfo *groupInfo,
128 : : Tuplesortstate *sortState)
129 : : {
130 : : TuplesortInstrumentation sort_instr;
131 : :
132 : 96 : groupInfo->groupCount++;
133 : :
134 : 96 : tuplesort_get_stats(sortState, &sort_instr);
135 : :
136 : : /* Calculate total and maximum memory and disk space used. */
137 [ - + - ]: 96 : switch (sort_instr.spaceType)
138 : : {
139 : 0 : case SORT_SPACE_TYPE_DISK:
140 : 0 : groupInfo->totalDiskSpaceUsed += sort_instr.spaceUsed;
141 [ # # ]: 0 : if (sort_instr.spaceUsed > groupInfo->maxDiskSpaceUsed)
142 : 0 : groupInfo->maxDiskSpaceUsed = sort_instr.spaceUsed;
143 : :
144 : 0 : break;
145 : 96 : case SORT_SPACE_TYPE_MEMORY:
146 : 96 : groupInfo->totalMemorySpaceUsed += sort_instr.spaceUsed;
147 [ + + ]: 96 : if (sort_instr.spaceUsed > groupInfo->maxMemorySpaceUsed)
148 : 36 : groupInfo->maxMemorySpaceUsed = sort_instr.spaceUsed;
149 : :
150 : 96 : break;
151 : : }
152 : :
153 : : /* Track each sort method we've used. */
154 : 96 : groupInfo->sortMethods |= sort_instr.sortMethod;
155 : 96 : }
156 : :
157 : : /* ----------------------------------------------------------------
158 : : * preparePresortedCols
159 : : *
160 : : * Prepare information for presorted_keys comparisons.
161 : : * ----------------------------------------------------------------
162 : : */
163 : : static void
164 : 580 : preparePresortedCols(IncrementalSortState *node)
165 : : {
166 : 580 : IncrementalSort *plannode = castNode(IncrementalSort, node->ss.ps.plan);
167 : :
168 : 580 : node->presorted_keys = palloc_array(PresortedKeyData, plannode->nPresortedCols);
169 : :
170 : : /* Pre-cache comparison functions for each pre-sorted key. */
171 [ + + ]: 1164 : for (int i = 0; i < plannode->nPresortedCols; i++)
172 : : {
173 : : Oid equalityOp,
174 : : equalityFunc;
175 : : PresortedKeyData *key;
176 : :
177 : 584 : key = &node->presorted_keys[i];
178 : 584 : key->attno = plannode->sort.sortColIdx[i];
179 : :
180 : 584 : equalityOp = get_equality_op_for_ordering_op(plannode->sort.sortOperators[i],
181 : : NULL);
182 [ - + ]: 584 : if (!OidIsValid(equalityOp))
183 [ # # ]: 0 : elog(ERROR, "missing equality operator for ordering operator %u",
184 : : plannode->sort.sortOperators[i]);
185 : :
186 : 584 : equalityFunc = get_opcode(equalityOp);
187 [ - + ]: 584 : if (!OidIsValid(equalityFunc))
188 [ # # ]: 0 : elog(ERROR, "missing function for operator %u", equalityOp);
189 : :
190 : : /* Lookup the comparison function */
191 : 584 : fmgr_info_cxt(equalityFunc, &key->flinfo, CurrentMemoryContext);
192 : :
193 : : /* We can initialize the callinfo just once and re-use it */
194 : 584 : key->fcinfo = palloc0(SizeForFunctionCallInfo(2));
195 : 584 : InitFunctionCallInfoData(*key->fcinfo, &key->flinfo, 2,
196 : : plannode->sort.collations[i], NULL, NULL);
197 : 584 : key->fcinfo->args[0].isnull = false;
198 : 584 : key->fcinfo->args[1].isnull = false;
199 : : }
200 : 580 : }
201 : :
202 : : /* ----------------------------------------------------------------
203 : : * isCurrentGroup
204 : : *
205 : : * Check whether a given tuple belongs to the current sort group by comparing
206 : : * the presorted column values to the pivot tuple of the current group.
207 : : * ----------------------------------------------------------------
208 : : */
209 : : static bool
210 : 332180 : isCurrentGroup(IncrementalSortState *node, TupleTableSlot *pivot, TupleTableSlot *tuple)
211 : : {
212 : : int nPresortedCols;
213 : :
214 : 332180 : nPresortedCols = castNode(IncrementalSort, node->ss.ps.plan)->nPresortedCols;
215 : :
216 : : /*
217 : : * That the input is sorted by keys * (0, ... n) implies that the tail
218 : : * keys are more likely to change. Therefore we do our comparison starting
219 : : * from the last pre-sorted column to optimize for early detection of
220 : : * inequality and minimizing the number of function calls..
221 : : */
222 [ + + ]: 662417 : for (int i = nPresortedCols - 1; i >= 0; i--)
223 : : {
224 : : Datum datumA,
225 : : datumB,
226 : : result;
227 : : bool isnullA,
228 : : isnullB;
229 : 332180 : AttrNumber attno = node->presorted_keys[i].attno;
230 : : PresortedKeyData *key;
231 : :
232 : 332180 : datumA = slot_getattr(pivot, attno, &isnullA);
233 : 332180 : datumB = slot_getattr(tuple, attno, &isnullB);
234 : :
235 : : /* Special case for NULL-vs-NULL, else use standard comparison */
236 [ + - - + ]: 332180 : if (isnullA || isnullB)
237 : : {
238 [ # # ]: 0 : if (isnullA == isnullB)
239 : 0 : continue;
240 : : else
241 : 1943 : return false;
242 : : }
243 : :
244 : 332180 : key = &node->presorted_keys[i];
245 : :
246 : 332180 : key->fcinfo->args[0].value = datumA;
247 : 332180 : key->fcinfo->args[1].value = datumB;
248 : :
249 : : /* just for paranoia's sake, we reset isnull each time */
250 : 332180 : key->fcinfo->isnull = false;
251 : :
252 : 332180 : result = FunctionCallInvoke(key->fcinfo);
253 : :
254 : : /* Check for null result, since caller is clearly not expecting one */
255 [ - + ]: 332180 : if (key->fcinfo->isnull)
256 [ # # ]: 0 : elog(ERROR, "function %u returned NULL", key->flinfo.fn_oid);
257 : :
258 [ + + ]: 332180 : if (!DatumGetBool(result))
259 : 1943 : return false;
260 : : }
261 : 330237 : return true;
262 : : }
263 : :
264 : : /* ----------------------------------------------------------------
265 : : * switchToPresortedPrefixMode
266 : : *
267 : : * When we determine that we've likely encountered a large batch of tuples all
268 : : * having the same presorted prefix values, we want to optimize tuplesort by
269 : : * only sorting on unsorted suffix keys.
270 : : *
271 : : * The problem is that we've already accumulated several tuples in another
272 : : * tuplesort configured to sort by all columns (assuming that there may be
273 : : * more than one prefix key group). So to switch to presorted prefix mode we
274 : : * have to go back and look at all the tuples we've already accumulated to
275 : : * verify they're all part of the same prefix key group before sorting them
276 : : * solely by unsorted suffix keys.
277 : : *
278 : : * While it's likely that all tuples already fetched are all part of a single
279 : : * prefix group, we also have to handle the possibility that there is at least
280 : : * one different prefix key group before the large prefix key group.
281 : : * ----------------------------------------------------------------
282 : : */
283 : : static void
284 : 371 : switchToPresortedPrefixMode(PlanState *pstate)
285 : : {
286 : 371 : IncrementalSortState *node = castNode(IncrementalSortState, pstate);
287 : : ScanDirection dir;
288 : : int64 nTuples;
289 : : TupleDesc tupDesc;
290 : : PlanState *outerNode;
291 : 371 : IncrementalSort *plannode = castNode(IncrementalSort, node->ss.ps.plan);
292 : :
293 : 371 : dir = node->ss.ps.state->es_direction;
294 : 371 : outerNode = outerPlanState(node);
295 : 371 : tupDesc = ExecGetResultType(outerNode);
296 : :
297 : : /* Configure the prefix sort state the first time around. */
298 [ + + ]: 371 : if (node->prefixsort_state == NULL)
299 : : {
300 : : Tuplesortstate *prefixsort_state;
301 : 74 : int nPresortedCols = plannode->nPresortedCols;
302 : :
303 : : /*
304 : : * Optimize the sort by assuming the prefix columns are all equal and
305 : : * thus we only need to sort by any remaining columns.
306 : : */
307 : 74 : prefixsort_state = tuplesort_begin_heap(tupDesc,
308 : 74 : plannode->sort.numCols - nPresortedCols,
309 : 74 : &(plannode->sort.sortColIdx[nPresortedCols]),
310 : 74 : &(plannode->sort.sortOperators[nPresortedCols]),
311 : 74 : &(plannode->sort.collations[nPresortedCols]),
312 : 74 : &(plannode->sort.nullsFirst[nPresortedCols]),
313 : : work_mem,
314 : : NULL,
315 [ + + ]: 74 : node->bounded ? TUPLESORT_ALLOWBOUNDED : TUPLESORT_NONE);
316 : 74 : node->prefixsort_state = prefixsort_state;
317 : : }
318 : : else
319 : : {
320 : : /* Next group of presorted data */
321 : 297 : tuplesort_reset(node->prefixsort_state);
322 : : }
323 : :
324 : : /*
325 : : * If the current node has a bound, then it's reasonably likely that a
326 : : * large prefix key group will benefit from bounded sort, so configure the
327 : : * tuplesort to allow for that optimization.
328 : : */
329 [ + + ]: 371 : if (node->bounded)
330 : : {
331 : : SO1_printf("Setting bound on presorted prefix tuplesort to: " INT64_FORMAT "\n",
332 : : node->bound - node->bound_Done);
333 : 121 : tuplesort_set_bound(node->prefixsort_state,
334 : 121 : node->bound - node->bound_Done);
335 : : }
336 : :
337 : : /*
338 : : * Copy as many tuples as we can (i.e., in the same prefix key group) from
339 : : * the full sort state to the prefix sort state.
340 : : */
341 [ + + ]: 16050 : for (nTuples = 0; nTuples < node->n_fullsort_remaining; nTuples++)
342 : : {
343 : : /*
344 : : * When we encounter multiple prefix key groups inside the full sort
345 : : * tuplesort we have to carry over the last read tuple into the next
346 : : * batch.
347 : : */
348 [ + + + - : 15803 : if (nTuples == 0 && !TupIsNull(node->transfer_tuple))
+ + ]
349 : : {
350 : 124 : tuplesort_puttupleslot(node->prefixsort_state, node->transfer_tuple);
351 : : /* The carried over tuple is our new group pivot tuple. */
352 : 124 : ExecCopySlot(node->group_pivot, node->transfer_tuple);
353 : : }
354 : : else
355 : : {
356 : 15679 : tuplesort_gettupleslot(node->fullsort_state,
357 : : ScanDirectionIsForward(dir),
358 : : false, node->transfer_tuple, NULL);
359 : :
360 : : /*
361 : : * If this is our first time through the loop, then we need to
362 : : * save the first tuple we get as our new group pivot.
363 : : */
364 [ + - + + ]: 15679 : if (TupIsNull(node->group_pivot))
365 : 247 : ExecCopySlot(node->group_pivot, node->transfer_tuple);
366 : :
367 [ + + ]: 15679 : if (isCurrentGroup(node, node->group_pivot, node->transfer_tuple))
368 : : {
369 : 15555 : tuplesort_puttupleslot(node->prefixsort_state, node->transfer_tuple);
370 : : }
371 : : else
372 : : {
373 : : /*
374 : : * The tuple isn't part of the current batch so we need to
375 : : * carry it over into the next batch of tuples we transfer out
376 : : * of the full sort tuplesort into the presorted prefix
377 : : * tuplesort. We don't actually have to do anything special to
378 : : * save the tuple since we've already loaded it into the
379 : : * node->transfer_tuple slot, and, even though that slot
380 : : * points to memory inside the full sort tuplesort, we can't
381 : : * reset that tuplesort anyway until we've fully transferred
382 : : * out its tuples, so this reference is safe. We do need to
383 : : * reset the group pivot tuple though since we've finished the
384 : : * current prefix key group.
385 : : */
386 : 124 : ExecClearTuple(node->group_pivot);
387 : :
388 : : /* Break out of for-loop early */
389 : 124 : break;
390 : : }
391 : : }
392 : : }
393 : :
394 : : /*
395 : : * Track how many tuples remain in the full sort batch so that we know if
396 : : * we need to sort multiple prefix key groups before processing tuples
397 : : * remaining in the large single prefix key group we think we've
398 : : * encountered.
399 : : */
400 : : SO1_printf("Moving " INT64_FORMAT " tuples to presorted prefix tuplesort\n", nTuples);
401 : 371 : node->n_fullsort_remaining -= nTuples;
402 : : SO1_printf("Setting n_fullsort_remaining to " INT64_FORMAT "\n", node->n_fullsort_remaining);
403 : :
404 [ + + ]: 371 : if (node->n_fullsort_remaining == 0)
405 : : {
406 : : /*
407 : : * We've found that all tuples remaining in the full sort batch are in
408 : : * the same prefix key group and moved all of those tuples into the
409 : : * presorted prefix tuplesort. We don't know that we've yet found the
410 : : * last tuple in the current prefix key group, so save our pivot
411 : : * comparison tuple and continue fetching tuples from the outer
412 : : * execution node to load into the presorted prefix tuplesort.
413 : : */
414 : 247 : ExecCopySlot(node->group_pivot, node->transfer_tuple);
415 : : SO_printf("Setting execution_status to INCSORT_LOADPREFIXSORT (switchToPresortedPrefixMode)\n");
416 : 247 : node->execution_status = INCSORT_LOADPREFIXSORT;
417 : :
418 : : /*
419 : : * Make sure we clear the transfer tuple slot so that next time we
420 : : * encounter a large prefix key group we don't incorrectly assume we
421 : : * have a tuple carried over from the previous group.
422 : : */
423 : 247 : ExecClearTuple(node->transfer_tuple);
424 : : }
425 : : else
426 : : {
427 : : /*
428 : : * We finished a group but didn't consume all of the tuples from the
429 : : * full sort state, so we'll sort this batch, let the outer node read
430 : : * out all of those tuples, and then come back around to find another
431 : : * batch.
432 : : */
433 : : SO1_printf("Sorting presorted prefix tuplesort with " INT64_FORMAT " tuples\n", nTuples);
434 : 124 : tuplesort_performsort(node->prefixsort_state);
435 : :
436 [ + + - + : 124 : INSTRUMENT_SORT_GROUP(node, prefixsort);
- - ]
437 : :
438 [ + + ]: 124 : if (node->bounded)
439 : : {
440 : : /*
441 : : * If the current node has a bound and we've already sorted n
442 : : * tuples, then the functional bound remaining is (original bound
443 : : * - n), so store the current number of processed tuples for use
444 : : * in configuring sorting bound.
445 : : */
446 : : SO2_printf("Changing bound_Done from " INT64_FORMAT " to " INT64_FORMAT "\n",
447 : : Min(node->bound, node->bound_Done + nTuples), node->bound_Done);
448 : 80 : node->bound_Done = Min(node->bound, node->bound_Done + nTuples);
449 : : }
450 : :
451 : : SO_printf("Setting execution_status to INCSORT_READPREFIXSORT (switchToPresortedPrefixMode)\n");
452 : 124 : node->execution_status = INCSORT_READPREFIXSORT;
453 : : }
454 : 371 : }
455 : :
456 : : /*
457 : : * Sorting many small groups with tuplesort is inefficient. In order to
458 : : * cope with this problem we don't start a new group until the current one
459 : : * contains at least DEFAULT_MIN_GROUP_SIZE tuples (unfortunately this also
460 : : * means we can't assume small groups of tuples all have the same prefix keys.)
461 : : * When we have a bound that's less than DEFAULT_MIN_GROUP_SIZE we start looking
462 : : * for the new group as soon as we've met our bound to avoid fetching more
463 : : * tuples than we absolutely have to fetch.
464 : : */
465 : : #define DEFAULT_MIN_GROUP_SIZE 32
466 : :
467 : : /*
468 : : * While we've optimized for small prefix key groups by not starting our prefix
469 : : * key comparisons until we've reached a minimum number of tuples, we don't want
470 : : * that optimization to cause us to lose out on the benefits of being able to
471 : : * assume a large group of tuples is fully presorted by its prefix keys.
472 : : * Therefore we use the DEFAULT_MAX_FULL_SORT_GROUP_SIZE cutoff as a heuristic
473 : : * for determining when we believe we've encountered a large group, and, if we
474 : : * get to that point without finding a new prefix key group we transition to
475 : : * presorted prefix key mode.
476 : : */
477 : : #define DEFAULT_MAX_FULL_SORT_GROUP_SIZE (2 * DEFAULT_MIN_GROUP_SIZE)
478 : :
479 : : /* ----------------------------------------------------------------
480 : : * ExecIncrementalSort
481 : : *
482 : : * Assuming that outer subtree returns tuple presorted by some prefix
483 : : * of target sort columns, performs incremental sort.
484 : : *
485 : : * Conditions:
486 : : * -- none.
487 : : *
488 : : * Initial States:
489 : : * -- the outer child is prepared to return the first tuple.
490 : : * ----------------------------------------------------------------
491 : : */
492 : : static TupleTableSlot *
493 : 354639 : ExecIncrementalSort(PlanState *pstate)
494 : : {
495 : 354639 : IncrementalSortState *node = castNode(IncrementalSortState, pstate);
496 : : EState *estate;
497 : : ScanDirection dir;
498 : : Tuplesortstate *read_sortstate;
499 : : Tuplesortstate *fullsort_state;
500 : : TupleTableSlot *slot;
501 : 354639 : IncrementalSort *plannode = (IncrementalSort *) node->ss.ps.plan;
502 : : PlanState *outerNode;
503 : : TupleDesc tupDesc;
504 : 354639 : int64 nTuples = 0;
505 : : int64 minGroupSize;
506 : :
507 [ - + ]: 354639 : CHECK_FOR_INTERRUPTS();
508 : :
509 : 354639 : estate = node->ss.ps.state;
510 : 354639 : dir = estate->es_direction;
511 : 354639 : fullsort_state = node->fullsort_state;
512 : :
513 : : /*
514 : : * If a previous iteration has sorted a batch, then we need to check to
515 : : * see if there are any remaining tuples in that batch that we can return
516 : : * before moving on to other execution states.
517 : : */
518 [ + + ]: 354639 : if (node->execution_status == INCSORT_READFULLSORT
519 [ + + ]: 295112 : || node->execution_status == INCSORT_READPREFIXSORT)
520 : : {
521 : : /*
522 : : * Return next tuple from the current sorted group set if available.
523 : : */
524 : 708110 : read_sortstate = node->execution_status == INCSORT_READFULLSORT ?
525 [ + + ]: 354055 : fullsort_state : node->prefixsort_state;
526 : 354055 : slot = node->ss.ps.ps_ResultTupleSlot;
527 : :
528 : : /*
529 : : * We have to populate the slot from the tuplesort before checking
530 : : * outerNodeDone because it will set the slot to NULL if no more
531 : : * tuples remain. If the tuplesort is empty, but we don't have any
532 : : * more tuples available for sort from the outer node, then
533 : : * outerNodeDone will have been set so we'll return that now-empty
534 : : * slot to the caller.
535 : : */
536 [ + + ]: 354055 : if (tuplesort_gettupleslot(read_sortstate, ScanDirectionIsForward(dir),
537 [ + + ]: 2277 : false, slot, NULL) || node->outerNodeDone)
538 : :
539 : : /*
540 : : * Note: there isn't a good test case for the node->outerNodeDone
541 : : * check directly, but we need it for any plan where the outer
542 : : * node will fail when trying to fetch too many tuples.
543 : : */
544 : 352189 : return slot;
545 [ + + ]: 1866 : else if (node->n_fullsort_remaining > 0)
546 : : {
547 : : /*
548 : : * When we transition to presorted prefix mode, we might have
549 : : * accumulated at least one additional prefix key group in the
550 : : * full sort tuplesort. The first call to
551 : : * switchToPresortedPrefixMode() will have pulled the first one of
552 : : * those groups out, and we've returned those tuples to the parent
553 : : * node, but if at this point we still have tuples remaining in
554 : : * the full sort state (i.e., n_fullsort_remaining > 0), then we
555 : : * need to re-execute the prefix mode transition function to pull
556 : : * out the next prefix key group.
557 : : */
558 : : SO1_printf("Re-calling switchToPresortedPrefixMode() because n_fullsort_remaining is > 0 (" INT64_FORMAT ")\n",
559 : : node->n_fullsort_remaining);
560 : 124 : switchToPresortedPrefixMode(pstate);
561 : : }
562 : : else
563 : : {
564 : : /*
565 : : * If we don't have any sorted tuples to read and we're not
566 : : * currently transitioning into presorted prefix sort mode, then
567 : : * it's time to start the process all over again by building a new
568 : : * group in the full sort state.
569 : : */
570 : : SO_printf("Setting execution_status to INCSORT_LOADFULLSORT (n_fullsort_remaining > 0)\n");
571 : 1742 : node->execution_status = INCSORT_LOADFULLSORT;
572 : : }
573 : : }
574 : :
575 : : /*
576 : : * Scan the subplan in the forward direction while creating the sorted
577 : : * data.
578 : : */
579 : 2450 : estate->es_direction = ForwardScanDirection;
580 : :
581 : 2450 : outerNode = outerPlanState(node);
582 : 2450 : tupDesc = ExecGetResultType(outerNode);
583 : :
584 : : /* Load tuples into the full sort state. */
585 [ + + ]: 2450 : if (node->execution_status == INCSORT_LOADFULLSORT)
586 : : {
587 : : /*
588 : : * Initialize sorting structures.
589 : : */
590 [ + + ]: 2326 : if (fullsort_state == NULL)
591 : : {
592 : : /*
593 : : * Initialize presorted column support structures for
594 : : * isCurrentGroup(). It's correct to do this along with the
595 : : * initial initialization for the full sort state (and not for the
596 : : * prefix sort state) since we always load the full sort state
597 : : * first.
598 : : */
599 : 580 : preparePresortedCols(node);
600 : :
601 : : /*
602 : : * Since we optimize small prefix key groups by accumulating a
603 : : * minimum number of tuples before sorting, we can't assume that a
604 : : * group of tuples all have the same prefix key values. Hence we
605 : : * setup the full sort tuplesort to sort by all requested sort
606 : : * keys.
607 : : */
608 : 580 : fullsort_state = tuplesort_begin_heap(tupDesc,
609 : : plannode->sort.numCols,
610 : : plannode->sort.sortColIdx,
611 : : plannode->sort.sortOperators,
612 : : plannode->sort.collations,
613 : : plannode->sort.nullsFirst,
614 : : work_mem,
615 : : NULL,
616 [ + + ]: 580 : node->bounded ?
617 : : TUPLESORT_ALLOWBOUNDED :
618 : : TUPLESORT_NONE);
619 : 580 : node->fullsort_state = fullsort_state;
620 : : }
621 : : else
622 : : {
623 : : /* Reset sort for the next batch. */
624 : 1746 : tuplesort_reset(fullsort_state);
625 : : }
626 : :
627 : : /*
628 : : * Calculate the remaining tuples left if bounded and configure both
629 : : * bounded sort and the minimum group size accordingly.
630 : : */
631 [ + + ]: 2326 : if (node->bounded)
632 : : {
633 : 141 : int64 currentBound = node->bound - node->bound_Done;
634 : :
635 : : /*
636 : : * Bounded sort isn't likely to be a useful optimization for full
637 : : * sort mode since we limit full sort mode to a relatively small
638 : : * number of tuples and tuplesort doesn't switch over to top-n
639 : : * heap sort anyway unless it hits (2 * bound) tuples.
640 : : */
641 [ + + ]: 141 : if (currentBound < DEFAULT_MIN_GROUP_SIZE)
642 : 52 : tuplesort_set_bound(fullsort_state, currentBound);
643 : :
644 : 141 : minGroupSize = Min(DEFAULT_MIN_GROUP_SIZE, currentBound);
645 : : }
646 : : else
647 : 2185 : minGroupSize = DEFAULT_MIN_GROUP_SIZE;
648 : :
649 : : /*
650 : : * Because we have to read the next tuple to find out that we've
651 : : * encountered a new prefix key group, on subsequent groups we have to
652 : : * carry over that extra tuple and add it to the new group's sort here
653 : : * before we read any new tuples from the outer node.
654 : : */
655 [ + - + + ]: 2326 : if (!TupIsNull(node->group_pivot))
656 : : {
657 : 1742 : tuplesort_puttupleslot(fullsort_state, node->group_pivot);
658 : 1742 : nTuples++;
659 : :
660 : : /*
661 : : * We're in full sort mode accumulating a minimum number of tuples
662 : : * and not checking for prefix key equality yet, so we can't
663 : : * assume the group pivot tuple will remain the same -- unless
664 : : * we're using a minimum group size of 1, in which case the pivot
665 : : * is obviously still the pivot.
666 : : */
667 [ + + ]: 1742 : if (nTuples != minGroupSize)
668 : 1734 : ExecClearTuple(node->group_pivot);
669 : : }
670 : :
671 : :
672 : : /*
673 : : * Pull as many tuples from the outer node as possible given our
674 : : * current operating mode.
675 : : */
676 : : for (;;)
677 : : {
678 : 76807 : slot = ExecProcNode(outerNode);
679 : :
680 : : /*
681 : : * If the outer node can't provide us any more tuples, then we can
682 : : * sort the current group and return those tuples.
683 : : */
684 [ + + + + ]: 76807 : if (TupIsNull(slot))
685 : : {
686 : : /*
687 : : * We need to know later if the outer node has completed to be
688 : : * able to distinguish between being done with a batch and
689 : : * being done with the whole node.
690 : : */
691 : 454 : node->outerNodeDone = true;
692 : :
693 : : SO1_printf("Sorting fullsort with " INT64_FORMAT " tuples\n", nTuples);
694 : 454 : tuplesort_performsort(fullsort_state);
695 : :
696 [ - + - - : 454 : INSTRUMENT_SORT_GROUP(node, fullsort);
- - ]
697 : :
698 : : SO_printf("Setting execution_status to INCSORT_READFULLSORT (final tuple)\n");
699 : 454 : node->execution_status = INCSORT_READFULLSORT;
700 : 454 : break;
701 : : }
702 : :
703 : : /* Accumulate the next group of presorted tuples. */
704 [ + + ]: 76353 : if (nTuples < minGroupSize)
705 : : {
706 : : /*
707 : : * If we haven't yet hit our target minimum group size, then
708 : : * we don't need to bother checking for inclusion in the
709 : : * current prefix group since at this point we'll assume that
710 : : * we'll full sort this batch to avoid a large number of very
711 : : * tiny (and thus inefficient) sorts.
712 : : */
713 : 59402 : tuplesort_puttupleslot(fullsort_state, slot);
714 : 59402 : nTuples++;
715 : :
716 : : /*
717 : : * If we've reached our minimum group size, then we need to
718 : : * store the most recent tuple as a pivot.
719 : : */
720 [ + + ]: 59402 : if (nTuples == minGroupSize)
721 : 1870 : ExecCopySlot(node->group_pivot, slot);
722 : : }
723 : : else
724 : : {
725 : : /*
726 : : * If we've already accumulated enough tuples to reach our
727 : : * minimum group size, then we need to compare any additional
728 : : * tuples to our pivot tuple to see if we reach the end of
729 : : * that prefix key group. Only after we find changed prefix
730 : : * keys can we guarantee sort stability of the tuples we've
731 : : * already accumulated.
732 : : */
733 [ + + ]: 16951 : if (isCurrentGroup(node, node->group_pivot, slot))
734 : : {
735 : : /*
736 : : * As long as the prefix keys match the pivot tuple then
737 : : * load the tuple into the tuplesort.
738 : : */
739 : 15326 : tuplesort_puttupleslot(fullsort_state, slot);
740 : 15326 : nTuples++;
741 : : }
742 : : else
743 : : {
744 : : /*
745 : : * Since the tuple we fetched isn't part of the current
746 : : * prefix key group we don't want to sort it as part of
747 : : * the current batch. Instead we use the group_pivot slot
748 : : * to carry it over to the next batch (even though we
749 : : * won't actually treat it as a group pivot).
750 : : */
751 : 1625 : ExecCopySlot(node->group_pivot, slot);
752 : :
753 [ + + ]: 1625 : if (node->bounded)
754 : : {
755 : : /*
756 : : * If the current node has a bound, and we've already
757 : : * sorted n tuples, then the functional bound
758 : : * remaining is (original bound - n), so store the
759 : : * current number of processed tuples for later use
760 : : * configuring the sort state's bound.
761 : : */
762 : : SO2_printf("Changing bound_Done from " INT64_FORMAT " to " INT64_FORMAT "\n",
763 : : node->bound_Done,
764 : : Min(node->bound, node->bound_Done + nTuples));
765 : 100 : node->bound_Done = Min(node->bound, node->bound_Done + nTuples);
766 : : }
767 : :
768 : : /*
769 : : * Once we find changed prefix keys we can complete the
770 : : * sort and transition modes to reading out the sorted
771 : : * tuples.
772 : : */
773 : : SO1_printf("Sorting fullsort tuplesort with " INT64_FORMAT " tuples\n",
774 : : nTuples);
775 : 1625 : tuplesort_performsort(fullsort_state);
776 : :
777 [ + + - + : 1625 : INSTRUMENT_SORT_GROUP(node, fullsort);
- - ]
778 : :
779 : : SO_printf("Setting execution_status to INCSORT_READFULLSORT (found end of group)\n");
780 : 1625 : node->execution_status = INCSORT_READFULLSORT;
781 : 1625 : break;
782 : : }
783 : : }
784 : :
785 : : /*
786 : : * Unless we've already transitioned modes to reading from the
787 : : * full sort state, then we assume that having read at least
788 : : * DEFAULT_MAX_FULL_SORT_GROUP_SIZE tuples means it's likely we're
789 : : * processing a large group of tuples all having equal prefix keys
790 : : * (but haven't yet found the final tuple in that prefix key
791 : : * group), so we need to transition into presorted prefix mode.
792 : : */
793 [ + + ]: 74728 : if (nTuples > DEFAULT_MAX_FULL_SORT_GROUP_SIZE &&
794 [ + - ]: 247 : node->execution_status != INCSORT_READFULLSORT)
795 : : {
796 : : /*
797 : : * The group pivot we have stored has already been put into
798 : : * the tuplesort; we don't want to carry it over. Since we
799 : : * haven't yet found the end of the prefix key group, it might
800 : : * seem like we should keep this, but we don't actually know
801 : : * how many prefix key groups might be represented in the full
802 : : * sort state, so we'll let the mode transition function
803 : : * manage this state for us.
804 : : */
805 : 247 : ExecClearTuple(node->group_pivot);
806 : :
807 : : /*
808 : : * Unfortunately the tuplesort API doesn't include a way to
809 : : * retrieve tuples unless a sort has been performed, so we
810 : : * perform the sort even though we could just as easily rely
811 : : * on FIFO retrieval semantics when transferring them to the
812 : : * presorted prefix tuplesort.
813 : : */
814 : : SO1_printf("Sorting fullsort tuplesort with " INT64_FORMAT " tuples\n", nTuples);
815 : 247 : tuplesort_performsort(fullsort_state);
816 : :
817 [ + + - + : 247 : INSTRUMENT_SORT_GROUP(node, fullsort);
- - ]
818 : :
819 : : /*
820 : : * If the full sort tuplesort happened to switch into top-n
821 : : * heapsort mode then we will only be able to retrieve
822 : : * currentBound tuples (since the tuplesort will have only
823 : : * retained the top-n tuples). This is safe even though we
824 : : * haven't yet completed fetching the current prefix key group
825 : : * because the tuples we've "lost" already sorted "below" the
826 : : * retained ones, and we're already contractually guaranteed
827 : : * to not need any more than the currentBound tuples.
828 : : */
829 [ + + ]: 247 : if (tuplesort_used_bound(node->fullsort_state))
830 : : {
831 : 8 : int64 currentBound = node->bound - node->bound_Done;
832 : :
833 : : SO2_printf("Read " INT64_FORMAT " tuples, but setting to " INT64_FORMAT " because we used bounded sort\n",
834 : : nTuples, Min(currentBound, nTuples));
835 : 8 : nTuples = Min(currentBound, nTuples);
836 : : }
837 : :
838 : : SO1_printf("Setting n_fullsort_remaining to " INT64_FORMAT " and calling switchToPresortedPrefixMode()\n",
839 : : nTuples);
840 : :
841 : : /*
842 : : * We might have multiple prefix key groups in the full sort
843 : : * state, so the mode transition function needs to know that
844 : : * it needs to move from the fullsort to presorted prefix
845 : : * sort.
846 : : */
847 : 247 : node->n_fullsort_remaining = nTuples;
848 : :
849 : : /* Transition the tuples to the presorted prefix tuplesort. */
850 : 247 : switchToPresortedPrefixMode(pstate);
851 : :
852 : : /*
853 : : * Since we know we had tuples to move to the presorted prefix
854 : : * tuplesort, we know that unless that transition has verified
855 : : * that all tuples belonged to the same prefix key group (in
856 : : * which case we can go straight to continuing to load tuples
857 : : * into that tuplesort), we should have a tuple to return
858 : : * here.
859 : : *
860 : : * Either way, the appropriate execution status should have
861 : : * been set by switchToPresortedPrefixMode(), so we can drop
862 : : * out of the loop here and let the appropriate path kick in.
863 : : */
864 : 247 : break;
865 : : }
866 : : }
867 : : }
868 : :
869 [ + + ]: 2450 : if (node->execution_status == INCSORT_LOADPREFIXSORT)
870 : : {
871 : : /*
872 : : * We only enter this state after the mode transition function has
873 : : * confirmed all remaining tuples from the full sort state have the
874 : : * same prefix and moved those tuples to the prefix sort state. That
875 : : * function has also set a group pivot tuple (which doesn't need to be
876 : : * carried over; it's already been put into the prefix sort state).
877 : : */
878 : : Assert(!TupIsNull(node->group_pivot));
879 : :
880 : : /*
881 : : * Read tuples from the outer node and load them into the prefix sort
882 : : * state until we encounter a tuple whose prefix keys don't match the
883 : : * current group_pivot tuple, since we can't guarantee sort stability
884 : : * until we have all tuples matching those prefix keys.
885 : : */
886 : : for (;;)
887 : : {
888 : 299603 : slot = ExecProcNode(outerNode);
889 : :
890 : : /*
891 : : * If we've exhausted tuples from the outer node we're done
892 : : * loading the prefix sort state.
893 : : */
894 [ + + + + ]: 299603 : if (TupIsNull(slot))
895 : : {
896 : : /*
897 : : * We need to know later if the outer node has completed to be
898 : : * able to distinguish between being done with a batch and
899 : : * being done with the whole node.
900 : : */
901 : 53 : node->outerNodeDone = true;
902 : 53 : break;
903 : : }
904 : :
905 : : /*
906 : : * If the tuple's prefix keys match our pivot tuple, we're not
907 : : * done yet and can load it into the prefix sort state. If not, we
908 : : * don't want to sort it as part of the current batch. Instead we
909 : : * use the group_pivot slot to carry it over to the next batch
910 : : * (even though we won't actually treat it as a group pivot).
911 : : */
912 [ + + ]: 299550 : if (isCurrentGroup(node, node->group_pivot, slot))
913 : : {
914 : 299356 : tuplesort_puttupleslot(node->prefixsort_state, slot);
915 : 299356 : nTuples++;
916 : : }
917 : : else
918 : : {
919 : 194 : ExecCopySlot(node->group_pivot, slot);
920 : 194 : break;
921 : : }
922 : : }
923 : :
924 : : /*
925 : : * Perform the sort and begin returning the tuples to the parent plan
926 : : * node.
927 : : */
928 : : SO1_printf("Sorting presorted prefix tuplesort with " INT64_FORMAT " tuples\n", nTuples);
929 : 247 : tuplesort_performsort(node->prefixsort_state);
930 : :
931 [ + + - + : 247 : INSTRUMENT_SORT_GROUP(node, prefixsort);
- - ]
932 : :
933 : : SO_printf("Setting execution_status to INCSORT_READPREFIXSORT (found end of group)\n");
934 : 247 : node->execution_status = INCSORT_READPREFIXSORT;
935 : :
936 [ + + ]: 247 : if (node->bounded)
937 : : {
938 : : /*
939 : : * If the current node has a bound, and we've already sorted n
940 : : * tuples, then the functional bound remaining is (original bound
941 : : * - n), so store the current number of processed tuples for use
942 : : * in configuring sorting bound.
943 : : */
944 : : SO2_printf("Changing bound_Done from " INT64_FORMAT " to " INT64_FORMAT "\n",
945 : : node->bound_Done,
946 : : Min(node->bound, node->bound_Done + nTuples));
947 : 41 : node->bound_Done = Min(node->bound, node->bound_Done + nTuples);
948 : : }
949 : : }
950 : :
951 : : /* Restore to user specified direction. */
952 : 2450 : estate->es_direction = dir;
953 : :
954 : : /*
955 : : * Get the first or next tuple from tuplesort. Returns NULL if no more
956 : : * tuples.
957 : : */
958 : 4900 : read_sortstate = node->execution_status == INCSORT_READFULLSORT ?
959 [ + + ]: 2450 : fullsort_state : node->prefixsort_state;
960 : 2450 : slot = node->ss.ps.ps_ResultTupleSlot;
961 : 2450 : (void) tuplesort_gettupleslot(read_sortstate, ScanDirectionIsForward(dir),
962 : : false, slot, NULL);
963 : 2450 : return slot;
964 : : }
965 : :
966 : : /* ----------------------------------------------------------------
967 : : * ExecInitIncrementalSort
968 : : *
969 : : * Creates the run-time state information for the sort node
970 : : * produced by the planner and initializes its outer subtree.
971 : : * ----------------------------------------------------------------
972 : : */
973 : : IncrementalSortState *
974 : 816 : ExecInitIncrementalSort(IncrementalSort *node, EState *estate, int eflags)
975 : : {
976 : : IncrementalSortState *incrsortstate;
977 : :
978 : : SO_printf("ExecInitIncrementalSort: initializing sort node\n");
979 : :
980 : : /*
981 : : * Incremental sort can't be used with EXEC_FLAG_BACKWARD or
982 : : * EXEC_FLAG_MARK, because the current sort state contains only one sort
983 : : * batch rather than the full result set.
984 : : */
985 : : Assert((eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)) == 0);
986 : :
987 : : /* Initialize state structure. */
988 : 816 : incrsortstate = makeNode(IncrementalSortState);
989 : 816 : incrsortstate->ss.ps.plan = (Plan *) node;
990 : 816 : incrsortstate->ss.ps.state = estate;
991 : 816 : incrsortstate->ss.ps.ExecProcNode = ExecIncrementalSort;
992 : :
993 : 816 : incrsortstate->execution_status = INCSORT_LOADFULLSORT;
994 : 816 : incrsortstate->bounded = false;
995 : 816 : incrsortstate->outerNodeDone = false;
996 : 816 : incrsortstate->bound_Done = 0;
997 : 816 : incrsortstate->fullsort_state = NULL;
998 : 816 : incrsortstate->prefixsort_state = NULL;
999 : 816 : incrsortstate->group_pivot = NULL;
1000 : 816 : incrsortstate->transfer_tuple = NULL;
1001 : 816 : incrsortstate->n_fullsort_remaining = 0;
1002 : 816 : incrsortstate->presorted_keys = NULL;
1003 : :
1004 [ - + ]: 816 : if (incrsortstate->ss.ps.instrument != NULL)
1005 : : {
1006 : 0 : IncrementalSortGroupInfo *fullsortGroupInfo =
1007 : : &incrsortstate->incsort_info.fullsortGroupInfo;
1008 : 0 : IncrementalSortGroupInfo *prefixsortGroupInfo =
1009 : : &incrsortstate->incsort_info.prefixsortGroupInfo;
1010 : :
1011 : 0 : fullsortGroupInfo->groupCount = 0;
1012 : 0 : fullsortGroupInfo->maxDiskSpaceUsed = 0;
1013 : 0 : fullsortGroupInfo->totalDiskSpaceUsed = 0;
1014 : 0 : fullsortGroupInfo->maxMemorySpaceUsed = 0;
1015 : 0 : fullsortGroupInfo->totalMemorySpaceUsed = 0;
1016 : 0 : fullsortGroupInfo->sortMethods = 0;
1017 : 0 : prefixsortGroupInfo->groupCount = 0;
1018 : 0 : prefixsortGroupInfo->maxDiskSpaceUsed = 0;
1019 : 0 : prefixsortGroupInfo->totalDiskSpaceUsed = 0;
1020 : 0 : prefixsortGroupInfo->maxMemorySpaceUsed = 0;
1021 : 0 : prefixsortGroupInfo->totalMemorySpaceUsed = 0;
1022 : 0 : prefixsortGroupInfo->sortMethods = 0;
1023 : : }
1024 : :
1025 : : /*
1026 : : * Miscellaneous initialization
1027 : : *
1028 : : * Sort nodes don't initialize their ExprContexts because they never call
1029 : : * ExecQual or ExecProject.
1030 : : */
1031 : :
1032 : : /*
1033 : : * Initialize child nodes.
1034 : : *
1035 : : * Incremental sort does not support backwards scans and mark/restore, so
1036 : : * we don't bother removing the flags from eflags here. We allow passing a
1037 : : * REWIND flag, because although incremental sort can't use it, the child
1038 : : * nodes may be able to do something more useful.
1039 : : */
1040 : 816 : outerPlanState(incrsortstate) = ExecInitNode(outerPlan(node), estate, eflags);
1041 : :
1042 : : /*
1043 : : * Initialize scan slot and type.
1044 : : */
1045 : 816 : ExecCreateScanSlotFromOuterPlan(estate, &incrsortstate->ss, &TTSOpsMinimalTuple);
1046 : :
1047 : : /*
1048 : : * Initialize return slot and type. No need to initialize projection info
1049 : : * because we don't do any projections.
1050 : : */
1051 : 816 : ExecInitResultTupleSlotTL(&incrsortstate->ss.ps, &TTSOpsMinimalTuple);
1052 : 816 : incrsortstate->ss.ps.ps_ProjInfo = NULL;
1053 : :
1054 : : /*
1055 : : * Initialize standalone slots to store a tuple for pivot prefix keys and
1056 : : * for carrying over a tuple from one batch to the next.
1057 : : */
1058 : 816 : incrsortstate->group_pivot =
1059 : 816 : MakeSingleTupleTableSlot(ExecGetResultType(outerPlanState(incrsortstate)),
1060 : : &TTSOpsMinimalTuple);
1061 : 816 : incrsortstate->transfer_tuple =
1062 : 816 : MakeSingleTupleTableSlot(ExecGetResultType(outerPlanState(incrsortstate)),
1063 : : &TTSOpsMinimalTuple);
1064 : :
1065 : : SO_printf("ExecInitIncrementalSort: sort node initialized\n");
1066 : :
1067 : 816 : return incrsortstate;
1068 : : }
1069 : :
1070 : : /* ----------------------------------------------------------------
1071 : : * ExecEndIncrementalSort(node)
1072 : : * ----------------------------------------------------------------
1073 : : */
1074 : : void
1075 : 816 : ExecEndIncrementalSort(IncrementalSortState *node)
1076 : : {
1077 : : SO_printf("ExecEndIncrementalSort: shutting down sort node\n");
1078 : :
1079 : 816 : ExecDropSingleTupleTableSlot(node->group_pivot);
1080 : 816 : ExecDropSingleTupleTableSlot(node->transfer_tuple);
1081 : :
1082 : : /*
1083 : : * Release tuplesort resources.
1084 : : */
1085 [ + + ]: 816 : if (node->fullsort_state != NULL)
1086 : : {
1087 : 580 : tuplesort_end(node->fullsort_state);
1088 : 580 : node->fullsort_state = NULL;
1089 : : }
1090 [ + + ]: 816 : if (node->prefixsort_state != NULL)
1091 : : {
1092 : 74 : tuplesort_end(node->prefixsort_state);
1093 : 74 : node->prefixsort_state = NULL;
1094 : : }
1095 : :
1096 : : /*
1097 : : * Shut down the subplan.
1098 : : */
1099 : 816 : ExecEndNode(outerPlanState(node));
1100 : :
1101 : : SO_printf("ExecEndIncrementalSort: sort node shutdown\n");
1102 : 816 : }
1103 : :
1104 : : void
1105 : 8 : ExecReScanIncrementalSort(IncrementalSortState *node)
1106 : : {
1107 : 8 : PlanState *outerPlan = outerPlanState(node);
1108 : :
1109 : : /*
1110 : : * Incremental sort doesn't support efficient rescan even when parameters
1111 : : * haven't changed (e.g., rewind) because unlike regular sort we don't
1112 : : * store all tuples at once for the full sort.
1113 : : *
1114 : : * So even if EXEC_FLAG_REWIND is set we just reset all of our state and
1115 : : * re-execute the sort along with the child node. Incremental sort itself
1116 : : * can't do anything smarter, but maybe the child nodes can.
1117 : : *
1118 : : * In theory if we've only filled the full sort with one batch (and
1119 : : * haven't reset it for a new batch yet) then we could efficiently rewind,
1120 : : * but that seems a narrow enough case that it's not worth handling
1121 : : * specially at this time.
1122 : : */
1123 : :
1124 : : /* must drop pointer to sort result tuple */
1125 : 8 : ExecClearTuple(node->ss.ps.ps_ResultTupleSlot);
1126 : :
1127 [ + - ]: 8 : if (node->group_pivot != NULL)
1128 : 8 : ExecClearTuple(node->group_pivot);
1129 [ + - ]: 8 : if (node->transfer_tuple != NULL)
1130 : 8 : ExecClearTuple(node->transfer_tuple);
1131 : :
1132 : 8 : node->outerNodeDone = false;
1133 : 8 : node->n_fullsort_remaining = 0;
1134 : 8 : node->bound_Done = 0;
1135 : :
1136 : 8 : node->execution_status = INCSORT_LOADFULLSORT;
1137 : :
1138 : : /*
1139 : : * If we've set up either of the sort states yet, we need to reset them.
1140 : : * We could end them and null out the pointers, but there's no reason to
1141 : : * repay the setup cost, and because ExecIncrementalSort guards presorted
1142 : : * column functions by checking to see if the full sort state has been
1143 : : * initialized yet, setting the sort states to null here might actually
1144 : : * cause a leak.
1145 : : */
1146 [ + + ]: 8 : if (node->fullsort_state != NULL)
1147 : 4 : tuplesort_reset(node->fullsort_state);
1148 [ + + ]: 8 : if (node->prefixsort_state != NULL)
1149 : 4 : tuplesort_reset(node->prefixsort_state);
1150 : :
1151 : : /*
1152 : : * If chgParam of subnode is not null, then the plan will be re-scanned by
1153 : : * the first ExecProcNode.
1154 : : */
1155 [ + - ]: 8 : if (outerPlan->chgParam == NULL)
1156 : 8 : ExecReScan(outerPlan);
1157 : 8 : }
1158 : :
1159 : : /* ----------------------------------------------------------------
1160 : : * Parallel Query Support
1161 : : * ----------------------------------------------------------------
1162 : : */
1163 : :
1164 : : /* ----------------------------------------------------------------
1165 : : * ExecSortEstimate
1166 : : *
1167 : : * Estimate space required to propagate sort statistics.
1168 : : * ----------------------------------------------------------------
1169 : : */
1170 : : void
1171 : 0 : ExecIncrementalSortEstimate(IncrementalSortState *node, ParallelContext *pcxt)
1172 : : {
1173 : : Size size;
1174 : :
1175 : : /* don't need this if not instrumenting or no workers */
1176 [ # # # # ]: 0 : if (!node->ss.ps.instrument || pcxt->nworkers == 0)
1177 : 0 : return;
1178 : :
1179 : 0 : size = mul_size(pcxt->nworkers, sizeof(IncrementalSortInfo));
1180 : 0 : size = add_size(size, offsetof(SharedIncrementalSortInfo, sinfo));
1181 : 0 : shm_toc_estimate_chunk(&pcxt->estimator, size);
1182 : 0 : shm_toc_estimate_keys(&pcxt->estimator, 1);
1183 : : }
1184 : :
1185 : : /* ----------------------------------------------------------------
1186 : : * ExecSortInitializeDSM
1187 : : *
1188 : : * Initialize DSM space for sort statistics.
1189 : : * ----------------------------------------------------------------
1190 : : */
1191 : : void
1192 : 0 : ExecIncrementalSortInitializeDSM(IncrementalSortState *node, ParallelContext *pcxt)
1193 : : {
1194 : : Size size;
1195 : :
1196 : : /* don't need this if not instrumenting or no workers */
1197 [ # # # # ]: 0 : if (!node->ss.ps.instrument || pcxt->nworkers == 0)
1198 : 0 : return;
1199 : :
1200 : 0 : size = offsetof(SharedIncrementalSortInfo, sinfo)
1201 : 0 : + pcxt->nworkers * sizeof(IncrementalSortInfo);
1202 : 0 : node->shared_info = shm_toc_allocate(pcxt->toc, size);
1203 : : /* ensure any unfilled slots will contain zeroes */
1204 : 0 : memset(node->shared_info, 0, size);
1205 : 0 : node->shared_info->num_workers = pcxt->nworkers;
1206 : 0 : shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id,
1207 : 0 : node->shared_info);
1208 : : }
1209 : :
1210 : : /* ----------------------------------------------------------------
1211 : : * ExecSortInitializeWorker
1212 : : *
1213 : : * Attach worker to DSM space for sort statistics.
1214 : : * ----------------------------------------------------------------
1215 : : */
1216 : : void
1217 : 0 : ExecIncrementalSortInitializeWorker(IncrementalSortState *node, ParallelWorkerContext *pwcxt)
1218 : : {
1219 : 0 : node->shared_info =
1220 : 0 : shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, true);
1221 : 0 : node->am_worker = true;
1222 : 0 : }
1223 : :
1224 : : /* ----------------------------------------------------------------
1225 : : * ExecSortRetrieveInstrumentation
1226 : : *
1227 : : * Transfer sort statistics from DSM to private memory.
1228 : : * ----------------------------------------------------------------
1229 : : */
1230 : : void
1231 : 0 : ExecIncrementalSortRetrieveInstrumentation(IncrementalSortState *node)
1232 : : {
1233 : : Size size;
1234 : : SharedIncrementalSortInfo *si;
1235 : :
1236 [ # # ]: 0 : if (node->shared_info == NULL)
1237 : 0 : return;
1238 : :
1239 : 0 : size = offsetof(SharedIncrementalSortInfo, sinfo)
1240 : 0 : + node->shared_info->num_workers * sizeof(IncrementalSortInfo);
1241 : 0 : si = palloc(size);
1242 : 0 : memcpy(si, node->shared_info, size);
1243 : 0 : node->shared_info = si;
1244 : : }
|
