Convert Postgres arrays to Perl arrays on PL/perl input arguments

More generally, arrays are turned in Perl array references, and row and
composite types are turned into Perl hash references.  This is done
recursively, in a way that's natural to every Perl programmer.

To avoid a backwards compatibility hit, the string representation of
each structure is also available if the function requests it.

Authors: Alexey Klyukin and Alex Hunsaker.
Some code cleanups by me.

13 files changed, 1227 insertions, 367 deletions

diff --git a/src/pl/plperl/GNUmakefile b/src/pl/plperl/GNUmakefile
index 01e585e4280..e86cb84dba2 100644
--- a/src/pl/plperl/GNUmakefile
+++ b/src/pl/plperl/GNUmakefile
@@ -41,7 +41,7 @@ PERLCHUNKS = plc_perlboot.pl plc_trusted.pl
 SHLIB_LINK = $(perl_embed_ldflags)
 
 REGRESS_OPTS = --dbname=$(PL_TESTDB) --load-language=plperl  --load-language=plperlu
-REGRESS = plperl plperl_trigger plperl_shared plperl_elog plperl_util plperl_init plperlu
+REGRESS = plperl plperl_trigger plperl_shared plperl_elog plperl_util plperl_init plperlu plperl_array
 # if Perl can support two interpreters in one backend,
 # test plperl-and-plperlu cases
 ifneq ($(PERL),)
diff --git a/src/pl/plperl/Util.xs b/src/pl/plperl/Util.xs
index 6c6e90faa77..eb1d15fc23e 100644
--- a/src/pl/plperl/Util.xs
+++ b/src/pl/plperl/Util.xs
@@ -198,6 +198,20 @@ looks_like_number(sv)
     OUTPUT:
     RETVAL
 
+SV *
+encode_typed_literal(sv, typname)
+	SV 	   *sv
+	char   *typname;
+	PREINIT:
+		char 	*outstr;
+	CODE:
+		outstr = plperl_sv_to_literal(sv, typname);
+		if (outstr == NULL)
+			RETVAL = &PL_sv_undef;
+		else
+			RETVAL = cstr2sv(outstr);
+	OUTPUT:
+	RETVAL
 
 BOOT:
     items = 0;  /* avoid 'unused variable' warning */
diff --git a/src/pl/plperl/expected/plperl.out b/src/pl/plperl/expected/plperl.out
index d95f646e060..5c1cd8cebfd 100644
--- a/src/pl/plperl/expected/plperl.out
+++ b/src/pl/plperl/expected/plperl.out
@@ -69,7 +69,8 @@ SELECT * FROM perl_set_int(5);
             5
 (6 rows)
 
-CREATE TYPE testrowperl AS (f1 integer, f2 text, f3 text);
+CREATE TYPE testnestperl AS (f5 integer[]);
+CREATE TYPE testrowperl AS (f1 integer, f2 text, f3 text, f4 testnestperl);
 CREATE OR REPLACE FUNCTION perl_row() RETURNS testrowperl AS $$
     return undef;
 $$ LANGUAGE plperl;
@@ -80,24 +81,24 @@ SELECT perl_row();
 (1 row)
 
 SELECT * FROM perl_row();
- f1 | f2 | f3 
-----+----+----
-    |    | 
+ f1 | f2 | f3 | f4 
+----+----+----+----
+    |    |    | 
 (1 row)
 
 CREATE OR REPLACE FUNCTION perl_row() RETURNS testrowperl AS $$
-    return {f2 => 'hello', f1 => 1, f3 => 'world'};
+    return {f2 => 'hello', f1 => 1, f3 => 'world', 'f4' => { 'f5' => [[1]] } };
 $$ LANGUAGE plperl;
 SELECT perl_row();
-    perl_row     
------------------
- (1,hello,world)
+         perl_row          
+---------------------------
+ (1,hello,world,"({{1}})")
 (1 row)
 
 SELECT * FROM perl_row();
- f1 |  f2   |  f3   
-----+-------+-------
-  1 | hello | world
+ f1 |  f2   |  f3   |   f4    
+----+-------+-------+---------
+  1 | hello | world | ({{1}})
 (1 row)
 
 CREATE OR REPLACE FUNCTION perl_set() RETURNS SETOF testrowperl AS $$
@@ -109,15 +110,18 @@ SELECT perl_set();
 (0 rows)
 
 SELECT * FROM perl_set();
- f1 | f2 | f3 
-----+----+----
+ f1 | f2 | f3 | f4 
+----+----+----+----
 (0 rows)
 
 CREATE OR REPLACE FUNCTION perl_set() RETURNS SETOF testrowperl AS $$
     return [
         { f1 => 1, f2 => 'Hello', f3 =>  'World' },
         undef,
-        { f1 => 3, f2 => 'Hello', f3 =>  'PL/Perl' }
+        { f1 => 3, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => {} },
+        { f1 => 4, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => { 'f5' => undef }},
+        { f1 => 5, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => { 'f5' => '{1}' }},
+        { f1 => 6, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => { 'f5' => [1] }},
     ];
 $$  LANGUAGE plperl;
 SELECT perl_set();
@@ -129,25 +133,37 @@ CONTEXT:  PL/Perl function "perl_set"
 CREATE OR REPLACE FUNCTION perl_set() RETURNS SETOF testrowperl AS $$
     return [
         { f1 => 1, f2 => 'Hello', f3 =>  'World' },
-        { f1 => 2, f2 => 'Hello', f3 =>  'PostgreSQL' },
-        { f1 => 3, f2 => 'Hello', f3 =>  'PL/Perl' }
+        { f1 => 2, f2 => 'Hello', f3 =>  'PostgreSQL', 'f4' => undef },
+        { f1 => 3, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => {} },
+        { f1 => 4, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => { 'f5' => undef }},
+        { f1 => 5, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => { 'f5' => '{1}' }},
+        { f1 => 6, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => { 'f5' => [1] }},
+        { f1 => 7, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => '({1})' },
     ];
 $$  LANGUAGE plperl;
 SELECT perl_set();
-       perl_set       
-----------------------
- (1,Hello,World)
- (2,Hello,PostgreSQL)
- (3,Hello,PL/Perl)
-(3 rows)
+         perl_set          
+---------------------------
+ (1,Hello,World,)
+ (2,Hello,PostgreSQL,)
+ (3,Hello,PL/Perl,"()")
+ (4,Hello,PL/Perl,"()")
+ (5,Hello,PL/Perl,"({1})")
+ (6,Hello,PL/Perl,"({1})")
+ (7,Hello,PL/Perl,"({1})")
+(7 rows)
 
 SELECT * FROM perl_set();
- f1 |  f2   |     f3     
-----+-------+------------
-  1 | Hello | World
-  2 | Hello | PostgreSQL
-  3 | Hello | PL/Perl
-(3 rows)
+ f1 |  f2   |     f3     |  f4   
+----+-------+------------+-------
+  1 | Hello | World      | 
+  2 | Hello | PostgreSQL | 
+  3 | Hello | PL/Perl    | ()
+  4 | Hello | PL/Perl    | ()
+  5 | Hello | PL/Perl    | ({1})
+  6 | Hello | PL/Perl    | ({1})
+  7 | Hello | PL/Perl    | ({1})
+(7 rows)
 
 CREATE OR REPLACE FUNCTION perl_record() RETURNS record AS $$
     return undef;
@@ -162,14 +178,14 @@ SELECT * FROM perl_record();
 ERROR:  a column definition list is required for functions returning "record"
 LINE 1: SELECT * FROM perl_record();
                       ^
-SELECT * FROM perl_record() AS (f1 integer, f2 text, f3 text);
- f1 | f2 | f3 
-----+----+----
-    |    | 
+SELECT * FROM perl_record() AS (f1 integer, f2 text, f3 text, f4 testnestperl);
+ f1 | f2 | f3 | f4 
+----+----+----+----
+    |    |    | 
 (1 row)
 
 CREATE OR REPLACE FUNCTION perl_record() RETURNS record AS $$
-    return {f2 => 'hello', f1 => 1, f3 => 'world'};
+    return {f2 => 'hello', f1 => 1, f3 => 'world', 'f4' => { 'f5' => [1] } };
 $$ LANGUAGE plperl;
 SELECT perl_record();
 ERROR:  function returning record called in context that cannot accept type record
@@ -178,10 +194,10 @@ SELECT * FROM perl_record();
 ERROR:  a column definition list is required for functions returning "record"
 LINE 1: SELECT * FROM perl_record();
                       ^
-SELECT * FROM perl_record() AS (f1 integer, f2 text, f3 text);
- f1 |  f2   |  f3   
-----+-------+-------
-  1 | hello | world
+SELECT * FROM perl_record() AS (f1 integer, f2 text, f3 text, f4 testnestperl);
+ f1 |  f2   |  f3   |  f4   
+----+-------+-------+-------
+  1 | hello | world | ({1})
 (1 row)
 
 CREATE OR REPLACE FUNCTION perl_record_set() RETURNS SETOF record AS $$
@@ -474,7 +490,7 @@ SELECT * FROM recurse(3);
 (5 rows)
 
 ---
---- Test arrary return
+--- Test array return
 ---
 CREATE OR REPLACE FUNCTION  array_of_text() RETURNS TEXT[][]
 LANGUAGE plperl as $$
@@ -555,6 +571,32 @@ $$ LANGUAGE plperl;
 SELECT perl_spi_prepared_bad(4.35) as "double precision";
 ERROR:  type "does_not_exist" does not exist at line 2.
 CONTEXT:  PL/Perl function "perl_spi_prepared_bad"
+-- Test with a row type
+CREATE OR REPLACE FUNCTION perl_spi_prepared() RETURNS INTEGER AS $$
+   my $x = spi_prepare('select $1::footype AS a', 'footype');
+   my $q = spi_exec_prepared( $x, '(1, 2)');
+   spi_freeplan($x);
+return $q->{rows}->[0]->{a}->{x};
+$$ LANGUAGE plperl;
+SELECT * from perl_spi_prepared();
+ perl_spi_prepared 
+-------------------
+                 1
+(1 row)
+
+CREATE OR REPLACE FUNCTION perl_spi_prepared_row(footype) RETURNS footype AS $$
+   my $footype = shift;
+   my $x = spi_prepare('select $1 AS a', 'footype');
+   my $q = spi_exec_prepared( $x, {}, $footype );
+   spi_freeplan($x);
+return $q->{rows}->[0]->{a};
+$$ LANGUAGE plperl;
+SELECT * from perl_spi_prepared_row('(1, 2)');
+ x | y 
+---+---
+ 1 | 2
+(1 row)
+
 -- simple test of a DO block
 DO $$
   $a = 'This is a test';
diff --git a/src/pl/plperl/expected/plperl_array.out b/src/pl/plperl/expected/plperl_array.out
new file mode 100644
index 00000000000..be76f6cc240
--- /dev/null
+++ b/src/pl/plperl/expected/plperl_array.out
@@ -0,0 +1,222 @@
+CREATE OR REPLACE FUNCTION plperl_sum_array(INTEGER[]) RETURNS text AS $$
+	my $array_arg = shift;
+	my $result = 0;
+	my @arrays;
+
+	push @arrays, @$array_arg;
+
+	while (@arrays > 0) {
+		my $el = shift @arrays;
+		if (is_array_ref($el)) {
+			push @arrays, @$el;
+		} else {
+			$result += $el;
+		}
+	}
+	return $result.' '.$array_arg;
+$$ LANGUAGE plperl;
+select plperl_sum_array('{1,2,NULL}');
+ plperl_sum_array 
+------------------
+ 3 {1,2,NULL}
+(1 row)
+
+select plperl_sum_array('{}');
+ plperl_sum_array 
+------------------
+ 0 {}
+(1 row)
+
+select plperl_sum_array('{{1,2,3}, {4,5,6}}');
+   plperl_sum_array   
+----------------------
+ 21 {{1,2,3},{4,5,6}}
+(1 row)
+
+select plperl_sum_array('{{{1,2,3}, {4,5,6}}, {{7,8,9}, {10,11,12}}}');
+              plperl_sum_array               
+---------------------------------------------
+ 78 {{{1,2,3},{4,5,6}},{{7,8,9},{10,11,12}}}
+(1 row)
+
+-- check whether we can handle arrays of maximum dimension (6)
+select plperl_sum_array(ARRAY[[[[[[1,2],[3,4]],[[5,6],[7,8]]],[[[9,10],[11,12]],
+[[13,14],[15,16]]]],
+[[[[17,18],[19,20]],[[21,22],[23,24]]],[[[25,26],[27,28]],[[29,30],[31,32]]]]],
+[[[[[1,2],[3,4]],[[5,6],[7,8]]],[[[9,10],[11,12]],[[13,14],[15,16]]]],
+[[[[17,18],[19,20]],[[21,22],[23,24]]],[[[25,26],[27,28]],[[29,30],[31,32]]]]]]);
+                                                                                                                                                 plperl_sum_array                                                                                                                                                 
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ 1056 {{{{{{1,2},{3,4}},{{5,6},{7,8}}},{{{9,10},{11,12}},{{13,14},{15,16}}}},{{{{17,18},{19,20}},{{21,22},{23,24}}},{{{25,26},{27,28}},{{29,30},{31,32}}}}},{{{{{1,2},{3,4}},{{5,6},{7,8}}},{{{9,10},{11,12}},{{13,14},{15,16}}}},{{{{17,18},{19,20}},{{21,22},{23,24}}},{{{25,26},{27,28}},{{29,30},{31,32}}}}}}
+(1 row)
+
+-- what would we do with the arrays exceeding maximum dimension (7)
+select plperl_sum_array('{{{{{{{1,2},{3,4}},{{5,6},{7,8}}},{{{9,10},{11,12}},
+{{13,14},{15,16}}}},
+{{{{17,18},{19,20}},{{21,22},{23,24}}},{{{25,26},{27,28}},{{29,30},{31,32}}}}},
+{{{{{1,2},{3,4}},{{5,6},{7,8}}},{{{9,10},{11,12}},{{13,14},{15,16}}}},
+{{{{17,18},{19,20}},{{21,22},{23,24}}},{{{25,26},{27,28}},{{29,30},{31,32}}}}}},
+{{{{{{1,2},{3,4}},{{5,6},{7,8}}},{{{9,10},{11,12}},{{13,14},{15,16}}}},
+{{{{17,18},{19,20}},{{21,22},{23,24}}},{{{25,26},{27,28}},{{29,30},{31,32}}}}},
+{{{{{1,2},{3,4}},{{5,6},{7,8}}},{{{9,10},{11,12}},{{13,14},{15,16}}}},
+{{{{17,18},{19,20}},{{21,22},{23,24}}},{{{25,26},{27,28}},{{29,30},{31,32}}}}}}}'
+);
+ERROR:  number of array dimensions (7) exceeds the maximum allowed (6)
+LINE 1: select plperl_sum_array('{{{{{{{1,2},{3,4}},{{5,6},{7,8}}},{...
+                                ^
+select plperl_sum_array('{{{1,2,3}, {4,5,6,7}}, {{7,8,9}, {10, 11, 12}}}');
+ERROR:  multidimensional arrays must have array expressions with matching dimensions
+LINE 1: select plperl_sum_array('{{{1,2,3}, {4,5,6,7}}, {{7,8,9}, {1...
+                                ^
+CREATE OR REPLACE FUNCTION plperl_concat(TEXT[]) RETURNS TEXT AS $$
+	my $array_arg = shift;
+	my $result = "";
+	my @arrays;
+	
+	push @arrays, @$array_arg;
+	while (@arrays > 0) {
+		my $el = shift @arrays;
+		if (is_array_ref($el)) {
+			push @arrays, @$el;
+		} else {
+			$result .= $el;
+		}
+	}
+	return $result.' '.$array_arg;
+$$ LANGUAGE plperl;
+select plperl_concat('{"NULL","NULL","NULL''"}');
+            plperl_concat            
+-------------------------------------
+ NULLNULLNULL' {"NULL","NULL",NULL'}
+(1 row)
+
+select plperl_concat('{{NULL,NULL,NULL}}');
+    plperl_concat    
+---------------------
+  {{NULL,NULL,NULL}}
+(1 row)
+
+select plperl_concat('{"hello"," ","world!"}');
+          plperl_concat          
+---------------------------------
+ hello world! {hello," ",world!}
+(1 row)
+
+-- array of rows --
+CREATE TYPE foo AS (bar INTEGER, baz TEXT);
+CREATE OR REPLACE FUNCTION plperl_array_of_rows(foo[]) RETURNS TEXT AS $$
+	my $array_arg = shift;
+	my $result = "";
+	
+	for my $row_ref (@$array_arg) {
+		die "not a hash reference" unless (ref $row_ref eq "HASH");
+			$result .= $row_ref->{bar}." items of ".$row_ref->{baz}.";";
+	}
+	return $result .' '. $array_arg;
+$$ LANGUAGE plperl;
+select plperl_array_of_rows(ARRAY[ ROW(2, 'coffee'), ROW(0, 'sugar')]::foo[]);
+                      plperl_array_of_rows                      
+----------------------------------------------------------------
+ 2 items of coffee;0 items of sugar; {"(2,coffee)","(0,sugar)"}
+(1 row)
+
+-- composite type containing arrays
+CREATE TYPE rowfoo AS (bar INTEGER, baz INTEGER[]);
+CREATE OR REPLACE FUNCTION plperl_sum_row_elements(rowfoo) RETURNS TEXT AS $$
+	my $row_ref = shift;
+	my $result;
+	
+	if (ref $row_ref ne 'HASH') {
+		$result = 0;
+	}
+	else {
+		$result = $row_ref->{bar};
+		die "not an array reference".ref ($row_ref->{baz}) 
+		unless (is_array_ref($row_ref->{baz}));
+		# process a single-dimensional array
+		foreach my $elem (@{$row_ref->{baz}}) {
+			$result += $elem unless ref $elem;
+		}
+	}
+	return $result;
+$$ LANGUAGE plperl;
+select plperl_sum_row_elements(ROW(1, ARRAY[2,3,4,5,6,7,8,9,10])::rowfoo);
+ plperl_sum_row_elements 
+-------------------------
+ 55
+(1 row)
+
+-- composite type containing array of another composite type, which, in order,
+-- contains an array of integers.
+CREATE TYPE rowbar AS (foo rowfoo[]);
+CREATE OR REPLACE FUNCTION plperl_sum_array_of_rows(rowbar) RETURNS TEXT AS $$
+	my $rowfoo_ref = shift;
+	my $result = 0;
+	
+	if (ref $rowfoo_ref eq 'HASH') {
+		my $row_array_ref = $rowfoo_ref->{foo};
+		if (is_array_ref($row_array_ref)) {
+			foreach my $row_ref (@{$row_array_ref}) {
+				if (ref $row_ref eq 'HASH') {
+					$result += $row_ref->{bar};
+					die "not an array reference".ref ($row_ref->{baz}) 
+					unless (is_array_ref($row_ref->{baz}));
+					foreach my $elem (@{$row_ref->{baz}}) {
+						$result += $elem unless ref $elem;
+					}
+				}
+				else {
+					die "element baz is not a reference to a rowfoo";
+				}
+			}
+		} else {
+			die "not a reference to an array of rowfoo elements"
+		}
+	} else {
+		die "not a reference to type rowbar";
+	}
+	return $result;
+$$ LANGUAGE plperl;
+select plperl_sum_array_of_rows(ROW(ARRAY[ROW(1, ARRAY[2,3,4,5,6,7,8,9,10])::rowfoo, 
+ROW(11, ARRAY[12,13,14,15,16,17,18,19,20])::rowfoo])::rowbar);
+ plperl_sum_array_of_rows 
+--------------------------
+ 210
+(1 row)
+
+-- check arrays as out parameters
+CREATE OR REPLACE FUNCTION plperl_arrays_out(OUT INTEGER[]) AS $$
+	return [[1,2,3],[4,5,6]];
+$$ LANGUAGE plperl;
+select plperl_arrays_out();
+ plperl_arrays_out 
+-------------------
+ {{1,2,3},{4,5,6}}
+(1 row)
+
+-- check that we can return the array we passed in
+CREATE OR REPLACE FUNCTION plperl_arrays_inout(INTEGER[]) returns INTEGER[] AS $$
+	return shift;
+$$ LANGUAGE plperl;
+select plperl_arrays_inout('{{1}, {2}, {3}}');
+ plperl_arrays_inout 
+---------------------
+ {{1},{2},{3}}
+(1 row)
+
+-- make sure setof works
+create or replace function perl_setof_array(integer[]) returns setof integer[] language plperl as $$
+	my $arr = shift;
+	for my $r (@$arr) {
+		return_next $r;
+	}
+	return undef;
+$$;
+select perl_setof_array('{{1}, {2}, {3}}');
+ perl_setof_array 
+------------------
+ {1}
+ {2}
+ {3}
+(3 rows)
+
diff --git a/src/pl/plperl/expected/plperl_trigger.out b/src/pl/plperl/expected/plperl_trigger.out
index 3e549f7eefe..238e1b73363 100644
--- a/src/pl/plperl/expected/plperl_trigger.out
+++ b/src/pl/plperl/expected/plperl_trigger.out
@@ -1,13 +1,50 @@
 -- test plperl triggers
+CREATE TYPE rowcomp as (i int);
+CREATE TYPE rowcompnest as (rfoo rowcomp);
 CREATE TABLE trigger_test (
         i int,
-        v varchar
+        v varchar,
+		foo rowcompnest
 );
 CREATE OR REPLACE FUNCTION trigger_data() RETURNS trigger LANGUAGE plperl AS $$
 
   # make sure keys are sorted for consistent results - perl no longer
   # hashes in  repeatable fashion across runs
 
+  sub str {
+	  my $val = shift;
+
+	  if (!defined $val)
+	  {
+		  return 'NULL';
+	  }
+	  elsif (ref $val eq 'HASH')
+	  {
+		my $str = '';
+		foreach my $rowkey (sort keys %$val)
+		{
+		  $str .= ", " if $str;
+		  my $rowval = str($val->{$rowkey});
+		  $str .= "'$rowkey' => $rowval";
+		}
+		return '{'. $str .'}';
+	  }
+	  elsif (ref $val eq 'ARRAY')
+	  {
+		  my $str = '';
+		  for my $argval (@$val)
+		  {
+			  $str .= ", " if $str;
+			  $str .= str($argval);
+		  }
+		  return '['. $str .']';
+	  }
+	  else
+	  {
+		  return "'$val'";
+	  }
+  }
+
   foreach my $key (sort keys %$_TD)
   {
 
@@ -16,42 +53,14 @@ CREATE OR REPLACE FUNCTION trigger_data() RETURNS trigger LANGUAGE plperl AS $$
 	# relid is variable, so we can not use it repeatably
 	$val = "bogus:12345" if $key eq 'relid';
 
-	if (! defined $val)
-	{
-	  elog(NOTICE, "\$_TD->\{$key\} = NULL");
-	}
-	elsif (not ref $val)
-    {
-	  elog(NOTICE, "\$_TD->\{$key\} = '$val'");
-	}
-	elsif (ref $val eq 'HASH')
-	{
-	  my $str = "";
-	  foreach my $rowkey (sort keys %$val)
-	  {
-	    $str .= ", " if $str;
-	    my $rowval = $val->{$rowkey};
-	    $str .= "'$rowkey' => '$rowval'";
-      }
-	  elog(NOTICE, "\$_TD->\{$key\} = \{$str\}");
-	}
-	elsif (ref $val eq 'ARRAY')
-	{
-	  my $str = "";
-	  foreach my $argval (@$val)
-	  {
-	    $str .= ", " if $str;
-	    $str .= "'$argval'";
-      }
-	  elog(NOTICE, "\$_TD->\{$key\} = \[$str\]");
-	}
+	elog(NOTICE, "\$_TD->\{$key\} = ". str($val));
   }
   return undef; # allow statement to proceed;
 $$;
 CREATE TRIGGER show_trigger_data_trig
 BEFORE INSERT OR UPDATE OR DELETE ON trigger_test
 FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
-insert into trigger_test values(1,'insert');
+insert into trigger_test values(1,'insert', '("(1)")');
 NOTICE:  $_TD->{argc} = '2'
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{args} = ['23', 'skidoo']
@@ -62,7 +71,7 @@ NOTICE:  $_TD->{level} = 'ROW'
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{name} = 'show_trigger_data_trig'
 CONTEXT:  PL/Perl function "trigger_data"
-NOTICE:  $_TD->{new} = {'i' => '1', 'v' => 'insert'}
+NOTICE:  $_TD->{new} = {'foo' => {'rfoo' => {'i' => '1'}}, 'i' => '1', 'v' => 'insert'}
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{relid} = 'bogus:12345'
 CONTEXT:  PL/Perl function "trigger_data"
@@ -85,9 +94,9 @@ NOTICE:  $_TD->{level} = 'ROW'
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{name} = 'show_trigger_data_trig'
 CONTEXT:  PL/Perl function "trigger_data"
-NOTICE:  $_TD->{new} = {'i' => '1', 'v' => 'update'}
+NOTICE:  $_TD->{new} = {'foo' => {'rfoo' => {'i' => '1'}}, 'i' => '1', 'v' => 'update'}
 CONTEXT:  PL/Perl function "trigger_data"
-NOTICE:  $_TD->{old} = {'i' => '1', 'v' => 'insert'}
+NOTICE:  $_TD->{old} = {'foo' => {'rfoo' => {'i' => '1'}}, 'i' => '1', 'v' => 'insert'}
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{relid} = 'bogus:12345'
 CONTEXT:  PL/Perl function "trigger_data"
@@ -110,7 +119,7 @@ NOTICE:  $_TD->{level} = 'ROW'
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{name} = 'show_trigger_data_trig'
 CONTEXT:  PL/Perl function "trigger_data"
-NOTICE:  $_TD->{old} = {'i' => '1', 'v' => 'update'}
+NOTICE:  $_TD->{old} = {'foo' => {'rfoo' => {'i' => '1'}}, 'i' => '1', 'v' => 'update'}
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{relid} = 'bogus:12345'
 CONTEXT:  PL/Perl function "trigger_data"
@@ -123,12 +132,12 @@ CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{when} = 'BEFORE'
 CONTEXT:  PL/Perl function "trigger_data"
 DROP TRIGGER show_trigger_data_trig on trigger_test;
-insert into trigger_test values(1,'insert');
+insert into trigger_test values(1,'insert', '("(1)")');
 CREATE VIEW trigger_test_view AS SELECT * FROM trigger_test;
 CREATE TRIGGER show_trigger_data_trig
 INSTEAD OF INSERT OR UPDATE OR DELETE ON trigger_test_view
 FOR EACH ROW EXECUTE PROCEDURE trigger_data(24,'skidoo view');
-insert into trigger_test_view values(2,'insert');
+insert into trigger_test_view values(2,'insert', '("(2)")');
 NOTICE:  $_TD->{argc} = '2'
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{args} = ['24', 'skidoo view']
@@ -139,7 +148,7 @@ NOTICE:  $_TD->{level} = 'ROW'
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{name} = 'show_trigger_data_trig'
 CONTEXT:  PL/Perl function "trigger_data"
-NOTICE:  $_TD->{new} = {'i' => '2', 'v' => 'insert'}
+NOTICE:  $_TD->{new} = {'foo' => {'rfoo' => {'i' => '2'}}, 'i' => '2', 'v' => 'insert'}
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{relid} = 'bogus:12345'
 CONTEXT:  PL/Perl function "trigger_data"
@@ -151,7 +160,7 @@ NOTICE:  $_TD->{table_schema} = 'public'
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{when} = 'INSTEAD OF'
 CONTEXT:  PL/Perl function "trigger_data"
-update trigger_test_view set v = 'update' where i = 1;
+update trigger_test_view set v = 'update', foo = '("(3)")' where i = 1;
 NOTICE:  $_TD->{argc} = '2'
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{args} = ['24', 'skidoo view']
@@ -162,9 +171,9 @@ NOTICE:  $_TD->{level} = 'ROW'
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{name} = 'show_trigger_data_trig'
 CONTEXT:  PL/Perl function "trigger_data"
-NOTICE:  $_TD->{new} = {'i' => '1', 'v' => 'update'}
+NOTICE:  $_TD->{new} = {'foo' => {'rfoo' => {'i' => '3'}}, 'i' => '1', 'v' => 'update'}
 CONTEXT:  PL/Perl function "trigger_data"
-NOTICE:  $_TD->{old} = {'i' => '1', 'v' => 'insert'}
+NOTICE:  $_TD->{old} = {'foo' => {'rfoo' => {'i' => '1'}}, 'i' => '1', 'v' => 'insert'}
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{relid} = 'bogus:12345'
 CONTEXT:  PL/Perl function "trigger_data"
@@ -187,7 +196,7 @@ NOTICE:  $_TD->{level} = 'ROW'
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{name} = 'show_trigger_data_trig'
 CONTEXT:  PL/Perl function "trigger_data"
-NOTICE:  $_TD->{old} = {'i' => '1', 'v' => 'insert'}
+NOTICE:  $_TD->{old} = {'foo' => {'rfoo' => {'i' => '1'}}, 'i' => '1', 'v' => 'insert'}
 CONTEXT:  PL/Perl function "trigger_data"
 NOTICE:  $_TD->{relid} = 'bogus:12345'
 CONTEXT:  PL/Perl function "trigger_data"
@@ -211,6 +220,7 @@ CREATE OR REPLACE FUNCTION valid_id() RETURNS trigger AS $$
     elsif ($_TD->{new}{v} ne "immortal")
     {
         $_TD->{new}{v} .= "(modified by trigger)";
+		$_TD->{new}{foo}{rfoo}{i}++;
         return "MODIFY"; # Modify tuple and proceed INSERT/UPDATE command
     }
     else
@@ -220,29 +230,29 @@ CREATE OR REPLACE FUNCTION valid_id() RETURNS trigger AS $$
 $$ LANGUAGE plperl;
 CREATE TRIGGER "test_valid_id_trig" BEFORE INSERT OR UPDATE ON trigger_test
 FOR EACH ROW EXECUTE PROCEDURE "valid_id"();
-INSERT INTO trigger_test (i, v) VALUES (1,'first line');
-INSERT INTO trigger_test (i, v) VALUES (2,'second line');
-INSERT INTO trigger_test (i, v) VALUES (3,'third line');
-INSERT INTO trigger_test (i, v) VALUES (4,'immortal');
+INSERT INTO trigger_test (i, v, foo) VALUES (1,'first line', '("(1)")');
+INSERT INTO trigger_test (i, v, foo) VALUES (2,'second line', '("(2)")');
+INSERT INTO trigger_test (i, v, foo) VALUES (3,'third line', '("(3)")');
+INSERT INTO trigger_test (i, v, foo) VALUES (4,'immortal', '("(4)")');
 INSERT INTO trigger_test (i, v) VALUES (101,'bad id');
 SELECT * FROM trigger_test;
- i |                v                 
----+----------------------------------
- 1 | first line(modified by trigger)
- 2 | second line(modified by trigger)
- 3 | third line(modified by trigger)
- 4 | immortal
+ i |                v                 |   foo   
+---+----------------------------------+---------
+ 1 | first line(modified by trigger)  | ("(2)")
+ 2 | second line(modified by trigger) | ("(3)")
+ 3 | third line(modified by trigger)  | ("(4)")
+ 4 | immortal                         | ("(4)")
 (4 rows)
 
 UPDATE trigger_test SET i = 5 where i=3;
 UPDATE trigger_test SET i = 100 where i=1;
 SELECT * FROM trigger_test;
- i |                          v                           
----+------------------------------------------------------
- 1 | first line(modified by trigger)
- 2 | second line(modified by trigger)
- 4 | immortal
- 5 | third line(modified by trigger)(modified by trigger)
+ i |                          v                           |   foo   
+---+------------------------------------------------------+---------
+ 1 | first line(modified by trigger)                      | ("(2)")
+ 2 | second line(modified by trigger)                     | ("(3)")
+ 4 | immortal                                             | ("(4)")
+ 5 | third line(modified by trigger)(modified by trigger) | ("(5)")
 (4 rows)
 
 CREATE OR REPLACE FUNCTION immortal() RETURNS trigger AS $$
@@ -259,9 +269,9 @@ CREATE TRIGGER "immortal_trig" BEFORE DELETE ON trigger_test
 FOR EACH ROW EXECUTE PROCEDURE immortal('immortal');
 DELETE FROM trigger_test;
 SELECT * FROM trigger_test;
- i |    v     
----+----------
- 4 | immortal
+ i |    v     |   foo   
+---+----------+---------
+ 4 | immortal | ("(4)")
 (1 row)
 
 CREATE FUNCTION direct_trigger() RETURNS trigger AS $$
diff --git a/src/pl/plperl/expected/plperl_util.out b/src/pl/plperl/expected/plperl_util.out
index 6f16669b261..7cd027f33ec 100644
--- a/src/pl/plperl/expected/plperl_util.out
+++ b/src/pl/plperl/expected/plperl_util.out
@@ -169,3 +169,21 @@ select perl_looks_like_number();
  '': not number
 (11 rows)
 
+-- test encode_typed_literal
+create type perl_foo as (a integer, b text[]);
+create type perl_bar as (c perl_foo[]);
+create or replace function perl_encode_typed_literal() returns setof text language plperl as $$
+	return_next encode_typed_literal(undef, 'text');
+	return_next encode_typed_literal([[1,2,3],[3,2,1],[1,3,2]], 'integer[]');
+	return_next encode_typed_literal({a => 1, b => ['PL','/','Perl']}, 'perl_foo');
+	return_next encode_typed_literal({c => [{a => 9, b => ['PostgreSQL']}, {b => ['Postgres'], a => 1}]}, 'perl_bar');
+$$;
+select perl_encode_typed_literal();
+           perl_encode_typed_literal           
+-----------------------------------------------
+ 
+ {{1,2,3},{3,2,1},{1,3,2}}
+ (1,"{PL,/,Perl}")
+ ("{""(9,{PostgreSQL})"",""(1,{Postgres})""}")
+(4 rows)
+
diff --git a/src/pl/plperl/plc_perlboot.pl b/src/pl/plperl/plc_perlboot.pl
index ebf6b4b100f..67c656086cb 100644
--- a/src/pl/plperl/plc_perlboot.pl
+++ b/src/pl/plperl/plc_perlboot.pl
@@ -5,8 +5,45 @@ use vars qw(%_SHARED);
 
 PostgreSQL::InServer::Util::bootstrap();
 
-package PostgreSQL::InServer;
+# globals
+
+sub ::is_array_ref {
+	return ref($_[0]) =~ m/^(?:PostgreSQL::InServer::)?ARRAY$/;
+}
 
+sub ::encode_array_literal {
+	my ($arg, $delim) = @_;
+	return $arg unless(::is_array_ref($arg));
+	$delim = ', ' unless defined $delim;
+	my $res = '';
+	foreach my $elem (@$arg) {
+		$res .= $delim if length $res;
+		if (ref $elem) {
+			$res .= ::encode_array_literal($elem, $delim);
+		}
+		elsif (defined $elem) {
+			(my $str = $elem) =~ s/(["\\])/\\$1/g;
+			$res .= qq("$str");
+		}
+		else {
+			$res .= 'NULL';
+		}
+	}
+	return qq({$res});
+}
+
+sub ::encode_array_constructor {
+	my $arg = shift;
+	return ::quote_nullable($arg) unless ::is_array_ref($arg);
+	my $res = join ", ", map {
+		(ref $_) ? ::encode_array_constructor($_)
+		         : ::quote_nullable($_)
+	} @$arg;
+	return "ARRAY[$res]";
+}
+
+{
+package PostgreSQL::InServer;
 use strict;
 use warnings;
 
@@ -43,35 +80,26 @@ sub mkfunc {
 	return $ret;
 }
 
-sub ::encode_array_literal {
-	my ($arg, $delim) = @_;
-	return $arg
-		if ref $arg ne 'ARRAY';
-	$delim = ', ' unless defined $delim;
-	my $res = '';
-	foreach my $elem (@$arg) {
-		$res .= $delim if length $res;
-		if (ref $elem) {
-			$res .= ::encode_array_literal($elem, $delim);
-		}
-		elsif (defined $elem) {
-			(my $str = $elem) =~ s/(["\\])/\\$1/g;
-			$res .= qq("$str");
-		}
-		else {
-			$res .= 'NULL';
-		}
-	}
-	return qq({$res});
+1;
 }
 
-sub ::encode_array_constructor {
-	my $arg = shift;
-	return ::quote_nullable($arg)
-		if ref $arg ne 'ARRAY';
-	my $res = join ", ", map {
-		(ref $_) ? ::encode_array_constructor($_)
-		         : ::quote_nullable($_)
-	} @$arg;
-	return "ARRAY[$res]";
+{
+package PostgreSQL::InServer::ARRAY;
+use strict;
+use warnings;
+
+use overload
+	'""'=>\&to_str,
+	'@{}'=>\&to_arr;
+
+sub to_str {
+	my $self = shift;
+	return ::encode_typed_literal($self->{'array'}, $self->{'typeoid'});
+}
+
+sub to_arr {
+	return shift->{'array'};
+}
+
+1;
 }
diff --git a/src/pl/plperl/plperl.c b/src/pl/plperl/plperl.c
index 5bc8db76472..5f40f1e501a 100644
--- a/src/pl/plperl/plperl.c
+++ b/src/pl/plperl/plperl.c
@@ -109,6 +109,7 @@ typedef struct plperl_proc_desc
 	int			nargs;
 	FmgrInfo	arg_out_func[FUNC_MAX_ARGS];
 	bool		arg_is_rowtype[FUNC_MAX_ARGS];
+	Oid			arg_arraytype[FUNC_MAX_ARGS];	/* InvalidOid if not an array */
 	SV		   *reference;
 } plperl_proc_desc;
 
@@ -179,6 +180,19 @@ typedef struct plperl_query_entry
 } plperl_query_entry;
 
 /**********************************************************************
+ * Information for PostgreSQL - Perl array conversion.
+ **********************************************************************/
+typedef struct plperl_array_info
+{
+	int			ndims;
+	bool		elem_is_rowtype;	/* 't' if element type is a rowtype */
+	Datum	   *elements;
+	bool	   *nulls;
+	int		   *nelems;
+	FmgrInfo	proc;
+} plperl_array_info;
+
+/**********************************************************************
  * Global data
  **********************************************************************/
 
@@ -221,6 +235,19 @@ static Datum plperl_trigger_handler(PG_FUNCTION_ARGS);
 static plperl_proc_desc *compile_plperl_function(Oid fn_oid, bool is_trigger);
 
 static SV  *plperl_hash_from_tuple(HeapTuple tuple, TupleDesc tupdesc);
+static SV  *plperl_hash_from_datum(Datum attr);
+static SV  *plperl_ref_from_pg_array(Datum arg, Oid typid);
+static SV  *split_array(plperl_array_info *info, int first, int last, int nest);
+static SV  *make_array_ref(plperl_array_info *info, int first, int last);
+static SV  *get_perl_array_ref(SV *sv);
+static Datum plperl_sv_to_datum(SV *sv, FmgrInfo *func, Oid typid,
+				   Oid typioparam, int32 typmod, bool *isnull);
+static void _sv_to_datum_finfo(FmgrInfo *fcinfo, Oid typid, Oid *typioparam);
+static Datum plperl_array_to_datum(SV *src, Oid typid);
+static ArrayBuildState *_array_to_datum(AV *av, int *ndims, int *dims,
+			  int cur_depth, ArrayBuildState *astate, Oid typid, Oid atypid);
+static Datum plperl_hash_to_datum(SV *src, TupleDesc td);
+
 static void plperl_init_shared_libs(pTHX);
 static void plperl_trusted_init(void);
 static void plperl_untrusted_init(void);
@@ -960,12 +987,14 @@ static HeapTuple
 plperl_build_tuple_result(HV *perlhash, AttInMetadata *attinmeta)
 {
 	TupleDesc	td = attinmeta->tupdesc;
-	char	  **values;
+	Datum	   *values;
+	bool	   *nulls;
 	HE		   *he;
 	HeapTuple	tup;
-	int			i;
 
-	values = (char **) palloc0(td->natts * sizeof(char *));
+	values = palloc0(sizeof(Datum) * td->natts);
+	nulls = palloc(sizeof(bool) * td->natts);
+	memset(nulls, true, sizeof(bool) * td->natts);
 
 	hv_iterinit(perlhash);
 	while ((he = hv_iternext(perlhash)))
@@ -973,65 +1002,378 @@ plperl_build_tuple_result(HV *perlhash, AttInMetadata *attinmeta)
 		SV		   *val = HeVAL(he);
 		char	   *key = hek2cstr(he);
 		int			attn = SPI_fnumber(td, key);
+		bool		isnull;
 
 		if (attn <= 0 || td->attrs[attn - 1]->attisdropped)
 			ereport(ERROR,
 					(errcode(ERRCODE_UNDEFINED_COLUMN),
 					 errmsg("Perl hash contains nonexistent column \"%s\"",
 							key)));
-		if (SvOK(val))
-		{
-			values[attn - 1] = sv2cstr(val);
-		}
+
+		values[attn - 1] = plperl_sv_to_datum(val,
+											  NULL,
+											  td->attrs[attn - 1]->atttypid,
+											  InvalidOid,
+											  td->attrs[attn - 1]->atttypmod,
+											  &isnull);
+		nulls[attn - 1] = isnull;
 
 		pfree(key);
 	}
 	hv_iterinit(perlhash);
 
-	tup = BuildTupleFromCStrings(attinmeta, values);
+	tup = heap_form_tuple(td, values, nulls);
+	pfree(values);
+	pfree(nulls);
+	return tup;
+}
 
-	for (i = 0; i < td->natts; i++)
+/* convert a hash reference to a datum */
+static Datum
+plperl_hash_to_datum(SV *src, TupleDesc td)
+{
+	AttInMetadata *attinmeta = TupleDescGetAttInMetadata(td);
+	HeapTuple	tup = plperl_build_tuple_result((HV *) SvRV(src), attinmeta);
+
+	return HeapTupleGetDatum(tup);
+}
+
+/*
+ * if we are an array ref return the reference. this is special in that if we
+ * are a PostgreSQL::InServer::ARRAY object we will return the 'magic' array.
+ */
+static SV  *
+get_perl_array_ref(SV *sv)
+{
+	if (SvOK(sv) && SvROK(sv))
 	{
-		if (values[i])
-			pfree(values[i]);
+		if (SvTYPE(SvRV(sv)) == SVt_PVAV)
+			return sv;
+		else if (sv_isa(sv, "PostgreSQL::InServer::ARRAY"))
+		{
+			HV		   *hv = (HV *) SvRV(sv);
+			SV		  **sav = hv_fetch_string(hv, "array");
+
+			if (*sav && SvOK(*sav) && SvROK(*sav) &&
+				SvTYPE(SvRV(*sav)) == SVt_PVAV)
+				return *sav;
+
+			elog(ERROR, "could not get array reference from PostgreSQL::InServer::ARRAY object");
+		}
 	}
-	pfree(values);
+	return NULL;
+}
 
-	return tup;
+/*
+ * helper function for plperl_array_to_datum, does the main recursing
+ */
+static ArrayBuildState *
+_array_to_datum(AV *av, int *ndims, int *dims, int cur_depth,
+				ArrayBuildState *astate, Oid typid, Oid atypid)
+{
+	int			i = 0;
+	int			len = av_len(av) + 1;
+
+	if (len == 0)
+		astate = accumArrayResult(astate, (Datum) 0, true, atypid, NULL);
+
+	for (i = 0; i < len; i++)
+	{
+		SV		  **svp = av_fetch(av, i, FALSE);
+		SV		   *sav = svp ? get_perl_array_ref(*svp) : NULL;
+
+		if (sav)
+		{
+			AV		   *nav = (AV *) SvRV(sav);
+
+			if (cur_depth + 1 > MAXDIM)
+				ereport(ERROR,
+						(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+						 errmsg("number of array dimensions (%d) exceeds the maximum allowed (%d)",
+								cur_depth + 1, MAXDIM)));
+
+			/* size based off the first element */
+			if (i == 0 && *ndims == cur_depth)
+			{
+				dims[*ndims] = av_len(nav) + 1;
+				(*ndims)++;
+			}
+			else
+			{
+				if (av_len(nav) + 1 != dims[cur_depth])
+					ereport(ERROR,
+							(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
+							 errmsg("multidimensional arrays must have array expressions with matching dimensions")));
+			}
+
+			astate = _array_to_datum(nav, ndims, dims, cur_depth + 1, astate,
+									 typid, atypid);
+		}
+		else
+		{
+			bool		isnull;
+			Datum		dat = plperl_sv_to_datum(svp ? *svp : NULL, NULL,
+												 atypid, 0, -1, &isnull);
+
+			astate = accumArrayResult(astate, dat, isnull, atypid, NULL);
+		}
+	}
+
+	return astate;
+}
+
+/*
+ * convert perl array ref to a datum
+ */
+static Datum
+plperl_array_to_datum(SV *src, Oid typid)
+{
+	ArrayBuildState *astate = NULL;
+	Oid			atypid;
+	int			dims[MAXDIM];
+	int			lbs[MAXDIM];
+	int			ndims = 1;
+	int			i;
+
+	atypid = get_element_type(typid);
+	if (!atypid)
+		atypid = typid;
+
+	memset(dims, 0, sizeof(dims));
+	dims[0] = av_len((AV *) SvRV(src)) + 1;
+
+	astate = _array_to_datum((AV *) SvRV(src), &ndims, dims, 1, astate, typid,
+							 atypid);
+
+	for (i = 0; i < ndims; i++)
+		lbs[i] = 1;
+
+	return makeMdArrayResult(astate, ndims, dims, lbs, CurrentMemoryContext, true);
+}
+
+static void
+_sv_to_datum_finfo(FmgrInfo *fcinfo, Oid typid, Oid *typioparam)
+{
+	Oid			typinput;
+
+	/* XXX would be better to cache these lookups */
+	getTypeInputInfo(typid,
+					 &typinput, typioparam);
+	fmgr_info(typinput, fcinfo);
+}
+
+/*
+ * convert a sv to datum
+ * fcinfo and typioparam are optional and will be looked-up if needed
+ */
+static Datum
+plperl_sv_to_datum(SV *sv, FmgrInfo *finfo, Oid typid, Oid typioparam,
+				   int32 typmod, bool *isnull)
+{
+	FmgrInfo	tmp;
+
+	/* we might recurse */
+	check_stack_depth();
+
+	if (isnull)
+		*isnull = false;
+
+	if (!sv || !SvOK(sv))
+	{
+		if (!finfo)
+		{
+			_sv_to_datum_finfo(&tmp, typid, &typioparam);
+			finfo = &tmp;
+		}
+		if (isnull)
+			*isnull = true;
+		return InputFunctionCall(finfo, NULL, typioparam, typmod);
+	}
+	else if (SvROK(sv))
+	{
+		SV		   *sav = get_perl_array_ref(sv);
+
+		if (sav)
+		{
+			return plperl_array_to_datum(sav, typid);
+		}
+		else if (SvTYPE(SvRV(sv)) == SVt_PVHV)
+		{
+			TupleDesc	td = lookup_rowtype_tupdesc(typid, typmod);
+			Datum		ret = plperl_hash_to_datum(sv, td);
+
+			ReleaseTupleDesc(td);
+			return ret;
+		}
+
+		ereport(ERROR,
+				(errcode(ERRCODE_DATATYPE_MISMATCH),
+				 errmsg("PL/Perl function must return reference to hash or array")));
+		return (Datum) 0;		/* shut up compiler */
+	}
+	else
+	{
+		Datum		ret;
+		char	   *str = sv2cstr(sv);
+
+		if (!finfo)
+		{
+			_sv_to_datum_finfo(&tmp, typid, &typioparam);
+			finfo = &tmp;
+		}
+
+		ret = InputFunctionCall(finfo, str, typioparam, typmod);
+		pfree(str);
+
+		return ret;
+	}
+}
+
+/* Convert the perl SV to a string returned by the type output function */
+char *
+plperl_sv_to_literal(SV *sv, char *fqtypename)
+{
+	Datum		str = CStringGetDatum(fqtypename);
+	Oid			typid = DirectFunctionCall1(regtypein, str);
+	Oid			typoutput;
+	Datum		datum;
+	bool		typisvarlena,
+				isnull;
+
+	if (!OidIsValid(typid))
+		elog(ERROR, "lookup failed for type %s", fqtypename);
+
+	datum = plperl_sv_to_datum(sv, NULL, typid, 0, -1, &isnull);
+
+	if (isnull)
+		return NULL;
+
+	getTypeOutputInfo(typid,
+					  &typoutput, &typisvarlena);
+
+	return OidOutputFunctionCall(typoutput, datum);
 }
 
 /*
- * convert perl array to postgres string representation
+ * Convert PostgreSQL array datum to a perl array reference.
+ *
+ * typid is arg's OID, which must be an array type.
  */
 static SV  *
-plperl_convert_to_pg_array(SV *src)
+plperl_ref_from_pg_array(Datum arg, Oid typid)
 {
-	SV		   *rv;
-	int			count;
+	ArrayType  *ar = DatumGetArrayTypeP(arg);
+	Oid			elementtype = ARR_ELEMTYPE(ar);
+	int16		typlen;
+	bool		typbyval;
+	char		typalign,
+				typdelim;
+	Oid			typioparam;
+	Oid			typoutputfunc;
+	int			i,
+				nitems,
+			   *dims;
+	plperl_array_info *info;
+	SV		   *av;
+	HV		   *hv;
 
-	dSP;
+	info = palloc(sizeof(plperl_array_info));
 
-	PUSHMARK(SP);
-	XPUSHs(src);
-	PUTBACK;
+	/* get element type information, including output conversion function */
+	get_type_io_data(elementtype, IOFunc_output,
+					 &typlen, &typbyval, &typalign,
+					 &typdelim, &typioparam, &typoutputfunc);
 
-	count = perl_call_pv("::encode_array_literal", G_SCALAR);
+	perm_fmgr_info(typoutputfunc, &info->proc);
 
-	SPAGAIN;
+	info->elem_is_rowtype = type_is_rowtype(elementtype);
 
-	if (count != 1)
-		elog(ERROR, "unexpected encode_array_literal failure");
+	/* Get the number and bounds of array dimensions */
+	info->ndims = ARR_NDIM(ar);
+	dims = ARR_DIMS(ar);
 
-	rv = POPs;
+	deconstruct_array(ar, elementtype, typlen, typbyval,
+					  typalign, &info->elements, &info->nulls,
+					  &nitems);
 
-	PUTBACK;
+	/* Get total number of elements in each dimension */
+	info->nelems = palloc(sizeof(int) * info->ndims);
+	info->nelems[0] = nitems;
+	for (i = 1; i < info->ndims; i++)
+		info->nelems[i] = info->nelems[i - 1] / dims[i - 1];
 
-	return rv;
+	av = split_array(info, 0, nitems, 0);
+
+	hv = newHV();
+	(void) hv_store(hv, "array", 5, av, 0);
+	(void) hv_store(hv, "typeoid", 7, newSViv(typid), 0);
+
+	return sv_bless(newRV_noinc((SV *) hv),
+					gv_stashpv("PostgreSQL::InServer::ARRAY", 0));
 }
 
+/*
+ * Recursively form array references from splices of the initial array
+ */
+static SV  *
+split_array(plperl_array_info *info, int first, int last, int nest)
+{
+	int			i;
+	AV		   *result;
 
-/* Set up the arguments for a trigger call. */
+	/* since this function recurses, it could be driven to stack overflow */
+	check_stack_depth();
+
+	/*
+	 * Base case, return a reference to a single-dimensional array
+	 */
+	if (nest >= info->ndims - 1)
+		return make_array_ref(info, first, last);
+
+	result = newAV();
+	for (i = first; i < last; i += info->nelems[nest + 1])
+	{
+		/* Recursively form references to arrays of lower dimensions */
+		SV		   *ref = split_array(info, i, i + info->nelems[nest + 1], nest + 1);
+
+		av_push(result, ref);
+	}
+	return newRV_noinc((SV *) result);
+}
+
+/*
+ * Create a Perl reference from a one-dimensional C array, converting
+ * composite type elements to hash references.
+ */
+static SV  *
+make_array_ref(plperl_array_info *info, int first, int last)
+{
+	int			i;
+	AV		   *result = newAV();
+
+	for (i = first; i < last; i++)
+	{
+		if (info->nulls[i])
+			av_push(result, &PL_sv_undef);
+		else
+		{
+			Datum		itemvalue = info->elements[i];
 
+			/* Handle composite type elements */
+			if (info->elem_is_rowtype)
+				av_push(result, plperl_hash_from_datum(itemvalue));
+			else
+			{
+				char	   *val = OutputFunctionCall(&info->proc, itemvalue);
+
+				av_push(result, cstr2sv(val));
+			}
+		}
+	}
+	return newRV_noinc((SV *) result);
+}
+
+/* Set up the arguments for a trigger call. */
 static SV  *
 plperl_trigger_build_args(FunctionCallInfo fcinfo)
 {
@@ -1174,12 +1516,9 @@ plperl_modify_tuple(HV *hvTD, TriggerData *tdata, HeapTuple otup)
 	hv_iterinit(hvNew);
 	while ((he = hv_iternext(hvNew)))
 	{
-		Oid			typinput;
-		Oid			typioparam;
-		int32		atttypmod;
-		FmgrInfo	finfo;
-		SV		   *val = HeVAL(he);
+		bool		isnull;
 		char	   *key = hek2cstr(he);
+		SV		   *val = HeVAL(he);
 		int			attn = SPI_fnumber(tupdesc, key);
 
 		if (attn <= 0 || tupdesc->attrs[attn - 1]->attisdropped)
@@ -1187,30 +1526,15 @@ plperl_modify_tuple(HV *hvTD, TriggerData *tdata, HeapTuple otup)
 					(errcode(ERRCODE_UNDEFINED_COLUMN),
 					 errmsg("Perl hash contains nonexistent column \"%s\"",
 							key)));
-		/* XXX would be better to cache these lookups */
-		getTypeInputInfo(tupdesc->attrs[attn - 1]->atttypid,
-						 &typinput, &typioparam);
-		fmgr_info(typinput, &finfo);
-		atttypmod = tupdesc->attrs[attn - 1]->atttypmod;
-		if (SvOK(val))
-		{
-			char	   *str = sv2cstr(val);
-
-			modvalues[slotsused] = InputFunctionCall(&finfo,
-													 str,
-													 typioparam,
-													 atttypmod);
-			modnulls[slotsused] = ' ';
-			pfree(str);
-		}
-		else
-		{
-			modvalues[slotsused] = InputFunctionCall(&finfo,
-													 NULL,
-													 typioparam,
-													 atttypmod);
-			modnulls[slotsused] = 'n';
-		}
+
+		modvalues[slotsused] = plperl_sv_to_datum(val,
+												  NULL,
+										  tupdesc->attrs[attn - 1]->atttypid,
+												  InvalidOid,
+										 tupdesc->attrs[attn - 1]->atttypmod,
+												  &isnull);
+
+		modnulls[slotsused] = isnull ? 'n' : ' ';
 		modattrs[slotsused] = attn;
 		slotsused++;
 
@@ -1530,7 +1854,6 @@ plperl_call_perl_func(plperl_proc_desc *desc, FunctionCallInfo fcinfo)
 	SV		   *retval;
 	int			i;
 	int			count;
-	SV		   *sv;
 
 	ENTER;
 	SAVETMPS;
@@ -1544,35 +1867,27 @@ plperl_call_perl_func(plperl_proc_desc *desc, FunctionCallInfo fcinfo)
 			PUSHs(&PL_sv_undef);
 		else if (desc->arg_is_rowtype[i])
 		{
-			HeapTupleHeader td;
-			Oid			tupType;
-			int32		tupTypmod;
-			TupleDesc	tupdesc;
-			HeapTupleData tmptup;
-			SV		   *hashref;
-
-			td = DatumGetHeapTupleHeader(fcinfo->arg[i]);
-			/* Extract rowtype info and find a tupdesc */
-			tupType = HeapTupleHeaderGetTypeId(td);
-			tupTypmod = HeapTupleHeaderGetTypMod(td);
-			tupdesc = lookup_rowtype_tupdesc(tupType, tupTypmod);
-			/* Build a temporary HeapTuple control structure */
-			tmptup.t_len = HeapTupleHeaderGetDatumLength(td);
-			tmptup.t_data = td;
-
-			hashref = plperl_hash_from_tuple(&tmptup, tupdesc);
-			PUSHs(sv_2mortal(hashref));
-			ReleaseTupleDesc(tupdesc);
+			SV		   *sv = plperl_hash_from_datum(fcinfo->arg[i]);
+
+			PUSHs(sv_2mortal(sv));
 		}
 		else
 		{
-			char	   *tmp;
+			SV		   *sv;
+
+			if (OidIsValid(desc->arg_arraytype[i]))
+				sv = plperl_ref_from_pg_array(fcinfo->arg[i], desc->arg_arraytype[i]);
+			else
+			{
+				char	   *tmp;
+
+				tmp = OutputFunctionCall(&(desc->arg_out_func[i]),
+										 fcinfo->arg[i]);
+				sv = cstr2sv(tmp);
+				pfree(tmp);
+			}
 
-			tmp = OutputFunctionCall(&(desc->arg_out_func[i]),
-									 fcinfo->arg[i]);
-			sv = cstr2sv(tmp);
 			PUSHs(sv_2mortal(sv));
-			pfree(tmp);
 		}
 	}
 	PUTBACK;
@@ -1677,8 +1992,8 @@ plperl_func_handler(PG_FUNCTION_ARGS)
 	SV		   *perlret;
 	Datum		retval;
 	ReturnSetInfo *rsi;
-	SV		   *array_ret = NULL;
 	ErrorContextCallback pl_error_context;
+	bool		has_retval = false;
 
 	/*
 	 * Create the call_data beforing connecting to SPI, so that it is not
@@ -1728,19 +2043,20 @@ plperl_func_handler(PG_FUNCTION_ARGS)
 
 	if (prodesc->fn_retisset)
 	{
+		SV		   *sav;
+
 		/*
 		 * If the Perl function returned an arrayref, we pretend that it
 		 * called return_next() for each element of the array, to handle old
 		 * SRFs that didn't know about return_next(). Any other sort of return
 		 * value is an error, except undef which means return an empty set.
 		 */
-		if (SvOK(perlret) &&
-			SvROK(perlret) &&
-			SvTYPE(SvRV(perlret)) == SVt_PVAV)
+		sav = get_perl_array_ref(perlret);
+		if (sav)
 		{
 			int			i = 0;
 			SV		  **svp = 0;
-			AV		   *rav = (AV *) SvRV(perlret);
+			AV		   *rav = (AV *) SvRV(sav);
 
 			while ((svp = av_fetch(rav, i, FALSE)) != NULL)
 			{
@@ -1763,22 +2079,18 @@ plperl_func_handler(PG_FUNCTION_ARGS)
 			rsi->setDesc = current_call_data->ret_tdesc;
 		}
 		retval = (Datum) 0;
+		has_retval = true;
 	}
 	else if (!SvOK(perlret))
 	{
 		/* Return NULL if Perl code returned undef */
 		if (rsi && IsA(rsi, ReturnSetInfo))
 			rsi->isDone = ExprEndResult;
-		retval = InputFunctionCall(&prodesc->result_in_func, NULL,
-								   prodesc->result_typioparam, -1);
-		fcinfo->isnull = true;
 	}
 	else if (prodesc->fn_retistuple)
 	{
 		/* Return a perl hash converted to a Datum */
 		TupleDesc	td;
-		AttInMetadata *attinmeta;
-		HeapTuple	tup;
 
 		if (!SvOK(perlret) || !SvROK(perlret) ||
 			SvTYPE(SvRV(perlret)) != SVt_PVHV)
@@ -1798,35 +2110,26 @@ plperl_func_handler(PG_FUNCTION_ARGS)
 							"that cannot accept type record")));
 		}
 
-		attinmeta = TupleDescGetAttInMetadata(td);
-		tup = plperl_build_tuple_result((HV *) SvRV(perlret), attinmeta);
-		retval = HeapTupleGetDatum(tup);
+		retval = plperl_hash_to_datum(perlret, td);
+		has_retval = true;
 	}
-	else
-	{
-		/* Return a perl string converted to a Datum */
-		char	   *str;
 
-		if (prodesc->fn_retisarray && SvROK(perlret) &&
-			SvTYPE(SvRV(perlret)) == SVt_PVAV)
-		{
-			array_ret = plperl_convert_to_pg_array(perlret);
-			SvREFCNT_dec(perlret);
-			perlret = array_ret;
-		}
+	if (!has_retval)
+	{
+		bool		isnull;
 
-		str = sv2cstr(perlret);
-		retval = InputFunctionCall(&prodesc->result_in_func,
-								   str,
-								   prodesc->result_typioparam, -1);
-		pfree(str);
+		retval = plperl_sv_to_datum(perlret,
+									&prodesc->result_in_func,
+									prodesc->result_oid,
+									prodesc->result_typioparam, -1, &isnull);
+		fcinfo->isnull = isnull;
+		has_retval = true;
 	}
 
 	/* Restore the previous error callback */
 	error_context_stack = pl_error_context.previous;
 
-	if (array_ret == NULL)
-		SvREFCNT_dec(perlret);
+	SvREFCNT_dec(perlret);
 
 	return retval;
 }
@@ -2181,6 +2484,12 @@ compile_plperl_function(Oid fn_oid, bool is_trigger)
 								   &(prodesc->arg_out_func[i]));
 				}
 
+				/* Identify array attributes */
+				if (typeStruct->typelem != 0 && typeStruct->typlen == -1)
+					prodesc->arg_arraytype[i] = procStruct->proargtypes.values[i];
+				else
+					prodesc->arg_arraytype[i] = InvalidOid;
+
 				ReleaseSysCache(typeTup);
 			}
 		}
@@ -2234,26 +2543,54 @@ compile_plperl_function(Oid fn_oid, bool is_trigger)
 	return prodesc;
 }
 
+/* Build a hash from a given composite/row datum */
+static SV  *
+plperl_hash_from_datum(Datum attr)
+{
+	HeapTupleHeader td;
+	Oid			tupType;
+	int32		tupTypmod;
+	TupleDesc	tupdesc;
+	HeapTupleData tmptup;
+	SV		   *sv;
 
-/* Build a hash from all attributes of a given tuple. */
+	td = DatumGetHeapTupleHeader(attr);
+
+	/* Extract rowtype info and find a tupdesc */
+	tupType = HeapTupleHeaderGetTypeId(td);
+	tupTypmod = HeapTupleHeaderGetTypMod(td);
+	tupdesc = lookup_rowtype_tupdesc(tupType, tupTypmod);
+
+	/* Build a temporary HeapTuple control structure */
+	tmptup.t_len = HeapTupleHeaderGetDatumLength(td);
+	tmptup.t_data = td;
 
+	sv = plperl_hash_from_tuple(&tmptup, tupdesc);
+	ReleaseTupleDesc(tupdesc);
+
+	return sv;
+}
+
+/* Build a hash from all attributes of a given tuple. */
 static SV  *
 plperl_hash_from_tuple(HeapTuple tuple, TupleDesc tupdesc)
 {
 	HV		   *hv;
 	int			i;
 
+	/* since this function recurses, it could be driven to stack overflow */
+	check_stack_depth();
+
 	hv = newHV();
 	hv_ksplit(hv, tupdesc->natts);		/* pre-grow the hash */
 
 	for (i = 0; i < tupdesc->natts; i++)
 	{
 		Datum		attr;
-		bool		isnull;
+		bool		isnull,
+					typisvarlena;
 		char	   *attname;
-		char	   *outputstr;
 		Oid			typoutput;
-		bool		typisvarlena;
 
 		if (tupdesc->attrs[i]->attisdropped)
 			continue;
@@ -2264,21 +2601,38 @@ plperl_hash_from_tuple(HeapTuple tuple, TupleDesc tupdesc)
 		if (isnull)
 		{
 			/* Store (attname => undef) and move on. */
-			hv_store_string(hv, attname, newSV(0));
+			hv_store_string(hv, attname, &PL_sv_undef);
 			continue;
 		}
 
-		/* XXX should have a way to cache these lookups */
-		getTypeOutputInfo(tupdesc->attrs[i]->atttypid,
-						  &typoutput, &typisvarlena);
+		if (type_is_rowtype(tupdesc->attrs[i]->atttypid))
+		{
+			SV		   *sv = plperl_hash_from_datum(attr);
+
+			hv_store_string(hv, attname, sv);
+		}
+		else
+		{
+			SV		   *sv;
+
+			if (OidIsValid(get_base_element_type(tupdesc->attrs[i]->atttypid)))
+				sv = plperl_ref_from_pg_array(attr, tupdesc->attrs[i]->atttypid);
+			else
+			{
+				char	   *outputstr;
 
-		outputstr = OidOutputFunctionCall(typoutput, attr);
+				/* XXX should have a way to cache these lookups */
+				getTypeOutputInfo(tupdesc->attrs[i]->atttypid,
+								  &typoutput, &typisvarlena);
 
-		hv_store_string(hv, attname, cstr2sv(outputstr));
+				outputstr = OidOutputFunctionCall(typoutput, attr);
+				sv = cstr2sv(outputstr);
+				pfree(outputstr);
+			}
 
-		pfree(outputstr);
+			hv_store_string(hv, attname, sv);
+		}
 	}
-
 	return newRV_noinc((SV *) hv);
 }
 
@@ -2507,29 +2861,11 @@ plperl_return_next(SV *sv)
 		Datum		ret;
 		bool		isNull;
 
-		if (SvOK(sv))
-		{
-			char	   *str;
-
-			if (prodesc->fn_retisarray && SvROK(sv) &&
-				SvTYPE(SvRV(sv)) == SVt_PVAV)
-			{
-				sv = plperl_convert_to_pg_array(sv);
-			}
-
-			str = sv2cstr(sv);
-			ret = InputFunctionCall(&prodesc->result_in_func,
-									str,
-									prodesc->result_typioparam, -1);
-			isNull = false;
-			pfree(str);
-		}
-		else
-		{
-			ret = InputFunctionCall(&prodesc->result_in_func, NULL,
-									prodesc->result_typioparam, -1);
-			isNull = true;
-		}
+		ret = plperl_sv_to_datum(sv,
+								 &prodesc->result_in_func,
+								 prodesc->result_oid,
+								 prodesc->result_typioparam,
+								 -1, &isNull);
 
 		tuplestore_putvalues(current_call_data->tuple_store,
 							 current_call_data->ret_tdesc,
@@ -2910,7 +3246,7 @@ plperl_spi_exec_prepared(char *query, HV *attr, int argc, SV **argv)
 		if (attr != NULL)
 		{
 			sv = hv_fetch_string(attr, "limit");
-			if (*sv && SvIOK(*sv))
+			if (sv && *sv && SvIOK(*sv))
 				limit = SvIV(*sv);
 		}
 		/************************************************************
@@ -2929,25 +3265,14 @@ plperl_spi_exec_prepared(char *query, HV *attr, int argc, SV **argv)
 
 		for (i = 0; i < argc; i++)
 		{
-			if (SvOK(argv[i]))
-			{
-				char	   *str = sv2cstr(argv[i]);
-
-				argvalues[i] = InputFunctionCall(&qdesc->arginfuncs[i],
-												 str,
-												 qdesc->argtypioparams[i],
-												 -1);
-				nulls[i] = ' ';
-				pfree(str);
-			}
-			else
-			{
-				argvalues[i] = InputFunctionCall(&qdesc->arginfuncs[i],
-												 NULL,
-												 qdesc->argtypioparams[i],
-												 -1);
-				nulls[i] = 'n';
-			}
+			bool		isnull;
+
+			argvalues[i] = plperl_sv_to_datum(argv[i],
+											  &qdesc->arginfuncs[i],
+											  qdesc->argtypes[i],
+											  qdesc->argtypioparams[i],
+											  -1, &isnull);
+			nulls[i] = isnull ? 'n' : ' ';
 		}
 
 		/************************************************************
@@ -3065,25 +3390,14 @@ plperl_spi_query_prepared(char *query, int argc, SV **argv)
 
 		for (i = 0; i < argc; i++)
 		{
-			if (SvOK(argv[i]))
-			{
-				char	   *str = sv2cstr(argv[i]);
-
-				argvalues[i] = InputFunctionCall(&qdesc->arginfuncs[i],
-												 str,
-												 qdesc->argtypioparams[i],
-												 -1);
-				nulls[i] = ' ';
-				pfree(str);
-			}
-			else
-			{
-				argvalues[i] = InputFunctionCall(&qdesc->arginfuncs[i],
-												 NULL,
-												 qdesc->argtypioparams[i],
-												 -1);
-				nulls[i] = 'n';
-			}
+			bool		isnull;
+
+			argvalues[i] = plperl_sv_to_datum(argv[i],
+											  &qdesc->arginfuncs[i],
+											  qdesc->argtypes[i],
+											  qdesc->argtypioparams[i],
+											  -1, &isnull);
+			nulls[i] = isnull ? 'n' : ' ';
 		}
 
 		/************************************************************
diff --git a/src/pl/plperl/plperl.h b/src/pl/plperl/plperl.h
index 1e0bad101aa..65b27a344f0 100644
--- a/src/pl/plperl/plperl.h
+++ b/src/pl/plperl/plperl.h
@@ -59,6 +59,7 @@ HV		   *plperl_spi_exec_prepared(char *, HV *, int, SV **);
 SV		   *plperl_spi_query_prepared(char *, int, SV **);
 void		plperl_spi_freeplan(char *);
 void		plperl_spi_cursor_close(char *);
+char 	   *plperl_sv_to_literal(SV *, char *);
 
 
 
diff --git a/src/pl/plperl/sql/plperl.sql b/src/pl/plperl/sql/plperl.sql
index 22ac0bb4512..4aaca2a27c5 100644
--- a/src/pl/plperl/sql/plperl.sql
+++ b/src/pl/plperl/sql/plperl.sql
@@ -32,7 +32,8 @@ SELECT perl_set_int(5);
 SELECT * FROM perl_set_int(5);
 
 
-CREATE TYPE testrowperl AS (f1 integer, f2 text, f3 text);
+CREATE TYPE testnestperl AS (f5 integer[]);
+CREATE TYPE testrowperl AS (f1 integer, f2 text, f3 text, f4 testnestperl);
 
 CREATE OR REPLACE FUNCTION perl_row() RETURNS testrowperl AS $$
     return undef;
@@ -41,8 +42,9 @@ $$ LANGUAGE plperl;
 SELECT perl_row();
 SELECT * FROM perl_row();
 
+
 CREATE OR REPLACE FUNCTION perl_row() RETURNS testrowperl AS $$
-    return {f2 => 'hello', f1 => 1, f3 => 'world'};
+    return {f2 => 'hello', f1 => 1, f3 => 'world', 'f4' => { 'f5' => [[1]] } };
 $$ LANGUAGE plperl;
 
 SELECT perl_row();
@@ -60,7 +62,10 @@ CREATE OR REPLACE FUNCTION perl_set() RETURNS SETOF testrowperl AS $$
     return [
         { f1 => 1, f2 => 'Hello', f3 =>  'World' },
         undef,
-        { f1 => 3, f2 => 'Hello', f3 =>  'PL/Perl' }
+        { f1 => 3, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => {} },
+        { f1 => 4, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => { 'f5' => undef }},
+        { f1 => 5, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => { 'f5' => '{1}' }},
+        { f1 => 6, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => { 'f5' => [1] }},
     ];
 $$  LANGUAGE plperl;
 
@@ -70,31 +75,33 @@ SELECT * FROM perl_set();
 CREATE OR REPLACE FUNCTION perl_set() RETURNS SETOF testrowperl AS $$
     return [
         { f1 => 1, f2 => 'Hello', f3 =>  'World' },
-        { f1 => 2, f2 => 'Hello', f3 =>  'PostgreSQL' },
-        { f1 => 3, f2 => 'Hello', f3 =>  'PL/Perl' }
+        { f1 => 2, f2 => 'Hello', f3 =>  'PostgreSQL', 'f4' => undef },
+        { f1 => 3, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => {} },
+        { f1 => 4, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => { 'f5' => undef }},
+        { f1 => 5, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => { 'f5' => '{1}' }},
+        { f1 => 6, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => { 'f5' => [1] }},
+        { f1 => 7, f2 => 'Hello', f3 =>  'PL/Perl', 'f4' => '({1})' },
     ];
 $$  LANGUAGE plperl;
 
 SELECT perl_set();
 SELECT * FROM perl_set();
 
-
-
 CREATE OR REPLACE FUNCTION perl_record() RETURNS record AS $$
     return undef;
 $$ LANGUAGE plperl;
 
 SELECT perl_record();
 SELECT * FROM perl_record();
-SELECT * FROM perl_record() AS (f1 integer, f2 text, f3 text);
+SELECT * FROM perl_record() AS (f1 integer, f2 text, f3 text, f4 testnestperl);
 
 CREATE OR REPLACE FUNCTION perl_record() RETURNS record AS $$
-    return {f2 => 'hello', f1 => 1, f3 => 'world'};
+    return {f2 => 'hello', f1 => 1, f3 => 'world', 'f4' => { 'f5' => [1] } };
 $$ LANGUAGE plperl;
 
 SELECT perl_record();
 SELECT * FROM perl_record();
-SELECT * FROM perl_record() AS (f1 integer, f2 text, f3 text);
+SELECT * FROM perl_record() AS (f1 integer, f2 text, f3 text, f4 testnestperl);
 
 
 CREATE OR REPLACE FUNCTION perl_record_set() RETURNS SETOF record AS $$
@@ -297,7 +304,7 @@ SELECT * FROM recurse(3);
 
 
 ---
---- Test arrary return
+--- Test array return
 ---
 CREATE OR REPLACE FUNCTION  array_of_text() RETURNS TEXT[][]
 LANGUAGE plperl as $$
@@ -361,6 +368,24 @@ CREATE OR REPLACE FUNCTION perl_spi_prepared_bad(double precision) RETURNS doubl
 $$ LANGUAGE plperl;
 SELECT perl_spi_prepared_bad(4.35) as "double precision";
 
+-- Test with a row type
+CREATE OR REPLACE FUNCTION perl_spi_prepared() RETURNS INTEGER AS $$
+   my $x = spi_prepare('select $1::footype AS a', 'footype');
+   my $q = spi_exec_prepared( $x, '(1, 2)');
+   spi_freeplan($x);
+return $q->{rows}->[0]->{a}->{x};
+$$ LANGUAGE plperl;
+SELECT * from perl_spi_prepared();
+
+CREATE OR REPLACE FUNCTION perl_spi_prepared_row(footype) RETURNS footype AS $$
+   my $footype = shift;
+   my $x = spi_prepare('select $1 AS a', 'footype');
+   my $q = spi_exec_prepared( $x, {}, $footype );
+   spi_freeplan($x);
+return $q->{rows}->[0]->{a};
+$$ LANGUAGE plperl;
+SELECT * from perl_spi_prepared_row('(1, 2)');
+
 -- simple test of a DO block
 DO $$
   $a = 'This is a test';
diff --git a/src/pl/plperl/sql/plperl_array.sql b/src/pl/plperl/sql/plperl_array.sql
new file mode 100644
index 00000000000..bc67c1ad0df
--- /dev/null
+++ b/src/pl/plperl/sql/plperl_array.sql
@@ -0,0 +1,164 @@
+CREATE OR REPLACE FUNCTION plperl_sum_array(INTEGER[]) RETURNS text AS $$
+	my $array_arg = shift;
+	my $result = 0;
+	my @arrays;
+
+	push @arrays, @$array_arg;
+
+	while (@arrays > 0) {
+		my $el = shift @arrays;
+		if (is_array_ref($el)) {
+			push @arrays, @$el;
+		} else {
+			$result += $el;
+		}
+	}
+	return $result.' '.$array_arg;
+$$ LANGUAGE plperl;
+
+select plperl_sum_array('{1,2,NULL}');
+select plperl_sum_array('{}');
+select plperl_sum_array('{{1,2,3}, {4,5,6}}');
+select plperl_sum_array('{{{1,2,3}, {4,5,6}}, {{7,8,9}, {10,11,12}}}');
+
+-- check whether we can handle arrays of maximum dimension (6)
+select plperl_sum_array(ARRAY[[[[[[1,2],[3,4]],[[5,6],[7,8]]],[[[9,10],[11,12]],
+[[13,14],[15,16]]]],
+[[[[17,18],[19,20]],[[21,22],[23,24]]],[[[25,26],[27,28]],[[29,30],[31,32]]]]],
+[[[[[1,2],[3,4]],[[5,6],[7,8]]],[[[9,10],[11,12]],[[13,14],[15,16]]]],
+[[[[17,18],[19,20]],[[21,22],[23,24]]],[[[25,26],[27,28]],[[29,30],[31,32]]]]]]);
+
+-- what would we do with the arrays exceeding maximum dimension (7)
+select plperl_sum_array('{{{{{{{1,2},{3,4}},{{5,6},{7,8}}},{{{9,10},{11,12}},
+{{13,14},{15,16}}}},
+{{{{17,18},{19,20}},{{21,22},{23,24}}},{{{25,26},{27,28}},{{29,30},{31,32}}}}},
+{{{{{1,2},{3,4}},{{5,6},{7,8}}},{{{9,10},{11,12}},{{13,14},{15,16}}}},
+{{{{17,18},{19,20}},{{21,22},{23,24}}},{{{25,26},{27,28}},{{29,30},{31,32}}}}}},
+{{{{{{1,2},{3,4}},{{5,6},{7,8}}},{{{9,10},{11,12}},{{13,14},{15,16}}}},
+{{{{17,18},{19,20}},{{21,22},{23,24}}},{{{25,26},{27,28}},{{29,30},{31,32}}}}},
+{{{{{1,2},{3,4}},{{5,6},{7,8}}},{{{9,10},{11,12}},{{13,14},{15,16}}}},
+{{{{17,18},{19,20}},{{21,22},{23,24}}},{{{25,26},{27,28}},{{29,30},{31,32}}}}}}}'
+);
+
+select plperl_sum_array('{{{1,2,3}, {4,5,6,7}}, {{7,8,9}, {10, 11, 12}}}');
+
+CREATE OR REPLACE FUNCTION plperl_concat(TEXT[]) RETURNS TEXT AS $$
+	my $array_arg = shift;
+	my $result = "";
+	my @arrays;
+	
+	push @arrays, @$array_arg;
+	while (@arrays > 0) {
+		my $el = shift @arrays;
+		if (is_array_ref($el)) {
+			push @arrays, @$el;
+		} else {
+			$result .= $el;
+		}
+	}
+	return $result.' '.$array_arg;
+$$ LANGUAGE plperl;
+
+select plperl_concat('{"NULL","NULL","NULL''"}');
+select plperl_concat('{{NULL,NULL,NULL}}');
+select plperl_concat('{"hello"," ","world!"}');
+
+-- array of rows --
+CREATE TYPE foo AS (bar INTEGER, baz TEXT);
+CREATE OR REPLACE FUNCTION plperl_array_of_rows(foo[]) RETURNS TEXT AS $$
+	my $array_arg = shift;
+	my $result = "";
+	
+	for my $row_ref (@$array_arg) {
+		die "not a hash reference" unless (ref $row_ref eq "HASH");
+			$result .= $row_ref->{bar}." items of ".$row_ref->{baz}.";";
+	}
+	return $result .' '. $array_arg;
+$$ LANGUAGE plperl;
+
+select plperl_array_of_rows(ARRAY[ ROW(2, 'coffee'), ROW(0, 'sugar')]::foo[]);
+
+-- composite type containing arrays
+CREATE TYPE rowfoo AS (bar INTEGER, baz INTEGER[]);
+
+CREATE OR REPLACE FUNCTION plperl_sum_row_elements(rowfoo) RETURNS TEXT AS $$
+	my $row_ref = shift;
+	my $result;
+	
+	if (ref $row_ref ne 'HASH') {
+		$result = 0;
+	}
+	else {
+		$result = $row_ref->{bar};
+		die "not an array reference".ref ($row_ref->{baz}) 
+		unless (is_array_ref($row_ref->{baz}));
+		# process a single-dimensional array
+		foreach my $elem (@{$row_ref->{baz}}) {
+			$result += $elem unless ref $elem;
+		}
+	}
+	return $result;
+$$ LANGUAGE plperl;
+
+select plperl_sum_row_elements(ROW(1, ARRAY[2,3,4,5,6,7,8,9,10])::rowfoo);
+
+-- composite type containing array of another composite type, which, in order,
+-- contains an array of integers.
+CREATE TYPE rowbar AS (foo rowfoo[]);
+
+CREATE OR REPLACE FUNCTION plperl_sum_array_of_rows(rowbar) RETURNS TEXT AS $$
+	my $rowfoo_ref = shift;
+	my $result = 0;
+	
+	if (ref $rowfoo_ref eq 'HASH') {
+		my $row_array_ref = $rowfoo_ref->{foo};
+		if (is_array_ref($row_array_ref)) {
+			foreach my $row_ref (@{$row_array_ref}) {
+				if (ref $row_ref eq 'HASH') {
+					$result += $row_ref->{bar};
+					die "not an array reference".ref ($row_ref->{baz}) 
+					unless (is_array_ref($row_ref->{baz}));
+					foreach my $elem (@{$row_ref->{baz}}) {
+						$result += $elem unless ref $elem;
+					}
+				}
+				else {
+					die "element baz is not a reference to a rowfoo";
+				}
+			}
+		} else {
+			die "not a reference to an array of rowfoo elements"
+		}
+	} else {
+		die "not a reference to type rowbar";
+	}
+	return $result;
+$$ LANGUAGE plperl;
+
+select plperl_sum_array_of_rows(ROW(ARRAY[ROW(1, ARRAY[2,3,4,5,6,7,8,9,10])::rowfoo, 
+ROW(11, ARRAY[12,13,14,15,16,17,18,19,20])::rowfoo])::rowbar);
+
+-- check arrays as out parameters
+CREATE OR REPLACE FUNCTION plperl_arrays_out(OUT INTEGER[]) AS $$
+	return [[1,2,3],[4,5,6]];
+$$ LANGUAGE plperl;
+
+select plperl_arrays_out();
+
+-- check that we can return the array we passed in
+CREATE OR REPLACE FUNCTION plperl_arrays_inout(INTEGER[]) returns INTEGER[] AS $$
+	return shift;
+$$ LANGUAGE plperl;
+
+select plperl_arrays_inout('{{1}, {2}, {3}}');
+
+-- make sure setof works
+create or replace function perl_setof_array(integer[]) returns setof integer[] language plperl as $$
+	my $arr = shift;
+	for my $r (@$arr) {
+		return_next $r;
+	}
+	return undef;
+$$;
+
+select perl_setof_array('{{1}, {2}, {3}}');
diff --git a/src/pl/plperl/sql/plperl_trigger.sql b/src/pl/plperl/sql/plperl_trigger.sql
index 1583a42544b..3b9bf89f8e6 100644
--- a/src/pl/plperl/sql/plperl_trigger.sql
+++ b/src/pl/plperl/sql/plperl_trigger.sql
@@ -1,8 +1,11 @@
 -- test plperl triggers
 
+CREATE TYPE rowcomp as (i int);
+CREATE TYPE rowcompnest as (rfoo rowcomp);
 CREATE TABLE trigger_test (
         i int,
-        v varchar
+        v varchar,
+		foo rowcompnest
 );
 
 CREATE OR REPLACE FUNCTION trigger_data() RETURNS trigger LANGUAGE plperl AS $$
@@ -10,6 +13,40 @@ CREATE OR REPLACE FUNCTION trigger_data() RETURNS trigger LANGUAGE plperl AS $$
   # make sure keys are sorted for consistent results - perl no longer
   # hashes in  repeatable fashion across runs
 
+  sub str {
+	  my $val = shift;
+
+	  if (!defined $val)
+	  {
+		  return 'NULL';
+	  }
+	  elsif (ref $val eq 'HASH')
+	  {
+		my $str = '';
+		foreach my $rowkey (sort keys %$val)
+		{
+		  $str .= ", " if $str;
+		  my $rowval = str($val->{$rowkey});
+		  $str .= "'$rowkey' => $rowval";
+		}
+		return '{'. $str .'}';
+	  }
+	  elsif (ref $val eq 'ARRAY')
+	  {
+		  my $str = '';
+		  for my $argval (@$val)
+		  {
+			  $str .= ", " if $str;
+			  $str .= str($argval);
+		  }
+		  return '['. $str .']';
+	  }
+	  else
+	  {
+		  return "'$val'";
+	  }
+  }
+
   foreach my $key (sort keys %$_TD)
   {
 
@@ -18,35 +55,7 @@ CREATE OR REPLACE FUNCTION trigger_data() RETURNS trigger LANGUAGE plperl AS $$
 	# relid is variable, so we can not use it repeatably
 	$val = "bogus:12345" if $key eq 'relid';
 
-	if (! defined $val)
-	{
-	  elog(NOTICE, "\$_TD->\{$key\} = NULL");
-	}
-	elsif (not ref $val)
-    {
-	  elog(NOTICE, "\$_TD->\{$key\} = '$val'");
-	}
-	elsif (ref $val eq 'HASH')
-	{
-	  my $str = "";
-	  foreach my $rowkey (sort keys %$val)
-	  {
-	    $str .= ", " if $str;
-	    my $rowval = $val->{$rowkey};
-	    $str .= "'$rowkey' => '$rowval'";
-      }
-	  elog(NOTICE, "\$_TD->\{$key\} = \{$str\}");
-	}
-	elsif (ref $val eq 'ARRAY')
-	{
-	  my $str = "";
-	  foreach my $argval (@$val)
-	  {
-	    $str .= ", " if $str;
-	    $str .= "'$argval'";
-      }
-	  elog(NOTICE, "\$_TD->\{$key\} = \[$str\]");
-	}
+	elog(NOTICE, "\$_TD->\{$key\} = ". str($val));
   }
   return undef; # allow statement to proceed;
 $$;
@@ -55,21 +64,21 @@ CREATE TRIGGER show_trigger_data_trig
 BEFORE INSERT OR UPDATE OR DELETE ON trigger_test
 FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
 
-insert into trigger_test values(1,'insert');
+insert into trigger_test values(1,'insert', '("(1)")');
 update trigger_test set v = 'update' where i = 1;
 delete from trigger_test;
 
 DROP TRIGGER show_trigger_data_trig on trigger_test;
 
-insert into trigger_test values(1,'insert');
+insert into trigger_test values(1,'insert', '("(1)")');
 CREATE VIEW trigger_test_view AS SELECT * FROM trigger_test;
 
 CREATE TRIGGER show_trigger_data_trig
 INSTEAD OF INSERT OR UPDATE OR DELETE ON trigger_test_view
 FOR EACH ROW EXECUTE PROCEDURE trigger_data(24,'skidoo view');
 
-insert into trigger_test_view values(2,'insert');
-update trigger_test_view set v = 'update' where i = 1;
+insert into trigger_test_view values(2,'insert', '("(2)")');
+update trigger_test_view set v = 'update', foo = '("(3)")' where i = 1;
 delete from trigger_test_view;
 
 DROP VIEW trigger_test_view;
@@ -86,6 +95,7 @@ CREATE OR REPLACE FUNCTION valid_id() RETURNS trigger AS $$
     elsif ($_TD->{new}{v} ne "immortal")
     {
         $_TD->{new}{v} .= "(modified by trigger)";
+		$_TD->{new}{foo}{rfoo}{i}++;
         return "MODIFY"; # Modify tuple and proceed INSERT/UPDATE command
     }
     else
@@ -97,10 +107,10 @@ $$ LANGUAGE plperl;
 CREATE TRIGGER "test_valid_id_trig" BEFORE INSERT OR UPDATE ON trigger_test
 FOR EACH ROW EXECUTE PROCEDURE "valid_id"();
 
-INSERT INTO trigger_test (i, v) VALUES (1,'first line');
-INSERT INTO trigger_test (i, v) VALUES (2,'second line');
-INSERT INTO trigger_test (i, v) VALUES (3,'third line');
-INSERT INTO trigger_test (i, v) VALUES (4,'immortal');
+INSERT INTO trigger_test (i, v, foo) VALUES (1,'first line', '("(1)")');
+INSERT INTO trigger_test (i, v, foo) VALUES (2,'second line', '("(2)")');
+INSERT INTO trigger_test (i, v, foo) VALUES (3,'third line', '("(3)")');
+INSERT INTO trigger_test (i, v, foo) VALUES (4,'immortal', '("(4)")');
 
 INSERT INTO trigger_test (i, v) VALUES (101,'bad id');
 
diff --git a/src/pl/plperl/sql/plperl_util.sql b/src/pl/plperl/sql/plperl_util.sql
index 6a810d8dd28..143d0478020 100644
--- a/src/pl/plperl/sql/plperl_util.sql
+++ b/src/pl/plperl/sql/plperl_util.sql
@@ -98,3 +98,15 @@ create or replace function perl_looks_like_number() returns setof text language
 $$;
 
 select perl_looks_like_number();
+
+-- test encode_typed_literal
+create type perl_foo as (a integer, b text[]);
+create type perl_bar as (c perl_foo[]);
+create or replace function perl_encode_typed_literal() returns setof text language plperl as $$
+	return_next encode_typed_literal(undef, 'text');
+	return_next encode_typed_literal([[1,2,3],[3,2,1],[1,3,2]], 'integer[]');
+	return_next encode_typed_literal({a => 1, b => ['PL','/','Perl']}, 'perl_foo');
+	return_next encode_typed_literal({c => [{a => 9, b => ['PostgreSQL']}, {b => ['Postgres'], a => 1}]}, 'perl_bar');
+$$;
+
+select perl_encode_typed_literal();