/*
* 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.
+ * crash we can lose (skip over) as many values as we pre-logged.
*/
#define SEQ_LOG_VALS 32
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() */
+ /* note that increment is zero until we first do read_seq_tuple() */
} SeqTableData;
typedef SeqTableData *SeqTable;
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 Form_pg_sequence read_seq_tuple(SeqTable elm, Relation rel,
+ Buffer *buf, HeapTuple seqtuple);
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);
case SEQ_COL_LOG:
coldef->typeName = makeTypeNameFromOid(INT8OID, -1);
coldef->colname = "log_cnt";
- value[i - 1] = Int64GetDatum((int64) 1);
+ value[i - 1] = Int64GetDatum((int64) 0);
break;
case SEQ_COL_CYCLE:
coldef->typeName = makeTypeNameFromOid(BOOLOID, -1);
SeqTable elm;
Form_pg_sequence seq;
Buffer buf;
- Page page;
+ HeapTupleData seqtuple;
HeapTuple tuple;
- HeapTupleData tupledata;
- ItemId lp;
/*
* Read the old sequence. This does a bit more work than really
* indeed a sequence.
*/
init_sequence(seq_relid, &elm, &seq_rel);
- read_info(elm, seq_rel, &buf);
+ (void) read_seq_tuple(elm, seq_rel, &buf, &seqtuple);
/*
* Copy the existing sequence tuple.
*/
- page = BufferGetPage(buf);
- lp = PageGetItemId(page, FirstOffsetNumber);
- Assert(ItemIdIsNormal(lp));
-
- tupledata.t_data = (HeapTupleHeader) PageGetItem(page, lp);
- tupledata.t_len = ItemIdGetLength(lp);
- tuple = heap_copytuple(&tupledata);
+ tuple = heap_copytuple(&seqtuple);
/* Now we're done with the old page */
UnlockReleaseBuffer(buf);
seq = (Form_pg_sequence) GETSTRUCT(tuple);
seq->last_value = seq->start_value;
seq->is_called = false;
- seq->log_cnt = 1;
+ seq->log_cnt = 0;
/*
* Create a new storage file for the sequence. We want to keep the
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;
SeqTable elm;
Relation seqrel;
Buffer buf;
- Page page;
+ HeapTupleData seqtuple;
Form_pg_sequence seq;
FormData_pg_sequence new;
List *owned_by;
stmt->sequence->relname);
/* lock page' buffer and read tuple into new sequence structure */
- seq = read_info(elm, seqrel, &buf);
- page = BufferGetPage(buf);
+ seq = read_seq_tuple(elm, seqrel, &buf, &seqtuple);
/* Copy old values of options into workspace */
memcpy(&new, seq, sizeof(FormData_pg_sequence));
elm->cached = elm->last;
/* Now okay to update the on-disk tuple */
- memcpy(seq, &new, sizeof(FormData_pg_sequence));
-
START_CRIT_SECTION();
+ memcpy(seq, &new, sizeof(FormData_pg_sequence));
+
MarkBufferDirty(buf);
/* XLOG stuff */
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].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].data = (char *) seqtuple.t_data;
+ rdata[1].len = seqtuple.t_len;
rdata[1].buffer = InvalidBuffer;
rdata[1].next = NULL;
Relation seqrel;
Buffer buf;
Page page;
+ HeapTupleData seqtuple;
Form_pg_sequence seq;
int64 incby,
maxv,
}
/* lock page' buffer and read tuple */
- seq = read_info(elm, seqrel, &buf);
+ seq = read_seq_tuple(elm, seqrel, &buf, &seqtuple);
page = BufferGetPage(buf);
last = next = result = seq->last_value;
if (!seq->is_called)
{
- rescnt++; /* last_value if not called */
+ rescnt++; /* return last_value if not is_called */
fetch--;
- log--;
}
/*
* checkpoint would fail to advance the sequence past the logged values.
* In this case we may as well fetch extra values.
*/
- if (log < fetch)
+ if (log < fetch || !seq->is_called)
{
/* forced log to satisfy local demand for values */
fetch = log = fetch + SEQ_LOG_VALS;
last_used_seq = elm;
+ /* ready to change the on-disk (or really, in-buffer) tuple */
START_CRIT_SECTION();
+ /*
+ * We must mark the buffer dirty before doing XLogInsert(); see notes in
+ * SyncOneBuffer(). However, we don't apply the desired changes just yet.
+ * This looks like a violation of the buffer update protocol, but it is
+ * in fact safe because we hold exclusive lock on the buffer. Any other
+ * process, including a checkpoint, that tries to examine the buffer
+ * contents will block until we release the lock, and then will see the
+ * final state that we install below.
+ */
MarkBufferDirty(buf);
/* XLOG stuff */
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]);
+ /*
+ * We don't log the current state of the tuple, but rather the state
+ * as it would appear after "log" more fetches. This lets us skip
+ * that many future WAL records, at the cost that we lose those
+ * sequence values if we crash.
+ */
/* 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;
+ 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 *) seqtuple.t_data;
+ rdata[1].len = seqtuple.t_len;
rdata[1].buffer = InvalidBuffer;
rdata[1].next = NULL;
PageSetTLI(page, ThisTimeLineID);
}
- /* update on-disk data */
+ /* Now update sequence tuple to the intended final state */
seq->last_value = last; /* last fetched number */
seq->is_called = true;
seq->log_cnt = log; /* how much is logged */
SeqTable elm;
Relation seqrel;
Buffer buf;
+ HeapTupleData seqtuple;
Form_pg_sequence seq;
/* open and AccessShareLock sequence */
PreventCommandIfReadOnly("setval()");
/* lock page' buffer and read tuple */
- seq = read_info(elm, seqrel, &buf);
+ seq = read_seq_tuple(elm, seqrel, &buf, &seqtuple);
if ((next < seq->min_value) || (next > seq->max_value))
{
/* In any case, forget any future cached numbers */
elm->cached = elm->last;
+ /* ready to change the on-disk (or really, in-buffer) tuple */
START_CRIT_SECTION();
+ seq->last_value = next; /* last fetched number */
+ seq->is_called = iscalled;
+ seq->log_cnt = 0;
+
MarkBufferDirty(buf);
/* XLOG stuff */
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].data = (char *) seqtuple.t_data;
+ rdata[1].len = seqtuple.t_len;
rdata[1].buffer = InvalidBuffer;
rdata[1].next = NULL;
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);
}
-/* Given an opened relation, lock the page buffer and find the tuple */
+/*
+ * Given an opened sequence relation, lock the page buffer and find the tuple
+ *
+ * *buf receives the reference to the pinned-and-ex-locked buffer
+ * *seqtuple receives the reference to the sequence tuple proper
+ * (this arg should point to a local variable of type HeapTupleData)
+ *
+ * Function's return value points to the data payload of the tuple
+ */
static Form_pg_sequence
-read_info(SeqTable elm, Relation rel, Buffer *buf)
+read_seq_tuple(SeqTable elm, Relation rel, Buffer *buf, HeapTuple seqtuple)
{
Page page;
ItemId lp;
- HeapTupleData tuple;
sequence_magic *sm;
Form_pg_sequence seq;
lp = PageGetItemId(page, FirstOffsetNumber);
Assert(ItemIdIsNormal(lp));
- tuple.t_data = (HeapTupleHeader) PageGetItem(page, lp);
+
+ /* Note we currently only bother to set these two fields of *seqtuple */
+ seqtuple->t_data = (HeapTupleHeader) PageGetItem(page, lp);
+ seqtuple->t_len = ItemIdGetLength(lp);
/*
* Previous releases of Postgres neglected to prevent SELECT FOR UPDATE on
* bit update, ie, don't bother to WAL-log it, since we can certainly do
* this again if the update gets lost.
*/
- if (HeapTupleHeaderGetXmax(tuple.t_data) != InvalidTransactionId)
+ if (HeapTupleHeaderGetXmax(seqtuple->t_data) != InvalidTransactionId)
{
- HeapTupleHeaderSetXmax(tuple.t_data, InvalidTransactionId);
- tuple.t_data->t_infomask &= ~HEAP_XMAX_COMMITTED;
- tuple.t_data->t_infomask |= HEAP_XMAX_INVALID;
+ HeapTupleHeaderSetXmax(seqtuple->t_data, InvalidTransactionId);
+ seqtuple->t_data->t_infomask &= ~HEAP_XMAX_COMMITTED;
+ seqtuple->t_data->t_infomask |= HEAP_XMAX_INVALID;
SetBufferCommitInfoNeedsSave(*buf);
}
- seq = (Form_pg_sequence) GETSTRUCT(&tuple);
+ seq = (Form_pg_sequence) GETSTRUCT(seqtuple);
/* this is a handy place to update our copy of the increment */
elm->increment = seq->increment_by;
defel->defname);
}
+ /*
+ * We must reset log_cnt when isInit or when changing any parameters
+ * that would affect future nextval allocations.
+ */
+ if (isInit)
+ new->log_cnt = 0;
+
/* INCREMENT BY */
if (increment_by != NULL)
{
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("INCREMENT must not be zero")));
+ new->log_cnt = 0;
}
else if (isInit)
new->increment_by = 1;
{
new->is_cycled = intVal(is_cycled->arg);
Assert(BoolIsValid(new->is_cycled));
+ new->log_cnt = 0;
}
else if (isInit)
new->is_cycled = false;
/* MAXVALUE (null arg means NO MAXVALUE) */
if (max_value != NULL && max_value->arg)
+ {
new->max_value = defGetInt64(max_value);
+ new->log_cnt = 0;
+ }
else if (isInit || max_value != NULL)
{
if (new->increment_by > 0)
new->max_value = SEQ_MAXVALUE; /* ascending seq */
else
new->max_value = -1; /* descending seq */
+ new->log_cnt = 0;
}
/* MINVALUE (null arg means NO MINVALUE) */
if (min_value != NULL && min_value->arg)
+ {
new->min_value = defGetInt64(min_value);
+ new->log_cnt = 0;
+ }
else if (isInit || min_value != NULL)
{
if (new->increment_by > 0)
new->min_value = 1; /* ascending seq */
else
new->min_value = SEQ_MINVALUE; /* descending seq */
+ new->log_cnt = 0;
}
/* crosscheck min/max */
else
new->last_value = new->start_value;
new->is_called = false;
- new->log_cnt = 1;
+ new->log_cnt = 0;
}
else if (isInit)
{
new->last_value = new->start_value;
new->is_called = false;
- new->log_cnt = 1;
}
/* crosscheck RESTART (or current value, if changing MIN/MAX) */
errmsg("CACHE (%s) must be greater than zero",
buf)));
}
+ new->log_cnt = 0;
}
else if (isInit)
new->cache_value = 1;
SeqTable elm;
Relation seqrel;
Buffer buf;
+ HeapTupleData seqtuple;
Form_pg_sequence seq;
/* open and AccessShareLock sequence */
memset(isnull, 0, sizeof(isnull));
- seq = read_info(elm, seqrel, &buf);
+ seq = read_seq_tuple(elm, seqrel, &buf, &seqtuple);
values[0] = Int64GetDatum(seq->start_value);
values[1] = Int64GetDatum(seq->min_value);
item = (char *) xlrec + sizeof(xl_seq_rec);
itemsz = record->xl_len - sizeof(xl_seq_rec);
- itemsz = MAXALIGN(itemsz);
+
if (PageAddItem(localpage, (Item) item, itemsz,
FirstOffsetNumber, false, false) == InvalidOffsetNumber)
elog(PANIC, "seq_redo: failed to add item to page");
projects / postgresql / commitdiff