why can't plpgsql return a row-expression?

diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 3b5b3bb..155e6f9 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -2454,9 +2454,60 @@ exec_stmt_return(PLpgSQL_execstate *estate, PLpgSQL_stmt_return *stmt)
 			exec_run_select(estate, stmt->expr, 1, NULL);
 			if (estate->eval_processed > 0)
 			{
-				estate->retval = PointerGetDatum(estate->eval_tuptable->vals[0]);
-				estate->rettupdesc = estate->eval_tuptable->tupdesc;
-				estate->retisnull = false;
+				TupleDesc tupdesc;
+				HeapTuple tuple;
+				bool isnull;
+
+				tuple = estate->eval_tuptable->vals[0];
+				tupdesc = estate->eval_tuptable->tupdesc;
+				isnull = false;
+				/*
+				 * Expression in the return statement can be of row expr. In that case
+				 * resultant tuple will have rowtype in a single column. so we would
+				 * need to extract and return the inner tuple. i.e.
+				 *
+				 * return (1, 'hello'::varchar);
+				 * where function's return type is a composite type such as
+				 * "footype(int, varchar)".
+				 *
+				 * Note that we do not do this if functions return type is record. That
+				 * is to comply with existing behavior.
+				 */
+				if (tupdesc->natts == 1 &&
+					estate->fn_rettype != RECORDOID &&
+					type_is_rowtype(tupdesc->attrs[0]->atttypid))
+				{
+					HeapTupleData	htup;
+					HeapTupleHeader td;
+					Oid		tupType;
+					int32	tupTypmod;
+					Datum	value;
+
+					isnull = true;
+					if (HeapTupleIsValid(tuple))
+					{
+						value = heap_getattr(tuple, 1, tupdesc, &isnull);
+						if (!isnull)
+						{
+							/* Source is a tuple Datum, so safe to do this: */
+							td = DatumGetHeapTupleHeader(value);
+							/* Extract rowtype info and find a tupdesc */
+							tupType = HeapTupleHeaderGetTypeId(td);
+							tupTypmod = HeapTupleHeaderGetTypMod(td);
+							tupdesc = lookup_rowtype_tupdesc_copy(tupType, tupTypmod);
+							/* Build a HeapTuple control structure */
+							htup.t_len = HeapTupleHeaderGetDatumLength(td);
+							ItemPointerSetInvalid(&(htup.t_self));
+							htup.t_tableOid = InvalidOid;
+							htup.t_data = td;
+							tuple = heap_copytuple(&htup);
+						}
+					}
+				}
+
+				estate->retval = PointerGetDatum(tuple);
+				estate->rettupdesc = tupdesc;
+				estate->retisnull = isnull;
 			}
 		}
 		else
diff --git a/src/pl/plpgsql/src/pl_gram.y b/src/pl/plpgsql/src/pl_gram.y
index cf164d0..bfc8bfa 100644
--- a/src/pl/plpgsql/src/pl_gram.y
+++ b/src/pl/plpgsql/src/pl_gram.y
@@ -2926,7 +2926,8 @@ make_return_stmt(int location)
 	}
 	else if (plpgsql_curr_compile->fn_retistuple)
 	{
-		switch (yylex())
+		int tok = yylex();
+		switch (tok)
 		{
 			case K_NULL:
 				/* we allow this to support RETURN NULL in triggers */
@@ -2944,10 +2945,13 @@ make_return_stmt(int location)
 				break;
 
 			default:
-				ereport(ERROR,
-						(errcode(ERRCODE_DATATYPE_MISMATCH),
-						 errmsg("RETURN must specify a record or row variable in function returning row"),
-						 parser_errposition(yylloc)));
+				/*
+				 * Allow use of expression in return statement for functions returning
+				 * row types.
+				 */
+				plpgsql_push_back_token(tok);
+				new->expr = read_sql_expression(';', ";");
+				return (PLpgSQL_stmt *) new;
 				break;
 		}
 		if (yylex() != ';')
diff --git a/src/test/regress/serial_schedule b/src/test/regress/serial_schedule
index be789e3..7ea2a26 100644
--- a/src/test/regress/serial_schedule
+++ b/src/test/regress/serial_schedule
@@ -134,3 +134,4 @@ test: largeobject
 test: with
 test: xml
 test: stats
+test: return_rowtype
diff --git a/src/test/regress/expected/return_rowtype.out b/src/test/regress/expected/return_rowtype.out
new file mode 100644
index 0000000..aa6edf1
--- /dev/null
+++ b/src/test/regress/expected/return_rowtype.out
@@ -0,0 +1,151 @@
+--create an composite type
+create type footype as (x int, y varchar);
+--test: use of variable of composite type in return statement
+create or replace function foo() returns footype as $$
+declare
+  v footype;
+begin
+  v := (1, 'hello');
+  return v;
+end;
+$$ language plpgsql;
+select foo();
+    foo    
+-----------
+ (1,hello)
+(1 row)
+
+--test: use row expr in return statement
+create or replace function foo() returns footype as $$
+begin
+  return (1, 'hello'::varchar);
+end;
+$$ language plpgsql;
+select foo();
+    foo    
+-----------
+ (1,hello)
+(1 row)
+
+DO $$
+declare
+  v footype;
+begin
+  v := foo();
+  raise info 'x = %', v.x;
+  raise info 'y = %', v.y;
+end;
+$$;
+INFO:  x = 1
+INFO:  y = hello
+drop function foo();
+--create a table
+create table footab(x int, y varchar(10));
+--test: return a row expr
+create or replace function foorec() returns footab as $$
+begin
+  return (1, 'hello'::varchar(10));
+end;
+$$ language plpgsql;
+DO $$
+declare
+  v footab%rowtype;
+begin
+  v := foorec();
+  raise info 'x = %', v.x;
+  raise info 'y = %', v.y;
+end; $$;
+INFO:  x = 1
+INFO:  y = hello
+drop function foorec();
+drop table footab;
+--test: return a row expr as record, (ensure record behavior is not changed)
+create or replace function foorec() returns record as $$
+declare
+  v record;
+begin
+  v := (1, 'hello');
+  return v;
+end;
+$$ language plpgsql;
+select foorec();
+  foorec   
+-----------
+ (1,hello)
+(1 row)
+
+DO $$
+declare
+  v record;
+begin
+  v := foorec();
+  raise info 'rec = %', v;
+end; $$;
+INFO:  rec = (1,hello)
+--test: return row expr in return statement.
+create or replace function foorec() returns record as $$
+begin
+  return (1, 'hello');
+end;
+$$ language plpgsql;
+select foorec();
+    foorec     
+---------------
+ ("(1,hello)")
+(1 row)
+
+DO $$
+declare
+  v record;
+begin
+  v := foorec();
+  raise info 'rec = %', v;
+end; $$;
+INFO:  rec = ("(1,hello)")
+drop function foorec();
+--test: use invalid expr in return statement.
+create or replace function foo() returns footype as $$
+begin
+  return 1 + 1;
+end;
+$$ language plpgsql;
+DO $$
+declare
+  v footype;
+begin
+  v := foo();
+  raise info 'x = %', v.x;
+  raise info 'y = %', v.y;
+end; $$;
+ERROR:  returned record type does not match expected record type
+DETAIL:  Number of returned columns (1) does not match expected column count (2).
+CONTEXT:  PL/pgSQL function foo() while casting return value to function's return type
+PL/pgSQL function inline_code_block line 5 at assignment
+--test: use invalid expr in return statement
+create or replace function foo() returns footype as $$
+declare
+  v record;
+  v2 record;
+begin
+  v := (1, 'hello');
+  v2 := (1, 'hello');
+  return (v || v2);
+end;
+$$ language plpgsql;
+DO $$
+declare
+  v footype;
+begin
+  v := foo();
+  raise info 'x = %', v.x;
+  raise info 'y = %', v.y;
+end; $$;
+ERROR:  operator does not exist: record || record
+LINE 1: SELECT (v || v2)
+                  ^
+HINT:  No operator matches the given name and argument type(s). You might need to add explicit type casts.
+QUERY:  SELECT (v || v2)
+CONTEXT:  PL/pgSQL function foo() line 8 at RETURN
+PL/pgSQL function inline_code_block line 5 at assignment
+drop function foo();
+drop type footype;
diff --git a/src/test/regress/sql/return_rowtype.sql b/src/test/regress/sql/return_rowtype.sql
new file mode 100644
index 0000000..8ba2c5d
--- /dev/null
+++ b/src/test/regress/sql/return_rowtype.sql
@@ -0,0 +1,137 @@
+--create an composite type
+create type footype as (x int, y varchar);
+
+--test: use of variable of composite type in return statement
+create or replace function foo() returns footype as $$
+declare
+  v footype;
+begin
+  v := (1, 'hello');
+  return v;
+end;
+$$ language plpgsql;
+
+select foo();
+
+--test: use row expr in return statement
+create or replace function foo() returns footype as $$
+begin
+  return (1, 'hello'::varchar);
+end;
+$$ language plpgsql;
+
+select foo();
+
+DO $$
+declare
+  v footype;
+begin
+  v := foo();
+  raise info 'x = %', v.x;
+  raise info 'y = %', v.y;
+end;
+$$;
+
+drop function foo();
+
+--create a table
+create table footab(x int, y varchar(10));
+
+--test: return a row expr
+create or replace function foorec() returns footab as $$
+begin
+  return (1, 'hello'::varchar(10));
+end;
+$$ language plpgsql;
+
+DO $$
+declare
+  v footab%rowtype;
+begin
+  v := foorec();
+  raise info 'x = %', v.x;
+  raise info 'y = %', v.y;
+end; $$;
+
+drop function foorec();
+drop table footab;
+
+--test: return a row expr as record, (ensure record behavior is not changed)
+create or replace function foorec() returns record as $$
+declare
+  v record;
+begin
+  v := (1, 'hello');
+  return v;
+end;
+$$ language plpgsql;
+
+select foorec();
+
+DO $$
+declare
+  v record;
+begin
+  v := foorec();
+  raise info 'rec = %', v;
+end; $$;
+
+--test: return row expr in return statement.
+create or replace function foorec() returns record as $$
+begin
+  return (1, 'hello');
+end;
+$$ language plpgsql;
+
+select foorec();
+
+DO $$
+declare
+  v record;
+begin
+  v := foorec();
+  raise info 'rec = %', v;
+end; $$;
+
+drop function foorec();
+
+--test: use invalid expr in return statement.
+create or replace function foo() returns footype as $$
+begin
+  return 1 + 1;
+end;
+$$ language plpgsql;
+
+DO $$
+declare
+  v footype;
+begin
+  v := foo();
+  raise info 'x = %', v.x;
+  raise info 'y = %', v.y;
+end; $$;
+
+--test: use invalid expr in return statement
+create or replace function foo() returns footype as $$
+declare
+  v record;
+  v2 record;
+begin
+  v := (1, 'hello');
+  v2 := (1, 'hello');
+  return (v || v2);
+end;
+$$ language plpgsql;
+
+DO $$
+declare
+  v footype;
+begin
+  v := foo();
+  raise info 'x = %', v.x;
+  raise info 'y = %', v.y;
+end; $$;
+
+drop function foo();
+drop type footype;
+