Re: why can't plpgsql return a row-expression?

diff --git a/doc/src/sgml/plpgsql.sgml b/doc/src/sgml/plpgsql.sgml
index 07fba57..58b1489 100644
--- a/doc/src/sgml/plpgsql.sgml
+++ b/doc/src/sgml/plpgsql.sgml
@@ -1571,11 +1571,9 @@ RETURN <replaceable>expression</replaceable>;
      </para>
 
      <para>
-      When returning a scalar type, any expression can be used. The
-      expression's result will be automatically cast into the
-      function's return type as described for assignments. To return a
-      composite (row) value, you must write a record or row variable
-      as the <replaceable>expression</replaceable>.
+      In <command>RETURN</command> statement you can use any expression.
+       The expression's result will be automatically cast into the
+      function's return type as described for assignments.
      </para>
 
      <para>
@@ -1600,6 +1598,21 @@ RETURN <replaceable>expression</replaceable>;
       however.  In those cases a <command>RETURN</command> statement is
       automatically executed if the top-level block finishes.
      </para>
+
+     <para>
+      Some examples:
+
+<programlisting>
+-- functions returning a scalar type
+RETURN 1 + 2;
+RETURN scalar_var;
+
+-- functions returning a composite type
+RETURN composite_type_var;
+RETURN (1, 2, 'three');
+</programlisting>
+
+     </para>
     </sect3>
 
     <sect3>
diff --git a/src/backend/access/common/tupconvert.c b/src/backend/access/common/tupconvert.c
index f813432..0f5ac50 100644
--- a/src/backend/access/common/tupconvert.c
+++ b/src/backend/access/common/tupconvert.c
@@ -22,7 +22,9 @@
 
 #include "access/htup_details.h"
 #include "access/tupconvert.h"
+#include "parser/parse_coerce.h"
 #include "utils/builtins.h"
+#include "utils/lsyscache.h"
 
 
 /*
@@ -69,6 +71,9 @@ convert_tuples_by_position(TupleDesc indesc,
 {
 	TupleConversionMap *map;
 	AttrNumber *attrMap;
+	Oid		   *typoutput;
+	Oid		   *typinput;
+	Oid		   *typioparam;
 	int			nincols;
 	int			noutcols;
 	int			n;
@@ -78,7 +83,10 @@ convert_tuples_by_position(TupleDesc indesc,
 
 	/* Verify compatibility and prepare attribute-number map */
 	n = outdesc->natts;
-	attrMap = (AttrNumber *) palloc0(n * sizeof(AttrNumber));
+	attrMap 	= (AttrNumber *) palloc0(n * sizeof(AttrNumber));
+	typinput 	= (Oid *) palloc0(n * sizeof(Oid));
+	typioparam	= (Oid *) palloc(n * sizeof(Oid));
+	typoutput 	= (Oid *) palloc((indesc->natts + 1) * sizeof(Oid)); /* +1 for NULL */
 	j = 0;						/* j is next physical input attribute */
 	nincols = noutcols = 0;		/* these count non-dropped attributes */
 	same = true;
@@ -100,8 +108,8 @@ convert_tuples_by_position(TupleDesc indesc,
 				continue;
 			nincols++;
 			/* Found matching column, check type */
-			if (atttypid != att->atttypid ||
-				(atttypmod != att->atttypmod && atttypmod >= 0))
+			if ((atttypmod != att->atttypmod && atttypmod >= 0) ||
+				!can_coerce_type(1, &atttypid, &(att->atttypid), COERCE_IMPLICIT_CAST))
 				ereport(ERROR,
 						(errcode(ERRCODE_DATATYPE_MISMATCH),
 						 errmsg_internal("%s", _(msg)),
@@ -111,6 +119,17 @@ convert_tuples_by_position(TupleDesc indesc,
 								   format_type_with_typemod(atttypid,
 															atttypmod),
 								   noutcols)));
+			/*
+			 * get the needed I/O functions to perform cast later in
+			 * do_convert_tuple, no need if type id's are same.
+			 */
+			if (atttypid != att->atttypid)
+			{
+				bool typIsVarlena;
+
+				getTypeOutputInfo(att->atttypid, &typoutput[j + 1], &typIsVarlena);
+				getTypeInputInfo(atttypid, &typinput[i], &typioparam[i]);
+			}
 			attrMap[i] = (AttrNumber) (j + 1);
 			j++;
 			break;
@@ -183,9 +202,12 @@ convert_tuples_by_position(TupleDesc indesc,
 	/* preallocate workspace for Datum arrays */
 	map->outvalues = (Datum *) palloc(n * sizeof(Datum));
 	map->outisnull = (bool *) palloc(n * sizeof(bool));
+	map->typinput = typinput;
+	map->typioparam = typioparam;
 	n = indesc->natts + 1;		/* +1 for NULL */
 	map->invalues = (Datum *) palloc(n * sizeof(Datum));
 	map->inisnull = (bool *) palloc(n * sizeof(bool));
+	map->typoutput = typoutput;
 	map->invalues[0] = (Datum) 0;		/* set up the NULL entry */
 	map->inisnull[0] = true;
 
@@ -206,6 +228,9 @@ convert_tuples_by_name(TupleDesc indesc,
 {
 	TupleConversionMap *map;
 	AttrNumber *attrMap;
+	Oid		   *typoutput;
+	Oid		   *typinput;
+	Oid		   *typioparam;
 	int			n;
 	int			i;
 	bool		same;
@@ -213,6 +238,9 @@ convert_tuples_by_name(TupleDesc indesc,
 	/* Verify compatibility and prepare attribute-number map */
 	n = outdesc->natts;
 	attrMap = (AttrNumber *) palloc0(n * sizeof(AttrNumber));
+	typinput 	= (Oid *) palloc0(n * sizeof(Oid));
+	typioparam	= (Oid *) palloc(n * sizeof(Oid));
+	typoutput 	= (Oid *) palloc((indesc->natts + 1) * sizeof(Oid)); /* +1 for NULL */
 	for (i = 0; i < n; i++)
 	{
 		Form_pg_attribute att = outdesc->attrs[i];
@@ -234,7 +262,8 @@ convert_tuples_by_name(TupleDesc indesc,
 			if (strcmp(attname, NameStr(att->attname)) == 0)
 			{
 				/* Found it, check type */
-				if (atttypid != att->atttypid || atttypmod != att->atttypmod)
+				if (atttypmod != att->atttypmod ||
+					!can_coerce_type(1, &atttypid, &(att->atttypid), COERCE_IMPLICIT_CAST))
 					ereport(ERROR,
 							(errcode(ERRCODE_DATATYPE_MISMATCH),
 							 errmsg_internal("%s", _(msg)),
@@ -242,6 +271,17 @@ convert_tuples_by_name(TupleDesc indesc,
 									   attname,
 									   format_type_be(outdesc->tdtypeid),
 									   format_type_be(indesc->tdtypeid))));
+				/*
+				 * get the needed I/O functions to perform cast later in
+				 * do_convert_tuple, no need if type id's are same.
+				 */
+				if (atttypid != att->atttypid)
+				{
+					bool typIsVarlena;
+
+					getTypeOutputInfo(att->atttypid, &typoutput[j + 1], &typIsVarlena);
+					getTypeInputInfo(atttypid, &typinput[i], &typioparam[i]);
+				}
 				attrMap[i] = (AttrNumber) (j + 1);
 				break;
 			}
@@ -303,9 +343,12 @@ convert_tuples_by_name(TupleDesc indesc,
 	/* preallocate workspace for Datum arrays */
 	map->outvalues = (Datum *) palloc(n * sizeof(Datum));
 	map->outisnull = (bool *) palloc(n * sizeof(bool));
+	map->typinput = typinput;
+	map->typioparam = typioparam;
 	n = indesc->natts + 1;		/* +1 for NULL */
 	map->invalues = (Datum *) palloc(n * sizeof(Datum));
 	map->inisnull = (bool *) palloc(n * sizeof(bool));
+	map->typoutput = typoutput;
 	map->invalues[0] = (Datum) 0;		/* set up the NULL entry */
 	map->inisnull[0] = true;
 
@@ -342,6 +385,21 @@ do_convert_tuple(HeapTuple tuple, TupleConversionMap *map)
 
 		outvalues[i] = invalues[j];
 		outisnull[i] = inisnull[j];
+
+		/* Perform the casting, if necessary. */
+		if (!inisnull[j] &&
+			OidIsValid(map->typinput[i]))
+		{
+			Form_pg_attribute att = map->outdesc->attrs[i];
+			char *strval;
+
+			strval = OidOutputFunctionCall(map->typoutput[j], invalues[j]);
+			outvalues[i] = OidInputFunctionCall(map->typinput[i],
+											 strval,
+											 map->typioparam[i],
+											 att->atttypmod);
+			pfree(strval);
+		}
 	}
 
 	/*
@@ -360,7 +418,10 @@ free_conversion_map(TupleConversionMap *map)
 	pfree(map->attrMap);
 	pfree(map->invalues);
 	pfree(map->inisnull);
+	pfree(map->typoutput);
 	pfree(map->outvalues);
 	pfree(map->outisnull);
+	pfree(map->typinput);
+	pfree(map->typioparam);
 	pfree(map);
 }
diff --git a/src/include/access/tupconvert.h b/src/include/access/tupconvert.h
index 342dbb4..61ef7ae 100644
--- a/src/include/access/tupconvert.h
+++ b/src/include/access/tupconvert.h
@@ -25,8 +25,11 @@ typedef struct TupleConversionMap
 	AttrNumber *attrMap;		/* indexes of input fields, or 0 for null */
 	Datum	   *invalues;		/* workspace for deconstructing source */
 	bool	   *inisnull;
+	Oid		   *typoutput;		/* output functions for source type */
 	Datum	   *outvalues;		/* workspace for constructing result */
 	bool	   *outisnull;
+	Oid		   *typinput;		/* input functions for result type */
+	Oid		   *typioparam;
 } TupleConversionMap;
 
 
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 3b5b3bb..d1c0acd 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -214,6 +214,7 @@ static void free_params_data(PreparedParamsData *ppd);
 static Portal exec_dynquery_with_params(PLpgSQL_execstate *estate,
 						  PLpgSQL_expr *dynquery, List *params,
 						  const char *portalname, int cursorOptions);
+static HeapTuple get_tuple_from_datum(Datum value, TupleDesc *rettupdesc);
 
 
 /* ----------
@@ -275,23 +276,11 @@ plpgsql_exec_function(PLpgSQL_function *func, FunctionCallInfo fcinfo)
 
 					if (!fcinfo->argnull[i])
 					{
-						HeapTupleHeader td;
-						Oid			tupType;
-						int32		tupTypmod;
 						TupleDesc	tupdesc;
-						HeapTupleData tmptup;
-
-						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);
-						ItemPointerSetInvalid(&(tmptup.t_self));
-						tmptup.t_tableOid = InvalidOid;
-						tmptup.t_data = td;
-						exec_move_row(&estate, NULL, row, &tmptup, tupdesc);
+						HeapTuple	tuple;
+
+						tuple = get_tuple_from_datum(fcinfo->arg[i], &tupdesc);
+						exec_move_row(&estate, NULL, row, tuple, tupdesc);
 						ReleaseTupleDesc(tupdesc);
 					}
 					else
@@ -2449,24 +2438,30 @@ exec_stmt_return(PLpgSQL_execstate *estate, PLpgSQL_stmt_return *stmt)
 
 	if (stmt->expr != NULL)
 	{
+		estate->retval = exec_eval_expr(estate, stmt->expr,
+										&(estate->retisnull),
+										&(estate->rettype));
 		if (estate->retistuple)
 		{
-			exec_run_select(estate, stmt->expr, 1, NULL);
-			if (estate->eval_processed > 0)
+			/* Source must be of RECORD or composite type */
+			if (!type_is_rowtype(estate->rettype))
+				ereport(ERROR,
+						(errcode(ERRCODE_DATATYPE_MISMATCH),
+						errmsg("cannot return non-composite value from composite type returning function")));
+
+			if (!estate->retisnull)
 			{
-				estate->retval = PointerGetDatum(estate->eval_tuptable->vals[0]);
-				estate->rettupdesc = estate->eval_tuptable->tupdesc;
-				estate->retisnull = false;
+				HeapTuple	tuple;
+				TupleDesc	tupdesc;
+
+				tuple = get_tuple_from_datum(estate->retval, &tupdesc);
+				estate->retval = PointerGetDatum(tuple);
+				estate->rettupdesc = CreateTupleDescCopy(tupdesc);
+				ReleaseTupleDesc(tupdesc);
 			}
 		}
-		else
-		{
-			/* Normal case for scalar results */
-			estate->retval = exec_eval_expr(estate, stmt->expr,
-											&(estate->retisnull),
-											&(estate->rettype));
-		}
 
+		/* Else, the expr is of scalar type and has been evaluated. simply return. */
 		return PLPGSQL_RC_RETURN;
 	}
 
@@ -2593,26 +2588,51 @@ exec_stmt_return_next(PLpgSQL_execstate *estate,
 		bool		isNull;
 		Oid			rettype;
 
-		if (natts != 1)
-			ereport(ERROR,
-					(errcode(ERRCODE_DATATYPE_MISMATCH),
-					 errmsg("wrong result type supplied in RETURN NEXT")));
-
 		retval = exec_eval_expr(estate,
 								stmt->expr,
 								&isNull,
 								&rettype);
 
-		/* coerce type if needed */
-		retval = exec_simple_cast_value(estate,
-										retval,
-										rettype,
-										tupdesc->attrs[0]->atttypid,
-										tupdesc->attrs[0]->atttypmod,
-										isNull);
+		/* Check if expr is of RECORD or composite type */
+		if (type_is_rowtype(rettype))
+		{
+			TupleConversionMap *tupmap;
+			TupleDesc retdesc;
+
+			tuple = get_tuple_from_datum(retval, &retdesc);
+			tupmap = convert_tuples_by_position(retdesc,
+												estate->rettupdesc,
+												gettext_noop("returned record type does not match expected record type"));
+			/* it might need conversion */
+			if (tupmap)
+			{
+				tuple = do_convert_tuple(tuple, tupmap);
+				free_conversion_map(tupmap);
+				free_tuple = true;
+			}
+
+			ReleaseTupleDesc(retdesc);
+		}
+		else
+		{
+			/* Normal case for scalar results */
+
+			if (natts != 1)
+				ereport(ERROR,
+						(errcode(ERRCODE_DATATYPE_MISMATCH),
+						errmsg("wrong result type supplied in RETURN NEXT")));
+
+			/* coerce type if needed */
+			retval = exec_simple_cast_value(estate,
+											retval,
+											rettype,
+											tupdesc->attrs[0]->atttypid,
+											tupdesc->attrs[0]->atttypmod,
+											isNull);
 
-		tuplestore_putvalues(estate->tuple_store, tupdesc,
-							 &retval, &isNull);
+			tuplestore_putvalues(estate->tuple_store, tupdesc,
+								&retval, &isNull);
+		}
 	}
 	else
 	{
@@ -3901,29 +3921,16 @@ exec_assign_value(PLpgSQL_execstate *estate,
 				}
 				else
 				{
-					HeapTupleHeader td;
-					Oid			tupType;
-					int32		tupTypmod;
 					TupleDesc	tupdesc;
-					HeapTupleData tmptup;
+					HeapTuple	tuple;
 
 					/* Source must be of RECORD or composite type */
 					if (!type_is_rowtype(valtype))
 						ereport(ERROR,
 								(errcode(ERRCODE_DATATYPE_MISMATCH),
 								 errmsg("cannot assign non-composite value to a row variable")));
-					/* 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(tupType, tupTypmod);
-					/* Build a temporary HeapTuple control structure */
-					tmptup.t_len = HeapTupleHeaderGetDatumLength(td);
-					ItemPointerSetInvalid(&(tmptup.t_self));
-					tmptup.t_tableOid = InvalidOid;
-					tmptup.t_data = td;
-					exec_move_row(estate, NULL, row, &tmptup, tupdesc);
+					tuple = get_tuple_from_datum(value, &tupdesc);
+					exec_move_row(estate, NULL, row, tuple, tupdesc);
 					ReleaseTupleDesc(tupdesc);
 				}
 				break;
@@ -3943,11 +3950,8 @@ exec_assign_value(PLpgSQL_execstate *estate,
 				}
 				else
 				{
-					HeapTupleHeader td;
-					Oid			tupType;
-					int32		tupTypmod;
 					TupleDesc	tupdesc;
-					HeapTupleData tmptup;
+					HeapTuple	tuple;
 
 					/* Source must be of RECORD or composite type */
 					if (!type_is_rowtype(valtype))
@@ -3955,18 +3959,8 @@ exec_assign_value(PLpgSQL_execstate *estate,
 								(errcode(ERRCODE_DATATYPE_MISMATCH),
 								 errmsg("cannot assign non-composite value to a record variable")));
 
-					/* 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(tupType, tupTypmod);
-					/* Build a temporary HeapTuple control structure */
-					tmptup.t_len = HeapTupleHeaderGetDatumLength(td);
-					ItemPointerSetInvalid(&(tmptup.t_self));
-					tmptup.t_tableOid = InvalidOid;
-					tmptup.t_data = td;
-					exec_move_row(estate, rec, NULL, &tmptup, tupdesc);
+					tuple = get_tuple_from_datum(value, &tupdesc);
+					exec_move_row(estate, rec, NULL, tuple, tupdesc);
 					ReleaseTupleDesc(tupdesc);
 				}
 				break;
@@ -6280,3 +6274,38 @@ exec_dynquery_with_params(PLpgSQL_execstate *estate,
 
 	return portal;
 }
+
+/* ----------
+ * get_tuple_from_datum		get a tuple from the rowtype datum
+ *
+ * If rettupdesc isn't NULL, it will receive a pointer to TupleDesc
+ * of rowtype.
+ *
+ * Note: its caller's responsibility to check 'value' is rowtype datum.
+ * ----------
+ */
+static HeapTuple
+get_tuple_from_datum(Datum value, TupleDesc *rettupdesc)
+{
+	HeapTupleHeader td;
+	Oid			tupType;
+	int32		tupTypmod;
+	HeapTupleData tmptup;
+
+	td = DatumGetHeapTupleHeader(value);
+	/* Extract rowtype info and find a tupdesc */
+	if (rettupdesc)
+	{
+		tupType = HeapTupleHeaderGetTypeId(td);
+		tupTypmod = HeapTupleHeaderGetTypMod(td);
+		*rettupdesc = lookup_rowtype_tupdesc(tupType, tupTypmod);
+	}
+
+	/* Build a temporary HeapTuple control structure */
+	tmptup.t_len = HeapTupleHeaderGetDatumLength(td);
+	ItemPointerSetInvalid(&(tmptup.t_self));
+	tmptup.t_tableOid = InvalidOid;
+	tmptup.t_data = td;
+
+	return heap_copytuple(&tmptup);
+}
diff --git a/src/pl/plpgsql/src/pl_gram.y b/src/pl/plpgsql/src/pl_gram.y
index cf164d0..c3b59e8 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() != ';')
@@ -2994,7 +2998,8 @@ make_return_next_stmt(int location)
 	}
 	else if (plpgsql_curr_compile->fn_retistuple)
 	{
-		switch (yylex())
+		int tok = yylex();
+		switch (tok)
 		{
 			case T_DATUM:
 				if (yylval.wdatum.datum->dtype == PLPGSQL_DTYPE_ROW ||
@@ -3008,10 +3013,13 @@ make_return_next_stmt(int location)
 				break;
 
 			default:
-				ereport(ERROR,
-						(errcode(ERRCODE_DATATYPE_MISMATCH),
-						 errmsg("RETURN NEXT 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..b008538
--- /dev/null
+++ b/src/test/regress/expected/return_rowtype.out
@@ -0,0 +1,135 @@
+--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: row expr in RETURN NEXT statement.
+create or replace function foo() returns setof footype as $$
+begin
+  for i in 1..10
+  loop
+    return next (1, 'hello');
+  end loop;
+  return;
+end;
+$$ language plpgsql;
+drop function foo();
+--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:  cannot return non-composite value from composite type returning function
+CONTEXT:  PL/pgSQL function foo() line 3 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..79a6553
--- /dev/null
+++ b/src/test/regress/sql/return_rowtype.sql
@@ -0,0 +1,128 @@
+--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: row expr in RETURN NEXT statement.
+create or replace function foo() returns setof footype as $$
+begin
+  for i in 1..10
+  loop
+    return next (1, 'hello');
+  end loop;
+  return;
+end;
+$$ language plpgsql;
+
+drop function foo();
+
+--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; $$;
+
+drop function foo();
+drop type footype;

diff --git a/doc/src/sgml/plpgsql.sgml b/doc/src/sgml/plpgsql.sgml
index 07fba57..58b1489 100644
--- a/doc/src/sgml/plpgsql.sgml
+++ b/doc/src/sgml/plpgsql.sgml
@@ -1571,11 +1571,9 @@ RETURN <replaceable>expression</replaceable>;
      </para>
 
      <para>
-      When returning a scalar type, any expression can be used. The
-      expression's result will be automatically cast into the
-      function's return type as described for assignments. To return a
-      composite (row) value, you must write a record or row variable
-      as the <replaceable>expression</replaceable>.
+      In <command>RETURN</command> statement you can use any expression.
+       The expression's result will be automatically cast into the
+      function's return type as described for assignments.
      </para>
 
      <para>
@@ -1600,6 +1598,21 @@ RETURN <replaceable>expression</replaceable>;
       however.  In those cases a <command>RETURN</command> statement is
       automatically executed if the top-level block finishes.
      </para>
+
+     <para>
+      Some examples:
+
+<programlisting>
+-- functions returning a scalar type
+RETURN 1 + 2;
+RETURN scalar_var;
+
+-- functions returning a composite type
+RETURN composite_type_var;
+RETURN (1, 2, 'three');
+</programlisting>
+
+     </para>
     </sect3>
 
     <sect3>
diff --git a/src/backend/access/common/tupconvert.c b/src/backend/access/common/tupconvert.c
index f813432..6a9cc64 100644
--- a/src/backend/access/common/tupconvert.c
+++ b/src/backend/access/common/tupconvert.c
@@ -22,7 +22,9 @@
 
 #include "access/htup_details.h"
 #include "access/tupconvert.h"
+#include "parser/parse_coerce.h"
 #include "utils/builtins.h"
+#include "utils/lsyscache.h"
 
 
 /*
@@ -69,6 +71,9 @@ convert_tuples_by_position(TupleDesc indesc,
 {
 	TupleConversionMap *map;
 	AttrNumber *attrMap;
+	Oid		   *typoutput;
+	Oid		   *typinput;
+	Oid		   *typioparam;
 	int			nincols;
 	int			noutcols;
 	int			n;
@@ -78,7 +83,10 @@ convert_tuples_by_position(TupleDesc indesc,
 
 	/* Verify compatibility and prepare attribute-number map */
 	n = outdesc->natts;
-	attrMap = (AttrNumber *) palloc0(n * sizeof(AttrNumber));
+	attrMap 	= (AttrNumber *) palloc0(n * sizeof(AttrNumber));
+	typinput 	= (Oid *) palloc0(n * sizeof(Oid));
+	typioparam	= (Oid *) palloc(n * sizeof(Oid));
+	typoutput 	= (Oid *) palloc((indesc->natts + 1) * sizeof(Oid)); /* +1 for NULL */
 	j = 0;						/* j is next physical input attribute */
 	nincols = noutcols = 0;		/* these count non-dropped attributes */
 	same = true;
@@ -100,8 +108,8 @@ convert_tuples_by_position(TupleDesc indesc,
 				continue;
 			nincols++;
 			/* Found matching column, check type */
-			if (atttypid != att->atttypid ||
-				(atttypmod != att->atttypmod && atttypmod >= 0))
+			if ((atttypmod != att->atttypmod && atttypmod >= 0) ||
+				!can_coerce_type(1, &atttypid, &(att->atttypid), COERCE_IMPLICIT_CAST))
 				ereport(ERROR,
 						(errcode(ERRCODE_DATATYPE_MISMATCH),
 						 errmsg_internal("%s", _(msg)),
@@ -111,6 +119,17 @@ convert_tuples_by_position(TupleDesc indesc,
 								   format_type_with_typemod(atttypid,
 															atttypmod),
 								   noutcols)));
+			/*
+			 * get the needed I/O functions to perform cast later in
+			 * do_convert_tuple, no need if type id's are same.
+			 */
+			if (atttypid != att->atttypid)
+			{
+				bool typIsVarlena;
+
+				getTypeOutputInfo(att->atttypid, &typoutput[j + 1], &typIsVarlena);
+				getTypeInputInfo(atttypid, &typinput[i], &typioparam[i]);
+			}
 			attrMap[i] = (AttrNumber) (j + 1);
 			j++;
 			break;
@@ -147,6 +166,13 @@ convert_tuples_by_position(TupleDesc indesc,
 	{
 		for (i = 0; i < n; i++)
 		{
+			/* If typinput is set, it also means we need to convert */
+			if (OidIsValid(typinput[i]))
+			{
+				same = false;
+				break;	
+			}
+
 			if (attrMap[i] == (i + 1))
 				continue;
 
@@ -183,9 +209,12 @@ convert_tuples_by_position(TupleDesc indesc,
 	/* preallocate workspace for Datum arrays */
 	map->outvalues = (Datum *) palloc(n * sizeof(Datum));
 	map->outisnull = (bool *) palloc(n * sizeof(bool));
+	map->typinput = typinput;
+	map->typioparam = typioparam;
 	n = indesc->natts + 1;		/* +1 for NULL */
 	map->invalues = (Datum *) palloc(n * sizeof(Datum));
 	map->inisnull = (bool *) palloc(n * sizeof(bool));
+	map->typoutput = typoutput;
 	map->invalues[0] = (Datum) 0;		/* set up the NULL entry */
 	map->inisnull[0] = true;
 
@@ -206,6 +235,9 @@ convert_tuples_by_name(TupleDesc indesc,
 {
 	TupleConversionMap *map;
 	AttrNumber *attrMap;
+	Oid		   *typoutput;
+	Oid		   *typinput;
+	Oid		   *typioparam;
 	int			n;
 	int			i;
 	bool		same;
@@ -213,6 +245,9 @@ convert_tuples_by_name(TupleDesc indesc,
 	/* Verify compatibility and prepare attribute-number map */
 	n = outdesc->natts;
 	attrMap = (AttrNumber *) palloc0(n * sizeof(AttrNumber));
+	typinput 	= (Oid *) palloc0(n * sizeof(Oid));
+	typioparam	= (Oid *) palloc(n * sizeof(Oid));
+	typoutput 	= (Oid *) palloc((indesc->natts + 1) * sizeof(Oid)); /* +1 for NULL */
 	for (i = 0; i < n; i++)
 	{
 		Form_pg_attribute att = outdesc->attrs[i];
@@ -234,7 +269,8 @@ convert_tuples_by_name(TupleDesc indesc,
 			if (strcmp(attname, NameStr(att->attname)) == 0)
 			{
 				/* Found it, check type */
-				if (atttypid != att->atttypid || atttypmod != att->atttypmod)
+				if (atttypmod != att->atttypmod ||
+					!can_coerce_type(1, &atttypid, &(att->atttypid), COERCE_IMPLICIT_CAST))
 					ereport(ERROR,
 							(errcode(ERRCODE_DATATYPE_MISMATCH),
 							 errmsg_internal("%s", _(msg)),
@@ -242,6 +278,17 @@ convert_tuples_by_name(TupleDesc indesc,
 									   attname,
 									   format_type_be(outdesc->tdtypeid),
 									   format_type_be(indesc->tdtypeid))));
+				/*
+				 * get the needed I/O functions to perform cast later in
+				 * do_convert_tuple, no need if type id's are same.
+				 */
+				if (atttypid != att->atttypid)
+				{
+					bool typIsVarlena;
+
+					getTypeOutputInfo(att->atttypid, &typoutput[j + 1], &typIsVarlena);
+					getTypeInputInfo(atttypid, &typinput[i], &typioparam[i]);
+				}
 				attrMap[i] = (AttrNumber) (j + 1);
 				break;
 			}
@@ -267,6 +314,13 @@ convert_tuples_by_name(TupleDesc indesc,
 		same = true;
 		for (i = 0; i < n; i++)
 		{
+			/* If typinput is set, it also means we need to convert */
+			if (OidIsValid(typinput[i]))
+			{
+				same = false;
+				break;	
+			}
+
 			if (attrMap[i] == (i + 1))
 				continue;
 
@@ -303,9 +357,12 @@ convert_tuples_by_name(TupleDesc indesc,
 	/* preallocate workspace for Datum arrays */
 	map->outvalues = (Datum *) palloc(n * sizeof(Datum));
 	map->outisnull = (bool *) palloc(n * sizeof(bool));
+	map->typinput = typinput;
+	map->typioparam = typioparam;
 	n = indesc->natts + 1;		/* +1 for NULL */
 	map->invalues = (Datum *) palloc(n * sizeof(Datum));
 	map->inisnull = (bool *) palloc(n * sizeof(bool));
+	map->typoutput = typoutput;
 	map->invalues[0] = (Datum) 0;		/* set up the NULL entry */
 	map->inisnull[0] = true;
 
@@ -342,6 +399,21 @@ do_convert_tuple(HeapTuple tuple, TupleConversionMap *map)
 
 		outvalues[i] = invalues[j];
 		outisnull[i] = inisnull[j];
+
+		/* Perform the casting, if necessary. */
+		if (!inisnull[j] &&
+			OidIsValid(map->typinput[i]))
+		{
+			Form_pg_attribute att = map->outdesc->attrs[i];
+			char *strval;
+
+			strval = OidOutputFunctionCall(map->typoutput[j], invalues[j]);
+			outvalues[i] = OidInputFunctionCall(map->typinput[i],
+											 strval,
+											 map->typioparam[i],
+											 att->atttypmod);
+			pfree(strval);
+		}
 	}
 
 	/*
@@ -360,7 +432,10 @@ free_conversion_map(TupleConversionMap *map)
 	pfree(map->attrMap);
 	pfree(map->invalues);
 	pfree(map->inisnull);
+	pfree(map->typoutput);
 	pfree(map->outvalues);
 	pfree(map->outisnull);
+	pfree(map->typinput);
+	pfree(map->typioparam);
 	pfree(map);
 }
diff --git a/src/include/access/tupconvert.h b/src/include/access/tupconvert.h
index 342dbb4..61ef7ae 100644
--- a/src/include/access/tupconvert.h
+++ b/src/include/access/tupconvert.h
@@ -25,8 +25,11 @@ typedef struct TupleConversionMap
 	AttrNumber *attrMap;		/* indexes of input fields, or 0 for null */
 	Datum	   *invalues;		/* workspace for deconstructing source */
 	bool	   *inisnull;
+	Oid		   *typoutput;		/* output functions for source type */
 	Datum	   *outvalues;		/* workspace for constructing result */
 	bool	   *outisnull;
+	Oid		   *typinput;		/* input functions for result type */
+	Oid		   *typioparam;
 } TupleConversionMap;
 
 
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 3b5b3bb..d1c0acd 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -214,6 +214,7 @@ static void free_params_data(PreparedParamsData *ppd);
 static Portal exec_dynquery_with_params(PLpgSQL_execstate *estate,
 						  PLpgSQL_expr *dynquery, List *params,
 						  const char *portalname, int cursorOptions);
+static HeapTuple get_tuple_from_datum(Datum value, TupleDesc *rettupdesc);
 
 
 /* ----------
@@ -275,23 +276,11 @@ plpgsql_exec_function(PLpgSQL_function *func, FunctionCallInfo fcinfo)
 
 					if (!fcinfo->argnull[i])
 					{
-						HeapTupleHeader td;
-						Oid			tupType;
-						int32		tupTypmod;
 						TupleDesc	tupdesc;
-						HeapTupleData tmptup;
-
-						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);
-						ItemPointerSetInvalid(&(tmptup.t_self));
-						tmptup.t_tableOid = InvalidOid;
-						tmptup.t_data = td;
-						exec_move_row(&estate, NULL, row, &tmptup, tupdesc);
+						HeapTuple	tuple;
+
+						tuple = get_tuple_from_datum(fcinfo->arg[i], &tupdesc);
+						exec_move_row(&estate, NULL, row, tuple, tupdesc);
 						ReleaseTupleDesc(tupdesc);
 					}
 					else
@@ -2449,24 +2438,30 @@ exec_stmt_return(PLpgSQL_execstate *estate, PLpgSQL_stmt_return *stmt)
 
 	if (stmt->expr != NULL)
 	{
+		estate->retval = exec_eval_expr(estate, stmt->expr,
+										&(estate->retisnull),
+										&(estate->rettype));
 		if (estate->retistuple)
 		{
-			exec_run_select(estate, stmt->expr, 1, NULL);
-			if (estate->eval_processed > 0)
+			/* Source must be of RECORD or composite type */
+			if (!type_is_rowtype(estate->rettype))
+				ereport(ERROR,
+						(errcode(ERRCODE_DATATYPE_MISMATCH),
+						errmsg("cannot return non-composite value from composite type returning function")));
+
+			if (!estate->retisnull)
 			{
-				estate->retval = PointerGetDatum(estate->eval_tuptable->vals[0]);
-				estate->rettupdesc = estate->eval_tuptable->tupdesc;
-				estate->retisnull = false;
+				HeapTuple	tuple;
+				TupleDesc	tupdesc;
+
+				tuple = get_tuple_from_datum(estate->retval, &tupdesc);
+				estate->retval = PointerGetDatum(tuple);
+				estate->rettupdesc = CreateTupleDescCopy(tupdesc);
+				ReleaseTupleDesc(tupdesc);
 			}
 		}
-		else
-		{
-			/* Normal case for scalar results */
-			estate->retval = exec_eval_expr(estate, stmt->expr,
-											&(estate->retisnull),
-											&(estate->rettype));
-		}
 
+		/* Else, the expr is of scalar type and has been evaluated. simply return. */
 		return PLPGSQL_RC_RETURN;
 	}
 
@@ -2593,26 +2588,51 @@ exec_stmt_return_next(PLpgSQL_execstate *estate,
 		bool		isNull;
 		Oid			rettype;
 
-		if (natts != 1)
-			ereport(ERROR,
-					(errcode(ERRCODE_DATATYPE_MISMATCH),
-					 errmsg("wrong result type supplied in RETURN NEXT")));
-
 		retval = exec_eval_expr(estate,
 								stmt->expr,
 								&isNull,
 								&rettype);
 
-		/* coerce type if needed */
-		retval = exec_simple_cast_value(estate,
-										retval,
-										rettype,
-										tupdesc->attrs[0]->atttypid,
-										tupdesc->attrs[0]->atttypmod,
-										isNull);
+		/* Check if expr is of RECORD or composite type */
+		if (type_is_rowtype(rettype))
+		{
+			TupleConversionMap *tupmap;
+			TupleDesc retdesc;
+
+			tuple = get_tuple_from_datum(retval, &retdesc);
+			tupmap = convert_tuples_by_position(retdesc,
+												estate->rettupdesc,
+												gettext_noop("returned record type does not match expected record type"));
+			/* it might need conversion */
+			if (tupmap)
+			{
+				tuple = do_convert_tuple(tuple, tupmap);
+				free_conversion_map(tupmap);
+				free_tuple = true;
+			}
+
+			ReleaseTupleDesc(retdesc);
+		}
+		else
+		{
+			/* Normal case for scalar results */
+
+			if (natts != 1)
+				ereport(ERROR,
+						(errcode(ERRCODE_DATATYPE_MISMATCH),
+						errmsg("wrong result type supplied in RETURN NEXT")));
+
+			/* coerce type if needed */
+			retval = exec_simple_cast_value(estate,
+											retval,
+											rettype,
+											tupdesc->attrs[0]->atttypid,
+											tupdesc->attrs[0]->atttypmod,
+											isNull);
 
-		tuplestore_putvalues(estate->tuple_store, tupdesc,
-							 &retval, &isNull);
+			tuplestore_putvalues(estate->tuple_store, tupdesc,
+								&retval, &isNull);
+		}
 	}
 	else
 	{
@@ -3901,29 +3921,16 @@ exec_assign_value(PLpgSQL_execstate *estate,
 				}
 				else
 				{
-					HeapTupleHeader td;
-					Oid			tupType;
-					int32		tupTypmod;
 					TupleDesc	tupdesc;
-					HeapTupleData tmptup;
+					HeapTuple	tuple;
 
 					/* Source must be of RECORD or composite type */
 					if (!type_is_rowtype(valtype))
 						ereport(ERROR,
 								(errcode(ERRCODE_DATATYPE_MISMATCH),
 								 errmsg("cannot assign non-composite value to a row variable")));
-					/* 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(tupType, tupTypmod);
-					/* Build a temporary HeapTuple control structure */
-					tmptup.t_len = HeapTupleHeaderGetDatumLength(td);
-					ItemPointerSetInvalid(&(tmptup.t_self));
-					tmptup.t_tableOid = InvalidOid;
-					tmptup.t_data = td;
-					exec_move_row(estate, NULL, row, &tmptup, tupdesc);
+					tuple = get_tuple_from_datum(value, &tupdesc);
+					exec_move_row(estate, NULL, row, tuple, tupdesc);
 					ReleaseTupleDesc(tupdesc);
 				}
 				break;
@@ -3943,11 +3950,8 @@ exec_assign_value(PLpgSQL_execstate *estate,
 				}
 				else
 				{
-					HeapTupleHeader td;
-					Oid			tupType;
-					int32		tupTypmod;
 					TupleDesc	tupdesc;
-					HeapTupleData tmptup;
+					HeapTuple	tuple;
 
 					/* Source must be of RECORD or composite type */
 					if (!type_is_rowtype(valtype))
@@ -3955,18 +3959,8 @@ exec_assign_value(PLpgSQL_execstate *estate,
 								(errcode(ERRCODE_DATATYPE_MISMATCH),
 								 errmsg("cannot assign non-composite value to a record variable")));
 
-					/* 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(tupType, tupTypmod);
-					/* Build a temporary HeapTuple control structure */
-					tmptup.t_len = HeapTupleHeaderGetDatumLength(td);
-					ItemPointerSetInvalid(&(tmptup.t_self));
-					tmptup.t_tableOid = InvalidOid;
-					tmptup.t_data = td;
-					exec_move_row(estate, rec, NULL, &tmptup, tupdesc);
+					tuple = get_tuple_from_datum(value, &tupdesc);
+					exec_move_row(estate, rec, NULL, tuple, tupdesc);
 					ReleaseTupleDesc(tupdesc);
 				}
 				break;
@@ -6280,3 +6274,38 @@ exec_dynquery_with_params(PLpgSQL_execstate *estate,
 
 	return portal;
 }
+
+/* ----------
+ * get_tuple_from_datum		get a tuple from the rowtype datum
+ *
+ * If rettupdesc isn't NULL, it will receive a pointer to TupleDesc
+ * of rowtype.
+ *
+ * Note: its caller's responsibility to check 'value' is rowtype datum.
+ * ----------
+ */
+static HeapTuple
+get_tuple_from_datum(Datum value, TupleDesc *rettupdesc)
+{
+	HeapTupleHeader td;
+	Oid			tupType;
+	int32		tupTypmod;
+	HeapTupleData tmptup;
+
+	td = DatumGetHeapTupleHeader(value);
+	/* Extract rowtype info and find a tupdesc */
+	if (rettupdesc)
+	{
+		tupType = HeapTupleHeaderGetTypeId(td);
+		tupTypmod = HeapTupleHeaderGetTypMod(td);
+		*rettupdesc = lookup_rowtype_tupdesc(tupType, tupTypmod);
+	}
+
+	/* Build a temporary HeapTuple control structure */
+	tmptup.t_len = HeapTupleHeaderGetDatumLength(td);
+	ItemPointerSetInvalid(&(tmptup.t_self));
+	tmptup.t_tableOid = InvalidOid;
+	tmptup.t_data = td;
+
+	return heap_copytuple(&tmptup);
+}
diff --git a/src/pl/plpgsql/src/pl_gram.y b/src/pl/plpgsql/src/pl_gram.y
index cf164d0..c3b59e8 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() != ';')
@@ -2994,7 +2998,8 @@ make_return_next_stmt(int location)
 	}
 	else if (plpgsql_curr_compile->fn_retistuple)
 	{
-		switch (yylex())
+		int tok = yylex();
+		switch (tok)
 		{
 			case T_DATUM:
 				if (yylval.wdatum.datum->dtype == PLPGSQL_DTYPE_ROW ||
@@ -3008,10 +3013,13 @@ make_return_next_stmt(int location)
 				break;
 
 			default:
-				ereport(ERROR,
-						(errcode(ERRCODE_DATATYPE_MISMATCH),
-						 errmsg("RETURN NEXT 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..effd6ac
--- /dev/null
+++ b/src/test/regress/expected/return_rowtype.out
@@ -0,0 +1,167 @@
+--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: row expr in RETURN NEXT statement.
+create or replace function foo() returns setof footype as $$
+begin
+  for i in 1..10
+  loop
+    return next (1, 'hello');
+  end loop;
+  return;
+end;
+$$ language plpgsql;
+drop function foo();
+--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:  cannot return non-composite value from composite type returning function
+CONTEXT:  PL/pgSQL function foo() line 3 at RETURN
+PL/pgSQL function inline_code_block line 5 at assignment
+drop function foo();
+drop type footype;
+--test: return record of integers but receive as numeric
+create or replace function foo(i integer)
+returns setof record as $$
+declare r record;
+begin
+  r := (10,20);
+  for i in 1..10 loop
+    return next r;
+  end loop;
+  return;
+end;
+$$ language plpgsql;
+select sum(a) from generate_series(1,3000) g(i), lateral foo(i) as (a numeric, b numeric);
+  sum   
+--------
+ 300000
+(1 row)
+
+create or replace function fo(i integer)
+returns record as $$
+declare r record;
+begin
+  r := (10,20); return r;
+end;
+$$ language plpgsql;
+select sum(a) from generate_series(1,3000) g(i), lateral fo(i) as (a int, b numeric);
+  sum  
+-------
+ 30000
+(1 row)
+
+drop function foo(integer);
diff --git a/src/test/regress/sql/return_rowtype.sql b/src/test/regress/sql/return_rowtype.sql
new file mode 100644
index 0000000..7776262
--- /dev/null
+++ b/src/test/regress/sql/return_rowtype.sql
@@ -0,0 +1,155 @@
+--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: row expr in RETURN NEXT statement.
+create or replace function foo() returns setof footype as $$
+begin
+  for i in 1..10
+  loop
+    return next (1, 'hello');
+  end loop;
+  return;
+end;
+$$ language plpgsql;
+
+drop function foo();
+
+--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; $$;
+
+drop function foo();
+drop type footype;
+
+--test: return record of integers but receive as numeric
+create or replace function foo(i integer)
+returns setof record as $$
+declare r record;
+begin
+  r := (10,20);
+  for i in 1..10 loop
+    return next r;
+  end loop;
+  return;
+end;
+$$ language plpgsql;
+
+select sum(a) from generate_series(1,3000) g(i), lateral foo(i) as (a numeric, b numeric);
+
+create or replace function fo(i integer)
+returns record as $$
+declare r record;
+begin
+  r := (10,20); return r;
+end;
+$$ language plpgsql;
+
+select sum(a) from generate_series(1,3000) g(i), lateral fo(i) as (a int, b numeric);
+
+drop function foo(integer);

diff --git a/doc/src/sgml/plpgsql.sgml b/doc/src/sgml/plpgsql.sgml
index 07fba57..338f7c2 100644
--- a/doc/src/sgml/plpgsql.sgml
+++ b/doc/src/sgml/plpgsql.sgml
@@ -1571,11 +1571,10 @@ RETURN <replaceable>expression</replaceable>;
      </para>
 
      <para>
-      When returning a scalar type, any expression can be used. The
-      expression's result will be automatically cast into the
-      function's return type as described for assignments. To return a
-      composite (row) value, you must write a record or row variable
-      as the <replaceable>expression</replaceable>.
+      In <command>RETURN</command> statement you can use any expression.
+      The expression's result will be automatically cast into the
+      function's return type as described for assignments. However if
+      row expression is used, you may have to do explicit casting.
      </para>
 
      <para>
@@ -1600,6 +1599,21 @@ RETURN <replaceable>expression</replaceable>;
       however.  In those cases a <command>RETURN</command> statement is
       automatically executed if the top-level block finishes.
      </para>
+
+     <para>
+      Some examples:
+
+<programlisting>
+-- functions returning a scalar type
+RETURN 1 + 2;
+RETURN scalar_var;
+
+-- functions returning a composite type
+RETURN composite_type_var;
+RETURN (1, 2, 'three');
+</programlisting>
+
+     </para>
     </sect3>
 
     <sect3>
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 3b5b3bb..d1c0acd 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -214,6 +214,7 @@ static void free_params_data(PreparedParamsData *ppd);
 static Portal exec_dynquery_with_params(PLpgSQL_execstate *estate,
 						  PLpgSQL_expr *dynquery, List *params,
 						  const char *portalname, int cursorOptions);
+static HeapTuple get_tuple_from_datum(Datum value, TupleDesc *rettupdesc);
 
 
 /* ----------
@@ -275,23 +276,11 @@ plpgsql_exec_function(PLpgSQL_function *func, FunctionCallInfo fcinfo)
 
 					if (!fcinfo->argnull[i])
 					{
-						HeapTupleHeader td;
-						Oid			tupType;
-						int32		tupTypmod;
 						TupleDesc	tupdesc;
-						HeapTupleData tmptup;
-
-						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);
-						ItemPointerSetInvalid(&(tmptup.t_self));
-						tmptup.t_tableOid = InvalidOid;
-						tmptup.t_data = td;
-						exec_move_row(&estate, NULL, row, &tmptup, tupdesc);
+						HeapTuple	tuple;
+
+						tuple = get_tuple_from_datum(fcinfo->arg[i], &tupdesc);
+						exec_move_row(&estate, NULL, row, tuple, tupdesc);
 						ReleaseTupleDesc(tupdesc);
 					}
 					else
@@ -2449,24 +2438,30 @@ exec_stmt_return(PLpgSQL_execstate *estate, PLpgSQL_stmt_return *stmt)
 
 	if (stmt->expr != NULL)
 	{
+		estate->retval = exec_eval_expr(estate, stmt->expr,
+										&(estate->retisnull),
+										&(estate->rettype));
 		if (estate->retistuple)
 		{
-			exec_run_select(estate, stmt->expr, 1, NULL);
-			if (estate->eval_processed > 0)
+			/* Source must be of RECORD or composite type */
+			if (!type_is_rowtype(estate->rettype))
+				ereport(ERROR,
+						(errcode(ERRCODE_DATATYPE_MISMATCH),
+						errmsg("cannot return non-composite value from composite type returning function")));
+
+			if (!estate->retisnull)
 			{
-				estate->retval = PointerGetDatum(estate->eval_tuptable->vals[0]);
-				estate->rettupdesc = estate->eval_tuptable->tupdesc;
-				estate->retisnull = false;
+				HeapTuple	tuple;
+				TupleDesc	tupdesc;
+
+				tuple = get_tuple_from_datum(estate->retval, &tupdesc);
+				estate->retval = PointerGetDatum(tuple);
+				estate->rettupdesc = CreateTupleDescCopy(tupdesc);
+				ReleaseTupleDesc(tupdesc);
 			}
 		}
-		else
-		{
-			/* Normal case for scalar results */
-			estate->retval = exec_eval_expr(estate, stmt->expr,
-											&(estate->retisnull),
-											&(estate->rettype));
-		}
 
+		/* Else, the expr is of scalar type and has been evaluated. simply return. */
 		return PLPGSQL_RC_RETURN;
 	}
 
@@ -2593,26 +2588,51 @@ exec_stmt_return_next(PLpgSQL_execstate *estate,
 		bool		isNull;
 		Oid			rettype;
 
-		if (natts != 1)
-			ereport(ERROR,
-					(errcode(ERRCODE_DATATYPE_MISMATCH),
-					 errmsg("wrong result type supplied in RETURN NEXT")));
-
 		retval = exec_eval_expr(estate,
 								stmt->expr,
 								&isNull,
 								&rettype);
 
-		/* coerce type if needed */
-		retval = exec_simple_cast_value(estate,
-										retval,
-										rettype,
-										tupdesc->attrs[0]->atttypid,
-										tupdesc->attrs[0]->atttypmod,
-										isNull);
+		/* Check if expr is of RECORD or composite type */
+		if (type_is_rowtype(rettype))
+		{
+			TupleConversionMap *tupmap;
+			TupleDesc retdesc;
+
+			tuple = get_tuple_from_datum(retval, &retdesc);
+			tupmap = convert_tuples_by_position(retdesc,
+												estate->rettupdesc,
+												gettext_noop("returned record type does not match expected record type"));
+			/* it might need conversion */
+			if (tupmap)
+			{
+				tuple = do_convert_tuple(tuple, tupmap);
+				free_conversion_map(tupmap);
+				free_tuple = true;
+			}
+
+			ReleaseTupleDesc(retdesc);
+		}
+		else
+		{
+			/* Normal case for scalar results */
+
+			if (natts != 1)
+				ereport(ERROR,
+						(errcode(ERRCODE_DATATYPE_MISMATCH),
+						errmsg("wrong result type supplied in RETURN NEXT")));
+
+			/* coerce type if needed */
+			retval = exec_simple_cast_value(estate,
+											retval,
+											rettype,
+											tupdesc->attrs[0]->atttypid,
+											tupdesc->attrs[0]->atttypmod,
+											isNull);
 
-		tuplestore_putvalues(estate->tuple_store, tupdesc,
-							 &retval, &isNull);
+			tuplestore_putvalues(estate->tuple_store, tupdesc,
+								&retval, &isNull);
+		}
 	}
 	else
 	{
@@ -3901,29 +3921,16 @@ exec_assign_value(PLpgSQL_execstate *estate,
 				}
 				else
 				{
-					HeapTupleHeader td;
-					Oid			tupType;
-					int32		tupTypmod;
 					TupleDesc	tupdesc;
-					HeapTupleData tmptup;
+					HeapTuple	tuple;
 
 					/* Source must be of RECORD or composite type */
 					if (!type_is_rowtype(valtype))
 						ereport(ERROR,
 								(errcode(ERRCODE_DATATYPE_MISMATCH),
 								 errmsg("cannot assign non-composite value to a row variable")));
-					/* 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(tupType, tupTypmod);
-					/* Build a temporary HeapTuple control structure */
-					tmptup.t_len = HeapTupleHeaderGetDatumLength(td);
-					ItemPointerSetInvalid(&(tmptup.t_self));
-					tmptup.t_tableOid = InvalidOid;
-					tmptup.t_data = td;
-					exec_move_row(estate, NULL, row, &tmptup, tupdesc);
+					tuple = get_tuple_from_datum(value, &tupdesc);
+					exec_move_row(estate, NULL, row, tuple, tupdesc);
 					ReleaseTupleDesc(tupdesc);
 				}
 				break;
@@ -3943,11 +3950,8 @@ exec_assign_value(PLpgSQL_execstate *estate,
 				}
 				else
 				{
-					HeapTupleHeader td;
-					Oid			tupType;
-					int32		tupTypmod;
 					TupleDesc	tupdesc;
-					HeapTupleData tmptup;
+					HeapTuple	tuple;
 
 					/* Source must be of RECORD or composite type */
 					if (!type_is_rowtype(valtype))
@@ -3955,18 +3959,8 @@ exec_assign_value(PLpgSQL_execstate *estate,
 								(errcode(ERRCODE_DATATYPE_MISMATCH),
 								 errmsg("cannot assign non-composite value to a record variable")));
 
-					/* 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(tupType, tupTypmod);
-					/* Build a temporary HeapTuple control structure */
-					tmptup.t_len = HeapTupleHeaderGetDatumLength(td);
-					ItemPointerSetInvalid(&(tmptup.t_self));
-					tmptup.t_tableOid = InvalidOid;
-					tmptup.t_data = td;
-					exec_move_row(estate, rec, NULL, &tmptup, tupdesc);
+					tuple = get_tuple_from_datum(value, &tupdesc);
+					exec_move_row(estate, rec, NULL, tuple, tupdesc);
 					ReleaseTupleDesc(tupdesc);
 				}
 				break;
@@ -6280,3 +6274,38 @@ exec_dynquery_with_params(PLpgSQL_execstate *estate,
 
 	return portal;
 }
+
+/* ----------
+ * get_tuple_from_datum		get a tuple from the rowtype datum
+ *
+ * If rettupdesc isn't NULL, it will receive a pointer to TupleDesc
+ * of rowtype.
+ *
+ * Note: its caller's responsibility to check 'value' is rowtype datum.
+ * ----------
+ */
+static HeapTuple
+get_tuple_from_datum(Datum value, TupleDesc *rettupdesc)
+{
+	HeapTupleHeader td;
+	Oid			tupType;
+	int32		tupTypmod;
+	HeapTupleData tmptup;
+
+	td = DatumGetHeapTupleHeader(value);
+	/* Extract rowtype info and find a tupdesc */
+	if (rettupdesc)
+	{
+		tupType = HeapTupleHeaderGetTypeId(td);
+		tupTypmod = HeapTupleHeaderGetTypMod(td);
+		*rettupdesc = lookup_rowtype_tupdesc(tupType, tupTypmod);
+	}
+
+	/* Build a temporary HeapTuple control structure */
+	tmptup.t_len = HeapTupleHeaderGetDatumLength(td);
+	ItemPointerSetInvalid(&(tmptup.t_self));
+	tmptup.t_tableOid = InvalidOid;
+	tmptup.t_data = td;
+
+	return heap_copytuple(&tmptup);
+}
diff --git a/src/pl/plpgsql/src/pl_gram.y b/src/pl/plpgsql/src/pl_gram.y
index cf164d0..c3b59e8 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() != ';')
@@ -2994,7 +2998,8 @@ make_return_next_stmt(int location)
 	}
 	else if (plpgsql_curr_compile->fn_retistuple)
 	{
-		switch (yylex())
+		int tok = yylex();
+		switch (tok)
 		{
 			case T_DATUM:
 				if (yylval.wdatum.datum->dtype == PLPGSQL_DTYPE_ROW ||
@@ -3008,10 +3013,13 @@ make_return_next_stmt(int location)
 				break;
 
 			default:
-				ereport(ERROR,
-						(errcode(ERRCODE_DATATYPE_MISMATCH),
-						 errmsg("RETURN NEXT 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..85baaed
--- /dev/null
+++ b/src/test/regress/expected/return_rowtype.out
@@ -0,0 +1,135 @@
+--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.
+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: row expr in RETURN NEXT statement.
+create or replace function foo() returns setof footype as $$
+begin
+  for i in 1..10
+  loop
+    return next (1, 'hello');
+  end loop;
+  return;
+end;
+$$ language plpgsql;
+drop function foo();
+--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:  cannot return non-composite value from composite type returning function
+CONTEXT:  PL/pgSQL function foo() line 3 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..7289f30
--- /dev/null
+++ b/src/test/regress/sql/return_rowtype.sql
@@ -0,0 +1,128 @@
+--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.
+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: row expr in RETURN NEXT statement.
+create or replace function foo() returns setof footype as $$
+begin
+  for i in 1..10
+  loop
+    return next (1, 'hello');
+  end loop;
+  return;
+end;
+$$ language plpgsql;
+
+drop function foo();
+
+--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; $$;
+
+drop function foo();
+drop type footype;