v11-0004-preliminary-work-for-pg_restore-exclude-datab.no-cfbot

From 77df1aa11496b8ef0f9547144aa5f5c8af6f5e7c Mon Sep 17 00:00:00 2001
From: jian he <jian.universality@gmail.com>
Date: Thu, 23 Jan 2025 17:03:40 +0800
Subject: [PATCH v11 4/4] preliminary work for pg_restore
 --exclude-database=PATTERN

---
 src/bin/pg_dump/pg_restore.c | 120 +++++++++--------------------------
 1 file changed, 29 insertions(+), 91 deletions(-)

diff --git a/src/bin/pg_dump/pg_restore.c b/src/bin/pg_dump/pg_restore.c
index a715448dce..0a1a7c1ecd 100644
--- a/src/bin/pg_dump/pg_restore.c
+++ b/src/bin/pg_dump/pg_restore.c
@@ -71,9 +71,6 @@ typedef struct SimpleDatabaseOidList
 	SimpleDatabaseOidListCell *tail;
 } SimpleDatabaseOidList;
 
-static void
-simple_db_oid_list_append(SimpleDatabaseOidList *list, Oid db_oid, const char *dbname);
-
 static void usage(const char *progname);
 static void read_restore_filters(const char *filename, RestoreOptions *opts);
 static bool IsFileExistsInDirectory(const char *dir, const char *filename);
@@ -81,22 +78,23 @@ static bool restoreOneDatabase(const char *inputFileSpec, RestoreOptions *opts,
 							   int numWorkers, bool append_data);
 static int ReadOneStatement(StringInfo inBuf, FILE *pfile);
 static int restoreAllDatabases(PGconn *conn, const char *dumpdirpath,
-	SimpleStringList db_exclude_patterns, RestoreOptions *opts, int numWorkers);
+							   RestoreOptions *opts, int numWorkers);
 static void execute_global_sql_commands(PGconn *conn, const char *dumpdirpath,
-		const char *outfile);
-static int filter_dbnames_for_restore(PGconn *conn,
-		SimpleDatabaseOidList *dbname_oid_list, SimpleStringList db_exclude_patterns);
+										const char *outfile);
+static int filter_dbnames_for_restore(SimpleDatabaseOidList *dbname_oid_list);
 static int get_dbname_oid_list_from_mfile(const char *dumpdirpath,
 										  SimpleDatabaseOidList *dbname_oid_list);
 static void simple_db_oid_list_append(SimpleDatabaseOidList *list, Oid db_oid,
 									  const char *dbname);
-static bool is_full_pattern(PGconn *conn, const char *str, const char *ptrn);
 static void simple_string_full_list_delete(SimpleStringList *list);
 static void simple_db_oid_full_list_delete(SimpleDatabaseOidList *list);
 static void simple_db_oid_list_delete(SimpleDatabaseOidList *list,
 									  SimpleDatabaseOidListCell *cell,
 									  SimpleDatabaseOidListCell *prev);
 
+static SimpleStringList database_exclude_names = {NULL, NULL};
+static SimpleStringList db_exclude_patterns = {NULL, NULL};
+
 int
 main(int argc, char **argv)
 {
@@ -118,8 +116,7 @@ main(int argc, char **argv)
 	static int	strict_names = 0;
 	bool		data_only = false;
 	bool		schema_only = false;
-	SimpleStringList    db_exclude_patterns = {NULL, NULL};
-	bool				globals_only = false;
+	bool		globals_only = false;
 
 	struct option cmdopts[] = {
 		{"clean", 0, NULL, 'c'},
@@ -491,8 +488,7 @@ main(int argc, char **argv)
 			else
 			{
 				/* Now restore all the databases from map.dat file. */
-				exit_code = restoreAllDatabases(conn, inputFileSpec, db_exclude_patterns,
-						opts, numWorkers);
+				exit_code = restoreAllDatabases(conn, inputFileSpec, opts, numWorkers);
 			}
 
 			/* Free db pattern list. */
@@ -798,14 +794,13 @@ ReadOneStatement(StringInfo inBuf, FILE *pfile)
 /*
  * filter_dbnames_for_restore
  *
- * This will remove names from all dblist that are given with exclude-database
- * option.
+ * This will remove database entries from dbname_oid_list that are matching with --exclude-database
+ * pattern.
  *
- * returns number of dbnames those will be restored.
+ * returns number of databases that will be restored.
  */
 static int
-filter_dbnames_for_restore(PGconn *conn, SimpleDatabaseOidList *dbname_oid_list,
-		SimpleStringList db_exclude_patterns)
+filter_dbnames_for_restore(SimpleDatabaseOidList *dbname_oid_list)
 {
 	SimpleDatabaseOidListCell	*dboid_cell = dbname_oid_list->head;
 	SimpleDatabaseOidListCell	*dboidprecell = NULL;
@@ -822,10 +817,9 @@ filter_dbnames_for_restore(PGconn *conn, SimpleDatabaseOidList *dbname_oid_list,
 		SimpleDatabaseOidListCell	*next = dboid_cell->next;
 
 		/* Now match this dbname with exclude-database list. */
-		for (SimpleStringListCell *celldb = db_exclude_patterns.head; celldb; celldb = celldb->next)
+		for (SimpleStringListCell *celldb = database_exclude_names.head; celldb; celldb = celldb->next)
 		{
-			if ((conn && is_full_pattern(conn, dboid_cell->db_name, celldb->val)) ||
-					(!conn && pg_strcasecmp(dboid_cell->db_name, celldb->val) == 0))
+			if (pg_strcasecmp(dboid_cell->db_name, celldb->val) == 0)
 			{
 				/*
 				 * As we need to skip this dbname so set flag to remove it from
@@ -937,9 +931,13 @@ get_dbname_oid_list_from_mfile(const char *dumpdirpath, SimpleDatabaseOidList *d
  */
 static int
 restoreAllDatabases(PGconn *conn, const char *dumpdirpath,
-		SimpleStringList db_exclude_patterns, RestoreOptions *opts,
-		int numWorkers)
+					RestoreOptions *opts, int numWorkers)
 {
+	/*
+	 * dbname_oid_list initially stores all the databases to be restored from
+	 * the map.dat file.  It then filters out any databases that match the
+	 * --exclude-database pattern.
+	*/
 	SimpleDatabaseOidList			dbname_oid_list = {NULL, NULL};
 	SimpleDatabaseOidListCell		*dboid_cell;
 	int								exit_code = 0;
@@ -948,12 +946,11 @@ restoreAllDatabases(PGconn *conn, const char *dumpdirpath,
 
 	num_total_db = get_dbname_oid_list_from_mfile(dumpdirpath, &dbname_oid_list);
 
+	pg_log_info("found total %d database in map.dat file", num_total_db);
 	/* If map.dat has no entry, return from here. */
 	if (dbname_oid_list.head == NULL)
 		return 0;
 
-	pg_log_info("found total %d database names in map.dat file", num_total_db);
-
 	if (!conn)
 	{
 		pg_log_info("trying to connect database \"postgres\"  to dump into out file");
@@ -982,18 +979,17 @@ restoreAllDatabases(PGconn *conn, const char *dumpdirpath,
 	 * Skip any explicitly excluded database.  If there is no database
 	 * connection, then just consider pattern as simple name to compare.
 	 */
-	num_db_restore = filter_dbnames_for_restore(conn, &dbname_oid_list,
-			db_exclude_patterns);
+	expand_dbname_patterns(conn, &db_exclude_patterns, &database_exclude_names);
+	num_db_restore = filter_dbnames_for_restore(&dbname_oid_list);
 
 	/* Close the db connection as we are done globals and patterns. */
 	if (conn)
 		PQfinish(conn);
 
-	/* Exit if no db needs to be restored. */
-	if (dbname_oid_list.head == NULL)
-		return 0;
-
 	pg_log_info("needs to restore %d databases out of %d databases", num_db_restore, num_total_db);
+	/* Exit if no db needs to be restored. */
+	if (num_db_restore == 0)
+		return 0;
 
 	/*
 	 * To restore multiple databases, -C (create database) option should be specified
@@ -1006,7 +1002,7 @@ restoreAllDatabases(PGconn *conn, const char *dumpdirpath,
 	if (num_db_restore > 100)
 	{
 		simple_db_oid_full_list_delete(&dbname_oid_list);
-		pg_fatal("cound not restore more than 100 databases by single pg_restore, here total db:%d", num_db_restore);
+		pg_fatal("cound not restore more than 100 databases by single pg_restore, here total databases:%d", num_db_restore);
 	}
 
 	/*
@@ -1042,12 +1038,9 @@ restoreAllDatabases(PGconn *conn, const char *dumpdirpath,
 			exit_code = dbexit_code;
 
 		dboid_cell = dboid_cell->next;
-	} /* end while */
+	}
 
-	/* Log number of processed databases.*/
-	pg_log_info("number of restored databases are %d", num_db_restore);
-
-	/* Free dbname and dboid list. */
+	/* Free dbname_oid_list */
 	simple_db_oid_full_list_delete(&dbname_oid_list);
 
 	return exit_code;
@@ -1218,58 +1211,3 @@ simple_db_oid_list_delete(SimpleDatabaseOidList *list, SimpleDatabaseOidListCell
 		pfree(cell);
 	}
 }
-
-/*
- * is_full_pattern
- *
- * This uses substring function to make 1st string from pattern.
- * Outstring of substring function is compared with 1st string to
- * validate this pattern.
- *
- * Returns true if 1st string can be constructed from given pattern.
- *
- */
-static bool
-is_full_pattern(PGconn *conn, const char *str, const char *ptrn)
-{
-	PQExpBuffer		query;
-	PGresult		*result;
-
-	query = createPQExpBuffer();
-
-	printfPQExpBuffer(query,
-			"SELECT substring ( "
-			" '%s' , "
-			" '%s' ) ", str, ptrn);
-
-   result = executeQuery(conn, query->data);
-
-	if (PQresultStatus(result) == PGRES_TUPLES_OK)
-	{
-		if (PQntuples(result) == 1)
-		{
-			const char	*outstr = NULL;
-
-			/*
-			 * If output string of substring function is matches with str, then
-			 * we can construct str from pattern.
-			 */
-			if (!PQgetisnull(result, 0, 0))
-				outstr = PQgetvalue(result, 0, 0);
-
-			if (outstr && pg_strcasecmp(outstr, str) == 0)
-			{
-				PQclear(result);
-				destroyPQExpBuffer(query);
-				return true;
-			}
-		}
-	}
-	else
-		pg_log_error("could not execute query: \"%s\" \nCommand was: \"%s\"", PQerrorMessage(conn), query->data);
-
-	PQclear(result);
-	destroyPQExpBuffer(query);
-
-	return false;
-}
-- 
2.34.1