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