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