[フレーム]

nodeRecursiveunion.c

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 *

12 *

13 *

14 * IDENTIFICATION

15 * src/backend/executor/nodeRecursiveunion.c

16 *

17 *-------------------------------------------------------------------------

18 */

19#include "postgres.h"

20

21#include "executor/executor.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 */

31static void

32 build_hash_table(RecursiveUnionState *rustate)

33{

34 RecursiveUnion *node = (RecursiveUnion *) rustate->ps.plan;

35 TupleDesc desc = ExecGetResultType(outerPlanState(rustate));

36

37 Assert(node->numCols > 0);

38 Assert(node->numGroups > 0);

39

40 /*

41 * If both child plans deliver the same fixed tuple slot type, we can tell

42 * BuildTupleHashTable to expect that slot type as input. Otherwise,

43 * we'll pass NULL denoting that any slot type is possible.

44 */

45 rustate->hashtable = BuildTupleHashTable(&rustate->ps,

46 desc,

47 ExecGetCommonChildSlotOps(&rustate->ps),

48 node->numCols,

49 node->dupColIdx,

50 rustate->eqfuncoids,

51 rustate->hashfunctions,

52 node->dupCollations,

53 node->numGroups,

54 0,

55 rustate->ps.state->es_query_cxt,

56 rustate->tableContext,

57 rustate->tempContext,

58 false);

59}

60

61

62/* ----------------------------------------------------------------

63 * ExecRecursiveUnion(node)

64 *

65 * Scans the recursive query sequentially and returns the next

66 * qualifying tuple.

67 *

68 * 1. evaluate non recursive term and assign the result to RT

69 *

70 * 2. execute recursive terms

71 *

72 * 2.1 WT := RT

73 * 2.2 while WT is not empty repeat 2.3 to 2.6. if WT is empty returns RT

74 * 2.3 replace the name of recursive term with WT

75 * 2.4 evaluate the recursive term and store into WT

76 * 2.5 append WT to RT

77 * 2.6 go back to 2.2

78 * ----------------------------------------------------------------

79 */

80static TupleTableSlot *

81 ExecRecursiveUnion(PlanState *pstate)

82{

83 RecursiveUnionState *node = castNode(RecursiveUnionState, pstate);

84 PlanState *outerPlan = outerPlanState(node);

85 PlanState *innerPlan = innerPlanState(node);

86 RecursiveUnion *plan = (RecursiveUnion *) node->ps.plan;

87 TupleTableSlot *slot;

88 bool isnew;

89

90 CHECK_FOR_INTERRUPTS();

91

92 /* 1. Evaluate non-recursive term */

93 if (!node->recursing)

94 {

95 for (;;)

96 {

97 slot = ExecProcNode(outerPlan);

98 if (TupIsNull(slot))

99 break;

100 if (plan->numCols > 0)

101 {

102 /* Find or build hashtable entry for this tuple's group */

103 LookupTupleHashEntry(node->hashtable, slot, &isnew, NULL);

104 /* Must reset temp context after each hashtable lookup */

105 MemoryContextReset(node->tempContext);

106 /* Ignore tuple if already seen */

107 if (!isnew)

108 continue;

109 }

110 /* Each non-duplicate tuple goes to the working table ... */

111 tuplestore_puttupleslot(node->working_table, slot);

112 /* ... and to the caller */

113 return slot;

114 }

115 node->recursing = true;

116 }

117

118 /* 2. Execute recursive term */

119 for (;;)

120 {

121 slot = ExecProcNode(innerPlan);

122 if (TupIsNull(slot))

123 {

124 Tuplestorestate *swaptemp;

125

126 /* Done if there's nothing in the intermediate table */

127 if (node->intermediate_empty)

128 break;

129

130 /*

131 * Now we let the intermediate table become the work table. We

132 * need a fresh intermediate table, so delete the tuples from the

133 * current working table and use that as the new intermediate

134 * table. This saves a round of free/malloc from creating a new

135 * tuple store.

136 */

137 tuplestore_clear(node->working_table);

138

139 swaptemp = node->working_table;

140 node->working_table = node->intermediate_table;

141 node->intermediate_table = swaptemp;

142

143 /* mark the intermediate table as empty */

144 node->intermediate_empty = true;

145

146 /* reset the recursive term */

147 innerPlan->chgParam = bms_add_member(innerPlan->chgParam,

148 plan->wtParam);

149

150 /* and continue fetching from recursive term */

151 continue;

152 }

153

154 if (plan->numCols > 0)

155 {

156 /* Find or build hashtable entry for this tuple's group */

157 LookupTupleHashEntry(node->hashtable, slot, &isnew, NULL);

158 /* Must reset temp context after each hashtable lookup */

159 MemoryContextReset(node->tempContext);

160 /* Ignore tuple if already seen */

161 if (!isnew)

162 continue;

163 }

164

165 /* Else, tuple is good; stash it in intermediate table ... */

166 node->intermediate_empty = false;

167 tuplestore_puttupleslot(node->intermediate_table, slot);

168 /* ... and return it */

169 return slot;

170 }

171

172 return NULL;

173}

174

175/* ----------------------------------------------------------------

176 * ExecInitRecursiveUnion

177 * ----------------------------------------------------------------

178 */

179RecursiveUnionState *

180 ExecInitRecursiveUnion(RecursiveUnion *node, EState *estate, int eflags)

181{

182 RecursiveUnionState *rustate;

183 ParamExecData *prmdata;

184

185 /* check for unsupported flags */

186 Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));

187

188 /*

189 * create state structure

190 */

191 rustate = makeNode(RecursiveUnionState);

192 rustate->ps.plan = (Plan *) node;

193 rustate->ps.state = estate;

194 rustate->ps.ExecProcNode = ExecRecursiveUnion;

195

196 rustate->eqfuncoids = NULL;

197 rustate->hashfunctions = NULL;

198 rustate->hashtable = NULL;

199 rustate->tempContext = NULL;

200 rustate->tableContext = NULL;

201

202 /* initialize processing state */

203 rustate->recursing = false;

204 rustate->intermediate_empty = true;

205 rustate->working_table = tuplestore_begin_heap(false, false, work_mem);

206 rustate->intermediate_table = tuplestore_begin_heap(false, false, work_mem);

207

208 /*

209 * If hashing, we need a per-tuple memory context for comparisons, and a

210 * longer-lived context to store the hash table. The table can't just be

211 * kept in the per-query context because we want to be able to throw it

212 * away when rescanning.

213 */

214 if (node->numCols > 0)

215 {

216 rustate->tempContext =

217 AllocSetContextCreate(CurrentMemoryContext,

218 "RecursiveUnion",

219 ALLOCSET_DEFAULT_SIZES);

220 rustate->tableContext =

221 AllocSetContextCreate(CurrentMemoryContext,

222 "RecursiveUnion hash table",

223 ALLOCSET_DEFAULT_SIZES);

224 }

225

226 /*

227 * Make the state structure available to descendant WorkTableScan nodes

228 * via the Param slot reserved for it.

229 */

230 prmdata = &(estate->es_param_exec_vals[node->wtParam]);

231 Assert(prmdata->execPlan == NULL);

232 prmdata->value = PointerGetDatum(rustate);

233 prmdata->isnull = false;

234

235 /*

236 * Miscellaneous initialization

237 *

238 * RecursiveUnion plans don't have expression contexts because they never

239 * call ExecQual or ExecProject.

240 */

241 Assert(node->plan.qual == NIL);

242

243 /*

244 * RecursiveUnion nodes still have Result slots, which hold pointers to

245 * tuples, so we have to initialize them.

246 */

247 ExecInitResultTypeTL(&rustate->ps);

248

249 /*

250 * Initialize result tuple type. (Note: we have to set up the result type

251 * before initializing child nodes, because nodeWorktablescan.c expects it

252 * to be valid.)

253 */

254 rustate->ps.ps_ProjInfo = NULL;

255

256 /*

257 * initialize child nodes

258 */

259 outerPlanState(rustate) = ExecInitNode(outerPlan(node), estate, eflags);

260 innerPlanState(rustate) = ExecInitNode(innerPlan(node), estate, eflags);

261

262 /*

263 * If hashing, precompute fmgr lookup data for inner loop, and create the

264 * hash table.

265 */

266 if (node->numCols > 0)

267 {

268 execTuplesHashPrepare(node->numCols,

269 node->dupOperators,

270 &rustate->eqfuncoids,

271 &rustate->hashfunctions);

272 build_hash_table(rustate);

273 }

274

275 return rustate;

276}

277

278/* ----------------------------------------------------------------

279 * ExecEndRecursiveUnion

280 *

281 * frees any storage allocated through C routines.

282 * ----------------------------------------------------------------

283 */

284void

285 ExecEndRecursiveUnion(RecursiveUnionState *node)

286{

287 /* Release tuplestores */

288 tuplestore_end(node->working_table);

289 tuplestore_end(node->intermediate_table);

290

291 /* free subsidiary stuff including hashtable */

292 if (node->tempContext)

293 MemoryContextDelete(node->tempContext);

294 if (node->tableContext)

295 MemoryContextDelete(node->tableContext);

296

297 /*

298 * close down subplans

299 */

300 ExecEndNode(outerPlanState(node));

301 ExecEndNode(innerPlanState(node));

302}

303

304/* ----------------------------------------------------------------

305 * ExecReScanRecursiveUnion

306 *

307 * Rescans the relation.

308 * ----------------------------------------------------------------

309 */

310void

311 ExecReScanRecursiveUnion(RecursiveUnionState *node)

312{

313 PlanState *outerPlan = outerPlanState(node);

314 PlanState *innerPlan = innerPlanState(node);

315 RecursiveUnion *plan = (RecursiveUnion *) node->ps.plan;

316

317 /*

318 * Set recursive term's chgParam to tell it that we'll modify the working

319 * table and therefore it has to rescan.

320 */

321 innerPlan->chgParam = bms_add_member(innerPlan->chgParam, plan->wtParam);

322

323 /*

324 * if chgParam of subnode is not null then plan will be re-scanned by

325 * first ExecProcNode. Because of above, we only have to do this to the

326 * non-recursive term.

327 */

328 if (outerPlan->chgParam == NULL)

329 ExecReScan(outerPlan);

330

331 /* Release any hashtable storage */

332 if (node->tableContext)

333 MemoryContextReset(node->tableContext);

334

335 /* Empty hashtable if needed */

336 if (plan->numCols > 0)

337 ResetTupleHashTable(node->hashtable);

338

339 /* reset processing state */

340 node->recursing = false;

341 node->intermediate_empty = true;

342 tuplestore_clear(node->working_table);

343 tuplestore_clear(node->intermediate_table);

344}

bms_add_member

Bitmapset * bms_add_member(Bitmapset *a, int x)

Definition: bitmapset.c:815

ExecReScan

void ExecReScan(PlanState *node)

Definition: execAmi.c:77

execTuplesHashPrepare

void execTuplesHashPrepare(int numCols, const Oid *eqOperators, Oid **eqFuncOids, FmgrInfo **hashFunctions)

Definition: execGrouping.c:97

LookupTupleHashEntry

TupleHashEntry LookupTupleHashEntry(TupleHashTable hashtable, TupleTableSlot *slot, bool *isnew, uint32 *hash)

Definition: execGrouping.c:301

BuildTupleHashTable

TupleHashTable BuildTupleHashTable(PlanState *parent, TupleDesc inputDesc, const TupleTableSlotOps *inputOps, int numCols, AttrNumber *keyColIdx, const Oid *eqfuncoids, FmgrInfo *hashfunctions, Oid *collations, long nbuckets, Size additionalsize, MemoryContext metacxt, MemoryContext tablecxt, MemoryContext tempcxt, bool use_variable_hash_iv)

Definition: execGrouping.c:167

ResetTupleHashTable

void ResetTupleHashTable(TupleHashTable hashtable)

Definition: execGrouping.c:280

ExecEndNode

void ExecEndNode(PlanState *node)

Definition: execProcnode.c:562

ExecInitNode

PlanState * ExecInitNode(Plan *node, EState *estate, int eflags)

Definition: execProcnode.c:142

ExecInitResultTypeTL

void ExecInitResultTypeTL(PlanState *planstate)