diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c
index 427b0d5..2a895a6 100644
--- a/src/backend/storage/ipc/ipci.c
+++ b/src/backend/storage/ipc/ipci.c
@@ -44,6 +44,7 @@
 #include "storage/procsignal.h"
 #include "storage/sinvaladt.h"
 #include "storage/spin.h"
+#include "utils/shared_meta_cache.h"
 #include "utils/snapmgr.h"
 
 /* GUCs */
@@ -147,6 +148,7 @@ CreateSharedMemoryAndSemaphores(void)
 		size = add_size(size, BTreeShmemSize());
 		size = add_size(size, SyncScanShmemSize());
 		size = add_size(size, AsyncShmemSize());
+		size = add_size(size, GlobalMetaCacheShmemSize());
 #ifdef EXEC_BACKEND
 		size = add_size(size, ShmemBackendArraySize());
 #endif
@@ -263,6 +265,7 @@ CreateSharedMemoryAndSemaphores(void)
 	BTreeShmemInit();
 	SyncScanShmemInit();
 	AsyncShmemInit();
+	GlobalMetaCacheShmemInit();
 
 #ifdef EXEC_BACKEND
 
diff --git a/src/backend/storage/lmgr/lwlock.c b/src/backend/storage/lmgr/lwlock.c
index d07ce60..008f409 100644
--- a/src/backend/storage/lmgr/lwlock.c
+++ b/src/backend/storage/lmgr/lwlock.c
@@ -87,6 +87,7 @@
 #include "storage/proclist.h"
 #include "storage/spin.h"
 #include "utils/memutils.h"
+#include "utils/shared_meta_cache.h"
 
 #ifdef LWLOCK_STATS
 #include "utils/hsearch.h"
@@ -522,6 +523,14 @@ RegisterLWLockTranches(void)
 	LWLockRegisterTranche(LWTRANCHE_PARALLEL_APPEND, "parallel_append");
 	LWLockRegisterTranche(LWTRANCHE_PARALLEL_HASH_JOIN, "parallel_hash_join");
 	LWLockRegisterTranche(LWTRANCHE_SXACT, "serializable_xact");
+	LWLockRegisterTranche(LWTRANCHE_SHARED_MCXT, "shared_mcxt");
+
+	if (IS_GLOBAL_META_CACHE)
+	{
+		LWLockRegisterTranche(LWTRANCHE_META_CACHE_MAP, "meta_cache_map");
+		LWLockRegisterTranche(LWTRANCHE_GLOBAL_CATCACHE, "global_metacache");
+		LWLockRegisterTranche(LWTRANCHE_META_CACHE_HANDLE, "meta_cache_handle");
+	}
 
 	/* Register named tranches. */
 	for (i = 0; i < NamedLWLockTrancheRequests; i++)
diff --git a/src/backend/utils/cache/Makefile b/src/backend/utils/cache/Makefile
index 38e46d2..1e430dd 100644
--- a/src/backend/utils/cache/Makefile
+++ b/src/backend/utils/cache/Makefile
@@ -23,6 +23,7 @@ OBJS = \
 	relcache.o \
 	relfilenodemap.o \
 	relmapper.o \
+	shared_cache.o \
 	spccache.o \
 	syscache.o \
 	ts_cache.o \
diff --git a/src/backend/utils/cache/catcache.c b/src/backend/utils/cache/catcache.c
index 64776e3..0d807c3 100644
--- a/src/backend/utils/cache/catcache.c
+++ b/src/backend/utils/cache/catcache.c
@@ -39,6 +39,14 @@
 #include "utils/resowner_private.h"
 #include "utils/syscache.h"
 
+#define CATC_HANDLE_VALID(localCatCTup) \
+	((localCatCTup)->lct_generation == localCatCTup->lct_handle->generation)
+
+#define CATC_IS_INVALID_DUMMY(localCatCTup) \
+	((localCatCTup)->lct_skip_global_subxid != InvalidSubTransactionId)
+
+/**/
+#define MAX_CC_BUCKET_LOCKS 64
 
  /* #define CACHEDEBUG */	/* turns DEBUG elogs on */
 
@@ -60,6 +68,39 @@
 #define CACHE_elog(...)
 #endif
 
+
+/* structs for Global CatCache */
+struct GlobalCatCache
+{
+	CatCache gcc_catcache; 	/* normal CatCache */
+	slock_t gcc_mutex; 	/* lock for CatCache itself */
+	LWLock *gcc_bucket_locks;	/* locks for buckets */
+	HandleHeader gcc_handle_hdr; /* handle */
+};
+
+/* struct for Global CatCtup */
+struct LocalCatCTup
+{
+	CatCTup lct_ct;		/* regular CatCTup */
+	Handle *lct_handle; /* handle pointing to global cache entry */
+	int lct_generation; /* handle generation to check handle validity */
+	bool lct_committed; /* committed or not */
+	/*
+	 * lct_skip_global_subxid is xid this entry is invalidated. In the
+	 * middle of xact, regular catctup is removed or gets dead by local
+	 * invalidation messages. However, SharedLocal cache
+	 */
+	SubTransactionId lct_skip_global_subxid;  /* lct was invalidated in this xaxt */
+};
+
+/* struct for Global CatCtup */
+struct GlobalCatCTup
+{
+	CatCTup gct_ct;    	/* regular CatCTup */
+	Handle *gct_handle; /* handle */
+	slock_t gct_mutex; 	/* lock for refcount/dead of CatCTup */
+};
+
 /* Cache management header --- pointer is NULL until created */
 static CatCacheHeader *CacheHdr = NULL;
 
@@ -79,10 +120,12 @@ static uint32 CatalogCacheComputeHashValue(CatCache *cache, int nkeys,
 										   Datum v1, Datum v2, Datum v3, Datum v4);
 static uint32 CatalogCacheComputeTupleHashValue(CatCache *cache, int nkeys,
 												HeapTuple tuple);
-static inline bool CatalogCacheCompareTuple(const CatCache *cache, int nkeys,
+static inline bool CatalogCacheCompareTuple(const CCFastEqualFN *gcc_fastequal,
+											int nkeys,
 											const Datum *cachekeys,
 											const Datum *searchkeys);
 
+
 #ifdef CATCACHE_STATS
 static void CatCachePrintStats(int code, Datum arg);
 #endif
@@ -92,13 +135,45 @@ static void CatalogCacheInitializeCache(CatCache *cache);
 static CatCTup *CatalogCacheCreateEntry(CatCache *cache, HeapTuple ntp,
 										Datum *arguments,
 										uint32 hashValue, Index hashIndex,
-										bool negative);
+										bool negative, Handle* handle);
 
 static void CatCacheFreeKeys(TupleDesc tupdesc, int nkeys, int *attnos,
 							 Datum *keys);
 static void CatCacheCopyKeys(TupleDesc tupdesc, int nkeys, int *attnos,
 							 Datum *srckeys, Datum *dstkeys);
 
+static void InitLocalCatCacheInternal(CatCache *cache);
+static void InitGlobalCatCache(CatCache *local_cache, GlobalCatCacheMapKey *keyPtr);
+static GlobalCatCache *InitGlobalCatCacheInternal(CatCache *local_cache);
+static GlobalCatCache *AttachGlobalCatCache(GlobalCatCacheMapKey *keyPtr);
+
+static CatCTup *
+SearchCatCacheBucket(CatCache *cache, int nkeys, Datum *searchkeys,
+					 uint32 hashValue, Index hashIndex,
+					 const CCFastEqualFN *gcc_fastequal);
+
+static CatCTup *
+CreateCatCTupWithTuple(CatCache *cache, HeapTuple ntp);
+static CatCTup *CreateCatCTupWithoutTuple(CatCache *cache, Datum *arguments);
+static CatCTup *
+LocalCacheGetValidEntry(LocalCatCTup *lct);
+
+static LWLock *
+GetCatCacheBucketLock(GlobalCatCache *gcp, uint32 hashValue);
+static void ReleaseLocalCtlistItem(CatCache *cache, List *ctlist);
+static void IncreaceGlobalCatCTupRefCount(GlobalCatCTup *gct);
+static void GlobalCatCacheRemoveCTup(GlobalCatCache *gcp, GlobalCatCTup *gct);
+static void
+GlobalCatCTupDecrementAndRemove(GlobalCatCache *gcp, GlobalCatCTup *gct);
+static void CatCacheInvalidateInternal(CatCache *local_cach, CatCTup *ct);
+static void	CreateInvalCatalogCache(CatCache *cache, uint32 hashValue);
+static void ResetRemoveCatalogCache(CatCache *cache);
+static void SetSkipGlobalSubxid(CatCache *cache);
+static void ResetSkipGlobalSubxid(CatCache *cache);
+static bool SkipGlobalSubxidIsOwnOrChild(CatCache *cache);
+static void ResetGlobalCatalogCache(CatCache *local_cache);
+static void GlobalCatCacheInvalInternal(GlobalCatCache *gcp,
+										CatCTup *global_ct);
 
 /*
  *					internal support functions
@@ -371,11 +446,10 @@ CatalogCacheComputeTupleHashValue(CatCache *cache, int nkeys, HeapTuple tuple)
  * Compare a tuple to the passed arguments.
  */
 static inline bool
-CatalogCacheCompareTuple(const CatCache *cache, int nkeys,
+CatalogCacheCompareTuple(const CCFastEqualFN *cc_fastequal, int nkeys,
 						 const Datum *cachekeys,
 						 const Datum *searchkeys)
 {
-	const CCFastEqualFN *cc_fastequal = cache->cc_fastequal;
 	int			i;
 
 	for (i = 0; i < nkeys; i++)
@@ -388,7 +462,6 @@ CatalogCacheCompareTuple(const CatCache *cache, int nkeys,
 
 
 #ifdef CATCACHE_STATS
-
 static void
 CatCachePrintStats(int code, Datum arg)
 {
@@ -420,7 +493,8 @@ CatCachePrintStats(int code, Datum arg)
 			 cache->cc_searches - cache->cc_hits - cache->cc_neg_hits,
 			 cache->cc_invals,
 			 cache->cc_lsearches,
-			 cache->cc_lhits);
+			 cache->cc_lhits
+			);
 		cc_searches += cache->cc_searches;
 		cc_hits += cache->cc_hits;
 		cc_neg_hits += cache->cc_neg_hits;
@@ -458,6 +532,7 @@ CatCacheRemoveCTup(CatCache *cache, CatCTup *ct)
 {
 	Assert(ct->refcount == 0);
 	Assert(ct->my_cache == cache);
+	Assert(cache->cc_type != CC_SHAREDGLOBAL);
 
 	if (ct->c_list)
 	{
@@ -482,6 +557,14 @@ CatCacheRemoveCTup(CatCache *cache, CatCTup *ct)
 		CatCacheFreeKeys(cache->cc_tupdesc, cache->cc_nkeys,
 						 cache->cc_keyno, ct->keys);
 
+	if (cache->cc_type == CC_SHAREDLOCAL)
+	{
+		/* Committed tuple is not negative but has keys. */
+		if (((LocalCatCTup *)ct)->lct_committed)
+			CatCacheFreeKeys(cache->cc_tupdesc, cache->cc_nkeys,
+							 cache->cc_keyno, ct->keys);
+	}
+
 	pfree(ct);
 
 	--cache->cc_ntup;
@@ -489,6 +572,85 @@ CatCacheRemoveCTup(CatCache *cache, CatCTup *ct)
 }
 
 /*
+ *		GlobalCatCTupDecrementAndRemove
+ *
+ * Release the reference counter of global cache entry. If cache entry is
+ * dead and referenced by one process alone, it's removed.
+ */
+static void
+GlobalCatCTupDecrementAndRemove(GlobalCatCache *gcp, GlobalCatCTup *gct)
+{
+	CatCTup *ct = (CatCTup *)gct;
+	LWLock *bucket_lock;
+
+	Assert(ct->my_cache == (CatCache *)gcp);
+	Assert(ct->my_cache->cc_type == CC_SHAREDGLOBAL);
+
+	SpinLockAcquire(&gct->gct_mutex);
+
+	Assert(ct->refcount > 0);
+
+	if (ct->dead && ct->refcount == 1)
+		SpinLockRelease(&gct->gct_mutex);
+	else
+	{
+		ct->refcount--;
+		SpinLockRelease(&gct->gct_mutex);
+		return;
+	}
+
+	/*
+	 * Here spin lock is not needed. Dead cache entry is never newly fetched.
+	 * So dead cache with refcount one is referenced only by this process.
+	 */
+	ct->refcount--;
+
+	bucket_lock = GetCatCacheBucketLock(gcp, ct->hash_value);
+
+	LWLockAcquire(bucket_lock, LW_EXCLUSIVE);
+	GlobalCatCacheRemoveCTup(gcp, gct);
+	LWLockRelease(bucket_lock);
+}
+
+/*
+ *		GlobalCatCacheRemoveCTup
+ *
+ * Unlink and delete the given global cache entry.
+ * This assumes that the bucket lock is aquired by the caller because multiple
+ * entries in one bucket are possibly removed recursively.
+ */
+static void
+GlobalCatCacheRemoveCTup(GlobalCatCache *gcp, GlobalCatCTup *gct)
+{
+	CatCTup *ct = (CatCTup *)gct;
+	LWLock *handle_lock;
+
+	Assert(ct->refcount == 0);
+
+	handle_lock = GET_HANDLE_LOCK(gct->gct_handle);
+	LWLockAcquire(handle_lock, LW_EXCLUSIVE);
+
+	/* delete from hash table */
+	dlist_delete(&ct->cache_elem);
+
+	/* invalidate handle */
+	Assert(gct->gct_handle->entry == gct);
+	InvalCacheHandle(&gcp->gcc_handle_hdr, gct->gct_handle);
+	LWLockRelease(handle_lock);
+
+	/*
+	 * Free after releasing hadnle lock is safe because no other process cannot
+	 * access this entry here.
+	 */
+	pfree(gct);
+
+	SpinLockAcquire(&gcp->gcc_mutex);
+	((CatCache *)gcp)->cc_ntup--;
+	SpinLockRelease(&gcp->gcc_mutex);
+}
+
+
+/*
  *		CatCacheRemoveCList
  *
  * Unlink and delete the given cache list entry
@@ -502,11 +664,16 @@ CatCacheRemoveCList(CatCache *cache, CatCList *cl)
 
 	Assert(cl->refcount == 0);
 	Assert(cl->my_cache == cache);
+	Assert(cache->cc_type != CC_SHAREDGLOBAL);
 
 	/* delink from member tuples */
 	for (i = cl->n_members; --i >= 0;)
 	{
-		CatCTup    *ct = cl->members[i];
+		CatCTup    *ct;
+		if (cache->cc_type == CC_REGULAR)
+			ct = cl->members[i];
+		else
+			ct = (CatCTup *)cl->shared_local_members[i];
 
 		Assert(ct->c_list == cl);
 		ct->c_list = NULL;
@@ -526,6 +693,9 @@ CatCacheRemoveCList(CatCache *cache, CatCList *cl)
 	CatCacheFreeKeys(cache->cc_tupdesc, cl->nkeys,
 					 cache->cc_keyno, cl->keys);
 
+	if (cache->cc_type == CC_SHAREDLOCAL)
+		pfree(cl->shared_local_members);
+
 	pfree(cl);
 }
 
@@ -549,10 +719,12 @@ CatCacheRemoveCList(CatCache *cache, CatCList *cl)
  *	This routine is only quasi-public: it should only be used by inval.c.
  */
 void
-CatCacheInvalidate(CatCache *cache, uint32 hashValue)
+CatCacheInvalidate(CatCache *cache, uint32 hashValue, InvalPhase invalPhase)
 {
 	Index		hashIndex;
 	dlist_mutable_iter iter;
+	LocalCatCTup *lct;
+	bool found_invaldummy = false;
 
 	CACHE_elog(DEBUG2, "CatCacheInvalidate: called");
 
@@ -585,22 +757,164 @@ CatCacheInvalidate(CatCache *cache, uint32 hashValue)
 
 		if (hashValue == ct->hash_value)
 		{
-			if (ct->refcount > 0 ||
-				(ct->c_list && ct->c_list->refcount > 0))
+			switch (invalPhase)
 			{
-				ct->dead = true;
-				/* list, if any, was marked dead above */
-				Assert(ct->c_list == NULL || ct->c_list->dead);
-			}
-			else
-				CatCacheRemoveCTup(cache, ct);
-			CACHE_elog(DEBUG2, "CatCacheInvalidate: invalidated");
+				case INVAL_COMMON:
+				case INVAL_GMC_ABORT_XACT:
+					CatCacheInvalidateInternal(cache, ct);
+
+					CACHE_elog(DEBUG2, "CatCacheInvalidate: invalidated");
 #ifdef CATCACHE_STATS
-			cache->cc_invals++;
+					cache->cc_invals++;
 #endif
+					break;
+
+				case INVAL_GMC_EO_COMMAND:
+				case INVAL_GMC_COMMIT_SUBXACT:
+					lct = (LocalCatCTup *)ct;
+
+					if (CATC_IS_INVALID_DUMMY(lct))
+					{
+						found_invaldummy = true;
+						continue;
+					}
+
+					CatCacheInvalidateInternal(cache, ct);
+					break;
+
+				case INVAL_GMC_COMMIT_XACT:
+					lct = (LocalCatCTup *)ct;
+
+					if (lct->lct_committed || !CATC_IS_INVALID_DUMMY(lct))
+						continue;
+
+					CatCacheInvalidateInternal(cache, ct);
+					break;
+
+				case INVAL_GMC_ABORT_SUBXACT:
+					lct = (LocalCatCTup *)ct;
+
+					/*
+					 * if found one has skip-flag and it's set by parent's
+					 * transaction, that dummy cache cannot be erased.
+					 */
+					if (lct->lct_skip_global_subxid != InvalidSubTransactionId &&
+						lct->lct_skip_global_subxid < GetCurrentSubTransactionId())
+						continue;
+
+					CatCacheInvalidateInternal(cache, ct);
+					break;
+			}
+			/* could be multiple matches, so keep looking! */
+		}
+	}
+
+	/*
+	 * Create a dummy cache with subxid to prevent upcoming command from
+	 * searching global cache instead of actual system catalog.
+	 */
+	if (invalPhase == INVAL_GMC_EO_COMMAND && !found_invaldummy)
+		CreateInvalCatalogCache(cache, hashValue);
+
+}
+
+static void
+CatCacheInvalidateInternal(CatCache *local_cache, CatCTup *ct)
+{
+	Assert(local_cache->cc_type != CC_SHAREDGLOBAL);
+
+	if (ct->refcount > 0 ||
+		(ct->c_list && ct->c_list->refcount > 0))
+	{
+		ct->dead = true;
+		/* list, if any, was marked dead in CatCacheInvalidate */
+		Assert(ct->c_list == NULL || ct->c_list->dead);
+	}
+	else
+		CatCacheRemoveCTup(local_cache, ct);
+}
+
+static void
+CreateInvalCatalogCache(CatCache *cache, uint32 hashValue)
+{
+	Index hashIndex;
+
+	Assert(cache->cc_type == CC_SHAREDLOCAL);
+
+	hashIndex = HASH_INDEX(hashValue, cache->cc_nbuckets);
+
+	CatalogCacheCreateEntry(cache, NULL, NULL,
+							hashValue, hashIndex,
+							false, NULL);
+
+}
+
+void
+GlobalCatCacheInvalidate(CatCache *local_cache, uint32 hashValue)
+{
+	CatCache *global_cache;
+	Index		hashIndex;
+	dlist_mutable_iter iter;
+	LWLock *bucket_lock;
+
+	CACHE_elog(DEBUG2, "GlobalCatCacheInvalidate: called");
+
+	/*
+	 * We don't bother to check whether the cache has finished initialization
+	 * yet; if not, there will be no entries in it so no problem.
+	 *
+	 * We don't also care CatCList because there is no global CatCList.
+	 */
+
+	/*
+	 * inspect the proper hash bucket of global hash table
+	 */
+
+	/* First acquire lock for bucket */
+	bucket_lock = GetCatCacheBucketLock(local_cache->gcp, hashValue);
+
+	LWLockAcquire(bucket_lock, LW_EXCLUSIVE);
+
+	global_cache = (CatCache *)local_cache->gcp;
+
+	hashIndex = HASH_INDEX(hashValue, global_cache->cc_nbuckets);
+
+	dlist_foreach_modify(iter, &global_cache->cc_bucket[hashIndex])
+	{
+		CatCTup    *ct = dlist_container(CatCTup, cache_elem, iter.cur);
+
+		if (hashValue == ct->hash_value)
+		{
+			GlobalCatCacheInvalInternal(local_cache->gcp, ct);
+
+			CACHE_elog(DEBUG2, "GlobalCatCacheInvalidate: invalidated");
+
 			/* could be multiple matches, so keep looking! */
 		}
 	}
+
+	LWLockRelease(bucket_lock);
+}
+
+/*
+ * Lock to bucket should be aquired by the caller
+ */
+static void
+GlobalCatCacheInvalInternal(GlobalCatCache *gcp, CatCTup *global_ct)
+{
+	slock_t *gct_mutex;
+	GlobalCatCTup *gct = (GlobalCatCTup *)global_ct;
+
+	if (global_ct->refcount > 0)
+	{
+		/* Get lock for CatCTup */
+		gct_mutex = &gct->gct_mutex;
+		SpinLockAcquire(gct_mutex);
+		global_ct->dead = true;
+		SpinLockRelease(gct_mutex);
+	}
+	else
+		GlobalCatCacheRemoveCTup(gcp, gct);
 }
 
 /* ----------------------------------------------------------------
@@ -629,7 +943,6 @@ CreateCacheMemoryContext(void)
 												   ALLOCSET_DEFAULT_SIZES);
 }
 
-
 /*
  *		ResetCatalogCache
  *
@@ -639,10 +952,9 @@ CreateCacheMemoryContext(void)
  * However, it shouldn't need to be efficient; we don't invoke it often.
  */
 static void
-ResetCatalogCache(CatCache *cache)
+ResetCatalogCache(CatCache *cache, InvalPhase invalPhase)
 {
 	dlist_mutable_iter iter;
-	int			i;
 
 	/* Remove each list in this cache, or at least mark it dead */
 	dlist_foreach_modify(iter, &cache->cc_lists)
@@ -655,6 +967,65 @@ ResetCatalogCache(CatCache *cache)
 			CatCacheRemoveCList(cache, cl);
 	}
 
+	switch (invalPhase)
+	{
+		case INVAL_COMMON:
+		case INVAL_GMC_ABORT_XACT: /* this label is not used */
+
+			/*
+			 * When GMC is off and some cases when GMC is on; accepting sinval
+			 * or abort transaction.
+			 */
+			ResetRemoveCatalogCache(cache);
+
+			if (cache->cc_type == CC_SHAREDLOCAL)
+				ResetSkipGlobalSubxid(cache);
+
+			break;
+
+		case INVAL_GMC_EO_COMMAND:
+		case INVAL_GMC_COMMIT_SUBXACT: /* this label is not used */
+			ResetRemoveCatalogCache(cache);
+
+			/* Mark skip flag not to search global cache */
+			SetSkipGlobalSubxid(cache);
+			break;
+
+		case INVAL_GMC_COMMIT_XACT:
+			/* ResetRemoveCatalogCache is not called  */
+			ResetGlobalCatalogCache(cache);
+			ResetSkipGlobalSubxid(cache);
+			break;
+
+		case INVAL_GMC_ABORT_SUBXACT:
+			ResetGlobalCatalogCache(cache);
+
+			/*
+			 * Reset skip-flag only if it's set by current subtransaction or
+			 * children's subtransaction not to affect parent's subtransaction.
+			 */
+			if (SkipGlobalSubxidIsOwnOrChild(cache))
+				ResetSkipGlobalSubxid(cache);
+
+			break;
+	}
+}
+
+
+/*
+ * ResetRemoveCatalogCache
+ *   Workhorse for ResetRemoveCatalogCache
+ *   This fucntion invalidates all caches and reset skip_global_subxid only if
+ *   GMC is on.
+ */
+static void
+ResetRemoveCatalogCache(CatCache *cache)
+{
+	dlist_mutable_iter iter;
+	int  i;
+
+	Assert(cache->cc_type != CC_SHAREDGLOBAL);
+
 	/* Remove each tuple in this cache, or at least mark it dead */
 	for (i = 0; i < cache->cc_nbuckets; i++)
 	{
@@ -678,6 +1049,69 @@ ResetCatalogCache(CatCache *cache)
 #endif
 		}
 	}
+
+}
+
+static void
+SetSkipGlobalSubxid(CatCache *cache)
+{
+	Assert(cache->cc_type == CC_SHAREDLOCAL);
+
+	cache->cc_skip_global_subxid = GetCurrentSubTransactionId();
+}
+
+static void
+ResetSkipGlobalSubxid(CatCache *cache)
+{
+	Assert(cache->cc_type == CC_SHAREDLOCAL);
+
+	cache->cc_skip_global_subxid = InvalidSubTransactionId;
+}
+
+static bool
+SkipGlobalSubxidIsOwnOrChild(CatCache *cache)
+{
+	return cache->cc_skip_global_subxid >= GetCurrentSubTransactionId();
+}
+
+static void
+ResetGlobalCatalogCache(CatCache *local_cache)
+{
+	CatCache *global_cache;
+	GlobalCatCache *gcp;
+	dlist_mutable_iter iter;
+	int			i;
+	int num_locks;
+
+	Assert(local_cache->cc_type == CC_SHAREDLOCAL);
+
+	/* global cache doesn't have cc_lists */
+
+	/* Get the corresponding cache and aquire locks */
+	gcp = local_cache->gcp;
+	global_cache = (CatCache *)gcp;
+
+	num_locks = MAX_CC_BUCKET_LOCKS;
+
+	for (i = 0; i < num_locks; ++i)
+		LWLockAcquire(&gcp->gcc_bucket_locks[i], LW_EXCLUSIVE);
+
+	/* Remove each tuple in this cache, or at least mark it dead */
+	for (i = 0; i < global_cache->cc_nbuckets; i++)
+	{
+		dlist_head *bucket = &global_cache->cc_bucket[i];
+
+		dlist_foreach_modify(iter, bucket)
+		{
+			CatCTup    *ct = dlist_container(CatCTup, cache_elem, iter.cur);
+
+			GlobalCatCacheInvalInternal(gcp, ct);
+		}
+	}
+
+	/* Release locks in reverse order */
+	for (i = num_locks - 1 ; i >= 0; --i)
+		LWLockRelease(&gcp->gcc_bucket_locks[i]);
 }
 
 /*
@@ -696,7 +1130,7 @@ ResetCatalogCaches(void)
 	{
 		CatCache   *cache = slist_container(CatCache, cc_next, iter.cur);
 
-		ResetCatalogCache(cache);
+		ResetCatalogCache(cache, INVAL_COMMON);
 	}
 
 	CACHE_elog(DEBUG2, "end of ResetCatalogCaches call");
@@ -716,7 +1150,7 @@ ResetCatalogCaches(void)
  *	this assumes the tupdesc of a cachable system table will not change...)
  */
 void
-CatalogCacheFlushCatalog(Oid catId)
+CatalogCacheFlushCatalog(Oid catId, InvalPhase invalPhase)
 {
 	slist_iter	iter;
 
@@ -730,7 +1164,7 @@ CatalogCacheFlushCatalog(Oid catId)
 		if (cache->cc_reloid == catId)
 		{
 			/* Yes, so flush all its contents */
-			ResetCatalogCache(cache);
+			ResetCatalogCache(cache, invalPhase);
 
 			/* Tell inval.c to call syscache callbacks for this cache */
 			CallSyscacheCallbacks(cache->id, 0);
@@ -740,6 +1174,7 @@ CatalogCacheFlushCatalog(Oid catId)
 	CACHE_elog(DEBUG2, "end of CatalogCacheFlushCatalog call");
 }
 
+
 /*
  *		InitCatCache
  *
@@ -817,6 +1252,11 @@ InitCatCache(int id,
 	cp = (CatCache *) CACHELINEALIGN(palloc0(sz));
 	cp->cc_bucket = palloc0(nbuckets * sizeof(dlist_head));
 
+	if (!IS_GLOBAL_META_CACHE)
+		cp->cc_type = CC_REGULAR;
+	else
+		cp->cc_type = CC_SHAREDLOCAL;
+
 	/*
 	 * initialize the cache's relation information for the relation
 	 * corresponding to this cache, and initialize some of the new cache's
@@ -831,6 +1271,8 @@ InitCatCache(int id,
 	cp->cc_ntup = 0;
 	cp->cc_nbuckets = nbuckets;
 	cp->cc_nkeys = nkeys;
+	cp->gcp = (GlobalCatCache *) NULL;
+	cp->cc_skip_global_subxid = InvalidSubTransactionId;
 	for (i = 0; i < nkeys; ++i)
 		cp->cc_keyno[i] = key[i];
 
@@ -860,15 +1302,36 @@ static void
 RehashCatCache(CatCache *cp)
 {
 	dlist_head *newbucket;
-	int			newnbuckets;
+	int		    newnbuckets;
 	int			i;
+	MemoryContext Context = NULL;
+	MemoryContext TempGlobalCatCacheContext = NULL;
 
-	elog(DEBUG1, "rehashing catalog cache id %d for %s; %d tups, %d buckets",
-		 cp->id, cp->cc_relname, cp->cc_ntup, cp->cc_nbuckets);
+	elog(DEBUG1, "rehashing catalog cache id %d for %s; %d tups, %d buckets"
+		 ", %d cc_type",
+		 cp->id, cp->cc_relname, cp->cc_ntup, cp->cc_nbuckets, cp->cc_type);
 
 	/* Allocate a new, larger, hash table. */
 	newnbuckets = cp->cc_nbuckets * 2;
-	newbucket = (dlist_head *) MemoryContextAllocZero(CacheMemoryContext, newnbuckets * sizeof(dlist_head));
+
+	if (cp->cc_type == CC_REGULAR || cp->cc_type == CC_SHAREDLOCAL)
+		Context = CacheMemoryContext;
+	else /* CC_SHAREDGLOBAL */
+	{
+		/*
+		 * CurrentContext is assumed to be transient so that it will be
+		 * cleaned up on error.
+		 */
+		TempGlobalCatCacheContext = MemoryContextClone(GlobalCatCacheContext,
+													   CurrentMemoryContext);
+		Context = TempGlobalCatCacheContext;
+	}
+
+	newbucket = (dlist_head *) MemoryContextAllocZero(Context,
+													  newnbuckets *
+													  sizeof(dlist_head));
+
+
 
 	/* Move all entries from old hash table to new. */
 	for (i = 0; i < cp->cc_nbuckets; i++)
@@ -889,6 +1352,9 @@ RehashCatCache(CatCache *cp)
 	pfree(cp->cc_bucket);
 	cp->cc_nbuckets = newnbuckets;
 	cp->cc_bucket = newbucket;
+
+	if (cp->cc_type == CC_SHAREDGLOBAL)
+		ShmRetailContextMoveChunk(TempGlobalCatCacheContext, GlobalCatCacheContext);
 }
 
 /*
@@ -922,102 +1388,26 @@ do { \
 static void
 CatalogCacheInitializeCache(CatCache *cache)
 {
-	Relation	relation;
-	MemoryContext oldcxt;
-	TupleDesc	tupdesc;
-	int			i;
+	GlobalCatCacheMapKey key;
 
 	CatalogCacheInitializeCache_DEBUG1;
 
-	relation = table_open(cache->cc_reloid, AccessShareLock);
-
-	/*
-	 * switch to the cache context so our allocations do not vanish at the end
-	 * of a transaction
-	 */
-	Assert(CacheMemoryContext != NULL);
-
-	oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
+	if (cache->cc_type == CC_REGULAR)
+		InitLocalCatCacheInternal(cache);
+	else
+	{
+		Assert(cache->cc_type == CC_SHAREDLOCAL);
 
-	/*
-	 * copy the relcache's tuple descriptor to permanent cache storage
-	 */
-	tupdesc = CreateTupleDescCopyConstr(RelationGetDescr(relation));
+		/* Fill in local CatCache members */
+		InitLocalCatCacheInternal(cache);
 
-	/*
-	 * save the relation's name and relisshared flag, too (cc_relname is used
-	 * only for debugging purposes)
-	 */
-	cache->cc_relname = pstrdup(RelationGetRelationName(relation));
-	cache->cc_relisshared = RelationGetForm(relation)->relisshared;
-
-	/*
-	 * return to the caller's memory context and close the rel
-	 */
-	MemoryContextSwitchTo(oldcxt);
-
-	table_close(relation, AccessShareLock);
-
-	CACHE_elog(DEBUG2, "CatalogCacheInitializeCache: %s, %d keys",
-			   cache->cc_relname, cache->cc_nkeys);
-
-	/*
-	 * initialize cache's key information
-	 */
-	for (i = 0; i < cache->cc_nkeys; ++i)
-	{
-		Oid			keytype;
-		RegProcedure eqfunc;
-
-		CatalogCacheInitializeCache_DEBUG2;
-
-		if (cache->cc_keyno[i] > 0)
-		{
-			Form_pg_attribute attr = TupleDescAttr(tupdesc,
-												   cache->cc_keyno[i] - 1);
-
-			keytype = attr->atttypid;
-			/* cache key columns should always be NOT NULL */
-			Assert(attr->attnotnull);
-		}
-		else
-		{
-			if (cache->cc_keyno[i] < 0)
-				elog(FATAL, "sys attributes are not supported in caches");
-			keytype = OIDOID;
-		}
-
-		GetCCHashEqFuncs(keytype,
-						 &cache->cc_hashfunc[i],
-						 &eqfunc,
-						 &cache->cc_fastequal[i]);
-
-		/*
-		 * Do equality-function lookup (we assume this won't need a catalog
-		 * lookup for any supported type)
-		 */
-		fmgr_info_cxt(eqfunc,
-					  &cache->cc_skey[i].sk_func,
-					  CacheMemoryContext);
-
-		/* Initialize sk_attno suitably for HeapKeyTest() and heap scans */
-		cache->cc_skey[i].sk_attno = cache->cc_keyno[i];
-
-		/* Fill in sk_strategy as well --- always standard equality */
-		cache->cc_skey[i].sk_strategy = BTEqualStrategyNumber;
-		cache->cc_skey[i].sk_subtype = InvalidOid;
-		/* If a catcache key requires a collation, it must be C collation */
-		cache->cc_skey[i].sk_collation = C_COLLATION_OID;
-
-		CACHE_elog(DEBUG2, "CatalogCacheInitializeCache %s %d %p",
-				   cache->cc_relname, i, cache);
-	}
-
-	/*
-	 * mark this cache fully initialized
-	 */
-	cache->cc_tupdesc = tupdesc;
-}
+		/* Attach to global CatCache. If not exist, initialize it */
+		GlobalCatCacheMapInitKey(cache, &key);
+		cache->gcp = AttachGlobalCatCache(&key);
+		if (!cache->gcp)
+			InitGlobalCatCache(cache, &key);
+	}
+}
 
 /*
  * InitCatCachePhase2 -- external interface for CatalogCacheInitializeCache
@@ -1195,7 +1585,7 @@ SearchCatCache4(CatCache *cache,
  * Work-horse for SearchCatCache/SearchCatCacheN.
  */
 static inline HeapTuple
-SearchCatCacheInternal(CatCache *cache,
+SearchCatCacheInternal(CatCache *local_cache,
 					   int nkeys,
 					   Datum v1,
 					   Datum v2,
@@ -1204,24 +1594,27 @@ SearchCatCacheInternal(CatCache *cache,
 {
 	Datum		arguments[CATCACHE_MAXKEYS];
 	uint32		hashValue;
-	Index		hashIndex;
-	dlist_iter	iter;
-	dlist_head *bucket;
-	CatCTup    *ct;
+	Index		local_hashIndex;
+	Index 		global_hashIndex;
+	CatCache *global_cache;
+	CatCTup    *local_ct;
+	CatCTup	   *global_ct;
 
 	/* Make sure we're in an xact, even if this ends up being a cache hit */
 	Assert(IsTransactionState());
 
-	Assert(cache->cc_nkeys == nkeys);
+	Assert(local_cache->cc_nkeys == nkeys);
+
+	Assert(local_cache->cc_type != CC_SHAREDGLOBAL);
 
 	/*
 	 * one-time startup overhead for each cache
 	 */
-	if (unlikely(cache->cc_tupdesc == NULL))
-		CatalogCacheInitializeCache(cache);
+	if (unlikely(local_cache->cc_tupdesc == NULL))
+		CatalogCacheInitializeCache(local_cache);
 
 #ifdef CATCACHE_STATS
-	cache->cc_searches++;
+	local_cache->cc_searches++;
 #endif
 
 	/* Initialize local parameter array */
@@ -1233,70 +1626,108 @@ SearchCatCacheInternal(CatCache *cache,
 	/*
 	 * find the hash bucket in which to look for the tuple
 	 */
-	hashValue = CatalogCacheComputeHashValue(cache, nkeys, v1, v2, v3, v4);
-	hashIndex = HASH_INDEX(hashValue, cache->cc_nbuckets);
+	hashValue = CatalogCacheComputeHashValue(local_cache, nkeys,
+											 v1, v2, v3, v4);
+	local_hashIndex = HASH_INDEX(hashValue, local_cache->cc_nbuckets);
+
+	/* Search local hash table. If not found, local_ct is NULL */
+	local_ct = SearchCatCacheBucket(local_cache, nkeys, arguments,
+									hashValue, local_hashIndex, NULL);
 
 	/*
-	 * scan the hash bucket until we find a match or exhaust our tuples
-	 *
-	 * Note: it's okay to use dlist_foreach here, even though we modify the
-	 * dlist within the loop, because we don't continue the loop afterwards.
+	 * In case of traditional CatCache, if entry is not found locally,
+	 * consult the actual catalog.
 	 */
-	bucket = &cache->cc_bucket[hashIndex];
-	dlist_foreach(iter, bucket)
+	if (local_cache->cc_type == CC_REGULAR)
 	{
-		ct = dlist_container(CatCTup, cache_elem, iter.cur);
-
-		if (ct->dead)
-			continue;			/* ignore dead entries */
+		/* if ct is found but negative, return NULL */
+		if (local_ct)
+			return !local_ct->negative ? &local_ct->tuple : NULL;
+		else
+			return SearchCatCacheMiss(local_cache, nkeys, hashValue,
+									  local_hashIndex, v1, v2, v3, v4);
+	}
+	else if (local_cache->cc_type == CC_SHAREDLOCAL)
+	{
+		if (local_ct)
+		{
+			CatCTup *temp_ct;
+			LocalCatCTup *lct = (LocalCatCTup *)local_ct;
 
-		if (ct->hash_value != hashValue)
-			continue;			/* quickly skip entry if wrong hash val */
+			if (CATC_IS_INVALID_DUMMY(lct))
+				return SearchCatCacheMiss(local_cache, nkeys, hashValue,
+										  local_hashIndex, v1, v2, v3, v4);
 
-		if (!CatalogCacheCompareTuple(cache, nkeys, ct->keys, arguments))
-			continue;
 
-		/*
-		 * We found a match in the cache.  Move it to the front of the list
-		 * for its hashbucket, in order to speed subsequent searches.  (The
-		 * most frequently accessed elements in any hashbucket will tend to be
-		 * near the front of the hashbucket's list.)
-		 */
-		dlist_move_head(bucket, &ct->cache_elem);
+			/* Get global cache or local uncommitted cache */
+			temp_ct = LocalCacheGetValidEntry(lct);
 
-		/*
-		 * If it's a positive entry, bump its refcount and return it. If it's
-		 * negative, we can report failure to the caller.
-		 */
-		if (!ct->negative)
+			/* if it's valid, early return, otherwise seach global cache */
+			if (temp_ct)
+				return !temp_ct->negative ? &temp_ct->tuple : NULL;
+		}
+		else
 		{
-			ResourceOwnerEnlargeCatCacheRefs(CurrentResourceOwner);
-			ct->refcount++;
-			ResourceOwnerRememberCatCacheRef(CurrentResourceOwner, &ct->tuple);
+			/*
+			 * Cache entry is not found in local hash table. This could be
+			 * because all caches are invalidated by CatalogCacheFlushCatalog.
+			 * Since system catalog is updated in this case, global cache is
+			 * outdated for this transaction.
+			 */
+			if (local_cache->cc_skip_global_subxid != InvalidSubTransactionId)
+				return SearchCatCacheMiss(local_cache, nkeys, hashValue,
+										  local_hashIndex, v1, v2, v3, v4);
 
-			CACHE_elog(DEBUG2, "SearchCatCache(%s): found in bucket %d",
-					   cache->cc_relname, hashIndex);
+		}
+	}
 
-#ifdef CATCACHE_STATS
-			cache->cc_hits++;
-#endif
+	CACHE_elog(DEBUG2, "try to search global cache entry");
+	/*
+	 * Now valid local cache entry is not obtained. Let's search the global hash
+	 * bucket. hashIndex is generally diffrent from local one because of hash
+	 * table expansion.
+	 */
+	global_cache = (CatCache *)local_cache->gcp;
+	global_hashIndex = HASH_INDEX(hashValue, global_cache->cc_nbuckets);
 
-			return &ct->tuple;
-		}
-		else
-		{
-			CACHE_elog(DEBUG2, "SearchCatCache(%s): found neg entry in bucket %d",
-					   cache->cc_relname, hashIndex);
+	/* global cache refcount is bumped if found */
+	global_ct = SearchCatCacheBucket(global_cache,
+									 nkeys, arguments,
+									 hashValue, global_hashIndex,
+									 local_cache->cc_fastequal);
 
-#ifdef CATCACHE_STATS
-			cache->cc_neg_hits++;
-#endif
 
-			return NULL;
-		}
+	/*
+	 * If global entry is found, register it to local table. If not,
+	 * consult actual system catalog and register it to local table.
+	 */
+	if (global_ct)
+	{
+		Handle *handle;
+
+		/*
+		 * Create local cache entry based on global cache entry. Refcount of
+		 * local entry is not bumped because it doesn't have tuple which
+		 * refcount would protect.
+		 */
+		handle = ((GlobalCatCTup *)global_ct)->gct_handle;
+
+		local_ct = CatalogCacheCreateEntry(local_cache,
+										   &global_ct->tuple, arguments,
+										   hashValue, local_hashIndex,
+										   global_ct->negative,
+										   handle);
+
+		/* There is no negative entry in GlobalCatCache */
+		return &global_ct->tuple;
 	}
+	else
+	{
+		/* Create local entry and also global entry if needed */
+		return SearchCatCacheMiss(local_cache, nkeys, hashValue,
+								  local_hashIndex, v1, v2, v3, v4);
 
-	return SearchCatCacheMiss(cache, nkeys, hashValue, hashIndex, v1, v2, v3, v4);
+	}
 }
 
 /*
@@ -1320,8 +1751,14 @@ SearchCatCacheMiss(CatCache *cache,
 	Relation	relation;
 	SysScanDesc scandesc;
 	HeapTuple	ntp;
-	CatCTup    *ct;
+	CatCTup    *local_ct;
 	Datum		arguments[CATCACHE_MAXKEYS];
+	GlobalCatCTup *gct = NULL;
+	Index global_hashIndex;
+	Handle* handle;
+	HeapTuple tuple = NULL;
+
+	Assert(cache->cc_type != CC_SHAREDGLOBAL);
 
 	/* Initialize local parameter array */
 	arguments[0] = v1;
@@ -1363,17 +1800,58 @@ SearchCatCacheMiss(CatCache *cache,
 								  nkeys,
 								  cur_skey);
 
-	ct = NULL;
+	local_ct = NULL;
 
 	while (HeapTupleIsValid(ntp = systable_getnext(scandesc)))
 	{
-		ct = CatalogCacheCreateEntry(cache, ntp, arguments,
-									 hashValue, hashIndex,
-									 false);
-		/* immediately set the refcount to 1 */
-		ResourceOwnerEnlargeCatCacheRefs(CurrentResourceOwner);
-		ct->refcount++;
-		ResourceOwnerRememberCatCacheRef(CurrentResourceOwner, &ct->tuple);
+		if (cache->cc_type == CC_REGULAR)
+			local_ct = CatalogCacheCreateEntry(cache, ntp, arguments,
+											   hashValue, hashIndex,
+											   false, NULL);
+		else if (cache->cc_type == CC_SHAREDLOCAL)
+		{
+			/* create global cache entry first then local one */
+
+			global_hashIndex = HASH_INDEX(hashValue,
+										  ((CatCache *)cache->gcp)->cc_nbuckets);
+
+			/* set the refcount of global cache entry to 1 here */
+			gct = (GlobalCatCTup *)
+				CatalogCacheCreateEntry((CatCache *)cache->gcp, ntp, arguments,
+										hashValue, global_hashIndex,
+										false, NULL);
+
+			/* If tuple is uncommitted, it's not located in GlobalCatCache */
+			if (gct)
+				handle = gct->gct_handle;
+			else
+				handle = NULL;
+
+			local_ct = CatalogCacheCreateEntry(cache, ntp, arguments,
+											   hashValue, hashIndex,
+											   false, handle);
+
+		}
+		else
+			elog(FATAL, "The target of CatCacheSearchMiss is not SHAREDGLOBAL");
+
+		/*
+		 * If local cache entry is a pointer to global entry, keep refcount to
+		 * zero because there is no need to protect tuple which doesn't exist.
+		 */
+		if (cache->cc_type == CC_REGULAR
+			||(cache->cc_type == CC_SHAREDLOCAL &&
+			   !((LocalCatCTup *)local_ct)->lct_committed))
+		{
+			/* immediately set the local refcount to 1 */
+			ResourceOwnerEnlargeCatCacheRefs(CurrentResourceOwner);
+			local_ct->refcount++;
+			ResourceOwnerRememberCatCacheRef(CurrentResourceOwner, &local_ct->tuple);
+
+			CACHE_elog(DEBUG2, "SearchCatCacheMiss local refcount is bumped: %s, %d tuple %p",
+					   local_ct->my_cache->cc_relname, local_ct->refcount,
+					   &local_ct->tuple);
+		}
 		break;					/* assume only one match */
 	}
 
@@ -1391,14 +1869,14 @@ SearchCatCacheMiss(CatCache *cache,
 	 * gets created later.  (Bootstrap doesn't do UPDATEs, so it doesn't need
 	 * cache inval for that.)
 	 */
-	if (ct == NULL)
+	if (local_ct == NULL)
 	{
 		if (IsBootstrapProcessingMode())
 			return NULL;
 
-		ct = CatalogCacheCreateEntry(cache, NULL, arguments,
-									 hashValue, hashIndex,
-									 true);
+		local_ct = CatalogCacheCreateEntry(cache, NULL, arguments,
+										   hashValue, hashIndex,
+										   true, NULL);
 
 		CACHE_elog(DEBUG2, "SearchCatCache(%s): Contains %d/%d tuples",
 				   cache->cc_relname, cache->cc_ntup, CacheHdr->ch_ntup);
@@ -1422,7 +1900,18 @@ SearchCatCacheMiss(CatCache *cache,
 	cache->cc_newloads++;
 #endif
 
-	return &ct->tuple;
+
+	if (cache->cc_type == CC_REGULAR)
+		tuple = &local_ct->tuple;
+	else if (cache->cc_type == CC_SHAREDLOCAL)
+	{
+		if (gct)
+			tuple = &((CatCTup *)gct)->tuple;
+		else
+			tuple = &local_ct->tuple;
+	}
+
+	return tuple;
 }
 
 /*
@@ -1439,6 +1928,7 @@ SearchCatCacheMiss(CatCache *cache,
 void
 ReleaseCatCache(HeapTuple tuple)
 {
+	CatCache *cache;
 	CatCTup    *ct = (CatCTup *) (((char *) tuple) -
 								  offsetof(CatCTup, tuple));
 
@@ -1446,8 +1936,21 @@ ReleaseCatCache(HeapTuple tuple)
 	Assert(ct->ct_magic == CT_MAGIC);
 	Assert(ct->refcount > 0);
 
-	ct->refcount--;
-	ResourceOwnerForgetCatCacheRef(CurrentResourceOwner, &ct->tuple);
+	cache = ct->my_cache;
+
+	if (cache->cc_type == CC_REGULAR ||
+		(cache->cc_type == CC_SHAREDLOCAL &&
+		 !((LocalCatCTup *)ct)->lct_committed))
+	{
+		/*
+		 * CC_REGULAR and CC_SHAREDLOCAL case.
+		 * Note that CC_SHAREDLOCAL might have uncommitted tuple locally
+		 */
+		ct->refcount--;
+		ResourceOwnerForgetCatCacheRef(CurrentResourceOwner, &ct->tuple);
+	}
+	else if (cache->cc_type == CC_SHAREDGLOBAL)
+		GlobalCatCTupDecrementAndRemove((GlobalCatCache *)cache, (GlobalCatCTup *)ct);
 
 	if (
 #ifndef CATCACHE_FORCE_RELEASE
@@ -1455,7 +1958,7 @@ ReleaseCatCache(HeapTuple tuple)
 #endif
 		ct->refcount == 0 &&
 		(ct->c_list == NULL || ct->c_list->refcount == 0))
-		CatCacheRemoveCTup(ct->my_cache, ct);
+		CatCacheRemoveCTup(cache, ct);
 }
 
 
@@ -1511,16 +2014,20 @@ SearchCatCacheList(CatCache *cache,
 {
 	Datum		v4 = 0;			/* dummy last-column value */
 	Datum		arguments[CATCACHE_MAXKEYS];
+	Datum		lct_arguments[CATCACHE_MAXKEYS];
 	uint32		lHashValue;
 	dlist_iter	iter;
 	CatCList   *cl;
 	CatCTup    *ct;
+	CatCTup    *local_ct;
+	LocalCatCTup *lct;
 	List	   *volatile ctlist;
 	ListCell   *ctlist_item;
 	int			nmembers;
 	bool		ordered;
 	HeapTuple	ntp;
 	MemoryContext oldcxt;
+	const CCFastEqualFN *cc_fastequal;
 	int			i;
 
 	/*
@@ -1570,10 +2077,60 @@ SearchCatCacheList(CatCache *cache,
 		if (cl->nkeys != nkeys)
 			continue;
 
-		if (!CatalogCacheCompareTuple(cache, nkeys, cl->keys, arguments))
+		cc_fastequal =  cache->cc_fastequal;
+		if (!CatalogCacheCompareTuple(cc_fastequal, nkeys, cl->keys, arguments))
 			continue;
 
 		/*
+		 * Cache entries may be stale. If there's invalid entry, the entry and
+		 * its list are removed or marked as dead. Moreover, we need to decrease
+		 * the refcounts of valid entries, which are already bumped.
+		 */
+		if (cache->cc_type == CC_SHAREDLOCAL)
+		{
+			CatCTup *temp_ct;
+
+			for (i = 0; i < cl->n_members; i++)
+			{
+				/* Note global refcount is raised if it's valid */
+				temp_ct = LocalCacheGetValidEntry(cl->shared_local_members[i]);
+
+				/* If it's valid, go through. If not, do clean up. */
+				if (!temp_ct)
+				{
+					LocalCatCTup *temp_lct;
+					GlobalCatCTup *temp_gct;
+
+					for (--i ; i >= 0; i--)
+					{
+						temp_lct = cl->shared_local_members[i];
+
+						/* Refcount of local entry is not increased */
+						if (!temp_lct->lct_committed || ((CatCTup *)temp_lct)->negative)
+							continue;
+
+						temp_gct = temp_lct->lct_handle->entry;
+
+						GlobalCatCTupDecrementAndRemove((GlobalCatCache *)(((CatCTup *)temp_gct)->my_cache),
+														temp_gct);
+					}
+
+					/* cl is not valid eitehr */
+					if (cl->refcount > 0)
+						cl->dead = true;
+					else
+						CatCacheRemoveCList(cache, cl);
+
+					break;
+				}
+			}
+
+			/* if last temp_ct is not valid, break dlist_foreach loop */
+			if (!temp_ct)
+				break;
+		}
+
+		/*
 		 * We found a matching list.  Move the list to the front of the
 		 * cache's list-of-lists, to speed subsequent searches.  (We do not
 		 * move the members to the fronts of their hashbucket lists, however,
@@ -1649,29 +2206,66 @@ SearchCatCacheList(CatCache *cache,
 			/*
 			 * See if there's an entry for this tuple already.
 			 */
-			ct = NULL;
-			hashValue = CatalogCacheComputeTupleHashValue(cache, cache->cc_nkeys, ntp);
+			local_ct = NULL;
+			hashValue = CatalogCacheComputeTupleHashValue(cache,
+														  cache->cc_nkeys, ntp);
 			hashIndex = HASH_INDEX(hashValue, cache->cc_nbuckets);
 
 			bucket = &cache->cc_bucket[hashIndex];
+
 			dlist_foreach(iter, bucket)
 			{
-				ct = dlist_container(CatCTup, cache_elem, iter.cur);
+				local_ct = dlist_container(CatCTup, cache_elem, iter.cur);
 
-				if (ct->dead || ct->negative)
+				if (local_ct->dead || local_ct->negative)
 					continue;	/* ignore dead and negative entries */
 
-				if (ct->hash_value != hashValue)
+				if (local_ct->hash_value != hashValue)
 					continue;	/* quickly skip entry if wrong hash val */
 
-				if (!ItemPointerEquals(&(ct->tuple.t_self), &(ntp->t_self)))
-					continue;	/* not same tuple */
+				if (cache->cc_type == CC_REGULAR)
+				{
+					if (!ItemPointerEquals(&(local_ct->tuple.t_self), &(ntp->t_self)))
+						continue;	/* not same tuple */
+				}
+				else if (cache->cc_type == CC_SHAREDLOCAL)
+				{
+					/*
+					 * ct is global cache entry or local uncommitted entry.
+					 * Refcount of global cache entry is bumped if it's valid.
+					 * If ct is NULL, ct is not valid and removed or marked
+					 * as dead.
+					 */
+					ct = LocalCacheGetValidEntry((LocalCatCTup *)local_ct);
+
+					if (CATC_IS_INVALID_DUMMY((LocalCatCTup *)local_ct))
+						continue;
+
+					if (ct)
+					{
+						if (!ItemPointerEquals(&(ct->tuple.t_self),
+											   &(ntp->t_self)))
+						{
+
+							/* Release refcount of unmatched global entry */
+							if (ct->my_cache->cc_type == CC_SHAREDGLOBAL)
+								GlobalCatCTupDecrementAndRemove((GlobalCatCache *)ct->my_cache,
+																(GlobalCatCTup *)ct);
+
+								continue;	/* not same tuple */
+						}
+					}
+					else
+						continue;
+				}
+				else
+					elog(FATAL, "CatCList is not located in shared memory");
 
 				/*
 				 * Found a match, but can't use it if it belongs to another
 				 * list already
 				 */
-				if (ct->c_list)
+				if (local_ct->c_list)
 					continue;
 
 				found = true;
@@ -1680,16 +2274,74 @@ SearchCatCacheList(CatCache *cache,
 
 			if (!found)
 			{
-				/* We didn't find a usable entry, so make a new one */
-				ct = CatalogCacheCreateEntry(cache, ntp, arguments,
-											 hashValue, hashIndex,
-											 false);
+				Index global_hashIndex;
+				GlobalCatCTup *gct;
+				Handle* handle;
+
+				if (cache->cc_type == CC_REGULAR)
+				{
+					/* We didn't find a usable entry, so make a new one */
+					local_ct = CatalogCacheCreateEntry(cache, ntp, arguments,
+													   hashValue, hashIndex,
+													   false, NULL);
+				}
+				else if (cache->cc_type == CC_SHAREDLOCAL)
+				{
+					global_hashIndex = HASH_INDEX(hashValue,
+												  ((CatCache *)cache->gcp)->
+												  cc_nbuckets);
+
+					gct = (GlobalCatCTup *)
+						CatalogCacheCreateEntry((CatCache *)cache->gcp,
+												ntp, arguments,
+												hashValue, global_hashIndex,
+												false, NULL);
+
+					/*
+					 * If tuple is uncommitted, it's not in GlobalCatCache
+					 * and ct has tuple locally.
+					 */
+					if (gct)
+						handle = gct->gct_handle;
+					else
+						handle = NULL;
+
+					/*
+					 * Local cache entry stores keys but arguments are partial
+					 * key and committed local cache entry doen't
+					 * have heapTuple. So keys cannot point to heapTuple.
+					 * So fill in arguments fully here and copy them.
+					 */
+					for (i = 0; i < nkeys; i++)
+						lct_arguments[i] = arguments[i];
+
+					for (i = nkeys; i < cache->cc_nkeys; i++)
+					{
+						Datum		atp;
+						bool		isnull;
+
+						atp = heap_getattr(ntp,
+										   cache->cc_keyno[i],
+										   cache->cc_tupdesc,
+										   &isnull);
+						Assert(!isnull);
+						lct_arguments[i] = atp;
+					}
+
+					local_ct = CatalogCacheCreateEntry(cache, ntp, lct_arguments,
+													   hashValue, hashIndex,
+													   false, handle);
+
+				}
 			}
 
-			/* Careful here: add entry to ctlist, then bump its refcount */
-			/* This way leaves state correct if lappend runs out of memory */
-			ctlist = lappend(ctlist, ct);
-			ct->refcount++;
+			/* Careful here: add entry to ctlist, then bump its refcount
+			 * This way leaves state correct if lappend runs out of memory.
+			 * The refcount of global cache entry is already bumped in
+			 * LocalCacheGetValidEntry.
+			 */
+			ctlist = lappend(ctlist, local_ct);
+			local_ct->refcount++;
 		}
 
 		systable_endscan(scandesc);
@@ -1702,6 +2354,12 @@ SearchCatCacheList(CatCache *cache,
 		cl = (CatCList *)
 			palloc(offsetof(CatCList, members) + nmembers * sizeof(CatCTup *));
 
+		if (cache->cc_type == CC_REGULAR)
+			cl->shared_local_members = NULL;
+		else if (cache->cc_type == CC_SHAREDLOCAL)
+			cl->shared_local_members = (LocalCatCTup **)
+				palloc(nmembers * sizeof(LocalCatCTup *));
+
 		/* Extract key values */
 		CatCacheCopyKeys(cache->cc_tupdesc, nkeys, cache->cc_keyno,
 						 arguments, cl->keys);
@@ -1718,19 +2376,37 @@ SearchCatCacheList(CatCache *cache,
 	}
 	PG_CATCH();
 	{
-		foreach(ctlist_item, ctlist)
+		GlobalCatCTup *gct;
+		GlobalCatCache *gcp;
+
+		/* Release refernce of local and global cache entry if it exists */
+		if (cache->cc_type == CC_REGULAR)
+			ReleaseLocalCtlistItem(cache, ctlist);
+		else if (cache->cc_type == CC_SHAREDLOCAL)
 		{
-			ct = (CatCTup *) lfirst(ctlist_item);
-			Assert(ct->c_list == NULL);
-			Assert(ct->refcount > 0);
-			ct->refcount--;
-			if (
-#ifndef CATCACHE_FORCE_RELEASE
-				ct->dead &&
-#endif
-				ct->refcount == 0 &&
-				(ct->c_list == NULL || ct->c_list->refcount == 0))
-				CatCacheRemoveCTup(cache, ct);
+			/*
+			 * Release global cache entry. If it's uncommitted entry release
+			 * local refcount.
+			 */
+			foreach(ctlist_item, ctlist)
+			{
+				ct = (CatCTup *) lfirst(ctlist_item);
+
+				Assert(ct->c_list == NULL);
+				Assert(ct->refcount > 0);
+
+				lct = (LocalCatCTup *)ct;
+
+				if (lct->lct_committed)
+				{
+					gct = lct->lct_handle->entry;
+					gcp = (GlobalCatCache *)(((CatCTup *)gct)->my_cache);
+
+					GlobalCatCTupDecrementAndRemove(gcp, gct);
+				}
+				else
+					ReleaseLocalCtlistItem(cache, ctlist);
+			}
 		}
 
 		PG_RE_THROW();
@@ -1746,20 +2422,57 @@ SearchCatCacheList(CatCache *cache,
 	cl->hash_value = lHashValue;
 	cl->n_members = nmembers;
 
-	i = 0;
-	foreach(ctlist_item, ctlist)
+
+	if (cache->cc_type == CC_REGULAR)
 	{
-		cl->members[i++] = ct = (CatCTup *) lfirst(ctlist_item);
-		Assert(ct->c_list == NULL);
-		ct->c_list = cl;
-		/* release the temporary refcount on the member */
-		Assert(ct->refcount > 0);
-		ct->refcount--;
-		/* mark list dead if any members already dead */
-		if (ct->dead)
-			cl->dead = true;
+		i = 0;
+		foreach(ctlist_item, ctlist)
+		{
+			cl->members[i++] = ct = (CatCTup *) lfirst(ctlist_item);
+			Assert(ct->c_list == NULL);
+			ct->c_list = cl;
+			/* release the temporary refcount on the member */
+			Assert(ct->refcount > 0);
+			ct->refcount--;
+			/* mark list dead if any members already dead */
+			if (ct->dead)
+				cl->dead = true;
+		}
+		Assert(i == nmembers);
+	}
+	else if (cache->cc_type == CC_SHAREDLOCAL)
+	{
+		/* Set local cache entries to the list */
+		i = 0;
+		foreach(ctlist_item, ctlist)
+		{
+			ct = lfirst(ctlist_item);
+			cl->shared_local_members[i] = lct = (LocalCatCTup *)ct;
+
+			Assert(ct->c_list == NULL);
+			ct->c_list = cl;
+
+			/* release the temporary refcount on the member */
+			Assert(ct->refcount > 0);
+			ct->refcount--;
+
+			/* mark list dead if any members already dead */
+			if (ct->dead)
+				cl->dead = true;
+
+			/* Set global cache entries or uncommitted local one to the list */
+			if (lct->lct_committed)
+			{
+				cl->members[i] = (CatCTup *)lct->lct_handle->entry;
+				Assert(cl->members[i]->refcount > 0);
+			}
+			else
+				cl->members[i] = ct;
+
+			i++;
+		}
+		Assert(i == nmembers);
 	}
-	Assert(i == nmembers);
 
 	dlist_push_head(&cache->cc_lists, &cl->cache_elem);
 
@@ -1767,13 +2480,42 @@ SearchCatCacheList(CatCache *cache,
 	cl->refcount++;
 	ResourceOwnerRememberCatCacheListRef(CurrentResourceOwner, cl);
 
-	CACHE_elog(DEBUG2, "SearchCatCacheList(%s): made list of %d members",
+	CACHE_elog(DEBUG2, "SearchCatCacheList (%s): made list of %d members",
 			   cache->cc_relname, nmembers);
 
 	return cl;
 }
 
 /*
+ *	ReleaseLocalCtlistItem
+ *
+ *  Workhorse of SearchCatCacheList.
+ *	Decrement the temporarily raised reference count of each ctlist item.
+ */
+static void
+ReleaseLocalCtlistItem(CatCache *cache, List *ctlist)
+{
+	CatCTup		*ct;
+	ListCell 	*ctlist_item;
+
+	foreach(ctlist_item, ctlist)
+	{
+		ct = (CatCTup *) lfirst(ctlist_item);
+		Assert(ct->c_list == NULL);
+		Assert(ct->refcount > 0);
+		ct->refcount--;
+		if (
+#ifndef CATCACHE_FORCE_RELEASE
+			ct->dead &&
+#endif
+			ct->refcount == 0 &&
+			(ct->c_list == NULL || ct->c_list->refcount == 0))
+			CatCacheRemoveCTup(cache, ct);
+
+	}
+}
+
+/*
  *	ReleaseCatCacheList
  *
  *	Decrement the reference count of a catcache list.
@@ -1787,6 +2529,28 @@ ReleaseCatCacheList(CatCList *list)
 	list->refcount--;
 	ResourceOwnerForgetCatCacheListRef(CurrentResourceOwner, list);
 
+	/*
+	 * Decrement the reference count of global cache entry.
+	 * members[] points to global entry but cannot be cast to GlobalCatCTup
+	 * so traverse from shared_local_members
+	 */
+	if (list->shared_local_members)
+	{
+		GlobalCatCTup *gct;
+		GlobalCatCache *gcp;
+
+		for (int i = 0; i < list->n_members; i++)
+		{
+			if (!list->shared_local_members[i]->lct_committed)
+				continue;
+
+			gct = (GlobalCatCTup *)list->members[i];
+			gcp = (GlobalCatCache *)list->members[i]->my_cache;
+
+			GlobalCatCTupDecrementAndRemove(gcp, gct);
+		}
+	}
+
 	if (
 #ifndef CATCACHE_FORCE_RELEASE
 		list->dead &&
@@ -1804,77 +2568,126 @@ ReleaseCatCacheList(CatCList *list)
 static CatCTup *
 CatalogCacheCreateEntry(CatCache *cache, HeapTuple ntp, Datum *arguments,
 						uint32 hashValue, Index hashIndex,
-						bool negative)
+						bool negative, Handle* handle)
 {
 	CatCTup    *ct;
-	HeapTuple	dtp;
-	MemoryContext oldcxt;
+	MemoryContext oldcxt = NULL;
+	MemoryContext TempGlobalCatCacheContext = NULL;
+	LWLock *bucket_lock = NULL;
+	int i;
+	LocalCatCTup *lct;
+	GlobalCatCTup *gct;
+
 
-	/* negative entries have no tuple associated */
-	if (ntp)
+	/* Allocate memory for CatCTup and the cached tuple in one go */
+	if (cache->cc_type == CC_REGULAR || cache->cc_type == CC_SHAREDLOCAL)
+		oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
+	else if (cache->cc_type == CC_SHAREDGLOBAL)
 	{
-		int			i;
+			TempGlobalCatCacheContext = MemoryContextClone(GlobalCatCacheContext,
+														   CurrentMemoryContext);
 
-		Assert(!negative);
+		oldcxt = MemoryContextSwitchTo(TempGlobalCatCacheContext);
+	}
 
-		/*
-		 * If there are any out-of-line toasted fields in the tuple, expand
-		 * them in-line.  This saves cycles during later use of the catcache
-		 * entry, and also protects us against the possibility of the toast
-		 * tuples being freed before we attempt to fetch them, in case of
-		 * something using a slightly stale catcache entry.
-		 */
-		if (HeapTupleHasExternal(ntp))
-			dtp = toast_flatten_tuple(ntp, cache->cc_tupdesc);
+	if (cache->cc_type == CC_REGULAR)
+	{
+		if (ntp)
+			ct = CreateCatCTupWithTuple(cache, ntp);
 		else
-			dtp = ntp;
+			ct = CreateCatCTupWithoutTuple(cache, arguments);
+	}
+	else if (cache->cc_type == CC_SHAREDLOCAL)
+	{
+		if (ntp)
+		{
+			/*
+			 * If the tuple is made by its own transaction (not committed
+			 * yet), the entry should not be placed in Global CatCache
+			 * hash table to prevent other process from seeing uncommitted
+			 * cache entry. So just put it in local hash table.
+			 */
+			if (HeapTupleHeaderXminCommitted(ntp->t_data))
+			{
+				/* Create local tuple, which has keys but doesn't tuple */
+				ct = CreateCatCTupWithoutTuple(cache, arguments);
+				lct =  (LocalCatCTup *) ct;
+				lct->lct_committed = true;
+
+				Assert(handle);
+				Assert(!handle->free);
+				lct->lct_handle = handle;
+				lct->lct_generation = handle->generation;
+				lct->lct_skip_global_subxid = InvalidSubTransactionId;
+			}
+			else
+			{
+				/*
+				 * Assume global cache entry and its handle doesn't exist,
+				 * because we get uncommitted tuple.
+				 */
+				Assert(!handle);
 
-		/* Allocate memory for CatCTup and the cached tuple in one go */
-		oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
+				ct = CreateCatCTupWithTuple(cache, ntp);
+				lct = (LocalCatCTup *) ct;
 
-		ct = (CatCTup *) palloc(sizeof(CatCTup) +
-								MAXIMUM_ALIGNOF + dtp->t_len);
-		ct->tuple.t_len = dtp->t_len;
-		ct->tuple.t_self = dtp->t_self;
-		ct->tuple.t_tableOid = dtp->t_tableOid;
-		ct->tuple.t_data = (HeapTupleHeader)
-			MAXALIGN(((char *) ct) + sizeof(CatCTup));
-		/* copy tuple contents */
-		memcpy((char *) ct->tuple.t_data,
-			   (const char *) dtp->t_data,
-			   dtp->t_len);
-		MemoryContextSwitchTo(oldcxt);
+				lct->lct_committed = false;
+				lct->lct_handle = NULL;
+				lct->lct_generation = 0;
+				lct->lct_skip_global_subxid = InvalidSubTransactionId;
+			}
+		}
+		else
+		{
+			/* Negative cache is not in shared memory */
+			Assert(!handle);
 
-		if (dtp != ntp)
-			heap_freetuple(dtp);
+			/* Negative cache case */
+			ct = CreateCatCTupWithoutTuple(cache, arguments);
+			lct = (LocalCatCTup *) ct;
+
+			lct->lct_committed = false;
+			lct->lct_handle = NULL;
+			lct->lct_generation = 0;
+
+			/* inval_tuple doesn't have information about keys */
+			if (arguments)
+				lct->lct_skip_global_subxid = InvalidSubTransactionId;
+			else
+				lct->lct_skip_global_subxid = GetCurrentSubTransactionId();
 
-		/* extract keys - they'll point into the tuple if not by-value */
-		for (i = 0; i < cache->cc_nkeys; i++)
-		{
-			Datum		atp;
-			bool		isnull;
-
-			atp = heap_getattr(&ct->tuple,
-							   cache->cc_keyno[i],
-							   cache->cc_tupdesc,
-							   &isnull);
-			Assert(!isnull);
-			ct->keys[i] = atp;
 		}
 	}
-	else
+	else /* shared global */
 	{
-		Assert(negative);
-		oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
-		ct = (CatCTup *) palloc(sizeof(CatCTup));
+		if (!ntp)
+			elog(FATAL, "Negative global cache entry is not supported");
 
 		/*
-		 * Store keys - they'll point into separately allocated memory if not
-		 * by-value.
+		 * If the tuple is made by its own transaction (not committed
+		 * yet), the entry should not be placed in Global CatCache
+		 * hash table to prevent other process from seeing uncommitted
+		 * cache entry. So return NULL.
 		 */
-		CatCacheCopyKeys(cache->cc_tupdesc, cache->cc_nkeys, cache->cc_keyno,
-						 arguments, ct->keys);
-		MemoryContextSwitchTo(oldcxt);
+		if (!HeapTupleHeaderXminCommitted(ntp->t_data))
+		{
+			ShmRetailContextMoveChunk(TempGlobalCatCacheContext,
+									  GlobalCatCacheContext);
+
+			MemoryContextSwitchTo(oldcxt);
+			return NULL;
+		}
+
+		ct =  CreateCatCTupWithTuple(cache, ntp);
+
+		gct = (GlobalCatCTup *)ct;
+
+		gct->gct_handle =
+			GetCacheHandle(&((GlobalCatCache *)cache)->gcc_handle_hdr,
+						   (void *)gct);
+
+		/* Initailize lock for GlobalCatCTup */
+		SpinLockInit(&gct->gct_mutex);
 	}
 
 	/*
@@ -1889,18 +2702,82 @@ CatalogCacheCreateEntry(CatCache *cache, HeapTuple ntp, Datum *arguments,
 	ct->negative = negative;
 	ct->hash_value = hashValue;
 
+	MemoryContextSwitchTo(oldcxt);
+
+	if (cache->cc_type == CC_SHAREDGLOBAL)
+	{
+		/* bump global refcount as soon as possible. No lock is needed here yet */
+		ct->refcount++;
+
+		CACHE_elog(DEBUG2, "CatalogCacheCreateEntry global refcount is bumped: %s, %d tuple %p",
+				   ct->my_cache->cc_relname, ct->refcount,
+				   &ct->tuple);
+
+		/* Get lock for bucket */
+		bucket_lock = GetCatCacheBucketLock((GlobalCatCache *)cache, hashValue);
+		LWLockAcquire(bucket_lock, LW_EXCLUSIVE);
+		hashIndex = HASH_INDEX(hashValue, cache->cc_nbuckets);
+	}
+
+	/* Now insert entry to hash table */
 	dlist_push_head(&cache->cc_bucket[hashIndex], &ct->cache_elem);
 
+	if (cache->cc_type == CC_SHAREDGLOBAL)
+	{
+		ShmRetailContextMoveChunk(TempGlobalCatCacheContext, GlobalCatCacheContext);
+		LWLockRelease(bucket_lock);
+		SpinLockAcquire(&((GlobalCatCache *)cache)->gcc_mutex);
+	}
+
 	cache->cc_ntup++;
-	CacheHdr->ch_ntup++;
+
+	if (cache->cc_type == CC_SHAREDGLOBAL)
+		SpinLockRelease(&((GlobalCatCache *)cache)->gcc_mutex);
+	else
+		CacheHdr->ch_ntup++;
 
 	/*
 	 * If the hash table has become too full, enlarge the buckets array. Quite
 	 * arbitrarily, we enlarge when fill factor > 2.
 	 */
 	if (cache->cc_ntup > cache->cc_nbuckets * 2)
+	{
+		GlobalCatCache *gcp;
+		LWLock *old_bucket_locks = NULL;
+		int num_locks;
+
+		/* Enlarge bucket locks as well */
+		if (cache->cc_type == CC_SHAREDGLOBAL)
+		{
+			gcp = (GlobalCatCache *) cache;
+
+			/* bucket locks will be changed after rehasing */
+
+			old_bucket_locks = gcp->gcc_bucket_locks;
+			num_locks = MAX_CC_BUCKET_LOCKS;
+
+			for (i = 0; i < num_locks; ++i)
+				LWLockAcquire(&old_bucket_locks[i], LW_EXCLUSIVE);
+
+			/* other process might already enlarge the hash table */
+			if (cache->cc_ntup <= cache->cc_nbuckets * 2)
+			{
+				/* Release locks in reverse order */
+				for (i = num_locks - 1 ; i >= 0; --i)
+					LWLockRelease(&old_bucket_locks[i]);
+			}
+		}
+
 		RehashCatCache(cache);
 
+		if (cache->cc_type == CC_SHAREDGLOBAL)
+		{
+			/* Release locks in reverse order */
+			for (i = num_locks - 1 ; i >= 0; --i)
+				LWLockRelease(&old_bucket_locks[i]);
+		}
+	}
+
 	return ct;
 }
 
@@ -2076,6 +2953,9 @@ PrintCatCacheLeakWarning(HeapTuple tuple)
 	/* Safety check to ensure we were handed a cache entry */
 	Assert(ct->ct_magic == CT_MAGIC);
 
+	/* CC_SHAREDLOCAL is ok if it's uncommitted tuple */
+	Assert(ct->my_cache->cc_type != CC_SHAREDGLOBAL);
+
 	elog(WARNING, "cache reference leak: cache %s (%d), tuple %u/%u has count %d",
 		 ct->my_cache->cc_relname, ct->my_cache->id,
 		 ItemPointerGetBlockNumber(&(tuple->t_self)),
@@ -2090,3 +2970,593 @@ PrintCatCacheListLeakWarning(CatCList *list)
 		 list->my_cache->cc_relname, list->my_cache->id,
 		 list, list->refcount);
 }
+
+/*
+ * Functions for shared catalog cache
+ */
+
+
+/*
+ * Initialize GlobalCatCache.
+ */
+static void
+InitGlobalCatCache(CatCache *local_cache, GlobalCatCacheMapKey *keyPtr)
+{
+	uint32 hashcode;
+	LWLock *map_lock;
+	GlobalCatCache *gcp;
+	GlobalCatCacheMapEnt *entry;
+	MemoryContext oldcxt;
+	MemoryContext TempGlobalCatCacheContext;
+
+	Assert(local_cache->gcp == NULL);
+
+	/* Get hashcode and lock instance */
+	hashcode = GlobalCatCacheMapHashCode(keyPtr);
+	map_lock = GlobalCatCachePartitionLock(hashcode);
+
+	/* Acquire lock for partition of GlobalCatCacheMap */
+	LWLockAcquire(map_lock, LW_EXCLUSIVE);
+
+	/*
+	 * If someone already registered GlobalCatCache to GlobalCatCacheMap,
+	 * attach and return.
+	 */
+	entry = GlobalCatCacheMapLookup(keyPtr, hashcode);
+
+	if (entry)
+	{
+		local_cache->gcp = entry->gcp;
+		LWLockRelease(map_lock);
+		return;
+	}
+
+	/* Allocate memory and initialize GlobalCatCache */
+
+	TempGlobalCatCacheContext = MemoryContextClone(GlobalCatCacheContext,
+												   CurrentMemoryContext);
+
+	oldcxt = MemoryContextSwitchTo(TempGlobalCatCacheContext);
+	gcp = InitGlobalCatCacheInternal(local_cache);
+	MemoryContextSwitchTo(oldcxt);
+	ShmRetailContextMoveChunk(TempGlobalCatCacheContext, GlobalCatCacheContext);
+
+	/* Insert GlobalCatCache to GlobalCatCacheMap */
+	if (!GlobalCatCacheMapInsert(keyPtr, hashcode, gcp))
+		elog(FATAL, "GlobalCatCache oid: %u, cacheid: %d"
+			 "is already registered to GlobalCatCacheMap",
+			 keyPtr->dbId, keyPtr->cacheId);
+
+	/* Now initialization is done */
+	local_cache->gcp = gcp;
+
+	LWLockRelease(map_lock);
+}
+
+/*
+ * Allocate and build GlobalCatCache
+ */
+static GlobalCatCache *
+InitGlobalCatCacheInternal(CatCache *local_cache)
+{
+	GlobalCatCache *gcp;
+	CatCache *global_cache;
+	size_t sz;
+	int num_locks;
+	int i;
+
+	/*
+	 * Allocate memory for GlobalCatCache and hash buckets as well.
+	 */
+	sz = sizeof(GlobalCatCache) + PG_CACHE_LINE_SIZE;
+	gcp = (GlobalCatCache *) CACHELINEALIGN(palloc0(sz));
+	global_cache = (CatCache *)gcp;
+
+	global_cache->cc_nbuckets = local_cache->cc_nbuckets;
+
+	global_cache->cc_bucket =
+		palloc0(global_cache->cc_nbuckets * sizeof(dlist_head));
+
+	/*
+	 * Allocate memory for bucket locks.
+	 * nbuckets of GlobalCatCache is same as local one at first.
+	 */
+	gcp->gcc_bucket_locks =
+		(LWLock *) palloc(MAX_CC_BUCKET_LOCKS * sizeof(LWLock));
+
+	global_cache->id = local_cache->id;
+	global_cache->cc_type = CC_SHAREDGLOBAL;
+	global_cache->cc_relname = pstrdup(local_cache->cc_relname);
+	global_cache->cc_reloid = local_cache->cc_reloid;
+	global_cache->cc_indexoid = local_cache->cc_indexoid;
+	global_cache->cc_relisshared = local_cache->cc_relisshared;
+	global_cache->cc_ntup = 0;
+	global_cache->cc_nkeys = local_cache->cc_nkeys;
+
+	global_cache->cc_tupdesc = CreateTupleDescCopyConstr(local_cache->cc_tupdesc);
+
+	/* following members are not used in GlobalCatCache */
+	for (i = 0; i < local_cache->cc_nkeys; ++i)
+		global_cache->cc_keyno[i] = local_cache->cc_keyno[i];
+
+	/* Initialize lock for GlobalCatCache members */
+	SpinLockInit(&gcp->gcc_mutex);
+
+	/* Initialize lock partitions of hash table */
+	/* if (global_cache->cc_nbuckets > MAX_CC_BUCKET_LOCKS) */
+	/* 	num_locks = MAX_CC_BUCKET_LOCKS; */
+	/* else */
+	/* 	num_locks = global_cache->cc_nbuckets; */
+	num_locks = MAX_CC_BUCKET_LOCKS;
+
+	for (i = 0; i < num_locks ; ++i)
+		LWLockInitialize(&gcp->gcc_bucket_locks[i],
+						 LWTRANCHE_GLOBAL_CATCACHE);
+
+	/* Initailize handler for Global CatCache entries */
+	InitCacheHandle(&gcp->gcc_handle_hdr, (void *)gcp);
+
+	return gcp;
+}
+
+
+/*
+ * Link local CatCache to global CatCache if available.
+ */
+static GlobalCatCache *
+AttachGlobalCatCache(GlobalCatCacheMapKey *keyPtr)
+{
+	uint32 hashcode;
+	LWLock *map_lock;
+	GlobalCatCacheMapEnt* entry;
+
+	hashcode = GlobalCatCacheMapHashCode(keyPtr);
+	map_lock = GlobalCatCachePartitionLock(hashcode);
+
+	LWLockAcquire(map_lock, LW_SHARED);
+	entry = GlobalCatCacheMapLookup(keyPtr, hashcode);
+	LWLockRelease(map_lock);
+
+	if (!entry)
+		return NULL;
+
+	return entry->gcp;
+}
+
+
+/*
+ * Workhorse of CatalogCacheInitializeCache.
+ * This initailize CatCache whose type is CC_REGULAR and CC_SHAREDLOCAL,
+ * consulting relation and filling in actual members of CatCache.
+ */
+static void
+InitLocalCatCacheInternal(CatCache *cache)
+{
+	Relation	relation;
+	MemoryContext oldcxt;
+	TupleDesc	tupdesc;
+	int			i;
+
+	Assert(cache->cc_type != CC_SHAREDGLOBAL);
+
+	relation = table_open(cache->cc_reloid, AccessShareLock);
+
+	/*
+	 * switch to the cache context so our allocations do not vanish
+	 * at the end of a transaction
+	 */
+	Assert(CacheMemoryContext != NULL);
+	oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
+
+	/*
+	 * copy the relcache's tuple descriptor to permanent cache storage
+	 */
+	tupdesc = CreateTupleDescCopyConstr(RelationGetDescr(relation));
+
+	/*
+	 * save the relation's name and relisshared flag, too (cc_relname is used
+	 * only for debugging purposes)
+	 */
+	cache->cc_relname = pstrdup(RelationGetRelationName(relation));
+	cache->cc_relisshared = RelationGetForm(relation)->relisshared;
+
+	table_close(relation, AccessShareLock);
+
+	CACHE_elog(DEBUG2, "CatalogCacheInitializeCache: %s, %d keys",
+			   cache->cc_relname, cache->cc_nkeys);
+
+	/*
+	 * initialize cache's key information
+	 */
+	for (i = 0; i < cache->cc_nkeys; ++i)
+	{
+		Oid			keytype;
+		RegProcedure eqfunc;
+
+		CatalogCacheInitializeCache_DEBUG2;
+
+		if (cache->cc_keyno[i] > 0)
+		{
+			Form_pg_attribute attr = TupleDescAttr(tupdesc,
+												   cache->cc_keyno[i] - 1);
+
+			keytype = attr->atttypid;
+			/* cache key columns should always be NOT NULL */
+			Assert(attr->attnotnull);
+		}
+		else
+		{
+			if (cache->cc_keyno[i] < 0)
+				elog(FATAL, "sys attributes are not supported in caches");
+			keytype = OIDOID;
+		}
+
+		GetCCHashEqFuncs(keytype,
+						 &cache->cc_hashfunc[i],
+						 &eqfunc,
+						 &cache->cc_fastequal[i]);
+
+		/*
+		 * Do equality-function lookup (we assume this won't need a catalog
+		 * lookup for any supported type)
+		 */
+		fmgr_info_cxt(eqfunc,
+					  &cache->cc_skey[i].sk_func,
+					  CacheMemoryContext);
+
+		/* Initialize sk_attno suitably for HeapKeyTest() and heap scans */
+		cache->cc_skey[i].sk_attno = cache->cc_keyno[i];
+
+		/* Fill in sk_strategy as well --- always standard equality */
+		cache->cc_skey[i].sk_strategy = BTEqualStrategyNumber;
+		cache->cc_skey[i].sk_subtype = InvalidOid;
+		/* If a catcache key requires a collation, it must be C collation */
+		cache->cc_skey[i].sk_collation = C_COLLATION_OID;
+
+		CACHE_elog(DEBUG2, "CatalogCacheInitializeCache %s %d %p",
+				   cache->cc_relname, i, cache);
+	}
+
+	/*
+	 * mark this cache fully initialized
+	 */
+	cache->cc_tupdesc = tupdesc;
+
+	MemoryContextSwitchTo(oldcxt);
+}
+
+/*
+ * Workhorse for SearchCatCacheInternal().
+ * Search the  hash bucket of local/global catcache.  This returns
+ * local/global catctup even if it's a negative one. If not found, return
+ * NULL.
+ * gcc_fastequal is passed only CatCache type is GlobalCatCache
+ * because GlobalCatCache doesn't have function pointer.
+ */
+static CatCTup *
+SearchCatCacheBucket(CatCache *cache, int nkeys, Datum *searchkeys,
+					 uint32 hashValue, Index hashIndex,
+					 const CCFastEqualFN *gcc_fastequal)
+{
+	dlist_head *bucket;
+	dlist_iter	iter;
+	CatCTup* ct;
+	GlobalCatCache *gcp;
+	LWLock *bucket_lock = NULL;
+	const CCFastEqualFN *cc_fastequal;
+	CatCTup *inval_ct = NULL;
+
+	/*
+	 * If catcache is shared among process and located in shared memory
+	 * get lock for bucket and calculate hashIndex
+	 */
+	if (cache->cc_type == CC_SHAREDGLOBAL)
+	{
+		cc_fastequal = gcc_fastequal;
+
+		gcp = (GlobalCatCache *)cache;
+		bucket_lock = GetCatCacheBucketLock(gcp, hashValue);
+		LWLockAcquire(bucket_lock, LW_SHARED);
+
+		hashIndex = HASH_INDEX(hashValue, cache->cc_nbuckets);
+	}
+	else
+		cc_fastequal = cache->cc_fastequal;
+
+	bucket = &cache->cc_bucket[hashIndex];
+
+
+	/*
+	 * scan the hash bucket until we find a match or exhaust our tuples
+	 *
+	 * Note: it's okay to use dlist_foreach here, even though we modify the
+	 * dlist within the loop, because we don't continue the loop afterwards.
+	 */
+	dlist_foreach(iter, bucket)
+	{
+		ct = dlist_container(CatCTup, cache_elem, iter.cur);
+
+		if (ct->hash_value != hashValue)
+			continue;			/* quickly skip entry if wrong hash val */
+
+		/*  Continue. Valid one could be found after we find dummy one. */
+		if (cache->cc_type == CC_SHAREDLOCAL &&  CATC_IS_INVALID_DUMMY((LocalCatCTup *)ct))
+		{
+			inval_ct = ct;
+			continue;
+		}
+
+
+		if (!CatalogCacheCompareTuple(cc_fastequal, nkeys, ct->keys, searchkeys))
+			continue;
+
+		if (ct->dead)
+			continue;			/* ignore dead entries */
+
+		/*
+		 * We found a match in the cache.  Move it to the front of the list
+		 * for its hashbucket, in order to speed subsequent searches.  (The
+		 * most frequently accessed elements in any hashbucket will tend to be
+		 * near the front of the hashbucket's list.) In case of global cache,
+		 * we don't move cache to avoid acquiring exclusive lock.
+		 */
+		if (cache->cc_type != CC_SHAREDGLOBAL)
+			dlist_move_head(bucket, &ct->cache_elem);
+
+		/*
+		 * If it's a positive entry, bump its refcount and return it. If it's
+		 * negative, we can report failure to the caller. If local cache entry
+		 * is a pointer to global entry, keep refcount to zero because there is
+		 * no need to protect tuple which doesn't exist.
+		 */
+		if (!ct->negative)
+		{
+			if (cache->cc_type == CC_REGULAR ||
+				(cache->cc_type == CC_SHAREDLOCAL &&
+				 !((LocalCatCTup *)ct)->lct_committed))
+			{
+				ResourceOwnerEnlargeCatCacheRefs(CurrentResourceOwner);
+				ct->refcount++;
+				ResourceOwnerRememberCatCacheRef(CurrentResourceOwner,
+												 &ct->tuple);
+			}
+			else if (cache->cc_type == CC_SHAREDGLOBAL)
+				IncreaceGlobalCatCTupRefCount((GlobalCatCTup *)ct);
+
+			CACHE_elog(DEBUG2, "SearchCatCacheBucket (%s): found in %s bucket %d",
+					   cache->cc_relname,
+					   ct->my_cache->cc_type == CC_SHAREDGLOBAL ? "global" : "local",
+					   hashIndex);
+
+#ifdef CATCACHE_STATS
+			cache->cc_hits++;
+#endif
+		}
+		else
+		{
+			CACHE_elog(DEBUG2, "SearchCatCache (%s): found neg entry in bucket %d",
+					   cache->cc_relname, hashIndex);
+
+#ifdef CATCACHE_STATS
+			cache->cc_neg_hits++;
+#endif
+		}
+
+		/* Release lock for catctup and bucket */
+		if (cache->cc_type == CC_SHAREDGLOBAL)
+			LWLockRelease(bucket_lock);
+
+		/* return ct in both cases of positive or negative entry */
+		return ct;
+	}
+	/* Reaching here means we cannot find the cache entry in this bucket */
+	if (cache->cc_type == CC_SHAREDGLOBAL)
+		LWLockRelease(bucket_lock);
+
+	/**/
+	return inval_ct;
+}
+
+
+/*
+ * LocalCacheGetValidEntry
+ *
+ * Return the valid CatCTup entry.
+ */
+static CatCTup *
+LocalCacheGetValidEntry(LocalCatCTup *lct)
+{
+	CatCTup *local_ct;
+	GlobalCatCTup *gct;
+	LWLock *handle_lock;
+
+	CACHE_elog(DEBUG2, "LocalCacheGetValidEntry: called");
+
+	/* To access the CatCTup members */
+	local_ct = (CatCTup *)lct;
+
+	/*
+	 * If local cache entry holds its tuple locally or negative, return itself.
+	 * Note refcount of uncommitted one is assumed to be bumped in caller.
+	 */
+	if (local_ct->negative || !lct->lct_committed)
+		return local_ct;
+
+	handle_lock = GET_HANDLE_LOCK(lct->lct_handle);
+
+	/*
+	 * Check if handle is still valid. If it's invalid,
+	 * remove the local entry or mark it as dead.
+	 */
+	LWLockAcquire(handle_lock, LW_SHARED);
+
+	if (CATC_HANDLE_VALID(lct))
+	{
+		gct = lct->lct_handle->entry;
+
+		if (!((CatCTup *)gct)->dead)
+		{
+			/* bump global refcount to keep it in shared memory */
+			IncreaceGlobalCatCTupRefCount(gct);
+
+			LWLockRelease(handle_lock);
+			/* global cache entry is not negative */
+			return &gct->gct_ct;
+		}
+	}
+
+	LWLockRelease(handle_lock);
+
+	/* reaching here means gct is not valid entry */
+
+	if (local_ct->refcount > 0 ||
+		(local_ct->c_list && local_ct->c_list->refcount > 0))
+	{
+		local_ct->dead = true;
+
+		if (local_ct->c_list)
+			local_ct->c_list->dead = true;
+	}
+	else
+		CatCacheRemoveCTup(local_ct->my_cache, local_ct);
+
+	return NULL;
+}
+
+/*
+ * CreateCatCTupWithTuple
+ *
+ * Workhorse for CatalogCacheCreateEntry.
+ * Create catctup struct and following tuple, which is non-negative Regular
+ * entry, SharedLocal uncommitted entry ,or SharedGlobal entry. MemoryContext
+ * should be already switched appropriately.
+ */
+static CatCTup *
+CreateCatCTupWithTuple(CatCache *cache, HeapTuple ntp)
+{
+	CatCTup *ct;
+	HeapTuple	dtp;
+	int i;
+	size_t ct_size;
+
+	/*
+	 * If there are any out-of-line toasted fields in the tuple, expand
+	 * them in-line.  This saves cycles during later use of the catcache
+	 * entry, and also protects us against the possibility of the toast
+	 * tuples being freed before we attempt to fetch them, in case of
+	 * something using a slightly stale catcache entry.
+	 */
+	if (HeapTupleHasExternal(ntp))
+		dtp = toast_flatten_tuple(ntp, cache->cc_tupdesc);
+	else
+		dtp = ntp;
+
+	/*
+	 * To enable type cast among CatCTup, LocalCatCTup and GlobalCatCTup,
+	 * allocate tuple information just after each struct. Members specific to
+	 * LocalCatCTup and GlobalCatCTup is initialized in the caller function.
+	 */
+	if (cache->cc_type == CC_REGULAR)
+		ct_size = sizeof(CatCTup);
+	else if (cache->cc_type == CC_SHAREDLOCAL)
+		ct_size = sizeof(LocalCatCTup);
+	else if (cache->cc_type == CC_SHAREDGLOBAL)
+		ct_size = sizeof(GlobalCatCTup);
+	else
+		elog(FATAL, "unknown cache type %d",cache->cc_type);
+
+	/* Assumed MemoryContext is already switched properly */
+	ct = (CatCTup *) palloc(ct_size +
+							MAXIMUM_ALIGNOF + dtp->t_len);
+
+	ct->tuple.t_len = dtp->t_len;
+	ct->tuple.t_self = dtp->t_self;
+	ct->tuple.t_tableOid = dtp->t_tableOid;
+	ct->tuple.t_data = (HeapTupleHeader)
+		MAXALIGN(((char *) ct) + ct_size);
+	/* copy tuple contents */
+	memcpy((char *) ct->tuple.t_data,
+		   (const char *) dtp->t_data,
+		   dtp->t_len);
+
+	if (dtp != ntp)
+		heap_freetuple(dtp);
+
+	/* extract keys - they'll point into the tuple if not by-value */
+	for (i = 0; i < cache->cc_nkeys; i++)
+	{
+		Datum		atp;
+		bool		isnull;
+
+		atp = heap_getattr(&ct->tuple,
+						   cache->cc_keyno[i],
+						   cache->cc_tupdesc,
+						   &isnull);
+		Assert(!isnull);
+		ct->keys[i] = atp;
+	}
+
+	return ct;
+}
+
+/*
+ * CreateCatCTupWithoutTuple
+ *
+ * Workhorse for CatalogCacheCreateEntry.
+ * Create only catctup struct, which is negative Regular entry or SharedLocal
+ * committed entry. MemoryContext should be already switched appropriately.
+ */
+static CatCTup *
+CreateCatCTupWithoutTuple(CatCache *cache, Datum *arguments)
+{
+	CatCTup *ct;
+
+	/*
+	 * Members of LocalCatCTup is initialized in the caller function.
+	 * Assumed MemoryContext is already switched properly.
+	 */
+	if (cache->cc_type == CC_REGULAR)
+		ct = (CatCTup *) palloc(sizeof(CatCTup));
+	else if (cache->cc_type == CC_SHAREDLOCAL)
+		ct = (CatCTup *) palloc(sizeof(LocalCatCTup));
+	else
+		elog(FATAL, "Global cache entry without tuple is not supported");
+
+	/*
+	 * Store keys - they'll point into separately allocated memory if not
+	 * by-value. If arguments is NULL, it's skip-global dummy.
+	 */
+	if (arguments)
+		CatCacheCopyKeys(cache->cc_tupdesc, cache->cc_nkeys, cache->cc_keyno,
+						 arguments, ct->keys);
+
+	return ct;
+}
+
+
+/*
+ * Get partition lock of Global CatCache hash table
+ */
+static LWLock *
+GetCatCacheBucketLock(GlobalCatCache *gcp, uint32 hashValue)
+{
+	Index hashIndex = HASH_INDEX(hashValue, ((CatCache *)gcp)->cc_nbuckets);
+
+	return &gcp->gcc_bucket_locks[(hashIndex % MAX_CC_BUCKET_LOCKS)];
+}
+
+/*
+ * Bump global refcount to keep it in shared memory
+ */
+static void
+IncreaceGlobalCatCTupRefCount(GlobalCatCTup *gct)
+{
+	SpinLockAcquire(&gct->gct_mutex);
+	((CatCTup *)gct)->refcount++;
+	SpinLockRelease(&gct->gct_mutex);
+
+	CACHE_elog(DEBUG2, "IncreaseGlobalCatCTupRfCount() global refcount is bumped: %s, %d tuple %p",
+			   ((CatCTup *)gct)->my_cache->cc_relname,
+			   ((CatCTup *)gct)->refcount,
+			   &((CatCTup *)gct)->tuple);
+}
diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c
index 591dd33..4c63805 100644
--- a/src/backend/utils/cache/inval.c
+++ b/src/backend/utils/cache/inval.c
@@ -106,6 +106,7 @@
 #include "storage/sinval.h"
 #include "storage/smgr.h"
 #include "utils/catcache.h"
+#include "utils/shared_meta_cache.h"
 #include "utils/inval.h"
 #include "utils/memdebug.h"
 #include "utils/memutils.h"
@@ -561,7 +562,7 @@ LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg)
 		{
 			InvalidateCatalogSnapshot();
 
-			SysCacheInvalidate(msg->cc.id, msg->cc.hashValue);
+			SysCacheInvalidate(msg->cc.id, msg->cc.hashValue, INVAL_COMMON);
 
 			CallSyscacheCallbacks(msg->cc.id, msg->cc.hashValue);
 		}
@@ -572,7 +573,7 @@ LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg)
 		{
 			InvalidateCatalogSnapshot();
 
-			CatalogCacheFlushCatalog(msg->cat.catId);
+			CatalogCacheFlushCatalog(msg->cat.catId, INVAL_COMMON);
 
 			/* CatalogCacheFlushCatalog calls CallSyscacheCallbacks as needed */
 		}
@@ -629,6 +630,102 @@ LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg)
 }
 
 /*
+ * GMCEOCommandExecInvalMessage
+ *
+ */
+void
+GMCEOCommandExecInvalMessage(SharedInvalidationMessage *msg)
+{
+	if (msg->id >= 0)
+	{
+		if (msg->cc.dbId == MyDatabaseId || msg->cc.dbId == InvalidOid)
+		{
+			InvalidateCatalogSnapshot();
+
+			/* not end of transaction */
+			SysCacheInvalidate(msg->cc.id, msg->cc.hashValue, INVAL_GMC_EO_COMMAND);
+
+			CallSyscacheCallbacks(msg->cc.id, msg->cc.hashValue);
+		}
+	}
+	else if (msg->id == SHAREDINVALCATALOG_ID)
+	{
+		if (msg->cat.dbId == MyDatabaseId || msg->cat.dbId == InvalidOid)
+		{
+			InvalidateCatalogSnapshot();
+
+			CatalogCacheFlushCatalog(msg->cat.catId, INVAL_GMC_EO_COMMAND);
+
+			/* CatalogCacheFlushCatalog calls CallSyscacheCallbacks as needed */
+		}
+	}
+	else
+		LocalExecuteInvalidationMessage(msg);
+
+}
+
+void
+GMCEOXactExecInvalMessage(SharedInvalidationMessage *msg)
+{
+	if (msg->id >= 0)
+	{
+		if (msg->cc.dbId == MyDatabaseId || msg->cc.dbId == InvalidOid)
+		{
+			//InvalidateCatalogSnapshot() must not be executed
+			/* end of transaction */
+			SysCacheInvalidate(msg->cc.id, msg->cc.hashValue, INVAL_GMC_EO_COMMAND);
+
+			/* Only the global caches entires are flushed */
+			GlobalSysCacheInvalidate(msg->cc.id, msg->cc.hashValue);
+
+			CallSyscacheCallbacks(msg->cc.id, msg->cc.hashValue);
+		}
+	}
+	else if (msg->id == SHAREDINVALCATALOG_ID)
+	{
+		if (msg->cat.dbId == MyDatabaseId || msg->cat.dbId == InvalidOid)
+		{
+			//InvalidateCatalogSnapshot() must not be executed
+
+			CatalogCacheFlushCatalog(msg->cat.catId, INVAL_GMC_COMMIT_XACT);
+
+			/* CatalogCacheFlushCatalog calls CallSyscacheCallbacks as needed */
+		}
+	}
+}
+
+void
+GMCAbortSubXactExecInvalMessage(SharedInvalidationMessage *msg)
+{
+	if (msg->id >= 0)
+	{
+		if (msg->cc.dbId == MyDatabaseId || msg->cc.dbId == InvalidOid)
+		{
+			InvalidateCatalogSnapshot();
+			/* end of transaction */
+			SysCacheInvalidate(msg->cc.id, msg->cc.hashValue, INVAL_GMC_ABORT_SUBXACT);
+
+			CallSyscacheCallbacks(msg->cc.id, msg->cc.hashValue);
+		}
+	}
+	else if (msg->id == SHAREDINVALCATALOG_ID)
+	{
+		if (msg->cat.dbId == MyDatabaseId || msg->cat.dbId == InvalidOid)
+		{
+			InvalidateCatalogSnapshot();
+
+			CatalogCacheFlushCatalog(msg->cat.catId, INVAL_GMC_ABORT_SUBXACT);
+
+			/* CatalogCacheFlushCatalog calls CallSyscacheCallbacks as needed */
+		}
+	}
+	else
+		LocalExecuteInvalidationMessage(msg);
+
+}
+
+
+/*
  *		InvalidateSystemCaches
  *
  *		This blows away all tuples in the system catalog caches and
@@ -968,6 +1065,16 @@ AtEOXact_Inval(bool isCommit)
 		AppendInvalidationMessages(&transInvalInfo->PriorCmdInvalidMsgs,
 								   &transInvalInfo->CurrentCmdInvalidMsgs);
 
+		/*
+		 * If catcache is global mode, try to remove global catcache entry.
+		 * And also local catcaches are removed since  cache tuples added
+		 * in this transaction are located in local CatCache rather than
+		 * global CatCache. We don't want to hold entries locally after commit.
+		 */
+		if (IS_GLOBAL_META_CACHE)
+			ProcessInvalidationMessages(&transInvalInfo->PriorCmdInvalidMsgs,
+										GMCEOXactExecInvalMessage);
+
 		ProcessInvalidationMessagesMulti(&transInvalInfo->PriorCmdInvalidMsgs,
 										 SendSharedInvalidMessages);
 
@@ -1055,8 +1162,12 @@ AtEOSubXact_Inval(bool isCommit)
 	}
 	else
 	{
-		ProcessInvalidationMessages(&myInfo->PriorCmdInvalidMsgs,
-									LocalExecuteInvalidationMessage);
+		if (IS_GLOBAL_META_CACHE)
+			ProcessInvalidationMessages(&myInfo->PriorCmdInvalidMsgs,
+										GMCAbortSubXactExecInvalMessage);
+		else
+			ProcessInvalidationMessages(&myInfo->PriorCmdInvalidMsgs,
+										LocalExecuteInvalidationMessage);
 
 		/* Pop the transaction state stack */
 		transInvalInfo = myInfo->parent;
@@ -1092,8 +1203,13 @@ CommandEndInvalidationMessages(void)
 	if (transInvalInfo == NULL)
 		return;
 
-	ProcessInvalidationMessages(&transInvalInfo->CurrentCmdInvalidMsgs,
-								LocalExecuteInvalidationMessage);
+	if (IS_GLOBAL_META_CACHE)
+		ProcessInvalidationMessages(&transInvalInfo->CurrentCmdInvalidMsgs,
+									GMCEOCommandExecInvalMessage);
+	else
+		ProcessInvalidationMessages(&transInvalInfo->CurrentCmdInvalidMsgs,
+									LocalExecuteInvalidationMessage);
+
 	AppendInvalidationMessages(&transInvalInfo->PriorCmdInvalidMsgs,
 							   &transInvalInfo->CurrentCmdInvalidMsgs);
 }
diff --git a/src/backend/utils/cache/shared_cache.c b/src/backend/utils/cache/shared_cache.c
new file mode 100644
index 0000000..a18add1
--- /dev/null
+++ b/src/backend/utils/cache/shared_cache.c
@@ -0,0 +1,523 @@
+/*-------------------------------------------------------------------------
+ *
+ * shared_cache.c
+ *	  Utilities for global system catalog cache and global relation cache
+ *
+ * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/utils/cache/globalcache.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "storage/shmem.h"
+#include "storage/spin.h"
+#include "utils/hsearch.h"
+#include "utils/memutils.h"
+#include "utils/shared_meta_cache.h"
+#include "utils/syscache.h"
+
+#include "miscadmin.h"
+
+
+#define NUM_DB 16 /* arbiary number */
+
+/* shared cache map to find the global catalog or relation hash table */
+static HTAB *GlobalCatCacheMap;
+static HTAB *GlobalRelCacheMap;
+
+//static MemoryContext TempGlobalCacheContext = NULL;
+
+static HandleBlock *InitHandleBlock(void);
+
+static void InitGlobalCatCacheMap(void);
+static void InitGlobalRelCacheMap(void);
+static Size GlobalCatCacheMapSize(void);
+static Size GlobalRelCacheMapSize(void);
+
+GlobalCatCacheMapLock *globalCatCacheMapLock = NULL;
+GlobalRelCacheMapLock *globalRelCacheMapLock = NULL;
+
+/*
+ * --------------------------------------------------------------
+ * Utilities for cache handle
+ * --------------------------------------------------------------
+ */
+
+/*
+ * InitCacheHanlde()
+ * 		Initialize HandleHeader with one free block.
+ *
+ */
+void
+InitCacheHandle(HandleHeader *header, void *owner)
+{
+	HandleBlock *block;
+
+	header->owner = owner;
+
+	LWLockInitialize(&header->freelist_lock, LWTRANCHE_CACHE_HANDLE);
+
+	dlist_init(&header->freelist);
+
+	/* Caller should switch to appropriate shared context */
+	block = InitHandleBlock();
+
+	dlist_push_head(&header->freelist, &block->node);
+}
+
+static HandleBlock *
+InitHandleBlock(void)
+{
+	HandleBlock *block;
+	int i;
+
+	Assert(IsA((CurrentMemoryContext), ShmRetailContext) ||
+		   IsA((CurrentMemoryContext), ShmZoneContext));
+
+	/* Assume MemoryContext is already switched */
+	block = (HandleBlock *) palloc0(sizeof(HandleBlock));
+
+	block->nfree = HANDLE_BLOCK_SIZE;
+
+	block->infreelist = true;
+
+	/* Initialize locks for partition of hash table */
+	SpinLockInit(&block->mutex);
+
+	LWLockInitialize(&block->handle_lock, LWTRANCHE_META_CACHE_HANDLE);
+
+	/* Initailize handle handles */
+	for (i = 0; i < HANDLE_BLOCK_SIZE; ++i)
+	{
+		block->handles[i].free = true;
+		block->handles[i].block = block;
+	}
+	return block;
+}
+
+
+/*
+ * InvalCacheHandle()
+ *  Free the used handle to recycle it.
+ *	When block is not in freelist, push the block to tail of list.
+ */
+void
+InvalCacheHandle(HandleHeader *header, Handle *handle)
+{
+	/* Get lock for the handle and its block */
+	SpinLockAcquire(&handle->block->mutex);
+
+	/* If this function is concurrently called, the handle might be already freed */
+	if (handle->free)
+	{
+		SpinLockRelease(&handle->block->mutex);
+		return;
+	}
+
+	handle->free = true;
+	handle->generation++;
+	handle->entry = NULL;
+	handle->block->nfree++;
+	SpinLockRelease(&handle->block->mutex);
+
+	/*
+	 * If the block is already in the freelist, do nothing and exit.
+	 * If not, push the block to the list. Check the block is in the list
+	 * or not twice to reduce the locking window.
+	 */
+	if (!handle->block->infreelist)
+	{
+		LWLockAcquire(&header->freelist_lock, LW_EXCLUSIVE);
+
+		/*
+		 * The block might be already pushed concurrently. There is also
+		 * possibility that someone already got the free handle before get LWLock.
+		 */
+		if (!handle->block->infreelist && handle->block->nfree != 0)
+		{
+			handle->block->infreelist = true;
+			dlist_push_tail(&header->freelist, &handle->block->node);
+		}
+		LWLockRelease(&header->freelist_lock);
+	}
+}
+
+
+/*
+ * GetCacheHandle()
+ * 	Return Handle. If not found, build a new free block.
+ */
+Handle *
+GetCacheHandle(HandleHeader *header, void *entry)
+{
+	Handle *handle = NULL;
+	HandleBlock *block = NULL;
+	int i;
+
+	LWLockAcquire(&header->freelist_lock, LW_SHARED);
+
+	/*
+	 * Get the first block of freelist. If no available free block, build a new
+	 * one. To reduce the lock contention, exclusive locking duration is as
+	 * small as possible.  Because a newly added block may become full, we
+	 * check if freelist is empyt recursively.
+	 */
+	while (dlist_is_empty(&header->freelist))
+	{
+		MemoryContext old_context;
+
+		LWLockRelease(&header->freelist_lock);
+
+		/* Append the newly built block */
+		LWLockAcquire(&header->freelist_lock, LW_EXCLUSIVE);
+
+		/* TempGlobalCacheContext = MemoryContextClone(GlobalCacheContext, */
+		/* 											CurTransactionContext); */
+
+		/* old_context = MemoryContextSwitchTo(TempGlobalCacheContext); */
+
+		old_context = MemoryContextSwitchTo(GlobalCacheContext);
+
+		block = InitHandleBlock();
+
+		MemoryContextSwitchTo(old_context);
+
+		/* ShmRetailContextMoveChunk(TempGlobalCacheContext, GlobalCacheContext); */
+
+		dlist_push_head(&header->freelist, &block->node);
+
+		LWLockRelease(&header->freelist_lock);
+
+		LWLockAcquire(&header->freelist_lock, LW_SHARED);
+	}
+
+	block = dlist_head_element(HandleBlock, node, &header->freelist);
+
+	/* Search linearly to find free handle */
+	for (i = 0; i < HANDLE_BLOCK_SIZE; ++i)
+	{
+		if (!block->handles[i].free)
+			continue;
+
+		/* Get lock for handle and its owning block */
+		SpinLockAcquire(&block->mutex);
+
+		/* Check the handle again because it might be already occupied */
+		if (!block->handles[i].free)
+		{
+			SpinLockRelease(&block->mutex);
+			continue;
+		}
+
+		block->nfree--;
+		block->handles[i].generation++;
+		block->handles[i].free = false;
+		handle = &block->handles[i];
+
+		SpinLockRelease(&block->mutex);
+
+		break;
+	}
+
+	LWLockRelease(&header->freelist_lock);
+
+
+	/*
+	 * If the block becomes full, delete from free list and mark it. To reduce
+	 * the lock duration, nfree is checked before lock is acquired and checked
+	 * again after acquation of lock.
+	 */
+	if (block->nfree == 0)
+	{
+		LWLockAcquire(&header->freelist_lock, LW_EXCLUSIVE);
+
+		if (block->infreelist && block->nfree == 0)
+		{
+			block->infreelist = false;
+			dlist_delete(&block->node);
+		}
+		LWLockRelease(&header->freelist_lock);
+	}
+
+	/* Set link from handle to entry */
+	handle->entry = entry;
+
+	return handle;
+}
+
+/*
+ * --------------------------------------------------------------
+ * Utilities for both global catalog and relation cache
+ * --------------------------------------------------------------
+ */
+
+/*
+ * GlobalMetaCacheShmemInit
+ *
+ * This is called during shared-memory initialization
+ *
+ */
+void
+GlobalMetaCacheShmemInit(void)
+{
+	/* do nothing if catalog cache is not global */
+	if (!IS_GLOBAL_META_CACHE)
+		return;
+
+	/* Create GMC area and GlobalCacheContext */
+	GlobalCacheContextShmemInit(SHM_METACACHE_SIZE_MB);
+
+	/* Initialize child contexts of GlobalCacheContext */
+	GlobalCatCacheContextShmemInit();
+	GlobalRelCacheContextShmemInit();
+
+	/* Initialize Maps to associate tag with CatCache and Relation hash table */
+	InitGlobalCatCacheMap();
+	InitGlobalRelCacheMap();
+}
+
+Size
+GlobalMetaCacheShmemSize(void)
+{
+	Size size = 0;
+
+	/* do nothing if catalog cache is not shared */
+	if (!IS_GLOBAL_META_CACHE)
+		return 0;
+
+	/* size of global catalog cache area */
+	size = add_size(size, SHM_METACACHE_SIZE_MB);
+
+	/* size of shared MemoryContext */
+	size = add_size(size, GlobalCacheContextSize());
+	size = add_size(size, GlobalCatCacheContextSize());
+	size = add_size(size, GlobalRelCacheContextSize());
+
+	/* size of lookup hash table */
+	size = add_size(size, GlobalCatCacheMapSize());
+	size = add_size(size, GlobalRelCacheMapSize());
+
+	return size;
+}
+
+
+/*
+ * --------------------------------------------------------------
+ * Utilities for hash table map of global catalog cache
+ * --------------------------------------------------------------
+ */
+
+static void
+InitGlobalCatCacheMap(void)
+{
+	HASHCTL info;
+	size_t init_size;
+	size_t max_size;
+	bool found;
+	int i;
+
+	/* assume no locking is needed yet */
+
+	info.keysize = sizeof(GlobalCatCacheMapKey);
+	info.entrysize = sizeof(GlobalCatCacheMapEnt);
+	info.num_partitions = NUM_MAP_PARTITIONS;
+
+	/*
+	 * Number of CatCache table times DB. Actually, this is slightly more than
+	 * we need because it counts dublicatedly shared catalog cache.
+	 */
+	init_size = max_size = SysCacheSize * NUM_DB;
+
+
+	GlobalCatCacheMap = ShmemInitHash("Global CatCache Map",
+									  init_size, max_size,
+									  &info,
+									  HASH_ELEM | HASH_BLOBS |HASH_PARTITION);
+
+	/* Get the pointer to LWLocks for parition */
+	globalCatCacheMapLock = (GlobalCatCacheMapLock *)
+		ShmemInitStruct("GlobalCatCacheMapLock",
+						sizeof(GlobalCatCacheMapLock),	&found);
+
+	if (!found)
+	{
+		Assert(!IsUnderPostmaster);
+
+		for (i = 0; i < NUM_MAP_PARTITIONS; i++)
+			LWLockInitialize(&globalCatCacheMapLock->locks[i],
+							 LWTRANCHE_META_CACHE_MAP);
+	}
+}
+
+/*
+ * Estimate space needed for mapping hashtable
+ */
+static Size
+GlobalCatCacheMapSize(void)
+{
+	/* size of lookup hash table. see comments in InitGlobalCatCacheMap */
+	return hash_estimate_size((SysCacheSize * NUM_DB), sizeof(GlobalCatCacheMapEnt));
+}
+
+uint32
+GlobalCatCacheMapHashCode(GlobalCatCacheMapKey *keyPtr)
+{
+	return get_hash_value(GlobalCatCacheMap, (void *) keyPtr);
+}
+
+
+GlobalCatCacheMapEnt*
+GlobalCatCacheMapLookup(GlobalCatCacheMapKey *keyPtr, uint32 hashcode)
+{
+	GlobalCatCacheMapEnt *result;
+
+	result = (GlobalCatCacheMapEnt *)
+		hash_search_with_hash_value(GlobalCatCacheMap,
+									(void *) keyPtr,
+									hashcode,
+									HASH_FIND,
+									NULL);
+
+	if (!result)
+		return NULL;
+
+	return result;
+}
+
+bool
+GlobalCatCacheMapInsert(GlobalCatCacheMapKey *keyPtr,
+						uint32 hashcode, GlobalCatCache *gcp)
+{
+	GlobalCatCacheMapEnt *result;
+	bool found;
+
+	result = (GlobalCatCacheMapEnt *)
+		hash_search_with_hash_value(GlobalCatCacheMap,
+									(void *) keyPtr,
+									hashcode,
+									HASH_ENTER,
+									&found);
+
+	if (found)
+		return false;
+
+	result->gcp = gcp;
+
+	return true;
+}
+
+void
+GlobalCatCacheMapInitKey(CatCache *lcp, GlobalCatCacheMapKey *keyPtr)
+{
+	Assert(lcp->cc_type == CC_SHAREDLOCAL);
+
+	keyPtr->dbId = lcp->cc_relisshared ? (Oid) 0 : MyDatabaseId;
+	keyPtr->cacheId = lcp->id;
+}
+
+/*
+ * --------------------------------------------------------------
+ * Utilities for hash table map of global catalog cache
+ * --------------------------------------------------------------
+ */
+static void
+InitGlobalRelCacheMap(void)
+{
+	HASHCTL info;
+	size_t init_size;
+	size_t max_size;
+	bool found;
+	int i;
+
+	/* assume no locking is needed yet */
+	info.keysize = sizeof(GlobalRelCacheMapKey);
+	info.entrysize = sizeof(GlobalRelCacheMapEnt);
+	info.num_partitions = NUM_MAP_PARTITIONS;
+
+	init_size = NUM_DB;
+	max_size = 2 * NUM_DB;
+
+	GlobalRelCacheMap = ShmemInitHash("Global Relcache Map",
+									  init_size, max_size,
+									  &info,
+									  HASH_ELEM | HASH_BLOBS |HASH_PARTITION);
+
+	/* Get the pointer to LWLocks for parition */
+	globalRelCacheMapLock = (GlobalRelCacheMapLock *)
+		ShmemInitStruct("GlobalRelacacheMapLock",
+						sizeof(GlobalRelCacheMapLock),&found);
+
+	if (!found)
+	{
+		Assert(!IsUnderPostmaster);
+
+		for (i = 0; i < NUM_MAP_PARTITIONS; i++)
+			LWLockInitialize(&globalRelCacheMapLock->locks[i],
+							 LWTRANCHE_META_CACHE_MAP);
+	}
+}
+
+/*
+ * Estimate space needed for mapping hashtable
+ */
+static Size
+GlobalRelCacheMapSize(void)
+{
+	size_t max_size = 2 * NUM_DB;
+	return hash_estimate_size(max_size, sizeof(GlobalRelCacheMapEnt));
+}
+
+uint32
+GlobalRelCacheMapHashCode(GlobalRelCacheMapKey *keyPtr)
+{
+	return get_hash_value(GlobalRelCacheMap, (void *) keyPtr);
+}
+
+
+GlobalRelCacheMapEnt*
+GlobalRelCacheMapLookup(GlobalRelCacheMapKey *keyPtr, uint32 hashcode)
+{
+	GlobalRelCacheMapEnt *result;
+
+	result = (GlobalRelCacheMapEnt *)
+		hash_search_with_hash_value(GlobalRelCacheMap,
+									(void *) keyPtr,
+									hashcode,
+									HASH_FIND,
+									NULL);
+
+	if (!result)
+		return NULL;
+
+	return result;
+}
+
+bool
+GlobalRelCacheMapInsert(GlobalRelCacheMapKey *keyPtr,
+						uint32 hashcode,
+						dshash_table_handle global_relcache_handle,
+						HandleHeader *cache_handle_header)
+{
+	GlobalRelCacheMapEnt *result;
+	bool found;
+
+	result = (GlobalRelCacheMapEnt *)
+		hash_search_with_hash_value(GlobalRelCacheMap,
+									(void *) keyPtr,
+									hashcode,
+									HASH_ENTER,
+									&found);
+
+	if (found)
+		return false;
+
+	result->global_relcache_handle = global_relcache_handle;
+	result->cache_handle_header = cache_handle_header;
+
+	return true;
+}
diff --git a/src/backend/utils/cache/syscache.c b/src/backend/utils/cache/syscache.c
index 53d9ddf..bba711f 100644
--- a/src/backend/utils/cache/syscache.c
+++ b/src/backend/utils/cache/syscache.c
@@ -997,6 +997,8 @@ static int	SysCacheSupportingRelOidSize;
 
 static int	oid_compare(const void *a, const void *b);
 
+static inline bool
+SysCacheInvalidteCheck(int cacheId, uint32 hashValue);
 
 /*
  * InitCatalogCache - initialize the caches
@@ -1446,7 +1448,7 @@ SearchSysCacheList(int cacheId, int nkeys,
  *	This routine is only quasi-public: it should only be used by inval.c.
  */
 void
-SysCacheInvalidate(int cacheId, uint32 hashValue)
+SysCacheInvalidate(int cacheId, uint32 hashValue, InvalPhase invalPhase)
 {
 	if (cacheId < 0 || cacheId >= SysCacheSize)
 		elog(ERROR, "invalid cache ID: %d", cacheId);
@@ -1455,9 +1457,31 @@ SysCacheInvalidate(int cacheId, uint32 hashValue)
 	if (!PointerIsValid(SysCache[cacheId]))
 		return;
 
-	CatCacheInvalidate(SysCache[cacheId], hashValue);
+	CatCacheInvalidate(SysCache[cacheId], hashValue, invalPhase);
 }
 
+
+void
+GlobalSysCacheInvalidate(int cacheId, uint32 hashValue)
+{
+	if (!SysCacheInvalidteCheck(cacheId, hashValue))
+		return;
+
+	GlobalCatCacheInvalidate(SysCache[cacheId], hashValue);
+}
+
+static inline bool
+SysCacheInvalidteCheck(int cacheId, uint32 hashValue)
+{
+	if (cacheId < 0 || cacheId >= SysCacheSize)
+		elog(ERROR, "invalid cache ID: %d", cacheId);
+
+	/* if this cache isn't initialized yet, no need to do anything */
+	if (PointerIsValid(SysCache[cacheId]))
+		return true;
+	else
+		return false;
+}
 /*
  * Certain relations that do not have system caches send snapshot invalidation
  * messages in lieu of catcache messages.  This is for the benefit of
diff --git a/src/backend/utils/init/globals.c b/src/backend/utils/init/globals.c
index b1f6291..e32cef5 100644
--- a/src/backend/utils/init/globals.c
+++ b/src/backend/utils/init/globals.c
@@ -148,3 +148,5 @@ int			VacuumCostBalance = 0;	/* working state for vacuum */
 bool		VacuumCostActive = false;
 
 double		vacuum_cleanup_index_scale_factor;
+
+int shared_metacache_mem = 0;
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 9630866..3b312de 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -83,6 +83,7 @@
 #include "tsearch/ts_cache.h"
 #include "utils/builtins.h"
 #include "utils/bytea.h"
+#include "utils/shared_meta_cache.h"
 #include "utils/float.h"
 #include "utils/guc_tables.h"
 #include "utils/memutils.h"
@@ -2352,6 +2353,20 @@ static struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+#if !defined(RELCACHE_FORCE_RELEASE) || !defined(CLOBBER_CACHE_ALWAYS) || \
+	!defined(CLOBBER_CACHE_RECURSIVELY) || !defined(CLOBBER_CACHE_RECURSIVELY)
+	{
+		{"shared_metacache_mem", PGC_POSTMASTER, RESOURCES_MEM,
+			gettext_noop("Sets the memory for global system catlog cache and relation cache."),
+			gettext_noop("A value of 0 disables this feature and meta cache is built per backend."),
+			GUC_UNIT_MB
+		},
+		&shared_metacache_mem,
+		0, 0, INT_MAX,
+		NULL, NULL, NULL
+	},
+#endif
+
 	/*
 	 * We use the hopefully-safely-small value of 100kB as the compiled-in
 	 * default for max_stack_depth.  InitializeGUCOptions will increase it if
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index e1048c0..a35cfaf 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -131,6 +131,8 @@
 #maintenance_work_mem = 64MB		# min 1MB
 #autovacuum_work_mem = -1		# min 1MB, or -1 to use maintenance_work_mem
 #logical_decoding_work_mem = 64MB	# min 64kB
+#shared_metacache_mem = 0MB		# zero disables the feature
+					# (change requires restart)
 #max_stack_depth = 2MB			# min 100kB
 #shared_memory_type = mmap		# the default is the first option
 					# supported by the operating system:
diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.h
index 8fda8e4..dce17e7 100644
--- a/src/include/storage/lwlock.h
+++ b/src/include/storage/lwlock.h
@@ -220,6 +220,11 @@ typedef enum BuiltinTrancheIds
 	LWTRANCHE_TBM,
 	LWTRANCHE_PARALLEL_APPEND,
 	LWTRANCHE_SXACT,
+	LWTRANCHE_SHARED_MCXT,
+	LWTRANCHE_CACHE_HANDLE,
+	LWTRANCHE_META_CACHE_MAP,
+	LWTRANCHE_GLOBAL_CATCACHE,
+	LWTRANCHE_META_CACHE_HANDLE,
 	LWTRANCHE_FIRST_USER_DEFINED
 }			BuiltinTrancheIds;
 
diff --git a/src/include/storage/sinval.h b/src/include/storage/sinval.h
index 903818b..8e84985 100644
--- a/src/include/storage/sinval.h
+++ b/src/include/storage/sinval.h
@@ -149,5 +149,7 @@ extern void ProcessCommittedInvalidationMessages(SharedInvalidationMessage *msgs
 												 Oid dbid, Oid tsid);
 
 extern void LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg);
-
+extern void GMCEOXactExecInvalMessage(SharedInvalidationMessage *msg);
+extern void GMCAbortSubXactExecInvalMessage(SharedInvalidationMessage *msg);
+extern void GMCEOCommandExecInvalMessage(SharedInvalidationMessage *msg);
 #endif							/* SINVAL_H */
diff --git a/src/include/utils/catcache.h b/src/include/utils/catcache.h
index f4aa316..7f4ddec 100644
--- a/src/include/utils/catcache.h
+++ b/src/include/utils/catcache.h
@@ -34,6 +34,7 @@
 
 #define CATCACHE_MAXKEYS		4
 
+#define GET_CACHEID(catcache) (catcache->id)
 
 /* function computing a datum's hash */
 typedef uint32 (*CCHashFN) (Datum datum);
@@ -41,9 +42,39 @@ typedef uint32 (*CCHashFN) (Datum datum);
 /* function computing equality of two datums */
 typedef bool (*CCFastEqualFN) (Datum a, Datum b);
 
+/* Forward declarations */
+typedef struct GlobalCatCache GlobalCatCache;
+typedef struct LocalCatCTup LocalCatCTup;
+typedef struct GlobalCatCTup GlobalCatCTup;
+
+typedef enum InvalPhase InvalPhase;
+
+/*
+ * CatCache type
+ * CC_REGULAR: CatCache and its entry are located in CacheMemoryContext
+ *
+ * CC_SHAREDLOCAL: CatCache and its entry are located in CacheMemoryContext
+ *  but its entry doesn't have actual tuple locally but points to handle
+ *  for global cache entry located in shared memory. When cache tuple
+ *  is uncommitted, the entry has the tuple locally and doesn't point to
+ *  its handle. Negative entry is storaed locally and doesn't use its
+ *  handle either.
+ *
+ * CC_SHAREDGLOBAL: CatCache and its entry are located in shared memory.
+ *   Negative entry is not stored. Each cache entry has corresponding
+ *   handle and can be accessed via handle.
+ */
+typedef enum
+{
+	CC_REGULAR,
+	CC_SHAREDLOCAL,
+	CC_SHAREDGLOBAL
+} CatCacheType;
+
 typedef struct catcache
 {
 	int			id;				/* cache identifier --- see syscache.h */
+	CatCacheType cc_type;		/* type of catcache --- see CatCacheType */
 	int			cc_nbuckets;	/* # of hash buckets in this cache */
 	TupleDesc	cc_tupdesc;		/* tuple descriptor (copied from reldesc) */
 	dlist_head *cc_bucket;		/* hash buckets */
@@ -62,6 +93,11 @@ typedef struct catcache
 	ScanKeyData cc_skey[CATCACHE_MAXKEYS];	/* precomputed key info for heap
 											 * scans */
 
+	/* materials for global meta cache */
+	GlobalCatCache *gcp; 		/* Global CatCache */
+	SubTransactionId cc_skip_global_subxid; /* subxid when all cache
+											 * is invalidated */
+
 	/*
 	 * Keep these at the end, so that compiling catcache.c with CATCACHE_STATS
 	 * doesn't break ABI for other modules
@@ -89,7 +125,6 @@ typedef struct catctup
 #define CT_MAGIC   0x57261502
 
 	uint32		hash_value;		/* hash value for this tuple's keys */
-
 	/*
 	 * Lookup keys for the entry. By-reference datums point into the tuple for
 	 * positive cache entries, and are separately allocated for negative ones.
@@ -175,6 +210,7 @@ typedef struct catclist
 	short		nkeys;			/* number of lookup keys specified */
 	int			n_members;		/* number of member tuples */
 	CatCache   *my_cache;		/* link to owning catcache */
+	LocalCatCTup  **shared_local_members; /* used when catcache is global */
 	CatCTup    *members[FLEXIBLE_ARRAY_MEMBER]; /* members */
 } CatCList;
 
@@ -188,6 +224,7 @@ typedef struct catcacheheader
 
 /* this extern duplicates utils/memutils.h... */
 extern PGDLLIMPORT MemoryContext CacheMemoryContext;
+extern PGDLLIMPORT MemoryContext GlobalCatCacheContext;
 
 extern void CreateCacheMemoryContext(void);
 
@@ -218,8 +255,9 @@ extern CatCList *SearchCatCacheList(CatCache *cache, int nkeys,
 extern void ReleaseCatCacheList(CatCList *list);
 
 extern void ResetCatalogCaches(void);
-extern void CatalogCacheFlushCatalog(Oid catId);
-extern void CatCacheInvalidate(CatCache *cache, uint32 hashValue);
+extern void CatalogCacheFlushCatalog(Oid catId, InvalPhase invalPhase);
+extern void CatCacheInvalidate(CatCache *cache, uint32 hashValue, InvalPhase invalPhase);
+extern void GlobalCatCacheInvalidate(CatCache *cache, uint32 hashValue);
 extern void PrepareToInvalidateCacheTuple(Relation relation,
 										  HeapTuple tuple,
 										  HeapTuple newtuple,
@@ -228,4 +266,7 @@ extern void PrepareToInvalidateCacheTuple(Relation relation,
 extern void PrintCatCacheLeakWarning(HeapTuple tuple);
 extern void PrintCatCacheListLeakWarning(CatCList *list);
 
+//ideriha
+extern void PrintAllCatCacheInfo(void);
+
 #endif							/* CATCACHE_H */
diff --git a/src/include/utils/shared_meta_cache.h b/src/include/utils/shared_meta_cache.h
new file mode 100644
index 0000000..f2bd7fb
--- /dev/null
+++ b/src/include/utils/shared_meta_cache.h
@@ -0,0 +1,170 @@
+/*-------------------------------------------------------------------------
+ *
+ * shared_meta_cache.h
+ *	  Routines for shared catalog/relation cache
+ *
+ * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/utils/shared_meta_cache.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef SHARED_META_CACHE_H
+#define SHARED_META_CACHE_H
+
+#include "lib/ilist.h"
+#include "lib/dshash.h"
+#include "storage/lwlock.h"
+#include "utils/catcache.h"
+
+/* Number of partitions of the shared cache mapping hashtable */
+#define NUM_MAP_PARTITIONS  8
+#define IS_GLOBAL_META_CACHE (shared_metacache_mem != 0)
+
+/* invalidation phase used in shared catalog/relation cache */
+typedef enum InvalPhase
+{
+	INVAL_COMMON, /* 1. when meta cache is not shared
+				   * 2. when sinval is read */
+	INVAL_GMC_COMMIT_XACT, /* when transaction is committed */
+	INVAL_GMC_ABORT_XACT,
+	INVAL_GMC_COMMIT_SUBXACT,
+	INVAL_GMC_ABORT_SUBXACT, /* when sub-transaction is aborted */
+	INVAL_GMC_EO_COMMAND /* end of command */
+} InvalPhase;
+
+
+/*
+ * Declarations for cache handle
+ */
+typedef struct HandleHeader HandleHeader;
+typedef struct HandleBlock HandleBlock;
+typedef struct Handle Handle;
+
+struct Handle
+{
+	bool free; 			/* is chunk free? */
+	int generation;     /* how many times this chunk is recycled */
+	HandleBlock *block; /* block owning this chunk */
+	void *entry; 	/* link to global catalog/relation cache entry */
+};
+
+struct HandleBlock
+{
+	dlist_node node; /* dlist node */
+	uint32 nfree; 	/* number of free handles */
+	bool infreelist;   	/* if it's a member of freelist or not */
+	slock_t mutex; /* lock for Handle members and above except for infreelist */
+#define HANDLE_BLOCK_SIZE 32
+	Handle handles[HANDLE_BLOCK_SIZE];
+	LWLock handle_lock; /* lock for get/invalidate handle operation */
+};
+
+struct HandleHeader
+{
+	dlist_head freelist;  /* head of the free block list */
+	LWLock freelist_lock; /* lock for freelist and HandleBlock->infreelist */
+	void *owner; /* Global Hash table of either catcache or relcache */
+};
+
+/*
+ * Declarations for shared catalog cache map
+ */
+
+/* If CatCache is shared across database, it's pushed to list whose dbid is 0 */
+typedef struct GlobalCatCacheMapKey
+{
+	Oid dbId; 		/* database oid, if shared cache, 0 */
+	int cacheId; 	/* catcache id */
+} GlobalCatCacheMapKey;
+
+typedef struct GlobalCatCacheMapEnt
+{
+	GlobalCatCacheMapKey key; /* tag of a CatCache hash table */
+	GlobalCatCache *gcp;   /* associated CatCache hash table */
+} GlobalCatCacheMapEnt;
+
+/* LWLock for each partition */
+typedef struct GlobalCatCacheMapLock
+{
+	LWLock locks[NUM_MAP_PARTITIONS];
+} GlobalCatCacheMapLock;
+
+extern GlobalCatCacheMapLock *globalCatCacheMapLock;
+
+/*
+ * Declarations for shared relation cache map
+ */
+typedef Oid GlobalRelCacheMapKey;
+
+typedef struct GlobalRelCacheMapEnt
+{
+	GlobalRelCacheMapKey key; /* key of a RelCache hash table */
+	dshash_table_handle global_relcache_handle; /* RelCache hash table handle */
+	HandleHeader *cache_handle_header; /* header for handles */
+} GlobalRelCacheMapEnt;
+
+typedef struct GlobalRelCacheMapLock
+{
+	LWLock locks[NUM_MAP_PARTITIONS];
+} GlobalRelCacheMapLock;
+
+extern GlobalRelCacheMapLock *globalRelCacheMapLock;
+
+/* GUC parameter: 0 indicates catalog/relation cache is built per backend */
+extern int shared_metacache_mem;
+
+/* Unit is megabyte */
+#define SHM_METACACHE_SIZE_MB ((size_t)(1024 * 1024 * shared_metacache_mem))
+#define GET_HANDLE_LOCK(handle) (&handle->block->handle_lock)
+
+/*
+ * The shared cache mapping table is partitioned to reduce contention.
+ * To determine which partition lock a given key requires, compute the key's
+ * hash code with GlobalCatCacheMapHashCode(), then apply
+ * SharedCachePartitionLock().
+ * NB: NUM_MAP_PARTITIONS must be a power of 2!
+ */
+#define GlobalCacheMapHashPartition(hashcode) \
+	((hashcode) % NUM_MAP_PARTITIONS)
+#define GlobalCatCachePartitionLock(hashcode) \
+	(&globalCatCacheMapLock->locks[GlobalCacheMapHashPartition(hashcode)])
+#define GlobalRelCachePartitionLock(hashcode) \
+	(&globalRelCacheMapLock->locks[GlobalCacheMapHashPartition(hashcode)])
+
+extern void GlobalMetaCacheShmemInit(void);
+extern Size GlobalMetaCacheShmemSize(void);
+
+/* routines for cache handle */
+extern void InitCacheHandle(HandleHeader *header, void *owner);
+extern void InvalCacheHandle(HandleHeader *header, Handle *chunk);
+extern Handle *GetCacheHandle(HandleHeader *header, void *entry);
+
+/* routines for shared hash map */
+extern void GlobalCatCacheMapInitKey(CatCache *lcp,
+									 GlobalCatCacheMapKey *keyPtr);
+
+extern uint32 GlobalCatCacheMapHashCode(GlobalCatCacheMapKey *keyPtr);
+
+extern GlobalCatCacheMapEnt*
+GlobalCatCacheMapLookup(GlobalCatCacheMapKey *keyPtr, uint32 hashcode);
+
+extern bool
+GlobalCatCacheMapInsert(GlobalCatCacheMapKey *keyPtr,
+						uint32 hashcode,
+						GlobalCatCache *gcp);
+
+extern uint32
+GlobalRelCacheMapHashCode(GlobalRelCacheMapKey *keyPtr);
+
+extern GlobalRelCacheMapEnt*
+GlobalRelCacheMapLookup(GlobalRelCacheMapKey *keyPtr, uint32 hashcode);
+
+extern bool
+GlobalRelCacheMapInsert(GlobalRelCacheMapKey *keyPtr,
+						uint32 hashcode,
+						dshash_table_handle global_relcache_handle,
+						HandleHeader *cache_handle_header);
+
+#endif							/* SHARED_META_CACHE_H */
diff --git a/src/include/utils/syscache.h b/src/include/utils/syscache.h
index f27b73d..8b2a46d 100644
--- a/src/include/utils/syscache.h
+++ b/src/include/utils/syscache.h
@@ -18,6 +18,7 @@
 
 #include "access/attnum.h"
 #include "access/htup.h"
+#include "utils/shared_meta_cache.h"
 /* we intentionally do not include utils/catcache.h here */
 
 /*
@@ -160,12 +161,15 @@ struct catclist;
 extern struct catclist *SearchSysCacheList(int cacheId, int nkeys,
 										   Datum key1, Datum key2, Datum key3);
 
-extern void SysCacheInvalidate(int cacheId, uint32 hashValue);
+extern void SysCacheInvalidate(int cacheId, uint32 hashValue, InvalPhase invalPhase);
 
 extern bool RelationInvalidatesSnapshotsOnly(Oid relid);
 extern bool RelationHasSysCache(Oid relid);
 extern bool RelationSupportsSysCache(Oid relid);
 
+/* For GlobalCatCache */
+extern void GlobalSysCacheInvalidate(int cacheId, uint32 hashValue);
+
 /*
  * The use of the macros below rather than direct calls to the corresponding
  * functions is encouraged, as it insulates the caller from changes in the
