Avoid unnecessary catalog updates in ALTER SEQUENCE

ALTER SEQUENCE can do nontransactional changes to the sequence (RESTART
clause) and transactional updates to the pg_sequence catalog (most other
clauses).  When just calling RESTART, the code would still needlessly do
a catalog update without any changes.  This would entangle that
operation in the concurrency issues of a catalog update (causing either
locking or concurrency errors, depending on how that issue is to be
resolved).

Fix by keeping track during options parsing whether a catalog update is
needed, and skip it if not.

Reported-by: Jason Petersen <jason@citusdata.com>

diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c
index acd4e359bb4521fc64bc05f0007263970904d9cf..4a56926b5306b47ac1e7c80c5c6fa6488157b658 100644 (file)
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -101,6 +101,7 @@ static Form_pg_sequence_data read_seq_tuple(Relation rel,
 static void init_params(ParseState *pstate, List *options, bool for_identity,
                                                bool isInit,
                                                Form_pg_sequence seqform,
+                                               bool *changed_seqform,
                                                Form_pg_sequence_data seqdataform, List **owned_by);
 static void do_setval(Oid relid, int64 next, bool iscalled);
 static void process_owned_by(Relation seqrel, List *owned_by, bool for_identity);
@@ -115,6 +116,7 @@ DefineSequence(ParseState *pstate, CreateSeqStmt *seq)
 {
        FormData_pg_sequence seqform;
        FormData_pg_sequence_data seqdataform;
+       bool            changed_seqform = false; /* not used here */
        List       *owned_by;
        CreateStmt *stmt = makeNode(CreateStmt);
        Oid                     seqoid;
@@ -153,7 +155,7 @@ DefineSequence(ParseState *pstate, CreateSeqStmt *seq)
        }
 
        /* Check and set all option values */
-       init_params(pstate, seq->options, seq->for_identity, true, &seqform, &seqdataform, &owned_by);
+       init_params(pstate, seq->options, seq->for_identity, true, &seqform, &changed_seqform, &seqdataform, &owned_by);
 
        /*
         * Create relation (and fill value[] and null[] for the tuple)
@@ -418,6 +420,7 @@ AlterSequence(ParseState *pstate, AlterSeqStmt *stmt)
        Form_pg_sequence seqform;
        Form_pg_sequence_data seqdata;
        FormData_pg_sequence_data newseqdata;
+       bool            changed_seqform = false;
        List       *owned_by;
        ObjectAddress address;
        Relation        rel;
@@ -443,7 +446,7 @@ AlterSequence(ParseState *pstate, AlterSeqStmt *stmt)
        /* lock page' buffer and read tuple into new sequence structure */
        seqdata = read_seq_tuple(seqrel, &buf, &seqdatatuple);
 
-       /* Copy old values of options into workspace */
+       /* Copy old sequence data into workspace */
        memcpy(&newseqdata, seqdata, sizeof(FormData_pg_sequence_data));
 
        rel = heap_open(SequenceRelationId, RowExclusiveLock);
@@ -456,7 +459,7 @@ AlterSequence(ParseState *pstate, AlterSeqStmt *stmt)
        seqform = (Form_pg_sequence) GETSTRUCT(tuple);
 
        /* Check and set new values */
-       init_params(pstate, stmt->options, stmt->for_identity, false, seqform, &newseqdata, &owned_by);
+       init_params(pstate, stmt->options, stmt->for_identity, false, seqform, &changed_seqform, &newseqdata, &owned_by);
 
        /* Clear local cache so that we don't think we have cached numbers */
        /* Note that we do not change the currval() state */
@@ -507,7 +510,8 @@ AlterSequence(ParseState *pstate, AlterSeqStmt *stmt)
 
        relation_close(seqrel, NoLock);
 
-       CatalogTupleUpdate(rel, &tuple->t_self, tuple);
+       if (changed_seqform)
+               CatalogTupleUpdate(rel, &tuple->t_self, tuple);
        heap_close(rel, RowExclusiveLock);
 
        return address;
@@ -1213,9 +1217,12 @@ read_seq_tuple(Relation rel, Buffer *buf, HeapTuple seqdatatuple)
 }
 
 /*
- * init_params: process the options list of CREATE or ALTER SEQUENCE,
- * and store the values into appropriate fields of *new.  Also set
- * *owned_by to any OWNED BY option, or to NIL if there is none.
+ * init_params: process the options list of CREATE or ALTER SEQUENCE, and
+ * store the values into appropriate fields of seqform, for changes that go
+ * into the pg_sequence catalog, and seqdataform for changes to the sequence
+ * relation itself.  Set *changed_seqform to true if seqform was changed
+ * (interesting for ALTER SEQUENCE).  Also set *owned_by to any OWNED BY
+ * option, or to NIL if there is none.
  *
  * If isInit is true, fill any unspecified options with default values;
  * otherwise, do not change existing options that aren't explicitly overridden.
@@ -1224,7 +1231,9 @@ static void
 init_params(ParseState *pstate, List *options, bool for_identity,
                        bool isInit,
                        Form_pg_sequence seqform,
-                       Form_pg_sequence_data seqdataform, List **owned_by)
+                       bool *changed_seqform,
+                       Form_pg_sequence_data seqdataform,
+                       List **owned_by)
 {
        DefElem    *as_type = NULL;
        DefElem    *start_value = NULL;
@@ -1342,6 +1351,8 @@ init_params(ParseState *pstate, List *options, bool for_identity,
                                 defel->defname);
        }
 
+       *changed_seqform = false;
+
        /*
         * We must reset log_cnt when isInit or when changing any parameters that
         * would affect future nextval allocations.
@@ -1382,14 +1393,19 @@ init_params(ParseState *pstate, List *options, bool for_identity,
                }
 
                seqform->seqtypid = newtypid;
+               *changed_seqform = true;
        }
        else if (isInit)
+       {
                seqform->seqtypid = INT8OID;
+               *changed_seqform = true;
+       }
 
        /* INCREMENT BY */
        if (increment_by != NULL)
        {
                seqform->seqincrement = defGetInt64(increment_by);
+               *changed_seqform = true;
                if (seqform->seqincrement == 0)
                        ereport(ERROR,
                                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
@@ -1397,22 +1413,30 @@ init_params(ParseState *pstate, List *options, bool for_identity,
                seqdataform->log_cnt = 0;
        }
        else if (isInit)
+       {
                seqform->seqincrement = 1;
+               *changed_seqform = true;
+       }
 
        /* CYCLE */
        if (is_cycled != NULL)
        {
                seqform->seqcycle = intVal(is_cycled->arg);
+               *changed_seqform = true;
                Assert(BoolIsValid(seqform->seqcycle));
                seqdataform->log_cnt = 0;
        }
        else if (isInit)
+       {
                seqform->seqcycle = false;
+               *changed_seqform = true;
+       }
 
        /* MAXVALUE (null arg means NO MAXVALUE) */
        if (max_value != NULL && max_value->arg)
        {
                seqform->seqmax = defGetInt64(max_value);
+               *changed_seqform = true;
                seqdataform->log_cnt = 0;
        }
        else if (isInit || max_value != NULL || reset_max_value)
@@ -1429,6 +1453,7 @@ init_params(ParseState *pstate, List *options, bool for_identity,
                }
                else
                        seqform->seqmax = -1;   /* descending seq */
+               *changed_seqform = true;
                seqdataform->log_cnt = 0;
        }
 
@@ -1450,6 +1475,7 @@ init_params(ParseState *pstate, List *options, bool for_identity,
        if (min_value != NULL && min_value->arg)
        {
                seqform->seqmin = defGetInt64(min_value);
+               *changed_seqform = true;
                seqdataform->log_cnt = 0;
        }
        else if (isInit || min_value != NULL || reset_min_value)
@@ -1466,6 +1492,7 @@ init_params(ParseState *pstate, List *options, bool for_identity,
                }
                else
                        seqform->seqmin = 1; /* ascending seq */
+               *changed_seqform = true;
                seqdataform->log_cnt = 0;
        }
 
@@ -1499,13 +1526,17 @@ init_params(ParseState *pstate, List *options, bool for_identity,
 
        /* START WITH */
        if (start_value != NULL)
+       {
                seqform->seqstart = defGetInt64(start_value);
+               *changed_seqform = true;
+       }
        else if (isInit)
        {
                if (seqform->seqincrement > 0)
                        seqform->seqstart = seqform->seqmin;    /* ascending seq */
                else
                        seqform->seqstart = seqform->seqmax;    /* descending seq */
+               *changed_seqform = true;
        }
 
        /* crosscheck START */
@@ -1580,6 +1611,7 @@ init_params(ParseState *pstate, List *options, bool for_identity,
        if (cache_value != NULL)
        {
                seqform->seqcache = defGetInt64(cache_value);
+               *changed_seqform = true;
                if (seqform->seqcache <= 0)
                {
                        char            buf[100];
@@ -1593,7 +1625,10 @@ init_params(ParseState *pstate, List *options, bool for_identity,
                seqdataform->log_cnt = 0;
        }
        else if (isInit)
+       {
                seqform->seqcache = 1;
+               *changed_seqform = true;
+       }
 }
 
 /*