Check to ensure the number of primary key fields supplied does not

exceed the total number of non-dropped source table fields for
dblink_build_sql_*(). Addresses bug report from Rushabh Lathia.

Backpatch all the way to the 7.3 branch.

1 files changed, 55 insertions, 0 deletions

diff --git a/contrib/dblink/dblink.c b/contrib/dblink/dblink.c
index 0a2d70e6b69..18271b24f76 100644
--- a/contrib/dblink/dblink.c
+++ b/contrib/dblink/dblink.c
@@ -88,6 +88,7 @@ static Oid	get_relid_from_relname(text *relname_text);
 static char *generate_relation_name(Oid relid);
 static char *connstr_strip_password(const char *connstr);
 static void dblink_security_check(PGconn *conn, remoteConn *rcon, const char *connstr);
+static int get_nondropped_natts(Oid relid);
 
 /* Global */
 List	   *res_id = NIL;
@@ -985,6 +986,7 @@ dblink_build_sql_insert(PG_FUNCTION_ARGS)
 	int16		typlen;
 	bool		typbyval;
 	char		typalign;
+	int			nondropped_natts;
 
 	relname_text = PG_GETARG_TEXT_P(0);
 
@@ -1016,6 +1018,15 @@ dblink_build_sql_insert(PG_FUNCTION_ARGS)
 				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 				 errmsg("number of key attributes must be > 0")));
 
+	/*
+	 * ensure we don't ask for more pk attributes than we have
+	 * non-dropped columns
+	 */
+	nondropped_natts = get_nondropped_natts(relid);
+	if (pknumatts > nondropped_natts)
+		ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR),
+				errmsg("number of primary key fields exceeds number of specified relation attributes")));
+
 	src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
 	tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
 
@@ -1132,6 +1143,7 @@ dblink_build_sql_delete(PG_FUNCTION_ARGS)
 	int16		typlen;
 	bool		typbyval;
 	char		typalign;
+	int			nondropped_natts;
 
 	relname_text = PG_GETARG_TEXT_P(0);
 
@@ -1163,6 +1175,15 @@ dblink_build_sql_delete(PG_FUNCTION_ARGS)
 				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 				 errmsg("number of key attributes must be > 0")));
 
+	/*
+	 * ensure we don't ask for more pk attributes than we have
+	 * non-dropped columns
+	 */
+	nondropped_natts = get_nondropped_natts(relid);
+	if (pknumatts > nondropped_natts)
+		ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR),
+				errmsg("number of primary key fields exceeds number of specified relation attributes")));
+
 	tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
 
 	/*
@@ -1254,6 +1275,7 @@ dblink_build_sql_update(PG_FUNCTION_ARGS)
 	int16		typlen;
 	bool		typbyval;
 	char		typalign;
+	int			nondropped_natts;
 
 	relname_text = PG_GETARG_TEXT_P(0);
 
@@ -1285,6 +1307,15 @@ dblink_build_sql_update(PG_FUNCTION_ARGS)
 				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 				 errmsg("number of key attributes must be > 0")));
 
+	/*
+	 * ensure we don't ask for more pk attributes than we have
+	 * non-dropped columns
+	 */
+	nondropped_natts = get_nondropped_natts(relid);
+	if (pknumatts > nondropped_natts)
+		ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR),
+				errmsg("number of primary key fields exceeds number of specified relation attributes")));
+
 	src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
 	tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
 
@@ -2134,3 +2165,27 @@ dblink_security_check(PGconn *conn, remoteConn *rcon, const char *connstr)
 			PQfinish(conn);
 	}
 }
+
+static int
+get_nondropped_natts(Oid relid)
+{
+	int			nondropped_natts = 0;
+	TupleDesc	tupdesc;
+	Relation	rel;
+	int			natts;
+	int			i;
+
+	rel = relation_open(relid, AccessShareLock);
+	tupdesc = rel->rd_att;
+	natts = tupdesc->natts;
+
+	for (i = 0; i < natts; i++)
+	{
+		if (tupdesc->attrs[i]->attisdropped)
+			continue;
+		nondropped_natts++;
+	}
+
+	relation_close(rel, AccessShareLock);
+	return nondropped_natts;
+}