src/backend/executor/spi.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * spi.c
   4  *                              Server Programming Interface
   5  *
   6  * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        src/backend/executor/spi.c
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 #include "postgres.h"
  16
  17 #include "access/htup_details.h"
  18 #include "access/printtup.h"
  19 #include "access/sysattr.h"
  20 #include "access/xact.h"
  21 #include "catalog/heap.h"
  22 #include "catalog/pg_type.h"
  23 #include "commands/trigger.h"
  24 #include "executor/executor.h"
  25 #include "executor/spi_priv.h"
  26 #include "tcop/pquery.h"
  27 #include "tcop/utility.h"
  28 #include "utils/builtins.h"
  29 #include "utils/datum.h"
  30 #include "utils/lsyscache.h"
  31 #include "utils/memutils.h"
  32 #include "utils/rel.h"
  33 #include "utils/snapmgr.h"
  34 #include "utils/syscache.h"
  35 #include "utils/typcache.h"
  36
  37
  38 /*
  39  * These global variables are part of the API for various SPI functions
  40  * (a horrible API choice, but it's too late now).  To reduce the risk of
  41  * interference between different SPI callers, we save and restore them
  42  * when entering/exiting a SPI nesting level.
  43  */
  44 uint64          SPI_processed = 0;
  45 SPITupleTable *SPI_tuptable = NULL;
  46 int                     SPI_result = 0;
  47
  48 static _SPI_connection *_SPI_stack = NULL;
  49 static _SPI_connection *_SPI_current = NULL;
  50 static int      _SPI_stack_depth = 0;   /* allocated size of _SPI_stack */
  51 static int      _SPI_connected = -1;    /* current stack index */
  52
  53 typedef struct SPICallbackArg
  54 {
  55         const char *query;
  56         RawParseMode mode;
  57 } SPICallbackArg;
  58
  59 static Portal SPI_cursor_open_internal(const char *name, SPIPlanPtr plan,
  60                                                                            ParamListInfo paramLI, bool read_only);
  61
  62 static void _SPI_prepare_plan(const char *src, SPIPlanPtr plan);
  63
  64 static void _SPI_prepare_oneshot_plan(const char *src, SPIPlanPtr plan);
  65
  66 static int      _SPI_execute_plan(SPIPlanPtr plan, const SPIExecuteOptions *options,
  67                                                           Snapshot snapshot, Snapshot crosscheck_snapshot,
  68                                                           bool fire_triggers);
  69
  70 static ParamListInfo _SPI_convert_params(int nargs, Oid *argtypes,
  71                                                                                  Datum *Values, const char *Nulls);
  72
  73 static int      _SPI_pquery(QueryDesc *queryDesc, bool fire_triggers, uint64 tcount);
  74
  75 static void _SPI_error_callback(void *arg);
  76
  77 static void _SPI_cursor_operation(Portal portal,
  78                                                                   FetchDirection direction, long count,
  79                                                                   DestReceiver *dest);
  80
  81 static SPIPlanPtr _SPI_make_plan_non_temp(SPIPlanPtr plan);
  82 static SPIPlanPtr _SPI_save_plan(SPIPlanPtr plan);
  83
  84 static int      _SPI_begin_call(bool use_exec);
  85 static int      _SPI_end_call(bool use_exec);
  86 static MemoryContext _SPI_execmem(void);
  87 static MemoryContext _SPI_procmem(void);
  88 static bool _SPI_checktuples(void);
  89
  90
  91 /* =================== interface functions =================== */
  92
  93 int
  94 SPI_connect(void)
  95 {
  96         return SPI_connect_ext(0);
  97 }
  98
  99 int
 100 SPI_connect_ext(int options)
 101 {
 102         int                     newdepth;
 103
 104         /* Enlarge stack if necessary */
 105         if (_SPI_stack == NULL)
 106         {
 107                 if (_SPI_connected != -1 || _SPI_stack_depth != 0)
 108                         elog(ERROR, "SPI stack corrupted");
 109                 newdepth = 16;
 110                 _SPI_stack = (_SPI_connection *)
 111                         MemoryContextAlloc(TopMemoryContext,
 112                                                            newdepth * sizeof(_SPI_connection));
 113                 _SPI_stack_depth = newdepth;
 114         }
 115         else
 116         {
 117                 if (_SPI_stack_depth <= 0 || _SPI_stack_depth <= _SPI_connected)
 118                         elog(ERROR, "SPI stack corrupted");
 119                 if (_SPI_stack_depth == _SPI_connected + 1)
 120                 {
 121                         newdepth = _SPI_stack_depth * 2;
 122                         _SPI_stack = (_SPI_connection *)
 123                                 repalloc(_SPI_stack,
 124                                                  newdepth * sizeof(_SPI_connection));
 125                         _SPI_stack_depth = newdepth;
 126                 }
 127         }
 128
 129         /* Enter new stack level */
 130         _SPI_connected++;
 131         Assert(_SPI_connected >= 0 && _SPI_connected < _SPI_stack_depth);
 132
 133         _SPI_current = &(_SPI_stack[_SPI_connected]);
 134         _SPI_current->processed = 0;
 135         _SPI_current->tuptable = NULL;
 136         _SPI_current->execSubid = InvalidSubTransactionId;
 137         slist_init(&_SPI_current->tuptables);
 138         _SPI_current->procCxt = NULL;   /* in case we fail to create 'em */
 139         _SPI_current->execCxt = NULL;
 140         _SPI_current->connectSubid = GetCurrentSubTransactionId();
 141         _SPI_current->queryEnv = NULL;
 142         _SPI_current->atomic = (options & SPI_OPT_NONATOMIC ? false : true);
 143         _SPI_current->internal_xact = false;
 144         _SPI_current->outer_processed = SPI_processed;
 145         _SPI_current->outer_tuptable = SPI_tuptable;
 146         _SPI_current->outer_result = SPI_result;
 147
 148         /*
 149          * Create memory contexts for this procedure
 150          *
 151          * In atomic contexts (the normal case), we use TopTransactionContext,
 152          * otherwise PortalContext, so that it lives across transaction
 153          * boundaries.
 154          *
 155          * XXX It could be better to use PortalContext as the parent context in
 156          * all cases, but we may not be inside a portal (consider deferred-trigger
 157          * execution).  Perhaps CurTransactionContext could be an option?  For now
 158          * it doesn't matter because we clean up explicitly in AtEOSubXact_SPI();
 159          * but see also AtEOXact_SPI().
 160          */
 161         _SPI_current->procCxt = AllocSetContextCreate(_SPI_current->atomic ? TopTransactionContext : PortalContext,
 162                                                                                                   "SPI Proc",
 163                                                                                                   ALLOCSET_DEFAULT_SIZES);
 164         _SPI_current->execCxt = AllocSetContextCreate(_SPI_current->atomic ? TopTransactionContext : _SPI_current->procCxt,
 165                                                                                                   "SPI Exec",
 166                                                                                                   ALLOCSET_DEFAULT_SIZES);
 167         /* ... and switch to procedure's context */
 168         _SPI_current->savedcxt = MemoryContextSwitchTo(_SPI_current->procCxt);
 169
 170         /*
 171          * Reset API global variables so that current caller cannot accidentally
 172          * depend on state of an outer caller.
 173          */
 174         SPI_processed = 0;
 175         SPI_tuptable = NULL;
 176         SPI_result = 0;
 177
 178         return SPI_OK_CONNECT;
 179 }
 180
 181 int
 182 SPI_finish(void)
 183 {
 184         int                     res;
 185
 186         res = _SPI_begin_call(false);   /* just check we're connected */
 187         if (res < 0)
 188                 return res;
 189
 190         /* Restore memory context as it was before procedure call */
 191         MemoryContextSwitchTo(_SPI_current->savedcxt);
 192
 193         /* Release memory used in procedure call (including tuptables) */
 194         MemoryContextDelete(_SPI_current->execCxt);
 195         _SPI_current->execCxt = NULL;
 196         MemoryContextDelete(_SPI_current->procCxt);
 197         _SPI_current->procCxt = NULL;
 198
 199         /*
 200          * Restore outer API variables, especially SPI_tuptable which is probably
 201          * pointing at a just-deleted tuptable
 202          */
 203         SPI_processed = _SPI_current->outer_processed;
 204         SPI_tuptable = _SPI_current->outer_tuptable;
 205         SPI_result = _SPI_current->outer_result;
 206
 207         /* Exit stack level */
 208         _SPI_connected--;
 209         if (_SPI_connected < 0)
 210                 _SPI_current = NULL;
 211         else
 212                 _SPI_current = &(_SPI_stack[_SPI_connected]);
 213
 214         return SPI_OK_FINISH;
 215 }
 216
 217 /*
 218  * SPI_start_transaction is a no-op, kept for backwards compatibility.
 219  * SPI callers are *always* inside a transaction.
 220  */
 221 void
 222 SPI_start_transaction(void)
 223 {
 224 }
 225
 226 static void
 227 _SPI_commit(bool chain)
 228 {
 229         MemoryContext oldcontext = CurrentMemoryContext;
 230         SavedTransactionCharacteristics savetc;
 231
 232         /*
 233          * Complain if we are in a context that doesn't permit transaction
 234          * termination.  (Note: here and _SPI_rollback should be the only places
 235          * that throw ERRCODE_INVALID_TRANSACTION_TERMINATION, so that callers can
 236          * test for that with security that they know what happened.)
 237          */
 238         if (_SPI_current->atomic)
 239                 ereport(ERROR,
 240                                 (errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
 241                                  errmsg("invalid transaction termination")));
 242
 243         /*
 244          * This restriction is required by PLs implemented on top of SPI.  They
 245          * use subtransactions to establish exception blocks that are supposed to
 246          * be rolled back together if there is an error.  Terminating the
 247          * top-level transaction in such a block violates that idea.  A future PL
 248          * implementation might have different ideas about this, in which case
 249          * this restriction would have to be refined or the check possibly be
 250          * moved out of SPI into the PLs.  Note however that the code below relies
 251          * on not being within a subtransaction.
 252          */
 253         if (IsSubTransaction())
 254                 ereport(ERROR,
 255                                 (errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
 256                                  errmsg("cannot commit while a subtransaction is active")));
 257
 258         if (chain)
 259                 SaveTransactionCharacteristics(&savetc);
 260
 261         /* Catch any error occurring during the COMMIT */
 262         PG_TRY();
 263         {
 264                 /* Protect current SPI stack entry against deletion */
 265                 _SPI_current->internal_xact = true;
 266
 267                 /*
 268                  * Hold any pinned portals that any PLs might be using.  We have to do
 269                  * this before changing transaction state, since this will run
 270                  * user-defined code that might throw an error.
 271                  */
 272                 HoldPinnedPortals();
 273
 274                 /* Release snapshots associated with portals */
 275                 ForgetPortalSnapshots();
 276
 277                 /* Do the deed */
 278                 CommitTransactionCommand();
 279
 280                 /* Immediately start a new transaction */
 281                 StartTransactionCommand();
 282                 if (chain)
 283                         RestoreTransactionCharacteristics(&savetc);
 284
 285                 MemoryContextSwitchTo(oldcontext);
 286
 287                 _SPI_current->internal_xact = false;
 288         }
 289         PG_CATCH();
 290         {
 291                 ErrorData  *edata;
 292
 293                 /* Save error info in caller's context */
 294                 MemoryContextSwitchTo(oldcontext);
 295                 edata = CopyErrorData();
 296                 FlushErrorState();
 297
 298                 /*
 299                  * Abort the failed transaction.  If this fails too, we'll just
 300                  * propagate the error out ... there's not that much we can do.
 301                  */
 302                 AbortCurrentTransaction();
 303
 304                 /* ... and start a new one */
 305                 StartTransactionCommand();
 306                 if (chain)
 307                         RestoreTransactionCharacteristics(&savetc);
 308
 309                 MemoryContextSwitchTo(oldcontext);
 310
 311                 _SPI_current->internal_xact = false;
 312
 313                 /* Now that we've cleaned up the transaction, re-throw the error */
 314                 ReThrowError(edata);
 315         }
 316         PG_END_TRY();
 317 }
 318
 319 void
 320 SPI_commit(void)
 321 {
 322         _SPI_commit(false);
 323 }
 324
 325 void
 326 SPI_commit_and_chain(void)
 327 {
 328         _SPI_commit(true);
 329 }
 330
 331 static void
 332 _SPI_rollback(bool chain)
 333 {
 334         MemoryContext oldcontext = CurrentMemoryContext;
 335         SavedTransactionCharacteristics savetc;
 336
 337         /* see comments in _SPI_commit() */
 338         if (_SPI_current->atomic)
 339                 ereport(ERROR,
 340                                 (errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
 341                                  errmsg("invalid transaction termination")));
 342
 343         /* see comments in _SPI_commit() */
 344         if (IsSubTransaction())
 345                 ereport(ERROR,
 346                                 (errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
 347                                  errmsg("cannot roll back while a subtransaction is active")));
 348
 349         if (chain)
 350                 SaveTransactionCharacteristics(&savetc);
 351
 352         /* Catch any error occurring during the ROLLBACK */
 353         PG_TRY();
 354         {
 355                 /* Protect current SPI stack entry against deletion */
 356                 _SPI_current->internal_xact = true;
 357
 358                 /*
 359                  * Hold any pinned portals that any PLs might be using.  We have to do
 360                  * this before changing transaction state, since this will run
 361                  * user-defined code that might throw an error, and in any case
 362                  * couldn't be run in an already-aborted transaction.
 363                  */
 364                 HoldPinnedPortals();
 365
 366                 /* Release snapshots associated with portals */
 367                 ForgetPortalSnapshots();
 368
 369                 /* Do the deed */
 370                 AbortCurrentTransaction();
 371
 372                 /* Immediately start a new transaction */
 373                 StartTransactionCommand();
 374                 if (chain)
 375                         RestoreTransactionCharacteristics(&savetc);
 376
 377                 MemoryContextSwitchTo(oldcontext);
 378
 379                 _SPI_current->internal_xact = false;
 380         }
 381         PG_CATCH();
 382         {
 383                 ErrorData  *edata;
 384
 385                 /* Save error info in caller's context */
 386                 MemoryContextSwitchTo(oldcontext);
 387                 edata = CopyErrorData();
 388                 FlushErrorState();
 389
 390                 /*
 391                  * Try again to abort the failed transaction.  If this fails too,
 392                  * we'll just propagate the error out ... there's not that much we can
 393                  * do.
 394                  */
 395                 AbortCurrentTransaction();
 396
 397                 /* ... and start a new one */
 398                 StartTransactionCommand();
 399                 if (chain)
 400                         RestoreTransactionCharacteristics(&savetc);
 401
 402                 MemoryContextSwitchTo(oldcontext);
 403
 404                 _SPI_current->internal_xact = false;
 405
 406                 /* Now that we've cleaned up the transaction, re-throw the error */
 407                 ReThrowError(edata);
 408         }
 409         PG_END_TRY();
 410 }
 411
 412 void
 413 SPI_rollback(void)
 414 {
 415         _SPI_rollback(false);
 416 }
 417
 418 void
 419 SPI_rollback_and_chain(void)
 420 {
 421         _SPI_rollback(true);
 422 }
 423
 424 /*
 425  * Clean up SPI state at transaction commit or abort.
 426  */
 427 void
 428 AtEOXact_SPI(bool isCommit)
 429 {
 430         bool            found = false;
 431
 432         /*
 433          * Pop stack entries, stopping if we find one marked internal_xact (that
 434          * one belongs to the caller of SPI_commit or SPI_rollback).
 435          */
 436         while (_SPI_connected >= 0)
 437         {
 438                 _SPI_connection *connection = &(_SPI_stack[_SPI_connected]);
 439
 440                 if (connection->internal_xact)
 441                         break;
 442
 443                 found = true;
 444
 445                 /*
 446                  * We need not release the procedure's memory contexts explicitly, as
 447                  * they'll go away automatically when their parent context does; see
 448                  * notes in SPI_connect_ext.
 449                  */
 450
 451                 /*
 452                  * Restore outer global variables and pop the stack entry.  Unlike
 453                  * SPI_finish(), we don't risk switching to memory contexts that might
 454                  * be already gone.
 455                  */
 456                 SPI_processed = connection->outer_processed;
 457                 SPI_tuptable = connection->outer_tuptable;
 458                 SPI_result = connection->outer_result;
 459
 460                 _SPI_connected--;
 461                 if (_SPI_connected < 0)
 462                         _SPI_current = NULL;
 463                 else
 464                         _SPI_current = &(_SPI_stack[_SPI_connected]);
 465         }
 466
 467         /* We should only find entries to pop during an ABORT. */
 468         if (found && isCommit)
 469                 ereport(WARNING,
 470                                 (errcode(ERRCODE_WARNING),
 471                                  errmsg("transaction left non-empty SPI stack"),
 472                                  errhint("Check for missing \"SPI_finish\" calls.")));
 473 }
 474
 475 /*
 476  * Clean up SPI state at subtransaction commit or abort.
 477  *
 478  * During commit, there shouldn't be any unclosed entries remaining from
 479  * the current subtransaction; we emit a warning if any are found.
 480  */
 481 void
 482 AtEOSubXact_SPI(bool isCommit, SubTransactionId mySubid)
 483 {
 484         bool            found = false;
 485
 486         while (_SPI_connected >= 0)
 487         {
 488                 _SPI_connection *connection = &(_SPI_stack[_SPI_connected]);
 489
 490                 if (connection->connectSubid != mySubid)
 491                         break;                          /* couldn't be any underneath it either */
 492
 493                 if (connection->internal_xact)
 494                         break;
 495
 496                 found = true;
 497
 498                 /*
 499                  * Release procedure memory explicitly (see note in SPI_connect)
 500                  */
 501                 if (connection->execCxt)
 502                 {
 503                         MemoryContextDelete(connection->execCxt);
 504                         connection->execCxt = NULL;
 505                 }
 506                 if (connection->procCxt)
 507                 {
 508                         MemoryContextDelete(connection->procCxt);
 509                         connection->procCxt = NULL;
 510                 }
 511
 512                 /*
 513                  * Restore outer global variables and pop the stack entry.  Unlike
 514                  * SPI_finish(), we don't risk switching to memory contexts that might
 515                  * be already gone.
 516                  */
 517                 SPI_processed = connection->outer_processed;
 518                 SPI_tuptable = connection->outer_tuptable;
 519                 SPI_result = connection->outer_result;
 520
 521                 _SPI_connected--;
 522                 if (_SPI_connected < 0)
 523                         _SPI_current = NULL;
 524                 else
 525                         _SPI_current = &(_SPI_stack[_SPI_connected]);
 526         }
 527
 528         if (found && isCommit)
 529                 ereport(WARNING,
 530                                 (errcode(ERRCODE_WARNING),
 531                                  errmsg("subtransaction left non-empty SPI stack"),
 532                                  errhint("Check for missing \"SPI_finish\" calls.")));
 533
 534         /*
 535          * If we are aborting a subtransaction and there is an open SPI context
 536          * surrounding the subxact, clean up to prevent memory leakage.
 537          */
 538         if (_SPI_current && !isCommit)
 539         {
 540                 slist_mutable_iter siter;
 541
 542                 /*
 543                  * Throw away executor state if current executor operation was started
 544                  * within current subxact (essentially, force a _SPI_end_call(true)).
 545                  */
 546                 if (_SPI_current->execSubid >= mySubid)
 547                 {
 548                         _SPI_current->execSubid = InvalidSubTransactionId;
 549                         MemoryContextReset(_SPI_current->execCxt);
 550                 }
 551
 552                 /* throw away any tuple tables created within current subxact */
 553                 slist_foreach_modify(siter, &_SPI_current->tuptables)
 554                 {
 555                         SPITupleTable *tuptable;
 556
 557                         tuptable = slist_container(SPITupleTable, next, siter.cur);
 558                         if (tuptable->subid >= mySubid)
 559                         {
 560                                 /*
 561                                  * If we used SPI_freetuptable() here, its internal search of
 562                                  * the tuptables list would make this operation O(N^2).
 563                                  * Instead, just free the tuptable manually.  This should
 564                                  * match what SPI_freetuptable() does.
 565                                  */
 566                                 slist_delete_current(&siter);
 567                                 if (tuptable == _SPI_current->tuptable)
 568                                         _SPI_current->tuptable = NULL;
 569                                 if (tuptable == SPI_tuptable)
 570                                         SPI_tuptable = NULL;
 571                                 MemoryContextDelete(tuptable->tuptabcxt);
 572                         }
 573                 }
 574         }
 575 }
 576
 577 /*
 578  * Are we executing inside a procedure (that is, a nonatomic SPI context)?
 579  */
 580 bool
 581 SPI_inside_nonatomic_context(void)
 582 {
 583         if (_SPI_current == NULL)
 584                 return false;                   /* not in any SPI context at all */
 585         /* these tests must match _SPI_commit's opinion of what's atomic: */
 586         if (_SPI_current->atomic)
 587                 return false;                   /* it's atomic (ie function not procedure) */
 588         if (IsSubTransaction())
 589                 return false;                   /* if within subtransaction, it's atomic */
 590         return true;
 591 }
 592
 593
 594 /* Parse, plan, and execute a query string */
 595 int
 596 SPI_execute(const char *src, bool read_only, long tcount)
 597 {
 598         _SPI_plan       plan;
 599         SPIExecuteOptions options;
 600         int                     res;
 601
 602         if (src == NULL || tcount < 0)
 603                 return SPI_ERROR_ARGUMENT;
 604
 605         res = _SPI_begin_call(true);
 606         if (res < 0)
 607                 return res;
 608
 609         memset(&plan, 0, sizeof(_SPI_plan));
 610         plan.magic = _SPI_PLAN_MAGIC;
 611         plan.parse_mode = RAW_PARSE_DEFAULT;
 612         plan.cursor_options = CURSOR_OPT_PARALLEL_OK;
 613
 614         _SPI_prepare_oneshot_plan(src, &plan);
 615
 616         memset(&options, 0, sizeof(options));
 617         options.read_only = read_only;
 618         options.tcount = tcount;
 619
 620         res = _SPI_execute_plan(&plan, &options,
 621                                                         InvalidSnapshot, InvalidSnapshot,
 622                                                         true);
 623
 624         _SPI_end_call(true);
 625         return res;
 626 }
 627
 628 /* Obsolete version of SPI_execute */
 629 int
 630 SPI_exec(const char *src, long tcount)
 631 {
 632         return SPI_execute(src, false, tcount);
 633 }
 634
 635 /* Parse, plan, and execute a query string, with extensible options */
 636 int
 637 SPI_execute_extended(const char *src,
 638                                          const SPIExecuteOptions *options)
 639 {
 640         int                     res;
 641         _SPI_plan       plan;
 642
 643         if (src == NULL || options == NULL)
 644                 return SPI_ERROR_ARGUMENT;
 645
 646         res = _SPI_begin_call(true);
 647         if (res < 0)
 648                 return res;
 649
 650         memset(&plan, 0, sizeof(_SPI_plan));
 651         plan.magic = _SPI_PLAN_MAGIC;
 652         plan.parse_mode = RAW_PARSE_DEFAULT;
 653         plan.cursor_options = CURSOR_OPT_PARALLEL_OK;
 654         if (options->params)
 655         {
 656                 plan.parserSetup = options->params->parserSetup;
 657                 plan.parserSetupArg = options->params->parserSetupArg;
 658         }
 659
 660         _SPI_prepare_oneshot_plan(src, &plan);
 661
 662         res = _SPI_execute_plan(&plan, options,
 663                                                         InvalidSnapshot, InvalidSnapshot,
 664                                                         true);
 665
 666         _SPI_end_call(true);
 667         return res;
 668 }
 669
 670 /* Execute a previously prepared plan */
 671 int
 672 SPI_execute_plan(SPIPlanPtr plan, Datum *Values, const char *Nulls,
 673                                  bool read_only, long tcount)
 674 {
 675         SPIExecuteOptions options;
 676         int                     res;
 677
 678         if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC || tcount < 0)
 679                 return SPI_ERROR_ARGUMENT;
 680
 681         if (plan->nargs > 0 && Values == NULL)
 682                 return SPI_ERROR_PARAM;
 683
 684         res = _SPI_begin_call(true);
 685         if (res < 0)
 686                 return res;
 687
 688         memset(&options, 0, sizeof(options));
 689         options.params = _SPI_convert_params(plan->nargs, plan->argtypes,
 690                                                                                  Values, Nulls);
 691         options.read_only = read_only;
 692         options.tcount = tcount;
 693
 694         res = _SPI_execute_plan(plan, &options,
 695                                                         InvalidSnapshot, InvalidSnapshot,
 696                                                         true);
 697
 698         _SPI_end_call(true);
 699         return res;
 700 }
 701
 702 /* Obsolete version of SPI_execute_plan */
 703 int
 704 SPI_execp(SPIPlanPtr plan, Datum *Values, const char *Nulls, long tcount)
 705 {
 706         return SPI_execute_plan(plan, Values, Nulls, false, tcount);
 707 }
 708
 709 /* Execute a previously prepared plan */
 710 int
 711 SPI_execute_plan_extended(SPIPlanPtr plan,
 712                                                   const SPIExecuteOptions *options)
 713 {
 714         int                     res;
 715
 716         if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC || options == NULL)
 717                 return SPI_ERROR_ARGUMENT;
 718
 719         res = _SPI_begin_call(true);
 720         if (res < 0)
 721                 return res;
 722
 723         res = _SPI_execute_plan(plan, options,
 724                                                         InvalidSnapshot, InvalidSnapshot,
 725                                                         true);
 726
 727         _SPI_end_call(true);
 728         return res;
 729 }
 730
 731 /* Execute a previously prepared plan */
 732 int
 733 SPI_execute_plan_with_paramlist(SPIPlanPtr plan, ParamListInfo params,
 734                                                                 bool read_only, long tcount)
 735 {
 736         SPIExecuteOptions options;
 737         int                     res;
 738
 739         if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC || tcount < 0)
 740                 return SPI_ERROR_ARGUMENT;
 741
 742         res = _SPI_begin_call(true);
 743         if (res < 0)
 744                 return res;
 745
 746         memset(&options, 0, sizeof(options));
 747         options.params = params;
 748         options.read_only = read_only;
 749         options.tcount = tcount;
 750
 751         res = _SPI_execute_plan(plan, &options,
 752                                                         InvalidSnapshot, InvalidSnapshot,
 753                                                         true);
 754
 755         _SPI_end_call(true);
 756         return res;
 757 }
 758
 759 /*
 760  * SPI_execute_snapshot -- identical to SPI_execute_plan, except that we allow
 761  * the caller to specify exactly which snapshots to use, which will be
 762  * registered here.  Also, the caller may specify that AFTER triggers should be
 763  * queued as part of the outer query rather than being fired immediately at the
 764  * end of the command.
 765  *
 766  * This is currently not documented in spi.sgml because it is only intended
 767  * for use by RI triggers.
 768  *
 769  * Passing snapshot == InvalidSnapshot will select the normal behavior of
 770  * fetching a new snapshot for each query.
 771  */
 772 int
 773 SPI_execute_snapshot(SPIPlanPtr plan,
 774                                          Datum *Values, const char *Nulls,
 775                                          Snapshot snapshot, Snapshot crosscheck_snapshot,
 776                                          bool read_only, bool fire_triggers, long tcount)
 777 {
 778         SPIExecuteOptions options;
 779         int                     res;
 780
 781         if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC || tcount < 0)
 782                 return SPI_ERROR_ARGUMENT;
 783
 784         if (plan->nargs > 0 && Values == NULL)
 785                 return SPI_ERROR_PARAM;
 786
 787         res = _SPI_begin_call(true);
 788         if (res < 0)
 789                 return res;
 790
 791         memset(&options, 0, sizeof(options));
 792         options.params = _SPI_convert_params(plan->nargs, plan->argtypes,
 793                                                                                  Values, Nulls);
 794         options.read_only = read_only;
 795         options.tcount = tcount;
 796
 797         res = _SPI_execute_plan(plan, &options,
 798                                                         snapshot, crosscheck_snapshot,
 799                                                         fire_triggers);
 800
 801         _SPI_end_call(true);
 802         return res;
 803 }
 804
 805 /*
 806  * SPI_execute_with_args -- plan and execute a query with supplied arguments
 807  *
 808  * This is functionally equivalent to SPI_prepare followed by
 809  * SPI_execute_plan.
 810  */
 811 int
 812 SPI_execute_with_args(const char *src,
 813                                           int nargs, Oid *argtypes,
 814                                           Datum *Values, const char *Nulls,
 815                                           bool read_only, long tcount)
 816 {
 817         int                     res;
 818         _SPI_plan       plan;
 819         ParamListInfo paramLI;
 820         SPIExecuteOptions options;
 821
 822         if (src == NULL || nargs < 0 || tcount < 0)
 823                 return SPI_ERROR_ARGUMENT;
 824
 825         if (nargs > 0 && (argtypes == NULL || Values == NULL))
 826                 return SPI_ERROR_PARAM;
 827
 828         res = _SPI_begin_call(true);
 829         if (res < 0)
 830                 return res;
 831
 832         memset(&plan, 0, sizeof(_SPI_plan));
 833         plan.magic = _SPI_PLAN_MAGIC;
 834         plan.parse_mode = RAW_PARSE_DEFAULT;
 835         plan.cursor_options = CURSOR_OPT_PARALLEL_OK;
 836         plan.nargs = nargs;
 837         plan.argtypes = argtypes;
 838         plan.parserSetup = NULL;
 839         plan.parserSetupArg = NULL;
 840
 841         paramLI = _SPI_convert_params(nargs, argtypes,
 842                                                                   Values, Nulls);
 843
 844         _SPI_prepare_oneshot_plan(src, &plan);
 845
 846         memset(&options, 0, sizeof(options));
 847         options.params = paramLI;
 848         options.read_only = read_only;
 849         options.tcount = tcount;
 850
 851         res = _SPI_execute_plan(&plan, &options,
 852                                                         InvalidSnapshot, InvalidSnapshot,
 853                                                         true);
 854
 855         _SPI_end_call(true);
 856         return res;
 857 }
 858
 859 SPIPlanPtr
 860 SPI_prepare(const char *src, int nargs, Oid *argtypes)
 861 {
 862         return SPI_prepare_cursor(src, nargs, argtypes, 0);
 863 }
 864
 865 SPIPlanPtr
 866 SPI_prepare_cursor(const char *src, int nargs, Oid *argtypes,
 867                                    int cursorOptions)
 868 {
 869         _SPI_plan       plan;
 870         SPIPlanPtr      result;
 871
 872         if (src == NULL || nargs < 0 || (nargs > 0 && argtypes == NULL))
 873         {
 874                 SPI_result = SPI_ERROR_ARGUMENT;
 875                 return NULL;
 876         }
 877
 878         SPI_result = _SPI_begin_call(true);
 879         if (SPI_result < 0)
 880                 return NULL;
 881
 882         memset(&plan, 0, sizeof(_SPI_plan));
 883         plan.magic = _SPI_PLAN_MAGIC;
 884         plan.parse_mode = RAW_PARSE_DEFAULT;
 885         plan.cursor_options = cursorOptions;
 886         plan.nargs = nargs;
 887         plan.argtypes = argtypes;
 888         plan.parserSetup = NULL;
 889         plan.parserSetupArg = NULL;
 890
 891         _SPI_prepare_plan(src, &plan);
 892
 893         /* copy plan to procedure context */
 894         result = _SPI_make_plan_non_temp(&plan);
 895
 896         _SPI_end_call(true);
 897
 898         return result;
 899 }
 900
 901 SPIPlanPtr
 902 SPI_prepare_extended(const char *src,
 903                                          const SPIPrepareOptions *options)
 904 {
 905         _SPI_plan       plan;
 906         SPIPlanPtr      result;
 907
 908         if (src == NULL || options == NULL)
 909         {
 910                 SPI_result = SPI_ERROR_ARGUMENT;
 911                 return NULL;
 912         }
 913
 914         SPI_result = _SPI_begin_call(true);
 915         if (SPI_result < 0)
 916                 return NULL;
 917
 918         memset(&plan, 0, sizeof(_SPI_plan));
 919         plan.magic = _SPI_PLAN_MAGIC;
 920         plan.parse_mode = options->parseMode;
 921         plan.cursor_options = options->cursorOptions;
 922         plan.nargs = 0;
 923         plan.argtypes = NULL;
 924         plan.parserSetup = options->parserSetup;
 925         plan.parserSetupArg = options->parserSetupArg;
 926
 927         _SPI_prepare_plan(src, &plan);
 928
 929         /* copy plan to procedure context */
 930         result = _SPI_make_plan_non_temp(&plan);
 931
 932         _SPI_end_call(true);
 933
 934         return result;
 935 }
 936
 937 SPIPlanPtr
 938 SPI_prepare_params(const char *src,
 939                                    ParserSetupHook parserSetup,
 940                                    void *parserSetupArg,
 941                                    int cursorOptions)
 942 {
 943         _SPI_plan       plan;
 944         SPIPlanPtr      result;
 945
 946         if (src == NULL)
 947         {
 948                 SPI_result = SPI_ERROR_ARGUMENT;
 949                 return NULL;
 950         }
 951
 952         SPI_result = _SPI_begin_call(true);
 953         if (SPI_result < 0)
 954                 return NULL;
 955
 956         memset(&plan, 0, sizeof(_SPI_plan));
 957         plan.magic = _SPI_PLAN_MAGIC;
 958         plan.parse_mode = RAW_PARSE_DEFAULT;
 959         plan.cursor_options = cursorOptions;
 960         plan.nargs = 0;
 961         plan.argtypes = NULL;
 962         plan.parserSetup = parserSetup;
 963         plan.parserSetupArg = parserSetupArg;
 964
 965         _SPI_prepare_plan(src, &plan);
 966
 967         /* copy plan to procedure context */
 968         result = _SPI_make_plan_non_temp(&plan);
 969
 970         _SPI_end_call(true);
 971
 972         return result;
 973 }
 974
 975 int
 976 SPI_keepplan(SPIPlanPtr plan)
 977 {
 978         ListCell   *lc;
 979
 980         if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC ||
 981                 plan->saved || plan->oneshot)
 982                 return SPI_ERROR_ARGUMENT;
 983
 984         /*
 985          * Mark it saved, reparent it under CacheMemoryContext, and mark all the
 986          * component CachedPlanSources as saved.  This sequence cannot fail
 987          * partway through, so there's no risk of long-term memory leakage.
 988          */
 989         plan->saved = true;
 990         MemoryContextSetParent(plan->plancxt, CacheMemoryContext);
 991
 992         foreach(lc, plan->plancache_list)
 993         {
 994                 CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc);
 995
 996                 SaveCachedPlan(plansource);
 997         }
 998
 999         return 0;
1000 }
1001
1002 SPIPlanPtr
1003 SPI_saveplan(SPIPlanPtr plan)
1004 {
1005         SPIPlanPtr      newplan;
1006
1007         if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC)
1008         {
1009                 SPI_result = SPI_ERROR_ARGUMENT;
1010                 return NULL;
1011         }
1012
1013         SPI_result = _SPI_begin_call(false);    /* don't change context */
1014         if (SPI_result < 0)
1015                 return NULL;
1016
1017         newplan = _SPI_save_plan(plan);
1018
1019         SPI_result = _SPI_end_call(false);
1020
1021         return newplan;
1022 }
1023
1024 int
1025 SPI_freeplan(SPIPlanPtr plan)
1026 {
1027         ListCell   *lc;
1028
1029         if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC)
1030                 return SPI_ERROR_ARGUMENT;
1031
1032         /* Release the plancache entries */
1033         foreach(lc, plan->plancache_list)
1034         {
1035                 CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc);
1036
1037                 DropCachedPlan(plansource);
1038         }
1039
1040         /* Now get rid of the _SPI_plan and subsidiary data in its plancxt */
1041         MemoryContextDelete(plan->plancxt);
1042
1043         return 0;
1044 }
1045
1046 HeapTuple
1047 SPI_copytuple(HeapTuple tuple)
1048 {
1049         MemoryContext oldcxt;
1050         HeapTuple       ctuple;
1051
1052         if (tuple == NULL)
1053         {
1054                 SPI_result = SPI_ERROR_ARGUMENT;
1055                 return NULL;
1056         }
1057
1058         if (_SPI_current == NULL)
1059         {
1060                 SPI_result = SPI_ERROR_UNCONNECTED;
1061                 return NULL;
1062         }
1063
1064         oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt);
1065
1066         ctuple = heap_copytuple(tuple);
1067
1068         MemoryContextSwitchTo(oldcxt);
1069
1070         return ctuple;
1071 }
1072
1073 HeapTupleHeader
1074 SPI_returntuple(HeapTuple tuple, TupleDesc tupdesc)
1075 {
1076         MemoryContext oldcxt;
1077         HeapTupleHeader dtup;
1078
1079         if (tuple == NULL || tupdesc == NULL)
1080         {
1081                 SPI_result = SPI_ERROR_ARGUMENT;
1082                 return NULL;
1083         }
1084
1085         if (_SPI_current == NULL)
1086         {
1087                 SPI_result = SPI_ERROR_UNCONNECTED;
1088                 return NULL;
1089         }
1090
1091         /* For RECORD results, make sure a typmod has been assigned */
1092         if (tupdesc->tdtypeid == RECORDOID &&
1093                 tupdesc->tdtypmod < 0)
1094                 assign_record_type_typmod(tupdesc);
1095
1096         oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt);
1097
1098         dtup = DatumGetHeapTupleHeader(heap_copy_tuple_as_datum(tuple, tupdesc));
1099
1100         MemoryContextSwitchTo(oldcxt);
1101
1102         return dtup;
1103 }
1104
1105 HeapTuple
1106 SPI_modifytuple(Relation rel, HeapTuple tuple, int natts, int *attnum,
1107                                 Datum *Values, const char *Nulls)
1108 {
1109         MemoryContext oldcxt;
1110         HeapTuple       mtuple;
1111         int                     numberOfAttributes;
1112         Datum      *v;
1113         bool       *n;
1114         int                     i;
1115
1116         if (rel == NULL || tuple == NULL || natts < 0 || attnum == NULL || Values == NULL)
1117         {
1118                 SPI_result = SPI_ERROR_ARGUMENT;
1119                 return NULL;
1120         }
1121
1122         if (_SPI_current == NULL)
1123         {
1124                 SPI_result = SPI_ERROR_UNCONNECTED;
1125                 return NULL;
1126         }
1127
1128         oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt);
1129
1130         SPI_result = 0;
1131
1132         numberOfAttributes = rel->rd_att->natts;
1133         v = (Datum *) palloc(numberOfAttributes * sizeof(Datum));
1134         n = (bool *) palloc(numberOfAttributes * sizeof(bool));
1135
1136         /* fetch old values and nulls */
1137         heap_deform_tuple(tuple, rel->rd_att, v, n);
1138
1139         /* replace values and nulls */
1140         for (i = 0; i < natts; i++)
1141         {
1142                 if (attnum[i] <= 0 || attnum[i] > numberOfAttributes)
1143                         break;
1144                 v[attnum[i] - 1] = Values[i];
1145                 n[attnum[i] - 1] = (Nulls && Nulls[i] == 'n');
1146         }
1147
1148         if (i == natts)                         /* no errors in *attnum */
1149         {
1150                 mtuple = heap_form_tuple(rel->rd_att, v, n);
1151
1152                 /*
1153                  * copy the identification info of the old tuple: t_ctid, t_self, and
1154                  * OID (if any)
1155                  */
1156                 mtuple->t_data->t_ctid = tuple->t_data->t_ctid;
1157                 mtuple->t_self = tuple->t_self;
1158                 mtuple->t_tableOid = tuple->t_tableOid;
1159         }
1160         else
1161         {
1162                 mtuple = NULL;
1163                 SPI_result = SPI_ERROR_NOATTRIBUTE;
1164         }
1165
1166         pfree(v);
1167         pfree(n);
1168
1169         MemoryContextSwitchTo(oldcxt);
1170
1171         return mtuple;
1172 }
1173
1174 int
1175 SPI_fnumber(TupleDesc tupdesc, const char *fname)
1176 {
1177         int                     res;
1178         const FormData_pg_attribute *sysatt;
1179
1180         for (res = 0; res < tupdesc->natts; res++)
1181         {
1182                 Form_pg_attribute attr = TupleDescAttr(tupdesc, res);
1183
1184                 if (namestrcmp(&attr->attname, fname) == 0 &&
1185                         !attr->attisdropped)
1186                         return res + 1;
1187         }
1188
1189         sysatt = SystemAttributeByName(fname);
1190         if (sysatt != NULL)
1191                 return sysatt->attnum;
1192
1193         /* SPI_ERROR_NOATTRIBUTE is different from all sys column numbers */
1194         return SPI_ERROR_NOATTRIBUTE;
1195 }
1196
1197 char *
1198 SPI_fname(TupleDesc tupdesc, int fnumber)
1199 {
1200         const FormData_pg_attribute *att;
1201
1202         SPI_result = 0;
1203
1204         if (fnumber > tupdesc->natts || fnumber == 0 ||
1205                 fnumber <= FirstLowInvalidHeapAttributeNumber)
1206         {
1207                 SPI_result = SPI_ERROR_NOATTRIBUTE;
1208                 return NULL;
1209         }
1210
1211         if (fnumber > 0)
1212                 att = TupleDescAttr(tupdesc, fnumber - 1);
1213         else
1214                 att = SystemAttributeDefinition(fnumber);
1215
1216         return pstrdup(NameStr(att->attname));
1217 }
1218
1219 char *
1220 SPI_getvalue(HeapTuple tuple, TupleDesc tupdesc, int fnumber)
1221 {
1222         Datum           val;
1223         bool            isnull;
1224         Oid                     typoid,
1225                                 foutoid;
1226         bool            typisvarlena;
1227
1228         SPI_result = 0;
1229
1230         if (fnumber > tupdesc->natts || fnumber == 0 ||
1231                 fnumber <= FirstLowInvalidHeapAttributeNumber)
1232         {
1233                 SPI_result = SPI_ERROR_NOATTRIBUTE;
1234                 return NULL;
1235         }
1236
1237         val = heap_getattr(tuple, fnumber, tupdesc, &isnull);
1238         if (isnull)
1239                 return NULL;
1240
1241         if (fnumber > 0)
1242                 typoid = TupleDescAttr(tupdesc, fnumber - 1)->atttypid;
1243         else
1244                 typoid = (SystemAttributeDefinition(fnumber))->atttypid;
1245
1246         getTypeOutputInfo(typoid, &foutoid, &typisvarlena);
1247
1248         return OidOutputFunctionCall(foutoid, val);
1249 }
1250
1251 Datum
1252 SPI_getbinval(HeapTuple tuple, TupleDesc tupdesc, int fnumber, bool *isnull)
1253 {
1254         SPI_result = 0;
1255
1256         if (fnumber > tupdesc->natts || fnumber == 0 ||
1257                 fnumber <= FirstLowInvalidHeapAttributeNumber)
1258         {
1259                 SPI_result = SPI_ERROR_NOATTRIBUTE;
1260                 *isnull = true;
1261                 return (Datum) NULL;
1262         }
1263
1264         return heap_getattr(tuple, fnumber, tupdesc, isnull);
1265 }
1266
1267 char *
1268 SPI_gettype(TupleDesc tupdesc, int fnumber)
1269 {
1270         Oid                     typoid;
1271         HeapTuple       typeTuple;
1272         char       *result;
1273
1274         SPI_result = 0;
1275
1276         if (fnumber > tupdesc->natts || fnumber == 0 ||
1277                 fnumber <= FirstLowInvalidHeapAttributeNumber)
1278         {
1279                 SPI_result = SPI_ERROR_NOATTRIBUTE;
1280                 return NULL;
1281         }
1282
1283         if (fnumber > 0)
1284                 typoid = TupleDescAttr(tupdesc, fnumber - 1)->atttypid;
1285         else
1286                 typoid = (SystemAttributeDefinition(fnumber))->atttypid;
1287
1288         typeTuple = SearchSysCache1(TYPEOID, ObjectIdGetDatum(typoid));
1289
1290         if (!HeapTupleIsValid(typeTuple))
1291         {
1292                 SPI_result = SPI_ERROR_TYPUNKNOWN;
1293                 return NULL;
1294         }
1295
1296         result = pstrdup(NameStr(((Form_pg_type) GETSTRUCT(typeTuple))->typname));
1297         ReleaseSysCache(typeTuple);
1298         return result;
1299 }
1300
1301 /*
1302  * Get the data type OID for a column.
1303  *
1304  * There's nothing similar for typmod and typcollation.  The rare consumers
1305  * thereof should inspect the TupleDesc directly.
1306  */
1307 Oid
1308 SPI_gettypeid(TupleDesc tupdesc, int fnumber)
1309 {
1310         SPI_result = 0;
1311
1312         if (fnumber > tupdesc->natts || fnumber == 0 ||
1313                 fnumber <= FirstLowInvalidHeapAttributeNumber)
1314         {
1315                 SPI_result = SPI_ERROR_NOATTRIBUTE;
1316                 return InvalidOid;
1317         }
1318
1319         if (fnumber > 0)
1320                 return TupleDescAttr(tupdesc, fnumber - 1)->atttypid;
1321         else
1322                 return (SystemAttributeDefinition(fnumber))->atttypid;
1323 }
1324
1325 char *
1326 SPI_getrelname(Relation rel)
1327 {
1328         return pstrdup(RelationGetRelationName(rel));
1329 }
1330
1331 char *
1332 SPI_getnspname(Relation rel)
1333 {
1334         return get_namespace_name(RelationGetNamespace(rel));
1335 }
1336
1337 void *
1338 SPI_palloc(Size size)
1339 {
1340         if (_SPI_current == NULL)
1341                 elog(ERROR, "SPI_palloc called while not connected to SPI");
1342
1343         return MemoryContextAlloc(_SPI_current->savedcxt, size);
1344 }
1345
1346 void *
1347 SPI_repalloc(void *pointer, Size size)
1348 {
1349         /* No longer need to worry which context chunk was in... */
1350         return repalloc(pointer, size);
1351 }
1352
1353 void
1354 SPI_pfree(void *pointer)
1355 {
1356         /* No longer need to worry which context chunk was in... */
1357         pfree(pointer);
1358 }
1359
1360 Datum
1361 SPI_datumTransfer(Datum value, bool typByVal, int typLen)
1362 {
1363         MemoryContext oldcxt;
1364         Datum           result;
1365
1366         if (_SPI_current == NULL)
1367                 elog(ERROR, "SPI_datumTransfer called while not connected to SPI");
1368
1369         oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt);
1370
1371         result = datumTransfer(value, typByVal, typLen);
1372
1373         MemoryContextSwitchTo(oldcxt);
1374
1375         return result;
1376 }
1377
1378 void
1379 SPI_freetuple(HeapTuple tuple)
1380 {
1381         /* No longer need to worry which context tuple was in... */
1382         heap_freetuple(tuple);
1383 }
1384
1385 void
1386 SPI_freetuptable(SPITupleTable *tuptable)
1387 {
1388         bool            found = false;
1389
1390         /* ignore call if NULL pointer */
1391         if (tuptable == NULL)
1392                 return;
1393
1394         /*
1395          * Search only the topmost SPI context for a matching tuple table.
1396          */
1397         if (_SPI_current != NULL)
1398         {
1399                 slist_mutable_iter siter;
1400
1401                 /* find tuptable in active list, then remove it */
1402                 slist_foreach_modify(siter, &_SPI_current->tuptables)
1403                 {
1404                         SPITupleTable *tt;
1405
1406                         tt = slist_container(SPITupleTable, next, siter.cur);
1407                         if (tt == tuptable)
1408                         {
1409                                 slist_delete_current(&siter);
1410                                 found = true;
1411                                 break;
1412                         }
1413                 }
1414         }
1415
1416         /*
1417          * Refuse the deletion if we didn't find it in the topmost SPI context.
1418          * This is primarily a guard against double deletion, but might prevent
1419          * other errors as well.  Since the worst consequence of not deleting a
1420          * tuptable would be a transient memory leak, this is just a WARNING.
1421          */
1422         if (!found)
1423         {
1424                 elog(WARNING, "attempt to delete invalid SPITupleTable %p", tuptable);
1425                 return;
1426         }
1427
1428         /* for safety, reset global variables that might point at tuptable */
1429         if (tuptable == _SPI_current->tuptable)
1430                 _SPI_current->tuptable = NULL;
1431         if (tuptable == SPI_tuptable)
1432                 SPI_tuptable = NULL;
1433
1434         /* release all memory belonging to tuptable */
1435         MemoryContextDelete(tuptable->tuptabcxt);
1436 }
1437
1438
1439 /*
1440  * SPI_cursor_open()
1441  *
1442  *      Open a prepared SPI plan as a portal
1443  */
1444 Portal
1445 SPI_cursor_open(const char *name, SPIPlanPtr plan,
1446                                 Datum *Values, const char *Nulls,
1447                                 bool read_only)
1448 {
1449         Portal          portal;
1450         ParamListInfo paramLI;
1451
1452         /* build transient ParamListInfo in caller's context */
1453         paramLI = _SPI_convert_params(plan->nargs, plan->argtypes,
1454                                                                   Values, Nulls);
1455
1456         portal = SPI_cursor_open_internal(name, plan, paramLI, read_only);
1457
1458         /* done with the transient ParamListInfo */
1459         if (paramLI)
1460                 pfree(paramLI);
1461
1462         return portal;
1463 }
1464
1465
1466 /*
1467  * SPI_cursor_open_with_args()
1468  *
1469  * Parse and plan a query and open it as a portal.
1470  */
1471 Portal
1472 SPI_cursor_open_with_args(const char *name,
1473                                                   const char *src,
1474                                                   int nargs, Oid *argtypes,
1475                                                   Datum *Values, const char *Nulls,
1476                                                   bool read_only, int cursorOptions)
1477 {
1478         Portal          result;
1479         _SPI_plan       plan;
1480         ParamListInfo paramLI;
1481
1482         if (src == NULL || nargs < 0)
1483                 elog(ERROR, "SPI_cursor_open_with_args called with invalid arguments");
1484
1485         if (nargs > 0 && (argtypes == NULL || Values == NULL))
1486                 elog(ERROR, "SPI_cursor_open_with_args called with missing parameters");
1487
1488         SPI_result = _SPI_begin_call(true);
1489         if (SPI_result < 0)
1490                 elog(ERROR, "SPI_cursor_open_with_args called while not connected");
1491
1492         memset(&plan, 0, sizeof(_SPI_plan));
1493         plan.magic = _SPI_PLAN_MAGIC;
1494         plan.parse_mode = RAW_PARSE_DEFAULT;
1495         plan.cursor_options = cursorOptions;
1496         plan.nargs = nargs;
1497         plan.argtypes = argtypes;
1498         plan.parserSetup = NULL;
1499         plan.parserSetupArg = NULL;
1500
1501         /* build transient ParamListInfo in executor context */
1502         paramLI = _SPI_convert_params(nargs, argtypes,
1503                                                                   Values, Nulls);
1504
1505         _SPI_prepare_plan(src, &plan);
1506
1507         /* We needn't copy the plan; SPI_cursor_open_internal will do so */
1508
1509         result = SPI_cursor_open_internal(name, &plan, paramLI, read_only);
1510
1511         /* And clean up */
1512         _SPI_end_call(true);
1513
1514         return result;
1515 }
1516
1517
1518 /*
1519  * SPI_cursor_open_with_paramlist()
1520  *
1521  *      Same as SPI_cursor_open except that parameters (if any) are passed
1522  *      as a ParamListInfo, which supports dynamic parameter set determination
1523  */
1524 Portal
1525 SPI_cursor_open_with_paramlist(const char *name, SPIPlanPtr plan,
1526                                                            ParamListInfo params, bool read_only)
1527 {
1528         return SPI_cursor_open_internal(name, plan, params, read_only);
1529 }
1530
1531 /* Parse a query and open it as a cursor */
1532 Portal
1533 SPI_cursor_parse_open(const char *name,
1534                                           const char *src,
1535                                           const SPIParseOpenOptions *options)
1536 {
1537         Portal          result;
1538         _SPI_plan       plan;
1539
1540         if (src == NULL || options == NULL)
1541                 elog(ERROR, "SPI_cursor_parse_open called with invalid arguments");
1542
1543         SPI_result = _SPI_begin_call(true);
1544         if (SPI_result < 0)
1545                 elog(ERROR, "SPI_cursor_parse_open called while not connected");
1546
1547         memset(&plan, 0, sizeof(_SPI_plan));
1548         plan.magic = _SPI_PLAN_MAGIC;
1549         plan.parse_mode = RAW_PARSE_DEFAULT;
1550         plan.cursor_options = options->cursorOptions;
1551         if (options->params)
1552         {
1553                 plan.parserSetup = options->params->parserSetup;
1554                 plan.parserSetupArg = options->params->parserSetupArg;
1555         }
1556
1557         _SPI_prepare_plan(src, &plan);
1558
1559         /* We needn't copy the plan; SPI_cursor_open_internal will do so */
1560
1561         result = SPI_cursor_open_internal(name, &plan,
1562                                                                           options->params, options->read_only);
1563
1564         /* And clean up */
1565         _SPI_end_call(true);
1566
1567         return result;
1568 }
1569
1570
1571 /*
1572  * SPI_cursor_open_internal()
1573  *
1574  *      Common code for SPI_cursor_open variants
1575  */
1576 static Portal
1577 SPI_cursor_open_internal(const char *name, SPIPlanPtr plan,
1578                                                  ParamListInfo paramLI, bool read_only)
1579 {
1580         CachedPlanSource *plansource;
1581         CachedPlan *cplan;
1582         List       *stmt_list;
1583         char       *query_string;
1584         Snapshot        snapshot;
1585         MemoryContext oldcontext;
1586         Portal          portal;
1587         SPICallbackArg spicallbackarg;
1588         ErrorContextCallback spierrcontext;
1589
1590         /*
1591          * Check that the plan is something the Portal code will special-case as
1592          * returning one tupleset.
1593          */
1594         if (!SPI_is_cursor_plan(plan))
1595         {
1596                 /* try to give a good error message */
1597                 const char *cmdtag;
1598
1599                 if (list_length(plan->plancache_list) != 1)
1600                         ereport(ERROR,
1601                                         (errcode(ERRCODE_INVALID_CURSOR_DEFINITION),
1602                                          errmsg("cannot open multi-query plan as cursor")));
1603                 plansource = (CachedPlanSource *) linitial(plan->plancache_list);
1604                 /* A SELECT that fails SPI_is_cursor_plan() must be SELECT INTO */
1605                 if (plansource->commandTag == CMDTAG_SELECT)
1606                         cmdtag = "SELECT INTO";
1607                 else
1608                         cmdtag = GetCommandTagName(plansource->commandTag);
1609                 ereport(ERROR,
1610                                 (errcode(ERRCODE_INVALID_CURSOR_DEFINITION),
1611                 /* translator: %s is name of a SQL command, eg INSERT */
1612                                  errmsg("cannot open %s query as cursor", cmdtag)));
1613         }
1614
1615         Assert(list_length(plan->plancache_list) == 1);
1616         plansource = (CachedPlanSource *) linitial(plan->plancache_list);
1617
1618         /* Push the SPI stack */
1619         if (_SPI_begin_call(true) < 0)
1620                 elog(ERROR, "SPI_cursor_open called while not connected");
1621
1622         /* Reset SPI result (note we deliberately don't touch lastoid) */
1623         SPI_processed = 0;
1624         SPI_tuptable = NULL;
1625         _SPI_current->processed = 0;
1626         _SPI_current->tuptable = NULL;
1627
1628         /* Create the portal */
1629         if (name == NULL || name[0] == '\0')
1630         {
1631                 /* Use a random nonconflicting name */
1632                 portal = CreateNewPortal();
1633         }
1634         else
1635         {
1636                 /* In this path, error if portal of same name already exists */
1637                 portal = CreatePortal(name, false, false);
1638         }
1639
1640         /* Copy the plan's query string into the portal */
1641         query_string = MemoryContextStrdup(portal->portalContext,
1642                                                                            plansource->query_string);
1643
1644         /*
1645          * Setup error traceback support for ereport(), in case GetCachedPlan
1646          * throws an error.
1647          */
1648         spicallbackarg.query = plansource->query_string;
1649         spicallbackarg.mode = plan->parse_mode;
1650         spierrcontext.callback = _SPI_error_callback;
1651         spierrcontext.arg = &spicallbackarg;
1652         spierrcontext.previous = error_context_stack;
1653         error_context_stack = &spierrcontext;
1654
1655         /*
1656          * Note: for a saved plan, we mustn't have any failure occur between
1657          * GetCachedPlan and PortalDefineQuery; that would result in leaking our
1658          * plancache refcount.
1659          */
1660
1661         /* Replan if needed, and increment plan refcount for portal */
1662         cplan = GetCachedPlan(plansource, paramLI, NULL, _SPI_current->queryEnv);
1663         stmt_list = cplan->stmt_list;
1664
1665         if (!plan->saved)
1666         {
1667                 /*
1668                  * We don't want the portal to depend on an unsaved CachedPlanSource,
1669                  * so must copy the plan into the portal's context.  An error here
1670                  * will result in leaking our refcount on the plan, but it doesn't
1671                  * matter because the plan is unsaved and hence transient anyway.
1672                  */
1673                 oldcontext = MemoryContextSwitchTo(portal->portalContext);
1674                 stmt_list = copyObject(stmt_list);
1675                 MemoryContextSwitchTo(oldcontext);
1676                 ReleaseCachedPlan(cplan, NULL);
1677                 cplan = NULL;                   /* portal shouldn't depend on cplan */
1678         }
1679
1680         /*
1681          * Set up the portal.
1682          */
1683         PortalDefineQuery(portal,
1684                                           NULL,         /* no statement name */
1685                                           query_string,
1686                                           plansource->commandTag,
1687                                           stmt_list,
1688                                           cplan);
1689
1690         /*
1691          * Set up options for portal.  Default SCROLL type is chosen the same way
1692          * as PerformCursorOpen does it.
1693          */
1694         portal->cursorOptions = plan->cursor_options;
1695         if (!(portal->cursorOptions & (CURSOR_OPT_SCROLL | CURSOR_OPT_NO_SCROLL)))
1696         {
1697                 if (list_length(stmt_list) == 1 &&
1698                         linitial_node(PlannedStmt, stmt_list)->commandType != CMD_UTILITY &&
1699                         linitial_node(PlannedStmt, stmt_list)->rowMarks == NIL &&
1700                         ExecSupportsBackwardScan(linitial_node(PlannedStmt, stmt_list)->planTree))
1701                         portal->cursorOptions |= CURSOR_OPT_SCROLL;
1702                 else
1703                         portal->cursorOptions |= CURSOR_OPT_NO_SCROLL;
1704         }
1705
1706         /*
1707          * Disallow SCROLL with SELECT FOR UPDATE.  This is not redundant with the
1708          * check in transformDeclareCursorStmt because the cursor options might
1709          * not have come through there.
1710          */
1711         if (portal->cursorOptions & CURSOR_OPT_SCROLL)
1712         {
1713                 if (list_length(stmt_list) == 1 &&
1714                         linitial_node(PlannedStmt, stmt_list)->commandType != CMD_UTILITY &&
1715                         linitial_node(PlannedStmt, stmt_list)->rowMarks != NIL)
1716                         ereport(ERROR,
1717                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1718                                          errmsg("DECLARE SCROLL CURSOR ... FOR UPDATE/SHARE is not supported"),
1719                                          errdetail("Scrollable cursors must be READ ONLY.")));
1720         }
1721
1722         /* Make current query environment available to portal at execution time. */
1723         portal->queryEnv = _SPI_current->queryEnv;
1724
1725         /*
1726          * If told to be read-only, we'd better check for read-only queries. This
1727          * can't be done earlier because we need to look at the finished, planned
1728          * queries.  (In particular, we don't want to do it between GetCachedPlan
1729          * and PortalDefineQuery, because throwing an error between those steps
1730          * would result in leaking our plancache refcount.)
1731          */
1732         if (read_only)
1733         {
1734                 ListCell   *lc;
1735
1736                 foreach(lc, stmt_list)
1737                 {
1738                         PlannedStmt *pstmt = lfirst_node(PlannedStmt, lc);
1739
1740                         if (!CommandIsReadOnly(pstmt))
1741                                 ereport(ERROR,
1742                                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1743                                 /* translator: %s is a SQL statement name */
1744                                                  errmsg("%s is not allowed in a non-volatile function",
1745                                                                 CreateCommandName((Node *) pstmt))));
1746                 }
1747         }
1748
1749         /* Set up the snapshot to use. */
1750         if (read_only)
1751                 snapshot = GetActiveSnapshot();
1752         else
1753         {
1754                 CommandCounterIncrement();
1755                 snapshot = GetTransactionSnapshot();
1756         }
1757
1758         /*
1759          * If the plan has parameters, copy them into the portal.  Note that this
1760          * must be done after revalidating the plan, because in dynamic parameter
1761          * cases the set of parameters could have changed during re-parsing.
1762          */
1763         if (paramLI)
1764         {
1765                 oldcontext = MemoryContextSwitchTo(portal->portalContext);
1766                 paramLI = copyParamList(paramLI);
1767                 MemoryContextSwitchTo(oldcontext);
1768         }
1769
1770         /*
1771          * Start portal execution.
1772          */
1773         PortalStart(portal, paramLI, 0, snapshot);
1774
1775         Assert(portal->strategy != PORTAL_MULTI_QUERY);
1776
1777         /* Pop the error context stack */
1778         error_context_stack = spierrcontext.previous;
1779
1780         /* Pop the SPI stack */
1781         _SPI_end_call(true);
1782
1783         /* Return the created portal */
1784         return portal;
1785 }
1786
1787
1788 /*
1789  * SPI_cursor_find()
1790  *
1791  *      Find the portal of an existing open cursor
1792  */
1793 Portal
1794 SPI_cursor_find(const char *name)
1795 {
1796         return GetPortalByName(name);
1797 }
1798
1799
1800 /*
1801  * SPI_cursor_fetch()
1802  *
1803  *      Fetch rows in a cursor
1804  */
1805 void
1806 SPI_cursor_fetch(Portal portal, bool forward, long count)
1807 {
1808         _SPI_cursor_operation(portal,
1809                                                   forward ? FETCH_FORWARD : FETCH_BACKWARD, count,
1810                                                   CreateDestReceiver(DestSPI));
1811         /* we know that the DestSPI receiver doesn't need a destroy call */
1812 }
1813
1814
1815 /*
1816  * SPI_cursor_move()
1817  *
1818  *      Move in a cursor
1819  */
1820 void
1821 SPI_cursor_move(Portal portal, bool forward, long count)
1822 {
1823         _SPI_cursor_operation(portal,
1824                                                   forward ? FETCH_FORWARD : FETCH_BACKWARD, count,
1825                                                   None_Receiver);
1826 }
1827
1828
1829 /*
1830  * SPI_scroll_cursor_fetch()
1831  *
1832  *      Fetch rows in a scrollable cursor
1833  */
1834 void
1835 SPI_scroll_cursor_fetch(Portal portal, FetchDirection direction, long count)
1836 {
1837         _SPI_cursor_operation(portal,
1838                                                   direction, count,
1839                                                   CreateDestReceiver(DestSPI));
1840         /* we know that the DestSPI receiver doesn't need a destroy call */
1841 }
1842
1843
1844 /*
1845  * SPI_scroll_cursor_move()
1846  *
1847  *      Move in a scrollable cursor
1848  */
1849 void
1850 SPI_scroll_cursor_move(Portal portal, FetchDirection direction, long count)
1851 {
1852         _SPI_cursor_operation(portal, direction, count, None_Receiver);
1853 }
1854
1855
1856 /*
1857  * SPI_cursor_close()
1858  *
1859  *      Close a cursor
1860  */
1861 void
1862 SPI_cursor_close(Portal portal)
1863 {
1864         if (!PortalIsValid(portal))
1865                 elog(ERROR, "invalid portal in SPI cursor operation");
1866
1867         PortalDrop(portal, false);
1868 }
1869
1870 /*
1871  * Returns the Oid representing the type id for argument at argIndex. First
1872  * parameter is at index zero.
1873  */
1874 Oid
1875 SPI_getargtypeid(SPIPlanPtr plan, int argIndex)
1876 {
1877         if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC ||
1878                 argIndex < 0 || argIndex >= plan->nargs)
1879         {
1880                 SPI_result = SPI_ERROR_ARGUMENT;
1881                 return InvalidOid;
1882         }
1883         return plan->argtypes[argIndex];
1884 }
1885
1886 /*
1887  * Returns the number of arguments for the prepared plan.
1888  */
1889 int
1890 SPI_getargcount(SPIPlanPtr plan)
1891 {
1892         if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC)
1893         {
1894                 SPI_result = SPI_ERROR_ARGUMENT;
1895                 return -1;
1896         }
1897         return plan->nargs;
1898 }
1899
1900 /*
1901  * Returns true if the plan contains exactly one command
1902  * and that command returns tuples to the caller (eg, SELECT or
1903  * INSERT ... RETURNING, but not SELECT ... INTO). In essence,
1904  * the result indicates if the command can be used with SPI_cursor_open
1905  *
1906  * Parameters
1907  *        plan: A plan previously prepared using SPI_prepare
1908  */
1909 bool
1910 SPI_is_cursor_plan(SPIPlanPtr plan)
1911 {
1912         CachedPlanSource *plansource;
1913
1914         if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC)
1915         {
1916                 SPI_result = SPI_ERROR_ARGUMENT;
1917                 return false;
1918         }
1919
1920         if (list_length(plan->plancache_list) != 1)
1921         {
1922                 SPI_result = 0;
1923                 return false;                   /* not exactly 1 pre-rewrite command */
1924         }
1925         plansource = (CachedPlanSource *) linitial(plan->plancache_list);
1926
1927         /*
1928          * We used to force revalidation of the cached plan here, but that seems
1929          * unnecessary: invalidation could mean a change in the rowtype of the
1930          * tuples returned by a plan, but not whether it returns tuples at all.
1931          */
1932         SPI_result = 0;
1933
1934         /* Does it return tuples? */
1935         if (plansource->resultDesc)
1936                 return true;
1937
1938         return false;
1939 }
1940
1941 /*
1942  * SPI_plan_is_valid --- test whether a SPI plan is currently valid
1943  * (that is, not marked as being in need of revalidation).
1944  *
1945  * See notes for CachedPlanIsValid before using this.
1946  */
1947 bool
1948 SPI_plan_is_valid(SPIPlanPtr plan)
1949 {
1950         ListCell   *lc;
1951
1952         Assert(plan->magic == _SPI_PLAN_MAGIC);
1953
1954         foreach(lc, plan->plancache_list)
1955         {
1956                 CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc);
1957
1958                 if (!CachedPlanIsValid(plansource))
1959                         return false;
1960         }
1961         return true;
1962 }
1963
1964 /*
1965  * SPI_result_code_string --- convert any SPI return code to a string
1966  *
1967  * This is often useful in error messages.  Most callers will probably
1968  * only pass negative (error-case) codes, but for generality we recognize
1969  * the success codes too.
1970  */
1971 const char *
1972 SPI_result_code_string(int code)
1973 {
1974         static char buf[64];
1975
1976         switch (code)
1977         {
1978                 case SPI_ERROR_CONNECT:
1979                         return "SPI_ERROR_CONNECT";
1980                 case SPI_ERROR_COPY:
1981                         return "SPI_ERROR_COPY";
1982                 case SPI_ERROR_OPUNKNOWN:
1983                         return "SPI_ERROR_OPUNKNOWN";
1984                 case SPI_ERROR_UNCONNECTED:
1985                         return "SPI_ERROR_UNCONNECTED";
1986                 case SPI_ERROR_ARGUMENT:
1987                         return "SPI_ERROR_ARGUMENT";
1988                 case SPI_ERROR_PARAM:
1989                         return "SPI_ERROR_PARAM";
1990                 case SPI_ERROR_TRANSACTION:
1991                         return "SPI_ERROR_TRANSACTION";
1992                 case SPI_ERROR_NOATTRIBUTE:
1993                         return "SPI_ERROR_NOATTRIBUTE";
1994                 case SPI_ERROR_NOOUTFUNC:
1995                         return "SPI_ERROR_NOOUTFUNC";
1996                 case SPI_ERROR_TYPUNKNOWN:
1997                         return "SPI_ERROR_TYPUNKNOWN";
1998                 case SPI_ERROR_REL_DUPLICATE:
1999                         return "SPI_ERROR_REL_DUPLICATE";
2000                 case SPI_ERROR_REL_NOT_FOUND:
2001                         return "SPI_ERROR_REL_NOT_FOUND";
2002                 case SPI_OK_CONNECT:
2003                         return "SPI_OK_CONNECT";
2004                 case SPI_OK_FINISH:
2005                         return "SPI_OK_FINISH";
2006                 case SPI_OK_FETCH:
2007                         return "SPI_OK_FETCH";
2008                 case SPI_OK_UTILITY:
2009                         return "SPI_OK_UTILITY";
2010                 case SPI_OK_SELECT:
2011                         return "SPI_OK_SELECT";
2012                 case SPI_OK_SELINTO:
2013                         return "SPI_OK_SELINTO";
2014                 case SPI_OK_INSERT:
2015                         return "SPI_OK_INSERT";
2016                 case SPI_OK_DELETE:
2017                         return "SPI_OK_DELETE";
2018                 case SPI_OK_UPDATE:
2019                         return "SPI_OK_UPDATE";
2020                 case SPI_OK_CURSOR:
2021                         return "SPI_OK_CURSOR";
2022                 case SPI_OK_INSERT_RETURNING:
2023                         return "SPI_OK_INSERT_RETURNING";
2024                 case SPI_OK_DELETE_RETURNING:
2025                         return "SPI_OK_DELETE_RETURNING";
2026                 case SPI_OK_UPDATE_RETURNING:
2027                         return "SPI_OK_UPDATE_RETURNING";
2028                 case SPI_OK_REWRITTEN:
2029                         return "SPI_OK_REWRITTEN";
2030                 case SPI_OK_REL_REGISTER:
2031                         return "SPI_OK_REL_REGISTER";
2032                 case SPI_OK_REL_UNREGISTER:
2033                         return "SPI_OK_REL_UNREGISTER";
2034                 case SPI_OK_TD_REGISTER:
2035                         return "SPI_OK_TD_REGISTER";
2036                 case SPI_OK_MERGE:
2037                         return "SPI_OK_MERGE";
2038                 case SPI_OK_MERGE_RETURNING:
2039                         return "SPI_OK_MERGE_RETURNING";
2040         }
2041         /* Unrecognized code ... return something useful ... */
2042         sprintf(buf, "Unrecognized SPI code %d", code);
2043         return buf;
2044 }
2045
2046 /*
2047  * SPI_plan_get_plan_sources --- get a SPI plan's underlying list of
2048  * CachedPlanSources.
2049  *
2050  * CAUTION: there is no check on whether the CachedPlanSources are up-to-date.
2051  *
2052  * This is exported so that PL/pgSQL can use it (this beats letting PL/pgSQL
2053  * look directly into the SPIPlan for itself).  It's not documented in
2054  * spi.sgml because we'd just as soon not have too many places using this.
2055  */
2056 List *
2057 SPI_plan_get_plan_sources(SPIPlanPtr plan)
2058 {
2059         Assert(plan->magic == _SPI_PLAN_MAGIC);
2060         return plan->plancache_list;
2061 }
2062
2063 /*
2064  * SPI_plan_get_cached_plan --- get a SPI plan's generic CachedPlan,
2065  * if the SPI plan contains exactly one CachedPlanSource.  If not,
2066  * return NULL.
2067  *
2068  * The plan's refcount is incremented (and logged in CurrentResourceOwner,
2069  * if it's a saved plan).  Caller is responsible for doing ReleaseCachedPlan.
2070  *
2071  * This is exported so that PL/pgSQL can use it (this beats letting PL/pgSQL
2072  * look directly into the SPIPlan for itself).  It's not documented in
2073  * spi.sgml because we'd just as soon not have too many places using this.
2074  */
2075 CachedPlan *
2076 SPI_plan_get_cached_plan(SPIPlanPtr plan)
2077 {
2078         CachedPlanSource *plansource;
2079         CachedPlan *cplan;
2080         SPICallbackArg spicallbackarg;
2081         ErrorContextCallback spierrcontext;
2082
2083         Assert(plan->magic == _SPI_PLAN_MAGIC);
2084
2085         /* Can't support one-shot plans here */
2086         if (plan->oneshot)
2087                 return NULL;
2088
2089         /* Must have exactly one CachedPlanSource */
2090         if (list_length(plan->plancache_list) != 1)
2091                 return NULL;
2092         plansource = (CachedPlanSource *) linitial(plan->plancache_list);
2093
2094         /* Setup error traceback support for ereport() */
2095         spicallbackarg.query = plansource->query_string;
2096         spicallbackarg.mode = plan->parse_mode;
2097         spierrcontext.callback = _SPI_error_callback;
2098         spierrcontext.arg = &spicallbackarg;
2099         spierrcontext.previous = error_context_stack;
2100         error_context_stack = &spierrcontext;
2101
2102         /* Get the generic plan for the query */
2103         cplan = GetCachedPlan(plansource, NULL,
2104                                                   plan->saved ? CurrentResourceOwner : NULL,
2105                                                   _SPI_current->queryEnv);
2106         Assert(cplan == plansource->gplan);
2107
2108         /* Pop the error context stack */
2109         error_context_stack = spierrcontext.previous;
2110
2111         return cplan;
2112 }
2113
2114
2115 /* =================== private functions =================== */
2116
2117 /*
2118  * spi_dest_startup
2119  *              Initialize to receive tuples from Executor into SPITupleTable
2120  *              of current SPI procedure
2121  */
2122 void
2123 spi_dest_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
2124 {
2125         SPITupleTable *tuptable;
2126         MemoryContext oldcxt;
2127         MemoryContext tuptabcxt;
2128
2129         if (_SPI_current == NULL)
2130                 elog(ERROR, "spi_dest_startup called while not connected to SPI");
2131
2132         if (_SPI_current->tuptable != NULL)
2133                 elog(ERROR, "improper call to spi_dest_startup");
2134
2135         /* We create the tuple table context as a child of procCxt */
2136
2137         oldcxt = _SPI_procmem();        /* switch to procedure memory context */
2138
2139         tuptabcxt = AllocSetContextCreate(CurrentMemoryContext,
2140                                                                           "SPI TupTable",
2141                                                                           ALLOCSET_DEFAULT_SIZES);
2142         MemoryContextSwitchTo(tuptabcxt);
2143
2144         _SPI_current->tuptable = tuptable = (SPITupleTable *)
2145                 palloc0(sizeof(SPITupleTable));
2146         tuptable->tuptabcxt = tuptabcxt;
2147         tuptable->subid = GetCurrentSubTransactionId();
2148
2149         /*
2150          * The tuptable is now valid enough to be freed by AtEOSubXact_SPI, so put
2151          * it onto the SPI context's tuptables list.  This will ensure it's not
2152          * leaked even in the unlikely event the following few lines fail.
2153          */
2154         slist_push_head(&_SPI_current->tuptables, &tuptable->next);
2155
2156         /* set up initial allocations */
2157         tuptable->alloced = 128;
2158         tuptable->vals = (HeapTuple *) palloc(tuptable->alloced * sizeof(HeapTuple));
2159         tuptable->numvals = 0;
2160         tuptable->tupdesc = CreateTupleDescCopy(typeinfo);
2161
2162         MemoryContextSwitchTo(oldcxt);
2163 }
2164
2165 /*
2166  * spi_printtup
2167  *              store tuple retrieved by Executor into SPITupleTable
2168  *              of current SPI procedure
2169  */
2170 bool
2171 spi_printtup(TupleTableSlot *slot, DestReceiver *self)
2172 {
2173         SPITupleTable *tuptable;
2174         MemoryContext oldcxt;
2175
2176         if (_SPI_current == NULL)
2177                 elog(ERROR, "spi_printtup called while not connected to SPI");
2178
2179         tuptable = _SPI_current->tuptable;
2180         if (tuptable == NULL)
2181                 elog(ERROR, "improper call to spi_printtup");
2182
2183         oldcxt = MemoryContextSwitchTo(tuptable->tuptabcxt);
2184
2185         if (tuptable->numvals >= tuptable->alloced)
2186         {
2187                 /* Double the size of the pointer array */
2188                 uint64          newalloced = tuptable->alloced * 2;
2189
2190                 tuptable->vals = (HeapTuple *) repalloc_huge(tuptable->vals,
2191                                                                                                          newalloced * sizeof(HeapTuple));
2192                 tuptable->alloced = newalloced;
2193         }
2194
2195         tuptable->vals[tuptable->numvals] = ExecCopySlotHeapTuple(slot);
2196         (tuptable->numvals)++;
2197
2198         MemoryContextSwitchTo(oldcxt);
2199
2200         return true;
2201 }
2202
2203 /*
2204  * Static functions
2205  */
2206
2207 /*
2208  * Parse and analyze a querystring.
2209  *
2210  * At entry, plan->argtypes and plan->nargs (or alternatively plan->parserSetup
2211  * and plan->parserSetupArg) must be valid, as must plan->parse_mode and
2212  * plan->cursor_options.
2213  *
2214  * Results are stored into *plan (specifically, plan->plancache_list).
2215  * Note that the result data is all in CurrentMemoryContext or child contexts
2216  * thereof; in practice this means it is in the SPI executor context, and
2217  * what we are creating is a "temporary" SPIPlan.  Cruft generated during
2218  * parsing is also left in CurrentMemoryContext.
2219  */
2220 static void
2221 _SPI_prepare_plan(const char *src, SPIPlanPtr plan)
2222 {
2223         List       *raw_parsetree_list;
2224         List       *plancache_list;
2225         ListCell   *list_item;
2226         SPICallbackArg spicallbackarg;
2227         ErrorContextCallback spierrcontext;
2228
2229         /*
2230          * Setup error traceback support for ereport()
2231          */
2232         spicallbackarg.query = src;
2233         spicallbackarg.mode = plan->parse_mode;
2234         spierrcontext.callback = _SPI_error_callback;
2235         spierrcontext.arg = &spicallbackarg;
2236         spierrcontext.previous = error_context_stack;
2237         error_context_stack = &spierrcontext;
2238
2239         /*
2240          * Parse the request string into a list of raw parse trees.
2241          */
2242         raw_parsetree_list = raw_parser(src, plan->parse_mode);
2243
2244         /*
2245          * Do parse analysis and rule rewrite for each raw parsetree, storing the
2246          * results into unsaved plancache entries.
2247          */
2248         plancache_list = NIL;
2249
2250         foreach(list_item, raw_parsetree_list)
2251         {
2252                 RawStmt    *parsetree = lfirst_node(RawStmt, list_item);
2253                 List       *stmt_list;
2254                 CachedPlanSource *plansource;
2255
2256                 /*
2257                  * Create the CachedPlanSource before we do parse analysis, since it
2258                  * needs to see the unmodified raw parse tree.
2259                  */
2260                 plansource = CreateCachedPlan(parsetree,
2261                                                                           src,
2262                                                                           CreateCommandTag(parsetree->stmt));
2263
2264                 /*
2265                  * Parameter datatypes are driven by parserSetup hook if provided,
2266                  * otherwise we use the fixed parameter list.
2267                  */
2268                 if (plan->parserSetup != NULL)
2269                 {
2270                         Assert(plan->nargs == 0);
2271                         stmt_list = pg_analyze_and_rewrite_withcb(parsetree,
2272                                                                                                           src,
2273                                                                                                           plan->parserSetup,
2274                                                                                                           plan->parserSetupArg,
2275                                                                                                           _SPI_current->queryEnv);
2276                 }
2277                 else
2278                 {
2279                         stmt_list = pg_analyze_and_rewrite_fixedparams(parsetree,
2280                                                                                                                    src,
2281                                                                                                                    plan->argtypes,
2282                                                                                                                    plan->nargs,
2283                                                                                                                    _SPI_current->queryEnv);
2284                 }
2285
2286                 /* Finish filling in the CachedPlanSource */
2287                 CompleteCachedPlan(plansource,
2288                                                    stmt_list,
2289                                                    NULL,
2290                                                    plan->argtypes,
2291                                                    plan->nargs,
2292                                                    plan->parserSetup,
2293                                                    plan->parserSetupArg,
2294                                                    plan->cursor_options,
2295                                                    false);      /* not fixed result */
2296
2297                 plancache_list = lappend(plancache_list, plansource);
2298         }
2299
2300         plan->plancache_list = plancache_list;
2301         plan->oneshot = false;
2302
2303         /*
2304          * Pop the error context stack
2305          */
2306         error_context_stack = spierrcontext.previous;
2307 }
2308
2309 /*
2310  * Parse, but don't analyze, a querystring.
2311  *
2312  * This is a stripped-down version of _SPI_prepare_plan that only does the
2313  * initial raw parsing.  It creates "one shot" CachedPlanSources
2314  * that still require parse analysis before execution is possible.
2315  *
2316  * The advantage of using the "one shot" form of CachedPlanSource is that
2317  * we eliminate data copying and invalidation overhead.  Postponing parse
2318  * analysis also prevents issues if some of the raw parsetrees are DDL
2319  * commands that affect validity of later parsetrees.  Both of these
2320  * attributes are good things for SPI_execute() and similar cases.
2321  *
2322  * Results are stored into *plan (specifically, plan->plancache_list).
2323  * Note that the result data is all in CurrentMemoryContext or child contexts
2324  * thereof; in practice this means it is in the SPI executor context, and
2325  * what we are creating is a "temporary" SPIPlan.  Cruft generated during
2326  * parsing is also left in CurrentMemoryContext.
2327  */
2328 static void
2329 _SPI_prepare_oneshot_plan(const char *src, SPIPlanPtr plan)
2330 {
2331         List       *raw_parsetree_list;
2332         List       *plancache_list;
2333         ListCell   *list_item;
2334         SPICallbackArg spicallbackarg;
2335         ErrorContextCallback spierrcontext;
2336
2337         /*
2338          * Setup error traceback support for ereport()
2339          */
2340         spicallbackarg.query = src;
2341         spicallbackarg.mode = plan->parse_mode;
2342         spierrcontext.callback = _SPI_error_callback;
2343         spierrcontext.arg = &spicallbackarg;
2344         spierrcontext.previous = error_context_stack;
2345         error_context_stack = &spierrcontext;
2346
2347         /*
2348          * Parse the request string into a list of raw parse trees.
2349          */
2350         raw_parsetree_list = raw_parser(src, plan->parse_mode);
2351
2352         /*
2353          * Construct plancache entries, but don't do parse analysis yet.
2354          */
2355         plancache_list = NIL;
2356
2357         foreach(list_item, raw_parsetree_list)
2358         {
2359                 RawStmt    *parsetree = lfirst_node(RawStmt, list_item);
2360                 CachedPlanSource *plansource;
2361
2362                 plansource = CreateOneShotCachedPlan(parsetree,
2363                                                                                          src,
2364                                                                                          CreateCommandTag(parsetree->stmt));
2365
2366                 plancache_list = lappend(plancache_list, plansource);
2367         }
2368
2369         plan->plancache_list = plancache_list;
2370         plan->oneshot = true;
2371
2372         /*
2373          * Pop the error context stack
2374          */
2375         error_context_stack = spierrcontext.previous;
2376 }
2377
2378 /*
2379  * _SPI_execute_plan: execute the given plan with the given options
2380  *
2381  * options contains options accessible from outside SPI:
2382  * params: parameter values to pass to query
2383  * read_only: true for read-only execution (no CommandCounterIncrement)
2384  * allow_nonatomic: true to allow nonatomic CALL/DO execution
2385  * must_return_tuples: throw error if query doesn't return tuples
2386  * tcount: execution tuple-count limit, or 0 for none
2387  * dest: DestReceiver to receive output, or NULL for normal SPI output
2388  * owner: ResourceOwner that will be used to hold refcount on plan;
2389  *              if NULL, CurrentResourceOwner is used (ignored for non-saved plan)
2390  *
2391  * Additional, only-internally-accessible options:
2392  * snapshot: query snapshot to use, or InvalidSnapshot for the normal
2393  *              behavior of taking a new snapshot for each query.
2394  * crosscheck_snapshot: for RI use, all others pass InvalidSnapshot
2395  * fire_triggers: true to fire AFTER triggers at end of query (normal case);
2396  *              false means any AFTER triggers are postponed to end of outer query
2397  */
2398 static int
2399 _SPI_execute_plan(SPIPlanPtr plan, const SPIExecuteOptions *options,
2400                                   Snapshot snapshot, Snapshot crosscheck_snapshot,
2401                                   bool fire_triggers)
2402 {
2403         int                     my_res = 0;
2404         uint64          my_processed = 0;
2405         SPITupleTable *my_tuptable = NULL;
2406         int                     res = 0;
2407         bool            allow_nonatomic;
2408         bool            pushed_active_snap = false;
2409         ResourceOwner plan_owner = options->owner;
2410         SPICallbackArg spicallbackarg;
2411         ErrorContextCallback spierrcontext;
2412         CachedPlan *cplan = NULL;
2413         ListCell   *lc1;
2414
2415         /*
2416          * We allow nonatomic behavior only if options->allow_nonatomic is set
2417          * *and* the SPI_OPT_NONATOMIC flag was given when connecting and we are
2418          * not inside a subtransaction.  The latter two tests match whether
2419          * _SPI_commit() would allow a commit; see there for more commentary.
2420          */
2421         allow_nonatomic = options->allow_nonatomic &&
2422                 !_SPI_current->atomic && !IsSubTransaction();
2423
2424         /*
2425          * Setup error traceback support for ereport()
2426          */
2427         spicallbackarg.query = NULL;    /* we'll fill this below */
2428         spicallbackarg.mode = plan->parse_mode;
2429         spierrcontext.callback = _SPI_error_callback;
2430         spierrcontext.arg = &spicallbackarg;
2431         spierrcontext.previous = error_context_stack;
2432         error_context_stack = &spierrcontext;
2433
2434         /*
2435          * We support four distinct snapshot management behaviors:
2436          *
2437          * snapshot != InvalidSnapshot, read_only = true: use exactly the given
2438          * snapshot.
2439          *
2440          * snapshot != InvalidSnapshot, read_only = false: use the given snapshot,
2441          * modified by advancing its command ID before each querytree.
2442          *
2443          * snapshot == InvalidSnapshot, read_only = true: do nothing for queries
2444          * that require no snapshot.  For those that do, ensure that a Portal
2445          * snapshot exists; then use that, or use the entry-time ActiveSnapshot if
2446          * that exists and is different.
2447          *
2448          * snapshot == InvalidSnapshot, read_only = false: do nothing for queries
2449          * that require no snapshot.  For those that do, ensure that a Portal
2450          * snapshot exists; then, in atomic execution (!allow_nonatomic) take a
2451          * full new snapshot for each user command, and advance its command ID
2452          * before each querytree within the command.  In allow_nonatomic mode we
2453          * just use the Portal snapshot unmodified.
2454          *
2455          * In the first two cases, we can just push the snap onto the stack once
2456          * for the whole plan list.
2457          *
2458          * Note that snapshot != InvalidSnapshot implies an atomic execution
2459          * context.
2460          */
2461         if (snapshot != InvalidSnapshot)
2462         {
2463                 /* this intentionally tests the options field not the derived value */
2464                 Assert(!options->allow_nonatomic);
2465                 if (options->read_only)
2466                 {
2467                         PushActiveSnapshot(snapshot);
2468                         pushed_active_snap = true;
2469                 }
2470                 else
2471                 {
2472                         /* Make sure we have a private copy of the snapshot to modify */
2473                         PushCopiedSnapshot(snapshot);
2474                         pushed_active_snap = true;
2475                 }
2476         }
2477
2478         /*
2479          * Ensure that we have a resource owner if plan is saved, and not if it
2480          * isn't.
2481          */
2482         if (!plan->saved)
2483                 plan_owner = NULL;
2484         else if (plan_owner == NULL)
2485                 plan_owner = CurrentResourceOwner;
2486
2487         /*
2488          * We interpret must_return_tuples as "there must be at least one query,
2489          * and all of them must return tuples".  This is a bit laxer than
2490          * SPI_is_cursor_plan's check, but there seems no reason to enforce that
2491          * there be only one query.
2492          */
2493         if (options->must_return_tuples && plan->plancache_list == NIL)
2494                 ereport(ERROR,
2495                                 (errcode(ERRCODE_SYNTAX_ERROR),
2496                                  errmsg("empty query does not return tuples")));
2497
2498         foreach(lc1, plan->plancache_list)
2499         {
2500                 CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc1);
2501                 List       *stmt_list;
2502                 ListCell   *lc2;
2503
2504                 spicallbackarg.query = plansource->query_string;
2505
2506                 /*
2507                  * If this is a one-shot plan, we still need to do parse analysis.
2508                  */
2509                 if (plan->oneshot)
2510                 {
2511                         RawStmt    *parsetree = plansource->raw_parse_tree;
2512                         const char *src = plansource->query_string;
2513                         List       *querytree_list;
2514
2515                         /*
2516                          * Parameter datatypes are driven by parserSetup hook if provided,
2517                          * otherwise we use the fixed parameter list.
2518                          */
2519                         if (parsetree == NULL)
2520                                 querytree_list = NIL;
2521                         else if (plan->parserSetup != NULL)
2522                         {
2523                                 Assert(plan->nargs == 0);
2524                                 querytree_list = pg_analyze_and_rewrite_withcb(parsetree,
2525                                                                                                                            src,
2526                                                                                                                            plan->parserSetup,
2527                                                                                                                            plan->parserSetupArg,
2528                                                                                                                            _SPI_current->queryEnv);
2529                         }
2530                         else
2531                         {
2532                                 querytree_list = pg_analyze_and_rewrite_fixedparams(parsetree,
2533                                                                                                                                         src,
2534                                                                                                                                         plan->argtypes,
2535                                                                                                                                         plan->nargs,
2536                                                                                                                                         _SPI_current->queryEnv);
2537                         }
2538
2539                         /* Finish filling in the CachedPlanSource */
2540                         CompleteCachedPlan(plansource,
2541                                                            querytree_list,
2542                                                            NULL,
2543                                                            plan->argtypes,
2544                                                            plan->nargs,
2545                                                            plan->parserSetup,
2546                                                            plan->parserSetupArg,
2547                                                            plan->cursor_options,
2548                                                            false);      /* not fixed result */
2549                 }
2550
2551                 /*
2552                  * If asked to, complain when query does not return tuples.
2553                  * (Replanning can't change this, so we can check it before that.
2554                  * However, we can't check it till after parse analysis, so in the
2555                  * case of a one-shot plan this is the earliest we could check.)
2556                  */
2557                 if (options->must_return_tuples && !plansource->resultDesc)
2558                 {
2559                         /* try to give a good error message */
2560                         const char *cmdtag;
2561
2562                         /* A SELECT without resultDesc must be SELECT INTO */
2563                         if (plansource->commandTag == CMDTAG_SELECT)
2564                                 cmdtag = "SELECT INTO";
2565                         else
2566                                 cmdtag = GetCommandTagName(plansource->commandTag);
2567                         ereport(ERROR,
2568                                         (errcode(ERRCODE_SYNTAX_ERROR),
2569                         /* translator: %s is name of a SQL command, eg INSERT */
2570                                          errmsg("%s query does not return tuples", cmdtag)));
2571                 }
2572
2573                 /*
2574                  * Replan if needed, and increment plan refcount.  If it's a saved
2575                  * plan, the refcount must be backed by the plan_owner.
2576                  */
2577                 cplan = GetCachedPlan(plansource, options->params,
2578                                                           plan_owner, _SPI_current->queryEnv);
2579
2580                 stmt_list = cplan->stmt_list;
2581
2582                 /*
2583                  * If we weren't given a specific snapshot to use, and the statement
2584                  * list requires a snapshot, set that up.
2585                  */
2586                 if (snapshot == InvalidSnapshot &&
2587                         (list_length(stmt_list) > 1 ||
2588                          (list_length(stmt_list) == 1 &&
2589                           PlannedStmtRequiresSnapshot(linitial_node(PlannedStmt,
2590                                                                                                                 stmt_list)))))
2591                 {
2592                         /*
2593                          * First, ensure there's a Portal-level snapshot.  This back-fills
2594                          * the snapshot stack in case the previous operation was a COMMIT
2595                          * or ROLLBACK inside a procedure or DO block.  (We can't put back
2596                          * the Portal snapshot any sooner, or we'd break cases like doing
2597                          * SET or LOCK just after COMMIT.)  It's enough to check once per
2598                          * statement list, since COMMIT/ROLLBACK/CALL/DO can't appear
2599                          * within a multi-statement list.
2600                          */
2601                         EnsurePortalSnapshotExists();
2602
2603                         /*
2604                          * In the default non-read-only case, get a new per-statement-list
2605                          * snapshot, replacing any that we pushed in a previous cycle.
2606                          * Skip it when doing non-atomic execution, though (we rely
2607                          * entirely on the Portal snapshot in that case).
2608                          */
2609                         if (!options->read_only && !allow_nonatomic)
2610                         {
2611                                 if (pushed_active_snap)
2612                                         PopActiveSnapshot();
2613                                 PushActiveSnapshot(GetTransactionSnapshot());
2614                                 pushed_active_snap = true;
2615                         }
2616                 }
2617
2618                 foreach(lc2, stmt_list)
2619                 {
2620                         PlannedStmt *stmt = lfirst_node(PlannedStmt, lc2);
2621                         bool            canSetTag = stmt->canSetTag;
2622                         DestReceiver *dest;
2623
2624                         /*
2625                          * Reset output state.  (Note that if a non-SPI receiver is used,
2626                          * _SPI_current->processed will stay zero, and that's what we'll
2627                          * report to the caller.  It's the receiver's job to count tuples
2628                          * in that case.)
2629                          */
2630                         _SPI_current->processed = 0;
2631                         _SPI_current->tuptable = NULL;
2632
2633                         /* Check for unsupported cases. */
2634                         if (stmt->utilityStmt)
2635                         {
2636                                 if (IsA(stmt->utilityStmt, CopyStmt))
2637                                 {
2638                                         CopyStmt   *cstmt = (CopyStmt *) stmt->utilityStmt;
2639
2640                                         if (cstmt->filename == NULL)
2641                                         {
2642                                                 my_res = SPI_ERROR_COPY;
2643                                                 goto fail;
2644                                         }
2645                                 }
2646                                 else if (IsA(stmt->utilityStmt, TransactionStmt))
2647                                 {
2648                                         my_res = SPI_ERROR_TRANSACTION;
2649                                         goto fail;
2650                                 }
2651                         }
2652
2653                         if (options->read_only && !CommandIsReadOnly(stmt))
2654                                 ereport(ERROR,
2655                                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2656                                 /* translator: %s is a SQL statement name */
2657                                                  errmsg("%s is not allowed in a non-volatile function",
2658                                                                 CreateCommandName((Node *) stmt))));
2659
2660                         /*
2661                          * If not read-only mode, advance the command counter before each
2662                          * command and update the snapshot.  (But skip it if the snapshot
2663                          * isn't under our control.)
2664                          */
2665                         if (!options->read_only && pushed_active_snap)
2666                         {
2667                                 CommandCounterIncrement();
2668                                 UpdateActiveSnapshotCommandId();
2669                         }
2670
2671                         /*
2672                          * Select appropriate tuple receiver.  Output from non-canSetTag
2673                          * subqueries always goes to the bit bucket.
2674                          */
2675                         if (!canSetTag)
2676                                 dest = CreateDestReceiver(DestNone);
2677                         else if (options->dest)
2678                                 dest = options->dest;
2679                         else
2680                                 dest = CreateDestReceiver(DestSPI);
2681
2682                         if (stmt->utilityStmt == NULL)
2683                         {
2684                                 QueryDesc  *qdesc;
2685                                 Snapshot        snap;
2686
2687                                 if (ActiveSnapshotSet())
2688                                         snap = GetActiveSnapshot();
2689                                 else
2690                                         snap = InvalidSnapshot;
2691
2692                                 qdesc = CreateQueryDesc(stmt,
2693                                                                                 plansource->query_string,
2694                                                                                 snap, crosscheck_snapshot,
2695                                                                                 dest,
2696                                                                                 options->params,
2697                                                                                 _SPI_current->queryEnv,
2698                                                                                 0);
2699                                 res = _SPI_pquery(qdesc, fire_triggers,
2700                                                                   canSetTag ? options->tcount : 0);
2701                                 FreeQueryDesc(qdesc);
2702                         }
2703                         else
2704                         {
2705                                 ProcessUtilityContext context;
2706                                 QueryCompletion qc;
2707
2708                                 /*
2709                                  * If we're not allowing nonatomic operations, tell
2710                                  * ProcessUtility this is an atomic execution context.
2711                                  */
2712                                 if (allow_nonatomic)
2713                                         context = PROCESS_UTILITY_QUERY_NONATOMIC;
2714                                 else
2715                                         context = PROCESS_UTILITY_QUERY;
2716
2717                                 InitializeQueryCompletion(&qc);
2718                                 ProcessUtility(stmt,
2719                                                            plansource->query_string,
2720                                                            true,        /* protect plancache's node tree */
2721                                                            context,
2722                                                            options->params,
2723                                                            _SPI_current->queryEnv,
2724                                                            dest,
2725                                                            &qc);
2726
2727                                 /* Update "processed" if stmt returned tuples */
2728                                 if (_SPI_current->tuptable)
2729                                         _SPI_current->processed = _SPI_current->tuptable->numvals;
2730
2731                                 res = SPI_OK_UTILITY;
2732
2733                                 /*
2734                                  * Some utility statements return a row count, even though the
2735                                  * tuples are not returned to the caller.
2736                                  */
2737                                 if (IsA(stmt->utilityStmt, CreateTableAsStmt))
2738                                 {
2739                                         CreateTableAsStmt *ctastmt = (CreateTableAsStmt *) stmt->utilityStmt;
2740
2741                                         if (qc.commandTag == CMDTAG_SELECT)
2742                                                 _SPI_current->processed = qc.nprocessed;
2743                                         else
2744                                         {
2745                                                 /*
2746                                                  * Must be an IF NOT EXISTS that did nothing, or a
2747                                                  * CREATE ... WITH NO DATA.
2748                                                  */
2749                                                 Assert(ctastmt->if_not_exists ||
2750                                                            ctastmt->into->skipData);
2751                                                 _SPI_current->processed = 0;
2752                                         }
2753
2754                                         /*
2755                                          * For historical reasons, if CREATE TABLE AS was spelled
2756                                          * as SELECT INTO, return a special return code.
2757                                          */
2758                                         if (ctastmt->is_select_into)
2759                                                 res = SPI_OK_SELINTO;
2760                                 }
2761                                 else if (IsA(stmt->utilityStmt, CopyStmt))
2762                                 {
2763                                         Assert(qc.commandTag == CMDTAG_COPY);
2764                                         _SPI_current->processed = qc.nprocessed;
2765                                 }
2766                         }
2767
2768                         /*
2769                          * The last canSetTag query sets the status values returned to the
2770                          * caller.  Be careful to free any tuptables not returned, to
2771                          * avoid intra-transaction memory leak.
2772                          */
2773                         if (canSetTag)
2774                         {
2775                                 my_processed = _SPI_current->processed;
2776                                 SPI_freetuptable(my_tuptable);
2777                                 my_tuptable = _SPI_current->tuptable;
2778                                 my_res = res;
2779                         }
2780                         else
2781                         {
2782                                 SPI_freetuptable(_SPI_current->tuptable);
2783                                 _SPI_current->tuptable = NULL;
2784                         }
2785
2786                         /*
2787                          * We don't issue a destroy call to the receiver.  The SPI and
2788                          * None receivers would ignore it anyway, while if the caller
2789                          * supplied a receiver, it's not our job to destroy it.
2790                          */
2791
2792                         if (res < 0)
2793                         {
2794                                 my_res = res;
2795                                 goto fail;
2796                         }
2797                 }
2798
2799                 /* Done with this plan, so release refcount */
2800                 ReleaseCachedPlan(cplan, plan_owner);
2801                 cplan = NULL;
2802
2803                 /*
2804                  * If not read-only mode, advance the command counter after the last
2805                  * command.  This ensures that its effects are visible, in case it was
2806                  * DDL that would affect the next CachedPlanSource.
2807                  */
2808                 if (!options->read_only)
2809                         CommandCounterIncrement();
2810         }
2811
2812 fail:
2813
2814         /* Pop the snapshot off the stack if we pushed one */
2815         if (pushed_active_snap)
2816                 PopActiveSnapshot();
2817
2818         /* We no longer need the cached plan refcount, if any */
2819         if (cplan)
2820                 ReleaseCachedPlan(cplan, plan_owner);
2821
2822         /*
2823          * Pop the error context stack
2824          */
2825         error_context_stack = spierrcontext.previous;
2826
2827         /* Save results for caller */
2828         SPI_processed = my_processed;
2829         SPI_tuptable = my_tuptable;
2830
2831         /* tuptable now is caller's responsibility, not SPI's */
2832         _SPI_current->tuptable = NULL;
2833
2834         /*
2835          * If none of the queries had canSetTag, return SPI_OK_REWRITTEN. Prior to
2836          * 8.4, we used return the last query's result code, but not its auxiliary
2837          * results, but that's confusing.
2838          */
2839         if (my_res == 0)
2840                 my_res = SPI_OK_REWRITTEN;
2841
2842         return my_res;
2843 }
2844
2845 /*
2846  * Convert arrays of query parameters to form wanted by planner and executor
2847  */
2848 static ParamListInfo
2849 _SPI_convert_params(int nargs, Oid *argtypes,
2850                                         Datum *Values, const char *Nulls)
2851 {
2852         ParamListInfo paramLI;
2853
2854         if (nargs > 0)
2855         {
2856                 paramLI = makeParamList(nargs);
2857
2858                 for (int i = 0; i < nargs; i++)
2859                 {
2860                         ParamExternData *prm = &paramLI->params[i];
2861
2862                         prm->value = Values[i];
2863                         prm->isnull = (Nulls && Nulls[i] == 'n');
2864                         prm->pflags = PARAM_FLAG_CONST;
2865                         prm->ptype = argtypes[i];
2866                 }
2867         }
2868         else
2869                 paramLI = NULL;
2870         return paramLI;
2871 }
2872
2873 static int
2874 _SPI_pquery(QueryDesc *queryDesc, bool fire_triggers, uint64 tcount)
2875 {
2876         int                     operation = queryDesc->operation;
2877         int                     eflags;
2878         int                     res;
2879
2880         switch (operation)
2881         {
2882                 case CMD_SELECT:
2883                         if (queryDesc->dest->mydest == DestNone)
2884                         {
2885                                 /* Don't return SPI_OK_SELECT if we're discarding result */
2886                                 res = SPI_OK_UTILITY;
2887                         }
2888                         else
2889                                 res = SPI_OK_SELECT;
2890                         break;
2891                 case CMD_INSERT:
2892                         if (queryDesc->plannedstmt->hasReturning)
2893                                 res = SPI_OK_INSERT_RETURNING;
2894                         else
2895                                 res = SPI_OK_INSERT;
2896                         break;
2897                 case CMD_DELETE:
2898                         if (queryDesc->plannedstmt->hasReturning)
2899                                 res = SPI_OK_DELETE_RETURNING;
2900                         else
2901                                 res = SPI_OK_DELETE;
2902                         break;
2903                 case CMD_UPDATE:
2904                         if (queryDesc->plannedstmt->hasReturning)
2905                                 res = SPI_OK_UPDATE_RETURNING;
2906                         else
2907                                 res = SPI_OK_UPDATE;
2908                         break;
2909                 case CMD_MERGE:
2910                         if (queryDesc->plannedstmt->hasReturning)
2911                                 res = SPI_OK_MERGE_RETURNING;
2912                         else
2913                                 res = SPI_OK_MERGE;
2914                         break;
2915                 default:
2916                         return SPI_ERROR_OPUNKNOWN;
2917         }
2918
2919 #ifdef SPI_EXECUTOR_STATS
2920         if (ShowExecutorStats)
2921                 ResetUsage();
2922 #endif
2923
2924         /* Select execution options */
2925         if (fire_triggers)
2926                 eflags = 0;                             /* default run-to-completion flags */
2927         else
2928                 eflags = EXEC_FLAG_SKIP_TRIGGERS;
2929
2930         ExecutorStart(queryDesc, eflags);
2931
2932         ExecutorRun(queryDesc, ForwardScanDirection, tcount);
2933
2934         _SPI_current->processed = queryDesc->estate->es_processed;
2935
2936         if ((res == SPI_OK_SELECT || queryDesc->plannedstmt->hasReturning) &&
2937                 queryDesc->dest->mydest == DestSPI)
2938         {
2939                 if (_SPI_checktuples())
2940                         elog(ERROR, "consistency check on SPI tuple count failed");
2941         }
2942
2943         ExecutorFinish(queryDesc);
2944         ExecutorEnd(queryDesc);
2945         /* FreeQueryDesc is done by the caller */
2946
2947 #ifdef SPI_EXECUTOR_STATS
2948         if (ShowExecutorStats)
2949                 ShowUsage("SPI EXECUTOR STATS");
2950 #endif
2951
2952         return res;
2953 }
2954
2955 /*
2956  * _SPI_error_callback
2957  *
2958  * Add context information when a query invoked via SPI fails
2959  */
2960 static void
2961 _SPI_error_callback(void *arg)
2962 {
2963         SPICallbackArg *carg = (SPICallbackArg *) arg;
2964         const char *query = carg->query;
2965         int                     syntaxerrposition;
2966
2967         if (query == NULL)                      /* in case arg wasn't set yet */
2968                 return;
2969
2970         /*
2971          * If there is a syntax error position, convert to internal syntax error;
2972          * otherwise treat the query as an item of context stack
2973          */
2974         syntaxerrposition = geterrposition();
2975         if (syntaxerrposition > 0)
2976         {
2977                 errposition(0);
2978                 internalerrposition(syntaxerrposition);
2979                 internalerrquery(query);
2980         }
2981         else
2982         {
2983                 /* Use the parse mode to decide how to describe the query */
2984                 switch (carg->mode)
2985                 {
2986                         case RAW_PARSE_PLPGSQL_EXPR:
2987                                 errcontext("PL/pgSQL expression \"%s\"", query);
2988                                 break;
2989                         case RAW_PARSE_PLPGSQL_ASSIGN1:
2990                         case RAW_PARSE_PLPGSQL_ASSIGN2:
2991                         case RAW_PARSE_PLPGSQL_ASSIGN3:
2992                                 errcontext("PL/pgSQL assignment \"%s\"", query);
2993                                 break;
2994                         default:
2995                                 errcontext("SQL statement \"%s\"", query);
2996                                 break;
2997                 }
2998         }
2999 }
3000
3001 /*
3002  * _SPI_cursor_operation()
3003  *
3004  *      Do a FETCH or MOVE in a cursor
3005  */
3006 static void
3007 _SPI_cursor_operation(Portal portal, FetchDirection direction, long count,
3008                                           DestReceiver *dest)
3009 {
3010         uint64          nfetched;
3011
3012         /* Check that the portal is valid */
3013         if (!PortalIsValid(portal))
3014                 elog(ERROR, "invalid portal in SPI cursor operation");
3015
3016         /* Push the SPI stack */
3017         if (_SPI_begin_call(true) < 0)
3018                 elog(ERROR, "SPI cursor operation called while not connected");
3019
3020         /* Reset the SPI result (note we deliberately don't touch lastoid) */
3021         SPI_processed = 0;
3022         SPI_tuptable = NULL;
3023         _SPI_current->processed = 0;
3024         _SPI_current->tuptable = NULL;
3025
3026         /* Run the cursor */
3027         nfetched = PortalRunFetch(portal,
3028                                                           direction,
3029                                                           count,
3030                                                           dest);
3031
3032         /*
3033          * Think not to combine this store with the preceding function call. If
3034          * the portal contains calls to functions that use SPI, then _SPI_stack is
3035          * likely to move around while the portal runs.  When control returns,
3036          * _SPI_current will point to the correct stack entry... but the pointer
3037          * may be different than it was beforehand. So we must be sure to re-fetch
3038          * the pointer after the function call completes.
3039          */
3040         _SPI_current->processed = nfetched;
3041
3042         if (dest->mydest == DestSPI && _SPI_checktuples())
3043                 elog(ERROR, "consistency check on SPI tuple count failed");
3044
3045         /* Put the result into place for access by caller */
3046         SPI_processed = _SPI_current->processed;
3047         SPI_tuptable = _SPI_current->tuptable;
3048
3049         /* tuptable now is caller's responsibility, not SPI's */
3050         _SPI_current->tuptable = NULL;
3051
3052         /* Pop the SPI stack */
3053         _SPI_end_call(true);
3054 }
3055
3056
3057 static MemoryContext
3058 _SPI_execmem(void)
3059 {
3060         return MemoryContextSwitchTo(_SPI_current->execCxt);
3061 }
3062
3063 static MemoryContext
3064 _SPI_procmem(void)
3065 {
3066         return MemoryContextSwitchTo(_SPI_current->procCxt);
3067 }
3068
3069 /*
3070  * _SPI_begin_call: begin a SPI operation within a connected procedure
3071  *
3072  * use_exec is true if we intend to make use of the procedure's execCxt
3073  * during this SPI operation.  We'll switch into that context, and arrange
3074  * for it to be cleaned up at _SPI_end_call or if an error occurs.
3075  */
3076 static int
3077 _SPI_begin_call(bool use_exec)
3078 {
3079         if (_SPI_current == NULL)
3080                 return SPI_ERROR_UNCONNECTED;
3081
3082         if (use_exec)
3083         {
3084                 /* remember when the Executor operation started */
3085                 _SPI_current->execSubid = GetCurrentSubTransactionId();
3086                 /* switch to the Executor memory context */
3087                 _SPI_execmem();
3088         }
3089
3090         return 0;
3091 }
3092
3093 /*
3094  * _SPI_end_call: end a SPI operation within a connected procedure
3095  *
3096  * use_exec must be the same as in the previous _SPI_begin_call
3097  *
3098  * Note: this currently has no failure return cases, so callers don't check
3099  */
3100 static int
3101 _SPI_end_call(bool use_exec)
3102 {
3103         if (use_exec)
3104         {
3105                 /* switch to the procedure memory context */
3106                 _SPI_procmem();
3107                 /* mark Executor context no longer in use */
3108                 _SPI_current->execSubid = InvalidSubTransactionId;
3109                 /* and free Executor memory */
3110                 MemoryContextReset(_SPI_current->execCxt);
3111         }
3112
3113         return 0;
3114 }
3115
3116 static bool
3117 _SPI_checktuples(void)
3118 {
3119         uint64          processed = _SPI_current->processed;
3120         SPITupleTable *tuptable = _SPI_current->tuptable;
3121         bool            failed = false;
3122
3123         if (tuptable == NULL)           /* spi_dest_startup was not called */
3124                 failed = true;
3125         else if (processed != tuptable->numvals)
3126                 failed = true;
3127
3128         return failed;
3129 }
3130
3131 /*
3132  * Convert a "temporary" SPIPlan into an "unsaved" plan.
3133  *
3134  * The passed _SPI_plan struct is on the stack, and all its subsidiary data
3135  * is in or under the current SPI executor context.  Copy the plan into the
3136  * SPI procedure context so it will survive _SPI_end_call().  To minimize
3137  * data copying, this destructively modifies the input plan, by taking the
3138  * plancache entries away from it and reparenting them to the new SPIPlan.
3139  */
3140 static SPIPlanPtr
3141 _SPI_make_plan_non_temp(SPIPlanPtr plan)
3142 {
3143         SPIPlanPtr      newplan;
3144         MemoryContext parentcxt = _SPI_current->procCxt;
3145         MemoryContext plancxt;
3146         MemoryContext oldcxt;
3147         ListCell   *lc;
3148
3149         /* Assert the input is a temporary SPIPlan */
3150         Assert(plan->magic == _SPI_PLAN_MAGIC);
3151         Assert(plan->plancxt == NULL);
3152         /* One-shot plans can't be saved */
3153         Assert(!plan->oneshot);
3154
3155         /*
3156          * Create a memory context for the plan, underneath the procedure context.
3157          * We don't expect the plan to be very large.
3158          */
3159         plancxt = AllocSetContextCreate(parentcxt,
3160                                                                         "SPI Plan",
3161                                                                         ALLOCSET_SMALL_SIZES);
3162         oldcxt = MemoryContextSwitchTo(plancxt);
3163
3164         /* Copy the _SPI_plan struct and subsidiary data into the new context */
3165         newplan = (SPIPlanPtr) palloc0(sizeof(_SPI_plan));
3166         newplan->magic = _SPI_PLAN_MAGIC;
3167         newplan->plancxt = plancxt;
3168         newplan->parse_mode = plan->parse_mode;
3169         newplan->cursor_options = plan->cursor_options;
3170         newplan->nargs = plan->nargs;
3171         if (plan->nargs > 0)
3172         {
3173                 newplan->argtypes = (Oid *) palloc(plan->nargs * sizeof(Oid));
3174                 memcpy(newplan->argtypes, plan->argtypes, plan->nargs * sizeof(Oid));
3175         }
3176         else
3177                 newplan->argtypes = NULL;
3178         newplan->parserSetup = plan->parserSetup;
3179         newplan->parserSetupArg = plan->parserSetupArg;
3180
3181         /*
3182          * Reparent all the CachedPlanSources into the procedure context.  In
3183          * theory this could fail partway through due to the pallocs, but we don't
3184          * care too much since both the procedure context and the executor context
3185          * would go away on error.
3186          */
3187         foreach(lc, plan->plancache_list)
3188         {
3189                 CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc);
3190
3191                 CachedPlanSetParentContext(plansource, parentcxt);
3192
3193                 /* Build new list, with list cells in plancxt */
3194                 newplan->plancache_list = lappend(newplan->plancache_list, plansource);
3195         }
3196
3197         MemoryContextSwitchTo(oldcxt);
3198
3199         /* For safety, unlink the CachedPlanSources from the temporary plan */
3200         plan->plancache_list = NIL;
3201
3202         return newplan;
3203 }
3204
3205 /*
3206  * Make a "saved" copy of the given plan.
3207  */
3208 static SPIPlanPtr
3209 _SPI_save_plan(SPIPlanPtr plan)
3210 {
3211         SPIPlanPtr      newplan;
3212         MemoryContext plancxt;
3213         MemoryContext oldcxt;
3214         ListCell   *lc;
3215
3216         /* One-shot plans can't be saved */
3217         Assert(!plan->oneshot);
3218
3219         /*
3220          * Create a memory context for the plan.  We don't expect the plan to be
3221          * very large, so use smaller-than-default alloc parameters.  It's a
3222          * transient context until we finish copying everything.
3223          */
3224         plancxt = AllocSetContextCreate(CurrentMemoryContext,
3225                                                                         "SPI Plan",
3226                                                                         ALLOCSET_SMALL_SIZES);
3227         oldcxt = MemoryContextSwitchTo(plancxt);
3228
3229         /* Copy the SPI plan into its own context */
3230         newplan = (SPIPlanPtr) palloc0(sizeof(_SPI_plan));
3231         newplan->magic = _SPI_PLAN_MAGIC;
3232         newplan->plancxt = plancxt;
3233         newplan->parse_mode = plan->parse_mode;
3234         newplan->cursor_options = plan->cursor_options;
3235         newplan->nargs = plan->nargs;
3236         if (plan->nargs > 0)
3237         {
3238                 newplan->argtypes = (Oid *) palloc(plan->nargs * sizeof(Oid));
3239                 memcpy(newplan->argtypes, plan->argtypes, plan->nargs * sizeof(Oid));
3240         }
3241         else
3242                 newplan->argtypes = NULL;
3243         newplan->parserSetup = plan->parserSetup;
3244         newplan->parserSetupArg = plan->parserSetupArg;
3245
3246         /* Copy all the plancache entries */
3247         foreach(lc, plan->plancache_list)
3248         {
3249                 CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc);
3250                 CachedPlanSource *newsource;
3251
3252                 newsource = CopyCachedPlan(plansource);
3253                 newplan->plancache_list = lappend(newplan->plancache_list, newsource);
3254         }
3255
3256         MemoryContextSwitchTo(oldcxt);
3257
3258         /*
3259          * Mark it saved, reparent it under CacheMemoryContext, and mark all the
3260          * component CachedPlanSources as saved.  This sequence cannot fail
3261          * partway through, so there's no risk of long-term memory leakage.
3262          */
3263         newplan->saved = true;
3264         MemoryContextSetParent(newplan->plancxt, CacheMemoryContext);
3265
3266         foreach(lc, newplan->plancache_list)
3267         {
3268                 CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc);
3269
3270                 SaveCachedPlan(plansource);
3271         }
3272
3273         return newplan;
3274 }
3275
3276 /*
3277  * Internal lookup of ephemeral named relation by name.
3278  */
3279 static EphemeralNamedRelation
3280 _SPI_find_ENR_by_name(const char *name)
3281 {
3282         /* internal static function; any error is bug in SPI itself */
3283         Assert(name != NULL);
3284
3285         /* fast exit if no tuplestores have been added */
3286         if (_SPI_current->queryEnv == NULL)
3287                 return NULL;
3288
3289         return get_ENR(_SPI_current->queryEnv, name);
3290 }
3291
3292 /*
3293  * Register an ephemeral named relation for use by the planner and executor on
3294  * subsequent calls using this SPI connection.
3295  */
3296 int
3297 SPI_register_relation(EphemeralNamedRelation enr)
3298 {
3299         EphemeralNamedRelation match;
3300         int                     res;
3301
3302         if (enr == NULL || enr->md.name == NULL)
3303                 return SPI_ERROR_ARGUMENT;
3304
3305         res = _SPI_begin_call(false);   /* keep current memory context */
3306         if (res < 0)
3307                 return res;
3308
3309         match = _SPI_find_ENR_by_name(enr->md.name);
3310         if (match)
3311                 res = SPI_ERROR_REL_DUPLICATE;
3312         else
3313         {
3314                 if (_SPI_current->queryEnv == NULL)
3315                         _SPI_current->queryEnv = create_queryEnv();
3316
3317                 register_ENR(_SPI_current->queryEnv, enr);
3318                 res = SPI_OK_REL_REGISTER;
3319         }
3320
3321         _SPI_end_call(false);
3322
3323         return res;
3324 }
3325
3326 /*
3327  * Unregister an ephemeral named relation by name.  This will probably be a
3328  * rarely used function, since SPI_finish will clear it automatically.
3329  */
3330 int
3331 SPI_unregister_relation(const char *name)
3332 {
3333         EphemeralNamedRelation match;
3334         int                     res;
3335
3336         if (name == NULL)
3337                 return SPI_ERROR_ARGUMENT;
3338
3339         res = _SPI_begin_call(false);   /* keep current memory context */
3340         if (res < 0)
3341                 return res;
3342
3343         match = _SPI_find_ENR_by_name(name);
3344         if (match)
3345         {
3346                 unregister_ENR(_SPI_current->queryEnv, match->md.name);
3347                 res = SPI_OK_REL_UNREGISTER;
3348         }
3349         else
3350                 res = SPI_ERROR_REL_NOT_FOUND;
3351
3352         _SPI_end_call(false);
3353
3354         return res;
3355 }
3356
3357 /*
3358  * Register the transient relations from 'tdata' using this SPI connection.
3359  * This should be called by PL implementations' trigger handlers after
3360  * connecting, in order to make transition tables visible to any queries run
3361  * in this connection.
3362  */
3363 int
3364 SPI_register_trigger_data(TriggerData *tdata)
3365 {
3366         if (tdata == NULL)
3367                 return SPI_ERROR_ARGUMENT;
3368
3369         if (tdata->tg_newtable)
3370         {
3371                 EphemeralNamedRelation enr =
3372                         palloc(sizeof(EphemeralNamedRelationData));
3373                 int                     rc;
3374
3375                 enr->md.name = tdata->tg_trigger->tgnewtable;
3376                 enr->md.reliddesc = tdata->tg_relation->rd_id;
3377                 enr->md.tupdesc = NULL;
3378                 enr->md.enrtype = ENR_NAMED_TUPLESTORE;
3379                 enr->md.enrtuples = tuplestore_tuple_count(tdata->tg_newtable);
3380                 enr->reldata = tdata->tg_newtable;
3381                 rc = SPI_register_relation(enr);
3382                 if (rc != SPI_OK_REL_REGISTER)
3383                         return rc;
3384         }
3385
3386         if (tdata->tg_oldtable)
3387         {
3388                 EphemeralNamedRelation enr =
3389                         palloc(sizeof(EphemeralNamedRelationData));
3390                 int                     rc;
3391
3392                 enr->md.name = tdata->tg_trigger->tgoldtable;
3393                 enr->md.reliddesc = tdata->tg_relation->rd_id;
3394                 enr->md.tupdesc = NULL;
3395                 enr->md.enrtype = ENR_NAMED_TUPLESTORE;
3396                 enr->md.enrtuples = tuplestore_tuple_count(tdata->tg_oldtable);
3397                 enr->reldata = tdata->tg_oldtable;
3398                 rc = SPI_register_relation(enr);
3399                 if (rc != SPI_OK_REL_REGISTER)
3400                         return rc;
3401         }
3402
3403         return SPI_OK_TD_REGISTER;
3404 }