Fix dblink_build_sql_insert() and related functions to handle dropped

columns correctly.  In passing, get rid of some dead logic in the
underlying get_sql_insert() etc functions --- there is no caller that
will pass null value-arrays to them.

Per bug report from Robert Voinea.

3 files changed, 99 insertions, 38 deletions

diff --git a/contrib/dblink/dblink.c b/contrib/dblink/dblink.c
index b737c68003a..7cb8cd4a2d6 100644
--- a/contrib/dblink/dblink.c
+++ b/contrib/dblink/dblink.c
@@ -1642,7 +1642,7 @@ get_sql_insert(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals
 	appendStringInfo(str, ") VALUES(");
 
 	/*
-	 * remember attvals are 1 based
+	 * Note: i is physical column number (counting from 0).
 	 */
 	needComma = false;
 	for (i = 0; i < natts; i++)
@@ -1653,12 +1653,9 @@ get_sql_insert(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals
 		if (needComma)
 			appendStringInfo(str, ",");
 
-		if (tgt_pkattvals != NULL)
-			key = get_attnum_pk_pos(pkattnums, pknumatts, i);
-		else
-			key = -1;
+		key = get_attnum_pk_pos(pkattnums, pknumatts, i);
 
-		if (key > -1)
+		if (key >= 0)
 			val = pstrdup(tgt_pkattvals[key]);
 		else
 			val = SPI_getvalue(tuple, tupdesc, i + 1);
@@ -1688,7 +1685,6 @@ get_sql_delete(Relation rel, int *pkattnums, int pknumatts, char **tgt_pkattvals
 	TupleDesc	tupdesc;
 	StringInfo	str = makeStringInfo();
 	char	   *sql;
-	char	   *val = NULL;
 	int			i;
 
 	/* get relation name including any needed schema prefix and quoting */
@@ -1707,17 +1703,9 @@ get_sql_delete(Relation rel, int *pkattnums, int pknumatts, char **tgt_pkattvals
 		appendStringInfo(str, "%s",
 		   quote_ident_cstr(NameStr(tupdesc->attrs[pkattnum]->attname)));
 
-		if (tgt_pkattvals != NULL)
-			val = pstrdup(tgt_pkattvals[i]);
-		else
-			/* internal error */
-			elog(ERROR, "target key array must not be NULL");
-
-		if (val != NULL)
-		{
-			appendStringInfo(str, " = %s", quote_literal_cstr(val));
-			pfree(val);
-		}
+		if (tgt_pkattvals[i] != NULL)
+			appendStringInfo(str, " = %s",
+							 quote_literal_cstr(tgt_pkattvals[i]));
 		else
 			appendStringInfo(str, " IS NULL");
 	}
@@ -1769,12 +1757,9 @@ get_sql_update(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals
 		appendStringInfo(str, "%s = ",
 					  quote_ident_cstr(NameStr(tupdesc->attrs[i]->attname)));
 
-		if (tgt_pkattvals != NULL)
-			key = get_attnum_pk_pos(pkattnums, pknumatts, i);
-		else
-			key = -1;
+		key = get_attnum_pk_pos(pkattnums, pknumatts, i);
 
-		if (key > -1)
+		if (key >= 0)
 			val = pstrdup(tgt_pkattvals[key]);
 		else
 			val = SPI_getvalue(tuple, tupdesc, i + 1);
@@ -1801,16 +1786,10 @@ get_sql_update(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals
 		appendStringInfo(str, "%s",
 		   quote_ident_cstr(NameStr(tupdesc->attrs[pkattnum]->attname)));
 
-		if (tgt_pkattvals != NULL)
-			val = pstrdup(tgt_pkattvals[i]);
-		else
-			val = SPI_getvalue(tuple, tupdesc, pkattnum + 1);
+		val = tgt_pkattvals[i];
 
 		if (val != NULL)
-		{
 			appendStringInfo(str, " = %s", quote_literal_cstr(val));
-			pfree(val);
-		}
 		else
 			appendStringInfo(str, " IS NULL");
 	}
@@ -1878,6 +1857,7 @@ get_tuple_of_interest(Relation rel, int *pkattnums, int pknumatts, char **src_pk
 {
 	char	   *relname;
 	TupleDesc	tupdesc;
+	int			natts;
 	StringInfo	str = makeStringInfo();
 	char	   *sql = NULL;
 	int			ret;
@@ -1885,11 +1865,6 @@ get_tuple_of_interest(Relation rel, int *pkattnums, int pknumatts, char **src_pk
 	int			i;
 	char	   *val = NULL;
 
-	/* get relation name including any needed schema prefix and quoting */
-	relname = generate_relation_name(rel);
-
-	tupdesc = rel->rd_att;
-
 	/*
 	 * Connect to SPI manager
 	 */
@@ -1897,11 +1872,34 @@ get_tuple_of_interest(Relation rel, int *pkattnums, int pknumatts, char **src_pk
 		/* internal error */
 		elog(ERROR, "SPI connect failure - returned %d", ret);
 
+	/* get relation name including any needed schema prefix and quoting */
+	relname = generate_relation_name(rel);
+
+	tupdesc = rel->rd_att;
+	natts = tupdesc->natts;
+
 	/*
-	 * Build sql statement to look up tuple of interest Use src_pkattvals as
-	 * the criteria.
+	 * Build sql statement to look up tuple of interest, ie, the one matching
+	 * src_pkattvals.  We used to use "SELECT *" here, but it's simpler to
+	 * generate a result tuple that matches the table's physical structure,
+	 * with NULLs for any dropped columns.  Otherwise we have to deal with
+	 * two different tupdescs and everything's very confusing.
 	 */
-	appendStringInfo(str, "SELECT * FROM %s WHERE ", relname);
+	appendStringInfoString(str, "SELECT ");
+
+	for (i = 0; i < natts; i++)
+	{
+		if (i > 0)
+			appendStringInfoString(str, ", ");
+
+		if (tupdesc->attrs[i]->attisdropped)
+			appendStringInfoString(str, "NULL");
+		else
+			appendStringInfoString(str,
+								   quote_ident_cstr(NameStr(tupdesc->attrs[i]->attname)));
+	}
+
+	appendStringInfo(str, " FROM %s WHERE ", relname);
 
 	for (i = 0; i < pknumatts; i++)
 	{
diff --git a/contrib/dblink/expected/dblink.out b/contrib/dblink/expected/dblink.out
index fa9a0feb28d..51a9bfe3030 100644
--- a/contrib/dblink/expected/dblink.out
+++ b/contrib/dblink/expected/dblink.out
@@ -678,3 +678,40 @@ SELECT dblink_disconnect('myconn');
 -- should get 'connection "myconn" not available' error
 SELECT dblink_disconnect('myconn');
 ERROR:  connection "myconn" not available
+-- test dropped columns in dblink_build_sql_insert, dblink_build_sql_update
+CREATE TEMP TABLE test_dropped
+(
+	col1 INT NOT NULL DEFAULT 111,
+	id SERIAL PRIMARY KEY,
+	col2 INT NOT NULL DEFAULT 112,
+	col2b INT NOT NULL DEFAULT 113
+);
+NOTICE:  CREATE TABLE will create implicit sequence "test_dropped_id_seq" for serial column "test_dropped.id"
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "test_dropped_pkey" for table "test_dropped"
+INSERT INTO test_dropped VALUES(default);
+ALTER TABLE test_dropped
+	DROP COLUMN col1,
+	DROP COLUMN col2,
+	ADD COLUMN col3 VARCHAR(10) NOT NULL DEFAULT 'foo',
+	ADD COLUMN col4 INT NOT NULL DEFAULT 42;
+SELECT dblink_build_sql_insert('test_dropped', '2', 1,
+                               ARRAY['1'::TEXT], ARRAY['2'::TEXT]);
+                          dblink_build_sql_insert                          
+---------------------------------------------------------------------------
+ INSERT INTO test_dropped(id,col2b,col3,col4) VALUES('2','113','foo','42')
+(1 row)
+
+SELECT dblink_build_sql_update('test_dropped', '2', 1,
+                               ARRAY['1'::TEXT], ARRAY['2'::TEXT]);
+                                  dblink_build_sql_update                                  
+-------------------------------------------------------------------------------------------
+ UPDATE test_dropped SET id = '2', col2b = '113', col3 = 'foo', col4 = '42' WHERE id = '2'
+(1 row)
+
+SELECT dblink_build_sql_delete('test_dropped', '2', 1,
+                               ARRAY['2'::TEXT]);
+         dblink_build_sql_delete         
+-----------------------------------------
+ DELETE FROM test_dropped WHERE id = '2'
+(1 row)
+
diff --git a/contrib/dblink/sql/dblink.sql b/contrib/dblink/sql/dblink.sql
index 1987da8da27..bb493480ffd 100644
--- a/contrib/dblink/sql/dblink.sql
+++ b/contrib/dblink/sql/dblink.sql
@@ -325,3 +325,29 @@ SELECT dblink_disconnect('myconn');
 -- close the named persistent connection again
 -- should get 'connection "myconn" not available' error
 SELECT dblink_disconnect('myconn');
+
+-- test dropped columns in dblink_build_sql_insert, dblink_build_sql_update
+CREATE TEMP TABLE test_dropped
+(
+	col1 INT NOT NULL DEFAULT 111,
+	id SERIAL PRIMARY KEY,
+	col2 INT NOT NULL DEFAULT 112,
+	col2b INT NOT NULL DEFAULT 113
+);
+
+INSERT INTO test_dropped VALUES(default);
+
+ALTER TABLE test_dropped
+	DROP COLUMN col1,
+	DROP COLUMN col2,
+	ADD COLUMN col3 VARCHAR(10) NOT NULL DEFAULT 'foo',
+	ADD COLUMN col4 INT NOT NULL DEFAULT 42;
+
+SELECT dblink_build_sql_insert('test_dropped', '2', 1,
+                               ARRAY['1'::TEXT], ARRAY['2'::TEXT]);
+
+SELECT dblink_build_sql_update('test_dropped', '2', 1,
+                               ARRAY['1'::TEXT], ARRAY['2'::TEXT]);
+
+SELECT dblink_build_sql_delete('test_dropped', '2', 1,
+                               ARRAY['2'::TEXT]);