diff --git a/src/backend/access/index/genam.c b/src/backend/access/index/genam.c
index 64023eaea5..adfcfc53b6 100644
--- a/src/backend/access/index/genam.c
+++ b/src/backend/access/index/genam.c
@@ -561,7 +561,7 @@ systable_getnext(SysScanDesc sysscan)
  * good crosscheck that the caller is interested in the right tuple.
  */
 bool
-systable_recheck_tuple(SysScanDesc sysscan, HeapTuple tup)
+systable_recheck_tuple(SysScanDesc sysscan, HeapTuple tup, bool dirtysnap)
 {
 	Snapshot	freshsnap;
 	bool		result;
@@ -574,6 +574,9 @@ systable_recheck_tuple(SysScanDesc sysscan, HeapTuple tup)
 	 * acquire snapshots, so we need not register the snapshot.  Those
 	 * facilities are too low-level to have any business scanning tables.
 	 */
+
+	UseDirtyCatalogSnapshot = dirtysnap;
+
 	freshsnap = GetCatalogSnapshot(RelationGetRelid(sysscan->heap_rel));
 
 	result = table_tuple_satisfies_snapshot(sysscan->heap_rel,
diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c
index fe9c714257..3729393e46 100644
--- a/src/backend/catalog/dependency.c
+++ b/src/backend/catalog/dependency.c
@@ -83,6 +83,7 @@
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"
 
+#include "utils/snapmgr.h"
 
 /*
  * Deletion processing requires additional state for each ObjectAddress that
@@ -105,6 +106,7 @@ typedef struct
 #define DEPFLAG_REVERSE		0x0040	/* reverse internal/extension link */
 #define DEPFLAG_IS_PART		0x0080	/* has a partition dependency */
 #define DEPFLAG_SUBOBJECT	0x0100	/* subobject of another deletable object */
+#define DEPFLAG_DIRTY		0x0200	/* reached thanks to dirty snapshot */
 
 
 /* expansible list of ObjectAddresses */
@@ -491,6 +493,7 @@ findDependentObjects(const ObjectAddress *object,
 	int			maxDependentObjects;
 	ObjectAddressStack mystack;
 	ObjectAddressExtra extra;
+	Snapshot dirtySnapshot;
 
 	/*
 	 * If the target object is already being visited in an outer recursion
@@ -566,8 +569,17 @@ findDependentObjects(const ObjectAddress *object,
 		nkeys = 2;
 	}
 
+	/*
+	 * We use a dirty snapshot so that we see all potential dependencies,
+	 * committed or not. Without doing this we would miss objects created
+	 * during in-flight transactions.
+	 */
+	UseDirtyCatalogSnapshot = true;
+
+	dirtySnapshot = GetCatalogSnapshot(RelationGetRelid(*depRel));
+
 	scan = systable_beginscan(*depRel, DependDependerIndexId, true,
-							  NULL, nkeys, key);
+							  dirtySnapshot, nkeys, key);
 
 	/* initialize variables that loop may fill */
 	memset(&owningObject, 0, sizeof(owningObject));
@@ -581,6 +593,16 @@ findDependentObjects(const ObjectAddress *object,
 		otherObject.objectId = foundDep->refobjid;
 		otherObject.objectSubId = foundDep->refobjsubid;
 
+		/*
+		 * dirtySnapshot->xmin is set to the tuple's xmin
+		 * if that is another transaction that's still in
+		 * progress; or to InvalidTransactionId if the
+		 * tuple's xmin is committed good, committed dead,
+		 * or my own xact. See snapshot.h comments.
+		 */
+		if(TransactionIdIsValid(dirtySnapshot->xmin))
+			objflags |= DEPFLAG_DIRTY;
+
 		/*
 		 * When scanning dependencies of a whole object, we may find rows
 		 * linking sub-objects of the object to the object itself.  (Normally,
@@ -704,7 +726,7 @@ findDependentObjects(const ObjectAddress *object,
 				 * interesting anymore.  We test this by checking the
 				 * pg_depend entry (see notes below).
 				 */
-				if (!systable_recheck_tuple(scan, tup))
+				if (!systable_recheck_tuple(scan, tup, true))
 				{
 					systable_endscan(scan);
 					ReleaseDeletionLock(&otherObject);
@@ -859,8 +881,18 @@ findDependentObjects(const ObjectAddress *object,
 	else
 		nkeys = 2;
 
+	/*
+	 * We use a dirty snapshot so that we see all potential dependencies,
+	 * committed or not. Without doing this we would miss objects created
+	 * during in-flight transactions.
+	 */
+
+	UseDirtyCatalogSnapshot = true;
+
+	dirtySnapshot = GetCatalogSnapshot(RelationGetRelid(*depRel));
+
 	scan = systable_beginscan(*depRel, DependReferenceIndexId, true,
-							  NULL, nkeys, key);
+							  dirtySnapshot, nkeys, key);
 
 	while (HeapTupleIsValid(tup = systable_getnext(scan)))
 	{
@@ -893,7 +925,7 @@ findDependentObjects(const ObjectAddress *object,
 		 * if the pg_depend tuple we are looking at is still live. (If the
 		 * object got deleted, the tuple would have been deleted too.)
 		 */
-		if (!systable_recheck_tuple(scan, tup))
+		if (!systable_recheck_tuple(scan, tup, true))
 		{
 			/* release the now-useless lock */
 			ReleaseDeletionLock(&otherObject);
@@ -931,6 +963,10 @@ findDependentObjects(const ObjectAddress *object,
 				break;
 		}
 
+		/* This tuple is for an in-flight transaction. */
+		if(TransactionIdIsValid(dirtySnapshot->xmin))
+			subflags |= DEPFLAG_DIRTY;
+
 		/* And add it to the pending-objects list */
 		if (numDependentObjects >= maxDependentObjects)
 		{
@@ -1025,6 +1061,7 @@ reportDependentObjects(const ObjectAddresses *targetObjects,
 	int			numReportedClient = 0;
 	int			numNotReportedClient = 0;
 	int			i;
+	int			num_dirty;
 
 	/*
 	 * If we need to delete any partition-dependent objects, make sure that
@@ -1036,10 +1073,16 @@ reportDependentObjects(const ObjectAddresses *targetObjects,
 	 * trigger this complaint is to explicitly try to delete one partition of
 	 * a partitioned object.
 	 */
+
+	num_dirty = 0;
+
 	for (i = 0; i < targetObjects->numrefs; i++)
 	{
 		const ObjectAddressExtra *extra = &targetObjects->extras[i];
 
+		if (extra->flags & DEPFLAG_DIRTY)
+			num_dirty++;
+
 		if ((extra->flags & DEPFLAG_IS_PART) &&
 			!(extra->flags & DEPFLAG_PARTITION))
 		{
@@ -1061,6 +1104,7 @@ reportDependentObjects(const ObjectAddresses *targetObjects,
 	 * the work.
 	 */
 	if (behavior == DROP_CASCADE &&
+		num_dirty == 0 &&
 		!message_level_is_interesting(msglevel))
 		return;
 
@@ -1092,6 +1136,10 @@ reportDependentObjects(const ObjectAddresses *targetObjects,
 		if (extra->flags & DEPFLAG_SUBOBJECT)
 			continue;
 
+		/* We need a dirty snapshot to get its description. */
+		if (extra->flags & DEPFLAG_DIRTY)
+			UseDirtyCatalogSnapshot = true;
+
 		objDesc = getObjectDescription(obj, false);
 
 		/* An object being dropped concurrently doesn't need to be reported */
@@ -1118,7 +1166,7 @@ reportDependentObjects(const ObjectAddresses *targetObjects,
 					(errmsg_internal("drop auto-cascades to %s",
 									 objDesc)));
 		}
-		else if (behavior == DROP_RESTRICT)
+		else if (behavior == DROP_RESTRICT || (behavior == DROP_CASCADE && num_dirty > 0))
 		{
 			char	   *otherDesc = getObjectDescription(&extra->dependee,
 														 false);
@@ -1130,8 +1178,12 @@ reportDependentObjects(const ObjectAddresses *targetObjects,
 					/* separate entries with a newline */
 					if (clientdetail.len != 0)
 						appendStringInfoChar(&clientdetail, '\n');
-					appendStringInfo(&clientdetail, _("%s depends on %s"),
-									 objDesc, otherDesc);
+					if (!(extra->flags & DEPFLAG_DIRTY))
+						appendStringInfo(&clientdetail, _("%s depends on %s"),
+										 objDesc, otherDesc);
+					else
+						appendStringInfo(&clientdetail, _("%s (not yet committed) depends on %s"),
+										 objDesc, otherDesc);
 					numReportedClient++;
 				}
 				else
@@ -1139,8 +1191,12 @@ reportDependentObjects(const ObjectAddresses *targetObjects,
 				/* separate entries with a newline */
 				if (logdetail.len != 0)
 					appendStringInfoChar(&logdetail, '\n');
-				appendStringInfo(&logdetail, _("%s depends on %s"),
-								 objDesc, otherDesc);
+				if (!(extra->flags & DEPFLAG_DIRTY))
+					appendStringInfo(&logdetail, _("%s depends on %s"),
+									 objDesc, otherDesc);
+				else
+					appendStringInfo(&logdetail, _("%s (not yet committed) depends on %s"),
+									 objDesc, otherDesc);
 				pfree(otherDesc);
 			}
 			else
@@ -1180,6 +1236,12 @@ reportDependentObjects(const ObjectAddresses *targetObjects,
 
 	if (!ok)
 	{
+		const char *hint_msg;
+		if (num_dirty > 0)
+			hint_msg = "DROP and DROP CASCADE won't work when there are uncommitted dependencies.";
+		else
+			hint_msg = "Use DROP ... CASCADE to drop the dependent objects too.";
+
 		if (origObject)
 			ereport(ERROR,
 					(errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
@@ -1187,14 +1249,14 @@ reportDependentObjects(const ObjectAddresses *targetObjects,
 							getObjectDescription(origObject, false)),
 					 errdetail("%s", clientdetail.data),
 					 errdetail_log("%s", logdetail.data),
-					 errhint("Use DROP ... CASCADE to drop the dependent objects too.")));
+					 errhint("%s", hint_msg)));
 		else
 			ereport(ERROR,
 					(errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
 					 errmsg("cannot drop desired object(s) because other objects depend on them"),
 					 errdetail("%s", clientdetail.data),
 					 errdetail_log("%s", logdetail.data),
-					 errhint("Use DROP ... CASCADE to drop the dependent objects too.")));
+					 errhint("%s", hint_msg)));
 	}
 	else if (numReportedClient > 1)
 	{
diff --git a/src/backend/catalog/pg_depend.c b/src/backend/catalog/pg_depend.c
index 5f37bf6d10..7fdd8c1fe4 100644
--- a/src/backend/catalog/pg_depend.c
+++ b/src/backend/catalog/pg_depend.c
@@ -28,6 +28,7 @@
 #include "utils/fmgroids.h"
 #include "utils/lsyscache.h"
 #include "utils/rel.h"
+#include "utils/snapmgr.h"
 
 
 static bool isObjectPinned(const ObjectAddress *object);
@@ -65,6 +66,12 @@ recordMultipleDependencies(const ObjectAddress *depender,
 				max_slots,
 				slot_init_count,
 				slot_stored_count;
+	Relation    catalog;
+	HeapTuple   tuple;
+	Oid         oidIndexId;
+	SysScanDesc scan;
+	ScanKeyData skey;
+	Snapshot dirtySnapshot;
 
 	if (nreferenced <= 0)
 		return;					/* nothing to do */
@@ -79,6 +86,68 @@ recordMultipleDependencies(const ObjectAddress *depender,
 	if (IsBootstrapProcessingMode())
 		return;
 
+	catalog = table_open(referenced->classId, AccessShareLock);
+	oidIndexId = get_object_oid_index(referenced->classId);
+
+	Assert(OidIsValid(oidIndexId));
+
+	/*
+	 * We use a dirty snapshot so that we see all potential dependencies,
+	 * committed or not. Without doing this we would miss objects created
+	 * during in-flight transactions.
+	 */
+	UseDirtyCatalogSnapshot = true;
+
+	dirtySnapshot = GetCatalogSnapshot(RelationGetRelid(catalog));
+
+	ScanKeyInit(&skey,
+				get_object_attnum_oid(referenced->classId),
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(referenced->objectId));
+
+	scan = systable_beginscan(catalog, oidIndexId, true,
+							  dirtySnapshot, 1, &skey);
+
+	tuple = systable_getnext(scan);
+
+	/*
+	 * dirtySnapshot->xmax is set to the tuple's xmax
+	 * if that is another transaction that's still in
+	 * progress; or to InvalidTransactionId if the
+	 * tuple's xmax is committed good, committed dead,
+	 * or my own xact. See snapshot.h comments.
+	 */
+	if (HeapTupleIsValid(tuple) && TransactionIdIsValid(dirtySnapshot->xmax))
+	{
+		char *dependerDesc;
+		char *referencedDesc;
+		StringInfoData detail;
+
+		initStringInfo(&detail);
+
+		/* We need a dirty snapshot to get its description. */
+		UseDirtyCatalogSnapshot = true;
+		dependerDesc = getObjectDescription(depender, false);
+
+		referencedDesc = getObjectDescription(referenced, false);
+
+
+		appendStringInfo(&detail, _("%s depends on %s (dependency not yet committed)"),
+									dependerDesc,
+									referencedDesc);
+
+		ereport(ERROR,
+				(errcode(ERRCODE_DEPENDENT_OBJECTS_UNCOMMITTED_DEPENDENCY),
+				errmsg("cannot create %s because it depends of other objects uncommitted dependencies",
+				dependerDesc)),
+				errdetail("%s", detail.data),
+				errdetail_log("%s", detail.data),
+				errhint("%s", "CREATE won't work as long as there is uncommitted dependencies."));
+	}
+
+	systable_endscan(scan);
+	table_close(catalog, AccessShareLock);
+
 	dependDesc = table_open(DependRelationId, RowExclusiveLock);
 
 	/*
diff --git a/src/backend/catalog/pg_shdepend.c b/src/backend/catalog/pg_shdepend.c
index c20b1fbb96..5338d52a72 100644
--- a/src/backend/catalog/pg_shdepend.c
+++ b/src/backend/catalog/pg_shdepend.c
@@ -67,6 +67,7 @@
 #include "utils/fmgroids.h"
 #include "utils/memutils.h"
 #include "utils/syscache.h"
+#include "utils/snapmgr.h"
 
 typedef enum
 {
@@ -123,6 +124,12 @@ recordSharedDependencyOn(ObjectAddress *depender,
 						 SharedDependencyType deptype)
 {
 	Relation	sdepRel;
+	Relation    catalog;
+	HeapTuple   tuple;
+	Oid         oidIndexId;
+	SysScanDesc scan;
+	ScanKeyData skey;
+	Snapshot dirtySnapshot;
 
 	/*
 	 * Objects in pg_shdepend can't have SubIds.
@@ -137,6 +144,67 @@ recordSharedDependencyOn(ObjectAddress *depender,
 	if (IsBootstrapProcessingMode())
 		return;
 
+	catalog = table_open(referenced->classId, AccessShareLock);
+	oidIndexId = get_object_oid_index(referenced->classId);
+
+	Assert(OidIsValid(oidIndexId));
+
+	/*
+	 * We use a dirty snapshot so that we see all potential dependencies,
+	 * committed or not. Without doing this we would miss objects created
+	 * during in-flight transactions.
+	 */
+	UseDirtyCatalogSnapshot = true;
+
+	dirtySnapshot = GetCatalogSnapshot(RelationGetRelid(catalog));
+
+	ScanKeyInit(&skey,
+				get_object_attnum_oid(referenced->classId),
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(referenced->objectId));
+
+	scan = systable_beginscan(catalog, oidIndexId, true,
+							  dirtySnapshot, 1, &skey);
+
+	tuple = systable_getnext(scan);
+
+	/*
+	 * dirtySnapshot->xmax is set to the tuple's xmax
+	 * if that is another transaction that's still in
+	 * progress; or to InvalidTransactionId if the
+	 * tuple's xmax is committed good, committed dead,
+	 * or my own xact. See snapshot.h comments.
+	 */
+	if (HeapTupleIsValid(tuple) && TransactionIdIsValid(dirtySnapshot->xmax))
+	{
+		char *dependerDesc;
+		char *referencedDesc;
+		StringInfoData detail;
+
+		initStringInfo(&detail);
+
+		/* We need a dirty snapshot to get its description. */
+		UseDirtyCatalogSnapshot = true;
+		referencedDesc = getObjectDescription(referenced, false);
+
+		dependerDesc = getObjectDescription(depender, false);
+
+		appendStringInfo(&detail, _("%s depends on %s (modifications not yet committed)"),
+									dependerDesc,
+									referencedDesc);
+
+		ereport(ERROR,
+				(errcode(ERRCODE_DEPENDENT_OBJECTS_UNCOMMITTED_DEPENDENCY),
+				errmsg("cannot create %s because it depends of other objects uncommitted dependencies",
+				dependerDesc)),
+				errdetail("%s", detail.data),
+				errdetail_log("%s", detail.data),
+				errhint("%s", "CREATE won't work as long as there is uncommitted modification dependencies."));
+	}
+
+	systable_endscan(scan);
+	table_close(catalog, AccessShareLock);
+
 	sdepRel = table_open(SharedDependRelationId, RowExclusiveLock);
 
 	/* If the referenced object is pinned, do nothing. */
@@ -1316,6 +1384,7 @@ shdepDropOwned(List *roleids, DropBehavior behavior)
 		ScanKeyData key[2];
 		SysScanDesc scan;
 		HeapTuple	tuple;
+		Snapshot dirtySnapshot;
 
 		/* Doesn't work for pinned objects */
 		if (IsPinnedObject(AuthIdRelationId, roleid))
@@ -1333,6 +1402,15 @@ shdepDropOwned(List *roleids, DropBehavior behavior)
 							getObjectDescription(&obj, false))));
 		}
 
+		/*
+		 * We use a dirty snapshot so that we see all potential dependencies,
+		 * committed or not. Without doing this we would miss objects created
+		 * during in-flight transactions.
+		 */
+		UseDirtyCatalogSnapshot = true;
+
+		dirtySnapshot = GetCatalogSnapshot(RelationGetRelid(sdepRel));
+
 		ScanKeyInit(&key[0],
 					Anum_pg_shdepend_refclassid,
 					BTEqualStrategyNumber, F_OIDEQ,
@@ -1343,7 +1421,7 @@ shdepDropOwned(List *roleids, DropBehavior behavior)
 					ObjectIdGetDatum(roleid));
 
 		scan = systable_beginscan(sdepRel, SharedDependReferenceIndexId, true,
-								  NULL, 2, key);
+								  dirtySnapshot, 2, key);
 
 		while ((tuple = systable_getnext(scan)) != NULL)
 		{
@@ -1390,11 +1468,43 @@ shdepDropOwned(List *roleids, DropBehavior behavior)
 						 * object in that case.
 						 */
 						AcquireDeletionLock(&obj, 0);
-						if (!systable_recheck_tuple(scan, tuple))
+						if (!systable_recheck_tuple(scan, tuple, true))
 						{
 							ReleaseDeletionLock(&obj);
 							break;
 						}
+
+						/*
+						 * dirtySnapshot->xmin is set to the tuple's xmin
+						 * if that is another transaction that's still in
+						 * progress; or to InvalidTransactionId if the
+						 * tuple's xmin is committed good, committed dead,
+						 * or my own xact. See snapshot.h comments.
+						 */
+						if(TransactionIdIsValid(dirtySnapshot->xmin))
+						{
+							ObjectAddress roleobj;
+							char *roledesc;
+							char *objdesc;
+
+							roleobj.classId = AuthIdRelationId;
+							roleobj.objectId = roleid;
+							roleobj.objectSubId = 0;
+
+							roledesc = getObjectDescription(&roleobj, false);
+
+							/* We need a dirty snapshot to get its description. */
+							UseDirtyCatalogSnapshot = true;
+							objdesc = getObjectDescription(&obj, false);
+
+							ereport(ERROR,
+									(errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
+									errmsg("cannot drop objects owned by %s because other uncommitted objects depend on it",
+									roledesc),
+									errdetail("%s (not yet committed) depends on %s", objdesc, roledesc),
+									errdetail_log("%s (not yet committed) depends on %s", objdesc, roledesc),
+									errhint("Commit or rollback %s creation.", objdesc)));
+						}
 						add_exact_object_address(&obj, deleteobjs);
 					}
 					break;
@@ -1407,11 +1517,44 @@ shdepDropOwned(List *roleids, DropBehavior behavior)
 						obj.objectSubId = sdepForm->objsubid;
 						/* as above */
 						AcquireDeletionLock(&obj, 0);
-						if (!systable_recheck_tuple(scan, tuple))
+						if (!systable_recheck_tuple(scan, tuple, true))
 						{
 							ReleaseDeletionLock(&obj);
 							break;
 						}
+
+						/*
+						 * dirtySnapshot->xmin is set to the tuple's xmin
+						 * if that is another transaction that's still in
+						 * progress; or to InvalidTransactionId if the
+						 * tuple's xmin is committed good, committed dead,
+						 * or my own xact. See snapshot.h comments.
+						 */
+						if(TransactionIdIsValid(dirtySnapshot->xmin))
+						{
+
+							ObjectAddress roleobj;
+							char *roledesc;
+							char *objdesc;
+
+							roleobj.classId = AuthIdRelationId;
+							roleobj.objectId = roleid;
+							roleobj.objectSubId = 0;
+
+							roledesc = getObjectDescription(&roleobj, false);
+
+							/* We need a dirty snapshot to get its description. */
+							UseDirtyCatalogSnapshot = true;
+							objdesc = getObjectDescription(&obj, false);
+
+							ereport(ERROR,
+									(errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
+									errmsg("cannot drop objects owned by %s because other uncommitted objects depend on it",
+									roledesc),
+									errdetail("%s (not yet committed) depends on %s", objdesc, roledesc),
+									errdetail_log("%s (not yet committed) depends on %s", objdesc, roledesc),
+									errhint("Commit or rollback %s creation.", objdesc)));
+						}
 						add_exact_object_address(&obj, deleteobjs);
 					}
 					break;
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index a9945c80eb..1c3c69060e 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -2261,7 +2261,11 @@ GetSnapshotData(Snapshot snapshot)
 	 */
 	LWLockAcquire(ProcArrayLock, LW_SHARED);
 
-	if (GetSnapshotDataReuse(snapshot))
+	/*
+	 * A dirty snapshot is not a candidate for GetSnapshotDataReuse
+	 * as its xmin and/or xmax may have changed
+	 */
+	if (!IsDirtySnapshot(snapshot) && GetSnapshotDataReuse(snapshot))
 	{
 		LWLockRelease(ProcArrayLock);
 		return snapshot;
diff --git a/src/backend/utils/errcodes.txt b/src/backend/utils/errcodes.txt
index 9874a77805..842cd1e854 100644
--- a/src/backend/utils/errcodes.txt
+++ b/src/backend/utils/errcodes.txt
@@ -270,6 +270,7 @@ Section: Class 2B - Dependent Privilege Descriptors Still Exist
 
 2B000    E    ERRCODE_DEPENDENT_PRIVILEGE_DESCRIPTORS_STILL_EXIST            dependent_privilege_descriptors_still_exist
 2BP01    E    ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST                          dependent_objects_still_exist
+2BP02    E    ERRCODE_DEPENDENT_OBJECTS_UNCOMMITTED_DEPENDENCY               dependent_objects_uncommitted_dependency
 
 Section: Class 2D - Invalid Transaction Termination
 
diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c
index 5001efdf7a..836d8f3508 100644
--- a/src/backend/utils/time/snapmgr.c
+++ b/src/backend/utils/time/snapmgr.c
@@ -95,6 +95,7 @@ volatile OldSnapshotControlData *oldSnapshotControl;
 static SnapshotData CurrentSnapshotData = {SNAPSHOT_MVCC};
 static SnapshotData SecondarySnapshotData = {SNAPSHOT_MVCC};
 SnapshotData CatalogSnapshotData = {SNAPSHOT_MVCC};
+SnapshotData DirtyCatalogSnapshotData = {SNAPSHOT_DIRTY};
 SnapshotData SnapshotSelfData = {SNAPSHOT_SELF};
 SnapshotData SnapshotAnyData = {SNAPSHOT_ANY};
 
@@ -102,6 +103,7 @@ SnapshotData SnapshotAnyData = {SNAPSHOT_ANY};
 static Snapshot CurrentSnapshot = NULL;
 static Snapshot SecondarySnapshot = NULL;
 static Snapshot CatalogSnapshot = NULL;
+static Snapshot DirtyCatalogSnapshot = NULL;
 static Snapshot HistoricSnapshot = NULL;
 
 /*
@@ -147,6 +149,7 @@ static pairingheap RegisteredSnapshots = {&xmin_cmp, NULL, NULL};
 
 /* first GetTransactionSnapshot call in a transaction? */
 bool		FirstSnapshotSet = false;
+bool		UseDirtyCatalogSnapshot = false;
 
 /*
  * Remember the serializable transaction snapshot, if any.  We cannot trust
@@ -310,6 +313,7 @@ GetTransactionSnapshot(void)
 
 	/* Don't allow catalog snapshot to be older than xact snapshot. */
 	InvalidateCatalogSnapshot();
+	InvalidateDirtyCatalogSnapshot();
 
 	CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData);
 
@@ -407,6 +411,9 @@ GetCatalogSnapshot(Oid relid)
 Snapshot
 GetNonHistoricCatalogSnapshot(Oid relid)
 {
+	Snapshot *snap;
+	SnapshotData *snapdata;
+
 	/*
 	 * If the caller is trying to scan a relation that has no syscache, no
 	 * catcache invalidations will be sent when it is updated.  For a few key
@@ -414,15 +421,31 @@ GetNonHistoricCatalogSnapshot(Oid relid)
 	 * scan a relation for which neither catcache nor snapshot invalidations
 	 * are sent, we must refresh the snapshot every time.
 	 */
-	if (CatalogSnapshot &&
+	if (!UseDirtyCatalogSnapshot)
+	{
+		snap = &CatalogSnapshot;
+		snapdata = &CatalogSnapshotData;
+	}
+	else
+	{
+		snap = &DirtyCatalogSnapshot;
+		snapdata = &DirtyCatalogSnapshotData;
+	}
+
+	if (*snap &&
 		!RelationInvalidatesSnapshotsOnly(relid) &&
 		!RelationHasSysCache(relid))
-		InvalidateCatalogSnapshot();
+	{
+		if (!UseDirtyCatalogSnapshot)
+			InvalidateCatalogSnapshot();
+		else
+			InvalidateDirtyCatalogSnapshot();
+	}
 
-	if (CatalogSnapshot == NULL)
+	if (*snap == NULL)
 	{
 		/* Get new snapshot. */
-		CatalogSnapshot = GetSnapshotData(&CatalogSnapshotData);
+		*snap = GetSnapshotData(snapdata);
 
 		/*
 		 * Make sure the catalog snapshot will be accounted for in decisions
@@ -436,10 +459,11 @@ GetNonHistoricCatalogSnapshot(Oid relid)
 		 * NB: it had better be impossible for this to throw error, since the
 		 * CatalogSnapshot pointer is already valid.
 		 */
-		pairingheap_add(&RegisteredSnapshots, &CatalogSnapshot->ph_node);
+		pairingheap_add(&RegisteredSnapshots, &snapdata->ph_node);
 	}
 
-	return CatalogSnapshot;
+	UseDirtyCatalogSnapshot = false;
+	return *snap;
 }
 
 /*
@@ -463,6 +487,26 @@ InvalidateCatalogSnapshot(void)
 	}
 }
 
+/*
+ * InvalidateDirtyCatalogSnapshot
+ *		Mark the current catalog snapshot, if any, as invalid
+ *
+ * We could change this API to allow the caller to provide more fine-grained
+ * invalidation details, so that a change to relation A wouldn't prevent us
+ * from using our cached snapshot to scan relation B, but so far there's no
+ * evidence that the CPU cycles we spent tracking such fine details would be
+ * well-spent.
+ */
+void
+InvalidateDirtyCatalogSnapshot(void)
+{
+	if (DirtyCatalogSnapshot)
+	{
+		pairingheap_remove(&RegisteredSnapshots, &DirtyCatalogSnapshot->ph_node);
+		DirtyCatalogSnapshot = NULL;
+		SnapshotResetXmin();
+	}
+}
 /*
  * InvalidateCatalogSnapshotConditionally
  *		Drop catalog snapshot if it's the only one we have
@@ -1072,6 +1116,7 @@ AtEOXact_Snapshot(bool isCommit, bool resetXmin)
 
 	/* Drop catalog snapshot if any */
 	InvalidateCatalogSnapshot();
+	InvalidateDirtyCatalogSnapshot();
 
 	/* On commit, complain about leftover snapshots */
 	if (isCommit)
@@ -1098,6 +1143,7 @@ AtEOXact_Snapshot(bool isCommit, bool resetXmin)
 	SecondarySnapshot = NULL;
 
 	FirstSnapshotSet = false;
+	UseDirtyCatalogSnapshot = false;
 
 	/*
 	 * During normal commit processing, we call ProcArrayEndTransaction() to
diff --git a/src/include/access/genam.h b/src/include/access/genam.h
index 480a4762f5..0c0db3fb74 100644
--- a/src/include/access/genam.h
+++ b/src/include/access/genam.h
@@ -218,7 +218,7 @@ extern SysScanDesc systable_beginscan(Relation heapRelation,
 									  Snapshot snapshot,
 									  int nkeys, ScanKey key);
 extern HeapTuple systable_getnext(SysScanDesc sysscan);
-extern bool systable_recheck_tuple(SysScanDesc sysscan, HeapTuple tup);
+extern bool systable_recheck_tuple(SysScanDesc sysscan, HeapTuple tup, bool dirtysnap);
 extern void systable_endscan(SysScanDesc sysscan);
 extern SysScanDesc systable_beginscan_ordered(Relation heapRelation,
 											  Relation indexRelation,
diff --git a/src/include/utils/snapmgr.h b/src/include/utils/snapmgr.h
index c6a176cc95..64a26cac9d 100644
--- a/src/include/utils/snapmgr.h
+++ b/src/include/utils/snapmgr.h
@@ -54,6 +54,12 @@ extern TimestampTz GetOldSnapshotThresholdTimestamp(void);
 extern void SnapshotTooOldMagicForTest(void);
 
 extern bool FirstSnapshotSet;
+/*
+ * Used to cover cases where a dirty
+ * catalog snapshot is needed. Currently
+ * used to avoid orphaned dependencies.
+ */
+extern bool UseDirtyCatalogSnapshot;
 
 extern PGDLLIMPORT TransactionId TransactionXmin;
 extern PGDLLIMPORT TransactionId RecentXmin;
@@ -62,6 +68,7 @@ extern PGDLLIMPORT TransactionId RecentXmin;
 extern PGDLLIMPORT SnapshotData SnapshotSelfData;
 extern PGDLLIMPORT SnapshotData SnapshotAnyData;
 extern PGDLLIMPORT SnapshotData CatalogSnapshotData;
+extern PGDLLIMPORT SnapshotData DirtyCatalogSnapshotData;
 
 #define SnapshotSelf		(&SnapshotSelfData)
 #define SnapshotAny			(&SnapshotAnyData)
@@ -97,6 +104,10 @@ extern PGDLLIMPORT SnapshotData CatalogSnapshotData;
 	((snapshot)->snapshot_type == SNAPSHOT_MVCC || \
 	 (snapshot)->snapshot_type == SNAPSHOT_HISTORIC_MVCC)
 
+/* This macro encodes the knowledge of which snapshots are Dirty */
+#define IsDirtySnapshot(snapshot)  \
+	((snapshot)->snapshot_type == SNAPSHOT_DIRTY)
+
 static inline bool
 OldSnapshotThresholdActive(void)
 {
@@ -111,6 +122,7 @@ extern Snapshot GetOldestSnapshot(void);
 extern Snapshot GetCatalogSnapshot(Oid relid);
 extern Snapshot GetNonHistoricCatalogSnapshot(Oid relid);
 extern void InvalidateCatalogSnapshot(void);
+extern void InvalidateDirtyCatalogSnapshot(void);
 extern void InvalidateCatalogSnapshotConditionally(void);
 
 extern void PushActiveSnapshot(Snapshot snapshot);
diff --git a/src/test/modules/Makefile b/src/test/modules/Makefile
index dffc79b2d9..dd162fb37a 100644
--- a/src/test/modules/Makefile
+++ b/src/test/modules/Makefile
@@ -16,6 +16,7 @@ SUBDIRS = \
 		  spgist_name_ops \
 		  test_bloomfilter \
 		  test_ddl_deparse \
+		  test_dependencies \
 		  test_extensions \
 		  test_ginpostinglist \
 		  test_integerset \
diff --git a/src/test/modules/test_dependencies/.gitignore b/src/test/modules/test_dependencies/.gitignore
new file mode 100644
index 0000000000..bf000faac4
--- /dev/null
+++ b/src/test/modules/test_dependencies/.gitignore
@@ -0,0 +1,3 @@
+# Generated subdirectories
+/log/
+/output_iso
diff --git a/src/test/modules/test_dependencies/Makefile b/src/test/modules/test_dependencies/Makefile
new file mode 100644
index 0000000000..a871431e81
--- /dev/null
+++ b/src/test/modules/test_dependencies/Makefile
@@ -0,0 +1,14 @@
+# src/test/modules/test_dependencies/Makefile
+
+ISOLATION = test_dependencies
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = src/test/modules/test_dependencies
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/src/test/modules/test_dependencies/expected/test_dependencies.out b/src/test/modules/test_dependencies/expected/test_dependencies.out
new file mode 100644
index 0000000000..b86cd6a007
--- /dev/null
+++ b/src/test/modules/test_dependencies/expected/test_dependencies.out
@@ -0,0 +1,44 @@
+Parsed test spec with 2 sessions
+
+starting permutation: s1_begin s1_create_function_in_schema s2_drop_schema s1_commit
+step s1_begin: BEGIN;
+step s1_create_function_in_schema: CREATE OR REPLACE FUNCTION testschema.functime() RETURNS TIMESTAMP AS $$
+DECLARE
+   outTS TIMESTAMP;
+BEGIN
+   RETURN CURRENT_TIMESTAMP;
+END;
+$$ LANGUAGE plpgsql volatile;
+
+step s2_drop_schema: drop schema testschema;
+ERROR:  cannot drop schema testschema because other objects depend on it
+step s1_commit: COMMIT;
+
+starting permutation: s2_begin s2_drop_schema s1_create_function_in_schema s2_commit
+step s2_begin: BEGIN;
+step s2_drop_schema: drop schema testschema;
+step s1_create_function_in_schema: CREATE OR REPLACE FUNCTION testschema.functime() RETURNS TIMESTAMP AS $$
+DECLARE
+   outTS TIMESTAMP;
+BEGIN
+   RETURN CURRENT_TIMESTAMP;
+END;
+$$ LANGUAGE plpgsql volatile;
+
+ERROR:  cannot create function testschema.functime() because it depends of other objects uncommitted dependencies
+step s2_commit: COMMIT;
+
+starting permutation: s1_set_session_authorization s1_begin s1_create_function s2_drop_owned_by_toorph s1_commit
+step s1_set_session_authorization: set SESSION AUTHORIZATION toorph;
+step s1_begin: BEGIN;
+step s1_create_function: CREATE OR REPLACE FUNCTION functime() RETURNS TIMESTAMP AS $$
+DECLARE
+   outTS TIMESTAMP;
+BEGIN
+   RETURN CURRENT_TIMESTAMP;
+END;
+$$ LANGUAGE plpgsql volatile;
+
+step s2_drop_owned_by_toorph: drop owned by toorph;
+ERROR:  cannot drop objects owned by role toorph because other uncommitted objects depend on it
+step s1_commit: COMMIT;
diff --git a/src/test/modules/test_dependencies/specs/test_dependencies.spec b/src/test/modules/test_dependencies/specs/test_dependencies.spec
new file mode 100644
index 0000000000..de54571389
--- /dev/null
+++ b/src/test/modules/test_dependencies/specs/test_dependencies.spec
@@ -0,0 +1,48 @@
+setup
+{
+  CREATE SCHEMA testschema;
+  CREATE USER toorph;
+  GRANT ALL PRIVILEGES ON schema public TO toorph;
+}
+
+teardown
+{
+  DROP FUNCTION IF EXISTS functime();
+  DROP FUNCTION IF EXISTS testschema.functime();
+  DROP SCHEMA IF EXISTS testschema;
+  REVOKE ALL PRIVILEGES ON schema public from toorph;
+  DROP USER toorph;
+}
+
+session "s1"
+
+step "s1_begin" { BEGIN; }
+step "s1_create_function_in_schema" { CREATE OR REPLACE FUNCTION testschema.functime() RETURNS TIMESTAMP AS $$
+DECLARE
+   outTS TIMESTAMP;
+BEGIN
+   RETURN CURRENT_TIMESTAMP;
+END;
+$$ LANGUAGE plpgsql volatile;
+ }
+step "s1_create_function" { CREATE OR REPLACE FUNCTION functime() RETURNS TIMESTAMP AS $$
+DECLARE
+   outTS TIMESTAMP;
+BEGIN
+   RETURN CURRENT_TIMESTAMP;
+END;
+$$ LANGUAGE plpgsql volatile;
+ }
+step "s1_set_session_authorization" { set SESSION AUTHORIZATION toorph; }
+step "s1_commit" { COMMIT; }
+
+session "s2"
+
+step "s2_begin" { BEGIN; }
+step "s2_drop_schema" { drop schema testschema; }
+step "s2_drop_owned_by_toorph" { drop owned by toorph; }
+step "s2_commit" { COMMIT; }
+
+permutation "s1_begin" "s1_create_function_in_schema" "s2_drop_schema" "s1_commit"
+permutation "s2_begin" "s2_drop_schema" "s1_create_function_in_schema" "s2_commit"
+permutation "s1_set_session_authorization" "s1_begin" "s1_create_function" "s2_drop_owned_by_toorph" "s1_commit"
diff --git a/src/tools/msvc/Mkvcbuild.pm b/src/tools/msvc/Mkvcbuild.pm
index 404c45a6f3..d3c3a18331 100644
--- a/src/tools/msvc/Mkvcbuild.pm
+++ b/src/tools/msvc/Mkvcbuild.pm
@@ -47,7 +47,7 @@ my @contrib_excludes = (
 	'hstore_plperl',    'hstore_plpython',
 	'intagg',           'jsonb_plperl',
 	'jsonb_plpython',   'ltree_plpython',
-	'sepgsql',
+	'sepgsql',			'test_dependencies',
 	'brin',             'test_extensions',
 	'test_misc',        'test_pg_dump',
 	'snapshot_too_old', 'unsafe_tests');