From c08aac9acb13025d70c14f4d8b185ce990db28dc Mon Sep 17 00:00:00 2001 From: Petr Jelinek Date: Fri, 4 Mar 2016 15:03:44 +0100 Subject: [PATCH 2/2] gapless seq --- contrib/Makefile | 1 + contrib/gapless_seq/Makefile | 63 ++++ contrib/gapless_seq/expected/concurrency.out | 31 ++ contrib/gapless_seq/expected/gapless_seq.out | 145 ++++++++ contrib/gapless_seq/gapless_seq--1.0.sql | 57 +++ contrib/gapless_seq/gapless_seq.c | 531 +++++++++++++++++++++++++++ contrib/gapless_seq/gapless_seq.control | 6 + contrib/gapless_seq/specs/concurrency.spec | 29 ++ contrib/gapless_seq/sql/gapless_seq.sql | 61 +++ doc/src/sgml/contrib.sgml | 1 + doc/src/sgml/filelist.sgml | 1 + doc/src/sgml/gapless-seq.sgml | 24 ++ 12 files changed, 950 insertions(+) create mode 100644 contrib/gapless_seq/Makefile create mode 100644 contrib/gapless_seq/expected/concurrency.out create mode 100644 contrib/gapless_seq/expected/gapless_seq.out create mode 100644 contrib/gapless_seq/gapless_seq--1.0.sql create mode 100644 contrib/gapless_seq/gapless_seq.c create mode 100644 contrib/gapless_seq/gapless_seq.control create mode 100644 contrib/gapless_seq/specs/concurrency.spec create mode 100644 contrib/gapless_seq/sql/gapless_seq.sql create mode 100644 doc/src/sgml/gapless-seq.sgml diff --git a/contrib/Makefile b/contrib/Makefile index d12dd63..d2a0620 100644 --- a/contrib/Makefile +++ b/contrib/Makefile @@ -19,6 +19,7 @@ SUBDIRS = \ earthdistance \ file_fdw \ fuzzystrmatch \ + gapless_seq \ hstore \ intagg \ intarray \ diff --git a/contrib/gapless_seq/Makefile b/contrib/gapless_seq/Makefile new file mode 100644 index 0000000..9378b93 --- /dev/null +++ b/contrib/gapless_seq/Makefile @@ -0,0 +1,63 @@ +# contrib/gapless_seq/Makefile + +MODULE_big = gapless_seq +OBJS = gapless_seq.o +PG_CPPFLAGS = -I$(libpq_srcdir) + +EXTENSION = gapless_seq +DATA = gapless_seq--1.0.sql + +EXTRA_CLEAN = $(pg_regress_clean_files) ./regression_output ./isolation_output + +ifdef USE_PGXS +PG_CONFIG = pg_config +PGXS := $(shell $(PG_CONFIG) --pgxs) +include $(PGXS) +else +subdir = contrib/gapless_seq +top_builddir = ../.. +include $(top_builddir)/src/Makefile.global +include $(top_srcdir)/contrib/contrib-global.mk +endif + +check: regresscheck isolationcheck + +submake-regress: + $(MAKE) -C $(top_builddir)/src/test/regress all + +submake-isolation: + $(MAKE) -C $(top_builddir)/src/test/isolation all + +submake-gapless_seq: + $(MAKE) -C $(top_builddir)/contrib/gapless_seq + +REGRESSCHECKS=gapless_seq + +regresscheck: all | submake-regress submake-gapless_seq temp-install + $(MKDIR_P) regression_output + $(pg_regress_check) \ + --temp-instance=./tmp_check \ + --outputdir=./regression_output \ + $(REGRESSCHECKS) + +regresscheck-install-force: | submake-regress submake-gapless_seq temp-install + $(pg_regress_installcheck) \ + $(REGRESSCHECKS) + +ISOLATIONCHECKS=concurrency + +isolationcheck: all | submake-isolation submake-gapless_seq temp-install + $(MKDIR_P) isolation_output + $(pg_isolation_regress_check) \ + --outputdir=./isolation_output \ + $(ISOLATIONCHECKS) + +isolationcheck-install-force: all | submake-isolation submake-gapless_seq temp-install + $(pg_isolation_regress_installcheck) \ + $(ISOLATIONCHECKS) + +PHONY: submake-gapless_seq submake-regress check \ + regresscheck regresscheck-install-force \ + isolationcheck isolationcheck-install-force + +temp-install: EXTRA_INSTALL=contrib/gapless_seq diff --git a/contrib/gapless_seq/expected/concurrency.out b/contrib/gapless_seq/expected/concurrency.out new file mode 100644 index 0000000..ec6a098 --- /dev/null +++ b/contrib/gapless_seq/expected/concurrency.out @@ -0,0 +1,31 @@ +Parsed test spec with 3 sessions + +starting permutation: s1_begin s1_nextval s2_begin s2_nextval s1_commit s2_commit +step s1_begin: BEGIN; +step s1_nextval: SELECT nextval('test_gapless'::regclass); +nextval + +1 +step s2_begin: BEGIN; +step s2_nextval: SELECT nextval('test_gapless'::regclass); +step s1_commit: COMMIT; +step s2_nextval: <... completed> +nextval + +2 +step s2_commit: COMMIT; + +starting permutation: s3_begin s3_nextval s2_begin s2_nextval s3_rollback s2_commit +step s3_begin: BEGIN; +step s3_nextval: SELECT nextval('test_gapless'::regclass); +nextval + +1 +step s2_begin: BEGIN; +step s2_nextval: SELECT nextval('test_gapless'::regclass); +step s3_rollback: ROLLBACK; +step s2_nextval: <... completed> +nextval + +1 +step s2_commit: COMMIT; diff --git a/contrib/gapless_seq/expected/gapless_seq.out b/contrib/gapless_seq/expected/gapless_seq.out new file mode 100644 index 0000000..130104a --- /dev/null +++ b/contrib/gapless_seq/expected/gapless_seq.out @@ -0,0 +1,145 @@ +CREATE EXTENSION gapless_seq; +CREATE SEQUENCE test_gapless USING gapless; +SELECT nextval('test_gapless'::regclass); + nextval +--------- + 1 +(1 row) + +BEGIN; + SELECT nextval('test_gapless'::regclass); + nextval +--------- + 2 +(1 row) + + SELECT nextval('test_gapless'::regclass); + nextval +--------- + 3 +(1 row) + + SELECT nextval('test_gapless'::regclass); + nextval +--------- + 4 +(1 row) + +ROLLBACK; +SELECT nextval('test_gapless'::regclass); + nextval +--------- + 2 +(1 row) + +CREATE SEQUENCE test_alter_seq USING local; +SELECT nextval('test_alter_seq'::regclass); + nextval +--------- + 1 +(1 row) + +ALTER SEQUENCE test_alter_seq USING gapless; +SELECT nextval('test_alter_seq'::regclass); + nextval +--------- + 2 +(1 row) + +BEGIN; + SELECT nextval('test_alter_seq'::regclass); + nextval +--------- + 3 +(1 row) + +ROLLBACK; +SELECT nextval('test_alter_seq'::regclass); + nextval +--------- + 3 +(1 row) + +BEGIN; + SELECT nextval('test_alter_seq'::regclass); + nextval +--------- + 4 +(1 row) + + SAVEPOINT mysp; + SELECT nextval('test_alter_seq'::regclass); + nextval +--------- + 5 +(1 row) + + ROLLBACK TO SAVEPOINT mysp; + SAVEPOINT mysp2; + SELECT nextval('test_alter_seq'::regclass); + nextval +--------- + 5 +(1 row) + + RELEASE SAVEPOINT mysp2; + SELECT nextval('test_alter_seq'::regclass); + nextval +--------- + 6 +(1 row) + +COMMIT; +ALTER SEQUENCE test_alter_seq RESTART 100 USING local; +SELECT nextval('test_alter_seq'::regclass); + nextval +--------- + 100 +(1 row) + +-- check dump/restore +SELECT pg_sequence_get_state('test_gapless'); + pg_sequence_get_state +----------------------- + (2,t) +(1 row) + +SELECT pg_sequence_set_state('test_gapless', pg_sequence_get_state('test_gapless')); + pg_sequence_set_state +----------------------- + +(1 row) + +SELECT pg_sequence_get_state('test_gapless'); + pg_sequence_get_state +----------------------- + (2,t) +(1 row) + +-- check that event trigger works correctly +SELECT last_value FROM gapless_seq.seqam_gapless_values ORDER BY seqid; + last_value +------------ + 2 + 6 +(2 rows) + +DROP SEQUENCE test_gapless; +SELECT last_value FROM gapless_seq.seqam_gapless_values ORDER BY seqid; + last_value +------------ +(0 rows) + +CREATE SEQUENCE test_gapless USING gapless; +-- should fail due to deps +DROP ACCESS METHOD gapless; +ERROR: cannot drop access method gapless because extension gapless_seq requires it +HINT: You can drop extension gapless_seq instead. +-- likewise +DROP EXTENSION gapless_seq; +ERROR: cannot drop extension gapless_seq because other objects depend on it +DETAIL: sequence test_gapless depends on access method gapless +HINT: Use DROP ... CASCADE to drop the dependent objects too. +-- success +DROP EXTENSION gapless_seq CASCADE; +NOTICE: drop cascades to sequence test_gapless diff --git a/contrib/gapless_seq/gapless_seq--1.0.sql b/contrib/gapless_seq/gapless_seq--1.0.sql new file mode 100644 index 0000000..3b41f86 --- /dev/null +++ b/contrib/gapless_seq/gapless_seq--1.0.sql @@ -0,0 +1,57 @@ +-- complain if script is sourced in psql, rather than via CREATE EXTENSION +\echo Use "CREATE EXTENSION gapless_seq" to load this file. \quit + +CREATE OR REPLACE FUNCTION seqam_gapless_handler(internal) +RETURNS SEQ_AM_HANDLER +LANGUAGE C +STABLE STRICT +AS 'MODULE_PATHNAME' +; + +CREATE OR REPLACE FUNCTION seqam_gapless_state_in(cstring) +RETURNS seqam_gapless_state +LANGUAGE C +STABLE STRICT +AS 'MODULE_PATHNAME' +; + +CREATE OR REPLACE FUNCTION seqam_gapless_state_out(seqam_gapless_state) +RETURNS cstring +LANGUAGE C +STABLE STRICT +AS 'MODULE_PATHNAME' +; + +CREATE TYPE seqam_gapless_state ( + INPUT = seqam_gapless_state_in, + OUTPUT = seqam_gapless_state_out, + INTERNALLENGTH = 16 +); +COMMENT ON TYPE seqam_gapless_state IS 'state for gapless sequence am'; + +CREATE TABLE seqam_gapless_values ( + seqid oid PRIMARY KEY, + last_value bigint NOT NULL, + is_called bool NOT NULL +); + +CREATE OR REPLACE FUNCTION gapless_seq_clean_sequence_value() +RETURNS event_trigger +LANGUAGE plpgsql +AS $$ +BEGIN + -- just delete all the value data that don't have corresponding + -- gapless sequence (either DELETEd or ALTERed to different AM) + DELETE FROM gapless_seq.seqam_gapless_values WHERE seqid NOT IN ( + SELECT oid FROM pg_class WHERE relkind = 'S' AND relam = ( + SELECT oid FROM pg_am WHERE amname = 'gapless' AND amtype = 'S' + ) + ); +END; +$$; + +CREATE EVENT TRIGGER gapless_seq_clean_sequence_value ON sql_drop + WHEN TAG IN ('DROP SEQUENCE') + EXECUTE PROCEDURE gapless_seq_clean_sequence_value(); + +CREATE ACCESS METHOD gapless TYPE SEQUENCE HANDLER seqam_gapless_handler; diff --git a/contrib/gapless_seq/gapless_seq.c b/contrib/gapless_seq/gapless_seq.c new file mode 100644 index 0000000..57f1570 --- /dev/null +++ b/contrib/gapless_seq/gapless_seq.c @@ -0,0 +1,531 @@ +/*------------------------------------------------------------------------- + * + * gapless_seq.c + * + * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * + * IDENTIFICATION + * contrib/gapless_seq/gapless_seq.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "access/genam.h" +#include "access/htup_details.h" +#include "access/seqamapi.h" +#include "access/transam.h" +#include "access/xact.h" +#include "catalog/indexing.h" +#include "catalog/namespace.h" +#include "catalog/pg_type.h" +#include "commands/sequence.h" +#include "funcapi.h" +#include "nodes/makefuncs.h" +#include "storage/procarray.h" +#include "storage/lmgr.h" +#include "utils/builtins.h" +#include "utils/fmgroids.h" +#include "utils/int8.h" +#include "utils/lsyscache.h" +#include "utils/memutils.h" +#include "utils/rel.h" +#include "utils/syscache.h" + +PG_MODULE_MAGIC; + +/*------------------------------------------------------------ + * + * Sequence Access Manager / Gapless sequence implementation + * + *------------------------------------------------------------ + */ + +typedef struct GaplessSequenceState +{ + int64 last_value; + uint32 xid; + bool is_called; +} GaplessSequenceState; + +typedef struct GaplessValue +{ + Oid seqid; + int64 last_value; + bool is_called; +} GaplessValue; + +#define GAPLESS_SEQ_NAMESPACE "gapless_seq" +#define VALUES_TABLE_NAME "seqam_gapless_values" +#define VALUES_TABLE_COLUMNS 3 + +static bytea *seqam_gapless_reloptions(Datum reloptions, bool validate); +static Datum seqam_gapless_init(Relation seqrel, Form_pg_sequence seq, + int64 restart_value, bool restart_requested, + bool is_init); +static int64 seqam_gapless_alloc(Relation seqrel, SequenceHandle *seqh, + int64 nrequested, int64 *last); +static void seqam_gapless_setval(Relation seqrel, SequenceHandle *seqh, + int64 new_value); +static Datum seqam_gapless_get_state(Relation seqrel, SequenceHandle *seqh); +static void seqam_gapless_set_state(Relation seqrel, SequenceHandle *seqh, + Datum amstate); + +PG_FUNCTION_INFO_V1(seqam_gapless_handler); +PG_FUNCTION_INFO_V1(seqam_gapless_state_in); +PG_FUNCTION_INFO_V1(seqam_gapless_state_out); + +static FormData_pg_sequence *wait_for_sequence(SequenceHandle *seqh, + TransactionId local_xid); +static Relation open_values_rel(void); +static HeapTuple get_last_value_tup(Relation rel, Oid seqid); +static void set_last_value_tup(Relation rel, Oid seqid, HeapTuple oldtuple, + int64 last_value, bool is_called); + +Datum +seqam_gapless_state_in(PG_FUNCTION_ARGS) +{ + char *state_str = PG_GETARG_CSTRING(0); + char *ptr, *end; + Datum d; + GaplessSequenceState *state = palloc(sizeof(LocalSequenceState)); + + ptr = state_str; + while (*ptr && isspace((unsigned char) *ptr)) + ptr++; + + if (*ptr != '(') + goto malformed; + ptr++; + + end = ptr; + while (*end != ',' && *end != ')') + { + char ch = *end++; + + if (ch == '\0') + goto malformed; + } + + d = DirectFunctionCall1(int8in, + CStringGetDatum(pnstrdup(ptr, end-ptr))); + state->last_value = DatumGetInt64(d); + + /* is_called is optional */ + if (*end == ',') + { + ptr = ++end; + + end = state_str+strlen(state_str)-1; + + if (*end != ')') + goto malformed; + + d = DirectFunctionCall1(boolin, + CStringGetDatum(pnstrdup(ptr, end-ptr))); + state->is_called = DatumGetBool(d); + } + else + state->is_called = true; + + state->xid = InvalidTransactionId; + + PG_RETURN_POINTER(state); + +malformed: + ereport(ERROR, + (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION), + errmsg("malformed seqam_gapless_state literal: \"%s\"", + state_str))); +} + +Datum +seqam_gapless_state_out(PG_FUNCTION_ARGS) +{ + GaplessSequenceState *state; + StringInfoData buf; + + state = (GaplessSequenceState *) PG_GETARG_POINTER(0); + + initStringInfo(&buf); + + appendStringInfo(&buf, "("UINT64_FORMAT",%s)", + state->last_value, state->is_called ? "t" : "f"); + + PG_RETURN_CSTRING(buf.data); +} + +/* + * Handler function for the gapless sequence access method. + */ +Datum +seqam_gapless_handler(PG_FUNCTION_ARGS) +{ + SeqAmRoutine *seqam = makeNode(SeqAmRoutine); + Oid nspid, + typid; + + nspid = get_namespace_oid("gapless_seq", false); + typid = GetSysCacheOid2(TYPENAMENSP, + PointerGetDatum("seqam_gapless_state"), + ObjectIdGetDatum(nspid)); + + if (!OidIsValid(typid)) + elog(ERROR, "cache lookup failed for type \"seqam_gapless_state\""); + + seqam->StateTypeOid = typid; + seqam->amoptions = seqam_gapless_reloptions; + seqam->Init = seqam_gapless_init; + seqam->Alloc = seqam_gapless_alloc; + seqam->Setval = seqam_gapless_setval; + seqam->GetState = seqam_gapless_get_state; + seqam->SetState = seqam_gapless_set_state; + + PG_RETURN_POINTER(seqam); +} + +/* + * seqam_gapless_reloptions() + * + * Parse and verify the reloptions of a gapless sequence. + */ +static bytea * +seqam_gapless_reloptions(Datum reloptions, bool validate) +{ + if (validate && PointerIsValid(DatumGetPointer(reloptions))) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("gapless sequence does not accept any storage parameters"))); + + return NULL; +} + +/* + * seqam_gapless_init() + * + * Initialize gapless sequence + * + */ +static Datum +seqam_gapless_init(Relation seqrel, Form_pg_sequence seq, int64 restart_value, + bool restart_requested, bool is_init) +{ + Oid seqrelid = seqrel->rd_id; + Relation valrel; + HeapTuple valtuple; + GaplessSequenceState *seqstate; + + if (is_init) + seqstate = palloc0(sizeof(GaplessSequenceState)); + else + seqstate = (GaplessSequenceState *) seq->amstate; + + /* + * If this is a new sequence or RESTART was provided in ALTER we should + * reset our state to that new starting point. + */ + if (is_init || restart_requested) + { + seqstate->last_value = restart_value; + seqstate->is_called = false; + } + + seqstate->xid = GetTopTransactionId(); + + /* Load current value if this is existing sequence. */ + valrel = open_values_rel(); + valtuple = get_last_value_tup(valrel, seqrelid); + + /* + * If this is new sequence or restart was provided or if there is + * no previous stored value for the sequence we should store it in + * the values table. + */ + if (is_init || restart_requested || !HeapTupleIsValid(valtuple)) + set_last_value_tup(valrel, seqrelid, valtuple, seqstate->last_value, + seqstate->is_called); + + /* Now we are done with values relation, but keep the lock. */ + heap_close(valrel, NoLock); + + return PointerGetDatum(seqstate); +} + +/* + * seqam_gapless_alloc() + * + * Allocate new value for gapless sequence. + */ +static int64 +seqam_gapless_alloc(Relation seqrel, SequenceHandle *seqh, int64 nrequested, + int64 *last) +{ + int64 result; + Oid seqrelid = RelationGetRelid(seqrel); + Relation valrel; + HeapTuple tuple; + Form_pg_sequence seq; + GaplessSequenceState *seqstate; + TransactionId local_xid = GetTopTransactionId(); + + /* Wait until the sequence is locked by us. */ + seq = wait_for_sequence(seqh, local_xid); + seqstate = (GaplessSequenceState *) seq->amstate; + + /* Read the last value from our transactional table (if any). */ + valrel = open_values_rel(); + tuple = get_last_value_tup(valrel, seqrelid); + + /* Last value found get next value. */ + if (HeapTupleIsValid(tuple)) + { + GaplessValue *v = (GaplessValue *) GETSTRUCT(tuple); + result = v->last_value; + + if (v->is_called) + (void) sequence_increment(seqrel, &result, 1, seq->min_value, + seq->max_value, + seq->increment_by, + seq->is_cycled, true); + } + else /* No last value, start from beginning. */ + result = seq->start_value; + + /* Insert or update the last value tuple. */ + set_last_value_tup(valrel, seqrelid, tuple, result, true); + + /* Now we are done with values relation, but keep the lock. */ + heap_close(valrel, NoLock); + + /* We always WAL log for gapless sequence. */ + sequence_start_update(seqh, true); + seqstate->last_value = result; + seqstate->is_called = true; + if (seqstate->xid != local_xid) + seqstate->xid = local_xid; + sequence_save_state(seqh, PointerGetDatum(seqstate), true); + sequence_finish_update(seqh); + + *last = result; + + return result; +} + +/* + * seqam_gapless_setval() + * + * Setval support (we don't allow setval on gapless) + */ +static void +seqam_gapless_setval(Relation seqrel, SequenceHandle *seqh, int64 new_value) +{ + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("setval() is not supported for gapless sequences"))); +} + +/* + * seqam_gapless_get_state() + * + * Dump state of a gapless sequence (for pg_dump) + */ +static Datum +seqam_gapless_get_state(Relation seqrel, SequenceHandle *seqh) +{ + Oid seqrelid = RelationGetRelid(seqrel); + Relation valrel; + HeapTuple tuple; + GaplessSequenceState *seqstate = palloc0(sizeof(GaplessSequenceState)); + + /* + * Get the last value from the values table, if not found use the values + * from sequence typle. + */ + valrel = open_values_rel(); + tuple = get_last_value_tup(valrel, seqrelid); + heap_close(valrel, RowExclusiveLock); + + if (HeapTupleIsValid(tuple)) + { + GaplessValue *v = (GaplessValue *) GETSTRUCT(tuple); + seqstate->last_value = v->last_value; + seqstate->is_called = v->is_called; + } + else + { + Datum amstate = sequence_read_state(seqh); + + memcpy(seqstate, DatumGetPointer(amstate), + sizeof(GaplessSequenceState)); + sequence_release_tuple(seqh); + } + + return PointerGetDatum(seqstate); +} + +/* + * seqam_gapless_set_state() + * + * Restore previously dumpred state of gapless sequence + */ +static void +seqam_gapless_set_state(Relation seqrel, SequenceHandle *seqh, + Datum amstate) +{ + Oid seqrelid = RelationGetRelid(seqrel); + Relation valrel; + HeapTuple tuple; + FormData_pg_sequence *seq; + GaplessSequenceState *seqstate; + TransactionId local_xid = GetTopTransactionId(); + + /* Wait until the sequence is locked by us. */ + seq = wait_for_sequence(seqh, local_xid); + seqstate = (GaplessSequenceState *) DatumGetPointer(amstate); + + sequence_check_range(seqstate->last_value, seq->min_value, seq->max_value, + "last_value"); + + /* Read the last value from our transactional table (if any). */ + valrel = open_values_rel(); + tuple = get_last_value_tup(valrel, seqrelid); + + /* Insert or update the last value tuple. */ + set_last_value_tup(valrel, seqrelid, tuple, seqstate->last_value, + seqstate->is_called); + + /* Now we are done with values relation, but keep the lock. */ + heap_close(valrel, NoLock); + + /* Save to updated sequence. */ + sequence_start_update(seqh, true); + sequence_save_state(seqh, amstate, true); + sequence_finish_update(seqh); + + sequence_release_tuple(seqh); +} + +/* + * Lock the sequence for current transaction. + */ +static FormData_pg_sequence * +wait_for_sequence(SequenceHandle *seqh, TransactionId local_xid) +{ + FormData_pg_sequence *seq = sequence_read_options(seqh); + GaplessSequenceState *seqstate; + + seqstate = (GaplessSequenceState *) seq->amstate; + + /* + * Read and lock the sequence for our transaction, there can't be any + * concurrent transactions accessing the sequence at the same time. + */ + while (seqstate->xid != local_xid && + TransactionIdIsInProgress(seqstate->xid)) + { + /* + * Release tuple to avoid dead locks and wait for the concurrent tx + * to finish. + */ + sequence_release_tuple(seqh); + XactLockTableWait(seqstate->xid, NULL, NULL, XLTW_None); + /* Reread the sequence. */ + seq = sequence_read_options(seqh); + } + + return seq; +} + +/* + * Open the relation used for storing last value in RowExclusive lock mode. + */ +static Relation +open_values_rel(void) +{ + RangeVar *rv; + Oid valrelid; + Relation valrel; + + rv = makeRangeVar(GAPLESS_SEQ_NAMESPACE, VALUES_TABLE_NAME, -1); + valrelid = RangeVarGetRelid(rv, RowExclusiveLock, false); + valrel = heap_open(valrelid, RowExclusiveLock); + + return valrel; +} + +/* + * Read the last value tuple from the values table. + * + * Can return NULL if tuple is not found. + */ +static HeapTuple +get_last_value_tup(Relation rel, Oid seqid) +{ + ScanKey key; + SysScanDesc scan; + HeapTuple tuple; + + key = (ScanKey) palloc(sizeof(ScanKeyData) * 1); + + ScanKeyInit(&key[0], + 1, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(seqid) + ); + + /* FIXME: should use index */ + scan = systable_beginscan(rel, 0, true, NULL, 1, key); + tuple = systable_getnext(scan); + if (HeapTupleIsValid(tuple)) + tuple = heap_copytuple(tuple); + systable_endscan(scan); + + return tuple; +} + +/* + * Insert or update the last value tuple. + * + * The write access to the table must be serialized by the wait_for_sequence + * so that we don't have to have any retry scheme here. +*/ +static void +set_last_value_tup(Relation rel, Oid seqid, HeapTuple oldtuple, + int64 last_value, bool is_called) +{ + bool nulls[VALUES_TABLE_COLUMNS]; + Datum values[VALUES_TABLE_COLUMNS]; + TupleDesc tupDesc = RelationGetDescr(rel); + HeapTuple tuple; + + if (!HeapTupleIsValid(oldtuple)) + { + memset(nulls, false, VALUES_TABLE_COLUMNS * sizeof(bool)); + values[0] = ObjectIdGetDatum(seqid); + values[1] = Int64GetDatum(last_value); + values[2] = BoolGetDatum(is_called); + + tuple = heap_form_tuple(tupDesc, values, nulls); + simple_heap_insert(rel, tuple); + } + else + { + bool replaces[VALUES_TABLE_COLUMNS]; + + replaces[0] = false; + replaces[1] = true; + replaces[2] = true; + + nulls[1] = false; + nulls[2] = false; + values[1] = Int64GetDatum(last_value); + values[2] = BoolGetDatum(is_called); + + tuple = heap_modify_tuple(oldtuple, tupDesc, values, nulls, replaces); + simple_heap_update(rel, &tuple->t_self, tuple); + } + + CatalogUpdateIndexes(rel, tuple); +} diff --git a/contrib/gapless_seq/gapless_seq.control b/contrib/gapless_seq/gapless_seq.control new file mode 100644 index 0000000..85da739 --- /dev/null +++ b/contrib/gapless_seq/gapless_seq.control @@ -0,0 +1,6 @@ +# Gapless sequence extension +comment = 'Gapless Sequence AM' +default_version = '1.0' +module_pathname = '$libdir/gapless_seq' +relocatable = false +schema = gapless_seq diff --git a/contrib/gapless_seq/specs/concurrency.spec b/contrib/gapless_seq/specs/concurrency.spec new file mode 100644 index 0000000..3a0cc57 --- /dev/null +++ b/contrib/gapless_seq/specs/concurrency.spec @@ -0,0 +1,29 @@ +setup +{ + CREATE EXTENSION IF NOT EXISTS gapless_seq; + DROP SEQUENCE IF EXISTS test_gapless; + CREATE SEQUENCE test_gapless USING gapless; +} + +teardown +{ + DROP SEQUENCE test_gapless; +} + +session "s1" +step "s1_begin" { BEGIN; } +step "s1_nextval" { SELECT nextval('test_gapless'::regclass); } +step "s1_commit" { COMMIT; } + +session "s2" +step "s2_begin" { BEGIN; } +step "s2_nextval" { SELECT nextval('test_gapless'::regclass); } +step "s2_commit" { COMMIT; } + +session "s3" +step "s3_begin" { BEGIN; } +step "s3_nextval" { SELECT nextval('test_gapless'::regclass); } +step "s3_rollback" { ROLLBACK; } + +permutation "s1_begin" "s1_nextval" "s2_begin" "s2_nextval" "s1_commit" "s2_commit" +permutation "s3_begin" "s3_nextval" "s2_begin" "s2_nextval" "s3_rollback" "s2_commit" diff --git a/contrib/gapless_seq/sql/gapless_seq.sql b/contrib/gapless_seq/sql/gapless_seq.sql new file mode 100644 index 0000000..b49cec6 --- /dev/null +++ b/contrib/gapless_seq/sql/gapless_seq.sql @@ -0,0 +1,61 @@ +CREATE EXTENSION gapless_seq; + +CREATE SEQUENCE test_gapless USING gapless; + +SELECT nextval('test_gapless'::regclass); + +BEGIN; + SELECT nextval('test_gapless'::regclass); + SELECT nextval('test_gapless'::regclass); + SELECT nextval('test_gapless'::regclass); +ROLLBACK; + +SELECT nextval('test_gapless'::regclass); + +CREATE SEQUENCE test_alter_seq USING local; + +SELECT nextval('test_alter_seq'::regclass); + +ALTER SEQUENCE test_alter_seq USING gapless; + +SELECT nextval('test_alter_seq'::regclass); + +BEGIN; + SELECT nextval('test_alter_seq'::regclass); +ROLLBACK; + +SELECT nextval('test_alter_seq'::regclass); + +BEGIN; + SELECT nextval('test_alter_seq'::regclass); + SAVEPOINT mysp; + SELECT nextval('test_alter_seq'::regclass); + ROLLBACK TO SAVEPOINT mysp; + SAVEPOINT mysp2; + SELECT nextval('test_alter_seq'::regclass); + RELEASE SAVEPOINT mysp2; + SELECT nextval('test_alter_seq'::regclass); +COMMIT; + +ALTER SEQUENCE test_alter_seq RESTART 100 USING local; + +SELECT nextval('test_alter_seq'::regclass); + +-- check dump/restore +SELECT pg_sequence_get_state('test_gapless'); +SELECT pg_sequence_set_state('test_gapless', pg_sequence_get_state('test_gapless')); +SELECT pg_sequence_get_state('test_gapless'); + +-- check that event trigger works correctly +SELECT last_value FROM gapless_seq.seqam_gapless_values ORDER BY seqid; +DROP SEQUENCE test_gapless; +SELECT last_value FROM gapless_seq.seqam_gapless_values ORDER BY seqid; + +CREATE SEQUENCE test_gapless USING gapless; + +-- should fail due to deps +DROP ACCESS METHOD gapless; +-- likewise +DROP EXTENSION gapless_seq; +-- success +DROP EXTENSION gapless_seq CASCADE; diff --git a/doc/src/sgml/contrib.sgml b/doc/src/sgml/contrib.sgml index 4e3f337..9852933 100644 --- a/doc/src/sgml/contrib.sgml +++ b/doc/src/sgml/contrib.sgml @@ -116,6 +116,7 @@ CREATE EXTENSION module_name FROM unpackaged; &earthdistance; &file-fdw; &fuzzystrmatch; + &gapless-seq; &hstore; &intagg; &intarray; diff --git a/doc/src/sgml/filelist.sgml b/doc/src/sgml/filelist.sgml index fd355dc..2681437 100644 --- a/doc/src/sgml/filelist.sgml +++ b/doc/src/sgml/filelist.sgml @@ -119,6 +119,7 @@ + diff --git a/doc/src/sgml/gapless-seq.sgml b/doc/src/sgml/gapless-seq.sgml new file mode 100644 index 0000000..44cc224 --- /dev/null +++ b/doc/src/sgml/gapless-seq.sgml @@ -0,0 +1,24 @@ + + + gapless_seq + + + gapless_seq + + + + gapless_seq provides a sequence implementation that never + has gaps in the sequence of values it produces, even after rollback or + a crash. + + + + The consequence of this capability is that every nextval() request + writes a WAL record recording the latest state of the sequence and hold + lock on the sequence itself, so any concurrent nextval() calls have to wait + for the transaction to finish. This could be very costly and is not + recommended for general usage except in specific applications that require + this feature. + + + -- 1.9.1