@ -259,7 +259,8 @@ static void AlterIndexNamespaces(Relation classRel, Relation rel,
static void AlterSeqNamespaces ( Relation classRel , Relation rel ,
Oid oldNspOid , Oid newNspOid ,
const char * newNspName , LOCKMODE lockmode ) ;
static void ATExecValidateConstraint ( Relation rel , char * constrName ) ;
static void ATExecValidateConstraint ( Relation rel , char * constrName ,
bool recurse , bool recursing , LOCKMODE lockmode ) ;
static int transformColumnNameList ( Oid relId , List * colList ,
int16 * attnums , Oid * atttypids ) ;
static int transformFkeyGetPrimaryKey ( Relation pkrel , Oid * indexOid ,
@ -270,6 +271,7 @@ static Oid transformFkeyCheckAttrs(Relation pkrel,
int numattrs , int16 * attnums ,
Oid * opclasses ) ;
static void checkFkeyPermissions ( Relation rel , int16 * attnums , int natts ) ;
static void validateCheckConstraint ( Relation rel , HeapTuple constrtup ) ;
static void validateForeignKeyConstraint ( char * conname ,
Relation rel , Relation pkrel ,
Oid pkindOid , Oid constraintOid ) ;
@ -561,6 +563,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId)
cooked - > name = NULL ;
cooked - > attnum = attnum ;
cooked - > expr = colDef - > cooked_default ;
cooked - > skip_validation = false ;
cooked - > is_local = true ; /* not used for defaults */
cooked - > inhcount = 0 ; /* ditto */
cookedDefaults = lappend ( cookedDefaults , cooked ) ;
@ -1568,6 +1571,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
cooked - > name = pstrdup ( name ) ;
cooked - > attnum = 0 ; /* not used for constraints */
cooked - > expr = expr ;
cooked - > skip_validation = false ;
cooked - > is_local = false ;
cooked - > inhcount = 1 ;
constraints = lappend ( constraints , cooked ) ;
@ -2933,7 +2937,14 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
ATPrepAddInherit ( rel ) ;
pass = AT_PASS_MISC ;
break ;
case AT_ValidateConstraint :
case AT_ValidateConstraint : /* VALIDATE CONSTRAINT */
ATSimplePermissions ( rel , ATT_TABLE ) ;
/* Recursion occurs during execution phase */
/* No command-specific prep needed except saving recurse flag */
if ( recurse )
cmd - > subtype = AT_ValidateConstraintRecurse ;
pass = AT_PASS_MISC ;
break ;
case AT_EnableTrig : /* ENABLE TRIGGER variants */
case AT_EnableAlwaysTrig :
case AT_EnableReplicaTrig :
@ -3098,8 +3109,12 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
case AT_AddIndexConstraint : /* ADD CONSTRAINT USING INDEX */
ATExecAddIndexConstraint ( tab , rel , ( IndexStmt * ) cmd - > def , lockmode ) ;
break ;
case AT_ValidateConstraint :
ATExecValidateConstraint ( rel , cmd - > name ) ;
case AT_ValidateConstraint : /* VALIDATE CONSTRAINT */
ATExecValidateConstraint ( rel , cmd - > name , false , false , lockmode ) ;
break ;
case AT_ValidateConstraintRecurse : /* VALIDATE CONSTRAINT with
* recursion */
ATExecValidateConstraint ( rel , cmd - > name , true , false , lockmode ) ;
break ;
case AT_DropConstraint : /* DROP CONSTRAINT */
ATExecDropConstraint ( rel , cmd - > name , cmd - > behavior ,
@ -5384,19 +5399,23 @@ ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
list_make1 ( copyObject ( constr ) ) ,
recursing , ! recursing ) ;
/* Add each constraint to Phase 3's queue */
/* Add each to-be-validated constraint to Phase 3's queue */
foreach ( lcon , newcons )
{
CookedConstraint * ccon = ( CookedConstraint * ) lfirst ( lcon ) ;
NewConstraint * newcon ;
newcon = ( NewConstraint * ) palloc0 ( sizeof ( NewConstraint ) ) ;
newcon - > name = ccon - > name ;
newcon - > contype = ccon - > contype ;
/* ExecQual wants implicit-AND format */
newcon - > qual = ( Node * ) make_ands_implicit ( ( Expr * ) ccon - > expr ) ;
if ( ! ccon - > skip_validation )
{
NewConstraint * newcon ;
tab - > constraints = lappend ( tab - > constraints , newcon ) ;
newcon = ( NewConstraint * ) palloc0 ( sizeof ( NewConstraint ) ) ;
newcon - > name = ccon - > name ;
newcon - > contype = ccon - > contype ;
/* ExecQual wants implicit-AND format */
newcon - > qual = ( Node * ) make_ands_implicit ( ( Expr * ) ccon - > expr ) ;
tab - > constraints = lappend ( tab - > constraints , newcon ) ;
}
/* Save the actually assigned name if it was defaulted */
if ( constr - > conname = = NULL )
@ -5755,9 +5774,15 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
/*
* ALTER TABLE VALIDATE CONSTRAINT
*
* XXX The reason we handle recursion here rather than at Phase 1 is because
* there ' s no good way to skip recursing when handling foreign keys : there is
* no need to lock children in that case , yet we wouldn ' t be able to avoid
* doing so at that level .
*/
static void
ATExecValidateConstraint ( Relation rel , char * constrName )
ATExecValidateConstraint ( Relation rel , char * constrName , bool recurse ,
bool recursing , LOCKMODE lockmode )
{
Relation conrel ;
SysScanDesc scan ;
@ -5781,8 +5806,7 @@ ATExecValidateConstraint(Relation rel, char *constrName)
while ( HeapTupleIsValid ( tuple = systable_getnext ( scan ) ) )
{
con = ( Form_pg_constraint ) GETSTRUCT ( tuple ) ;
if ( con - > contype = = CONSTRAINT_FOREIGN & &
strcmp ( NameStr ( con - > conname ) , constrName ) = = 0 )
if ( strcmp ( NameStr ( con - > conname ) , constrName ) = = 0 )
{
found = true ;
break ;
@ -5792,39 +5816,110 @@ ATExecValidateConstraint(Relation rel, char *constrName)
if ( ! found )
ereport ( ERROR ,
( errcode ( ERRCODE_UNDEFINED_OBJECT ) ,
errmsg ( " foreign key constraint \" %s \" of relation \" %s \" does not exist " ,
errmsg ( " constraint \" %s \" of relation \" %s \" does not exist " ,
constrName , RelationGetRelationName ( rel ) ) ) ) ;
if ( con - > contype ! = CONSTRAINT_FOREIGN & &
con - > contype ! = CONSTRAINT_CHECK )
ereport ( ERROR ,
( errcode ( ERRCODE_WRONG_OBJECT_TYPE ) ,
errmsg ( " constraint \" %s \" of relation \" %s \" is not a foreign key or check constraint " ,
constrName , RelationGetRelationName ( rel ) ) ) ) ;
if ( ! con - > convalidated )
{
Oid conid = HeapTupleGetOid ( tuple ) ;
HeapTuple copyTuple = heap_copytuple ( tuple ) ;
Form_pg_constraint copy_con = ( Form_pg_constraint ) GETSTRUCT ( copyTuple ) ;
Relation refrel ;
HeapTuple copyTuple ;
Form_pg_constraint copy_con ;
/*
* Triggers are already in place on both tables , so a concurrent write
* that alters the result here is not possible . Normally we can run a
* query here to do the validation , which would only require
* AccessShareLock . In some cases , it is possible that we might need
* to fire triggers to perform the check , so we take a lock at
* RowShareLock level just in case .
*/
refrel = heap_open ( con - > confrelid , RowShareLock ) ;
if ( con - > contype = = CONSTRAINT_FOREIGN )
{
Oid conid = HeapTupleGetOid ( tuple ) ;
Relation refrel ;
validateForeignKeyConstraint ( constrName , rel , refrel ,
con - > conindid ,
conid ) ;
/*
* Triggers are already in place on both tables , so a concurrent write
* that alters the result here is not possible . Normally we can run a
* query here to do the validation , which would only require
* AccessShareLock . In some cases , it is possible that we might need
* to fire triggers to perform the check , so we take a lock at
* RowShareLock level just in case .
*/
refrel = heap_open ( con - > confrelid , RowShareLock ) ;
validateForeignKeyConstraint ( constrName , rel , refrel ,
con - > conindid ,
conid ) ;
heap_close ( refrel , NoLock ) ;
/*
* Foreign keys do not inherit , so we purposedly ignore the
* recursion bit here
*/
}
else if ( con - > contype = = CONSTRAINT_CHECK )
{
List * children = NIL ;
ListCell * child ;
/*
* If we ' re recursing , the parent has already done this , so skip
* it .
*/
if ( ! recursing )
children = find_all_inheritors ( RelationGetRelid ( rel ) ,
lockmode , NULL ) ;
/*
* For CHECK constraints , we must ensure that we only mark the
* constraint as validated on the parent if it ' s already validated
* on the children .
*
* We recurse before validating on the parent , to reduce risk of
* deadlocks .
*/
foreach ( child , children )
{
Oid childoid = lfirst_oid ( child ) ;
Relation childrel ;
if ( childoid = = RelationGetRelid ( rel ) )
continue ;
/*
* If we are told not to recurse , there had better not be any
* child tables ; else the addition would put them out of step .
*/
if ( ! recurse )
ereport ( ERROR ,
( errcode ( ERRCODE_INVALID_TABLE_DEFINITION ) ,
errmsg ( " constraint must be validated on child tables too " ) ) ) ;
/* find_all_inheritors already got lock */
childrel = heap_open ( childoid , NoLock ) ;
ATExecValidateConstraint ( childrel , constrName , false ,
true , lockmode ) ;
heap_close ( childrel , NoLock ) ;
}
validateCheckConstraint ( rel , tuple ) ;
/*
* Invalidate relcache so that others see the new validated
* constraint .
*/
CacheInvalidateRelcache ( rel ) ;
}
/*
* Now update the catalog , while we have the door open .
*/
copyTuple = heap_copytuple ( tuple ) ;
copy_con = ( Form_pg_constraint ) GETSTRUCT ( copyTuple ) ;
copy_con - > convalidated = true ;
simple_heap_update ( conrel , & copyTuple - > t_self , copyTuple ) ;
CatalogUpdateIndexes ( conrel , copyTuple ) ;
heap_freetuple ( copyTuple ) ;
heap_close ( refrel , NoLock ) ;
}
systable_endscan ( scan ) ;
@ -6130,6 +6225,79 @@ checkFkeyPermissions(Relation rel, int16 *attnums, int natts)
}
}
/*
* Scan the existing rows in a table to verify they meet a proposed
* CHECK constraint .
*
* The caller must have opened and locked the relation appropriately .
*/
static void
validateCheckConstraint ( Relation rel , HeapTuple constrtup )
{
EState * estate ;
Datum val ;
char * conbin ;
Expr * origexpr ;
List * exprstate ;
TupleDesc tupdesc ;
HeapScanDesc scan ;
HeapTuple tuple ;
ExprContext * econtext ;
MemoryContext oldcxt ;
TupleTableSlot * slot ;
Form_pg_constraint constrForm ;
bool isnull ;
constrForm = ( Form_pg_constraint ) GETSTRUCT ( constrtup ) ;
estate = CreateExecutorState ( ) ;
/*
* XXX this tuple doesn ' t really come from a syscache , but this doesn ' t
* matter to SysCacheGetAttr , because it only wants to be able to fetch the
* tupdesc
*/
val = SysCacheGetAttr ( CONSTROID , constrtup , Anum_pg_constraint_conbin ,
& isnull ) ;
if ( isnull )
elog ( ERROR , " null conbin for constraint %u " ,
HeapTupleGetOid ( constrtup ) ) ;
conbin = TextDatumGetCString ( val ) ;
origexpr = ( Expr * ) stringToNode ( conbin ) ;
exprstate = ( List * )
ExecPrepareExpr ( ( Expr * ) make_ands_implicit ( origexpr ) , estate ) ;
econtext = GetPerTupleExprContext ( estate ) ;
tupdesc = RelationGetDescr ( rel ) ;
slot = MakeSingleTupleTableSlot ( tupdesc ) ;
econtext - > ecxt_scantuple = slot ;
scan = heap_beginscan ( rel , SnapshotNow , 0 , NULL ) ;
/*
* Switch to per - tuple memory context and reset it for each tuple
* produced , so we don ' t leak memory .
*/
oldcxt = MemoryContextSwitchTo ( GetPerTupleMemoryContext ( estate ) ) ;
while ( ( tuple = heap_getnext ( scan , ForwardScanDirection ) ) ! = NULL )
{
ExecStoreTuple ( tuple , slot , InvalidBuffer , false ) ;
if ( ! ExecQual ( exprstate , econtext , true ) )
ereport ( ERROR ,
( errcode ( ERRCODE_CHECK_VIOLATION ) ,
errmsg ( " check constraint \" %s \" is violated by some row " ,
NameStr ( constrForm - > conname ) ) ) ) ;
ResetExprContext ( econtext ) ;
}
MemoryContextSwitchTo ( oldcxt ) ;
heap_endscan ( scan ) ;
ExecDropSingleTupleTableSlot ( slot ) ;
FreeExecutorState ( estate ) ;
}
/*
* Scan the existing rows in a table to verify they meet a proposed FK
* constraint .
Alvaro Herrera
14 years ago