Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * pquery.c
4 : * POSTGRES process query command code
5 : *
6 : * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
7 : * Portions Copyright (c) 1994, Regents of the University of California
8 : *
9 : *
10 : * IDENTIFICATION
11 : * src/backend/tcop/pquery.c
12 : *
13 : *-------------------------------------------------------------------------
14 : */
15 :
16 : #include "postgres.h"
17 :
18 : #include <limits.h>
19 :
20 : #include "access/xact.h"
21 : #include "commands/prepare.h"
22 : #include "executor/tstoreReceiver.h"
23 : #include "miscadmin.h"
24 : #include "pg_trace.h"
25 : #include "tcop/pquery.h"
26 : #include "tcop/utility.h"
27 : #include "utils/memutils.h"
28 : #include "utils/snapmgr.h"
29 :
30 :
31 : /*
32 : * ActivePortal is the currently executing Portal (the most closely nested,
33 : * if there are several).
34 : */
35 : Portal ActivePortal = NULL;
36 :
37 :
38 : static void ProcessQuery(PlannedStmt *plan,
39 : const char *sourceText,
40 : ParamListInfo params,
41 : QueryEnvironment *queryEnv,
42 : DestReceiver *dest,
43 : char *completionTag);
44 : static void FillPortalStore(Portal portal, bool isTopLevel);
45 : static uint64 RunFromStore(Portal portal, ScanDirection direction, uint64 count,
46 : DestReceiver *dest);
47 : static uint64 PortalRunSelect(Portal portal, bool forward, long count,
48 : DestReceiver *dest);
49 : static void PortalRunUtility(Portal portal, PlannedStmt *pstmt,
50 : bool isTopLevel, bool setHoldSnapshot,
51 : DestReceiver *dest, char *completionTag);
52 : static void PortalRunMulti(Portal portal,
53 : bool isTopLevel, bool setHoldSnapshot,
54 : DestReceiver *dest, DestReceiver *altdest,
55 : char *completionTag);
56 : static uint64 DoPortalRunFetch(Portal portal,
57 : FetchDirection fdirection,
58 : long count,
59 : DestReceiver *dest);
60 : static void DoPortalRewind(Portal portal);
61 :
62 :
63 : /*
64 : * CreateQueryDesc
65 : */
66 : QueryDesc *
67 349202 : CreateQueryDesc(PlannedStmt *plannedstmt,
68 : const char *sourceText,
69 : Snapshot snapshot,
70 : Snapshot crosscheck_snapshot,
71 : DestReceiver *dest,
72 : ParamListInfo params,
73 : QueryEnvironment *queryEnv,
74 : int instrument_options)
75 : {
76 349202 : QueryDesc *qd = (QueryDesc *) palloc(sizeof(QueryDesc));
77 :
78 349202 : qd->operation = plannedstmt->commandType; /* operation */
79 349202 : qd->plannedstmt = plannedstmt; /* plan */
80 349202 : qd->sourceText = sourceText; /* query text */
81 349202 : qd->snapshot = RegisterSnapshot(snapshot); /* snapshot */
82 : /* RI check snapshot */
83 349202 : qd->crosscheck_snapshot = RegisterSnapshot(crosscheck_snapshot);
84 349202 : qd->dest = dest; /* output dest */
85 349202 : qd->params = params; /* parameter values passed into query */
86 349202 : qd->queryEnv = queryEnv;
87 349202 : qd->instrument_options = instrument_options; /* instrumentation wanted? */
88 :
89 : /* null these fields until set by ExecutorStart */
90 349202 : qd->tupDesc = NULL;
91 349202 : qd->estate = NULL;
92 349202 : qd->planstate = NULL;
93 349202 : qd->totaltime = NULL;
94 :
95 : /* not yet executed */
96 349202 : qd->already_executed = false;
97 :
98 349202 : return qd;
99 : }
100 :
101 : /*
102 : * FreeQueryDesc
103 : */
104 : void
105 336232 : FreeQueryDesc(QueryDesc *qdesc)
106 : {
107 : /* Can't be a live query */
108 : Assert(qdesc->estate == NULL);
109 :
110 : /* forget our snapshots */
111 336232 : UnregisterSnapshot(qdesc->snapshot);
112 336232 : UnregisterSnapshot(qdesc->crosscheck_snapshot);
113 :
114 : /* Only the QueryDesc itself need be freed */
115 336232 : pfree(qdesc);
116 336232 : }
117 :
118 :
119 : /*
120 : * ProcessQuery
121 : * Execute a single plannable query within a PORTAL_MULTI_QUERY,
122 : * PORTAL_ONE_RETURNING, or PORTAL_ONE_MOD_WITH portal
123 : *
124 : * plan: the plan tree for the query
125 : * sourceText: the source text of the query
126 : * params: any parameters needed
127 : * dest: where to send results
128 : * completionTag: points to a buffer of size COMPLETION_TAG_BUFSIZE
129 : * in which to store a command completion status string.
130 : *
131 : * completionTag may be NULL if caller doesn't want a status string.
132 : *
133 : * Must be called in a memory context that will be reset or deleted on
134 : * error; otherwise the executor's memory usage will be leaked.
135 : */
136 : static void
137 67216 : ProcessQuery(PlannedStmt *plan,
138 : const char *sourceText,
139 : ParamListInfo params,
140 : QueryEnvironment *queryEnv,
141 : DestReceiver *dest,
142 : char *completionTag)
143 : {
144 : QueryDesc *queryDesc;
145 :
146 : /*
147 : * Create the QueryDesc object
148 : */
149 67216 : queryDesc = CreateQueryDesc(plan, sourceText,
150 : GetActiveSnapshot(), InvalidSnapshot,
151 : dest, params, queryEnv, 0);
152 :
153 : /*
154 : * Call ExecutorStart to prepare the plan for execution
155 : */
156 67216 : ExecutorStart(queryDesc, 0);
157 :
158 : /*
159 : * Run the plan to completion.
160 : */
161 66986 : ExecutorRun(queryDesc, ForwardScanDirection, 0L, true);
162 :
163 : /*
164 : * Build command completion status string, if caller wants one.
165 : */
166 65310 : if (completionTag)
167 : {
168 : Oid lastOid;
169 :
170 64938 : switch (queryDesc->operation)
171 : {
172 : case CMD_SELECT:
173 76 : snprintf(completionTag, COMPLETION_TAG_BUFSIZE,
174 : "SELECT " UINT64_FORMAT,
175 76 : queryDesc->estate->es_processed);
176 76 : break;
177 : case CMD_INSERT:
178 : /* lastoid doesn't exist anymore */
179 54210 : lastOid = InvalidOid;
180 54210 : snprintf(completionTag, COMPLETION_TAG_BUFSIZE,
181 : "INSERT %u " UINT64_FORMAT,
182 54210 : lastOid, queryDesc->estate->es_processed);
183 54210 : break;
184 : case CMD_UPDATE:
185 8222 : snprintf(completionTag, COMPLETION_TAG_BUFSIZE,
186 : "UPDATE " UINT64_FORMAT,
187 8222 : queryDesc->estate->es_processed);
188 8222 : break;
189 : case CMD_DELETE:
190 2430 : snprintf(completionTag, COMPLETION_TAG_BUFSIZE,
191 : "DELETE " UINT64_FORMAT,
192 2430 : queryDesc->estate->es_processed);
193 2430 : break;
194 : default:
195 0 : strcpy(completionTag, "???");
196 0 : break;
197 : }
198 : }
199 :
200 : /*
201 : * Now, we close down all the scans and free allocated resources.
202 : */
203 65310 : ExecutorFinish(queryDesc);
204 64796 : ExecutorEnd(queryDesc);
205 :
206 64796 : FreeQueryDesc(queryDesc);
207 64796 : }
208 :
209 : /*
210 : * ChoosePortalStrategy
211 : * Select portal execution strategy given the intended statement list.
212 : *
213 : * The list elements can be Querys or PlannedStmts.
214 : * That's more general than portals need, but plancache.c uses this too.
215 : *
216 : * See the comments in portal.h.
217 : */
218 : PortalStrategy
219 544506 : ChoosePortalStrategy(List *stmts)
220 : {
221 : int nSetTag;
222 : ListCell *lc;
223 :
224 : /*
225 : * PORTAL_ONE_SELECT and PORTAL_UTIL_SELECT need only consider the
226 : * single-statement case, since there are no rewrite rules that can add
227 : * auxiliary queries to a SELECT or a utility command. PORTAL_ONE_MOD_WITH
228 : * likewise allows only one top-level statement.
229 : */
230 544506 : if (list_length(stmts) == 1)
231 : {
232 544318 : Node *stmt = (Node *) linitial(stmts);
233 :
234 544318 : if (IsA(stmt, Query))
235 : {
236 39830 : Query *query = (Query *) stmt;
237 :
238 39830 : if (query->canSetTag)
239 : {
240 39830 : if (query->commandType == CMD_SELECT)
241 : {
242 28394 : if (query->hasModifyingCTE)
243 0 : return PORTAL_ONE_MOD_WITH;
244 : else
245 28394 : return PORTAL_ONE_SELECT;
246 : }
247 11436 : if (query->commandType == CMD_UTILITY)
248 : {
249 4682 : if (UtilityReturnsTuples(query->utilityStmt))
250 2424 : return PORTAL_UTIL_SELECT;
251 : /* it can't be ONE_RETURNING, so give up */
252 2258 : return PORTAL_MULTI_QUERY;
253 : }
254 : }
255 : }
256 504488 : else if (IsA(stmt, PlannedStmt))
257 : {
258 504488 : PlannedStmt *pstmt = (PlannedStmt *) stmt;
259 :
260 504488 : if (pstmt->canSetTag)
261 : {
262 504484 : if (pstmt->commandType == CMD_SELECT)
263 : {
264 142620 : if (pstmt->hasModifyingCTE)
265 84 : return PORTAL_ONE_MOD_WITH;
266 : else
267 142536 : return PORTAL_ONE_SELECT;
268 : }
269 361864 : if (pstmt->commandType == CMD_UTILITY)
270 : {
271 295288 : if (UtilityReturnsTuples(pstmt->utilityStmt))
272 14730 : return PORTAL_UTIL_SELECT;
273 : /* it can't be ONE_RETURNING, so give up */
274 280558 : return PORTAL_MULTI_QUERY;
275 : }
276 : }
277 : }
278 : else
279 0 : elog(ERROR, "unrecognized node type: %d", (int) nodeTag(stmt));
280 : }
281 :
282 : /*
283 : * PORTAL_ONE_RETURNING has to allow auxiliary queries added by rewrite.
284 : * Choose PORTAL_ONE_RETURNING if there is exactly one canSetTag query and
285 : * it has a RETURNING list.
286 : */
287 73522 : nSetTag = 0;
288 75064 : foreach(lc, stmts)
289 : {
290 73626 : Node *stmt = (Node *) lfirst(lc);
291 :
292 73626 : if (IsA(stmt, Query))
293 : {
294 6754 : Query *query = (Query *) stmt;
295 :
296 6754 : if (query->canSetTag)
297 : {
298 6754 : if (++nSetTag > 1)
299 72084 : return PORTAL_MULTI_QUERY; /* no need to look further */
300 13508 : if (query->commandType == CMD_UTILITY ||
301 6754 : query->returningList == NIL)
302 6588 : return PORTAL_MULTI_QUERY; /* no need to look further */
303 : }
304 : }
305 66872 : else if (IsA(stmt, PlannedStmt))
306 : {
307 66872 : PlannedStmt *pstmt = (PlannedStmt *) stmt;
308 :
309 66872 : if (pstmt->canSetTag)
310 : {
311 66760 : if (++nSetTag > 1)
312 0 : return PORTAL_MULTI_QUERY; /* no need to look further */
313 133520 : if (pstmt->commandType == CMD_UTILITY ||
314 66760 : !pstmt->hasReturning)
315 65496 : return PORTAL_MULTI_QUERY; /* no need to look further */
316 : }
317 : }
318 : else
319 0 : elog(ERROR, "unrecognized node type: %d", (int) nodeTag(stmt));
320 : }
321 1438 : if (nSetTag == 1)
322 1430 : return PORTAL_ONE_RETURNING;
323 :
324 : /* Else, it's the general case... */
325 8 : return PORTAL_MULTI_QUERY;
326 : }
327 :
328 : /*
329 : * FetchPortalTargetList
330 : * Given a portal that returns tuples, extract the query targetlist.
331 : * Returns NIL if the portal doesn't have a determinable targetlist.
332 : *
333 : * Note: do not modify the result.
334 : */
335 : List *
336 155132 : FetchPortalTargetList(Portal portal)
337 : {
338 : /* no point in looking if we determined it doesn't return tuples */
339 155132 : if (portal->strategy == PORTAL_MULTI_QUERY)
340 4 : return NIL;
341 : /* get the primary statement and find out what it returns */
342 155128 : return FetchStatementTargetList((Node *) PortalGetPrimaryStmt(portal));
343 : }
344 :
345 : /*
346 : * FetchStatementTargetList
347 : * Given a statement that returns tuples, extract the query targetlist.
348 : * Returns NIL if the statement doesn't have a determinable targetlist.
349 : *
350 : * This can be applied to a Query or a PlannedStmt.
351 : * That's more general than portals need, but plancache.c uses this too.
352 : *
353 : * Note: do not modify the result.
354 : *
355 : * XXX be careful to keep this in sync with UtilityReturnsTuples.
356 : */
357 : List *
358 155928 : FetchStatementTargetList(Node *stmt)
359 : {
360 155928 : if (stmt == NULL)
361 0 : return NIL;
362 155928 : if (IsA(stmt, Query))
363 : {
364 800 : Query *query = (Query *) stmt;
365 :
366 800 : if (query->commandType == CMD_UTILITY)
367 : {
368 : /* transfer attention to utility statement */
369 8 : stmt = query->utilityStmt;
370 : }
371 : else
372 : {
373 792 : if (query->commandType == CMD_SELECT)
374 792 : return query->targetList;
375 0 : if (query->returningList)
376 0 : return query->returningList;
377 0 : return NIL;
378 : }
379 : }
380 155136 : if (IsA(stmt, PlannedStmt))
381 : {
382 155128 : PlannedStmt *pstmt = (PlannedStmt *) stmt;
383 :
384 155128 : if (pstmt->commandType == CMD_UTILITY)
385 : {
386 : /* transfer attention to utility statement */
387 12248 : stmt = pstmt->utilityStmt;
388 : }
389 : else
390 : {
391 142880 : if (pstmt->commandType == CMD_SELECT)
392 141676 : return pstmt->planTree->targetlist;
393 1204 : if (pstmt->hasReturning)
394 1204 : return pstmt->planTree->targetlist;
395 0 : return NIL;
396 : }
397 : }
398 12256 : if (IsA(stmt, FetchStmt))
399 : {
400 4216 : FetchStmt *fstmt = (FetchStmt *) stmt;
401 : Portal subportal;
402 :
403 : Assert(!fstmt->ismove);
404 4216 : subportal = GetPortalByName(fstmt->portalname);
405 : Assert(PortalIsValid(subportal));
406 4216 : return FetchPortalTargetList(subportal);
407 : }
408 8040 : if (IsA(stmt, ExecuteStmt))
409 : {
410 692 : ExecuteStmt *estmt = (ExecuteStmt *) stmt;
411 : PreparedStatement *entry;
412 :
413 692 : entry = FetchPreparedStatement(estmt->name, true);
414 692 : return FetchPreparedStatementTargetList(entry);
415 : }
416 7348 : return NIL;
417 : }
418 :
419 : /*
420 : * PortalStart
421 : * Prepare a portal for execution.
422 : *
423 : * Caller must already have created the portal, done PortalDefineQuery(),
424 : * and adjusted portal options if needed.
425 : *
426 : * If parameters are needed by the query, they must be passed in "params"
427 : * (caller is responsible for giving them appropriate lifetime).
428 : *
429 : * The caller can also provide an initial set of "eflags" to be passed to
430 : * ExecutorStart (but note these can be modified internally, and they are
431 : * currently only honored for PORTAL_ONE_SELECT portals). Most callers
432 : * should simply pass zero.
433 : *
434 : * The caller can optionally pass a snapshot to be used; pass InvalidSnapshot
435 : * for the normal behavior of setting a new snapshot. This parameter is
436 : * presently ignored for non-PORTAL_ONE_SELECT portals (it's only intended
437 : * to be used for cursors).
438 : *
439 : * On return, portal is ready to accept PortalRun() calls, and the result
440 : * tupdesc (if any) is known.
441 : */
442 : void
443 504676 : PortalStart(Portal portal, ParamListInfo params,
444 : int eflags, Snapshot snapshot)
445 : {
446 : Portal saveActivePortal;
447 : ResourceOwner saveResourceOwner;
448 : MemoryContext savePortalContext;
449 : MemoryContext oldContext;
450 : QueryDesc *queryDesc;
451 : int myeflags;
452 :
453 : AssertArg(PortalIsValid(portal));
454 : AssertState(portal->status == PORTAL_DEFINED);
455 :
456 : /*
457 : * Set up global portal context pointers.
458 : */
459 504676 : saveActivePortal = ActivePortal;
460 504676 : saveResourceOwner = CurrentResourceOwner;
461 504676 : savePortalContext = PortalContext;
462 504676 : PG_TRY();
463 : {
464 504676 : ActivePortal = portal;
465 504676 : if (portal->resowner)
466 504676 : CurrentResourceOwner = portal->resowner;
467 504676 : PortalContext = portal->portalContext;
468 :
469 504676 : oldContext = MemoryContextSwitchTo(PortalContext);
470 :
471 : /* Must remember portal param list, if any */
472 504676 : portal->portalParams = params;
473 :
474 : /*
475 : * Determine the portal execution strategy
476 : */
477 504676 : portal->strategy = ChoosePortalStrategy(portal->stmts);
478 :
479 : /*
480 : * Fire her up according to the strategy
481 : */
482 504676 : switch (portal->strategy)
483 : {
484 : case PORTAL_ONE_SELECT:
485 :
486 : /* Must set snapshot before starting executor. */
487 142536 : if (snapshot)
488 4576 : PushActiveSnapshot(snapshot);
489 : else
490 137960 : PushActiveSnapshot(GetTransactionSnapshot());
491 :
492 : /*
493 : * Create QueryDesc in portal's context; for the moment, set
494 : * the destination to DestNone.
495 : */
496 142536 : queryDesc = CreateQueryDesc(linitial_node(PlannedStmt, portal->stmts),
497 : portal->sourceText,
498 : GetActiveSnapshot(),
499 : InvalidSnapshot,
500 : None_Receiver,
501 : params,
502 : portal->queryEnv,
503 : 0);
504 :
505 : /*
506 : * If it's a scrollable cursor, executor needs to support
507 : * REWIND and backwards scan, as well as whatever the caller
508 : * might've asked for.
509 : */
510 142536 : if (portal->cursorOptions & CURSOR_OPT_SCROLL)
511 2396 : myeflags = eflags | EXEC_FLAG_REWIND | EXEC_FLAG_BACKWARD;
512 : else
513 140140 : myeflags = eflags;
514 :
515 : /*
516 : * Call ExecutorStart to prepare the plan for execution
517 : */
518 142536 : ExecutorStart(queryDesc, myeflags);
519 :
520 : /*
521 : * This tells PortalCleanup to shut down the executor
522 : */
523 142282 : portal->queryDesc = queryDesc;
524 :
525 : /*
526 : * Remember tuple descriptor (computed by ExecutorStart)
527 : */
528 142282 : portal->tupDesc = queryDesc->tupDesc;
529 :
530 : /*
531 : * Reset cursor position data to "start of query"
532 : */
533 142282 : portal->atStart = true;
534 142282 : portal->atEnd = false; /* allow fetches */
535 142282 : portal->portalPos = 0;
536 :
537 142282 : PopActiveSnapshot();
538 142282 : break;
539 :
540 : case PORTAL_ONE_RETURNING:
541 : case PORTAL_ONE_MOD_WITH:
542 :
543 : /*
544 : * We don't start the executor until we are told to run the
545 : * portal. We do need to set up the result tupdesc.
546 : */
547 : {
548 : PlannedStmt *pstmt;
549 :
550 1348 : pstmt = PortalGetPrimaryStmt(portal);
551 1348 : portal->tupDesc =
552 1348 : ExecCleanTypeFromTL(pstmt->planTree->targetlist);
553 : }
554 :
555 : /*
556 : * Reset cursor position data to "start of query"
557 : */
558 1348 : portal->atStart = true;
559 1348 : portal->atEnd = false; /* allow fetches */
560 1348 : portal->portalPos = 0;
561 1348 : break;
562 :
563 : case PORTAL_UTIL_SELECT:
564 :
565 : /*
566 : * We don't set snapshot here, because PortalRunUtility will
567 : * take care of it if needed.
568 : */
569 : {
570 14730 : PlannedStmt *pstmt = PortalGetPrimaryStmt(portal);
571 :
572 : Assert(pstmt->commandType == CMD_UTILITY);
573 14730 : portal->tupDesc = UtilityTupleDescriptor(pstmt->utilityStmt);
574 : }
575 :
576 : /*
577 : * Reset cursor position data to "start of query"
578 : */
579 14728 : portal->atStart = true;
580 14728 : portal->atEnd = false; /* allow fetches */
581 14728 : portal->portalPos = 0;
582 14728 : break;
583 :
584 : case PORTAL_MULTI_QUERY:
585 : /* Need do nothing now */
586 346062 : portal->tupDesc = NULL;
587 346062 : break;
588 : }
589 : }
590 256 : PG_CATCH();
591 : {
592 : /* Uncaught error while executing portal: mark it dead */
593 256 : MarkPortalFailed(portal);
594 :
595 : /* Restore global vars and propagate error */
596 256 : ActivePortal = saveActivePortal;
597 256 : CurrentResourceOwner = saveResourceOwner;
598 256 : PortalContext = savePortalContext;
599 :
600 256 : PG_RE_THROW();
601 : }
602 504420 : PG_END_TRY();
603 :
604 504420 : MemoryContextSwitchTo(oldContext);
605 :
606 504420 : ActivePortal = saveActivePortal;
607 504420 : CurrentResourceOwner = saveResourceOwner;
608 504420 : PortalContext = savePortalContext;
609 :
610 504420 : portal->status = PORTAL_READY;
611 504420 : }
612 :
613 : /*
614 : * PortalSetResultFormat
615 : * Select the format codes for a portal's output.
616 : *
617 : * This must be run after PortalStart for a portal that will be read by
618 : * a DestRemote or DestRemoteExecute destination. It is not presently needed
619 : * for other destination types.
620 : *
621 : * formats[] is the client format request, as per Bind message conventions.
622 : */
623 : void
624 497410 : PortalSetResultFormat(Portal portal, int nFormats, int16 *formats)
625 : {
626 : int natts;
627 : int i;
628 :
629 : /* Do nothing if portal won't return tuples */
630 497410 : if (portal->tupDesc == NULL)
631 345994 : return;
632 151416 : natts = portal->tupDesc->natts;
633 151416 : portal->formats = (int16 *)
634 151416 : MemoryContextAlloc(portal->portalContext,
635 : natts * sizeof(int16));
636 151416 : if (nFormats > 1)
637 : {
638 : /* format specified for each column */
639 0 : if (nFormats != natts)
640 0 : ereport(ERROR,
641 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
642 : errmsg("bind message has %d result formats but query has %d columns",
643 : nFormats, natts)));
644 0 : memcpy(portal->formats, formats, natts * sizeof(int16));
645 : }
646 151416 : else if (nFormats > 0)
647 : {
648 : /* single format specified, use for all columns */
649 151416 : int16 format1 = formats[0];
650 :
651 686766 : for (i = 0; i < natts; i++)
652 535350 : portal->formats[i] = format1;
653 : }
654 : else
655 : {
656 : /* use default format for all columns */
657 0 : for (i = 0; i < natts; i++)
658 0 : portal->formats[i] = 0;
659 : }
660 : }
661 :
662 : /*
663 : * PortalRun
664 : * Run a portal's query or queries.
665 : *
666 : * count <= 0 is interpreted as a no-op: the destination gets started up
667 : * and shut down, but nothing else happens. Also, count == FETCH_ALL is
668 : * interpreted as "all rows". Note that count is ignored in multi-query
669 : * situations, where we always run the portal to completion.
670 : *
671 : * isTopLevel: true if query is being executed at backend "top level"
672 : * (that is, directly from a client command message)
673 : *
674 : * dest: where to send output of primary (canSetTag) query
675 : *
676 : * altdest: where to send output of non-primary queries
677 : *
678 : * completionTag: points to a buffer of size COMPLETION_TAG_BUFSIZE
679 : * in which to store a command completion status string.
680 : * May be NULL if caller doesn't want a status string.
681 : *
682 : * Returns true if the portal's execution is complete, false if it was
683 : * suspended due to exhaustion of the count parameter.
684 : */
685 : bool
686 498192 : PortalRun(Portal portal, long count, bool isTopLevel, bool run_once,
687 : DestReceiver *dest, DestReceiver *altdest,
688 : char *completionTag)
689 : {
690 : bool result;
691 : uint64 nprocessed;
692 : ResourceOwner saveTopTransactionResourceOwner;
693 : MemoryContext saveTopTransactionContext;
694 : Portal saveActivePortal;
695 : ResourceOwner saveResourceOwner;
696 : MemoryContext savePortalContext;
697 : MemoryContext saveMemoryContext;
698 :
699 : AssertArg(PortalIsValid(portal));
700 :
701 : TRACE_POSTGRESQL_QUERY_EXECUTE_START();
702 :
703 : /* Initialize completion tag to empty string */
704 498192 : if (completionTag)
705 498192 : completionTag[0] = '\0';
706 :
707 498192 : if (log_executor_stats && portal->strategy != PORTAL_MULTI_QUERY)
708 : {
709 0 : elog(DEBUG3, "PortalRun");
710 : /* PORTAL_MULTI_QUERY logs its own stats per query */
711 0 : ResetUsage();
712 : }
713 :
714 : /*
715 : * Check for improper portal use, and mark portal active.
716 : */
717 498192 : MarkPortalActive(portal);
718 :
719 : /* Set run_once flag. Shouldn't be clear if previously set. */
720 : Assert(!portal->run_once || run_once);
721 498192 : portal->run_once = run_once;
722 :
723 : /*
724 : * Set up global portal context pointers.
725 : *
726 : * We have to play a special game here to support utility commands like
727 : * VACUUM and CLUSTER, which internally start and commit transactions.
728 : * When we are called to execute such a command, CurrentResourceOwner will
729 : * be pointing to the TopTransactionResourceOwner --- which will be
730 : * destroyed and replaced in the course of the internal commit and
731 : * restart. So we need to be prepared to restore it as pointing to the
732 : * exit-time TopTransactionResourceOwner. (Ain't that ugly? This idea of
733 : * internally starting whole new transactions is not good.)
734 : * CurrentMemoryContext has a similar problem, but the other pointers we
735 : * save here will be NULL or pointing to longer-lived objects.
736 : */
737 498192 : saveTopTransactionResourceOwner = TopTransactionResourceOwner;
738 498192 : saveTopTransactionContext = TopTransactionContext;
739 498192 : saveActivePortal = ActivePortal;
740 498192 : saveResourceOwner = CurrentResourceOwner;
741 498192 : savePortalContext = PortalContext;
742 498192 : saveMemoryContext = CurrentMemoryContext;
743 498192 : PG_TRY();
744 : {
745 498192 : ActivePortal = portal;
746 498192 : if (portal->resowner)
747 498192 : CurrentResourceOwner = portal->resowner;
748 498192 : PortalContext = portal->portalContext;
749 :
750 498192 : MemoryContextSwitchTo(PortalContext);
751 :
752 498192 : switch (portal->strategy)
753 : {
754 : case PORTAL_ONE_SELECT:
755 : case PORTAL_ONE_RETURNING:
756 : case PORTAL_ONE_MOD_WITH:
757 : case PORTAL_UTIL_SELECT:
758 :
759 : /*
760 : * If we have not yet run the command, do so, storing its
761 : * results in the portal's tuplestore. But we don't do that
762 : * for the PORTAL_ONE_SELECT case.
763 : */
764 152130 : if (portal->strategy != PORTAL_ONE_SELECT && !portal->holdStore)
765 13710 : FillPortalStore(portal, isTopLevel);
766 :
767 : /*
768 : * Now fetch desired portion of results.
769 : */
770 151946 : nprocessed = PortalRunSelect(portal, true, count, dest);
771 :
772 : /*
773 : * If the portal result contains a command tag and the caller
774 : * gave us a pointer to store it, copy it. Patch the "SELECT"
775 : * tag to also provide the rowcount.
776 : */
777 149150 : if (completionTag && portal->commandTag)
778 : {
779 149150 : if (strcmp(portal->commandTag, "SELECT") == 0)
780 135624 : snprintf(completionTag, COMPLETION_TAG_BUFSIZE,
781 : "SELECT " UINT64_FORMAT, nprocessed);
782 : else
783 13526 : strcpy(completionTag, portal->commandTag);
784 : }
785 :
786 : /* Mark portal not active */
787 149150 : portal->status = PORTAL_READY;
788 :
789 : /*
790 : * Since it's a forward fetch, say DONE iff atEnd is now true.
791 : */
792 149150 : result = portal->atEnd;
793 149150 : break;
794 :
795 : case PORTAL_MULTI_QUERY:
796 346062 : PortalRunMulti(portal, isTopLevel, false,
797 : dest, altdest, completionTag);
798 :
799 : /* Prevent portal's commands from being re-executed */
800 336954 : MarkPortalDone(portal);
801 :
802 : /* Always complete at end of RunMulti */
803 336954 : result = true;
804 336954 : break;
805 :
806 : default:
807 0 : elog(ERROR, "unrecognized portal strategy: %d",
808 : (int) portal->strategy);
809 : result = false; /* keep compiler quiet */
810 : break;
811 : }
812 : }
813 12088 : PG_CATCH();
814 : {
815 : /* Uncaught error while executing portal: mark it dead */
816 12088 : MarkPortalFailed(portal);
817 :
818 : /* Restore global vars and propagate error */
819 12088 : if (saveMemoryContext == saveTopTransactionContext)
820 11936 : MemoryContextSwitchTo(TopTransactionContext);
821 : else
822 152 : MemoryContextSwitchTo(saveMemoryContext);
823 12088 : ActivePortal = saveActivePortal;
824 12088 : if (saveResourceOwner == saveTopTransactionResourceOwner)
825 11950 : CurrentResourceOwner = TopTransactionResourceOwner;
826 : else
827 138 : CurrentResourceOwner = saveResourceOwner;
828 12088 : PortalContext = savePortalContext;
829 :
830 12088 : PG_RE_THROW();
831 : }
832 486104 : PG_END_TRY();
833 :
834 486104 : if (saveMemoryContext == saveTopTransactionContext)
835 456686 : MemoryContextSwitchTo(TopTransactionContext);
836 : else
837 29418 : MemoryContextSwitchTo(saveMemoryContext);
838 486104 : ActivePortal = saveActivePortal;
839 486104 : if (saveResourceOwner == saveTopTransactionResourceOwner)
840 482526 : CurrentResourceOwner = TopTransactionResourceOwner;
841 : else
842 3578 : CurrentResourceOwner = saveResourceOwner;
843 486104 : PortalContext = savePortalContext;
844 :
845 486104 : if (log_executor_stats && portal->strategy != PORTAL_MULTI_QUERY)
846 0 : ShowUsage("EXECUTOR STATISTICS");
847 :
848 : TRACE_POSTGRESQL_QUERY_EXECUTE_DONE();
849 :
850 486104 : return result;
851 : }
852 :
853 : /*
854 : * PortalRunSelect
855 : * Execute a portal's query in PORTAL_ONE_SELECT mode, and also
856 : * when fetching from a completed holdStore in PORTAL_ONE_RETURNING,
857 : * PORTAL_ONE_MOD_WITH, and PORTAL_UTIL_SELECT cases.
858 : *
859 : * This handles simple N-rows-forward-or-backward cases. For more complex
860 : * nonsequential access to a portal, see PortalRunFetch.
861 : *
862 : * count <= 0 is interpreted as a no-op: the destination gets started up
863 : * and shut down, but nothing else happens. Also, count == FETCH_ALL is
864 : * interpreted as "all rows". (cf FetchStmt.howMany)
865 : *
866 : * Caller must already have validated the Portal and done appropriate
867 : * setup (cf. PortalRun).
868 : *
869 : * Returns number of rows processed (suitable for use in result tag)
870 : */
871 : static uint64
872 164756 : PortalRunSelect(Portal portal,
873 : bool forward,
874 : long count,
875 : DestReceiver *dest)
876 : {
877 : QueryDesc *queryDesc;
878 : ScanDirection direction;
879 : uint64 nprocessed;
880 :
881 : /*
882 : * NB: queryDesc will be NULL if we are fetching from a held cursor or a
883 : * completed utility query; can't use it in that path.
884 : */
885 164756 : queryDesc = portal->queryDesc;
886 :
887 : /* Caller messed up if we have neither a ready query nor held data. */
888 : Assert(queryDesc || portal->holdStore);
889 :
890 : /*
891 : * Force the queryDesc destination to the right thing. This supports
892 : * MOVE, for example, which will pass in dest = DestNone. This is okay to
893 : * change as long as we do it on every fetch. (The Executor must not
894 : * assume that dest never changes.)
895 : */
896 164756 : if (queryDesc)
897 146358 : queryDesc->dest = dest;
898 :
899 : /*
900 : * Determine which direction to go in, and check to see if we're already
901 : * at the end of the available tuples in that direction. If so, set the
902 : * direction to NoMovement to avoid trying to fetch any tuples. (This
903 : * check exists because not all plan node types are robust about being
904 : * called again if they've already returned NULL once.) Then call the
905 : * executor (we must not skip this, because the destination needs to see a
906 : * setup and shutdown even if no tuples are available). Finally, update
907 : * the portal position state depending on the number of tuples that were
908 : * retrieved.
909 : */
910 164756 : if (forward)
911 : {
912 164380 : if (portal->atEnd || count <= 0)
913 : {
914 3818 : direction = NoMovementScanDirection;
915 3818 : count = 0; /* don't pass negative count to executor */
916 : }
917 : else
918 160562 : direction = ForwardScanDirection;
919 :
920 : /* In the executor, zero count processes all rows */
921 164380 : if (count == FETCH_ALL)
922 152130 : count = 0;
923 :
924 164380 : if (portal->holdStore)
925 18390 : nprocessed = RunFromStore(portal, direction, (uint64) count, dest);
926 : else
927 : {
928 145990 : PushActiveSnapshot(queryDesc->snapshot);
929 145990 : ExecutorRun(queryDesc, direction, (uint64) count,
930 145990 : portal->run_once);
931 143172 : nprocessed = queryDesc->estate->es_processed;
932 143172 : PopActiveSnapshot();
933 : }
934 :
935 161562 : if (!ScanDirectionIsNoMovement(direction))
936 : {
937 157744 : if (nprocessed > 0)
938 134942 : portal->atStart = false; /* OK to go backward now */
939 157744 : if (count == 0 || nprocessed < (uint64) count)
940 155008 : portal->atEnd = true; /* we retrieved 'em all */
941 157744 : portal->portalPos += nprocessed;
942 : }
943 : }
944 : else
945 : {
946 376 : if (portal->cursorOptions & CURSOR_OPT_NO_SCROLL)
947 12 : ereport(ERROR,
948 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
949 : errmsg("cursor can only scan forward"),
950 : errhint("Declare it with SCROLL option to enable backward scan.")));
951 :
952 364 : if (portal->atStart || count <= 0)
953 : {
954 48 : direction = NoMovementScanDirection;
955 48 : count = 0; /* don't pass negative count to executor */
956 : }
957 : else
958 316 : direction = BackwardScanDirection;
959 :
960 : /* In the executor, zero count processes all rows */
961 364 : if (count == FETCH_ALL)
962 32 : count = 0;
963 :
964 364 : if (portal->holdStore)
965 8 : nprocessed = RunFromStore(portal, direction, (uint64) count, dest);
966 : else
967 : {
968 356 : PushActiveSnapshot(queryDesc->snapshot);
969 356 : ExecutorRun(queryDesc, direction, (uint64) count,
970 356 : portal->run_once);
971 356 : nprocessed = queryDesc->estate->es_processed;
972 356 : PopActiveSnapshot();
973 : }
974 :
975 364 : if (!ScanDirectionIsNoMovement(direction))
976 : {
977 316 : if (nprocessed > 0 && portal->atEnd)
978 : {
979 88 : portal->atEnd = false; /* OK to go forward now */
980 88 : portal->portalPos++; /* adjust for endpoint case */
981 : }
982 316 : if (count == 0 || nprocessed < (uint64) count)
983 : {
984 116 : portal->atStart = true; /* we retrieved 'em all */
985 116 : portal->portalPos = 0;
986 : }
987 : else
988 : {
989 200 : portal->portalPos -= nprocessed;
990 : }
991 : }
992 : }
993 :
994 161926 : return nprocessed;
995 : }
996 :
997 : /*
998 : * FillPortalStore
999 : * Run the query and load result tuples into the portal's tuple store.
1000 : *
1001 : * This is used for PORTAL_ONE_RETURNING, PORTAL_ONE_MOD_WITH, and
1002 : * PORTAL_UTIL_SELECT cases only.
1003 : */
1004 : static void
1005 16076 : FillPortalStore(Portal portal, bool isTopLevel)
1006 : {
1007 : DestReceiver *treceiver;
1008 : char completionTag[COMPLETION_TAG_BUFSIZE];
1009 :
1010 16076 : PortalCreateHoldStore(portal);
1011 16076 : treceiver = CreateDestReceiver(DestTuplestore);
1012 16076 : SetTuplestoreDestReceiverParams(treceiver,
1013 : portal->holdStore,
1014 : portal->holdContext,
1015 : false);
1016 :
1017 16076 : completionTag[0] = '\0';
1018 :
1019 16076 : switch (portal->strategy)
1020 : {
1021 : case PORTAL_ONE_RETURNING:
1022 : case PORTAL_ONE_MOD_WITH:
1023 :
1024 : /*
1025 : * Run the portal to completion just as for the default
1026 : * PORTAL_MULTI_QUERY case, but send the primary query's output to
1027 : * the tuplestore. Auxiliary query outputs are discarded. Set the
1028 : * portal's holdSnapshot to the snapshot used (or a copy of it).
1029 : */
1030 1348 : PortalRunMulti(portal, isTopLevel, true,
1031 : treceiver, None_Receiver, completionTag);
1032 1280 : break;
1033 :
1034 : case PORTAL_UTIL_SELECT:
1035 14728 : PortalRunUtility(portal, linitial_node(PlannedStmt, portal->stmts),
1036 : isTopLevel, true, treceiver, completionTag);
1037 14612 : break;
1038 :
1039 : default:
1040 0 : elog(ERROR, "unsupported portal strategy: %d",
1041 : (int) portal->strategy);
1042 : break;
1043 : }
1044 :
1045 : /* Override default completion tag with actual command result */
1046 15892 : if (completionTag[0] != '\0')
1047 6184 : portal->commandTag = pstrdup(completionTag);
1048 :
1049 15892 : treceiver->rDestroy(treceiver);
1050 15892 : }
1051 :
1052 : /*
1053 : * RunFromStore
1054 : * Fetch tuples from the portal's tuple store.
1055 : *
1056 : * Calling conventions are similar to ExecutorRun, except that we
1057 : * do not depend on having a queryDesc or estate. Therefore we return the
1058 : * number of tuples processed as the result, not in estate->es_processed.
1059 : *
1060 : * One difference from ExecutorRun is that the destination receiver functions
1061 : * are run in the caller's memory context (since we have no estate). Watch
1062 : * out for memory leaks.
1063 : */
1064 : static uint64
1065 18398 : RunFromStore(Portal portal, ScanDirection direction, uint64 count,
1066 : DestReceiver *dest)
1067 : {
1068 18398 : uint64 current_tuple_count = 0;
1069 : TupleTableSlot *slot;
1070 :
1071 18398 : slot = MakeSingleTupleTableSlot(portal->tupDesc, &TTSOpsMinimalTuple);
1072 :
1073 18398 : dest->rStartup(dest, CMD_SELECT, portal->tupDesc);
1074 :
1075 18398 : if (ScanDirectionIsNoMovement(direction))
1076 : {
1077 : /* do nothing except start/stop the destination */
1078 : }
1079 : else
1080 : {
1081 16028 : bool forward = ScanDirectionIsForward(direction);
1082 :
1083 : for (;;)
1084 252808 : {
1085 : MemoryContext oldcontext;
1086 : bool ok;
1087 :
1088 268836 : oldcontext = MemoryContextSwitchTo(portal->holdContext);
1089 :
1090 268836 : ok = tuplestore_gettupleslot(portal->holdStore, forward, false,
1091 : slot);
1092 :
1093 268836 : MemoryContextSwitchTo(oldcontext);
1094 :
1095 268836 : if (!ok)
1096 15920 : break;
1097 :
1098 : /*
1099 : * If we are not able to send the tuple, we assume the destination
1100 : * has closed and no more tuples can be sent. If that's the case,
1101 : * end the loop.
1102 : */
1103 252916 : if (!dest->receiveSlot(slot, dest))
1104 0 : break;
1105 :
1106 252916 : ExecClearTuple(slot);
1107 :
1108 : /*
1109 : * check our tuple count.. if we've processed the proper number
1110 : * then quit, else loop again and process more tuples. Zero count
1111 : * means no limit.
1112 : */
1113 252916 : current_tuple_count++;
1114 252916 : if (count && count == current_tuple_count)
1115 108 : break;
1116 : }
1117 : }
1118 :
1119 18398 : dest->rShutdown(dest);
1120 :
1121 18398 : ExecDropSingleTupleTableSlot(slot);
1122 :
1123 18398 : return current_tuple_count;
1124 : }
1125 :
1126 : /*
1127 : * PortalRunUtility
1128 : * Execute a utility statement inside a portal.
1129 : */
1130 : static void
1131 295286 : PortalRunUtility(Portal portal, PlannedStmt *pstmt,
1132 : bool isTopLevel, bool setHoldSnapshot,
1133 : DestReceiver *dest, char *completionTag)
1134 : {
1135 295286 : Node *utilityStmt = pstmt->utilityStmt;
1136 : Snapshot snapshot;
1137 :
1138 : /*
1139 : * Set snapshot if utility stmt needs one. Most reliable way to do this
1140 : * seems to be to enumerate those that do not need one; this is a short
1141 : * list. Transaction control, LOCK, and SET must *not* set a snapshot
1142 : * since they need to be executable at the start of a transaction-snapshot
1143 : * mode transaction without freezing a snapshot. By extension we allow
1144 : * SHOW not to set a snapshot. The other stmts listed are just efficiency
1145 : * hacks. Beware of listing anything that can modify the database --- if,
1146 : * say, it has to update an index with expressions that invoke
1147 : * user-defined functions, then it had better have a snapshot.
1148 : */
1149 823062 : if (!(IsA(utilityStmt, TransactionStmt) ||
1150 544414 : IsA(utilityStmt, LockStmt) ||
1151 528774 : IsA(utilityStmt, VariableSetStmt) ||
1152 516680 : IsA(utilityStmt, VariableShowStmt) ||
1153 516168 : IsA(utilityStmt, ConstraintsSetStmt) ||
1154 : /* efficiency hacks from here down */
1155 511768 : IsA(utilityStmt, FetchStmt) ||
1156 507396 : IsA(utilityStmt, ListenStmt) ||
1157 507280 : IsA(utilityStmt, NotifyStmt) ||
1158 253600 : IsA(utilityStmt, UnlistenStmt) ||
1159 253572 : IsA(utilityStmt, CheckPointStmt)))
1160 : {
1161 253494 : snapshot = GetTransactionSnapshot();
1162 : /* If told to, register the snapshot we're using and save in portal */
1163 253494 : if (setHoldSnapshot)
1164 : {
1165 10022 : snapshot = RegisterSnapshot(snapshot);
1166 10022 : portal->holdSnapshot = snapshot;
1167 : }
1168 253494 : PushActiveSnapshot(snapshot);
1169 : /* PushActiveSnapshot might have copied the snapshot */
1170 253494 : snapshot = GetActiveSnapshot();
1171 : }
1172 : else
1173 41792 : snapshot = NULL;
1174 :
1175 590572 : ProcessUtility(pstmt,
1176 : portal->sourceText,
1177 295286 : isTopLevel ? PROCESS_UTILITY_TOPLEVEL : PROCESS_UTILITY_QUERY,
1178 : portal->portalParams,
1179 : portal->queryEnv,
1180 : dest,
1181 : completionTag);
1182 :
1183 : /* Some utility statements may change context on us */
1184 288414 : MemoryContextSwitchTo(portal->portalContext);
1185 :
1186 : /*
1187 : * Some utility commands may pop the ActiveSnapshot stack from under us,
1188 : * so be careful to only pop the stack if our snapshot is still at the
1189 : * top.
1190 : */
1191 531052 : if (snapshot != NULL && ActiveSnapshotSet() &&
1192 242638 : snapshot == GetActiveSnapshot())
1193 242638 : PopActiveSnapshot();
1194 288414 : }
1195 :
1196 : /*
1197 : * PortalRunMulti
1198 : * Execute a portal's queries in the general case (multi queries
1199 : * or non-SELECT-like queries)
1200 : */
1201 : static void
1202 347410 : PortalRunMulti(Portal portal,
1203 : bool isTopLevel, bool setHoldSnapshot,
1204 : DestReceiver *dest, DestReceiver *altdest,
1205 : char *completionTag)
1206 : {
1207 347410 : bool active_snapshot_set = false;
1208 : ListCell *stmtlist_item;
1209 :
1210 : /*
1211 : * If the destination is DestRemoteExecute, change to DestNone. The
1212 : * reason is that the client won't be expecting any tuples, and indeed has
1213 : * no way to know what they are, since there is no provision for Describe
1214 : * to send a RowDescription message when this portal execution strategy is
1215 : * in effect. This presently will only affect SELECT commands added to
1216 : * non-SELECT queries by rewrite rules: such commands will be executed,
1217 : * but the results will be discarded unless you use "simple Query"
1218 : * protocol.
1219 : */
1220 347410 : if (dest->mydest == DestRemoteExecute)
1221 8424 : dest = None_Receiver;
1222 347410 : if (altdest->mydest == DestRemoteExecute)
1223 8424 : altdest = None_Receiver;
1224 :
1225 : /*
1226 : * Loop to handle the individual queries generated from a single parsetree
1227 : * by analysis and rewrite.
1228 : */
1229 686008 : foreach(stmtlist_item, portal->stmts)
1230 : {
1231 347774 : PlannedStmt *pstmt = lfirst_node(PlannedStmt, stmtlist_item);
1232 :
1233 : /*
1234 : * If we got a cancel signal in prior command, quit
1235 : */
1236 347774 : CHECK_FOR_INTERRUPTS();
1237 :
1238 347774 : if (pstmt->utilityStmt == NULL)
1239 : {
1240 : /*
1241 : * process a plannable query.
1242 : */
1243 : TRACE_POSTGRESQL_QUERY_EXECUTE_START();
1244 :
1245 67216 : if (log_executor_stats)
1246 0 : ResetUsage();
1247 :
1248 : /*
1249 : * Must always have a snapshot for plannable queries. First time
1250 : * through, take a new snapshot; for subsequent queries in the
1251 : * same portal, just update the snapshot's copy of the command
1252 : * counter.
1253 : */
1254 67216 : if (!active_snapshot_set)
1255 : {
1256 66852 : Snapshot snapshot = GetTransactionSnapshot();
1257 :
1258 : /* If told to, register the snapshot and save in portal */
1259 66852 : if (setHoldSnapshot)
1260 : {
1261 1348 : snapshot = RegisterSnapshot(snapshot);
1262 1348 : portal->holdSnapshot = snapshot;
1263 : }
1264 :
1265 : /*
1266 : * We can't have the holdSnapshot also be the active one,
1267 : * because UpdateActiveSnapshotCommandId would complain. So
1268 : * force an extra snapshot copy. Plain PushActiveSnapshot
1269 : * would have copied the transaction snapshot anyway, so this
1270 : * only adds a copy step when setHoldSnapshot is true. (It's
1271 : * okay for the command ID of the active snapshot to diverge
1272 : * from what holdSnapshot has.)
1273 : */
1274 66852 : PushCopiedSnapshot(snapshot);
1275 66852 : active_snapshot_set = true;
1276 : }
1277 : else
1278 364 : UpdateActiveSnapshotCommandId();
1279 :
1280 67216 : if (pstmt->canSetTag)
1281 : {
1282 : /* statement can set tag string */
1283 66844 : ProcessQuery(pstmt,
1284 : portal->sourceText,
1285 : portal->portalParams,
1286 : portal->queryEnv,
1287 : dest, completionTag);
1288 : }
1289 : else
1290 : {
1291 : /* stmt added by rewrite cannot set tag */
1292 372 : ProcessQuery(pstmt,
1293 : portal->sourceText,
1294 : portal->portalParams,
1295 : portal->queryEnv,
1296 : altdest, NULL);
1297 : }
1298 :
1299 64796 : if (log_executor_stats)
1300 0 : ShowUsage("EXECUTOR STATISTICS");
1301 :
1302 : TRACE_POSTGRESQL_QUERY_EXECUTE_DONE();
1303 : }
1304 : else
1305 : {
1306 : /*
1307 : * process utility functions (create, destroy, etc..)
1308 : *
1309 : * We must not set a snapshot here for utility commands (if one is
1310 : * needed, PortalRunUtility will do it). If a utility command is
1311 : * alone in a portal then everything's fine. The only case where
1312 : * a utility command can be part of a longer list is that rules
1313 : * are allowed to include NotifyStmt. NotifyStmt doesn't care
1314 : * whether it has a snapshot or not, so we just leave the current
1315 : * snapshot alone if we have one.
1316 : */
1317 280558 : if (pstmt->canSetTag)
1318 : {
1319 : Assert(!active_snapshot_set);
1320 : /* statement can set tag string */
1321 280558 : PortalRunUtility(portal, pstmt, isTopLevel, false,
1322 : dest, completionTag);
1323 : }
1324 : else
1325 : {
1326 : Assert(IsA(pstmt->utilityStmt, NotifyStmt));
1327 : /* stmt added by rewrite cannot set tag */
1328 0 : PortalRunUtility(portal, pstmt, isTopLevel, false,
1329 : altdest, NULL);
1330 : }
1331 : }
1332 :
1333 : /*
1334 : * Increment command counter between queries, but not after the last
1335 : * one.
1336 : */
1337 338598 : if (lnext(portal->stmts, stmtlist_item) != NULL)
1338 364 : CommandCounterIncrement();
1339 :
1340 : /*
1341 : * Clear subsidiary contexts to recover temporary memory.
1342 : */
1343 : Assert(portal->portalContext == CurrentMemoryContext);
1344 :
1345 338598 : MemoryContextDeleteChildren(portal->portalContext);
1346 : }
1347 :
1348 : /* Pop the snapshot if we pushed one. */
1349 338234 : if (active_snapshot_set)
1350 64432 : PopActiveSnapshot();
1351 :
1352 : /*
1353 : * If a command completion tag was supplied, use it. Otherwise use the
1354 : * portal's commandTag as the default completion tag.
1355 : *
1356 : * Exception: Clients expect INSERT/UPDATE/DELETE tags to have counts, so
1357 : * fake them with zeros. This can happen with DO INSTEAD rules if there
1358 : * is no replacement query of the same type as the original. We print "0
1359 : * 0" here because technically there is no query of the matching tag type,
1360 : * and printing a non-zero count for a different query type seems wrong,
1361 : * e.g. an INSERT that does an UPDATE instead should not print "0 1" if
1362 : * one row was updated. See QueryRewrite(), step 3, for details.
1363 : */
1364 338234 : if (completionTag && completionTag[0] == '\0')
1365 : {
1366 268198 : if (portal->commandTag)
1367 268198 : strcpy(completionTag, portal->commandTag);
1368 268198 : if (strcmp(completionTag, "SELECT") == 0)
1369 0 : sprintf(completionTag, "SELECT 0 0");
1370 268198 : else if (strcmp(completionTag, "INSERT") == 0)
1371 4 : strcpy(completionTag, "INSERT 0 0");
1372 268194 : else if (strcmp(completionTag, "UPDATE") == 0)
1373 0 : strcpy(completionTag, "UPDATE 0");
1374 268194 : else if (strcmp(completionTag, "DELETE") == 0)
1375 4 : strcpy(completionTag, "DELETE 0");
1376 : }
1377 338234 : }
1378 :
1379 : /*
1380 : * PortalRunFetch
1381 : * Variant form of PortalRun that supports SQL FETCH directions.
1382 : *
1383 : * Note: we presently assume that no callers of this want isTopLevel = true.
1384 : *
1385 : * count <= 0 is interpreted as a no-op: the destination gets started up
1386 : * and shut down, but nothing else happens. Also, count == FETCH_ALL is
1387 : * interpreted as "all rows". (cf FetchStmt.howMany)
1388 : *
1389 : * Returns number of rows processed (suitable for use in result tag)
1390 : */
1391 : uint64
1392 12758 : PortalRunFetch(Portal portal,
1393 : FetchDirection fdirection,
1394 : long count,
1395 : DestReceiver *dest)
1396 : {
1397 : uint64 result;
1398 : Portal saveActivePortal;
1399 : ResourceOwner saveResourceOwner;
1400 : MemoryContext savePortalContext;
1401 : MemoryContext oldContext;
1402 :
1403 : AssertArg(PortalIsValid(portal));
1404 :
1405 : /*
1406 : * Check for improper portal use, and mark portal active.
1407 : */
1408 12758 : MarkPortalActive(portal);
1409 :
1410 : /* If supporting FETCH, portal can't be run-once. */
1411 : Assert(!portal->run_once);
1412 :
1413 : /*
1414 : * Set up global portal context pointers.
1415 : */
1416 12746 : saveActivePortal = ActivePortal;
1417 12746 : saveResourceOwner = CurrentResourceOwner;
1418 12746 : savePortalContext = PortalContext;
1419 12746 : PG_TRY();
1420 : {
1421 12746 : ActivePortal = portal;
1422 12746 : if (portal->resowner)
1423 12644 : CurrentResourceOwner = portal->resowner;
1424 12746 : PortalContext = portal->portalContext;
1425 :
1426 12746 : oldContext = MemoryContextSwitchTo(PortalContext);
1427 :
1428 12746 : switch (portal->strategy)
1429 : {
1430 : case PORTAL_ONE_SELECT:
1431 7976 : result = DoPortalRunFetch(portal, fdirection, count, dest);
1432 7942 : break;
1433 :
1434 : case PORTAL_ONE_RETURNING:
1435 : case PORTAL_ONE_MOD_WITH:
1436 : case PORTAL_UTIL_SELECT:
1437 :
1438 : /*
1439 : * If we have not yet run the command, do so, storing its
1440 : * results in the portal's tuplestore.
1441 : */
1442 4770 : if (!portal->holdStore)
1443 2366 : FillPortalStore(portal, false /* isTopLevel */ );
1444 :
1445 : /*
1446 : * Now fetch desired portion of results.
1447 : */
1448 4770 : result = DoPortalRunFetch(portal, fdirection, count, dest);
1449 4770 : break;
1450 :
1451 : default:
1452 0 : elog(ERROR, "unsupported portal strategy");
1453 : result = 0; /* keep compiler quiet */
1454 : break;
1455 : }
1456 : }
1457 34 : PG_CATCH();
1458 : {
1459 : /* Uncaught error while executing portal: mark it dead */
1460 34 : MarkPortalFailed(portal);
1461 :
1462 : /* Restore global vars and propagate error */
1463 34 : ActivePortal = saveActivePortal;
1464 34 : CurrentResourceOwner = saveResourceOwner;
1465 34 : PortalContext = savePortalContext;
1466 :
1467 34 : PG_RE_THROW();
1468 : }
1469 12712 : PG_END_TRY();
1470 :
1471 12712 : MemoryContextSwitchTo(oldContext);
1472 :
1473 : /* Mark portal not active */
1474 12712 : portal->status = PORTAL_READY;
1475 :
1476 12712 : ActivePortal = saveActivePortal;
1477 12712 : CurrentResourceOwner = saveResourceOwner;
1478 12712 : PortalContext = savePortalContext;
1479 :
1480 12712 : return result;
1481 : }
1482 :
1483 : /*
1484 : * DoPortalRunFetch
1485 : * Guts of PortalRunFetch --- the portal context is already set up
1486 : *
1487 : * count <= 0 is interpreted as a no-op: the destination gets started up
1488 : * and shut down, but nothing else happens. Also, count == FETCH_ALL is
1489 : * interpreted as "all rows". (cf FetchStmt.howMany)
1490 : *
1491 : * Returns number of rows processed (suitable for use in result tag)
1492 : */
1493 : static uint64
1494 12746 : DoPortalRunFetch(Portal portal,
1495 : FetchDirection fdirection,
1496 : long count,
1497 : DestReceiver *dest)
1498 : {
1499 : bool forward;
1500 :
1501 : Assert(portal->strategy == PORTAL_ONE_SELECT ||
1502 : portal->strategy == PORTAL_ONE_RETURNING ||
1503 : portal->strategy == PORTAL_ONE_MOD_WITH ||
1504 : portal->strategy == PORTAL_UTIL_SELECT);
1505 :
1506 12746 : switch (fdirection)
1507 : {
1508 : case FETCH_FORWARD:
1509 12270 : if (count < 0)
1510 : {
1511 0 : fdirection = FETCH_BACKWARD;
1512 0 : count = -count;
1513 : }
1514 : /* fall out of switch to share code with FETCH_BACKWARD */
1515 12270 : break;
1516 : case FETCH_BACKWARD:
1517 312 : if (count < 0)
1518 : {
1519 0 : fdirection = FETCH_FORWARD;
1520 0 : count = -count;
1521 : }
1522 : /* fall out of switch to share code with FETCH_FORWARD */
1523 312 : break;
1524 : case FETCH_ABSOLUTE:
1525 108 : if (count > 0)
1526 : {
1527 : /*
1528 : * Definition: Rewind to start, advance count-1 rows, return
1529 : * next row (if any).
1530 : *
1531 : * In practice, if the goal is less than halfway back to the
1532 : * start, it's better to scan from where we are.
1533 : *
1534 : * Also, if current portalPos is outside the range of "long",
1535 : * do it the hard way to avoid possible overflow of the count
1536 : * argument to PortalRunSelect. We must exclude exactly
1537 : * LONG_MAX, as well, lest the count look like FETCH_ALL.
1538 : *
1539 : * In any case, we arrange to fetch the target row going
1540 : * forwards.
1541 : */
1542 72 : if ((uint64) (count - 1) <= portal->portalPos / 2 ||
1543 12 : portal->portalPos >= (uint64) LONG_MAX)
1544 : {
1545 48 : DoPortalRewind(portal);
1546 96 : if (count > 1)
1547 0 : PortalRunSelect(portal, true, count - 1,
1548 : None_Receiver);
1549 : }
1550 : else
1551 : {
1552 12 : long pos = (long) portal->portalPos;
1553 :
1554 12 : if (portal->atEnd)
1555 0 : pos++; /* need one extra fetch if off end */
1556 12 : if (count <= pos)
1557 4 : PortalRunSelect(portal, false, pos - count + 1,
1558 : None_Receiver);
1559 8 : else if (count > pos + 1)
1560 8 : PortalRunSelect(portal, true, count - pos - 1,
1561 : None_Receiver);
1562 : }
1563 60 : return PortalRunSelect(portal, true, 1L, dest);
1564 : }
1565 48 : else if (count < 0)
1566 : {
1567 : /*
1568 : * Definition: Advance to end, back up abs(count)-1 rows,
1569 : * return prior row (if any). We could optimize this if we
1570 : * knew in advance where the end was, but typically we won't.
1571 : * (Is it worth considering case where count > half of size of
1572 : * query? We could rewind once we know the size ...)
1573 : */
1574 32 : PortalRunSelect(portal, true, FETCH_ALL, None_Receiver);
1575 32 : if (count < -1)
1576 0 : PortalRunSelect(portal, false, -count - 1, None_Receiver);
1577 32 : return PortalRunSelect(portal, false, 1L, dest);
1578 : }
1579 : else
1580 : {
1581 : /* count == 0 */
1582 : /* Rewind to start, return zero rows */
1583 16 : DoPortalRewind(portal);
1584 16 : return PortalRunSelect(portal, true, 0L, dest);
1585 : }
1586 : break;
1587 : case FETCH_RELATIVE:
1588 56 : if (count > 0)
1589 : {
1590 : /*
1591 : * Definition: advance count-1 rows, return next row (if any).
1592 : */
1593 24 : if (count > 1)
1594 16 : PortalRunSelect(portal, true, count - 1, None_Receiver);
1595 24 : return PortalRunSelect(portal, true, 1L, dest);
1596 : }
1597 32 : else if (count < 0)
1598 : {
1599 : /*
1600 : * Definition: back up abs(count)-1 rows, return prior row (if
1601 : * any).
1602 : */
1603 20 : if (count < -1)
1604 12 : PortalRunSelect(portal, false, -count - 1, None_Receiver);
1605 20 : return PortalRunSelect(portal, false, 1L, dest);
1606 : }
1607 : else
1608 : {
1609 : /* count == 0 */
1610 : /* Same as FETCH FORWARD 0, so fall out of switch */
1611 12 : fdirection = FETCH_FORWARD;
1612 : }
1613 12 : break;
1614 : default:
1615 0 : elog(ERROR, "bogus direction");
1616 : break;
1617 : }
1618 :
1619 : /*
1620 : * Get here with fdirection == FETCH_FORWARD or FETCH_BACKWARD, and count
1621 : * >= 0.
1622 : */
1623 12594 : forward = (fdirection == FETCH_FORWARD);
1624 :
1625 : /*
1626 : * Zero count means to re-fetch the current row, if any (per SQL)
1627 : */
1628 12594 : if (count == 0)
1629 : {
1630 : bool on_row;
1631 :
1632 : /* Are we sitting on a row? */
1633 12 : on_row = (!portal->atStart && !portal->atEnd);
1634 :
1635 12 : if (dest->mydest == DestNone)
1636 : {
1637 : /* MOVE 0 returns 0/1 based on if FETCH 0 would return a row */
1638 0 : return on_row ? 1 : 0;
1639 : }
1640 : else
1641 : {
1642 : /*
1643 : * If we are sitting on a row, back up one so we can re-fetch it.
1644 : * If we are not sitting on a row, we still have to start up and
1645 : * shut down the executor so that the destination is initialized
1646 : * and shut down correctly; so keep going. To PortalRunSelect,
1647 : * count == 0 means we will retrieve no row.
1648 : */
1649 12 : if (on_row)
1650 : {
1651 12 : PortalRunSelect(portal, false, 1L, None_Receiver);
1652 : /* Set up to fetch one row forward */
1653 8 : count = 1;
1654 8 : forward = true;
1655 : }
1656 : }
1657 : }
1658 :
1659 : /*
1660 : * Optimize MOVE BACKWARD ALL into a Rewind.
1661 : */
1662 12590 : if (!forward && count == FETCH_ALL && dest->mydest == DestNone)
1663 : {
1664 16 : uint64 result = portal->portalPos;
1665 :
1666 16 : if (result > 0 && !portal->atEnd)
1667 0 : result--;
1668 16 : DoPortalRewind(portal);
1669 16 : return result;
1670 : }
1671 :
1672 12574 : return PortalRunSelect(portal, forward, count, dest);
1673 : }
1674 :
1675 : /*
1676 : * DoPortalRewind - rewind a Portal to starting point
1677 : */
1678 : static void
1679 80 : DoPortalRewind(Portal portal)
1680 : {
1681 : QueryDesc *queryDesc;
1682 :
1683 : /* Rewind holdStore, if we have one */
1684 80 : if (portal->holdStore)
1685 : {
1686 : MemoryContext oldcontext;
1687 :
1688 4 : oldcontext = MemoryContextSwitchTo(portal->holdContext);
1689 4 : tuplestore_rescan(portal->holdStore);
1690 4 : MemoryContextSwitchTo(oldcontext);
1691 : }
1692 :
1693 : /* Rewind executor, if active */
1694 80 : queryDesc = portal->queryDesc;
1695 80 : if (queryDesc)
1696 : {
1697 76 : PushActiveSnapshot(queryDesc->snapshot);
1698 76 : ExecutorRewind(queryDesc);
1699 76 : PopActiveSnapshot();
1700 : }
1701 :
1702 80 : portal->atStart = true;
1703 80 : portal->atEnd = false;
1704 80 : portal->portalPos = 0;
1705 80 : }
|
