diff --git a/doc/src/sgml/plperl.sgml b/doc/src/sgml/plperl.sgml
index b9668103ecd..a94163e7be6 100644
--- a/doc/src/sgml/plperl.sgml
+++ b/doc/src/sgml/plperl.sgml
@@ -1,4 +1,4 @@
-<!-- $PostgreSQL: pgsql/doc/src/sgml/plperl.sgml,v 2.58 2006/10/23 18:10:31 petere Exp $ -->
+<!-- $PostgreSQL: pgsql/doc/src/sgml/plperl.sgml,v 2.59 2006/11/13 17:13:56 adunstan Exp $ -->
 
  <chapter id="plperl">
   <title>PL/Perl - Perl Procedural Language</title>
@@ -646,6 +646,25 @@ $$ LANGUAGE plperl;
    If the above function was created by a superuser using the language
    <literal>plperlu</>, execution would succeed.
   </para>
+
+  <note>
+    <para>
+	  For security reasons, to stop a leak of privileged operations from
+      <application>PL/PerlU</> to <application>PL/Perl</>, these two languages
+	  have to run in separate instances of the Perl interpreter. If your
+	  Perl installation has been appropriately compiled, this is not a problem.
+	  However, not all installations are compiled with the requisite flags.
+	  If <productname>PostgreSQL</> detects that this is the case then it will
+	  not start a second interpreter, but instead create an error. In
+	  consequence, in such an installation, you cannot use both 
+	  <application>PL/PerlU</> and <application>PL/Perl</> in the same backend
+	  process. The remedy for this is to obtain a Perl installation created
+	  with the appropriate flags, namely either <literal>usemultiplicity</> or
+	  both <literal>usethreads</> and <literal>useithreads</>. 
+	  For more details,see the <literal>perlembed</> manual page.
+    </para>
+  </note>
+  
  </sect1>
 
  <sect1 id="plperl-triggers">
diff --git a/doc/src/sgml/release.sgml b/doc/src/sgml/release.sgml
index 58b4eaf50b0..78a72cea008 100644
--- a/doc/src/sgml/release.sgml
+++ b/doc/src/sgml/release.sgml
@@ -1,4 +1,4 @@
-<!-- $PostgreSQL: pgsql/doc/src/sgml/release.sgml,v 1.482 2006/11/06 17:00:27 tgl Exp $ -->
+<!-- $PostgreSQL: pgsql/doc/src/sgml/release.sgml,v 1.483 2006/11/13 17:13:56 adunstan Exp $ -->
 <!--
 
 Typical markup:
@@ -407,6 +407,21 @@ links to the main documentation.
        </para>
       </listitem>
 
+	  <listitem>
+	    <para>
+		  Data can no longer be shared between a PL/Perl function and a 
+		  PL/PerlU function, and modules used by a /PerlU function are no 
+		  longer available to PL/Perl functions.
+		</para>
+		<para>
+		  Some perl installations have not been compiled with the correct flags
+		  to allow multiple interpreters to exist within a single process.
+		  In this situation PL/Perl and PL/PerlU cannot both be used in a 
+		  single backend. The solution is to get a Perl installation which 
+		  supports multiple interpreters. (Andrew)  
+ 	    </para>
+      </listitem>
+
       <listitem>
        <para>
         In <filename>contrib/xml2/</>, rename <function>xml_valid()</> to
@@ -1743,9 +1758,22 @@ links to the main documentation.
        <para>
         Previously, it was lexical, which caused unexpected sharing
         violations.
-       </para>
+       </para>	   
       </listitem>
 
+	  <listitem>
+	    <para>
+		  Run PL/Perl and PL/PerlU in separate interpreters, for security 
+		  reasons.
+		</para>
+		<para>
+		  In consequence, they can no longer share data nor loaded modules.
+		  Also, if Perl has not been compiled with the requisite flags to
+		  allow multiple interpreters, only one of these lamguages can be used
+		  in any given backend process. (Andrew)
+	    </para>
+	  </listitem>
+
      </itemizedlist>
 
     </sect3>
diff --git a/src/pl/plperl/plperl.c b/src/pl/plperl/plperl.c
index 83332b92cde..0ca7f9b1f64 100644
--- a/src/pl/plperl/plperl.c
+++ b/src/pl/plperl/plperl.c
@@ -1,7 +1,7 @@
 /**********************************************************************
  * plperl.c - perl as a procedural language for PostgreSQL
  *
- *	  $PostgreSQL: pgsql/src/pl/plperl/plperl.c,v 1.121 2006/10/19 18:32:47 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/pl/plperl/plperl.c,v 1.122 2006/11/13 17:13:57 adunstan Exp $
  *
  **********************************************************************/
 
@@ -27,6 +27,7 @@
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/typcache.h"
+#include "utils/hsearch.h"
 
 /* perl stuff */
 #include "plperl.h"
@@ -55,6 +56,14 @@ typedef struct plperl_proc_desc
 	SV		   *reference;
 } plperl_proc_desc;
 
+/* hash table entry for proc desc  */
+
+typedef struct plperl_proc_entry
+{
+	char proc_name[NAMEDATALEN];
+	plperl_proc_desc *proc_data;
+} plperl_proc_entry;
+
 /*
  * The information we cache for the duration of a single call to a
  * function.
@@ -82,13 +91,38 @@ typedef struct plperl_query_desc
 	Oid		   *argtypioparams;
 } plperl_query_desc;
 
+/* hash table entry for query desc  */
+
+typedef struct plperl_query_entry
+{
+	char query_name[NAMEDATALEN];
+	plperl_query_desc *query_data;
+} plperl_query_entry;
+
 /**********************************************************************
  * Global data
  **********************************************************************/
+
+typedef enum
+{
+	INTERP_NONE,
+	INTERP_HELD,
+	INTERP_TRUSTED,
+	INTERP_UNTRUSTED,
+	INTERP_BOTH
+} InterpState;
+
+static InterpState interp_state = INTERP_NONE;
+static bool can_run_two = false;
+
 static bool plperl_safe_init_done = false;
-static PerlInterpreter *plperl_interp = NULL;
-static HV  *plperl_proc_hash = NULL;
-static HV  *plperl_query_hash = NULL;
+static PerlInterpreter *plperl_trusted_interp = NULL;
+static PerlInterpreter *plperl_untrusted_interp = NULL;
+static PerlInterpreter *plperl_held_interp = NULL;
+static bool can_run_two;
+static bool trusted_context;
+static HTAB  *plperl_proc_hash = NULL;
+static HTAB  *plperl_query_hash = NULL;
 
 static bool plperl_use_strict = false;
 
@@ -144,6 +178,7 @@ _PG_init(void)
 {
 	/* Be sure we do initialization only once (should be redundant now) */
 	static bool inited = false;
+    HASHCTL     hash_ctl;
 
 	if (inited)
 		return;
@@ -157,6 +192,22 @@ _PG_init(void)
 
 	EmitWarningsOnPlaceholders("plperl");
 
+	MemSet(&hash_ctl, 0, sizeof(hash_ctl));
+
+	hash_ctl.keysize = NAMEDATALEN;
+	hash_ctl.entrysize = sizeof(plperl_proc_entry);
+
+	plperl_proc_hash = hash_create("PLPerl Procedures",
+								   32,
+								   &hash_ctl,
+								   HASH_ELEM);
+
+	hash_ctl.entrysize = sizeof(plperl_query_entry);
+	plperl_query_hash = hash_create("PLPerl Queries",
+									32,
+									&hash_ctl,
+									HASH_ELEM);
+
 	plperl_init_interp();
 
 	inited = true;
@@ -235,6 +286,90 @@ _PG_init(void)
 	"      elog(ERROR,'trusted Perl functions disabled - " \
 	"      please upgrade Perl Safe module to version 2.09 or later');}]); }"
 
+#define TEST_FOR_MULTI \
+	"use Config; " \
+	"$Config{usemultiplicity} eq 'define' or "  \
+    "($Config{usethreads} eq 'define' " \
+	" and $Config{useithreads} eq 'define')"
+
+
+/********************************************************************
+ *
+ * We start out by creating a "held" interpreter that we can use in
+ * trusted or untrusted mode (but not both) as the need arises. Later, we
+ * assign that interpreter if it is available to either the trusted or 
+ * untrusted interpreter. If it has already been assigned, and we need to
+ * create the other interpreter, we do that if we can, or error out.
+ * We detect if it is safe to run two interpreters during the setup of the
+ * dummy interpreter.
+ */
+
+
+static void 
+check_interp(bool trusted)
+{
+	if (interp_state == INTERP_HELD)
+	{
+		if (trusted)
+		{
+			plperl_trusted_interp = plperl_held_interp;
+			interp_state = INTERP_TRUSTED;
+		}
+		else
+		{
+			plperl_untrusted_interp = plperl_held_interp;
+			interp_state = INTERP_UNTRUSTED;
+		}
+		plperl_held_interp = NULL;
+		trusted_context = trusted;
+	}
+	else if (interp_state == INTERP_BOTH || 
+			 (trusted && interp_state == INTERP_TRUSTED) ||
+			 (!trusted && interp_state == INTERP_UNTRUSTED))
+	{
+		if (trusted_context != trusted)
+		{
+			if (trusted)
+				PERL_SET_CONTEXT(plperl_trusted_interp);
+			else
+				PERL_SET_CONTEXT(plperl_untrusted_interp);
+			trusted_context = trusted;
+		}
+	}
+	else if (can_run_two)
+	{
+		PERL_SET_CONTEXT(plperl_held_interp);
+		plperl_init_interp();
+		if (trusted)
+			plperl_trusted_interp = plperl_held_interp;
+		else
+			plperl_untrusted_interp = plperl_held_interp;
+		interp_state = INTERP_BOTH;
+		plperl_held_interp = NULL;
+		trusted_context = trusted;
+	}
+	else
+	{
+		elog(ERROR, 
+			 "can not allocate second Perl interpreter on this platform");
+
+	}
+	
+}
+
+
+static void
+restore_context (bool old_context)
+{
+	if (trusted_context != old_context)
+	{
+		if (old_context)
+			PERL_SET_CONTEXT(plperl_trusted_interp);
+		else
+			PERL_SET_CONTEXT(plperl_untrusted_interp);
+		trusted_context = old_context;
+	}
+}
 
 static void
 plperl_init_interp(void)
@@ -285,16 +420,24 @@ plperl_init_interp(void)
 	save_time = loc ? pstrdup(loc) : NULL;
 #endif
 
-	plperl_interp = perl_alloc();
-	if (!plperl_interp)
+
+	plperl_held_interp = perl_alloc();
+	if (!plperl_held_interp)
 		elog(ERROR, "could not allocate Perl interpreter");
 
-	perl_construct(plperl_interp);
-	perl_parse(plperl_interp, plperl_init_shared_libs, 3, embedding, NULL);
-	perl_run(plperl_interp);
+	perl_construct(plperl_held_interp);
+	perl_parse(plperl_held_interp, plperl_init_shared_libs, 
+			   3, embedding, NULL);
+	perl_run(plperl_held_interp);
 
-	plperl_proc_hash = newHV();
-	plperl_query_hash = newHV();
+	if (interp_state == INTERP_NONE)
+	{
+		SV *res;
+
+		res = eval_pv(TEST_FOR_MULTI,TRUE);
+		can_run_two = SvIV(res); 
+		interp_state = INTERP_HELD;
+	}
 
 #ifdef WIN32
 
@@ -1009,6 +1152,7 @@ plperl_func_handler(PG_FUNCTION_ARGS)
 	Datum		retval;
 	ReturnSetInfo *rsi;
 	SV		   *array_ret = NULL;
+	bool       oldcontext = trusted_context;
 
 	/*
 	 * Create the call_data beforing connecting to SPI, so that it is not
@@ -1037,6 +1181,8 @@ plperl_func_handler(PG_FUNCTION_ARGS)
 							"cannot accept a set")));
 	}
 
+	check_interp(prodesc->lanpltrusted);
+
 	perlret = plperl_call_perl_func(prodesc, fcinfo);
 
 	/************************************************************
@@ -1146,6 +1292,8 @@ plperl_func_handler(PG_FUNCTION_ARGS)
 		SvREFCNT_dec(perlret);
 
 	current_call_data = NULL;
+	restore_context(oldcontext);
+
 	return retval;
 }
 
@@ -1158,6 +1306,7 @@ plperl_trigger_handler(PG_FUNCTION_ARGS)
 	Datum		retval;
 	SV		   *svTD;
 	HV		   *hvTD;
+	bool       oldcontext = trusted_context;
 
 	/*
 	 * Create the call_data beforing connecting to SPI, so that it is not
@@ -1174,6 +1323,8 @@ plperl_trigger_handler(PG_FUNCTION_ARGS)
 	prodesc = compile_plperl_function(fcinfo->flinfo->fn_oid, true);
 	current_call_data->prodesc = prodesc;
 
+	check_interp(prodesc->lanpltrusted);
+
 	svTD = plperl_trigger_build_args(fcinfo);
 	perlret = plperl_call_perl_trigger_func(prodesc, fcinfo, svTD);
 	hvTD = (HV *) SvRV(svTD);
@@ -1244,6 +1395,7 @@ plperl_trigger_handler(PG_FUNCTION_ARGS)
 		SvREFCNT_dec(perlret);
 
 	current_call_data = NULL;
+	restore_context(oldcontext);
 	return retval;
 }
 
@@ -1256,7 +1408,9 @@ compile_plperl_function(Oid fn_oid, bool is_trigger)
 	char		internal_proname[64];
 	plperl_proc_desc *prodesc = NULL;
 	int			i;
-	SV		  **svp;
+	plperl_proc_entry *hash_entry;
+	bool found;
+	bool oldcontext = trusted_context;
 
 	/* We'll need the pg_proc tuple in any case... */
 	procTup = SearchSysCache(PROCOID,
@@ -1277,12 +1431,14 @@ compile_plperl_function(Oid fn_oid, bool is_trigger)
 	/************************************************************
 	 * Lookup the internal proc name in the hashtable
 	 ************************************************************/
-	svp = hv_fetch_string(plperl_proc_hash, internal_proname);
-	if (svp)
+	hash_entry = hash_search(plperl_proc_hash, internal_proname, 
+							 HASH_FIND, NULL);
+
+	if (hash_entry)
 	{
 		bool		uptodate;
 
-		prodesc = INT2PTR(plperl_proc_desc *, SvUV(*svp));
+		prodesc = hash_entry->proc_data;
 
 		/************************************************************
 		 * If it's present, must check whether it's still up to date.
@@ -1294,8 +1450,10 @@ compile_plperl_function(Oid fn_oid, bool is_trigger)
 
 		if (!uptodate)
 		{
-			/* need we delete old entry? */
+			free(prodesc); /* are we leaking memory here? */
 			prodesc = NULL;
+			hash_search(plperl_proc_hash, internal_proname,
+						HASH_REMOVE,NULL);
 		}
 	}
 
@@ -1469,7 +1627,13 @@ compile_plperl_function(Oid fn_oid, bool is_trigger)
 		/************************************************************
 		 * Create the procedure in the interpreter
 		 ************************************************************/
+
+		check_interp(prodesc->lanpltrusted);
+
 		prodesc->reference = plperl_create_sub(proc_source, prodesc->lanpltrusted);
+
+		restore_context(oldcontext);
+
 		pfree(proc_source);
 		if (!prodesc->reference)	/* can this happen? */
 		{
@@ -1479,8 +1643,9 @@ compile_plperl_function(Oid fn_oid, bool is_trigger)
 				 internal_proname);
 		}
 
-		hv_store_string(plperl_proc_hash, internal_proname,
-						newSVuv(PTR2UV(prodesc)));
+		hash_entry = hash_search(plperl_proc_hash, internal_proname,
+								 HASH_ENTER, &found);
+		hash_entry->proc_data = prodesc;
 	}
 
 	ReleaseSysCache(procTup);
@@ -1939,6 +2104,8 @@ SV *
 plperl_spi_prepare(char *query, int argc, SV **argv)
 {
 	plperl_query_desc *qdesc;
+	plperl_query_entry *hash_entry;
+	bool        found;
 	void	   *plan;
 	int			i;
 
@@ -2051,7 +2218,10 @@ plperl_spi_prepare(char *query, int argc, SV **argv)
 	 * Insert a hashtable entry for the plan and return
 	 * the key to the caller.
 	 ************************************************************/
-	hv_store_string(plperl_query_hash, qdesc->qname, newSVuv(PTR2UV(qdesc)));
+
+	hash_entry = hash_search(plperl_query_hash, qdesc->qname,
+							 HASH_ENTER,&found);
+	hash_entry->query_data = qdesc;
 
 	return newSVstring(qdesc->qname);
 }
@@ -2067,6 +2237,7 @@ plperl_spi_exec_prepared(char *query, HV *attr, int argc, SV **argv)
 	char	   *nulls;
 	Datum	   *argvalues;
 	plperl_query_desc *qdesc;
+	plperl_query_entry *hash_entry;
 
 	/*
 	 * Execute the query inside a sub-transaction, so we can cope with errors
@@ -2084,13 +2255,14 @@ plperl_spi_exec_prepared(char *query, HV *attr, int argc, SV **argv)
 		/************************************************************
 		 * Fetch the saved plan descriptor, see if it's o.k.
 		 ************************************************************/
-		sv = hv_fetch_string(plperl_query_hash, query);
-		if (sv == NULL)
-			elog(ERROR, "spi_exec_prepared: Invalid prepared query passed");
-		if (*sv == NULL || !SvOK(*sv))
-			elog(ERROR, "spi_exec_prepared: panic - plperl_query_hash value corrupted");
 
-		qdesc = INT2PTR(plperl_query_desc *, SvUV(*sv));
+		hash_entry = hash_search(plperl_query_hash, query,
+										 HASH_FIND,NULL);
+		if (hash_entry == NULL)
+			elog(ERROR, "spi_exec_prepared: Invalid prepared query passed");
+
+		qdesc = hash_entry->query_data;
+
 		if (qdesc == NULL)
 			elog(ERROR, "spi_exec_prepared: panic - plperl_query_hash value vanished");
 
@@ -2201,11 +2373,11 @@ plperl_spi_exec_prepared(char *query, HV *attr, int argc, SV **argv)
 SV *
 plperl_spi_query_prepared(char *query, int argc, SV **argv)
 {
-	SV		  **sv;
 	int			i;
 	char	   *nulls;
 	Datum	   *argvalues;
 	plperl_query_desc *qdesc;
+	plperl_query_entry *hash_entry;
 	SV		   *cursor;
 	Portal		portal = NULL;
 
@@ -2225,13 +2397,13 @@ plperl_spi_query_prepared(char *query, int argc, SV **argv)
 		/************************************************************
 		 * Fetch the saved plan descriptor, see if it's o.k.
 		 ************************************************************/
-		sv = hv_fetch_string(plperl_query_hash, query);
-		if (sv == NULL)
-			elog(ERROR, "spi_query_prepared: Invalid prepared query passed");
-		if (*sv == NULL || !SvOK(*sv))
-			elog(ERROR, "spi_query_prepared: panic - plperl_query_hash value corrupted");
+		hash_entry = hash_search(plperl_query_hash, query,
+										 HASH_FIND,NULL);
+		if (hash_entry == NULL)
+			elog(ERROR, "spi_exec_prepared: Invalid prepared query passed");
+
+		qdesc = hash_entry->query_data;
 
-		qdesc = INT2PTR(plperl_query_desc *, SvUV(*sv));
 		if (qdesc == NULL)
 			elog(ERROR, "spi_query_prepared: panic - plperl_query_hash value vanished");
 
@@ -2335,17 +2507,17 @@ plperl_spi_query_prepared(char *query, int argc, SV **argv)
 void
 plperl_spi_freeplan(char *query)
 {
-	SV		  **sv;
 	void	   *plan;
 	plperl_query_desc *qdesc;
+	plperl_query_entry *hash_entry;
 
-	sv = hv_fetch_string(plperl_query_hash, query);
-	if (sv == NULL)
-		elog(ERROR, "spi_exec_freeplan: Invalid prepared query passed");
-	if (*sv == NULL || !SvOK(*sv))
-		elog(ERROR, "spi_exec_freeplan: panic - plperl_query_hash value corrupted");
+	hash_entry = hash_search(plperl_query_hash, query,
+										 HASH_FIND,NULL);
+	if (hash_entry == NULL)
+		elog(ERROR, "spi_exec_prepared: Invalid prepared query passed");
+
+	qdesc = hash_entry->query_data;
 
-	qdesc = INT2PTR(plperl_query_desc *, SvUV(*sv));
 	if (qdesc == NULL)
 		elog(ERROR, "spi_exec_freeplan: panic - plperl_query_hash value vanished");
 
@@ -2353,7 +2525,9 @@ plperl_spi_freeplan(char *query)
 	 * free all memory before SPI_freeplan, so if it dies, nothing will be
 	 * left over
 	 */
-	hv_delete(plperl_query_hash, query, strlen(query), G_DISCARD);
+	hash_search(plperl_query_hash, query, 
+				HASH_REMOVE,NULL);
+
 	plan = qdesc->plan;
 	free(qdesc->argtypes);
 	free(qdesc->arginfuncs);