From 711428b9865c44ce6eeb9640dac07c2a977e283c Mon Sep 17 00:00:00 2001
From: Hari Babu <kommi.haribabu@gmail.com>
Date: Thu, 29 Mar 2018 16:59:20 +1100
Subject: [PATCH 10/14] table rewrite functionality

All the heap rewrite functionality is moved into heap table AM
and exposed them with table AM API. Currenlty these API's are used
only by the cluster command operation.

The logical rewrite mapping code is currently left as it is,
this needs to be handled separately.
---
 src/backend/access/heap/heapam_handler.c |  6 ++++++
 src/backend/access/table/tableam.c       | 33 ++++++++++++++++++++++++++++++++
 src/backend/commands/cluster.c           | 32 +++++++++++++++----------------
 src/include/access/heapam.h              |  9 +++++++++
 src/include/access/rewriteheap.h         | 11 -----------
 src/include/access/tableam.h             |  8 ++++++++
 src/include/access/tableam_common.h      |  2 ++
 src/include/access/tableamapi.h          | 13 +++++++++++++
 8 files changed, 86 insertions(+), 28 deletions(-)

diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index ba49551a07..53123927da 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -22,6 +22,7 @@
 
 #include "access/heapam.h"
 #include "access/relscan.h"
+#include "access/rewriteheap.h"
 #include "access/tableamapi.h"
 #include "pgstat.h"
 #include "storage/lmgr.h"
@@ -388,5 +389,10 @@ heap_tableam_handler(PG_FUNCTION_ARGS)
 	amroutine->freebulkinsertstate = FreeBulkInsertState;
 	amroutine->releasebulkinsertstate = ReleaseBulkInsertStatePin;
 
+	amroutine->begin_heap_rewrite = begin_heap_rewrite;
+	amroutine->end_heap_rewrite = end_heap_rewrite;
+	amroutine->rewrite_heap_tuple = rewrite_heap_tuple;
+	amroutine->rewrite_heap_dead_tuple = rewrite_heap_dead_tuple;
+
 	PG_RETURN_POINTER(amroutine);
 }
diff --git a/src/backend/access/table/tableam.c b/src/backend/access/table/tableam.c
index 25db521dca..f1ba658f17 100644
--- a/src/backend/access/table/tableam.c
+++ b/src/backend/access/table/tableam.c
@@ -412,3 +412,36 @@ table_releasebulkinsertstate(Relation rel, BulkInsertState bistate)
 {
 	rel->rd_tableamroutine->releasebulkinsertstate(bistate);
 }
+
+/*
+ * -------------------
+ * storage tuple rewrite functions
+ * -------------------
+ */
+RewriteState
+table_begin_rewrite(Relation OldHeap, Relation NewHeap,
+				   TransactionId OldestXmin, TransactionId FreezeXid,
+				   MultiXactId MultiXactCutoff, bool use_wal)
+{
+	return NewHeap->rd_tableamroutine->begin_heap_rewrite(OldHeap, NewHeap,
+			OldestXmin, FreezeXid, MultiXactCutoff, use_wal);
+}
+
+void
+table_end_rewrite(Relation rel, RewriteState state)
+{
+	rel->rd_tableamroutine->end_heap_rewrite(state);
+}
+
+void
+table_rewrite_tuple(Relation rel, RewriteState state, HeapTuple oldTuple,
+				   HeapTuple newTuple)
+{
+	rel->rd_tableamroutine->rewrite_heap_tuple(state, oldTuple, newTuple);
+}
+
+bool
+table_rewrite_dead_tuple(Relation rel, RewriteState state, HeapTuple oldTuple)
+{
+	return rel->rd_tableamroutine->rewrite_heap_dead_tuple(state, oldTuple);
+}
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index c0b88bb01f..556f745012 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -74,9 +74,8 @@ static void copy_heap_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex,
 			   TransactionId *pFreezeXid, MultiXactId *pCutoffMulti);
 static List *get_tables_to_cluster(MemoryContext cluster_context);
 static void reform_and_rewrite_tuple(HeapTuple tuple,
-						 TupleDesc oldTupDesc, TupleDesc newTupDesc,
-						 Datum *values, bool *isnull,
-						 bool newRelHasOids, RewriteState rwstate);
+						 Relation OldHeap, Relation NewHeap,
+						 Datum *values, bool *isnull, RewriteState rwstate);
 
 
 /*---------------------------------------------------------------------------
@@ -893,7 +892,7 @@ copy_heap_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex, bool verbose,
 	is_system_catalog = IsSystemRelation(OldHeap);
 
 	/* Initialize the rewrite operation */
-	rwstate = begin_heap_rewrite(OldHeap, NewHeap, OldestXmin, FreezeXid,
+	rwstate = table_begin_rewrite(OldHeap, NewHeap, OldestXmin, FreezeXid,
 								 MultiXactCutoff, use_wal);
 
 	/*
@@ -1043,7 +1042,7 @@ copy_heap_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex, bool verbose,
 		{
 			tups_vacuumed += 1;
 			/* heap rewrite module still needs to see it... */
-			if (rewrite_heap_dead_tuple(rwstate, tuple))
+			if (table_rewrite_dead_tuple(NewHeap, rwstate, tuple))
 			{
 				/* A previous recently-dead tuple is now known dead */
 				tups_vacuumed += 1;
@@ -1057,9 +1056,8 @@ copy_heap_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex, bool verbose,
 			tuplesort_putheaptuple(tuplesort, tuple);
 		else
 			reform_and_rewrite_tuple(tuple,
-									 oldTupDesc, newTupDesc,
-									 values, isnull,
-									 NewHeap->rd_rel->relhasoids, rwstate);
+									 OldHeap, NewHeap,
+									 values, isnull, rwstate);
 	}
 
 	if (indexScan != NULL)
@@ -1086,16 +1084,15 @@ copy_heap_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex, bool verbose,
 				break;
 
 			reform_and_rewrite_tuple(tuple,
-									 oldTupDesc, newTupDesc,
-									 values, isnull,
-									 NewHeap->rd_rel->relhasoids, rwstate);
+									 OldHeap, NewHeap,
+									 values, isnull, rwstate);
 		}
 
 		tuplesort_end(tuplesort);
 	}
 
 	/* Write out any remaining tuples, and fsync if needed */
-	end_heap_rewrite(rwstate);
+	table_end_rewrite(NewHeap, rwstate);
 
 	/* Reset rd_toastoid just to be tidy --- it shouldn't be looked at again */
 	NewHeap->rd_toastoid = InvalidOid;
@@ -1759,10 +1756,11 @@ get_tables_to_cluster(MemoryContext cluster_context)
  */
 static void
 reform_and_rewrite_tuple(HeapTuple tuple,
-						 TupleDesc oldTupDesc, TupleDesc newTupDesc,
-						 Datum *values, bool *isnull,
-						 bool newRelHasOids, RewriteState rwstate)
+						 Relation OldHeap, Relation NewHeap,
+						 Datum *values, bool *isnull, RewriteState rwstate)
 {
+	TupleDesc oldTupDesc = RelationGetDescr(OldHeap);
+	TupleDesc newTupDesc = RelationGetDescr(NewHeap);
 	HeapTuple	copiedTuple;
 	int			i;
 
@@ -1778,11 +1776,11 @@ reform_and_rewrite_tuple(HeapTuple tuple,
 	copiedTuple = heap_form_tuple(newTupDesc, values, isnull);
 
 	/* Preserve OID, if any */
-	if (newRelHasOids)
+	if (NewHeap->rd_rel->relhasoids)
 		HeapTupleSetOid(copiedTuple, HeapTupleGetOid(tuple));
 
 	/* The heap rewrite module does the rest */
-	rewrite_heap_tuple(rwstate, tuple, copiedTuple);
+	table_rewrite_tuple(NewHeap, rwstate, tuple, copiedTuple);
 
 	heap_freetuple(copiedTuple);
 }
diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index f9a0602b86..354ac62fb1 100644
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -212,4 +212,13 @@ extern bool HeapTupleHeaderIsOnlyLocked(HeapTupleHeader tuple);
 extern bool XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
 extern bool HeapTupleIsSurelyDead(HeapTuple htup, TransactionId OldestXmin);
 
+/* in heap/rewriteheap.c */
+extern RewriteState begin_heap_rewrite(Relation OldHeap, Relation NewHeap,
+				   TransactionId OldestXmin, TransactionId FreezeXid,
+				   MultiXactId MultiXactCutoff, bool use_wal);
+extern void end_heap_rewrite(RewriteState state);
+extern void rewrite_heap_tuple(RewriteState state, HeapTuple oldTuple,
+				   HeapTuple newTuple);
+extern bool rewrite_heap_dead_tuple(RewriteState state, HeapTuple oldTuple);
+
 #endif							/* HEAPAM_H */
diff --git a/src/include/access/rewriteheap.h b/src/include/access/rewriteheap.h
index 6d7f669cbc..c610914133 100644
--- a/src/include/access/rewriteheap.h
+++ b/src/include/access/rewriteheap.h
@@ -18,17 +18,6 @@
 #include "storage/relfilenode.h"
 #include "utils/relcache.h"
 
-/* struct definition is private to rewriteheap.c */
-typedef struct RewriteStateData *RewriteState;
-
-extern RewriteState begin_heap_rewrite(Relation OldHeap, Relation NewHeap,
-				   TransactionId OldestXmin, TransactionId FreezeXid,
-				   MultiXactId MultiXactCutoff, bool use_wal);
-extern void end_heap_rewrite(RewriteState state);
-extern void rewrite_heap_tuple(RewriteState state, HeapTuple oldTuple,
-				   HeapTuple newTuple);
-extern bool rewrite_heap_dead_tuple(RewriteState state, HeapTuple oldTuple);
-
 /*
  * On-Disk data format for an individual logical rewrite mapping.
  */
diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
index 1027bcfba8..54f7c41108 100644
--- a/src/include/access/tableam.h
+++ b/src/include/access/tableam.h
@@ -122,4 +122,12 @@ extern BulkInsertState table_getbulkinsertstate(Relation rel);
 extern void table_freebulkinsertstate(Relation rel, BulkInsertState bistate);
 extern void table_releasebulkinsertstate(Relation rel, BulkInsertState bistate);
 
+extern RewriteState table_begin_rewrite(Relation OldHeap, Relation NewHeap,
+				   TransactionId OldestXmin, TransactionId FreezeXid,
+				   MultiXactId MultiXactCutoff, bool use_wal);
+extern void table_end_rewrite(Relation rel, RewriteState state);
+extern void table_rewrite_tuple(Relation rel, RewriteState state, HeapTuple oldTuple,
+				   HeapTuple newTuple);
+extern bool table_rewrite_dead_tuple(Relation rel, RewriteState state, HeapTuple oldTuple);
+
 #endif		/* TABLEAM_H */
diff --git a/src/include/access/tableam_common.h b/src/include/access/tableam_common.h
index 3c2ce82df3..b8fdfddb31 100644
--- a/src/include/access/tableam_common.h
+++ b/src/include/access/tableam_common.h
@@ -41,6 +41,8 @@ typedef enum
 
 typedef struct BulkInsertStateData *BulkInsertState;
 
+/* struct definition is private to rewriteheap.c */
+typedef struct RewriteStateData *RewriteState;
 
 /*
  * slot table AM routine functions
diff --git a/src/include/access/tableamapi.h b/src/include/access/tableamapi.h
index adc3057eca..9e43db0259 100644
--- a/src/include/access/tableamapi.h
+++ b/src/include/access/tableamapi.h
@@ -85,6 +85,14 @@ typedef BulkInsertState (*GetBulkInsertState_function) (void);
 typedef void (*FreeBulkInsertState_function) (BulkInsertState bistate);
 typedef void (*ReleaseBulkInsertState_function) (BulkInsertState bistate);
 
+typedef RewriteState (*BeginHeapRewrite_function) (Relation OldHeap, Relation NewHeap,
+				   TransactionId OldestXmin, TransactionId FreezeXid,
+				   MultiXactId MultiXactCutoff, bool use_wal);
+typedef void (*EndHeapRewrite_function) (RewriteState state);
+typedef void (*RewriteHeapTuple_function) (RewriteState state, HeapTuple oldTuple,
+				   HeapTuple newTuple);
+typedef bool (*RewriteHeapDeadTuple_function) (RewriteState state, HeapTuple oldTuple);
+
 typedef TableScanDesc (*ScanBegin_function) (Relation relation,
 											Snapshot snapshot,
 											int nkeys, ScanKey key,
@@ -161,6 +169,11 @@ typedef struct TableAmRoutine
 	FreeBulkInsertState_function freebulkinsertstate;
 	ReleaseBulkInsertState_function releasebulkinsertstate;
 
+	BeginHeapRewrite_function begin_heap_rewrite;
+	EndHeapRewrite_function end_heap_rewrite;
+	RewriteHeapTuple_function rewrite_heap_tuple;
+	RewriteHeapDeadTuple_function rewrite_heap_dead_tuple;
+
 	/* Operations on relation scans */
 	ScanBegin_function scan_begin;
 	ScanGetParallelheapscandesc_function scan_get_parallelheapscandesc;
-- 
2.16.1.windows.4

