From c8a21e0d7cb0d8340a885be0a53255060c982ec1 Mon Sep 17 00:00:00 2001
From: Ashwin Agrawal <aagrawal@pivotal.io>
Date: Tue, 8 Mar 2016 15:10:20 -0800
Subject: [PATCH] Validate gp_relation_node tuple after index fetch.

This commit makes sure while accessing gp_relation_node through its
index, sanity check is always performed to verify the tuple being
operated on is the intended tuple, else for any reason index is broken
and provide bad tuple fail the operation instead of causing damage.
For some scenarios like delete gp_relation_code tuple case it adds extra tuple
deform call which was not done earlier but doesn't seem heavy enough to
be performed on ddl operation.
---
 src/backend/commands/tablecmds.c   | 14 +++++++-----
 src/backend/utils/cache/relcache.c | 34 ++++++++++++++++++------------
 src/include/utils/relationnode.h   |  4 ----
 3 files changed, 30 insertions(+), 22 deletions(-)

diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 1c9a3cf297..eb0adc8f08 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -10872,7 +10872,9 @@ ATExecSetTableSpace_BufferPool(
 	int64 newPersistentSerialNum;
 	SMgrRelation dstrel;
 	bool useWal;
-	
+	ItemPointerData oldPersistentTid;
+	int64 oldPersistentSerialNum;
+
 	PersistentFileSysRelStorageMgr localRelStorageMgr;
 	PersistentFileSysRelBufpoolKind relBufpoolKind;
 	
@@ -10900,10 +10902,12 @@ ATExecSetTableSpace_BufferPool(
 			 (!useWal ? "true" : "false"));
 	
 	/* Fetch relation's gp_relation_node row */
-	nodeTuple = ScanGpRelationNodeTuple(
-							gp_relation_node,
-							rel->rd_rel->relfilenode,
-							/* segmentFileNum */ 0);
+	nodeTuple = FetchGpRelationNodeTuple(
+					gp_relation_node,
+					rel->rd_rel->relfilenode,
+					/* segmentFileNum */ 0,
+					&oldPersistentTid,
+					&oldPersistentSerialNum);
 	if (!HeapTupleIsValid(nodeTuple))
 		elog(ERROR, "cache lookup failed for relation %u, tablespace %u, relation file node %u", 
 		     tableOid,
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index a6215a0485..950794ab0b 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -356,11 +356,8 @@ GpRelationNodeBeginScan(
 HeapTuple
 GpRelationNodeGetNext(
 	GpRelationNodeScan 	*gpRelationNodeScan,
-
 	int32				*segmentFileNum,
-
 	ItemPointer			persistentTid,
-
 	int64				*persistentSerialNum)
 {
 	HeapTuple tuple;
@@ -394,7 +391,8 @@ GpRelationNodeGetNext(
 						persistentTid,
 						persistentSerialNum);
 	if (actualRelationNode != gpRelationNodeScan->relfilenode)
-		elog(FATAL, "Mismatch in node tuple for gp_relation_node for relation %u, relfilenode %u, relation node %u",
+		elog(FATAL, "Index on gp_relation_node broken."
+			   "Mismatch in node tuple for gp_relation_node for relation %u, relfilenode %u, relation node %u",
 			 gpRelationNodeScan->relationId, 
 			 gpRelationNodeScan->relfilenode,
 			 actualRelationNode);
@@ -411,8 +409,7 @@ GpRelationNodeEndScan(
 	systable_endscan((SysScanDesc)gpRelationNodeScan->scan);
 }
 
-
-HeapTuple
+static HeapTuple
 ScanGpRelationNodeTuple(
 	Relation 	gp_relation_node,
 	Oid 		relfilenode,
@@ -516,8 +513,15 @@ FetchGpRelationNodeTuple(
 						&createMirrorDataLossTrackingSessionNum,
 						persistentTid,
 						persistentSerialNum);
-	Assert (actualRelationNode == relfilenode);
 	
+	if (actualRelationNode != relfilenode)
+	{
+		elog(ERROR, "Index on gp_relation_node broken."
+			   "Mismatch in node tuple for gp_relation_node intended relfilenode %u, fetched relfilenode %u",
+			 relfilenode,
+			 actualRelationNode);
+	}
+
 	return tuple;
 }
 
@@ -532,21 +536,25 @@ DeleteGpRelationNodeTuple(
 {
 	Relation	gp_relation_node;
 	HeapTuple	tuple;
+	ItemPointerData     persistentTid;
+	int64               persistentSerialNum;
 
-	/* Grab an appropriate lock on the pg_class relation */
+	/* Grab an appropriate lock on the gp_relation_node relation */
 	gp_relation_node = heap_open(GpRelationNodeRelationId, RowExclusiveLock);
 
-	tuple = ScanGpRelationNodeTuple(
-						gp_relation_node,
-						relation->rd_rel->relfilenode,
-						segmentFileNum);
+	tuple = FetchGpRelationNodeTuple(gp_relation_node,
+				relation->rd_rel->relfilenode,
+				segmentFileNum,
+				&persistentTid,
+				&persistentSerialNum);
+
 	if (!HeapTupleIsValid(tuple))
 		elog(ERROR, "could not find node tuple for relation %u, relation file node %u, segment file #%d",
 			 RelationGetRelid(relation),
 			 relation->rd_rel->relfilenode,
 			 segmentFileNum);
 
-	/* delete the relation tuple from pg_class, and finish up */
+	/* delete the relation tuple from gp_relation_node, and finish up */
 	simple_heap_delete(gp_relation_node, &tuple->t_self);
 	heap_freetuple(tuple);
 
diff --git a/src/include/utils/relationnode.h b/src/include/utils/relationnode.h
index b12ebb1207..baff09553f 100644
--- a/src/include/utils/relationnode.h
+++ b/src/include/utils/relationnode.h
@@ -44,10 +44,6 @@ extern void
 GpRelationNodeEndScan(
 	GpRelationNodeScan 	*gpRelationNodeScan);
 
-extern HeapTuple ScanGpRelationNodeTuple(
-	Relation 	gp_relation_node,
-	Oid 		relationNode,
-	int32		segmentFileNum);
 extern HeapTuple FetchGpRelationNodeTuple(
 	Relation 	gp_relation_node,
 	Oid 		relationNode,
-- 
GitLab