Allow dbname to be written as part of connstring via pg_basebackup's -R option.

Commit cca97ce6a6 allowed dbname in pg_basebackup connstring and in this
commit we allow it to be written in postgresql.auto.conf when -R option is
used. The database name in the connection string will be used by the
logical replication slot synchronization on standby.

The dbname will be recorded only if specified explicitly in the connection
string or environment variable.

Masahiko Sawada hasn't reviewed the code in detail but endorsed the idea.

Author: Vignesh C, Kuroda Hayato
Reviewed-by: Amit Kapila
Discussion: https://postgr.es/m/CAB8KJ=hdKdg+UeXhReeHpHA6N6v3e0qFF+ZsPFHk9_ThWKf=2A@mail.gmail.com

doc/src/sgml/ref/pg_basebackup.sgml

@@ -243,7 +243,11 @@ PostgreSQL documentation
         The <filename>postgresql.auto.conf</filename> file will record the connection
         settings and, if specified, the replication slot
         that <application>pg_basebackup</application> is using, so that
-        streaming replication will use the same settings later on.
+        streaming replication and <link linkend="logicaldecoding-replication-slots-synchronization">
+        logical replication slot synchronization</link> will use the same
+        settings later on. The dbname will be recorded only if the dbname was
+        specified explicitly in the connection string or <link linkend="libpq-envars">
+        environment variable</link>.
        </para>
 
       </listitem>
@@ -809,7 +813,9 @@ PostgreSQL documentation
         name in the connection string will be ignored
         by <productname>PostgreSQL</productname>. Middleware, or proxies, used in
         connecting to <productname>PostgreSQL</productname> might however
-        utilize the value.
+        utilize the value. The database name specified in connection string can
+        also be used by <link linkend="logicaldecoding-replication-slots-synchronization">
+        logical replication slot synchronization</link>.
        </para>
       </listitem>
      </varlistentry>

src/bin/pg_basebackup/pg_basebackup.c

@@ -1807,10 +1807,18 @@ BaseBackup(char *compression_algorithm, char *compression_detail,
 	}
 
 	/*
-	 * Build contents of configuration file if requested
+	 * Build contents of configuration file if requested.
+	 *
+	 * Note that we don't use the dbname from key-value pair in conn as that
+	 * would have been filled by the default dbname (dbname=replication) in
+	 * case the user didn't specify the one. The dbname written in the config
+	 * file as part of primary_conninfo would be used by slotsync worker which
+	 * doesn't use a replication connection so the default won't work for it.
 	 */
 	if (writerecoveryconf)
-		recoveryconfcontents = GenerateRecoveryConfig(conn, replication_slot);
+		recoveryconfcontents = GenerateRecoveryConfig(conn,
+													  replication_slot,
+													  GetDbnameFromConnectionOptions());
 
 	/*
 	 * Run IDENTIFY_SYSTEM so we can get the timeline

src/bin/pg_basebackup/streamutil.c

@@ -34,6 +34,7 @@
 int			WalSegSz;
 
 static bool RetrieveDataDirCreatePerm(PGconn *conn);
+static void FindDbnameInConnParams(PQconninfoOption *conn_opts, char **dbname);
 
 /* SHOW command for replication connection was introduced in version 10 */
 #define MINIMUM_VERSION_FOR_SHOW_CMD 100000
@@ -267,6 +268,75 @@ GetConnection(void)
 	return tmpconn;
 }
 
+/*
+ * FindDbnameInConnParams
+ *
+ * This is a helper function for GetDbnameFromConnectionOptions(). Extract
+ * the value of dbname from PQconninfoOption parameters.
+ */
+static void
+FindDbnameInConnParams(PQconninfoOption *conn_opts, char **dbname)
+{
+	PQconninfoOption *conn_opt;
+
+	Assert(dbname != NULL);
+
+	for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
+	{
+		if ((strcmp(conn_opt->keyword, "dbname") == 0) &&
+			conn_opt->val != NULL && conn_opt->val[0] != '\0')
+			*dbname = pg_strdup(conn_opt->val);
+	}
+}
+
+/*
+ * GetDbnameFromConnectionOptions
+ *
+ * This is a special purpose function to retrieve the dbname from either the
+ * connection_string specified by the user or from the environment variables.
+ *
+ * We follow GetConnection() to fetch the dbname from various connection
+ * options.
+ *
+ * Returns NULL, if dbname is not specified by the user in the above
+ * mentioned connection options.
+ */
+char *
+GetDbnameFromConnectionOptions(void)
+{
+	PQconninfoOption *conn_opts = NULL;
+	char	   *err_msg = NULL;
+	char	   *dbname = NULL;
+
+	/* First try to get the dbname from connection string. */
+	if (connection_string)
+	{
+		conn_opts = PQconninfoParse(connection_string, &err_msg);
+		if (conn_opts == NULL)
+			pg_fatal("%s", err_msg);
+
+		FindDbnameInConnParams(conn_opts, &dbname);
+		if (dbname)
+		{
+			PQconninfoFree(conn_opts);
+			return dbname;
+		}
+	}
+
+	/*
+	 * Next try to get the dbname from default values that are available from
+	 * the environment.
+	 */
+	conn_opts = PQconndefaults();
+	if (conn_opts == NULL)
+		pg_fatal("out of memory");
+
+	FindDbnameInConnParams(conn_opts, &dbname);
+
+	PQconninfoFree(conn_opts);
+	return dbname;
+}
+
 /*
  * From version 10, explicitly set wal segment size using SHOW wal_segment_size
  * since ControlFile is not accessible here.

src/bin/pg_basebackup/streamutil.h

@@ -31,6 +31,8 @@ extern PGconn *conn;
 
 extern PGconn *GetConnection(void);
 
+extern char *GetDbnameFromConnectionOptions(void);
+
 /* Replication commands */
 extern bool CreateReplicationSlot(PGconn *conn, const char *slot_name,
 								  const char *plugin, bool is_temporary,

src/bin/pg_basebackup/t/010_pg_basebackup.pl

@@ -783,6 +783,19 @@ my $checksum = $node->safe_psql('postgres', 'SHOW data_checksums;');
 is($checksum, 'on', 'checksums are enabled');
 rmtree("$tempdir/backupxs_sl_R");
 
+$node->command_ok(
+	[
+		@pg_basebackup_defs, '-D', "$tempdir/backup_dbname_R", '-X',
+		'stream', '-d', "dbname=db1", '-R',
+	],
+	'pg_basebackup with dbname and -R runs');
+like(
+	slurp_file("$tempdir/backup_dbname_R/postgresql.auto.conf"),
+	qr/dbname=db1/m,
+	'recovery conf file sets dbname');
+
+rmtree("$tempdir/backup_dbname_R");
+
 # create tables to corrupt and get their relfilenodes
 my $file_corrupt1 = $node->safe_psql('postgres',
 	q{CREATE TABLE corrupt1 AS SELECT a FROM generate_series(1,10000) AS a; ALTER TABLE corrupt1 SET (autovacuum_enabled=false); SELECT pg_relation_filepath('corrupt1')}

src/bin/pg_rewind/pg_rewind.c

@@ -451,7 +451,7 @@ main(int argc, char **argv)
 		pg_log_info("no rewind required");
 		if (writerecoveryconf && !dry_run)
 			WriteRecoveryConfig(conn, datadir_target,
-								GenerateRecoveryConfig(conn, NULL));
+								GenerateRecoveryConfig(conn, NULL, NULL));
 		exit(0);
 	}
 
@@ -525,7 +525,7 @@ main(int argc, char **argv)
 	/* Also update the standby configuration, if requested. */
 	if (writerecoveryconf && !dry_run)
 		WriteRecoveryConfig(conn, datadir_target,
-							GenerateRecoveryConfig(conn, NULL));
+							GenerateRecoveryConfig(conn, NULL, NULL));
 
 	/* don't need the source connection anymore */
 	source->destroy(source);

src/fe_utils/recovery_gen.c

@@ -18,9 +18,14 @@ static char *escape_quotes(const char *src);
 /*
  * Write recovery configuration contents into a fresh PQExpBuffer, and
  * return it.
+ *
+ * This accepts the dbname which will be appended to the primary_conninfo.
+ * The dbname will be ignored by walreciever process but slotsync worker uses
+ * it to connect to the primary server.
  */
 PQExpBuffer
-GenerateRecoveryConfig(PGconn *pgconn, const char *replication_slot)
+GenerateRecoveryConfig(PGconn *pgconn, const char *replication_slot,
+					   char *dbname)
 {
 	PQconninfoOption *connOptions;
 	PQExpBufferData conninfo_buf;
@@ -66,6 +71,20 @@ GenerateRecoveryConfig(PGconn *pgconn, const char *replication_slot)
 		appendPQExpBuffer(&conninfo_buf, "%s=", opt->keyword);
 		appendConnStrVal(&conninfo_buf, opt->val);
 	}
+
+	if (dbname)
+	{
+		/*
+		 * If dbname is specified in the connection, append the dbname. This
+		 * will be used later for logical replication slot synchronization.
+		 */
+		if (conninfo_buf.len != 0)
+			appendPQExpBufferChar(&conninfo_buf, ' ');
+
+		appendPQExpBuffer(&conninfo_buf, "%s=", "dbname");
+		appendConnStrVal(&conninfo_buf, dbname);
+	}
+
 	if (PQExpBufferDataBroken(conninfo_buf))
 		pg_fatal("out of memory");
 

src/include/fe_utils/recovery_gen.h

@@ -21,7 +21,8 @@
 #define MINIMUM_VERSION_FOR_RECOVERY_GUC 120000
 
 extern PQExpBuffer GenerateRecoveryConfig(PGconn *pgconn,
-										  const char *replication_slot);
+										  const char *replication_slot,
+										  char *dbname);
 extern void WriteRecoveryConfig(PGconn *pgconn, const char *target_dir,
 								PQExpBuffer contents);