review: plpgsql return a row-expression

diff --git a/src/backend/access/common/tupconvert.c b/src/backend/access/common/tupconvert.c
index f813432..2901f47 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"
 
 
 /*
@@ -100,8 +102,7 @@ convert_tuples_by_position(TupleDesc indesc,
 				continue;
 			nincols++;
 			/* Found matching column, check type */
-			if (atttypid != att->atttypid ||
-				(atttypmod != att->atttypmod && atttypmod >= 0))
+			if (!can_coerce_type(1, &atttypid, &(att->atttypid), COERCE_IMPLICIT_CAST))
 				ereport(ERROR,
 						(errcode(ERRCODE_DATATYPE_MISMATCH),
 						 errmsg_internal("%s", _(msg)),
@@ -234,7 +235,7 @@ 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 (!can_coerce_type(1, &atttypid, &(att->atttypid), COERCE_IMPLICIT_CAST))
 					ereport(ERROR,
 							(errcode(ERRCODE_DATATYPE_MISMATCH),
 							 errmsg_internal("%s", _(msg)),
@@ -339,9 +340,40 @@ do_convert_tuple(HeapTuple tuple, TupleConversionMap *map)
 	for (i = 0; i < outnatts; i++)
 	{
 		int			j = attrMap[i];
+		TupleDesc	indesc = map->indesc;
+		TupleDesc	outdesc = map->outdesc;
+		Oid			reqtype = outdesc->attrs[i]->atttypid;
+		int32		reqtypmod = outdesc->attrs[i]->atttypmod;
 
 		outvalues[i] = invalues[j];
 		outisnull[i] = inisnull[j];
+
+		/*
+		 * If types are not exactly matched, do the casting. We have already
+		 * performed the can_coerce_type() check. So It should be safe.
+		 */
+		if (!inisnull[j] &&
+			(reqtype != indesc->attrs[j - 1]->atttypid ||
+			 reqtypmod != indesc->attrs[j - 1]->atttypmod))
+		{
+			Oid			typinput;
+			Oid			typoutput;
+			Oid			typioparam;
+			Oid			valtype = indesc->attrs[j - 1]->atttypid;
+			bool		typIsVarlena;
+			FmgrInfo	finfo_input;
+			char		*strval;
+
+			getTypeOutputInfo(valtype, &typoutput, &typIsVarlena);
+			getTypeInputInfo(reqtype, &typinput, &typioparam);
+			fmgr_info(typinput, &finfo_input);
+
+			strval = OidOutputFunctionCall(typoutput, invalues[j]);
+			outvalues[i] = InputFunctionCall(&finfo_input,
+											 strval,
+											 typioparam,
+											 reqtypmod);
+		}
 	}
 
 	/*
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..2901f47 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"
 
 
 /*
@@ -100,8 +102,7 @@ convert_tuples_by_position(TupleDesc indesc,
 				continue;
 			nincols++;
 			/* Found matching column, check type */
-			if (atttypid != att->atttypid ||
-				(atttypmod != att->atttypmod && atttypmod >= 0))
+			if (!can_coerce_type(1, &atttypid, &(att->atttypid), COERCE_IMPLICIT_CAST))
 				ereport(ERROR,
 						(errcode(ERRCODE_DATATYPE_MISMATCH),
 						 errmsg_internal("%s", _(msg)),
@@ -234,7 +235,7 @@ 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 (!can_coerce_type(1, &atttypid, &(att->atttypid), COERCE_IMPLICIT_CAST))
 					ereport(ERROR,
 							(errcode(ERRCODE_DATATYPE_MISMATCH),
 							 errmsg_internal("%s", _(msg)),
@@ -339,9 +340,40 @@ do_convert_tuple(HeapTuple tuple, TupleConversionMap *map)
 	for (i = 0; i < outnatts; i++)
 	{
 		int			j = attrMap[i];
+		TupleDesc	indesc = map->indesc;
+		TupleDesc	outdesc = map->outdesc;
+		Oid			reqtype = outdesc->attrs[i]->atttypid;
+		int32		reqtypmod = outdesc->attrs[i]->atttypmod;
 
 		outvalues[i] = invalues[j];
 		outisnull[i] = inisnull[j];
+
+		/*
+		 * If types are not exactly matched, do the casting. We have already
+		 * performed the can_coerce_type() check. So It should be safe.
+		 */
+		if (!inisnull[j] &&
+			(reqtype != indesc->attrs[j - 1]->atttypid ||
+			 reqtypmod != indesc->attrs[j - 1]->atttypmod))
+		{
+			Oid			typinput;
+			Oid			typoutput;
+			Oid			typioparam;
+			Oid			valtype = indesc->attrs[j - 1]->atttypid;
+			bool		typIsVarlena;
+			FmgrInfo	finfo_input;
+			char		*strval;
+
+			getTypeOutputInfo(valtype, &typoutput, &typIsVarlena);
+			getTypeInputInfo(reqtype, &typinput, &typioparam);
+			fmgr_info(typinput, &finfo_input);
+
+			strval = OidOutputFunctionCall(typoutput, invalues[j]);
+			outvalues[i] = InputFunctionCall(&finfo_input,
+											 strval,
+											 typioparam,
+											 reqtypmod);
+		}
 	}
 
 	/*
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;