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