pg_upgrade: freeze master data directory before copying to segments

This allows the data to be visible on the segments. The segements
should not interpret any transaction id from master during or after
upgrade.
Co-authored-by: NAsim R P <apraveen@pivotal.io>
Co-authored-by: NKalen Krempely <kkrempely@pivotal.io>

contrib/pg_upgrade/pg_upgrade.c

@@ -48,6 +48,7 @@ static void prepare_new_databases(void);
 static void create_new_objects(void);
 static void copy_clog_xlog_xid(void);
 static void set_frozenxids(bool minmxid_only);
+static void freeze_master_data(void);
 static void setup(char *argv0, bool *live_check);
 static void cleanup(void);
 static void	get_restricted_token(const char *progname);
@@ -103,7 +104,7 @@ main(int argc, char **argv)
 	adjust_data_dir(&new_cluster);
 
 	setup(argv[0], &live_check);
-	
+
 	report_progress(NULL, CHECK, "Checking cluster compatibility");
 	output_check_banner(live_check);
 
@@ -164,6 +165,12 @@ main(int argc, char **argv)
 	 */
 	restore_aosegment_tables();
 
+	if (user_opts.segment_mode == DISPATCHER)
+	{
+		/* freeze master data *right before* stopping */
+		freeze_master_data();
+	}
+
 	stop_postmaster(false);
 
 	/*
@@ -608,6 +615,127 @@ create_new_objects(void)
 	}
 }
 
+
+/*
+ * Greenplum upgrade involves copying the MASTER_DATA_DIRECTORY to
+ * each primary segment. We need to freeze the master data *after* the master
+ * schema has been restored to allow the data to be visible on the segments.
+ * All databases need to be frozen including those where datallowconn is false.
+ *
+ * Note: No further updates should occur after freezing the master data
+ * directory.
+ */
+static void
+freeze_master_data(void)
+{
+	PGconn 			*conn;
+	PGconn			*conn_template1;
+	PGresult		*dbres;
+	PGresult		*txid_res;
+	PGresult		*dbage_res;
+	int				dbnum;
+	int				ntups;
+	int				i_datallowconn;
+	int				i_datname;
+	TransactionId	txid_before;
+	TransactionId	txid_after;
+	int32 			txns_from_freeze;
+
+	prep_status("Freezing all rows in new master after pg_restore");
+
+	/* Temporarily allow connections to all databases for vacuum freeze */
+	conn_template1 = connectToServer(&new_cluster, "template1");
+
+	PQclear(executeQueryOrDie(conn_template1, "set allow_system_table_mods=true"));
+
+	dbres = executeQueryOrDie(conn_template1,
+							  "SELECT datname, datallowconn "
+							  "FROM	pg_catalog.pg_database");
+
+	i_datname = PQfnumber(dbres, "datname");
+	i_datallowconn = PQfnumber(dbres, "datallowconn");
+
+	ntups = PQntuples(dbres);
+	for (dbnum = 0; dbnum < ntups; dbnum++)
+	{
+		char *datallowconn = PQgetvalue(dbres, dbnum, i_datallowconn);
+		char *datname = PQgetvalue(dbres, dbnum, i_datname);
+		char *escaped_datname = pg_malloc(strlen(datname) * 2 + 1);
+		PQescapeStringConn(conn_template1, escaped_datname, datname, strlen(datname), NULL);
+
+		/* For vacuum freeze, temporarily set datallowconn to true. */
+		if (strcmp(datallowconn, "f") == 0)
+			PQclear(executeQueryOrDie(conn_template1,
+									  "UPDATE pg_catalog.pg_database "
+									  "SET datallowconn = true "
+									  "WHERE datname = '%s'", escaped_datname));
+
+		conn = connectToServer(&new_cluster, datname);
+
+		/* Obtain txid_current before vacuum freeze. */
+		txid_res = executeQueryOrDie(conn, "SELECT txid_current()");
+		txid_before = str2uint(PQgetvalue(txid_res, 0, PQfnumber(txid_res, "txid_current")));
+		PQclear(txid_res);
+
+		PQclear(executeQueryOrDie(conn, "VACUUM FREEZE"));
+
+		/*
+		 * Obtain txid_current and age after vacuum freeze.
+		 *
+		 * Note: It is important that this occurs before any other transactions
+		 * are executed so verification succeeds.
+		 */
+		dbage_res = executeQueryOrDie(conn,
+									  "SELECT txid_current(), age(datfrozenxid) "
+									  "FROM pg_catalog.pg_database "
+									  "WHERE datname = '%s'", escaped_datname);
+		txid_after = str2uint(PQgetvalue(dbage_res, 0, PQfnumber(dbage_res, "txid_current")));
+		uint datfrozenxid_age = str2uint(PQgetvalue(dbage_res, 0, PQfnumber(dbage_res, "age")));
+		PQclear(dbage_res);
+
+		/*
+		 * Verify that the database was frozen by checking that the database age
+		 * is less than the number of transactions taken by "VACUUM FREEZE".
+		 * This implies that all transaction ids that are older than the
+		 * "VACUUM FREEZE" transaction are frozen, and that the oldest
+		 * transaction in the database is newer than the "VACUUM FREEZE"
+		 * transaction.
+		 */
+		txns_from_freeze = txid_after - txid_before;
+		if (txns_from_freeze < 0)
+		{
+			/* Needed if a wrap around occurs between txid after and before. */
+			txns_from_freeze = INT32_MAX - Abs(txns_from_freeze);
+		}
+
+		/* Reset datallowconn flag before possibly raising an error. */
+		if (strcmp(datallowconn, "f") == 0)
+			PQclear(executeQueryOrDie(conn_template1,
+									  "UPDATE pg_catalog.pg_database "
+									  "SET datallowconn = false "
+									  "WHERE datname = '%s'", escaped_datname));
+
+		pg_free(escaped_datname);
+		PQfinish(conn);
+
+		if (datfrozenxid_age > txns_from_freeze)
+		{
+			PQfinish(conn_template1);
+			pg_fatal("Error database '%s' was not properly frozen. Database age of %d is older than %d.\n",
+					 datname, datfrozenxid_age, txns_from_freeze);
+		}
+	}
+
+	/* Freeze the tuples updated from resetting datallowconn flag */
+	PQclear(executeQueryOrDie(conn_template1, "VACUUM FREEZE pg_catalog.pg_database"));
+
+	PQclear(dbres);
+
+	PQfinish(conn_template1);
+
+	check_ok();
+}
+
 /*
  * Delete the given subdirectory contents from the new cluster
  */
@@ -807,9 +935,12 @@ set_frozenxids(bool minmxid_only)
 	ntups = PQntuples(dbres);
 	for (dbnum = 0; dbnum < ntups; dbnum++)
 	{
+
 		char	   *datname = PQgetvalue(dbres, dbnum, i_datname);
+		char	   *escaped_datname = NULL;
 		char	   *datallowconn = PQgetvalue(dbres, dbnum, i_datallowconn);
 
+
 		/*
 		 * We must update databases where datallowconn = false, e.g.
 		 * template0, because autovacuum increments their datfrozenxids,
@@ -818,10 +949,14 @@ set_frozenxids(bool minmxid_only)
 		 * this, we temporarily change datallowconn.
 		 */
 		if (strcmp(datallowconn, "f") == 0)
+		{
+			escaped_datname = pg_malloc(strlen(datname) * 2 + 1);
+			PQescapeStringConn(conn_template1, escaped_datname, datname, strlen(datname), NULL);
 			PQclear(executeQueryOrDie(conn_template1,
 									  "UPDATE pg_catalog.pg_database "
 									  "SET	datallowconn = true "
-									  "WHERE datname = '%s'", datname));
+									  "WHERE datname = '%s'", escaped_datname));
+		}
 
 		conn = connectToServer(&new_cluster, datname);
 
@@ -861,10 +996,13 @@ set_frozenxids(bool minmxid_only)
 
 		/* Reset datallowconn flag */
 		if (strcmp(datallowconn, "f") == 0)
+		{
 			PQclear(executeQueryOrDie(conn_template1,
 									  "UPDATE pg_catalog.pg_database "
 									  "SET	datallowconn = false "
-									  "WHERE datname = '%s'", datname));
+									  "WHERE datname = '%s'", escaped_datname));
+			pg_free(escaped_datname);
+		}
 	}
 
 	PQclear(dbres);