Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * nodeMergeAppend.c
4 : : * routines to handle MergeAppend nodes.
5 : : *
6 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : *
10 : : * IDENTIFICATION
11 : : * src/backend/executor/nodeMergeAppend.c
12 : : *
13 : : *-------------------------------------------------------------------------
14 : : */
15 : : /*
16 : : * INTERFACE ROUTINES
17 : : * ExecInitMergeAppend - initialize the MergeAppend node
18 : : * ExecMergeAppend - retrieve the next tuple from the node
19 : : * ExecEndMergeAppend - shut down the MergeAppend node
20 : : * ExecReScanMergeAppend - rescan the MergeAppend node
21 : : *
22 : : * NOTES
23 : : * A MergeAppend node contains a list of one or more subplans.
24 : : * These are each expected to deliver tuples that are sorted according
25 : : * to a common sort key. The MergeAppend node merges these streams
26 : : * to produce output sorted the same way.
27 : : *
28 : : * MergeAppend nodes don't make use of their left and right
29 : : * subtrees, rather they maintain a list of subplans so
30 : : * a typical MergeAppend node looks like this in the plan tree:
31 : : *
32 : : * ...
33 : : * /
34 : : * MergeAppend---+------+------+--- nil
35 : : * / \ | | |
36 : : * nil nil ... ... ...
37 : : * subplans
38 : : */
39 : :
40 : : #include "postgres.h"
41 : :
42 : : #include "executor/executor.h"
43 : : #include "executor/execPartition.h"
44 : : #include "executor/nodeMergeAppend.h"
45 : : #include "lib/binaryheap.h"
46 : : #include "miscadmin.h"
47 : : #include "utils/sortsupport.h"
48 : :
49 : : /*
50 : : * We have one slot for each item in the heap array. We use SlotNumber
51 : : * to store slot indexes. This doesn't actually provide any formal
52 : : * type-safety, but it makes the code more self-documenting.
53 : : */
54 : : typedef int32 SlotNumber;
55 : :
56 : : static TupleTableSlot *ExecMergeAppend(PlanState *pstate);
57 : : static int heap_compare_slots(Datum a, Datum b, void *arg);
58 : :
59 : :
60 : : /* ----------------------------------------------------------------
61 : : * ExecInitMergeAppend
62 : : *
63 : : * Begin all of the subscans of the MergeAppend node.
64 : : * ----------------------------------------------------------------
65 : : */
66 : : MergeAppendState *
67 : 390 : ExecInitMergeAppend(MergeAppend *node, EState *estate, int eflags)
68 : : {
69 : 390 : MergeAppendState *mergestate = makeNode(MergeAppendState);
70 : : PlanState **mergeplanstates;
71 : : const TupleTableSlotOps *mergeops;
72 : : Bitmapset *validsubplans;
73 : : int nplans;
74 : : int i,
75 : : j;
76 : :
77 : : /* check for unsupported flags */
78 : : Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
79 : :
80 : : /*
81 : : * create new MergeAppendState for our node
82 : : */
83 : 390 : mergestate->ps.plan = (Plan *) node;
84 : 390 : mergestate->ps.state = estate;
85 : 390 : mergestate->ps.ExecProcNode = ExecMergeAppend;
86 : :
87 : : /* If run-time partition pruning is enabled, then set that up now */
88 [ + + ]: 390 : if (node->part_prune_index >= 0)
89 : : {
90 : : PartitionPruneState *prunestate;
91 : :
92 : : /*
93 : : * Set up pruning data structure. This also initializes the set of
94 : : * subplans to initialize (validsubplans) by taking into account the
95 : : * result of performing initial pruning if any.
96 : : */
97 : 44 : prunestate = ExecInitPartitionExecPruning(&mergestate->ps,
98 : 44 : list_length(node->mergeplans),
99 : : node->part_prune_index,
100 : : node->apprelids,
101 : : &validsubplans);
102 : 44 : mergestate->ms_prune_state = prunestate;
103 : 44 : nplans = bms_num_members(validsubplans);
104 : :
105 : : /*
106 : : * When no run-time pruning is required and there's at least one
107 : : * subplan, we can fill ms_valid_subplans immediately, preventing
108 : : * later calls to ExecFindMatchingSubPlans.
109 : : */
110 [ + + + + ]: 44 : if (!prunestate->do_exec_prune && nplans > 0)
111 : 24 : mergestate->ms_valid_subplans = bms_add_range(NULL, 0, nplans - 1);
112 : : }
113 : : else
114 : : {
115 : 346 : nplans = list_length(node->mergeplans);
116 : :
117 : : /*
118 : : * When run-time partition pruning is not enabled we can just mark all
119 : : * subplans as valid; they must also all be initialized.
120 : : */
121 : : Assert(nplans > 0);
122 : 346 : mergestate->ms_valid_subplans = validsubplans =
123 : 346 : bms_add_range(NULL, 0, nplans - 1);
124 : 346 : mergestate->ms_prune_state = NULL;
125 : : }
126 : :
127 : 390 : mergeplanstates = palloc_array(PlanState *, nplans);
128 : 390 : mergestate->mergeplans = mergeplanstates;
129 : 390 : mergestate->ms_nplans = nplans;
130 : :
131 : 390 : mergestate->ms_slots = palloc0_array(TupleTableSlot *, nplans);
132 : 390 : mergestate->ms_heap = binaryheap_allocate(nplans, heap_compare_slots,
133 : : mergestate);
134 : :
135 : : /*
136 : : * call ExecInitNode on each of the valid plans to be executed and save
137 : : * the results into the mergeplanstates array.
138 : : */
139 : 390 : j = 0;
140 : 390 : i = -1;
141 [ + + ]: 1465 : while ((i = bms_next_member(validsubplans, i)) >= 0)
142 : : {
143 : 1075 : Plan *initNode = (Plan *) list_nth(node->mergeplans, i);
144 : :
145 : 1075 : mergeplanstates[j++] = ExecInitNode(initNode, estate, eflags);
146 : : }
147 : :
148 : : /*
149 : : * Initialize MergeAppend's result tuple type and slot. If the child
150 : : * plans all produce the same fixed slot type, we can use that slot type;
151 : : * otherwise make a virtual slot. (Note that the result slot itself is
152 : : * used only to return a null tuple at end of execution; real tuples are
153 : : * returned to the caller in the children's own result slots. What we are
154 : : * doing here is allowing the parent plan node to optimize if the
155 : : * MergeAppend will return only one kind of slot.)
156 : : */
157 : 390 : mergeops = ExecGetCommonSlotOps(mergeplanstates, j);
158 [ + + ]: 390 : if (mergeops != NULL)
159 : : {
160 : 289 : ExecInitResultTupleSlotTL(&mergestate->ps, mergeops);
161 : : }
162 : : else
163 : : {
164 : 101 : ExecInitResultTupleSlotTL(&mergestate->ps, &TTSOpsVirtual);
165 : : /* show that the output slot type is not fixed */
166 : 101 : mergestate->ps.resultopsset = true;
167 : 101 : mergestate->ps.resultopsfixed = false;
168 : : }
169 : :
170 : : /*
171 : : * Miscellaneous initialization
172 : : */
173 : 390 : mergestate->ps.ps_ProjInfo = NULL;
174 : :
175 : : /*
176 : : * initialize sort-key information
177 : : */
178 : 390 : mergestate->ms_nkeys = node->numCols;
179 : 390 : mergestate->ms_sortkeys = palloc0_array(SortSupportData, node->numCols);
180 : :
181 [ + + ]: 849 : for (i = 0; i < node->numCols; i++)
182 : : {
183 : 459 : SortSupport sortKey = mergestate->ms_sortkeys + i;
184 : :
185 : 459 : sortKey->ssup_cxt = CurrentMemoryContext;
186 : 459 : sortKey->ssup_collation = node->collations[i];
187 : 459 : sortKey->ssup_nulls_first = node->nullsFirst[i];
188 : 459 : sortKey->ssup_attno = node->sortColIdx[i];
189 : :
190 : : /*
191 : : * It isn't feasible to perform abbreviated key conversion, since
192 : : * tuples are pulled into mergestate's binary heap as needed. It
193 : : * would likely be counter-productive to convert tuples into an
194 : : * abbreviated representation as they're pulled up, so opt out of that
195 : : * additional optimization entirely.
196 : : */
197 : 459 : sortKey->abbreviate = false;
198 : :
199 : 459 : PrepareSortSupportFromOrderingOp(node->sortOperators[i], sortKey);
200 : : }
201 : :
202 : : /*
203 : : * initialize to show we have not run the subplans yet
204 : : */
205 : 390 : mergestate->ms_initialized = false;
206 : :
207 : 390 : return mergestate;
208 : : }
209 : :
210 : : /* ----------------------------------------------------------------
211 : : * ExecMergeAppend
212 : : *
213 : : * Handles iteration over multiple subplans.
214 : : * ----------------------------------------------------------------
215 : : */
216 : : static TupleTableSlot *
217 : 107937 : ExecMergeAppend(PlanState *pstate)
218 : : {
219 : 107937 : MergeAppendState *node = castNode(MergeAppendState, pstate);
220 : : TupleTableSlot *result;
221 : : SlotNumber i;
222 : :
223 [ - + ]: 107937 : CHECK_FOR_INTERRUPTS();
224 : :
225 [ + + ]: 107937 : if (!node->ms_initialized)
226 : : {
227 : : /* Nothing to do if all subplans were pruned */
228 [ + + ]: 184 : if (node->ms_nplans == 0)
229 : 12 : return ExecClearTuple(node->ps.ps_ResultTupleSlot);
230 : :
231 : : /*
232 : : * If we've yet to determine the valid subplans then do so now. If
233 : : * run-time pruning is disabled then the valid subplans will always be
234 : : * set to all subplans.
235 : : */
236 [ + + ]: 172 : if (node->ms_valid_subplans == NULL)
237 : 8 : node->ms_valid_subplans =
238 : 8 : ExecFindMatchingSubPlans(node->ms_prune_state, false, NULL);
239 : :
240 : : /*
241 : : * First time through: pull the first tuple from each valid subplan,
242 : : * and set up the heap.
243 : : */
244 : 172 : i = -1;
245 [ + + ]: 661 : while ((i = bms_next_member(node->ms_valid_subplans, i)) >= 0)
246 : : {
247 : 489 : node->ms_slots[i] = ExecProcNode(node->mergeplans[i]);
248 [ + + + + ]: 489 : if (!TupIsNull(node->ms_slots[i]))
249 : 385 : binaryheap_add_unordered(node->ms_heap, Int32GetDatum(i));
250 : : }
251 : 172 : binaryheap_build(node->ms_heap);
252 : 172 : node->ms_initialized = true;
253 : : }
254 : : else
255 : : {
256 : : /*
257 : : * Otherwise, pull the next tuple from whichever subplan we returned
258 : : * from last time, and reinsert the subplan index into the heap,
259 : : * because it might now compare differently against the existing
260 : : * elements of the heap. (We could perhaps simplify the logic a bit
261 : : * by doing this before returning from the prior call, but it's better
262 : : * to not pull tuples until necessary.)
263 : : */
264 : 107753 : i = DatumGetInt32(binaryheap_first(node->ms_heap));
265 : 107753 : node->ms_slots[i] = ExecProcNode(node->mergeplans[i]);
266 [ + + + + ]: 107753 : if (!TupIsNull(node->ms_slots[i]))
267 : 107504 : binaryheap_replace_first(node->ms_heap, Int32GetDatum(i));
268 : : else
269 : 249 : (void) binaryheap_remove_first(node->ms_heap);
270 : : }
271 : :
272 [ + + ]: 107925 : if (binaryheap_empty(node->ms_heap))
273 : : {
274 : : /* All the subplans are exhausted, and so is the heap */
275 : 124 : result = ExecClearTuple(node->ps.ps_ResultTupleSlot);
276 : : }
277 : : else
278 : : {
279 : 107801 : i = DatumGetInt32(binaryheap_first(node->ms_heap));
280 : 107801 : result = node->ms_slots[i];
281 : : }
282 : :
283 : 107925 : return result;
284 : : }
285 : :
286 : : /*
287 : : * Compare the tuples in the two given slots.
288 : : */
289 : : static int32
290 : 98248 : heap_compare_slots(Datum a, Datum b, void *arg)
291 : : {
292 : 98248 : MergeAppendState *node = (MergeAppendState *) arg;
293 : 98248 : SlotNumber slot1 = DatumGetInt32(a);
294 : 98248 : SlotNumber slot2 = DatumGetInt32(b);
295 : :
296 : 98248 : TupleTableSlot *s1 = node->ms_slots[slot1];
297 : 98248 : TupleTableSlot *s2 = node->ms_slots[slot2];
298 : : int nkey;
299 : :
300 : : Assert(!TupIsNull(s1));
301 : : Assert(!TupIsNull(s2));
302 : :
303 [ + + ]: 130784 : for (nkey = 0; nkey < node->ms_nkeys; nkey++)
304 : : {
305 : 98248 : SortSupport sortKey = node->ms_sortkeys + nkey;
306 : 98248 : AttrNumber attno = sortKey->ssup_attno;
307 : : Datum datum1,
308 : : datum2;
309 : : bool isNull1,
310 : : isNull2;
311 : : int compare;
312 : :
313 : 98248 : datum1 = slot_getattr(s1, attno, &isNull1);
314 : 98248 : datum2 = slot_getattr(s2, attno, &isNull2);
315 : :
316 : 98248 : compare = ApplySortComparator(datum1, isNull1,
317 : : datum2, isNull2,
318 : : sortKey);
319 [ + + ]: 98248 : if (compare != 0)
320 : : {
321 [ + + ]: 65712 : INVERT_COMPARE_RESULT(compare);
322 : 65712 : return compare;
323 : : }
324 : : }
325 : 32536 : return 0;
326 : : }
327 : :
328 : : /* ----------------------------------------------------------------
329 : : * ExecEndMergeAppend
330 : : *
331 : : * Shuts down the subscans of the MergeAppend node.
332 : : *
333 : : * Returns nothing of interest.
334 : : * ----------------------------------------------------------------
335 : : */
336 : : void
337 : 390 : ExecEndMergeAppend(MergeAppendState *node)
338 : : {
339 : : PlanState **mergeplans;
340 : : int nplans;
341 : : int i;
342 : :
343 : : /*
344 : : * get information from the node
345 : : */
346 : 390 : mergeplans = node->mergeplans;
347 : 390 : nplans = node->ms_nplans;
348 : :
349 : : /*
350 : : * shut down each of the subscans
351 : : */
352 [ + + ]: 1465 : for (i = 0; i < nplans; i++)
353 : 1075 : ExecEndNode(mergeplans[i]);
354 : 390 : }
355 : :
356 : : void
357 : 12 : ExecReScanMergeAppend(MergeAppendState *node)
358 : : {
359 : : int i;
360 : :
361 : : /*
362 : : * If any PARAM_EXEC Params used in pruning expressions have changed, then
363 : : * we'd better unset the valid subplans so that they are reselected for
364 : : * the new parameter values.
365 : : */
366 [ - + - - ]: 12 : if (node->ms_prune_state &&
367 : 0 : bms_overlap(node->ps.chgParam,
368 : 0 : node->ms_prune_state->execparamids))
369 : : {
370 : 0 : bms_free(node->ms_valid_subplans);
371 : 0 : node->ms_valid_subplans = NULL;
372 : : }
373 : :
374 [ + + ]: 36 : for (i = 0; i < node->ms_nplans; i++)
375 : : {
376 : 24 : PlanState *subnode = node->mergeplans[i];
377 : :
378 : : /*
379 : : * ExecReScan doesn't know about my subplans, so I have to do
380 : : * changed-parameter signaling myself.
381 : : */
382 [ + - ]: 24 : if (node->ps.chgParam != NULL)
383 : 24 : UpdateChangedParamSet(subnode, node->ps.chgParam);
384 : :
385 : : /*
386 : : * If chgParam of subnode is not null then plan will be re-scanned by
387 : : * first ExecProcNode.
388 : : */
389 [ - + ]: 24 : if (subnode->chgParam == NULL)
390 : 0 : ExecReScan(subnode);
391 : : }
392 : 12 : binaryheap_reset(node->ms_heap);
393 : 12 : node->ms_initialized = false;
394 : 12 : }
|
