implement prepared queries in plperl - Mailing list pgsql-patches

--
Sincerely,
    Dmitry Karasik


diff -rcN plperl.cvs/SPI.xs plperl.0/SPI.xs
*** plperl.cvs/SPI.xs    Thu Oct 27 12:34:29 2005
--- plperl.0/SPI.xs    Thu Dec  8 10:35:38 2005
***************
*** 146,150 ****
--- 146,226 ----
      OUTPUT:
          RETVAL

+ SV*
+ spi_spi_prepare(query, ...)
+     char* query;
+     CODE:
+         int i;
+         SV** argv;
+         if (items < 1)
+             Perl_croak(aTHX_ "Usage: spi_prepare(query, ...)");
+         argv = ( SV**) palloc(( items - 1) * sizeof(SV*));
+         if ( argv == NULL)
+             Perl_croak(aTHX_ "spi_prepare: not enough memory");
+         for ( i = 1; i < items; i++)
+             argv[i - 1] = ST(i);
+         RETVAL = plperl_spi_prepare(query, items - 1, argv);
+         pfree( argv);
+     OUTPUT:
+         RETVAL
+
+ SV*
+ spi_spi_exec_prepared(query, ...)
+     char * query;
+     PREINIT:
+         HV *ret_hash;
+     CODE:
+         HV *attr = NULL;
+         int i, offset = 1, argc;
+         SV ** argv;
+         if ( items < 1)
+             Perl_croak(aTHX_ "Usage: spi_exec_prepared(query, [\\%%attr,] [\\@bind_values]");
+         if ( items > 1 && SvROK( ST( 1)) && SvTYPE( SvRV( ST( 1))) == SVt_PVHV) {
+             attr = ( HV*) SvRV(ST(1));
+             offset++;
+         }
+         argc = items - offset;
+         argv = ( SV**) palloc( argc * sizeof(SV*));
+         if ( argv == NULL)
+             Perl_croak(aTHX_ "spi_exec_prepared: not enough memory");
+         for ( i = 0; offset < items; offset++, i++)
+             argv[i] = ST(offset);
+         ret_hash = plperl_spi_exec_prepared(query, attr, argc, argv);
+         RETVAL = newRV_noinc((SV*)ret_hash);
+         pfree( argv);
+     OUTPUT:
+         RETVAL
+
+ SV*
+ spi_spi_query_prepared(query, ...)
+     char * query;
+     CODE:
+         int i;
+         SV ** argv;
+         if ( items < 1)
+             Perl_croak(aTHX_ "Usage: spi_query_prepared(query, [\\@bind_values]");
+         argv = ( SV**) palloc(( items - 1) * sizeof(SV*));
+         if ( argv == NULL)
+             Perl_croak(aTHX_ "spi_query_prepared: not enough memory");
+         for ( i = 1; i < items; i++)
+             argv[i - 1] = ST(i);
+         RETVAL = plperl_spi_query_prepared(query, items - 1, argv);
+         pfree( argv);
+     OUTPUT:
+         RETVAL
+
+ void
+ spi_spi_freeplan(query)
+     char *query;
+     CODE:
+         plperl_spi_freeplan(query);
+
+ void
+ spi_spi_cursor_close(cursor)
+     char *cursor;
+     CODE:
+         plperl_spi_cursor_close(cursor);
+
+
  BOOT:
      items = 0;  /* avoid 'unused variable' warning */
diff -rcN plperl.cvs/expected/plperl.out plperl.0/expected/plperl.out
*** plperl.cvs/expected/plperl.out    Tue Nov 22 11:48:57 2005
--- plperl.0/expected/plperl.out    Thu Dec  8 10:35:57 2005
***************
*** 367,372 ****
--- 367,386 ----
               2
  (2 rows)

+ --
+ -- Test spi_fetchrow abort
+ --
+ CREATE OR REPLACE FUNCTION perl_spi_func2() RETURNS INTEGER AS $$
+ my $x = spi_query("select 1 as a union select 2 as a");
+ spi_cursor_close( $x);
+ return 0;
+ $$ LANGUAGE plperl;
+ SELECT * from perl_spi_func2();
+  perl_spi_func2
+ ----------------
+               0
+ (1 row)
+
  ---
  --- Test recursion via SPI
  ---
***************
*** 419,422 ****
--- 433,470 ----
  ---------------------------------------
   {{"a\"b",NULL,"c,d"},{"e\\f",NULL,g}}
  (1 row)
+
+ --
+ -- Test spi_prepare/spi_exec_prepared/spi_freeplan
+ --
+ CREATE OR REPLACE FUNCTION perl_spi_prepared(INTEGER) RETURNS INTEGER AS $$
+    my $x = spi_prepare('select $1 AS a', 'INT4');
+    my $q = spi_exec_prepared( $x, $_[0] + 1);
+    spi_freeplan($x);
+ return $q->{rows}->[0]->{a};
+ $$ LANGUAGE plperl;
+ SELECT * from perl_spi_prepared(42);
+  perl_spi_prepared
+ -------------------
+                 43
+ (1 row)
+
+ --
+ -- Test spi_prepare/spi_query_prepared/spi_freeplan
+ --
+ CREATE OR REPLACE FUNCTION perl_spi_prepared_set(INTEGER, INTEGER) RETURNS SETOF INTEGER AS $$
+   my $x = spi_prepare('SELECT $1 AS a union select $2 as a', 'INT4', 'INT4');
+   my $q = spi_query_prepared( $x, 1+$_[0], 2+$_[1]);
+   while (defined (my $y = spi_fetchrow($q))) {
+       return_next $y->{a};
+   }
+   spi_freeplan($x);
+   return;
+ $$ LANGUAGE plperl;
+ SELECT * from perl_spi_prepared_set(1,2);
+  perl_spi_prepared_set
+ -----------------------
+                      2
+                      4
+ (2 rows)

diff -rcN plperl.cvs/plperl.c plperl.0/plperl.c
*** plperl.cvs/plperl.c    Thu Dec  1 13:49:22 2005
--- plperl.0/plperl.c    Thu Dec  8 10:51:31 2005
***************
*** 55,60 ****
--- 55,61 ----
  #include "utils/typcache.h"
  #include "miscadmin.h"
  #include "mb/pg_wchar.h"
+ #include "parser/parse_type.h"

  /* perl stuff */
  #include "EXTERN.h"
***************
*** 92,97 ****
--- 93,110 ----
      SV           *reference;
  } plperl_proc_desc;

+ /**********************************************************************
+  * The information we cache about prepared and saved plans
+  **********************************************************************/
+ typedef struct plperl_query_desc
+ {
+     char        qname[sizeof(long) * 2 + 1];
+     void       *plan;
+     int            nargs;
+     Oid           *argtypes;
+     FmgrInfo   *arginfuncs;
+     Oid           *argtypioparams;
+ } plperl_query_desc;

  /**********************************************************************
   * Global data
***************
*** 100,105 ****
--- 113,119 ----
  static bool plperl_safe_init_done = false;
  static PerlInterpreter *plperl_interp = NULL;
  static HV  *plperl_proc_hash = NULL;
+ static HV  *plperl_query_hash = NULL;

  static bool plperl_use_strict = false;

***************
*** 229,235 ****
      "$PLContainer->permit_only(':default');" \
      "$PLContainer->permit(qw[:base_math !:base_io sort time]);" \
      "$PLContainer->share(qw[&elog &spi_exec_query &return_next " \
!     "&spi_query &spi_fetchrow " \
      "&_plperl_to_pg_array " \
      "&DEBUG &LOG &INFO &NOTICE &WARNING &ERROR %_SHARED ]);" \
      "sub ::mksafefunc {" \
--- 243,250 ----
      "$PLContainer->permit_only(':default');" \
      "$PLContainer->permit(qw[:base_math !:base_io sort time]);" \
      "$PLContainer->share(qw[&elog &spi_exec_query &return_next " \
!     "&spi_query &spi_fetchrow &spi_cursor_close " \
!     "&spi_prepare &spi_exec_prepared &spi_query_prepared &spi_freeplan " \
      "&_plperl_to_pg_array " \
      "&DEBUG &LOG &INFO &NOTICE &WARNING &ERROR %_SHARED ]);" \
      "sub ::mksafefunc {" \
***************
*** 269,274 ****
--- 284,290 ----
      perl_run(plperl_interp);

      plperl_proc_hash = newHV();
+     plperl_query_hash = newHV();
  }


***************
*** 1184,1190 ****
      {
          bool        uptodate;

!         prodesc = (plperl_proc_desc *) SvIV(*svp);

          /************************************************************
           * If it's present, must check whether it's still up to date.
--- 1200,1206 ----
      {
          bool        uptodate;

!         prodesc = INT2PTR( plperl_proc_desc *, SvUV(*svp));

          /************************************************************
           * If it's present, must check whether it's still up to date.
***************
*** 1382,1388 ****
          }

          hv_store(plperl_proc_hash, internal_proname, proname_len,
!                  newSViv((IV) prodesc), 0);
      }

      ReleaseSysCache(procTup);
--- 1398,1404 ----
          }

          hv_store(plperl_proc_hash, internal_proname, proname_len,
!                  newSVuv( PTR2UV( prodesc)), 0);
      }

      ReleaseSysCache(procTup);
***************
*** 1654,1669 ****
      PG_TRY();
      {
          void       *plan;
!         Portal        portal = NULL;

          /* Create a cursor for the query */
          plan = SPI_prepare(query, 0, NULL);
!         if (plan)
!             portal = SPI_cursor_open(NULL, plan, NULL, NULL, false);
!         if (portal)
!             cursor = newSVpv(portal->name, 0);
!         else
!             cursor = newSV(0);

          /* Commit the inner transaction, return to outer xact context */
          ReleaseCurrentSubTransaction();
--- 1670,1689 ----
      PG_TRY();
      {
          void       *plan;
!         Portal        portal;

          /* Create a cursor for the query */
          plan = SPI_prepare(query, 0, NULL);
!         if ( plan == NULL)
!             elog(ERROR, "SPI_prepare() failed:%s",
!                 SPI_result_code_string(SPI_result));
!
!         portal = SPI_cursor_open(NULL, plan, NULL, NULL, false);
!         SPI_freeplan( plan);
!         if ( portal == NULL)
!             elog(ERROR, "SPI_cursor_open() failed:%s",
!                 SPI_result_code_string(SPI_result));
!         cursor = newSVpv(portal->name, 0);

          /* Commit the inner transaction, return to outer xact context */
          ReleaseCurrentSubTransaction();
***************
*** 1730,1743 ****
          Portal        p = SPI_cursor_find(cursor);

          if (!p)
!             row = newSV(0);
          else
          {
              SPI_cursor_fetch(p, true, 1);
              if (SPI_processed == 0)
              {
                  SPI_cursor_close(p);
!                 row = newSV(0);
              }
              else
              {
--- 1750,1763 ----
          Portal        p = SPI_cursor_find(cursor);

          if (!p)
!             row = &PL_sv_undef;
          else
          {
              SPI_cursor_fetch(p, true, 1);
              if (SPI_processed == 0)
              {
                  SPI_cursor_close(p);
!                 row = &PL_sv_undef;
              }
              else
              {
***************
*** 1788,1791 ****
--- 1808,2242 ----
      PG_END_TRY();

      return row;
+ }
+
+ void
+ plperl_spi_cursor_close(char *cursor)
+ {
+     Portal p = SPI_cursor_find(cursor);
+     if (p)
+         SPI_cursor_close(p);
+ }
+
+ SV *
+ plperl_spi_prepare(char* query, int argc, SV ** argv)
+ {
+     plperl_query_desc *qdesc;
+     void       *plan;
+     int            i;
+     HeapTuple    typeTup;
+
+     MemoryContext oldcontext = CurrentMemoryContext;
+     ResourceOwner oldowner = CurrentResourceOwner;
+
+     BeginInternalSubTransaction(NULL);
+     MemoryContextSwitchTo(oldcontext);
+
+     /************************************************************
+      * Allocate the new querydesc structure
+      ************************************************************/
+     qdesc = (plperl_query_desc *) malloc(sizeof(plperl_query_desc));
+     MemSet(qdesc, 0, sizeof(plperl_query_desc));
+     snprintf(qdesc-> qname, sizeof(qdesc-> qname), "%lx", (long) qdesc);
+     qdesc-> nargs = argc;
+     qdesc-> argtypes = (Oid *) malloc(argc * sizeof(Oid));
+     qdesc-> arginfuncs = (FmgrInfo *) malloc(argc * sizeof(FmgrInfo));
+     qdesc-> argtypioparams = (Oid *) malloc(argc * sizeof(Oid));
+
+     PG_TRY();
+     {
+         /************************************************************
+          * Lookup the argument types by name in the system cache
+          * and remember the required information for input conversion
+          ************************************************************/
+         for (i = 0; i < argc; i++)
+         {
+             char       *argcopy;
+             List       *names = NIL;
+             ListCell   *l;
+             TypeName   *typename;
+
+             /************************************************************
+              * Use SplitIdentifierString() on a copy of the type name,
+              * turn the resulting pointer list into a TypeName node
+              * and call typenameType() to get the pg_type tuple.
+              ************************************************************/
+             argcopy = pstrdup(SvPV(argv[i],PL_na));
+             SplitIdentifierString(argcopy, '.', &names);
+             typename = makeNode(TypeName);
+             foreach(l, names)
+                 typename->names = lappend(typename->names, makeString(lfirst(l)));
+
+             typeTup = typenameType(typename);
+             qdesc->argtypes[i] = HeapTupleGetOid(typeTup);
+             perm_fmgr_info(((Form_pg_type) GETSTRUCT(typeTup))->typinput,
+                            &(qdesc->arginfuncs[i]));
+             qdesc->argtypioparams[i] = getTypeIOParam(typeTup);
+             ReleaseSysCache(typeTup);
+
+             list_free(typename->names);
+             pfree(typename);
+             list_free(names);
+             pfree(argcopy);
+         }
+
+         /************************************************************
+          * Prepare the plan and check for errors
+          ************************************************************/
+         plan = SPI_prepare(query, argc, qdesc->argtypes);
+
+         if (plan == NULL)
+             elog(ERROR, "SPI_prepare() failed:%s",
+                 SPI_result_code_string(SPI_result));
+
+         /************************************************************
+          * Save the plan into permanent memory (right now it's in the
+          * SPI procCxt, which will go away at function end).
+          ************************************************************/
+         qdesc->plan = SPI_saveplan(plan);
+         if (qdesc->plan == NULL)
+             elog(ERROR, "SPI_saveplan() failed: %s",
+                 SPI_result_code_string(SPI_result));
+
+         /* Release the procCxt copy to avoid within-function memory leak */
+         SPI_freeplan(plan);
+
+         /* Commit the inner transaction, return to outer xact context */
+         ReleaseCurrentSubTransaction();
+         MemoryContextSwitchTo(oldcontext);
+         CurrentResourceOwner = oldowner;
+         /*
+          * AtEOSubXact_SPI() should not have popped any SPI context,
+          * but just in case it did, make sure we remain connected.
+          */
+         SPI_restore_connection();
+     }
+     PG_CATCH();
+     {
+         ErrorData  *edata;
+
+         free(qdesc-> argtypes);
+         free(qdesc-> arginfuncs);
+         free(qdesc-> argtypioparams);
+         free(qdesc);
+
+         /* Save error info */
+         MemoryContextSwitchTo(oldcontext);
+         edata = CopyErrorData();
+         FlushErrorState();
+
+         /* Abort the inner transaction */
+         RollbackAndReleaseCurrentSubTransaction();
+         MemoryContextSwitchTo(oldcontext);
+         CurrentResourceOwner = oldowner;
+
+         /*
+          * If AtEOSubXact_SPI() popped any SPI context of the subxact,
+          * it will have left us in a disconnected state.  We need this
+          * hack to return to connected state.
+          */
+         SPI_restore_connection();
+
+         /* Punt the error to Perl */
+         croak("%s", edata->message);
+
+         /* Can't get here, but keep compiler quiet */
+         return NULL;
+     }
+     PG_END_TRY();
+
+     /************************************************************
+      * Insert a hashtable entry for the plan and return
+      * the key to the caller.
+      ************************************************************/
+     hv_store( plperl_query_hash, qdesc->qname, strlen(qdesc->qname), newSVuv( PTR2UV( qdesc)), 0);
+
+     return newSVpv( qdesc->qname, strlen(qdesc->qname));
+ }
+
+ HV *
+ plperl_spi_exec_prepared(char* query, HV * attr, int argc, SV ** argv)
+ {
+     HV           *ret_hv;
+     SV **sv;
+     int i, limit, spi_rv;
+     char * nulls;
+     Datum       *argvalues;
+     plperl_query_desc *qdesc;
+
+     /*
+      * Execute the query inside a sub-transaction, so we can cope with
+      * errors sanely
+      */
+     MemoryContext oldcontext = CurrentMemoryContext;
+     ResourceOwner oldowner = CurrentResourceOwner;
+
+     BeginInternalSubTransaction(NULL);
+     /* Want to run inside function's memory context */
+     MemoryContextSwitchTo(oldcontext);
+
+     PG_TRY();
+     {
+         /************************************************************
+          * Fetch the saved plan descriptor, see if it's o.k.
+          ************************************************************/
+         sv = hv_fetch(plperl_query_hash, query, strlen(query), 0);
+         if ( sv == NULL)
+             elog(ERROR, "spi_exec_prepared: Invalid prepared query passed");
+         if ( *sv == NULL || !SvOK( *sv))
+             elog(ERROR, "spi_exec_prepared: panic - plperl_query_hash value corrupted");
+
+         qdesc = INT2PTR( plperl_query_desc *, SvUV(*sv));
+         if ( qdesc == NULL)
+             elog(ERROR, "spi_exec_prepared: panic - plperl_query_hash value vanished");
+
+         if ( qdesc-> nargs != argc)
+             elog(ERROR, "spi_exec_prepared: expected %d argument(s), %d passed",
+                 qdesc-> nargs, argc);
+
+         /************************************************************
+          * Parse eventual attributes
+          ************************************************************/
+         limit = 0;
+         if ( attr != NULL) {
+             sv = hv_fetch( attr, "limit", 5, 0);
+             if ( *sv && SvIOK( *sv))
+                 limit = SvIV( *sv);
+         }
+         /************************************************************
+          * Set up arguments
+          ************************************************************/
+         if ( argc > 0) {
+             nulls = (char *)palloc( argc);
+             argvalues = (Datum *) palloc(argc * sizeof(Datum));
+             if ( nulls == NULL || argvalues == NULL)
+                 elog(ERROR, "spi_exec_prepared: not enough memory");
+         } else {
+             nulls = NULL;
+             argvalues = NULL;
+         }
+
+         for ( i = 0; i < argc; i++) {
+             if ( SvTYPE( argv[i]) != SVt_NULL) {
+                 argvalues[i] =
+                     FunctionCall3( &qdesc->arginfuncs[i],
+                           CStringGetDatum( SvPV( argv[i], PL_na)),
+                           ObjectIdGetDatum( qdesc->argtypioparams[i]),
+                           Int32GetDatum(-1)
+                     );
+                 nulls[i] = ' ';
+             } else {
+                 argvalues[i] = (Datum) 0;
+                 nulls[i] = 'n';
+             }
+         }
+
+         /************************************************************
+          * go
+          ************************************************************/
+         spi_rv = SPI_execute_plan(qdesc-> plan, argvalues, nulls,
+                              plperl_current_prodesc->fn_readonly, limit);
+         ret_hv = plperl_spi_execute_fetch_result(SPI_tuptable, SPI_processed,
+                                                  spi_rv);
+         if ( argc > 0) {
+             pfree( argvalues);
+             pfree( nulls);
+         }
+
+         /* Commit the inner transaction, return to outer xact context */
+         ReleaseCurrentSubTransaction();
+         MemoryContextSwitchTo(oldcontext);
+         CurrentResourceOwner = oldowner;
+         /*
+          * AtEOSubXact_SPI() should not have popped any SPI context,
+          * but just in case it did, make sure we remain connected.
+          */
+         SPI_restore_connection();
+     }
+     PG_CATCH();
+     {
+         ErrorData  *edata;
+
+         /* Save error info */
+         MemoryContextSwitchTo(oldcontext);
+         edata = CopyErrorData();
+         FlushErrorState();
+
+         /* Abort the inner transaction */
+         RollbackAndReleaseCurrentSubTransaction();
+         MemoryContextSwitchTo(oldcontext);
+         CurrentResourceOwner = oldowner;
+
+         /*
+          * If AtEOSubXact_SPI() popped any SPI context of the subxact,
+          * it will have left us in a disconnected state.  We need this
+          * hack to return to connected state.
+          */
+         SPI_restore_connection();
+
+         /* Punt the error to Perl */
+         croak("%s", edata->message);
+
+         /* Can't get here, but keep compiler quiet */
+         return NULL;
+     }
+     PG_END_TRY();
+
+     return ret_hv;
+ }
+
+ SV *
+ plperl_spi_query_prepared(char* query, int argc, SV ** argv)
+ {
+     SV **sv;
+     int i;
+     char * nulls;
+     Datum       *argvalues;
+     plperl_query_desc *qdesc;
+     SV *cursor;
+     Portal portal = NULL;
+
+     /*
+      * Execute the query inside a sub-transaction, so we can cope with
+      * errors sanely
+      */
+     MemoryContext oldcontext = CurrentMemoryContext;
+     ResourceOwner oldowner = CurrentResourceOwner;
+
+     BeginInternalSubTransaction(NULL);
+     /* Want to run inside function's memory context */
+     MemoryContextSwitchTo(oldcontext);
+
+     PG_TRY();
+     {
+         /************************************************************
+          * Fetch the saved plan descriptor, see if it's o.k.
+          ************************************************************/
+         sv = hv_fetch(plperl_query_hash, query, strlen(query), 0);
+         if ( sv == NULL)
+             elog(ERROR, "spi_query_prepared: Invalid prepared query passed");
+         if ( *sv == NULL || !SvOK( *sv))
+             elog(ERROR, "spi_query_prepared: panic - plperl_query_hash value corrupted");
+
+         qdesc = INT2PTR( plperl_query_desc *, SvUV(*sv));
+         if ( qdesc == NULL)
+             elog(ERROR, "spi_query_prepared: panic - plperl_query_hash value vanished");
+
+         if ( qdesc-> nargs != argc)
+             elog(ERROR, "spi_query_prepared: expected %d argument(s), %d passed",
+                 qdesc-> nargs, argc);
+
+         /************************************************************
+          * Set up arguments
+          ************************************************************/
+         if ( argc > 0) {
+             nulls = (char *)palloc( argc);
+             argvalues = (Datum *) palloc(argc * sizeof(Datum));
+             if ( nulls == NULL || argvalues == NULL)
+                 elog(ERROR, "spi_query_prepared: not enough memory");
+         } else {
+             nulls = NULL;
+             argvalues = NULL;
+         }
+
+         for ( i = 0; i < argc; i++) {
+             if ( SvTYPE( argv[i]) != SVt_NULL) {
+                 argvalues[i] =
+                     FunctionCall3( &qdesc->arginfuncs[i],
+                           CStringGetDatum( SvPV( argv[i], PL_na)),
+                           ObjectIdGetDatum( qdesc->argtypioparams[i]),
+                           Int32GetDatum(-1)
+                     );
+                 nulls[i] = ' ';
+             } else {
+                 argvalues[i] = (Datum) 0;
+                 nulls[i] = 'n';
+             }
+         }
+
+         /************************************************************
+          * go
+          ************************************************************/
+         portal = SPI_cursor_open(NULL, qdesc-> plan, argvalues, nulls,
+                             plperl_current_prodesc->fn_readonly);
+         if ( argc > 0) {
+             pfree( argvalues);
+             pfree( nulls);
+         }
+         if ( portal == NULL)
+             elog(ERROR, "SPI_cursor_open() failed:%s",
+                 SPI_result_code_string(SPI_result));
+
+         cursor = newSVpv(portal->name, 0);
+
+         /* Commit the inner transaction, return to outer xact context */
+         ReleaseCurrentSubTransaction();
+         MemoryContextSwitchTo(oldcontext);
+         CurrentResourceOwner = oldowner;
+         /*
+          * AtEOSubXact_SPI() should not have popped any SPI context,
+          * but just in case it did, make sure we remain connected.
+          */
+         SPI_restore_connection();
+     }
+     PG_CATCH();
+     {
+         ErrorData  *edata;
+
+         /* Save error info */
+         MemoryContextSwitchTo(oldcontext);
+         edata = CopyErrorData();
+         FlushErrorState();
+
+         /* Abort the inner transaction */
+         RollbackAndReleaseCurrentSubTransaction();
+         MemoryContextSwitchTo(oldcontext);
+         CurrentResourceOwner = oldowner;
+
+         /*
+          * If AtEOSubXact_SPI() popped any SPI context of the subxact,
+          * it will have left us in a disconnected state.  We need this
+          * hack to return to connected state.
+          */
+         SPI_restore_connection();
+
+         /* Punt the error to Perl */
+         croak("%s", edata->message);
+
+         /* Can't get here, but keep compiler quiet */
+         return NULL;
+     }
+     PG_END_TRY();
+
+     return cursor;
+ }
+
+ void
+ plperl_spi_freeplan(char *query)
+ {
+     SV ** sv;
+     void * plan;
+     plperl_query_desc *qdesc;
+
+     sv = hv_fetch(plperl_query_hash, query, strlen(query), 0);
+     if ( sv == NULL)
+         elog(ERROR, "spi_exec_freeplan: Invalid prepared query passed");
+     if ( *sv == NULL || !SvOK( *sv))
+         elog(ERROR, "spi_exec_freeplan: panic - plperl_query_hash value corrupted");
+
+     qdesc = INT2PTR( plperl_query_desc *, SvUV(*sv));
+     if ( qdesc == NULL)
+         elog(ERROR, "spi_exec_freeplan: panic - plperl_query_hash value vanished");
+
+     /*
+     *    free all memory before SPI_freeplan, so if it dies, nothing will be left over
+     */
+     hv_delete(plperl_query_hash, query, strlen(query), G_DISCARD);
+     plan = qdesc-> plan;
+     free(qdesc-> argtypes);
+     free(qdesc-> arginfuncs);
+     free(qdesc-> argtypioparams);
+     free(qdesc);
+
+     SPI_freeplan( plan);
  }
diff -rcN plperl.cvs/spi_internal.h plperl.0/spi_internal.h
*** plperl.cvs/spi_internal.h    Thu Oct 27 12:34:30 2005
--- plperl.0/spi_internal.h    Thu Dec  8 10:35:57 2005
***************
*** 20,22 ****
--- 20,27 ----
  void        plperl_return_next(SV *);
  SV           *plperl_spi_query(char *);
  SV           *plperl_spi_fetchrow(char *);
+ SV *plperl_spi_prepare(char *, int, SV **);
+ HV *plperl_spi_exec_prepared(char *, HV *, int, SV **);
+ SV *plperl_spi_query_prepared(char *, int, SV **);
+ void plperl_spi_freeplan(char *);
+ void plperl_spi_cursor_close(char *);
diff -rcN plperl.cvs/sql/plperl.sql plperl.0/sql/plperl.sql
*** plperl.cvs/sql/plperl.sql    Tue Nov 22 11:48:57 2005
--- plperl.0/sql/plperl.sql    Thu Dec  8 10:36:00 2005
***************
*** 261,266 ****
--- 261,276 ----
  $$ LANGUAGE plperl;
  SELECT * from perl_spi_func();

+ --
+ -- Test spi_fetchrow abort
+ --
+ CREATE OR REPLACE FUNCTION perl_spi_func2() RETURNS INTEGER AS $$
+ my $x = spi_query("select 1 as a union select 2 as a");
+ spi_cursor_close( $x);
+ return 0;
+ $$ LANGUAGE plperl;
+ SELECT * from perl_spi_func2();
+

  ---
  --- Test recursion via SPI
***************
*** 300,303 ****
      return [['a"b',undef,'c,d'],['e\\f',undef,'g']];
  $$;

! SELECT array_of_text();
--- 310,339 ----
      return [['a"b',undef,'c,d'],['e\\f',undef,'g']];
  $$;

! SELECT array_of_text();
!
! --
! -- Test spi_prepare/spi_exec_prepared/spi_freeplan
! --
! CREATE OR REPLACE FUNCTION perl_spi_prepared(INTEGER) RETURNS INTEGER AS $$
!    my $x = spi_prepare('select $1 AS a', 'INT4');
!    my $q = spi_exec_prepared( $x, $_[0] + 1);
!    spi_freeplan($x);
! return $q->{rows}->[0]->{a};
! $$ LANGUAGE plperl;
! SELECT * from perl_spi_prepared(42);
!
! --
! -- Test spi_prepare/spi_query_prepared/spi_freeplan
! --
! CREATE OR REPLACE FUNCTION perl_spi_prepared_set(INTEGER, INTEGER) RETURNS SETOF INTEGER AS $$
!   my $x = spi_prepare('SELECT $1 AS a union select $2 as a', 'INT4', 'INT4');
!   my $q = spi_query_prepared( $x, 1+$_[0], 2+$_[1]);
!   while (defined (my $y = spi_fetchrow($q))) {
!       return_next $y->{a};
!   }
!   spi_freeplan($x);
!   return;
! $$ LANGUAGE plperl;
! SELECT * from perl_spi_prepared_set(1,2);
!