[PATCH v5 2/5] Introduce a new syntax to add/drop publications

agora inbox for [email protected]  
help / color / mirror / Atom feed

[PATCH v5 2/5] Introduce a new syntax to add/drop publications
6+ messages / 2 participants
[nested] [flat]

* [PATCH v5 2/5] Introduce a new syntax to add/drop publications
@ 2021-02-13 05:11  Japin Li <[email protected]>
  0 siblings, 0 replies; 6+ messages in thread

From: Japin Li @ 2021-02-13 05:11 UTC (permalink / raw)

At present, if we want to update publications in subscription, we can
use SET PUBLICATION, however, it requires supply all publications that
exists and the new publications if we want to add new publications, it's
inconvenient.  The new syntax only supply the new publications.  When
the refresh is true, it only refresh the new publications.
---
 src/backend/commands/subscriptioncmds.c | 144 ++++++++++++++++++++++++
 src/backend/parser/gram.y               |  20 ++++
 src/include/nodes/parsenodes.h          |   2 +
 3 files changed, 166 insertions(+)

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 5cf874e0b4..ad271deb64 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -47,6 +47,7 @@
 #include "utils/syscache.h"
 
 static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
+static List *merge_subpublications(HeapTuple tuple, List *newpublist, bool addpub);
 static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err);
 
 
@@ -962,6 +963,53 @@ AlterSubscription(AlterSubscriptionStmt *stmt, bool isTopLevel)
 				break;
 			}
 
+		case ALTER_SUBSCRIPTION_ADD_PUBLICATION:
+		case ALTER_SUBSCRIPTION_DROP_PUBLICATION:
+			{
+				bool    copy_data = false;
+				bool    isadd = stmt->kind == ALTER_SUBSCRIPTION_ADD_PUBLICATION;
+				bool    refresh;
+				List   *publist = NIL;
+
+				publist = merge_subpublications(tup, stmt->publication, isadd);
+
+				parse_subscription_options(stmt->options,
+										   NULL,	/* no "connect" */
+										   NULL, NULL,	/* no "enabled" */
+										   NULL,	/* no "create_slot" */
+										   NULL, NULL,	/* no "slot_name" */
+										   isadd ? &copy_data : NULL,	/* for drop, no "copy_data" */
+										   NULL,	/* no "synchronous_commit" */
+										   &refresh,
+										   NULL, NULL,	/* no "binary" */
+										   NULL, NULL); /* no "streaming" */
+
+				values[Anum_pg_subscription_subpublications - 1] =
+					publicationListToArray(publist);
+				replaces[Anum_pg_subscription_subpublications - 1] = true;
+
+				update_tuple = true;
+
+				/* Refresh if user asked us to. */
+				if (refresh)
+				{
+					if (!sub->enabled)
+						ereport(ERROR,
+								(errcode(ERRCODE_SYNTAX_ERROR),
+								 errmsg("ALTER SUBSCRIPTION with refresh is not allowed for disabled subscriptions"),
+								 errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION ... WITH (refresh = false).")));
+
+					PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION with refresh");
+
+					/* Only refresh the added/dropped list of publications. */
+					sub->publications = stmt->publication;
+
+					AlterSubscription_refresh(sub, copy_data);
+				}
+
+				break;
+			}
+
 		case ALTER_SUBSCRIPTION_REFRESH:
 			{
 				bool		copy_data;
@@ -1546,3 +1594,99 @@ ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err)
 			 errhint("Use %s to disassociate the subscription from the slot.",
 					 "ALTER SUBSCRIPTION ... SET (slot_name = NONE)")));
 }
+
+/*
+ * Merge ccurrent subscription's publications and user specified publications
+ * by ADD/DROP PUBLICATIONS.
+ *
+ * If isadd == true, we will add the list of publications into current
+ * subscription's publications.  Otherwise, we will delete the list of
+ * publications from current subscription's publications.
+ */
+static List *
+merge_subpublications(HeapTuple tuple, List *newpublist, bool addpub)
+{
+	Datum	datum;
+	bool	isnull;
+	List   *publist = NIL;
+	List   *errlist = NIL;
+	ListCell	*lc;
+
+	/* Get publications */
+	datum = SysCacheGetAttr(SUBSCRIPTIONOID,
+							tuple,
+							Anum_pg_subscription_subpublications,
+							&isnull);
+	Assert(!isnull);
+	publist = textarray_to_stringlist(DatumGetArrayTypeP(datum));
+
+	foreach(lc, newpublist)
+	{
+		char		*name = strVal(lfirst(lc));
+		ListCell	*cell = NULL;
+
+		foreach(cell, publist)
+		{
+			char	*pubname = strVal(lfirst(cell));
+
+			if (strcmp(name, pubname) == 0)
+			{
+				if (addpub)
+					errlist = lappend(errlist, makeString(name));
+				else
+					publist = list_delete_cell(publist, cell);
+				break;
+			}
+		}
+
+		if (addpub && cell == NULL)
+			publist = lappend(publist, makeString(name));
+		else if (!addpub && cell == NULL)
+			errlist = lappend(errlist, makeString(name));
+	}
+
+	if (errlist != NIL)
+	{
+		StringInfoData	errstr;
+		bool			first = true;
+		int				len = list_length(errlist);
+
+		initStringInfo(&errstr);
+		if (len == 1)
+			appendStringInfo(&errstr, "publication name");
+		else
+			appendStringInfo(&errstr, "publication names");
+
+		appendStringInfoString(&errstr, " \"");
+		foreach(lc, errlist)
+		{
+			char	*name = strVal(lfirst(lc));
+
+			if (first)
+				appendStringInfo(&errstr, "%s", name);
+			else
+				appendStringInfo(&errstr, ", %s", name);
+
+			first = false;
+		}
+		appendStringInfoChar(&errstr, '"');
+
+		if (addpub)
+			appendStringInfo(&errstr, " %s already in subscription",
+							 len == 1 ? "is" : "are");
+		else
+			appendStringInfo(&errstr, " %s no exist in subscription",
+							 len == 1 ? "does" : "do");
+
+		ereport(ERROR,
+				(errcode(ERRCODE_SYNTAX_ERROR),
+				 errmsg("%s", errstr.data)));
+	}
+
+	if (publist == NIL)
+		ereport(ERROR,
+				(errcode(ERRCODE_SYNTAX_ERROR),
+				 errmsg("subscription must contain at least one publication")));
+
+	return publist;
+}
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index dd72a9fc3c..e45f98d353 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -9609,6 +9609,26 @@ AlterSubscriptionStmt:
 					n->options = $6;
 					$$ = (Node *)n;
 				}
+			| ALTER SUBSCRIPTION name ADD_P PUBLICATION name_list opt_definition
+				{
+					AlterSubscriptionStmt *n =
+						makeNode(AlterSubscriptionStmt);
+					n->kind = ALTER_SUBSCRIPTION_ADD_PUBLICATION;
+					n->subname = $3;
+					n->publication = $6;
+					n->options = $7;
+					$$ = (Node *)n;
+				}
+			| ALTER SUBSCRIPTION name DROP PUBLICATION name_list opt_definition
+				{
+					AlterSubscriptionStmt *n =
+						makeNode(AlterSubscriptionStmt);
+					n->kind = ALTER_SUBSCRIPTION_DROP_PUBLICATION;
+					n->subname = $3;
+					n->publication = $6;
+					n->options = $7;
+					$$ = (Node *)n;
+				}
 			| ALTER SUBSCRIPTION name SET PUBLICATION name_list opt_definition
 				{
 					AlterSubscriptionStmt *n =
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 236832a2ca..e109607936 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3580,6 +3580,8 @@ typedef enum AlterSubscriptionType
 	ALTER_SUBSCRIPTION_OPTIONS,
 	ALTER_SUBSCRIPTION_CONNECTION,
 	ALTER_SUBSCRIPTION_PUBLICATION,
+	ALTER_SUBSCRIPTION_ADD_PUBLICATION,
+	ALTER_SUBSCRIPTION_DROP_PUBLICATION,
 	ALTER_SUBSCRIPTION_REFRESH,
 	ALTER_SUBSCRIPTION_ENABLED
 } AlterSubscriptionType;
-- 
2.25.1


--=-=-=
Content-Type: text/x-patch
Content-Disposition: attachment;
 filename=v5-0003-Test-ALTER-SUBSCRIPTION-.-ADD-DROP-PUBLICATION.patch



^ permalink  raw  reply  [nested|flat] 6+ messages in thread

* [PATCH v7 10/11] parallelize tables with oids check in pg_upgrade
@ 2024-07-09 02:42  Nathan Bossart <[email protected]>
  0 siblings, 0 replies; 6+ messages in thread

From: Nathan Bossart @ 2024-07-09 02:42 UTC (permalink / raw)

---
 src/bin/pg_upgrade/check.c | 90 ++++++++++++++++++++------------------
 1 file changed, 48 insertions(+), 42 deletions(-)

diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.c
index 32c29aa881..410d8b0cad 100644
--- a/src/bin/pg_upgrade/check.c
+++ b/src/bin/pg_upgrade/check.c
@@ -1520,15 +1520,58 @@ check_for_incompatible_polymorphics(ClusterInfo *cluster)
 	termPQExpBuffer(&state.old_polymorphics);
 }
 
+static char *
+with_oids_query(void *arg)
+{
+	return pg_strdup("SELECT n.nspname, c.relname "
+					 "FROM   pg_catalog.pg_class c, "
+					 "       pg_catalog.pg_namespace n "
+					 "WHERE  c.relnamespace = n.oid AND "
+					 "       c.relhasoids AND"
+					 "       n.nspname NOT IN ('pg_catalog')");
+}
+
+static void
+with_oids_process(DbInfo *dbinfo, PGresult *res, void *arg)
+{
+	bool		db_used = false;
+	int			ntups = PQntuples(res);
+	char		output_path[MAXPGPATH];
+	int			i_nspname = PQfnumber(res, "nspname");
+	int			i_relname = PQfnumber(res, "relname");
+	FILE	  **script = (FILE **) arg;
+
+	if (!ntups)
+		return;
+
+	snprintf(output_path, sizeof(output_path), "%s/%s",
+			 log_opts.basedir,
+			 "tables_with_oids.txt");
+
+	for (int rowno = 0; rowno < ntups; rowno++)
+	{
+		if (*script == NULL && (*script = fopen_priv(output_path, "w")) == NULL)
+			pg_fatal("could not open file \"%s\": %m", output_path);
+		if (!db_used)
+		{
+			fprintf(*script, "In database: %s\n", dbinfo->db_name);
+			db_used = true;
+		}
+		fprintf(*script, "  %s.%s\n",
+				PQgetvalue(res, rowno, i_nspname),
+				PQgetvalue(res, rowno, i_relname));
+	}
+}
+
 /*
  * Verify that no tables are declared WITH OIDS.
  */
 static void
 check_for_tables_with_oids(ClusterInfo *cluster)
 {
-	int			dbnum;
 	FILE	   *script = NULL;
 	char		output_path[MAXPGPATH];
+	AsyncTask  *task = async_task_create();
 
 	prep_status("Checking for tables WITH OIDS");
 
@@ -1536,47 +1579,10 @@ check_for_tables_with_oids(ClusterInfo *cluster)
 			 log_opts.basedir,
 			 "tables_with_oids.txt");
 
-	/* Find any tables declared WITH OIDS */
-	for (dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)
-	{
-		PGresult   *res;
-		bool		db_used = false;
-		int			ntups;
-		int			rowno;
-		int			i_nspname,
-					i_relname;
-		DbInfo	   *active_db = &cluster->dbarr.dbs[dbnum];
-		PGconn	   *conn = connectToServer(cluster, active_db->db_name);
-
-		res = executeQueryOrDie(conn,
-								"SELECT n.nspname, c.relname "
-								"FROM	pg_catalog.pg_class c, "
-								"		pg_catalog.pg_namespace n "
-								"WHERE	c.relnamespace = n.oid AND "
-								"		c.relhasoids AND"
-								"       n.nspname NOT IN ('pg_catalog')");
-
-		ntups = PQntuples(res);
-		i_nspname = PQfnumber(res, "nspname");
-		i_relname = PQfnumber(res, "relname");
-		for (rowno = 0; rowno < ntups; rowno++)
-		{
-			if (script == NULL && (script = fopen_priv(output_path, "w")) == NULL)
-				pg_fatal("could not open file \"%s\": %m", output_path);
-			if (!db_used)
-			{
-				fprintf(script, "In database: %s\n", active_db->db_name);
-				db_used = true;
-			}
-			fprintf(script, "  %s.%s\n",
-					PQgetvalue(res, rowno, i_nspname),
-					PQgetvalue(res, rowno, i_relname));
-		}
-
-		PQclear(res);
-
-		PQfinish(conn);
-	}
+	async_task_add_step(task, with_oids_query,
+						with_oids_process, true, &script);
+	async_task_run(task, cluster);
+	async_task_free(task);
 
 	if (script)
 	{
-- 
2.39.3 (Apple Git-146)


--71XO4bcTeIVOb3nX
Content-Type: text/plain; charset=us-ascii
Content-Disposition: attachment;
	filename="v7-0011-parallelize-user-defined-encoding-conversions-che.patch"



^ permalink  raw  reply  [nested|flat] 6+ messages in thread

* [PATCH v8 10/11] parallelize tables with oids check in pg_upgrade
@ 2024-07-09 02:42  Nathan Bossart <[email protected]>
  0 siblings, 0 replies; 6+ messages in thread

From: Nathan Bossart @ 2024-07-09 02:42 UTC (permalink / raw)

---
 src/bin/pg_upgrade/check.c | 85 +++++++++++++++++++-------------------
 1 file changed, 43 insertions(+), 42 deletions(-)

diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.c
index af7d093581..4156257843 100644
--- a/src/bin/pg_upgrade/check.c
+++ b/src/bin/pg_upgrade/check.c
@@ -1521,15 +1521,53 @@ check_for_incompatible_polymorphics(ClusterInfo *cluster)
 	pg_free(query);
 }
 
+static void
+with_oids_process(DbInfo *dbinfo, PGresult *res, void *arg)
+{
+	bool		db_used = false;
+	int			ntups = PQntuples(res);
+	char		output_path[MAXPGPATH];
+	int			i_nspname = PQfnumber(res, "nspname");
+	int			i_relname = PQfnumber(res, "relname");
+	FILE	  **script = (FILE **) arg;
+
+	if (!ntups)
+		return;
+
+	snprintf(output_path, sizeof(output_path), "%s/%s",
+			 log_opts.basedir,
+			 "tables_with_oids.txt");
+
+	for (int rowno = 0; rowno < ntups; rowno++)
+	{
+		if (*script == NULL && (*script = fopen_priv(output_path, "w")) == NULL)
+			pg_fatal("could not open file \"%s\": %m", output_path);
+		if (!db_used)
+		{
+			fprintf(*script, "In database: %s\n", dbinfo->db_name);
+			db_used = true;
+		}
+		fprintf(*script, "  %s.%s\n",
+				PQgetvalue(res, rowno, i_nspname),
+				PQgetvalue(res, rowno, i_relname));
+	}
+}
+
 /*
  * Verify that no tables are declared WITH OIDS.
  */
 static void
 check_for_tables_with_oids(ClusterInfo *cluster)
 {
-	int			dbnum;
 	FILE	   *script = NULL;
 	char		output_path[MAXPGPATH];
+	AsyncTask  *task = async_task_create();
+	const char *query = "SELECT n.nspname, c.relname "
+		"FROM   pg_catalog.pg_class c, "
+		"       pg_catalog.pg_namespace n "
+		"WHERE  c.relnamespace = n.oid AND "
+		"       c.relhasoids AND"
+		"       n.nspname NOT IN ('pg_catalog')";
 
 	prep_status("Checking for tables WITH OIDS");
 
@@ -1537,47 +1575,10 @@ check_for_tables_with_oids(ClusterInfo *cluster)
 			 log_opts.basedir,
 			 "tables_with_oids.txt");
 
-	/* Find any tables declared WITH OIDS */
-	for (dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)
-	{
-		PGresult   *res;
-		bool		db_used = false;
-		int			ntups;
-		int			rowno;
-		int			i_nspname,
-					i_relname;
-		DbInfo	   *active_db = &cluster->dbarr.dbs[dbnum];
-		PGconn	   *conn = connectToServer(cluster, active_db->db_name);
-
-		res = executeQueryOrDie(conn,
-								"SELECT n.nspname, c.relname "
-								"FROM	pg_catalog.pg_class c, "
-								"		pg_catalog.pg_namespace n "
-								"WHERE	c.relnamespace = n.oid AND "
-								"		c.relhasoids AND"
-								"       n.nspname NOT IN ('pg_catalog')");
-
-		ntups = PQntuples(res);
-		i_nspname = PQfnumber(res, "nspname");
-		i_relname = PQfnumber(res, "relname");
-		for (rowno = 0; rowno < ntups; rowno++)
-		{
-			if (script == NULL && (script = fopen_priv(output_path, "w")) == NULL)
-				pg_fatal("could not open file \"%s\": %m", output_path);
-			if (!db_used)
-			{
-				fprintf(script, "In database: %s\n", active_db->db_name);
-				db_used = true;
-			}
-			fprintf(script, "  %s.%s\n",
-					PQgetvalue(res, rowno, i_nspname),
-					PQgetvalue(res, rowno, i_relname));
-		}
-
-		PQclear(res);
-
-		PQfinish(conn);
-	}
+	async_task_add_step(task, query,
+						with_oids_process, true, &script);
+	async_task_run(task, cluster);
+	async_task_free(task);
 
 	if (script)
 	{
-- 
2.39.3 (Apple Git-146)


--7bj13dLpkt2vgUoO
Content-Type: text/plain; charset=us-ascii
Content-Disposition: attachment;
	filename="v8-0011-parallelize-user-defined-encoding-conversions-che.patch"



^ permalink  raw  reply  [nested|flat] 6+ messages in thread

* [PATCH v9 10/11] parallelize tables with oids check in pg_upgrade
@ 2024-07-09 02:42  Nathan Bossart <[email protected]>
  0 siblings, 0 replies; 6+ messages in thread

From: Nathan Bossart @ 2024-07-09 02:42 UTC (permalink / raw)

---
 src/bin/pg_upgrade/check.c | 85 +++++++++++++++++++-------------------
 1 file changed, 43 insertions(+), 42 deletions(-)

diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.c
index af7d093581..4156257843 100644
--- a/src/bin/pg_upgrade/check.c
+++ b/src/bin/pg_upgrade/check.c
@@ -1521,15 +1521,53 @@ check_for_incompatible_polymorphics(ClusterInfo *cluster)
 	pg_free(query);
 }
 
+static void
+with_oids_process(DbInfo *dbinfo, PGresult *res, void *arg)
+{
+	bool		db_used = false;
+	int			ntups = PQntuples(res);
+	char		output_path[MAXPGPATH];
+	int			i_nspname = PQfnumber(res, "nspname");
+	int			i_relname = PQfnumber(res, "relname");
+	FILE	  **script = (FILE **) arg;
+
+	if (!ntups)
+		return;
+
+	snprintf(output_path, sizeof(output_path), "%s/%s",
+			 log_opts.basedir,
+			 "tables_with_oids.txt");
+
+	for (int rowno = 0; rowno < ntups; rowno++)
+	{
+		if (*script == NULL && (*script = fopen_priv(output_path, "w")) == NULL)
+			pg_fatal("could not open file \"%s\": %m", output_path);
+		if (!db_used)
+		{
+			fprintf(*script, "In database: %s\n", dbinfo->db_name);
+			db_used = true;
+		}
+		fprintf(*script, "  %s.%s\n",
+				PQgetvalue(res, rowno, i_nspname),
+				PQgetvalue(res, rowno, i_relname));
+	}
+}
+
 /*
  * Verify that no tables are declared WITH OIDS.
  */
 static void
 check_for_tables_with_oids(ClusterInfo *cluster)
 {
-	int			dbnum;
 	FILE	   *script = NULL;
 	char		output_path[MAXPGPATH];
+	AsyncTask  *task = async_task_create();
+	const char *query = "SELECT n.nspname, c.relname "
+		"FROM   pg_catalog.pg_class c, "
+		"       pg_catalog.pg_namespace n "
+		"WHERE  c.relnamespace = n.oid AND "
+		"       c.relhasoids AND"
+		"       n.nspname NOT IN ('pg_catalog')";
 
 	prep_status("Checking for tables WITH OIDS");
 
@@ -1537,47 +1575,10 @@ check_for_tables_with_oids(ClusterInfo *cluster)
 			 log_opts.basedir,
 			 "tables_with_oids.txt");
 
-	/* Find any tables declared WITH OIDS */
-	for (dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)
-	{
-		PGresult   *res;
-		bool		db_used = false;
-		int			ntups;
-		int			rowno;
-		int			i_nspname,
-					i_relname;
-		DbInfo	   *active_db = &cluster->dbarr.dbs[dbnum];
-		PGconn	   *conn = connectToServer(cluster, active_db->db_name);
-
-		res = executeQueryOrDie(conn,
-								"SELECT n.nspname, c.relname "
-								"FROM	pg_catalog.pg_class c, "
-								"		pg_catalog.pg_namespace n "
-								"WHERE	c.relnamespace = n.oid AND "
-								"		c.relhasoids AND"
-								"       n.nspname NOT IN ('pg_catalog')");
-
-		ntups = PQntuples(res);
-		i_nspname = PQfnumber(res, "nspname");
-		i_relname = PQfnumber(res, "relname");
-		for (rowno = 0; rowno < ntups; rowno++)
-		{
-			if (script == NULL && (script = fopen_priv(output_path, "w")) == NULL)
-				pg_fatal("could not open file \"%s\": %m", output_path);
-			if (!db_used)
-			{
-				fprintf(script, "In database: %s\n", active_db->db_name);
-				db_used = true;
-			}
-			fprintf(script, "  %s.%s\n",
-					PQgetvalue(res, rowno, i_nspname),
-					PQgetvalue(res, rowno, i_relname));
-		}
-
-		PQclear(res);
-
-		PQfinish(conn);
-	}
+	async_task_add_step(task, query,
+						with_oids_process, true, &script);
+	async_task_run(task, cluster);
+	async_task_free(task);
 
 	if (script)
 	{
-- 
2.39.3 (Apple Git-146)


--lJQ6g+oBmWDgF+NV
Content-Type: text/plain; charset=us-ascii
Content-Disposition: attachment;
	filename="v9-0011-parallelize-user-defined-encoding-conversions-che.patch"



^ permalink  raw  reply  [nested|flat] 6+ messages in thread

* [PATCH v4 11/12] parallelize tables with oids check in pg_upgrade
@ 2024-07-09 02:42  Nathan Bossart <[email protected]>
  0 siblings, 0 replies; 6+ messages in thread

From: Nathan Bossart @ 2024-07-09 02:42 UTC (permalink / raw)

---
 src/bin/pg_upgrade/check.c | 90 ++++++++++++++++++++------------------
 1 file changed, 48 insertions(+), 42 deletions(-)

diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.c
index 6c11732e70..639a62d426 100644
--- a/src/bin/pg_upgrade/check.c
+++ b/src/bin/pg_upgrade/check.c
@@ -1518,15 +1518,58 @@ check_for_incompatible_polymorphics(ClusterInfo *cluster)
 	termPQExpBuffer(&state.old_polymorphics);
 }
 
+static char *
+with_oids_query(DbInfo *dbinfo, void *arg)
+{
+	return pg_strdup("SELECT n.nspname, c.relname "
+					 "FROM   pg_catalog.pg_class c, "
+					 "       pg_catalog.pg_namespace n "
+					 "WHERE  c.relnamespace = n.oid AND "
+					 "       c.relhasoids AND"
+					 "       n.nspname NOT IN ('pg_catalog')");
+}
+
+static void
+with_oids_process(DbInfo *dbinfo, PGresult *res, void *arg)
+{
+	bool		db_used = false;
+	int			ntups = PQntuples(res);
+	char		output_path[MAXPGPATH];
+	int			i_nspname = PQfnumber(res, "nspname");
+	int			i_relname = PQfnumber(res, "relname");
+	FILE	  **script = (FILE **) arg;
+
+	if (!ntups)
+		return;
+
+	snprintf(output_path, sizeof(output_path), "%s/%s",
+			 log_opts.basedir,
+			 "tables_with_oids.txt");
+
+	for (int rowno = 0; rowno < ntups; rowno++)
+	{
+		if (*script == NULL && (*script = fopen_priv(output_path, "w")) == NULL)
+			pg_fatal("could not open file \"%s\": %m", output_path);
+		if (!db_used)
+		{
+			fprintf(*script, "In database: %s\n", dbinfo->db_name);
+			db_used = true;
+		}
+		fprintf(*script, "  %s.%s\n",
+				PQgetvalue(res, rowno, i_nspname),
+				PQgetvalue(res, rowno, i_relname));
+	}
+}
+
 /*
  * Verify that no tables are declared WITH OIDS.
  */
 static void
 check_for_tables_with_oids(ClusterInfo *cluster)
 {
-	int			dbnum;
 	FILE	   *script = NULL;
 	char		output_path[MAXPGPATH];
+	AsyncTask  *task = async_task_create();
 
 	prep_status("Checking for tables WITH OIDS");
 
@@ -1534,47 +1577,10 @@ check_for_tables_with_oids(ClusterInfo *cluster)
 			 log_opts.basedir,
 			 "tables_with_oids.txt");
 
-	/* Find any tables declared WITH OIDS */
-	for (dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)
-	{
-		PGresult   *res;
-		bool		db_used = false;
-		int			ntups;
-		int			rowno;
-		int			i_nspname,
-					i_relname;
-		DbInfo	   *active_db = &cluster->dbarr.dbs[dbnum];
-		PGconn	   *conn = connectToServer(cluster, active_db->db_name);
-
-		res = executeQueryOrDie(conn,
-								"SELECT n.nspname, c.relname "
-								"FROM	pg_catalog.pg_class c, "
-								"		pg_catalog.pg_namespace n "
-								"WHERE	c.relnamespace = n.oid AND "
-								"		c.relhasoids AND"
-								"       n.nspname NOT IN ('pg_catalog')");
-
-		ntups = PQntuples(res);
-		i_nspname = PQfnumber(res, "nspname");
-		i_relname = PQfnumber(res, "relname");
-		for (rowno = 0; rowno < ntups; rowno++)
-		{
-			if (script == NULL && (script = fopen_priv(output_path, "w")) == NULL)
-				pg_fatal("could not open file \"%s\": %m", output_path);
-			if (!db_used)
-			{
-				fprintf(script, "In database: %s\n", active_db->db_name);
-				db_used = true;
-			}
-			fprintf(script, "  %s.%s\n",
-					PQgetvalue(res, rowno, i_nspname),
-					PQgetvalue(res, rowno, i_relname));
-		}
-
-		PQclear(res);
-
-		PQfinish(conn);
-	}
+	async_task_add_step(task, with_oids_query,
+						with_oids_process, true, &script);
+	async_task_run(task, cluster);
+	async_task_free(task);
 
 	if (script)
 	{
-- 
2.39.3 (Apple Git-146)


--+rVtD9QSctUEvLG2
Content-Type: text/plain; charset=us-ascii
Content-Disposition: attachment;
	filename="v4-0012-parallelize-user-defined-encoding-conversions-che.patch"



^ permalink  raw  reply  [nested|flat] 6+ messages in thread

* [PATCH v5 12/13] parallelize tables with oids check in pg_upgrade
@ 2024-07-09 02:42  Nathan Bossart <[email protected]>
  0 siblings, 0 replies; 6+ messages in thread

From: Nathan Bossart @ 2024-07-09 02:42 UTC (permalink / raw)

---
 src/bin/pg_upgrade/check.c | 90 ++++++++++++++++++++------------------
 1 file changed, 48 insertions(+), 42 deletions(-)

diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.c
index 614464dfd3..a09eafa16e 100644
--- a/src/bin/pg_upgrade/check.c
+++ b/src/bin/pg_upgrade/check.c
@@ -1518,15 +1518,58 @@ check_for_incompatible_polymorphics(ClusterInfo *cluster)
 	termPQExpBuffer(&state.old_polymorphics);
 }
 
+static char *
+with_oids_query(void *arg)
+{
+	return pg_strdup("SELECT n.nspname, c.relname "
+					 "FROM   pg_catalog.pg_class c, "
+					 "       pg_catalog.pg_namespace n "
+					 "WHERE  c.relnamespace = n.oid AND "
+					 "       c.relhasoids AND"
+					 "       n.nspname NOT IN ('pg_catalog')");
+}
+
+static void
+with_oids_process(DbInfo *dbinfo, PGresult *res, void *arg)
+{
+	bool		db_used = false;
+	int			ntups = PQntuples(res);
+	char		output_path[MAXPGPATH];
+	int			i_nspname = PQfnumber(res, "nspname");
+	int			i_relname = PQfnumber(res, "relname");
+	FILE	  **script = (FILE **) arg;
+
+	if (!ntups)
+		return;
+
+	snprintf(output_path, sizeof(output_path), "%s/%s",
+			 log_opts.basedir,
+			 "tables_with_oids.txt");
+
+	for (int rowno = 0; rowno < ntups; rowno++)
+	{
+		if (*script == NULL && (*script = fopen_priv(output_path, "w")) == NULL)
+			pg_fatal("could not open file \"%s\": %m", output_path);
+		if (!db_used)
+		{
+			fprintf(*script, "In database: %s\n", dbinfo->db_name);
+			db_used = true;
+		}
+		fprintf(*script, "  %s.%s\n",
+				PQgetvalue(res, rowno, i_nspname),
+				PQgetvalue(res, rowno, i_relname));
+	}
+}
+
 /*
  * Verify that no tables are declared WITH OIDS.
  */
 static void
 check_for_tables_with_oids(ClusterInfo *cluster)
 {
-	int			dbnum;
 	FILE	   *script = NULL;
 	char		output_path[MAXPGPATH];
+	AsyncTask  *task = async_task_create();
 
 	prep_status("Checking for tables WITH OIDS");
 
@@ -1534,47 +1577,10 @@ check_for_tables_with_oids(ClusterInfo *cluster)
 			 log_opts.basedir,
 			 "tables_with_oids.txt");
 
-	/* Find any tables declared WITH OIDS */
-	for (dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)
-	{
-		PGresult   *res;
-		bool		db_used = false;
-		int			ntups;
-		int			rowno;
-		int			i_nspname,
-					i_relname;
-		DbInfo	   *active_db = &cluster->dbarr.dbs[dbnum];
-		PGconn	   *conn = connectToServer(cluster, active_db->db_name);
-
-		res = executeQueryOrDie(conn,
-								"SELECT n.nspname, c.relname "
-								"FROM	pg_catalog.pg_class c, "
-								"		pg_catalog.pg_namespace n "
-								"WHERE	c.relnamespace = n.oid AND "
-								"		c.relhasoids AND"
-								"       n.nspname NOT IN ('pg_catalog')");
-
-		ntups = PQntuples(res);
-		i_nspname = PQfnumber(res, "nspname");
-		i_relname = PQfnumber(res, "relname");
-		for (rowno = 0; rowno < ntups; rowno++)
-		{
-			if (script == NULL && (script = fopen_priv(output_path, "w")) == NULL)
-				pg_fatal("could not open file \"%s\": %m", output_path);
-			if (!db_used)
-			{
-				fprintf(script, "In database: %s\n", active_db->db_name);
-				db_used = true;
-			}
-			fprintf(script, "  %s.%s\n",
-					PQgetvalue(res, rowno, i_nspname),
-					PQgetvalue(res, rowno, i_relname));
-		}
-
-		PQclear(res);
-
-		PQfinish(conn);
-	}
+	async_task_add_step(task, with_oids_query,
+						with_oids_process, true, &script);
+	async_task_run(task, cluster);
+	async_task_free(task);
 
 	if (script)
 	{
-- 
2.39.3 (Apple Git-146)


--378n8yA1gW08N3vi
Content-Type: text/plain; charset=us-ascii
Content-Disposition: attachment;
	filename="v5-0013-parallelize-user-defined-encoding-conversions-che.patch"



^ permalink  raw  reply  [nested|flat] 6+ messages in thread

end of thread, other threads:[~2024-07-09 02:42 UTC | newest]

Thread overview: 6+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2021-02-13 05:11 [PATCH v5 2/5] Introduce a new syntax to add/drop publications Japin Li <[email protected]>
2024-07-09 02:42 [PATCH v4 11/12] parallelize tables with oids check in pg_upgrade Nathan Bossart <[email protected]>
2024-07-09 02:42 [PATCH v5 12/13] parallelize tables with oids check in pg_upgrade Nathan Bossart <[email protected]>
2024-07-09 02:42 [PATCH v7 10/11] parallelize tables with oids check in pg_upgrade Nathan Bossart <[email protected]>
2024-07-09 02:42 [PATCH v8 10/11] parallelize tables with oids check in pg_upgrade Nathan Bossart <[email protected]>
2024-07-09 02:42 [PATCH v9 10/11] parallelize tables with oids check in pg_upgrade Nathan Bossart <[email protected]>

This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox