Re-implement pl/pgsql's expression and assignment parsing.

Invent new RawParseModes that allow the core grammar to handle
pl/pgsql expressions and assignments directly, and thereby get rid
of a lot of hackery in pl/pgsql's parser.  This moves a good deal
of knowledge about pl/pgsql into the core code: notably, we have to
invent a CoercionContext that matches pl/pgsql's (rather dubious)
historical behavior for assignment coercions.  That's getting away
from the original idea of pl/pgsql as an arm's-length extension of
the core, but really we crossed that bridge a long time ago.

The main advantage of doing this is that we can now use the core
parser to generate FieldStore and/or SubscriptingRef nodes to handle
assignments to pl/pgsql variables that are records or arrays.  That
fixes a number of cases that had never been implemented in pl/pgsql
assignment, such as nested records and array slicing, and it allows
pl/pgsql assignment to support the datatype-specific subscripting
behaviors introduced in commit c7aba7c14.

There are cosmetic benefits too: when a syntax error occurs in a
pl/pgsql expression, the error report no longer includes the confusing
"SELECT" keyword that used to get prefixed to the expression text.
Also, there seem to be some small speed gains.

Discussion: https://postgr.es/m/4165684.1607707277@sss.pgh.pa.us

10 files changed, 503 insertions, 106 deletions

diff --git a/src/pl/plpgsql/src/Makefile b/src/pl/plpgsql/src/Makefile
index 193df8a0108..9946abbc1de 100644
--- a/src/pl/plpgsql/src/Makefile
+++ b/src/pl/plpgsql/src/Makefile
@@ -32,7 +32,7 @@ DATA = plpgsql.control plpgsql--1.0.sql
 
 REGRESS_OPTS = --dbname=$(PL_TESTDB)
 
-REGRESS = plpgsql_call plpgsql_control plpgsql_copy plpgsql_domain \
+REGRESS = plpgsql_array plpgsql_call plpgsql_control plpgsql_copy plpgsql_domain \
 	plpgsql_record plpgsql_cache plpgsql_simple plpgsql_transaction \
 	plpgsql_trap plpgsql_trigger plpgsql_varprops
 
diff --git a/src/pl/plpgsql/src/expected/plpgsql_array.out b/src/pl/plpgsql/src/expected/plpgsql_array.out
new file mode 100644
index 00000000000..5f28b4f685b
--- /dev/null
+++ b/src/pl/plpgsql/src/expected/plpgsql_array.out
@@ -0,0 +1,94 @@
+--
+-- Tests for PL/pgSQL handling of array variables
+--
+-- We also check arrays of composites here, so this has some overlap
+-- with the plpgsql_record tests.
+--
+create type complex as (r float8, i float8);
+create type quadarray as (c1 complex[], c2 complex);
+do $$ declare a int[];
+begin a := array[1,2]; a[3] := 4; raise notice 'a = %', a; end$$;
+NOTICE:  a = {1,2,4}
+do $$ declare a int[];
+begin a[3] := 4; raise notice 'a = %', a; end$$;
+NOTICE:  a = [3:3]={4}
+do $$ declare a int[];
+begin a[1][4] := 4; raise notice 'a = %', a; end$$;
+NOTICE:  a = [1:1][4:4]={{4}}
+do $$ declare a int[];
+begin a[1] := 23::text; raise notice 'a = %', a; end$$;  -- lax typing
+NOTICE:  a = {23}
+do $$ declare a int[];
+begin a := array[1,2]; a[2:3] := array[3,4]; raise notice 'a = %', a; end$$;
+NOTICE:  a = {1,3,4}
+do $$ declare a int[];
+begin a := array[1,2]; a[2] := a[2] + 1; raise notice 'a = %', a; end$$;
+NOTICE:  a = {1,3}
+do $$ declare a int[];
+begin a[1:2] := array[3,4]; raise notice 'a = %', a; end$$;
+NOTICE:  a = {3,4}
+do $$ declare a int[];
+begin a[1:2] := 4; raise notice 'a = %', a; end$$;  -- error
+ERROR:  malformed array literal: "4"
+DETAIL:  Array value must start with "{" or dimension information.
+CONTEXT:  PL/pgSQL function inline_code_block line 2 at assignment
+do $$ declare a complex[];
+begin a[1] := (1,2); a[1].i := 11; raise notice 'a = %', a; end$$;
+NOTICE:  a = {"(1,11)"}
+do $$ declare a complex[];
+begin a[1].i := 11; raise notice 'a = %, a[1].i = %', a, a[1].i; end$$;
+NOTICE:  a = {"(,11)"}, a[1].i = 11
+-- perhaps this ought to work, but for now it doesn't:
+do $$ declare a complex[];
+begin a[1:2].i := array[11,12]; raise notice 'a = %', a; end$$;
+ERROR:  cannot assign to field "i" of column "a" because its type complex[] is not a composite type
+LINE 1: a[1:2].i := array[11,12]
+        ^
+QUERY:  a[1:2].i := array[11,12]
+CONTEXT:  PL/pgSQL function inline_code_block line 2 at assignment
+do $$ declare a quadarray;
+begin a.c1[1].i := 11; raise notice 'a = %, a.c1[1].i = %', a, a.c1[1].i; end$$;
+NOTICE:  a = ("{""(,11)""}",), a.c1[1].i = 11
+do $$ declare a int[];
+begin a := array_agg(x) from (values(1),(2),(3)) v(x); raise notice 'a = %', a; end$$;
+NOTICE:  a = {1,2,3}
+create temp table onecol as select array[1,2] as f1;
+do $$ declare a int[];
+begin a := f1 from onecol; raise notice 'a = %', a; end$$;
+NOTICE:  a = {1,2}
+do $$ declare a int[];
+begin a := * from onecol for update; raise notice 'a = %', a; end$$;
+NOTICE:  a = {1,2}
+-- error cases:
+do $$ declare a int[];
+begin a := from onecol; raise notice 'a = %', a; end$$;
+ERROR:  assignment source returned 0 columns
+CONTEXT:  PL/pgSQL assignment "a := from onecol"
+PL/pgSQL function inline_code_block line 2 at assignment
+do $$ declare a int[];
+begin a := f1, f1 from onecol; raise notice 'a = %', a; end$$;
+ERROR:  assignment source returned 2 columns
+CONTEXT:  PL/pgSQL assignment "a := f1, f1 from onecol"
+PL/pgSQL function inline_code_block line 2 at assignment
+insert into onecol values(array[11]);
+do $$ declare a int[];
+begin a := f1 from onecol; raise notice 'a = %', a; end$$;
+ERROR:  query "a := f1 from onecol" returned more than one row
+CONTEXT:  PL/pgSQL function inline_code_block line 2 at assignment
+do $$ declare a int[];
+begin a := f1 from onecol limit 1; raise notice 'a = %', a; end$$;
+NOTICE:  a = {1,2}
+do $$ declare a real;
+begin a[1] := 2; raise notice 'a = %', a; end$$;
+ERROR:  cannot subscript type real because it does not support subscripting
+LINE 1: a[1] := 2
+        ^
+QUERY:  a[1] := 2
+CONTEXT:  PL/pgSQL function inline_code_block line 2 at assignment
+do $$ declare a complex;
+begin a.r[1] := 2; raise notice 'a = %', a; end$$;
+ERROR:  cannot subscript type double precision because it does not support subscripting
+LINE 1: a.r[1] := 2
+        ^
+QUERY:  a.r[1] := 2
+CONTEXT:  PL/pgSQL function inline_code_block line 2 at assignment
diff --git a/src/pl/plpgsql/src/expected/plpgsql_record.out b/src/pl/plpgsql/src/expected/plpgsql_record.out
index cf6089cbb21..6e835c0751b 100644
--- a/src/pl/plpgsql/src/expected/plpgsql_record.out
+++ b/src/pl/plpgsql/src/expected/plpgsql_record.out
@@ -3,6 +3,7 @@
 --
 create type two_int4s as (f1 int4, f2 int4);
 create type two_int8s as (q1 int8, q2 int8);
+create type nested_int8s as (c1 two_int8s, c2 two_int8s);
 -- base-case return of a composite type
 create function retc(int) returns two_int8s language plpgsql as
 $$ begin return row($1,1)::two_int8s; end $$;
@@ -82,6 +83,88 @@ begin
 end$$;
 NOTICE:  c4 = (1,2)
 NOTICE:  c8 = (1,2)
+do $$ declare c two_int8s; d nested_int8s;
+begin
+  c := row(1,2);
+  d := row(c, row(c.q1, c.q2+1));
+  raise notice 'c = %, d = %', c, d;
+  c.q1 := 10;
+  d.c1 := row(11,12);
+  d.c2.q2 := 42;
+  raise notice 'c = %, d = %', c, d;
+  raise notice 'c.q1 = %, d.c2 = %', c.q1, d.c2;
+  raise notice '(d).c2.q2 = %', (d).c2.q2;  -- doesn't work without parens
+  raise notice '(d.c2).q2 = %', (d.c2).q2;  -- doesn't work without parens
+end$$;
+NOTICE:  c = (1,2), d = ("(1,2)","(1,3)")
+NOTICE:  c = (10,2), d = ("(11,12)","(1,42)")
+NOTICE:  c.q1 = 10, d.c2 = (1,42)
+NOTICE:  (d).c2.q2 = 42
+NOTICE:  (d.c2).q2 = 42
+-- block-qualified naming
+do $$ <<b>> declare c two_int8s; d nested_int8s;
+begin
+  b.c := row(1,2);
+  b.d := row(b.c, row(b.c.q1, b.c.q2+1));
+  raise notice 'b.c = %, b.d = %', b.c, b.d;
+  b.c.q1 := 10;
+  b.d.c1 := row(11,12);
+  b.d.c2.q2 := 42;
+  raise notice 'b.c = %, b.d = %', b.c, b.d;
+  raise notice 'b.c.q1 = %, b.d.c2 = %', b.c.q1, b.d.c2;
+  raise notice '(b.d).c2.q2 = %', (b.d).c2.q2;  -- doesn't work without parens
+  raise notice '(b.d.c2).q2 = %', (b.d.c2).q2;  -- doesn't work without parens
+end$$;
+NOTICE:  b.c = (1,2), b.d = ("(1,2)","(1,3)")
+NOTICE:  b.c = (10,2), b.d = ("(11,12)","(1,42)")
+NOTICE:  b.c.q1 = 10, b.d.c2 = (1,42)
+NOTICE:  (b.d).c2.q2 = 42
+NOTICE:  (b.d.c2).q2 = 42
+-- error cases
+do $$ declare c two_int8s; begin c.x = 1; end $$;
+ERROR:  record "c" has no field "x"
+CONTEXT:  PL/pgSQL assignment "c.x = 1"
+PL/pgSQL function inline_code_block line 1 at assignment
+do $$ declare c nested_int8s; begin c.x = 1; end $$;
+ERROR:  record "c" has no field "x"
+CONTEXT:  PL/pgSQL assignment "c.x = 1"
+PL/pgSQL function inline_code_block line 1 at assignment
+do $$ declare c nested_int8s; begin c.x.q1 = 1; end $$;
+ERROR:  record "c" has no field "x"
+CONTEXT:  PL/pgSQL assignment "c.x.q1 = 1"
+PL/pgSQL function inline_code_block line 1 at assignment
+do $$ declare c nested_int8s; begin c.c2.x = 1; end $$;
+ERROR:  cannot assign to field "x" of column "c" because there is no such column in data type two_int8s
+LINE 1: c.c2.x = 1
+        ^
+QUERY:  c.c2.x = 1
+CONTEXT:  PL/pgSQL function inline_code_block line 1 at assignment
+do $$ declare c nested_int8s; begin d.c2.x = 1; end $$;
+ERROR:  "d.c2.x" is not a known variable
+LINE 1: do $$ declare c nested_int8s; begin d.c2.x = 1; end $$;
+                                            ^
+do $$ <<b>> declare c two_int8s; begin b.c.x = 1; end $$;
+ERROR:  record "c" has no field "x"
+CONTEXT:  PL/pgSQL assignment "b.c.x = 1"
+PL/pgSQL function inline_code_block line 1 at assignment
+do $$ <<b>> declare c nested_int8s; begin b.c.x = 1; end $$;
+ERROR:  record "c" has no field "x"
+CONTEXT:  PL/pgSQL assignment "b.c.x = 1"
+PL/pgSQL function inline_code_block line 1 at assignment
+do $$ <<b>> declare c nested_int8s; begin b.c.x.q1 = 1; end $$;
+ERROR:  record "c" has no field "x"
+CONTEXT:  PL/pgSQL assignment "b.c.x.q1 = 1"
+PL/pgSQL function inline_code_block line 1 at assignment
+do $$ <<b>> declare c nested_int8s; begin b.c.c2.x = 1; end $$;
+ERROR:  cannot assign to field "x" of column "b" because there is no such column in data type two_int8s
+LINE 1: b.c.c2.x = 1
+        ^
+QUERY:  b.c.c2.x = 1
+CONTEXT:  PL/pgSQL function inline_code_block line 1 at assignment
+do $$ <<b>> declare c nested_int8s; begin b.d.c2.x = 1; end $$;
+ERROR:  "b.d.c2" is not a known variable
+LINE 1: do $$ <<b>> declare c nested_int8s; begin b.d.c2.x = 1; end ...
+                                                  ^
 -- check passing composite result to another function
 create function getq1(two_int8s) returns int8 language plpgsql as $$
 declare r two_int8s; begin r := $1; return r.q1; end $$;
@@ -188,7 +271,7 @@ NOTICE:  r1.q1 = <NULL>
 NOTICE:  r1.q2 = <NULL>
 NOTICE:  r1 = <NULL>
 ERROR:  record "r1" has no field "nosuchfield"
-CONTEXT:  SQL statement "SELECT r1.nosuchfield"
+CONTEXT:  SQL expression "r1.nosuchfield"
 PL/pgSQL function inline_code_block line 7 at RAISE
 -- records, not so much
 do $$
@@ -202,7 +285,7 @@ end$$;
 NOTICE:  r1 = <NULL>
 ERROR:  record "r1" is not assigned yet
 DETAIL:  The tuple structure of a not-yet-assigned record is indeterminate.
-CONTEXT:  SQL statement "SELECT r1.f1"
+CONTEXT:  SQL expression "r1.f1"
 PL/pgSQL function inline_code_block line 5 at RAISE
 -- but OK if you assign first
 do $$
@@ -220,7 +303,7 @@ NOTICE:  r1.f1 = 1
 NOTICE:  r1.f2 = 2
 NOTICE:  r1 = (1,2)
 ERROR:  record "r1" has no field "nosuchfield"
-CONTEXT:  SQL statement "SELECT r1.nosuchfield"
+CONTEXT:  SQL expression "r1.nosuchfield"
 PL/pgSQL function inline_code_block line 9 at RAISE
 -- check repeated assignments to composite fields
 create table some_table (id int, data text);
@@ -431,7 +514,7 @@ create function getf3(x mutable) returns int language plpgsql as
 $$ begin return x.f3; end $$;
 select getf3(null::mutable);  -- doesn't work yet
 ERROR:  record "x" has no field "f3"
-CONTEXT:  SQL statement "SELECT x.f3"
+CONTEXT:  SQL expression "x.f3"
 PL/pgSQL function getf3(mutable) line 1 at RETURN
 alter table mutable add column f3 int;
 select getf3(null::mutable);  -- now it works
diff --git a/src/pl/plpgsql/src/expected/plpgsql_varprops.out b/src/pl/plpgsql/src/expected/plpgsql_varprops.out
index 18f03d75b42..3801dccc95a 100644
--- a/src/pl/plpgsql/src/expected/plpgsql_varprops.out
+++ b/src/pl/plpgsql/src/expected/plpgsql_varprops.out
@@ -76,7 +76,7 @@ begin
   raise notice 'x = %', x;
 end$$;
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
+CONTEXT:  SQL expression "1/0"
 PL/pgSQL function inline_code_block line 3 during statement block local variable initialization
 do $$
 declare x bigint[] := array[1,3,5];
diff --git a/src/pl/plpgsql/src/pl_comp.c b/src/pl/plpgsql/src/pl_comp.c
index 81979c2961b..5336793a933 100644
--- a/src/pl/plpgsql/src/pl_comp.c
+++ b/src/pl/plpgsql/src/pl_comp.c
@@ -1458,7 +1458,8 @@ plpgsql_parse_dblword(char *word1, char *word2,
 	/*
 	 * We should do nothing in DECLARE sections.  In SQL expressions, we
 	 * really only need to make sure that RECFIELD datums are created when
-	 * needed.
+	 * needed.  In all the cases handled by this function, returning a T_DATUM
+	 * with a two-word idents string is the right thing.
 	 */
 	if (plpgsql_IdentifierLookup != IDENTIFIER_LOOKUP_DECLARE)
 	{
@@ -1532,40 +1533,53 @@ plpgsql_parse_tripword(char *word1, char *word2, char *word3,
 	List	   *idents;
 	int			nnames;
 
-	idents = list_make3(makeString(word1),
-						makeString(word2),
-						makeString(word3));
-
 	/*
-	 * We should do nothing in DECLARE sections.  In SQL expressions, we
-	 * really only need to make sure that RECFIELD datums are created when
-	 * needed.
+	 * We should do nothing in DECLARE sections.  In SQL expressions, we need
+	 * to make sure that RECFIELD datums are created when needed, and we need
+	 * to be careful about how many names are reported as belonging to the
+	 * T_DATUM: the third word could be a sub-field reference, which we don't
+	 * care about here.
 	 */
 	if (plpgsql_IdentifierLookup != IDENTIFIER_LOOKUP_DECLARE)
 	{
 		/*
-		 * Do a lookup in the current namespace stack. Must find a qualified
+		 * Do a lookup in the current namespace stack.  Must find a record
 		 * reference, else ignore.
 		 */
 		ns = plpgsql_ns_lookup(plpgsql_ns_top(), false,
 							   word1, word2, word3,
 							   &nnames);
-		if (ns != NULL && nnames == 2)
+		if (ns != NULL)
 		{
 			switch (ns->itemtype)
 			{
 				case PLPGSQL_NSTYPE_REC:
 					{
-						/*
-						 * words 1/2 are a record name, so third word could be
-						 * a field in this record.
-						 */
 						PLpgSQL_rec *rec;
 						PLpgSQL_recfield *new;
 
 						rec = (PLpgSQL_rec *) (plpgsql_Datums[ns->itemno]);
-						new = plpgsql_build_recfield(rec, word3);
-
+						if (nnames == 1)
+						{
+							/*
+							 * First word is a record name, so second word
+							 * could be a field in this record (and the third,
+							 * a sub-field).  We build a RECFIELD datum
+							 * whether it is or not --- any error will be
+							 * detected later.
+							 */
+							new = plpgsql_build_recfield(rec, word2);
+							idents = list_make2(makeString(word1),
+												makeString(word2));
+						}
+						else
+						{
+							/* Block-qualified reference to record variable. */
+							new = plpgsql_build_recfield(rec, word3);
+							idents = list_make3(makeString(word1),
+												makeString(word2),
+												makeString(word3));
+						}
 						wdatum->datum = (PLpgSQL_datum *) new;
 						wdatum->ident = NULL;
 						wdatum->quoted = false; /* not used */
@@ -1580,6 +1594,9 @@ plpgsql_parse_tripword(char *word1, char *word2, char *word3,
 	}
 
 	/* Nothing found */
+	idents = list_make3(makeString(word1),
+						makeString(word2),
+						makeString(word3));
 	cword->idents = idents;
 	return false;
 }
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 4a51fb6d9f1..5e4dbd25a2a 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -4182,7 +4182,7 @@ exec_prepare_plan(PLpgSQL_execstate *estate,
 	memset(&options, 0, sizeof(options));
 	options.parserSetup = (ParserSetupHook) plpgsql_parser_setup;
 	options.parserSetupArg = (void *) expr;
-	options.parseMode = RAW_PARSE_DEFAULT;
+	options.parseMode = expr->parseMode;
 	options.cursorOptions = cursorOptions;
 	plan = SPI_prepare_extended(expr->query, &options);
 	if (plan == NULL)
@@ -8006,10 +8006,14 @@ get_cast_hashentry(PLpgSQL_execstate *estate,
 		placeholder->collation = get_typcollation(srctype);
 
 		/*
-		 * Apply coercion.  We use ASSIGNMENT coercion because that's the
-		 * closest match to plpgsql's historical behavior; in particular,
-		 * EXPLICIT coercion would allow silent truncation to a destination
-		 * varchar/bpchar's length, which we do not want.
+		 * Apply coercion.  We use the special coercion context
+		 * COERCION_PLPGSQL to match plpgsql's historical behavior, namely
+		 * that any cast not available at ASSIGNMENT level will be implemented
+		 * as an I/O coercion.  (It's somewhat dubious that we prefer I/O
+		 * coercion over cast pathways that exist at EXPLICIT level.  Changing
+		 * that would cause assorted minor behavioral differences though, and
+		 * a user who wants the explicit-cast behavior can always write an
+		 * explicit cast.)
 		 *
 		 * If source type is UNKNOWN, coerce_to_target_type will fail (it only
 		 * expects to see that for Const input nodes), so don't call it; we'll
@@ -8022,7 +8026,7 @@ get_cast_hashentry(PLpgSQL_execstate *estate,
 			cast_expr = coerce_to_target_type(NULL,
 											  (Node *) placeholder, srctype,
 											  dsttype, dsttypmod,
-											  COERCION_ASSIGNMENT,
+											  COERCION_PLPGSQL,
 											  COERCE_IMPLICIT_CAST,
 											  -1);
 
@@ -8030,7 +8034,8 @@ get_cast_hashentry(PLpgSQL_execstate *estate,
 		 * If there's no cast path according to the parser, fall back to using
 		 * an I/O coercion; this is semantically dubious but matches plpgsql's
 		 * historical behavior.  We would need something of the sort for
-		 * UNKNOWN literals in any case.
+		 * UNKNOWN literals in any case.  (This is probably now only reachable
+		 * in the case where srctype is UNKNOWN/RECORD.)
 		 */
 		if (cast_expr == NULL)
 		{
@@ -8339,7 +8344,8 @@ exec_check_rw_parameter(PLpgSQL_expr *expr, int target_dno)
 		return;
 
 	/*
-	 * Top level of expression must be a simple FuncExpr or OpExpr.
+	 * Top level of expression must be a simple FuncExpr, OpExpr, or
+	 * SubscriptingRef.
 	 */
 	if (IsA(expr->expr_simple_expr, FuncExpr))
 	{
@@ -8355,6 +8361,33 @@ exec_check_rw_parameter(PLpgSQL_expr *expr, int target_dno)
 		funcid = opexpr->opfuncid;
 		fargs = opexpr->args;
 	}
+	else if (IsA(expr->expr_simple_expr, SubscriptingRef))
+	{
+		SubscriptingRef *sbsref = (SubscriptingRef *) expr->expr_simple_expr;
+
+		/* We only trust standard varlena arrays to be safe */
+		if (get_typsubscript(sbsref->refcontainertype, NULL) !=
+			F_ARRAY_SUBSCRIPT_HANDLER)
+			return;
+
+		/* refexpr can be a simple Param, otherwise must not contain target */
+		if (!(sbsref->refexpr && IsA(sbsref->refexpr, Param)) &&
+			contains_target_param((Node *) sbsref->refexpr, &target_dno))
+			return;
+
+		/* the other subexpressions must not contain target */
+		if (contains_target_param((Node *) sbsref->refupperindexpr,
+								  &target_dno) ||
+			contains_target_param((Node *) sbsref->reflowerindexpr,
+								  &target_dno) ||
+			contains_target_param((Node *) sbsref->refassgnexpr,
+								  &target_dno))
+			return;
+
+		/* OK, we can pass target as a read-write parameter */
+		expr->rwparam = target_dno;
+		return;
+	}
 	else
 		return;
 
diff --git a/src/pl/plpgsql/src/pl_gram.y b/src/pl/plpgsql/src/pl_gram.y
index c09576efff5..ad248bc7648 100644
--- a/src/pl/plpgsql/src/pl_gram.y
+++ b/src/pl/plpgsql/src/pl_gram.y
@@ -51,7 +51,6 @@
 typedef struct
 {
 	int			location;
-	int			leaderlen;
 } sql_error_callback_arg;
 
 #define parser_errposition(pos)  plpgsql_scanner_errposition(pos)
@@ -67,7 +66,7 @@ static	PLpgSQL_expr	*read_sql_construct(int until,
 											int until2,
 											int until3,
 											const char *expected,
-											const char *sqlstart,
+											RawParseMode parsemode,
 											bool isexpression,
 											bool valid_sql,
 											bool trim,
@@ -78,7 +77,7 @@ static	PLpgSQL_expr	*read_sql_expression(int until,
 static	PLpgSQL_expr	*read_sql_expression2(int until, int until2,
 											  const char *expected,
 											  int *endtoken);
-static	PLpgSQL_expr	*read_sql_stmt(const char *sqlstart);
+static	PLpgSQL_expr	*read_sql_stmt(void);
 static	PLpgSQL_type	*read_datatype(int tok);
 static	PLpgSQL_stmt	*make_execsql_stmt(int firsttoken, int location);
 static	PLpgSQL_stmt_fetch *read_fetch_direction(void);
@@ -99,8 +98,8 @@ static	PLpgSQL_row		*read_into_scalar_list(char *initial_name,
 static	PLpgSQL_row		*make_scalar_list1(char *initial_name,
 										   PLpgSQL_datum *initial_datum,
 										   int lineno, int location);
-static	void			 check_sql_expr(const char *stmt, int location,
-										int leaderlen);
+static	void			 check_sql_expr(const char *stmt,
+										RawParseMode parseMode, int location);
 static	void			 plpgsql_sql_error_callback(void *arg);
 static	PLpgSQL_type	*parse_datatype(const char *string, int location);
 static	void			 check_labels(const char *start_label,
@@ -540,7 +539,7 @@ decl_statement	: decl_varname decl_const decl_datatype decl_collate decl_notnull
 					{
 						PLpgSQL_var *new;
 						PLpgSQL_expr *curname_def;
-						char		buf[1024];
+						char		buf[NAMEDATALEN * 2 + 64];
 						char		*cp1;
 						char		*cp2;
 
@@ -557,9 +556,9 @@ decl_statement	: decl_varname decl_const decl_datatype decl_collate decl_notnull
 
 						curname_def = palloc0(sizeof(PLpgSQL_expr));
 
-						strcpy(buf, "SELECT ");
+						/* Note: refname has been truncated to NAMEDATALEN */
 						cp1 = new->refname;
-						cp2 = buf + strlen(buf);
+						cp2 = buf;
 						/*
 						 * Don't trust standard_conforming_strings here;
 						 * it might change before we use the string.
@@ -575,6 +574,7 @@ decl_statement	: decl_varname decl_const decl_datatype decl_collate decl_notnull
 						}
 						strcpy(cp2, "'::pg_catalog.refcursor");
 						curname_def->query = pstrdup(buf);
+						curname_def->parseMode = RAW_PARSE_PLPGSQL_EXPR;
 						new->default_val = curname_def;
 
 						new->cursor_explicit_expr = $7;
@@ -602,7 +602,7 @@ opt_scrollable :
 
 decl_cursor_query :
 					{
-						$$ = read_sql_stmt("");
+						$$ = read_sql_stmt();
 					}
 				;
 
@@ -904,15 +904,37 @@ proc_stmt		: pl_block ';'
 						{ $$ = $1; }
 				;
 
-stmt_perform	: K_PERFORM expr_until_semi
+stmt_perform	: K_PERFORM
 					{
 						PLpgSQL_stmt_perform *new;
+						int		startloc;
 
 						new = palloc0(sizeof(PLpgSQL_stmt_perform));
 						new->cmd_type = PLPGSQL_STMT_PERFORM;
 						new->lineno   = plpgsql_location_to_lineno(@1);
 						new->stmtid = ++plpgsql_curr_compile->nstatements;
-						new->expr  = $2;
+						plpgsql_push_back_token(K_PERFORM);
+
+						/*
+						 * Since PERFORM isn't legal SQL, we have to cheat to
+						 * the extent of substituting "SELECT" for "PERFORM"
+						 * in the parsed text.  It does not seem worth
+						 * inventing a separate parse mode for this one case.
+						 * We can't do syntax-checking until after we make the
+						 * substitution.
+						 */
+						new->expr = read_sql_construct(';', 0, 0, ";",
+													   RAW_PARSE_DEFAULT,
+													   false, false, true,
+													   &startloc, NULL);
+						/* overwrite "perform" ... */
+						memcpy(new->expr->query, " SELECT", 7);
+						/* left-justify to get rid of the leading space */
+						memmove(new->expr->query, new->expr->query + 1,
+								strlen(new->expr->query));
+						/* offset syntax error position to account for that */
+						check_sql_expr(new->expr->query, new->expr->parseMode,
+									   startloc + 1);
 
 						$$ = (PLpgSQL_stmt *)new;
 					}
@@ -926,7 +948,8 @@ stmt_call		: K_CALL
 						new->cmd_type = PLPGSQL_STMT_CALL;
 						new->lineno = plpgsql_location_to_lineno(@1);
 						new->stmtid = ++plpgsql_curr_compile->nstatements;
-						new->expr = read_sql_stmt("CALL ");
+						plpgsql_push_back_token(K_CALL);
+						new->expr = read_sql_stmt();
 						new->is_call = true;
 
 						$$ = (PLpgSQL_stmt *)new;
@@ -941,7 +964,8 @@ stmt_call		: K_CALL
 						new->cmd_type = PLPGSQL_STMT_CALL;
 						new->lineno = plpgsql_location_to_lineno(@1);
 						new->stmtid = ++plpgsql_curr_compile->nstatements;
-						new->expr = read_sql_stmt("DO ");
+						plpgsql_push_back_token(K_DO);
+						new->expr = read_sql_stmt();
 						new->is_call = false;
 
 						$$ = (PLpgSQL_stmt *)new;
@@ -949,16 +973,40 @@ stmt_call		: K_CALL
 					}
 				;
 
-stmt_assign		: assign_var assign_operator expr_until_semi
+stmt_assign		: T_DATUM
 					{
 						PLpgSQL_stmt_assign *new;
+						RawParseMode pmode;
+
+						/* see how many names identify the datum */
+						switch ($1.ident ? 1 : list_length($1.idents))
+						{
+							case 1:
+								pmode = RAW_PARSE_PLPGSQL_ASSIGN1;
+								break;
+							case 2:
+								pmode = RAW_PARSE_PLPGSQL_ASSIGN2;
+								break;
+							case 3:
+								pmode = RAW_PARSE_PLPGSQL_ASSIGN3;
+								break;
+							default:
+								elog(ERROR, "unexpected number of names");
+								pmode = 0; /* keep compiler quiet */
+						}
 
+						check_assignable($1.datum, @1);
 						new = palloc0(sizeof(PLpgSQL_stmt_assign));
 						new->cmd_type = PLPGSQL_STMT_ASSIGN;
 						new->lineno   = plpgsql_location_to_lineno(@1);
 						new->stmtid = ++plpgsql_curr_compile->nstatements;
-						new->varno = $1->dno;
-						new->expr  = $3;
+						new->varno = $1.datum->dno;
+						/* Push back the head name to include it in the stmt */
+						plpgsql_push_back_token(T_DATUM);
+						new->expr = read_sql_construct(';', 0, 0, ";",
+													   pmode,
+													   false, true, true,
+													   NULL, NULL);
 
 						$$ = (PLpgSQL_stmt *)new;
 					}
@@ -1452,16 +1500,16 @@ for_control		: for_variable K_IN
 
 							/*
 							 * Read tokens until we see either a ".."
-							 * or a LOOP. The text we read may not
-							 * necessarily be a well-formed SQL
-							 * statement, so we need to invoke
-							 * read_sql_construct directly.
+							 * or a LOOP.  The text we read may be either
+							 * an expression or a whole SQL statement, so
+							 * we need to invoke read_sql_construct directly,
+							 * and tell it not to check syntax yet.
 							 */
 							expr1 = read_sql_construct(DOT_DOT,
 													   K_LOOP,
 													   0,
 													   "LOOP",
-													   "SELECT ",
+													   RAW_PARSE_DEFAULT,
 													   true,
 													   false,
 													   true,
@@ -1476,8 +1524,13 @@ for_control		: for_variable K_IN
 								PLpgSQL_var			*fvar;
 								PLpgSQL_stmt_fori	*new;
 
-								/* Check first expression is well-formed */
-								check_sql_expr(expr1->query, expr1loc, 7);
+								/*
+								 * Relabel first expression as an expression;
+								 * then we can check its syntax.
+								 */
+								expr1->parseMode = RAW_PARSE_PLPGSQL_EXPR;
+								check_sql_expr(expr1->query, expr1->parseMode,
+											   expr1loc);
 
 								/* Read and check the second one */
 								expr2 = read_sql_expression2(K_LOOP, K_BY,
@@ -1522,12 +1575,8 @@ for_control		: for_variable K_IN
 							else
 							{
 								/*
-								 * No "..", so it must be a query loop. We've
-								 * prefixed an extra SELECT to the query text,
-								 * so we need to remove that before performing
-								 * syntax checking.
+								 * No "..", so it must be a query loop.
 								 */
-								char				*tmp_query;
 								PLpgSQL_stmt_fors	*new;
 
 								if (reverse)
@@ -1536,12 +1585,9 @@ for_control		: for_variable K_IN
 											 errmsg("cannot specify REVERSE in query FOR loop"),
 											 parser_errposition(tokloc)));
 
-								Assert(strncmp(expr1->query, "SELECT ", 7) == 0);
-								tmp_query = pstrdup(expr1->query + 7);
-								pfree(expr1->query);
-								expr1->query = tmp_query;
-
-								check_sql_expr(expr1->query, expr1loc, 0);
+								/* Check syntax as a regular query */
+								check_sql_expr(expr1->query, expr1->parseMode,
+											   expr1loc);
 
 								new = palloc0(sizeof(PLpgSQL_stmt_fors));
 								new->cmd_type = PLPGSQL_STMT_FORS;
@@ -1870,7 +1916,7 @@ stmt_raise		: K_RAISE
 
 									expr = read_sql_construct(',', ';', K_USING,
 															  ", or ; or USING",
-															  "SELECT ",
+															  RAW_PARSE_PLPGSQL_EXPR,
 															  true, true, true,
 															  NULL, &tok);
 									new->params = lappend(new->params, expr);
@@ -1958,7 +2004,7 @@ loop_body		: proc_sect K_END K_LOOP opt_label ';'
  * variable.  (The composite case is probably a syntax error, but we'll let
  * the core parser decide that.)  Normally, we should assume that such a
  * word is a SQL statement keyword that isn't also a plpgsql keyword.
- * However, if the next token is assignment or '[', it can't be a valid
+ * However, if the next token is assignment or '[' or '.', it can't be a valid
  * SQL statement, and what we're probably looking at is an intended variable
  * assignment.  Give an appropriate complaint for that, instead of letting
  * the core parser throw an unhelpful "syntax error".
@@ -1977,7 +2023,8 @@ stmt_execsql	: K_IMPORT
 
 						tok = yylex();
 						plpgsql_push_back_token(tok);
-						if (tok == '=' || tok == COLON_EQUALS || tok == '[')
+						if (tok == '=' || tok == COLON_EQUALS ||
+							tok == '[' || tok == '.')
 							word_is_not_variable(&($1), @1);
 						$$ = make_execsql_stmt(T_WORD, @1);
 					}
@@ -1987,7 +2034,8 @@ stmt_execsql	: K_IMPORT
 
 						tok = yylex();
 						plpgsql_push_back_token(tok);
-						if (tok == '=' || tok == COLON_EQUALS || tok == '[')
+						if (tok == '=' || tok == COLON_EQUALS ||
+							tok == '[' || tok == '.')
 							cword_is_not_variable(&($1), @1);
 						$$ = make_execsql_stmt(T_CWORD, @1);
 					}
@@ -2001,7 +2049,7 @@ stmt_dynexecute : K_EXECUTE
 
 						expr = read_sql_construct(K_INTO, K_USING, ';',
 												  "INTO or USING or ;",
-												  "SELECT ",
+												  RAW_PARSE_PLPGSQL_EXPR,
 												  true, true, true,
 												  NULL, &endtoken);
 
@@ -2040,7 +2088,7 @@ stmt_dynexecute : K_EXECUTE
 								{
 									expr = read_sql_construct(',', ';', K_INTO,
 															  ", or ; or INTO",
-															  "SELECT ",
+															  RAW_PARSE_PLPGSQL_EXPR,
 															  true, true, true,
 															  NULL, &endtoken);
 									new->params = lappend(new->params, expr);
@@ -2122,7 +2170,7 @@ stmt_open		: K_OPEN cursor_variable
 							else
 							{
 								plpgsql_push_back_token(tok);
-								new->query = read_sql_stmt("");
+								new->query = read_sql_stmt();
 							}
 						}
 						else
@@ -2246,8 +2294,8 @@ stmt_set	: K_SET
 						new->cmd_type = PLPGSQL_STMT_SET;
 						new->lineno = plpgsql_location_to_lineno(@1);
 						new->stmtid = ++plpgsql_curr_compile->nstatements;
-
-						new->expr = read_sql_stmt("SET ");
+						plpgsql_push_back_token(K_SET);
+						new->expr = read_sql_stmt();
 
 						$$ = (PLpgSQL_stmt *)new;
 					}
@@ -2259,7 +2307,8 @@ stmt_set	: K_SET
 						new->cmd_type = PLPGSQL_STMT_SET;
 						new->lineno = plpgsql_location_to_lineno(@1);
 						new->stmtid = ++plpgsql_curr_compile->nstatements;
-						new->expr = read_sql_stmt("RESET ");
+						plpgsql_push_back_token(K_RESET);
+						new->expr = read_sql_stmt();
 
 						$$ = (PLpgSQL_stmt *)new;
 					}
@@ -2656,7 +2705,8 @@ static PLpgSQL_expr *
 read_sql_expression(int until, const char *expected)
 {
 	return read_sql_construct(until, 0, 0, expected,
-							  "SELECT ", true, true, true, NULL, NULL);
+							  RAW_PARSE_PLPGSQL_EXPR,
+							  true, true, true, NULL, NULL);
 }
 
 /* Convenience routine to read an expression with two possible terminators */
@@ -2665,15 +2715,17 @@ read_sql_expression2(int until, int until2, const char *expected,
 					 int *endtoken)
 {
 	return read_sql_construct(until, until2, 0, expected,
-							  "SELECT ", true, true, true, NULL, endtoken);
+							  RAW_PARSE_PLPGSQL_EXPR,
+							  true, true, true, NULL, endtoken);
 }
 
 /* Convenience routine to read a SQL statement that must end with ';' */
 static PLpgSQL_expr *
-read_sql_stmt(const char *sqlstart)
+read_sql_stmt(void)
 {
 	return read_sql_construct(';', 0, 0, ";",
-							  sqlstart, false, true, true, NULL, NULL);
+							  RAW_PARSE_DEFAULT,
+							  false, true, true, NULL, NULL);
 }
 
 /*
@@ -2683,9 +2735,9 @@ read_sql_stmt(const char *sqlstart)
  * until2:		token code for alternate terminator (pass 0 if none)
  * until3:		token code for another alternate terminator (pass 0 if none)
  * expected:	text to use in complaining that terminator was not found
- * sqlstart:	text to prefix to the accumulated SQL text
+ * parsemode:	raw_parser() mode to use
  * isexpression: whether to say we're reading an "expression" or a "statement"
- * valid_sql:   whether to check the syntax of the expr (prefixed with sqlstart)
+ * valid_sql:   whether to check the syntax of the expr
  * trim:		trim trailing whitespace
  * startloc:	if not NULL, location of first token is stored at *startloc
  * endtoken:	if not NULL, ending token is stored at *endtoken
@@ -2696,7 +2748,7 @@ read_sql_construct(int until,
 				   int until2,
 				   int until3,
 				   const char *expected,
-				   const char *sqlstart,
+				   RawParseMode parsemode,
 				   bool isexpression,
 				   bool valid_sql,
 				   bool trim,
@@ -2711,7 +2763,6 @@ read_sql_construct(int until,
 	PLpgSQL_expr		*expr;
 
 	initStringInfo(&ds);
-	appendStringInfoString(&ds, sqlstart);
 
 	/* special lookup mode for identifiers within the SQL text */
 	save_IdentifierLookup = plpgsql_IdentifierLookup;
@@ -2787,6 +2838,7 @@ read_sql_construct(int until,
 
 	expr = palloc0(sizeof(PLpgSQL_expr));
 	expr->query			= pstrdup(ds.data);
+	expr->parseMode		= parsemode;
 	expr->plan			= NULL;
 	expr->paramnos		= NULL;
 	expr->rwparam		= -1;
@@ -2794,7 +2846,7 @@ read_sql_construct(int until,
 	pfree(ds.data);
 
 	if (valid_sql)
-		check_sql_expr(expr->query, startlocation, strlen(sqlstart));
+		check_sql_expr(expr->query, expr->parseMode, startlocation);
 
 	return expr;
 }
@@ -3033,13 +3085,14 @@ make_execsql_stmt(int firsttoken, int location)
 
 	expr = palloc0(sizeof(PLpgSQL_expr));
 	expr->query			= pstrdup(ds.data);
+	expr->parseMode		= RAW_PARSE_DEFAULT;
 	expr->plan			= NULL;
 	expr->paramnos		= NULL;
 	expr->rwparam		= -1;
 	expr->ns			= plpgsql_ns_top();
 	pfree(ds.data);
 
-	check_sql_expr(expr->query, location, 0);
+	check_sql_expr(expr->query, expr->parseMode, location);
 
 	execsql = palloc(sizeof(PLpgSQL_stmt_execsql));
 	execsql->cmd_type = PLPGSQL_STMT_EXECSQL;
@@ -3382,7 +3435,7 @@ make_return_query_stmt(int location)
 	{
 		/* ordinary static query */
 		plpgsql_push_back_token(tok);
-		new->query = read_sql_stmt("");
+		new->query = read_sql_stmt();
 	}
 	else
 	{
@@ -3637,13 +3690,12 @@ make_scalar_list1(char *initial_name,
  * borders. So it is best to bail out as early as we can.
  *
  * It is assumed that "stmt" represents a copy of the function source text
- * beginning at offset "location", with leader text of length "leaderlen"
- * (typically "SELECT ") prefixed to the source text.  We use this assumption
- * to transpose any error cursor position back to the function source text.
+ * beginning at offset "location".  We use this assumption to transpose
+ * any error cursor position back to the function source text.
  * If no error cursor is provided, we'll just point at "location".
  */
 static void
-check_sql_expr(const char *stmt, int location, int leaderlen)
+check_sql_expr(const char *stmt, RawParseMode parseMode, int location)
 {
 	sql_error_callback_arg cbarg;
 	ErrorContextCallback  syntax_errcontext;
@@ -3653,7 +3705,6 @@ check_sql_expr(const char *stmt, int location, int leaderlen)
 		return;
 
 	cbarg.location = location;
-	cbarg.leaderlen = leaderlen;
 
 	syntax_errcontext.callback = plpgsql_sql_error_callback;
 	syntax_errcontext.arg = &cbarg;
@@ -3661,7 +3712,7 @@ check_sql_expr(const char *stmt, int location, int leaderlen)
 	error_context_stack = &syntax_errcontext;
 
 	oldCxt = MemoryContextSwitchTo(plpgsql_compile_tmp_cxt);
-	(void) raw_parser(stmt, RAW_PARSE_DEFAULT);
+	(void) raw_parser(stmt, parseMode);
 	MemoryContextSwitchTo(oldCxt);
 
 	/* Restore former ereport callback */
@@ -3686,12 +3737,12 @@ plpgsql_sql_error_callback(void *arg)
 	 * Note we are dealing with 1-based character numbers at this point.
 	 */
 	errpos = geterrposition();
-	if (errpos > cbarg->leaderlen)
+	if (errpos > 0)
 	{
 		int		myerrpos = getinternalerrposition();
 
 		if (myerrpos > 0)		/* safety check */
-			internalerrposition(myerrpos + errpos - cbarg->leaderlen - 1);
+			internalerrposition(myerrpos + errpos - 1);
 	}
 
 	/* In any case, flush errposition --- we want internalerrposition only */
@@ -3717,7 +3768,6 @@ parse_datatype(const char *string, int location)
 	ErrorContextCallback  syntax_errcontext;
 
 	cbarg.location = location;
-	cbarg.leaderlen = 0;
 
 	syntax_errcontext.callback = plpgsql_sql_error_callback;
 	syntax_errcontext.arg = &cbarg;
@@ -3780,7 +3830,6 @@ read_cursor_args(PLpgSQL_var *cursor, int until)
 	int			argc;
 	char	  **argv;
 	StringInfoData ds;
-	char	   *sqlstart = "SELECT ";
 	bool		any_named = false;
 
 	tok = yylex();
@@ -3881,12 +3930,12 @@ read_cursor_args(PLpgSQL_var *cursor, int until)
 		 */
 		item = read_sql_construct(',', ')', 0,
 								  ",\" or \")",
-								  sqlstart,
+								  RAW_PARSE_PLPGSQL_EXPR,
 								  true, true,
 								  false, /* do not trim */
 								  NULL, &endtoken);
 
-		argv[argpos] = item->query + strlen(sqlstart);
+		argv[argpos] = item->query;
 
 		if (endtoken == ')' && !(argc == row->nfields - 1))
 			ereport(ERROR,
@@ -3905,7 +3954,6 @@ read_cursor_args(PLpgSQL_var *cursor, int until)
 
 	/* Make positional argument list */
 	initStringInfo(&ds);
-	appendStringInfoString(&ds, sqlstart);
 	for (argc = 0; argc < row->nfields; argc++)
 	{
 		Assert(argv[argc] != NULL);
@@ -3921,10 +3969,10 @@ read_cursor_args(PLpgSQL_var *cursor, int until)
 		if (argc < row->nfields - 1)
 			appendStringInfoString(&ds, ", ");
 	}
-	appendStringInfoChar(&ds, ';');
 
 	expr = palloc0(sizeof(PLpgSQL_expr));
 	expr->query			= pstrdup(ds.data);
+	expr->parseMode		= RAW_PARSE_PLPGSQL_EXPR;
 	expr->plan			= NULL;
 	expr->paramnos		= NULL;
 	expr->rwparam		= -1;
@@ -4097,14 +4145,14 @@ make_case(int location, PLpgSQL_expr *t_expr,
 			PLpgSQL_expr *expr = cwt->expr;
 			StringInfoData	ds;
 
-			/* copy expression query without SELECT keyword (expr->query + 7) */
-			Assert(strncmp(expr->query, "SELECT ", 7) == 0);
+			/* We expect to have expressions not statements */
+			Assert(expr->parseMode == RAW_PARSE_PLPGSQL_EXPR);
 
-			/* And do the string hacking */
+			/* Do the string hacking */
 			initStringInfo(&ds);
 
-			appendStringInfo(&ds, "SELECT \"%s\" IN (%s)",
-							 varname, expr->query + 7);
+			appendStringInfo(&ds, "\"%s\" IN (%s)",
+							 varname, expr->query);
 
 			pfree(expr->query);
 			expr->query = pstrdup(ds.data);
diff --git a/src/pl/plpgsql/src/plpgsql.h b/src/pl/plpgsql/src/plpgsql.h
index dee175b7bf6..f80d023a5ad 100644
--- a/src/pl/plpgsql/src/plpgsql.h
+++ b/src/pl/plpgsql/src/plpgsql.h
@@ -218,8 +218,9 @@ typedef struct PLpgSQL_type
  */
 typedef struct PLpgSQL_expr
 {
-	char	   *query;
-	SPIPlanPtr	plan;
+	char	   *query;			/* query string, verbatim from function body */
+	RawParseMode parseMode;		/* raw_parser() mode to use */
+	SPIPlanPtr	plan;			/* plan, or NULL if not made yet */
 	Bitmapset  *paramnos;		/* all dnos referenced by this query */
 	int			rwparam;		/* dno of read/write param, or -1 if none */
 
diff --git a/src/pl/plpgsql/src/sql/plpgsql_array.sql b/src/pl/plpgsql/src/sql/plpgsql_array.sql
new file mode 100644
index 00000000000..4c3f26be101
--- /dev/null
+++ b/src/pl/plpgsql/src/sql/plpgsql_array.sql
@@ -0,0 +1,79 @@
+--
+-- Tests for PL/pgSQL handling of array variables
+--
+-- We also check arrays of composites here, so this has some overlap
+-- with the plpgsql_record tests.
+--
+
+create type complex as (r float8, i float8);
+create type quadarray as (c1 complex[], c2 complex);
+
+do $$ declare a int[];
+begin a := array[1,2]; a[3] := 4; raise notice 'a = %', a; end$$;
+
+do $$ declare a int[];
+begin a[3] := 4; raise notice 'a = %', a; end$$;
+
+do $$ declare a int[];
+begin a[1][4] := 4; raise notice 'a = %', a; end$$;
+
+do $$ declare a int[];
+begin a[1] := 23::text; raise notice 'a = %', a; end$$;  -- lax typing
+
+do $$ declare a int[];
+begin a := array[1,2]; a[2:3] := array[3,4]; raise notice 'a = %', a; end$$;
+
+do $$ declare a int[];
+begin a := array[1,2]; a[2] := a[2] + 1; raise notice 'a = %', a; end$$;
+
+do $$ declare a int[];
+begin a[1:2] := array[3,4]; raise notice 'a = %', a; end$$;
+
+do $$ declare a int[];
+begin a[1:2] := 4; raise notice 'a = %', a; end$$;  -- error
+
+do $$ declare a complex[];
+begin a[1] := (1,2); a[1].i := 11; raise notice 'a = %', a; end$$;
+
+do $$ declare a complex[];
+begin a[1].i := 11; raise notice 'a = %, a[1].i = %', a, a[1].i; end$$;
+
+-- perhaps this ought to work, but for now it doesn't:
+do $$ declare a complex[];
+begin a[1:2].i := array[11,12]; raise notice 'a = %', a; end$$;
+
+do $$ declare a quadarray;
+begin a.c1[1].i := 11; raise notice 'a = %, a.c1[1].i = %', a, a.c1[1].i; end$$;
+
+do $$ declare a int[];
+begin a := array_agg(x) from (values(1),(2),(3)) v(x); raise notice 'a = %', a; end$$;
+
+create temp table onecol as select array[1,2] as f1;
+
+do $$ declare a int[];
+begin a := f1 from onecol; raise notice 'a = %', a; end$$;
+
+do $$ declare a int[];
+begin a := * from onecol for update; raise notice 'a = %', a; end$$;
+
+-- error cases:
+
+do $$ declare a int[];
+begin a := from onecol; raise notice 'a = %', a; end$$;
+
+do $$ declare a int[];
+begin a := f1, f1 from onecol; raise notice 'a = %', a; end$$;
+
+insert into onecol values(array[11]);
+
+do $$ declare a int[];
+begin a := f1 from onecol; raise notice 'a = %', a; end$$;
+
+do $$ declare a int[];
+begin a := f1 from onecol limit 1; raise notice 'a = %', a; end$$;
+
+do $$ declare a real;
+begin a[1] := 2; raise notice 'a = %', a; end$$;
+
+do $$ declare a complex;
+begin a.r[1] := 2; raise notice 'a = %', a; end$$;
diff --git a/src/pl/plpgsql/src/sql/plpgsql_record.sql b/src/pl/plpgsql/src/sql/plpgsql_record.sql
index 128846e6108..be10f00b1e6 100644
--- a/src/pl/plpgsql/src/sql/plpgsql_record.sql
+++ b/src/pl/plpgsql/src/sql/plpgsql_record.sql
@@ -4,6 +4,7 @@
 
 create type two_int4s as (f1 int4, f2 int4);
 create type two_int8s as (q1 int8, q2 int8);
+create type nested_int8s as (c1 two_int8s, c2 two_int8s);
 
 -- base-case return of a composite type
 create function retc(int) returns two_int8s language plpgsql as
@@ -59,6 +60,47 @@ begin
   raise notice 'c8 = %', c8;
 end$$;
 
+do $$ declare c two_int8s; d nested_int8s;
+begin
+  c := row(1,2);
+  d := row(c, row(c.q1, c.q2+1));
+  raise notice 'c = %, d = %', c, d;
+  c.q1 := 10;
+  d.c1 := row(11,12);
+  d.c2.q2 := 42;
+  raise notice 'c = %, d = %', c, d;
+  raise notice 'c.q1 = %, d.c2 = %', c.q1, d.c2;
+  raise notice '(d).c2.q2 = %', (d).c2.q2;  -- doesn't work without parens
+  raise notice '(d.c2).q2 = %', (d.c2).q2;  -- doesn't work without parens
+end$$;
+
+-- block-qualified naming
+do $$ <<b>> declare c two_int8s; d nested_int8s;
+begin
+  b.c := row(1,2);
+  b.d := row(b.c, row(b.c.q1, b.c.q2+1));
+  raise notice 'b.c = %, b.d = %', b.c, b.d;
+  b.c.q1 := 10;
+  b.d.c1 := row(11,12);
+  b.d.c2.q2 := 42;
+  raise notice 'b.c = %, b.d = %', b.c, b.d;
+  raise notice 'b.c.q1 = %, b.d.c2 = %', b.c.q1, b.d.c2;
+  raise notice '(b.d).c2.q2 = %', (b.d).c2.q2;  -- doesn't work without parens
+  raise notice '(b.d.c2).q2 = %', (b.d.c2).q2;  -- doesn't work without parens
+end$$;
+
+-- error cases
+do $$ declare c two_int8s; begin c.x = 1; end $$;
+do $$ declare c nested_int8s; begin c.x = 1; end $$;
+do $$ declare c nested_int8s; begin c.x.q1 = 1; end $$;
+do $$ declare c nested_int8s; begin c.c2.x = 1; end $$;
+do $$ declare c nested_int8s; begin d.c2.x = 1; end $$;
+do $$ <<b>> declare c two_int8s; begin b.c.x = 1; end $$;
+do $$ <<b>> declare c nested_int8s; begin b.c.x = 1; end $$;
+do $$ <<b>> declare c nested_int8s; begin b.c.x.q1 = 1; end $$;
+do $$ <<b>> declare c nested_int8s; begin b.c.c2.x = 1; end $$;
+do $$ <<b>> declare c nested_int8s; begin b.d.c2.x = 1; end $$;
+
 -- check passing composite result to another function
 create function getq1(two_int8s) returns int8 language plpgsql as $$
 declare r two_int8s; begin r := $1; return r.q1; end $$;