PostgreSQL Source Code git master
All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
libpqwalreceiver.c File Reference
#include "postgres.h"
#include <unistd.h>
#include <sys/time.h>
#include "common/connect.h"
#include "funcapi.h"
#include "libpq-fe.h"
#include "libpq/libpq-be-fe-helpers.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "pqexpbuffer.h"
#include "replication/walreceiver.h"
#include "storage/latch.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/tuplestore.h"
Include dependency graph for libpqwalreceiver.c:

Go to the source code of this file.

Data Structures

struct  WalReceiverConn
 

Functions

 PG_MODULE_MAGIC_EXT (.name="libpqwalreceiver",.version=PG_VERSION)
 
static WalReceiverConnlibpqrcv_connect (const char *conninfo, bool replication, bool logical, bool must_use_password, const char *appname, char **err)
 
static void libpqrcv_check_conninfo (const char *conninfo, bool must_use_password)
 
static char * libpqrcv_get_conninfo (WalReceiverConn *conn)
 
static void libpqrcv_get_senderinfo (WalReceiverConn *conn, char **sender_host, int *sender_port)
 
static char * libpqrcv_identify_system (WalReceiverConn *conn, TimeLineID *primary_tli)
 
static char * libpqrcv_get_dbname_from_conninfo (const char *connInfo)
 
static int libpqrcv_server_version (WalReceiverConn *conn)
 
static void libpqrcv_readtimelinehistoryfile (WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *len)
 
static bool libpqrcv_startstreaming (WalReceiverConn *conn, const WalRcvStreamOptions *options)
 
static void libpqrcv_endstreaming (WalReceiverConn *conn, TimeLineID *next_tli)
 
static int libpqrcv_receive (WalReceiverConn *conn, char **buffer, pgsocket *wait_fd)
 
static void libpqrcv_send (WalReceiverConn *conn, const char *buffer, int nbytes)
 
static char * libpqrcv_create_slot (WalReceiverConn *conn, const char *slotname, bool temporary, bool two_phase, bool failover, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
 
static void libpqrcv_alter_slot (WalReceiverConn *conn, const char *slotname, const bool *failover, const bool *two_phase)
 
static pid_t libpqrcv_get_backend_pid (WalReceiverConn *conn)
 
static WalRcvExecResultlibpqrcv_exec (WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
 
static void libpqrcv_disconnect (WalReceiverConn *conn)
 
static char * stringlist_to_identifierstr (PGconn *conn, List *strings)
 
void _PG_init (void)
 
static void libpqrcv_processTuples (PGresult *pgres, WalRcvExecResult *walres, const int nRetTypes, const Oid *retTypes)
 

Variables

static WalReceiverFunctionsType PQWalReceiverFunctions
 

Function Documentation

◆ _PG_init()

void _PG_init ( void  )

Definition at line 123 of file libpqwalreceiver.c.

124{
125 if (WalReceiverFunctions != NULL)
126 elog(ERROR, "libpqwalreceiver already loaded");
128}
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:225
static WalReceiverFunctionsType PQWalReceiverFunctions
WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:94

References elog, ERROR, PQWalReceiverFunctions, and WalReceiverFunctions.

◆ libpqrcv_alter_slot()

static void libpqrcv_alter_slot ( WalReceiverConn conn,
const char *  slotname,
const bool *  failover,
const bool *  two_phase 
)
static

Definition at line 1000 of file libpqwalreceiver.c.

1002{
1003 StringInfoData cmd;
1004 PGresult *res;
1005
1006 initStringInfo(&cmd);
1007 appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( ",
1008 quote_identifier(slotname));
1009
1010 if (failover)
1011 appendStringInfo(&cmd, "FAILOVER %s",
1012 *failover ? "true" : "false");
1013
1014 if (failover && two_phase)
1015 appendStringInfoString(&cmd, ", ");
1016
1017 if (two_phase)
1018 appendStringInfo(&cmd, "TWO_PHASE %s",
1019 *two_phase ? "true" : "false");
1020
1021 appendStringInfoString(&cmd, " );");
1022
1023 res = libpqsrv_exec(conn->streamConn, cmd.data,
1024 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
1025 pfree(cmd.data);
1026
1027 if (PQresultStatus(res) != PGRES_COMMAND_OK)
1028 ereport(ERROR,
1029 (errcode(ERRCODE_PROTOCOL_VIOLATION),
1030 errmsg("could not alter replication slot \"%s\": %s",
1031 slotname, pchomp(PQerrorMessage(conn->streamConn)))));
1032
1033 PQclear(res);
1034}
int errcode(int sqlerrcode)
Definition: elog.c:854
int errmsg(const char *fmt,...)
Definition: elog.c:1071
#define ereport(elevel,...)
Definition: elog.h:149
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:7625
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:3411
void PQclear(PGresult *res)
Definition: fe-exec.c:721
static PGresult * libpqsrv_exec(PGconn *conn, const char *query, uint32 wait_event_info)
@ PGRES_COMMAND_OK
Definition: libpq-fe.h:125
void pfree(void *pointer)
Definition: mcxt.c:2152
char * pchomp(const char *in)
Definition: mcxt.c:2355
static bool two_phase
static bool failover
const char * quote_identifier(const char *ident)
Definition: ruleutils.c:13019
PGconn * conn
Definition: streamutil.c:52
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:145
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:230
void initStringInfo(StringInfo str)
Definition: stringinfo.c:97

References appendStringInfo(), appendStringInfoString(), conn, StringInfoData::data, ereport, errcode(), errmsg(), ERROR, failover, initStringInfo(), libpqsrv_exec(), pchomp(), pfree(), PGRES_COMMAND_OK, PQclear(), PQerrorMessage(), PQresultStatus(), quote_identifier(), and two_phase.

◆ libpqrcv_check_conninfo()

static void libpqrcv_check_conninfo ( const char *  conninfo,
bool  must_use_password 
)
static

Definition at line 280 of file libpqwalreceiver.c.

281{
282 PQconninfoOption *opts = NULL;
283 PQconninfoOption *opt;
284 char *err = NULL;
285
286 opts = PQconninfoParse(conninfo, &err);
287 if (opts == NULL)
288 {
289 /* The error string is malloc'd, so we must free it explicitly */
290 char *errcopy = err ? pstrdup(err) : "out of memory";
291
292 PQfreemem(err);
294 (errcode(ERRCODE_SYNTAX_ERROR),
295 errmsg("invalid connection string syntax: %s", errcopy)));
296 }
297
298 if (must_use_password)
299 {
300 bool uses_password = false;
301
302 for (opt = opts; opt->keyword != NULL; ++opt)
303 {
304 /* Ignore connection options that are not present. */
305 if (opt->val == NULL)
306 continue;
307
308 if (strcmp(opt->keyword, "password") == 0 && opt->val[0] != '\0')
309 {
310 uses_password = true;
311 break;
312 }
313 }
314
315 if (!uses_password)
316 {
317 /* malloc'd, so we must free it explicitly */
319
321 (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
322 errmsg("password is required"),
323 errdetail("Non-superusers must provide a password in the connection string.")));
324 }
325 }
326
328}
int errdetail(const char *fmt,...)
Definition: elog.c:1204
void err(int eval, const char *fmt,...)
Definition: err.c:43
void PQconninfoFree(PQconninfoOption *connOptions)
Definition: fe-connect.c:7440
PQconninfoOption * PQconninfoParse(const char *conninfo, char **errmsg)
Definition: fe-connect.c:6156
void PQfreemem(void *ptr)
Definition: fe-exec.c:4032
char * pstrdup(const char *in)
Definition: mcxt.c:2327
static AmcheckOptions opts
Definition: pg_amcheck.c:112

References ereport, err(), errcode(), errdetail(), errmsg(), ERROR, _PQconninfoOption::keyword, opts, PQconninfoFree(), PQconninfoParse(), PQfreemem(), pstrdup(), and _PQconninfoOption::val.

Referenced by libpqrcv_connect().

◆ libpqrcv_connect()

static WalReceiverConn * libpqrcv_connect ( const char *  conninfo,
bool  replication,
bool  logical,
bool  must_use_password,
const char *  appname,
char **  err 
)
static

Definition at line 146 of file libpqwalreceiver.c.

148{
150 const char *keys[6];
151 const char *vals[6];
152 int i = 0;
153
154 /*
155 * Re-validate connection string. The validation already happened at DDL
156 * time, but the subscription owner may have changed. If we don't recheck
157 * with the correct must_use_password, it's possible that the connection
158 * will obtain the password from a different source, such as PGPASSFILE or
159 * PGPASSWORD.
160 */
161 libpqrcv_check_conninfo(conninfo, must_use_password);
162
163 /*
164 * We use the expand_dbname parameter to process the connection string (or
165 * URI), and pass some extra options.
166 */
167 keys[i] = "dbname";
168 vals[i] = conninfo;
169
170 /* We can not have logical without replication */
171 Assert(replication || !logical);
172
173 if (replication)
174 {
175 keys[++i] = "replication";
176 vals[i] = logical ? "database" : "true";
177
178 if (logical)
179 {
180 /* Tell the publisher to translate to our encoding */
181 keys[++i] = "client_encoding";
182 vals[i] = GetDatabaseEncodingName();
183
184 /*
185 * Force assorted GUC parameters to settings that ensure that the
186 * publisher will output data values in a form that is unambiguous
187 * to the subscriber. (We don't want to modify the subscriber's
188 * GUC settings, since that might surprise user-defined code
189 * running in the subscriber, such as triggers.) This should
190 * match what pg_dump does.
191 */
192 keys[++i] = "options";
193 vals[i] = "-c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3";
194 }
195 else
196 {
197 /*
198 * The database name is ignored by the server in replication mode,
199 * but specify "replication" for .pgpass lookup.
200 */
201 keys[++i] = "dbname";
202 vals[i] = "replication";
203 }
204 }
205
206 keys[++i] = "fallback_application_name";
207 vals[i] = appname;
208
209 keys[++i] = NULL;
210 vals[i] = NULL;
211
212 Assert(i < lengthof(keys));
213
214 conn = palloc0(sizeof(WalReceiverConn));
215 conn->streamConn =
216 libpqsrv_connect_params(keys, vals,
217 /* expand_dbname = */ true,
218 WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
219
220 if (PQstatus(conn->streamConn) != CONNECTION_OK)
221 goto bad_connection_errmsg;
222
223 if (must_use_password && !PQconnectionUsedPassword(conn->streamConn))
224 {
225 libpqsrv_disconnect(conn->streamConn);
226 pfree(conn);
227
229 (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
230 errmsg("password is required"),
231 errdetail("Non-superuser cannot connect if the server does not request a password."),
232 errhint("Target server's authentication method must be changed, or set password_required=false in the subscription parameters.")));
233 }
234
235 /*
236 * Set always-secure search path for the cases where the connection is
237 * used to run SQL queries, so malicious users can't get control.
238 */
239 if (!replication || logical)
240 {
241 PGresult *res;
242
243 res = libpqsrv_exec(conn->streamConn,
245 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
247 {
248 PQclear(res);
249 *err = psprintf(_("could not clear search path: %s"),
250 pchomp(PQerrorMessage(conn->streamConn)));
251 goto bad_connection;
252 }
253 PQclear(res);
254 }
255
256 conn->logical = logical;
257
258 return conn;
259
260 /* error path, using libpq's error message */
261bad_connection_errmsg:
262 *err = pchomp(PQerrorMessage(conn->streamConn));
263
264 /* error path, error already set */
265bad_connection:
266 libpqsrv_disconnect(conn->streamConn);
267 pfree(conn);
268 return NULL;
269}
#define lengthof(array)
Definition: c.h:759
#define ALWAYS_SECURE_SEARCH_PATH_SQL
Definition: connect.h:25
int errhint(const char *fmt,...)
Definition: elog.c:1318
#define _(x)
Definition: elog.c:91
int PQconnectionUsedPassword(const PGconn *conn)
Definition: fe-connect.c:7693
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:7562
Assert(PointerIsAligned(start, uint64))
int i
Definition: isn.c:77
static PGconn * libpqsrv_connect_params(const char *const *keywords, const char *const *values, int expand_dbname, uint32 wait_event_info)
static void libpqsrv_disconnect(PGconn *conn)
@ CONNECTION_OK
Definition: libpq-fe.h:84
@ PGRES_TUPLES_OK
Definition: libpq-fe.h:128
static void libpqrcv_check_conninfo(const char *conninfo, bool must_use_password)
const char * GetDatabaseEncodingName(void)
Definition: mbutils.c:1267
void * palloc0(Size size)
Definition: mcxt.c:1975
char * psprintf(const char *fmt,...)
Definition: psprintf.c:43

References _, ALWAYS_SECURE_SEARCH_PATH_SQL, Assert(), conn, CONNECTION_OK, ereport, err(), errcode(), errdetail(), errhint(), errmsg(), ERROR, GetDatabaseEncodingName(), i, lengthof, libpqrcv_check_conninfo(), libpqsrv_connect_params(), libpqsrv_disconnect(), libpqsrv_exec(), palloc0(), pchomp(), pfree(), PGRES_TUPLES_OK, PQclear(), PQconnectionUsedPassword(), PQerrorMessage(), PQresultStatus(), PQstatus(), and psprintf().

◆ libpqrcv_create_slot()

static char * libpqrcv_create_slot ( WalReceiverConn conn,
const char *  slotname,
bool  temporary,
bool  two_phase,
bool  failover,
CRSSnapshotAction  snapshot_action,
XLogRecPtr lsn 
)
static

Definition at line 885 of file libpqwalreceiver.c.

888{
889 PGresult *res;
890 StringInfoData cmd;
891 char *snapshot;
892 int use_new_options_syntax;
893
894 use_new_options_syntax = (PQserverVersion(conn->streamConn) >= 150000);
895
896 initStringInfo(&cmd);
897
898 appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\"", slotname);
899
900 if (temporary)
901 appendStringInfoString(&cmd, " TEMPORARY");
902
903 if (conn->logical)
904 {
905 appendStringInfoString(&cmd, " LOGICAL pgoutput ");
906 if (use_new_options_syntax)
907 appendStringInfoChar(&cmd, '(');
908 if (two_phase)
909 {
910 appendStringInfoString(&cmd, "TWO_PHASE");
911 if (use_new_options_syntax)
912 appendStringInfoString(&cmd, ", ");
913 else
914 appendStringInfoChar(&cmd, ' ');
915 }
916
917 if (failover)
918 {
919 appendStringInfoString(&cmd, "FAILOVER");
920 if (use_new_options_syntax)
921 appendStringInfoString(&cmd, ", ");
922 else
923 appendStringInfoChar(&cmd, ' ');
924 }
925
926 if (use_new_options_syntax)
927 {
928 switch (snapshot_action)
929 {
931 appendStringInfoString(&cmd, "SNAPSHOT 'export'");
932 break;
934 appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
935 break;
936 case CRS_USE_SNAPSHOT:
937 appendStringInfoString(&cmd, "SNAPSHOT 'use'");
938 break;
939 }
940 }
941 else
942 {
943 switch (snapshot_action)
944 {
946 appendStringInfoString(&cmd, "EXPORT_SNAPSHOT");
947 break;
949 appendStringInfoString(&cmd, "NOEXPORT_SNAPSHOT");
950 break;
951 case CRS_USE_SNAPSHOT:
952 appendStringInfoString(&cmd, "USE_SNAPSHOT");
953 break;
954 }
955 }
956
957 if (use_new_options_syntax)
958 appendStringInfoChar(&cmd, ')');
959 }
960 else
961 {
962 if (use_new_options_syntax)
963 appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
964 else
965 appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
966 }
967
968 res = libpqsrv_exec(conn->streamConn,
969 cmd.data,
970 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
971 pfree(cmd.data);
972
974 {
975 PQclear(res);
977 (errcode(ERRCODE_PROTOCOL_VIOLATION),
978 errmsg("could not create replication slot \"%s\": %s",
979 slotname, pchomp(PQerrorMessage(conn->streamConn)))));
980 }
981
982 if (lsn)
984 CStringGetDatum(PQgetvalue(res, 0, 1))));
985
986 if (!PQgetisnull(res, 0, 2))
987 snapshot = pstrdup(PQgetvalue(res, 0, 2));
988 else
989 snapshot = NULL;
990
991 PQclear(res);
992
993 return snapshot;
994}
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:7615
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3876
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3901
Datum DirectFunctionCall1Coll(PGFunction func, Oid collation, Datum arg1)
Definition: fmgr.c:792
Datum pg_lsn_in(PG_FUNCTION_ARGS)
Definition: pg_lsn.c:63
static XLogRecPtr DatumGetLSN(Datum X)
Definition: pg_lsn.h:22
static Datum CStringGetDatum(const char *X)
Definition: postgres.h:355
#define InvalidOid
Definition: postgres_ext.h:35
void appendStringInfoChar(StringInfo str, char ch)
Definition: stringinfo.c:242
@ CRS_USE_SNAPSHOT
Definition: walsender.h:24
@ CRS_NOEXPORT_SNAPSHOT
Definition: walsender.h:23
@ CRS_EXPORT_SNAPSHOT
Definition: walsender.h:22

References appendStringInfo(), appendStringInfoChar(), appendStringInfoString(), conn, CRS_EXPORT_SNAPSHOT, CRS_NOEXPORT_SNAPSHOT, CRS_USE_SNAPSHOT, CStringGetDatum(), StringInfoData::data, DatumGetLSN(), DirectFunctionCall1Coll(), ereport, errcode(), errmsg(), ERROR, failover, initStringInfo(), InvalidOid, libpqsrv_exec(), pchomp(), pfree(), pg_lsn_in(), PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQgetisnull(), PQgetvalue(), PQresultStatus(), PQserverVersion(), pstrdup(), and two_phase.

◆ libpqrcv_disconnect()

static void libpqrcv_disconnect ( WalReceiverConn conn)
static

Definition at line 753 of file libpqwalreceiver.c.

754{
755 libpqsrv_disconnect(conn->streamConn);
756 PQfreemem(conn->recvBuf);
757 pfree(conn);
758}

References conn, libpqsrv_disconnect(), pfree(), and PQfreemem().

◆ libpqrcv_endstreaming()

static void libpqrcv_endstreaming ( WalReceiverConn conn,
TimeLineID next_tli 
)
static

Definition at line 623 of file libpqwalreceiver.c.

624{
625 PGresult *res;
626
627 /*
628 * Send copy-end message. As in libpqsrv_exec, this could theoretically
629 * block, but the risk seems small.
630 */
631 if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
632 PQflush(conn->streamConn))
634 (errcode(ERRCODE_CONNECTION_FAILURE),
635 errmsg("could not send end-of-streaming message to primary: %s",
636 pchomp(PQerrorMessage(conn->streamConn)))));
637
638 *next_tli = 0;
639
640 /*
641 * After COPY is finished, we should receive a result set indicating the
642 * next timeline's ID, or just CommandComplete if the server was shut
643 * down.
644 *
645 * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
646 * also possible in case we aborted the copy in mid-stream.
647 */
648 res = libpqsrv_get_result(conn->streamConn,
649 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
651 {
652 /*
653 * Read the next timeline's ID. The server also sends the timeline's
654 * starting point, but it is ignored.
655 */
656 if (PQnfields(res) < 2 || PQntuples(res) != 1)
658 (errcode(ERRCODE_PROTOCOL_VIOLATION),
659 errmsg("unexpected result set after end-of-streaming")));
660 *next_tli = pg_strtoint32(PQgetvalue(res, 0, 0));
661 PQclear(res);
662
663 /* the result set should be followed by CommandComplete */
664 res = libpqsrv_get_result(conn->streamConn,
665 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
666 }
667 else if (PQresultStatus(res) == PGRES_COPY_OUT)
668 {
669 PQclear(res);
670
671 /* End the copy */
672 if (PQendcopy(conn->streamConn))
674 (errcode(ERRCODE_CONNECTION_FAILURE),
675 errmsg("error while shutting down streaming COPY: %s",
676 pchomp(PQerrorMessage(conn->streamConn)))));
677
678 /* CommandComplete should follow */
679 res = libpqsrv_get_result(conn->streamConn,
680 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
681 }
682
685 (errcode(ERRCODE_PROTOCOL_VIOLATION),
686 errmsg("error reading result of streaming command: %s",
687 pchomp(PQerrorMessage(conn->streamConn)))));
688 PQclear(res);
689
690 /* Verify that there are no more results */
691 res = libpqsrv_get_result(conn->streamConn,
692 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
693 if (res != NULL)
695 (errcode(ERRCODE_PROTOCOL_VIOLATION),
696 errmsg("unexpected result after CommandComplete: %s",
697 pchomp(PQerrorMessage(conn->streamConn)))));
698}
int PQflush(PGconn *conn)
Definition: fe-exec.c:4000
int PQendcopy(PGconn *conn)
Definition: fe-exec.c:2949
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2749
int PQntuples(const PGresult *res)
Definition: fe-exec.c:3481
int PQnfields(const PGresult *res)
Definition: fe-exec.c:3489
static PGresult * libpqsrv_get_result(PGconn *conn, uint32 wait_event_info)
@ PGRES_COPY_OUT
Definition: libpq-fe.h:131
int32 pg_strtoint32(const char *s)
Definition: numutils.c:383

References conn, ereport, errcode(), errmsg(), ERROR, libpqsrv_get_result(), pchomp(), pg_strtoint32(), PGRES_COMMAND_OK, PGRES_COPY_OUT, PGRES_TUPLES_OK, PQclear(), PQendcopy(), PQerrorMessage(), PQflush(), PQgetvalue(), PQnfields(), PQntuples(), PQputCopyEnd(), and PQresultStatus().

◆ libpqrcv_exec()

static WalRcvExecResult * libpqrcv_exec ( WalReceiverConn conn,
const char *  query,
const int  nRetTypes,
const Oid retTypes 
)
static

Definition at line 1125 of file libpqwalreceiver.c.

1127{
1128 PGresult *pgres = NULL;
1129 WalRcvExecResult *walres = palloc0(sizeof(WalRcvExecResult));
1130 char *diag_sqlstate;
1131
1132 if (MyDatabaseId == InvalidOid)
1133 ereport(ERROR,
1134 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1135 errmsg("the query interface requires a database connection")));
1136
1137 pgres = libpqsrv_exec(conn->streamConn,
1138 query,
1139 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
1140
1141 switch (PQresultStatus(pgres))
1142 {
1143 case PGRES_TUPLES_OK:
1144 case PGRES_SINGLE_TUPLE:
1145 case PGRES_TUPLES_CHUNK:
1146 walres->status = WALRCV_OK_TUPLES;
1147 libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
1148 break;
1149
1150 case PGRES_COPY_IN:
1151 walres->status = WALRCV_OK_COPY_IN;
1152 break;
1153
1154 case PGRES_COPY_OUT:
1155 walres->status = WALRCV_OK_COPY_OUT;
1156 break;
1157
1158 case PGRES_COPY_BOTH:
1159 walres->status = WALRCV_OK_COPY_BOTH;
1160 break;
1161
1162 case PGRES_COMMAND_OK:
1163 walres->status = WALRCV_OK_COMMAND;
1164 break;
1165
1166 /* Empty query is considered error. */
1167 case PGRES_EMPTY_QUERY:
1168 walres->status = WALRCV_ERROR;
1169 walres->err = _("empty query");
1170 break;
1171
1174 walres->status = WALRCV_ERROR;
1175 walres->err = _("unexpected pipeline mode");
1176 break;
1177
1179 case PGRES_FATAL_ERROR:
1180 case PGRES_BAD_RESPONSE:
1181 walres->status = WALRCV_ERROR;
1182 walres->err = pchomp(PQerrorMessage(conn->streamConn));
1183 diag_sqlstate = PQresultErrorField(pgres, PG_DIAG_SQLSTATE);
1184 if (diag_sqlstate)
1185 walres->sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
1186 diag_sqlstate[1],
1187 diag_sqlstate[2],
1188 diag_sqlstate[3],
1189 diag_sqlstate[4]);
1190 break;
1191 }
1192
1193 PQclear(pgres);
1194
1195 return walres;
1196}
#define MAKE_SQLSTATE(ch1, ch2, ch3, ch4, ch5)
Definition: elog.h:56
char * PQresultErrorField(const PGresult *res, int fieldcode)
Definition: fe-exec.c:3466
Oid MyDatabaseId
Definition: globals.c:95
@ PGRES_COPY_IN
Definition: libpq-fe.h:132
@ PGRES_COPY_BOTH
Definition: libpq-fe.h:137
@ PGRES_TUPLES_CHUNK
Definition: libpq-fe.h:142
@ PGRES_FATAL_ERROR
Definition: libpq-fe.h:136
@ PGRES_SINGLE_TUPLE
Definition: libpq-fe.h:138
@ PGRES_EMPTY_QUERY
Definition: libpq-fe.h:124
@ PGRES_PIPELINE_SYNC
Definition: libpq-fe.h:139
@ PGRES_BAD_RESPONSE
Definition: libpq-fe.h:133
@ PGRES_PIPELINE_ABORTED
Definition: libpq-fe.h:140
@ PGRES_NONFATAL_ERROR
Definition: libpq-fe.h:135
static void libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres, const int nRetTypes, const Oid *retTypes)
#define PG_DIAG_SQLSTATE
Definition: postgres_ext.h:52
WalRcvExecStatus status
Definition: walreceiver.h:220
@ WALRCV_OK_COPY_IN
Definition: walreceiver.h:208
@ WALRCV_OK_COMMAND
Definition: walreceiver.h:205
@ WALRCV_ERROR
Definition: walreceiver.h:204
@ WALRCV_OK_TUPLES
Definition: walreceiver.h:207
@ WALRCV_OK_COPY_OUT
Definition: walreceiver.h:209
@ WALRCV_OK_COPY_BOTH
Definition: walreceiver.h:210

References _, conn, ereport, WalRcvExecResult::err, errcode(), errmsg(), ERROR, InvalidOid, libpqrcv_processTuples(), libpqsrv_exec(), MAKE_SQLSTATE, MyDatabaseId, palloc0(), pchomp(), PG_DIAG_SQLSTATE, PGRES_BAD_RESPONSE, PGRES_COMMAND_OK, PGRES_COPY_BOTH, PGRES_COPY_IN, PGRES_COPY_OUT, PGRES_EMPTY_QUERY, PGRES_FATAL_ERROR, PGRES_NONFATAL_ERROR, PGRES_PIPELINE_ABORTED, PGRES_PIPELINE_SYNC, PGRES_SINGLE_TUPLE, PGRES_TUPLES_CHUNK, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQresultErrorField(), PQresultStatus(), WalRcvExecResult::sqlstate, WalRcvExecResult::status, WALRCV_ERROR, WALRCV_OK_COMMAND, WALRCV_OK_COPY_BOTH, WALRCV_OK_COPY_IN, WALRCV_OK_COPY_OUT, and WALRCV_OK_TUPLES.

◆ libpqrcv_get_backend_pid()

static pid_t libpqrcv_get_backend_pid ( WalReceiverConn conn)
static

Definition at line 1040 of file libpqwalreceiver.c.

1041{
1042 return PQbackendPID(conn->streamConn);
1043}
int PQbackendPID(const PGconn *conn)
Definition: fe-connect.c:7661

References conn, and PQbackendPID().

◆ libpqrcv_get_conninfo()

static char * libpqrcv_get_conninfo ( WalReceiverConn conn)
static

Definition at line 335 of file libpqwalreceiver.c.

336{
337 PQconninfoOption *conn_opts;
338 PQconninfoOption *conn_opt;
340 char *retval;
341
342 Assert(conn->streamConn != NULL);
343
345 conn_opts = PQconninfo(conn->streamConn);
346
347 if (conn_opts == NULL)
349 (errcode(ERRCODE_OUT_OF_MEMORY),
350 errmsg("could not parse connection string: %s",
351 _("out of memory"))));
352
353 /* build a clean connection string from pieces */
354 for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
355 {
356 bool obfuscate;
357
358 /* Skip debug and empty options */
359 if (strchr(conn_opt->dispchar, 'D') ||
360 conn_opt->val == NULL ||
361 conn_opt->val[0] == '\0')
362 continue;
363
364 /* Obfuscate security-sensitive options */
365 obfuscate = strchr(conn_opt->dispchar, '*') != NULL;
366
367 appendPQExpBuffer(&buf, "%s%s=%s",
368 buf.len == 0 ? "" : " ",
369 conn_opt->keyword,
370 obfuscate ? "********" : conn_opt->val);
371 }
372
373 PQconninfoFree(conn_opts);
374
375 retval = PQExpBufferDataBroken(buf) ? NULL : pstrdup(buf.data);
377 return retval;
378}
PQconninfoOption * PQconninfo(PGconn *conn)
Definition: fe-connect.c:7396
static char * buf
Definition: pg_test_fsync.c:72
void initPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:90
void appendPQExpBuffer(PQExpBuffer str, const char *fmt,...)
Definition: pqexpbuffer.c:265
void termPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:129
#define PQExpBufferDataBroken(buf)
Definition: pqexpbuffer.h:67

References _, appendPQExpBuffer(), Assert(), buf, conn, _PQconninfoOption::dispchar, ereport, errcode(), errmsg(), ERROR, initPQExpBuffer(), _PQconninfoOption::keyword, PQconninfo(), PQconninfoFree(), PQExpBufferDataBroken, pstrdup(), termPQExpBuffer(), and _PQconninfoOption::val.

◆ libpqrcv_get_dbname_from_conninfo()

static char * libpqrcv_get_dbname_from_conninfo ( const char *  connInfo)
static

Definition at line 468 of file libpqwalreceiver.c.

469{
471 char *dbname = NULL;
472 char *err = NULL;
473
474 opts = PQconninfoParse(connInfo, &err);
475 if (opts == NULL)
476 {
477 /* The error string is malloc'd, so we must free it explicitly */
478 char *errcopy = err ? pstrdup(err) : "out of memory";
479
480 PQfreemem(err);
482 (errcode(ERRCODE_SYNTAX_ERROR),
483 errmsg("invalid connection string syntax: %s", errcopy)));
484 }
485
486 for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
487 {
488 /*
489 * If multiple dbnames are specified, then the last one will be
490 * returned
491 */
492 if (strcmp(opt->keyword, "dbname") == 0 && opt->val &&
493 *opt->val)
494 {
495 if (dbname)
496 pfree(dbname);
497
498 dbname = pstrdup(opt->val);
499 }
500 }
501
503 return dbname;
504}
char * dbname
Definition: streamutil.c:49

References dbname, ereport, err(), errcode(), errmsg(), ERROR, opts, pfree(), PQconninfoFree(), PQconninfoParse(), PQfreemem(), and pstrdup().

◆ libpqrcv_get_senderinfo()

static void libpqrcv_get_senderinfo ( WalReceiverConn conn,
char **  sender_host,
int *  sender_port 
)
static

Definition at line 384 of file libpqwalreceiver.c.

386{
387 char *ret = NULL;
388
389 *sender_host = NULL;
390 *sender_port = 0;
391
392 Assert(conn->streamConn != NULL);
393
394 ret = PQhost(conn->streamConn);
395 if (ret && strlen(ret) != 0)
396 *sender_host = pstrdup(ret);
397
398 ret = PQport(conn->streamConn);
399 if (ret && strlen(ret) != 0)
400 *sender_port = atoi(ret);
401}
char * PQport(const PGconn *conn)
Definition: fe-connect.c:7530
char * PQhost(const PGconn *conn)
Definition: fe-connect.c:7494

References Assert(), conn, PQhost(), PQport(), and pstrdup().

◆ libpqrcv_identify_system()

static char * libpqrcv_identify_system ( WalReceiverConn conn,
TimeLineID primary_tli 
)
static

Definition at line 408 of file libpqwalreceiver.c.

409{
410 PGresult *res;
411 char *primary_sysid;
412
413 /*
414 * Get the system identifier and timeline ID as a DataRow message from the
415 * primary server.
416 */
417 res = libpqsrv_exec(conn->streamConn,
418 "IDENTIFY_SYSTEM",
419 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
421 {
422 PQclear(res);
424 (errcode(ERRCODE_PROTOCOL_VIOLATION),
425 errmsg("could not receive database system identifier and timeline ID from "
426 "the primary server: %s",
427 pchomp(PQerrorMessage(conn->streamConn)))));
428 }
429
430 /*
431 * IDENTIFY_SYSTEM returns 3 columns in 9.3 and earlier, and 4 columns in
432 * 9.4 and onwards.
433 */
434 if (PQnfields(res) < 3 || PQntuples(res) != 1)
435 {
436 int ntuples = PQntuples(res);
437 int nfields = PQnfields(res);
438
439 PQclear(res);
441 (errcode(ERRCODE_PROTOCOL_VIOLATION),
442 errmsg("invalid response from primary server"),
443 errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
444 ntuples, nfields, 1, 3)));
445 }
446 primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
447 *primary_tli = pg_strtoint32(PQgetvalue(res, 0, 1));
448 PQclear(res);
449
450 return primary_sysid;
451}

References conn, ereport, errcode(), errdetail(), errmsg(), ERROR, libpqsrv_exec(), pchomp(), pg_strtoint32(), PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQgetvalue(), PQnfields(), PQntuples(), PQresultStatus(), and pstrdup().

◆ libpqrcv_processTuples()

static void libpqrcv_processTuples ( PGresult pgres,
WalRcvExecResult walres,
const int  nRetTypes,
const Oid retTypes 
)
static

Definition at line 1049 of file libpqwalreceiver.c.

1051{
1052 int tupn;
1053 int coln;
1054 int nfields = PQnfields(pgres);
1055 HeapTuple tuple;
1056 AttInMetadata *attinmeta;
1057 MemoryContext rowcontext;
1058 MemoryContext oldcontext;
1059
1060 /* Make sure we got expected number of fields. */
1061 if (nfields != nRetTypes)
1062 ereport(ERROR,
1063 (errcode(ERRCODE_PROTOCOL_VIOLATION),
1064 errmsg("invalid query response"),
1065 errdetail("Expected %d fields, got %d fields.",
1066 nRetTypes, nfields)));
1067
1068 walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
1069
1070 /* Create tuple descriptor corresponding to expected result. */
1071 walres->tupledesc = CreateTemplateTupleDesc(nRetTypes);
1072 for (coln = 0; coln < nRetTypes; coln++)
1073 TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
1074 PQfname(pgres, coln), retTypes[coln], -1, 0);
1075 attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
1076
1077 /* No point in doing more here if there were no tuples returned. */
1078 if (PQntuples(pgres) == 0)
1079 return;
1080
1081 /* Create temporary context for local allocations. */
1083 "libpqrcv query result context",
1085
1086 /* Process returned rows. */
1087 for (tupn = 0; tupn < PQntuples(pgres); tupn++)
1088 {
1089 char *cstrs[MaxTupleAttributeNumber];
1090
1092
1093 /* Do the allocations in temporary context. */
1094 oldcontext = MemoryContextSwitchTo(rowcontext);
1095
1096 /*
1097 * Fill cstrs with null-terminated strings of column values.
1098 */
1099 for (coln = 0; coln < nfields; coln++)
1100 {
1101 if (PQgetisnull(pgres, tupn, coln))
1102 cstrs[coln] = NULL;
1103 else
1104 cstrs[coln] = PQgetvalue(pgres, tupn, coln);
1105 }
1106
1107 /* Convert row to a tuple, and add it to the tuplestore */
1108 tuple = BuildTupleFromCStrings(attinmeta, cstrs);
1109 tuplestore_puttuple(walres->tuplestore, tuple);
1110
1111 /* Clean up */
1112 MemoryContextSwitchTo(oldcontext);
1113 MemoryContextReset(rowcontext);
1114 }
1115
1116 MemoryContextDelete(rowcontext);
1117}
int16 AttrNumber
Definition: attnum.h:21
HeapTuple BuildTupleFromCStrings(AttInMetadata *attinmeta, char **values)
Definition: execTuples.c:2324
AttInMetadata * TupleDescGetAttInMetadata(TupleDesc tupdesc)
Definition: execTuples.c:2275
char * PQfname(const PGresult *res, int field_num)
Definition: fe-exec.c:3567
int work_mem
Definition: globals.c:132
#define MaxTupleAttributeNumber
Definition: htup_details.h:34
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:414
MemoryContext CurrentMemoryContext
Definition: mcxt.c:159
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:485
#define AllocSetContextCreate
Definition: memutils.h:149
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:180
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:123
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:124
Tuplestorestate * tuplestore
Definition: walreceiver.h:223
TupleDesc tupledesc
Definition: walreceiver.h:224
TupleDesc CreateTemplateTupleDesc(int natts)
Definition: tupdesc.c:175
void TupleDescInitEntry(TupleDesc desc, AttrNumber attributeNumber, const char *attributeName, Oid oidtypeid, int32 typmod, int attdim)
Definition: tupdesc.c:835
Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
Definition: tuplestore.c:330
void tuplestore_puttuple(Tuplestorestate *state, HeapTuple tuple)
Definition: tuplestore.c:764

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, BuildTupleFromCStrings(), CHECK_FOR_INTERRUPTS, CreateTemplateTupleDesc(), CurrentMemoryContext, ereport, errcode(), errdetail(), errmsg(), ERROR, MaxTupleAttributeNumber, MemoryContextDelete(), MemoryContextReset(), MemoryContextSwitchTo(), PQfname(), PQgetisnull(), PQgetvalue(), PQnfields(), PQntuples(), WalRcvExecResult::tupledesc, TupleDescGetAttInMetadata(), TupleDescInitEntry(), WalRcvExecResult::tuplestore, tuplestore_begin_heap(), tuplestore_puttuple(), and work_mem.

Referenced by libpqrcv_exec().

◆ libpqrcv_readtimelinehistoryfile()

static void libpqrcv_readtimelinehistoryfile ( WalReceiverConn conn,
TimeLineID  tli,
char **  filename,
char **  content,
int *  len 
)
static

Definition at line 704 of file libpqwalreceiver.c.

707{
708 PGresult *res;
709 char cmd[64];
710
711 Assert(!conn->logical);
712
713 /*
714 * Request the primary to send over the history file for given timeline.
715 */
716 snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
717 res = libpqsrv_exec(conn->streamConn,
718 cmd,
719 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
721 {
722 PQclear(res);
724 (errcode(ERRCODE_PROTOCOL_VIOLATION),
725 errmsg("could not receive timeline history file from "
726 "the primary server: %s",
727 pchomp(PQerrorMessage(conn->streamConn)))));
728 }
729 if (PQnfields(res) != 2 || PQntuples(res) != 1)
730 {
731 int ntuples = PQntuples(res);
732 int nfields = PQnfields(res);
733
734 PQclear(res);
736 (errcode(ERRCODE_PROTOCOL_VIOLATION),
737 errmsg("invalid response from primary server"),
738 errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
739 ntuples, nfields)));
740 }
741 *filename = pstrdup(PQgetvalue(res, 0, 0));
742
743 *len = PQgetlength(res, 0, 1);
744 *content = palloc(*len);
745 memcpy(*content, PQgetvalue(res, 0, 1), *len);
746 PQclear(res);
747}
int PQgetlength(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3887
void * palloc(Size size)
Definition: mcxt.c:1945
const void size_t len
static char * filename
Definition: pg_dumpall.c:123
#define snprintf
Definition: port.h:239

References Assert(), conn, ereport, errcode(), errdetail(), errmsg(), ERROR, filename, len, libpqsrv_exec(), palloc(), pchomp(), PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQgetlength(), PQgetvalue(), PQnfields(), PQntuples(), PQresultStatus(), pstrdup(), and snprintf.

◆ libpqrcv_receive()

static int libpqrcv_receive ( WalReceiverConn conn,
char **  buffer,
pgsocket wait_fd 
)
static

Definition at line 777 of file libpqwalreceiver.c.

779{
780 int rawlen;
781
782 PQfreemem(conn->recvBuf);
783 conn->recvBuf = NULL;
784
785 /* Try to receive a CopyData message */
786 rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
787 if (rawlen == 0)
788 {
789 /* Try consuming some data. */
790 if (PQconsumeInput(conn->streamConn) == 0)
792 (errcode(ERRCODE_CONNECTION_FAILURE),
793 errmsg("could not receive data from WAL stream: %s",
794 pchomp(PQerrorMessage(conn->streamConn)))));
795
796 /* Now that we've consumed some input, try again */
797 rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
798 if (rawlen == 0)
799 {
800 /* Tell caller to try again when our socket is ready. */
801 *wait_fd = PQsocket(conn->streamConn);
802 return 0;
803 }
804 }
805 if (rawlen == -1) /* end-of-streaming or error */
806 {
807 PGresult *res;
808
809 res = libpqsrv_get_result(conn->streamConn,
810 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
812 {
813 PQclear(res);
814
815 /* Verify that there are no more results. */
816 res = libpqsrv_get_result(conn->streamConn,
817 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
818 if (res != NULL)
819 {
820 PQclear(res);
821
822 /*
823 * If the other side closed the connection orderly (otherwise
824 * we'd seen an error, or PGRES_COPY_IN) don't report an error
825 * here, but let callers deal with it.
826 */
827 if (PQstatus(conn->streamConn) == CONNECTION_BAD)
828 return -1;
829
831 (errcode(ERRCODE_PROTOCOL_VIOLATION),
832 errmsg("unexpected result after CommandComplete: %s",
833 PQerrorMessage(conn->streamConn))));
834 }
835
836 return -1;
837 }
838 else if (PQresultStatus(res) == PGRES_COPY_IN)
839 {
840 PQclear(res);
841 return -1;
842 }
843 else
844 {
845 PQclear(res);
847 (errcode(ERRCODE_PROTOCOL_VIOLATION),
848 errmsg("could not receive data from WAL stream: %s",
849 pchomp(PQerrorMessage(conn->streamConn)))));
850 }
851 }
852 if (rawlen < -1)
854 (errcode(ERRCODE_PROTOCOL_VIOLATION),
855 errmsg("could not receive data from WAL stream: %s",
856 pchomp(PQerrorMessage(conn->streamConn)))));
857
858 /* Return received messages to caller */
859 *buffer = conn->recvBuf;
860 return rawlen;
861}
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:7651
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1984
int PQgetCopyData(PGconn *conn, char **buffer, int async)
Definition: fe-exec.c:2816
@ CONNECTION_BAD
Definition: libpq-fe.h:85

References conn, CONNECTION_BAD, ereport, errcode(), errmsg(), ERROR, libpqsrv_get_result(), pchomp(), PGRES_COMMAND_OK, PGRES_COPY_IN, PQclear(), PQconsumeInput(), PQerrorMessage(), PQfreemem(), PQgetCopyData(), PQresultStatus(), PQsocket(), and PQstatus().

◆ libpqrcv_send()

static void libpqrcv_send ( WalReceiverConn conn,
const char *  buffer,
int  nbytes 
)
static

Definition at line 869 of file libpqwalreceiver.c.

870{
871 if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
872 PQflush(conn->streamConn))
874 (errcode(ERRCODE_CONNECTION_FAILURE),
875 errmsg("could not send data to WAL stream: %s",
876 pchomp(PQerrorMessage(conn->streamConn)))));
877}
int PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
Definition: fe-exec.c:2695

References conn, ereport, errcode(), errmsg(), ERROR, pchomp(), PQerrorMessage(), PQflush(), and PQputCopyData().

◆ libpqrcv_server_version()

static int libpqrcv_server_version ( WalReceiverConn conn)
static

Definition at line 457 of file libpqwalreceiver.c.

458{
459 return PQserverVersion(conn->streamConn);
460}

References conn, and PQserverVersion().

◆ libpqrcv_startstreaming()

static bool libpqrcv_startstreaming ( WalReceiverConn conn,
const WalRcvStreamOptions options 
)
static

Definition at line 517 of file libpqwalreceiver.c.

519{
520 StringInfoData cmd;
521 PGresult *res;
522
523 Assert(options->logical == conn->logical);
524 Assert(options->slotname || !options->logical);
525
526 initStringInfo(&cmd);
527
528 /* Build the command. */
529 appendStringInfoString(&cmd, "START_REPLICATION");
530 if (options->slotname != NULL)
531 appendStringInfo(&cmd, " SLOT \"%s\"",
532 options->slotname);
533
534 if (options->logical)
535 appendStringInfoString(&cmd, " LOGICAL");
536
537 appendStringInfo(&cmd, " %X/%X", LSN_FORMAT_ARGS(options->startpoint));
538
539 /*
540 * Additional options are different depending on if we are doing logical
541 * or physical replication.
542 */
543 if (options->logical)
544 {
545 char *pubnames_str;
546 List *pubnames;
547 char *pubnames_literal;
548
549 appendStringInfoString(&cmd, " (");
550
551 appendStringInfo(&cmd, "proto_version '%u'",
552 options->proto.logical.proto_version);
553
554 if (options->proto.logical.streaming_str)
555 appendStringInfo(&cmd, ", streaming '%s'",
556 options->proto.logical.streaming_str);
557
558 if (options->proto.logical.twophase &&
559 PQserverVersion(conn->streamConn) >= 150000)
560 appendStringInfoString(&cmd, ", two_phase 'on'");
561
562 if (options->proto.logical.origin &&
563 PQserverVersion(conn->streamConn) >= 160000)
564 appendStringInfo(&cmd, ", origin '%s'",
565 options->proto.logical.origin);
566
567 pubnames = options->proto.logical.publication_names;
568 pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
569 if (!pubnames_str)
571 (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
572 errmsg("could not start WAL streaming: %s",
573 pchomp(PQerrorMessage(conn->streamConn)))));
574 pubnames_literal = PQescapeLiteral(conn->streamConn, pubnames_str,
575 strlen(pubnames_str));
576 if (!pubnames_literal)
578 (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
579 errmsg("could not start WAL streaming: %s",
580 pchomp(PQerrorMessage(conn->streamConn)))));
581 appendStringInfo(&cmd, ", publication_names %s", pubnames_literal);
582 PQfreemem(pubnames_literal);
583 pfree(pubnames_str);
584
585 if (options->proto.logical.binary &&
586 PQserverVersion(conn->streamConn) >= 140000)
587 appendStringInfoString(&cmd, ", binary 'true'");
588
589 appendStringInfoChar(&cmd, ')');
590 }
591 else
592 appendStringInfo(&cmd, " TIMELINE %u",
593 options->proto.physical.startpointTLI);
594
595 /* Start streaming. */
596 res = libpqsrv_exec(conn->streamConn,
597 cmd.data,
598 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
599 pfree(cmd.data);
600
602 {
603 PQclear(res);
604 return false;
605 }
606 else if (PQresultStatus(res) != PGRES_COPY_BOTH)
607 {
608 PQclear(res);
610 (errcode(ERRCODE_PROTOCOL_VIOLATION),
611 errmsg("could not start WAL streaming: %s",
612 pchomp(PQerrorMessage(conn->streamConn)))));
613 }
614 PQclear(res);
615 return true;
616}
char * PQescapeLiteral(PGconn *conn, const char *str, size_t len)
Definition: fe-exec.c:4363
static char * stringlist_to_identifierstr(PGconn *conn, List *strings)
Definition: pg_list.h:54
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43

References appendStringInfo(), appendStringInfoChar(), appendStringInfoString(), Assert(), conn, StringInfoData::data, ereport, errcode(), errmsg(), ERROR, initStringInfo(), libpqsrv_exec(), LSN_FORMAT_ARGS, pchomp(), pfree(), PGRES_COMMAND_OK, PGRES_COPY_BOTH, PQclear(), PQerrorMessage(), PQescapeLiteral(), PQfreemem(), PQresultStatus(), PQserverVersion(), and stringlist_to_identifierstr().

◆ PG_MODULE_MAGIC_EXT()

PG_MODULE_MAGIC_EXT ( name = "libpqwalreceiver",
version = PG_VERSION 
)

◆ stringlist_to_identifierstr()

static char * stringlist_to_identifierstr ( PGconn conn,
List strings 
)
static

Definition at line 1207 of file libpqwalreceiver.c.

1208{
1209 ListCell *lc;
1210 StringInfoData res;
1211 bool first = true;
1212
1213 initStringInfo(&res);
1214
1215 foreach(lc, strings)
1216 {
1217 char *val = strVal(lfirst(lc));
1218 char *val_escaped;
1219
1220 if (first)
1221 first = false;
1222 else
1223 appendStringInfoChar(&res, ',');
1224
1225 val_escaped = PQescapeIdentifier(conn, val, strlen(val));
1226 if (!val_escaped)
1227 {
1228 free(res.data);
1229 return NULL;
1230 }
1231 appendStringInfoString(&res, val_escaped);
1232 PQfreemem(val_escaped);
1233 }
1234
1235 return res.data;
1236}
char * PQescapeIdentifier(PGconn *conn, const char *str, size_t len)
Definition: fe-exec.c:4369
#define free(a)
Definition: header.h:65
long val
Definition: informix.c:689
#define lfirst(lc)
Definition: pg_list.h:172
#define strVal(v)
Definition: value.h:82

References appendStringInfoChar(), appendStringInfoString(), conn, StringInfoData::data, free, initStringInfo(), lfirst, PQescapeIdentifier(), PQfreemem(), strVal, and val.

Referenced by libpqrcv_startstreaming().

Variable Documentation

◆ PQWalReceiverFunctions

WalReceiverFunctionsType PQWalReceiverFunctions
static
Initial value:
= {
.walrcv_connect = libpqrcv_connect,
.walrcv_check_conninfo = libpqrcv_check_conninfo,
.walrcv_get_conninfo = libpqrcv_get_conninfo,
.walrcv_get_senderinfo = libpqrcv_get_senderinfo,
.walrcv_identify_system = libpqrcv_identify_system,
.walrcv_server_version = libpqrcv_server_version,
.walrcv_readtimelinehistoryfile = libpqrcv_readtimelinehistoryfile,
.walrcv_startstreaming = libpqrcv_startstreaming,
.walrcv_endstreaming = libpqrcv_endstreaming,
.walrcv_receive = libpqrcv_receive,
.walrcv_send = libpqrcv_send,
.walrcv_create_slot = libpqrcv_create_slot,
.walrcv_alter_slot = libpqrcv_alter_slot,
.walrcv_get_dbname_from_conninfo = libpqrcv_get_dbname_from_conninfo,
.walrcv_get_backend_pid = libpqrcv_get_backend_pid,
.walrcv_exec = libpqrcv_exec,
.walrcv_disconnect = libpqrcv_disconnect
}
static WalReceiverConn * libpqrcv_connect(const char *conninfo, bool replication, bool logical, bool must_use_password, const char *appname, char **err)
static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *len)
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn)
static void libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
static void libpqrcv_disconnect(WalReceiverConn *conn)
static char * libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli)
static char * libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname, bool temporary, bool two_phase, bool failover, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
static WalRcvExecResult * libpqrcv_exec(WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
static void libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
static void libpqrcv_get_senderinfo(WalReceiverConn *conn, char **sender_host, int *sender_port)
static int libpqrcv_server_version(WalReceiverConn *conn)
static bool libpqrcv_startstreaming(WalReceiverConn *conn, const WalRcvStreamOptions *options)
static int libpqrcv_receive(WalReceiverConn *conn, char **buffer, pgsocket *wait_fd)
static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname, const bool *failover, const bool *two_phase)
static char * libpqrcv_get_conninfo(WalReceiverConn *conn)
static char * libpqrcv_get_dbname_from_conninfo(const char *connInfo)

Definition at line 96 of file libpqwalreceiver.c.

Referenced by _PG_init().