speed up pg_upgrade with large number of tables

diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.c
index 27924159d6..570500824d 100644
--- a/src/bin/pg_upgrade/check.c
+++ b/src/bin/pg_upgrade/check.c
@@ -590,6 +590,9 @@ check_and_dump_old_cluster(bool live_check)
 
 	get_loadable_libraries();
 
+	/* skip the below check */
+	if (user_opts.skip_check)
+		goto generate_old_dump;
 
 	/*
 	 * Check for various failure cases
@@ -660,6 +663,8 @@ check_and_dump_old_cluster(bool live_check)
 	if (GET_MAJOR_VERSION(old_cluster.major_version) <= 905)
 		check_for_pg_role_prefix(&old_cluster);
 
+generate_old_dump:
+
 	/*
 	 * While not a check option, we do this now because this is the only time
 	 * the old server is running.
diff --git a/src/bin/pg_upgrade/option.c b/src/bin/pg_upgrade/option.c
index 548ea4e623..a85b6a96fd 100644
--- a/src/bin/pg_upgrade/option.c
+++ b/src/bin/pg_upgrade/option.c
@@ -60,6 +60,7 @@ parseCommandLine(int argc, char *argv[])
 		{"copy", no_argument, NULL, 2},
 		{"copy-file-range", no_argument, NULL, 3},
 		{"sync-method", required_argument, NULL, 4},
+		{"skip-check", no_argument, NULL, 5},
 
 		{NULL, 0, NULL, 0}
 	};
@@ -211,6 +212,9 @@ parseCommandLine(int argc, char *argv[])
 					exit(1);
 				user_opts.sync_method = pg_strdup(optarg);
 				break;
+			case 5:
+				user_opts.skip_check = true;
+				break;
 
 			default:
 				fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
@@ -289,6 +293,7 @@ usage(void)
 	printf(_("  -B, --new-bindir=BINDIR       new cluster executable directory (default\n"
 			 "                                same directory as pg_upgrade)\n"));
 	printf(_("  -c, --check                   check clusters only, don't change any data\n"));
+	printf(_("  --skip-check                  skip the check in actual upgrade \n"));
 	printf(_("  -d, --old-datadir=DATADIR     old cluster data directory\n"));
 	printf(_("  -D, --new-datadir=DATADIR     new cluster data directory\n"));
 	printf(_("  -j, --jobs=NUM                number of simultaneous processes or threads to use\n"));
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c
index af370768b6..e0071d7725 100644
--- a/src/bin/pg_upgrade/pg_upgrade.c
+++ b/src/bin/pg_upgrade/pg_upgrade.c
@@ -125,6 +125,12 @@ main(int argc, char **argv)
 
 	setup(argv[0], &live_check);
 
+	if (live_check && user_opts.skip_check)
+	{
+		pg_fatal("skip-check can't be used in check mode. Please turn off "
+				 "skip-check if you want to perform a live check .\n");
+	}
+
 	output_check_banner(live_check);
 
 	check_cluster_versions();
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h
index 8afe240bdf..adf4ab7e1c 100644
--- a/src/bin/pg_upgrade/pg_upgrade.h
+++ b/src/bin/pg_upgrade/pg_upgrade.h
@@ -327,6 +327,9 @@ typedef struct
 	int			jobs;			/* number of processes/threads to use */
 	char	   *socketdir;		/* directory to use for Unix sockets */
 	char	   *sync_method;
+	bool		skip_check; 	/* true -> skip check in actual upgrade,
+								 * assuming the various failure cases are
+								 * impossible. */
 } UserOpts;
 
 typedef struct

diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index 0400a50777..8a1427cd9b 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -2717,28 +2717,25 @@ FastPathTransferRelationLocks(LockMethod lockMethodTable, const LOCKTAG *locktag
 		PGPROC	   *proc = &ProcGlobal->allProcs[i];
 		uint32		f;
 
-		LWLockAcquire(&proc->fpInfoLock, LW_EXCLUSIVE);
-
 		/*
 		 * If the target backend isn't referencing the same database as the
 		 * lock, then we needn't examine the individual relation IDs at all;
 		 * none of them can be relevant.
 		 *
 		 * proc->databaseId is set at backend startup time and never changes
-		 * thereafter, so it might be safe to perform this test before
-		 * acquiring &proc->fpInfoLock.  In particular, it's certainly safe to
+		 * thereafter, so it is safe to perform this test before acquiring
+		 * &proc->fpInfoLock.  In particular, it's certainly safe to
 		 * assume that if the target backend holds any fast-path locks, it
 		 * must have performed a memory-fencing operation (in particular, an
-		 * LWLock acquisition) since setting proc->databaseId.  However, it's
-		 * less clear that our backend is certain to have performed a memory
-		 * fencing operation since the other backend set proc->databaseId.  So
-		 * for now, we test it after acquiring the LWLock just to be safe.
+		 * LWLock acquisition) since setting proc->databaseId.  So once a read
+		 * barrier is performed, we can see the true proc->database if the
+		 * target backend has acquired its fpInfoLock.
 		 */
+		pg_read_barrier();
 		if (proc->databaseId != locktag->locktag_field1)
-		{
-			LWLockRelease(&proc->fpInfoLock);
 			continue;
-		}
+
+		LWLockAcquire(&proc->fpInfoLock, LW_EXCLUSIVE);
 
 		for (f = 0; f < FP_LOCK_SLOTS_PER_BACKEND; f++)
 		{
@@ -2964,21 +2961,19 @@ GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode, int *countp)
 			if (proc == MyProc)
 				continue;
 
-			LWLockAcquire(&proc->fpInfoLock, LW_SHARED);
-
 			/*
 			 * If the target backend isn't referencing the same database as
 			 * the lock, then we needn't examine the individual relation IDs
 			 * at all; none of them can be relevant.
 			 *
 			 * See FastPathTransferRelationLocks() for discussion of why we do
-			 * this test after acquiring the lock.
+			 * pg_read_barrier here.
 			 */
+			pg_read_barrier();
 			if (proc->databaseId != locktag->locktag_field1)
-			{
-				LWLockRelease(&proc->fpInfoLock);
 				continue;
-			}
+
+			LWLockAcquire(&proc->fpInfoLock, LW_SHARED);
 
 			for (f = 0; f < FP_LOCK_SLOTS_PER_BACKEND; f++)
 			{

diff --git a/src/backend/access/common/relation.c b/src/backend/access/common/relation.c
index d8a313a2c9..d5b41b3887 100644
--- a/src/backend/access/common/relation.c
+++ b/src/backend/access/common/relation.c
@@ -35,7 +35,7 @@
  *		If lockmode is not "NoLock", the specified kind of lock is
  *		obtained on the relation.  (Generally, NoLock should only be
  *		used if the caller knows it has some appropriate lock on the
- *		relation already.)
+ *		relation already or the relation is only visible to the caller.)
  *
  *		An error is raised if the relation does not exist.
  *
@@ -61,12 +61,14 @@ relation_open(Oid relationId, LOCKMODE lockmode)
 		elog(ERROR, "could not open relation with OID %u", relationId);
 
 	/*
-	 * If we didn't get the lock ourselves, assert that caller holds one,
-	 * except in bootstrap mode where no locks are used.
+	 * If we didn't get the lock ourselves, assert that caller holds one or
+	 * this relation is created in current trasaction except in bootstrap
+	 * mode where no locks are used.
 	 */
 	Assert(lockmode != NoLock ||
 		   IsBootstrapProcessingMode() ||
-		   CheckRelationLockedByMe(r, AccessShareLock, true));
+		   CheckRelationLockedByMe(r, AccessShareLock, true) ||
+		   OidIsValid(r->rd_createSubid));
 
 	/* Make note that we've accessed a temporary relation */
 	if (RelationUsesLocalBuffers(r))
diff --git a/src/backend/catalog/toasting.c b/src/backend/catalog/toasting.c
index 738bc46ae8..37f73e98bb 100644
--- a/src/backend/catalog/toasting.c
+++ b/src/backend/catalog/toasting.c
@@ -269,8 +269,11 @@ create_toast_table(Relation rel, Oid toastOid, Oid toastIndexOid,
 	/* make the toast relation visible, else table_open will fail */
 	CommandCounterIncrement();
 
-	/* ShareLock is not really needed here, but take it anyway */
-	toast_rel = table_open(toast_relid, ShareLock);
+	/*
+	 * ShareLock is not really needed here, since this is the only transaction
+	 * the relation is visible to.
+	 */
+	toast_rel = table_open(toast_relid, NoLock);
 
 	/*
 	 * Create unique index on chunk_id, chunk_seq.