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.

contrib/dblink/dblink.c

@@ -8,7 +8,7 @@
  * Darko Prenosil <Darko.Prenosil@finteh.hr>
  * Shridhar Daithankar <shridhar_daithankar@persistent.co.in>
  *
- * $PostgreSQL: pgsql/contrib/dblink/dblink.c,v 1.82.2.1 2009/09/12 23:21:13 joe Exp $
+ * $PostgreSQL: pgsql/contrib/dblink/dblink.c,v 1.82.2.2 2010/02/03 23:01:23 joe Exp $
  * Copyright (c) 2001-2009, PostgreSQL Global Development Group
  * ALL RIGHTS RESERVED;
  *
@@ -100,6 +100,7 @@ static void dblink_security_check(PGconn *conn, remoteConn *rconn);
 static void dblink_res_error(const char *conname, PGresult *res, const char *dblink_context_msg, bool fail);
 static char *get_connect_string(const char *servername);
 static char *escape_param_str(const char *from);
+static int get_nondropped_natts(Oid relid);
 
 /* Global */
 static remoteConn *pconn = NULL;
@@ -1367,6 +1368,7 @@ dblink_build_sql_insert(PG_FUNCTION_ARGS)
 	int			src_nitems;
 	int			tgt_nitems;
 	char	   *sql;
+	int			nondropped_natts;
 
 	/*
 	 * Convert relname to rel OID.
@@ -1394,6 +1396,15 @@ dblink_build_sql_insert(PG_FUNCTION_ARGS)
 				 errmsg("input for number of primary key " \
 						"attributes too large")));
 
+	/*
+	 * 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")));
+
 	/*
 	 * Source array is made up of key values that will be used to locate the
 	 * tuple of interest from the local system.
@@ -1459,6 +1470,7 @@ dblink_build_sql_delete(PG_FUNCTION_ARGS)
 	int2vector *pkattnums = (int2vector *) PG_GETARG_POINTER(1);
 	int32		pknumatts_tmp = PG_GETARG_INT32(2);
 	ArrayType  *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
+	int			nondropped_natts;
 	Oid			relid;
 	int16		pknumatts = 0;
 	char	  **tgt_pkattvals;
@@ -1491,6 +1503,15 @@ dblink_build_sql_delete(PG_FUNCTION_ARGS)
 				 errmsg("input for number of primary key " \
 						"attributes too large")));
 
+	/*
+	 * 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")));
+
 	/*
 	 * Target array is made up of key values that will be used to build the
 	 * SQL string for use on the remote system.
@@ -1546,6 +1567,7 @@ dblink_build_sql_update(PG_FUNCTION_ARGS)
 	int32		pknumatts_tmp = PG_GETARG_INT32(2);
 	ArrayType  *src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
 	ArrayType  *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
+	int			nondropped_natts;
 	Oid			relid;
 	int16		pknumatts = 0;
 	char	  **src_pkattvals;
@@ -1580,6 +1602,15 @@ dblink_build_sql_update(PG_FUNCTION_ARGS)
 				 errmsg("input for number of primary key " \
 						"attributes too large")));
 
+	/*
+	 * 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")));
+
 	/*
 	 * Source array is made up of key values that will be used to locate the
 	 * tuple of interest from the local system.
@@ -2467,3 +2498,28 @@ escape_param_str(const char *str)
 
 	return buf->data;
 }
+
+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;
+}
+

contrib/dblink/expected/dblink.out

@@ -39,6 +39,9 @@ SELECT dblink_build_sql_insert('foo','1 2',2,'{"0", "a"}','{"99", "xyz"}');
  INSERT INTO foo(f1,f2,f3) VALUES('99','xyz','{a0,b0,c0}')
 (1 row)
 
+-- too many pk fields, should fail
+SELECT dblink_build_sql_insert('foo','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}','{"99", "xyz", "{za0,zb0,zc0}"}');
+ERROR:  number of primary key fields exceeds number of specified relation attributes
 -- build an update statement based on a local tuple,
 -- replacing the primary key values with new ones
 SELECT dblink_build_sql_update('foo','1 2',2,'{"0", "a"}','{"99", "xyz"}');
@@ -47,6 +50,9 @@ SELECT dblink_build_sql_update('foo','1 2',2,'{"0", "a"}','{"99", "xyz"}');
  UPDATE foo SET f1 = '99', f2 = 'xyz', f3 = '{a0,b0,c0}' WHERE f1 = '99' AND f2 = 'xyz'
 (1 row)
 
+-- too many pk fields, should fail
+SELECT dblink_build_sql_update('foo','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}','{"99", "xyz", "{za0,zb0,zc0}"}');
+ERROR:  number of primary key fields exceeds number of specified relation attributes
 -- build a delete statement based on a local tuple,
 SELECT dblink_build_sql_delete('foo','1 2',2,'{"0", "a"}');
            dblink_build_sql_delete           
@@ -54,6 +60,9 @@ SELECT dblink_build_sql_delete('foo','1 2',2,'{"0", "a"}');
  DELETE FROM foo WHERE f1 = '0' AND f2 = 'a'
 (1 row)
 
+-- too many pk fields, should fail
+SELECT dblink_build_sql_delete('foo','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}');
+ERROR:  number of primary key fields exceeds number of specified relation attributes
 -- retest using a quoted and schema qualified table
 CREATE SCHEMA "MySchema";
 CREATE TABLE "MySchema"."Foo"(f1 int, f2 text, f3 text[], primary key (f1,f2));

contrib/dblink/sql/dblink.sql

@@ -34,13 +34,19 @@ FROM dblink_get_pkey('foo');
 -- build an insert statement based on a local tuple,
 -- replacing the primary key values with new ones
 SELECT dblink_build_sql_insert('foo','1 2',2,'{"0", "a"}','{"99", "xyz"}');
+-- too many pk fields, should fail
+SELECT dblink_build_sql_insert('foo','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}','{"99", "xyz", "{za0,zb0,zc0}"}');
 
 -- build an update statement based on a local tuple,
 -- replacing the primary key values with new ones
 SELECT dblink_build_sql_update('foo','1 2',2,'{"0", "a"}','{"99", "xyz"}');
+-- too many pk fields, should fail
+SELECT dblink_build_sql_update('foo','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}','{"99", "xyz", "{za0,zb0,zc0}"}');
 
 -- build a delete statement based on a local tuple,
 SELECT dblink_build_sql_delete('foo','1 2',2,'{"0", "a"}');
+-- too many pk fields, should fail
+SELECT dblink_build_sql_delete('foo','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}');
 
 -- retest using a quoted and schema qualified table
 CREATE SCHEMA "MySchema";