src/backend/commands/sequence.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * sequence.c
   4  *        PostgreSQL sequences support code.
   5  *
   6  * Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        src/backend/commands/sequence.c
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 #include "postgres.h"
  16
  17 #include "access/bufmask.h"
  18 #include "access/htup_details.h"
  19 #include "access/multixact.h"
  20 #include "access/relation.h"
  21 #include "access/table.h"
  22 #include "access/transam.h"
  23 #include "access/xact.h"
  24 #include "access/xlog.h"
  25 #include "access/xloginsert.h"
  26 #include "access/xlogutils.h"
  27 #include "catalog/dependency.h"
  28 #include "catalog/indexing.h"
  29 #include "catalog/namespace.h"
  30 #include "catalog/objectaccess.h"
  31 #include "catalog/pg_sequence.h"
  32 #include "catalog/pg_type.h"
  33 #include "catalog/storage_xlog.h"
  34 #include "commands/defrem.h"
  35 #include "commands/sequence.h"
  36 #include "commands/tablecmds.h"
  37 #include "funcapi.h"
  38 #include "miscadmin.h"
  39 #include "nodes/makefuncs.h"
  40 #include "parser/parse_type.h"
  41 #include "storage/lmgr.h"
  42 #include "storage/proc.h"
  43 #include "storage/smgr.h"
  44 #include "utils/acl.h"
  45 #include "utils/builtins.h"
  46 #include "utils/lsyscache.h"
  47 #include "utils/resowner.h"
  48 #include "utils/syscache.h"
  49 #include "utils/varlena.h"
  50
  51
  52 /*
  53  * We don't want to log each fetching of a value from a sequence,
  54  * so we pre-log a few fetches in advance. In the event of
  55  * crash we can lose (skip over) as many values as we pre-logged.
  56  */
  57 #define SEQ_LOG_VALS    32
  58
  59 /*
  60  * The "special area" of a sequence's buffer page looks like this.
  61  */
  62 #define SEQ_MAGIC         0x1717
  63
  64 typedef struct sequence_magic
  65 {
  66         uint32          magic;
  67 } sequence_magic;
  68
  69 /*
  70  * We store a SeqTable item for every sequence we have touched in the current
  71  * session.  This is needed to hold onto nextval/currval state.  (We can't
  72  * rely on the relcache, since it's only, well, a cache, and may decide to
  73  * discard entries.)
  74  */
  75 typedef struct SeqTableData
  76 {
  77         Oid                     relid;                  /* pg_class OID of this sequence (hash key) */
  78         RelFileNumber filenumber;       /* last seen relfilenumber of this sequence */
  79         LocalTransactionId lxid;        /* xact in which we last did a seq op */
  80         bool            last_valid;             /* do we have a valid "last" value? */
  81         int64           last;                   /* value last returned by nextval */
  82         int64           cached;                 /* last value already cached for nextval */
  83         /* if last != cached, we have not used up all the cached values */
  84         int64           increment;              /* copy of sequence's increment field */
  85         /* note that increment is zero until we first do nextval_internal() */
  86 } SeqTableData;
  87
  88 typedef SeqTableData *SeqTable;
  89
  90 static HTAB *seqhashtab = NULL; /* hash table for SeqTable items */
  91
  92 /*
  93  * last_used_seq is updated by nextval() to point to the last used
  94  * sequence.
  95  */
  96 static SeqTableData *last_used_seq = NULL;
  97
  98 static void fill_seq_with_data(Relation rel, HeapTuple tuple);
  99 static void fill_seq_fork_with_data(Relation rel, HeapTuple tuple, ForkNumber forkNum);
 100 static Relation lock_and_open_sequence(SeqTable seq);
 101 static void create_seq_hashtable(void);
 102 static void init_sequence(Oid relid, SeqTable *p_elm, Relation *p_rel);
 103 static Form_pg_sequence_data read_seq_tuple(Relation rel,
 104                                                                                         Buffer *buf, HeapTuple seqdatatuple);
 105 static void init_params(ParseState *pstate, List *options, bool for_identity,
 106                                                 bool isInit,
 107                                                 Form_pg_sequence seqform,
 108                                                 Form_pg_sequence_data seqdataform,
 109                                                 bool *need_seq_rewrite,
 110                                                 List **owned_by);
 111 static void do_setval(Oid relid, int64 next, bool iscalled);
 112 static void process_owned_by(Relation seqrel, List *owned_by, bool for_identity);
 113
 114
 115 /*
 116  * DefineSequence
 117  *                              Creates a new sequence relation
 118  */
 119 ObjectAddress
 120 DefineSequence(ParseState *pstate, CreateSeqStmt *seq)
 121 {
 122         FormData_pg_sequence seqform;
 123         FormData_pg_sequence_data seqdataform;
 124         bool            need_seq_rewrite;
 125         List       *owned_by;
 126         CreateStmt *stmt = makeNode(CreateStmt);
 127         Oid                     seqoid;
 128         ObjectAddress address;
 129         Relation        rel;
 130         HeapTuple       tuple;
 131         TupleDesc       tupDesc;
 132         Datum           value[SEQ_COL_LASTCOL];
 133         bool            null[SEQ_COL_LASTCOL];
 134         Datum           pgs_values[Natts_pg_sequence];
 135         bool            pgs_nulls[Natts_pg_sequence];
 136         int                     i;
 137
 138         /*
 139          * If if_not_exists was given and a relation with the same name already
 140          * exists, bail out. (Note: we needn't check this when not if_not_exists,
 141          * because DefineRelation will complain anyway.)
 142          */
 143         if (seq->if_not_exists)
 144         {
 145                 RangeVarGetAndCheckCreationNamespace(seq->sequence, NoLock, &seqoid);
 146                 if (OidIsValid(seqoid))
 147                 {
 148                         /*
 149                          * If we are in an extension script, insist that the pre-existing
 150                          * object be a member of the extension, to avoid security risks.
 151                          */
 152                         ObjectAddressSet(address, RelationRelationId, seqoid);
 153                         checkMembershipInCurrentExtension(&address);
 154
 155                         /* OK to skip */
 156                         ereport(NOTICE,
 157                                         (errcode(ERRCODE_DUPLICATE_TABLE),
 158                                          errmsg("relation \"%s\" already exists, skipping",
 159                                                         seq->sequence->relname)));
 160                         return InvalidObjectAddress;
 161                 }
 162         }
 163
 164         /* Check and set all option values */
 165         init_params(pstate, seq->options, seq->for_identity, true,
 166                                 &seqform, &seqdataform,
 167                                 &need_seq_rewrite, &owned_by);
 168
 169         /*
 170          * Create relation (and fill value[] and null[] for the tuple)
 171          */
 172         stmt->tableElts = NIL;
 173         for (i = SEQ_COL_FIRSTCOL; i <= SEQ_COL_LASTCOL; i++)
 174         {
 175                 ColumnDef  *coldef = NULL;
 176
 177                 switch (i)
 178                 {
 179                         case SEQ_COL_LASTVAL:
 180                                 coldef = makeColumnDef("last_value", INT8OID, -1, InvalidOid);
 181                                 value[i - 1] = Int64GetDatumFast(seqdataform.last_value);
 182                                 break;
 183                         case SEQ_COL_LOG:
 184                                 coldef = makeColumnDef("log_cnt", INT8OID, -1, InvalidOid);
 185                                 value[i - 1] = Int64GetDatum((int64) 0);
 186                                 break;
 187                         case SEQ_COL_CALLED:
 188                                 coldef = makeColumnDef("is_called", BOOLOID, -1, InvalidOid);
 189                                 value[i - 1] = BoolGetDatum(false);
 190                                 break;
 191                 }
 192
 193                 coldef->is_not_null = true;
 194                 null[i - 1] = false;
 195
 196                 stmt->tableElts = lappend(stmt->tableElts, coldef);
 197         }
 198
 199         stmt->relation = seq->sequence;
 200         stmt->inhRelations = NIL;
 201         stmt->constraints = NIL;
 202         stmt->options = NIL;
 203         stmt->oncommit = ONCOMMIT_NOOP;
 204         stmt->tablespacename = NULL;
 205         stmt->if_not_exists = seq->if_not_exists;
 206
 207         address = DefineRelation(stmt, RELKIND_SEQUENCE, seq->ownerId, NULL, NULL);
 208         seqoid = address.objectId;
 209         Assert(seqoid != InvalidOid);
 210
 211         rel = table_open(seqoid, AccessExclusiveLock);
 212         tupDesc = RelationGetDescr(rel);
 213
 214         /* now initialize the sequence's data */
 215         tuple = heap_form_tuple(tupDesc, value, null);
 216         fill_seq_with_data(rel, tuple);
 217
 218         /* process OWNED BY if given */
 219         if (owned_by)
 220                 process_owned_by(rel, owned_by, seq->for_identity);
 221
 222         table_close(rel, NoLock);
 223
 224         /* fill in pg_sequence */
 225         rel = table_open(SequenceRelationId, RowExclusiveLock);
 226         tupDesc = RelationGetDescr(rel);
 227
 228         memset(pgs_nulls, 0, sizeof(pgs_nulls));
 229
 230         pgs_values[Anum_pg_sequence_seqrelid - 1] = ObjectIdGetDatum(seqoid);
 231         pgs_values[Anum_pg_sequence_seqtypid - 1] = ObjectIdGetDatum(seqform.seqtypid);
 232         pgs_values[Anum_pg_sequence_seqstart - 1] = Int64GetDatumFast(seqform.seqstart);
 233         pgs_values[Anum_pg_sequence_seqincrement - 1] = Int64GetDatumFast(seqform.seqincrement);
 234         pgs_values[Anum_pg_sequence_seqmax - 1] = Int64GetDatumFast(seqform.seqmax);
 235         pgs_values[Anum_pg_sequence_seqmin - 1] = Int64GetDatumFast(seqform.seqmin);
 236         pgs_values[Anum_pg_sequence_seqcache - 1] = Int64GetDatumFast(seqform.seqcache);
 237         pgs_values[Anum_pg_sequence_seqcycle - 1] = BoolGetDatum(seqform.seqcycle);
 238
 239         tuple = heap_form_tuple(tupDesc, pgs_values, pgs_nulls);
 240         CatalogTupleInsert(rel, tuple);
 241
 242         heap_freetuple(tuple);
 243         table_close(rel, RowExclusiveLock);
 244
 245         return address;
 246 }
 247
 248 /*
 249  * Reset a sequence to its initial value.
 250  *
 251  * The change is made transactionally, so that on failure of the current
 252  * transaction, the sequence will be restored to its previous state.
 253  * We do that by creating a whole new relfilenumber for the sequence; so this
 254  * works much like the rewriting forms of ALTER TABLE.
 255  *
 256  * Caller is assumed to have acquired AccessExclusiveLock on the sequence,
 257  * which must not be released until end of transaction.  Caller is also
 258  * responsible for permissions checking.
 259  */
 260 void
 261 ResetSequence(Oid seq_relid)
 262 {
 263         Relation        seq_rel;
 264         SeqTable        elm;
 265         Form_pg_sequence_data seq;
 266         Buffer          buf;
 267         HeapTupleData seqdatatuple;
 268         HeapTuple       tuple;
 269         HeapTuple       pgstuple;
 270         Form_pg_sequence pgsform;
 271         int64           startv;
 272
 273         /*
 274          * Read the old sequence.  This does a bit more work than really
 275          * necessary, but it's simple, and we do want to double-check that it's
 276          * indeed a sequence.
 277          */
 278         init_sequence(seq_relid, &elm, &seq_rel);
 279         (void) read_seq_tuple(seq_rel, &buf, &seqdatatuple);
 280
 281         pgstuple = SearchSysCache1(SEQRELID, ObjectIdGetDatum(seq_relid));
 282         if (!HeapTupleIsValid(pgstuple))
 283                 elog(ERROR, "cache lookup failed for sequence %u", seq_relid);
 284         pgsform = (Form_pg_sequence) GETSTRUCT(pgstuple);
 285         startv = pgsform->seqstart;
 286         ReleaseSysCache(pgstuple);
 287
 288         /*
 289          * Copy the existing sequence tuple.
 290          */
 291         tuple = heap_copytuple(&seqdatatuple);
 292
 293         /* Now we're done with the old page */
 294         UnlockReleaseBuffer(buf);
 295
 296         /*
 297          * Modify the copied tuple to execute the restart (compare the RESTART
 298          * action in AlterSequence)
 299          */
 300         seq = (Form_pg_sequence_data) GETSTRUCT(tuple);
 301         seq->last_value = startv;
 302         seq->is_called = false;
 303         seq->log_cnt = 0;
 304
 305         /*
 306          * Create a new storage file for the sequence.
 307          */
 308         RelationSetNewRelfilenumber(seq_rel, seq_rel->rd_rel->relpersistence);
 309
 310         /*
 311          * Ensure sequence's relfrozenxid is at 0, since it won't contain any
 312          * unfrozen XIDs.  Same with relminmxid, since a sequence will never
 313          * contain multixacts.
 314          */
 315         Assert(seq_rel->rd_rel->relfrozenxid == InvalidTransactionId);
 316         Assert(seq_rel->rd_rel->relminmxid == InvalidMultiXactId);
 317
 318         /*
 319          * Insert the modified tuple into the new storage file.
 320          */
 321         fill_seq_with_data(seq_rel, tuple);
 322
 323         /* Clear local cache so that we don't think we have cached numbers */
 324         /* Note that we do not change the currval() state */
 325         elm->cached = elm->last;
 326
 327         relation_close(seq_rel, NoLock);
 328 }
 329
 330 /*
 331  * Initialize a sequence's relation with the specified tuple as content
 332  *
 333  * This handles unlogged sequences by writing to both the main and the init
 334  * fork as necessary.
 335  */
 336 static void
 337 fill_seq_with_data(Relation rel, HeapTuple tuple)
 338 {
 339         fill_seq_fork_with_data(rel, tuple, MAIN_FORKNUM);
 340
 341         if (rel->rd_rel->relpersistence == RELPERSISTENCE_UNLOGGED)
 342         {
 343                 SMgrRelation srel;
 344
 345                 srel = smgropen(rel->rd_locator, InvalidBackendId);
 346                 smgrcreate(srel, INIT_FORKNUM, false);
 347                 log_smgrcreate(&rel->rd_locator, INIT_FORKNUM);
 348                 fill_seq_fork_with_data(rel, tuple, INIT_FORKNUM);
 349                 FlushRelationBuffers(rel);
 350                 smgrclose(srel);
 351         }
 352 }
 353
 354 /*
 355  * Initialize a sequence's relation fork with the specified tuple as content
 356  */
 357 static void
 358 fill_seq_fork_with_data(Relation rel, HeapTuple tuple, ForkNumber forkNum)
 359 {
 360         Buffer          buf;
 361         Page            page;
 362         sequence_magic *sm;
 363         OffsetNumber offnum;
 364
 365         /* Initialize first page of relation with special magic number */
 366
 367         buf = ExtendBufferedRel(BMR_REL(rel), forkNum, NULL,
 368                                                         EB_LOCK_FIRST | EB_SKIP_EXTENSION_LOCK);
 369         Assert(BufferGetBlockNumber(buf) == 0);
 370
 371         page = BufferGetPage(buf);
 372
 373         PageInit(page, BufferGetPageSize(buf), sizeof(sequence_magic));
 374         sm = (sequence_magic *) PageGetSpecialPointer(page);
 375         sm->magic = SEQ_MAGIC;
 376
 377         /* Now insert sequence tuple */
 378
 379         /*
 380          * Since VACUUM does not process sequences, we have to force the tuple to
 381          * have xmin = FrozenTransactionId now.  Otherwise it would become
 382          * invisible to SELECTs after 2G transactions.  It is okay to do this
 383          * because if the current transaction aborts, no other xact will ever
 384          * examine the sequence tuple anyway.
 385          */
 386         HeapTupleHeaderSetXmin(tuple->t_data, FrozenTransactionId);
 387         HeapTupleHeaderSetXminFrozen(tuple->t_data);
 388         HeapTupleHeaderSetCmin(tuple->t_data, FirstCommandId);
 389         HeapTupleHeaderSetXmax(tuple->t_data, InvalidTransactionId);
 390         tuple->t_data->t_infomask |= HEAP_XMAX_INVALID;
 391         ItemPointerSet(&tuple->t_data->t_ctid, 0, FirstOffsetNumber);
 392
 393         /* check the comment above nextval_internal()'s equivalent call. */
 394         if (RelationNeedsWAL(rel))
 395                 GetTopTransactionId();
 396
 397         START_CRIT_SECTION();
 398
 399         MarkBufferDirty(buf);
 400
 401         offnum = PageAddItem(page, (Item) tuple->t_data, tuple->t_len,
 402                                                  InvalidOffsetNumber, false, false);
 403         if (offnum != FirstOffsetNumber)
 404                 elog(ERROR, "failed to add sequence tuple to page");
 405
 406         /* XLOG stuff */
 407         if (RelationNeedsWAL(rel) || forkNum == INIT_FORKNUM)
 408         {
 409                 xl_seq_rec      xlrec;
 410                 XLogRecPtr      recptr;
 411
 412                 XLogBeginInsert();
 413                 XLogRegisterBuffer(0, buf, REGBUF_WILL_INIT);
 414
 415                 xlrec.locator = rel->rd_locator;
 416
 417                 XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
 418                 XLogRegisterData((char *) tuple->t_data, tuple->t_len);
 419
 420                 recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
 421
 422                 PageSetLSN(page, recptr);
 423         }
 424
 425         END_CRIT_SECTION();
 426
 427         UnlockReleaseBuffer(buf);
 428 }
 429
 430 /*
 431  * AlterSequence
 432  *
 433  * Modify the definition of a sequence relation
 434  */
 435 ObjectAddress
 436 AlterSequence(ParseState *pstate, AlterSeqStmt *stmt)
 437 {
 438         Oid                     relid;
 439         SeqTable        elm;
 440         Relation        seqrel;
 441         Buffer          buf;
 442         HeapTupleData datatuple;
 443         Form_pg_sequence seqform;
 444         Form_pg_sequence_data newdataform;
 445         bool            need_seq_rewrite;
 446         List       *owned_by;
 447         ObjectAddress address;
 448         Relation        rel;
 449         HeapTuple       seqtuple;
 450         HeapTuple       newdatatuple;
 451
 452         /* Open and lock sequence, and check for ownership along the way. */
 453         relid = RangeVarGetRelidExtended(stmt->sequence,
 454                                                                          ShareRowExclusiveLock,
 455                                                                          stmt->missing_ok ? RVR_MISSING_OK : 0,
 456                                                                          RangeVarCallbackOwnsRelation,
 457                                                                          NULL);
 458         if (relid == InvalidOid)
 459         {
 460                 ereport(NOTICE,
 461                                 (errmsg("relation \"%s\" does not exist, skipping",
 462                                                 stmt->sequence->relname)));
 463                 return InvalidObjectAddress;
 464         }
 465
 466         init_sequence(relid, &elm, &seqrel);
 467
 468         rel = table_open(SequenceRelationId, RowExclusiveLock);
 469         seqtuple = SearchSysCacheCopy1(SEQRELID,
 470                                                                    ObjectIdGetDatum(relid));
 471         if (!HeapTupleIsValid(seqtuple))
 472                 elog(ERROR, "cache lookup failed for sequence %u",
 473                          relid);
 474
 475         seqform = (Form_pg_sequence) GETSTRUCT(seqtuple);
 476
 477         /* lock page buffer and read tuple into new sequence structure */
 478         (void) read_seq_tuple(seqrel, &buf, &datatuple);
 479
 480         /* copy the existing sequence data tuple, so it can be modified locally */
 481         newdatatuple = heap_copytuple(&datatuple);
 482         newdataform = (Form_pg_sequence_data) GETSTRUCT(newdatatuple);
 483
 484         UnlockReleaseBuffer(buf);
 485
 486         /* Check and set new values */
 487         init_params(pstate, stmt->options, stmt->for_identity, false,
 488                                 seqform, newdataform,
 489                                 &need_seq_rewrite, &owned_by);
 490
 491         /* Clear local cache so that we don't think we have cached numbers */
 492         /* Note that we do not change the currval() state */
 493         elm->cached = elm->last;
 494
 495         /* If needed, rewrite the sequence relation itself */
 496         if (need_seq_rewrite)
 497         {
 498                 /* check the comment above nextval_internal()'s equivalent call. */
 499                 if (RelationNeedsWAL(seqrel))
 500                         GetTopTransactionId();
 501
 502                 /*
 503                  * Create a new storage file for the sequence, making the state
 504                  * changes transactional.
 505                  */
 506                 RelationSetNewRelfilenumber(seqrel, seqrel->rd_rel->relpersistence);
 507
 508                 /*
 509                  * Ensure sequence's relfrozenxid is at 0, since it won't contain any
 510                  * unfrozen XIDs.  Same with relminmxid, since a sequence will never
 511                  * contain multixacts.
 512                  */
 513                 Assert(seqrel->rd_rel->relfrozenxid == InvalidTransactionId);
 514                 Assert(seqrel->rd_rel->relminmxid == InvalidMultiXactId);
 515
 516                 /*
 517                  * Insert the modified tuple into the new storage file.
 518                  */
 519                 fill_seq_with_data(seqrel, newdatatuple);
 520         }
 521
 522         /* process OWNED BY if given */
 523         if (owned_by)
 524                 process_owned_by(seqrel, owned_by, stmt->for_identity);
 525
 526         /* update the pg_sequence tuple (we could skip this in some cases...) */
 527         CatalogTupleUpdate(rel, &seqtuple->t_self, seqtuple);
 528
 529         InvokeObjectPostAlterHook(RelationRelationId, relid, 0);
 530
 531         ObjectAddressSet(address, RelationRelationId, relid);
 532
 533         table_close(rel, RowExclusiveLock);
 534         relation_close(seqrel, NoLock);
 535
 536         return address;
 537 }
 538
 539 void
 540 SequenceChangePersistence(Oid relid, char newrelpersistence)
 541 {
 542         SeqTable        elm;
 543         Relation        seqrel;
 544         Buffer          buf;
 545         HeapTupleData seqdatatuple;
 546
 547         init_sequence(relid, &elm, &seqrel);
 548
 549         /* check the comment above nextval_internal()'s equivalent call. */
 550         if (RelationNeedsWAL(seqrel))
 551                 GetTopTransactionId();
 552
 553         (void) read_seq_tuple(seqrel, &buf, &seqdatatuple);
 554         RelationSetNewRelfilenumber(seqrel, newrelpersistence);
 555         fill_seq_with_data(seqrel, &seqdatatuple);
 556         UnlockReleaseBuffer(buf);
 557
 558         relation_close(seqrel, NoLock);
 559 }
 560
 561 void
 562 DeleteSequenceTuple(Oid relid)
 563 {
 564         Relation        rel;
 565         HeapTuple       tuple;
 566
 567         rel = table_open(SequenceRelationId, RowExclusiveLock);
 568
 569         tuple = SearchSysCache1(SEQRELID, ObjectIdGetDatum(relid));
 570         if (!HeapTupleIsValid(tuple))
 571                 elog(ERROR, "cache lookup failed for sequence %u", relid);
 572
 573         CatalogTupleDelete(rel, &tuple->t_self);
 574
 575         ReleaseSysCache(tuple);
 576         table_close(rel, RowExclusiveLock);
 577 }
 578
 579 /*
 580  * Note: nextval with a text argument is no longer exported as a pg_proc
 581  * entry, but we keep it around to ease porting of C code that may have
 582  * called the function directly.
 583  */
 584 Datum
 585 nextval(PG_FUNCTION_ARGS)
 586 {
 587         text       *seqin = PG_GETARG_TEXT_PP(0);
 588         RangeVar   *sequence;
 589         Oid                     relid;
 590
 591         sequence = makeRangeVarFromNameList(textToQualifiedNameList(seqin));
 592
 593         /*
 594          * XXX: This is not safe in the presence of concurrent DDL, but acquiring
 595          * a lock here is more expensive than letting nextval_internal do it,
 596          * since the latter maintains a cache that keeps us from hitting the lock
 597          * manager more than once per transaction.  It's not clear whether the
 598          * performance penalty is material in practice, but for now, we do it this
 599          * way.
 600          */
 601         relid = RangeVarGetRelid(sequence, NoLock, false);
 602
 603         PG_RETURN_INT64(nextval_internal(relid, true));
 604 }
 605
 606 Datum
 607 nextval_oid(PG_FUNCTION_ARGS)
 608 {
 609         Oid                     relid = PG_GETARG_OID(0);
 610
 611         PG_RETURN_INT64(nextval_internal(relid, true));
 612 }
 613
 614 int64
 615 nextval_internal(Oid relid, bool check_permissions)
 616 {
 617         SeqTable        elm;
 618         Relation        seqrel;
 619         Buffer          buf;
 620         Page            page;
 621         HeapTuple       pgstuple;
 622         Form_pg_sequence pgsform;
 623         HeapTupleData seqdatatuple;
 624         Form_pg_sequence_data seq;
 625         int64           incby,
 626                                 maxv,
 627                                 minv,
 628                                 cache,
 629                                 log,
 630                                 fetch,
 631                                 last;
 632         int64           result,
 633                                 next,
 634                                 rescnt = 0;
 635         bool            cycle;
 636         bool            logit = false;
 637
 638         /* open and lock sequence */
 639         init_sequence(relid, &elm, &seqrel);
 640
 641         if (check_permissions &&
 642                 pg_class_aclcheck(elm->relid, GetUserId(),
 643                                                   ACL_USAGE | ACL_UPDATE) != ACLCHECK_OK)
 644                 ereport(ERROR,
 645                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 646                                  errmsg("permission denied for sequence %s",
 647                                                 RelationGetRelationName(seqrel))));
 648
 649         /* read-only transactions may only modify temp sequences */
 650         if (!seqrel->rd_islocaltemp)
 651                 PreventCommandIfReadOnly("nextval()");
 652
 653         /*
 654          * Forbid this during parallel operation because, to make it work, the
 655          * cooperating backends would need to share the backend-local cached
 656          * sequence information.  Currently, we don't support that.
 657          */
 658         PreventCommandIfParallelMode("nextval()");
 659
 660         if (elm->last != elm->cached)   /* some numbers were cached */
 661         {
 662                 Assert(elm->last_valid);
 663                 Assert(elm->increment != 0);
 664                 elm->last += elm->increment;
 665                 relation_close(seqrel, NoLock);
 666                 last_used_seq = elm;
 667                 return elm->last;
 668         }
 669
 670         pgstuple = SearchSysCache1(SEQRELID, ObjectIdGetDatum(relid));
 671         if (!HeapTupleIsValid(pgstuple))
 672                 elog(ERROR, "cache lookup failed for sequence %u", relid);
 673         pgsform = (Form_pg_sequence) GETSTRUCT(pgstuple);
 674         incby = pgsform->seqincrement;
 675         maxv = pgsform->seqmax;
 676         minv = pgsform->seqmin;
 677         cache = pgsform->seqcache;
 678         cycle = pgsform->seqcycle;
 679         ReleaseSysCache(pgstuple);
 680
 681         /* lock page buffer and read tuple */
 682         seq = read_seq_tuple(seqrel, &buf, &seqdatatuple);
 683         page = BufferGetPage(buf);
 684
 685         elm->increment = incby;
 686         last = next = result = seq->last_value;
 687         fetch = cache;
 688         log = seq->log_cnt;
 689
 690         if (!seq->is_called)
 691         {
 692                 rescnt++;                               /* return last_value if not is_called */
 693                 fetch--;
 694         }
 695
 696         /*
 697          * Decide whether we should emit a WAL log record.  If so, force up the
 698          * fetch count to grab SEQ_LOG_VALS more values than we actually need to
 699          * cache.  (These will then be usable without logging.)
 700          *
 701          * If this is the first nextval after a checkpoint, we must force a new
 702          * WAL record to be written anyway, else replay starting from the
 703          * checkpoint would fail to advance the sequence past the logged values.
 704          * In this case we may as well fetch extra values.
 705          */
 706         if (log < fetch || !seq->is_called)
 707         {
 708                 /* forced log to satisfy local demand for values */
 709                 fetch = log = fetch + SEQ_LOG_VALS;
 710                 logit = true;
 711         }
 712         else
 713         {
 714                 XLogRecPtr      redoptr = GetRedoRecPtr();
 715
 716                 if (PageGetLSN(page) <= redoptr)
 717                 {
 718                         /* last update of seq was before checkpoint */
 719                         fetch = log = fetch + SEQ_LOG_VALS;
 720                         logit = true;
 721                 }
 722         }
 723
 724         while (fetch)                           /* try to fetch cache [+ log ] numbers */
 725         {
 726                 /*
 727                  * Check MAXVALUE for ascending sequences and MINVALUE for descending
 728                  * sequences
 729                  */
 730                 if (incby > 0)
 731                 {
 732                         /* ascending sequence */
 733                         if ((maxv >= 0 && next > maxv - incby) ||
 734                                 (maxv < 0 && next + incby > maxv))
 735                         {
 736                                 if (rescnt > 0)
 737                                         break;          /* stop fetching */
 738                                 if (!cycle)
 739                                         ereport(ERROR,
 740                                                         (errcode(ERRCODE_SEQUENCE_GENERATOR_LIMIT_EXCEEDED),
 741                                                          errmsg("nextval: reached maximum value of sequence \"%s\" (%lld)",
 742                                                                         RelationGetRelationName(seqrel),
 743                                                                         (long long) maxv)));
 744                                 next = minv;
 745                         }
 746                         else
 747                                 next += incby;
 748                 }
 749                 else
 750                 {
 751                         /* descending sequence */
 752                         if ((minv < 0 && next < minv - incby) ||
 753                                 (minv >= 0 && next + incby < minv))
 754                         {
 755                                 if (rescnt > 0)
 756                                         break;          /* stop fetching */
 757                                 if (!cycle)
 758                                         ereport(ERROR,
 759                                                         (errcode(ERRCODE_SEQUENCE_GENERATOR_LIMIT_EXCEEDED),
 760                                                          errmsg("nextval: reached minimum value of sequence \"%s\" (%lld)",
 761                                                                         RelationGetRelationName(seqrel),
 762                                                                         (long long) minv)));
 763                                 next = maxv;
 764                         }
 765                         else
 766                                 next += incby;
 767                 }
 768                 fetch--;
 769                 if (rescnt < cache)
 770                 {
 771                         log--;
 772                         rescnt++;
 773                         last = next;
 774                         if (rescnt == 1)        /* if it's first result - */
 775                                 result = next;  /* it's what to return */
 776                 }
 777         }
 778
 779         log -= fetch;                           /* adjust for any unfetched numbers */
 780         Assert(log >= 0);
 781
 782         /* save info in local cache */
 783         elm->last = result;                     /* last returned number */
 784         elm->cached = last;                     /* last fetched number */
 785         elm->last_valid = true;
 786
 787         last_used_seq = elm;
 788
 789         /*
 790          * If something needs to be WAL logged, acquire an xid, so this
 791          * transaction's commit will trigger a WAL flush and wait for syncrep.
 792          * It's sufficient to ensure the toplevel transaction has an xid, no need
 793          * to assign xids subxacts, that'll already trigger an appropriate wait.
 794          * (Have to do that here, so we're outside the critical section)
 795          */
 796         if (logit && RelationNeedsWAL(seqrel))
 797                 GetTopTransactionId();
 798
 799         /* ready to change the on-disk (or really, in-buffer) tuple */
 800         START_CRIT_SECTION();
 801
 802         /*
 803          * We must mark the buffer dirty before doing XLogInsert(); see notes in
 804          * SyncOneBuffer().  However, we don't apply the desired changes just yet.
 805          * This looks like a violation of the buffer update protocol, but it is in
 806          * fact safe because we hold exclusive lock on the buffer.  Any other
 807          * process, including a checkpoint, that tries to examine the buffer
 808          * contents will block until we release the lock, and then will see the
 809          * final state that we install below.
 810          */
 811         MarkBufferDirty(buf);
 812
 813         /* XLOG stuff */
 814         if (logit && RelationNeedsWAL(seqrel))
 815         {
 816                 xl_seq_rec      xlrec;
 817                 XLogRecPtr      recptr;
 818
 819                 /*
 820                  * We don't log the current state of the tuple, but rather the state
 821                  * as it would appear after "log" more fetches.  This lets us skip
 822                  * that many future WAL records, at the cost that we lose those
 823                  * sequence values if we crash.
 824                  */
 825                 XLogBeginInsert();
 826                 XLogRegisterBuffer(0, buf, REGBUF_WILL_INIT);
 827
 828                 /* set values that will be saved in xlog */
 829                 seq->last_value = next;
 830                 seq->is_called = true;
 831                 seq->log_cnt = 0;
 832
 833                 xlrec.locator = seqrel->rd_locator;
 834
 835                 XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
 836                 XLogRegisterData((char *) seqdatatuple.t_data, seqdatatuple.t_len);
 837
 838                 recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
 839
 840                 PageSetLSN(page, recptr);
 841         }
 842
 843         /* Now update sequence tuple to the intended final state */
 844         seq->last_value = last;         /* last fetched number */
 845         seq->is_called = true;
 846         seq->log_cnt = log;                     /* how much is logged */
 847
 848         END_CRIT_SECTION();
 849
 850         UnlockReleaseBuffer(buf);
 851
 852         relation_close(seqrel, NoLock);
 853
 854         return result;
 855 }
 856
 857 Datum
 858 currval_oid(PG_FUNCTION_ARGS)
 859 {
 860         Oid                     relid = PG_GETARG_OID(0);
 861         int64           result;
 862         SeqTable        elm;
 863         Relation        seqrel;
 864
 865         /* open and lock sequence */
 866         init_sequence(relid, &elm, &seqrel);
 867
 868         if (pg_class_aclcheck(elm->relid, GetUserId(),
 869                                                   ACL_SELECT | ACL_USAGE) != ACLCHECK_OK)
 870                 ereport(ERROR,
 871                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 872                                  errmsg("permission denied for sequence %s",
 873                                                 RelationGetRelationName(seqrel))));
 874
 875         if (!elm->last_valid)
 876                 ereport(ERROR,
 877                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 878                                  errmsg("currval of sequence \"%s\" is not yet defined in this session",
 879                                                 RelationGetRelationName(seqrel))));
 880
 881         result = elm->last;
 882
 883         relation_close(seqrel, NoLock);
 884
 885         PG_RETURN_INT64(result);
 886 }
 887
 888 Datum
 889 lastval(PG_FUNCTION_ARGS)
 890 {
 891         Relation        seqrel;
 892         int64           result;
 893
 894         if (last_used_seq == NULL)
 895                 ereport(ERROR,
 896                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 897                                  errmsg("lastval is not yet defined in this session")));
 898
 899         /* Someone may have dropped the sequence since the last nextval() */
 900         if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(last_used_seq->relid)))
 901                 ereport(ERROR,
 902                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 903                                  errmsg("lastval is not yet defined in this session")));
 904
 905         seqrel = lock_and_open_sequence(last_used_seq);
 906
 907         /* nextval() must have already been called for this sequence */
 908         Assert(last_used_seq->last_valid);
 909
 910         if (pg_class_aclcheck(last_used_seq->relid, GetUserId(),
 911                                                   ACL_SELECT | ACL_USAGE) != ACLCHECK_OK)
 912                 ereport(ERROR,
 913                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 914                                  errmsg("permission denied for sequence %s",
 915                                                 RelationGetRelationName(seqrel))));
 916
 917         result = last_used_seq->last;
 918         relation_close(seqrel, NoLock);
 919
 920         PG_RETURN_INT64(result);
 921 }
 922
 923 /*
 924  * Main internal procedure that handles 2 & 3 arg forms of SETVAL.
 925  *
 926  * Note that the 3 arg version (which sets the is_called flag) is
 927  * only for use in pg_dump, and setting the is_called flag may not
 928  * work if multiple users are attached to the database and referencing
 929  * the sequence (unlikely if pg_dump is restoring it).
 930  *
 931  * It is necessary to have the 3 arg version so that pg_dump can
 932  * restore the state of a sequence exactly during data-only restores -
 933  * it is the only way to clear the is_called flag in an existing
 934  * sequence.
 935  */
 936 static void
 937 do_setval(Oid relid, int64 next, bool iscalled)
 938 {
 939         SeqTable        elm;
 940         Relation        seqrel;
 941         Buffer          buf;
 942         HeapTupleData seqdatatuple;
 943         Form_pg_sequence_data seq;
 944         HeapTuple       pgstuple;
 945         Form_pg_sequence pgsform;
 946         int64           maxv,
 947                                 minv;
 948
 949         /* open and lock sequence */
 950         init_sequence(relid, &elm, &seqrel);
 951
 952         if (pg_class_aclcheck(elm->relid, GetUserId(), ACL_UPDATE) != ACLCHECK_OK)
 953                 ereport(ERROR,
 954                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 955                                  errmsg("permission denied for sequence %s",
 956                                                 RelationGetRelationName(seqrel))));
 957
 958         pgstuple = SearchSysCache1(SEQRELID, ObjectIdGetDatum(relid));
 959         if (!HeapTupleIsValid(pgstuple))
 960                 elog(ERROR, "cache lookup failed for sequence %u", relid);
 961         pgsform = (Form_pg_sequence) GETSTRUCT(pgstuple);
 962         maxv = pgsform->seqmax;
 963         minv = pgsform->seqmin;
 964         ReleaseSysCache(pgstuple);
 965
 966         /* read-only transactions may only modify temp sequences */
 967         if (!seqrel->rd_islocaltemp)
 968                 PreventCommandIfReadOnly("setval()");
 969
 970         /*
 971          * Forbid this during parallel operation because, to make it work, the
 972          * cooperating backends would need to share the backend-local cached
 973          * sequence information.  Currently, we don't support that.
 974          */
 975         PreventCommandIfParallelMode("setval()");
 976
 977         /* lock page buffer and read tuple */
 978         seq = read_seq_tuple(seqrel, &buf, &seqdatatuple);
 979
 980         if ((next < minv) || (next > maxv))
 981                 ereport(ERROR,
 982                                 (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
 983                                  errmsg("setval: value %lld is out of bounds for sequence \"%s\" (%lld..%lld)",
 984                                                 (long long) next, RelationGetRelationName(seqrel),
 985                                                 (long long) minv, (long long) maxv)));
 986
 987         /* Set the currval() state only if iscalled = true */
 988         if (iscalled)
 989         {
 990                 elm->last = next;               /* last returned number */
 991                 elm->last_valid = true;
 992         }
 993
 994         /* In any case, forget any future cached numbers */
 995         elm->cached = elm->last;
 996
 997         /* check the comment above nextval_internal()'s equivalent call. */
 998         if (RelationNeedsWAL(seqrel))
 999                 GetTopTransactionId();
1000
1001         /* ready to change the on-disk (or really, in-buffer) tuple */
1002         START_CRIT_SECTION();
1003
1004         seq->last_value = next;         /* last fetched number */
1005         seq->is_called = iscalled;
1006         seq->log_cnt = 0;
1007
1008         MarkBufferDirty(buf);
1009
1010         /* XLOG stuff */
1011         if (RelationNeedsWAL(seqrel))
1012         {
1013                 xl_seq_rec      xlrec;
1014                 XLogRecPtr      recptr;
1015                 Page            page = BufferGetPage(buf);
1016
1017                 XLogBeginInsert();
1018                 XLogRegisterBuffer(0, buf, REGBUF_WILL_INIT);
1019
1020                 xlrec.locator = seqrel->rd_locator;
1021                 XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
1022                 XLogRegisterData((char *) seqdatatuple.t_data, seqdatatuple.t_len);
1023
1024                 recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
1025
1026                 PageSetLSN(page, recptr);
1027         }
1028
1029         END_CRIT_SECTION();
1030
1031         UnlockReleaseBuffer(buf);
1032
1033         relation_close(seqrel, NoLock);
1034 }
1035
1036 /*
1037  * Implement the 2 arg setval procedure.
1038  * See do_setval for discussion.
1039  */
1040 Datum
1041 setval_oid(PG_FUNCTION_ARGS)
1042 {
1043         Oid                     relid = PG_GETARG_OID(0);
1044         int64           next = PG_GETARG_INT64(1);
1045
1046         do_setval(relid, next, true);
1047
1048         PG_RETURN_INT64(next);
1049 }
1050
1051 /*
1052  * Implement the 3 arg setval procedure.
1053  * See do_setval for discussion.
1054  */
1055 Datum
1056 setval3_oid(PG_FUNCTION_ARGS)
1057 {
1058         Oid                     relid = PG_GETARG_OID(0);
1059         int64           next = PG_GETARG_INT64(1);
1060         bool            iscalled = PG_GETARG_BOOL(2);
1061
1062         do_setval(relid, next, iscalled);
1063
1064         PG_RETURN_INT64(next);
1065 }
1066
1067
1068 /*
1069  * Open the sequence and acquire lock if needed
1070  *
1071  * If we haven't touched the sequence already in this transaction,
1072  * we need to acquire a lock.  We arrange for the lock to
1073  * be owned by the top transaction, so that we don't need to do it
1074  * more than once per xact.
1075  */
1076 static Relation
1077 lock_and_open_sequence(SeqTable seq)
1078 {
1079         LocalTransactionId thislxid = MyProc->lxid;
1080
1081         /* Get the lock if not already held in this xact */
1082         if (seq->lxid != thislxid)
1083         {
1084                 ResourceOwner currentOwner;
1085
1086                 currentOwner = CurrentResourceOwner;
1087                 CurrentResourceOwner = TopTransactionResourceOwner;
1088
1089                 LockRelationOid(seq->relid, RowExclusiveLock);
1090
1091                 CurrentResourceOwner = currentOwner;
1092
1093                 /* Flag that we have a lock in the current xact */
1094                 seq->lxid = thislxid;
1095         }
1096
1097         /* We now know we have the lock, and can safely open the rel */
1098         return relation_open(seq->relid, NoLock);
1099 }
1100
1101 /*
1102  * Creates the hash table for storing sequence data
1103  */
1104 static void
1105 create_seq_hashtable(void)
1106 {
1107         HASHCTL         ctl;
1108
1109         ctl.keysize = sizeof(Oid);
1110         ctl.entrysize = sizeof(SeqTableData);
1111
1112         seqhashtab = hash_create("Sequence values", 16, &ctl,
1113                                                          HASH_ELEM | HASH_BLOBS);
1114 }
1115
1116 /*
1117  * Given a relation OID, open and lock the sequence.  p_elm and p_rel are
1118  * output parameters.
1119  */
1120 static void
1121 init_sequence(Oid relid, SeqTable *p_elm, Relation *p_rel)
1122 {
1123         SeqTable        elm;
1124         Relation        seqrel;
1125         bool            found;
1126
1127         /* Find or create a hash table entry for this sequence */
1128         if (seqhashtab == NULL)
1129                 create_seq_hashtable();
1130
1131         elm = (SeqTable) hash_search(seqhashtab, &relid, HASH_ENTER, &found);
1132
1133         /*
1134          * Initialize the new hash table entry if it did not exist already.
1135          *
1136          * NOTE: seqhashtab entries are stored for the life of a backend (unless
1137          * explicitly discarded with DISCARD). If the sequence itself is deleted
1138          * then the entry becomes wasted memory, but it's small enough that this
1139          * should not matter.
1140          */
1141         if (!found)
1142         {
1143                 /* relid already filled in */
1144                 elm->filenumber = InvalidRelFileNumber;
1145                 elm->lxid = InvalidLocalTransactionId;
1146                 elm->last_valid = false;
1147                 elm->last = elm->cached = 0;
1148         }
1149
1150         /*
1151          * Open the sequence relation.
1152          */
1153         seqrel = lock_and_open_sequence(elm);
1154
1155         if (seqrel->rd_rel->relkind != RELKIND_SEQUENCE)
1156                 ereport(ERROR,
1157                                 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1158                                  errmsg("\"%s\" is not a sequence",
1159                                                 RelationGetRelationName(seqrel))));
1160
1161         /*
1162          * If the sequence has been transactionally replaced since we last saw it,
1163          * discard any cached-but-unissued values.  We do not touch the currval()
1164          * state, however.
1165          */
1166         if (seqrel->rd_rel->relfilenode != elm->filenumber)
1167         {
1168                 elm->filenumber = seqrel->rd_rel->relfilenode;
1169                 elm->cached = elm->last;
1170         }
1171
1172         /* Return results */
1173         *p_elm = elm;
1174         *p_rel = seqrel;
1175 }
1176
1177
1178 /*
1179  * Given an opened sequence relation, lock the page buffer and find the tuple
1180  *
1181  * *buf receives the reference to the pinned-and-ex-locked buffer
1182  * *seqdatatuple receives the reference to the sequence tuple proper
1183  *              (this arg should point to a local variable of type HeapTupleData)
1184  *
1185  * Function's return value points to the data payload of the tuple
1186  */
1187 static Form_pg_sequence_data
1188 read_seq_tuple(Relation rel, Buffer *buf, HeapTuple seqdatatuple)
1189 {
1190         Page            page;
1191         ItemId          lp;
1192         sequence_magic *sm;
1193         Form_pg_sequence_data seq;
1194
1195         *buf = ReadBuffer(rel, 0);
1196         LockBuffer(*buf, BUFFER_LOCK_EXCLUSIVE);
1197
1198         page = BufferGetPage(*buf);
1199         sm = (sequence_magic *) PageGetSpecialPointer(page);
1200
1201         if (sm->magic != SEQ_MAGIC)
1202                 elog(ERROR, "bad magic number in sequence \"%s\": %08X",
1203                          RelationGetRelationName(rel), sm->magic);
1204
1205         lp = PageGetItemId(page, FirstOffsetNumber);
1206         Assert(ItemIdIsNormal(lp));
1207
1208         /* Note we currently only bother to set these two fields of *seqdatatuple */
1209         seqdatatuple->t_data = (HeapTupleHeader) PageGetItem(page, lp);
1210         seqdatatuple->t_len = ItemIdGetLength(lp);
1211
1212         /*
1213          * Previous releases of Postgres neglected to prevent SELECT FOR UPDATE on
1214          * a sequence, which would leave a non-frozen XID in the sequence tuple's
1215          * xmax, which eventually leads to clog access failures or worse. If we
1216          * see this has happened, clean up after it.  We treat this like a hint
1217          * bit update, ie, don't bother to WAL-log it, since we can certainly do
1218          * this again if the update gets lost.
1219          */
1220         Assert(!(seqdatatuple->t_data->t_infomask & HEAP_XMAX_IS_MULTI));
1221         if (HeapTupleHeaderGetRawXmax(seqdatatuple->t_data) != InvalidTransactionId)
1222         {
1223                 HeapTupleHeaderSetXmax(seqdatatuple->t_data, InvalidTransactionId);
1224                 seqdatatuple->t_data->t_infomask &= ~HEAP_XMAX_COMMITTED;
1225                 seqdatatuple->t_data->t_infomask |= HEAP_XMAX_INVALID;
1226                 MarkBufferDirtyHint(*buf, true);
1227         }
1228
1229         seq = (Form_pg_sequence_data) GETSTRUCT(seqdatatuple);
1230
1231         return seq;
1232 }
1233
1234 /*
1235  * init_params: process the options list of CREATE or ALTER SEQUENCE, and
1236  * store the values into appropriate fields of seqform, for changes that go
1237  * into the pg_sequence catalog, and fields of seqdataform for changes to the
1238  * sequence relation itself.  Set *need_seq_rewrite to true if we changed any
1239  * parameters that require rewriting the sequence's relation (interesting for
1240  * ALTER SEQUENCE).  Also set *owned_by to any OWNED BY option, or to NIL if
1241  * there is none.
1242  *
1243  * If isInit is true, fill any unspecified options with default values;
1244  * otherwise, do not change existing options that aren't explicitly overridden.
1245  *
1246  * Note: we force a sequence rewrite whenever we change parameters that affect
1247  * generation of future sequence values, even if the seqdataform per se is not
1248  * changed.  This allows ALTER SEQUENCE to behave transactionally.  Currently,
1249  * the only option that doesn't cause that is OWNED BY.  It's *necessary* for
1250  * ALTER SEQUENCE OWNED BY to not rewrite the sequence, because that would
1251  * break pg_upgrade by causing unwanted changes in the sequence's
1252  * relfilenumber.
1253  */
1254 static void
1255 init_params(ParseState *pstate, List *options, bool for_identity,
1256                         bool isInit,
1257                         Form_pg_sequence seqform,
1258                         Form_pg_sequence_data seqdataform,
1259                         bool *need_seq_rewrite,
1260                         List **owned_by)
1261 {
1262         DefElem    *as_type = NULL;
1263         DefElem    *start_value = NULL;
1264         DefElem    *restart_value = NULL;
1265         DefElem    *increment_by = NULL;
1266         DefElem    *max_value = NULL;
1267         DefElem    *min_value = NULL;
1268         DefElem    *cache_value = NULL;
1269         DefElem    *is_cycled = NULL;
1270         ListCell   *option;
1271         bool            reset_max_value = false;
1272         bool            reset_min_value = false;
1273
1274         *need_seq_rewrite = false;
1275         *owned_by = NIL;
1276
1277         foreach(option, options)
1278         {
1279                 DefElem    *defel = (DefElem *) lfirst(option);
1280
1281                 if (strcmp(defel->defname, "as") == 0)
1282                 {
1283                         if (as_type)
1284                                 errorConflictingDefElem(defel, pstate);
1285                         as_type = defel;
1286                         *need_seq_rewrite = true;
1287                 }
1288                 else if (strcmp(defel->defname, "increment") == 0)
1289                 {
1290                         if (increment_by)
1291                                 errorConflictingDefElem(defel, pstate);
1292                         increment_by = defel;
1293                         *need_seq_rewrite = true;
1294                 }
1295                 else if (strcmp(defel->defname, "start") == 0)
1296                 {
1297                         if (start_value)
1298                                 errorConflictingDefElem(defel, pstate);
1299                         start_value = defel;
1300                         *need_seq_rewrite = true;
1301                 }
1302                 else if (strcmp(defel->defname, "restart") == 0)
1303                 {
1304                         if (restart_value)
1305                                 errorConflictingDefElem(defel, pstate);
1306                         restart_value = defel;
1307                         *need_seq_rewrite = true;
1308                 }
1309                 else if (strcmp(defel->defname, "maxvalue") == 0)
1310                 {
1311                         if (max_value)
1312                                 errorConflictingDefElem(defel, pstate);
1313                         max_value = defel;
1314                         *need_seq_rewrite = true;
1315                 }
1316                 else if (strcmp(defel->defname, "minvalue") == 0)
1317                 {
1318                         if (min_value)
1319                                 errorConflictingDefElem(defel, pstate);
1320                         min_value = defel;
1321                         *need_seq_rewrite = true;
1322                 }
1323                 else if (strcmp(defel->defname, "cache") == 0)
1324                 {
1325                         if (cache_value)
1326                                 errorConflictingDefElem(defel, pstate);
1327                         cache_value = defel;
1328                         *need_seq_rewrite = true;
1329                 }
1330                 else if (strcmp(defel->defname, "cycle") == 0)
1331                 {
1332                         if (is_cycled)
1333                                 errorConflictingDefElem(defel, pstate);
1334                         is_cycled = defel;
1335                         *need_seq_rewrite = true;
1336                 }
1337                 else if (strcmp(defel->defname, "owned_by") == 0)
1338                 {
1339                         if (*owned_by)
1340                                 errorConflictingDefElem(defel, pstate);
1341                         *owned_by = defGetQualifiedName(defel);
1342                 }
1343                 else if (strcmp(defel->defname, "sequence_name") == 0)
1344                 {
1345                         /*
1346                          * The parser allows this, but it is only for identity columns, in
1347                          * which case it is filtered out in parse_utilcmd.c.  We only get
1348                          * here if someone puts it into a CREATE SEQUENCE.
1349                          */
1350                         ereport(ERROR,
1351                                         (errcode(ERRCODE_SYNTAX_ERROR),
1352                                          errmsg("invalid sequence option SEQUENCE NAME"),
1353                                          parser_errposition(pstate, defel->location)));
1354                 }
1355                 else
1356                         elog(ERROR, "option \"%s\" not recognized",
1357                                  defel->defname);
1358         }
1359
1360         /*
1361          * We must reset log_cnt when isInit or when changing any parameters that
1362          * would affect future nextval allocations.
1363          */
1364         if (isInit)
1365                 seqdataform->log_cnt = 0;
1366
1367         /* AS type */
1368         if (as_type != NULL)
1369         {
1370                 Oid                     newtypid = typenameTypeId(pstate, defGetTypeName(as_type));
1371
1372                 if (newtypid != INT2OID &&
1373                         newtypid != INT4OID &&
1374                         newtypid != INT8OID)
1375                         ereport(ERROR,
1376                                         (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1377                                          for_identity
1378                                          ? errmsg("identity column type must be smallint, integer, or bigint")
1379                                          : errmsg("sequence type must be smallint, integer, or bigint")));
1380
1381                 if (!isInit)
1382                 {
1383                         /*
1384                          * When changing type and the old sequence min/max values were the
1385                          * min/max of the old type, adjust sequence min/max values to
1386                          * min/max of new type.  (Otherwise, the user chose explicit
1387                          * min/max values, which we'll leave alone.)
1388                          */
1389                         if ((seqform->seqtypid == INT2OID && seqform->seqmax == PG_INT16_MAX) ||
1390                                 (seqform->seqtypid == INT4OID && seqform->seqmax == PG_INT32_MAX) ||
1391                                 (seqform->seqtypid == INT8OID && seqform->seqmax == PG_INT64_MAX))
1392                                 reset_max_value = true;
1393                         if ((seqform->seqtypid == INT2OID && seqform->seqmin == PG_INT16_MIN) ||
1394                                 (seqform->seqtypid == INT4OID && seqform->seqmin == PG_INT32_MIN) ||
1395                                 (seqform->seqtypid == INT8OID && seqform->seqmin == PG_INT64_MIN))
1396                                 reset_min_value = true;
1397                 }
1398
1399                 seqform->seqtypid = newtypid;
1400         }
1401         else if (isInit)
1402         {
1403                 seqform->seqtypid = INT8OID;
1404         }
1405
1406         /* INCREMENT BY */
1407         if (increment_by != NULL)
1408         {
1409                 seqform->seqincrement = defGetInt64(increment_by);
1410                 if (seqform->seqincrement == 0)
1411                         ereport(ERROR,
1412                                         (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1413                                          errmsg("INCREMENT must not be zero")));
1414                 seqdataform->log_cnt = 0;
1415         }
1416         else if (isInit)
1417         {
1418                 seqform->seqincrement = 1;
1419         }
1420
1421         /* CYCLE */
1422         if (is_cycled != NULL)
1423         {
1424                 seqform->seqcycle = boolVal(is_cycled->arg);
1425                 Assert(BoolIsValid(seqform->seqcycle));
1426                 seqdataform->log_cnt = 0;
1427         }
1428         else if (isInit)
1429         {
1430                 seqform->seqcycle = false;
1431         }
1432
1433         /* MAXVALUE (null arg means NO MAXVALUE) */
1434         if (max_value != NULL && max_value->arg)
1435         {
1436                 seqform->seqmax = defGetInt64(max_value);
1437                 seqdataform->log_cnt = 0;
1438         }
1439         else if (isInit || max_value != NULL || reset_max_value)
1440         {
1441                 if (seqform->seqincrement > 0 || reset_max_value)
1442                 {
1443                         /* ascending seq */
1444                         if (seqform->seqtypid == INT2OID)
1445                                 seqform->seqmax = PG_INT16_MAX;
1446                         else if (seqform->seqtypid == INT4OID)
1447                                 seqform->seqmax = PG_INT32_MAX;
1448                         else
1449                                 seqform->seqmax = PG_INT64_MAX;
1450                 }
1451                 else
1452                         seqform->seqmax = -1;   /* descending seq */
1453                 seqdataform->log_cnt = 0;
1454         }
1455
1456         /* Validate maximum value.  No need to check INT8 as seqmax is an int64 */
1457         if ((seqform->seqtypid == INT2OID && (seqform->seqmax < PG_INT16_MIN || seqform->seqmax > PG_INT16_MAX))
1458                 || (seqform->seqtypid == INT4OID && (seqform->seqmax < PG_INT32_MIN || seqform->seqmax > PG_INT32_MAX)))
1459                 ereport(ERROR,
1460                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1461                                  errmsg("MAXVALUE (%lld) is out of range for sequence data type %s",
1462                                                 (long long) seqform->seqmax,
1463                                                 format_type_be(seqform->seqtypid))));
1464
1465         /* MINVALUE (null arg means NO MINVALUE) */
1466         if (min_value != NULL && min_value->arg)
1467         {
1468                 seqform->seqmin = defGetInt64(min_value);
1469                 seqdataform->log_cnt = 0;
1470         }
1471         else if (isInit || min_value != NULL || reset_min_value)
1472         {
1473                 if (seqform->seqincrement < 0 || reset_min_value)
1474                 {
1475                         /* descending seq */
1476                         if (seqform->seqtypid == INT2OID)
1477                                 seqform->seqmin = PG_INT16_MIN;
1478                         else if (seqform->seqtypid == INT4OID)
1479                                 seqform->seqmin = PG_INT32_MIN;
1480                         else
1481                                 seqform->seqmin = PG_INT64_MIN;
1482                 }
1483                 else
1484                         seqform->seqmin = 1;    /* ascending seq */
1485                 seqdataform->log_cnt = 0;
1486         }
1487
1488         /* Validate minimum value.  No need to check INT8 as seqmin is an int64 */
1489         if ((seqform->seqtypid == INT2OID && (seqform->seqmin < PG_INT16_MIN || seqform->seqmin > PG_INT16_MAX))
1490                 || (seqform->seqtypid == INT4OID && (seqform->seqmin < PG_INT32_MIN || seqform->seqmin > PG_INT32_MAX)))
1491                 ereport(ERROR,
1492                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1493                                  errmsg("MINVALUE (%lld) is out of range for sequence data type %s",
1494                                                 (long long) seqform->seqmin,
1495                                                 format_type_be(seqform->seqtypid))));
1496
1497         /* crosscheck min/max */
1498         if (seqform->seqmin >= seqform->seqmax)
1499                 ereport(ERROR,
1500                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1501                                  errmsg("MINVALUE (%lld) must be less than MAXVALUE (%lld)",
1502                                                 (long long) seqform->seqmin,
1503                                                 (long long) seqform->seqmax)));
1504
1505         /* START WITH */
1506         if (start_value != NULL)
1507         {
1508                 seqform->seqstart = defGetInt64(start_value);
1509         }
1510         else if (isInit)
1511         {
1512                 if (seqform->seqincrement > 0)
1513                         seqform->seqstart = seqform->seqmin;    /* ascending seq */
1514                 else
1515                         seqform->seqstart = seqform->seqmax;    /* descending seq */
1516         }
1517
1518         /* crosscheck START */
1519         if (seqform->seqstart < seqform->seqmin)
1520                 ereport(ERROR,
1521                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1522                                  errmsg("START value (%lld) cannot be less than MINVALUE (%lld)",
1523                                                 (long long) seqform->seqstart,
1524                                                 (long long) seqform->seqmin)));
1525         if (seqform->seqstart > seqform->seqmax)
1526                 ereport(ERROR,
1527                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1528                                  errmsg("START value (%lld) cannot be greater than MAXVALUE (%lld)",
1529                                                 (long long) seqform->seqstart,
1530                                                 (long long) seqform->seqmax)));
1531
1532         /* RESTART [WITH] */
1533         if (restart_value != NULL)
1534         {
1535                 if (restart_value->arg != NULL)
1536                         seqdataform->last_value = defGetInt64(restart_value);
1537                 else
1538                         seqdataform->last_value = seqform->seqstart;
1539                 seqdataform->is_called = false;
1540                 seqdataform->log_cnt = 0;
1541         }
1542         else if (isInit)
1543         {
1544                 seqdataform->last_value = seqform->seqstart;
1545                 seqdataform->is_called = false;
1546         }
1547
1548         /* crosscheck RESTART (or current value, if changing MIN/MAX) */
1549         if (seqdataform->last_value < seqform->seqmin)
1550                 ereport(ERROR,
1551                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1552                                  errmsg("RESTART value (%lld) cannot be less than MINVALUE (%lld)",
1553                                                 (long long) seqdataform->last_value,
1554                                                 (long long) seqform->seqmin)));
1555         if (seqdataform->last_value > seqform->seqmax)
1556                 ereport(ERROR,
1557                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1558                                  errmsg("RESTART value (%lld) cannot be greater than MAXVALUE (%lld)",
1559                                                 (long long) seqdataform->last_value,
1560                                                 (long long) seqform->seqmax)));
1561
1562         /* CACHE */
1563         if (cache_value != NULL)
1564         {
1565                 seqform->seqcache = defGetInt64(cache_value);
1566                 if (seqform->seqcache <= 0)
1567                         ereport(ERROR,
1568                                         (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1569                                          errmsg("CACHE (%lld) must be greater than zero",
1570                                                         (long long) seqform->seqcache)));
1571                 seqdataform->log_cnt = 0;
1572         }
1573         else if (isInit)
1574         {
1575                 seqform->seqcache = 1;
1576         }
1577 }
1578
1579 /*
1580  * Process an OWNED BY option for CREATE/ALTER SEQUENCE
1581  *
1582  * Ownership permissions on the sequence are already checked,
1583  * but if we are establishing a new owned-by dependency, we must
1584  * enforce that the referenced table has the same owner and namespace
1585  * as the sequence.
1586  */
1587 static void
1588 process_owned_by(Relation seqrel, List *owned_by, bool for_identity)
1589 {
1590         DependencyType deptype;
1591         int                     nnames;
1592         Relation        tablerel;
1593         AttrNumber      attnum;
1594
1595         deptype = for_identity ? DEPENDENCY_INTERNAL : DEPENDENCY_AUTO;
1596
1597         nnames = list_length(owned_by);
1598         Assert(nnames > 0);
1599         if (nnames == 1)
1600         {
1601                 /* Must be OWNED BY NONE */
1602                 if (strcmp(strVal(linitial(owned_by)), "none") != 0)
1603                         ereport(ERROR,
1604                                         (errcode(ERRCODE_SYNTAX_ERROR),
1605                                          errmsg("invalid OWNED BY option"),
1606                                          errhint("Specify OWNED BY table.column or OWNED BY NONE.")));
1607                 tablerel = NULL;
1608                 attnum = 0;
1609         }
1610         else
1611         {
1612                 List       *relname;
1613                 char       *attrname;
1614                 RangeVar   *rel;
1615
1616                 /* Separate relname and attr name */
1617                 relname = list_copy_head(owned_by, nnames - 1);
1618                 attrname = strVal(llast(owned_by));
1619
1620                 /* Open and lock rel to ensure it won't go away meanwhile */
1621                 rel = makeRangeVarFromNameList(relname);
1622                 tablerel = relation_openrv(rel, AccessShareLock);
1623
1624                 /* Must be a regular or foreign table */
1625                 if (!(tablerel->rd_rel->relkind == RELKIND_RELATION ||
1626                           tablerel->rd_rel->relkind == RELKIND_FOREIGN_TABLE ||
1627                           tablerel->rd_rel->relkind == RELKIND_VIEW ||
1628                           tablerel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE))
1629                         ereport(ERROR,
1630                                         (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1631                                          errmsg("sequence cannot be owned by relation \"%s\"",
1632                                                         RelationGetRelationName(tablerel)),
1633                                          errdetail_relkind_not_supported(tablerel->rd_rel->relkind)));
1634
1635                 /* We insist on same owner and schema */
1636                 if (seqrel->rd_rel->relowner != tablerel->rd_rel->relowner)
1637                         ereport(ERROR,
1638                                         (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1639                                          errmsg("sequence must have same owner as table it is linked to")));
1640                 if (RelationGetNamespace(seqrel) != RelationGetNamespace(tablerel))
1641                         ereport(ERROR,
1642                                         (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1643                                          errmsg("sequence must be in same schema as table it is linked to")));
1644
1645                 /* Now, fetch the attribute number from the system cache */
1646                 attnum = get_attnum(RelationGetRelid(tablerel), attrname);
1647                 if (attnum == InvalidAttrNumber)
1648                         ereport(ERROR,
1649                                         (errcode(ERRCODE_UNDEFINED_COLUMN),
1650                                          errmsg("column \"%s\" of relation \"%s\" does not exist",
1651                                                         attrname, RelationGetRelationName(tablerel))));
1652         }
1653
1654         /*
1655          * Catch user explicitly running OWNED BY on identity sequence.
1656          */
1657         if (deptype == DEPENDENCY_AUTO)
1658         {
1659                 Oid                     tableId;
1660                 int32           colId;
1661
1662                 if (sequenceIsOwned(RelationGetRelid(seqrel), DEPENDENCY_INTERNAL, &tableId, &colId))
1663                         ereport(ERROR,
1664                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1665                                          errmsg("cannot change ownership of identity sequence"),
1666                                          errdetail("Sequence \"%s\" is linked to table \"%s\".",
1667                                                            RelationGetRelationName(seqrel),
1668                                                            get_rel_name(tableId))));
1669         }
1670
1671         /*
1672          * OK, we are ready to update pg_depend.  First remove any existing
1673          * dependencies for the sequence, then optionally add a new one.
1674          */
1675         deleteDependencyRecordsForClass(RelationRelationId, RelationGetRelid(seqrel),
1676                                                                         RelationRelationId, deptype);
1677
1678         if (tablerel)
1679         {
1680                 ObjectAddress refobject,
1681                                         depobject;
1682
1683                 refobject.classId = RelationRelationId;
1684                 refobject.objectId = RelationGetRelid(tablerel);
1685                 refobject.objectSubId = attnum;
1686                 depobject.classId = RelationRelationId;
1687                 depobject.objectId = RelationGetRelid(seqrel);
1688                 depobject.objectSubId = 0;
1689                 recordDependencyOn(&depobject, &refobject, deptype);
1690         }
1691
1692         /* Done, but hold lock until commit */
1693         if (tablerel)
1694                 relation_close(tablerel, NoLock);
1695 }
1696
1697
1698 /*
1699  * Return sequence parameters in a list of the form created by the parser.
1700  */
1701 List *
1702 sequence_options(Oid relid)
1703 {
1704         HeapTuple       pgstuple;
1705         Form_pg_sequence pgsform;
1706         List       *options = NIL;
1707
1708         pgstuple = SearchSysCache1(SEQRELID, ObjectIdGetDatum(relid));
1709         if (!HeapTupleIsValid(pgstuple))
1710                 elog(ERROR, "cache lookup failed for sequence %u", relid);
1711         pgsform = (Form_pg_sequence) GETSTRUCT(pgstuple);
1712
1713         /* Use makeFloat() for 64-bit integers, like gram.y does. */
1714         options = lappend(options,
1715                                           makeDefElem("cache", (Node *) makeFloat(psprintf(INT64_FORMAT, pgsform->seqcache)), -1));
1716         options = lappend(options,
1717                                           makeDefElem("cycle", (Node *) makeBoolean(pgsform->seqcycle), -1));
1718         options = lappend(options,
1719                                           makeDefElem("increment", (Node *) makeFloat(psprintf(INT64_FORMAT, pgsform->seqincrement)), -1));
1720         options = lappend(options,
1721                                           makeDefElem("maxvalue", (Node *) makeFloat(psprintf(INT64_FORMAT, pgsform->seqmax)), -1));
1722         options = lappend(options,
1723                                           makeDefElem("minvalue", (Node *) makeFloat(psprintf(INT64_FORMAT, pgsform->seqmin)), -1));
1724         options = lappend(options,
1725                                           makeDefElem("start", (Node *) makeFloat(psprintf(INT64_FORMAT, pgsform->seqstart)), -1));
1726
1727         ReleaseSysCache(pgstuple);
1728
1729         return options;
1730 }
1731
1732 /*
1733  * Return sequence parameters (formerly for use by information schema)
1734  */
1735 Datum
1736 pg_sequence_parameters(PG_FUNCTION_ARGS)
1737 {
1738         Oid                     relid = PG_GETARG_OID(0);
1739         TupleDesc       tupdesc;
1740         Datum           values[7];
1741         bool            isnull[7];
1742         HeapTuple       pgstuple;
1743         Form_pg_sequence pgsform;
1744
1745         if (pg_class_aclcheck(relid, GetUserId(), ACL_SELECT | ACL_UPDATE | ACL_USAGE) != ACLCHECK_OK)
1746                 ereport(ERROR,
1747                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1748                                  errmsg("permission denied for sequence %s",
1749                                                 get_rel_name(relid))));
1750
1751         if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
1752                 elog(ERROR, "return type must be a row type");
1753
1754         memset(isnull, 0, sizeof(isnull));
1755
1756         pgstuple = SearchSysCache1(SEQRELID, ObjectIdGetDatum(relid));
1757         if (!HeapTupleIsValid(pgstuple))
1758                 elog(ERROR, "cache lookup failed for sequence %u", relid);
1759         pgsform = (Form_pg_sequence) GETSTRUCT(pgstuple);
1760
1761         values[0] = Int64GetDatum(pgsform->seqstart);
1762         values[1] = Int64GetDatum(pgsform->seqmin);
1763         values[2] = Int64GetDatum(pgsform->seqmax);
1764         values[3] = Int64GetDatum(pgsform->seqincrement);
1765         values[4] = BoolGetDatum(pgsform->seqcycle);
1766         values[5] = Int64GetDatum(pgsform->seqcache);
1767         values[6] = ObjectIdGetDatum(pgsform->seqtypid);
1768
1769         ReleaseSysCache(pgstuple);
1770
1771         return HeapTupleGetDatum(heap_form_tuple(tupdesc, values, isnull));
1772 }
1773
1774 /*
1775  * Return the last value from the sequence
1776  *
1777  * Note: This has a completely different meaning than lastval().
1778  */
1779 Datum
1780 pg_sequence_last_value(PG_FUNCTION_ARGS)
1781 {
1782         Oid                     relid = PG_GETARG_OID(0);
1783         SeqTable        elm;
1784         Relation        seqrel;
1785         Buffer          buf;
1786         HeapTupleData seqtuple;
1787         Form_pg_sequence_data seq;
1788         bool            is_called;
1789         int64           result;
1790
1791         /* open and lock sequence */
1792         init_sequence(relid, &elm, &seqrel);
1793
1794         if (pg_class_aclcheck(relid, GetUserId(), ACL_SELECT | ACL_USAGE) != ACLCHECK_OK)
1795                 ereport(ERROR,
1796                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1797                                  errmsg("permission denied for sequence %s",
1798                                                 RelationGetRelationName(seqrel))));
1799
1800         seq = read_seq_tuple(seqrel, &buf, &seqtuple);
1801
1802         is_called = seq->is_called;
1803         result = seq->last_value;
1804
1805         UnlockReleaseBuffer(buf);
1806         relation_close(seqrel, NoLock);
1807
1808         if (is_called)
1809                 PG_RETURN_INT64(result);
1810         else
1811                 PG_RETURN_NULL();
1812 }
1813
1814
1815 void
1816 seq_redo(XLogReaderState *record)
1817 {
1818         XLogRecPtr      lsn = record->EndRecPtr;
1819         uint8           info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
1820         Buffer          buffer;
1821         Page            page;
1822         Page            localpage;
1823         char       *item;
1824         Size            itemsz;
1825         xl_seq_rec *xlrec = (xl_seq_rec *) XLogRecGetData(record);
1826         sequence_magic *sm;
1827
1828         if (info != XLOG_SEQ_LOG)
1829                 elog(PANIC, "seq_redo: unknown op code %u", info);
1830
1831         buffer = XLogInitBufferForRedo(record, 0);
1832         page = (Page) BufferGetPage(buffer);
1833
1834         /*
1835          * We always reinit the page.  However, since this WAL record type is also
1836          * used for updating sequences, it's possible that a hot-standby backend
1837          * is examining the page concurrently; so we mustn't transiently trash the
1838          * buffer.  The solution is to build the correct new page contents in
1839          * local workspace and then memcpy into the buffer.  Then only bytes that
1840          * are supposed to change will change, even transiently. We must palloc
1841          * the local page for alignment reasons.
1842          */
1843         localpage = (Page) palloc(BufferGetPageSize(buffer));
1844
1845         PageInit(localpage, BufferGetPageSize(buffer), sizeof(sequence_magic));
1846         sm = (sequence_magic *) PageGetSpecialPointer(localpage);
1847         sm->magic = SEQ_MAGIC;
1848
1849         item = (char *) xlrec + sizeof(xl_seq_rec);
1850         itemsz = XLogRecGetDataLen(record) - sizeof(xl_seq_rec);
1851
1852         if (PageAddItem(localpage, (Item) item, itemsz,
1853                                         FirstOffsetNumber, false, false) == InvalidOffsetNumber)
1854                 elog(PANIC, "seq_redo: failed to add item to page");
1855
1856         PageSetLSN(localpage, lsn);
1857
1858         memcpy(page, localpage, BufferGetPageSize(buffer));
1859         MarkBufferDirty(buffer);
1860         UnlockReleaseBuffer(buffer);
1861
1862         pfree(localpage);
1863 }
1864
1865 /*
1866  * Flush cached sequence information.
1867  */
1868 void
1869 ResetSequenceCaches(void)
1870 {
1871         if (seqhashtab)
1872         {
1873                 hash_destroy(seqhashtab);
1874                 seqhashtab = NULL;
1875         }
1876
1877         last_used_seq = NULL;
1878 }
1879
1880 /*
1881  * Mask a Sequence page before performing consistency checks on it.
1882  */
1883 void
1884 seq_mask(char *page, BlockNumber blkno)
1885 {
1886         mask_page_lsn_and_checksum(page);
1887
1888         mask_unused_space(page);
1889 }