From 2dbbda02e7e688311e161a912a0ce00cde9bb6fc Mon Sep 17 00:00:00 2001
From: Simon Riggs <simon@2ndQuadrant.com>
Date: Wed, 28 Jul 2010 05:22:24 +0000
Subject: [PATCH] Reduce lock levels of CREATE TRIGGER and some ALTER TABLE,
CREATE RULE actions. Avoid hard-coding lockmode used for many altering DDL
commands, allowing easier future changes of lock levels. Implementation of
initial analysis on DDL sub-commands, so that many lock levels are now at
ShareUpdateExclusiveLock or ShareRowExclusiveLock, allowing certain DDL not
to block reads/writes. First of number of planned changes in this area;
additional docs required when full project complete.
---
doc/src/sgml/mvcc.sgml | 18 +-
src/backend/catalog/pg_shdepend.c | 4 +-
src/backend/commands/alter.c | 4 +-
src/backend/commands/tablecmds.c | 559 ++++++++++++++++++----------
src/backend/commands/trigger.c | 20 +-
src/backend/commands/typecmds.c | 4 +-
src/backend/parser/parse_utilcmd.c | 23 +-
src/backend/rewrite/rewriteDefine.c | 13 +-
src/backend/utils/adt/ri_triggers.c | 4 +-
src/include/commands/tablecmds.h | 9 +-
10 files changed, 426 insertions(+), 232 deletions(-)
diff --git a/doc/src/sgml/mvcc.sgml b/doc/src/sgml/mvcc.sgml
index 341351e46ff..e72a32c58ee 100644
--- a/doc/src/sgml/mvcc.sgml
+++ b/doc/src/sgml/mvcc.sgml
@@ -1,4 +1,4 @@
-<!-- $PostgreSQL: pgsql/doc/src/sgml/mvcc.sgml,v 2.75 2010/05/03 15:35:30 alvherre Exp $ -->
+<!-- $PostgreSQL: pgsql/doc/src/sgml/mvcc.sgml,v 2.76 2010/07/28 05:22:24 sriggs Exp $ -->
<chapter id="mvcc">
<title>Concurrency Control</title>
@@ -532,7 +532,7 @@ SELECT SUM(value) FROM mytab WHERE class = 2;
most <productname>PostgreSQL</productname> commands automatically
acquire locks of appropriate modes to ensure that referenced
tables are not dropped or modified in incompatible ways while the
- command executes. (For example, <command>ALTER TABLE</> cannot safely be
+ command executes. (For example, <command>TRUNCATE</> cannot safely be
executed concurrently with other operations on the same table, so it
obtains an exclusive lock on the table to enforce that.)
</para>
@@ -695,8 +695,9 @@ SELECT SUM(value) FROM mytab WHERE class = 2;
</para>
<para>
- This lock mode is not automatically acquired by any
- <productname>PostgreSQL</productname> command.
+ Acquired by <command>CREATE TRIGGER</command>,
+ <command>CREATE RULE</command> (except for <literal>ON SELECT</>
+ rules) and in some cases <command>ALTER TABLE</command>.
</para>
</listitem>
</varlistentry>
@@ -742,11 +743,12 @@ SELECT SUM(value) FROM mytab WHERE class = 2;
</para>
<para>
- Acquired by the <command>ALTER TABLE</command>, <command>DROP
- TABLE</command>, <command>TRUNCATE</command>, <command>REINDEX</command>,
+ Acquired by the <command>DROP TABLE</command>,
+ <command>TRUNCATE</command>, <command>REINDEX</command>,
<command>CLUSTER</command>, and <command>VACUUM FULL</command>
- commands. This is also the default lock mode for <command>LOCK
- TABLE</command> statements that do not specify a mode explicitly.
+ commands, as well as most variants of <command>ALTER TABLE</>.
+ This is also the default lock mode for <command>LOCK TABLE</command>
+ statements that do not specify a mode explicitly.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/catalog/pg_shdepend.c b/src/backend/catalog/pg_shdepend.c
index e525a551ffe..a03fc0d4e2e 100644
--- a/src/backend/catalog/pg_shdepend.c
+++ b/src/backend/catalog/pg_shdepend.c
@@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/catalog/pg_shdepend.c,v 1.43 2010/07/06 19:18:56 momjian Exp $
+ * $PostgreSQL: pgsql/src/backend/catalog/pg_shdepend.c,v 1.44 2010/07/28 05:22:24 sriggs Exp $
*
*-------------------------------------------------------------------------
*/
@@ -1346,7 +1346,7 @@ shdepReassignOwned(List *roleids, Oid newrole)
* owned sequences, etc when we happen to visit them
* before their parent table.
*/
- ATExecChangeOwner(sdepForm->objid, newrole, true);
+ ATExecChangeOwner(sdepForm->objid, newrole, true, AccessExclusiveLock);
break;
case ProcedureRelationId:
diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c
index cecdfa5c6d4..edf05b6eaa3 100644
--- a/src/backend/commands/alter.c
+++ b/src/backend/commands/alter.c
@@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/commands/alter.c,v 1.36 2010/06/13 17:43:12 rhaas Exp $
+ * $PostgreSQL: pgsql/src/backend/commands/alter.c,v 1.37 2010/07/28 05:22:24 sriggs Exp $
*
*-------------------------------------------------------------------------
*/
@@ -190,7 +190,7 @@ ExecAlterObjectSchemaStmt(AlterObjectSchemaStmt *stmt)
case OBJECT_VIEW:
CheckRelationOwnership(stmt->relation, true);
AlterTableNamespace(stmt->relation, stmt->newschema,
- stmt->objectType);
+ stmt->objectType, AccessExclusiveLock);
break;
case OBJECT_TYPE:
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 2fdce989c01..1496b6eb159 100644
--- 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.334 2010/07/25 23:21:21 rhaas Exp $
+ * $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.335 2010/07/28 05:22:24 sriggs Exp $
*
*-------------------------------------------------------------------------
*/
@@ -64,6 +64,7 @@
#include "rewrite/rewriteHandler.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
+#include "storage/lock.h"
#include "storage/smgr.h"
#include "utils/acl.h"
#include "utils/builtins.h"
@@ -237,7 +238,7 @@ static void AlterIndexNamespaces(Relation classRel, Relation rel,
Oid oldNspOid, Oid newNspOid);
static void AlterSeqNamespaces(Relation classRel, Relation rel,
Oid oldNspOid, Oid newNspOid,
- const char *newNspName);
+ const char *newNspName, LOCKMODE lockmode);
static int transformColumnNameList(Oid relId, List *colList,
int16 *attnums, Oid *atttypids);
static int transformFkeyGetPrimaryKey(Relation pkrel, Oid *indexOid,
@@ -253,84 +254,84 @@ static void validateForeignKeyConstraint(Constraint *fkconstraint,
Oid pkindOid, Oid constraintOid);
static void createForeignKeyTriggers(Relation rel, Constraint *fkconstraint,
Oid constraintOid, Oid indexOid);
-static void ATController(Relation rel, List *cmds, bool recurse);
+static void ATController(Relation rel, List *cmds, bool recurse, LOCKMODE lockmode);
static void ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
- bool recurse, bool recursing);
-static void ATRewriteCatalogs(List **wqueue);
+ bool recurse, bool recursing, LOCKMODE lockmode);
+static void ATRewriteCatalogs(List **wqueue, LOCKMODE lockmode);
static void ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
- AlterTableCmd *cmd);
-static void ATRewriteTables(List **wqueue);
-static void ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap);
+ AlterTableCmd *cmd, LOCKMODE lockmode);
+static void ATRewriteTables(List **wqueue, LOCKMODE lockmode);
+static void ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode);
static AlteredTableInfo *ATGetQueueEntry(List **wqueue, Relation rel);
static void ATSimplePermissions(Relation rel, bool allowView);
static void ATSimplePermissionsRelationOrIndex(Relation rel);
static void ATSimpleRecursion(List **wqueue, Relation rel,
- AlterTableCmd *cmd, bool recurse);
+ AlterTableCmd *cmd, bool recurse, LOCKMODE lockmode);
static void ATOneLevelRecursion(List **wqueue, Relation rel,
- AlterTableCmd *cmd);
+ AlterTableCmd *cmd, LOCKMODE lockmode);
static void ATPrepAddColumn(List **wqueue, Relation rel, bool recurse,
- AlterTableCmd *cmd);
+ AlterTableCmd *cmd, LOCKMODE lockmode);
static void ATExecAddColumn(AlteredTableInfo *tab, Relation rel,
- ColumnDef *colDef, bool isOid);
+ ColumnDef *colDef, bool isOid, LOCKMODE lockmode);
static void add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid);
static void ATPrepAddOids(List **wqueue, Relation rel, bool recurse,
- AlterTableCmd *cmd);
-static void ATExecDropNotNull(Relation rel, const char *colName);
+ AlterTableCmd *cmd, LOCKMODE lockmode);
+static void ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode);
static void ATExecSetNotNull(AlteredTableInfo *tab, Relation rel,
- const char *colName);
+ const char *colName, LOCKMODE lockmode);
static void ATExecColumnDefault(Relation rel, const char *colName,
- Node *newDefault);
+ Node *newDefault, LOCKMODE lockmode);
static void ATPrepSetStatistics(Relation rel, const char *colName,
- Node *newValue);
+ Node *newValue, LOCKMODE lockmode);
static void ATExecSetStatistics(Relation rel, const char *colName,
- Node *newValue);
+ Node *newValue, LOCKMODE lockmode);
static void ATExecSetOptions(Relation rel, const char *colName,
- Node *options, bool isReset);
+ Node *options, bool isReset, LOCKMODE lockmode);
static void ATExecSetStorage(Relation rel, const char *colName,
- Node *newValue);
+ Node *newValue, LOCKMODE lockmode);
static void ATPrepDropColumn(Relation rel, bool recurse, AlterTableCmd *cmd);
static void ATExecDropColumn(List **wqueue, Relation rel, const char *colName,
DropBehavior behavior,
bool recurse, bool recursing,
- bool missing_ok);
+ bool missing_ok, LOCKMODE lockmode);
static void ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
- IndexStmt *stmt, bool is_rebuild);
+ IndexStmt *stmt, bool is_rebuild, LOCKMODE lockmode);
static void ATExecAddConstraint(List **wqueue,
AlteredTableInfo *tab, Relation rel,
- Constraint *newConstraint, bool recurse);
+ Constraint *newConstraint, bool recurse, LOCKMODE lockmode);
static void ATAddCheckConstraint(List **wqueue,
AlteredTableInfo *tab, Relation rel,
Constraint *constr,
- bool recurse, bool recursing);
+ bool recurse, bool recursing, LOCKMODE lockmode);
static void ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
- Constraint *fkconstraint);
+ Constraint *fkconstraint, LOCKMODE lockmode);
static void ATExecDropConstraint(Relation rel, const char *constrName,
DropBehavior behavior,
bool recurse, bool recursing,
- bool missing_ok);
+ bool missing_ok, LOCKMODE lockmode);
static void ATPrepAlterColumnType(List **wqueue,
AlteredTableInfo *tab, Relation rel,
bool recurse, bool recursing,
- AlterTableCmd *cmd);
+ AlterTableCmd *cmd, LOCKMODE lockmode);
static void ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
- const char *colName, TypeName *typeName);
-static void ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab);
-static void ATPostAlterTypeParse(char *cmd, List **wqueue);
+ const char *colName, TypeName *typeName, LOCKMODE lockmode);
+static void ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab, LOCKMODE lockmode);
+static void ATPostAlterTypeParse(char *cmd, List **wqueue, LOCKMODE lockmode);
static void change_owner_recurse_to_sequences(Oid relationOid,
- Oid newOwnerId);
-static void ATExecClusterOn(Relation rel, const char *indexName);
-static void ATExecDropCluster(Relation rel);
+ Oid newOwnerId, LOCKMODE lockmode);
+static void ATExecClusterOn(Relation rel, const char *indexName, LOCKMODE lockmode);
+static void ATExecDropCluster(Relation rel, LOCKMODE lockmode);
static void ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel,
- char *tablespacename);
-static void ATExecSetTableSpace(Oid tableOid, Oid newTableSpace);
-static void ATExecSetRelOptions(Relation rel, List *defList, bool isReset);
+ char *tablespacename, LOCKMODE lockmode);
+static void ATExecSetTableSpace(Oid tableOid, Oid newTableSpace, LOCKMODE lockmode);
+static void ATExecSetRelOptions(Relation rel, List *defList, bool isReset, LOCKMODE lockmode);
static void ATExecEnableDisableTrigger(Relation rel, char *trigname,
- char fires_when, bool skip_system);
+ char fires_when, bool skip_system, LOCKMODE lockmode);
static void ATExecEnableDisableRule(Relation rel, char *rulename,
- char fires_when);
+ char fires_when, LOCKMODE lockmode);
static void ATPrepAddInherit(Relation child_rel);
-static void ATExecAddInherit(Relation child_rel, RangeVar *parent);
-static void ATExecDropInherit(Relation rel, RangeVar *parent);
+static void ATExecAddInherit(Relation child_rel, RangeVar *parent, LOCKMODE lockmode);
+static void ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode);
static void copy_relation_data(SMgrRelation rel, SMgrRelation dst,
ForkNumber forkNum, bool istemp);
static const char *storage_name(char c);
@@ -2249,8 +2250,8 @@ RenameRelationInternal(Oid myrelid, const char *newrelname, Oid namespaceId)
* Disallow ALTER TABLE (and similar commands) when the current backend has
* any open reference to the target table besides the one just acquired by
* the calling command; this implies there's an open cursor or active plan.
- * We need this check because our AccessExclusiveLock doesn't protect us
- * against stomping on our own foot, only other people's feet!
+ * We need this check because our lock doesn't protect us against stomping
+ * on our own foot, only other people's feet!
*
* For ALTER TABLE, the only case known to cause serious trouble is ALTER
* COLUMN TYPE, and some changes are obviously pretty benign, so this could
@@ -2327,11 +2328,27 @@ CheckTableNotInUse(Relation rel, const char *stmt)
*
* Thanks to the magic of MVCC, an error anywhere along the way rolls back
* the whole operation; we don't have to do anything special to clean up.
+ *
+ * We lock the table as the first action, with an appropriate lock level
+ * for the subcommands requested. Any subcommand that needs to rewrite
+ * tuples in the table forces the whole command to be executed with
+ * AccessExclusiveLock. If all subcommands do not require rewrite table
+ * then we may be able to use lower lock levels. We pass the lock level down
+ * so that we can apply it recursively to inherited tables. Note that the
+ * lock level we want as we recurse may well be higher than required for
+ * that specific subcommand. So we pass down the overall lock requirement,
+ * rather than reassess it at lower levels.
*/
void
AlterTable(AlterTableStmt *stmt)
{
- Relation rel = relation_openrv(stmt->relation, AccessExclusiveLock);
+ Relation rel;
+ LOCKMODE lockmode = AlterTableGetLockLevel(stmt->cmds);
+
+ /*
+ * Acquire same level of lock as already acquired during parsing.
+ */
+ rel = relation_openrv(stmt->relation, lockmode);
CheckTableNotInUse(rel, "ALTER TABLE");
@@ -2374,7 +2391,8 @@ AlterTable(AlterTableStmt *stmt)
elog(ERROR, "unrecognized object type: %d", (int) stmt->relkind);
}
- ATController(rel, stmt->cmds, interpretInhOption(stmt->relation->inhOpt));
+ ATController(rel, stmt->cmds, interpretInhOption(stmt->relation->inhOpt),
+ lockmode);
}
/*
@@ -2391,13 +2409,169 @@ AlterTable(AlterTableStmt *stmt)
void
AlterTableInternal(Oid relid, List *cmds, bool recurse)
{
- Relation rel = relation_open(relid, AccessExclusiveLock);
+ Relation rel;
+ LOCKMODE lockmode = AlterTableGetLockLevel(cmds);
- ATController(rel, cmds, recurse);
+ rel = relation_open(relid, lockmode);
+
+ ATController(rel, cmds, recurse, lockmode);
+}
+
+/*
+ * AlterTableGetLockLevel
+ *
+ * Sets the overall lock level required for the supplied list of subcommands.
+ * Policy for doing this set according to needs of AlterTable(), see
+ * comments there for overall explanation.
+ *
+ * Function is called before and after parsing, so it must give same
+ * answer each time it is called. Some subcommands are transformed
+ * into other subcommand types, so the transform must never be made to a
+ * lower lock level than previously assigned. All transforms are noted below.
+ *
+ * Since this is called before we lock the table we cannot use table metadata
+ * to influence the type of lock we acquire.
+ *
+ * There should be no lockmodes hardcoded into the subcommand functions. All
+ * lockmode decisions for ALTER TABLE are made here only. The one exception is
+ * ALTER TABLE RENAME which is treated as a different statement type T_RenameStmt
+ * and does not travel through this section of code and cannot be combined with
+ * any of the subcommands given here.
+ */
+LOCKMODE
+AlterTableGetLockLevel(List *cmds)
+{
+ ListCell *lcmd;
+ LOCKMODE lockmode = ShareUpdateExclusiveLock;
+
+ foreach(lcmd, cmds)
+ {
+ AlterTableCmd *cmd = (AlterTableCmd *) lfirst(lcmd);
+ LOCKMODE cmd_lockmode = AccessExclusiveLock; /* default for compiler */
+
+ switch (cmd->subtype)
+ {
+ /*
+ * Need AccessExclusiveLock for these subcommands because they
+ * affect or potentially affect both read and write operations.
+ *
+ * New subcommand types should be added here by default.
+ */
+ case AT_AddColumn: /* may rewrite heap, in some cases and visible to SELECT */
+ case AT_DropColumn: /* change visible to SELECT */
+ case AT_AddColumnToView: /* CREATE VIEW */
+ case AT_AlterColumnType: /* must rewrite heap */
+ case AT_DropConstraint: /* as DROP INDEX */
+ case AT_AddOids: /* must rewrite heap */
+ case AT_DropOids: /* calls AT_DropColumn */
+ case AT_EnableAlwaysRule: /* may change SELECT rules */
+ case AT_EnableReplicaRule: /* may change SELECT rules */
+ case AT_EnableRule: /* may change SELECT rules */
+ case AT_DisableRule: /* may change SELECT rules */
+ case AT_ChangeOwner: /* change visible to SELECT */
+ case AT_SetTableSpace: /* must rewrite heap */
+ case AT_DropNotNull: /* may change some SQL plans */
+ case AT_SetNotNull:
+ cmd_lockmode = AccessExclusiveLock;
+ break;
+
+ /*
+ * These subcommands affect write operations only.
+ */
+ case AT_ColumnDefault:
+ case AT_ProcessedConstraint: /* becomes AT_AddConstraint */
+ case AT_AddConstraintRecurse: /* becomes AT_AddConstraint */
+ case AT_EnableTrig:
+ case AT_EnableAlwaysTrig:
+ case AT_EnableReplicaTrig:
+ case AT_EnableTrigAll:
+ case AT_EnableTrigUser:
+ case AT_DisableTrig:
+ case AT_DisableTrigAll:
+ case AT_DisableTrigUser:
+ case AT_AddIndex: /* from ADD CONSTRAINT */
+ cmd_lockmode = ShareRowExclusiveLock;
+ break;
+
+ case AT_AddConstraint:
+ if (IsA(cmd->def, Constraint))
+ {
+ Constraint *con = (Constraint *) cmd->def;
+
+ switch (con->contype)
+ {
+ case CONSTR_EXCLUSION:
+ case CONSTR_PRIMARY:
+ case CONSTR_UNIQUE:
+ /*
+ * Cases essentially the same as CREATE INDEX. We
+ * could reduce the lock strength to ShareLock if we
+ * can work out how to allow concurrent catalog updates.
+ */
+ cmd_lockmode = ShareRowExclusiveLock;
+ break;
+ case CONSTR_FOREIGN:
+ /*
+ * We add triggers to both tables when we add a
+ * Foreign Key, so the lock level must be at least
+ * as strong as CREATE TRIGGER.
+ */
+ cmd_lockmode = ShareRowExclusiveLock;
+ break;
+
+ default:
+ cmd_lockmode = ShareRowExclusiveLock;
+ }
+ }
+ break;
+
+ /*
+ * These subcommands affect inheritance behaviour. Queries started before us
+ * will continue to see the old inheritance behaviour, while queries started
+ * after we commit will see new behaviour. No need to prevent reads or writes
+ * to the subtable while we hook it up though. In both cases the parent table
+ * is locked with AccessShareLock.
+ */
+ case AT_AddInherit:
+ case AT_DropInherit:
+ cmd_lockmode = ShareUpdateExclusiveLock;
+ break;
+
+ /*
+ * These subcommands affect general strategies for performance and maintenance,
+ * though don't change the semantic results from normal data reads and writes.
+ * Delaying an ALTER TABLE behind currently active writes only delays the point
+ * where the new strategy begins to take effect, so there is no benefit in waiting.
+ * In thise case the minimum restriction applies: we don't currently allow
+ * concurrent catalog updates.
+ */
+ case AT_SetStatistics:
+ case AT_ClusterOn:
+ case AT_DropCluster:
+ case AT_SetRelOptions:
+ case AT_ResetRelOptions:
+ case AT_SetStorage:
+ cmd_lockmode = ShareUpdateExclusiveLock;
+ break;
+
+ default: /* oops */
+ elog(ERROR, "unrecognized alter table type: %d",
+ (int) cmd->subtype);
+ break;
+ }
+
+ /*
+ * Take the greatest lockmode from any subcommand
+ */
+ if (cmd_lockmode > lockmode)
+ lockmode = cmd_lockmode;
+ }
+
+ return lockmode;
}
static void
-ATController(Relation rel, List *cmds, bool recurse)
+ATController(Relation rel, List *cmds, bool recurse, LOCKMODE lockmode)
{
List *wqueue = NIL;
ListCell *lcmd;
@@ -2407,17 +2581,17 @@ ATController(Relation rel, List *cmds, bool recurse)
{
AlterTableCmd *cmd = (AlterTableCmd *) lfirst(lcmd);
- ATPrepCmd(&wqueue, rel, cmd, recurse, false);
+ ATPrepCmd(&wqueue, rel, cmd, recurse, false, lockmode);
}
/* Close the relation, but keep lock until commit */
relation_close(rel, NoLock);
/* Phase 2: update system catalogs */
- ATRewriteCatalogs(&wqueue);
+ ATRewriteCatalogs(&wqueue, lockmode);
/* Phase 3: scan/rewrite tables as needed */
- ATRewriteTables(&wqueue);
+ ATRewriteTables(&wqueue, lockmode);
}
/*
@@ -2426,12 +2600,12 @@ ATController(Relation rel, List *cmds, bool recurse)
* Traffic cop for ALTER TABLE Phase 1 operations, including simple
* recursion and permission checks.
*
- * Caller must have acquired AccessExclusiveLock on relation already.
+ * Caller must have acquired appropriate lock type on relation already.
* This lock should be held until commit.
*/
static void
ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
- bool recurse, bool recursing)
+ bool recurse, bool recursing, LOCKMODE lockmode)
{
AlteredTableInfo *tab;
int pass;
@@ -2456,14 +2630,14 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
case AT_AddColumn: /* ADD COLUMN */
ATSimplePermissions(rel, false);
/* Performs own recursion */
- ATPrepAddColumn(wqueue, rel, recurse, cmd);
+ ATPrepAddColumn(wqueue, rel, recurse, cmd, lockmode);
pass = AT_PASS_ADD_COL;
break;
case AT_AddColumnToView: /* add column via CREATE OR REPLACE
* VIEW */
ATSimplePermissions(rel, true);
/* Performs own recursion */
- ATPrepAddColumn(wqueue, rel, recurse, cmd);
+ ATPrepAddColumn(wqueue, rel, recurse, cmd, lockmode);
pass = AT_PASS_ADD_COL;
break;
case AT_ColumnDefault: /* ALTER COLUMN DEFAULT */
@@ -2475,26 +2649,26 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
* rules.
*/
ATSimplePermissions(rel, true);
- ATSimpleRecursion(wqueue, rel, cmd, recurse);
+ ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
/* No command-specific prep needed */
pass = cmd->def ? AT_PASS_ADD_CONSTR : AT_PASS_DROP;
break;
case AT_DropNotNull: /* ALTER COLUMN DROP NOT NULL */
ATSimplePermissions(rel, false);
- ATSimpleRecursion(wqueue, rel, cmd, recurse);
+ ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
/* No command-specific prep needed */
pass = AT_PASS_DROP;
break;
case AT_SetNotNull: /* ALTER COLUMN SET NOT NULL */
ATSimplePermissions(rel, false);
- ATSimpleRecursion(wqueue, rel, cmd, recurse);
+ ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
/* No command-specific prep needed */
pass = AT_PASS_ADD_CONSTR;
break;
case AT_SetStatistics: /* ALTER COLUMN SET STATISTICS */
- ATSimpleRecursion(wqueue, rel, cmd, recurse);
+ ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
/* Performs own permission checks */
- ATPrepSetStatistics(rel, cmd->name, cmd->def);
+ ATPrepSetStatistics(rel, cmd->name, cmd->def, lockmode);
pass = AT_PASS_COL_ATTRS;
break;
case AT_SetOptions: /* ALTER COLUMN SET ( options ) */
@@ -2505,7 +2679,7 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
break;
case AT_SetStorage: /* ALTER COLUMN SET STORAGE */
ATSimplePermissions(rel, false);
- ATSimpleRecursion(wqueue, rel, cmd, recurse);
+ ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
/* No command-specific prep needed */
pass = AT_PASS_COL_ATTRS;
break;
@@ -2540,7 +2714,7 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
case AT_AlterColumnType: /* ALTER COLUMN TYPE */
ATSimplePermissions(rel, false);
/* Performs own recursion */
- ATPrepAlterColumnType(wqueue, tab, rel, recurse, recursing, cmd);
+ ATPrepAlterColumnType(wqueue, tab, rel, recurse, recursing, cmd, lockmode);
pass = AT_PASS_ALTER_TYPE;
break;
case AT_ChangeOwner: /* ALTER OWNER */
@@ -2559,7 +2733,7 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
ATSimplePermissions(rel, false);
/* Performs own recursion */
if (!rel->rd_rel->relhasoids || recursing)
- ATPrepAddOids(wqueue, rel, recurse, cmd);
+ ATPrepAddOids(wqueue, rel, recurse, cmd, lockmode);
pass = AT_PASS_ADD_COL;
break;
case AT_DropOids: /* SET WITHOUT OIDS */
@@ -2572,14 +2746,14 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
dropCmd->subtype = AT_DropColumn;
dropCmd->name = pstrdup("oid");
dropCmd->behavior = cmd->behavior;
- ATPrepCmd(wqueue, rel, dropCmd, recurse, false);
+ ATPrepCmd(wqueue, rel, dropCmd, recurse, false, lockmode);
}
pass = AT_PASS_DROP;
break;
case AT_SetTableSpace: /* SET TABLESPACE */
ATSimplePermissionsRelationOrIndex(rel);
/* This command never recurses */
- ATPrepSetTableSpace(tab, rel, cmd->name);
+ ATPrepSetTableSpace(tab, rel, cmd->name, lockmode);
pass = AT_PASS_MISC; /* doesn't actually matter */
break;
case AT_SetRelOptions: /* SET (...) */
@@ -2632,7 +2806,7 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
* conflicts).
*/
static void
-ATRewriteCatalogs(List **wqueue)
+ATRewriteCatalogs(List **wqueue, LOCKMODE lockmode)
{
int pass;
ListCell *ltab;
@@ -2658,12 +2832,12 @@ ATRewriteCatalogs(List **wqueue)
continue;
/*
- * Exclusive lock was obtained by phase 1, needn't get it again
+ * Appropriate lock was obtained by phase 1, needn't get it again
*/
rel = relation_open(tab->relid, NoLock);
foreach(lcmd, subcmds)
- ATExecCmd(wqueue, tab, rel, (AlterTableCmd *) lfirst(lcmd));
+ ATExecCmd(wqueue, tab, rel, (AlterTableCmd *) lfirst(lcmd), lockmode);
/*
* After the ALTER TYPE pass, do cleanup work (this is not done in
@@ -2671,7 +2845,7 @@ ATRewriteCatalogs(List **wqueue)
* multiple columns of a table are altered).
*/
if (pass == AT_PASS_ALTER_TYPE)
- ATPostAlterTypeCleanup(wqueue, tab);
+ ATPostAlterTypeCleanup(wqueue, tab, lockmode);
relation_close(rel, NoLock);
}
@@ -2698,86 +2872,86 @@ ATRewriteCatalogs(List **wqueue)
*/
static void
ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
- AlterTableCmd *cmd)
+ AlterTableCmd *cmd, LOCKMODE lockmode)
{
switch (cmd->subtype)
{
case AT_AddColumn: /* ADD COLUMN */
case AT_AddColumnToView: /* add column via CREATE OR REPLACE
* VIEW */
- ATExecAddColumn(tab, rel, (ColumnDef *) cmd->def, false);
+ ATExecAddColumn(tab, rel, (ColumnDef *) cmd->def, false, lockmode);
break;
case AT_ColumnDefault: /* ALTER COLUMN DEFAULT */
- ATExecColumnDefault(rel, cmd->name, cmd->def);
+ ATExecColumnDefault(rel, cmd->name, cmd->def, lockmode);
break;
case AT_DropNotNull: /* ALTER COLUMN DROP NOT NULL */
- ATExecDropNotNull(rel, cmd->name);
+ ATExecDropNotNull(rel, cmd->name, lockmode);
break;
case AT_SetNotNull: /* ALTER COLUMN SET NOT NULL */
- ATExecSetNotNull(tab, rel, cmd->name);
+ ATExecSetNotNull(tab, rel, cmd->name, lockmode);
break;
case AT_SetStatistics: /* ALTER COLUMN SET STATISTICS */
- ATExecSetStatistics(rel, cmd->name, cmd->def);
+ ATExecSetStatistics(rel, cmd->name, cmd->def, lockmode);
break;
case AT_SetOptions: /* ALTER COLUMN SET ( options ) */
- ATExecSetOptions(rel, cmd->name, cmd->def, false);
+ ATExecSetOptions(rel, cmd->name, cmd->def, false, lockmode);
break;
case AT_ResetOptions: /* ALTER COLUMN RESET ( options ) */
- ATExecSetOptions(rel, cmd->name, cmd->def, true);
+ ATExecSetOptions(rel, cmd->name, cmd->def, true, lockmode);
break;
case AT_SetStorage: /* ALTER COLUMN SET STORAGE */
- ATExecSetStorage(rel, cmd->name, cmd->def);
+ ATExecSetStorage(rel, cmd->name, cmd->def, lockmode);
break;
case AT_DropColumn: /* DROP COLUMN */
ATExecDropColumn(wqueue, rel, cmd->name,
- cmd->behavior, false, false, cmd->missing_ok);
+ cmd->behavior, false, false, cmd->missing_ok, lockmode);
break;
case AT_DropColumnRecurse: /* DROP COLUMN with recursion */
ATExecDropColumn(wqueue, rel, cmd->name,
- cmd->behavior, true, false, cmd->missing_ok);
+ cmd->behavior, true, false, cmd->missing_ok, lockmode);
break;
case AT_AddIndex: /* ADD INDEX */
- ATExecAddIndex(tab, rel, (IndexStmt *) cmd->def, false);
+ ATExecAddIndex(tab, rel, (IndexStmt *) cmd->def, false, lockmode);
break;
case AT_ReAddIndex: /* ADD INDEX */
- ATExecAddIndex(tab, rel, (IndexStmt *) cmd->def, true);
+ ATExecAddIndex(tab, rel, (IndexStmt *) cmd->def, true, lockmode);
break;
case AT_AddConstraint: /* ADD CONSTRAINT */
ATExecAddConstraint(wqueue, tab, rel, (Constraint *) cmd->def,
- false);
+ false, lockmode);
break;
case AT_AddConstraintRecurse: /* ADD CONSTRAINT with recursion */
ATExecAddConstraint(wqueue, tab, rel, (Constraint *) cmd->def,
- true);
+ true, lockmode);
break;
case AT_DropConstraint: /* DROP CONSTRAINT */
ATExecDropConstraint(rel, cmd->name, cmd->behavior,
false, false,
- cmd->missing_ok);
+ cmd->missing_ok, lockmode);
break;
case AT_DropConstraintRecurse: /* DROP CONSTRAINT with recursion */
ATExecDropConstraint(rel, cmd->name, cmd->behavior,
true, false,
- cmd->missing_ok);
+ cmd->missing_ok, lockmode);
break;
case AT_AlterColumnType: /* ALTER COLUMN TYPE */
- ATExecAlterColumnType(tab, rel, cmd->name, (TypeName *) cmd->def);
+ ATExecAlterColumnType(tab, rel, cmd->name, (TypeName *) cmd->def, lockmode);
break;
case AT_ChangeOwner: /* ALTER OWNER */
ATExecChangeOwner(RelationGetRelid(rel),
get_roleid_checked(cmd->name),
- false);
+ false, lockmode);
break;
case AT_ClusterOn: /* CLUSTER ON */
- ATExecClusterOn(rel, cmd->name);
+ ATExecClusterOn(rel, cmd->name, lockmode);
break;
case AT_DropCluster: /* SET WITHOUT CLUSTER */
- ATExecDropCluster(rel);
+ ATExecDropCluster(rel, lockmode);
break;
case AT_AddOids: /* SET WITH OIDS */
/* Use the ADD COLUMN code, unless prep decided to do nothing */
if (cmd->def != NULL)
- ATExecAddColumn(tab, rel, (ColumnDef *) cmd->def, true);
+ ATExecAddColumn(tab, rel, (ColumnDef *) cmd->def, true, lockmode);
break;
case AT_DropOids: /* SET WITHOUT OIDS */
@@ -2793,67 +2967,67 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
*/
break;
case AT_SetRelOptions: /* SET (...) */
- ATExecSetRelOptions(rel, (List *) cmd->def, false);
+ ATExecSetRelOptions(rel, (List *) cmd->def, false, lockmode);
break;
case AT_ResetRelOptions: /* RESET (...) */
- ATExecSetRelOptions(rel, (List *) cmd->def, true);
+ ATExecSetRelOptions(rel, (List *) cmd->def, true, lockmode);
break;
case AT_EnableTrig: /* ENABLE TRIGGER name */
ATExecEnableDisableTrigger(rel, cmd->name,
- TRIGGER_FIRES_ON_ORIGIN, false);
+ TRIGGER_FIRES_ON_ORIGIN, false, lockmode);
break;
case AT_EnableAlwaysTrig: /* ENABLE ALWAYS TRIGGER name */
ATExecEnableDisableTrigger(rel, cmd->name,
- TRIGGER_FIRES_ALWAYS, false);
+ TRIGGER_FIRES_ALWAYS, false, lockmode);
break;
case AT_EnableReplicaTrig: /* ENABLE REPLICA TRIGGER name */
ATExecEnableDisableTrigger(rel, cmd->name,
- TRIGGER_FIRES_ON_REPLICA, false);
+ TRIGGER_FIRES_ON_REPLICA, false, lockmode);
break;
case AT_DisableTrig: /* DISABLE TRIGGER name */
ATExecEnableDisableTrigger(rel, cmd->name,
- TRIGGER_DISABLED, false);
+ TRIGGER_DISABLED, false, lockmode);
break;
case AT_EnableTrigAll: /* ENABLE TRIGGER ALL */
ATExecEnableDisableTrigger(rel, NULL,
- TRIGGER_FIRES_ON_ORIGIN, false);
+ TRIGGER_FIRES_ON_ORIGIN, false, lockmode);
break;
case AT_DisableTrigAll: /* DISABLE TRIGGER ALL */
ATExecEnableDisableTrigger(rel, NULL,
- TRIGGER_DISABLED, false);
+ TRIGGER_DISABLED, false, lockmode);
break;
case AT_EnableTrigUser: /* ENABLE TRIGGER USER */
ATExecEnableDisableTrigger(rel, NULL,
- TRIGGER_FIRES_ON_ORIGIN, true);
+ TRIGGER_FIRES_ON_ORIGIN, true, lockmode);
break;
case AT_DisableTrigUser: /* DISABLE TRIGGER USER */
ATExecEnableDisableTrigger(rel, NULL,
- TRIGGER_DISABLED, true);
+ TRIGGER_DISABLED, true, lockmode);
break;
case AT_EnableRule: /* ENABLE RULE name */
ATExecEnableDisableRule(rel, cmd->name,
- RULE_FIRES_ON_ORIGIN);
+ RULE_FIRES_ON_ORIGIN, lockmode);
break;
case AT_EnableAlwaysRule: /* ENABLE ALWAYS RULE name */
ATExecEnableDisableRule(rel, cmd->name,
- RULE_FIRES_ALWAYS);
+ RULE_FIRES_ALWAYS, lockmode);
break;
case AT_EnableReplicaRule: /* ENABLE REPLICA RULE name */
ATExecEnableDisableRule(rel, cmd->name,
- RULE_FIRES_ON_REPLICA);
+ RULE_FIRES_ON_REPLICA, lockmode);
break;
case AT_DisableRule: /* DISABLE RULE name */
ATExecEnableDisableRule(rel, cmd->name,
- RULE_DISABLED);
+ RULE_DISABLED, lockmode);
break;
case AT_AddInherit:
- ATExecAddInherit(rel, (RangeVar *) cmd->def);
+ ATExecAddInherit(rel, (RangeVar *) cmd->def, lockmode);
break;
case AT_DropInherit:
- ATExecDropInherit(rel, (RangeVar *) cmd->def);
+ ATExecDropInherit(rel, (RangeVar *) cmd->def, lockmode);
break;
default: /* oops */
elog(ERROR, "unrecognized alter table type: %d",
@@ -2872,7 +3046,7 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
* ATRewriteTables: ALTER TABLE phase 3
*/
static void
-ATRewriteTables(List **wqueue)
+ATRewriteTables(List **wqueue, LOCKMODE lockmode)
{
ListCell *ltab;
@@ -2933,7 +3107,7 @@ ATRewriteTables(List **wqueue)
* modifications, and test the current data within the table
* against new constraints generated by ALTER TABLE commands.
*/
- ATRewriteTable(tab, OIDNewHeap);
+ ATRewriteTable(tab, OIDNewHeap, lockmode);
/*
* Swap the physical files of the old and new heaps, then rebuild
@@ -2952,14 +3126,14 @@ ATRewriteTables(List **wqueue)
* generated by ALTER TABLE commands, but don't rebuild data.
*/
if (tab->constraints != NIL || tab->new_notnull)
- ATRewriteTable(tab, InvalidOid);
+ ATRewriteTable(tab, InvalidOid, lockmode);
/*
* If we had SET TABLESPACE but no reason to reconstruct tuples,
* just do a block-by-block copy.
*/
if (tab->newTableSpace)
- ATExecSetTableSpace(tab->relid, tab->newTableSpace);
+ ATExecSetTableSpace(tab->relid, tab->newTableSpace, lockmode);
}
}
@@ -2986,12 +3160,14 @@ ATRewriteTables(List **wqueue)
Relation refrel;
if (rel == NULL)
- {
/* Long since locked, no need for another */
rel = heap_open(tab->relid, NoLock);
- }
- refrel = heap_open(con->refrelid, RowShareLock);
+ /*
+ * We're adding a trigger to both tables, so the lock level
+ * here should sensibly reflect that.
+ */
+ refrel = heap_open(con->refrelid, ShareRowExclusiveLock);
validateForeignKeyConstraint(fkconstraint, rel, refrel,
con->refindid,
@@ -3012,7 +3188,7 @@ ATRewriteTables(List **wqueue)
* OIDNewHeap is InvalidOid if we don't need to rewrite
*/
static void
-ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap)
+ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode)
{
Relation oldrel;
Relation newrel;
@@ -3036,7 +3212,7 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap)
newTupDesc = RelationGetDescr(oldrel); /* includes all mods */
if (OidIsValid(OIDNewHeap))
- newrel = heap_open(OIDNewHeap, AccessExclusiveLock);
+ newrel = heap_open(OIDNewHeap, lockmode);
else
newrel = NULL;
@@ -3402,7 +3578,7 @@ ATSimplePermissionsRelationOrIndex(Relation rel)
*/
static void
ATSimpleRecursion(List **wqueue, Relation rel,
- AlterTableCmd *cmd, bool recurse)
+ AlterTableCmd *cmd, bool recurse, LOCKMODE lockmode)
{
/*
* Propagate to children if desired. Non-table relations never have
@@ -3414,7 +3590,7 @@ ATSimpleRecursion(List **wqueue, Relation rel,
ListCell *child;
List *children;
- children = find_all_inheritors(relid, AccessExclusiveLock, NULL);
+ children = find_all_inheritors(relid, lockmode, NULL);
/*
* find_all_inheritors does the recursive search of the inheritance
@@ -3431,7 +3607,7 @@ ATSimpleRecursion(List **wqueue, Relation rel,
/* find_all_inheritors already got lock */
childrel = relation_open(childrelid, NoLock);
CheckTableNotInUse(childrel, "ALTER TABLE");
- ATPrepCmd(wqueue, childrel, cmd, false, true);
+ ATPrepCmd(wqueue, childrel, cmd, false, true, lockmode);
relation_close(childrel, NoLock);
}
}
@@ -3447,13 +3623,13 @@ ATSimpleRecursion(List **wqueue, Relation rel,
*/
static void
ATOneLevelRecursion(List **wqueue, Relation rel,
- AlterTableCmd *cmd)
+ AlterTableCmd *cmd, LOCKMODE lockmode)
{
Oid relid = RelationGetRelid(rel);
ListCell *child;
List *children;
- children = find_inheritance_children(relid, AccessExclusiveLock);
+ children = find_inheritance_children(relid, lockmode);
foreach(child, children)
{
@@ -3463,7 +3639,7 @@ ATOneLevelRecursion(List **wqueue, Relation rel,
/* find_inheritance_children already got lock */
childrel = relation_open(childrelid, NoLock);
CheckTableNotInUse(childrel, "ALTER TABLE");
- ATPrepCmd(wqueue, childrel, cmd, true, true);
+ ATPrepCmd(wqueue, childrel, cmd, true, true, lockmode);
relation_close(childrel, NoLock);
}
}
@@ -3581,7 +3757,7 @@ find_composite_type_dependencies(Oid typeOid,
*/
static void
ATPrepAddColumn(List **wqueue, Relation rel, bool recurse,
- AlterTableCmd *cmd)
+ AlterTableCmd *cmd, LOCKMODE lockmode)
{
if (rel->rd_rel->reloftype)
ereport(ERROR,
@@ -3604,7 +3780,7 @@ ATPrepAddColumn(List **wqueue, Relation rel, bool recurse,
colDefChild->inhcount = 1;
colDefChild->is_local = false;
- ATOneLevelRecursion(wqueue, rel, childCmd);
+ ATOneLevelRecursion(wqueue, rel, childCmd, lockmode);
}
else
{
@@ -3621,7 +3797,7 @@ ATPrepAddColumn(List **wqueue, Relation rel, bool recurse,
static void
ATExecAddColumn(AlteredTableInfo *tab, Relation rel,
- ColumnDef *colDef, bool isOid)
+ ColumnDef *colDef, bool isOid, LOCKMODE lockmode)
{
Oid myrelid = RelationGetRelid(rel);
Relation pgclass,
@@ -3902,7 +4078,7 @@ add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid)
* to cons up a ColumnDef node because the ADD COLUMN code needs one.
*/
static void
-ATPrepAddOids(List **wqueue, Relation rel, bool recurse, AlterTableCmd *cmd)
+ATPrepAddOids(List **wqueue, Relation rel, bool recurse, AlterTableCmd *cmd, LOCKMODE lockmode)
{
/* If we're recursing to a child table, the ColumnDef is already set up */
if (cmd->def == NULL)
@@ -3917,14 +4093,14 @@ ATPrepAddOids(List **wqueue, Relation rel, bool recurse, AlterTableCmd *cmd)
cdef->storage = 0;
cmd->def = (Node *) cdef;
}
- ATPrepAddColumn(wqueue, rel, recurse, cmd);
+ ATPrepAddColumn(wqueue, rel, recurse, cmd, lockmode);
}
/*
* ALTER TABLE ALTER COLUMN DROP NOT NULL
*/
static void
-ATExecDropNotNull(Relation rel, const char *colName)
+ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode)
{
HeapTuple tuple;
AttrNumber attnum;
@@ -4016,7 +4192,7 @@ ATExecDropNotNull(Relation rel, const char *colName)
*/
static void
ATExecSetNotNull(AlteredTableInfo *tab, Relation rel,
- const char *colName)
+ const char *colName, LOCKMODE lockmode)
{
HeapTuple tuple;
AttrNumber attnum;
@@ -4068,7 +4244,7 @@ ATExecSetNotNull(AlteredTableInfo *tab, Relation rel,
*/
static void
ATExecColumnDefault(Relation rel, const char *colName,
- Node *newDefault)
+ Node *newDefault, LOCKMODE lockmode)
{
AttrNumber attnum;
@@ -4117,7 +4293,7 @@ ATExecColumnDefault(Relation rel, const char *colName,
* ALTER TABLE ALTER COLUMN SET STATISTICS
*/
static void
-ATPrepSetStatistics(Relation rel, const char *colName, Node *newValue)
+ATPrepSetStatistics(Relation rel, const char *colName, Node *newValue, LOCKMODE lockmode)
{
/*
* We do our own permission checking because (a) we want to allow SET
@@ -4139,7 +4315,7 @@ ATPrepSetStatistics(Relation rel, const char *colName, Node *newValue)
}
static void
-ATExecSetStatistics(Relation rel, const char *colName, Node *newValue)
+ATExecSetStatistics(Relation rel, const char *colName, Node *newValue, LOCKMODE lockmode)
{
int newtarget;
Relation attrelation;
@@ -4199,7 +4375,7 @@ ATExecSetStatistics(Relation rel, const char *colName, Node *newValue)
static void
ATExecSetOptions(Relation rel, const char *colName, Node *options,
- bool isReset)
+ bool isReset, LOCKMODE lockmode)
{
Relation attrelation;
HeapTuple tuple,
@@ -4263,7 +4439,7 @@ ATExecSetOptions(Relation rel, const char *colName, Node *options,
* ALTER TABLE ALTER COLUMN SET STORAGE
*/
static void
-ATExecSetStorage(Relation rel, const char *colName, Node *newValue)
+ATExecSetStorage(Relation rel, const char *colName, Node *newValue, LOCKMODE lockmode)
{
char *storagemode;
char newstorage;
@@ -4357,7 +4533,7 @@ static void
ATExecDropColumn(List **wqueue, Relation rel, const char *colName,
DropBehavior behavior,
bool recurse, bool recursing,
- bool missing_ok)
+ bool missing_ok, LOCKMODE lockmode)
{
HeapTuple tuple;
Form_pg_attribute targetatt;
@@ -4415,8 +4591,7 @@ ATExecDropColumn(List **wqueue, Relation rel, const char *colName,
* routines, we have to do this one level of recursion at a time; we can't
* use find_all_inheritors to do it in one pass.
*/
- children = find_inheritance_children(RelationGetRelid(rel),
- AccessExclusiveLock);
+ children = find_inheritance_children(RelationGetRelid(rel), lockmode);
if (children)
{
@@ -4456,7 +4631,7 @@ ATExecDropColumn(List **wqueue, Relation rel, const char *colName,
/* Time to delete this child column, too */
ATExecDropColumn(wqueue, childrel, colName,
behavior, true, true,
- false);
+ false, lockmode);
}
else
{
@@ -4551,7 +4726,7 @@ ATExecDropColumn(List **wqueue, Relation rel, const char *colName,
*/
static void
ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
- IndexStmt *stmt, bool is_rebuild)
+ IndexStmt *stmt, bool is_rebuild, LOCKMODE lockmode)
{
bool check_rights;
bool skip_build;
@@ -4594,7 +4769,7 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
*/
static void
ATExecAddConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
- Constraint *newConstraint, bool recurse)
+ Constraint *newConstraint, bool recurse, LOCKMODE lockmode)
{
Assert(IsA(newConstraint, Constraint));
@@ -4607,7 +4782,7 @@ ATExecAddConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
{
case CONSTR_CHECK:
ATAddCheckConstraint(wqueue, tab, rel,
- newConstraint, recurse, false);
+ newConstraint, recurse, false, lockmode);
break;
case CONSTR_FOREIGN:
@@ -4638,7 +4813,7 @@ ATExecAddConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
RelationGetNamespace(rel),
NIL);
- ATAddForeignKeyConstraint(tab, rel, newConstraint);
+ ATAddForeignKeyConstraint(tab, rel, newConstraint, lockmode);
break;
default:
@@ -4662,7 +4837,8 @@ ATExecAddConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
*/
static void
ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
- Constraint *constr, bool recurse, bool recursing)
+ Constraint *constr, bool recurse, bool recursing,
+ LOCKMODE lockmode)
{
List *newcons;
ListCell *lcon;
@@ -4717,8 +4893,7 @@ ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
* routines, we have to do this one level of recursion at a time; we can't
* use find_all_inheritors to do it in one pass.
*/
- children = find_inheritance_children(RelationGetRelid(rel),
- AccessExclusiveLock);
+ children = find_inheritance_children(RelationGetRelid(rel), lockmode);
/*
* If we are told not to recurse, there had better not be any child
@@ -4744,7 +4919,7 @@ ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
/* Recurse to child */
ATAddCheckConstraint(wqueue, childtab, childrel,
- constr, recurse, true);
+ constr, recurse, true, lockmode);
heap_close(childrel, NoLock);
}
@@ -4759,7 +4934,7 @@ ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
*/
static void
ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
- Constraint *fkconstraint)
+ Constraint *fkconstraint, LOCKMODE lockmode)
{
Relation pkrel;
int16 pkattnum[INDEX_MAX_KEYS];
@@ -4776,14 +4951,7 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
Oid indexOid;
Oid constrOid;
- /*
- * Grab an exclusive lock on the pk table, so that someone doesn't delete
- * rows out from under us. (Although a lesser lock would do for that
- * purpose, we'll need exclusive lock anyway to add triggers to the pk
- * table; trying to start with a lesser lock will just create a risk of
- * deadlock.)
- */
- pkrel = heap_openrv(fkconstraint->pktable, AccessExclusiveLock);
+ pkrel = heap_openrv(fkconstraint->pktable, lockmode);
/*
* Validity checks (permission checks wait till we have the column
@@ -5347,7 +5515,7 @@ checkFkeyPermissions(Relation rel, int16 *attnums, int natts)
* Scan the existing rows in a table to verify they meet a proposed FK
* constraint.
*
- * Caller must have opened and locked both relations.
+ * Caller must have opened and locked both relations appropriately.
*/
static void
validateForeignKeyConstraint(Constraint *fkconstraint,
@@ -5599,7 +5767,7 @@ static void
ATExecDropConstraint(Relation rel, const char *constrName,
DropBehavior behavior,
bool recurse, bool recursing,
- bool missing_ok)
+ bool missing_ok, LOCKMODE lockmode)
{
List *children;
ListCell *child;
@@ -5686,8 +5854,7 @@ ATExecDropConstraint(Relation rel, const char *constrName,
* use find_all_inheritors to do it in one pass.
*/
if (is_check_constraint)
- children = find_inheritance_children(RelationGetRelid(rel),
- AccessExclusiveLock);
+ children = find_inheritance_children(RelationGetRelid(rel), lockmode);
else
children = NIL;
@@ -5743,7 +5910,7 @@ ATExecDropConstraint(Relation rel, const char *constrName,
/* Time to delete this child constraint, too */
ATExecDropConstraint(childrel, constrName, behavior,
true, true,
- false);
+ false, lockmode);
}
else
{
@@ -5798,7 +5965,7 @@ static void
ATPrepAlterColumnType(List **wqueue,
AlteredTableInfo *tab, Relation rel,
bool recurse, bool recursing,
- AlterTableCmd *cmd)
+ AlterTableCmd *cmd, LOCKMODE lockmode)
{
char *colName = cmd->name;
TypeName *typeName = (TypeName *) cmd->def;
@@ -5925,7 +6092,7 @@ ATPrepAlterColumnType(List **wqueue,
* alter would put them out of step.
*/
if (recurse)
- ATSimpleRecursion(wqueue, rel, cmd, recurse);
+ ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
else if (!recursing &&
find_inheritance_children(RelationGetRelid(rel), NoLock) != NIL)
ereport(ERROR,
@@ -5936,7 +6103,7 @@ ATPrepAlterColumnType(List **wqueue,
static void
ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
- const char *colName, TypeName *typeName)
+ const char *colName, TypeName *typeName, LOCKMODE lockmode)
{
HeapTuple heapTup;
Form_pg_attribute attTup;
@@ -6284,7 +6451,7 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
* and constraints that depend on the altered columns.
*/
static void
-ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab)
+ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab, LOCKMODE lockmode)
{
ObjectAddress obj;
ListCell *l;
@@ -6298,9 +6465,9 @@ ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab)
* at the catalogs to detect the existing entry.
*/
foreach(l, tab->changedIndexDefs)
- ATPostAlterTypeParse((char *) lfirst(l), wqueue);
+ ATPostAlterTypeParse((char *) lfirst(l), wqueue, lockmode);
foreach(l, tab->changedConstraintDefs)
- ATPostAlterTypeParse((char *) lfirst(l), wqueue);
+ ATPostAlterTypeParse((char *) lfirst(l), wqueue, lockmode);
/*
* Now we can drop the existing constraints and indexes --- constraints
@@ -6333,7 +6500,7 @@ ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab)
}
static void
-ATPostAlterTypeParse(char *cmd, List **wqueue)
+ATPostAlterTypeParse(char *cmd, List **wqueue, LOCKMODE lockmode)
{
List *raw_parsetree_list;
List *querytree_list;
@@ -6380,7 +6547,7 @@ ATPostAlterTypeParse(char *cmd, List **wqueue)
IndexStmt *stmt = (IndexStmt *) stm;
AlterTableCmd *newcmd;
- rel = relation_openrv(stmt->relation, AccessExclusiveLock);
+ rel = relation_openrv(stmt->relation, lockmode);
tab = ATGetQueueEntry(wqueue, rel);
newcmd = makeNode(AlterTableCmd);
newcmd->subtype = AT_ReAddIndex;
@@ -6395,7 +6562,7 @@ ATPostAlterTypeParse(char *cmd, List **wqueue)
AlterTableStmt *stmt = (AlterTableStmt *) stm;
ListCell *lcmd;
- rel = relation_openrv(stmt->relation, AccessExclusiveLock);
+ rel = relation_openrv(stmt->relation, lockmode);
tab = ATGetQueueEntry(wqueue, rel);
foreach(lcmd, stmt->cmds)
{
@@ -6441,7 +6608,7 @@ ATPostAlterTypeParse(char *cmd, List **wqueue)
* free-standing composite type.
*/
void
-ATExecChangeOwner(Oid relationOid, Oid newOwnerId, bool recursing)
+ATExecChangeOwner(Oid relationOid, Oid newOwnerId, bool recursing, LOCKMODE lockmode)
{
Relation target_rel;
Relation class_rel;
@@ -6452,7 +6619,7 @@ ATExecChangeOwner(Oid relationOid, Oid newOwnerId, bool recursing)
* Get exclusive lock till end of transaction on the target table. Use
* relation_open so that we can work on indexes and sequences.
*/
- target_rel = relation_open(relationOid, AccessExclusiveLock);
+ target_rel = relation_open(relationOid, lockmode);
/* Get its pg_class tuple, too */
class_rel = heap_open(RelationRelationId, RowExclusiveLock);
@@ -6629,7 +6796,7 @@ ATExecChangeOwner(Oid relationOid, Oid newOwnerId, bool recursing)
/* For each index, recursively change its ownership */
foreach(i, index_oid_list)
- ATExecChangeOwner(lfirst_oid(i), newOwnerId, true);
+ ATExecChangeOwner(lfirst_oid(i), newOwnerId, true, lockmode);
list_free(index_oid_list);
}
@@ -6639,10 +6806,10 @@ ATExecChangeOwner(Oid relationOid, Oid newOwnerId, bool recursing)
/* If it has a toast table, recurse to change its ownership */
if (tuple_class->reltoastrelid != InvalidOid)
ATExecChangeOwner(tuple_class->reltoastrelid, newOwnerId,
- true);
+ true, lockmode);
/* If it has dependent sequences, recurse to change them too */
- change_owner_recurse_to_sequences(relationOid, newOwnerId);
+ change_owner_recurse_to_sequences(relationOid, newOwnerId, lockmode);
}
}
@@ -6659,7 +6826,7 @@ ATExecChangeOwner(Oid relationOid, Oid newOwnerId, bool recursing)
* ownership.
*/
static void
-change_owner_recurse_to_sequences(Oid relationOid, Oid newOwnerId)
+change_owner_recurse_to_sequences(Oid relationOid, Oid newOwnerId, LOCKMODE lockmode)
{
Relation depRel;
SysScanDesc scan;
@@ -6698,18 +6865,18 @@ change_owner_recurse_to_sequences(Oid relationOid, Oid newOwnerId)
continue;
/* Use relation_open just in case it's an index */
- seqRel = relation_open(depForm->objid, AccessExclusiveLock);
+ seqRel = relation_open(depForm->objid, lockmode);
/* skip non-sequence relations */
if (RelationGetForm(seqRel)->relkind != RELKIND_SEQUENCE)
{
/* No need to keep the lock */
- relation_close(seqRel, AccessExclusiveLock);
+ relation_close(seqRel, lockmode);
continue;
}
/* We don't need to close the sequence while we alter it. */
- ATExecChangeOwner(depForm->objid, newOwnerId, true);
+ ATExecChangeOwner(depForm->objid, newOwnerId, true, lockmode);
/* Now we can close it. Keep the lock till end of transaction. */
relation_close(seqRel, NoLock);
@@ -6726,7 +6893,7 @@ change_owner_recurse_to_sequences(Oid relationOid, Oid newOwnerId)
* The only thing we have to do is to change the indisclustered bits.
*/
static void
-ATExecClusterOn(Relation rel, const char *indexName)
+ATExecClusterOn(Relation rel, const char *indexName, LOCKMODE lockmode)
{
Oid indexOid;
@@ -6752,7 +6919,7 @@ ATExecClusterOn(Relation rel, const char *indexName)
* set and turn it off.
*/
static void
-ATExecDropCluster(Relation rel)
+ATExecDropCluster(Relation rel, LOCKMODE lockmode)
{
mark_index_clustered(rel, InvalidOid);
}
@@ -6761,7 +6928,7 @@ ATExecDropCluster(Relation rel)
* ALTER TABLE SET TABLESPACE
*/
static void
-ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel, char *tablespacename)
+ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel, char *tablespacename, LOCKMODE lockmode)
{
Oid tablespaceId;
AclResult aclresult;
@@ -6790,7 +6957,7 @@ ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel, char *tablespacename)
* ALTER TABLE/INDEX SET (...) or RESET (...)
*/
static void
-ATExecSetRelOptions(Relation rel, List *defList, bool isReset)
+ATExecSetRelOptions(Relation rel, List *defList, bool isReset, LOCKMODE lockmode)
{
Oid relid;
Relation pgclass;
@@ -6871,7 +7038,7 @@ ATExecSetRelOptions(Relation rel, List *defList, bool isReset)
Relation toastrel;
Oid toastid = rel->rd_rel->reltoastrelid;
- toastrel = heap_open(toastid, AccessExclusiveLock);
+ toastrel = heap_open(toastid, lockmode);
/* Get the old reloptions */
tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(toastid));
@@ -6918,7 +7085,7 @@ ATExecSetRelOptions(Relation rel, List *defList, bool isReset)
* rewriting to be done, so we just want to copy the data as fast as possible.
*/
static void
-ATExecSetTableSpace(Oid tableOid, Oid newTableSpace)
+ATExecSetTableSpace(Oid tableOid, Oid newTableSpace, LOCKMODE lockmode)
{
Relation rel;
Oid oldTableSpace;
@@ -6935,7 +7102,7 @@ ATExecSetTableSpace(Oid tableOid, Oid newTableSpace)
/*
* Need lock here in case we are recursing to toast table or index
*/
- rel = relation_open(tableOid, AccessExclusiveLock);
+ rel = relation_open(tableOid, lockmode);
/*
* No work if no change in tablespace.
@@ -7049,9 +7216,9 @@ ATExecSetTableSpace(Oid tableOid, Oid newTableSpace)
/* Move associated toast relation and/or index, too */
if (OidIsValid(reltoastrelid))
- ATExecSetTableSpace(reltoastrelid, newTableSpace);
+ ATExecSetTableSpace(reltoastrelid, newTableSpace, lockmode);
if (OidIsValid(reltoastidxid))
- ATExecSetTableSpace(reltoastidxid, newTableSpace);
+ ATExecSetTableSpace(reltoastidxid, newTableSpace, lockmode);
}
/*
@@ -7119,7 +7286,7 @@ copy_relation_data(SMgrRelation src, SMgrRelation dst,
*/
static void
ATExecEnableDisableTrigger(Relation rel, char *trigname,
- char fires_when, bool skip_system)
+ char fires_when, bool skip_system, LOCKMODE lockmode)
{
EnableDisableTrigger(rel, trigname, fires_when, skip_system);
}
@@ -7131,7 +7298,7 @@ ATExecEnableDisableTrigger(Relation rel, char *trigname,
*/
static void
ATExecEnableDisableRule(Relation rel, char *trigname,
- char fires_when)
+ char fires_when, LOCKMODE lockmode)
{
EnableDisableRule(rel, trigname, fires_when);
}
@@ -7153,7 +7320,7 @@ ATPrepAddInherit(Relation child_rel)
}
static void
-ATExecAddInherit(Relation child_rel, RangeVar *parent)
+ATExecAddInherit(Relation child_rel, RangeVar *parent, LOCKMODE lockmode)
{
Relation parent_rel,
catalogRelation;
@@ -7514,7 +7681,7 @@ MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel)
* exactly the same way.
*/
static void
-ATExecDropInherit(Relation rel, RangeVar *parent)
+ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode)
{
Relation parent_rel;
Relation catalogRelation;
@@ -7738,7 +7905,7 @@ ATExecDropInherit(Relation rel, RangeVar *parent)
*/
void
AlterTableNamespace(RangeVar *relation, const char *newschema,
- ObjectType stmttype)
+ ObjectType stmttype, LOCKMODE lockmode)
{
Relation rel;
Oid relid;
@@ -7746,7 +7913,7 @@ AlterTableNamespace(RangeVar *relation, const char *newschema,
Oid nspOid;
Relation classRel;
- rel = relation_openrv(relation, AccessExclusiveLock);
+ rel = relation_openrv(relation, lockmode);
relid = RelationGetRelid(rel);
oldNspOid = RelationGetNamespace(rel);
@@ -7855,7 +8022,7 @@ AlterTableNamespace(RangeVar *relation, const char *newschema,
if (rel->rd_rel->relkind == RELKIND_RELATION)
{
AlterIndexNamespaces(classRel, rel, oldNspOid, nspOid);
- AlterSeqNamespaces(classRel, rel, oldNspOid, nspOid, newschema);
+ AlterSeqNamespaces(classRel, rel, oldNspOid, nspOid, newschema, lockmode);
AlterConstraintNamespaces(relid, oldNspOid, nspOid, false);
}
@@ -7951,7 +8118,7 @@ AlterIndexNamespaces(Relation classRel, Relation rel,
*/
static void
AlterSeqNamespaces(Relation classRel, Relation rel,
- Oid oldNspOid, Oid newNspOid, const char *newNspName)
+ Oid oldNspOid, Oid newNspOid, const char *newNspName, LOCKMODE lockmode)
{
Relation depRel;
SysScanDesc scan;
@@ -7990,13 +8157,13 @@ AlterSeqNamespaces(Relation classRel, Relation rel,
continue;
/* Use relation_open just in case it's an index */
- seqRel = relation_open(depForm->objid, AccessExclusiveLock);
+ seqRel = relation_open(depForm->objid, lockmode);
/* skip non-sequence relations */
if (RelationGetForm(seqRel)->relkind != RELKIND_SEQUENCE)
{
/* No need to keep the lock */
- relation_close(seqRel, AccessExclusiveLock);
+ relation_close(seqRel, lockmode);
continue;
}
diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index aef7838eb6e..643b7b3a6f1 100644
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -7,7 +7,7 @@
* Portions Copyright (c) 1994, Regents of the University of California
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/commands/trigger.c,v 1.262 2010/02/26 02:00:39 momjian Exp $
+ * $PostgreSQL: pgsql/src/backend/commands/trigger.c,v 1.263 2010/07/28 05:22:24 sriggs Exp $
*
*-------------------------------------------------------------------------
*/
@@ -141,7 +141,14 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
ObjectAddress myself,
referenced;
- rel = heap_openrv(stmt->relation, AccessExclusiveLock);
+ /*
+ * ShareRowExclusiveLock is sufficient to prevent concurrent write activity
+ * to the relation, and thus to lock out any operations that might want to
+ * fire triggers on the relation. If we had ON SELECT triggers we would
+ * need to take an AccessExclusiveLock to add one of those, just as we do
+ * with ON SELECT rules.
+ */
+ rel = heap_openrv(stmt->relation, ShareRowExclusiveLock);
if (rel->rd_rel->relkind != RELKIND_RELATION)
ereport(ERROR,
@@ -417,7 +424,7 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
* can skip this for internally generated triggers, since the name
* modification above should be sufficient.
*
- * NOTE that this is cool only because we have AccessExclusiveLock on the
+ * NOTE that this is cool only because we have ShareRowExclusiveLock on the
* relation, so the trigger set won't be changing underneath us.
*/
if (!isInternal)
@@ -1051,11 +1058,14 @@ RemoveTriggerById(Oid trigOid)
elog(ERROR, "could not find tuple for trigger %u", trigOid);
/*
- * Open and exclusive-lock the relation the trigger belongs to.
+ * Open and lock the relation the trigger belongs to. As in
+ * CreateTrigger, this is sufficient to lock out all operations that
+ * could fire or add triggers; but it would need to be revisited if
+ * we had ON SELECT triggers.
*/
relid = ((Form_pg_trigger) GETSTRUCT(tup))->tgrelid;
- rel = heap_open(relid, AccessExclusiveLock);
+ rel = heap_open(relid, ShareRowExclusiveLock);
if (rel->rd_rel->relkind != RELKIND_RELATION)
ereport(ERROR,
diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c
index 5d94445910c..19d61a72568 100644
--- a/src/backend/commands/typecmds.c
+++ b/src/backend/commands/typecmds.c
@@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/commands/typecmds.c,v 1.149 2010/07/25 23:21:21 rhaas Exp $
+ * $PostgreSQL: pgsql/src/backend/commands/typecmds.c,v 1.150 2010/07/28 05:22:24 sriggs Exp $
*
* DESCRIPTION
* The "DefineFoo" routines take the parse tree and pick out the
@@ -2638,7 +2638,7 @@ AlterTypeOwner(List *names, Oid newOwnerId)
* AlterTypeOwnerInternal to take care of the pg_type entry(s).
*/
if (typTup->typtype == TYPTYPE_COMPOSITE)
- ATExecChangeOwner(typTup->typrelid, newOwnerId, true);
+ ATExecChangeOwner(typTup->typrelid, newOwnerId, true, AccessExclusiveLock);
else
{
/*
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index 1657096c6d8..a7e875b380c 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -19,7 +19,7 @@
* Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
- * $PostgreSQL: pgsql/src/backend/parser/parse_utilcmd.c,v 2.40 2010/02/26 02:00:53 momjian Exp $
+ * $PostgreSQL: pgsql/src/backend/parser/parse_utilcmd.c,v 2.41 2010/07/28 05:22:24 sriggs Exp $
*
*-------------------------------------------------------------------------
*/
@@ -53,6 +53,7 @@
#include "parser/parse_utilcmd.h"
#include "parser/parser.h"
#include "rewrite/rewriteManip.h"
+#include "storage/lock.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
@@ -1528,7 +1529,7 @@ transformFKConstraints(ParseState *pstate, CreateStmtContext *cxt,
}
/*
- * transformIndexStmt - parse analysis for CREATE INDEX
+ * transformIndexStmt - parse analysis for CREATE INDEX and ALTER TABLE
*
* Note: this is a no-op for an index not using either index expressions or
* a predicate expression. There are several code paths that create indexes
@@ -1554,7 +1555,8 @@ transformIndexStmt(IndexStmt *stmt, const char *queryString)
* because addRangeTableEntry() would acquire only AccessShareLock,
* leaving DefineIndex() needing to do a lock upgrade with consequent risk
* of deadlock. Make sure this stays in sync with the type of lock
- * DefineIndex() wants.
+ * DefineIndex() wants. If we are being called by ALTER TABLE, we will
+ * already hold a higher lock.
*/
rel = heap_openrv(stmt->relation,
(stmt->concurrent ? ShareUpdateExclusiveLock : ShareLock));
@@ -1919,6 +1921,7 @@ transformAlterTableStmt(AlterTableStmt *stmt, const char *queryString)
List *newcmds = NIL;
bool skipValidation = true;
AlterTableCmd *newcmd;
+ LOCKMODE lockmode;
/*
* We must not scribble on the passed-in AlterTableStmt, so copy it. (This
@@ -1927,13 +1930,19 @@ transformAlterTableStmt(AlterTableStmt *stmt, const char *queryString)
stmt = (AlterTableStmt *) copyObject(stmt);
/*
- * Acquire exclusive lock on the target relation, which will be held until
+ * Assign the appropriate lock level for this list of subcommands.
+ */
+ lockmode = AlterTableGetLockLevel(stmt->cmds);
+
+ /*
+ * Acquire appropriate lock on the target relation, which will be held until
* end of transaction. This ensures any decisions we make here based on
* the state of the relation will still be good at execution. We must get
- * exclusive lock now because execution will; taking a lower grade lock
- * now and trying to upgrade later risks deadlock.
+ * lock now because execution will later require it; taking a lower grade lock
+ * now and trying to upgrade later risks deadlock. Any new commands we add
+ * after this must not upgrade the lock level requested here.
*/
- rel = relation_openrv(stmt->relation, AccessExclusiveLock);
+ rel = relation_openrv(stmt->relation, lockmode);
/* Set up pstate */
pstate = make_parsestate(NULL);
diff --git a/src/backend/rewrite/rewriteDefine.c b/src/backend/rewrite/rewriteDefine.c
index 06469b31cf0..8e32aed13ec 100644
--- a/src/backend/rewrite/rewriteDefine.c
+++ b/src/backend/rewrite/rewriteDefine.c
@@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/rewrite/rewriteDefine.c,v 1.141 2010/02/14 18:42:15 rhaas Exp $
+ * $PostgreSQL: pgsql/src/backend/rewrite/rewriteDefine.c,v 1.142 2010/07/28 05:22:24 sriggs Exp $
*
*-------------------------------------------------------------------------
*/
@@ -236,11 +236,14 @@ DefineQueryRewrite(char *rulename,
/*
* If we are installing an ON SELECT rule, we had better grab
* AccessExclusiveLock to ensure no SELECTs are currently running on the
- * event relation. For other types of rules, it might be sufficient to
- * grab ShareLock to lock out insert/update/delete actions. But for now,
- * let's just grab AccessExclusiveLock all the time.
+ * event relation. For other types of rules, it is sufficient to
+ * grab ShareRowExclusiveLock to lock out insert/update/delete actions
+ * and to ensure that we lock out current CREATE RULE statements.
*/
- event_relation = heap_open(event_relid, AccessExclusiveLock);
+ if (event_type == CMD_SELECT)
+ event_relation = heap_open(event_relid, AccessExclusiveLock);
+ else
+ event_relation = heap_open(event_relid, ShareRowExclusiveLock);
/*
* Verify relation is of a type that rules can sensibly be applied to.
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
index 080cb9c1ac2..14d183d2a51 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -15,7 +15,7 @@
*
* Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
*
- * $PostgreSQL: pgsql/src/backend/utils/adt/ri_triggers.c,v 1.119 2010/07/22 00:47:52 rhaas Exp $
+ * $PostgreSQL: pgsql/src/backend/utils/adt/ri_triggers.c,v 1.120 2010/07/28 05:22:24 sriggs Exp $
*
* ----------
*/
@@ -2608,7 +2608,7 @@ RI_FKey_keyequal_upd_fk(Trigger *trigger, Relation fk_rel,
* This is not a trigger procedure, but is called during ALTER TABLE
* ADD FOREIGN KEY to validate the initial table contents.
*
- * We expect that an exclusive lock has been taken on rel and pkrel;
+ * We expect that a ShareRowExclusiveLock or higher has been taken on rel and pkrel;
* hence, we do not need to lock individual rows for the check.
*
* If the check fails because the current user doesn't have permissions
diff --git a/src/include/commands/tablecmds.h b/src/include/commands/tablecmds.h
index 45d1835548a..c94a46027ed 100644
--- a/src/include/commands/tablecmds.h
+++ b/src/include/commands/tablecmds.h
@@ -7,7 +7,7 @@
* Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
- * $PostgreSQL: pgsql/src/include/commands/tablecmds.h,v 1.46 2010/02/01 19:28:56 rhaas Exp $
+ * $PostgreSQL: pgsql/src/include/commands/tablecmds.h,v 1.47 2010/07/28 05:22:24 sriggs Exp $
*
*-------------------------------------------------------------------------
*/
@@ -15,6 +15,7 @@
#define TABLECMDS_H
#include "nodes/parsenodes.h"
+#include "storage/lock.h"
#include "utils/relcache.h"
@@ -24,12 +25,14 @@ extern void RemoveRelations(DropStmt *drop);
extern void AlterTable(AlterTableStmt *stmt);
-extern void ATExecChangeOwner(Oid relationOid, Oid newOwnerId, bool recursing);
+extern LOCKMODE AlterTableGetLockLevel(List *cmds);
+
+extern void ATExecChangeOwner(Oid relationOid, Oid newOwnerId, bool recursing, LOCKMODE lockmode);
extern void AlterTableInternal(Oid relid, List *cmds, bool recurse);
extern void AlterTableNamespace(RangeVar *relation, const char *newschema,
- ObjectType stmttype);
+ ObjectType stmttype, LOCKMODE lockmode);
extern void AlterRelationNamespaceInternal(Relation classRel, Oid relOid,
Oid oldNspOid, Oid newNspOid,
--
GitLab