Make the world safe (more or less) for dropped columns in plpgsql rowtypes.

Author: tglsfdc

src/backend/executor/execQual.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/executor/execQual.c,v 1.145 2003/09/25 06:57:59 petere Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/executor/execQual.c,v 1.146 2003/09/25 23:02:11 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -315,23 +315,6 @@ ExecEvalAggref(AggrefExprState *aggref, ExprContext *econtext, bool *isNull)
  *
  *		Returns a Datum whose value is the value of a range
  *		variable with respect to given expression context.
- *
- *
- *		As an entry condition, we expect that the datatype the
- *		plan expects to get (as told by our "variable" argument) is in
- *		fact the datatype of the attribute the plan says to fetch (as
- *		seen in the current context, identified by our "econtext"
- *		argument).
- *
- *		If we fetch a Type A attribute and Caller treats it as if it
- *		were Type B, there will be undefined results (e.g. crash).
- *		One way these might mismatch now is that we're accessing a
- *		catalog class and the type information in the pg_attribute
- *		class does not match the hardcoded pg_attribute information
- *		(in pg_attribute.h) for the class in question.
- *
- *		We have an Assert to make sure this entry condition is met.
- *
  * ---------------------------------------------------------------- */
 static Datum
 ExecEvalVar(Var *variable, ExprContext *econtext, bool *isNull)
@@ -369,11 +352,40 @@ ExecEvalVar(Var *variable, ExprContext *econtext, bool *isNull)
 
 	attnum = variable->varattno;
 
-	/* (See prolog for explanation of this Assert) */
-	Assert(attnum <= 0 ||
-		   (attnum - 1 <= tuple_type->natts - 1 &&
-			tuple_type->attrs[attnum - 1] != NULL &&
-		  variable->vartype == tuple_type->attrs[attnum - 1]->atttypid));
+	/*
+	 * Some checks that are only applied for user attribute numbers
+	 * (bogus system attnums will be caught inside heap_getattr).
+	 */
+	if (attnum > 0)
+	{
+		/*
+		 * This assert checks that the attnum is valid.
+		 */
+		Assert(attnum <= tuple_type->natts &&
+			   tuple_type->attrs[attnum - 1] != NULL);
+
+		/*
+		 * If the attribute's column has been dropped, we force a NULL result.
+		 * This case should not happen in normal use, but it could happen if
+		 * we are executing a plan cached before the column was dropped.
+		 */
+		if (tuple_type->attrs[attnum - 1]->attisdropped)
+		{
+			*isNull = true;
+			return (Datum) 0;
+		}
+
+		/*
+		 * This assert checks that the datatype the plan expects to get (as
+		 * told by our "variable" argument) is in fact the datatype of the
+		 * attribute being fetched (as seen in the current context, identified
+		 * by our "econtext" argument).  Otherwise crashes are likely.
+		 *
+		 * Note that we can't check dropped columns, since their atttypid
+		 * has been zeroed.
+		 */
+		Assert(variable->vartype == tuple_type->attrs[attnum - 1]->atttypid);
+	}
 
 	/*
 	 * If the attribute number is invalid, then we are supposed to return

src/backend/executor/nodeFunctionscan.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/executor/nodeFunctionscan.c,v 1.21 2003/09/25 06:57:59 petere Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/executor/nodeFunctionscan.c,v 1.22 2003/09/25 23:02:12 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -35,7 +35,7 @@
 
 
 static TupleTableSlot *FunctionNext(FunctionScanState *node);
-static bool tupledesc_mismatch(TupleDesc tupdesc1, TupleDesc tupdesc2);
+static bool tupledesc_match(TupleDesc dst_tupdesc, TupleDesc src_tupdesc);
 
 /* ----------------------------------------------------------------
  *						Scan Support
@@ -86,8 +86,7 @@ FunctionNext(FunctionScanState *node)
 		 * need to do this for functions returning RECORD, but might as
 		 * well do it always.
 		 */
-		if (funcTupdesc &&
-			tupledesc_mismatch(node->tupdesc, funcTupdesc))
+		if (funcTupdesc && !tupledesc_match(node->tupdesc, funcTupdesc))
 			ereport(ERROR,
 					(errcode(ERRCODE_DATATYPE_MISMATCH),
 					 errmsg("query-specified return row and actual function return row do not match")));
@@ -364,26 +363,36 @@ ExecFunctionReScan(FunctionScanState *node, ExprContext *exprCtxt)
 		tuplestore_rescan(node->tuplestorestate);
 }
 
-
+/*
+ * Check that function result tuple type (src_tupdesc) matches or can be
+ * considered to match what the query expects (dst_tupdesc).
+ *
+ * We really only care about number of attributes and data type.
+ * Also, we can ignore type mismatch on columns that are dropped in the
+ * destination type, so long as the physical storage matches.  This is
+ * helpful in some cases involving out-of-date cached plans.
+ */
 static bool
-tupledesc_mismatch(TupleDesc tupdesc1, TupleDesc tupdesc2)
+tupledesc_match(TupleDesc dst_tupdesc, TupleDesc src_tupdesc)
 {
 	int			i;
 
-	if (tupdesc1->natts != tupdesc2->natts)
-		return true;
+	if (dst_tupdesc->natts != src_tupdesc->natts)
+		return false;
 
-	for (i = 0; i < tupdesc1->natts; i++)
+	for (i = 0; i < dst_tupdesc->natts; i++)
 	{
-		Form_pg_attribute attr1 = tupdesc1->attrs[i];
-		Form_pg_attribute attr2 = tupdesc2->attrs[i];
-
-		/*
-		 * We really only care about number of attributes and data type
-		 */
-		if (attr1->atttypid != attr2->atttypid)
-			return true;
+		Form_pg_attribute dattr = dst_tupdesc->attrs[i];
+		Form_pg_attribute sattr = src_tupdesc->attrs[i];
+
+		if (dattr->atttypid == sattr->atttypid)
+			continue;			/* no worries */
+		if (!dattr->attisdropped)
+			return false;
+		if (dattr->attlen != sattr->attlen ||
+			dattr->attalign != sattr->attalign)
+			return false;
 	}
 
-	return false;
+	return true;
 }

src/pl/plpgsql/src/gram.y

@@ -4,7 +4,7 @@
  *						  procedural language
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/pl/plpgsql/src/gram.y,v 1.46 2003/07/27 21:49:54 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/pl/plpgsql/src/gram.y,v 1.47 2003/09/25 23:02:12 tgl Exp $
  *
  *	  This software is copyrighted by Jan Wieck - Hamburg.
  *
@@ -47,7 +47,6 @@ static	PLpgSQL_expr	*read_sql_stmt(const char *sqlstart);
 static	PLpgSQL_type	*read_datatype(int tok);
 static	PLpgSQL_stmt	*make_select_stmt(void);
 static	PLpgSQL_stmt	*make_fetch_stmt(void);
-static	PLpgSQL_expr	*make_tupret_expr(PLpgSQL_row *row);
 static	void check_assignable(PLpgSQL_datum *datum);
 
 %}
@@ -493,7 +492,7 @@ decl_cursor_arglist : decl_cursor_arg
 						new->dtype = PLPGSQL_DTYPE_ROW;
 						new->refname = strdup("*internal*");
 						new->lineno = plpgsql_scanner_lineno();
-						new->rowtypeclass = InvalidOid;
+						new->rowtupdesc = NULL;
 						/*
 						 * We make temporary fieldnames/varnos arrays that
 						 * are much bigger than necessary.  We will resize
@@ -1176,24 +1175,24 @@ stmt_return		: K_RETURN lno
 
 						new = malloc(sizeof(PLpgSQL_stmt_return));
 						memset(new, 0, sizeof(PLpgSQL_stmt_return));
+						new->expr = NULL;
+						new->retrecno	= -1;
+						new->retrowno	= -1;
 
 						if (plpgsql_curr_compile->fn_retistuple &&
 							!plpgsql_curr_compile->fn_retset)
 						{
-							new->retrecno	= -1;
 							switch (yylex())
 							{
 								case K_NULL:
-									new->expr = NULL;
 									break;
 
 								case T_ROW:
-									new->expr = make_tupret_expr(yylval.row);
+									new->retrowno = yylval.row->rowno;
 									break;
 
 								case T_RECORD:
 									new->retrecno = yylval.rec->recno;
-									new->expr = NULL;
 									break;
 
 								default:
@@ -1874,7 +1873,7 @@ make_select_stmt(void)
 					row->dtype = PLPGSQL_DTYPE_ROW;
 					row->refname = strdup("*internal*");
 					row->lineno = plpgsql_scanner_lineno();
-					row->rowtypeclass = InvalidOid;
+					row->rowtupdesc = NULL;
 					row->nfields = nfields;
 					row->fieldnames = malloc(sizeof(char *) * nfields);
 					row->varnos = malloc(sizeof(int) * nfields);
@@ -2007,7 +2006,7 @@ make_fetch_stmt(void)
 				row->dtype = PLPGSQL_DTYPE_ROW;
 				row->refname = strdup("*internal*");
 				row->lineno = plpgsql_scanner_lineno();
-				row->rowtypeclass = InvalidOid;
+				row->rowtupdesc = NULL;
 				row->nfields = nfields;
 				row->fieldnames = malloc(sizeof(char *) * nfields);
 				row->varnos = malloc(sizeof(int) * nfields);
@@ -2041,36 +2040,6 @@ make_fetch_stmt(void)
 }
 
 
-static PLpgSQL_expr *
-make_tupret_expr(PLpgSQL_row *row)
-{
-	PLpgSQL_dstring		ds;
-	PLpgSQL_expr		*expr;
-	int					i;
-	char				buf[16];
-
-	expr = malloc(sizeof(PLpgSQL_expr) + sizeof(int) * (row->nfields - 1));
-	expr->dtype			= PLPGSQL_DTYPE_EXPR;
-
-	plpgsql_dstring_init(&ds);
-	plpgsql_dstring_append(&ds, "SELECT ");
-
-	for (i = 0; i < row->nfields; i++)
-	{
-		sprintf(buf, "%s$%d", (i > 0) ? "," : "", i + 1);
-		plpgsql_dstring_append(&ds, buf);
-		expr->params[i] = row->varnos[i];
-	}
-
-	expr->query			= strdup(plpgsql_dstring_get(&ds));
-	expr->plan			= NULL;
-	expr->plan_argtypes = NULL;
-	expr->nparams		= row->nfields;
-
-	plpgsql_dstring_free(&ds);
-	return expr;
-}
-
 static void
 check_assignable(PLpgSQL_datum *datum)
 {