VCI - changes to postgres core

Author: Peter Smith

src/backend/access/heap/heapam.c

@@ -54,6 +54,7 @@
 #include "utils/spccache.h"
 #include "utils/syscache.h"
 
+void		(*add_index_delete_hook) (Relation indexRelation, ItemPointer heap_tid, TransactionId xmin) = NULL;
 
 static HeapTuple heap_prepare_insert(Relation relation, HeapTuple tup,
 									 TransactionId xid, CommandId cid, int options);
@@ -364,6 +365,9 @@ initscan(HeapScanDesc scan, ScanKey key, bool keep_startblock)
 	 * results for a non-MVCC snapshot, the caller must hold some higher-level
 	 * lock that ensures the interesting tuple(s) won't change.)
 	 */
+	if (keep_startblock)
+		goto skip_get_number_of_blocks;
+
 	if (scan->rs_base.rs_parallel != NULL)
 	{
 		bpscan = (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel;
@@ -372,6 +376,8 @@ initscan(HeapScanDesc scan, ScanKey key, bool keep_startblock)
 	else
 		scan->rs_nblocks = RelationGetNumberOfBlocks(scan->rs_base.rs_rd);
 
+skip_get_number_of_blocks:
+
 	/*
 	 * If the table is large relative to NBuffers, use a bulk-read access
 	 * strategy and enable synchronized scanning (see syncscan.c).  Although
@@ -2806,6 +2812,7 @@ heap_delete(Relation relation, ItemPointer tid,
 	bool		all_visible_cleared = false;
 	HeapTuple	old_key_tuple = NULL;	/* replica identity of the tuple */
 	bool		old_key_copied = false;
+	TransactionId old_xmin;
 
 	Assert(ItemPointerIsValid(tid));
 
@@ -3052,6 +3059,8 @@ heap_delete(Relation relation, ItemPointer tid,
 							  xid, LockTupleExclusive, true,
 							  &new_xmax, &new_infomask, &new_infomask2);
 
+	old_xmin = HeapTupleHeaderGetXmin(tp.t_data);
+
 	START_CRIT_SECTION();
 
 	/*
@@ -3197,6 +3206,9 @@ heap_delete(Relation relation, ItemPointer tid,
 	if (old_key_tuple != NULL && old_key_copied)
 		heap_freetuple(old_key_tuple);
 
+	if (add_index_delete_hook)
+		add_index_delete_hook(relation, tid, old_xmin);
+
 	return TM_Ok;
 }
 
@@ -3299,6 +3311,7 @@ heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
 				infomask2_old_tuple,
 				infomask_new_tuple,
 				infomask2_new_tuple;
+	TransactionId old_xmin;
 
 	Assert(ItemPointerIsValid(otid));
 
@@ -3745,6 +3758,8 @@ heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
 							  &xmax_old_tuple, &infomask_old_tuple,
 							  &infomask2_old_tuple);
 
+	old_xmin = HeapTupleHeaderGetRawXmin(oldtup.t_data);
+
 	/*
 	 * And also prepare an Xmax value for the new copy of the tuple.  If there
 	 * was no xmax previously, or there was one but all lockers are now gone,
@@ -4228,6 +4243,9 @@ heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
 	bms_free(modified_attrs);
 	bms_free(interesting_attrs);
 
+	if (add_index_delete_hook)
+		add_index_delete_hook(relation, otid, old_xmin);
+
 	return TM_Ok;
 }
 

src/backend/access/heap/heapam_handler.c

@@ -46,6 +46,9 @@
 #include "utils/builtins.h"
 #include "utils/rel.h"
 
+/* Preserve the original heap tuple that is passed to callback in heapam_index_build_range_scan() */
+HeapTuple	IndexHeapTuple;
+
 static void reform_and_rewrite_tuple(HeapTuple tuple,
 									 Relation OldHeap, Relation NewHeap,
 									 Datum *values, bool *isnull, RewriteState rwstate);
@@ -1658,6 +1661,7 @@ heapam_index_build_range_scan(Relation heapRelation,
 		 * some index AMs want to do further processing on the data first.  So
 		 * pass the values[] and isnull[] arrays, instead.
 		 */
+		IndexHeapTuple = heapTuple;
 
 		if (HeapTupleIsHeapOnly(heapTuple))
 		{

src/backend/access/heap/heapam_visibility.c

@@ -78,6 +78,8 @@
 #include "utils/builtins.h"
 #include "utils/snapmgr.h"
 
+bool		(*add_snapshot_satisfies_hook) (HeapTuple tup, Snapshot snapshot, Buffer buffer);
+
 
 /*
  * SetHintBits()
@@ -1791,6 +1793,10 @@ HeapTupleSatisfiesVisibility(HeapTuple htup, Snapshot snapshot, Buffer buffer)
 			return HeapTupleSatisfiesHistoricMVCC(htup, snapshot, buffer);
 		case SNAPSHOT_NON_VACUUMABLE:
 			return HeapTupleSatisfiesNonVacuumable(htup, snapshot, buffer);
+		case SNAPSHOT_VCI_WOS2ROS:
+		case SNAPSHOT_VCI_LOCALROS:
+			if (add_snapshot_satisfies_hook)
+				return add_snapshot_satisfies_hook(htup, snapshot, buffer);
 	}
 
 	return false;				/* keep compiler quiet */

src/backend/access/transam/xlogfuncs.c

@@ -607,6 +607,7 @@ pg_get_wal_replay_pause_state(PG_FUNCTION_ARGS)
 			statestr = "not paused";
 			break;
 		case RECOVERY_PAUSE_REQUESTED:
+		case RECOVERY_VCI_PAUSE_REQUESTED:
 			statestr = "pause requested";
 			break;
 		case RECOVERY_PAUSED:

src/backend/access/transam/xlogrecovery.c

@@ -2949,7 +2949,8 @@ recoveryPausesHere(bool endOfRecovery)
 		ereport(LOG,
 				(errmsg("pausing at the end of recovery"),
 				 errhint("Execute pg_wal_replay_resume() to promote.")));
-	else
+	/* If pause requested by VCI, the log is not output. */
+	else if (GetRecoveryPauseState() != RECOVERY_VCI_PAUSE_REQUESTED)
 		ereport(LOG,
 				(errmsg("recovery has paused"),
 				 errhint("Execute pg_wal_replay_resume() to continue.")));
@@ -3115,6 +3116,18 @@ SetRecoveryPause(bool recoveryPause)
 		ConditionVariableBroadcast(&XLogRecoveryCtl->recoveryNotPausedCV);
 }
 
+/* Set the recovery pause requested for VCI. */
+void
+SetVciRecoveryPause(void)
+{
+	SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+
+	if (XLogRecoveryCtl->recoveryPauseState == RECOVERY_NOT_PAUSED)
+		XLogRecoveryCtl->recoveryPauseState = RECOVERY_VCI_PAUSE_REQUESTED;
+
+	SpinLockRelease(&XLogRecoveryCtl->info_lck);
+}
+
 /*
  * Confirm the recovery pause by setting the recovery pause state to
  * RECOVERY_PAUSED.
@@ -3124,7 +3137,8 @@ ConfirmRecoveryPaused(void)
 {
 	/* If recovery pause is requested then set it paused */
 	SpinLockAcquire(&XLogRecoveryCtl->info_lck);
-	if (XLogRecoveryCtl->recoveryPauseState == RECOVERY_PAUSE_REQUESTED)
+	if (XLogRecoveryCtl->recoveryPauseState == RECOVERY_PAUSE_REQUESTED ||
+		XLogRecoveryCtl->recoveryPauseState == RECOVERY_VCI_PAUSE_REQUESTED)
 		XLogRecoveryCtl->recoveryPauseState = RECOVERY_PAUSED;
 	SpinLockRelease(&XLogRecoveryCtl->info_lck);
 }

src/backend/catalog/dependency.c

@@ -85,6 +85,7 @@
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"
 
+bool		(*add_drop_relation_hook) (const ObjectAddress *object, int flags) = NULL;
 
 /*
  * Deletion processing requires additional state for each ObjectAddress that
@@ -1357,6 +1358,12 @@ doDeletion(const ObjectAddress *object, int flags)
 			{
 				char		relKind = get_rel_relkind(object->objectId);
 
+				if (add_drop_relation_hook)
+				{
+					if (add_drop_relation_hook(object, flags))
+						break;
+				}
+
 				if (relKind == RELKIND_INDEX ||
 					relKind == RELKIND_PARTITIONED_INDEX)
 				{

src/backend/catalog/index.c

@@ -133,6 +133,7 @@ static void ResetReindexProcessing(void);
 static void SetReindexPending(List *indexes);
 static void RemoveReindexPending(Oid indexOid);
 
+bool		(*add_reindex_index_hook) (Relation) = NULL;
 
 /*
  * relationHasPrimaryKey
@@ -342,12 +343,17 @@ ConstructTupleDescriptor(Relation heapRelation,
 			/* Simple index column */
 			const FormData_pg_attribute *from;
 
-			Assert(atnum > 0);	/* should've been caught above */
-
 			if (atnum > natts)	/* safety check */
 				elog(ERROR, "invalid column number %d", atnum);
-			from = TupleDescAttr(heapTupDesc,
-								 AttrNumberGetAttrOffset(atnum));
+			if (atnum > 0)
+			{
+				from = TupleDescAttr(heapTupDesc,
+									 AttrNumberGetAttrOffset(atnum));
+			}
+			else
+			{
+				from = SystemAttributeDefinition(atnum);
+			}
 
 			to->atttypid = from->atttypid;
 			to->attlen = from->attlen;
@@ -3692,6 +3698,25 @@ reindex_index(const ReindexStmt *stmt, Oid indexId,
 		return;
 	}
 
+	if (add_reindex_index_hook)
+	{
+		if (!add_reindex_index_hook(iRel))
+		{
+			RemoveReindexPending(RelationGetRelid(iRel));
+
+			/* Roll back any GUC changes */
+			AtEOXact_GUC(false, save_nestlevel);
+
+			/* Restore userid and security context */
+			SetUserIdAndSecContext(save_userid, save_sec_context);
+
+			/* Close rels, but keep locks */
+			index_close(iRel, NoLock);
+			table_close(heapRelation, NoLock);
+			return;
+		}
+	}
+
 	if (progress)
 		pgstat_progress_update_param(PROGRESS_CREATEIDX_ACCESS_METHOD_OID,
 									 iRel->rd_rel->relam);

src/backend/commands/explain.c

@@ -1208,6 +1208,7 @@ ExplainPreScanNode(PlanState *planstate, Bitmapset **rels_used)
 										 ((ForeignScan *) plan)->fs_base_relids);
 			break;
 		case T_CustomScan:
+		case T_CustomPlanMarkPos:
 			*rels_used = bms_add_members(*rels_used,
 										 ((CustomScan *) plan)->custom_relids);
 			break;
@@ -1511,6 +1512,7 @@ ExplainNode(PlanState *planstate, List *ancestors,
 			}
 			break;
 		case T_CustomScan:
+		case T_CustomPlanMarkPos:
 			sname = "Custom Scan";
 			custom_name = ((CustomScan *) plan)->methods->CustomName;
 			if (custom_name)
@@ -1674,10 +1676,18 @@ ExplainNode(PlanState *planstate, List *ancestors,
 			ExplainScanTarget((Scan *) plan, es);
 			break;
 		case T_ForeignScan:
-		case T_CustomScan:
 			if (((Scan *) plan)->scanrelid > 0)
 				ExplainScanTarget((Scan *) plan, es);
 			break;
+		case T_CustomScan:
+		case T_CustomPlanMarkPos:
+			{
+				CustomScanState *css = (CustomScanState *) planstate;
+
+				if (css->methods->ExplainCustomPlanTargetRel)
+					css->methods->ExplainCustomPlanTargetRel(css, es);
+			}
+			break;
 		case T_IndexScan:
 			{
 				IndexScan  *indexscan = (IndexScan *) plan;
@@ -2149,6 +2159,7 @@ ExplainNode(PlanState *planstate, List *ancestors,
 			show_foreignscan_info((ForeignScanState *) planstate, es);
 			break;
 		case T_CustomScan:
+		case T_CustomPlanMarkPos:
 			{
 				CustomScanState *css = (CustomScanState *) planstate;
 
@@ -2410,6 +2421,7 @@ ExplainNode(PlanState *planstate, List *ancestors,
 						"Subquery", NULL, es);
 			break;
 		case T_CustomScan:
+		case T_CustomPlanMarkPos:
 			ExplainCustomChildren((CustomScanState *) planstate,
 								  ancestors, es);
 			break;
@@ -4428,6 +4440,7 @@ ExplainTargetRel(Plan *plan, Index rti, ExplainState *es)
 		case T_TidRangeScan:
 		case T_ForeignScan:
 		case T_CustomScan:
+		case T_CustomPlanMarkPos:
 		case T_ModifyTable:
 			/* Assert it's on a real relation */
 			Assert(rte->rtekind == RTE_RELATION);
@@ -5109,3 +5122,12 @@ ExplainFlushWorkersState(ExplainState *es)
 	pfree(wstate->worker_state_save);
 	pfree(wstate);
 }
+
+void
+ExplainPropertySortGroupKeys(PlanState *planstate, const char *qlabel,
+							 int nkeys, AttrNumber *keycols,
+							 List *ancestors, ExplainState *es)
+{
+	show_sort_group_keys(planstate, qlabel, nkeys, 0, keycols,
+						 NULL, NULL, NULL, ancestors, es);
+}