/*
 * Guts of language creation.
 */
static void
create_proc_lang(const char *languageName,
				 Oid handlerOid, Oid valOid, bool trusted)
{
	Relation	rel;
	TupleDesc	tupDesc;
	Datum		values[Natts_pg_language];
	char		nulls[Natts_pg_language];
	NameData	langname;
	HeapTuple	tup;
	ObjectAddress myself,
				referenced;

	/*
	 * Insert the new language into pg_language
	 */
	rel = heap_open(LanguageRelationId, RowExclusiveLock);
	tupDesc = rel->rd_att;

	memset(values, 0, sizeof(values));
	memset(nulls, ' ', sizeof(nulls));

	namestrcpy(&langname, languageName);
	values[Anum_pg_language_lanname - 1] = NameGetDatum(&langname);
	values[Anum_pg_language_lanispl - 1] = BoolGetDatum(true);
	values[Anum_pg_language_lanpltrusted - 1] = BoolGetDatum(trusted);
	values[Anum_pg_language_lanplcallfoid - 1] = ObjectIdGetDatum(handlerOid);
	values[Anum_pg_language_lanvalidator - 1] = ObjectIdGetDatum(valOid);
	nulls[Anum_pg_language_lanacl - 1] = 'n';

	tup = heap_formtuple(tupDesc, values, nulls);

	simple_heap_insert(rel, tup);

	CatalogUpdateIndexes(rel, tup);

	/*
	 * Create dependencies for language
	 */
	myself.classId = LanguageRelationId;
	myself.objectId = HeapTupleGetOid(tup);
	myself.objectSubId = 0;

	/* dependency on the PL handler function */
	referenced.classId = ProcedureRelationId;
	referenced.objectId = handlerOid;
	referenced.objectSubId = 0;
	recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);

	/* dependency on the validator function, if any */
	if (OidIsValid(valOid))
	{
		referenced.classId = ProcedureRelationId;
		referenced.objectId = valOid;
		referenced.objectSubId = 0;
		recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
	}

	heap_close(rel, RowExclusiveLock);
}

/*
 * LogTransactionRecord registers the fact that a transaction has been
 * prepared on a worker. The presence of this record indicates that the
 * prepared transaction should be committed.
 */
void
LogTransactionRecord(int groupId, char *transactionName)
{
	Relation pgDistTransaction = NULL;
	TupleDesc tupleDescriptor = NULL;
	HeapTuple heapTuple = NULL;
	Datum values[Natts_pg_dist_transaction];
	bool isNulls[Natts_pg_dist_transaction];

	/* form new transaction tuple */
	memset(values, 0, sizeof(values));
	memset(isNulls, false, sizeof(isNulls));

	values[Anum_pg_dist_transaction_groupid - 1] = Int32GetDatum(groupId);
	values[Anum_pg_dist_transaction_gid - 1] = CStringGetTextDatum(transactionName);

	/* open transaction relation and insert new tuple */
	pgDistTransaction = heap_open(DistTransactionRelationId(), RowExclusiveLock);

	tupleDescriptor = RelationGetDescr(pgDistTransaction);
	heapTuple = heap_form_tuple(tupleDescriptor, values, isNulls);

	simple_heap_insert(pgDistTransaction, heapTuple);
	CatalogUpdateIndexes(pgDistTransaction, heapTuple);
	CommandCounterIncrement();

	/* close relation and invalidate previous cache entry */
	heap_close(pgDistTransaction, RowExclusiveLock);
}

/* ----------------------------------------------------------------
 *		UpdateRelationRelation
 * ----------------------------------------------------------------
 */
static void
UpdateRelationRelation(Relation indexRelation)
{
	Relation	pg_class;
	HeapTuple	tuple;

	pg_class = heap_openr(RelationRelationName, RowExclusiveLock);

	/* XXX Natts_pg_class_fixed is a hack - see pg_class.h */
	tuple = heap_addheader(Natts_pg_class_fixed,
						   true,
						   CLASS_TUPLE_SIZE,
						   (void *) indexRelation->rd_rel);

	/*
	 * the new tuple must have the oid already chosen for the index. sure
	 * would be embarrassing to do this sort of thing in polite company.
	 */
	HeapTupleSetOid(tuple, RelationGetRelid(indexRelation));
	simple_heap_insert(pg_class, tuple);

	/* update the system catalog indexes */
	CatalogUpdateIndexes(pg_class, tuple);

	heap_freetuple(tuple);
	heap_close(pg_class, RowExclusiveLock);
}

/*
 * InsertAgLabelTuple - register the new label in ag_label
 *
 * See InsertPgClassTuple()
 */
static void
InsertAgLabelTuple(Relation ag_label_desc, Oid laboid, RangeVar *label,
				   Oid relid, char labkind)
{
	Oid			graphid = get_graphname_oid(label->schemaname);
	char	   *labname = label->relname;
	int32		labid;
	Datum		values[Natts_ag_label];
	bool		nulls[Natts_ag_label];
	HeapTuple	tup;

	AssertArg(labkind == LABEL_KIND_VERTEX || labkind == LABEL_KIND_EDGE);

	labid = (int32) GetNewLabelId(label->schemaname, graphid);

	values[Anum_ag_label_labname - 1] = CStringGetDatum(labname);
	values[Anum_ag_label_graphid - 1] = CStringGetDatum(graphid);
	values[Anum_ag_label_labid - 1] = Int32GetDatum(labid);
	values[Anum_ag_label_relid - 1] = ObjectIdGetDatum(relid);
	values[Anum_ag_label_labkind - 1] = CharGetDatum(labkind);

	memset(nulls, false, sizeof(nulls));

	tup = heap_form_tuple(RelationGetDescr(ag_label_desc), values, nulls);

	HeapTupleSetOid(tup, laboid);

	simple_heap_insert(ag_label_desc, tup);

	CatalogUpdateIndexes(ag_label_desc, tup);

	heap_freetuple(tup);
}

/* ---------------------
 * addProcCallback() - Add a new callback to pg_proc_callback
 *
 * Parameters:
 *    profnoid    - oid of the function that has a callback
 *    procallback - oid of the callback function
 *    promethod   - role the callback function plays
 *
 * Notes:
 *    This function does not maintain dependencies in pg_depend, that behavior
 *    is currently controlled in pg_proc.c
 * ---------------------
 */
void 
addProcCallback(Oid profnoid, Oid procallback, char promethod)
{
	Relation	rel;
	bool		nulls[Natts_pg_proc_callback];
	Datum		values[Natts_pg_proc_callback];
	HeapTuple   tup;
	
	Insist(OidIsValid(profnoid));
	Insist(OidIsValid(procallback));

	/* open pg_proc_callback */
	rel = heap_open(ProcCallbackRelationId, RowExclusiveLock);

	/* Build the tuple and insert it */
	nulls[Anum_pg_proc_callback_profnoid - 1]	  = false;
	nulls[Anum_pg_proc_callback_procallback - 1]  = false;
	nulls[Anum_pg_proc_callback_promethod - 1]	  = false;
	values[Anum_pg_proc_callback_profnoid - 1]	  = ObjectIdGetDatum(profnoid);
	values[Anum_pg_proc_callback_procallback - 1] = ObjectIdGetDatum(procallback);
	values[Anum_pg_proc_callback_promethod - 1]	  = CharGetDatum(promethod);

	tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
	
	/* Insert tuple into the relation */
	simple_heap_insert(rel, tup);
	CatalogUpdateIndexes(rel, tup);

	heap_close(rel, RowExclusiveLock);
}

/*
 * Sets the policy of a table into the gp_distribution_policy table
 * from a GpPolicy structure.
 */
void
GpPolicyStore(Oid tbloid, const GpPolicy *policy)
{
	Relation	gp_policy_rel;
	HeapTuple	gp_policy_tuple = NULL;

	ArrayType  *attrnums;

	bool		nulls[2];
	Datum		values[2];

	Insist(policy->ptype == POLICYTYPE_PARTITIONED);

    /*
     * Open and lock the gp_distribution_policy catalog.
     */
	gp_policy_rel = heap_open(GpPolicyRelationId, RowExclusiveLock);

	/*
	 * Convert C arrays into Postgres arrays.
	 */
	if (policy->nattrs > 0)
	{
		int			i;
		Datum	   *akey;

		akey = (Datum *) palloc(policy->nattrs * sizeof(Datum));
		for (i = 0; i < policy->nattrs; i++)
			akey[i] = Int16GetDatum(policy->attrs[i]);
		attrnums = construct_array(akey, policy->nattrs,
								   INT2OID, 2, true, 's');
	}
	else
	{
		attrnums = NULL;
	}

	nulls[0] = false;
	nulls[1] = false;
	values[0] = ObjectIdGetDatum(tbloid);

	if (attrnums)
		values[1] = PointerGetDatum(attrnums);
	else
		nulls[1] = true;

	gp_policy_tuple = heap_form_tuple(RelationGetDescr(gp_policy_rel), values, nulls);

	/* Insert tuple into the relation */
	simple_heap_insert(gp_policy_rel, gp_policy_tuple);
	CatalogUpdateIndexes(gp_policy_rel, gp_policy_tuple);

	/*
     * Close the gp_distribution_policy relcache entry without unlocking.
     * We have updated the catalog: consequently the lock must be held until
     * end of transaction.
     */
    heap_close(gp_policy_rel, NoLock);
}                               /* GpPolicyStore */

/*
 * CatalogTupleInsertWithInfo - as above, but with caller-supplied index info
 *
 * This should be used when it's important to amortize CatalogOpenIndexes/
 * CatalogCloseIndexes work across multiple insertions.  At some point we
 * might cache the CatalogIndexState data somewhere (perhaps in the relcache)
 * so that callers needn't trouble over this ... but we don't do so today.
 */
void
CatalogTupleInsertWithInfo(Relation heapRel, HeapTuple tup,
						   CatalogIndexState indstate)
{
	simple_heap_insert(heapRel, tup);

	CatalogIndexInsert(indstate, tup);
}

/*
 * Insert tuple represented in the slot to the relation, update the indexes,
 * and execute any constraints and per-row triggers.
 *
 * Caller is responsible for opening the indexes.
 */
void
ExecSimpleRelationInsert(EState *estate, TupleTableSlot *slot)
{
	bool		skip_tuple = false;
	HeapTuple	tuple;
	ResultRelInfo *resultRelInfo = estate->es_result_relation_info;
	Relation	rel = resultRelInfo->ri_RelationDesc;

	/* For now we support only tables. */
	Assert(rel->rd_rel->relkind == RELKIND_RELATION);

	CheckCmdReplicaIdentity(rel, CMD_INSERT);

	/* BEFORE ROW INSERT Triggers */
	if (resultRelInfo->ri_TrigDesc &&
		resultRelInfo->ri_TrigDesc->trig_insert_before_row)
	{
		slot = ExecBRInsertTriggers(estate, resultRelInfo, slot);

		if (slot == NULL)		/* "do nothing" */
			skip_tuple = true;
	}

	if (!skip_tuple)
	{
		List	   *recheckIndexes = NIL;

		/* Check the constraints of the tuple */
		if (rel->rd_att->constr)
			ExecConstraints(resultRelInfo, slot, estate);
		if (resultRelInfo->ri_PartitionCheck)
			ExecPartitionCheck(resultRelInfo, slot, estate, true);

		/* Materialize slot into a tuple that we can scribble upon. */
		tuple = ExecFetchSlotHeapTuple(slot, true, NULL);

		/* OK, store the tuple and create index entries for it */
		simple_heap_insert(rel, tuple);

		if (resultRelInfo->ri_NumIndices > 0)
			recheckIndexes = ExecInsertIndexTuples(slot, &(tuple->t_self),
												   estate, false, NULL,
												   NIL);

		/* AFTER ROW INSERT Triggers */
		ExecARInsertTriggers(estate, resultRelInfo, tuple,
							 recheckIndexes, NULL);

		/*
		 * XXX we should in theory pass a TransitionCaptureState object to the
		 * above to capture transition tuples, but after statement triggers
		 * don't actually get fired by replication yet anyway
		 */

		list_free(recheckIndexes);
	}
}

/*
 * PgxcClassCreate
 *		Create a pgxc_class entry
 */
void
PgxcClassCreate(Oid pcrelid,
				char pclocatortype,
				int pcattnum,
				int pchashalgorithm,
				int pchashbuckets,
				int numnodes,
				Oid *nodes)
{
	Relation	pgxcclassrel;
	HeapTuple	htup;
	bool		nulls[Natts_pgxc_class];
	Datum		values[Natts_pgxc_class];
	int		i;
	oidvector	*nodes_array;

	/* Build array of Oids to be inserted */
	nodes_array = buildoidvector(nodes, numnodes);

	/* Iterate through attributes initializing nulls and values */
	for (i = 0; i < Natts_pgxc_class; i++)
	{
		nulls[i]  = false;
		values[i] = (Datum) 0;
	}

	/* should not happen */
	if (pcrelid == InvalidOid)
	{
		elog(ERROR,"pgxc class relid invalid.");
		return;
	}

	values[Anum_pgxc_class_pcrelid - 1]   = ObjectIdGetDatum(pcrelid);
	values[Anum_pgxc_class_pclocatortype - 1] = CharGetDatum(pclocatortype);

	if (pclocatortype == LOCATOR_TYPE_HASH || pclocatortype == LOCATOR_TYPE_MODULO)
	{
		values[Anum_pgxc_class_pcattnum - 1] = UInt16GetDatum(pcattnum);
		values[Anum_pgxc_class_pchashalgorithm - 1] = UInt16GetDatum(pchashalgorithm);
		values[Anum_pgxc_class_pchashbuckets - 1] = UInt16GetDatum(pchashbuckets);
	}

	/* Node information */
	values[Anum_pgxc_class_nodes - 1] = PointerGetDatum(nodes_array);

	/* Open the relation for insertion */
	pgxcclassrel = heap_open(PgxcClassRelationId, RowExclusiveLock);

	htup = heap_form_tuple(pgxcclassrel->rd_att, values, nulls);

	(void) simple_heap_insert(pgxcclassrel, htup);

	CatalogUpdateIndexes(pgxcclassrel, htup);

	heap_close(pgxcclassrel, RowExclusiveLock);
}

/* ----------------
 * NamespaceCreate
 * ---------------
 */
Oid
NamespaceCreate(const char *nspName, Oid ownerId, Oid forceOid)
{
	Relation	nspdesc;
	HeapTuple	tup;
	Oid			nspoid;
	bool		nulls[Natts_pg_namespace];
	Datum		values[Natts_pg_namespace];
	NameData	nname;
	TupleDesc	tupDesc;
	int			i;

	/* sanity checks */
	if (!nspName)
		elog(ERROR, "no namespace name supplied");

	/* make sure there is no existing namespace of same name */
	if (SearchSysCacheExists(NAMESPACENAME,
							 PointerGetDatum(nspName),
							 0, 0, 0))
		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_SCHEMA),
				 errmsg("schema \"%s\" already exists", nspName)));

	/* initialize nulls and values */
	for (i = 0; i < Natts_pg_namespace; i++)
	{
		nulls[i] = false;
		values[i] = (Datum) 0;
	}
	namestrcpy(&nname, nspName);
	values[Anum_pg_namespace_nspname - 1] = NameGetDatum(&nname);
	values[Anum_pg_namespace_nspowner - 1] = ObjectIdGetDatum(ownerId);
	nulls[Anum_pg_namespace_nspacl - 1] = true;

	nspdesc = heap_open(NamespaceRelationId, RowExclusiveLock);
	tupDesc = nspdesc->rd_att;

	tup = heap_form_tuple(tupDesc, values, nulls);

	if (forceOid != InvalidOid)
		HeapTupleSetOid(tup, forceOid);		/* override heap_insert's OID
											 * selection */

	nspoid = simple_heap_insert(nspdesc, tup);
	Assert(OidIsValid(nspoid));

	CatalogUpdateIndexes(nspdesc, tup);

	heap_close(nspdesc, RowExclusiveLock);

	/* Record dependency on owner */
	recordDependencyOnOwner(NamespaceRelationId, nspoid, ownerId);

	return nspoid;
}

/*
 * Stores the visibility map entry.
 *
 * The entry/tuple is invalidated after this function call.
 *
 * Assumes that a valid visimap entry is passed.
 * Assumes that the entry corresponds to the latest tuple
 * returned by AppendOnlyVisimapStore_find.
 *
 * Should not be called twice in the same command.
 */ 
void
AppendOnlyVisimapStore_Store(
		AppendOnlyVisimapStore* visiMapStore,
		AppendOnlyVisimapEntry* visiMapEntry)
{
	MemoryContext oldContext;
	Relation visimapRelation; 
	TupleDesc heapTupleDesc;
	HeapTuple tuple;
	Datum values[Natts_pg_aovisimap];
	bool nulls[Natts_pg_aovisimap];

	Assert(visiMapStore);
	Assert(visiMapEntry);

	elogif (Debug_appendonly_print_visimap, LOG, 
			"Append-only visi map store: Store visimap entry: "
			"(segFileNum, firstRowNum) = (%u, " INT64_FORMAT ")",
			visiMapEntry->segmentFileNum, visiMapEntry->firstRowNum);

	oldContext = MemoryContextSwitchTo(visiMapStore->memoryContext);

	AppendOnlyVisimapEntry_Write(visiMapEntry, values,
			nulls);

	visimapRelation = visiMapStore->visimapRelation;
	heapTupleDesc = RelationGetDescr(visimapRelation);
	tuple = heap_form_tuple(heapTupleDesc,
							  values,
							  nulls);

	/*
	 * Write out the visimap entry to the relation.
	 * If this visimap entry already in the relation, we update
	 * the row. Otherwise, a new row is inserted.
	 */
	if (ItemPointerIsValid(&visiMapEntry->tupleTid))
	{
		simple_heap_update(visimapRelation, &visiMapEntry->tupleTid, tuple);
	}
	else
	{
		simple_heap_insert(visimapRelation, tuple);
	}

	CatalogUpdateIndexes(visimapRelation, tuple);
	
	heap_freetuple(tuple);
	
	MemoryContextSwitchTo(oldContext);

	// Invalidate the data after storing it.
	ItemPointerSetInvalid(&visiMapEntry->tupleTid);	
}

/*
 * CatalogTupleInsert - do heap and indexing work for a new catalog tuple
 *
 * Insert the tuple data in "tup" into the specified catalog relation.
 * The Oid of the inserted tuple is returned.
 *
 * This is a convenience routine for the common case of inserting a single
 * tuple in a system catalog; it inserts a new heap tuple, keeping indexes
 * current.  Avoid using it for multiple tuples, since opening the indexes
 * and building the index info structures is moderately expensive.
 * (Use CatalogTupleInsertWithInfo in such cases.)
 */
void
CatalogTupleInsert(Relation heapRel, HeapTuple tup)
{
	CatalogIndexState indstate;

	indstate = CatalogOpenIndexes(heapRel);

	simple_heap_insert(heapRel, tup);

	CatalogIndexInsert(indstate, tup);
	CatalogCloseIndexes(indstate);
}

/* AddUpdResqueueCapabilityEntryInternal:
 *
 * Internal function to add a new entry to pg_resqueuecapability, or
 * update an existing one.  Key cols are queueid, restypint.  If
 * old_tuple is set (ie not InvalidOid), the update the ressetting column,
 * else insert a new row.
 *
 */
static void
AddUpdResqueueCapabilityEntryInternal(
								  Relation		 resqueuecap_rel,
								  Oid			 queueid,
								  int			 resTypeInt,
								  char			*pResSetting,
								  Relation		 rel,
								  HeapTuple		 old_tuple)
{
	HeapTuple	new_tuple;
	Datum		values[Natts_pg_resqueuecapability];
	bool		isnull[Natts_pg_resqueuecapability];
	bool		new_record_repl[Natts_pg_resqueuecapability];

	MemSet(isnull, 0, sizeof(bool) * Natts_pg_resqueuecapability);
	MemSet(new_record_repl, 0, sizeof(bool) * Natts_pg_resqueuecapability);

	values[Anum_pg_resqueuecapability_resqueueid - 1]  = 
			ObjectIdGetDatum(queueid);
	values[Anum_pg_resqueuecapability_restypid - 1]   = resTypeInt;

	Assert(pResSetting);

	values[Anum_pg_resqueuecapability_ressetting - 1] = 
			CStringGetTextDatum(pResSetting);
	/* set this column to update */
	new_record_repl[Anum_pg_resqueuecapability_ressetting - 1]  = true;

	ValidateResqueueCapabilityEntry(resTypeInt, pResSetting);

	if (HeapTupleIsValid(old_tuple))
	{
		new_tuple = heap_modify_tuple(old_tuple,
									  RelationGetDescr(resqueuecap_rel),
									  values, isnull, new_record_repl);

		simple_heap_update(resqueuecap_rel, &old_tuple->t_self, new_tuple);
		CatalogUpdateIndexes(resqueuecap_rel, new_tuple);
	}
	else
	{
		new_tuple = heap_form_tuple(RelationGetDescr(resqueuecap_rel), values, isnull);

		simple_heap_insert(resqueuecap_rel, new_tuple);
		CatalogUpdateIndexes(resqueuecap_rel, new_tuple);
	}

	if (HeapTupleIsValid(old_tuple))
		heap_freetuple(new_tuple);
} /* end AddUpdResqueueCapabilityEntryInternal */

/*
 * gp_fastsequence is used to generate and keep track of row numbers for AO
 * and CO tables. Row numbers for AO/CO tables act as a component to form TID,
 * stored in index tuples and used during index scans to lookup intended
 * tuple. Hence this number must be monotonically incrementing value. Also
 * should not rollback irrespective of insert/update transaction aborting for
 * AO/CO table, as reusing row numbers even across aborted transactions would
 * yield wrong results for index scans. Also, entries in gp_fastsequence must
 * only exist for lifespan of the corresponding table.
 *
 * Given those special needs, this function inserts 2 initial rows to
 * fastsequence for segfile 0 (used for special cases like CTAS and ALTER) and
 * segfile 1. Only segfile 0 or segfile 1 can be used to insert tuples within
 * same transaction creating the table hence initial entry is only created for
 * these. Entries for rest of segfiles will get created with frozenXids during
 * inserts. These entries are inserted while creating the AO/CO table to
 * leverage MVCC to clear out gp_fastsequence entries incase of
 * aborts/failures. All future calls to insert_or_update_fastsequence() for
 * objmod 0 or objmod 1 will perform inplace updates to these tuples.
 */
void
InsertInitialFastSequenceEntries(Oid objid)
{
	Relation gp_fastsequence_rel;
	TupleDesc tupleDesc;
	Datum *values;
	bool *nulls;
	HeapTuple tuple = NULL;

	/*
	 * Open and lock the gp_fastsequence catalog table.
	 */
	gp_fastsequence_rel = heap_open(FastSequenceRelationId, RowExclusiveLock);
	tupleDesc = RelationGetDescr(gp_fastsequence_rel);

	values = palloc0(sizeof(Datum) * tupleDesc->natts);
	nulls = palloc0(sizeof(bool) * tupleDesc->natts);

	values[Anum_gp_fastsequence_objid - 1] = ObjectIdGetDatum(objid);
	values[Anum_gp_fastsequence_last_sequence - 1] = Int64GetDatum(0);

	/* Insert enrty for segfile 0 */
	values[Anum_gp_fastsequence_objmod - 1] = Int64GetDatum(RESERVED_SEGNO);
	tuple = heaptuple_form_to(tupleDesc, values, nulls, NULL, NULL);
	simple_heap_insert(gp_fastsequence_rel, tuple);
	CatalogUpdateIndexes(gp_fastsequence_rel, tuple);
	heap_freetuple(tuple);

	/* Insert entry for segfile 1 */
	values[Anum_gp_fastsequence_objmod - 1] = Int64GetDatum(1);
	tuple = heaptuple_form_to(tupleDesc, values, nulls, NULL, NULL);
	simple_heap_insert(gp_fastsequence_rel, tuple);
	CatalogUpdateIndexes(gp_fastsequence_rel, tuple);
	heap_freetuple(tuple);

	heap_close(gp_fastsequence_rel, RowExclusiveLock);
}

/*
 * Insert tuple represented in the slot to the relation, update the indexes,
 * and execute any constraints and per-row triggers.
 *
 * Caller is responsible for opening the indexes.
 */
void
ExecSimpleRelationInsert(EState *estate, TupleTableSlot *slot)
{
	bool		skip_tuple = false;
	HeapTuple	tuple;
	ResultRelInfo *resultRelInfo = estate->es_result_relation_info;
	Relation	rel = resultRelInfo->ri_RelationDesc;

	/* For now we support only tables. */
	Assert(rel->rd_rel->relkind == RELKIND_RELATION);

	CheckCmdReplicaIdentity(rel, CMD_INSERT);

	/* BEFORE ROW INSERT Triggers */
	if (resultRelInfo->ri_TrigDesc &&
		resultRelInfo->ri_TrigDesc->trig_insert_before_row)
	{
		slot = ExecBRInsertTriggers(estate, resultRelInfo, slot);

		if (slot == NULL)		/* "do nothing" */
			skip_tuple = true;
	}

	if (!skip_tuple)
	{
		List	   *recheckIndexes = NIL;

		/* Check the constraints of the tuple */
		if (rel->rd_att->constr)
			ExecConstraints(resultRelInfo, slot, estate);

		/* Store the slot into tuple that we can inspect. */
		tuple = ExecMaterializeSlot(slot);

		/* OK, store the tuple and create index entries for it */
		simple_heap_insert(rel, tuple);

		if (resultRelInfo->ri_NumIndices > 0)
			recheckIndexes = ExecInsertIndexTuples(slot, &(tuple->t_self),
												   estate, false, NULL,
												   NIL);

		/* AFTER ROW INSERT Triggers */
		ExecARInsertTriggers(estate, resultRelInfo, tuple,
							 recheckIndexes);

		list_free(recheckIndexes);
	}
}

/* ----------------------------------------------------------------
 * caql_insert()
 * during beginscan/endscan iteration, insert a tuple
 * NOTE: a separate call to CatalogUpdateIndexes after this will 
 * cause an error
 * ----------------------------------------------------------------
 */
Oid caql_insert(cqContext *pCtx, HeapTuple tup)
{
	Relation		 rel;
	Oid				 result;

	rel  = pCtx->cq_heap_rel;
	Assert(RelationIsValid(rel));

	result = simple_heap_insert(rel, tup);

	/* keep the catalog indexes up to date (if has any) */
	caql_UpdateIndexes(pCtx, rel, tup);

	return (result);
}

/* ----------------------------------------------------------------
 *		AppendAttributeTuples
 * ----------------------------------------------------------------
 */
static void
AppendAttributeTuples(Relation indexRelation, int numatts)
{
	Relation	pg_attribute;
	CatalogIndexState indstate;
	TupleDesc	indexTupDesc;
	HeapTuple	new_tuple;
	int			i;

	/*
	 * open the attribute relation and its indexes
	 */
	pg_attribute = heap_openr(AttributeRelationName, RowExclusiveLock);

	indstate = CatalogOpenIndexes(pg_attribute);

	/*
	 * insert data from new index's tupdesc into pg_attribute
	 */
	indexTupDesc = RelationGetDescr(indexRelation);

	for (i = 0; i < numatts; i++)
	{
		/*
		 * There used to be very grotty code here to set these fields, but
		 * I think it's unnecessary.  They should be set already.
		 */
		Assert(indexTupDesc->attrs[i]->attnum == i + 1);
		Assert(indexTupDesc->attrs[i]->attcacheoff == -1);

		new_tuple = heap_addheader(Natts_pg_attribute,
								   false,
								   ATTRIBUTE_TUPLE_SIZE,
								   (void *) indexTupDesc->attrs[i]);

		simple_heap_insert(pg_attribute, new_tuple);

		CatalogIndexInsert(indstate, new_tuple);

		heap_freetuple(new_tuple);
	}

	CatalogCloseIndexes(indstate);

	heap_close(pg_attribute, RowExclusiveLock);
}

/* ----------------------------------------------------------------
 * caql_insert()
 * during beginscan/endscan iteration, insert a tuple
 * NOTE: a separate call to CatalogUpdateIndexes after this will
 * cause an error
 * ----------------------------------------------------------------
 */
Oid
caql_insert(cqContext *pCtx, HeapTuple tup)
{
	Relation		 rel;
	Oid				 result;

	rel  = pCtx->cq_heap_rel;
	Assert(RelationIsValid(rel));

	disable_catalog_check(pCtx, tup);

	{
		caql_iud_switch(pCtx, 1, NULL, tup, true /* dontWait */);
		result = simple_heap_insert(rel, tup);

		/* keep the catalog indexes up to date (if has any) */
		caql_UpdateIndexes(pCtx, rel, tup);
	}
	return (result);
}

/*
 * CreatePipelineDatabaseCatalogEntry
 */
void
CreatePipelineDatabaseCatalogEntry(Oid dbid)
{
	Relation pipeline_database = heap_open(PipelineDatabaseRelationId, RowExclusiveLock);
	Datum values[Natts_pipeline_database];
	bool nulls[Natts_pipeline_database];
	HeapTuple tup;

	MemSet(nulls, 0, sizeof(nulls));

	values[Anum_pipeline_database_dbid - 1] = ObjectIdGetDatum(dbid);
	values[Anum_pipeline_database_cq_enabled - 1] = BoolGetDatum(continuous_queries_enabled);

	tup = heap_form_tuple(pipeline_database->rd_att, values, nulls);
	simple_heap_insert(pipeline_database, tup);
	CatalogUpdateIndexes(pipeline_database, tup);

	CommandCounterIncrement();

	heap_close(pipeline_database, NoLock);
}

/*
 * Create a large object having the given LO identifier.
 *
 * We create a new large object by inserting an entry into
 * pg_largeobject_metadata without any data pages, so that the object
 * will appear to exist with size 0.
 */
Oid
LargeObjectCreate(Oid loid)
{
	Relation	pg_lo_meta;
	HeapTuple	ntup;
	Oid			loid_new;
	Datum		values[Natts_pg_largeobject_metadata];
	bool		nulls[Natts_pg_largeobject_metadata];

	pg_lo_meta = heap_open(LargeObjectMetadataRelationId,
						   RowExclusiveLock);

	/*
	 * Insert metadata of the largeobject
	 */
	memset(values, 0, sizeof(values));
	memset(nulls, false, sizeof(nulls));

	values[Anum_pg_largeobject_metadata_lomowner - 1]
		= ObjectIdGetDatum(GetUserId());
	nulls[Anum_pg_largeobject_metadata_lomacl - 1] = true;

	ntup = heap_form_tuple(RelationGetDescr(pg_lo_meta),
						   values, nulls);
	if (OidIsValid(loid))
		HeapTupleSetOid(ntup, loid);

	loid_new = simple_heap_insert(pg_lo_meta, ntup);
	Assert(!OidIsValid(loid) || loid == loid_new);

	CatalogUpdateIndexes(pg_lo_meta, ntup);

	heap_freetuple(ntup);

	heap_close(pg_lo_meta, RowExclusiveLock);

	return loid_new;
}