granicus.if.org Git - postgresql/blob - src/backend/tcop/pquery.c

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