diff --git a/src/backend/access/appendonly/appendonly_compaction.c b/src/backend/access/appendonly/appendonly_compaction.c
index 0f4fb9d55619e5a38cb4c4f32edbdb099e279a1a..69d338dcce312af8a0d1e8c1d9e9777f3a81a0e2 100644
--- a/src/backend/access/appendonly/appendonly_compaction.c
+++ b/src/backend/access/appendonly/appendonly_compaction.c
@@ -281,7 +281,7 @@ AppendOnlyMoveTuple(MemTuple tuple,
 	tupleOid = MemTupleGetOid(tuple, mt_bind);
 	appendonly_insert(insertDesc,
 					  tuple,
-					  &tupleOid,
+					  tupleOid,
 					  &newAoTupleId);
 
 	/* insert index' tuples if needed */
diff --git a/src/backend/access/appendonly/appendonlyam.c b/src/backend/access/appendonly/appendonlyam.c
index 2269081efe38777b55f53435a536b308621ff84d..f2d5194d564f2a1a493aaa871c71c01924a0ce75 100755
--- a/src/backend/access/appendonly/appendonlyam.c
+++ b/src/backend/access/appendonly/appendonlyam.c
@@ -2652,11 +2652,9 @@ appendonly_update(AppendOnlyUpdateDesc aoUpdateDesc,
 	if (result != HeapTupleMayBeUpdated)
 		return result;
 
-	Oid			newOid = InvalidOid;	/* new oid should be old oid */
-
 	appendonly_insert(aoUpdateDesc->aoInsertDesc,
 					  memTuple,
-					  &newOid,
+					  InvalidOid,	/* new oid should be old oid */
 					  newAoTupleId);
 
 	return result;
@@ -2892,11 +2890,13 @@ aoInsertDesc->appendOnlyMetaDataSnapshot, //CONCERN:Safe to assume all block dir
   * The output parameter tupleOid is the OID assigned to the tuple (either here or by the
   * caller), or InvalidOid if no OID.  The header fields of *tup are updated
   * to match the stored tuple;
+  *
+  * Unlike heap_insert(), this function doesn't scribble on the input tuple.
   */
-void
+Oid
 appendonly_insert(AppendOnlyInsertDesc aoInsertDesc,
 				  MemTuple instup,
-				  Oid *tupleOid,
+				  Oid tupleOid,
 				  AOTupleId *aoTupleId)
 {
 	Relation	relation = aoInsertDesc->aoi_rel;
@@ -2919,19 +2919,6 @@ appendonly_insert(AppendOnlyInsertDesc aoInsertDesc,
 #endif
 
 	Insist(RelationIsAoRows(relation));
-	if (relation->rd_rel->relhasoids)
-	{
-		/*
-		 * If the object id of this tuple has already been assigned, trust the
-		 * caller.	There are a couple of ways this can happen.  At initial db
-		 * creation, the backend program sets oids for tuples. When we define
-		 * an index, we set the oid.  Finally, in the future, we may allow
-		 * users to set their own object ids in order to support a persistent
-		 * object store (objects need to contain pointers to one another).
-		 */
-		if (!OidIsValid(MemTupleGetOid(instup, aoInsertDesc->mt_bind)))
-			MemTupleSetOid(instup, aoInsertDesc->mt_bind, GetNewOid(relation));
-	}
 
 	if (aoInsertDesc->useNoToast)
 		need_toast = false;
@@ -2955,6 +2942,32 @@ appendonly_insert(AppendOnlyInsertDesc aoInsertDesc,
 	else
 		tup = instup;
 
+	if (relation->rd_rel->relhasoids)
+	{
+		/*
+		 * Don't modify the input tuple, so make a copy unless we already
+		 * made one. I'm not sure if the input tuple can point to any
+		 * permanent storage, so modifying it might be harmless, but better
+		 * safe than sorry. An AO table with OIDs is a weird beast anyway,
+		 * so performance of this case isn't important.
+		 */
+		if (tup == instup)
+			tup = memtuple_copy_to(instup, NULL, NULL);
+
+		/*
+		 * If the object id of this tuple has already been assigned, trust the
+		 * caller.	There are a couple of ways this can happen.  At initial db
+		 * creation, the backend program sets oids for tuples. When we define
+		 * an index, we set the oid.  Finally, in the future, we may allow
+		 * users to set their own object ids in order to support a persistent
+		 * object store (objects need to contain pointers to one another).
+		 */
+		if (!OidIsValid(tupleOid))
+			tupleOid = GetNewOid(relation);
+
+		MemTupleSetOid(tup, aoInsertDesc->mt_bind, tupleOid);
+	}
+
 	/*
 	 * get space to insert our next item (tuple)
 	 */
@@ -3101,7 +3114,7 @@ appendonly_insert(AppendOnlyInsertDesc aoInsertDesc,
 
 	Assert(aoInsertDesc->numSequences >= 0);
 
-	*tupleOid = MemTupleGetOid(tup, aoInsertDesc->mt_bind);
+	tupleOid = MemTupleGetOid(tup, aoInsertDesc->mt_bind);
 
 	AOTupleIdInit_Init(aoTupleId);
 	AOTupleIdInit_segmentFileNum(aoTupleId, aoInsertDesc->cur_segno);
@@ -3136,6 +3149,8 @@ appendonly_insert(AppendOnlyInsertDesc aoInsertDesc,
 
 	if (tup != instup)
 		pfree(tup);
+
+	return tupleOid;
 }
 
 /*
diff --git a/src/backend/access/external/fileam.c b/src/backend/access/external/fileam.c
index 463206cb396443f8d8083fc81ad265ec658b7e0c..c2e0dd02155dc7d9d90fa38957344a87e657691a 100644
--- a/src/backend/access/external/fileam.c
+++ b/src/backend/access/external/fileam.c
@@ -584,6 +584,7 @@ external_insert_init(Relation rel)
  * - transaction information is of no interest.
  * - tuples are sent always to the destination (local file or remote target).
  *
+ * Like heap_insert(), this function can modify the input tuple!
  */
 Oid
 external_insert(ExternalInsertDesc extInsertDesc, HeapTuple instup)
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index 4e2679123d4d3d2e3a01720357a926deb1aa3aad..b880b9510ce1598bf00baf3a03e56d7a6cd25285 100644
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -4907,13 +4907,12 @@ PROCESS_SEGMENT_DATA:
 					 */
 					if (relstorage == RELSTORAGE_AOROWS)
 					{
-						Oid			tupleOid;
 						MemTuple	mtuple;
 
 						mtuple = ExecFetchSlotMemTuple(slot, false);
 
 						/* inserting into an append only relation */
-						appendonly_insert(resultRelInfo->ri_aoInsertDesc, mtuple, &tupleOid, (AOTupleId *) &insertedTid);
+						appendonly_insert(resultRelInfo->ri_aoInsertDesc, mtuple, InvalidOid, (AOTupleId *) &insertedTid);
 					}
 					else if (relstorage == RELSTORAGE_AOCOLS)
 					{
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 7e0a072a261d592876fa5f9d7da77478bec41d41..b5633247096080d0c9c0698d47028923360d8ced 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -5337,10 +5337,9 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap)
 				/* Write the tuple out to the new relation */
 				if (newrel)
 				{
-					Oid 		tupleOid;
 					AOTupleId	aoTupleId;
 					
-					appendonly_insert(aoInsertDesc, mtuple, &tupleOid, &aoTupleId);
+					appendonly_insert(aoInsertDesc, mtuple, InvalidOid, &aoTupleId);
 				}
 
 				ResetExprContext(econtext);
@@ -14322,7 +14321,6 @@ split_rows(Relation intoa, Relation intob, Relation temprel)
 		else if (RelationIsAoRows(targetRelation))
 		{
 			MemTuple	mtuple;
-			Oid			tupleOid;
 
 			if (!(*targetAODescPtr))
 			{
@@ -14333,7 +14331,7 @@ split_rows(Relation intoa, Relation intob, Relation temprel)
 			}
 
 			mtuple = ExecFetchSlotMemTuple(targetSlot, false);
-			appendonly_insert(*targetAODescPtr, mtuple, &tupleOid, &aoTupleId);
+			appendonly_insert(*targetAODescPtr, mtuple, InvalidOid, &aoTupleId);
 
 			/* cache TID for later updating of indexes */
 			tid = (ItemPointer) &aoTupleId;
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 1a8356ebbb65a5449bdcd6fb290b93a3d72cc902..0015f08b6ff5951acc6b98fd2a6eb780cdb54a6b 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -3080,6 +3080,7 @@ ExecInsert(TupleTableSlot *slot,
 	bool		rel_is_aocols = false;
 	bool		rel_is_external = false;
 	ItemPointerData lastTid;
+	Oid			tuple_oid = InvalidOid;
 
 	/*
 	 * get information on the (current) result relation
@@ -3144,6 +3145,81 @@ ExecInsert(TupleTableSlot *slot,
 	rel_is_aorows = RelationIsAoRows(resultRelationDesc);
 	rel_is_external = RelationIsExternal(resultRelationDesc);
 
+	/*
+	 * Prepare the right kind of "insert desc".
+	 */
+	if (rel_is_aorows)
+	{
+		if (resultRelInfo->ri_aoInsertDesc == NULL)
+		{
+			/* Set the pre-assigned fileseg number to insert into */
+			ResultRelInfoSetSegno(resultRelInfo, estate->es_result_aosegnos);
+
+			resultRelInfo->ri_aoInsertDesc =
+				appendonly_insert_init(resultRelationDesc,
+									   resultRelInfo->ri_aosegno,
+									   false);
+		}
+	}
+	else if (rel_is_aocols)
+	{
+		if (resultRelInfo->ri_aocsInsertDesc == NULL)
+		{
+			ResultRelInfoSetSegno(resultRelInfo, estate->es_result_aosegnos);
+			resultRelInfo->ri_aocsInsertDesc = aocs_insert_init(resultRelationDesc,
+																resultRelInfo->ri_aosegno, false);
+		}
+	}
+	else if (rel_is_external)
+	{
+		if (resultRelInfo->ri_extInsertDesc == NULL)
+			resultRelInfo->ri_extInsertDesc = external_insert_init(resultRelationDesc);
+	}
+
+	/*
+	 * If the result relation has OIDs, force the tuple's OID to zero so that
+	 * heap_insert will assign a fresh OID.  Usually the OID already will be
+	 * zero at this point, but there are corner cases where the plan tree can
+	 * return a tuple extracted literally from some table with the same
+	 * rowtype.
+	 *
+	 * XXX if we ever wanted to allow users to assign their own OIDs to new
+	 * rows, this'd be the place to do it.  For the moment, we make a point
+	 * of doing this before calling triggers, so that a user-supplied trigger
+	 * could hack the OID if desired.
+	 *
+	 * GPDB: In PostgreSQL, here we set the Oid in the HeapTuple, which is a
+	 * local copy at this point. But in GPDB, we don't materialize the tuple
+	 * yet, because we might need a MemTuple or a HeapTuple depending on
+	 * what kind of a table this is (or neither for an AOCS table, since
+	 * aocs_insert() works directly off the slot). So we keep the Oid in a
+	 * local variable for now, and only set it in the tuple just before the
+	 * call to heap/appendonly/external_insert().
+	 */
+	if (resultRelationDesc->rd_rel->relhasoids)
+	{
+		tuple_oid = InvalidOid;
+
+		/*
+		 * But if this is really an UPDATE, try to preserve the old OID.
+		 */
+		if (isUpdate)
+		{
+			GenericTuple gtuple;
+
+			gtuple = ExecFetchSlotGenericTuple(slot, false);
+
+			if (!is_memtuple(gtuple))
+				tuple_oid = HeapTupleGetOid((HeapTuple) gtuple);
+			else
+			{
+				if (resultRelInfo->ri_aoInsertDesc)
+					tuple_oid = MemTupleGetOid((MemTuple) gtuple,
+											   resultRelInfo->ri_aoInsertDesc->mt_bind);
+			}
+		}
+	}
+
 	slot = reconstructMatchingTupleSlot(slot, resultRelInfo);
 
 	if (rel_is_external &&
@@ -3165,7 +3241,9 @@ ExecInsert(TupleTableSlot *slot,
 		HeapTuple	newtuple;
 		HeapTuple	tuple;
 
-		tuple = ExecFetchSlotHeapTuple(slot);
+		tuple = ExecMaterializeSlot(slot);
+		if (resultRelationDesc->rd_rel->relhasoids)
+			HeapTupleSetOid(tuple, tuple_oid);
 
 		newtuple = ExecBRInsertTriggers(estate, resultRelInfo, tuple);
 
@@ -3188,6 +3266,13 @@ ExecInsert(TupleTableSlot *slot,
 			newslot->tts_tableOid = slot->tts_tableOid; /* for constraints */
 			slot = newslot;
 			tuple = newtuple;
+
+			/*
+			 * since we keep the OID in a separate variable, also update that,
+			 * in case the trigger set it.
+			 */
+			if (resultRelationDesc->rd_rel->relhasoids)
+				tuple_oid = HeapTupleGetOid(newtuple);
 		}
 	}
 
@@ -3218,11 +3303,10 @@ ExecInsert(TupleTableSlot *slot,
 				appendonly_insert_init(resultRelationDesc,
 									   resultRelInfo->ri_aosegno,
 									   false);
-
 		}
 
 		mtuple = ExecFetchSlotMemTuple(slot, false);
-		appendonly_insert(resultRelInfo->ri_aoInsertDesc, mtuple, &newId, (AOTupleId *) &lastTid);
+		newId = appendonly_insert(resultRelInfo->ri_aoInsertDesc, mtuple, tuple_oid, (AOTupleId *) &lastTid);
 	}
 	else if (rel_is_aocols)
 	{
@@ -3244,7 +3328,13 @@ ExecInsert(TupleTableSlot *slot,
 		if (resultRelInfo->ri_extInsertDesc == NULL)
 			resultRelInfo->ri_extInsertDesc = external_insert_init(resultRelationDesc);
 
-		tuple = ExecFetchSlotHeapTuple(slot);
+		/*
+		 * get the heap tuple out of the tuple table slot, making sure we have a
+		 * writable copy. (external_insert() can scribble on the tuple)
+		 */
+		tuple = ExecMaterializeSlot(slot);
+		if (resultRelationDesc->rd_rel->relhasoids)
+			HeapTupleSetOid(tuple, tuple_oid);
 
 		newId = external_insert(resultRelInfo->ri_extInsertDesc, tuple);
 		ItemPointerSetInvalid(&lastTid);
@@ -3255,7 +3345,13 @@ ExecInsert(TupleTableSlot *slot,
 
 		Insist(rel_is_heap);
 
-		tuple = ExecFetchSlotHeapTuple(slot);
+		/*
+		 * get the heap tuple out of the tuple table slot, making sure we have a
+		 * writable copy. (heap_insert() will scribble on the tuple)
+		 */
+		tuple = ExecMaterializeSlot(slot);
+		if (resultRelationDesc->rd_rel->relhasoids)
+			HeapTupleSetOid(tuple, tuple_oid);
 
 		newId = heap_insert(resultRelationDesc,
 							tuple,
@@ -3282,7 +3378,7 @@ ExecInsert(TupleTableSlot *slot,
 		resultRelInfo->ri_TrigDesc->n_after_row[TRIGGER_EVENT_INSERT] > 0 &&
 		!isUpdate)
 	{
-		HeapTuple tuple = ExecFetchSlotHeapTuple(slot);
+		HeapTuple tuple = ExecMaterializeSlot(slot);
 
 		Assert(planGen == PLANGEN_PLANNER);
 
@@ -3821,7 +3917,7 @@ lreplace:;
 		{
 			HeapTuple tuple;
 
-			tuple = ExecFetchSlotHeapTuple(slot);
+			tuple = ExecMaterializeSlot(slot);
 
 			result = heap_update(resultRelationDesc, tupleid, tuple,
 							 &update_ctid, &update_xmax,
@@ -3925,13 +4021,11 @@ lreplace:;
 		/*
 		 * Persistent metadata path.
 		 */
-		persistentTuple = ExecCopySlotHeapTuple(slot);
+		persistentTuple = ExecMaterializeSlot(slot);
 		persistentTuple->t_self = *tupleid;
 
 		frozen_heap_inplace_update(resultRelationDesc, persistentTuple);
 		wasHotUpdate = false;
-
-		heap_freetuple(persistentTuple);
 	}
 
 	(estate->es_processed)++;
@@ -3961,7 +4055,7 @@ lreplace:;
 		resultRelInfo->ri_TrigDesc->n_after_row[TRIGGER_EVENT_UPDATE] > 0 &&
 		rel_is_heap)
 	{
-		HeapTuple tuple = ExecFetchSlotHeapTuple(slot);
+		HeapTuple tuple = ExecMaterializeSlot(slot);
 
 		ExecARUpdateTriggers(estate, resultRelInfo, tupleid, tuple);
 	}
@@ -5043,13 +5137,12 @@ intorel_receive(TupleTableSlot *slot, DestReceiver *self)
 	if (RelationIsAoRows(into_rel))
 	{
 		MemTuple	tuple = ExecCopySlotMemTuple(slot);
-		Oid			tupleOid;
 		AOTupleId	aoTupleId;
 
 		if (myState->ao_insertDesc == NULL)
 			myState->ao_insertDesc = appendonly_insert_init(into_rel, RESERVED_SEGNO, false);
 
-		appendonly_insert(myState->ao_insertDesc, tuple, &tupleOid, &aoTupleId);
+		appendonly_insert(myState->ao_insertDesc, tuple, InvalidOid, &aoTupleId);
 		pfree(tuple);
 	}
 	else if (RelationIsAoCols(into_rel))
@@ -5061,7 +5154,19 @@ intorel_receive(TupleTableSlot *slot, DestReceiver *self)
 	}
 	else
 	{
-		HeapTuple	tuple = ExecCopySlotHeapTuple(slot);
+		HeapTuple	tuple;
+
+		/*
+		 * get the heap tuple out of the tuple table slot, making sure we have a
+		 * writable copy
+		 */
+		tuple = ExecMaterializeSlot(slot);
+
+		/*
+		 * force assignment of new OID (see comments in ExecInsert)
+		 */
+		if (into_rel->rd_rel->relhasoids)
+			HeapTupleSetOid(tuple, InvalidOid);
 
 		heap_insert(into_rel,
 					tuple,
@@ -5071,8 +5176,6 @@ intorel_receive(TupleTableSlot *slot, DestReceiver *self)
 					GetCurrentTransactionId());
 
 		estate->es_into_relation_last_heap_tid = tuple->t_self;
-
-		heap_freetuple(tuple);
 	}
 
 	/* We know this is a newly created relation, so there are no indexes */
diff --git a/src/backend/executor/execTuples.c b/src/backend/executor/execTuples.c
index 84c0f18c65725a65438fdb10fdb68c8e47decb44..85401830395822ddc326de9e22ca0241622b49d1 100644
--- a/src/backend/executor/execTuples.c
+++ b/src/backend/executor/execTuples.c
@@ -703,9 +703,6 @@ ExecCopySlotMemTupleTo(TupleTableSlot *slot, MemoryContext pctxt, char *dest, un
 HeapTuple
 ExecFetchSlotHeapTuple(TupleTableSlot *slot)
 {
-	uint32 tuplen;
-	HeapTuple htup;
-
 	/*
 	 * sanity checks
 	 */
@@ -717,29 +714,10 @@ ExecFetchSlotHeapTuple(TupleTableSlot *slot)
 	if(slot->PRIVATE_tts_heaptuple)
 		return slot->PRIVATE_tts_heaptuple;
 
-	slot_getallattrs(slot);
-
-	Assert(TupHasVirtualTuple(slot));
-	Assert(slot->PRIVATE_tts_nvalid == slot->tts_tupleDescriptor->natts);
-
-	tuplen = slot->PRIVATE_tts_htup_buf_len;
-	htup = heaptuple_form_to(slot->tts_tupleDescriptor, slot_get_values(slot), slot_get_isnull(slot),
-			slot->PRIVATE_tts_htup_buf, &tuplen);
-	
-	if(!htup)
-	{
-		if(slot->PRIVATE_tts_htup_buf)
-			pfree(slot->PRIVATE_tts_htup_buf);
-		slot->PRIVATE_tts_htup_buf = (HeapTuple) MemoryContextAlloc(slot->tts_mcxt, tuplen);
-		slot->PRIVATE_tts_htup_buf_len = tuplen;
-
-		htup = heaptuple_form_to(slot->tts_tupleDescriptor, slot_get_values(slot), slot_get_isnull(slot),
-			slot->PRIVATE_tts_htup_buf, &tuplen);
-		Assert(htup);
-	}
-
-	slot->PRIVATE_tts_heaptuple = htup;
-	return htup;
+	/*
+	 * Otherwise materialize the slot...
+	 */
+	return ExecMaterializeSlot(slot);
 }
 
 /* --------------------------------
@@ -802,6 +780,69 @@ ExecFetchSlotMemTuple(TupleTableSlot *slot, bool inline_toast)
 	return newTuple;
 }
 
+/* --------------------------------
+ *		ExecMaterializeSlot
+ *			Force a slot into the "materialized" state.
+ *
+ *		This causes the slot's tuple to be a local copy not dependent on
+ *		any external storage.  A pointer to the contained tuple is returned.
+ *
+ *		A typical use for this operation is to prepare a computed tuple
+ *		for being stored on disk.  The original data may or may not be
+ *		virtual, but in any case we need a private copy for heap_insert
+ *		to scribble on.
+ * --------------------------------
+ */
+HeapTuple
+ExecMaterializeSlot(TupleTableSlot *slot)
+{
+	uint32		tuplen;
+	HeapTuple	htup;
+
+	/*
+	 * sanity checks
+	 */
+	Assert(!TupIsNull(slot));
+
+	/*
+	 * If we have a regular physical tuple, and it's locally palloc'd, we have
+	 * nothing to do.
+	 */
+	if (slot->PRIVATE_tts_heaptuple &&
+		(TupShouldFree(slot) || slot->PRIVATE_tts_heaptuple == slot->PRIVATE_tts_htup_buf))
+		return slot->PRIVATE_tts_heaptuple;
+
+	/*
+	 * Otherwise, copy or build a physical tuple, and store it into the slot.
+	 */
+	slot_getallattrs(slot);
+
+	Assert(slot->PRIVATE_tts_nvalid == slot->tts_tupleDescriptor->natts);
+
+	tuplen = slot->PRIVATE_tts_htup_buf_len;
+	htup = heaptuple_form_to(slot->tts_tupleDescriptor, slot_get_values(slot), slot_get_isnull(slot),
+			slot->PRIVATE_tts_htup_buf, &tuplen);
+
+	if (!htup)
+	{
+		/* enlarge the buffer and retry */
+		if (slot->PRIVATE_tts_htup_buf)
+			pfree(slot->PRIVATE_tts_htup_buf);
+		slot->PRIVATE_tts_htup_buf = (HeapTuple) MemoryContextAlloc(slot->tts_mcxt, tuplen);
+		slot->PRIVATE_tts_htup_buf_len = tuplen;
+
+		htup = heaptuple_form_to(slot->tts_tupleDescriptor,
+								 slot_get_values(slot),
+								 slot_get_isnull(slot),
+								 slot->PRIVATE_tts_htup_buf,
+								 &tuplen);
+		Assert(htup);
+	}
+
+	slot->PRIVATE_tts_heaptuple = htup;
+	return htup;
+}
+
 /* --------------------------------
  *		ExecCopySlot
  *			Copy the source slot's contents into the destination slot.
diff --git a/src/include/cdb/cdbappendonlyam.h b/src/include/cdb/cdbappendonlyam.h
index 70e09c083080f7bd6d5461ad6f87bbabba2817a1..8d9a4c9fe667d2d8dc4ef1b14739804e5098f354 100644
--- a/src/include/cdb/cdbappendonlyam.h
+++ b/src/include/cdb/cdbappendonlyam.h
@@ -351,10 +351,10 @@ extern bool appendonly_fetch(
 	TupleTableSlot *slot);
 extern void appendonly_fetch_finish(AppendOnlyFetchDesc aoFetchDesc);
 extern AppendOnlyInsertDesc appendonly_insert_init(Relation rel, int segno, bool update_mode);
-extern void appendonly_insert(
+extern Oid appendonly_insert(
 		AppendOnlyInsertDesc aoInsertDesc, 
 		MemTuple instup, 
-		Oid *tupleOid, 
+		Oid tupleOid,
 		AOTupleId *aoTupleId);
 extern void appendonly_insert_finish(AppendOnlyInsertDesc aoInsertDesc);
 extern BlockNumber RelationGuessNumberOfBlocks(double totalbytes);
diff --git a/src/include/executor/tuptable.h b/src/include/executor/tuptable.h
index ebc087b0974b78267d1244b108c9fa6223fe8763..d3c6b32f5e90fd59681f5a013c6e0ac1f53f88da 100644
--- a/src/include/executor/tuptable.h
+++ b/src/include/executor/tuptable.h
@@ -429,6 +429,7 @@ ExecCopyGenericTuple(TupleTableSlot *slot)
 	return (GenericTuple) ExecCopySlotMemTuple(slot);
 }
 
+extern HeapTuple ExecMaterializeSlot(TupleTableSlot *slot);
 extern TupleTableSlot *ExecCopySlot(TupleTableSlot *dstslot, TupleTableSlot *srcslot);
 
 extern void ExecModifyMemTuple(TupleTableSlot *slot, Datum *values, bool *isnull, bool *doRepl);
diff --git a/src/test/regress/expected/bfv_dml.out b/src/test/regress/expected/bfv_dml.out
index c66be7cfc9ca4fc61a1c8c11cf0197fd50d8574c..df8e4545a4b3698c1fc8f5c3eaf4678c68885f03 100644
--- a/src/test/regress/expected/bfv_dml.out
+++ b/src/test/regress/expected/bfv_dml.out
@@ -250,3 +250,39 @@ EXECUTE PROCEDURE bfv_dml_error_func();
 UPDATE bfv_dml_trigger_test SET t = 'bar';
 UPDATE bfv_dml_trigger_test SET id = id + 1;
 ERROR:  Cannot parallelize an UPDATE statement that updates the distribution columns
+--
+-- Verify that ExecInsert doesn't scribble on the old tuple, when the new
+-- tuple comes directly from the old table.
+--
+CREATE TABLE execinsert_test (id int4, t text) DISTRIBUTED BY (id);
+INSERT INTO execinsert_test values (1, 'foo');
+-- Insert another identical tuple, but roll it back. If the insertion
+-- incorrectly modified the xmin on the old tuple, then it will become
+-- invisible when we roll back.
+begin;
+INSERT INTO execinsert_test select * FROM execinsert_test;
+rollback;
+select * from execinsert_test;
+ id |  t  
+----+-----
+  1 | foo
+(1 row)
+
+drop table execinsert_test;
+-- Repeat with a hacked master-only table, just in case the planner decides
+-- to add a Motion node or something that hides the problem otherwise.
+CREATE TABLE execinsert_test (id int4, t text) DISTRIBUTED BY (id);
+set allow_system_table_mods='DML';
+delete from gp_distribution_policy where localoid='execinsert_test'::regclass;
+reset allow_system_table_mods;
+INSERT INTO execinsert_test values (1, 'foo');
+begin;
+INSERT INTO execinsert_test select * FROM execinsert_test;
+rollback;
+select * from execinsert_test;
+ id |  t  
+----+-----
+  1 | foo
+(1 row)
+
+drop table execinsert_test;
diff --git a/src/test/regress/expected/bfv_dml_optimizer.out b/src/test/regress/expected/bfv_dml_optimizer.out
index 153e745613c41c7f64d8bd6ef12998ba6bae008e..b8cb1aa60673520d34849cbbae388c7215628f20 100644
--- a/src/test/regress/expected/bfv_dml_optimizer.out
+++ b/src/test/regress/expected/bfv_dml_optimizer.out
@@ -267,3 +267,39 @@ FOR EACH ROW
 EXECUTE PROCEDURE bfv_dml_error_func();
 UPDATE bfv_dml_trigger_test SET t = 'bar';
 UPDATE bfv_dml_trigger_test SET id = id + 1;
+--
+-- Verify that ExecInsert doesn't scribble on the old tuple, when the new
+-- tuple comes directly from the old table.
+--
+CREATE TABLE execinsert_test (id int4, t text) DISTRIBUTED BY (id);
+INSERT INTO execinsert_test values (1, 'foo');
+-- Insert another identical tuple, but roll it back. If the insertion
+-- incorrectly modified the xmin on the old tuple, then it will become
+-- invisible when we roll back.
+begin;
+INSERT INTO execinsert_test select * FROM execinsert_test;
+rollback;
+select * from execinsert_test;
+ id |  t  
+----+-----
+  1 | foo
+(1 row)
+
+drop table execinsert_test;
+-- Repeat with a hacked master-only table, just in case the planner decides
+-- to add a Motion node or something that hides the problem otherwise.
+CREATE TABLE execinsert_test (id int4, t text) DISTRIBUTED BY (id);
+set allow_system_table_mods='DML';
+delete from gp_distribution_policy where localoid='execinsert_test'::regclass;
+reset allow_system_table_mods;
+INSERT INTO execinsert_test values (1, 'foo');
+begin;
+INSERT INTO execinsert_test select * FROM execinsert_test;
+rollback;
+select * from execinsert_test;
+ id |  t  
+----+-----
+  1 | foo
+(1 row)
+
+drop table execinsert_test;
diff --git a/src/test/regress/expected/qp_dml_oids.out b/src/test/regress/expected/qp_dml_oids.out
index 51b77b09d30ecf390ecb60d307f0ca67744196e4..a2e398eed93bdf38a59a747db0bc03efb72253bc 100644
--- a/src/test/regress/expected/qp_dml_oids.out
+++ b/src/test/regress/expected/qp_dml_oids.out
@@ -298,3 +298,16 @@ SELECT COUNT(distinct oid) FROM dml_ao where a = 10;
      2
 (1 row)
 
+--
+-- Check that new OIDs are generated even if the tuple being inserted came from
+-- the same relation and segment.
+--
+INSERT INTO dml_ao VALUES (11, 1, 'foo');
+INSERT INTO dml_ao VALUES (11, 2, 'bar');
+INSERT INTO dml_ao SELECT * FROM dml_ao WHERE a = 11 LIMIT 1;
+SELECT COUNT(DISTINCT oid) FROM dml_ao WHERE a = 11; -- all three rows should have different OID
+ count 
+-------
+     3
+(1 row)
+
diff --git a/src/test/regress/expected/qp_dml_oids_optimizer.out b/src/test/regress/expected/qp_dml_oids_optimizer.out
index 1997fb46490a133e194cecafe0e444a52df8b34b..6caf76c8510dd317f4dddc9903805c1187b3470b 100644
--- a/src/test/regress/expected/qp_dml_oids_optimizer.out
+++ b/src/test/regress/expected/qp_dml_oids_optimizer.out
@@ -296,3 +296,16 @@ SELECT COUNT(distinct oid) FROM dml_ao where a = 10;
      2
 (1 row)
 
+--
+-- Check that new OIDs are generated even if the tuple being inserted came from
+-- the same relation and segment.
+--
+INSERT INTO dml_ao VALUES (11, 1, 'foo');
+INSERT INTO dml_ao VALUES (11, 2, 'bar');
+INSERT INTO dml_ao SELECT * FROM dml_ao WHERE a = 11 LIMIT 1;
+SELECT COUNT(DISTINCT oid) FROM dml_ao WHERE a = 11; -- all three rows should have different OID
+ count 
+-------
+     3
+(1 row)
+
diff --git a/src/test/regress/greenplum_schedule b/src/test/regress/greenplum_schedule
index e954e2673569de7f785fe9ebfddaa74ac472da37..25c2d4816fbeb57600bb793707420f4714c6a047 100755
--- a/src/test/regress/greenplum_schedule
+++ b/src/test/regress/greenplum_schedule
@@ -113,7 +113,7 @@ test: tuple_serialization
 # temp tables
 test: bfv_cte bfv_joins bfv_subquery bfv_planner bfv_legacy bfv_temp bfv_dml
 
-test: qp_olap_mdqa qp_misc gp_recursive_cte qp_dml_joins qp_dml_oids
+test: qp_olap_mdqa qp_misc gp_recursive_cte qp_dml_joins qp_dml_oids trigger_sets_oid
 
 test: qp_misc_jiras qp_with_clause qp_executor qp_olap_windowerr qp_olap_window qp_derived_table qp_bitmapscan session_level_memory_consumption qp_dropped_cols
 test: qp_with_functional_inlining qp_with_functional_noinlining
diff --git a/src/test/regress/input/trigger_sets_oid.source b/src/test/regress/input/trigger_sets_oid.source
new file mode 100644
index 0000000000000000000000000000000000000000..1b5972110432091e0362737bcdcfa44b8c7a4e1f
--- /dev/null
+++ b/src/test/regress/input/trigger_sets_oid.source
@@ -0,0 +1,35 @@
+--
+-- Test setting the OID of a tuple in a BEFORE ROW INSERT trigger.
+--
+-- There's a comment in ExecInsert that says that that should work, although there
+-- is no code in PostgreSQL to test that. So let's test it.
+--
+
+CREATE OR REPLACE FUNCTION trigger_return_new_oid()
+RETURNS trigger
+LANGUAGE C VOLATILE EXECUTE ON MASTER AS '@abs_builddir@/regress@DLSUFFIX@', 'trigger_udf_return_new_oid';
+
+
+-- Heap table
+CREATE TABLE trigger_oid_tab(id int8, v text) WITH (oids = true);
+
+CREATE TRIGGER t BEFORE INSERT ON trigger_oid_tab
+FOR EACH ROW EXECUTE PROCEDURE trigger_return_new_oid(12345);
+
+INSERT INTO trigger_oid_tab values (1, 'foo');
+SELECT oid, * FROM trigger_oid_tab;
+
+DROP TABLE trigger_oid_tab;
+
+-- AO table
+CREATE TABLE trigger_oid_tab(id int8, v text) WITH (oids = true, appendonly = true);
+
+CREATE TRIGGER t BEFORE INSERT ON trigger_oid_tab
+FOR EACH ROW EXECUTE PROCEDURE trigger_return_new_oid(12345);
+
+INSERT INTO trigger_oid_tab values (1, 'foo');
+SELECT oid, * FROM trigger_oid_tab;
+
+DROP TABLE trigger_oid_tab;
+
+-- Column-oriented AO tables don't support OIDs, so no need to test that.
diff --git a/src/test/regress/output/trigger_sets_oid.source b/src/test/regress/output/trigger_sets_oid.source
new file mode 100644
index 0000000000000000000000000000000000000000..5d6472082bc7b5586ff157adb14e8094bee30e8f
--- /dev/null
+++ b/src/test/regress/output/trigger_sets_oid.source
@@ -0,0 +1,42 @@
+--
+-- Test setting the OID of a tuple in a BEFORE ROW INSERT trigger.
+--
+-- There's a comment in ExecInsert that says that that should work, although there
+-- is no code in PostgreSQL to test that. So let's test it.
+--
+CREATE OR REPLACE FUNCTION trigger_return_new_oid()
+RETURNS trigger
+LANGUAGE C VOLATILE EXECUTE ON MASTER AS '@abs_builddir@/regress@DLSUFFIX@', 'trigger_udf_return_new_oid';
+-- Heap table
+CREATE TABLE trigger_oid_tab(id int8, v text) WITH (oids = true);
+NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause -- Using column named 'id' as the Greenplum Database data distribution key for this table.
+HINT:  The 'DISTRIBUTED BY' clause determines the distribution of data. Make sure column(s) chosen are the optimal data distribution key to minimize skew.
+NOTICE:  OIDS=TRUE is not recommended for user-created tables. Use OIDS=FALSE to prevent wrap-around of the OID counter
+CREATE TRIGGER t BEFORE INSERT ON trigger_oid_tab
+FOR EACH ROW EXECUTE PROCEDURE trigger_return_new_oid(12345);
+INSERT INTO trigger_oid_tab values (1, 'foo');
+NOTICE:  trigger_udf_return_new_oid assigned OID 12345 to the new tuple  (seg0 127.0.0.1:40000 pid=25965)
+SELECT oid, * FROM trigger_oid_tab;
+  oid  | id |  v  
+-------+----+-----
+ 12345 |  1 | foo
+(1 row)
+
+DROP TABLE trigger_oid_tab;
+-- AO table
+CREATE TABLE trigger_oid_tab(id int8, v text) WITH (oids = true, appendonly = true);
+NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause -- Using column named 'id' as the Greenplum Database data distribution key for this table.
+HINT:  The 'DISTRIBUTED BY' clause determines the distribution of data. Make sure column(s) chosen are the optimal data distribution key to minimize skew.
+NOTICE:  OIDS=TRUE is not recommended for user-created tables. Use OIDS=FALSE to prevent wrap-around of the OID counter
+CREATE TRIGGER t BEFORE INSERT ON trigger_oid_tab
+FOR EACH ROW EXECUTE PROCEDURE trigger_return_new_oid(12345);
+INSERT INTO trigger_oid_tab values (1, 'foo');
+NOTICE:  trigger_udf_return_new_oid assigned OID 12345 to the new tuple  (seg0 127.0.0.1:40000 pid=25965)
+SELECT oid, * FROM trigger_oid_tab;
+  oid  | id |  v  
+-------+----+-----
+ 12345 |  1 | foo
+(1 row)
+
+DROP TABLE trigger_oid_tab;
+-- Column-oriented AO tables don't support OIDs, so no need to test that.
diff --git a/src/test/regress/regress_gp.c b/src/test/regress/regress_gp.c
index fa2fb08dc7eacd1f77e744ecef47cc0d30e1a211..803247ae865ae6ca2c477b69acc4996f851aebee 100644
--- a/src/test/regress/regress_gp.c
+++ b/src/test/regress/regress_gp.c
@@ -108,6 +108,7 @@ extern Datum check_foreign_key(PG_FUNCTION_ARGS);
 extern Datum autoinc(PG_FUNCTION_ARGS);
 static EPlan *find_plan(char *ident, EPlan ** eplan, int *nplans);
 
+extern Datum trigger_udf_return_new_oid(PG_FUNCTION_ARGS);
 
 
 PG_FUNCTION_INFO_V1(multiset_scalar_null);
@@ -1903,3 +1904,51 @@ find_plan(char *ident, EPlan ** eplan, int *nplans)
 
 	return (newp);
 }
+
+
+/*
+ * trigger_udf_return_new_oid
+ *
+ * A helper function to assign a specific OID to a tuple on INSERT.
+ */
+PG_FUNCTION_INFO_V1(trigger_udf_return_new_oid);
+Datum
+trigger_udf_return_new_oid(PG_FUNCTION_ARGS)
+{
+	TriggerData *trigdata = (TriggerData *) fcinfo->context;
+	Trigger    *trigger;
+	char	  **args;
+	HeapTuple	input_tuple;
+	HeapTuple	ret_tuple;
+	Oid			new_oid;
+
+	if (!CALLED_AS_TRIGGER(fcinfo))
+		elog(ERROR, "not fired by trigger manager");
+	if (!TRIGGER_FIRED_FOR_ROW(trigdata->tg_event))
+		elog(ERROR, "cannot process STATEMENT events");
+	if (!TRIGGER_FIRED_BEFORE(trigdata->tg_event))
+		elog(ERROR, "must be fired before event");
+
+	if (!TRIGGER_FIRED_BY_INSERT(trigdata->tg_event))
+		elog(ERROR, "trigger_udf_return_new_oid() called for a non-INSERT");
+
+	/*
+	 * Get the argument. (Trigger functions receive their arguments
+	 * differently than normal functions.)
+	 */
+	trigger = trigdata->tg_trigger;
+	if (trigger->tgnargs != 1)
+		elog(ERROR, "trigger_udf_return_new_oid called with invalid number of arguments (%d, expected 1)",
+			 trigger->tgnargs);
+	args = trigger->tgargs;
+	new_oid = DatumGetObjectId(DirectFunctionCall1(oidin,
+												   CStringGetDatum(args[0])));
+
+	elog(NOTICE, "trigger_udf_return_new_oid assigned OID %u to the new tuple", new_oid);
+
+	input_tuple = trigdata->tg_trigtuple;
+	ret_tuple = heap_copytuple(input_tuple);
+	HeapTupleSetOid(ret_tuple, new_oid);
+
+	return PointerGetDatum(ret_tuple);
+}
diff --git a/src/test/regress/sql/bfv_dml.sql b/src/test/regress/sql/bfv_dml.sql
index 9b2bb78e317ef453971191409fbf8c072e9923d0..33bd1924823531ef62eec5c355b5766dd5bdda05 100644
--- a/src/test/regress/sql/bfv_dml.sql
+++ b/src/test/regress/sql/bfv_dml.sql
@@ -164,3 +164,38 @@ EXECUTE PROCEDURE bfv_dml_error_func();
 
 UPDATE bfv_dml_trigger_test SET t = 'bar';
 UPDATE bfv_dml_trigger_test SET id = id + 1;
+
+
+--
+-- Verify that ExecInsert doesn't scribble on the old tuple, when the new
+-- tuple comes directly from the old table.
+--
+CREATE TABLE execinsert_test (id int4, t text) DISTRIBUTED BY (id);
+
+INSERT INTO execinsert_test values (1, 'foo');
+
+-- Insert another identical tuple, but roll it back. If the insertion
+-- incorrectly modified the xmin on the old tuple, then it will become
+-- invisible when we roll back.
+begin;
+INSERT INTO execinsert_test select * FROM execinsert_test;
+rollback;
+select * from execinsert_test;
+
+drop table execinsert_test;
+
+-- Repeat with a hacked master-only table, just in case the planner decides
+-- to add a Motion node or something that hides the problem otherwise.
+
+CREATE TABLE execinsert_test (id int4, t text) DISTRIBUTED BY (id);
+set allow_system_table_mods='DML';
+delete from gp_distribution_policy where localoid='execinsert_test'::regclass;
+reset allow_system_table_mods;
+
+INSERT INTO execinsert_test values (1, 'foo');
+begin;
+INSERT INTO execinsert_test select * FROM execinsert_test;
+rollback;
+select * from execinsert_test;
+
+drop table execinsert_test;
diff --git a/src/test/regress/sql/qp_dml_oids.sql b/src/test/regress/sql/qp_dml_oids.sql
index 4c2958866cc044083716a1fa17f2e872e14fe819..81c55a6c182146c92f2aa017e72b2d2e01b6a5d3 100644
--- a/src/test/regress/sql/qp_dml_oids.sql
+++ b/src/test/regress/sql/qp_dml_oids.sql
@@ -162,3 +162,12 @@ INSERT INTO dml_ao (a, b, c) VALUES (10, 1, repeat('x', 50000));
 INSERT INTO dml_ao (a, b, c) VALUES (10, 2, repeat('x', 50000));
 
 SELECT COUNT(distinct oid) FROM dml_ao where a = 10;
+
+--
+-- Check that new OIDs are generated even if the tuple being inserted came from
+-- the same relation and segment.
+--
+INSERT INTO dml_ao VALUES (11, 1, 'foo');
+INSERT INTO dml_ao VALUES (11, 2, 'bar');
+INSERT INTO dml_ao SELECT * FROM dml_ao WHERE a = 11 LIMIT 1;
+SELECT COUNT(DISTINCT oid) FROM dml_ao WHERE a = 11; -- all three rows should have different OID