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