TLA Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * nodeRecursiveunion.c
4 : * routines to handle RecursiveUnion nodes.
5 : *
6 : * To implement UNION (without ALL), we need a hashtable that stores tuples
7 : * already seen. The hash key is computed from the grouping columns.
8 : *
9 : *
10 : * Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
11 : * Portions Copyright (c) 1994, Regents of the University of California
12 : *
13 : *
14 : * IDENTIFICATION
15 : * src/backend/executor/nodeRecursiveunion.c
16 : *
17 : *-------------------------------------------------------------------------
18 : */
19 : #include "postgres.h"
20 :
21 : #include "executor/execdebug.h"
22 : #include "executor/nodeRecursiveunion.h"
23 : #include "miscadmin.h"
24 : #include "utils/memutils.h"
25 :
26 :
27 :
28 : /*
29 : * Initialize the hash table to empty.
30 : */
31 : static void
32 CBC 144 : build_hash_table(RecursiveUnionState *rustate)
33 : {
34 144 : RecursiveUnion *node = (RecursiveUnion *) rustate->ps.plan;
35 144 : TupleDesc desc = ExecGetResultType(outerPlanState(rustate));
36 :
37 144 : Assert(node->numCols > 0);
38 144 : Assert(node->numGroups > 0);
39 :
40 288 : rustate->hashtable = BuildTupleHashTableExt(&rustate->ps,
41 : desc,
42 : node->numCols,
43 : node->dupColIdx,
44 144 : rustate->eqfuncoids,
45 : rustate->hashfunctions,
46 : node->dupCollations,
47 : node->numGroups,
48 : 0,
49 144 : rustate->ps.state->es_query_cxt,
50 : rustate->tableContext,
51 : rustate->tempContext,
52 : false);
53 144 : }
54 :
55 :
56 : /* ----------------------------------------------------------------
57 : * ExecRecursiveUnion(node)
58 : *
59 : * Scans the recursive query sequentially and returns the next
60 : * qualifying tuple.
61 : *
62 : * 1. evaluate non recursive term and assign the result to RT
63 : *
64 : * 2. execute recursive terms
65 : *
66 : * 2.1 WT := RT
67 : * 2.2 while WT is not empty repeat 2.3 to 2.6. if WT is empty returns RT
68 : * 2.3 replace the name of recursive term with WT
69 : * 2.4 evaluate the recursive term and store into WT
70 : * 2.5 append WT to RT
71 : * 2.6 go back to 2.2
72 : * ----------------------------------------------------------------
73 : */
74 : static TupleTableSlot *
75 18189 : ExecRecursiveUnion(PlanState *pstate)
76 : {
77 18189 : RecursiveUnionState *node = castNode(RecursiveUnionState, pstate);
78 18189 : PlanState *outerPlan = outerPlanState(node);
79 18189 : PlanState *innerPlan = innerPlanState(node);
80 18189 : RecursiveUnion *plan = (RecursiveUnion *) node->ps.plan;
81 : TupleTableSlot *slot;
82 : bool isnew;
83 :
84 18189 : CHECK_FOR_INTERRUPTS();
85 :
86 : /* 1. Evaluate non-recursive term */
87 18189 : if (!node->recursing)
88 : {
89 : for (;;)
90 : {
91 3763 : slot = ExecProcNode(outerPlan);
92 3763 : if (TupIsNull(slot))
93 : break;
94 3433 : if (plan->numCols > 0)
95 : {
96 : /* Find or build hashtable entry for this tuple's group */
97 243 : LookupTupleHashEntry(node->hashtable, slot, &isnew, NULL);
98 : /* Must reset temp context after each hashtable lookup */
99 243 : MemoryContextReset(node->tempContext);
100 : /* Ignore tuple if already seen */
101 243 : if (!isnew)
102 4 : continue;
103 : }
104 : /* Each non-duplicate tuple goes to the working table ... */
105 3429 : tuplestore_puttupleslot(node->working_table, slot);
106 : /* ... and to the caller */
107 3429 : return slot;
108 : }
109 330 : node->recursing = true;
110 : }
111 :
112 : /* 2. Execute recursive term */
113 : for (;;)
114 : {
115 17892 : slot = ExecProcNode(innerPlan);
116 17892 : if (TupIsNull(slot))
117 : {
118 : /* Done if there's nothing in the intermediate table */
119 3407 : if (node->intermediate_empty)
120 309 : break;
121 :
122 : /* done with old working table ... */
123 3098 : tuplestore_end(node->working_table);
124 :
125 : /* intermediate table becomes working table */
126 3098 : node->working_table = node->intermediate_table;
127 :
128 : /* create new empty intermediate table */
129 3098 : node->intermediate_table = tuplestore_begin_heap(false, false,
130 : work_mem);
131 3098 : node->intermediate_empty = true;
132 :
133 : /* reset the recursive term */
134 3098 : innerPlan->chgParam = bms_add_member(innerPlan->chgParam,
135 : plan->wtParam);
136 :
137 : /* and continue fetching from recursive term */
138 3098 : continue;
139 : }
140 :
141 14485 : if (plan->numCols > 0)
142 : {
143 : /* Find or build hashtable entry for this tuple's group */
144 438 : LookupTupleHashEntry(node->hashtable, slot, &isnew, NULL);
145 : /* Must reset temp context after each hashtable lookup */
146 438 : MemoryContextReset(node->tempContext);
147 : /* Ignore tuple if already seen */
148 438 : if (!isnew)
149 34 : continue;
150 : }
151 :
152 : /* Else, tuple is good; stash it in intermediate table ... */
153 14451 : node->intermediate_empty = false;
154 14451 : tuplestore_puttupleslot(node->intermediate_table, slot);
155 : /* ... and return it */
156 14451 : return slot;
157 : }
158 :
159 309 : return NULL;
160 : }
161 :
162 : /* ----------------------------------------------------------------
163 : * ExecInitRecursiveUnion
164 : * ----------------------------------------------------------------
165 : */
166 : RecursiveUnionState *
167 354 : ExecInitRecursiveUnion(RecursiveUnion *node, EState *estate, int eflags)
168 : {
169 : RecursiveUnionState *rustate;
170 : ParamExecData *prmdata;
171 :
172 : /* check for unsupported flags */
173 354 : Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
174 :
175 : /*
176 : * create state structure
177 : */
178 354 : rustate = makeNode(RecursiveUnionState);
179 354 : rustate->ps.plan = (Plan *) node;
180 354 : rustate->ps.state = estate;
181 354 : rustate->ps.ExecProcNode = ExecRecursiveUnion;
182 :
183 354 : rustate->eqfuncoids = NULL;
184 354 : rustate->hashfunctions = NULL;
185 354 : rustate->hashtable = NULL;
186 354 : rustate->tempContext = NULL;
187 354 : rustate->tableContext = NULL;
188 :
189 : /* initialize processing state */
190 354 : rustate->recursing = false;
191 354 : rustate->intermediate_empty = true;
192 354 : rustate->working_table = tuplestore_begin_heap(false, false, work_mem);
193 354 : rustate->intermediate_table = tuplestore_begin_heap(false, false, work_mem);
194 :
195 : /*
196 : * If hashing, we need a per-tuple memory context for comparisons, and a
197 : * longer-lived context to store the hash table. The table can't just be
198 : * kept in the per-query context because we want to be able to throw it
199 : * away when rescanning.
200 : */
201 354 : if (node->numCols > 0)
202 : {
203 144 : rustate->tempContext =
204 144 : AllocSetContextCreate(CurrentMemoryContext,
205 : "RecursiveUnion",
206 : ALLOCSET_DEFAULT_SIZES);
207 144 : rustate->tableContext =
208 144 : AllocSetContextCreate(CurrentMemoryContext,
209 : "RecursiveUnion hash table",
210 : ALLOCSET_DEFAULT_SIZES);
211 : }
212 :
213 : /*
214 : * Make the state structure available to descendant WorkTableScan nodes
215 : * via the Param slot reserved for it.
216 : */
217 354 : prmdata = &(estate->es_param_exec_vals[node->wtParam]);
218 354 : Assert(prmdata->execPlan == NULL);
219 354 : prmdata->value = PointerGetDatum(rustate);
220 354 : prmdata->isnull = false;
221 :
222 : /*
223 : * Miscellaneous initialization
224 : *
225 : * RecursiveUnion plans don't have expression contexts because they never
226 : * call ExecQual or ExecProject.
227 : */
228 354 : Assert(node->plan.qual == NIL);
229 :
230 : /*
231 : * RecursiveUnion nodes still have Result slots, which hold pointers to
232 : * tuples, so we have to initialize them.
233 : */
234 354 : ExecInitResultTypeTL(&rustate->ps);
235 :
236 : /*
237 : * Initialize result tuple type. (Note: we have to set up the result type
238 : * before initializing child nodes, because nodeWorktablescan.c expects it
239 : * to be valid.)
240 : */
241 354 : rustate->ps.ps_ProjInfo = NULL;
242 :
243 : /*
244 : * initialize child nodes
245 : */
246 354 : outerPlanState(rustate) = ExecInitNode(outerPlan(node), estate, eflags);
247 354 : innerPlanState(rustate) = ExecInitNode(innerPlan(node), estate, eflags);
248 :
249 : /*
250 : * If hashing, precompute fmgr lookup data for inner loop, and create the
251 : * hash table.
252 : */
253 354 : if (node->numCols > 0)
254 : {
255 144 : execTuplesHashPrepare(node->numCols,
256 144 : node->dupOperators,
257 : &rustate->eqfuncoids,
258 : &rustate->hashfunctions);
259 144 : build_hash_table(rustate);
260 : }
261 :
262 354 : return rustate;
263 : }
264 :
265 : /* ----------------------------------------------------------------
266 : * ExecEndRecursiveUnion
267 : *
268 : * frees any storage allocated through C routines.
269 : * ----------------------------------------------------------------
270 : */
271 : void
272 354 : ExecEndRecursiveUnion(RecursiveUnionState *node)
273 : {
274 : /* Release tuplestores */
275 354 : tuplestore_end(node->working_table);
276 354 : tuplestore_end(node->intermediate_table);
277 :
278 : /* free subsidiary stuff including hashtable */
279 354 : if (node->tempContext)
280 144 : MemoryContextDelete(node->tempContext);
281 354 : if (node->tableContext)
282 144 : MemoryContextDelete(node->tableContext);
283 :
284 : /*
285 : * close down subplans
286 : */
287 354 : ExecEndNode(outerPlanState(node));
288 354 : ExecEndNode(innerPlanState(node));
289 354 : }
290 :
291 : /* ----------------------------------------------------------------
292 : * ExecReScanRecursiveUnion
293 : *
294 : * Rescans the relation.
295 : * ----------------------------------------------------------------
296 : */
297 : void
298 6 : ExecReScanRecursiveUnion(RecursiveUnionState *node)
299 : {
300 6 : PlanState *outerPlan = outerPlanState(node);
301 6 : PlanState *innerPlan = innerPlanState(node);
302 6 : RecursiveUnion *plan = (RecursiveUnion *) node->ps.plan;
303 :
304 : /*
305 : * Set recursive term's chgParam to tell it that we'll modify the working
306 : * table and therefore it has to rescan.
307 : */
308 6 : innerPlan->chgParam = bms_add_member(innerPlan->chgParam, plan->wtParam);
309 :
310 : /*
311 : * if chgParam of subnode is not null then plan will be re-scanned by
312 : * first ExecProcNode. Because of above, we only have to do this to the
313 : * non-recursive term.
314 : */
315 6 : if (outerPlan->chgParam == NULL)
316 UBC 0 : ExecReScan(outerPlan);
317 :
318 : /* Release any hashtable storage */
319 CBC 6 : if (node->tableContext)
320 6 : MemoryContextResetAndDeleteChildren(node->tableContext);
321 :
322 : /* Empty hashtable if needed */
323 6 : if (plan->numCols > 0)
324 6 : ResetTupleHashTable(node->hashtable);
325 :
326 : /* reset processing state */
327 6 : node->recursing = false;
328 6 : node->intermediate_empty = true;
329 6 : tuplestore_clear(node->working_table);
330 6 : tuplestore_clear(node->intermediate_table);
331 6 : }
|
