text: /*-------------------------------------------------------------------------
   *
   * sequence.c
   *	  PostgreSQL sequences support code.
   *
   * Portions Copyright (c) 2005-2008, Greenplum inc.
   * Portions Copyright (c) 2012-Present Pivotal Software, Inc.
   * Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
   * Portions Copyright (c) 1994, Regents of the University of California
   *
   *
   * IDENTIFICATION
   *	  src/backend/commands/sequence.c
   *
   *-------------------------------------------------------------------------
   */
  #include "postgres.h"
  
  #include "access/bufmask.h"
  #include "access/htup_details.h"
  #include "access/multixact.h"
  #include "access/transam.h"
  #include "access/xact.h"
  #include "access/xlog.h"
  #include "access/xloginsert.h"
  #include "access/xlogutils.h"
  #include "catalog/dependency.h"
  #include "catalog/heap.h"
  #include "catalog/namespace.h"
  #include "catalog/objectaccess.h"
  #include "catalog/pg_type.h"
  #include "commands/defrem.h"
  #include "commands/sequence.h"
  #include "commands/tablecmds.h"
  #include "funcapi.h"
  #include "libpq/libpq.h"
  #include "libpq/pqformat.h"
  #include "miscadmin.h"
  #include "storage/smgr.h"               /* RelationCloseSmgr -> smgrclose */
  #include "nodes/makefuncs.h"
  #include "storage/lmgr.h"
  #include "storage/proc.h"
  #include "storage/smgr.h"
  #include "utils/acl.h"
  #include "utils/builtins.h"
  #include "utils/formatting.h"
  #include "utils/lsyscache.h"
  #include "utils/resowner.h"
  #include "utils/syscache.h"
  
  #include "catalog/oid_dispatch.h"
  #include "cdb/cdbdisp_query.h"
  #include "cdb/cdbvars.h"
  #include "cdb/cdbmotion.h"
  #include "cdb/ml_ipc.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 (skip over) as many values 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.)
   */
  typedef struct SeqTableData
  {
  	Oid			relid;			/* pg_class OID of this sequence (hash key) */
  	Oid			filenode;		/* last seen relfilenode 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_seq_tuple() */
  } SeqTableData;
  
  typedef SeqTableData *SeqTable;
  
  static HTAB *seqhashtab = NULL; /* hash table for SeqTable items */
  
  /*
   * last_used_seq is updated by nextval() to point to the last used
   * sequence.
   */
  static SeqTableData *last_used_seq = NULL;
  
  static void fill_seq_with_data(Relation rel, HeapTuple tuple);
  static int64 nextval_internal(Oid relid, bool is_direct_request);
  static Relation open_share_lock(SeqTable seq);
  static void create_seq_hashtable(void);
  static void init_sequence(Oid relid, SeqTable *p_elm, Relation *p_rel);
  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);
  static void process_owned_by(Relation seqrel, List *owned_by);
  static void mask_seq_values(Page page);
  
  static void
  cdb_sequence_nextval_qe(Relation seqrel,
  						int64   *plast,
  						int64   *pcached,
  						int64   *pincrement,
  						bool    *pvalid);
  
  /*
   * DefineSequence
   *				Creates a new sequence relation
   */
  ObjectAddress
  DefineSequence(CreateSeqStmt *seq)
  {
  	FormData_pg_sequence new;
  	List	   *owned_by;
  	CreateStmt *stmt = makeNode(CreateStmt);
  	Oid			seqoid;
  	ObjectAddress address;
  	Relation	rel;
  	HeapTuple	tuple;
  	TupleDesc	tupDesc;
  	Datum		value[SEQ_COL_LASTCOL];
  	bool		null[SEQ_COL_LASTCOL];
  	int			i;
  	NameData	name;
  
  	bool shouldDispatch =  Gp_role == GP_ROLE_DISPATCH && !IsBootstrapProcessingMode();
  
  	/* Unlogged sequences are not implemented -- not clear if useful. */
  	if (seq->sequence->relpersistence == RELPERSISTENCE_UNLOGGED)
  		ereport(ERROR,
  				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
  				 errmsg("unlogged sequences are not supported")));
  
  	/*
  	 * If if_not_exists was given and a relation with the same name already
  	 * exists, bail out. (Note: we needn't check this when not if_not_exists,
  	 * because DefineRelation will complain anyway.)
  	 */
  	if (seq->if_not_exists)
  	{
  		RangeVarGetAndCheckCreationNamespace(seq->sequence, NoLock, &seqoid);
  		if (OidIsValid(seqoid))
  		{
  			ereport(NOTICE,
  					(errcode(ERRCODE_DUPLICATE_TABLE),
  					 errmsg("relation \"%s\" already exists, skipping",
  							seq->sequence->relname)));
  			return InvalidObjectAddress;
  		}
  	}
  
  	/* 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->is_from_type = false;
  		coldef->storage = 0;
  		coldef->raw_default = NULL;
  		coldef->cooked_default = NULL;
  		coldef->collClause = NULL;
  		coldef->collOid = InvalidOid;
  		coldef->constraints = NIL;
  		coldef->location = -1;
  
  		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) 0);
  				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->inhOids = NIL;
  	stmt->parentOidCount = 0;
  	stmt->options = NIL;
  	stmt->oncommit = ONCOMMIT_NOOP;
  	stmt->tablespacename = NULL;
  	stmt->if_not_exists = seq->if_not_exists;
  	stmt->relKind = RELKIND_SEQUENCE;
  	stmt->ownerid = GetUserId();
  
  	address = DefineRelation(stmt, RELKIND_SEQUENCE, seq->ownerId, NULL, RELSTORAGE_HEAP, false, true, NULL);
  	seqoid = address.objectId;
  	Assert(seqoid != InvalidOid);
  
  	/*
  	 * Open and lock the new sequence.  (This lock is redundant; an
  	 * AccessExclusiveLock was acquired above by DefineRelation and
  	 * won't be released until end of transaction.)
  	 *
  	 * CDB: Acquire lock on qDisp before dispatching to qExecs, so
  	 * qDisp can detect and resolve any deadlocks.
  	 */
  	rel = heap_open(seqoid, AccessExclusiveLock);
  	tupDesc = RelationGetDescr(rel);
  
  	/* now initialize the sequence's data */
  	tuple = heap_form_tuple(tupDesc, value, null);
  	fill_seq_with_data(rel, tuple);
  
  	/* Dispatch to segments */
  	if (shouldDispatch)
  	{
  		CdbDispatchUtilityStatement((Node *) seq,
  									DF_CANCEL_ON_ERROR|
  									DF_WITH_SNAPSHOT|
  									DF_NEED_TWO_PHASE,
  									GetAssignedOidsForDispatch(),
  									NULL);
  	}
  
  	/* process OWNED BY if given */
  	if (owned_by)
  		process_owned_by(rel, owned_by);
  
  	heap_close(rel, NoLock);
  
  	return address;
  }
  
  /*
   * Reset a sequence to its initial value.
   *
   * The change is made transactionally, so that on failure of the current
   * transaction, the sequence will be restored to its previous state.
   * We do that by creating a whole new relfilenode for the sequence; so this
   * works much like the rewriting forms of ALTER TABLE.
   *
   * Caller is assumed to have acquired AccessExclusiveLock on the sequence,
   * which must not be released until end of transaction.  Caller is also
   * responsible for permissions checking.
   */
  void
  ResetSequence(Oid seq_relid)
  {
  	Relation	seq_rel;
  	SeqTable	elm;
  	Form_pg_sequence seq;
  	Buffer		buf;
  	HeapTupleData seqtuple;
  	HeapTuple	tuple;
  
  	/*
  	 * Read the old sequence.  This does a bit more work than really
  	 * necessary, but it's simple, and we do want to double-check that it's
  	 * indeed a sequence.
  	 */
  	init_sequence(seq_relid, &elm, &seq_rel);
  	(void) read_seq_tuple(elm, seq_rel, &buf, &seqtuple);
  
  	/*
  	 * Copy the existing sequence tuple.
  	 */
  	tuple = heap_copytuple(&seqtuple);
  
  	/* Now we're done with the old page */
  	UnlockReleaseBuffer(buf);
  
  	/*
  	 * Modify the copied tuple to execute the restart (compare the RESTART
  	 * action in AlterSequence)
  	 */
  	seq = (Form_pg_sequence) GETSTRUCT(tuple);
  	seq->last_value = seq->start_value;
  	seq->is_called = false;
  	seq->log_cnt = 0;
  
  	/*
  	 * Create a new storage file for the sequence.  We want to keep the
  	 * sequence's relfrozenxid at 0, since it won't contain any unfrozen XIDs.
  	 * Same with relminmxid, since a sequence will never contain multixacts.
  	 */
  	RelationSetNewRelfilenode(seq_rel, seq_rel->rd_rel->relpersistence,
  							  InvalidTransactionId, InvalidMultiXactId);
  
  	/*
  	 * Insert the modified tuple into the new storage file.
  	 */
  	fill_seq_with_data(seq_rel, tuple);
  
  	/* 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;
  
  	relation_close(seq_rel, NoLock);
  }
  
  /*
   * Initialize a sequence's relation with the specified tuple as content
   */
  static void
  fill_seq_with_data(Relation rel, HeapTuple tuple)
  {
  	Buffer		buf;
  	Page		page;
  	sequence_magic *sm;
  	OffsetNumber offnum;
  
  	/* Initialize first page of relation with special magic number */
  
  	buf = ReadBuffer(rel, P_NEW);
  	Assert(BufferGetBlockNumber(buf) == 0);
  
  	page = BufferGetPage(buf);
  
  	LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
  
  	PageInit(page, BufferGetPageSize(buf), sizeof(sequence_magic));
  	sm = (sequence_magic *) PageGetSpecialPointer(page);
  	sm->magic = SEQ_MAGIC;
  
  	/* Now insert sequence tuple */
  
  	/*
  	 * 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.
  	 */
  	HeapTupleHeaderSetXmin(tuple->t_data, FrozenTransactionId);
  	HeapTupleHeaderSetXminFrozen(tuple->t_data);
  	HeapTupleHeaderSetCmin(tuple->t_data, FirstCommandId);
  	HeapTupleHeaderSetXmax(tuple->t_data, InvalidTransactionId);
  	tuple->t_data->t_infomask |= HEAP_XMAX_INVALID;
  	ItemPointerSet(&tuple->t_data->t_ctid, 0, FirstOffsetNumber);
  
  	/* check the comment above nextval_internal()'s equivalent call. */
  	if (RelationNeedsWAL(rel))
  		GetTopTransactionId();
  
  	START_CRIT_SECTION();
  
  	MarkBufferDirty(buf);
  
  	offnum = PageAddItem(page, (Item) tuple->t_data, tuple->t_len,
  						 InvalidOffsetNumber, false, false);
  	if (offnum != FirstOffsetNumber)
  		elog(ERROR, "failed to add sequence tuple to page");
  
  	/* XLOG stuff */
  	if (RelationNeedsWAL(rel))
  	{
  		xl_seq_rec	xlrec;
  		XLogRecPtr	recptr;
  
  		XLogBeginInsert();
  		XLogRegisterBuffer(0, buf, REGBUF_WILL_INIT);
  
  		xlrec.node = rel->rd_node;
  
  		XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
  		XLogRegisterData((char *) tuple->t_data, tuple->t_len);
  
  		recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
  
  		PageSetLSN(page, recptr);
  	}
  
  	END_CRIT_SECTION();
  
  	UnlockReleaseBuffer(buf);
  }
  
  /*
   * AlterSequence
   *
   * Modify the definition of a sequence relation
   */
  ObjectAddress
  AlterSequence(AlterSeqStmt *stmt)
  {
  	Oid			relid;
  	SeqTable	elm;
  	Relation	seqrel;
  	Buffer		buf;
  	HeapTupleData seqtuple;
  	Form_pg_sequence seq;
  	FormData_pg_sequence new;
  	List	   *owned_by;
  	ObjectAddress address;
  	bool        bSeqIsTemp = false;
  	int			numopts;
  	char	   *alter_subtype = "";		/* metadata tracking: kind of
  										   redundant to say "role" */
  
  	/* Open and lock sequence. */
  	relid = RangeVarGetRelid(stmt->sequence, AccessShareLock, stmt->missing_ok);
  	if (relid == InvalidOid)
  	{
  		ereport(NOTICE,
  				(errmsg("relation \"%s\" does not exist, skipping",
  						stmt->sequence->relname)));
  		return InvalidObjectAddress;
  	}
  
  	init_sequence(relid, &elm, &seqrel);
  
  	/* allow ALTER to sequence owner only */
  	if (!pg_class_ownercheck(relid, GetUserId()))
  		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
  					   stmt->sequence->relname);
  
  	/* lock page' buffer and read tuple into new sequence structure */
  	seq = read_seq_tuple(elm, seqrel, &buf, &seqtuple);
  
  	/* Copy old values of options into workspace */
  	memcpy(&new, seq, sizeof(FormData_pg_sequence));
  
  	/* Check and set new values */
  	init_params(stmt->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;
  
  	/* check the comment above nextval_internal()'s equivalent call. */
  	if (RelationNeedsWAL(seqrel))
  		GetTopTransactionId();
  
  	/* Now okay to update the on-disk tuple */
  	START_CRIT_SECTION();
  
  	memcpy(seq, &new, sizeof(FormData_pg_sequence));
  
  	MarkBufferDirty(buf);
  
  	/* XLOG stuff */
  	if (RelationNeedsWAL(seqrel))
  	{
  		xl_seq_rec	xlrec;
  		XLogRecPtr	recptr;
  		Page		page = BufferGetPage(buf);
  
  		XLogBeginInsert();
  		XLogRegisterBuffer(0, buf, REGBUF_WILL_INIT);
  
  		xlrec.node = seqrel->rd_node;
  		XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
  
  		XLogRegisterData((char *) seqtuple.t_data, seqtuple.t_len);
  
  		recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
  
  		PageSetLSN(page, recptr);
  	}
  
  	END_CRIT_SECTION();
  
  	UnlockReleaseBuffer(buf);
  
  	/* process OWNED BY if given */
  	if (owned_by)
  		process_owned_by(seqrel, owned_by);
  
  	bSeqIsTemp = (seqrel->rd_rel->relpersistence == RELPERSISTENCE_TEMP);
  
  	numopts = list_length(stmt->options);
  	if (numopts > 1)
  	{
  		alter_subtype = psprintf("%d OPTIONS", numopts);
  	}
  	else if (0 == numopts)
  	{
  		alter_subtype = "0 OPTIONS";
  	}
  	else if (Gp_role == GP_ROLE_DISPATCH && !bSeqIsTemp)
  	{
  		ListCell		*option = list_head(stmt->options);
  		DefElem			*defel	= (DefElem *) lfirst(option);
  		char			*tempo	= NULL;
  
  		alter_subtype = defel->defname;
  		if (0 == strcmp(alter_subtype, "owned_by"))
  			alter_subtype = "OWNED BY";
  
  		tempo = asc_toupper(alter_subtype, strlen(alter_subtype));
  
  		alter_subtype = tempo;
  	}
  
  	if (Gp_role == GP_ROLE_DISPATCH && !bSeqIsTemp)
  	{
  		/* MPP-6929: metadata tracking */
  		MetaTrackUpdObject(RelationRelationId,
  						   relid,
  						   GetUserId(),
  						   "ALTER", alter_subtype);
  	}
  
  	if (Gp_role == GP_ROLE_DISPATCH)
  		CdbDispatchUtilityStatement((Node *) stmt,
  									DF_CANCEL_ON_ERROR|
  									DF_WITH_SNAPSHOT|
  									DF_NEED_TWO_PHASE,
  									NIL,
  									NULL);
  	InvokeObjectPostAlterHook(RelationRelationId, relid, 0);
  
  	ObjectAddressSet(address, RelationRelationId, relid);
  
  	relation_close(seqrel, NoLock);
  
  	return address;
  }
  
  
  /*
   * 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));
  
  	/*
  	 * XXX: This is not safe in the presence of concurrent DDL, but acquiring
  	 * a lock here is more expensive than letting nextval_internal do it,
  	 * since the latter maintains a cache that keeps us from hitting the lock
  	 * manager more than once per transaction.  It's not clear whether the
  	 * performance penalty is material in practice, but for now, we do it this
  	 * way.
  	 */
  	relid = RangeVarGetRelid(sequence, NoLock, false);
  
  	PG_RETURN_INT64(nextval_internal(relid, false));
  }
  
  Datum
  nextval_oid(PG_FUNCTION_ARGS)
  {
  	Oid			relid = PG_GETARG_OID(0);
  
  	PG_RETURN_INT64(nextval_internal(relid, false));
  }
  
  void
  nextval_qd(Oid relid, int64 *plast, int64 *pcached, int64  *pincrement, bool *poverflow)
  {
  	Assert(IS_QUERY_DISPATCHER());
  
  	*plast = nextval_internal(relid, true);
  	*pcached = last_used_seq->cached;
  	*pincrement = last_used_seq->increment;
  	*poverflow = !last_used_seq->last_valid;
  }
  
  static int64
  nextval_internal(Oid relid, bool called_from_dispatcher)
  {
  	SeqTable	elm;
  	Relation	seqrel;
  	Buffer		buf;
  	Page		page;
  	HeapTupleData seqtuple;
  	Form_pg_sequence seq;
  	int64		incby,
  				maxv,
  				minv,
  				cache,
  				log,
  				fetch,
  				last;
  	int64		result,
  				next,
  				rescnt = 0;
  	bool            logit = false;
  
  	/* open and AccessShareLock sequence */
  	init_sequence(relid, &elm, &seqrel);
  
  	if (pg_class_aclcheck(elm->relid, GetUserId(),
  						  ACL_USAGE | ACL_UPDATE) != ACLCHECK_OK)
  		ereport(ERROR,
  				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
  				 errmsg("permission denied for sequence %s",
  						RelationGetRelationName(seqrel))));
  
  	/* read-only transactions may only modify temp sequences */
  	if (!seqrel->rd_islocaltemp)
  		PreventCommandIfReadOnly("nextval()");
  
  	/*
  	 * Forbid this during parallel operation because, to make it work, the
  	 * cooperating backends would need to share the backend-local cached
  	 * sequence information.  Currently, we don't support that.
  	 */
  	PreventCommandIfParallelMode("nextval()");
  
  	if (elm->last != elm->cached 		/* some numbers were cached */
  		&& !called_from_dispatcher)
  	{
  		Assert(elm->last_valid);
  		Assert(elm->increment != 0);
  		elm->last += elm->increment;
  		relation_close(seqrel, NoLock);
  		last_used_seq = elm;
  		return elm->last;
  	}
  
  	/* Update the sequence object. */
  	if (Gp_role == GP_ROLE_EXECUTE)
  	{
  		cdb_sequence_nextval_qe(seqrel,
  								&elm->last,
  								&elm->cached,
  								&elm->increment,
  								&elm->last_valid);
  
  		last_used_seq = elm;
  		relation_close(seqrel, NoLock);
  
  		return elm->last;
  	}
  
  	/* lock page' buffer and read tuple */
  	seq = read_seq_tuple(elm, seqrel, &buf, &seqtuple);
  	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++;				/* return last_value if not is_called */
  		fetch--;
  	}
  
  	/*
  	 * 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 || !seq->is_called)
  	{
  		/* forced log to satisfy local demand for values */
  		fetch = log = fetch + SEQ_LOG_VALS;
  		logit = true;
  	}
  	else
  	{
  		XLogRecPtr	redoptr = GetRedoRecPtr();
  
  		if (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;
  	elm->increment = incby;
  
  	last_used_seq = elm;
  
  	/*
  	 * If something needs to be WAL logged, acquire an xid, so this
  	 * transaction's commit will trigger a WAL flush and wait for syncrep.
  	 * It's sufficient to ensure the toplevel transaction has an xid, no need
  	 * to assign xids subxacts, that'll already trigger an appropriate wait.
  	 * (Have to do that here, so we're outside the critical section)
  	 */
  	if (logit && RelationNeedsWAL(seqrel))
  		GetTopTransactionId();
  
  	/* 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 */
  	if (logit && RelationNeedsWAL(seqrel))
  	{
  		xl_seq_rec	xlrec;
  		XLogRecPtr	recptr;
  
  		/*
  		 * 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.
  		 */
  		XLogBeginInsert();
  		XLogRegisterBuffer(0, buf, REGBUF_WILL_INIT);
  
  		/* set values that will be saved in xlog */
  		seq->last_value = next;
  		seq->is_called = true;
  		seq->log_cnt = 0;
  
  		xlrec.node = seqrel->rd_node;
  
  		XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
  		XLogRegisterData((char *) seqtuple.t_data, seqtuple.t_len);
  
  		recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
  
  		PageSetLSN(page, recptr);
  	}
  
  	/* 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 */
  
  	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;
  
  	/* For now, strictly forbidden on MPP. */
  	if (Gp_role == GP_ROLE_DISPATCH || Gp_role == GP_ROLE_EXECUTE)
  	{
  		ereport(ERROR,
  				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
  				 errmsg("currval() not supported")));
  	}
  
  	/* open and AccessShareLock sequence */
  	init_sequence(relid, &elm, &seqrel);
  
  	if (pg_class_aclcheck(elm->relid, GetUserId(),
  						  ACL_SELECT | 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;
  
  	/* For now, strictly forbidden on MPP. */
  	if (Gp_role == GP_ROLE_DISPATCH || Gp_role == GP_ROLE_EXECUTE)
  	{
  		ereport(ERROR,
  				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
  				 errmsg("lastval() not supported")));
  	}
  
  	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 | 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;
  	HeapTupleData seqtuple;
  	Form_pg_sequence seq;
  
  	if (Gp_role == GP_ROLE_EXECUTE)
  	{
  		ereport(ERROR,
  				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
  				 errmsg("setval() not supported in this context")));
  	}
  
  	/* 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))));
  
  	/* read-only transactions may only modify temp sequences */
  	if (!seqrel->rd_islocaltemp)
  		PreventCommandIfReadOnly("setval()");
  
  	/*
  	 * Forbid this during parallel operation because, to make it work, the
  	 * cooperating backends would need to share the backend-local cached
  	 * sequence information.  Currently, we don't support that.
  	 */
  	PreventCommandIfParallelMode("setval()");
  
  	/* lock page' buffer and read tuple */
  	seq = read_seq_tuple(elm, seqrel, &buf, &seqtuple);
  
  	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;
  
  	/* check the comment above nextval_internal()'s equivalent call. */
  	if (RelationNeedsWAL(seqrel))
  		GetTopTransactionId();
  
  	/* 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 */
  	if (RelationNeedsWAL(seqrel))
  	{
  		xl_seq_rec	xlrec;
  		XLogRecPtr	recptr;
  		Page		page = BufferGetPage(buf);
  
  		XLogBeginInsert();
  		XLogRegisterBuffer(0, buf, REGBUF_WILL_INIT);
  
  		xlrec.node = seqrel->rd_node;
  		XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
  		XLogRegisterData((char *) seqtuple.t_data, seqtuple.t_len);
  
  		recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
  
  		PageSetLSN(page, recptr);
  	}
  
  	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);
  }
  
  /*
   * Creates the hash table for storing sequence data
   */
  static void
  create_seq_hashtable(void)
  {
  	HASHCTL		ctl;
  
  	memset(&ctl, 0, sizeof(ctl));
  	ctl.keysize = sizeof(Oid);
  	ctl.entrysize = sizeof(SeqTableData);
  
  	seqhashtab = hash_create("Sequence values", 16, &ctl,
  							 HASH_ELEM | HASH_BLOBS);
  }
  
  /*
   * 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;
  	bool		found;
  
  	/* Find or create a hash table entry for this sequence */
  	if (seqhashtab == NULL)
  		create_seq_hashtable();
  
  	elm = (SeqTable) hash_search(seqhashtab, &relid, HASH_ENTER, &found);
  
  	/*
  	 * Initialize the new hash table entry if it did not exist already.
  	 *
  	 * NOTE: seqtable entries are stored for the life of a backend (unless
  	 * explicitly discarded with DISCARD). If the sequence itself is deleted
  	 * then the entry becomes wasted memory, but it's small enough that this
  	 * should not matter.
  	 */
  	if (!found)
  	{
  		/* relid already filled in */
  		elm->filenode = InvalidOid;
  		elm->lxid = InvalidLocalTransactionId;
  		elm->last_valid = false;
  		elm->last = elm->cached = elm->increment = 0;
  	}
  
  	/*
  	 * 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))));
  
  	/*
  	 * If the sequence has been transactionally replaced since we last saw it,
  	 * discard any cached-but-unissued values.  We do not touch the currval()
  	 * state, however.
  	 */
  	if (seqrel->rd_rel->relfilenode != elm->filenode)
  	{
  		elm->filenode = seqrel->rd_rel->relfilenode;
  		elm->cached = elm->last;
  	}
  
  	/* Return results */
  	*p_elm = elm;
  	*p_rel = seqrel;
  }
  
  
  /*
   * 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_seq_tuple(SeqTable elm, Relation rel, Buffer *buf, HeapTuple seqtuple)
  {
  	Page		page;
  	ItemId		lp;
  	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));
  
  	/* 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
  	 * a sequence, which would leave a non-frozen XID in the sequence tuple's
  	 * xmax, which eventually leads to clog access failures or worse. If we
  	 * see this has happened, clean up after it.  We treat this like a hint
  	 * bit update, ie, don't bother to WAL-log it, since we can certainly do
  	 * this again if the update gets lost.
  	 */
  	Assert(!(seqtuple->t_data->t_infomask & HEAP_XMAX_IS_MULTI));
  	if (HeapTupleHeaderGetRawXmax(seqtuple->t_data) != InvalidTransactionId)
  	{
  		HeapTupleHeaderSetXmax(seqtuple->t_data, InvalidTransactionId);
  		seqtuple->t_data->t_infomask &= ~HEAP_XMAX_COMMITTED;
  		seqtuple->t_data->t_infomask |= HEAP_XMAX_INVALID;
  		MarkBufferDirtyHint(*buf, true);
  	}
  
  	seq = (Form_pg_sequence) GETSTRUCT(seqtuple);
  
  	/* 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,
  			Form_pg_sequence new, List **owned_by)
  {
  	DefElem    *start_value = NULL;
  	DefElem    *restart_value = NULL;
  	DefElem    *increment_by = NULL;
  	DefElem    *max_value = NULL;
  	DefElem    *min_value = NULL;
  	DefElem    *cache_value = NULL;
  	DefElem    *is_cycled = NULL;
  	ListCell   *option;
  
  	*owned_by = NIL;
  
  	foreach(option, options)
  	{
  		DefElem    *defel = (DefElem *) lfirst(option);
  
  		if (strcmp(defel->defname, "increment") == 0)
  		{
  			if (increment_by)
  				ereport(ERROR,
  						(errcode(ERRCODE_SYNTAX_ERROR),
  						 errmsg("conflicting or redundant options")));
  			increment_by = defel;
  		}
  		else if (strcmp(defel->defname, "start") == 0)
  		{
  			if (start_value)
  				ereport(ERROR,
  						(errcode(ERRCODE_SYNTAX_ERROR),
  						 errmsg("conflicting or redundant options")));
  			start_value = defel;
  		}
  		else if (strcmp(defel->defname, "restart") == 0)
  		{
  			if (restart_value)
  				ereport(ERROR,
  						(errcode(ERRCODE_SYNTAX_ERROR),
  						 errmsg("conflicting or redundant options")));
  			restart_value = defel;
  		}
  		else if (strcmp(defel->defname, "maxvalue") == 0)
  		{
  			if (max_value)
  				ereport(ERROR,
  						(errcode(ERRCODE_SYNTAX_ERROR),
  						 errmsg("conflicting or redundant options")));
  			max_value = defel;
  		}
  		else if (strcmp(defel->defname, "minvalue") == 0)
  		{
  			if (min_value)
  				ereport(ERROR,
  						(errcode(ERRCODE_SYNTAX_ERROR),
  						 errmsg("conflicting or redundant options")));
  			min_value = defel;
  		}
  		else if (strcmp(defel->defname, "cache") == 0)
  		{
  			if (cache_value)
  				ereport(ERROR,
  						(errcode(ERRCODE_SYNTAX_ERROR),
  						 errmsg("conflicting or redundant options")));
  			cache_value = defel;
  		}
  		else if (strcmp(defel->defname, "cycle") == 0)
  		{
  			if (is_cycled)
  				ereport(ERROR,
  						(errcode(ERRCODE_SYNTAX_ERROR),
  						 errmsg("conflicting or redundant options")));
  			is_cycled = defel;
  		}
  		else if (strcmp(defel->defname, "owned_by") == 0)
  		{
  			if (*owned_by)
  				ereport(ERROR,
  						(errcode(ERRCODE_SYNTAX_ERROR),
  						 errmsg("conflicting or redundant options")));
  			*owned_by = defGetQualifiedName(defel);
  		}
  		else
  			elog(ERROR, "option \"%s\" not recognized",
  				 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)
  	{
  		new->increment_by = defGetInt64(increment_by);
  		if (new->increment_by == 0)
  			ereport(ERROR,
  					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
  					 errmsg("INCREMENT must not be zero")));
  		new->log_cnt = 0;
  	}
  	else if (isInit)
  		new->increment_by = 1;
  
  	/* CYCLE */
  	if (is_cycled != NULL)
  	{
  		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 */
  	if (new->min_value >= new->max_value)
  	{
  		char		bufm[100],
  					bufx[100];
  
  		snprintf(bufm, sizeof(bufm), INT64_FORMAT, new->min_value);
  		snprintf(bufx, sizeof(bufx), INT64_FORMAT, new->max_value);
  		ereport(ERROR,
  				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
  				 errmsg("MINVALUE (%s) must be less than MAXVALUE (%s)",
  						bufm, bufx)));
  	}
  
  	/* START WITH */
  	if (start_value != NULL)
  		new->start_value = defGetInt64(start_value);
  	else if (isInit)
  	{
  		if (new->increment_by > 0)
  			new->start_value = new->min_value;	/* ascending seq */
  		else
  			new->start_value = new->max_value;	/* descending seq */
  	}
  
  	/* crosscheck START */
  	if (new->start_value < new->min_value)
  	{
  		char		bufs[100],
  					bufm[100];
  
  		snprintf(bufs, sizeof(bufs), INT64_FORMAT, new->start_value);
  		snprintf(bufm, sizeof(bufm), INT64_FORMAT, new->min_value);
  		ereport(ERROR,
  				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
  				 errmsg("START value (%s) cannot be less than MINVALUE (%s)",
  						bufs, bufm)));
  	}
  	if (new->start_value > new->max_value)
  	{
  		char		bufs[100],
  					bufm[100];
  
  		snprintf(bufs, sizeof(bufs), INT64_FORMAT, new->start_value);
  		snprintf(bufm, sizeof(bufm), INT64_FORMAT, new->max_value);
  		ereport(ERROR,
  				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
  			  errmsg("START value (%s) cannot be greater than MAXVALUE (%s)",
  					 bufs, bufm)));
  	}
  
  	/* RESTART [WITH] */
  	if (restart_value != NULL)
  	{
  		if (restart_value->arg != NULL)
  			new->last_value = defGetInt64(restart_value);
  		else
  			new->last_value = new->start_value;
  		new->is_called = false;
  		new->log_cnt = 0;
  	}
  	else if (isInit)
  	{
  		new->last_value = new->start_value;
  		new->is_called = false;
  	}
  
  	/* crosscheck RESTART (or current value, if changing MIN/MAX) */
  	if (new->last_value < new->min_value)
  	{
  		char		bufs[100],
  					bufm[100];
  
  		snprintf(bufs, sizeof(bufs), INT64_FORMAT, new->last_value);
  		snprintf(bufm, sizeof(bufm), INT64_FORMAT, new->min_value);
  		ereport(ERROR,
  				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
  			   errmsg("RESTART value (%s) cannot be less than MINVALUE (%s)",
  					  bufs, bufm)));
  	}
  	if (new->last_value > new->max_value)
  	{
  		char		bufs[100],
  					bufm[100];
  
  		snprintf(bufs, sizeof(bufs), INT64_FORMAT, new->last_value);
  		snprintf(bufm, sizeof(bufm), INT64_FORMAT, new->max_value);
  		ereport(ERROR,
  				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
  			errmsg("RESTART value (%s) cannot be greater than MAXVALUE (%s)",
  				   bufs, bufm)));
  	}
  
  	/* CACHE */
  	if (cache_value != NULL)
  	{
  		new->cache_value = defGetInt64(cache_value);
  		if (new->cache_value <= 0)
  		{
  			char		buf[100];
  
  			snprintf(buf, sizeof(buf), INT64_FORMAT, new->cache_value);
  			ereport(ERROR,
  					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
  					 errmsg("CACHE (%s) must be greater than zero",
  							buf)));
  		}
  		new->log_cnt = 0;
  	}
  	else if (isInit)
  		new->cache_value = 1;
  }
  
  /*
   * Process an OWNED BY option for CREATE/ALTER SEQUENCE
   *
   * Ownership permissions on the sequence are already checked,
   * but if we are establishing a new owned-by dependency, we must
   * enforce that the referenced table has the same owner and namespace
   * as the sequence.
   */
  static void
  process_owned_by(Relation seqrel, List *owned_by)
  {
  	int			nnames;
  	Relation	tablerel;
  	AttrNumber	attnum;
  
  	nnames = list_length(owned_by);
  	Assert(nnames > 0);
  	if (nnames == 1)
  	{
  		/* Must be OWNED BY NONE */
  		if (strcmp(strVal(linitial(owned_by)), "none") != 0)
  			ereport(ERROR,
  					(errcode(ERRCODE_SYNTAX_ERROR),
  					 errmsg("invalid OWNED BY option"),
  				errhint("Specify OWNED BY table.column or OWNED BY NONE.")));
  		tablerel = NULL;
  		attnum = 0;
  	}
  	else
  	{
  		List	   *relname;
  		char	   *attrname;
  		RangeVar   *rel;
  
  		/* Separate relname and attr name */
  		relname = list_truncate(list_copy(owned_by), nnames - 1);
  		attrname = strVal(lfirst(list_tail(owned_by)));
  
  		/* Open and lock rel to ensure it won't go away meanwhile */
  		rel = makeRangeVarFromNameList(relname);
  		tablerel = relation_openrv(rel, AccessShareLock);
  
  		/* Must be a regular or foreign table */
  		if (!(tablerel->rd_rel->relkind == RELKIND_RELATION ||
  			  tablerel->rd_rel->relkind == RELKIND_FOREIGN_TABLE))
  			ereport(ERROR,
  					(errcode(ERRCODE_WRONG_OBJECT_TYPE),
  					 errmsg("referenced relation \"%s\" is not a table or foreign table",
  							RelationGetRelationName(tablerel))));
  
  		/* We insist on same owner and schema */
  		if (seqrel->rd_rel->relowner != tablerel->rd_rel->relowner)
  			ereport(ERROR,
  					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
  					 errmsg("sequence must have same owner as table it is linked to")));
  		if (RelationGetNamespace(seqrel) != RelationGetNamespace(tablerel))
  			ereport(ERROR,
  					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
  					 errmsg("sequence must be in same schema as table it is linked to")));
  
  		/* Now, fetch the attribute number from the system cache */
  		attnum = get_attnum(RelationGetRelid(tablerel), attrname);
  		if (attnum == InvalidAttrNumber)
  			ereport(ERROR,
  					(errcode(ERRCODE_UNDEFINED_COLUMN),
  					 errmsg("column \"%s\" of relation \"%s\" does not exist",
  							attrname, RelationGetRelationName(tablerel))));
  	}
  
  	/*
  	 * OK, we are ready to update pg_depend.  First remove any existing AUTO
  	 * dependencies for the sequence, then optionally add a new one.
  	 */
  	markSequenceUnowned(RelationGetRelid(seqrel));
  
  	if (tablerel)
  	{
  		ObjectAddress refobject,
  					depobject;
  
  		refobject.classId = RelationRelationId;
  		refobject.objectId = RelationGetRelid(tablerel);
  		refobject.objectSubId = attnum;
  		depobject.classId = RelationRelationId;
  		depobject.objectId = RelationGetRelid(seqrel);
  		depobject.objectSubId = 0;
  		recordDependencyOn(&depobject, &refobject, DEPENDENCY_AUTO);
  	}
  
  	/* Done, but hold lock until commit */
  	if (tablerel)
  		relation_close(tablerel, NoLock);
  }
  
  
  /*
   * Return sequence parameters, for use by information schema
   */
  Datum
  pg_sequence_parameters(PG_FUNCTION_ARGS)
  {
  	Oid			relid = PG_GETARG_OID(0);
  	TupleDesc	tupdesc;
  	Datum		values[5];
  	bool		isnull[5];
  	SeqTable	elm;
  	Relation	seqrel;
  	Buffer		buf;
  	HeapTupleData seqtuple;
  	Form_pg_sequence seq;
  
  	/* open and AccessShareLock sequence */
  	init_sequence(relid, &elm, &seqrel);
  
  	if (pg_class_aclcheck(relid, GetUserId(), ACL_SELECT | ACL_UPDATE | ACL_USAGE) != ACLCHECK_OK)
  		ereport(ERROR,
  				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
  				 errmsg("permission denied for sequence %s",
  						RelationGetRelationName(seqrel))));
  
  	tupdesc = CreateTemplateTupleDesc(5, false);
  	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "start_value",
  					   INT8OID, -1, 0);
  	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "minimum_value",
  					   INT8OID, -1, 0);
  	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "maximum_value",
  					   INT8OID, -1, 0);
  	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "increment",
  					   INT8OID, -1, 0);
  	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "cycle_option",
  					   BOOLOID, -1, 0);
  
  	BlessTupleDesc(tupdesc);
  
  	memset(isnull, 0, sizeof(isnull));
  
  	seq = read_seq_tuple(elm, seqrel, &buf, &seqtuple);
  
  	values[0] = Int64GetDatum(seq->start_value);
  	values[1] = Int64GetDatum(seq->min_value);
  	values[2] = Int64GetDatum(seq->max_value);
  	values[3] = Int64GetDatum(seq->increment_by);
  	values[4] = BoolGetDatum(seq->is_cycled);
  
  	UnlockReleaseBuffer(buf);
  	relation_close(seqrel, NoLock);
  
  	return HeapTupleGetDatum(heap_form_tuple(tupdesc, values, isnull));
  }
  
  
  void
  seq_redo(XLogReaderState *record)
  {
  	XLogRecPtr	lsn = record->EndRecPtr;
  	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
  	Buffer		buffer;
  	Page		page;
  	Page		localpage;
  	char	   *item;
  	Size		itemsz;
  	xl_seq_rec *xlrec = (xl_seq_rec *) XLogRecGetData(record);
  	sequence_magic *sm;
  
  	if (info != XLOG_SEQ_LOG)
  		elog(PANIC, "seq_redo: unknown op code %u", info);
  
  	buffer = XLogInitBufferForRedo(record, 0);
  	page = (Page) BufferGetPage(buffer);
  
  	/*
  	 * We always reinit the page.  However, since this WAL record type is also
  	 * used for updating sequences, it's possible that a hot-standby backend
  	 * is examining the page concurrently; so we mustn't transiently trash the
  	 * buffer.  The solution is to build the correct new page contents in
  	 * local workspace and then memcpy into the buffer.  Then only bytes that
  	 * are supposed to change will change, even transiently. We must palloc
  	 * the local page for alignment reasons.
  	 */
  	localpage = (Page) palloc(BufferGetPageSize(buffer));
  
  	PageInit(localpage, BufferGetPageSize(buffer), sizeof(sequence_magic));
  	sm = (sequence_magic *) PageGetSpecialPointer(localpage);
  	sm->magic = SEQ_MAGIC;
  
  	item = (char *) xlrec + sizeof(xl_seq_rec);
  	itemsz = XLogRecGetDataLen(record) - sizeof(xl_seq_rec);
  
  	if (PageAddItem(localpage, (Item) item, itemsz,
  					FirstOffsetNumber, false, false) == InvalidOffsetNumber)
  		elog(PANIC, "seq_redo: failed to add item to page");
  
  	PageSetLSN(localpage, lsn);
  
  	memcpy(page, localpage, BufferGetPageSize(buffer));
  	MarkBufferDirty(buffer);
  	UnlockReleaseBuffer(buffer);
  
  	pfree(localpage);
  }
  
  /*
   * Flush cached sequence information.
   */
  void
  ResetSequenceCaches(void)
  {
  	if (seqhashtab)
  	{
  		hash_destroy(seqhashtab);
  		seqhashtab = NULL;
  	}
  
  	last_used_seq = NULL;
  }
  
  /*
   * Mask last_value and log_cnt for consistency checking
   *
   * To avoid logging every fetch from a sequence, SEQ_LOG_VALS are pre-logged
   * and thus we need to mask the last_value and log_cnt during consistency
   * checks.
   */
  static void
  mask_seq_values(Page page)
  {
  	OffsetNumber 		i;
  	OffsetNumber 		maxoff;
  	Form_pg_sequence	seqtup;
  
  	maxoff = PageGetMaxOffsetNumber(page);
  
  	for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
  	{
  		HeapTupleData	htup;
  		ItemId			iid = PageGetItemId(page, i);
  
  		htup.t_data = (HeapTupleHeader) ((char *) page + ItemIdGetOffset(iid));
  		htup.t_len = ItemIdGetLength(iid);
  
  		seqtup = (Form_pg_sequence) GETSTRUCT(&htup);
  		MemSet(&seqtup->last_value, 0, sizeof(int64));
  		MemSet(&seqtup->log_cnt, 0, sizeof(int64));
  	}
  }
  
  /*
   * Mask a Sequence page before performing consistency checks on it.
   */
  void
  seq_mask(char *page, BlockNumber blkno)
  {
  	mask_page_lsn_and_checksum(page);
  
  	/*
  	 * last_value and log_cnt need to be masked to account for SEQ_LOG_VALS
  	 * skipped loggings of fetching
  	 */
  	mask_seq_values(page);
  
  	mask_unused_space(page);
  }
  
  /*
   * CDB: forward a nextval request from qExec to the QD
   */
  void
  cdb_sequence_nextval_qe(Relation	seqrel,
  						int64   *plast,
  						int64   *pcached,
  						int64   *pincrement,
  						bool    *pvalid)
  {
  	Oid oid;
  	int64 last;
  	int64 cached;
  	int64 increment;
  	char overflow;
  	char error;
  	unsigned char	qtype;
  	int retval;
  	char *current;
  	int *pint32;
  	StringInfoData buf;
  	Oid dbid = seqrel->rd_node.dbNode;
  	Oid seq_oid = seqrel->rd_id;
  
  	/*
  	 * Construct a nextval NOTIFY message to send to the QD using "nextval"
  	 * channel. Sends pq_beginmessage(..., 'A') to signal that this is a NOTIFY
  	 * message. Payload includes all info required to update the sequence
  	 * value.
  	 */
  	char payload[128];
  	snprintf(payload, sizeof(payload), "%d:%d", dbid, seq_oid);
  	pq_beginmessage(&buf, 'A');
  	pq_sendint(&buf, gp_session_id, sizeof(int32));
  	pq_sendstring(&buf, "nextval"); /* channel */
  	pq_sendstring(&buf, payload);
  	pq_endmessage(&buf);
  	pq_flush();
  
  	/*
  	 * Read nextval response from QD.
  	 */
  	do
  	{
  		pq_startmsgread();
  		retval = pq_getbyte_if_available(&qtype);
  		if (retval == 0)
  		{
  			pq_endmsgread();
  			CHECK_FOR_INTERRUPTS();
  		}
  
  		if (retval == EOF)
  			ereport(ERROR,
  					(errcode(ERRCODE_INTERNAL_ERROR),
  					 errmsg("nextval: QD closed the connection")));
  	} while (retval != 1);
  	Assert(qtype == SEQ_NEXTVAL_QUERY_RESPONSE);
  
  	initStringInfo(&buf);
  	if (pq_getmessage(&buf, 0) != 0)
  		elog(ERROR, "unable to parse nextval response from QD");
  
  	current = buf.data;
  
  	oid = ntohl(*((int32 *) current));
  	current += sizeof(int32);
  
  	pint32 = (int32 *) &last;
  	*pint32 = ntohl(*((int32 *) current + 1));
  	pint32++;
  	*pint32 = ntohl(*((int32 *) current));
  	current += sizeof(int64);
  
  	pint32 = (int32 *) &cached;
  	*pint32 = ntohl(*((int32 *) current + 1));
  	pint32++;
  	*pint32 = ntohl(*((int32 *) current));
  	current += sizeof(int64);
  
  	pint32 = (int32 *) &increment;
  	*pint32 = ntohl(*((int32 *) current + 1));
  	pint32++;
  	*pint32 = ntohl(*((int32 *) current));
  	current += sizeof(int64);
  
  	overflow = *current;
  	current += sizeof(char);
  	error = *current;
  
  	if (overflow == SEQ_NEXTVAL_TRUE)
  		ereport(ERROR,
  				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
  				 errmsg("nextval: reached %s value of sequence \"%s\" (" INT64_FORMAT ")",
  				 increment>0 ? "maximum":"minimum",
  				 RelationGetRelationName(seqrel), last)));
  
  	if (error == SEQ_NEXTVAL_TRUE)
  		ereport(ERROR,
  				(errcode(ERRCODE_INTERNAL_ERROR),
  						errmsg("nextval: QD encountered error")));
  
  	Assert(overflow == SEQ_NEXTVAL_FALSE);
  	Assert(error == SEQ_NEXTVAL_FALSE);
  
  	if (oid != seq_oid)
  		ereport(ERROR,
  				(errcode(ERRCODE_INTERNAL_ERROR),
  				 errmsg("nextval: request oid:%d of QE doesn't match the response oid:%d from QD",
  						seq_oid, oid)));
  
  	*plast = last;
  	*pcached = cached;
  	*pincrement = increment;
  	*pvalid = true;
  }
  