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.

diff --git a/contrib/dblink/dblink.c b/contrib/dblink/dblink.c
index b03b16a139a51ef7001ed0bc445cb2500e35c3ec..2478884f0a497d4b64d9967a78624f55760c37c3 100644 (file)
--- a/contrib/dblink/dblink.c
+++ b/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.60.2.9 2010/06/15 16:22:39 tgl Exp $
+ * $PostgreSQL: pgsql/contrib/dblink/dblink.c,v 1.60.2.10 2010/06/15 19:04:34 tgl Exp $
  * Copyright (c) 2001-2006, PostgreSQL Global Development Group
  * ALL RIGHTS RESERVED;
  *
@@ -1800,7 +1800,7 @@ get_sql_insert(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals
        appendStringInfo(&buf, ") VALUES(");
 
        /*
-        * remember attvals are 1 based
+        * Note: i is physical column number (counting from 0).
         */
        needComma = false;
        for (i = 0; i < natts; i++)
@@ -1811,12 +1811,9 @@ get_sql_insert(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals
                if (needComma)
                        appendStringInfo(&buf, ",");
 
-               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 = tgt_pkattvals[key] ? pstrdup(tgt_pkattvals[key]) : NULL;
                else
                        val = SPI_getvalue(tuple, tupdesc, i + 1);
@@ -1861,10 +1858,6 @@ get_sql_delete(Relation rel, int *pkattnums, int pknumatts, char **tgt_pkattvals
                appendStringInfoString(&buf,
                   quote_ident_cstr(NameStr(tupdesc->attrs[pkattnum]->attname)));
 
-               if (tgt_pkattvals == NULL)
-                       /* internal error */
-                       elog(ERROR, "target key array must not be NULL");
-
                if (tgt_pkattvals[i] != NULL)
                        appendStringInfo(&buf, " = %s",
                                                         quote_literal_cstr(tgt_pkattvals[i]));
@@ -1904,6 +1897,9 @@ get_sql_update(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals
 
        appendStringInfo(&buf, "UPDATE %s SET ", relname);
 
+       /*
+        * Note: i is physical column number (counting from 0).
+        */
        needComma = false;
        for (i = 0; i < natts; i++)
        {
@@ -1916,12 +1912,9 @@ get_sql_update(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals
                appendStringInfo(&buf, "%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 = tgt_pkattvals[key] ? pstrdup(tgt_pkattvals[key]) : NULL;
                else
                        val = SPI_getvalue(tuple, tupdesc, i + 1);
@@ -1948,16 +1941,10 @@ get_sql_update(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals
                appendStringInfo(&buf, "%s",
                   quote_ident_cstr(NameStr(tupdesc->attrs[pkattnum]->attname)));
 
-               if (tgt_pkattvals != NULL)
-                       val = tgt_pkattvals[i] ? pstrdup(tgt_pkattvals[i]) : NULL;
-               else
-                       val = SPI_getvalue(tuple, tupdesc, pkattnum + 1);
+               val = tgt_pkattvals[i];
 
                if (val != NULL)
-               {
                        appendStringInfo(&buf, " = %s", quote_literal_cstr(val));
-                       pfree(val);
-               }
                else
                        appendStringInfo(&buf, " IS NULL");
        }
@@ -2021,30 +2008,49 @@ get_tuple_of_interest(Relation rel, int *pkattnums, int pknumatts, char **src_pk
 {
        char       *relname;
        TupleDesc       tupdesc;
+       int                     natts;
        StringInfoData buf;
        int                     ret;
        HeapTuple       tuple;
        int                     i;
 
+       /*
+        * Connect to SPI manager
+        */
+       if ((ret = SPI_connect()) < 0)
+               /* internal error */
+               elog(ERROR, "SPI connect failure - returned %d", ret);
+
        initStringInfo(&buf);
 
        /* get relation name including any needed schema prefix and quoting */
        relname = generate_relation_name(rel);
 
        tupdesc = rel->rd_att;
+       natts = tupdesc->natts;
 
        /*
-        * Connect to SPI manager
+        * 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.
         */
-       if ((ret = SPI_connect()) < 0)
-               /* internal error */
-               elog(ERROR, "SPI connect failure - returned %d", ret);
+       appendStringInfoString(&buf, "SELECT ");
 
-       /*
-        * Build sql statement to look up tuple of interest Use src_pkattvals as
-        * the criteria.
-        */
-       appendStringInfo(&buf, "SELECT * FROM %s WHERE ", relname);
+       for (i = 0; i < natts; i++)
+       {
+               if (i > 0)
+                       appendStringInfoString(&buf, ", ");
+
+               if (tupdesc->attrs[i]->attisdropped)
+                       appendStringInfoString(&buf, "NULL");
+               else
+                       appendStringInfoString(&buf,
+                                                                  quote_ident_cstr(NameStr(tupdesc->attrs[i]->attname)));
+       }
+
+       appendStringInfo(&buf, " 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 4804d8f297de3e1346abfa1d262f8acf0adeca1c..af0534ceffb4ea478b5a0dccc86fab7a61b57dec 100644 (file)
--- a/contrib/dblink/expected/dblink.out
+++ b/contrib/dblink/expected/dblink.out
@@ -802,3 +802,40 @@ SELECT dblink_disconnect('dtest1');
  OK
 (1 row)
 
+-- 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 8572787d2b556fe25a659ded00b7fceff5bf0ede..4d4d49c8ba3bdcf03f6072338413d5e9b7c2b980 100644 (file)
--- a/contrib/dblink/sql/dblink.sql
+++ b/contrib/dblink/sql/dblink.sql
@@ -363,3 +363,29 @@ SELECT * from
 SELECT dblink_cancel_query('dtest1');
 SELECT dblink_error_message('dtest1');
 SELECT dblink_disconnect('dtest1');
+
+-- 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]);