Fix ALTER TABLE to check pre-existing NOT NULL constraints when rewriting

a table.  Otherwise a USING clause that yields NULL can leave the table
violating its constraint (possibly there are other cases too).  Per report
from Alexander Pravking.

diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index f67f1b62ec4ff68e65f5c6d2954b2e714a5f3df5..d34cc8824c19dc9832627e0fa64ef04d81f01de2 100644 (file)
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *       $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.192 2006/07/03 22:45:38 tgl Exp $
+ *       $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.193 2006/07/10 22:10:39 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -123,6 +123,7 @@ typedef struct AlteredTableInfo
        /* Information saved by Phases 1/2 for Phase 3: */
        List       *constraints;        /* List of NewConstraint */
        List       *newvals;            /* List of NewColumnValue */
+       bool            new_notnull;    /* T if we added new NOT NULL constraints */
        Oid                     newTableSpace;  /* new tablespace; 0 means no change */
        /* Objects to rebuild after completing ALTER TYPE operations */
        List       *changedConstraintOids;      /* OIDs of constraints to rebuild */
@@ -132,11 +133,11 @@ typedef struct AlteredTableInfo
 } AlteredTableInfo;
 
 /* Struct describing one new constraint to check in Phase 3 scan */
+/* Note: new NOT NULL constraints are handled elsewhere */
 typedef struct NewConstraint
 {
        char       *name;                       /* Constraint name, or NULL if none */
-       ConstrType      contype;                /* CHECK, NOT_NULL, or FOREIGN */
-       AttrNumber      attnum;                 /* only relevant for NOT_NULL */
+       ConstrType      contype;                /* CHECK or FOREIGN */
        Oid                     refrelid;               /* PK rel, if FOREIGN */
        Node       *qual;                       /* Check expr or FkConstraint struct */
        List       *qualstate;          /* Execution state for CHECK */
@@ -2438,7 +2439,7 @@ ATRewriteTables(List **wqueue)
                         * Test the current data within the table against new constraints
                         * generated by ALTER TABLE commands, but don't rebuild data.
                         */
-                       if (tab->constraints != NIL)
+                       if (tab->constraints != NIL || tab->new_notnull)
                                ATRewriteTable(tab, InvalidOid);
 
                        /*
@@ -2504,6 +2505,7 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap)
        TupleDesc       oldTupDesc;
        TupleDesc       newTupDesc;
        bool            needscan = false;
+       List       *notnull_attrs;
        int                     i;
        ListCell   *l;
        EState     *estate;
@@ -2554,9 +2556,6 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap)
                        case CONSTR_FOREIGN:
                                /* Nothing to do here */
                                break;
-                       case CONSTR_NOTNULL:
-                               needscan = true;
-                               break;
                        default:
                                elog(ERROR, "unrecognized constraint type: %d",
                                         (int) con->contype);
@@ -2572,6 +2571,25 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap)
                ex->exprstate = ExecPrepareExpr((Expr *) ex->expr, estate);
        }
 
+       notnull_attrs = NIL;
+       if (newrel || tab->new_notnull)
+       {
+               /*
+                * If we are rebuilding the tuples OR if we added any new NOT NULL
+                * constraints, check all not-null constraints.  This is a bit of
+                * overkill but it minimizes risk of bugs, and heap_attisnull is
+                * a pretty cheap test anyway.
+                */
+               for (i = 0; i < newTupDesc->natts; i++)
+               {
+                       if (newTupDesc->attrs[i]->attnotnull &&
+                               !newTupDesc->attrs[i]->attisdropped)
+                               notnull_attrs = lappend_int(notnull_attrs, i);
+               }
+               if (notnull_attrs)
+                       needscan = true;
+       }
+
        if (needscan)
        {
                ExprContext *econtext;
@@ -2672,6 +2690,17 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap)
                        ExecStoreTuple(tuple, newslot, InvalidBuffer, false);
                        econtext->ecxt_scantuple = newslot;
 
+                       foreach(l, notnull_attrs)
+                       {
+                               int             attn = lfirst_int(l);
+
+                               if (heap_attisnull(tuple, attn+1))
+                                       ereport(ERROR,
+                                                       (errcode(ERRCODE_NOT_NULL_VIOLATION),
+                                                        errmsg("column \"%s\" contains null values",
+                                                                       NameStr(newTupDesc->attrs[attn]->attname))));
+                       }
+
                        foreach(l, tab->constraints)
                        {
                                NewConstraint *con = lfirst(l);
@@ -2685,21 +2714,6 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap)
                                                                         errmsg("check constraint \"%s\" is violated by some row",
                                                                                        con->name)));
                                                break;
-                                       case CONSTR_NOTNULL:
-                                               {
-                                                       Datum           d;
-                                                       bool            isnull;
-
-                                                       d = heap_getattr(tuple, con->attnum, newTupDesc,
-                                                                                        &isnull);
-                                                       if (isnull)
-                                                               ereport(ERROR,
-                                                                               (errcode(ERRCODE_NOT_NULL_VIOLATION),
-                                                                errmsg("column \"%s\" contains null values",
-                                                                               get_attname(tab->relid,
-                                                                                                       con->attnum))));
-                                               }
-                                               break;
                                        case CONSTR_FOREIGN:
                                                /* Nothing to do here */
                                                break;
@@ -3398,7 +3412,6 @@ ATExecSetNotNull(AlteredTableInfo *tab, Relation rel,
        HeapTuple       tuple;
        AttrNumber      attnum;
        Relation        attr_rel;
-       NewConstraint *newcon;
 
        /*
         * lookup the attribute
@@ -3434,13 +3447,8 @@ ATExecSetNotNull(AlteredTableInfo *tab, Relation rel,
                /* keep the system catalog indexes current */
                CatalogUpdateIndexes(attr_rel, tuple);
 
-               /* Tell Phase 3 to test the constraint */
-               newcon = (NewConstraint *) palloc0(sizeof(NewConstraint));
-               newcon->contype = CONSTR_NOTNULL;
-               newcon->attnum = attnum;
-               newcon->name = "NOT NULL";
-
-               tab->constraints = lappend(tab->constraints, newcon);
+               /* Tell Phase 3 it needs to test the constraint */
+               tab->new_notnull = true;
        }
 
        heap_close(attr_rel, RowExclusiveLock);
@@ -3909,7 +3917,6 @@ ATExecAddConstraint(AlteredTableInfo *tab, Relation rel, Node *newConstraint)
                                                                newcon = (NewConstraint *) palloc0(sizeof(NewConstraint));
                                                                newcon->name = ccon->name;
                                                                newcon->contype = ccon->contype;
-                                                               newcon->attnum = ccon->attnum;
                                                                /* ExecQual wants implicit-AND format */
                                                                newcon->qual = (Node *)
                                                                        make_ands_implicit((Expr *) ccon->expr);