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-2003, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        $Header: /cvsroot/pgsql/src/backend/tcop/pquery.c,v 1.72 2003/08/12 18:23:21 tgl Exp $
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 #include "postgres.h"
  16
  17 #include "executor/executor.h"
  18 #include "miscadmin.h"
  19 #include "tcop/tcopprot.h"
  20 #include "tcop/pquery.h"
  21 #include "tcop/utility.h"
  22 #include "utils/guc.h"
  23 #include "utils/memutils.h"
  24
  25
  26 static uint32 RunFromStore(Portal portal, ScanDirection direction, long count,
  27                          DestReceiver *dest);
  28 static long PortalRunSelect(Portal portal, bool forward, long count,
  29                                 DestReceiver *dest);
  30 static void PortalRunUtility(Portal portal, Query *query,
  31                                  DestReceiver *dest, char *completionTag);
  32 static void PortalRunMulti(Portal portal,
  33                            DestReceiver *dest, DestReceiver *altdest,
  34                            char *completionTag);
  35 static long DoPortalRunFetch(Portal portal,
  36                                  FetchDirection fdirection,
  37                                  long count,
  38                                  DestReceiver *dest);
  39 static void DoPortalRewind(Portal portal);
  40
  41
  42 /*
  43  * CreateQueryDesc
  44  */
  45 QueryDesc *
  46 CreateQueryDesc(Query *parsetree,
  47                                 Plan *plantree,
  48                                 DestReceiver *dest,
  49                                 ParamListInfo params,
  50                                 bool doInstrument)
  51 {
  52         QueryDesc  *qd = (QueryDesc *) palloc(sizeof(QueryDesc));
  53
  54         qd->operation = parsetree->commandType;         /* operation */
  55         qd->parsetree = parsetree;      /* parse tree */
  56         qd->plantree = plantree;        /* plan */
  57         qd->dest = dest;                        /* output dest */
  58         qd->params = params;            /* parameter values passed into query */
  59         qd->doInstrument = doInstrument;        /* instrumentation wanted? */
  60
  61         /* null these fields until set by ExecutorStart */
  62         qd->tupDesc = NULL;
  63         qd->estate = NULL;
  64         qd->planstate = NULL;
  65
  66         return qd;
  67 }
  68
  69 /*
  70  * FreeQueryDesc
  71  */
  72 void
  73 FreeQueryDesc(QueryDesc *qdesc)
  74 {
  75         /* Can't be a live query */
  76         Assert(qdesc->estate == NULL);
  77         /* Only the QueryDesc itself need be freed */
  78         pfree(qdesc);
  79 }
  80
  81
  82 /*
  83  * ProcessQuery
  84  *              Execute a single query
  85  *
  86  *      parsetree: the query tree
  87  *      plan: the plan tree for the query
  88  *      params: any parameters needed
  89  *      dest: where to send results
  90  *      completionTag: points to a buffer of size COMPLETION_TAG_BUFSIZE
  91  *              in which to store a command completion status string.
  92  *
  93  * completionTag may be NULL if caller doesn't want a status string.
  94  *
  95  * Must be called in a memory context that will be reset or deleted on
  96  * error; otherwise the executor's memory usage will be leaked.
  97  */
  98 void
  99 ProcessQuery(Query *parsetree,
 100                          Plan *plan,
 101                          ParamListInfo params,
 102                          DestReceiver *dest,
 103                          char *completionTag)
 104 {
 105         int                     operation = parsetree->commandType;
 106         QueryDesc  *queryDesc;
 107
 108         /*
 109          * Check for special-case destinations
 110          */
 111         if (operation == CMD_SELECT)
 112         {
 113                 if (parsetree->into != NULL)
 114                 {
 115                         /*
 116                          * SELECT INTO table (a/k/a CREATE AS ... SELECT).
 117                          *
 118                          * Override the normal communication destination; execMain.c
 119                          * special-cases this case.  (Perhaps would be cleaner to have
 120                          * an additional destination type?)
 121                          */
 122                         dest = None_Receiver;
 123                 }
 124         }
 125
 126         /*
 127          * Create the QueryDesc object
 128          */
 129         queryDesc = CreateQueryDesc(parsetree, plan, dest, params, false);
 130
 131         /*
 132          * Call ExecStart to prepare the plan for execution
 133          */
 134         ExecutorStart(queryDesc, false);
 135
 136         /*
 137          * Run the plan to completion.
 138          */
 139         ExecutorRun(queryDesc, ForwardScanDirection, 0L);
 140
 141         /*
 142          * Build command completion status string, if caller wants one.
 143          */
 144         if (completionTag)
 145         {
 146                 Oid                     lastOid;
 147
 148                 switch (operation)
 149                 {
 150                         case CMD_SELECT:
 151                                 strcpy(completionTag, "SELECT");
 152                                 break;
 153                         case CMD_INSERT:
 154                                 if (queryDesc->estate->es_processed == 1)
 155                                         lastOid = queryDesc->estate->es_lastoid;
 156                                 else
 157                                         lastOid = InvalidOid;
 158                                 snprintf(completionTag, COMPLETION_TAG_BUFSIZE,
 159                                 "INSERT %u %u", lastOid, queryDesc->estate->es_processed);
 160                                 break;
 161                         case CMD_UPDATE:
 162                                 snprintf(completionTag, COMPLETION_TAG_BUFSIZE,
 163                                                  "UPDATE %u", queryDesc->estate->es_processed);
 164                                 break;
 165                         case CMD_DELETE:
 166                                 snprintf(completionTag, COMPLETION_TAG_BUFSIZE,
 167                                                  "DELETE %u", queryDesc->estate->es_processed);
 168                                 break;
 169                         default:
 170                                 strcpy(completionTag, "???");
 171                                 break;
 172                 }
 173         }
 174
 175         /*
 176          * Now, we close down all the scans and free allocated resources.
 177          */
 178         ExecutorEnd(queryDesc);
 179
 180         FreeQueryDesc(queryDesc);
 181 }
 182
 183 /*
 184  * ChoosePortalStrategy
 185  *              Select portal execution strategy given the intended query list.
 186  *
 187  * See the comments in portal.h.
 188  */
 189 PortalStrategy
 190 ChoosePortalStrategy(List *parseTrees)
 191 {
 192         PortalStrategy strategy;
 193
 194         strategy = PORTAL_MULTI_QUERY;          /* default assumption */
 195
 196         if (length(parseTrees) == 1)
 197         {
 198                 Query      *query = (Query *) lfirst(parseTrees);
 199
 200                 if (query->commandType == CMD_SELECT &&
 201                         query->canSetTag &&
 202                         query->into == NULL)
 203                         strategy = PORTAL_ONE_SELECT;
 204                 else if (query->commandType == CMD_UTILITY &&
 205                                  query->canSetTag &&
 206                                  query->utilityStmt != NULL)
 207                 {
 208                         if (UtilityReturnsTuples(query->utilityStmt))
 209                                 strategy = PORTAL_UTIL_SELECT;
 210                 }
 211         }
 212         return strategy;
 213 }
 214
 215 /*
 216  * PortalStart
 217  *              Prepare a portal for execution.
 218  *
 219  * Caller must already have created the portal, done PortalDefineQuery(),
 220  * and adjusted portal options if needed.  If parameters are needed by
 221  * the query, they must be passed in here (caller is responsible for
 222  * giving them appropriate lifetime).
 223  *
 224  * On return, portal is ready to accept PortalRun() calls, and the result
 225  * tupdesc (if any) is known.
 226  */
 227 void
 228 PortalStart(Portal portal, ParamListInfo params)
 229 {
 230         MemoryContext oldContext;
 231         QueryDesc  *queryDesc;
 232
 233         AssertArg(PortalIsValid(portal));
 234         AssertState(portal->queryContext != NULL);      /* query defined? */
 235         AssertState(!portal->portalReady);      /* else extra PortalStart */
 236
 237         oldContext = MemoryContextSwitchTo(PortalGetHeapMemory(portal));
 238
 239         /* Must remember portal param list, if any */
 240         portal->portalParams = params;
 241
 242         /*
 243          * Determine the portal execution strategy
 244          */
 245         portal->strategy = ChoosePortalStrategy(portal->parseTrees);
 246
 247         /*
 248          * Fire her up according to the strategy
 249          */
 250         switch (portal->strategy)
 251         {
 252                 case PORTAL_ONE_SELECT:
 253
 254                         /*
 255                          * Must set query snapshot before starting executor.
 256                          */
 257                         SetQuerySnapshot();
 258
 259                         /*
 260                          * Create QueryDesc in portal's context; for the moment, set
 261                          * the destination to None.
 262                          */
 263                         queryDesc = CreateQueryDesc((Query *) lfirst(portal->parseTrees),
 264                                                                           (Plan *) lfirst(portal->planTrees),
 265                                                                                 None_Receiver,
 266                                                                                 params,
 267                                                                                 false);
 268
 269                         /*
 270                          * Call ExecStart to prepare the plan for execution
 271                          */
 272                         ExecutorStart(queryDesc, false);
 273
 274                         /*
 275                          * This tells PortalCleanup to shut down the executor
 276                          */
 277                         portal->queryDesc = queryDesc;
 278
 279                         /*
 280                          * Remember tuple descriptor (computed by ExecutorStart)
 281                          */
 282                         portal->tupDesc = queryDesc->tupDesc;
 283
 284                         /*
 285                          * Reset cursor position data to "start of query"
 286                          */
 287                         portal->atStart = true;
 288                         portal->atEnd = false;          /* allow fetches */
 289                         portal->portalPos = 0;
 290                         portal->posOverflow = false;
 291                         break;
 292
 293                 case PORTAL_UTIL_SELECT:
 294
 295                         /*
 296                          * We don't set query snapshot here, because PortalRunUtility
 297                          * will take care of it.
 298                          */
 299                         portal->tupDesc =
 300                                 UtilityTupleDescriptor(((Query *) lfirst(portal->parseTrees))->utilityStmt);
 301
 302                         /*
 303                          * Reset cursor position data to "start of query"
 304                          */
 305                         portal->atStart = true;
 306                         portal->atEnd = false;          /* allow fetches */
 307                         portal->portalPos = 0;
 308                         portal->posOverflow = false;
 309                         break;
 310
 311                 case PORTAL_MULTI_QUERY:
 312                         /* Need do nothing now */
 313                         portal->tupDesc = NULL;
 314                         break;
 315         }
 316
 317         MemoryContextSwitchTo(oldContext);
 318
 319         portal->portalReady = true;
 320 }
 321
 322 /*
 323  * PortalSetResultFormat
 324  *              Select the format codes for a portal's output.
 325  *
 326  * This must be run after PortalStart for a portal that will be read by
 327  * a Remote or RemoteExecute destination.  It is not presently needed for
 328  * other destination types.
 329  *
 330  * formats[] is the client format request, as per Bind message conventions.
 331  */
 332 void
 333 PortalSetResultFormat(Portal portal, int nFormats, int16 *formats)
 334 {
 335         int                     natts;
 336         int                     i;
 337
 338         /* Do nothing if portal won't return tuples */
 339         if (portal->tupDesc == NULL)
 340                 return;
 341         natts = portal->tupDesc->natts;
 342         /* +1 avoids palloc(0) if no columns */
 343         portal->formats = (int16 *)
 344                 MemoryContextAlloc(PortalGetHeapMemory(portal),
 345                                                    (natts + 1) * sizeof(int16));
 346         if (nFormats > 1)
 347         {
 348                 /* format specified for each column */
 349                 if (nFormats != natts)
 350                         ereport(ERROR,
 351                                         (errcode(ERRCODE_PROTOCOL_VIOLATION),
 352                                          errmsg("bind message has %d result formats but query has %d columns",
 353                                                         nFormats, natts)));
 354                 memcpy(portal->formats, formats, natts * sizeof(int16));
 355         }
 356         else if (nFormats > 0)
 357         {
 358                 /* single format specified, use for all columns */
 359                 int16           format1 = formats[0];
 360
 361                 for (i = 0; i < natts; i++)
 362                         portal->formats[i] = format1;
 363         }
 364         else
 365         {
 366                 /* use default format for all columns */
 367                 for (i = 0; i < natts; i++)
 368                         portal->formats[i] = 0;
 369         }
 370 }
 371
 372 /*
 373  * PortalRun
 374  *              Run a portal's query or queries.
 375  *
 376  * count <= 0 is interpreted as a no-op: the destination gets started up
 377  * and shut down, but nothing else happens.  Also, count == FETCH_ALL is
 378  * interpreted as "all rows".  Note that count is ignored in multi-query
 379  * situations, where we always run the portal to completion.
 380  *
 381  * dest: where to send output of primary (canSetTag) query
 382  *
 383  * altdest: where to send output of non-primary queries
 384  *
 385  * completionTag: points to a buffer of size COMPLETION_TAG_BUFSIZE
 386  *              in which to store a command completion status string.
 387  *              May be NULL if caller doesn't want a status string.
 388  *
 389  * Returns TRUE if the portal's execution is complete, FALSE if it was
 390  * suspended due to exhaustion of the count parameter.
 391  */
 392 bool
 393 PortalRun(Portal portal, long count,
 394                   DestReceiver *dest, DestReceiver *altdest,
 395                   char *completionTag)
 396 {
 397         bool            result;
 398         MemoryContext savePortalContext;
 399         MemoryContext saveQueryContext;
 400         MemoryContext oldContext;
 401
 402         AssertArg(PortalIsValid(portal));
 403         AssertState(portal->portalReady);       /* else no PortalStart */
 404
 405         /* Initialize completion tag to empty string */
 406         if (completionTag)
 407                 completionTag[0] = '\0';
 408
 409         /*
 410          * Check for improper portal use, and mark portal active.
 411          */
 412         if (portal->portalDone)
 413                 ereport(ERROR,
 414                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 415                    errmsg("portal \"%s\" cannot be run anymore", portal->name)));
 416         if (portal->portalActive)
 417                 ereport(ERROR,
 418                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 419                                  errmsg("portal \"%s\" already active", portal->name)));
 420         portal->portalActive = true;
 421
 422         /*
 423          * Set global portal context pointers.
 424          */
 425         savePortalContext = PortalContext;
 426         PortalContext = PortalGetHeapMemory(portal);
 427         saveQueryContext = QueryContext;
 428         QueryContext = portal->queryContext;
 429
 430         oldContext = MemoryContextSwitchTo(PortalContext);
 431
 432         switch (portal->strategy)
 433         {
 434                 case PORTAL_ONE_SELECT:
 435                         (void) PortalRunSelect(portal, true, count, dest);
 436                         /* we know the query is supposed to set the tag */
 437                         if (completionTag && portal->commandTag)
 438                                 strcpy(completionTag, portal->commandTag);
 439
 440                         /*
 441                          * Since it's a forward fetch, say DONE iff atEnd is now true.
 442                          */
 443                         result = portal->atEnd;
 444                         break;
 445
 446                 case PORTAL_UTIL_SELECT:
 447
 448                         /*
 449                          * If we have not yet run the utility statement, do so,
 450                          * storing its results in the portal's tuplestore.
 451                          */
 452                         if (!portal->portalUtilReady)
 453                         {
 454                                 DestReceiver *treceiver;
 455
 456                                 PortalCreateHoldStore(portal);
 457                                 treceiver = CreateDestReceiver(Tuplestore, portal);
 458                                 PortalRunUtility(portal, lfirst(portal->parseTrees),
 459                                                                  treceiver, NULL);
 460                                 (*treceiver->rDestroy) (treceiver);
 461                                 portal->portalUtilReady = true;
 462                         }
 463
 464                         /*
 465                          * Now fetch desired portion of results.
 466                          */
 467                         (void) PortalRunSelect(portal, true, count, dest);
 468
 469                         /*
 470                          * We know the query is supposed to set the tag; we assume
 471                          * only the default tag is needed.
 472                          */
 473                         if (completionTag && portal->commandTag)
 474                                 strcpy(completionTag, portal->commandTag);
 475
 476                         /*
 477                          * Since it's a forward fetch, say DONE iff atEnd is now true.
 478                          */
 479                         result = portal->atEnd;
 480                         break;
 481
 482                 case PORTAL_MULTI_QUERY:
 483                         PortalRunMulti(portal, dest, altdest, completionTag);
 484                         /* Always complete at end of RunMulti */
 485                         result = true;
 486                         break;
 487
 488                 default:
 489                         elog(ERROR, "unrecognized portal strategy: %d",
 490                                  (int) portal->strategy);
 491                         result = false;         /* keep compiler quiet */
 492                         break;
 493         }
 494
 495         MemoryContextSwitchTo(oldContext);
 496
 497         /* Mark portal not active */
 498         portal->portalActive = false;
 499
 500         PortalContext = savePortalContext;
 501         QueryContext = saveQueryContext;
 502
 503         return result;
 504 }
 505
 506 /*
 507  * PortalRunSelect
 508  *              Execute a portal's query in SELECT cases (also UTIL_SELECT).
 509  *
 510  * This handles simple N-rows-forward-or-backward cases.  For more complex
 511  * nonsequential access to a portal, see PortalRunFetch.
 512  *
 513  * count <= 0 is interpreted as a no-op: the destination gets started up
 514  * and shut down, but nothing else happens.  Also, count == FETCH_ALL is
 515  * interpreted as "all rows".
 516  *
 517  * Caller must already have validated the Portal and done appropriate
 518  * setup (cf. PortalRun).
 519  *
 520  * Returns number of rows processed (suitable for use in result tag)
 521  */
 522 static long
 523 PortalRunSelect(Portal portal,
 524                                 bool forward,
 525                                 long count,
 526                                 DestReceiver *dest)
 527 {
 528         QueryDesc  *queryDesc;
 529         ScanDirection direction;
 530         uint32          nprocessed;
 531
 532         /*
 533          * NB: queryDesc will be NULL if we are fetching from a held cursor or
 534          * a completed utility query; can't use it in that path.
 535          */
 536         queryDesc = PortalGetQueryDesc(portal);
 537
 538         /* Caller messed up if we have neither a ready query nor held data. */
 539         Assert(queryDesc || portal->holdStore);
 540
 541         /*
 542          * Force the queryDesc destination to the right thing.  This supports
 543          * MOVE, for example, which will pass in dest = None.  This is okay to
 544          * change as long as we do it on every fetch.  (The Executor must not
 545          * assume that dest never changes.)
 546          */
 547         if (queryDesc)
 548                 queryDesc->dest = dest;
 549
 550         /*
 551          * Determine which direction to go in, and check to see if we're
 552          * already at the end of the available tuples in that direction.  If
 553          * so, set the direction to NoMovement to avoid trying to fetch any
 554          * tuples.      (This check exists because not all plan node types are
 555          * robust about being called again if they've already returned NULL
 556          * once.)  Then call the executor (we must not skip this, because the
 557          * destination needs to see a setup and shutdown even if no tuples are
 558          * available).  Finally, update the portal position state depending on
 559          * the number of tuples that were retrieved.
 560          */
 561         if (forward)
 562         {
 563                 if (portal->atEnd || count <= 0)
 564                         direction = NoMovementScanDirection;
 565                 else
 566                         direction = ForwardScanDirection;
 567
 568                 /* In the executor, zero count processes all rows */
 569                 if (count == FETCH_ALL)
 570                         count = 0;
 571
 572                 if (portal->holdStore)
 573                         nprocessed = RunFromStore(portal, direction, count, dest);
 574                 else
 575                 {
 576                         ExecutorRun(queryDesc, direction, count);
 577                         nprocessed = queryDesc->estate->es_processed;
 578                 }
 579
 580                 if (direction != NoMovementScanDirection)
 581                 {
 582                         long            oldPos;
 583
 584                         if (nprocessed > 0)
 585                                 portal->atStart = false;                /* OK to go backward now */
 586                         if (count == 0 ||
 587                                 (unsigned long) nprocessed < (unsigned long) count)
 588                                 portal->atEnd = true;   /* we retrieved 'em all */
 589                         oldPos = portal->portalPos;
 590                         portal->portalPos += nprocessed;
 591                         /* portalPos doesn't advance when we fall off the end */
 592                         if (portal->portalPos < oldPos)
 593                                 portal->posOverflow = true;
 594                 }
 595         }
 596         else
 597         {
 598                 if (portal->cursorOptions & CURSOR_OPT_NO_SCROLL)
 599                         ereport(ERROR,
 600                                         (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 601                                          errmsg("cursor can only scan forward"),
 602                                          errhint("Declare it with SCROLL option to enable backward scan.")));
 603
 604                 if (portal->atStart || count <= 0)
 605                         direction = NoMovementScanDirection;
 606                 else
 607                         direction = BackwardScanDirection;
 608
 609                 /* In the executor, zero count processes all rows */
 610                 if (count == FETCH_ALL)
 611                         count = 0;
 612
 613                 if (portal->holdStore)
 614                         nprocessed = RunFromStore(portal, direction, count, dest);
 615                 else
 616                 {
 617                         ExecutorRun(queryDesc, direction, count);
 618                         nprocessed = queryDesc->estate->es_processed;
 619                 }
 620
 621                 if (direction != NoMovementScanDirection)
 622                 {
 623                         if (nprocessed > 0 && portal->atEnd)
 624                         {
 625                                 portal->atEnd = false;  /* OK to go forward now */
 626                                 portal->portalPos++;    /* adjust for endpoint case */
 627                         }
 628                         if (count == 0 ||
 629                                 (unsigned long) nprocessed < (unsigned long) count)
 630                         {
 631                                 portal->atStart = true; /* we retrieved 'em all */
 632                                 portal->portalPos = 0;
 633                                 portal->posOverflow = false;
 634                         }
 635                         else
 636                         {
 637                                 long            oldPos;
 638
 639                                 oldPos = portal->portalPos;
 640                                 portal->portalPos -= nprocessed;
 641                                 if (portal->portalPos > oldPos ||
 642                                         portal->portalPos <= 0)
 643                                         portal->posOverflow = true;
 644                         }
 645                 }
 646         }
 647
 648         return nprocessed;
 649 }
 650
 651 /*
 652  * RunFromStore
 653  *              Fetch tuples from the portal's tuple store.
 654  *
 655  * Calling conventions are similar to ExecutorRun, except that we
 656  * do not depend on having a queryDesc or estate.  Therefore we return the
 657  * number of tuples processed as the result, not in estate->es_processed.
 658  *
 659  * One difference from ExecutorRun is that the destination receiver functions
 660  * are run in the caller's memory context (since we have no estate).  Watch
 661  * out for memory leaks.
 662  */
 663 static uint32
 664 RunFromStore(Portal portal, ScanDirection direction, long count,
 665                          DestReceiver *dest)
 666 {
 667         long            current_tuple_count = 0;
 668
 669         (*dest->rStartup) (dest, CMD_SELECT, portal->tupDesc);
 670
 671         if (direction == NoMovementScanDirection)
 672         {
 673                 /* do nothing except start/stop the destination */
 674         }
 675         else
 676         {
 677                 bool            forward = (direction == ForwardScanDirection);
 678
 679                 for (;;)
 680                 {
 681                         MemoryContext oldcontext;
 682                         HeapTuple       tup;
 683                         bool            should_free;
 684
 685                         oldcontext = MemoryContextSwitchTo(portal->holdContext);
 686
 687                         tup = tuplestore_getheaptuple(portal->holdStore, forward,
 688                                                                                   &should_free);
 689
 690                         MemoryContextSwitchTo(oldcontext);
 691
 692                         if (tup == NULL)
 693                                 break;
 694
 695                         (*dest->receiveTuple) (tup, portal->tupDesc, dest);
 696
 697                         if (should_free)
 698                                 pfree(tup);
 699
 700                         /*
 701                          * check our tuple count.. if we've processed the proper
 702                          * number then quit, else loop again and process more tuples.
 703                          * Zero count means no limit.
 704                          */
 705                         current_tuple_count++;
 706                         if (count && count == current_tuple_count)
 707                                 break;
 708                 }
 709         }
 710
 711         (*dest->rShutdown) (dest);
 712
 713         return (uint32) current_tuple_count;
 714 }
 715
 716 /*
 717  * PortalRunUtility
 718  *              Execute a utility statement inside a portal.
 719  */
 720 static void
 721 PortalRunUtility(Portal portal, Query *query,
 722                                  DestReceiver *dest, char *completionTag)
 723 {
 724         Node       *utilityStmt = query->utilityStmt;
 725
 726         ereport(DEBUG3,
 727                         (errmsg_internal("ProcessUtility")));
 728
 729         /*
 730          * Set snapshot if utility stmt needs one.      Most reliable way to do
 731          * this seems to be to enumerate those that do not need one; this is a
 732          * short list.  Transaction control, LOCK, and SET must *not* set a
 733          * snapshot since they need to be executable at the start of a
 734          * serializable transaction without freezing a snapshot.  By extension
 735          * we allow SHOW not to set a snapshot.  The other stmts listed are
 736          * just efficiency hacks.  Beware of listing anything that can modify
 737          * the database --- if, say, it has to update an index with
 738          * expressions that invoke user-defined functions, then it had better
 739          * have a snapshot.
 740          */
 741         if (!(IsA(utilityStmt, TransactionStmt) ||
 742                   IsA(utilityStmt, LockStmt) ||
 743                   IsA(utilityStmt, VariableSetStmt) ||
 744                   IsA(utilityStmt, VariableShowStmt) ||
 745                   IsA(utilityStmt, VariableResetStmt) ||
 746                   IsA(utilityStmt, ConstraintsSetStmt) ||
 747         /* efficiency hacks from here down */
 748                   IsA(utilityStmt, FetchStmt) ||
 749                   IsA(utilityStmt, ListenStmt) ||
 750                   IsA(utilityStmt, NotifyStmt) ||
 751                   IsA(utilityStmt, UnlistenStmt) ||
 752                   IsA(utilityStmt, CheckPointStmt)))
 753                 SetQuerySnapshot();
 754
 755         if (query->canSetTag)
 756         {
 757                 /* utility statement can override default tag string */
 758                 ProcessUtility(utilityStmt, dest, completionTag);
 759                 if (completionTag && completionTag[0] == '\0' && portal->commandTag)
 760                         strcpy(completionTag, portal->commandTag);      /* use the default */
 761         }
 762         else
 763         {
 764                 /* utility added by rewrite cannot set tag */
 765                 ProcessUtility(utilityStmt, dest, NULL);
 766         }
 767
 768         /* Some utility statements may change context on us */
 769         MemoryContextSwitchTo(PortalGetHeapMemory(portal));
 770 }
 771
 772 /*
 773  * PortalRunMulti
 774  *              Execute a portal's queries in the general case (multi queries).
 775  */
 776 static void
 777 PortalRunMulti(Portal portal,
 778                            DestReceiver *dest, DestReceiver *altdest,
 779                            char *completionTag)
 780 {
 781         List       *plantree_list = portal->planTrees;
 782         List       *querylist_item;
 783
 784         /*
 785          * If the destination is RemoteExecute, change to None.  The reason is
 786          * that the client won't be expecting any tuples, and indeed has no
 787          * way to know what they are, since there is no provision for Describe
 788          * to send a RowDescription message when this portal execution
 789          * strategy is in effect.  This presently will only affect SELECT
 790          * commands added to non-SELECT queries by rewrite rules: such
 791          * commands will be executed, but the results will be discarded unless
 792          * you use "simple Query" protocol.
 793          */
 794         if (dest->mydest == RemoteExecute)
 795                 dest = None_Receiver;
 796         if (altdest->mydest == RemoteExecute)
 797                 altdest = None_Receiver;
 798
 799         /*
 800          * Loop to handle the individual queries generated from a single
 801          * parsetree by analysis and rewrite.
 802          */
 803         foreach(querylist_item, portal->parseTrees)
 804         {
 805                 Query      *query = (Query *) lfirst(querylist_item);
 806                 Plan       *plan = (Plan *) lfirst(plantree_list);
 807
 808                 plantree_list = lnext(plantree_list);
 809
 810                 /*
 811                  * If we got a cancel signal in prior command, quit
 812                  */
 813                 CHECK_FOR_INTERRUPTS();
 814
 815                 if (query->commandType == CMD_UTILITY)
 816                 {
 817                         /*
 818                          * process utility functions (create, destroy, etc..)
 819                          */
 820                         Assert(plan == NULL);
 821
 822                         PortalRunUtility(portal, query,
 823                                                          query->canSetTag ? dest : altdest,
 824                                                          completionTag);
 825                 }
 826                 else
 827                 {
 828                         /*
 829                          * process a plannable query.
 830                          */
 831                         ereport(DEBUG3,
 832                                         (errmsg_internal("ProcessQuery")));
 833
 834                         /* Must always set snapshot for plannable queries */
 835                         SetQuerySnapshot();
 836
 837                         /*
 838                          * execute the plan
 839                          */
 840                         if (log_executor_stats)
 841                                 ResetUsage();
 842
 843                         if (query->canSetTag)
 844                         {
 845                                 /* statement can set tag string */
 846                                 ProcessQuery(query, plan,
 847                                                          portal->portalParams,
 848                                                          dest, completionTag);
 849                         }
 850                         else
 851                         {
 852                                 /* stmt added by rewrite cannot set tag */
 853                                 ProcessQuery(query, plan,
 854                                                          portal->portalParams,
 855                                                          altdest, NULL);
 856                         }
 857
 858                         if (log_executor_stats)
 859                                 ShowUsage("EXECUTOR STATISTICS");
 860                 }
 861
 862                 /*
 863                  * Increment command counter between queries, but not after the
 864                  * last one.
 865                  */
 866                 if (plantree_list != NIL)
 867                         CommandCounterIncrement();
 868
 869                 /*
 870                  * Clear subsidiary contexts to recover temporary memory.
 871                  */
 872                 Assert(PortalGetHeapMemory(portal) == CurrentMemoryContext);
 873
 874                 MemoryContextDeleteChildren(PortalGetHeapMemory(portal));
 875         }
 876
 877         /*
 878          * If a command completion tag was supplied, use it.  Otherwise use
 879          * the portal's commandTag as the default completion tag.
 880          *
 881          * Exception: clients will expect INSERT/UPDATE/DELETE tags to have
 882          * counts, so fake something up if necessary.  (This could happen if
 883          * the original query was replaced by a DO INSTEAD rule.)
 884          */
 885         if (completionTag && completionTag[0] == '\0')
 886         {
 887                 if (portal->commandTag)
 888                         strcpy(completionTag, portal->commandTag);
 889                 if (strcmp(completionTag, "INSERT") == 0)
 890                         strcpy(completionTag, "INSERT 0 0");
 891                 else if (strcmp(completionTag, "UPDATE") == 0)
 892                         strcpy(completionTag, "UPDATE 0");
 893                 else if (strcmp(completionTag, "DELETE") == 0)
 894                         strcpy(completionTag, "DELETE 0");
 895         }
 896
 897         /* Prevent portal's commands from being re-executed */
 898         portal->portalDone = true;
 899 }
 900
 901 /*
 902  * PortalRunFetch
 903  *              Variant form of PortalRun that supports SQL FETCH directions.
 904  *
 905  * Returns number of rows processed (suitable for use in result tag)
 906  */
 907 long
 908 PortalRunFetch(Portal portal,
 909                            FetchDirection fdirection,
 910                            long count,
 911                            DestReceiver *dest)
 912 {
 913         long            result;
 914         MemoryContext savePortalContext;
 915         MemoryContext saveQueryContext;
 916         MemoryContext oldContext;
 917
 918         AssertArg(PortalIsValid(portal));
 919         AssertState(portal->portalReady);       /* else no PortalStart */
 920
 921         /*
 922          * Check for improper portal use, and mark portal active.
 923          */
 924         if (portal->portalDone)
 925                 ereport(ERROR,
 926                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 927                    errmsg("portal \"%s\" cannot be run anymore", portal->name)));
 928         if (portal->portalActive)
 929                 ereport(ERROR,
 930                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 931                                  errmsg("portal \"%s\" already active", portal->name)));
 932         portal->portalActive = true;
 933
 934         /*
 935          * Set global portal context pointers.
 936          */
 937         savePortalContext = PortalContext;
 938         PortalContext = PortalGetHeapMemory(portal);
 939         saveQueryContext = QueryContext;
 940         QueryContext = portal->queryContext;
 941
 942         oldContext = MemoryContextSwitchTo(PortalContext);
 943
 944         switch (portal->strategy)
 945         {
 946                 case PORTAL_ONE_SELECT:
 947                         result = DoPortalRunFetch(portal, fdirection, count, dest);
 948                         break;
 949
 950                 default:
 951                         elog(ERROR, "unsupported portal strategy");
 952                         result = 0;                     /* keep compiler quiet */
 953                         break;
 954         }
 955
 956         MemoryContextSwitchTo(oldContext);
 957
 958         /* Mark portal not active */
 959         portal->portalActive = false;
 960
 961         PortalContext = savePortalContext;
 962         QueryContext = saveQueryContext;
 963
 964         return result;
 965 }
 966
 967 /*
 968  * DoPortalRunFetch
 969  *              Guts of PortalRunFetch --- the portal context is already set up
 970  *
 971  * Returns number of rows processed (suitable for use in result tag)
 972  */
 973 static long
 974 DoPortalRunFetch(Portal portal,
 975                                  FetchDirection fdirection,
 976                                  long count,
 977                                  DestReceiver *dest)
 978 {
 979         bool            forward;
 980
 981         Assert(portal->strategy == PORTAL_ONE_SELECT);
 982
 983         switch (fdirection)
 984         {
 985                 case FETCH_FORWARD:
 986                         if (count < 0)
 987                         {
 988                                 fdirection = FETCH_BACKWARD;
 989                                 count = -count;
 990                         }
 991                         /* fall out of switch to share code with FETCH_BACKWARD */
 992                         break;
 993                 case FETCH_BACKWARD:
 994                         if (count < 0)
 995                         {
 996                                 fdirection = FETCH_FORWARD;
 997                                 count = -count;
 998                         }
 999                         /* fall out of switch to share code with FETCH_FORWARD */
1000                         break;
1001                 case FETCH_ABSOLUTE:
1002                         if (count > 0)
1003                         {
1004                                 /*
1005                                  * Definition: Rewind to start, advance count-1 rows,
1006                                  * return next row (if any).  In practice, if the goal is
1007                                  * less than halfway back to the start, it's better to
1008                                  * scan from where we are.      In any case, we arrange to
1009                                  * fetch the target row going forwards.
1010                                  */
1011                                 if (portal->posOverflow || portal->portalPos == LONG_MAX ||
1012                                         count - 1 <= portal->portalPos / 2)
1013                                 {
1014                                         DoPortalRewind(portal);
1015                                         if (count > 1)
1016                                                 PortalRunSelect(portal, true, count - 1,
1017                                                                                 None_Receiver);
1018                                 }
1019                                 else
1020                                 {
1021                                         long            pos = portal->portalPos;
1022
1023                                         if (portal->atEnd)
1024                                                 pos++;  /* need one extra fetch if off end */
1025                                         if (count <= pos)
1026                                                 PortalRunSelect(portal, false, pos - count + 1,
1027                                                                                 None_Receiver);
1028                                         else if (count > pos + 1)
1029                                                 PortalRunSelect(portal, true, count - pos - 1,
1030                                                                                 None_Receiver);
1031                                 }
1032                                 return PortalRunSelect(portal, true, 1L, dest);
1033                         }
1034                         else if (count < 0)
1035                         {
1036                                 /*
1037                                  * Definition: Advance to end, back up abs(count)-1 rows,
1038                                  * return prior row (if any).  We could optimize this if
1039                                  * we knew in advance where the end was, but typically we
1040                                  * won't. (Is it worth considering case where count > half
1041                                  * of size of query?  We could rewind once we know the
1042                                  * size ...)
1043                                  */
1044                                 PortalRunSelect(portal, true, FETCH_ALL, None_Receiver);
1045                                 if (count < -1)
1046                                         PortalRunSelect(portal, false, -count - 1, None_Receiver);
1047                                 return PortalRunSelect(portal, false, 1L, dest);
1048                         }
1049                         else
1050 /* count == 0 */
1051                         {
1052                                 /* Rewind to start, return zero rows */
1053                                 DoPortalRewind(portal);
1054                                 return PortalRunSelect(portal, true, 0L, dest);
1055                         }
1056                         break;
1057                 case FETCH_RELATIVE:
1058                         if (count > 0)
1059                         {
1060                                 /*
1061                                  * Definition: advance count-1 rows, return next row (if
1062                                  * any).
1063                                  */
1064                                 if (count > 1)
1065                                         PortalRunSelect(portal, true, count - 1, None_Receiver);
1066                                 return PortalRunSelect(portal, true, 1L, dest);
1067                         }
1068                         else if (count < 0)
1069                         {
1070                                 /*
1071                                  * Definition: back up abs(count)-1 rows, return prior row
1072                                  * (if any).
1073                                  */
1074                                 if (count < -1)
1075                                         PortalRunSelect(portal, false, -count - 1, None_Receiver);
1076                                 return PortalRunSelect(portal, false, 1L, dest);
1077                         }
1078                         else
1079 /* count == 0 */
1080                         {
1081                                 /* Same as FETCH FORWARD 0, so fall out of switch */
1082                                 fdirection = FETCH_FORWARD;
1083                         }
1084                         break;
1085                 default:
1086                         elog(ERROR, "bogus direction");
1087                         break;
1088         }
1089
1090         /*
1091          * Get here with fdirection == FETCH_FORWARD or FETCH_BACKWARD, and
1092          * count >= 0.
1093          */
1094         forward = (fdirection == FETCH_FORWARD);
1095
1096         /*
1097          * Zero count means to re-fetch the current row, if any (per SQL92)
1098          */
1099         if (count == 0)
1100         {
1101                 bool            on_row;
1102
1103                 /* Are we sitting on a row? */
1104                 on_row = (!portal->atStart && !portal->atEnd);
1105
1106                 if (dest->mydest == None)
1107                 {
1108                         /* MOVE 0 returns 0/1 based on if FETCH 0 would return a row */
1109                         return on_row ? 1L : 0L;
1110                 }
1111                 else
1112                 {
1113                         /*
1114                          * If we are sitting on a row, back up one so we can re-fetch
1115                          * it. If we are not sitting on a row, we still have to start
1116                          * up and shut down the executor so that the destination is
1117                          * initialized and shut down correctly; so keep going.  To
1118                          * PortalRunSelect, count == 0 means we will retrieve no row.
1119                          */
1120                         if (on_row)
1121                         {
1122                                 PortalRunSelect(portal, false, 1L, None_Receiver);
1123                                 /* Set up to fetch one row forward */
1124                                 count = 1;
1125                                 forward = true;
1126                         }
1127                 }
1128         }
1129
1130         /*
1131          * Optimize MOVE BACKWARD ALL into a Rewind.
1132          */
1133         if (!forward && count == FETCH_ALL && dest->mydest == None)
1134         {
1135                 long            result = portal->portalPos;
1136
1137                 if (result > 0 && !portal->atEnd)
1138                         result--;
1139                 DoPortalRewind(portal);
1140                 /* result is bogus if pos had overflowed, but it's best we can do */
1141                 return result;
1142         }
1143
1144         return PortalRunSelect(portal, forward, count, dest);
1145 }
1146
1147 /*
1148  * DoPortalRewind - rewind a Portal to starting point
1149  */
1150 static void
1151 DoPortalRewind(Portal portal)
1152 {
1153         if (portal->holdStore)
1154         {
1155                 MemoryContext oldcontext;
1156
1157                 oldcontext = MemoryContextSwitchTo(portal->holdContext);
1158                 tuplestore_rescan(portal->holdStore);
1159                 MemoryContextSwitchTo(oldcontext);
1160         }
1161         if (PortalGetQueryDesc(portal))
1162                 ExecutorRewind(PortalGetQueryDesc(portal));
1163
1164         portal->atStart = true;
1165         portal->atEnd = false;
1166         portal->portalPos = 0;
1167         portal->posOverflow = false;
1168 }