-
Notifications
You must be signed in to change notification settings - Fork 4.5k
/
sequence.c
1393 lines (1187 loc) · 36.3 KB
/
sequence.c
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*-------------------------------------------------------------------------
*
* sequence.c
* PostgreSQL sequences support code.
*
* Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
*
* IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/commands/sequence.c,v 1.167 2010/02/19 06:29:19 heikki Exp $
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "access/heapam.h"
#include "access/transam.h"
#include "access/xact.h"
#include "access/xlogutils.h"
#include "catalog/dependency.h"
#include "catalog/namespace.h"
#include "catalog/pg_type.h"
#include "commands/defrem.h"
#include "commands/sequence.h"
#include "commands/tablecmds.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/smgr.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/resowner.h"
#include "utils/syscache.h"
/*
* We don't want to log each fetching of a value from a sequence,
* so we pre-log a few fetches in advance. In the event of
* crash we can lose as much as we pre-logged.
*/
#define SEQ_LOG_VALS 32
/*
* The "special area" of a sequence's buffer page looks like this.
*/
#define SEQ_MAGIC 0x1717
typedef struct sequence_magic
{
uint32 magic;
} sequence_magic;
/*
* We store a SeqTable item for every sequence we have touched in the current
* session. This is needed to hold onto nextval/currval state. (We can't
* rely on the relcache, since it's only, well, a cache, and may decide to
* discard entries.)
*
* XXX We use linear search to find pre-existing SeqTable entries. This is
* good when only a small number of sequences are touched in a session, but
* would suck with many different sequences. Perhaps use a hashtable someday.
*/
typedef struct SeqTableData
{
struct SeqTableData *next; /* link to next SeqTable object */
Oid relid; /* pg_class OID of this sequence */
LocalTransactionId lxid; /* xact in which we last did a seq op */
bool last_valid; /* do we have a valid "last" value? */
int64 last; /* value last returned by nextval */
int64 cached; /* last value already cached for nextval */
/* if last != cached, we have not used up all the cached values */
int64 increment; /* copy of sequence's increment field */
/* note that increment is zero until we first do read_info() */
} SeqTableData;
typedef SeqTableData *SeqTable;
static SeqTable seqtab = NULL; /* Head of list of SeqTable items */
/*
* last_used_seq is updated by nextval() to point to the last used
* sequence.
*/
static SeqTableData *last_used_seq = NULL;
static int64 nextval_internal(Oid relid);
static Relation open_share_lock(SeqTable seq);
static void init_sequence(Oid relid, SeqTable *p_elm, Relation *p_rel);
static Form_pg_sequence read_info(SeqTable elm, Relation rel, Buffer *buf);
static void init_params(List *options, bool isInit,
Form_pg_sequence new, List **owned_by);
static void do_setval(Oid relid, int64 next, bool iscalled);
static void process_owned_by(Relation seqrel, List *owned_by);
/*
* DefineSequence
* Creates a new sequence relation
*/
void
DefineSequence(CreateSeqStmt *seq)
{
FormData_pg_sequence new;
List *owned_by;
CreateStmt *stmt = makeNode(CreateStmt);
Oid seqoid;
Relation rel;
Buffer buf;
Page page;
sequence_magic *sm;
HeapTuple tuple;
TupleDesc tupDesc;
Datum value[SEQ_COL_LASTCOL];
bool null[SEQ_COL_LASTCOL];
int i;
NameData name;
/* Check and set all option values */
init_params(seq->options, true, &new, &owned_by);
/*
* Create relation (and fill value[] and null[] for the tuple)
*/
stmt->tableElts = NIL;
for (i = SEQ_COL_FIRSTCOL; i <= SEQ_COL_LASTCOL; i++)
{
ColumnDef *coldef = makeNode(ColumnDef);
coldef->inhcount = 0;
coldef->is_local = true;
coldef->is_not_null = true;
coldef->storage = 0;
coldef->raw_default = NULL;
coldef->cooked_default = NULL;
coldef->constraints = NIL;
null[i - 1] = false;
switch (i)
{
case SEQ_COL_NAME:
coldef->typeName = makeTypeNameFromOid(NAMEOID, -1);
coldef->colname = "sequence_name";
namestrcpy(&name, seq->sequence->relname);
value[i - 1] = NameGetDatum(&name);
break;
case SEQ_COL_LASTVAL:
coldef->typeName = makeTypeNameFromOid(INT8OID, -1);
coldef->colname = "last_value";
value[i - 1] = Int64GetDatumFast(new.last_value);
break;
case SEQ_COL_STARTVAL:
coldef->typeName = makeTypeNameFromOid(INT8OID, -1);
coldef->colname = "start_value";
value[i - 1] = Int64GetDatumFast(new.start_value);
break;
case SEQ_COL_INCBY:
coldef->typeName = makeTypeNameFromOid(INT8OID, -1);
coldef->colname = "increment_by";
value[i - 1] = Int64GetDatumFast(new.increment_by);
break;
case SEQ_COL_MAXVALUE:
coldef->typeName = makeTypeNameFromOid(INT8OID, -1);
coldef->colname = "max_value";
value[i - 1] = Int64GetDatumFast(new.max_value);
break;
case SEQ_COL_MINVALUE:
coldef->typeName = makeTypeNameFromOid(INT8OID, -1);
coldef->colname = "min_value";
value[i - 1] = Int64GetDatumFast(new.min_value);
break;
case SEQ_COL_CACHE:
coldef->typeName = makeTypeNameFromOid(INT8OID, -1);
coldef->colname = "cache_value";
value[i - 1] = Int64GetDatumFast(new.cache_value);
break;
case SEQ_COL_LOG:
coldef->typeName = makeTypeNameFromOid(INT8OID, -1);
coldef->colname = "log_cnt";
value[i - 1] = Int64GetDatum((int64) 1);
break;
case SEQ_COL_CYCLE:
coldef->typeName = makeTypeNameFromOid(BOOLOID, -1);
coldef->colname = "is_cycled";
value[i - 1] = BoolGetDatum(new.is_cycled);
break;
case SEQ_COL_CALLED:
coldef->typeName = makeTypeNameFromOid(BOOLOID, -1);
coldef->colname = "is_called";
value[i - 1] = BoolGetDatum(false);
break;
}
stmt->tableElts = lappend(stmt->tableElts, coldef);
}
stmt->relation = seq->sequence;
stmt->inhRelations = NIL;
stmt->constraints = NIL;
stmt->options = list_make1(defWithOids(false));
stmt->oncommit = ONCOMMIT_NOOP;
stmt->tablespacename = NULL;
seqoid = DefineRelation(stmt, RELKIND_SEQUENCE);
rel = heap_open(seqoid, AccessExclusiveLock);
tupDesc = RelationGetDescr(rel);
/* Initialize first page of relation with special magic number */
buf = ReadBuffer(rel, P_NEW);
Assert(BufferGetBlockNumber(buf) == 0);
page = BufferGetPage(buf);
PageInit(page, BufferGetPageSize(buf), sizeof(sequence_magic));
sm = (sequence_magic *) PageGetSpecialPointer(page);
sm->magic = SEQ_MAGIC;
/* hack: ensure heap_insert will insert on the just-created page */
RelationSetTargetBlock(rel, 0);
/* Now form & insert sequence tuple */
tuple = heap_form_tuple(tupDesc, value, null);
simple_heap_insert(rel, tuple);
Assert(ItemPointerGetOffsetNumber(&(tuple->t_self)) == FirstOffsetNumber);
/*
* Two special hacks here:
*
* 1. Since VACUUM does not process sequences, we have to force the tuple
* to have xmin = FrozenTransactionId now. Otherwise it would become
* invisible to SELECTs after 2G transactions. It is okay to do this
* because if the current transaction aborts, no other xact will ever
* examine the sequence tuple anyway.
*
* 2. Even though heap_insert emitted a WAL log record, we have to emit an
* XLOG_SEQ_LOG record too, since (a) the heap_insert record will not have
* the right xmin, and (b) REDO of the heap_insert record would re-init
* page and sequence magic number would be lost. This means two log
* records instead of one :-(
*/
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
START_CRIT_SECTION();
{
/*
* Note that the "tuple" structure is still just a local tuple record
* created by heap_form_tuple; its t_data pointer doesn't point at the
* disk buffer. To scribble on the disk buffer we need to fetch the
* item pointer. But do the same to the local tuple, since that will
* be the source for the WAL log record, below.
*/
ItemId itemId;
Item item;
itemId = PageGetItemId((Page) page, FirstOffsetNumber);
item = PageGetItem((Page) page, itemId);
HeapTupleHeaderSetXmin((HeapTupleHeader) item, FrozenTransactionId);
((HeapTupleHeader) item)->t_infomask |= HEAP_XMIN_COMMITTED;
HeapTupleHeaderSetXmin(tuple->t_data, FrozenTransactionId);
tuple->t_data->t_infomask |= HEAP_XMIN_COMMITTED;
}
MarkBufferDirty(buf);
/* XLOG stuff */
if (!rel->rd_istemp)
{
xl_seq_rec xlrec;
XLogRecPtr recptr;
XLogRecData rdata[2];
Form_pg_sequence newseq = (Form_pg_sequence) GETSTRUCT(tuple);
/* We do not log first nextval call, so "advance" sequence here */
/* Note we are scribbling on local tuple, not the disk buffer */
newseq->is_called = true;
newseq->log_cnt = 0;
xlrec.node = rel->rd_node;
rdata[0].data = (char *) &xlrec;
rdata[0].len = sizeof(xl_seq_rec);
rdata[0].buffer = InvalidBuffer;
rdata[0].next = &(rdata[1]);
rdata[1].data = (char *) tuple->t_data;
rdata[1].len = tuple->t_len;
rdata[1].buffer = InvalidBuffer;
rdata[1].next = NULL;
recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG, rdata);
PageSetLSN(page, recptr);
PageSetTLI(page, ThisTimeLineID);
}
END_CRIT_SECTION();
UnlockReleaseBuffer(buf);
/* process OWNED BY if given */
if (owned_by)
process_owned_by(rel, owned_by);
heap_close(rel, NoLock);
}
/*
* AlterSequence
*
* Modify the definition of a sequence relation
*/
void
AlterSequence(AlterSeqStmt *stmt)
{
Oid relid;
/* find sequence */
relid = RangeVarGetRelid(stmt->sequence, false);
/* allow ALTER to sequence owner only */
/* if you change this, see also callers of AlterSequenceInternal! */
if (!pg_class_ownercheck(relid, GetUserId()))
aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
stmt->sequence->relname);
/* do the work */
AlterSequenceInternal(relid, stmt->options);
}
/*
* AlterSequenceInternal
*
* Same as AlterSequence except that the sequence is specified by OID
* and we assume the caller already checked permissions.
*/
void
AlterSequenceInternal(Oid relid, List *options)
{
SeqTable elm;
Relation seqrel;
Buffer buf;
Page page;
Form_pg_sequence seq;
FormData_pg_sequence new;
List *owned_by;
/* open and AccessShareLock sequence */
init_sequence(relid, &elm, &seqrel);
/* lock page' buffer and read tuple into new sequence structure */
seq = read_info(elm, seqrel, &buf);
page = BufferGetPage(buf);
/* Copy old values of options into workspace */
memcpy(&new, seq, sizeof(FormData_pg_sequence));
/* Check and set new values */
init_params(options, false, &new, &owned_by);
/* Clear local cache so that we don't think we have cached numbers */
/* Note that we do not change the currval() state */
elm->cached = elm->last;
/* Now okay to update the on-disk tuple */
memcpy(seq, &new, sizeof(FormData_pg_sequence));
START_CRIT_SECTION();
MarkBufferDirty(buf);
/* XLOG stuff */
if (!seqrel->rd_istemp)
{
xl_seq_rec xlrec;
XLogRecPtr recptr;
XLogRecData rdata[2];
xlrec.node = seqrel->rd_node;
rdata[0].data = (char *) &xlrec;
rdata[0].len = sizeof(xl_seq_rec);
rdata[0].buffer = InvalidBuffer;
rdata[0].next = &(rdata[1]);
rdata[1].data = (char *) page + ((PageHeader) page)->pd_upper;
rdata[1].len = ((PageHeader) page)->pd_special -
((PageHeader) page)->pd_upper;
rdata[1].buffer = InvalidBuffer;
rdata[1].next = NULL;
recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG, rdata);
PageSetLSN(page, recptr);
PageSetTLI(page, ThisTimeLineID);
}
END_CRIT_SECTION();
UnlockReleaseBuffer(buf);
/* process OWNED BY if given */
if (owned_by)
process_owned_by(seqrel, owned_by);
relation_close(seqrel, NoLock);
}
/*
* Note: nextval with a text argument is no longer exported as a pg_proc
* entry, but we keep it around to ease porting of C code that may have
* called the function directly.
*/
Datum
nextval(PG_FUNCTION_ARGS)
{
text *seqin = PG_GETARG_TEXT_P(0);
RangeVar *sequence;
Oid relid;
sequence = makeRangeVarFromNameList(textToQualifiedNameList(seqin));
relid = RangeVarGetRelid(sequence, false);
PG_RETURN_INT64(nextval_internal(relid));
}
Datum
nextval_oid(PG_FUNCTION_ARGS)
{
Oid relid = PG_GETARG_OID(0);
PG_RETURN_INT64(nextval_internal(relid));
}
static int64
nextval_internal(Oid relid)
{
SeqTable elm;
Relation seqrel;
Buffer buf;
Page page;
Form_pg_sequence seq;
int64 incby,
maxv,
minv,
cache,
log,
fetch,
last;
int64 result,
next,
rescnt = 0;
bool logit = false;
/* nextval() writes to database and must be prevented during recovery */
PreventCommandDuringRecovery();
/* open and AccessShareLock sequence */
init_sequence(relid, &elm, &seqrel);
if (pg_class_aclcheck(elm->relid, GetUserId(), ACL_USAGE) != ACLCHECK_OK &&
pg_class_aclcheck(elm->relid, GetUserId(), ACL_UPDATE) != ACLCHECK_OK)
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("permission denied for sequence %s",
RelationGetRelationName(seqrel))));
if (elm->last != elm->cached) /* some numbers were cached */
{
Assert(elm->last_valid);
Assert(elm->increment != 0);
elm->last += elm->increment;
relation_close(seqrel, NoLock);
last_used_seq = elm;
return elm->last;
}
/* lock page' buffer and read tuple */
seq = read_info(elm, seqrel, &buf);
page = BufferGetPage(buf);
last = next = result = seq->last_value;
incby = seq->increment_by;
maxv = seq->max_value;
minv = seq->min_value;
fetch = cache = seq->cache_value;
log = seq->log_cnt;
if (!seq->is_called)
{
rescnt++; /* last_value if not called */
fetch--;
log--;
}
/*
* Decide whether we should emit a WAL log record. If so, force up the
* fetch count to grab SEQ_LOG_VALS more values than we actually need to
* cache. (These will then be usable without logging.)
*
* If this is the first nextval after a checkpoint, we must force a new
* WAL record to be written anyway, else replay starting from the
* checkpoint would fail to advance the sequence past the logged values.
* In this case we may as well fetch extra values.
*/
if (log < fetch)
{
/* forced log to satisfy local demand for values */
fetch = log = fetch + SEQ_LOG_VALS;
logit = true;
}
else
{
XLogRecPtr redoptr = GetRedoRecPtr();
if (XLByteLE(PageGetLSN(page), redoptr))
{
/* last update of seq was before checkpoint */
fetch = log = fetch + SEQ_LOG_VALS;
logit = true;
}
}
while (fetch) /* try to fetch cache [+ log ] numbers */
{
/*
* Check MAXVALUE for ascending sequences and MINVALUE for descending
* sequences
*/
if (incby > 0)
{
/* ascending sequence */
if ((maxv >= 0 && next > maxv - incby) ||
(maxv < 0 && next + incby > maxv))
{
if (rescnt > 0)
break; /* stop fetching */
if (!seq->is_cycled)
{
char buf[100];
snprintf(buf, sizeof(buf), INT64_FORMAT, maxv);
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("nextval: reached maximum value of sequence \"%s\" (%s)",
RelationGetRelationName(seqrel), buf)));
}
next = minv;
}
else
next += incby;
}
else
{
/* descending sequence */
if ((minv < 0 && next < minv - incby) ||
(minv >= 0 && next + incby < minv))
{
if (rescnt > 0)
break; /* stop fetching */
if (!seq->is_cycled)
{
char buf[100];
snprintf(buf, sizeof(buf), INT64_FORMAT, minv);
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("nextval: reached minimum value of sequence \"%s\" (%s)",
RelationGetRelationName(seqrel), buf)));
}
next = maxv;
}
else
next += incby;
}
fetch--;
if (rescnt < cache)
{
log--;
rescnt++;
last = next;
if (rescnt == 1) /* if it's first result - */
result = next; /* it's what to return */
}
}
log -= fetch; /* adjust for any unfetched numbers */
Assert(log >= 0);
/* save info in local cache */
elm->last = result; /* last returned number */
elm->cached = last; /* last fetched number */
elm->last_valid = true;
last_used_seq = elm;
START_CRIT_SECTION();
MarkBufferDirty(buf);
/* XLOG stuff */
if (logit && !seqrel->rd_istemp)
{
xl_seq_rec xlrec;
XLogRecPtr recptr;
XLogRecData rdata[2];
xlrec.node = seqrel->rd_node;
rdata[0].data = (char *) &xlrec;
rdata[0].len = sizeof(xl_seq_rec);
rdata[0].buffer = InvalidBuffer;
rdata[0].next = &(rdata[1]);
/* set values that will be saved in xlog */
seq->last_value = next;
seq->is_called = true;
seq->log_cnt = 0;
rdata[1].data = (char *) page + ((PageHeader) page)->pd_upper;
rdata[1].len = ((PageHeader) page)->pd_special -
((PageHeader) page)->pd_upper;
rdata[1].buffer = InvalidBuffer;
rdata[1].next = NULL;
recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG, rdata);
PageSetLSN(page, recptr);
PageSetTLI(page, ThisTimeLineID);
}
/* update on-disk data */
seq->last_value = last; /* last fetched number */
seq->is_called = true;
seq->log_cnt = log; /* how much is logged */
END_CRIT_SECTION();
UnlockReleaseBuffer(buf);
relation_close(seqrel, NoLock);
return result;
}
Datum
currval_oid(PG_FUNCTION_ARGS)
{
Oid relid = PG_GETARG_OID(0);
int64 result;
SeqTable elm;
Relation seqrel;
/* open and AccessShareLock sequence */
init_sequence(relid, &elm, &seqrel);
if (pg_class_aclcheck(elm->relid, GetUserId(), ACL_SELECT) != ACLCHECK_OK &&
pg_class_aclcheck(elm->relid, GetUserId(), ACL_USAGE) != ACLCHECK_OK)
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("permission denied for sequence %s",
RelationGetRelationName(seqrel))));
if (!elm->last_valid)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("currval of sequence \"%s\" is not yet defined in this session",
RelationGetRelationName(seqrel))));
result = elm->last;
relation_close(seqrel, NoLock);
PG_RETURN_INT64(result);
}
Datum
lastval(PG_FUNCTION_ARGS)
{
Relation seqrel;
int64 result;
if (last_used_seq == NULL)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("lastval is not yet defined in this session")));
/* Someone may have dropped the sequence since the last nextval() */
if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(last_used_seq->relid)))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("lastval is not yet defined in this session")));
seqrel = open_share_lock(last_used_seq);
/* nextval() must have already been called for this sequence */
Assert(last_used_seq->last_valid);
if (pg_class_aclcheck(last_used_seq->relid, GetUserId(), ACL_SELECT) != ACLCHECK_OK &&
pg_class_aclcheck(last_used_seq->relid, GetUserId(), ACL_USAGE) != ACLCHECK_OK)
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("permission denied for sequence %s",
RelationGetRelationName(seqrel))));
result = last_used_seq->last;
relation_close(seqrel, NoLock);
PG_RETURN_INT64(result);
}
/*
* Main internal procedure that handles 2 & 3 arg forms of SETVAL.
*
* Note that the 3 arg version (which sets the is_called flag) is
* only for use in pg_dump, and setting the is_called flag may not
* work if multiple users are attached to the database and referencing
* the sequence (unlikely if pg_dump is restoring it).
*
* It is necessary to have the 3 arg version so that pg_dump can
* restore the state of a sequence exactly during data-only restores -
* it is the only way to clear the is_called flag in an existing
* sequence.
*/
static void
do_setval(Oid relid, int64 next, bool iscalled)
{
SeqTable elm;
Relation seqrel;
Buffer buf;
Form_pg_sequence seq;
/* setval() writes to database and must be prevented during recovery */
PreventCommandDuringRecovery();
/* open and AccessShareLock sequence */
init_sequence(relid, &elm, &seqrel);
if (pg_class_aclcheck(elm->relid, GetUserId(), ACL_UPDATE) != ACLCHECK_OK)
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("permission denied for sequence %s",
RelationGetRelationName(seqrel))));
/* lock page' buffer and read tuple */
seq = read_info(elm, seqrel, &buf);
if ((next < seq->min_value) || (next > seq->max_value))
{
char bufv[100],
bufm[100],
bufx[100];
snprintf(bufv, sizeof(bufv), INT64_FORMAT, next);
snprintf(bufm, sizeof(bufm), INT64_FORMAT, seq->min_value);
snprintf(bufx, sizeof(bufx), INT64_FORMAT, seq->max_value);
ereport(ERROR,
(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
errmsg("setval: value %s is out of bounds for sequence \"%s\" (%s..%s)",
bufv, RelationGetRelationName(seqrel),
bufm, bufx)));
}
/* Set the currval() state only if iscalled = true */
if (iscalled)
{
elm->last = next; /* last returned number */
elm->last_valid = true;
}
/* In any case, forget any future cached numbers */
elm->cached = elm->last;
START_CRIT_SECTION();
MarkBufferDirty(buf);
/* XLOG stuff */
if (!seqrel->rd_istemp)
{
xl_seq_rec xlrec;
XLogRecPtr recptr;
XLogRecData rdata[2];
Page page = BufferGetPage(buf);
xlrec.node = seqrel->rd_node;
rdata[0].data = (char *) &xlrec;
rdata[0].len = sizeof(xl_seq_rec);
rdata[0].buffer = InvalidBuffer;
rdata[0].next = &(rdata[1]);
/* set values that will be saved in xlog */
seq->last_value = next;
seq->is_called = true;
seq->log_cnt = 0;
rdata[1].data = (char *) page + ((PageHeader) page)->pd_upper;
rdata[1].len = ((PageHeader) page)->pd_special -
((PageHeader) page)->pd_upper;
rdata[1].buffer = InvalidBuffer;
rdata[1].next = NULL;
recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG, rdata);
PageSetLSN(page, recptr);
PageSetTLI(page, ThisTimeLineID);
}
/* save info in sequence relation */
seq->last_value = next; /* last fetched number */
seq->is_called = iscalled;
seq->log_cnt = (iscalled) ? 0 : 1;
END_CRIT_SECTION();
UnlockReleaseBuffer(buf);
relation_close(seqrel, NoLock);
}
/*
* Implement the 2 arg setval procedure.
* See do_setval for discussion.
*/
Datum
setval_oid(PG_FUNCTION_ARGS)
{
Oid relid = PG_GETARG_OID(0);
int64 next = PG_GETARG_INT64(1);
do_setval(relid, next, true);
PG_RETURN_INT64(next);
}
/*
* Implement the 3 arg setval procedure.
* See do_setval for discussion.
*/
Datum
setval3_oid(PG_FUNCTION_ARGS)
{
Oid relid = PG_GETARG_OID(0);
int64 next = PG_GETARG_INT64(1);
bool iscalled = PG_GETARG_BOOL(2);
do_setval(relid, next, iscalled);
PG_RETURN_INT64(next);
}
/*
* Open the sequence and acquire AccessShareLock if needed
*
* If we haven't touched the sequence already in this transaction,
* we need to acquire AccessShareLock. We arrange for the lock to
* be owned by the top transaction, so that we don't need to do it
* more than once per xact.
*/
static Relation
open_share_lock(SeqTable seq)
{
LocalTransactionId thislxid = MyProc->lxid;
/* Get the lock if not already held in this xact */
if (seq->lxid != thislxid)
{
ResourceOwner currentOwner;
currentOwner = CurrentResourceOwner;
PG_TRY();
{
CurrentResourceOwner = TopTransactionResourceOwner;
LockRelationOid(seq->relid, AccessShareLock);
}
PG_CATCH();
{
/* Ensure CurrentResourceOwner is restored on error */
CurrentResourceOwner = currentOwner;
PG_RE_THROW();
}
PG_END_TRY();
CurrentResourceOwner = currentOwner;
/* Flag that we have a lock in the current xact */
seq->lxid = thislxid;
}
/* We now know we have AccessShareLock, and can safely open the rel */
return relation_open(seq->relid, NoLock);
}
/*
* Given a relation OID, open and lock the sequence. p_elm and p_rel are
* output parameters.
*/
static void
init_sequence(Oid relid, SeqTable *p_elm, Relation *p_rel)
{
SeqTable elm;
Relation seqrel;
/* Look to see if we already have a seqtable entry for relation */
for (elm = seqtab; elm != NULL; elm = elm->next)
{
if (elm->relid == relid)
break;
}
/*
* Allocate new seqtable entry if we didn't find one.
*
* NOTE: seqtable entries remain in the list for the life of a backend. If
* the sequence itself is deleted then the entry becomes wasted memory,
* but it's small enough that this should not matter.
*/
if (elm == NULL)
{
/*
* Time to make a new seqtable entry. These entries live as long as
* the backend does, so we use plain malloc for them.
*/
elm = (SeqTable) malloc(sizeof(SeqTableData));
if (elm == NULL)
ereport(ERROR,
(errcode(ERRCODE_OUT_OF_MEMORY),
errmsg("out of memory")));
elm->relid = relid;
elm->lxid = InvalidLocalTransactionId;
elm->last_valid = false;
elm->last = elm->cached = elm->increment = 0;
elm->next = seqtab;
seqtab = elm;
}
/*
* Open the sequence relation.
*/
seqrel = open_share_lock(elm);
if (seqrel->rd_rel->relkind != RELKIND_SEQUENCE)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("\"%s\" is not a sequence",
RelationGetRelationName(seqrel))));
*p_elm = elm;
*p_rel = seqrel;
}
/* Given an opened relation, lock the page buffer and find the tuple */
static Form_pg_sequence
read_info(SeqTable elm, Relation rel, Buffer *buf)
{
Page page;
ItemId lp;
HeapTupleData tuple;
sequence_magic *sm;
Form_pg_sequence seq;
*buf = ReadBuffer(rel, 0);
LockBuffer(*buf, BUFFER_LOCK_EXCLUSIVE);
page = BufferGetPage(*buf);
sm = (sequence_magic *) PageGetSpecialPointer(page);
if (sm->magic != SEQ_MAGIC)
elog(ERROR, "bad magic number in sequence \"%s\": %08X",
RelationGetRelationName(rel), sm->magic);
lp = PageGetItemId(page, FirstOffsetNumber);
Assert(ItemIdIsNormal(lp));
tuple.t_data = (HeapTupleHeader) PageGetItem(page, lp);
seq = (Form_pg_sequence) GETSTRUCT(&tuple);
/* this is a handy place to update our copy of the increment */
elm->increment = seq->increment_by;
return seq;
}
/*
* init_params: process the options list of CREATE or ALTER SEQUENCE,
* and store the values into appropriate fields of *new. Also set
* *owned_by to any OWNED BY option, or to NIL if there is none.
*
* If isInit is true, fill any unspecified options with default values;
* otherwise, do not change existing options that aren't explicitly overridden.
*/
static void
init_params(List *options, bool isInit,