granicus.if.org Git - postgresql/blob - src/backend/commands/sequence.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * sequence.c
   4  *        PostgreSQL sequences support code.
   5  *
   6  * Portions Copyright (c) 1996-2006, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        $PostgreSQL: pgsql/src/backend/commands/sequence.c,v 1.141 2006/10/06 17:13:58 petere Exp $
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 #include "postgres.h"
  16
  17 #include "access/heapam.h"
  18 #include "access/transam.h"
  19 #include "access/xact.h"
  20 #include "catalog/dependency.h"
  21 #include "catalog/namespace.h"
  22 #include "catalog/pg_type.h"
  23 #include "commands/defrem.h"
  24 #include "commands/sequence.h"
  25 #include "commands/tablecmds.h"
  26 #include "miscadmin.h"
  27 #include "nodes/makefuncs.h"
  28 #include "utils/acl.h"
  29 #include "utils/builtins.h"
  30 #include "utils/lsyscache.h"
  31 #include "utils/resowner.h"
  32 #include "utils/syscache.h"
  33
  34
  35 /*
  36  * We don't want to log each fetching of a value from a sequence,
  37  * so we pre-log a few fetches in advance. In the event of
  38  * crash we can lose as much as we pre-logged.
  39  */
  40 #define SEQ_LOG_VALS    32
  41
  42 /*
  43  * The "special area" of a sequence's buffer page looks like this.
  44  */
  45 #define SEQ_MAGIC         0x1717
  46
  47 typedef struct sequence_magic
  48 {
  49         uint32          magic;
  50 } sequence_magic;
  51
  52 /*
  53  * We store a SeqTable item for every sequence we have touched in the current
  54  * session.  This is needed to hold onto nextval/currval state.  (We can't
  55  * rely on the relcache, since it's only, well, a cache, and may decide to
  56  * discard entries.)
  57  *
  58  * XXX We use linear search to find pre-existing SeqTable entries.      This is
  59  * good when only a small number of sequences are touched in a session, but
  60  * would suck with many different sequences.  Perhaps use a hashtable someday.
  61  */
  62 typedef struct SeqTableData
  63 {
  64         struct SeqTableData *next;      /* link to next SeqTable object */
  65         Oid                     relid;                  /* pg_class OID of this sequence */
  66         TransactionId xid;                      /* xact in which we last did a seq op */
  67         int64           last;                   /* value last returned by nextval */
  68         int64           cached;                 /* last value already cached for nextval */
  69         /* if last != cached, we have not used up all the cached values */
  70         int64           increment;              /* copy of sequence's increment field */
  71 } SeqTableData;
  72
  73 typedef SeqTableData *SeqTable;
  74
  75 static SeqTable seqtab = NULL;  /* Head of list of SeqTable items */
  76
  77 /*
  78  * last_used_seq is updated by nextval() to point to the last used
  79  * sequence.
  80  */
  81 static SeqTableData *last_used_seq = NULL;
  82
  83 static int64 nextval_internal(Oid relid);
  84 static Relation open_share_lock(SeqTable seq);
  85 static void init_sequence(Oid relid, SeqTable *p_elm, Relation *p_rel);
  86 static Form_pg_sequence read_info(SeqTable elm, Relation rel, Buffer *buf);
  87 static void init_params(List *options, bool isInit,
  88                         Form_pg_sequence new, List **owned_by);
  89 static void do_setval(Oid relid, int64 next, bool iscalled);
  90 static void process_owned_by(Relation seqrel, List *owned_by);
  91
  92
  93 /*
  94  * DefineSequence
  95  *                              Creates a new sequence relation
  96  */
  97 void
  98 DefineSequence(CreateSeqStmt *seq)
  99 {
 100         FormData_pg_sequence new;
 101         List       *owned_by;
 102         CreateStmt *stmt = makeNode(CreateStmt);
 103         Oid                     seqoid;
 104         Relation        rel;
 105         Buffer          buf;
 106         PageHeader      page;
 107         sequence_magic *sm;
 108         HeapTuple       tuple;
 109         TupleDesc       tupDesc;
 110         Datum           value[SEQ_COL_LASTCOL];
 111         char            null[SEQ_COL_LASTCOL];
 112         int                     i;
 113         NameData        name;
 114
 115         /* Check and set all option values */
 116         init_params(seq->options, true, &new, &owned_by);
 117
 118         /*
 119          * Create relation (and fill *null & *value)
 120          */
 121         stmt->tableElts = NIL;
 122         for (i = SEQ_COL_FIRSTCOL; i <= SEQ_COL_LASTCOL; i++)
 123         {
 124                 ColumnDef  *coldef = makeNode(ColumnDef);
 125
 126                 coldef->inhcount = 0;
 127                 coldef->is_local = true;
 128                 coldef->is_not_null = true;
 129                 coldef->raw_default = NULL;
 130                 coldef->cooked_default = NULL;
 131                 coldef->constraints = NIL;
 132
 133                 null[i - 1] = ' ';
 134
 135                 switch (i)
 136                 {
 137                         case SEQ_COL_NAME:
 138                                 coldef->typename = makeTypeNameFromOid(NAMEOID, -1);
 139                                 coldef->colname = "sequence_name";
 140                                 namestrcpy(&name, seq->sequence->relname);
 141                                 value[i - 1] = NameGetDatum(&name);
 142                                 break;
 143                         case SEQ_COL_LASTVAL:
 144                                 coldef->typename = makeTypeNameFromOid(INT8OID, -1);
 145                                 coldef->colname = "last_value";
 146                                 value[i - 1] = Int64GetDatumFast(new.last_value);
 147                                 break;
 148                         case SEQ_COL_INCBY:
 149                                 coldef->typename = makeTypeNameFromOid(INT8OID, -1);
 150                                 coldef->colname = "increment_by";
 151                                 value[i - 1] = Int64GetDatumFast(new.increment_by);
 152                                 break;
 153                         case SEQ_COL_MAXVALUE:
 154                                 coldef->typename = makeTypeNameFromOid(INT8OID, -1);
 155                                 coldef->colname = "max_value";
 156                                 value[i - 1] = Int64GetDatumFast(new.max_value);
 157                                 break;
 158                         case SEQ_COL_MINVALUE:
 159                                 coldef->typename = makeTypeNameFromOid(INT8OID, -1);
 160                                 coldef->colname = "min_value";
 161                                 value[i - 1] = Int64GetDatumFast(new.min_value);
 162                                 break;
 163                         case SEQ_COL_CACHE:
 164                                 coldef->typename = makeTypeNameFromOid(INT8OID, -1);
 165                                 coldef->colname = "cache_value";
 166                                 value[i - 1] = Int64GetDatumFast(new.cache_value);
 167                                 break;
 168                         case SEQ_COL_LOG:
 169                                 coldef->typename = makeTypeNameFromOid(INT8OID, -1);
 170                                 coldef->colname = "log_cnt";
 171                                 value[i - 1] = Int64GetDatum((int64) 1);
 172                                 break;
 173                         case SEQ_COL_CYCLE:
 174                                 coldef->typename = makeTypeNameFromOid(BOOLOID, -1);
 175                                 coldef->colname = "is_cycled";
 176                                 value[i - 1] = BoolGetDatum(new.is_cycled);
 177                                 break;
 178                         case SEQ_COL_CALLED:
 179                                 coldef->typename = makeTypeNameFromOid(BOOLOID, -1);
 180                                 coldef->colname = "is_called";
 181                                 value[i - 1] = BoolGetDatum(false);
 182                                 break;
 183                 }
 184                 stmt->tableElts = lappend(stmt->tableElts, coldef);
 185         }
 186
 187         stmt->relation = seq->sequence;
 188         stmt->inhRelations = NIL;
 189         stmt->constraints = NIL;
 190         stmt->options = list_make1(defWithOids(false));
 191         stmt->oncommit = ONCOMMIT_NOOP;
 192         stmt->tablespacename = NULL;
 193
 194         seqoid = DefineRelation(stmt, RELKIND_SEQUENCE);
 195
 196         rel = heap_open(seqoid, AccessExclusiveLock);
 197         tupDesc = RelationGetDescr(rel);
 198
 199         /* Initialize first page of relation with special magic number */
 200
 201         buf = ReadBuffer(rel, P_NEW);
 202         Assert(BufferGetBlockNumber(buf) == 0);
 203
 204         page = (PageHeader) BufferGetPage(buf);
 205
 206         PageInit((Page) page, BufferGetPageSize(buf), sizeof(sequence_magic));
 207         sm = (sequence_magic *) PageGetSpecialPointer(page);
 208         sm->magic = SEQ_MAGIC;
 209
 210         /* hack: ensure heap_insert will insert on the just-created page */
 211         rel->rd_targblock = 0;
 212
 213         /* Now form & insert sequence tuple */
 214         tuple = heap_formtuple(tupDesc, value, null);
 215         simple_heap_insert(rel, tuple);
 216
 217         Assert(ItemPointerGetOffsetNumber(&(tuple->t_self)) == FirstOffsetNumber);
 218
 219         /*
 220          * Two special hacks here:
 221          *
 222          * 1. Since VACUUM does not process sequences, we have to force the tuple
 223          * to have xmin = FrozenTransactionId now.      Otherwise it would become
 224          * invisible to SELECTs after 2G transactions.  It is okay to do this
 225          * because if the current transaction aborts, no other xact will ever
 226          * examine the sequence tuple anyway.
 227          *
 228          * 2. Even though heap_insert emitted a WAL log record, we have to emit an
 229          * XLOG_SEQ_LOG record too, since (a) the heap_insert record will not have
 230          * the right xmin, and (b) REDO of the heap_insert record would re-init
 231          * page and sequence magic number would be lost.  This means two log
 232          * records instead of one :-(
 233          */
 234         LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
 235
 236         START_CRIT_SECTION();
 237
 238         {
 239                 /*
 240                  * Note that the "tuple" structure is still just a local tuple record
 241                  * created by heap_formtuple; its t_data pointer doesn't point at the
 242                  * disk buffer.  To scribble on the disk buffer we need to fetch the
 243                  * item pointer.  But do the same to the local tuple, since that will
 244                  * be the source for the WAL log record, below.
 245                  */
 246                 ItemId          itemId;
 247                 Item            item;
 248
 249                 itemId = PageGetItemId((Page) page, FirstOffsetNumber);
 250                 item = PageGetItem((Page) page, itemId);
 251
 252                 HeapTupleHeaderSetXmin((HeapTupleHeader) item, FrozenTransactionId);
 253                 ((HeapTupleHeader) item)->t_infomask |= HEAP_XMIN_COMMITTED;
 254
 255                 HeapTupleHeaderSetXmin(tuple->t_data, FrozenTransactionId);
 256                 tuple->t_data->t_infomask |= HEAP_XMIN_COMMITTED;
 257         }
 258
 259         MarkBufferDirty(buf);
 260
 261         /* XLOG stuff */
 262         if (!rel->rd_istemp)
 263         {
 264                 xl_seq_rec      xlrec;
 265                 XLogRecPtr      recptr;
 266                 XLogRecData rdata[2];
 267                 Form_pg_sequence newseq = (Form_pg_sequence) GETSTRUCT(tuple);
 268
 269                 /* We do not log first nextval call, so "advance" sequence here */
 270                 /* Note we are scribbling on local tuple, not the disk buffer */
 271                 newseq->is_called = true;
 272                 newseq->log_cnt = 0;
 273
 274                 xlrec.node = rel->rd_node;
 275                 rdata[0].data = (char *) &xlrec;
 276                 rdata[0].len = sizeof(xl_seq_rec);
 277                 rdata[0].buffer = InvalidBuffer;
 278                 rdata[0].next = &(rdata[1]);
 279
 280                 rdata[1].data = (char *) tuple->t_data;
 281                 rdata[1].len = tuple->t_len;
 282                 rdata[1].buffer = InvalidBuffer;
 283                 rdata[1].next = NULL;
 284
 285                 recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG | XLOG_NO_TRAN, rdata);
 286
 287                 PageSetLSN(page, recptr);
 288                 PageSetTLI(page, ThisTimeLineID);
 289         }
 290
 291         END_CRIT_SECTION();
 292
 293         UnlockReleaseBuffer(buf);
 294
 295         /* process OWNED BY if given */
 296         if (owned_by)
 297                 process_owned_by(rel, owned_by);
 298
 299         heap_close(rel, NoLock);
 300 }
 301
 302 /*
 303  * AlterSequence
 304  *
 305  * Modify the definition of a sequence relation
 306  */
 307 void
 308 AlterSequence(AlterSeqStmt *stmt)
 309 {
 310         Oid                     relid;
 311         SeqTable        elm;
 312         Relation        seqrel;
 313         Buffer          buf;
 314         Page            page;
 315         Form_pg_sequence seq;
 316         FormData_pg_sequence new;
 317         List       *owned_by;
 318
 319         /* open and AccessShareLock sequence */
 320         relid = RangeVarGetRelid(stmt->sequence, false);
 321         init_sequence(relid, &elm, &seqrel);
 322
 323         /* allow ALTER to sequence owner only */
 324         if (!pg_class_ownercheck(elm->relid, GetUserId()))
 325                 aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
 326                                            stmt->sequence->relname);
 327
 328         /* lock page' buffer and read tuple into new sequence structure */
 329         seq = read_info(elm, seqrel, &buf);
 330         page = BufferGetPage(buf);
 331
 332         /* Copy old values of options into workspace */
 333         memcpy(&new, seq, sizeof(FormData_pg_sequence));
 334
 335         /* Check and set new values */
 336         init_params(stmt->options, false, &new, &owned_by);
 337
 338         /* Now okay to update the on-disk tuple */
 339         memcpy(seq, &new, sizeof(FormData_pg_sequence));
 340
 341         /* Clear local cache so that we don't think we have cached numbers */
 342         elm->last = new.last_value; /* last returned number */
 343         elm->cached = new.last_value;           /* last cached number (forget cached
 344                                                                                  * values) */
 345
 346         START_CRIT_SECTION();
 347
 348         MarkBufferDirty(buf);
 349
 350         /* XLOG stuff */
 351         if (!seqrel->rd_istemp)
 352         {
 353                 xl_seq_rec      xlrec;
 354                 XLogRecPtr      recptr;
 355                 XLogRecData rdata[2];
 356
 357                 xlrec.node = seqrel->rd_node;
 358                 rdata[0].data = (char *) &xlrec;
 359                 rdata[0].len = sizeof(xl_seq_rec);
 360                 rdata[0].buffer = InvalidBuffer;
 361                 rdata[0].next = &(rdata[1]);
 362
 363                 rdata[1].data = (char *) page + ((PageHeader) page)->pd_upper;
 364                 rdata[1].len = ((PageHeader) page)->pd_special -
 365                         ((PageHeader) page)->pd_upper;
 366                 rdata[1].buffer = InvalidBuffer;
 367                 rdata[1].next = NULL;
 368
 369                 recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG | XLOG_NO_TRAN, rdata);
 370
 371                 PageSetLSN(page, recptr);
 372                 PageSetTLI(page, ThisTimeLineID);
 373         }
 374
 375         END_CRIT_SECTION();
 376
 377         UnlockReleaseBuffer(buf);
 378
 379         /* process OWNED BY if given */
 380         if (owned_by)
 381                 process_owned_by(seqrel, owned_by);
 382
 383         relation_close(seqrel, NoLock);
 384 }
 385
 386
 387 /*
 388  * Note: nextval with a text argument is no longer exported as a pg_proc
 389  * entry, but we keep it around to ease porting of C code that may have
 390  * called the function directly.
 391  */
 392 Datum
 393 nextval(PG_FUNCTION_ARGS)
 394 {
 395         text       *seqin = PG_GETARG_TEXT_P(0);
 396         RangeVar   *sequence;
 397         Oid                     relid;
 398
 399         sequence = makeRangeVarFromNameList(textToQualifiedNameList(seqin));
 400         relid = RangeVarGetRelid(sequence, false);
 401
 402         PG_RETURN_INT64(nextval_internal(relid));
 403 }
 404
 405 Datum
 406 nextval_oid(PG_FUNCTION_ARGS)
 407 {
 408         Oid                     relid = PG_GETARG_OID(0);
 409
 410         PG_RETURN_INT64(nextval_internal(relid));
 411 }
 412
 413 static int64
 414 nextval_internal(Oid relid)
 415 {
 416         SeqTable        elm;
 417         Relation        seqrel;
 418         Buffer          buf;
 419         Page            page;
 420         Form_pg_sequence seq;
 421         int64           incby,
 422                                 maxv,
 423                                 minv,
 424                                 cache,
 425                                 log,
 426                                 fetch,
 427                                 last;
 428         int64           result,
 429                                 next,
 430                                 rescnt = 0;
 431         bool            logit = false;
 432
 433         /* open and AccessShareLock sequence */
 434         init_sequence(relid, &elm, &seqrel);
 435
 436         if (pg_class_aclcheck(elm->relid, GetUserId(), ACL_USAGE) != ACLCHECK_OK &&
 437                 pg_class_aclcheck(elm->relid, GetUserId(), ACL_UPDATE) != ACLCHECK_OK)
 438                 ereport(ERROR,
 439                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 440                                  errmsg("permission denied for sequence %s",
 441                                                 RelationGetRelationName(seqrel))));
 442
 443         if (elm->last != elm->cached)           /* some numbers were cached */
 444         {
 445                 last_used_seq = elm;
 446                 elm->last += elm->increment;
 447                 relation_close(seqrel, NoLock);
 448                 return elm->last;
 449         }
 450
 451         /* lock page' buffer and read tuple */
 452         seq = read_info(elm, seqrel, &buf);
 453         page = BufferGetPage(buf);
 454
 455         last = next = result = seq->last_value;
 456         incby = seq->increment_by;
 457         maxv = seq->max_value;
 458         minv = seq->min_value;
 459         fetch = cache = seq->cache_value;
 460         log = seq->log_cnt;
 461
 462         if (!seq->is_called)
 463         {
 464                 rescnt++;                               /* last_value if not called */
 465                 fetch--;
 466                 log--;
 467         }
 468
 469         /*
 470          * Decide whether we should emit a WAL log record.      If so, force up the
 471          * fetch count to grab SEQ_LOG_VALS more values than we actually need to
 472          * cache.  (These will then be usable without logging.)
 473          *
 474          * If this is the first nextval after a checkpoint, we must force a new
 475          * WAL record to be written anyway, else replay starting from the
 476          * checkpoint would fail to advance the sequence past the logged values.
 477          * In this case we may as well fetch extra values.
 478          */
 479         if (log < fetch)
 480         {
 481                 /* forced log to satisfy local demand for values */
 482                 fetch = log = fetch + SEQ_LOG_VALS;
 483                 logit = true;
 484         }
 485         else
 486         {
 487                 XLogRecPtr      redoptr = GetRedoRecPtr();
 488
 489                 if (XLByteLE(PageGetLSN(page), redoptr))
 490                 {
 491                         /* last update of seq was before checkpoint */
 492                         fetch = log = fetch + SEQ_LOG_VALS;
 493                         logit = true;
 494                 }
 495         }
 496
 497         while (fetch)                           /* try to fetch cache [+ log ] numbers */
 498         {
 499                 /*
 500                  * Check MAXVALUE for ascending sequences and MINVALUE for descending
 501                  * sequences
 502                  */
 503                 if (incby > 0)
 504                 {
 505                         /* ascending sequence */
 506                         if ((maxv >= 0 && next > maxv - incby) ||
 507                                 (maxv < 0 && next + incby > maxv))
 508                         {
 509                                 if (rescnt > 0)
 510                                         break;          /* stop fetching */
 511                                 if (!seq->is_cycled)
 512                                 {
 513                                         char            buf[100];
 514
 515                                         snprintf(buf, sizeof(buf), INT64_FORMAT, maxv);
 516                                         ereport(ERROR,
 517                                                   (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 518                                                    errmsg("nextval: reached maximum value of sequence \"%s\" (%s)",
 519                                                                   RelationGetRelationName(seqrel), buf)));
 520                                 }
 521                                 next = minv;
 522                         }
 523                         else
 524                                 next += incby;
 525                 }
 526                 else
 527                 {
 528                         /* descending sequence */
 529                         if ((minv < 0 && next < minv - incby) ||
 530                                 (minv >= 0 && next + incby < minv))
 531                         {
 532                                 if (rescnt > 0)
 533                                         break;          /* stop fetching */
 534                                 if (!seq->is_cycled)
 535                                 {
 536                                         char            buf[100];
 537
 538                                         snprintf(buf, sizeof(buf), INT64_FORMAT, minv);
 539                                         ereport(ERROR,
 540                                                   (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 541                                                    errmsg("nextval: reached minimum value of sequence \"%s\" (%s)",
 542                                                                   RelationGetRelationName(seqrel), buf)));
 543                                 }
 544                                 next = maxv;
 545                         }
 546                         else
 547                                 next += incby;
 548                 }
 549                 fetch--;
 550                 if (rescnt < cache)
 551                 {
 552                         log--;
 553                         rescnt++;
 554                         last = next;
 555                         if (rescnt == 1)        /* if it's first result - */
 556                                 result = next;  /* it's what to return */
 557                 }
 558         }
 559
 560         log -= fetch;                           /* adjust for any unfetched numbers */
 561         Assert(log >= 0);
 562
 563         /* save info in local cache */
 564         elm->last = result;                     /* last returned number */
 565         elm->cached = last;                     /* last fetched number */
 566
 567         last_used_seq = elm;
 568
 569         START_CRIT_SECTION();
 570
 571         MarkBufferDirty(buf);
 572
 573         /* XLOG stuff */
 574         if (logit && !seqrel->rd_istemp)
 575         {
 576                 xl_seq_rec      xlrec;
 577                 XLogRecPtr      recptr;
 578                 XLogRecData rdata[2];
 579
 580                 xlrec.node = seqrel->rd_node;
 581                 rdata[0].data = (char *) &xlrec;
 582                 rdata[0].len = sizeof(xl_seq_rec);
 583                 rdata[0].buffer = InvalidBuffer;
 584                 rdata[0].next = &(rdata[1]);
 585
 586                 /* set values that will be saved in xlog */
 587                 seq->last_value = next;
 588                 seq->is_called = true;
 589                 seq->log_cnt = 0;
 590
 591                 rdata[1].data = (char *) page + ((PageHeader) page)->pd_upper;
 592                 rdata[1].len = ((PageHeader) page)->pd_special -
 593                         ((PageHeader) page)->pd_upper;
 594                 rdata[1].buffer = InvalidBuffer;
 595                 rdata[1].next = NULL;
 596
 597                 recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG | XLOG_NO_TRAN, rdata);
 598
 599                 PageSetLSN(page, recptr);
 600                 PageSetTLI(page, ThisTimeLineID);
 601         }
 602
 603         /* update on-disk data */
 604         seq->last_value = last;         /* last fetched number */
 605         seq->is_called = true;
 606         seq->log_cnt = log;                     /* how much is logged */
 607
 608         END_CRIT_SECTION();
 609
 610         UnlockReleaseBuffer(buf);
 611
 612         relation_close(seqrel, NoLock);
 613
 614         return result;
 615 }
 616
 617 Datum
 618 currval_oid(PG_FUNCTION_ARGS)
 619 {
 620         Oid                     relid = PG_GETARG_OID(0);
 621         int64           result;
 622         SeqTable        elm;
 623         Relation        seqrel;
 624
 625         /* open and AccessShareLock sequence */
 626         init_sequence(relid, &elm, &seqrel);
 627
 628         if (pg_class_aclcheck(elm->relid, GetUserId(), ACL_SELECT) != ACLCHECK_OK &&
 629                 pg_class_aclcheck(elm->relid, GetUserId(), ACL_USAGE) != ACLCHECK_OK)
 630                 ereport(ERROR,
 631                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 632                                  errmsg("permission denied for sequence %s",
 633                                                 RelationGetRelationName(seqrel))));
 634
 635         if (elm->increment == 0)        /* nextval/read_info were not called */
 636                 ereport(ERROR,
 637                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 638                                  errmsg("currval of sequence \"%s\" is not yet defined in this session",
 639                                                 RelationGetRelationName(seqrel))));
 640
 641         result = elm->last;
 642
 643         relation_close(seqrel, NoLock);
 644
 645         PG_RETURN_INT64(result);
 646 }
 647
 648 Datum
 649 lastval(PG_FUNCTION_ARGS)
 650 {
 651         Relation        seqrel;
 652         int64           result;
 653
 654         if (last_used_seq == NULL)
 655                 ereport(ERROR,
 656                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 657                                  errmsg("lastval is not yet defined in this session")));
 658
 659         /* Someone may have dropped the sequence since the last nextval() */
 660         if (!SearchSysCacheExists(RELOID,
 661                                                           ObjectIdGetDatum(last_used_seq->relid),
 662                                                           0, 0, 0))
 663                 ereport(ERROR,
 664                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 665                                  errmsg("lastval is not yet defined in this session")));
 666
 667         seqrel = open_share_lock(last_used_seq);
 668
 669         /* nextval() must have already been called for this sequence */
 670         Assert(last_used_seq->increment != 0);
 671
 672         if (pg_class_aclcheck(last_used_seq->relid, GetUserId(), ACL_SELECT) != ACLCHECK_OK &&
 673                 pg_class_aclcheck(last_used_seq->relid, GetUserId(), ACL_USAGE) != ACLCHECK_OK)
 674                 ereport(ERROR,
 675                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 676                                  errmsg("permission denied for sequence %s",
 677                                                 RelationGetRelationName(seqrel))));
 678
 679         result = last_used_seq->last;
 680         relation_close(seqrel, NoLock);
 681
 682         PG_RETURN_INT64(result);
 683 }
 684
 685 /*
 686  * Main internal procedure that handles 2 & 3 arg forms of SETVAL.
 687  *
 688  * Note that the 3 arg version (which sets the is_called flag) is
 689  * only for use in pg_dump, and setting the is_called flag may not
 690  * work if multiple users are attached to the database and referencing
 691  * the sequence (unlikely if pg_dump is restoring it).
 692  *
 693  * It is necessary to have the 3 arg version so that pg_dump can
 694  * restore the state of a sequence exactly during data-only restores -
 695  * it is the only way to clear the is_called flag in an existing
 696  * sequence.
 697  */
 698 static void
 699 do_setval(Oid relid, int64 next, bool iscalled)
 700 {
 701         SeqTable        elm;
 702         Relation        seqrel;
 703         Buffer          buf;
 704         Form_pg_sequence seq;
 705
 706         /* open and AccessShareLock sequence */
 707         init_sequence(relid, &elm, &seqrel);
 708
 709         if (pg_class_aclcheck(elm->relid, GetUserId(), ACL_UPDATE) != ACLCHECK_OK)
 710                 ereport(ERROR,
 711                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 712                                  errmsg("permission denied for sequence %s",
 713                                                 RelationGetRelationName(seqrel))));
 714
 715         /* lock page' buffer and read tuple */
 716         seq = read_info(elm, seqrel, &buf);
 717
 718         if ((next < seq->min_value) || (next > seq->max_value))
 719         {
 720                 char            bufv[100],
 721                                         bufm[100],
 722                                         bufx[100];
 723
 724                 snprintf(bufv, sizeof(bufv), INT64_FORMAT, next);
 725                 snprintf(bufm, sizeof(bufm), INT64_FORMAT, seq->min_value);
 726                 snprintf(bufx, sizeof(bufx), INT64_FORMAT, seq->max_value);
 727                 ereport(ERROR,
 728                                 (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
 729                                  errmsg("setval: value %s is out of bounds for sequence \"%s\" (%s..%s)",
 730                                                 bufv, RelationGetRelationName(seqrel),
 731                                                 bufm, bufx)));
 732         }
 733
 734         /* save info in local cache */
 735         elm->last = next;                       /* last returned number */
 736         elm->cached = next;                     /* last cached number (forget cached values) */
 737
 738         START_CRIT_SECTION();
 739
 740         MarkBufferDirty(buf);
 741
 742         /* XLOG stuff */
 743         if (!seqrel->rd_istemp)
 744         {
 745                 xl_seq_rec      xlrec;
 746                 XLogRecPtr      recptr;
 747                 XLogRecData rdata[2];
 748                 Page            page = BufferGetPage(buf);
 749
 750                 xlrec.node = seqrel->rd_node;
 751                 rdata[0].data = (char *) &xlrec;
 752                 rdata[0].len = sizeof(xl_seq_rec);
 753                 rdata[0].buffer = InvalidBuffer;
 754                 rdata[0].next = &(rdata[1]);
 755
 756                 /* set values that will be saved in xlog */
 757                 seq->last_value = next;
 758                 seq->is_called = true;
 759                 seq->log_cnt = 0;
 760
 761                 rdata[1].data = (char *) page + ((PageHeader) page)->pd_upper;
 762                 rdata[1].len = ((PageHeader) page)->pd_special -
 763                         ((PageHeader) page)->pd_upper;
 764                 rdata[1].buffer = InvalidBuffer;
 765                 rdata[1].next = NULL;
 766
 767                 recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG | XLOG_NO_TRAN, rdata);
 768
 769                 PageSetLSN(page, recptr);
 770                 PageSetTLI(page, ThisTimeLineID);
 771         }
 772
 773         /* save info in sequence relation */
 774         seq->last_value = next;         /* last fetched number */
 775         seq->is_called = iscalled;
 776         seq->log_cnt = (iscalled) ? 0 : 1;
 777
 778         END_CRIT_SECTION();
 779
 780         UnlockReleaseBuffer(buf);
 781
 782         relation_close(seqrel, NoLock);
 783 }
 784
 785 /*
 786  * Implement the 2 arg setval procedure.
 787  * See do_setval for discussion.
 788  */
 789 Datum
 790 setval_oid(PG_FUNCTION_ARGS)
 791 {
 792         Oid                     relid = PG_GETARG_OID(0);
 793         int64           next = PG_GETARG_INT64(1);
 794
 795         do_setval(relid, next, true);
 796
 797         PG_RETURN_INT64(next);
 798 }
 799
 800 /*
 801  * Implement the 3 arg setval procedure.
 802  * See do_setval for discussion.
 803  */
 804 Datum
 805 setval3_oid(PG_FUNCTION_ARGS)
 806 {
 807         Oid                     relid = PG_GETARG_OID(0);
 808         int64           next = PG_GETARG_INT64(1);
 809         bool            iscalled = PG_GETARG_BOOL(2);
 810
 811         do_setval(relid, next, iscalled);
 812
 813         PG_RETURN_INT64(next);
 814 }
 815
 816
 817 /*
 818  * Open the sequence and acquire AccessShareLock if needed
 819  *
 820  * If we haven't touched the sequence already in this transaction,
 821  * we need to acquire AccessShareLock.  We arrange for the lock to
 822  * be owned by the top transaction, so that we don't need to do it
 823  * more than once per xact.
 824  */
 825 static Relation
 826 open_share_lock(SeqTable seq)
 827 {
 828         TransactionId thisxid = GetTopTransactionId();
 829
 830         /* Get the lock if not already held in this xact */
 831         if (seq->xid != thisxid)
 832         {
 833                 ResourceOwner currentOwner;
 834
 835                 currentOwner = CurrentResourceOwner;
 836                 PG_TRY();
 837                 {
 838                         CurrentResourceOwner = TopTransactionResourceOwner;
 839                         LockRelationOid(seq->relid, AccessShareLock);
 840                 }
 841                 PG_CATCH();
 842                 {
 843                         /* Ensure CurrentResourceOwner is restored on error */
 844                         CurrentResourceOwner = currentOwner;
 845                         PG_RE_THROW();
 846                 }
 847                 PG_END_TRY();
 848                 CurrentResourceOwner = currentOwner;
 849
 850                 /* Flag that we have a lock in the current xact */
 851                 seq->xid = thisxid;
 852         }
 853
 854         /* We now know we have AccessShareLock, and can safely open the rel */
 855         return relation_open(seq->relid, NoLock);
 856 }
 857
 858 /*
 859  * Given a relation OID, open and lock the sequence.  p_elm and p_rel are
 860  * output parameters.
 861  */
 862 static void
 863 init_sequence(Oid relid, SeqTable *p_elm, Relation *p_rel)
 864 {
 865         SeqTable        elm;
 866         Relation        seqrel;
 867
 868         /* Look to see if we already have a seqtable entry for relation */
 869         for (elm = seqtab; elm != NULL; elm = elm->next)
 870         {
 871                 if (elm->relid == relid)
 872                         break;
 873         }
 874
 875         /*
 876          * Allocate new seqtable entry if we didn't find one.
 877          *
 878          * NOTE: seqtable entries remain in the list for the life of a backend. If
 879          * the sequence itself is deleted then the entry becomes wasted memory,
 880          * but it's small enough that this should not matter.
 881          */
 882         if (elm == NULL)
 883         {
 884                 /*
 885                  * Time to make a new seqtable entry.  These entries live as long as
 886                  * the backend does, so we use plain malloc for them.
 887                  */
 888                 elm = (SeqTable) malloc(sizeof(SeqTableData));
 889                 if (elm == NULL)
 890                         ereport(ERROR,
 891                                         (errcode(ERRCODE_OUT_OF_MEMORY),
 892                                          errmsg("out of memory")));
 893                 elm->relid = relid;
 894                 elm->xid = InvalidTransactionId;
 895                 /* increment is set to 0 until we do read_info (see currval) */
 896                 elm->last = elm->cached = elm->increment = 0;
 897                 elm->next = seqtab;
 898                 seqtab = elm;
 899         }
 900
 901         /*
 902          * Open the sequence relation.
 903          */
 904         seqrel = open_share_lock(elm);
 905
 906         if (seqrel->rd_rel->relkind != RELKIND_SEQUENCE)
 907                 ereport(ERROR,
 908                                 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
 909                                  errmsg("\"%s\" is not a sequence",
 910                                                 RelationGetRelationName(seqrel))));
 911
 912         *p_elm = elm;
 913         *p_rel = seqrel;
 914 }
 915
 916
 917 /* Given an opened relation, lock the page buffer and find the tuple */
 918 static Form_pg_sequence
 919 read_info(SeqTable elm, Relation rel, Buffer *buf)
 920 {
 921         PageHeader      page;
 922         ItemId          lp;
 923         HeapTupleData tuple;
 924         sequence_magic *sm;
 925         Form_pg_sequence seq;
 926
 927         *buf = ReadBuffer(rel, 0);
 928         LockBuffer(*buf, BUFFER_LOCK_EXCLUSIVE);
 929
 930         page = (PageHeader) BufferGetPage(*buf);
 931         sm = (sequence_magic *) PageGetSpecialPointer(page);
 932
 933         if (sm->magic != SEQ_MAGIC)
 934                 elog(ERROR, "bad magic number in sequence \"%s\": %08X",
 935                          RelationGetRelationName(rel), sm->magic);
 936
 937         lp = PageGetItemId(page, FirstOffsetNumber);
 938         Assert(ItemIdIsUsed(lp));
 939         tuple.t_data = (HeapTupleHeader) PageGetItem((Page) page, lp);
 940
 941         seq = (Form_pg_sequence) GETSTRUCT(&tuple);
 942
 943         elm->increment = seq->increment_by;
 944
 945         return seq;
 946 }
 947
 948 /*
 949  * init_params: process the options list of CREATE or ALTER SEQUENCE,
 950  * and store the values into appropriate fields of *new.  Also set
 951  * *owned_by to any OWNED BY option, or to NIL if there is none.
 952  *
 953  * If isInit is true, fill any unspecified options with default values;
 954  * otherwise, do not change existing options that aren't explicitly overridden.
 955  */
 956 static void
 957 init_params(List *options, bool isInit,
 958                         Form_pg_sequence new, List **owned_by)
 959 {
 960         DefElem    *last_value = NULL;
 961         DefElem    *increment_by = NULL;
 962         DefElem    *max_value = NULL;
 963         DefElem    *min_value = NULL;
 964         DefElem    *cache_value = NULL;
 965         DefElem    *is_cycled = NULL;
 966         ListCell   *option;
 967
 968         *owned_by = NIL;
 969
 970         foreach(option, options)
 971         {
 972                 DefElem    *defel = (DefElem *) lfirst(option);
 973
 974                 if (strcmp(defel->defname, "increment") == 0)
 975                 {
 976                         if (increment_by)
 977                                 ereport(ERROR,
 978                                                 (errcode(ERRCODE_SYNTAX_ERROR),
 979                                                  errmsg("conflicting or redundant options")));
 980                         increment_by = defel;
 981                 }
 982
 983                 /*
 984                  * start is for a new sequence restart is for alter
 985                  */
 986                 else if (strcmp(defel->defname, "start") == 0 ||
 987                                  strcmp(defel->defname, "restart") == 0)
 988                 {
 989                         if (last_value)
 990                                 ereport(ERROR,
 991                                                 (errcode(ERRCODE_SYNTAX_ERROR),
 992                                                  errmsg("conflicting or redundant options")));
 993                         last_value = defel;
 994                 }
 995                 else if (strcmp(defel->defname, "maxvalue") == 0)
 996                 {
 997                         if (max_value)
 998                                 ereport(ERROR,
 999                                                 (errcode(ERRCODE_SYNTAX_ERROR),
1000                                                  errmsg("conflicting or redundant options")));
1001                         max_value = defel;
1002                 }
1003                 else if (strcmp(defel->defname, "minvalue") == 0)
1004                 {
1005                         if (min_value)
1006                                 ereport(ERROR,
1007                                                 (errcode(ERRCODE_SYNTAX_ERROR),
1008                                                  errmsg("conflicting or redundant options")));
1009                         min_value = defel;
1010                 }
1011                 else if (strcmp(defel->defname, "cache") == 0)
1012                 {
1013                         if (cache_value)
1014                                 ereport(ERROR,
1015                                                 (errcode(ERRCODE_SYNTAX_ERROR),
1016                                                  errmsg("conflicting or redundant options")));
1017                         cache_value = defel;
1018                 }
1019                 else if (strcmp(defel->defname, "cycle") == 0)
1020                 {
1021                         if (is_cycled)
1022                                 ereport(ERROR,
1023                                                 (errcode(ERRCODE_SYNTAX_ERROR),
1024                                                  errmsg("conflicting or redundant options")));
1025                         is_cycled = defel;
1026                 }
1027                 else if (strcmp(defel->defname, "owned_by") == 0)
1028                 {
1029                         if (*owned_by)
1030                                 ereport(ERROR,
1031                                                 (errcode(ERRCODE_SYNTAX_ERROR),
1032                                                  errmsg("conflicting or redundant options")));
1033                         *owned_by = defGetQualifiedName(defel);
1034                 }
1035                 else
1036                         elog(ERROR, "option \"%s\" not recognized",
1037                                  defel->defname);
1038         }
1039
1040         /* INCREMENT BY */
1041         if (increment_by != NULL)
1042         {
1043                 new->increment_by = defGetInt64(increment_by);
1044                 if (new->increment_by == 0)
1045                         ereport(ERROR,
1046                                         (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1047                                          errmsg("INCREMENT must not be zero")));
1048         }
1049         else if (isInit)
1050                 new->increment_by = 1;
1051
1052         /* CYCLE */
1053         if (is_cycled != NULL)
1054         {
1055                 new->is_cycled = intVal(is_cycled->arg);
1056                 Assert(new->is_cycled == false || new->is_cycled == true);
1057         }
1058         else if (isInit)
1059                 new->is_cycled = false;
1060
1061         /* MAXVALUE (null arg means NO MAXVALUE) */
1062         if (max_value != NULL && max_value->arg)
1063                 new->max_value = defGetInt64(max_value);
1064         else if (isInit || max_value != NULL)
1065         {
1066                 if (new->increment_by > 0)
1067                         new->max_value = SEQ_MAXVALUE;          /* ascending seq */
1068                 else
1069                         new->max_value = -1;    /* descending seq */
1070         }
1071
1072         /* MINVALUE (null arg means NO MINVALUE) */
1073         if (min_value != NULL && min_value->arg)
1074                 new->min_value = defGetInt64(min_value);
1075         else if (isInit || min_value != NULL)
1076         {
1077                 if (new->increment_by > 0)
1078                         new->min_value = 1; /* ascending seq */
1079                 else
1080                         new->min_value = SEQ_MINVALUE;          /* descending seq */
1081         }
1082
1083         /* crosscheck min/max */
1084         if (new->min_value >= new->max_value)
1085         {
1086                 char            bufm[100],
1087                                         bufx[100];
1088
1089                 snprintf(bufm, sizeof(bufm), INT64_FORMAT, new->min_value);
1090                 snprintf(bufx, sizeof(bufx), INT64_FORMAT, new->max_value);
1091                 ereport(ERROR,
1092                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1093                                  errmsg("MINVALUE (%s) must be less than MAXVALUE (%s)",
1094                                                 bufm, bufx)));
1095         }
1096
1097         /* START WITH */
1098         if (last_value != NULL)
1099         {
1100                 new->last_value = defGetInt64(last_value);
1101                 new->is_called = false;
1102                 new->log_cnt = 1;
1103         }
1104         else if (isInit)
1105         {
1106                 if (new->increment_by > 0)
1107                         new->last_value = new->min_value;       /* ascending seq */
1108                 else
1109                         new->last_value = new->max_value;       /* descending seq */
1110                 new->is_called = false;
1111                 new->log_cnt = 1;
1112         }
1113
1114         /* crosscheck */
1115         if (new->last_value < new->min_value)
1116         {
1117                 char            bufs[100],
1118                                         bufm[100];
1119
1120                 snprintf(bufs, sizeof(bufs), INT64_FORMAT, new->last_value);
1121                 snprintf(bufm, sizeof(bufm), INT64_FORMAT, new->min_value);
1122                 ereport(ERROR,
1123                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1124                                  errmsg("START value (%s) can't be less than MINVALUE (%s)",
1125                                                 bufs, bufm)));
1126         }
1127         if (new->last_value > new->max_value)
1128         {
1129                 char            bufs[100],
1130                                         bufm[100];
1131
1132                 snprintf(bufs, sizeof(bufs), INT64_FORMAT, new->last_value);
1133                 snprintf(bufm, sizeof(bufm), INT64_FORMAT, new->max_value);
1134                 ereport(ERROR,
1135                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1136                            errmsg("START value (%s) can't be greater than MAXVALUE (%s)",
1137                                           bufs, bufm)));
1138         }
1139
1140         /* CACHE */
1141         if (cache_value != NULL)
1142         {
1143                 new->cache_value = defGetInt64(cache_value);
1144                 if (new->cache_value <= 0)
1145                 {
1146                         char            buf[100];
1147
1148                         snprintf(buf, sizeof(buf), INT64_FORMAT, new->cache_value);
1149                         ereport(ERROR,
1150                                         (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1151                                          errmsg("CACHE (%s) must be greater than zero",
1152                                                         buf)));
1153                 }
1154         }
1155         else if (isInit)
1156                 new->cache_value = 1;
1157 }
1158
1159 /*
1160  * Process an OWNED BY option for CREATE/ALTER SEQUENCE
1161  *
1162  * Ownership permissions on the sequence are already checked,
1163  * but if we are establishing a new owned-by dependency, we must
1164  * enforce that the referenced table has the same owner and namespace
1165  * as the sequence.
1166  */
1167 static void
1168 process_owned_by(Relation seqrel, List *owned_by)
1169 {
1170         int                     nnames;
1171         Relation        tablerel;
1172         AttrNumber      attnum;
1173
1174         nnames = list_length(owned_by);
1175         Assert(nnames > 0);
1176         if (nnames == 1)
1177         {
1178                 /* Must be OWNED BY NONE */
1179                 if (strcmp(strVal(linitial(owned_by)), "none") != 0)
1180                         ereport(ERROR,
1181                                         (errcode(ERRCODE_SYNTAX_ERROR),
1182                                          errmsg("invalid OWNED BY option"),
1183                                 errhint("Specify OWNED BY table.column or OWNED BY NONE.")));
1184                 tablerel = NULL;
1185                 attnum = 0;
1186         }
1187         else
1188         {
1189                 List       *relname;
1190                 char       *attrname;
1191                 RangeVar   *rel;
1192
1193                 /* Separate relname and attr name */
1194                 relname = list_truncate(list_copy(owned_by), nnames - 1);
1195                 attrname = strVal(lfirst(list_tail(owned_by)));
1196
1197                 /* Open and lock rel to ensure it won't go away meanwhile */
1198                 rel = makeRangeVarFromNameList(relname);
1199                 tablerel = relation_openrv(rel, AccessShareLock);
1200
1201                 /* Must be a regular table */
1202                 if (tablerel->rd_rel->relkind != RELKIND_RELATION)
1203                         ereport(ERROR,
1204                                         (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1205                                          errmsg("referenced relation \"%s\" is not a table",
1206                                                         RelationGetRelationName(tablerel))));
1207
1208                 /* We insist on same owner and schema */
1209                 if (seqrel->rd_rel->relowner != tablerel->rd_rel->relowner)
1210                         ereport(ERROR,
1211                                         (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1212                         errmsg("sequence must have same owner as table it is linked to")));
1213                 if (RelationGetNamespace(seqrel) != RelationGetNamespace(tablerel))
1214                         ereport(ERROR,
1215                                         (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1216                                          errmsg("sequence must be in same schema as table it is linked to")));
1217
1218                 /* Now, fetch the attribute number from the system cache */
1219                 attnum = get_attnum(RelationGetRelid(tablerel), attrname);
1220                 if (attnum == InvalidAttrNumber)
1221                         ereport(ERROR,
1222                                         (errcode(ERRCODE_UNDEFINED_COLUMN),
1223                                          errmsg("column \"%s\" of relation \"%s\" does not exist",
1224                                                         attrname, RelationGetRelationName(tablerel))));
1225         }
1226
1227         /*
1228          * OK, we are ready to update pg_depend.  First remove any existing AUTO
1229          * dependencies for the sequence, then optionally add a new one.
1230          */
1231         markSequenceUnowned(RelationGetRelid(seqrel));
1232
1233         if (tablerel)
1234         {
1235                 ObjectAddress refobject,
1236                                         depobject;
1237
1238                 refobject.classId = RelationRelationId;
1239                 refobject.objectId = RelationGetRelid(tablerel);
1240                 refobject.objectSubId = attnum;
1241                 depobject.classId = RelationRelationId;
1242                 depobject.objectId = RelationGetRelid(seqrel);
1243                 depobject.objectSubId = 0;
1244                 recordDependencyOn(&depobject, &refobject, DEPENDENCY_AUTO);
1245         }
1246
1247         /* Done, but hold lock until commit */
1248         if (tablerel)
1249                 relation_close(tablerel, NoLock);
1250 }
1251
1252
1253 void
1254 seq_redo(XLogRecPtr lsn, XLogRecord *record)
1255 {
1256         uint8           info = record->xl_info & ~XLR_INFO_MASK;
1257         Relation        reln;
1258         Buffer          buffer;
1259         Page            page;
1260         char       *item;
1261         Size            itemsz;
1262         xl_seq_rec *xlrec = (xl_seq_rec *) XLogRecGetData(record);
1263         sequence_magic *sm;
1264
1265         if (info != XLOG_SEQ_LOG)
1266                 elog(PANIC, "seq_redo: unknown op code %u", info);
1267
1268         reln = XLogOpenRelation(xlrec->node);
1269         buffer = XLogReadBuffer(reln, 0, true);
1270         Assert(BufferIsValid(buffer));
1271         page = (Page) BufferGetPage(buffer);
1272
1273         /* Always reinit the page and reinstall the magic number */
1274         /* See comments in DefineSequence */
1275         PageInit((Page) page, BufferGetPageSize(buffer), sizeof(sequence_magic));
1276         sm = (sequence_magic *) PageGetSpecialPointer(page);
1277         sm->magic = SEQ_MAGIC;
1278
1279         item = (char *) xlrec + sizeof(xl_seq_rec);
1280         itemsz = record->xl_len - sizeof(xl_seq_rec);
1281         itemsz = MAXALIGN(itemsz);
1282         if (PageAddItem(page, (Item) item, itemsz,
1283                                         FirstOffsetNumber, LP_USED) == InvalidOffsetNumber)
1284                 elog(PANIC, "seq_redo: failed to add item to page");
1285
1286         PageSetLSN(page, lsn);
1287         PageSetTLI(page, ThisTimeLineID);
1288         MarkBufferDirty(buffer);
1289         UnlockReleaseBuffer(buffer);
1290 }
1291
1292 void
1293 seq_desc(StringInfo buf, uint8 xl_info, char *rec)
1294 {
1295         uint8           info = xl_info & ~XLR_INFO_MASK;
1296         xl_seq_rec *xlrec = (xl_seq_rec *) rec;
1297
1298         if (info == XLOG_SEQ_LOG)
1299                 appendStringInfo(buf, "log: ");
1300         else
1301         {
1302                 appendStringInfo(buf, "UNKNOWN");
1303                 return;
1304         }
1305
1306         appendStringInfo(buf, "rel %u/%u/%u",
1307                            xlrec->node.spcNode, xlrec->node.dbNode, xlrec->node.relNode);
1308 }