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