53 bool replication,
bool logical,
54 bool must_use_password,
55 const char *appname,
char **
err);
57 bool must_use_password);
60 char **sender_host,
int *sender_port);
67 char **content,
int *
len);
84 const bool *failover,
const bool *
two_phase);
124 elog(
ERROR,
"libpqwalreceiver already loaded");
145 bool must_use_password,
const char *appname,
char **
err)
170 Assert(replication || !logical);
174 keys[++
i] =
"replication";
175 vals[
i] = logical ?
"database" :
"true";
180 keys[++
i] =
"client_encoding";
191 keys[++
i] =
"options";
192 vals[
i] =
"-c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3";
200 keys[++
i] =
"dbname";
201 vals[
i] =
"replication";
205 keys[++
i] =
"fallback_application_name";
217 goto bad_connection_errmsg;
244 WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
259 goto bad_connection_errmsg;
267 (
errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
268 errmsg(
"password is required"),
269 errdetail(
"Non-superuser cannot connect if the server does not request a password."),
270 errhint(
"Target server's authentication method must be changed, or set password_required=false in the subscription parameters.")));
277 if (!replication || logical)
293 conn->logical = logical;
298bad_connection_errmsg:
331 (
errcode(ERRCODE_SYNTAX_ERROR),
332 errmsg(
"invalid connection string syntax: %s", errcopy)));
335 if (must_use_password)
337 bool uses_password =
false;
342 if (opt->
val == NULL)
345 if (strcmp(opt->
keyword,
"password") == 0 && opt->
val[0] !=
'\0')
347 uses_password =
true;
358 (
errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
359 errmsg(
"password is required"),
360 errdetail(
"Non-superusers must provide a password in the connection string.")));
384 if (conn_opts == NULL)
386 (
errcode(ERRCODE_OUT_OF_MEMORY),
387 errmsg(
"could not parse connection string: %s",
388 _(
"out of memory"))));
391 for (conn_opt = conn_opts; conn_opt->
keyword != NULL; conn_opt++)
396 if (strchr(conn_opt->
dispchar,
'D') ||
397 conn_opt->
val == NULL ||
398 conn_opt->
val[0] ==
'\0')
402 obfuscate = strchr(conn_opt->
dispchar,
'*') != NULL;
405 buf.len == 0 ?
"" :
" ",
407 obfuscate ?
"********" : conn_opt->
val);
432 if (ret && strlen(ret) != 0)
436 if (ret && strlen(ret) != 0)
437 *sender_port = atoi(ret);
459 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
460 errmsg(
"could not receive database system identifier and timeline ID from "
461 "the primary server: %s",
476 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
477 errmsg(
"invalid response from primary server"),
478 errdetail(
"Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
479 ntuples, nfields, 1, 3)));
485 return primary_sysid;
517 (
errcode(ERRCODE_SYNTAX_ERROR),
518 errmsg(
"invalid connection string syntax: %s", errcopy)));
527 if (strcmp(opt->keyword,
"dbname") == 0 && opt->val &&
582 char *pubnames_literal;
587 options->proto.logical.proto_version);
589 if (
options->proto.logical.streaming_str)
591 options->proto.logical.streaming_str);
593 if (
options->proto.logical.twophase &&
597 if (
options->proto.logical.origin &&
600 options->proto.logical.origin);
602 pubnames =
options->proto.logical.publication_names;
606 (
errcode(ERRCODE_OUT_OF_MEMORY),
607 errmsg(
"could not start WAL streaming: %s",
610 strlen(pubnames_str));
611 if (!pubnames_literal)
613 (
errcode(ERRCODE_OUT_OF_MEMORY),
614 errmsg(
"could not start WAL streaming: %s",
620 if (
options->proto.logical.binary &&
628 options->proto.physical.startpointTLI);
643 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
644 errmsg(
"could not start WAL streaming: %s",
667 (
errcode(ERRCODE_CONNECTION_FAILURE),
668 errmsg(
"could not send end-of-streaming message to primary: %s",
690 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
691 errmsg(
"unexpected result set after end-of-streaming")));
705 (
errcode(ERRCODE_CONNECTION_FAILURE),
706 errmsg(
"error while shutting down streaming COPY: %s",
715 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
716 errmsg(
"error reading result of streaming command: %s",
724 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
725 errmsg(
"unexpected result after CommandComplete: %s",
735 char **content,
int *
len)
745 snprintf(cmd,
sizeof(cmd),
"TIMELINE_HISTORY %u", tli);
751 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
752 errmsg(
"could not receive timeline history file from "
753 "the primary server: %s",
763 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
764 errmsg(
"invalid response from primary server"),
765 errdetail(
"Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
856 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
911 conn->recvBuf = NULL;
920 (
errcode(ERRCODE_CONNECTION_FAILURE),
921 errmsg(
"could not receive data from WAL stream: %s",
957 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
958 errmsg(
"unexpected result after CommandComplete: %s",
973 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
974 errmsg(
"could not receive data from WAL stream: %s",
980 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
981 errmsg(
"could not receive data from WAL stream: %s",
985 *buffer =
conn->recvBuf;
1000 (
errcode(ERRCODE_CONNECTION_FAILURE),
1001 errmsg(
"could not send data to WAL stream: %s",
1012 bool temporary,
bool two_phase,
bool failover,
1018 int use_new_options_syntax;
1032 if (use_new_options_syntax)
1037 if (use_new_options_syntax)
1046 if (use_new_options_syntax)
1052 if (use_new_options_syntax)
1054 switch (snapshot_action)
1069 switch (snapshot_action)
1083 if (use_new_options_syntax)
1088 if (use_new_options_syntax)
1101 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1102 errmsg(
"could not create replication slot \"%s\": %s",
1125 const bool *failover,
const bool *
two_phase)
1136 *failover ?
"true" :
"false");
1152 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1153 errmsg(
"could not alter replication slot \"%s\": %s",
1173 const int nRetTypes,
const Oid *retTypes)
1184 if (nfields != nRetTypes)
1186 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1187 errmsg(
"invalid query response"),
1188 errdetail(
"Expected %d fields, got %d fields.",
1189 nRetTypes, nfields)));
1195 for (coln = 0; coln < nRetTypes; coln++)
1197 PQfname(pgres, coln), retTypes[coln], -1, 0);
1206 "libpqrcv query result context",
1210 for (tupn = 0; tupn <
PQntuples(pgres); tupn++)
1222 for (coln = 0; coln < nfields; coln++)
1249 const int nRetTypes,
const Oid *retTypes)
1253 char *diag_sqlstate;
1257 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1258 errmsg(
"the query interface requires a database connection")));
1290 walres->
err =
_(
"empty query");
1296 walres->
err =
_(
"unexpected pipeline mode");
1336 foreach(lc, strings)
#define Assert(condition)
#define ALWAYS_SECURE_SEARCH_PATH_SQL
static void PGresult * res
int errdetail(const char *fmt,...)
int errhint(const char *fmt,...)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
#define MAKE_SQLSTATE(ch1, ch2, ch3, ch4, ch5)
#define ereport(elevel,...)
void err(int eval, const char *fmt,...)
HeapTuple BuildTupleFromCStrings(AttInMetadata *attinmeta, char **values)
AttInMetadata * TupleDescGetAttInMetadata(TupleDesc tupdesc)
int PQserverVersion(const PGconn *conn)
char * PQport(const PGconn *conn)
char * PQhost(const PGconn *conn)
int PQconnectionUsedPassword(const PGconn *conn)
PQconninfoOption * PQconninfo(PGconn *conn)
PostgresPollingStatusType PQconnectPoll(PGconn *conn)
void PQconninfoFree(PQconninfoOption *connOptions)
PQconninfoOption * PQconninfoParse(const char *conninfo, char **errmsg)
ConnStatusType PQstatus(const PGconn *conn)
PGconn * PQconnectStartParams(const char *const *keywords, const char *const *values, int expand_dbname)
void PQfinish(PGconn *conn)
int PQbackendPID(const PGconn *conn)
char * PQerrorMessage(const PGconn *conn)
int PQsocket(const PGconn *conn)
int PQgetlength(const PGresult *res, int tup_num, int field_num)
int PQflush(PGconn *conn)
void PQfreemem(void *ptr)
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
PGresult * PQgetResult(PGconn *conn)
ExecStatusType PQresultStatus(const PGresult *res)
int PQendcopy(PGconn *conn)
int PQputCopyEnd(PGconn *conn, const char *errormsg)
int PQntuples(const PGresult *res)
int PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
char * PQfname(const PGresult *res, int field_num)
int PQconsumeInput(PGconn *conn)
char * PQescapeLiteral(PGconn *conn, const char *str, size_t len)
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
char * PQresultErrorField(const PGresult *res, int fieldcode)
int PQsendQuery(PGconn *conn, const char *query)
int PQisBusy(PGconn *conn)
char * PQescapeIdentifier(PGconn *conn, const char *str, size_t len)
int PQnfields(const PGresult *res)
int PQgetCopyData(PGconn *conn, char **buffer, int async)
Datum DirectFunctionCall1Coll(PGFunction func, Oid collation, Datum arg1)
#define MaxTupleAttributeNumber
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
void ResetLatch(Latch *latch)
#define WL_SOCKET_READABLE
#define WL_EXIT_ON_PM_DEATH
#define WL_SOCKET_CONNECTED
#define WL_SOCKET_WRITEABLE
PostgresPollingStatusType
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 PGresult * libpqrcv_PQgetResult(PGconn *streamConn)
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 char * stringlist_to_identifierstr(PGconn *conn, List *strings)
static void libpqrcv_check_conninfo(const char *conninfo, bool must_use_password)
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 WalReceiverFunctionsType PQWalReceiverFunctions
static char * libpqrcv_get_conninfo(WalReceiverConn *conn)
static void libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres, const int nRetTypes, const Oid *retTypes)
static char * libpqrcv_get_dbname_from_conninfo(const char *connInfo)
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
const char * GetDatabaseEncodingName(void)
void MemoryContextReset(MemoryContext context)
char * pstrdup(const char *in)
void pfree(void *pointer)
void * palloc0(Size size)
char * pchomp(const char *in)
MemoryContext CurrentMemoryContext
void MemoryContextDelete(MemoryContext context)
#define AllocSetContextCreate
#define ALLOCSET_DEFAULT_SIZES
int32 pg_strtoint32(const char *s)
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
static AmcheckOptions opts
Datum pg_lsn_in(PG_FUNCTION_ARGS)
static XLogRecPtr DatumGetLSN(Datum X)
static Datum CStringGetDatum(const char *X)
void initPQExpBuffer(PQExpBuffer str)
void appendPQExpBuffer(PQExpBuffer str, const char *fmt,...)
void termPQExpBuffer(PQExpBuffer str)
#define PQExpBufferDataBroken(buf)
char * psprintf(const char *fmt,...)
const char * quote_identifier(const char *ident)
void appendStringInfo(StringInfo str, const char *fmt,...)
void appendStringInfoString(StringInfo str, const char *s)
void appendStringInfoChar(StringInfo str, char ch)
void initStringInfo(StringInfo str)
Tuplestorestate * tuplestore
walrcv_connect_fn walrcv_connect
TupleDesc CreateTemplateTupleDesc(int natts)
void TupleDescInitEntry(TupleDesc desc, AttrNumber attributeNumber, const char *attributeName, Oid oidtypeid, int32 typmod, int attdim)
Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
void tuplestore_puttuple(Tuplestorestate *state, HeapTuple tuple)
void ProcessWalRcvInterrupts(void)
WalReceiverFunctionsType * WalReceiverFunctions
#define LSN_FORMAT_ARGS(lsn)