*** contrib/tablefunc/tablefunc.c.orig	Sun Aug 29 09:54:44 2004
--- contrib/tablefunc/tablefunc.c	Sun Sep 12 16:52:39 2004
***************
*** 386,392 ****
  			elog(ERROR, "crosstab: SPI_connect returned %d", ret);
  
  		/* Retrieve the desired rows */
! 		ret = SPI_exec(sql, 0);
  		proc = SPI_processed;
  
  		/* Check for qualifying tuples */
--- 386,392 ----
  			elog(ERROR, "crosstab: SPI_connect returned %d", ret);
  
  		/* Retrieve the desired rows */
! 		ret = SPI_execute(sql, true, 0);
  		proc = SPI_processed;
  
  		/* Check for qualifying tuples */
***************
*** 777,783 ****
  		elog(ERROR, "load_categories_hash: SPI_connect returned %d", ret);
  
  	/* Retrieve the category name rows */
! 	ret = SPI_exec(cats_sql, 0);
  	num_categories = proc = SPI_processed;
  
  	/* Check for qualifying tuples */
--- 777,783 ----
  		elog(ERROR, "load_categories_hash: SPI_connect returned %d", ret);
  
  	/* Retrieve the category name rows */
! 	ret = SPI_execute(cats_sql, true, 0);
  	num_categories = proc = SPI_processed;
  
  	/* Check for qualifying tuples */
***************
*** 855,861 ****
  		elog(ERROR, "get_crosstab_tuplestore: SPI_connect returned %d", ret);
  
  	/* Now retrieve the crosstab source rows */
! 	ret = SPI_exec(sql, 0);
  	proc = SPI_processed;
  
  	/* Check for qualifying tuples */
--- 855,861 ----
  		elog(ERROR, "get_crosstab_tuplestore: SPI_connect returned %d", ret);
  
  	/* Now retrieve the crosstab source rows */
! 	ret = SPI_execute(sql, true, 0);
  	proc = SPI_processed;
  
  	/* Check for qualifying tuples */
***************
*** 1376,1382 ****
  	}
  
  	/* Retrieve the desired rows */
! 	ret = SPI_exec(sql->data, 0);
  	proc = SPI_processed;
  
  	/* Check for qualifying tuples */
--- 1376,1382 ----
  	}
  
  	/* Retrieve the desired rows */
! 	ret = SPI_execute(sql->data, true, 0);
  	proc = SPI_processed;
  
  	/* Check for qualifying tuples */
*** doc/src/sgml/plpython.sgml.orig	Thu May 20 15:22:46 2004
--- doc/src/sgml/plpython.sgml	Sun Sep 12 16:50:25 2004
***************
*** 175,181 ****
     row number and column name.  It has these additional methods:
     <function>nrows</function> which returns the number of rows
     returned by the query, and <function>status</function> which is the
!    <function>SPI_exec()</function> return value.  The result object
     can be modified.
    </para>
  
--- 175,181 ----
     row number and column name.  It has these additional methods:
     <function>nrows</function> which returns the number of rows
     returned by the query, and <function>status</function> which is the
!    <function>SPI_execute()</function> return value.  The result object
     can be modified.
    </para>
  
*** doc/src/sgml/spi.sgml.orig	Thu Apr  1 16:28:43 2004
--- doc/src/sgml/spi.sgml	Sun Sep 12 16:50:25 2004
***************
*** 206,212 ****
  
   <refnamediv>
    <refname>SPI_push</refname>
!   <refpurpose>pushes SPI stack to allow recursive SPI calls</refpurpose>
   </refnamediv>
  
   <indexterm><primary>SPI_push</primary></indexterm>
--- 206,212 ----
  
   <refnamediv>
    <refname>SPI_push</refname>
!   <refpurpose>pushes SPI stack to allow recursive SPI usage</refpurpose>
   </refnamediv>
  
   <indexterm><primary>SPI_push</primary></indexterm>
***************
*** 221,228 ****
    <title>Description</title>
  
    <para>
!    <function>SPI_push</function> pushes a new environment on to the 
!    SPI call stack, allowing recursive calls to use a new environment.
    </para>
   </refsect1>
  
--- 221,244 ----
    <title>Description</title>
  
    <para>
!    <function>SPI_push</function> should be called before executing another
!    procedure that might itself wish to use SPI.
!    After <function>SPI_push</function>, SPI is no longer in a
!    <quote>connected</> state, and SPI function calls will be rejected unless
!    a fresh <function>SPI_connect</function> is done.  This ensures a clean
!    separation between your procedure's SPI state and that of another procedure
!    you call.  After the other procedure returns, call
!    <function>SPI_pop</function> to restore access to your own SPI state.
!   </para>
! 
!   <para>
!    Note that <function>SPI_execute</function> and related functions
!    automatically do the equivalent of <function>SPI_push</function> before
!    passing control back to the SQL execution engine, so it is not necessary
!    for you to worry about this when using those functions.
!    Only when you are directly calling arbitrary code that might contain
!    <function>SPI_connect</function> do you need to issue
!    <function>SPI_push</function> and <function>SPI_pop</function>.
    </para>
   </refsect1>
  
***************
*** 237,243 ****
  
   <refnamediv>
    <refname>SPI_pop</refname>
!   <refpurpose>pops SPI stack to allow recursive SPI calls</refpurpose>
   </refnamediv>
  
   <indexterm><primary>SPI_pop</primary></indexterm>
--- 253,259 ----
  
   <refnamediv>
    <refname>SPI_pop</refname>
!   <refpurpose>pops SPI stack to return from recursive SPI usage</refpurpose>
   </refnamediv>
  
   <indexterm><primary>SPI_pop</primary></indexterm>
***************
*** 253,259 ****
  
    <para>
     <function>SPI_pop</function> pops the previous environment from the 
!    SPI call stack.  For use when returning from recursive SPI calls.
    </para>
   </refsect1>
  
--- 269,275 ----
  
    <para>
     <function>SPI_pop</function> pops the previous environment from the 
!    SPI call stack.  See <function>SPI_push</function>.
    </para>
   </refsect1>
  
***************
*** 261,281 ****
  
  <!-- *********************************************** -->
  
! <refentry id="spi-spi-exec">
   <refmeta>
!   <refentrytitle>SPI_exec</refentrytitle>
   </refmeta>
  
   <refnamediv>
!   <refname>SPI_exec</refname>
    <refpurpose>execute a command</refpurpose>
   </refnamediv>
  
!  <indexterm><primary>SPI_exec</primary></indexterm>
  
   <refsynopsisdiv>
  <synopsis>
! int SPI_exec(const char * <parameter>command</parameter>, int <parameter>count</parameter>)
  </synopsis>
   </refsynopsisdiv>
  
--- 277,297 ----
  
  <!-- *********************************************** -->
  
! <refentry id="spi-spi-execute">
   <refmeta>
!   <refentrytitle>SPI_execute</refentrytitle>
   </refmeta>
  
   <refnamediv>
!   <refname>SPI_execute</refname>
    <refpurpose>execute a command</refpurpose>
   </refnamediv>
  
!  <indexterm><primary>SPI_execute</primary></indexterm>
  
   <refsynopsisdiv>
  <synopsis>
! int SPI_execute(const char * <parameter>command</parameter>, bool <parameter>read_only</parameter>, int <parameter>count</parameter>)
  </synopsis>
   </refsynopsisdiv>
  
***************
*** 283,290 ****
    <title>Description</title>
  
    <para>
!    <function>SPI_exec</function> executes the specified SQL command
!    for <parameter>count</parameter> rows.
    </para>
  
    <para>
--- 299,308 ----
    <title>Description</title>
  
    <para>
!    <function>SPI_execute</function> executes the specified SQL command
!    for <parameter>count</parameter> rows.  If <parameter>read_only</parameter>
!    is <literal>true</>, the command must be read-only, and execution overhead
!    is somewhat reduced.
    </para>
  
    <para>
***************
*** 295,317 ****
     will be executed is restricted (much like a
     <literal>LIMIT</literal> clause). For example,
  <programlisting>
! SPI_exec("INSERT INTO tab SELECT * FROM tab", 5);
  </programlisting>
     will allow at most 5 rows to be inserted into the table.
    </para>
  
    <para>
     You may pass multiple commands in one string, and the command may
!    be rewritten by rules. <function>SPI_exec</function> returns the
     result for the command executed last.
    </para>
  
    <para>
     The actual number of rows for which the (last) command was executed
     is returned in the global variable <varname>SPI_processed</varname>
     (unless the return value of the function is
     <symbol>SPI_OK_UTILITY</symbol>).  If the return value of the
!    function is <symbol>SPI_OK_SELECT</symbol> then you may the use
     global pointer <literal>SPITupleTable *SPI_tuptable</literal> to
     access the result rows.
    </para>
--- 313,366 ----
     will be executed is restricted (much like a
     <literal>LIMIT</literal> clause). For example,
  <programlisting>
! SPI_execute("INSERT INTO tab SELECT * FROM tab", false, 5);
  </programlisting>
     will allow at most 5 rows to be inserted into the table.
    </para>
  
    <para>
     You may pass multiple commands in one string, and the command may
!    be rewritten by rules. <function>SPI_execute</function> returns the
     result for the command executed last.
    </para>
  
    <para>
+    When <parameter>read_only</parameter> is <literal>false</>,
+    <function>SPI_execute</function> computes a new <quote>snapshot</>
+    before executing each command in the string, and increments the command
+    counter afterwards.  The snapshot does not actually change if the current
+    transaction isolation level is <literal>SERIALIZABLE</>, but in
+    <literal>READ COMMITTED</> mode the snapshot update allows each command to
+    see the results of newly committed transactions from other sessions.
+    This is essential for consistent behavior when the commands are modifying
+    the database.
+   </para>
+ 
+   <para>
+    When <parameter>read_only</parameter> is <literal>true</>,
+    <function>SPI_execute</function> does not update either the snapshot
+    or the command counter, and it allows only plain <command>SELECT</>
+    commands to appear in the command string.  The commands are executed
+    using the snapshot previously established for the surrounding query.
+    This execution mode is somewhat faster than the read/write mode due
+    to eliminating per-command overhead.  It also allows genuinely
+    <firstterm>stable</> functions to be built: since successive executions
+    will all use the same snapshot, there will be no change in the results.
+   </para>
+ 
+   <para>
+    It is generally unwise to mix read-only and read-write commands within
+    a single function using SPI; that could result in very confusing behavior,
+    since the read-only queries would not see the results of any database
+    updates done by the read-write queries.
+   </para>
+ 
+   <para>
     The actual number of rows for which the (last) command was executed
     is returned in the global variable <varname>SPI_processed</varname>
     (unless the return value of the function is
     <symbol>SPI_OK_UTILITY</symbol>).  If the return value of the
!    function is <symbol>SPI_OK_SELECT</symbol> then you may use the
     global pointer <literal>SPITupleTable *SPI_tuptable</literal> to
     access the result rows.
    </para>
***************
*** 330,336 ****
  } SPITupleTable;
  </programlisting>
     <structfield>vals</> is an array of pointers to rows.  (The number
!    of valid entries is given by <varname>SPI_processed</varname>).
     <structfield>tupdesc</> is a row descriptor which you may pass to
     SPI functions dealing with rows.  <structfield>tuptabcxt</>,
     <structfield>alloced</>, and <structfield>free</> are internal
--- 379,385 ----
  } SPITupleTable;
  </programlisting>
     <structfield>vals</> is an array of pointers to rows.  (The number
!    of valid entries is given by <varname>SPI_processed</varname>.)
     <structfield>tupdesc</> is a row descriptor which you may pass to
     SPI functions dealing with rows.  <structfield>tuptabcxt</>,
     <structfield>alloced</>, and <structfield>free</> are internal
***************
*** 359,364 ****
--- 408,422 ----
     </varlistentry>
  
     <varlistentry>
+     <term><literal>bool <parameter>read_only</parameter></literal></term>
+     <listitem>
+      <para>
+       <literal>true</> for read-only execution
+      </para>
+     </listitem>
+    </varlistentry>
+ 
+    <varlistentry>
      <term><literal>int <parameter>count</parameter></literal></term>
      <listitem>
       <para>
***************
*** 504,517 ****
    <title>Notes</title>
  
    <para>
!    The functions <function>SPI_exec</function>,
!    <function>SPI_execp</function>, and
!    <function>SPI_prepare</function> change both
     <varname>SPI_processed</varname> and
     <varname>SPI_tuptable</varname> (just the pointer, not the contents
     of the structure).  Save these two global variables into local
!    procedure variables if you need to access the result of
!    <function>SPI_exec</function> or <function>SPI_execp</function>
     across later calls.
    </para>
   </refsect1>
--- 562,576 ----
    <title>Notes</title>
  
    <para>
!    The functions <function>SPI_execute</function>,
!    <function>SPI_exec</function>,
!    <function>SPI_execute_plan</function>, and
!    <function>SPI_execp</function> change both
     <varname>SPI_processed</varname> and
     <varname>SPI_tuptable</varname> (just the pointer, not the contents
     of the structure).  Save these two global variables into local
!    procedure variables if you need to access the result table of
!    <function>SPI_execute</function> or a related function
     across later calls.
    </para>
   </refsect1>
***************
*** 519,524 ****
--- 578,647 ----
  
  <!-- *********************************************** -->
  
+ <refentry id="spi-spi-exec">
+  <refmeta>
+   <refentrytitle>SPI_exec</refentrytitle>
+  </refmeta>
+ 
+  <refnamediv>
+   <refname>SPI_exec</refname>
+   <refpurpose>execute a read/write command</refpurpose>
+  </refnamediv>
+ 
+  <indexterm><primary>SPI_exec</primary></indexterm>
+ 
+  <refsynopsisdiv>
+ <synopsis>
+ int SPI_exec(const char * <parameter>command</parameter>, int <parameter>count</parameter>)
+ </synopsis>
+  </refsynopsisdiv>
+ 
+  <refsect1>
+   <title>Description</title>
+ 
+   <para>
+    <function>SPI_exec</function> is the same as
+    <function>SPI_execute</function>, with the latter's
+    <parameter>read_only</parameter> parameter always taken as
+    <literal>false</>.
+   </para>
+  </refsect1>
+ 
+  <refsect1>
+   <title>Arguments</title>
+ 
+   <variablelist>
+    <varlistentry>
+     <term><literal>const char * <parameter>command</parameter></literal></term>
+     <listitem>
+      <para>
+       string containing command to execute
+      </para>
+     </listitem>
+    </varlistentry>
+ 
+    <varlistentry>
+     <term><literal>int <parameter>count</parameter></literal></term>
+     <listitem>
+      <para>
+       maximum number of rows to process or return
+      </para>
+     </listitem>
+    </varlistentry>
+   </variablelist>
+  </refsect1>
+ 
+  <refsect1>
+   <title>Return Value</title>
+ 
+   <para>
+    See <function>SPI_execute</function>.
+   </para>
+  </refsect1>
+ </refentry>
+ 
+ <!-- *********************************************** -->
+ 
  <refentry id="spi-spi-prepare">
   <refmeta>
    <refentrytitle>SPI_prepare</refentrytitle>
***************
*** 551,564 ****
     may be advantageous to perform the planning only once.
     <function>SPI_prepare</function> converts a command string into an
     execution plan that can be executed repeatedly using
!    <function>SPI_execp</function>.
    </para>
  
    <para>
     A prepared command can be generalized by writing parameters
     (<literal>$1</>, <literal>$2</>, etc.) in place of what would be
     constants in a normal command.  The actual values of the parameters
!    are then specified when <function>SPI_execp</function> is called.
     This allows the prepared command to be used over a wider range of
     situations than would be possible without parameters.
    </para>
--- 674,687 ----
     may be advantageous to perform the planning only once.
     <function>SPI_prepare</function> converts a command string into an
     execution plan that can be executed repeatedly using
!    <function>SPI_execute_plan</function>.
    </para>
  
    <para>
     A prepared command can be generalized by writing parameters
     (<literal>$1</>, <literal>$2</>, etc.) in place of what would be
     constants in a normal command.  The actual values of the parameters
!    are then specified when <function>SPI_execute_plan</function> is called.
     This allows the prepared command to be used over a wider range of
     situations than would be possible without parameters.
    </para>
***************
*** 610,619 ****
    <title>Return Value</title>
  
    <para>
!    <function>SPI_prepare</function> returns non-null pointer to an
!    execution plan.  On error, <symbol>NULL</symbol> will be returned.
!    In both cases, <varname>SPI_result</varname> will be set analogous
!    to the value returned by <function>SPI_exec</function>, except that
     it is set to <symbol>SPI_ERROR_ARGUMENT</symbol> if
     <parameter>command</parameter> is <symbol>NULL</symbol>, or if
     <parameter>nargs</> is less than 0, or if <parameter>nargs</> is
--- 733,742 ----
    <title>Return Value</title>
  
    <para>
!    <function>SPI_prepare</function> returns a non-null pointer to an
!    execution plan.  On error, <symbol>NULL</symbol> will be returned,
!    and <varname>SPI_result</varname> will be set to one of the same
!    error codes used by <function>SPI_execute</function>, except that
     it is set to <symbol>SPI_ERROR_ARGUMENT</symbol> if
     <parameter>command</parameter> is <symbol>NULL</symbol>, or if
     <parameter>nargs</> is less than 0, or if <parameter>nargs</> is
***************
*** 697,704 ****
  
   <refnamediv>
    <refname>SPI_getargtypeid</refname>
!   <refpurpose>returns the expected typeid for the specified argument when
!   executing a plan prepared by <function>SPI_prepare</function></refpurpose>
   </refnamediv>
  
   <indexterm><primary>SPI_getargtypeid</primary></indexterm>
--- 820,827 ----
  
   <refnamediv>
    <refname>SPI_getargtypeid</refname>
!   <refpurpose>returns the expected typeid for the specified argument of
!   a plan prepared by <function>SPI_prepare</function></refpurpose>
   </refnamediv>
  
   <indexterm><primary>SPI_getargtypeid</primary></indexterm>
***************
*** 765,772 ****
   <refnamediv>
    <refname>SPI_is_cursor_plan</refname>
    <refpurpose>returns <symbol>true</symbol> if a plan
!   prepared by <function>SPI_prepare</function> can be passed
!   as an argument to <function>SPI_cursor_open</function></refpurpose>
   </refnamediv>
  
   <indexterm><primary>SPI_is_cursor_plan</primary></indexterm>
--- 888,895 ----
   <refnamediv>
    <refname>SPI_is_cursor_plan</refname>
    <refpurpose>returns <symbol>true</symbol> if a plan
!   prepared by <function>SPI_prepare</function> can be used with
!   <function>SPI_cursor_open</function></refpurpose>
   </refnamediv>
  
   <indexterm><primary>SPI_is_cursor_plan</primary></indexterm>
***************
*** 784,790 ****
     <function>SPI_is_cursor_plan</function> returns <symbol>true</symbol>
     if a plan prepared by <function>SPI_prepare</function> can be passed
     as an argument to <function>SPI_cursor_open</function> and <symbol>
!    false</symbol> if that is not the case. The criteria is that the
     <parameter>plan</parameter> represents one single command and that this
     command is a <command>SELECT</command> without an <command>INTO</command>
     clause.
--- 907,913 ----
     <function>SPI_is_cursor_plan</function> returns <symbol>true</symbol>
     if a plan prepared by <function>SPI_prepare</function> can be passed
     as an argument to <function>SPI_cursor_open</function> and <symbol>
!    false</symbol> if that is not the case. The criteria are that the
     <parameter>plan</parameter> represents one single command and that this
     command is a <command>SELECT</command> without an <command>INTO</command>
     clause.
***************
*** 819,839 ****
  
  <!-- *********************************************** -->
  
! <refentry id="spi-spi-execp">
   <refmeta>
!   <refentrytitle>SPI_execp</refentrytitle>
   </refmeta>
  
   <refnamediv>
!   <refname>SPI_execp</refname>
    <refpurpose>executes a plan prepared by <function>SPI_prepare</function></refpurpose>
   </refnamediv>
  
!  <indexterm><primary>SPI_execp</primary></indexterm>
  
   <refsynopsisdiv>
  <synopsis>
! int SPI_execp(void * <parameter>plan</parameter>, Datum * <parameter>values</parameter>, const char * <parameter>nulls</parameter>, int <parameter>count</parameter>)
  </synopsis>
   </refsynopsisdiv>
  
--- 942,962 ----
  
  <!-- *********************************************** -->
  
! <refentry id="spi-spi-execute-plan">
   <refmeta>
!   <refentrytitle>SPI_execute_plan</refentrytitle>
   </refmeta>
  
   <refnamediv>
!   <refname>SPI_execute_plan</refname>
    <refpurpose>executes a plan prepared by <function>SPI_prepare</function></refpurpose>
   </refnamediv>
  
!  <indexterm><primary>SPI_execute_plan</primary></indexterm>
  
   <refsynopsisdiv>
  <synopsis>
! int SPI_execute_plan(void * <parameter>plan</parameter>, Datum * <parameter>values</parameter>, const char * <parameter>nulls</parameter>, bool <parameter>read_only</parameter>, int <parameter>count</parameter>)
  </synopsis>
   </refsynopsisdiv>
  
***************
*** 841,849 ****
    <title>Description</title>
  
    <para>
!    <function>SPI_execp</function> executes a plan prepared by
!    <function>SPI_prepare</function>.  <parameter>tcount</parameter>
!    has the same interpretation as in <function>SPI_exec</function>.
    </para>
   </refsect1>
  
--- 964,973 ----
    <title>Description</title>
  
    <para>
!    <function>SPI_execute_plan</function> executes a plan prepared by
!    <function>SPI_prepare</function>.  <parameter>read_only</parameter> and
!    <parameter>count</parameter> have the same interpretation as in
!    <function>SPI_execute</function>.
    </para>
   </refsect1>
  
***************
*** 864,870 ****
      <term><literal>Datum *<parameter>values</parameter></literal></term>
      <listitem>
       <para>
!       actual parameter values
       </para>
      </listitem>
     </varlistentry>
--- 988,995 ----
      <term><literal>Datum *<parameter>values</parameter></literal></term>
      <listitem>
       <para>
!       An array of actual parameter values.  Must have same length as the
!       plan's number of arguments.
       </para>
      </listitem>
     </varlistentry>
***************
*** 873,879 ****
      <term><literal>const char * <parameter>nulls</parameter></literal></term>
      <listitem>
       <para>
!       An array describing which parameters are null.
        <literal>n</literal> indicates a null value (entry in
        <parameter>values</> will be ignored); a space indicates a
        nonnull value (entry in <parameter>values</> is valid).
--- 998,1005 ----
      <term><literal>const char * <parameter>nulls</parameter></literal></term>
      <listitem>
       <para>
!       An array describing which parameters are null.  Must have same length as
!       the plan's number of arguments.
        <literal>n</literal> indicates a null value (entry in
        <parameter>values</> will be ignored); a space indicates a
        nonnull value (entry in <parameter>values</> is valid).
***************
*** 881,897 ****
  
       <para>
        If <parameter>nulls</parameter> is <symbol>NULL</symbol> then
!       <function>SPI_execp</function> assumes that no parameters are
        null.
       </para>
      </listitem>
     </varlistentry>
  
     <varlistentry>
      <term><literal>int <parameter>count</parameter></literal></term>
      <listitem>
       <para>
!       number of row for which plan is to be executed
       </para>
      </listitem>
     </varlistentry>
--- 1007,1032 ----
  
       <para>
        If <parameter>nulls</parameter> is <symbol>NULL</symbol> then
!       <function>SPI_execute_plan</function> assumes that no parameters are
        null.
       </para>
      </listitem>
     </varlistentry>
  
     <varlistentry>
+     <term><literal>bool <parameter>read_only</parameter></literal></term>
+     <listitem>
+      <para>
+       <literal>true</> for read-only execution
+      </para>
+     </listitem>
+    </varlistentry>
+ 
+    <varlistentry>
      <term><literal>int <parameter>count</parameter></literal></term>
      <listitem>
       <para>
!       maximum number of rows to process or return
       </para>
      </listitem>
     </varlistentry>
***************
*** 902,909 ****
    <title>Return Value</title>
  
    <para>
!    The return value is the same as for <function>SPI_exec</function>
!    or one of the following:
  
     <variablelist>
      <varlistentry>
--- 1037,1044 ----
    <title>Return Value</title>
  
    <para>
!    The return value is the same as for <function>SPI_execute</function>,
!    with the following additional possible error (negative) results:
  
     <variablelist>
      <varlistentry>
***************
*** 931,937 ****
    <para>
     <varname>SPI_processed</varname> and
     <varname>SPI_tuptable</varname> are set as in
!    <function>SPI_exec</function> if successful.
    </para>
   </refsect1>
  
--- 1066,1072 ----
    <para>
     <varname>SPI_processed</varname> and
     <varname>SPI_tuptable</varname> are set as in
!    <function>SPI_execute</function> if successful.
    </para>
   </refsect1>
  
***************
*** 941,947 ****
    <para>
     If one of the objects (a table, function, etc.) referenced by the
     prepared plan is dropped during the session then the result of
!    <function>SPI_execp</function> for this plan will be unpredictable.
    </para>
   </refsect1>
  </refentry>
--- 1076,1181 ----
    <para>
     If one of the objects (a table, function, etc.) referenced by the
     prepared plan is dropped during the session then the result of
!    <function>SPI_execute_plan</function> for this plan will be unpredictable.
!   </para>
!  </refsect1>
! </refentry>
! 
! <!-- *********************************************** -->
! 
! <refentry id="spi-spi-execp">
!  <refmeta>
!   <refentrytitle>SPI_execp</refentrytitle>
!  </refmeta>
! 
!  <refnamediv>
!   <refname>SPI_execp</refname>
!   <refpurpose>executes a plan in read/write mode</refpurpose>
!  </refnamediv>
! 
!  <indexterm><primary>SPI_execp</primary></indexterm>
! 
!  <refsynopsisdiv>
! <synopsis>
! int SPI_execp(void * <parameter>plan</parameter>, Datum * <parameter>values</parameter>, const char * <parameter>nulls</parameter>, int <parameter>count</parameter>)
! </synopsis>
!  </refsynopsisdiv>
! 
!  <refsect1>
!   <title>Description</title>
! 
!   <para>
!    <function>SPI_execp</function> is the same as
!    <function>SPI_execute_plan</function>, with the latter's
!    <parameter>read_only</parameter> parameter always taken as
!    <literal>false</>.
!   </para>
!  </refsect1>
! 
!  <refsect1>
!   <title>Arguments</title>
! 
!   <variablelist>
!    <varlistentry>
!     <term><literal>void * <parameter>plan</parameter></literal></term>
!     <listitem>
!      <para>
!       execution plan (returned by <function>SPI_prepare</function>)
!      </para>
!     </listitem>
!    </varlistentry>
! 
!    <varlistentry>
!     <term><literal>Datum *<parameter>values</parameter></literal></term>
!     <listitem>
!      <para>
!       An array of actual parameter values.  Must have same length as the
!       plan's number of arguments.
!      </para>
!     </listitem>
!    </varlistentry>
! 
!    <varlistentry>
!     <term><literal>const char * <parameter>nulls</parameter></literal></term>
!     <listitem>
!      <para>
!       An array describing which parameters are null.  Must have same length as
!       the plan's number of arguments.
!       <literal>n</literal> indicates a null value (entry in
!       <parameter>values</> will be ignored); a space indicates a
!       nonnull value (entry in <parameter>values</> is valid).
!      </para>
! 
!      <para>
!       If <parameter>nulls</parameter> is <symbol>NULL</symbol> then
!       <function>SPI_execp</function> assumes that no parameters are
!       null.
!      </para>
!     </listitem>
!    </varlistentry>
! 
!    <varlistentry>
!     <term><literal>int <parameter>count</parameter></literal></term>
!     <listitem>
!      <para>
!       maximum number of rows to process or return
!      </para>
!     </listitem>
!    </varlistentry>
!   </variablelist>
!  </refsect1>
! 
!  <refsect1>
!   <title>Return Value</title>
! 
!   <para>
!    See <function>SPI_execute_plan</function>.
!   </para>
! 
!   <para>
!    <varname>SPI_processed</varname> and
!    <varname>SPI_tuptable</varname> are set as in
!    <function>SPI_execute</function> if successful.
    </para>
   </refsect1>
  </refentry>
***************
*** 1168,1174 ****
    <para>
     <varname>SPI_processed</varname> and
     <varname>SPI_tuptable</varname> are set as in
!    <function>SPI_exec</function> if successful.
    </para>
   </refsect1>
  </refentry>
--- 1402,1408 ----
    <para>
     <varname>SPI_processed</varname> and
     <varname>SPI_tuptable</varname> are set as in
!    <function>SPI_execute</function> if successful.
    </para>
   </refsect1>
  </refentry>
***************
*** 1320,1326 ****
     your procedure in the current session.  You may save the pointer
     returned in a local variable.  Always check if this pointer is
     <symbol>NULL</symbol> or not either when preparing a plan or using
!    an already prepared plan in <function>SPI_execp</function>.
    </para>
   </refsect1>
  
--- 1554,1560 ----
     your procedure in the current session.  You may save the pointer
     returned in a local variable.  Always check if this pointer is
     <symbol>NULL</symbol> or not either when preparing a plan or using
!    an already prepared plan in <function>SPI_execute_plan</function>.
    </para>
   </refsect1>
  
***************
*** 1374,1380 ****
    <para>
     If one of the objects (a table, function, etc.) referenced by the
     prepared plan is dropped during the session then the results of
!    <function>SPI_execp</function> for this plan will be unpredictable.
    </para>
   </refsect1>
  </refentry>
--- 1608,1614 ----
    <para>
     If one of the objects (a table, function, etc.) referenced by the
     prepared plan is dropped during the session then the results of
!    <function>SPI_execute_plan</function> for this plan will be unpredictable.
    </para>
   </refsect1>
  </refentry>
***************
*** 1386,1392 ****
  
   <para>
    The functions described here provide an interface for extracting
!   information from result sets returned by <function>SPI_exec</> and
    other SPI functions.
   </para>
  
--- 1620,1626 ----
  
   <para>
    The functions described here provide an interface for extracting
!   information from result sets returned by <function>SPI_execute</> and
    other SPI functions.
   </para>
  
***************
*** 2360,2366 ****
      <term><literal>const char * <parameter>Nulls</parameter></literal></term>
      <listitem>
       <para>
!       which new values are null, if any (see <function>SPI_execp</function> for the format)
       </para>
      </listitem>
     </varlistentry>
--- 2594,2601 ----
      <term><literal>const char * <parameter>Nulls</parameter></literal></term>
      <listitem>
       <para>
!       which new values are null, if any (see
!       <function>SPI_execute_plan</function> for the format)
       </para>
      </listitem>
     </varlistentry>
***************
*** 2466,2472 ****
  
   <refnamediv>
    <refname>SPI_freetuptable</refname>
!   <refpurpose>free a row set created by <function>SPI_exec</> or a similar function</refpurpose>
   </refnamediv>
  
   <indexterm><primary>SPI_freetuptable</primary></indexterm>
--- 2701,2708 ----
  
   <refnamediv>
    <refname>SPI_freetuptable</refname>
!   <refpurpose>free a row set created by <function>SPI_execute</> or a similar
!   function</refpurpose>
   </refnamediv>
  
   <indexterm><primary>SPI_freetuptable</primary></indexterm>
***************
*** 2483,2489 ****
    <para>
     <function>SPI_freetuptable</function> frees a row set created by a
     prior SPI command execution function, such as
!    <function>SPI_exec</>.  Therefore, this function is usually called
     with the global variable <varname>SPI_tupletable</varname> as
     argument.
    </para>
--- 2719,2725 ----
    <para>
     <function>SPI_freetuptable</function> frees a row set created by a
     prior SPI command execution function, such as
!    <function>SPI_execute</>.  Therefore, this function is usually called
     with the global variable <varname>SPI_tupletable</varname> as
     argument.
    </para>
*** src/backend/access/transam/xact.c.orig	Fri Sep 10 14:39:55 2004
--- src/backend/access/transam/xact.c	Sat Sep 11 19:00:37 2004
***************
*** 401,411 ****
  				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
  				 errmsg("cannot have more than 2^32-1 commands in a transaction")));
  
! 	/* Propagate new command ID into query snapshots, if set */
! 	if (QuerySnapshot)
! 		QuerySnapshot->curcid = s->commandId;
  	if (SerializableSnapshot)
  		SerializableSnapshot->curcid = s->commandId;
  
  	/*
  	 * make cache changes visible to me.
--- 401,411 ----
  				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
  				 errmsg("cannot have more than 2^32-1 commands in a transaction")));
  
! 	/* Propagate new command ID into static snapshots, if set */
  	if (SerializableSnapshot)
  		SerializableSnapshot->curcid = s->commandId;
+ 	if (LatestSnapshot)
+ 		LatestSnapshot->curcid = s->commandId;
  
  	/*
  	 * make cache changes visible to me.
*** src/backend/commands/cluster.c.orig	Sun Aug 29 09:54:55 2004
--- src/backend/commands/cluster.c	Sat Sep 11 19:00:26 2004
***************
*** 202,209 ****
  
  			/* Start a new transaction for each relation. */
  			StartTransactionCommand();
! 			SetQuerySnapshot(); /* might be needed for functions in
! 								 * indexes */
  			cluster_rel(rvtc, true);
  			CommitTransactionCommand();
  		}
--- 202,209 ----
  
  			/* Start a new transaction for each relation. */
  			StartTransactionCommand();
! 			/* functions in indexes may want a snapshot set */
! 			ActiveSnapshot = CopySnapshot(GetTransactionSnapshot());
  			cluster_rel(rvtc, true);
  			CommitTransactionCommand();
  		}
*** src/backend/commands/copy.c.orig	Fri Sep 10 14:39:56 2004
--- src/backend/commands/copy.c	Sat Sep 11 19:10:35 2004
***************
*** 1182,1188 ****
  	Oid		   *typioparams;
  	bool	   *isvarlena;
  	char	   *string;
- 	Snapshot	mySnapshot;
  	ListCell   *cur;
  	MemoryContext oldcontext;
  	MemoryContext mycontext;
--- 1182,1187 ----
***************
*** 1260,1268 ****
  									strlen(null_print));
  	}
  
! 	mySnapshot = CopyQuerySnapshot();
! 
! 	scandesc = heap_beginscan(rel, mySnapshot, 0, NULL);
  
  	while ((tuple = heap_getnext(scandesc, ForwardScanDirection)) != NULL)
  	{
--- 1259,1265 ----
  									strlen(null_print));
  	}
  
! 	scandesc = heap_beginscan(rel, ActiveSnapshot, 0, NULL);
  
  	while ((tuple = heap_getnext(scandesc, ForwardScanDirection)) != NULL)
  	{
*** src/backend/commands/explain.c.orig	Fri Sep 10 14:39:56 2004
--- src/backend/commands/explain.c	Sat Sep 11 21:07:54 2004
***************
*** 180,186 ****
  	plan = planner(query, isCursor, cursorOptions, NULL);
  
  	/* Create a QueryDesc requesting no output */
! 	queryDesc = CreateQueryDesc(query, plan, None_Receiver, NULL,
  								stmt->analyze);
  
  	ExplainOnePlan(queryDesc, stmt, tstate);
--- 180,188 ----
  	plan = planner(query, isCursor, cursorOptions, NULL);
  
  	/* Create a QueryDesc requesting no output */
! 	queryDesc = CreateQueryDesc(query, plan,
! 								ActiveSnapshot, InvalidSnapshot,
! 								None_Receiver, NULL,
  								stmt->analyze);
  
  	ExplainOnePlan(queryDesc, stmt, tstate);
***************
*** 212,218 ****
  		AfterTriggerBeginQuery();
  
  	/* call ExecutorStart to prepare the plan for execution */
! 	ExecutorStart(queryDesc, false, !stmt->analyze);
  
  	/* Execute the plan for statistics if asked for */
  	if (stmt->analyze)
--- 214,220 ----
  		AfterTriggerBeginQuery();
  
  	/* call ExecutorStart to prepare the plan for execution */
! 	ExecutorStart(queryDesc, !stmt->analyze);
  
  	/* Execute the plan for statistics if asked for */
  	if (stmt->analyze)
*** src/backend/commands/indexcmds.c.orig	Sun Aug 29 09:54:55 2004
--- src/backend/commands/indexcmds.c	Sat Sep 11 19:00:27 2004
***************
*** 1060,1067 ****
  		Oid			relid = lfirst_oid(l);
  
  		StartTransactionCommand();
! 		SetQuerySnapshot();		/* might be needed for functions in
! 								 * indexes */
  		if (reindex_relation(relid, true))
  			ereport(NOTICE,
  					(errmsg("table \"%s\" was reindexed",
--- 1060,1067 ----
  		Oid			relid = lfirst_oid(l);
  
  		StartTransactionCommand();
! 		/* functions in indexes may want a snapshot set */
! 		ActiveSnapshot = CopySnapshot(GetTransactionSnapshot());
  		if (reindex_relation(relid, true))
  			ereport(NOTICE,
  					(errmsg("table \"%s\" was reindexed",
*** src/backend/commands/portalcmds.c.orig	Fri Sep 10 14:39:56 2004
--- src/backend/commands/portalcmds.c	Sat Sep 11 21:06:19 2004
***************
*** 135,141 ****
  	/*
  	 * Start execution, inserting parameters if any.
  	 */
! 	PortalStart(portal, params);
  
  	Assert(portal->strategy == PORTAL_ONE_SELECT);
  
--- 135,141 ----
  	/*
  	 * Start execution, inserting parameters if any.
  	 */
! 	PortalStart(portal, params, ActiveSnapshot);
  
  	Assert(portal->strategy == PORTAL_ONE_SELECT);
  
*** src/backend/commands/prepare.c.orig	Sun Aug 29 09:54:56 2004
--- src/backend/commands/prepare.c	Sat Sep 11 21:06:58 2004
***************
*** 186,192 ****
  	/*
  	 * Run the portal to completion.
  	 */
! 	PortalStart(portal, paramLI);
  
  	(void) PortalRun(portal, FETCH_ALL, dest, dest, completionTag);
  
--- 186,192 ----
  	/*
  	 * Run the portal to completion.
  	 */
! 	PortalStart(portal, paramLI, ActiveSnapshot);
  
  	(void) PortalRun(portal, FETCH_ALL, dest, dest, completionTag);
  
***************
*** 544,550 ****
  			}
  
  			/* Create a QueryDesc requesting no output */
! 			qdesc = CreateQueryDesc(query, plan, None_Receiver,
  									paramLI, stmt->analyze);
  
  			ExplainOnePlan(qdesc, stmt, tstate);
--- 544,552 ----
  			}
  
  			/* Create a QueryDesc requesting no output */
! 			qdesc = CreateQueryDesc(query, plan,
! 									ActiveSnapshot, InvalidSnapshot,
! 									None_Receiver,
  									paramLI, stmt->analyze);
  
  			ExplainOnePlan(qdesc, stmt, tstate);
*** src/backend/commands/trigger.c.orig	Fri Sep 10 14:39:56 2004
--- src/backend/commands/trigger.c	Sat Sep 11 23:27:51 2004
***************
*** 2094,2099 ****
--- 2094,2108 ----
  	TriggerDesc *trigdesc = NULL;
  	FmgrInfo   *finfo = NULL;
  
+ 	/*
+ 	 * Advance the command counter to be sure that the AFTER triggers
+ 	 * can see the results of the just-completed command.  Notice that
+ 	 * we do not CCI between triggers though; it's expected that any
+ 	 * read/write trigger will do a CCI after its update (at least if
+ 	 * it thinks other triggers should be able to see the update).
+ 	 */
+ 	CommandCounterIncrement();
+ 
  	/* Make a per-tuple memory context for trigger function calls */
  	per_tuple_context =
  		AllocSetContextCreate(CurrentMemoryContext,
*** src/backend/commands/vacuum.c.orig	Sun Aug 29 22:57:30 2004
--- src/backend/commands/vacuum.c	Sat Sep 11 19:00:29 2004
***************
*** 402,409 ****
  				if (use_own_xacts)
  				{
  					StartTransactionCommand();
! 					SetQuerySnapshot(); /* might be needed for functions
! 										 * in indexes */
  				}
  				else
  					old_context = MemoryContextSwitchTo(anl_context);
--- 402,409 ----
  				if (use_own_xacts)
  				{
  					StartTransactionCommand();
! 					/* functions in indexes may want a snapshot set */
! 					ActiveSnapshot = CopySnapshot(GetTransactionSnapshot());
  				}
  				else
  					old_context = MemoryContextSwitchTo(anl_context);
***************
*** 865,872 ****
  
  	/* Begin a transaction for vacuuming this relation */
  	StartTransactionCommand();
! 	SetQuerySnapshot();			/* might be needed for functions in
! 								 * indexes */
  
  	/*
  	 * Tell the cache replacement strategy that vacuum is causing all
--- 865,872 ----
  
  	/* Begin a transaction for vacuuming this relation */
  	StartTransactionCommand();
! 	/* functions in indexes may want a snapshot set */
! 	ActiveSnapshot = CopySnapshot(GetTransactionSnapshot());
  
  	/*
  	 * Tell the cache replacement strategy that vacuum is causing all
*** src/backend/executor/execMain.c.orig	Sat Sep 11 14:28:34 2004
--- src/backend/executor/execMain.c	Sat Sep 11 19:19:08 2004
***************
*** 106,120 ****
   * field of the QueryDesc is filled in to describe the tuples that will be
   * returned, and the internal fields (estate and planstate) are set up.
   *
-  * If useCurrentSnapshot is true, run the query with the latest available
-  * snapshot, instead of the normal QuerySnapshot.  Also, if it's an update
-  * or delete query, check that the rows to be updated or deleted would be
-  * visible to the normal QuerySnapshot.  (This is a special-case behavior
-  * needed for referential integrity updates in serializable transactions.
-  * We must check all currently-committed rows, but we want to throw a
-  * can't-serialize error if any rows that would need updates would not be
-  * visible under the normal serializable snapshot.)
-  *
   * If explainOnly is true, we are not actually intending to run the plan,
   * only to set up for EXPLAIN; so skip unwanted side-effects.
   *
--- 106,111 ----
***************
*** 123,129 ****
   * ----------------------------------------------------------------
   */
  void
! ExecutorStart(QueryDesc *queryDesc, bool useCurrentSnapshot, bool explainOnly)
  {
  	EState	   *estate;
  	MemoryContext oldcontext;
--- 114,120 ----
   * ----------------------------------------------------------------
   */
  void
! ExecutorStart(QueryDesc *queryDesc, bool explainOnly)
  {
  	EState	   *estate;
  	MemoryContext oldcontext;
***************
*** 156,183 ****
  		estate->es_param_exec_vals = (ParamExecData *)
  			palloc0(queryDesc->plantree->nParamExec * sizeof(ParamExecData));
  
- 	estate->es_instrument = queryDesc->doInstrument;
- 
  	/*
! 	 * Make our own private copy of the current query snapshot data.
! 	 *
! 	 * This "freezes" our idea of which tuples are good and which are not for
! 	 * the life of this query, even if it outlives the current command and
! 	 * current snapshot.
! 	 */
! 	if (useCurrentSnapshot)
! 	{
! 		/* RI update/delete query --- must use an up-to-date snapshot */
! 		estate->es_snapshot = CopyCurrentSnapshot();
! 		/* crosscheck updates/deletes against transaction snapshot */
! 		estate->es_crosscheck_snapshot = CopyQuerySnapshot();
! 	}
! 	else
! 	{
! 		/* normal query --- use query snapshot, no crosscheck */
! 		estate->es_snapshot = CopyQuerySnapshot();
! 		estate->es_crosscheck_snapshot = InvalidSnapshot;
! 	}
  
  	/*
  	 * Initialize the plan state tree
--- 147,158 ----
  		estate->es_param_exec_vals = (ParamExecData *)
  			palloc0(queryDesc->plantree->nParamExec * sizeof(ParamExecData));
  
  	/*
! 	 * Copy other important information into the EState
! 	 */
! 	estate->es_snapshot = queryDesc->snapshot;
! 	estate->es_crosscheck_snapshot = queryDesc->crosscheck_snapshot;
! 	estate->es_instrument = queryDesc->doInstrument;
  
  	/*
  	 * Initialize the plan state tree
***************
*** 1454,1459 ****
--- 1429,1439 ----
  
  	/*
  	 * delete the tuple
+ 	 *
+ 	 * Note: if es_crosscheck_snapshot isn't InvalidSnapshot, we check that
+ 	 * the row to be deleted is visible to that snapshot, and throw a can't-
+ 	 * serialize error if not.  This is a special-case behavior needed for
+ 	 * referential integrity updates in serializable transactions.
  	 */
  ldelete:;
  	result = heap_delete(resultRelationDesc, tupleid,
***************
*** 1591,1596 ****
--- 1571,1581 ----
  
  	/*
  	 * replace the heap tuple
+ 	 *
+ 	 * Note: if es_crosscheck_snapshot isn't InvalidSnapshot, we check that
+ 	 * the row to be updated is visible to that snapshot, and throw a can't-
+ 	 * serialize error if not.  This is a special-case behavior needed for
+ 	 * referential integrity updates in serializable transactions.
  	 */
  	result = heap_update(resultRelationDesc, tupleid, tuple,
  						 &ctid,
*** src/backend/executor/functions.c.orig	Fri Sep 10 14:39:57 2004
--- src/backend/executor/functions.c	Sat Sep 11 23:28:08 2004
***************
*** 65,70 ****
--- 65,71 ----
  	bool		typbyval;		/* true if return type is pass by value */
  	bool		returnsTuple;	/* true if returning whole tuple result */
  	bool		shutdown_reg;	/* true if registered shutdown callback */
+ 	bool		readonly_func;	/* true to run in "read only" mode */
  
  	ParamListInfo paramLI;		/* Param list representing current args */
  
***************
*** 76,86 ****
  
  
  /* non-export function prototypes */
! static execution_state *init_execution_state(List *queryTree_list);
  static void init_sql_fcache(FmgrInfo *finfo);
  static void postquel_start(execution_state *es, SQLFunctionCachePtr fcache);
  static TupleTableSlot *postquel_getnext(execution_state *es);
! static void postquel_end(execution_state *es);
  static void postquel_sub_params(SQLFunctionCachePtr fcache,
  					FunctionCallInfo fcinfo);
  static Datum postquel_execute(execution_state *es,
--- 77,88 ----
  
  
  /* non-export function prototypes */
! static execution_state *init_execution_state(List *queryTree_list,
! 											 bool readonly_func);
  static void init_sql_fcache(FmgrInfo *finfo);
  static void postquel_start(execution_state *es, SQLFunctionCachePtr fcache);
  static TupleTableSlot *postquel_getnext(execution_state *es);
! static void postquel_end(execution_state *es, SQLFunctionCachePtr fcache);
  static void postquel_sub_params(SQLFunctionCachePtr fcache,
  					FunctionCallInfo fcinfo);
  static Datum postquel_execute(execution_state *es,
***************
*** 91,97 ****
  
  
  static execution_state *
! init_execution_state(List *queryTree_list)
  {
  	execution_state *firstes = NULL;
  	execution_state *preves = NULL;
--- 93,99 ----
  
  
  static execution_state *
! init_execution_state(List *queryTree_list, bool readonly_func)
  {
  	execution_state *firstes = NULL;
  	execution_state *preves = NULL;
***************
*** 103,108 ****
--- 105,126 ----
  		Plan	   *planTree;
  		execution_state *newes;
  
+ 		/* Precheck all commands for validity in a function */
+ 		if (queryTree->commandType == CMD_UTILITY &&
+ 			IsA(queryTree->utilityStmt, TransactionStmt))
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ 					 /* translator: %s is a SQL statement name */
+ 					 errmsg("%s is not allowed in a SQL function",
+ 							CreateQueryTag(queryTree))));
+ 
+ 		if (readonly_func && !QueryIsReadOnly(queryTree))
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ 					 /* translator: %s is a SQL statement name */
+ 					 errmsg("%s is not allowed in a non-volatile function",
+ 							CreateQueryTag(queryTree))));
+ 
  		planTree = pg_plan_query(queryTree, NULL);
  
  		newes = (execution_state *) palloc(sizeof(execution_state));
***************
*** 172,177 ****
--- 190,199 ----
  
  	fcache->rettype = rettype;
  
+ 	/* Remember if function is STABLE/IMMUTABLE */
+ 	fcache->readonly_func =
+ 		(procedureStruct->provolatile != PROVOLATILE_VOLATILE);
+ 
  	/* Now look up the actual result type */
  	typeTuple = SearchSysCache(TYPEOID,
  							   ObjectIdGetDatum(rettype),
***************
*** 253,259 ****
  												   queryTree_list);
  
  	/* Finally, plan the queries */
! 	fcache->func_state = init_execution_state(queryTree_list);
  
  	pfree(src);
  
--- 275,282 ----
  												   queryTree_list);
  
  	/* Finally, plan the queries */
! 	fcache->func_state = init_execution_state(queryTree_list,
! 											  fcache->readonly_func);
  
  	pfree(src);
  
***************
*** 267,282 ****
  static void
  postquel_start(execution_state *es, SQLFunctionCachePtr fcache)
  {
  	Assert(es->qd == NULL);
  	es->qd = CreateQueryDesc(es->query, es->plan,
  							 None_Receiver,
  							 fcache->paramLI, false);
  
  	/* Utility commands don't need Executor. */
  	if (es->qd->operation != CMD_UTILITY)
  	{
  		AfterTriggerBeginQuery();
! 		ExecutorStart(es->qd, false, false);
  	}
  
  	es->status = F_EXEC_RUN;
--- 290,321 ----
  static void
  postquel_start(execution_state *es, SQLFunctionCachePtr fcache)
  {
+ 	Snapshot	snapshot;
+ 
  	Assert(es->qd == NULL);
+ 
+ 	/*
+ 	 * In a read-only function, use the surrounding query's snapshot;
+ 	 * otherwise take a new snapshot for each query.  We copy always
+ 	 * so that postquel_end can unconditionally do FreeSnapshot.
+ 	 */
+ 	if (fcache->readonly_func)
+ 		snapshot = CopySnapshot(ActiveSnapshot);
+ 	else
+ 		snapshot = CopySnapshot(GetTransactionSnapshot());
+ 
  	es->qd = CreateQueryDesc(es->query, es->plan,
+ 							 snapshot, InvalidSnapshot,
  							 None_Receiver,
  							 fcache->paramLI, false);
  
+ 	/* We assume we don't need to set up ActiveSnapshot for ExecutorStart */
+ 
  	/* Utility commands don't need Executor. */
  	if (es->qd->operation != CMD_UTILITY)
  	{
  		AfterTriggerBeginQuery();
! 		ExecutorStart(es->qd, false);
  	}
  
  	es->status = F_EXEC_RUN;
***************
*** 285,332 ****
  static TupleTableSlot *
  postquel_getnext(execution_state *es)
  {
  	long		count;
  
! 	if (es->qd->operation == CMD_UTILITY)
  	{
! 		/* Can't handle starting or committing a transaction */
! 		if (IsA(es->qd->parsetree->utilityStmt, TransactionStmt))
! 			ereport(ERROR,
! 					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! 					 errmsg("cannot begin/end transactions in SQL functions")));
! 		ProcessUtility(es->qd->parsetree->utilityStmt, es->qd->params,
! 					   es->qd->dest, NULL);
! 		return NULL;
  	}
  
! 	/*
! 	 * If it's the function's last command, and it's a SELECT, fetch one
! 	 * row at a time so we can return the results.	Otherwise just run it
! 	 * to completion.
! 	 */
! 	if (LAST_POSTQUEL_COMMAND(es) && es->qd->operation == CMD_SELECT)
! 		count = 1L;
! 	else
! 		count = 0L;
  
! 	return ExecutorRun(es->qd, ForwardScanDirection, count);
  }
  
  static void
! postquel_end(execution_state *es)
  {
  	/* mark status done to ensure we don't do ExecutorEnd twice */
  	es->status = F_EXEC_DONE;
  
  	/* Utility commands don't need Executor. */
  	if (es->qd->operation != CMD_UTILITY)
  	{
! 		ExecutorEnd(es->qd);
! 		AfterTriggerEndQuery();
  	}
  
  	FreeQueryDesc(es->qd);
  	es->qd = NULL;
  }
  
  /* Build ParamListInfo array representing current arguments */
--- 324,414 ----
  static TupleTableSlot *
  postquel_getnext(execution_state *es)
  {
+ 	TupleTableSlot *result;
+ 	Snapshot	saveActiveSnapshot;
  	long		count;
  
! 	/* Make our snapshot the active one for any called functions */
! 	saveActiveSnapshot = ActiveSnapshot;
! 	PG_TRY();
  	{
! 		ActiveSnapshot = es->qd->snapshot;
! 
! 		if (es->qd->operation == CMD_UTILITY)
! 		{
! 			ProcessUtility(es->qd->parsetree->utilityStmt, es->qd->params,
! 						   es->qd->dest, NULL);
! 			result = NULL;
! 		}
! 		else
! 		{
! 			/*
! 			 * If it's the function's last command, and it's a SELECT, fetch
! 			 * one row at a time so we can return the results. Otherwise just
! 			 * run it to completion.  (If we run to completion then
! 			 * ExecutorRun is guaranteed to return NULL.)
! 			 */
! 			if (LAST_POSTQUEL_COMMAND(es) && es->qd->operation == CMD_SELECT)
! 				count = 1L;
! 			else
! 				count = 0L;
! 
! 			result = ExecutorRun(es->qd, ForwardScanDirection, count);
! 		}
  	}
+ 	PG_CATCH();
+ 	{
+ 		/* Restore global vars and propagate error */
+ 		ActiveSnapshot = saveActiveSnapshot;
+ 		PG_RE_THROW();
+ 	}
+ 	PG_END_TRY();
  
! 	ActiveSnapshot = saveActiveSnapshot;
  
! 	return result;
  }
  
  static void
! postquel_end(execution_state *es, SQLFunctionCachePtr fcache)
  {
+ 	Snapshot	saveActiveSnapshot;
+ 
  	/* mark status done to ensure we don't do ExecutorEnd twice */
  	es->status = F_EXEC_DONE;
  
  	/* Utility commands don't need Executor. */
  	if (es->qd->operation != CMD_UTILITY)
  	{
! 		/* Make our snapshot the active one for any called functions */
! 		saveActiveSnapshot = ActiveSnapshot;
! 		PG_TRY();
! 		{
! 			ActiveSnapshot = es->qd->snapshot;
! 
! 			ExecutorEnd(es->qd);
! 			AfterTriggerEndQuery();
! 		}
! 		PG_CATCH();
! 		{
! 			/* Restore global vars and propagate error */
! 			ActiveSnapshot = saveActiveSnapshot;
! 			PG_RE_THROW();
! 		}
! 		PG_END_TRY();
! 		ActiveSnapshot = saveActiveSnapshot;
  	}
  
+ 	FreeSnapshot(es->qd->snapshot);
  	FreeQueryDesc(es->qd);
  	es->qd = NULL;
+ 
+ 	/*
+ 	 * We must do CommandCounterIncrement after each completed operation,
+ 	 * unless we are in a read-only function.
+ 	 */
+ 	if (!fcache->readonly_func)
+ 		CommandCounterIncrement();
  }
  
  /* Build ParamListInfo array representing current arguments */
***************
*** 368,373 ****
--- 450,457 ----
  				 SQLFunctionCachePtr fcache)
  {
  	TupleTableSlot *slot;
+ 	HeapTuple	tup;
+ 	TupleDesc	tupDesc;
  	Datum		value;
  
  	if (es->status == F_EXEC_START)
***************
*** 377,477 ****
  
  	if (TupIsNull(slot))
  	{
- 		postquel_end(es);
- 		fcinfo->isnull = true;
- 
  		/*
! 		 * If this isn't the last command for the function we have to
! 		 * increment the command counter so that subsequent commands can
! 		 * see changes made by previous ones.
  		 */
! 		if (!LAST_POSTQUEL_COMMAND(es))
! 			CommandCounterIncrement();
  		return (Datum) NULL;
  	}
  
! 	if (LAST_POSTQUEL_COMMAND(es))
  	{
  		/*
! 		 * Set up to return the function value.
  		 */
! 		HeapTuple	tup = slot->val;
! 		TupleDesc	tupDesc = slot->ttc_tupleDescriptor;
! 
! 		if (fcache->returnsTuple)
! 		{
! 			/*
! 			 * We are returning the whole tuple, so copy it into current
! 			 * execution context and make sure it is a valid Datum.
! 			 *
! 			 * XXX do we need to remove junk attrs from the result tuple?
! 			 * Probably OK to leave them, as long as they are at the end.
! 			 */
! 			HeapTupleHeader dtup;
! 			Oid			dtuptype;
! 			int32		dtuptypmod;
! 
! 			dtup = (HeapTupleHeader) palloc(tup->t_len);
! 			memcpy((char *) dtup, (char *) tup->t_data, tup->t_len);
  
! 			/*
! 			 * Use the declared return type if it's not RECORD; else take
! 			 * the type from the computed result, making sure a typmod has
! 			 * been assigned.
! 			 */
! 			if (fcache->rettype != RECORDOID)
! 			{
! 				/* function has a named composite return type */
! 				dtuptype = fcache->rettype;
! 				dtuptypmod = -1;
! 			}
! 			else
! 			{
! 				/* function is declared to return RECORD */
! 				if (tupDesc->tdtypeid == RECORDOID &&
! 					tupDesc->tdtypmod < 0)
! 					assign_record_type_typmod(tupDesc);
! 				dtuptype = tupDesc->tdtypeid;
! 				dtuptypmod = tupDesc->tdtypmod;
! 			}
! 
! 			HeapTupleHeaderSetDatumLength(dtup, tup->t_len);
! 			HeapTupleHeaderSetTypeId(dtup, dtuptype);
! 			HeapTupleHeaderSetTypMod(dtup, dtuptypmod);
  
! 			value = PointerGetDatum(dtup);
! 			fcinfo->isnull = false;
  		}
  		else
  		{
! 			/*
! 			 * Returning a scalar, which we have to extract from the first
! 			 * column of the SELECT result, and then copy into current
! 			 * execution context if needed.
! 			 */
! 			value = heap_getattr(tup, 1, tupDesc, &(fcinfo->isnull));
! 
! 			if (!fcinfo->isnull)
! 				value = datumCopy(value, fcache->typbyval, fcache->typlen);
  		}
  
  		/*
! 		 * If this is a single valued function we have to end the function
! 		 * execution now.
  		 */
! 		if (!fcinfo->flinfo->fn_retset)
! 			postquel_end(es);
  
! 		return value;
  	}
  
  	/*
! 	 * If this isn't the last command for the function, we don't return
! 	 * any results, but we have to increment the command counter so that
! 	 * subsequent commands can see changes made by previous ones.
  	 */
! 	CommandCounterIncrement();
! 	return (Datum) NULL;
  }
  
  Datum
--- 461,552 ----
  
  	if (TupIsNull(slot))
  	{
  		/*
! 		 * We fall out here for all cases except where we have obtained
! 		 * a row from a function's final SELECT.
  		 */
! 		postquel_end(es, fcache);
! 		fcinfo->isnull = true;
  		return (Datum) NULL;
  	}
  
! 	/*
! 	 * If we got a row from a command within the function it has to be
! 	 * the final command.  All others shouldn't be returning anything.
! 	 */
! 	Assert(LAST_POSTQUEL_COMMAND(es));
! 
! 	/*
! 	 * Set up to return the function value.
! 	 */
! 	tup = slot->val;
! 	tupDesc = slot->ttc_tupleDescriptor;
! 
! 	if (fcache->returnsTuple)
  	{
  		/*
! 		 * We are returning the whole tuple, so copy it into current
! 		 * execution context and make sure it is a valid Datum.
! 		 *
! 		 * XXX do we need to remove junk attrs from the result tuple?
! 		 * Probably OK to leave them, as long as they are at the end.
  		 */
! 		HeapTupleHeader dtup;
! 		Oid			dtuptype;
! 		int32		dtuptypmod;
  
! 		dtup = (HeapTupleHeader) palloc(tup->t_len);
! 		memcpy((char *) dtup, (char *) tup->t_data, tup->t_len);
  
! 		/*
! 		 * Use the declared return type if it's not RECORD; else take
! 		 * the type from the computed result, making sure a typmod has
! 		 * been assigned.
! 		 */
! 		if (fcache->rettype != RECORDOID)
! 		{
! 			/* function has a named composite return type */
! 			dtuptype = fcache->rettype;
! 			dtuptypmod = -1;
  		}
  		else
  		{
! 			/* function is declared to return RECORD */
! 			if (tupDesc->tdtypeid == RECORDOID &&
! 				tupDesc->tdtypmod < 0)
! 				assign_record_type_typmod(tupDesc);
! 			dtuptype = tupDesc->tdtypeid;
! 			dtuptypmod = tupDesc->tdtypmod;
  		}
  
+ 		HeapTupleHeaderSetDatumLength(dtup, tup->t_len);
+ 		HeapTupleHeaderSetTypeId(dtup, dtuptype);
+ 		HeapTupleHeaderSetTypMod(dtup, dtuptypmod);
+ 
+ 		value = PointerGetDatum(dtup);
+ 		fcinfo->isnull = false;
+ 	}
+ 	else
+ 	{
  		/*
! 		 * Returning a scalar, which we have to extract from the first
! 		 * column of the SELECT result, and then copy into current
! 		 * execution context if needed.
  		 */
! 		value = heap_getattr(tup, 1, tupDesc, &(fcinfo->isnull));
  
! 		if (!fcinfo->isnull)
! 			value = datumCopy(value, fcache->typbyval, fcache->typlen);
  	}
  
  	/*
! 	 * If this is a single valued function we have to end the function
! 	 * execution now.
  	 */
! 	if (!fcinfo->flinfo->fn_retset)
! 		postquel_end(es, fcache);
! 
! 	return value;
  }
  
  Datum
***************
*** 726,732 ****
  	{
  		/* Shut down anything still running */
  		if (es->status == F_EXEC_RUN)
! 			postquel_end(es);
  		/* Reset states to START in case we're called again */
  		es->status = F_EXEC_START;
  		es = es->next;
--- 801,807 ----
  	{
  		/* Shut down anything still running */
  		if (es->status == F_EXEC_RUN)
! 			postquel_end(es, fcache);
  		/* Reset states to START in case we're called again */
  		es->status = F_EXEC_START;
  		es = es->next;
*** src/backend/executor/spi.c.orig	Fri Sep 10 14:39:57 2004
--- src/backend/executor/spi.c	Sun Sep 12 16:50:11 2004
***************
*** 34,46 ****
  static int	_SPI_connected = -1;
  static int	_SPI_curid = -1;
  
! static int	_SPI_execute(const char *src, int tcount, _SPI_plan *plan);
! static int _SPI_pquery(QueryDesc *queryDesc, bool runit,
! 			bool useCurrentSnapshot, int tcount);
  
  static int _SPI_execute_plan(_SPI_plan *plan,
! 				  Datum *Values, const char *Nulls,
! 				  bool useCurrentSnapshot, int tcount);
  
  static void _SPI_error_callback(void *arg);
  
--- 34,47 ----
  static int	_SPI_connected = -1;
  static int	_SPI_curid = -1;
  
! static void _SPI_prepare_plan(const char *src, _SPI_plan *plan);
  
  static int _SPI_execute_plan(_SPI_plan *plan,
! 							 Datum *Values, const char *Nulls,
! 							 Snapshot snapshot, Snapshot crosscheck_snapshot,
! 							 bool read_only, int tcount);
! 
! static int _SPI_pquery(QueryDesc *queryDesc, int tcount);
  
  static void _SPI_error_callback(void *arg);
  
***************
*** 252,260 ****
  	_SPI_curid--;
  }
  
  int
! SPI_exec(const char *src, int tcount)
  {
  	int			res;
  
  	if (src == NULL || tcount < 0)
--- 253,263 ----
  	_SPI_curid--;
  }
  
+ /* Parse, plan, and execute a querystring */
  int
! SPI_execute(const char *src, bool read_only, int tcount)
  {
+ 	_SPI_plan	plan;
  	int			res;
  
  	if (src == NULL || tcount < 0)
***************
*** 264,277 ****
  	if (res < 0)
  		return res;
  
! 	res = _SPI_execute(src, tcount, NULL);
  
  	_SPI_end_call(true);
  	return res;
  }
  
  int
! SPI_execp(void *plan, Datum *Values, const char *Nulls, int tcount)
  {
  	int			res;
  
--- 267,298 ----
  	if (res < 0)
  		return res;
  
! 	plan.plancxt = NULL;		/* doesn't have own context */
! 	plan.query = src;
! 	plan.nargs = 0;
! 	plan.argtypes = NULL;
! 
! 	_SPI_prepare_plan(src, &plan);
! 
! 	res = _SPI_execute_plan(&plan, NULL, NULL,
! 							InvalidSnapshot, InvalidSnapshot,
! 							read_only, tcount);
  
  	_SPI_end_call(true);
  	return res;
  }
  
+ /* Obsolete version of SPI_execute */
  int
! SPI_exec(const char *src, int tcount)
! {
! 	return SPI_execute(src, false, tcount);
! }
! 
! /* Execute a previously prepared plan */
! int
! SPI_execute_plan(void *plan, Datum *Values, const char *Nulls,
! 				 bool read_only, int tcount)
  {
  	int			res;
  
***************
*** 285,305 ****
  	if (res < 0)
  		return res;
  
! 	res = _SPI_execute_plan((_SPI_plan *) plan, Values, Nulls, false, tcount);
  
  	_SPI_end_call(true);
  	return res;
  }
  
  /*
!  * SPI_execp_current -- identical to SPI_execp, except that we expose the
!  * Executor option to use a current snapshot instead of the normal
!  * QuerySnapshot.  This is currently not documented in spi.sgml because
!  * it is only intended for use by RI triggers.
   */
! int
! SPI_execp_current(void *plan, Datum *Values, const char *Nulls,
! 				  bool useCurrentSnapshot, int tcount)
  {
  	int			res;
  
--- 306,341 ----
  	if (res < 0)
  		return res;
  
! 	res = _SPI_execute_plan((_SPI_plan *) plan,
! 							Values, Nulls,
! 							InvalidSnapshot, InvalidSnapshot,
! 							read_only, tcount);
  
  	_SPI_end_call(true);
  	return res;
  }
  
+ /* Obsolete version of SPI_execute_plan */
+ int
+ SPI_execp(void *plan, Datum *Values, const char *Nulls, int tcount)
+ {
+ 	return SPI_execute_plan(plan, Values, Nulls, false, tcount);
+ }
+ 
  /*
!  * SPI_execute_snapshot -- identical to SPI_execute_plan, except that we allow
!  * the caller to specify exactly which snapshots to use.  This is currently
!  * not documented in spi.sgml because it is only intended for use by RI
!  * triggers.
!  *
!  * Passing snapshot == InvalidSnapshot will select the normal behavior of
!  * fetching a new snapshot for each query.
   */
! extern int
! SPI_execute_snapshot(void *plan,
! 					 Datum *Values, const char *Nulls,
! 					 Snapshot snapshot, Snapshot crosscheck_snapshot,
! 					 bool read_only, int tcount)
  {
  	int			res;
  
***************
*** 313,320 ****
  	if (res < 0)
  		return res;
  
! 	res = _SPI_execute_plan((_SPI_plan *) plan, Values, Nulls,
! 							useCurrentSnapshot, tcount);
  
  	_SPI_end_call(true);
  	return res;
--- 349,358 ----
  	if (res < 0)
  		return res;
  
! 	res = _SPI_execute_plan((_SPI_plan *) plan,
! 							Values, Nulls,
! 							snapshot, crosscheck_snapshot,
! 							read_only, tcount);
  
  	_SPI_end_call(true);
  	return res;
***************
*** 341,352 ****
  	plan.nargs = nargs;
  	plan.argtypes = argtypes;
  
! 	SPI_result = _SPI_execute(src, 0, &plan);
  
! 	if (SPI_result >= 0)		/* copy plan to procedure context */
! 		result = _SPI_copy_plan(&plan, _SPI_CPLAN_PROCXT);
! 	else
! 		result = NULL;
  
  	_SPI_end_call(true);
  
--- 379,388 ----
  	plan.nargs = nargs;
  	plan.argtypes = argtypes;
  
! 	_SPI_prepare_plan(src, &plan);
  
! 	/* copy plan to procedure context */
! 	result = _SPI_copy_plan(&plan, _SPI_CPLAN_PROCXT);
  
  	_SPI_end_call(true);
  
***************
*** 785,793 ****
  				(errcode(ERRCODE_INVALID_CURSOR_DEFINITION),
  				 errmsg("cannot open SELECT INTO query as cursor")));
  
- 	/* Increment CommandCounter to see changes made by now */
- 	CommandCounterIncrement();
- 
  	/* Reset SPI result */
  	SPI_processed = 0;
  	SPI_tuptable = NULL;
--- 821,826 ----
***************
*** 869,875 ****
  	/*
  	 * Start portal execution.
  	 */
! 	PortalStart(portal, paramLI);
  
  	Assert(portal->strategy == PORTAL_ONE_SELECT);
  
--- 902,908 ----
  	/*
  	 * Start portal execution.
  	 */
! 	PortalStart(portal, paramLI, InvalidSnapshot);
  
  	Assert(portal->strategy == PORTAL_ONE_SELECT);
  
***************
*** 1143,1179 ****
   */
  
  /*
!  * Plan and optionally execute a querystring.
   *
!  * If plan != NULL, just prepare plan trees and save them in *plan;
!  * else execute immediately.
   */
! static int
! _SPI_execute(const char *src, int tcount, _SPI_plan *plan)
  {
  	List	   *raw_parsetree_list;
  	List	   *query_list_list;
  	List	   *plan_list;
  	ListCell   *list_item;
  	ErrorContextCallback spierrcontext;
! 	int			nargs = 0;
! 	Oid		   *argtypes = NULL;
! 	int			res = 0;
! 
! 	if (plan)
! 	{
! 		nargs = plan->nargs;
! 		argtypes = plan->argtypes;
! 	}
! 
! 	/* Increment CommandCounter to see changes made by now */
! 	CommandCounterIncrement();
! 
! 	/* Reset state (only needed in case string is empty) */
! 	SPI_processed = 0;
! 	SPI_lastoid = InvalidOid;
! 	SPI_tuptable = NULL;
! 	_SPI_current->tuptable = NULL;
  
  	/*
  	 * Setup error traceback support for ereport()
--- 1176,1197 ----
   */
  
  /*
!  * Parse and plan a querystring.
   *
!  * At entry, plan->argtypes and plan->nargs must be valid.
!  *
!  * Query and plan lists are stored into *plan.
   */
! static void
! _SPI_prepare_plan(const char *src, _SPI_plan *plan)
  {
  	List	   *raw_parsetree_list;
  	List	   *query_list_list;
  	List	   *plan_list;
  	ListCell   *list_item;
  	ErrorContextCallback spierrcontext;
! 	Oid		   *argtypes = plan->argtypes;
! 	int			nargs = plan->nargs;
  
  	/*
  	 * Setup error traceback support for ereport()
***************
*** 1191,1199 ****
  	/*
  	 * Do parse analysis and rule rewrite for each raw parsetree.
  	 *
! 	 * We save the querytrees from each raw parsetree as a separate sublist.
! 	 * This allows _SPI_execute_plan() to know where the boundaries
! 	 * between original queries fall.
  	 */
  	query_list_list = NIL;
  	plan_list = NIL;
--- 1209,1217 ----
  	/*
  	 * Do parse analysis and rule rewrite for each raw parsetree.
  	 *
! 	 * We save the querytrees from each raw parsetree as a separate
! 	 * sublist.  This allows _SPI_execute_plan() to know where the
! 	 * boundaries between original queries fall.
  	 */
  	query_list_list = NIL;
  	plan_list = NIL;
***************
*** 1202,1404 ****
  	{
  		Node	   *parsetree = (Node *) lfirst(list_item);
  		List	   *query_list;
- 		ListCell   *query_list_item;
  
  		query_list = pg_analyze_and_rewrite(parsetree, argtypes, nargs);
  
  		query_list_list = lappend(query_list_list, query_list);
  
! 		/* Reset state for each original parsetree */
! 		/* (at most one of its querytrees will be marked canSetTag) */
! 		SPI_processed = 0;
! 		SPI_lastoid = InvalidOid;
! 		SPI_tuptable = NULL;
! 		_SPI_current->tuptable = NULL;
! 
! 		foreach(query_list_item, query_list)
! 		{
! 			Query	   *queryTree = (Query *) lfirst(query_list_item);
! 			Plan	   *planTree;
! 			QueryDesc  *qdesc;
! 			DestReceiver *dest;
! 
! 			planTree = pg_plan_query(queryTree, NULL);
! 			plan_list = lappend(plan_list, planTree);
! 
! 			dest = CreateDestReceiver(queryTree->canSetTag ? SPI : None, NULL);
! 			if (queryTree->commandType == CMD_UTILITY)
! 			{
! 				if (IsA(queryTree->utilityStmt, CopyStmt))
! 				{
! 					CopyStmt   *stmt = (CopyStmt *) queryTree->utilityStmt;
! 
! 					if (stmt->filename == NULL)
! 					{
! 						res = SPI_ERROR_COPY;
! 						goto fail;
! 					}
! 				}
! 				else if (IsA(queryTree->utilityStmt, DeclareCursorStmt) ||
! 						 IsA(queryTree->utilityStmt, ClosePortalStmt) ||
! 						 IsA(queryTree->utilityStmt, FetchStmt))
! 				{
! 					res = SPI_ERROR_CURSOR;
! 					goto fail;
! 				}
! 				else if (IsA(queryTree->utilityStmt, TransactionStmt))
! 				{
! 					res = SPI_ERROR_TRANSACTION;
! 					goto fail;
! 				}
! 				res = SPI_OK_UTILITY;
! 				if (plan == NULL)
! 				{
! 					ProcessUtility(queryTree->utilityStmt, NULL, dest, NULL);
! 					CommandCounterIncrement();
! 				}
! 			}
! 			else if (plan == NULL)
! 			{
! 				qdesc = CreateQueryDesc(queryTree, planTree, dest,
! 										NULL, false);
! 				res = _SPI_pquery(qdesc, true, false,
! 								  queryTree->canSetTag ? tcount : 0);
! 				if (res < 0)
! 					goto fail;
! 				CommandCounterIncrement();
! 			}
! 			else
! 			{
! 				qdesc = CreateQueryDesc(queryTree, planTree, dest,
! 										NULL, false);
! 				res = _SPI_pquery(qdesc, false, false, 0);
! 				if (res < 0)
! 					goto fail;
! 			}
! 		}
  	}
  
! 	if (plan)
! 	{
! 		plan->qtlist = query_list_list;
! 		plan->ptlist = plan_list;
! 	}
! 
! fail:
  
  	/*
  	 * Pop the error context stack
  	 */
  	error_context_stack = spierrcontext.previous;
- 
- 	return res;
  }
  
  static int
  _SPI_execute_plan(_SPI_plan *plan, Datum *Values, const char *Nulls,
! 				  bool useCurrentSnapshot, int tcount)
  {
! 	List	   *query_list_list = plan->qtlist;
! 	ListCell   *plan_list_item = list_head(plan->ptlist);
! 	ListCell   *query_list_list_item;
! 	ErrorContextCallback spierrcontext;
! 	int			nargs = plan->nargs;
! 	int			res = 0;
! 	ParamListInfo paramLI;
! 
! 	/* Increment CommandCounter to see changes made by now */
! 	CommandCounterIncrement();
  
! 	/* Convert parameters to form wanted by executor */
! 	if (nargs > 0)
! 	{
! 		int			k;
  
! 		paramLI = (ParamListInfo)
! 			palloc0((nargs + 1) * sizeof(ParamListInfoData));
! 
! 		for (k = 0; k < nargs; k++)
  		{
! 			paramLI[k].kind = PARAM_NUM;
! 			paramLI[k].id = k + 1;
! 			paramLI[k].ptype = plan->argtypes[k];
! 			paramLI[k].isnull = (Nulls && Nulls[k] == 'n');
! 			paramLI[k].value = Values[k];
! 		}
! 		paramLI[k].kind = PARAM_INVALID;
! 	}
! 	else
! 		paramLI = NULL;
  
! 	/* Reset state (only needed in case string is empty) */
! 	SPI_processed = 0;
! 	SPI_lastoid = InvalidOid;
! 	SPI_tuptable = NULL;
! 	_SPI_current->tuptable = NULL;
! 
! 	/*
! 	 * Setup error traceback support for ereport()
! 	 */
! 	spierrcontext.callback = _SPI_error_callback;
! 	spierrcontext.arg = (void *) plan->query;
! 	spierrcontext.previous = error_context_stack;
! 	error_context_stack = &spierrcontext;
  
! 	foreach(query_list_list_item, query_list_list)
! 	{
! 		List	   *query_list = lfirst(query_list_list_item);
! 		ListCell   *query_list_item;
  
! 		/* Reset state for each original parsetree */
! 		/* (at most one of its querytrees will be marked canSetTag) */
  		SPI_processed = 0;
  		SPI_lastoid = InvalidOid;
  		SPI_tuptable = NULL;
  		_SPI_current->tuptable = NULL;
  
! 		foreach(query_list_item, query_list)
  		{
! 			Query	   *queryTree = (Query *) lfirst(query_list_item);
! 			Plan	   *planTree;
! 			QueryDesc  *qdesc;
! 			DestReceiver *dest;
  
! 			planTree = lfirst(plan_list_item);
! 			plan_list_item = lnext(plan_list_item);
  
! 			dest = CreateDestReceiver(queryTree->canSetTag ? SPI : None, NULL);
! 			if (queryTree->commandType == CMD_UTILITY)
  			{
! 				ProcessUtility(queryTree->utilityStmt, paramLI, dest, NULL);
! 				res = SPI_OK_UTILITY;
! 				CommandCounterIncrement();
! 			}
! 			else
! 			{
! 				qdesc = CreateQueryDesc(queryTree, planTree, dest,
! 										paramLI, false);
! 				res = _SPI_pquery(qdesc, true, useCurrentSnapshot,
! 								  queryTree->canSetTag ? tcount : 0);
  				if (res < 0)
  					goto fail;
! 				CommandCounterIncrement();
  			}
  		}
- 	}
  
  fail:
  
! 	/*
! 	 * Pop the error context stack
! 	 */
! 	error_context_stack = spierrcontext.previous;
  
  	return res;
  }
  
  static int
! _SPI_pquery(QueryDesc *queryDesc, bool runit,
! 			bool useCurrentSnapshot, int tcount)
  {
  	int			operation = queryDesc->operation;
  	int			res;
--- 1220,1417 ----
  	{
  		Node	   *parsetree = (Node *) lfirst(list_item);
  		List	   *query_list;
  
  		query_list = pg_analyze_and_rewrite(parsetree, argtypes, nargs);
  
  		query_list_list = lappend(query_list_list, query_list);
  
! 		plan_list = list_concat(plan_list,
! 								pg_plan_queries(query_list, NULL, false));
  	}
  
! 	plan->qtlist = query_list_list;
! 	plan->ptlist = plan_list;
  
  	/*
  	 * Pop the error context stack
  	 */
  	error_context_stack = spierrcontext.previous;
  }
  
+ /*
+  * Execute the given plan with the given parameter values
+  *
+  * snapshot: query snapshot to use, or InvalidSnapshot for the normal
+  *		behavior of taking a new snapshot for each query.
+  * crosscheck_snapshot: for RI use, all others pass InvalidSnapshot
+  * read_only: TRUE for read-only execution (no CommandCounterIncrement)
+  * tcount: execution tuple-count limit, or 0 for none
+  */
  static int
  _SPI_execute_plan(_SPI_plan *plan, Datum *Values, const char *Nulls,
! 				  Snapshot snapshot, Snapshot crosscheck_snapshot,
! 				  bool read_only, int tcount)
  {
! 	volatile int res = 0;
! 	Snapshot	saveActiveSnapshot;
  
! 	/* Be sure to restore ActiveSnapshot on error exit */
! 	saveActiveSnapshot = ActiveSnapshot;
! 	PG_TRY();
! 	{
! 		List	   *query_list_list = plan->qtlist;
! 		ListCell   *plan_list_item = list_head(plan->ptlist);
! 		ListCell   *query_list_list_item;
! 		ErrorContextCallback spierrcontext;
! 		int			nargs = plan->nargs;
! 		ParamListInfo paramLI;
  
! 		/* Convert parameters to form wanted by executor */
! 		if (nargs > 0)
  		{
! 			int			k;
  
! 			paramLI = (ParamListInfo)
! 				palloc0((nargs + 1) * sizeof(ParamListInfoData));
  
! 			for (k = 0; k < nargs; k++)
! 			{
! 				paramLI[k].kind = PARAM_NUM;
! 				paramLI[k].id = k + 1;
! 				paramLI[k].ptype = plan->argtypes[k];
! 				paramLI[k].isnull = (Nulls && Nulls[k] == 'n');
! 				paramLI[k].value = Values[k];
! 			}
! 			paramLI[k].kind = PARAM_INVALID;
! 		}
! 		else
! 			paramLI = NULL;
  
! 		/* Reset state (only needed in case string is empty) */
  		SPI_processed = 0;
  		SPI_lastoid = InvalidOid;
  		SPI_tuptable = NULL;
  		_SPI_current->tuptable = NULL;
  
! 		/*
! 		 * Setup error traceback support for ereport()
! 		 */
! 		spierrcontext.callback = _SPI_error_callback;
! 		spierrcontext.arg = (void *) plan->query;
! 		spierrcontext.previous = error_context_stack;
! 		error_context_stack = &spierrcontext;
! 
! 		foreach(query_list_list_item, query_list_list)
  		{
! 			List	   *query_list = lfirst(query_list_list_item);
! 			ListCell   *query_list_item;
  
! 			/* Reset state for each original parsetree */
! 			/* (at most one of its querytrees will be marked canSetTag) */
! 			SPI_processed = 0;
! 			SPI_lastoid = InvalidOid;
! 			SPI_tuptable = NULL;
! 			_SPI_current->tuptable = NULL;
  
! 			foreach(query_list_item, query_list)
  			{
! 				Query	   *queryTree = (Query *) lfirst(query_list_item);
! 				Plan	   *planTree;
! 				QueryDesc  *qdesc;
! 				DestReceiver *dest;
! 
! 				planTree = lfirst(plan_list_item);
! 				plan_list_item = lnext(plan_list_item);
! 
! 				if (queryTree->commandType == CMD_UTILITY)
! 				{
! 					if (IsA(queryTree->utilityStmt, CopyStmt))
! 					{
! 						CopyStmt   *stmt = (CopyStmt *) queryTree->utilityStmt;
! 
! 						if (stmt->filename == NULL)
! 						{
! 							res = SPI_ERROR_COPY;
! 							goto fail;
! 						}
! 					}
! 					else if (IsA(queryTree->utilityStmt, DeclareCursorStmt) ||
! 							 IsA(queryTree->utilityStmt, ClosePortalStmt) ||
! 							 IsA(queryTree->utilityStmt, FetchStmt))
! 					{
! 						res = SPI_ERROR_CURSOR;
! 						goto fail;
! 					}
! 					else if (IsA(queryTree->utilityStmt, TransactionStmt))
! 					{
! 						res = SPI_ERROR_TRANSACTION;
! 						goto fail;
! 					}
! 				}
! 
! 				if (read_only && !QueryIsReadOnly(queryTree))
! 					ereport(ERROR,
! 							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! 							 /* translator: %s is a SQL statement name */
! 							 errmsg("%s is not allowed in a non-volatile function",
! 									CreateQueryTag(queryTree))));
! 
! 				dest = CreateDestReceiver(queryTree->canSetTag ? SPI : None,
! 										  NULL);
! 				if (snapshot == InvalidSnapshot)
! 					ActiveSnapshot = CopySnapshot(GetTransactionSnapshot());
! 				else
! 					ActiveSnapshot = CopySnapshot(snapshot);
! 
! 				if (queryTree->commandType == CMD_UTILITY)
! 				{
! 					ProcessUtility(queryTree->utilityStmt, paramLI,
! 								   dest, NULL);
! 					res = SPI_OK_UTILITY;
! 				}
! 				else
! 				{
! 					qdesc = CreateQueryDesc(queryTree, planTree,
! 											ActiveSnapshot,
! 											crosscheck_snapshot,
! 											dest,
! 											paramLI, false);
! 					res = _SPI_pquery(qdesc,
! 									  queryTree->canSetTag ? tcount : 0);
! 					FreeQueryDesc(qdesc);
! 				}
! 				FreeSnapshot(ActiveSnapshot);
! 				ActiveSnapshot = NULL;
! 				/* we know that the receiver doesn't need a destroy call */
  				if (res < 0)
  					goto fail;
! 				if (!read_only)
! 					CommandCounterIncrement();
  			}
  		}
  
  fail:
  
! 		/*
! 		 * Pop the error context stack
! 		 */
! 		error_context_stack = spierrcontext.previous;
! 	}
! 	PG_CATCH();
! 	{
! 		/* Restore global vars and propagate error */
! 		ActiveSnapshot = saveActiveSnapshot;
! 		PG_RE_THROW();
! 	}
! 	PG_END_TRY();
! 
! 	ActiveSnapshot = saveActiveSnapshot;
  
  	return res;
  }
  
  static int
! _SPI_pquery(QueryDesc *queryDesc, int tcount)
  {
  	int			operation = queryDesc->operation;
  	int			res;
***************
*** 1427,1435 ****
  			return SPI_ERROR_OPUNKNOWN;
  	}
  
- 	if (!runit)					/* plan preparation, don't execute */
- 		return res;
- 
  #ifdef SPI_EXECUTOR_STATS
  	if (ShowExecutorStats)
  		ResetUsage();
--- 1440,1445 ----
***************
*** 1437,1443 ****
  
  	AfterTriggerBeginQuery();
  
! 	ExecutorStart(queryDesc, useCurrentSnapshot, false);
  
  	ExecutorRun(queryDesc, ForwardScanDirection, (long) tcount);
  
--- 1447,1453 ----
  
  	AfterTriggerBeginQuery();
  
! 	ExecutorStart(queryDesc, false);
  
  	ExecutorRun(queryDesc, ForwardScanDirection, (long) tcount);
  
***************
*** 1466,1473 ****
  		/* Don't return SPI_OK_SELECT if we discarded the result */
  		res = SPI_OK_UTILITY;
  	}
- 
- 	FreeQueryDesc(queryDesc);
  
  #ifdef SPI_EXECUTOR_STATS
  	if (ShowExecutorStats)
--- 1476,1481 ----
*** src/backend/tcop/fastpath.c.orig	Sun Aug 29 09:55:21 2004
--- src/backend/tcop/fastpath.c	Sat Sep 11 19:18:05 2004
***************
*** 334,344 ****
  					   get_func_name(fid));
  
  	/*
- 	 * Set up a query snapshot in case function needs one.
- 	 */
- 	SetQuerySnapshot();
- 
- 	/*
  	 * Prepare function call info block and insert arguments.
  	 */
  	MemSet(&fcinfo, 0, sizeof(fcinfo));
--- 334,339 ----
*** src/backend/tcop/postgres.c.orig	Fri Sep 10 14:39:59 2004
--- src/backend/tcop/postgres.c	Sat Sep 11 19:17:49 2004
***************
*** 700,706 ****
  		{
  			if (needSnapshot)
  			{
! 				SetQuerySnapshot();
  				needSnapshot = false;
  			}
  			plan = pg_plan_query(query, boundParams);
--- 700,706 ----
  		{
  			if (needSnapshot)
  			{
! 				ActiveSnapshot = CopySnapshot(GetTransactionSnapshot());
  				needSnapshot = false;
  			}
  			plan = pg_plan_query(query, boundParams);
***************
*** 883,889 ****
  		/*
  		 * Start the portal.  No parameters here.
  		 */
! 		PortalStart(portal, NULL);
  
  		/*
  		 * Select the appropriate output format: text unless we are doing
--- 883,889 ----
  		/*
  		 * Start the portal.  No parameters here.
  		 */
! 		PortalStart(portal, NULL, InvalidSnapshot);
  
  		/*
  		 * Select the appropriate output format: text unless we are doing
***************
*** 1539,1545 ****
  					  pstmt->plan_list,
  					  pstmt->context);
  
! 	PortalStart(portal, params);
  
  	/*
  	 * Apply the result format requests to the portal.
--- 1539,1545 ----
  					  pstmt->plan_list,
  					  pstmt->context);
  
! 	PortalStart(portal, params, InvalidSnapshot);
  
  	/*
  	 * Apply the result format requests to the portal.
***************
*** 3026,3031 ****
--- 3026,3034 ----
  
  				/* switch back to message context */
  				MemoryContextSwitchTo(MessageContext);
+ 
+ 				/* set snapshot in case function needs one */
+ 				ActiveSnapshot = CopySnapshot(GetTransactionSnapshot());
  
  				if (HandleFunctionRequest(&input_message) == EOF)
  				{
*** src/backend/tcop/pquery.c.orig	Fri Sep 10 14:40:00 2004
--- src/backend/tcop/pquery.c	Sat Sep 11 21:00:49 2004
***************
*** 32,37 ****
--- 32,42 ----
  Portal		ActivePortal = NULL;
  
  
+ static void ProcessQuery(Query *parsetree,
+ 			 Plan *plan,
+ 			 ParamListInfo params,
+ 			 DestReceiver *dest,
+ 			 char *completionTag);
  static uint32 RunFromStore(Portal portal, ScanDirection direction, long count,
  			 DestReceiver *dest);
  static long PortalRunSelect(Portal portal, bool forward, long count,
***************
*** 54,59 ****
--- 59,66 ----
  QueryDesc *
  CreateQueryDesc(Query *parsetree,
  				Plan *plantree,
+ 				Snapshot snapshot,
+ 				Snapshot crosscheck_snapshot,
  				DestReceiver *dest,
  				ParamListInfo params,
  				bool doInstrument)
***************
*** 63,68 ****
--- 70,77 ----
  	qd->operation = parsetree->commandType;		/* operation */
  	qd->parsetree = parsetree;	/* parse tree */
  	qd->plantree = plantree;	/* plan */
+ 	qd->snapshot = snapshot;	/* snapshot */
+ 	qd->crosscheck_snapshot = crosscheck_snapshot; /* RI check snapshot */
  	qd->dest = dest;			/* output dest */
  	qd->params = params;		/* parameter values passed into query */
  	qd->doInstrument = doInstrument;	/* instrumentation wanted? */
***************
*** 90,96 ****
  
  /*
   * ProcessQuery
!  *		Execute a single query
   *
   *	parsetree: the query tree
   *	plan: the plan tree for the query
--- 99,105 ----
  
  /*
   * ProcessQuery
!  *		Execute a single plannable query within a PORTAL_MULTI_QUERY portal
   *
   *	parsetree: the query tree
   *	plan: the plan tree for the query
***************
*** 104,110 ****
   * Must be called in a memory context that will be reset or deleted on
   * error; otherwise the executor's memory usage will be leaked.
   */
! void
  ProcessQuery(Query *parsetree,
  			 Plan *plan,
  			 ParamListInfo params,
--- 113,119 ----
   * Must be called in a memory context that will be reset or deleted on
   * error; otherwise the executor's memory usage will be leaked.
   */
! static void
  ProcessQuery(Query *parsetree,
  			 Plan *plan,
  			 ParamListInfo params,
***************
*** 114,119 ****
--- 123,131 ----
  	int			operation = parsetree->commandType;
  	QueryDesc  *queryDesc;
  
+ 	ereport(DEBUG3,
+ 			(errmsg_internal("ProcessQuery")));
+ 
  	/*
  	 * Check for special-case destinations
  	 */
***************
*** 133,141 ****
  	}
  
  	/*
  	 * Create the QueryDesc object
  	 */
! 	queryDesc = CreateQueryDesc(parsetree, plan, dest, params, false);
  
  	/*
  	 * Set up to collect AFTER triggers
--- 145,161 ----
  	}
  
  	/*
+ 	 * Must always set snapshot for plannable queries.  Note we assume
+ 	 * that caller will take care of restoring ActiveSnapshot on exit/error.
+ 	 */
+ 	ActiveSnapshot = CopySnapshot(GetTransactionSnapshot());
+ 
+ 	/*
  	 * Create the QueryDesc object
  	 */
! 	queryDesc = CreateQueryDesc(parsetree, plan,
! 								ActiveSnapshot, InvalidSnapshot,
! 								dest, params, false);
  
  	/*
  	 * Set up to collect AFTER triggers
***************
*** 145,151 ****
  	/*
  	 * Call ExecStart to prepare the plan for execution
  	 */
! 	ExecutorStart(queryDesc, false, false);
  
  	/*
  	 * Run the plan to completion.
--- 165,171 ----
  	/*
  	 * Call ExecStart to prepare the plan for execution
  	 */
! 	ExecutorStart(queryDesc, false);
  
  	/*
  	 * Run the plan to completion.
***************
*** 195,200 ****
--- 215,223 ----
  	AfterTriggerEndQuery();
  
  	FreeQueryDesc(queryDesc);
+ 
+ 	FreeSnapshot(ActiveSnapshot);
+ 	ActiveSnapshot = NULL;
  }
  
  /*
***************
*** 238,250 ****
   * the query, they must be passed in here (caller is responsible for
   * giving them appropriate lifetime).
   *
   * On return, portal is ready to accept PortalRun() calls, and the result
   * tupdesc (if any) is known.
   */
  void
! PortalStart(Portal portal, ParamListInfo params)
  {
  	Portal		saveActivePortal;
  	ResourceOwner saveResourceOwner;
  	MemoryContext savePortalContext;
  	MemoryContext oldContext;
--- 261,279 ----
   * the query, they must be passed in here (caller is responsible for
   * giving them appropriate lifetime).
   *
+  * The caller can optionally pass a snapshot to be used; pass InvalidSnapshot
+  * for the normal behavior of setting a new snapshot.  This parameter is
+  * presently ignored for non-PORTAL_ONE_SELECT portals (it's only intended
+  * to be used for cursors).
+  *
   * On return, portal is ready to accept PortalRun() calls, and the result
   * tupdesc (if any) is known.
   */
  void
! PortalStart(Portal portal, ParamListInfo params, Snapshot snapshot)
  {
  	Portal		saveActivePortal;
+ 	Snapshot	saveActiveSnapshot;
  	ResourceOwner saveResourceOwner;
  	MemoryContext savePortalContext;
  	MemoryContext oldContext;
***************
*** 259,269 ****
--- 288,300 ----
  	 * QueryContext?)
  	 */
  	saveActivePortal = ActivePortal;
+ 	saveActiveSnapshot = ActiveSnapshot;
  	saveResourceOwner = CurrentResourceOwner;
  	savePortalContext = PortalContext;
  	PG_TRY();
  	{
  		ActivePortal = portal;
+ 		ActiveSnapshot = NULL;				/* will be set later */
  		CurrentResourceOwner = portal->resowner;
  		PortalContext = PortalGetHeapMemory(portal);
  
***************
*** 285,293 ****
  			case PORTAL_ONE_SELECT:
  
  				/*
! 				 * Must set query snapshot before starting executor.
  				 */
! 				SetQuerySnapshot();
  
  				/*
  				 * Create QueryDesc in portal's context; for the moment,
--- 316,328 ----
  			case PORTAL_ONE_SELECT:
  
  				/*
! 				 * Must set snapshot before starting executor.  Be sure to
! 				 * copy it into the portal's context.
  				 */
! 				if (snapshot)
! 					ActiveSnapshot = CopySnapshot(snapshot);
! 				else
! 					ActiveSnapshot = CopySnapshot(GetTransactionSnapshot());
  
  				/*
  				 * Create QueryDesc in portal's context; for the moment,
***************
*** 295,300 ****
--- 330,337 ----
  				 */
  				queryDesc = CreateQueryDesc((Query *) linitial(portal->parseTrees),
  									(Plan *) linitial(portal->planTrees),
+ 											ActiveSnapshot,
+ 											InvalidSnapshot,
  											None_Receiver,
  											params,
  											false);
***************
*** 309,315 ****
  				/*
  				 * Call ExecStart to prepare the plan for execution
  				 */
! 				ExecutorStart(queryDesc, false, false);
  
  				/*
  				 * This tells PortalCleanup to shut down the executor
--- 346,352 ----
  				/*
  				 * Call ExecStart to prepare the plan for execution
  				 */
! 				ExecutorStart(queryDesc, false);
  
  				/*
  				 * This tells PortalCleanup to shut down the executor
***************
*** 333,340 ****
  			case PORTAL_UTIL_SELECT:
  
  				/*
! 				 * We don't set query snapshot here, because
! 				 * PortalRunUtility will take care of it.
  				 */
  				portal->tupDesc =
  					UtilityTupleDescriptor(((Query *) linitial(portal->parseTrees))->utilityStmt);
--- 370,377 ----
  			case PORTAL_UTIL_SELECT:
  
  				/*
! 				 * We don't set snapshot here, because
! 				 * PortalRunUtility will take care of it if needed.
  				 */
  				portal->tupDesc =
  					UtilityTupleDescriptor(((Query *) linitial(portal->parseTrees))->utilityStmt);
***************
*** 361,366 ****
--- 398,404 ----
  
  		/* Restore global vars and propagate error */
  		ActivePortal = saveActivePortal;
+ 		ActiveSnapshot = saveActiveSnapshot;
  		CurrentResourceOwner = saveResourceOwner;
  		PortalContext = savePortalContext;
  
***************
*** 371,376 ****
--- 409,415 ----
  	MemoryContextSwitchTo(oldContext);
  
  	ActivePortal = saveActivePortal;
+ 	ActiveSnapshot = saveActiveSnapshot;
  	CurrentResourceOwner = saveResourceOwner;
  	PortalContext = savePortalContext;
  
***************
*** 453,458 ****
--- 492,498 ----
  {
  	bool		result;
  	Portal		saveActivePortal;
+ 	Snapshot	saveActiveSnapshot;
  	ResourceOwner saveResourceOwner;
  	MemoryContext savePortalContext;
  	MemoryContext saveQueryContext;
***************
*** 485,496 ****
--- 525,538 ----
  	 * Set up global portal context pointers.
  	 */
  	saveActivePortal = ActivePortal;
+ 	saveActiveSnapshot = ActiveSnapshot;
  	saveResourceOwner = CurrentResourceOwner;
  	savePortalContext = PortalContext;
  	saveQueryContext = QueryContext;
  	PG_TRY();
  	{
  		ActivePortal = portal;
+ 		ActiveSnapshot = NULL;				/* will be set later */
  		CurrentResourceOwner = portal->resowner;
  		PortalContext = PortalGetHeapMemory(portal);
  		QueryContext = portal->queryContext;
***************
*** 579,584 ****
--- 621,627 ----
  
  		/* Restore global vars and propagate error */
  		ActivePortal = saveActivePortal;
+ 		ActiveSnapshot = saveActiveSnapshot;
  		CurrentResourceOwner = saveResourceOwner;
  		PortalContext = savePortalContext;
  		QueryContext = saveQueryContext;
***************
*** 590,595 ****
--- 633,639 ----
  	MemoryContextSwitchTo(oldContext);
  
  	ActivePortal = saveActivePortal;
+ 	ActiveSnapshot = saveActiveSnapshot;
  	CurrentResourceOwner = saveResourceOwner;
  	PortalContext = savePortalContext;
  	QueryContext = saveQueryContext;
***************
*** 670,675 ****
--- 714,720 ----
  			nprocessed = RunFromStore(portal, direction, count, dest);
  		else
  		{
+ 			ActiveSnapshot = queryDesc->snapshot;
  			ExecutorRun(queryDesc, direction, count);
  			nprocessed = queryDesc->estate->es_processed;
  		}
***************
*** 711,716 ****
--- 756,762 ----
  			nprocessed = RunFromStore(portal, direction, count, dest);
  		else
  		{
+ 			ActiveSnapshot = queryDesc->snapshot;
  			ExecutorRun(queryDesc, direction, count);
  			nprocessed = queryDesc->estate->es_processed;
  		}
***************
*** 834,839 ****
--- 880,888 ----
  	 * the database --- if, say, it has to update an index with
  	 * expressions that invoke user-defined functions, then it had better
  	 * have a snapshot.
+ 	 *
+ 	 * Note we assume that caller will take care of restoring ActiveSnapshot
+ 	 * on exit/error.
  	 */
  	if (!(IsA(utilityStmt, TransactionStmt) ||
  		  IsA(utilityStmt, LockStmt) ||
***************
*** 847,853 ****
  		  IsA(utilityStmt, NotifyStmt) ||
  		  IsA(utilityStmt, UnlistenStmt) ||
  		  IsA(utilityStmt, CheckPointStmt)))
! 		SetQuerySnapshot();
  
  	if (query->canSetTag)
  	{
--- 896,904 ----
  		  IsA(utilityStmt, NotifyStmt) ||
  		  IsA(utilityStmt, UnlistenStmt) ||
  		  IsA(utilityStmt, CheckPointStmt)))
! 		ActiveSnapshot = CopySnapshot(GetTransactionSnapshot());
! 	else
! 		ActiveSnapshot = NULL;
  
  	if (query->canSetTag)
  	{
***************
*** 864,869 ****
--- 915,924 ----
  
  	/* Some utility statements may change context on us */
  	MemoryContextSwitchTo(PortalGetHeapMemory(portal));
+ 
+ 	if (ActiveSnapshot)
+ 		FreeSnapshot(ActiveSnapshot);
+ 	ActiveSnapshot = NULL;
  }
  
  /*
***************
*** 924,938 ****
  			/*
  			 * process a plannable query.
  			 */
- 			ereport(DEBUG3,
- 					(errmsg_internal("ProcessQuery")));
- 
- 			/* Must always set snapshot for plannable queries */
- 			SetQuerySnapshot();
- 
- 			/*
- 			 * execute the plan
- 			 */
  			if (log_executor_stats)
  				ResetUsage();
  
--- 979,984 ----
***************
*** 1005,1010 ****
--- 1051,1057 ----
  {
  	long		result;
  	Portal		saveActivePortal;
+ 	Snapshot	saveActiveSnapshot;
  	ResourceOwner saveResourceOwner;
  	MemoryContext savePortalContext;
  	MemoryContext saveQueryContext;
***************
*** 1025,1036 ****
--- 1072,1085 ----
  	 * Set up global portal context pointers.
  	 */
  	saveActivePortal = ActivePortal;
+ 	saveActiveSnapshot = ActiveSnapshot;
  	saveResourceOwner = CurrentResourceOwner;
  	savePortalContext = PortalContext;
  	saveQueryContext = QueryContext;
  	PG_TRY();
  	{
  		ActivePortal = portal;
+ 		ActiveSnapshot = NULL;				/* will be set later */
  		CurrentResourceOwner = portal->resowner;
  		PortalContext = PortalGetHeapMemory(portal);
  		QueryContext = portal->queryContext;
***************
*** 1056,1061 ****
--- 1105,1111 ----
  
  		/* Restore global vars and propagate error */
  		ActivePortal = saveActivePortal;
+ 		ActiveSnapshot = saveActiveSnapshot;
  		CurrentResourceOwner = saveResourceOwner;
  		PortalContext = savePortalContext;
  		QueryContext = saveQueryContext;
***************
*** 1070,1075 ****
--- 1120,1126 ----
  	portal->status = PORTAL_READY;
  
  	ActivePortal = saveActivePortal;
+ 	ActiveSnapshot = saveActiveSnapshot;
  	CurrentResourceOwner = saveResourceOwner;
  	PortalContext = savePortalContext;
  	QueryContext = saveQueryContext;
*** src/backend/tcop/utility.c.orig	Fri Sep 10 14:40:00 2004
--- src/backend/tcop/utility.c	Sat Sep 11 22:09:13 2004
***************
*** 222,227 ****
--- 222,267 ----
  }
  
  
+ /*
+  * QueryIsReadOnly: is an analyzed/rewritten query read-only?
+  *
+  * This is a much stricter test than we apply for XactReadOnly mode;
+  * the query must be *in truth* read-only, because the caller wishes
+  * not to do CommandCounterIncrement afterwards.
+  */
+ bool
+ QueryIsReadOnly(Query *parsetree)
+ {
+ 	switch (parsetree->commandType)
+ 	{
+ 		case CMD_SELECT:
+ 			if (parsetree->into != NULL)
+ 				return false;					/* SELECT INTO */
+ 			else if (parsetree->rowMarks != NIL)
+ 				return false;					/* SELECT FOR UPDATE */
+ 			else
+ 				return true;
+ 		case CMD_UPDATE:
+ 		case CMD_INSERT:
+ 		case CMD_DELETE:
+ 			return false;
+ 		case CMD_UTILITY:
+ 			/* For now, treat all utility commands as read/write */
+ 			return false;
+ 		default:
+ 			elog(WARNING, "unrecognized commandType: %d",
+ 				 (int) parsetree->commandType);
+ 			break;
+ 	}
+ 	return false;
+ }
+ 
+ /*
+  * check_xact_readonly: is a utility command read-only?
+  *
+  * Here we use the loose rules of XactReadOnly mode: no permanent effects
+  * on the database are allowed.
+  */
  static void
  check_xact_readonly(Node *parsetree)
  {
***************
*** 299,306 ****
   *	completionTag: points to a buffer of size COMPLETION_TAG_BUFSIZE
   *		in which to store a command completion status string.
   *
!  * completionTag is only set nonempty if we want to return a nondefault
!  * status (currently, only used for MOVE/FETCH).
   *
   * completionTag may be NULL if caller doesn't want a status string.
   */
--- 339,345 ----
   *	completionTag: points to a buffer of size COMPLETION_TAG_BUFSIZE
   *		in which to store a command completion status string.
   *
!  * completionTag is only set nonempty if we want to return a nondefault status.
   *
   * completionTag may be NULL if caller doesn't want a status string.
   */
***************
*** 1580,1585 ****
--- 1619,1672 ----
  		default:
  			elog(WARNING, "unrecognized node type: %d",
  				 (int) nodeTag(parsetree));
+ 			tag = "???";
+ 			break;
+ 	}
+ 
+ 	return tag;
+ }
+ 
+ /*
+  * CreateQueryTag
+  *		utility to get a string representation of a Query operation.
+  *
+  * This is exactly like CreateCommandTag, except it works on a Query
+  * that has already been through parse analysis (and possibly further).
+  */
+ const char *
+ CreateQueryTag(Query *parsetree)
+ {
+ 	const char *tag;
+ 
+ 	switch (parsetree->commandType)
+ 	{
+ 		case CMD_SELECT:
+ 			/*
+ 			 * We take a little extra care here so that the result will
+ 			 * be useful for complaints about read-only statements
+ 			 */
+ 			if (parsetree->into != NULL)
+ 				tag = "SELECT INTO";
+ 			else if (parsetree->rowMarks != NIL)
+ 				tag = "SELECT FOR UPDATE";
+ 			else
+ 				tag = "SELECT";
+ 			break;
+ 		case CMD_UPDATE:
+ 			tag = "UPDATE";
+ 			break;
+ 		case CMD_INSERT:
+ 			tag = "INSERT";
+ 			break;
+ 		case CMD_DELETE:
+ 			tag = "DELETE";
+ 			break;
+ 		case CMD_UTILITY:
+ 			tag = CreateCommandTag(parsetree->utilityStmt);
+ 			break;
+ 		default:
+ 			elog(WARNING, "unrecognized commandType: %d",
+ 				 (int) parsetree->commandType);
  			tag = "???";
  			break;
  	}
*** src/backend/utils/adt/ri_triggers.c.orig	Fri Sep 10 14:40:04 2004
--- src/backend/utils/adt/ri_triggers.c	Sun Sep 12 16:49:46 2004
***************
*** 2698,2713 ****
  		elog(ERROR, "SPI_prepare returned %d for %s", SPI_result, querystr);
  
  	/*
! 	 * Run the plan.  For safety we force a current query snapshot to be
! 	 * used.  (In serializable mode, this arguably violates
! 	 * serializability, but we really haven't got much choice.)  We need
! 	 * at most one tuple returned, so pass limit = 1.
! 	 */
! 	spi_result = SPI_execp_current(qplan, NULL, NULL, true, 1);
  
  	/* Check result */
  	if (spi_result != SPI_OK_SELECT)
! 		elog(ERROR, "SPI_execp_current returned %d", spi_result);
  
  	/* Did we find a tuple violating the constraint? */
  	if (SPI_processed > 0)
--- 2698,2717 ----
  		elog(ERROR, "SPI_prepare returned %d for %s", SPI_result, querystr);
  
  	/*
! 	 * Run the plan.  For safety we force a current snapshot to be used.
! 	 * (In serializable mode, this arguably violates serializability, but we
! 	 * really haven't got much choice.)  We need at most one tuple returned,
! 	 * so pass limit = 1.
! 	 */
! 	spi_result = SPI_execute_snapshot(qplan,
! 									  NULL, NULL,
! 									  CopySnapshot(GetLatestSnapshot()),
! 									  InvalidSnapshot,
! 									  false, 1);
  
  	/* Check result */
  	if (spi_result != SPI_OK_SELECT)
! 		elog(ERROR, "SPI_execute_snapshot returned %d", spi_result);
  
  	/* Did we find a tuple violating the constraint? */
  	if (SPI_processed > 0)
***************
*** 3043,3049 ****
  	Relation	query_rel,
  				source_rel;
  	int			key_idx;
! 	bool		useCurrentSnapshot;
  	int			limit;
  	int			spi_result;
  	AclId		save_uid;
--- 3047,3054 ----
  	Relation	query_rel,
  				source_rel;
  	int			key_idx;
! 	Snapshot	test_snapshot;
! 	Snapshot	crosscheck_snapshot;
  	int			limit;
  	int			spi_result;
  	AclId		save_uid;
***************
*** 3094,3114 ****
  	}
  
  	/*
! 	 * In READ COMMITTED mode, we just need to make sure the regular query
! 	 * snapshot is up-to-date, and we will see all rows that could be
! 	 * interesting.  In SERIALIZABLE mode, we can't update the regular
! 	 * query snapshot.	If the caller passes detectNewRows == false then
! 	 * it's okay to do the query with the transaction snapshot; otherwise
! 	 * we tell the executor to force a current snapshot (and error out if
! 	 * it finds any rows under current snapshot that wouldn't be visible
! 	 * per the transaction snapshot).
  	 */
! 	if (IsXactIsoLevelSerializable)
! 		useCurrentSnapshot = detectNewRows;
  	else
  	{
! 		SetQuerySnapshot();
! 		useCurrentSnapshot = false;
  	}
  
  	/*
--- 3099,3122 ----
  	}
  
  	/*
! 	 * In READ COMMITTED mode, we just need to use an up-to-date regular
! 	 * snapshot, and we will see all rows that could be interesting.
! 	 * But in SERIALIZABLE mode, we can't change the transaction snapshot.
! 	 * If the caller passes detectNewRows == false then it's okay to do the
! 	 * query with the transaction snapshot; otherwise we use a current
! 	 * snapshot, and tell the executor to error out if it finds any rows under
! 	 * the current snapshot that wouldn't be visible per the transaction
! 	 * snapshot.
  	 */
! 	if (IsXactIsoLevelSerializable && detectNewRows)
! 	{
! 		test_snapshot = CopySnapshot(GetLatestSnapshot());
! 		crosscheck_snapshot = CopySnapshot(GetTransactionSnapshot());
! 	}
  	else
  	{
! 		test_snapshot = CopySnapshot(GetTransactionSnapshot());
! 		crosscheck_snapshot = InvalidSnapshot;
  	}
  
  	/*
***************
*** 3124,3138 ****
  	SetUserId(RelationGetForm(query_rel)->relowner);
  
  	/* Finally we can run the query. */
! 	spi_result = SPI_execp_current(qplan, vals, nulls,
! 								   useCurrentSnapshot, limit);
  
  	/* Restore UID */
  	SetUserId(save_uid);
  
  	/* Check result */
  	if (spi_result < 0)
! 		elog(ERROR, "SPI_execp_current returned %d", spi_result);
  
  	if (expect_OK >= 0 && spi_result != expect_OK)
  		ri_ReportViolation(qkey, constrname ? constrname : "",
--- 3132,3148 ----
  	SetUserId(RelationGetForm(query_rel)->relowner);
  
  	/* Finally we can run the query. */
! 	spi_result = SPI_execute_snapshot(qplan,
! 									  vals, nulls,
! 									  test_snapshot, crosscheck_snapshot,
! 									  false, limit);
  
  	/* Restore UID */
  	SetUserId(save_uid);
  
  	/* Check result */
  	if (spi_result < 0)
! 		elog(ERROR, "SPI_execute_snapshot returned %d", spi_result);
  
  	if (expect_OK >= 0 && spi_result != expect_OK)
  		ri_ReportViolation(qkey, constrname ? constrname : "",
*** src/backend/utils/adt/ruleutils.c.orig	Wed Sep  1 19:58:38 2004
--- src/backend/utils/adt/ruleutils.c	Sun Sep 12 16:49:46 2004
***************
*** 290,296 ****
  	 */
  	args[0] = ObjectIdGetDatum(ruleoid);
  	nulls[0] = ' ';
! 	spirc = SPI_execp(plan_getrulebyoid, args, nulls, 1);
  	if (spirc != SPI_OK_SELECT)
  		elog(ERROR, "failed to get pg_rewrite tuple for rule %u", ruleoid);
  	if (SPI_processed != 1)
--- 290,296 ----
  	 */
  	args[0] = ObjectIdGetDatum(ruleoid);
  	nulls[0] = ' ';
! 	spirc = SPI_execute_plan(plan_getrulebyoid, args, nulls, true, 1);
  	if (spirc != SPI_OK_SELECT)
  		elog(ERROR, "failed to get pg_rewrite tuple for rule %u", ruleoid);
  	if (SPI_processed != 1)
***************
*** 425,431 ****
  	args[1] = PointerGetDatum(ViewSelectRuleName);
  	nulls[0] = ' ';
  	nulls[1] = ' ';
! 	spirc = SPI_execp(plan_getviewrule, args, nulls, 2);
  	if (spirc != SPI_OK_SELECT)
  		elog(ERROR, "failed to get pg_rewrite tuple for view %u", viewoid);
  	if (SPI_processed != 1)
--- 425,431 ----
  	args[1] = PointerGetDatum(ViewSelectRuleName);
  	nulls[0] = ' ';
  	nulls[1] = ' ';
! 	spirc = SPI_execute_plan(plan_getviewrule, args, nulls, true, 2);
  	if (spirc != SPI_OK_SELECT)
  		elog(ERROR, "failed to get pg_rewrite tuple for view %u", viewoid);
  	if (SPI_processed != 1)
*** src/backend/utils/time/tqual.c.orig	Sun Aug 29 09:55:38 2004
--- src/backend/utils/time/tqual.c	Sat Sep 11 18:59:52 2004
***************
*** 28,45 ****
  #include "utils/tqual.h"
  
  /*
!  * The SnapshotData structs are static to simplify memory allocation
   * (see the hack in GetSnapshotData to avoid repeated malloc/free).
   */
- static SnapshotData QuerySnapshotData;
- static SnapshotData SerializableSnapshotData;
- static SnapshotData CurrentSnapshotData;
  static SnapshotData SnapshotDirtyData;
  
  /* Externally visible pointers to valid snapshots: */
- Snapshot	QuerySnapshot = NULL;
- Snapshot	SerializableSnapshot = NULL;
  Snapshot	SnapshotDirty = &SnapshotDirtyData;
  
  /* These are updated by GetSnapshotData: */
  TransactionId RecentXmin = InvalidTransactionId;
--- 28,51 ----
  #include "utils/tqual.h"
  
  /*
!  * These SnapshotData structs are static to simplify memory allocation
   * (see the hack in GetSnapshotData to avoid repeated malloc/free).
   */
  static SnapshotData SnapshotDirtyData;
+ static SnapshotData SerializableSnapshotData;
+ static SnapshotData LatestSnapshotData;
  
  /* Externally visible pointers to valid snapshots: */
  Snapshot	SnapshotDirty = &SnapshotDirtyData;
+ Snapshot	SerializableSnapshot = NULL;
+ Snapshot	LatestSnapshot = NULL;
+ 
+ /*
+  * This pointer is not maintained by this module, but it's convenient
+  * to declare it here anyway.  Callers typically assign a copy of
+  * GetTransactionSnapshot's result to ActiveSnapshot.
+  */
+ Snapshot	ActiveSnapshot = NULL;
  
  /* These are updated by GetSnapshotData: */
  TransactionId RecentXmin = InvalidTransactionId;
***************
*** 1028,1128 ****
  
  
  /*
!  * SetQuerySnapshot
!  *		Initialize query snapshot for a new query
   *
   * The SerializableSnapshot is the first one taken in a transaction.
   * In serializable mode we just use that one throughout the transaction.
!  * In read-committed mode, we take a new snapshot at the start of each query.
   */
! void
! SetQuerySnapshot(void)
  {
! 	/* 1st call in xaction? */
  	if (SerializableSnapshot == NULL)
  	{
  		SerializableSnapshot = GetSnapshotData(&SerializableSnapshotData, true);
! 		QuerySnapshot = SerializableSnapshot;
! 		Assert(QuerySnapshot != NULL);
! 		return;
  	}
  
  	if (IsXactIsoLevelSerializable)
! 		QuerySnapshot = SerializableSnapshot;
! 	else
! 		QuerySnapshot = GetSnapshotData(&QuerySnapshotData, false);
  
! 	Assert(QuerySnapshot != NULL);
  }
  
  /*
!  * CopyQuerySnapshot
!  *		Copy the current query snapshot.
!  *
!  * Copying the snapshot is done so that a query is guaranteed to use a
!  * consistent snapshot for its entire execution life, even if the command
!  * counter is incremented or SetQuerySnapshot() is called while it runs
!  * (as could easily happen, due to triggers etc. executing queries).
!  *
!  * The copy is palloc'd in the current memory context.
   */
  Snapshot
! CopyQuerySnapshot(void)
  {
! 	Snapshot	snapshot;
! 
! 	if (QuerySnapshot == NULL)	/* should be set beforehand */
  		elog(ERROR, "no snapshot has been set");
  
! 	snapshot = (Snapshot) palloc(sizeof(SnapshotData));
! 	memcpy(snapshot, QuerySnapshot, sizeof(SnapshotData));
! 	if (snapshot->xcnt > 0)
! 	{
! 		snapshot->xip = (TransactionId *)
! 			palloc(snapshot->xcnt * sizeof(TransactionId));
! 		memcpy(snapshot->xip, QuerySnapshot->xip,
! 			   snapshot->xcnt * sizeof(TransactionId));
! 	}
! 	else
! 		snapshot->xip = NULL;
  
! 	return snapshot;
  }
  
  /*
!  * CopyCurrentSnapshot
!  *		Make a snapshot that is up-to-date as of the current instant,
!  *		and return a copy.
   *
   * The copy is palloc'd in the current memory context.
   */
  Snapshot
! CopyCurrentSnapshot(void)
  {
! 	Snapshot	currentSnapshot;
! 	Snapshot	snapshot;
! 
! 	if (QuerySnapshot == NULL)	/* should not be first call in xact */
! 		elog(ERROR, "no snapshot has been set");
  
! 	/* Update the static struct */
! 	currentSnapshot = GetSnapshotData(&CurrentSnapshotData, false);
! 	currentSnapshot->curcid = GetCurrentCommandId();
! 
! 	/* Make a copy */
! 	snapshot = (Snapshot) palloc(sizeof(SnapshotData));
! 	memcpy(snapshot, currentSnapshot, sizeof(SnapshotData));
  	if (snapshot->xcnt > 0)
  	{
! 		snapshot->xip = (TransactionId *)
! 			palloc(snapshot->xcnt * sizeof(TransactionId));
! 		memcpy(snapshot->xip, currentSnapshot->xip,
  			   snapshot->xcnt * sizeof(TransactionId));
  	}
  	else
! 		snapshot->xip = NULL;
  
! 	return snapshot;
  }
  
  /*
--- 1034,1127 ----
  
  
  /*
!  * GetTransactionSnapshot
!  *		Get the appropriate snapshot for a new query in a transaction.
   *
   * The SerializableSnapshot is the first one taken in a transaction.
   * In serializable mode we just use that one throughout the transaction.
!  * In read-committed mode, we take a new snapshot each time we are called.
!  *
!  * Note that the return value points at static storage that will be modified
!  * by future calls and by CommandCounterIncrement().  Callers should copy
!  * the result with CopySnapshot() if it is to be used very long.
   */
! Snapshot
! GetTransactionSnapshot(void)
  {
! 	/* First call in transaction? */
  	if (SerializableSnapshot == NULL)
  	{
  		SerializableSnapshot = GetSnapshotData(&SerializableSnapshotData, true);
! 		return SerializableSnapshot;
  	}
  
  	if (IsXactIsoLevelSerializable)
! 		return SerializableSnapshot;
  
! 	LatestSnapshot = GetSnapshotData(&LatestSnapshotData, false);
! 
! 	return LatestSnapshot;
  }
  
  /*
!  * GetLatestSnapshot
!  *		Get a snapshot that is up-to-date as of the current instant,
!  *		even if we are executing in SERIALIZABLE mode.
   */
  Snapshot
! GetLatestSnapshot(void)
  {
! 	/* Should not be first call in transaction */
! 	if (SerializableSnapshot == NULL)
  		elog(ERROR, "no snapshot has been set");
  
! 	LatestSnapshot = GetSnapshotData(&LatestSnapshotData, false);
  
! 	return LatestSnapshot;
  }
  
  /*
!  * CopySnapshot
!  *		Copy the given snapshot.
   *
   * The copy is palloc'd in the current memory context.
+  *
+  * Note that this will not work on "special" snapshots.
   */
  Snapshot
! CopySnapshot(Snapshot snapshot)
  {
! 	Snapshot	newsnap;
  
! 	/* We allocate any XID array needed in the same palloc block. */
! 	newsnap = (Snapshot) palloc(sizeof(SnapshotData) +
! 								snapshot->xcnt * sizeof(TransactionId));
! 	memcpy(newsnap, snapshot, sizeof(SnapshotData));
  	if (snapshot->xcnt > 0)
  	{
! 		newsnap->xip = (TransactionId *) (newsnap + 1);
! 		memcpy(newsnap->xip, snapshot->xip,
  			   snapshot->xcnt * sizeof(TransactionId));
  	}
  	else
! 		newsnap->xip = NULL;
  
! 	return newsnap;
! }
! 
! /*
!  * FreeSnapshot
!  *		Free a snapshot previously copied with CopySnapshot.
!  *
!  * This is currently identical to pfree, but is provided for cleanliness.
!  *
!  * Do *not* apply this to the results of GetTransactionSnapshot or
!  * GetLatestSnapshot.
!  */
! void
! FreeSnapshot(Snapshot snapshot)
! {
! 	pfree(snapshot);
  }
  
  /*
***************
*** 1133,1142 ****
  FreeXactSnapshot(void)
  {
  	/*
! 	 * We do not free the xip arrays for the snapshot structs; they will
! 	 * be reused soon.	So this is now just a state change to prevent
  	 * outside callers from accessing the snapshots.
  	 */
- 	QuerySnapshot = NULL;
  	SerializableSnapshot = NULL;
  }
--- 1132,1142 ----
  FreeXactSnapshot(void)
  {
  	/*
! 	 * We do not free the xip arrays for the static snapshot structs; they
! 	 * will be reused soon. So this is now just a state change to prevent
  	 * outside callers from accessing the snapshots.
  	 */
  	SerializableSnapshot = NULL;
+ 	LatestSnapshot = NULL;
+ 	ActiveSnapshot = NULL;		/* just for cleanliness */
  }
*** src/include/executor/execdesc.h.orig	Sun Aug 29 09:56:01 2004
--- src/include/executor/execdesc.h	Sat Sep 11 18:59:37 2004
***************
*** 33,38 ****
--- 33,40 ----
  	CmdType		operation;		/* CMD_SELECT, CMD_UPDATE, etc. */
  	Query	   *parsetree;		/* rewritten parsetree */
  	Plan	   *plantree;		/* planner's output */
+ 	Snapshot	snapshot;		/* snapshot to use for query */
+ 	Snapshot	crosscheck_snapshot;	/* crosscheck for RI update/delete */
  	DestReceiver *dest;			/* the destination for tuple output */
  	ParamListInfo params;		/* param values being passed in */
  	bool		doInstrument;	/* TRUE requests runtime instrumentation */
***************
*** 45,50 ****
--- 47,54 ----
  
  /* in pquery.c */
  extern QueryDesc *CreateQueryDesc(Query *parsetree, Plan *plantree,
+ 				Snapshot snapshot,
+ 				Snapshot crosscheck_snapshot,
  				DestReceiver *dest,
  				ParamListInfo params,
  				bool doInstrument);
*** src/include/executor/executor.h.orig	Sun Aug 29 09:56:01 2004
--- src/include/executor/executor.h	Sat Sep 11 19:12:22 2004
***************
*** 95,102 ****
  /*
   * prototypes from functions in execMain.c
   */
! extern void ExecutorStart(QueryDesc *queryDesc, bool useCurrentSnapshot,
! 			  bool explainOnly);
  extern TupleTableSlot *ExecutorRun(QueryDesc *queryDesc,
  			ScanDirection direction, long count);
  extern void ExecutorEnd(QueryDesc *queryDesc);
--- 95,101 ----
  /*
   * prototypes from functions in execMain.c
   */
! extern void ExecutorStart(QueryDesc *queryDesc, bool explainOnly);
  extern TupleTableSlot *ExecutorRun(QueryDesc *queryDesc,
  			ScanDirection direction, long count);
  extern void ExecutorEnd(QueryDesc *queryDesc);
*** src/include/executor/spi.h.orig	Sun Aug 29 09:56:02 2004
--- src/include/executor/spi.h	Sun Sep 12 16:49:38 2004
***************
*** 81,91 ****
  extern int	SPI_finish(void);
  extern void SPI_push(void);
  extern void SPI_pop(void);
  extern int	SPI_exec(const char *src, int tcount);
! extern int SPI_execp(void *plan, Datum *values, const char *Nulls,
! 		  int tcount);
! extern int SPI_execp_current(void *plan, Datum *values, const char *Nulls,
! 				  bool useCurrentSnapshot, int tcount);
  extern void *SPI_prepare(const char *src, int nargs, Oid *argtypes);
  extern void *SPI_saveplan(void *plan);
  extern int	SPI_freeplan(void *plan);
--- 81,97 ----
  extern int	SPI_finish(void);
  extern void SPI_push(void);
  extern void SPI_pop(void);
+ extern int	SPI_execute(const char *src, bool read_only, int tcount);
+ extern int	SPI_execute_plan(void *plan, Datum *Values, const char *Nulls,
+ 							 bool read_only, int tcount);
  extern int	SPI_exec(const char *src, int tcount);
! extern int	SPI_execp(void *plan, Datum *Values, const char *Nulls,
! 					  int tcount);
! extern int	SPI_execute_snapshot(void *plan,
! 								 Datum *Values, const char *Nulls,
! 								 Snapshot snapshot,
! 								 Snapshot crosscheck_snapshot,
! 								 bool read_only, int tcount);
  extern void *SPI_prepare(const char *src, int nargs, Oid *argtypes);
  extern void *SPI_saveplan(void *plan);
  extern int	SPI_freeplan(void *plan);
*** src/include/tcop/pquery.h.orig	Sun Aug 29 09:56:11 2004
--- src/include/tcop/pquery.h	Sat Sep 11 18:59:32 2004
***************
*** 20,34 ****
  extern DLLIMPORT Portal ActivePortal;
  
  
- extern void ProcessQuery(Query *parsetree,
- 			 Plan *plan,
- 			 ParamListInfo params,
- 			 DestReceiver *dest,
- 			 char *completionTag);
- 
  extern PortalStrategy ChoosePortalStrategy(List *parseTrees);
  
! extern void PortalStart(Portal portal, ParamListInfo params);
  
  extern void PortalSetResultFormat(Portal portal, int nFormats,
  					  int16 *formats);
--- 20,29 ----
  extern DLLIMPORT Portal ActivePortal;
  
  
  extern PortalStrategy ChoosePortalStrategy(List *parseTrees);
  
! extern void PortalStart(Portal portal, ParamListInfo params,
! 						Snapshot snapshot);
  
  extern void PortalSetResultFormat(Portal portal, int nFormats,
  					  int16 *formats);
*** src/include/tcop/utility.h.orig	Sun Aug 29 09:56:11 2004
--- src/include/tcop/utility.h	Sat Sep 11 22:09:07 2004
***************
*** 26,31 ****
--- 26,35 ----
  
  extern const char *CreateCommandTag(Node *parsetree);
  
+ extern const char *CreateQueryTag(Query *parsetree);
+ 
+ extern bool QueryIsReadOnly(Query *parsetree);
+ 
  extern void CheckRelationOwnership(RangeVar *rel, bool noCatalogs);
  
  #endif   /* UTILITY_H */
*** src/include/utils/tqual.h.orig	Sat Sep 11 14:28:34 2004
--- src/include/utils/tqual.h	Sat Sep 11 18:59:27 2004
***************
*** 54,61 ****
  #define SnapshotToast				((Snapshot) 0x4)
  
  extern DLLIMPORT Snapshot SnapshotDirty;
! extern DLLIMPORT Snapshot QuerySnapshot;
  extern DLLIMPORT Snapshot SerializableSnapshot;
  
  extern TransactionId RecentXmin;
  extern TransactionId RecentGlobalXmin;
--- 54,63 ----
  #define SnapshotToast				((Snapshot) 0x4)
  
  extern DLLIMPORT Snapshot SnapshotDirty;
! 
  extern DLLIMPORT Snapshot SerializableSnapshot;
+ extern DLLIMPORT Snapshot LatestSnapshot;
+ extern DLLIMPORT Snapshot ActiveSnapshot;
  
  extern TransactionId RecentXmin;
  extern TransactionId RecentGlobalXmin;
***************
*** 121,130 ****
  extern HTSV_Result HeapTupleSatisfiesVacuum(HeapTupleHeader tuple,
  						 TransactionId OldestXmin);
  
! extern Snapshot GetSnapshotData(Snapshot snapshot, bool serializable);
! extern void SetQuerySnapshot(void);
! extern Snapshot CopyQuerySnapshot(void);
! extern Snapshot CopyCurrentSnapshot(void);
  extern void FreeXactSnapshot(void);
  
  #endif   /* TQUAL_H */
--- 123,135 ----
  extern HTSV_Result HeapTupleSatisfiesVacuum(HeapTupleHeader tuple,
  						 TransactionId OldestXmin);
  
! extern Snapshot GetTransactionSnapshot(void);
! extern Snapshot GetLatestSnapshot(void);
! extern Snapshot CopySnapshot(Snapshot snapshot);
! extern void FreeSnapshot(Snapshot snapshot);
  extern void FreeXactSnapshot(void);
+ 
+ /* in sinval.c; declared here to avoid including tqual.h in sinval.h: */
+ extern Snapshot GetSnapshotData(Snapshot snapshot, bool serializable);
  
  #endif   /* TQUAL_H */
*** src/pl/plperl/plperl.c.orig	Sun Aug 29 22:58:08 2004
--- src/pl/plperl/plperl.c	Sun Sep 12 18:18:38 2004
***************
*** 53,58 ****
--- 53,59 ----
  #include "executor/spi.h"
  #include "fmgr.h"
  #include "tcop/tcopprot.h"
+ #include "utils/lsyscache.h"
  #include "utils/syscache.h"
  #include "utils/typcache.h"
  
***************
*** 77,82 ****
--- 78,84 ----
  	char	   *proname;
  	TransactionId fn_xmin;
  	CommandId	fn_cmin;
+ 	bool		fn_readonly;
  	bool		lanpltrusted;
  	bool		fn_retistuple;	/* true, if function returns tuple */
  	bool		fn_retisset;	/* true, if function returns set */
***************
*** 98,108 ****
  static bool plperl_safe_init_done = false;
  static PerlInterpreter *plperl_interp = NULL;
  static HV  *plperl_proc_hash = NULL;
- static AV  *g_row_keys = NULL;
  static AV  *g_column_keys = NULL;
  static SV  *srf_perlret = NULL; /* keep returned value */
  static int	g_attr_num = 0;
  
  /**********************************************************************
   * Forward declarations
   **********************************************************************/
--- 100,112 ----
  static bool plperl_safe_init_done = false;
  static PerlInterpreter *plperl_interp = NULL;
  static HV  *plperl_proc_hash = NULL;
  static AV  *g_column_keys = NULL;
  static SV  *srf_perlret = NULL; /* keep returned value */
  static int	g_attr_num = 0;
  
+ /* this is saved and restored by plperl_call_handler */
+ static plperl_proc_desc *plperl_current_prodesc = NULL;
+ 
  /**********************************************************************
   * Forward declarations
   **********************************************************************/
***************
*** 119,124 ****
--- 123,129 ----
  
  static SV  *plperl_build_tuple_argument(HeapTuple tuple, TupleDesc tupdesc);
  static void plperl_init_shared_libs(pTHX);
+ static HV  *plperl_spi_execute_fetch_result(SPITupleTable *, int, int);
  
  
  /*
***************
*** 435,441 ****
  plperl_get_keys(HV *hv)
  {
  	AV		   *ret;
- 	SV		  **svp;
  	int			key_count;
  	SV		   *val;
  	char	   *key;
--- 440,445 ----
***************
*** 445,451 ****
  	ret = newAV();
  
  	hv_iterinit(hv);
! 	while (val = hv_iternextsv(hv, (char **) &key, &klen))
  	{
  		av_store(ret, key_count, eval_pv(key, TRUE));
  		key_count++;
--- 449,455 ----
  	ret = newAV();
  
  	hv_iterinit(hv);
! 	while ((val = hv_iternextsv(hv, (char **) &key, &klen)))
  	{
  		av_store(ret, key_count, eval_pv(key, TRUE));
  		key_count++;
***************
*** 592,617 ****
  plperl_call_handler(PG_FUNCTION_ARGS)
  {
  	Datum		retval;
  
! 	/************************************************************
! 	 * Initialize interpreter
! 	 ************************************************************/
  	plperl_init_all();
  
! 	/************************************************************
! 	 * Connect to SPI manager
! 	 ************************************************************/
! 	if (SPI_connect() != SPI_OK_CONNECT)
! 		elog(ERROR, "could not connect to SPI manager");
  
! 	/************************************************************
! 	 * Determine if called as function or trigger and
! 	 * call appropriate subhandler
! 	 ************************************************************/
! 	if (CALLED_AS_TRIGGER(fcinfo))
! 		retval = PointerGetDatum(plperl_trigger_handler(fcinfo));
! 	else
! 		retval = plperl_func_handler(fcinfo);
  
  	return retval;
  }
--- 596,638 ----
  plperl_call_handler(PG_FUNCTION_ARGS)
  {
  	Datum		retval;
+ 	plperl_proc_desc *save_prodesc;
  
! 	/*
! 	 * Initialize interpreter if first time through
! 	 */
  	plperl_init_all();
  
! 	/*
! 	 * Ensure that static pointers are saved/restored properly
! 	 */
! 	save_prodesc = plperl_current_prodesc;
  
! 	PG_TRY();
! 	{
! 		/************************************************************
! 		 * Connect to SPI manager
! 		 ************************************************************/
! 		if (SPI_connect() != SPI_OK_CONNECT)
! 			elog(ERROR, "could not connect to SPI manager");
! 
! 		/************************************************************
! 		 * Determine if called as function or trigger and
! 		 * call appropriate subhandler
! 		 ************************************************************/
! 		if (CALLED_AS_TRIGGER(fcinfo))
! 			retval = PointerGetDatum(plperl_trigger_handler(fcinfo));
! 		else
! 			retval = plperl_func_handler(fcinfo);
! 	}
! 	PG_CATCH();
! 	{
! 		plperl_current_prodesc = save_prodesc;
! 		PG_RE_THROW();
! 	}
! 	PG_END_TRY();
! 
! 	plperl_current_prodesc = save_prodesc;
  
  	return retval;
  }
***************
*** 821,827 ****
  	SV		   *retval;
  	int			i;
  	int			count;
- 	char	   *ret_test;
  
  	ENTER;
  	SAVETMPS;
--- 842,847 ----
***************
*** 874,879 ****
--- 894,902 ----
  
  	/* Find or compile the function */
  	prodesc = compile_plperl_function(fcinfo->flinfo->fn_oid, false);
+ 
+ 	plperl_current_prodesc = prodesc;
+ 
  	/************************************************************
  	 * Call the Perl function if not returning set
  	 ************************************************************/
***************
*** 1002,1008 ****
  			{
  				HV		   *row_hv;
  				SV		  **svp;
- 				char	   *row_key;
  
  				svp = av_fetch(ret_av, call_cntr, FALSE);
  
--- 1025,1030 ----
***************
*** 1052,1058 ****
  		if (SRF_IS_FIRSTCALL())
  		{
  			MemoryContext oldcontext;
- 			int			i;
  
  			funcctx = SRF_FIRSTCALL_INIT();
  			oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
--- 1074,1079 ----
***************
*** 1067,1073 ****
  			Datum		result;
  			AV		   *array;
  			SV		  **svp;
- 			int			i;
  
  			array = (AV *) SvRV(perlret);
  			svp = av_fetch(array, funcctx->call_cntr, FALSE);
--- 1088,1093 ----
***************
*** 1158,1163 ****
--- 1178,1185 ----
  	/* Find or compile the function */
  	prodesc = compile_plperl_function(fcinfo->flinfo->fn_oid, true);
  
+ 	plperl_current_prodesc = prodesc;
+ 
  	/************************************************************
  	* Call the Perl function
  	************************************************************/
***************
*** 1323,1328 ****
--- 1345,1354 ----
  		prodesc->fn_xmin = HeapTupleHeaderGetXmin(procTup->t_data);
  		prodesc->fn_cmin = HeapTupleHeaderGetCmin(procTup->t_data);
  
+ 		/* Remember if function is STABLE/IMMUTABLE */
+ 		prodesc->fn_readonly =
+ 			(procStruct->provolatile != PROVOLATILE_VOLATILE);
+ 
  		/************************************************************
  		 * Lookup the pg_language tuple by Oid
  		 ************************************************************/
***************
*** 1559,1562 ****
--- 1585,1667 ----
  	sv_catpv(output, "}");
  	output = perl_eval_pv(SvPV(output, PL_na), TRUE);
  	return output;
+ }
+ 
+ 
+ HV *
+ plperl_spi_exec(char *query, int limit)
+ {
+ 	HV		   *ret_hv;
+ 	int			spi_rv;
+ 
+ 	spi_rv = SPI_execute(query, plperl_current_prodesc->fn_readonly, limit);
+ 	ret_hv = plperl_spi_execute_fetch_result(SPI_tuptable, SPI_processed, spi_rv);
+ 
+ 	return ret_hv;
+ }
+ 
+ static HV  *
+ plperl_hash_from_tuple(HeapTuple tuple, TupleDesc tupdesc)
+ {
+ 	int			i;
+ 	char	   *attname;
+ 	char	   *attdata;
+ 
+ 	HV		   *array;
+ 
+ 	array = newHV();
+ 
+ 	for (i = 0; i < tupdesc->natts; i++)
+ 	{
+ 		/************************************************************
+ 		* Get the attribute name
+ 		************************************************************/
+ 		attname = tupdesc->attrs[i]->attname.data;
+ 
+ 		/************************************************************
+ 		* Get the attributes value
+ 		************************************************************/
+ 		attdata = SPI_getvalue(tuple, tupdesc, i + 1);
+ 		if (attdata)
+ 			hv_store(array, attname, strlen(attname), newSVpv(attdata, 0), 0);
+ 		else
+ 			hv_store(array, attname, strlen(attname), newSVpv("undef", 0), 0);
+ 	}
+ 	return array;
+ }
+ 
+ static HV  *
+ plperl_spi_execute_fetch_result(SPITupleTable *tuptable, int processed, int status)
+ {
+ 	HV		   *result;
+ 
+ 	result = newHV();
+ 
+ 	hv_store(result, "status", strlen("status"),
+ 			 newSVpv((char *) SPI_result_code_string(status), 0), 0);
+ 	hv_store(result, "processed", strlen("processed"),
+ 			 newSViv(processed), 0);
+ 
+ 	if (status == SPI_OK_SELECT)
+ 	{
+ 		if (processed)
+ 		{
+ 			AV		   *rows;
+ 			HV		   *row;
+ 			int			i;
+ 
+ 			rows = newAV();
+ 			for (i = 0; i < processed; i++)
+ 			{
+ 				row = plperl_hash_from_tuple(tuptable->vals[i], tuptable->tupdesc);
+ 				av_store(rows, i, newRV_noinc((SV *) row));
+ 			}
+ 			hv_store(result, "rows", strlen("rows"),
+ 					 newRV_noinc((SV *) rows), 0);
+ 		}
+ 	}
+ 
+ 	SPI_freetuptable(tuptable);
+ 
+ 	return result;
  }
*** src/pl/plperl/spi_internal.c.orig	Sun Aug 29 09:56:35 2004
--- src/pl/plperl/spi_internal.c	Sun Sep 12 18:18:38 2004
***************
*** 1,15 ****
- #include "postgres.h"
- #include "executor/spi.h"
- #include "utils/syscache.h"
  /*
   * This kludge is necessary because of the conflicting
   * definitions of 'DEBUG' between postgres and perl.
   * we'll live.
   */
  
! #include "spi_internal.h"
  
! static HV  *plperl_spi_execute_fetch_result(SPITupleTable *, int, int);
  
  
  int
--- 1,12 ----
  /*
   * This kludge is necessary because of the conflicting
   * definitions of 'DEBUG' between postgres and perl.
   * we'll live.
   */
  
! #include "postgres.h"
  
! #include "spi_internal.h"
  
  
  int
***************
*** 46,127 ****
  spi_ERROR(void)
  {
  	return ERROR;
- }
- 
- HV *
- plperl_spi_exec(char *query, int limit)
- {
- 	HV		   *ret_hv;
- 	int			spi_rv;
- 
- 	spi_rv = SPI_exec(query, limit);
- 	ret_hv = plperl_spi_execute_fetch_result(SPI_tuptable, SPI_processed, spi_rv);
- 
- 	return ret_hv;
- }
- 
- static HV  *
- plperl_hash_from_tuple(HeapTuple tuple, TupleDesc tupdesc)
- {
- 	int			i;
- 	char	   *attname;
- 	char	   *attdata;
- 
- 	HV		   *array;
- 
- 	array = newHV();
- 
- 	for (i = 0; i < tupdesc->natts; i++)
- 	{
- 		/************************************************************
- 		* Get the attribute name
- 		************************************************************/
- 		attname = tupdesc->attrs[i]->attname.data;
- 
- 		/************************************************************
- 		* Get the attributes value
- 		************************************************************/
- 		attdata = SPI_getvalue(tuple, tupdesc, i + 1);
- 		if (attdata)
- 			hv_store(array, attname, strlen(attname), newSVpv(attdata, 0), 0);
- 		else
- 			hv_store(array, attname, strlen(attname), newSVpv("undef", 0), 0);
- 	}
- 	return array;
- }
- 
- static HV  *
- plperl_spi_execute_fetch_result(SPITupleTable *tuptable, int processed, int status)
- {
- 	HV		   *result;
- 
- 	result = newHV();
- 
- 	hv_store(result, "status", strlen("status"),
- 			 newSVpv((char *) SPI_result_code_string(status), 0), 0);
- 	hv_store(result, "processed", strlen("processed"),
- 			 newSViv(processed), 0);
- 
- 	if (status == SPI_OK_SELECT)
- 	{
- 		if (processed)
- 		{
- 			AV		   *rows;
- 			HV		   *row;
- 			int			i;
- 
- 			rows = newAV();
- 			for (i = 0; i < processed; i++)
- 			{
- 				row = plperl_hash_from_tuple(tuptable->vals[i], tuptable->tupdesc);
- 				av_store(rows, i, newRV_noinc((SV *) row));
- 			}
- 			hv_store(result, "rows", strlen("rows"),
- 					 newRV_noinc((SV *) rows), 0);
- 		}
- 	}
- 
- 	SPI_freetuptable(tuptable);
- 
- 	return result;
  }
--- 43,46 ----
*** src/pl/plperl/spi_internal.h.orig	Sun Aug 29 09:56:35 2004
--- src/pl/plperl/spi_internal.h	Sun Sep 12 18:18:39 2004
***************
*** 15,18 ****
--- 15,19 ----
  
  int			spi_ERROR(void);
  
+ /* this is actually in plperl.c */
  HV		   *plperl_spi_exec(char *, int);
*** src/pl/plpgsql/src/pl_comp.c.orig	Sun Aug 29 22:58:09 2004
--- src/pl/plpgsql/src/pl_comp.c	Sun Sep 12 17:04:30 2004
***************
*** 578,583 ****
--- 578,586 ----
  			break;
  	}
  
+ 	/* Remember if function is STABLE/IMMUTABLE */
+ 	function->fn_readonly = (procStruct->provolatile != PROVOLATILE_VOLATILE);
+ 
  	/*
  	 * Create the magic FOUND variable.
  	 */
*** src/pl/plpgsql/src/pl_exec.c.orig	Sun Aug 29 22:58:09 2004
--- src/pl/plpgsql/src/pl_exec.c	Sun Sep 12 17:04:30 2004
***************
*** 897,902 ****
--- 897,903 ----
  		 * sub-transaction
  		 */
  		MemoryContext oldcontext = CurrentMemoryContext;
+ 		ResourceOwner oldowner = CurrentResourceOwner;
  		volatile bool caught = false;
  		int			xrc;
  
***************
*** 907,918 ****
  		BeginInternalSubTransaction(NULL);
  		/* Want to run statements inside function's memory context */
  		MemoryContextSwitchTo(oldcontext);
  		if ((xrc = SPI_connect()) != SPI_OK_CONNECT)
  			elog(ERROR, "SPI_connect failed: %s",
  				 SPI_result_code_string(xrc));
  
  		PG_TRY();
! 		rc = exec_stmts(estate, block->body);
  		PG_CATCH();
  		{
  			ErrorData  *edata;
--- 908,922 ----
  		BeginInternalSubTransaction(NULL);
  		/* Want to run statements inside function's memory context */
  		MemoryContextSwitchTo(oldcontext);
+ 
  		if ((xrc = SPI_connect()) != SPI_OK_CONNECT)
  			elog(ERROR, "SPI_connect failed: %s",
  				 SPI_result_code_string(xrc));
  
  		PG_TRY();
! 		{
! 			rc = exec_stmts(estate, block->body);
! 		}
  		PG_CATCH();
  		{
  			ErrorData  *edata;
***************
*** 927,932 ****
--- 931,937 ----
  			/* Abort the inner transaction (and inner SPI connection) */
  			RollbackAndReleaseCurrentSubTransaction();
  			MemoryContextSwitchTo(oldcontext);
+ 			CurrentResourceOwner = oldowner;
  
  			SPI_pop();
  
***************
*** 958,965 ****
--- 963,973 ----
  			if ((xrc = SPI_finish()) != SPI_OK_FINISH)
  				elog(ERROR, "SPI_finish failed: %s",
  					 SPI_result_code_string(xrc));
+ 
  			ReleaseCurrentSubTransaction();
  			MemoryContextSwitchTo(oldcontext);
+ 			CurrentResourceOwner = oldowner;
+ 
  			SPI_pop();
  		}
  	}
***************
*** 1984,1989 ****
--- 1992,1999 ----
  	estate->retistuple = func->fn_retistuple;
  	estate->retisset = func->fn_retset;
  
+ 	estate->readonly_func = func->fn_readonly;
+ 
  	estate->rettupdesc = NULL;
  	estate->exitlabel = NULL;
  
***************
*** 2019,2025 ****
  static void
  exec_eval_cleanup(PLpgSQL_execstate *estate)
  {
! 	/* Clear result of a full SPI_exec */
  	if (estate->eval_tuptable != NULL)
  		SPI_freetuptable(estate->eval_tuptable);
  	estate->eval_tuptable = NULL;
--- 2029,2035 ----
  static void
  exec_eval_cleanup(PLpgSQL_execstate *estate)
  {
! 	/* Clear result of a full SPI_execute */
  	if (estate->eval_tuptable != NULL)
  		SPI_freetuptable(estate->eval_tuptable);
  	estate->eval_tuptable = NULL;
***************
*** 2120,2126 ****
  		exec_prepare_plan(estate, expr);
  
  	/*
! 	 * Now build up the values and nulls arguments for SPI_execp()
  	 */
  	values = (Datum *) palloc(expr->nparams * sizeof(Datum));
  	nulls = (char *) palloc(expr->nparams * sizeof(char));
--- 2130,2136 ----
  		exec_prepare_plan(estate, expr);
  
  	/*
! 	 * Now build up the values and nulls arguments for SPI_execute_plan()
  	 */
  	values = (Datum *) palloc(expr->nparams * sizeof(Datum));
  	nulls = (char *) palloc(expr->nparams * sizeof(char));
***************
*** 2142,2148 ****
  	/*
  	 * Execute the plan
  	 */
! 	rc = SPI_execp(expr->plan, values, nulls, 0);
  	switch (rc)
  	{
  		case SPI_OK_UTILITY:
--- 2152,2159 ----
  	/*
  	 * Execute the plan
  	 */
! 	rc = SPI_execute_plan(expr->plan, values, nulls,
! 						  estate->readonly_func, 0);
  	switch (rc)
  	{
  		case SPI_OK_UTILITY:
***************
*** 2168,2179 ****
  					 errhint("If you want to discard the results, use PERFORM instead.")));
  
  		default:
! 			elog(ERROR, "SPI_execp failed executing query \"%s\": %s",
  				 expr->query, SPI_result_code_string(rc));
  	}
  
  	/*
! 	 * Release any result tuples from SPI_execp (probably shouldn't be
  	 * any)
  	 */
  	SPI_freetuptable(SPI_tuptable);
--- 2179,2190 ----
  					 errhint("If you want to discard the results, use PERFORM instead.")));
  
  		default:
! 			elog(ERROR, "SPI_execute_plan failed executing query \"%s\": %s",
  				 expr->query, SPI_result_code_string(rc));
  	}
  
  	/*
! 	 * Release any result tuples from SPI_execute_plan (probably shouldn't be
  	 * any)
  	 */
  	SPI_freetuptable(SPI_tuptable);
***************
*** 2220,2230 ****
  	exec_eval_cleanup(estate);
  
  	/*
! 	 * Call SPI_exec() without preparing a saved plan. The returncode can
  	 * be any standard OK.	Note that while a SELECT is allowed, its
  	 * results will be discarded.
  	 */
! 	exec_res = SPI_exec(querystr, 0);
  	switch (exec_res)
  	{
  		case SPI_OK_SELECT:
--- 2231,2241 ----
  	exec_eval_cleanup(estate);
  
  	/*
! 	 * Call SPI_execute() without preparing a saved plan. The returncode can
  	 * be any standard OK.	Note that while a SELECT is allowed, its
  	 * results will be discarded.
  	 */
! 	exec_res = SPI_execute(querystr, estate->readonly_func, 0);
  	switch (exec_res)
  	{
  		case SPI_OK_SELECT:
***************
*** 2249,2255 ****
  			 * behavior is not consistent with SELECT INTO in a normal
  			 * plpgsql context. (We need to reimplement EXECUTE to parse
  			 * the string as a plpgsql command, not just feed it to
! 			 * SPI_exec.) However, CREATE AS should be allowed ... and
  			 * since it produces the same parsetree as SELECT INTO,
  			 * there's no way to tell the difference except to look at the
  			 * source text.  Wotta kluge!
--- 2260,2266 ----
  			 * behavior is not consistent with SELECT INTO in a normal
  			 * plpgsql context. (We need to reimplement EXECUTE to parse
  			 * the string as a plpgsql command, not just feed it to
! 			 * SPI_execute.) However, CREATE AS should be allowed ... and
  			 * since it produces the same parsetree as SELECT INTO,
  			 * there's no way to tell the difference except to look at the
  			 * source text.  Wotta kluge!
***************
*** 2284,2295 ****
  					 errhint("Use a BEGIN block with an EXCEPTION clause instead.")));
  
  		default:
! 			elog(ERROR, "SPI_exec failed executing query \"%s\": %s",
  				 querystr, SPI_result_code_string(exec_res));
  			break;
  	}
  
! 	/* Release any result from SPI_exec, as well as the querystring */
  	SPI_freetuptable(SPI_tuptable);
  	pfree(querystr);
  
--- 2295,2306 ----
  					 errhint("Use a BEGIN block with an EXCEPTION clause instead.")));
  
  		default:
! 			elog(ERROR, "SPI_execute failed executing query \"%s\": %s",
  				 querystr, SPI_result_code_string(exec_res));
  			break;
  	}
  
! 	/* Release any result from SPI_execute, as well as the querystring */
  	SPI_freetuptable(SPI_tuptable);
  	pfree(querystr);
  
***************
*** 3470,3476 ****
  		exec_prepare_plan(estate, expr);
  
  	/*
! 	 * Now build up the values and nulls arguments for SPI_execp()
  	 */
  	values = (Datum *) palloc(expr->nparams * sizeof(Datum));
  	nulls = (char *) palloc(expr->nparams * sizeof(char));
--- 3481,3487 ----
  		exec_prepare_plan(estate, expr);
  
  	/*
! 	 * Now build up the values and nulls arguments for SPI_execute_plan()
  	 */
  	values = (Datum *) palloc(expr->nparams * sizeof(Datum));
  	nulls = (char *) palloc(expr->nparams * sizeof(char));
***************
*** 3506,3512 ****
  	/*
  	 * Execute the query
  	 */
! 	rc = SPI_execp(expr->plan, values, nulls, maxtuples);
  	if (rc != SPI_OK_SELECT)
  		ereport(ERROR,
  				(errcode(ERRCODE_SYNTAX_ERROR),
--- 3517,3524 ----
  	/*
  	 * Execute the query
  	 */
! 	rc = SPI_execute_plan(expr->plan, values, nulls,
! 						  estate->readonly_func, maxtuples);
  	if (rc != SPI_OK_SELECT)
  		ereport(ERROR,
  				(errcode(ERRCODE_SYNTAX_ERROR),
*** src/pl/plpgsql/src/plpgsql.h.orig	Sun Aug 29 22:58:09 2004
--- src/pl/plpgsql/src/plpgsql.h	Sun Sep 12 17:04:30 2004
***************
*** 585,590 ****
--- 585,591 ----
  	Oid			fn_rettypioparam;
  	bool		fn_retistuple;
  	bool		fn_retset;
+ 	bool		fn_readonly;
  
  	int			fn_nargs;
  	int			fn_argvarnos[FUNC_MAX_ARGS];
***************
*** 614,619 ****
--- 615,622 ----
  	Oid			fn_rettype;		/* info about declared function rettype */
  	bool		retistuple;
  	bool		retisset;
+ 
+ 	bool		readonly_func;
  
  	TupleDesc	rettupdesc;
  	char	   *exitlabel;
*** src/pl/plpython/plpython.c.orig	Sun Aug 29 22:58:10 2004
--- src/pl/plpython/plpython.c	Sun Sep 12 17:57:31 2004
***************
*** 131,136 ****
--- 131,137 ----
  	char	   *pyname;			/* Python name of procedure */
  	TransactionId fn_xmin;
  	CommandId	fn_cmin;
+ 	bool		fn_readonly;
  	PLyTypeInfo result;			/* also used to store info for trigger
  								 * tuple type */
  	PLyTypeInfo args[FUNC_MAX_ARGS];
***************
*** 257,267 ****
  static int	PLy_first_call = 1;
  
  /*
!  * Last function called by postgres backend
!  *
!  * XXX replace this with errcontext mechanism
   */
! static PLyProcedure *PLy_last_procedure = NULL;
  
  /*
   * When a callback from Python into PG incurs an error, we temporarily store
--- 258,266 ----
  static int	PLy_first_call = 1;
  
  /*
!  * Currently active plpython function
   */
! static PLyProcedure *PLy_curr_procedure = NULL;
  
  /*
   * When a callback from Python into PG incurs an error, we temporarily store
***************
*** 322,327 ****
--- 321,327 ----
  plpython_call_handler(PG_FUNCTION_ARGS)
  {
  	Datum		retval;
+ 	PLyProcedure *save_curr_proc;
  	PLyProcedure *volatile proc = NULL;
  
  	PLy_init_all();
***************
*** 329,334 ****
--- 329,336 ----
  	if (SPI_connect() != SPI_OK_CONNECT)
  		elog(ERROR, "could not connect to SPI manager");
  
+ 	save_curr_proc = PLy_curr_procedure;
+ 
  	PG_TRY();
  	{
  		if (CALLED_AS_TRIGGER(fcinfo))
***************
*** 338,354 ****
--- 340,359 ----
  
  			proc = PLy_procedure_get(fcinfo,
  								   RelationGetRelid(tdata->tg_relation));
+ 			PLy_curr_procedure = proc;
  			trv = PLy_trigger_handler(fcinfo, proc);
  			retval = PointerGetDatum(trv);
  		}
  		else
  		{
  			proc = PLy_procedure_get(fcinfo, InvalidOid);
+ 			PLy_curr_procedure = proc;
  			retval = PLy_function_handler(fcinfo, proc);
  		}
  	}
  	PG_CATCH();
  	{
+ 		PLy_curr_procedure = save_curr_proc;
  		if (proc)
  		{
  			/* note: Py_DECREF needs braces around it, as of 2003/08 */
***************
*** 359,364 ****
--- 364,371 ----
  	}
  	PG_END_TRY();
  
+ 	PLy_curr_procedure = save_curr_proc;
+ 
  	Py_DECREF(proc->me);
  
  	return retval;
***************
*** 795,808 ****
  PLy_procedure_call(PLyProcedure * proc, char *kargs, PyObject * vargs)
  {
  	PyObject   *rv;
- 	PLyProcedure *current;
  
- 	current = PLy_last_procedure;
- 	PLy_last_procedure = proc;
  	PyDict_SetItemString(proc->globals, kargs, vargs);
  	rv = PyEval_EvalCode((PyCodeObject *) proc->code,
  						 proc->globals, proc->globals);
- 	PLy_last_procedure = current;
  
  	/*
  	 * If there was an error in a PG callback, propagate that no matter
--- 802,811 ----
***************
*** 1005,1010 ****
--- 1008,1016 ----
  	strcpy(proc->pyname, procName);
  	proc->fn_xmin = HeapTupleHeaderGetXmin(procTup->t_data);
  	proc->fn_cmin = HeapTupleHeaderGetCmin(procTup->t_data);
+ 	/* Remember if function is STABLE/IMMUTABLE */
+ 	proc->fn_readonly =
+ 		(procStruct->provolatile != PROVOLATILE_VOLATILE);
  	PLy_typeinfo_init(&proc->result);
  	for (i = 0; i < FUNC_MAX_ARGS; i++)
  		PLy_typeinfo_init(&proc->args[i]);
***************
*** 1935,1941 ****
  			PyErr_SetString(PLy_exc_spi_error,
  							"Unknown error in PLy_spi_prepare");
  		/* XXX this oughta be replaced with errcontext mechanism */
! 		PLy_elog(WARNING, "in function %s:", PLy_procedure_name(PLy_last_procedure));
  		return NULL;
  	}
  	PG_END_TRY();
--- 1941,1948 ----
  			PyErr_SetString(PLy_exc_spi_error,
  							"Unknown error in PLy_spi_prepare");
  		/* XXX this oughta be replaced with errcontext mechanism */
! 		PLy_elog(WARNING, "in function %s:",
! 				 PLy_procedure_name(PLy_curr_procedure));
  		return NULL;
  	}
  	PG_END_TRY();
***************
*** 2054,2060 ****
  			}
  		}
  
! 		rv = SPI_execp(plan->plan, plan->values, nulls, limit);
  
  		pfree(nulls);
  	}
--- 2061,2068 ----
  			}
  		}
  
! 		rv = SPI_execute_plan(plan->plan, plan->values, nulls,
! 							  PLy_curr_procedure->fn_readonly, limit);
  
  		pfree(nulls);
  	}
***************
*** 2080,2086 ****
  		if (!PyErr_Occurred())
  			PyErr_SetString(PLy_exc_error,
  							"Unknown error in PLy_spi_execute_plan");
! 		PLy_elog(WARNING, "in function %s:", PLy_procedure_name(PLy_last_procedure));
  		return NULL;
  	}
  	PG_END_TRY();
--- 2088,2096 ----
  		if (!PyErr_Occurred())
  			PyErr_SetString(PLy_exc_error,
  							"Unknown error in PLy_spi_execute_plan");
! 		/* XXX this oughta be replaced with errcontext mechanism */
! 		PLy_elog(WARNING, "in function %s:",
! 				 PLy_procedure_name(PLy_curr_procedure));
  		return NULL;
  	}
  	PG_END_TRY();
***************
*** 2098,2104 ****
  	if (rv < 0)
  	{
  		PLy_exception_set(PLy_exc_spi_error,
! 						  "SPI_execp failed: %s",
  						  SPI_result_code_string(rv));
  		return NULL;
  	}
--- 2108,2114 ----
  	if (rv < 0)
  	{
  		PLy_exception_set(PLy_exc_spi_error,
! 						  "SPI_execute_plan failed: %s",
  						  SPI_result_code_string(rv));
  		return NULL;
  	}
***************
*** 2114,2120 ****
  
  	oldcontext = CurrentMemoryContext;
  	PG_TRY();
! 	rv = SPI_exec(query, limit);
  	PG_CATCH();
  	{
  		MemoryContextSwitchTo(oldcontext);
--- 2124,2132 ----
  
  	oldcontext = CurrentMemoryContext;
  	PG_TRY();
! 	{
! 		rv = SPI_execute(query, PLy_curr_procedure->fn_readonly, limit);
! 	}
  	PG_CATCH();
  	{
  		MemoryContextSwitchTo(oldcontext);
***************
*** 2123,2129 ****
  		if (!PyErr_Occurred())
  			PyErr_SetString(PLy_exc_spi_error,
  							"Unknown error in PLy_spi_execute_query");
! 		PLy_elog(WARNING, "in function %s:", PLy_procedure_name(PLy_last_procedure));
  		return NULL;
  	}
  	PG_END_TRY();
--- 2135,2143 ----
  		if (!PyErr_Occurred())
  			PyErr_SetString(PLy_exc_spi_error,
  							"Unknown error in PLy_spi_execute_query");
! 		/* XXX this oughta be replaced with errcontext mechanism */
! 		PLy_elog(WARNING, "in function %s:",
! 				 PLy_procedure_name(PLy_curr_procedure));
  		return NULL;
  	}
  	PG_END_TRY();
***************
*** 2131,2137 ****
  	if (rv < 0)
  	{
  		PLy_exception_set(PLy_exc_spi_error,
! 						  "SPI_exec failed: %s",
  						  SPI_result_code_string(rv));
  		return NULL;
  	}
--- 2145,2151 ----
  	if (rv < 0)
  	{
  		PLy_exception_set(PLy_exc_spi_error,
! 						  "SPI_execute failed: %s",
  						  SPI_result_code_string(rv));
  		return NULL;
  	}
*** src/pl/tcl/pltcl.c.orig	Sun Aug 29 22:58:11 2004
--- src/pl/tcl/pltcl.c	Sun Sep 12 17:42:16 2004
***************
*** 87,99 ****
  					pfree(_pltcl_utf_dst); } while (0)
  #define UTF_U2E(x)	 (_pltcl_utf_dst=utf_u2e(_pltcl_utf_src=(x)))
  #define UTF_E2U(x)	 (_pltcl_utf_dst=utf_e2u(_pltcl_utf_src=(x)))
! #else							/* PLTCL_UTF */
  #define  UTF_BEGIN
  #define  UTF_END
  #define  UTF_U2E(x)  (x)
  #define  UTF_E2U(x)  (x)
  #endif   /* PLTCL_UTF */
  
  /**********************************************************************
   * The information we cache about loaded procedures
   **********************************************************************/
--- 87,103 ----
  					pfree(_pltcl_utf_dst); } while (0)
  #define UTF_U2E(x)	 (_pltcl_utf_dst=utf_u2e(_pltcl_utf_src=(x)))
  #define UTF_E2U(x)	 (_pltcl_utf_dst=utf_e2u(_pltcl_utf_src=(x)))
! 
! #else							/* !PLTCL_UTF */
! 
  #define  UTF_BEGIN
  #define  UTF_END
  #define  UTF_U2E(x)  (x)
  #define  UTF_E2U(x)  (x)
+ 
  #endif   /* PLTCL_UTF */
  
+ 
  /**********************************************************************
   * The information we cache about loaded procedures
   **********************************************************************/
***************
*** 102,107 ****
--- 106,112 ----
  	char	   *proname;
  	TransactionId fn_xmin;
  	CommandId	fn_cmin;
+ 	bool		fn_readonly;
  	bool		lanpltrusted;
  	FmgrInfo	result_in_func;
  	Oid			result_typioparam;
***************
*** 137,143 ****
--- 142,151 ----
  static Tcl_HashTable *pltcl_proc_hash = NULL;
  static Tcl_HashTable *pltcl_norm_query_hash = NULL;
  static Tcl_HashTable *pltcl_safe_query_hash = NULL;
+ 
+ /* these are saved and restored by pltcl_call_handler */
  static FunctionCallInfo pltcl_current_fcinfo = NULL;
+ static pltcl_proc_desc *pltcl_current_prodesc = NULL;
  
  /*
   * When a callback from Tcl into PG incurs an error, we temporarily store
***************
*** 179,189 ****
  static int pltcl_returnnull(ClientData cdata, Tcl_Interp *interp,
  				 int argc, CONST84 char *argv[]);
  
! static int pltcl_SPI_exec(ClientData cdata, Tcl_Interp *interp,
  			   int argc, CONST84 char *argv[]);
  static int pltcl_SPI_prepare(ClientData cdata, Tcl_Interp *interp,
  				  int argc, CONST84 char *argv[]);
! static int pltcl_SPI_execp(ClientData cdata, Tcl_Interp *interp,
  				int argc, CONST84 char *argv[]);
  static int pltcl_SPI_lastoid(ClientData cdata, Tcl_Interp *interp,
  				  int argc, CONST84 char *argv[]);
--- 187,197 ----
  static int pltcl_returnnull(ClientData cdata, Tcl_Interp *interp,
  				 int argc, CONST84 char *argv[]);
  
! static int pltcl_SPI_execute(ClientData cdata, Tcl_Interp *interp,
  			   int argc, CONST84 char *argv[]);
  static int pltcl_SPI_prepare(ClientData cdata, Tcl_Interp *interp,
  				  int argc, CONST84 char *argv[]);
! static int pltcl_SPI_execute_plan(ClientData cdata, Tcl_Interp *interp,
  				int argc, CONST84 char *argv[]);
  static int pltcl_SPI_lastoid(ClientData cdata, Tcl_Interp *interp,
  				  int argc, CONST84 char *argv[]);
***************
*** 307,317 ****
  					  pltcl_returnnull, NULL, NULL);
  
  	Tcl_CreateCommand(interp, "spi_exec",
! 					  pltcl_SPI_exec, NULL, NULL);
  	Tcl_CreateCommand(interp, "spi_prepare",
  					  pltcl_SPI_prepare, NULL, NULL);
  	Tcl_CreateCommand(interp, "spi_execp",
! 					  pltcl_SPI_execp, NULL, NULL);
  	Tcl_CreateCommand(interp, "spi_lastoid",
  					  pltcl_SPI_lastoid, NULL, NULL);
  }
--- 315,325 ----
  					  pltcl_returnnull, NULL, NULL);
  
  	Tcl_CreateCommand(interp, "spi_exec",
! 					  pltcl_SPI_execute, NULL, NULL);
  	Tcl_CreateCommand(interp, "spi_prepare",
  					  pltcl_SPI_prepare, NULL, NULL);
  	Tcl_CreateCommand(interp, "spi_execp",
! 					  pltcl_SPI_execute_plan, NULL, NULL);
  	Tcl_CreateCommand(interp, "spi_lastoid",
  					  pltcl_SPI_lastoid, NULL, NULL);
  }
***************
*** 334,341 ****
  	/************************************************************
  	 * Check if table pltcl_modules exists
  	 ************************************************************/
! 	spi_rc = SPI_exec("select 1 from pg_catalog.pg_class "
! 					  "where relname = 'pltcl_modules'", 1);
  	SPI_freetuptable(SPI_tuptable);
  	if (spi_rc != SPI_OK_SELECT)
  		elog(ERROR, "select from pg_class failed");
--- 342,350 ----
  	/************************************************************
  	 * Check if table pltcl_modules exists
  	 ************************************************************/
! 	spi_rc = SPI_execute("select 1 from pg_catalog.pg_class "
! 						 "where relname = 'pltcl_modules'",
! 						 false, 1);
  	SPI_freetuptable(SPI_tuptable);
  	if (spi_rc != SPI_OK_SELECT)
  		elog(ERROR, "select from pg_class failed");
***************
*** 348,356 ****
  	 ************************************************************/
  	Tcl_DStringInit(&unknown_src);
  
! 	spi_rc = SPI_exec("select modseq, modsrc from pltcl_modules "
! 					  "where modname = 'unknown' "
! 					  "order by modseq", 0);
  	if (spi_rc != SPI_OK_SELECT)
  		elog(ERROR, "select from pltcl_modules failed");
  
--- 357,366 ----
  	 ************************************************************/
  	Tcl_DStringInit(&unknown_src);
  
! 	spi_rc = SPI_execute("select modseq, modsrc from pltcl_modules "
! 						 "where modname = 'unknown' "
! 						 "order by modseq",
! 						 false, 0);
  	if (spi_rc != SPI_OK_SELECT)
  		elog(ERROR, "select from pltcl_modules failed");
  
***************
*** 405,434 ****
  {
  	Datum		retval;
  	FunctionCallInfo save_fcinfo;
  
! 	/************************************************************
  	 * Initialize interpreters if first time through
! 	 ************************************************************/
  	pltcl_init_all();
  
! 	/************************************************************
! 	 * Determine if called as function or trigger and
! 	 * call appropriate subhandler
! 	 ************************************************************/
  	save_fcinfo = pltcl_current_fcinfo;
  
! 	if (CALLED_AS_TRIGGER(fcinfo))
  	{
! 		pltcl_current_fcinfo = NULL;
! 		retval = PointerGetDatum(pltcl_trigger_handler(fcinfo));
  	}
! 	else
  	{
! 		pltcl_current_fcinfo = fcinfo;
! 		retval = pltcl_func_handler(fcinfo);
  	}
  
  	pltcl_current_fcinfo = save_fcinfo;
  
  	return retval;
  }
--- 415,460 ----
  {
  	Datum		retval;
  	FunctionCallInfo save_fcinfo;
+ 	pltcl_proc_desc *save_prodesc;
  
! 	/*
  	 * Initialize interpreters if first time through
! 	 */
  	pltcl_init_all();
  
! 	/*
! 	 * Ensure that static pointers are saved/restored properly
! 	 */
  	save_fcinfo = pltcl_current_fcinfo;
+ 	save_prodesc = pltcl_current_prodesc;
  
! 	PG_TRY();
  	{
! 		/*
! 		 * Determine if called as function or trigger and
! 		 * call appropriate subhandler
! 		 */
! 		if (CALLED_AS_TRIGGER(fcinfo))
! 		{
! 			pltcl_current_fcinfo = NULL;
! 			retval = PointerGetDatum(pltcl_trigger_handler(fcinfo));
! 		}
! 		else
! 		{
! 			pltcl_current_fcinfo = fcinfo;
! 			retval = pltcl_func_handler(fcinfo);
! 		}
  	}
! 	PG_CATCH();
  	{
! 		pltcl_current_fcinfo = save_fcinfo;
! 		pltcl_current_prodesc = save_prodesc;
! 		PG_RE_THROW();
  	}
+ 	PG_END_TRY();
  
  	pltcl_current_fcinfo = save_fcinfo;
+ 	pltcl_current_prodesc = save_prodesc;
  
  	return retval;
  }
***************
*** 467,472 ****
--- 493,500 ----
  	/* Find or compile the function */
  	prodesc = compile_pltcl_function(fcinfo->flinfo->fn_oid, InvalidOid);
  
+ 	pltcl_current_prodesc = prodesc;
+ 
  	if (prodesc->lanpltrusted)
  		interp = pltcl_safe_interp;
  	else
***************
*** 643,648 ****
--- 671,678 ----
  	prodesc = compile_pltcl_function(fcinfo->flinfo->fn_oid,
  								RelationGetRelid(trigdata->tg_relation));
  
+ 	pltcl_current_prodesc = prodesc;
+ 
  	if (prodesc->lanpltrusted)
  		interp = pltcl_safe_interp;
  	else
***************
*** 1030,1035 ****
--- 1060,1069 ----
  		prodesc->fn_xmin = HeapTupleHeaderGetXmin(procTup->t_data);
  		prodesc->fn_cmin = HeapTupleHeaderGetCmin(procTup->t_data);
  
+ 		/* Remember if function is STABLE/IMMUTABLE */
+ 		prodesc->fn_readonly =
+ 			(procStruct->provolatile != PROVOLATILE_VOLATILE);
+ 
  		/************************************************************
  		 * Lookup the pg_language tuple by Oid
  		 ************************************************************/
***************
*** 1336,1342 ****
  
  /**********************************************************************
   * pltcl_quote()	- quote literal strings that are to
!  *			  be used in SPI_exec query strings
   **********************************************************************/
  static int
  pltcl_quote(ClientData cdata, Tcl_Interp *interp,
--- 1370,1376 ----
  
  /**********************************************************************
   * pltcl_quote()	- quote literal strings that are to
!  *			  be used in SPI_execute query strings
   **********************************************************************/
  static int
  pltcl_quote(ClientData cdata, Tcl_Interp *interp,
***************
*** 1484,1495 ****
  
  
  /**********************************************************************
!  * pltcl_SPI_exec()		- The builtin SPI_exec command
   *				  for the Tcl interpreter
   **********************************************************************/
  static int
! pltcl_SPI_exec(ClientData cdata, Tcl_Interp *interp,
! 			   int argc, CONST84 char *argv[])
  {
  	volatile int my_rc;
  	int			spi_rc;
--- 1518,1529 ----
  
  
  /**********************************************************************
!  * pltcl_SPI_execute()		- The builtin SPI_execute command
   *				  for the Tcl interpreter
   **********************************************************************/
  static int
! pltcl_SPI_execute(ClientData cdata, Tcl_Interp *interp,
! 				  int argc, CONST84 char *argv[])
  {
  	volatile int my_rc;
  	int			spi_rc;
***************
*** 1570,1576 ****
  	PG_TRY();
  	{
  		UTF_BEGIN;
! 		spi_rc = SPI_exec(UTF_U2E(argv[query_idx]), count);
  		UTF_END;
  	}
  	PG_CATCH();
--- 1604,1611 ----
  	PG_TRY();
  	{
  		UTF_BEGIN;
! 		spi_rc = SPI_execute(UTF_U2E(argv[query_idx]),
! 							 pltcl_current_prodesc->fn_readonly, count);
  		UTF_END;
  	}
  	PG_CATCH();
***************
*** 1603,1609 ****
  			break;
  
  		default:
! 			Tcl_AppendResult(interp, "pltcl: SPI_exec failed: ",
  							 SPI_result_code_string(spi_rc), NULL);
  			SPI_freetuptable(SPI_tuptable);
  			return TCL_ERROR;
--- 1638,1644 ----
  			break;
  
  		default:
! 			Tcl_AppendResult(interp, "pltcl: SPI_execute failed: ",
  							 SPI_result_code_string(spi_rc), NULL);
  			SPI_freetuptable(SPI_tuptable);
  			return TCL_ERROR;
***************
*** 1840,1850 ****
  
  
  /**********************************************************************
!  * pltcl_SPI_execp()		- Execute a prepared plan
   **********************************************************************/
  static int
! pltcl_SPI_execp(ClientData cdata, Tcl_Interp *interp,
! 				int argc, CONST84 char *argv[])
  {
  	volatile int my_rc;
  	int			spi_rc;
--- 1875,1885 ----
  
  
  /**********************************************************************
!  * pltcl_SPI_execute_plan()		- Execute a prepared plan
   **********************************************************************/
  static int
! pltcl_SPI_execute_plan(ClientData cdata, Tcl_Interp *interp,
! 					   int argc, CONST84 char *argv[])
  {
  	volatile int my_rc;
  	int			spi_rc;
***************
*** 1992,1998 ****
  		}
  
  		/************************************************************
! 		 * Setup the value array for the SPI_execp() using
  		 * the type specific input functions
  		 ************************************************************/
  		oldcontext = CurrentMemoryContext;
--- 2027,2033 ----
  		}
  
  		/************************************************************
! 		 * Setup the value array for SPI_execute_plan() using
  		 * the type specific input functions
  		 ************************************************************/
  		oldcontext = CurrentMemoryContext;
***************
*** 2046,2052 ****
  	 ************************************************************/
  	oldcontext = CurrentMemoryContext;
  	PG_TRY();
! 	spi_rc = SPI_execp(qdesc->plan, argvalues, nulls, count);
  	PG_CATCH();
  	{
  		MemoryContextSwitchTo(oldcontext);
--- 2081,2090 ----
  	 ************************************************************/
  	oldcontext = CurrentMemoryContext;
  	PG_TRY();
! 	{
! 		spi_rc = SPI_execute_plan(qdesc->plan, argvalues, nulls,
! 								  pltcl_current_prodesc->fn_readonly, count);
! 	}
  	PG_CATCH();
  	{
  		MemoryContextSwitchTo(oldcontext);
***************
*** 2058,2064 ****
  	PG_END_TRY();
  
  	/************************************************************
! 	 * Check the return code from SPI_execp()
  	 ************************************************************/
  	switch (spi_rc)
  	{
--- 2096,2102 ----
  	PG_END_TRY();
  
  	/************************************************************
! 	 * Check the return code from SPI_execute_plan()
  	 ************************************************************/
  	switch (spi_rc)
  	{
***************
*** 2080,2086 ****
  			break;
  
  		default:
! 			Tcl_AppendResult(interp, "pltcl: SPI_execp failed: ",
  							 SPI_result_code_string(spi_rc), NULL);
  			SPI_freetuptable(SPI_tuptable);
  			return TCL_ERROR;
--- 2118,2124 ----
  			break;
  
  		default:
! 			Tcl_AppendResult(interp, "pltcl: SPI_execute_plan failed: ",
  							 SPI_result_code_string(spi_rc), NULL);
  			SPI_freetuptable(SPI_tuptable);
  			return TCL_ERROR;
*** src/test/regress/expected/transactions.out.orig	Mon Sep  6 13:46:34 2004
--- src/test/regress/expected/transactions.out	Sun Sep 12 17:27:18 2004
***************
*** 374,379 ****
--- 374,457 ----
  	FETCH 10 FROM c;
  ERROR:  portal "c" cannot be run
  COMMIT;
+ --
+ -- Check that "stable" functions are really stable.  They should not be
+ -- able to see the partial results of the calling query.  (Ideally we would
+ -- also check that they don't see commits of concurrent transactions, but
+ -- that's a mite hard to do within the limitations of pg_regress.)
+ --
+ select * from xacttest;
+   a  |    b    
+ -----+---------
+   56 |     7.8
+  100 |  99.097
+    0 | 0.09561
+   42 |  324.78
+  777 | 777.777
+ (5 rows)
+ 
+ create or replace function max_xacttest() returns smallint language sql as
+ 'select max(a) from xacttest' stable;
+ begin;
+ update xacttest set a = max_xacttest() + 10 where a > 0;
+ select * from xacttest;
+   a  |    b    
+ -----+---------
+    0 | 0.09561
+  787 |     7.8
+  787 |  99.097
+  787 |  324.78
+  787 | 777.777
+ (5 rows)
+ 
+ rollback;
+ -- But a volatile function can see the partial results of the calling query
+ create or replace function max_xacttest() returns smallint language sql as
+ 'select max(a) from xacttest' volatile;
+ begin;
+ update xacttest set a = max_xacttest() + 10 where a > 0;
+ select * from xacttest;
+   a  |    b    
+ -----+---------
+    0 | 0.09561
+  787 |     7.8
+  797 |  99.097
+  807 |  324.78
+  817 | 777.777
+ (5 rows)
+ 
+ rollback;
+ -- Now the same test with plpgsql (since it depends on SPI which is different)
+ create or replace function max_xacttest() returns smallint language plpgsql as
+ 'begin return max(a) from xacttest; end' stable;
+ begin;
+ update xacttest set a = max_xacttest() + 10 where a > 0;
+ select * from xacttest;
+   a  |    b    
+ -----+---------
+    0 | 0.09561
+  787 |     7.8
+  787 |  99.097
+  787 |  324.78
+  787 | 777.777
+ (5 rows)
+ 
+ rollback;
+ create or replace function max_xacttest() returns smallint language plpgsql as
+ 'begin return max(a) from xacttest; end' volatile;
+ begin;
+ update xacttest set a = max_xacttest() + 10 where a > 0;
+ select * from xacttest;
+   a  |    b    
+ -----+---------
+    0 | 0.09561
+  787 |     7.8
+  797 |  99.097
+  807 |  324.78
+  817 | 777.777
+ (5 rows)
+ 
+ rollback;
  -- test case for problems with dropping an open relation during abort
  BEGIN;
  	savepoint x;
*** src/test/regress/sql/transactions.sql.orig	Mon Sep  6 13:44:08 2004
--- src/test/regress/sql/transactions.sql	Sun Sep 12 17:25:11 2004
***************
*** 231,236 ****
--- 231,279 ----
  	FETCH 10 FROM c;
  COMMIT;
  
+ --
+ -- Check that "stable" functions are really stable.  They should not be
+ -- able to see the partial results of the calling query.  (Ideally we would
+ -- also check that they don't see commits of concurrent transactions, but
+ -- that's a mite hard to do within the limitations of pg_regress.)
+ --
+ select * from xacttest;
+ 
+ create or replace function max_xacttest() returns smallint language sql as
+ 'select max(a) from xacttest' stable;
+ 
+ begin;
+ update xacttest set a = max_xacttest() + 10 where a > 0;
+ select * from xacttest;
+ rollback;
+ 
+ -- But a volatile function can see the partial results of the calling query
+ create or replace function max_xacttest() returns smallint language sql as
+ 'select max(a) from xacttest' volatile;
+ 
+ begin;
+ update xacttest set a = max_xacttest() + 10 where a > 0;
+ select * from xacttest;
+ rollback;
+ 
+ -- Now the same test with plpgsql (since it depends on SPI which is different)
+ create or replace function max_xacttest() returns smallint language plpgsql as
+ 'begin return max(a) from xacttest; end' stable;
+ 
+ begin;
+ update xacttest set a = max_xacttest() + 10 where a > 0;
+ select * from xacttest;
+ rollback;
+ 
+ create or replace function max_xacttest() returns smallint language plpgsql as
+ 'begin return max(a) from xacttest; end' volatile;
+ 
+ begin;
+ update xacttest set a = max_xacttest() + 10 where a > 0;
+ select * from xacttest;
+ rollback;
+ 
+ 
  -- test case for problems with dropping an open relation during abort
  BEGIN;
  	savepoint x;