52 bool logical,
bool must_use_password,
53 const char *appname,
char **
err);
55 bool must_use_password);
58 char **sender_host,
int *sender_port);
64 char **content,
int *
len);
116 elog(
ERROR,
"libpqwalreceiver already loaded");
133 const char *appname,
char **
err)
146 keys[++
i] =
"replication";
147 vals[
i] = logical ?
"database" :
"true";
154 keys[++
i] =
"dbname";
155 vals[
i] =
"replication";
157 keys[++
i] =
"fallback_application_name";
162 keys[++
i] =
"client_encoding";
173 keys[++
i] =
"options";
174 vals[
i] =
"-c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3";
185 WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
188 goto bad_connection_errmsg;
196 (
errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
197 errmsg(
"password is required"),
198 errdetail(
"Non-superuser cannot connect if the server does not request a password."),
199 errhint(
"Target server's authentication method must be changed, or set password_required=false in the subscription parameters.")));
218 conn->logical = logical;
223 bad_connection_errmsg:
256 (
errcode(ERRCODE_SYNTAX_ERROR),
257 errmsg(
"invalid connection string syntax: %s", errcopy)));
260 if (must_use_password)
262 bool uses_password =
false;
267 if (opt->
val == NULL)
270 if (strcmp(opt->
keyword,
"password") == 0 && opt->
val[0] !=
'\0')
272 uses_password =
true;
279 (
errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
280 errmsg(
"password is required"),
281 errdetail(
"Non-superusers must provide a password in the connection string.")));
304 if (conn_opts == NULL)
306 (
errcode(ERRCODE_OUT_OF_MEMORY),
307 errmsg(
"could not parse connection string: %s",
308 _(
"out of memory"))));
311 for (conn_opt = conn_opts; conn_opt->
keyword != NULL; conn_opt++)
316 if (strchr(conn_opt->
dispchar,
'D') ||
317 conn_opt->
val == NULL ||
318 conn_opt->
val[0] ==
'\0')
322 obfuscate = strchr(conn_opt->
dispchar,
'*') != NULL;
325 buf.len == 0 ?
"" :
" ",
327 obfuscate ?
"********" : conn_opt->
val);
352 if (ret && strlen(ret) != 0)
356 if (ret && strlen(ret) != 0)
357 *sender_port = atoi(ret);
379 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
380 errmsg(
"could not receive database system identifier and timeline ID from "
381 "the primary server: %s",
391 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
392 errmsg(
"invalid response from primary server"),
393 errdetail(
"Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
394 ntuples, nfields, 3, 1)));
400 return primary_sysid;
453 char *pubnames_literal;
458 options->proto.logical.proto_version);
460 if (
options->proto.logical.streaming_str)
462 options->proto.logical.streaming_str);
464 if (
options->proto.logical.twophase &&
468 if (
options->proto.logical.origin &&
471 options->proto.logical.origin);
473 pubnames =
options->proto.logical.publication_names;
477 (
errcode(ERRCODE_OUT_OF_MEMORY),
478 errmsg(
"could not start WAL streaming: %s",
481 strlen(pubnames_str));
482 if (!pubnames_literal)
484 (
errcode(ERRCODE_OUT_OF_MEMORY),
485 errmsg(
"could not start WAL streaming: %s",
491 if (
options->proto.logical.binary &&
499 options->proto.physical.startpointTLI);
514 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
515 errmsg(
"could not start WAL streaming: %s",
538 (
errcode(ERRCODE_CONNECTION_FAILURE),
539 errmsg(
"could not send end-of-streaming message to primary: %s",
561 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
562 errmsg(
"unexpected result set after end-of-streaming")));
576 (
errcode(ERRCODE_CONNECTION_FAILURE),
577 errmsg(
"error while shutting down streaming COPY: %s",
586 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
587 errmsg(
"error reading result of streaming command: %s",
595 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
596 errmsg(
"unexpected result after CommandComplete: %s",
606 char **content,
int *
len)
616 snprintf(cmd,
sizeof(cmd),
"TIMELINE_HISTORY %u", tli);
622 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
623 errmsg(
"could not receive timeline history file from "
624 "the primary server: %s",
634 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
635 errmsg(
"invalid response from primary server"),
636 errdetail(
"Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
730 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
785 conn->recvBuf = NULL;
794 (
errcode(ERRCODE_CONNECTION_FAILURE),
795 errmsg(
"could not receive data from WAL stream: %s",
831 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
832 errmsg(
"unexpected result after CommandComplete: %s",
847 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
848 errmsg(
"could not receive data from WAL stream: %s",
854 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
855 errmsg(
"could not receive data from WAL stream: %s",
859 *buffer =
conn->recvBuf;
874 (
errcode(ERRCODE_CONNECTION_FAILURE),
875 errmsg(
"could not send data to WAL stream: %s",
892 int use_new_options_syntax;
906 if (use_new_options_syntax)
911 if (use_new_options_syntax)
917 if (use_new_options_syntax)
919 switch (snapshot_action)
934 switch (snapshot_action)
948 if (use_new_options_syntax)
953 if (use_new_options_syntax)
966 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
967 errmsg(
"could not create replication slot \"%s\": %s",
999 const int nRetTypes,
const Oid *retTypes)
1010 if (nfields != nRetTypes)
1012 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
1013 errmsg(
"invalid query response"),
1014 errdetail(
"Expected %d fields, got %d fields.",
1015 nRetTypes, nfields)));
1021 for (coln = 0; coln < nRetTypes; coln++)
1023 PQfname(pgres, coln), retTypes[coln], -1, 0);
1032 "libpqrcv query result context",
1036 for (tupn = 0; tupn <
PQntuples(pgres); tupn++)
1048 for (coln = 0; coln < nfields; coln++)
1075 const int nRetTypes,
const Oid *retTypes)
1079 char *diag_sqlstate;
1083 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1084 errmsg(
"the query interface requires a database connection")));
1115 walres->
err =
_(
"empty query");
1121 walres->
err =
_(
"unexpected pipeline mode");
1161 foreach(lc, strings)
#define ALWAYS_SECURE_SEARCH_PATH_SQL
static void PGresult * res
elog(ERROR, "%s: %s", p2, msg)
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)
PQconninfoOption * PQconninfoParse(const char *conninfo, char **errmsg)
char * PQhost(const PGconn *conn)
int PQconnectionUsedPassword(const PGconn *conn)
void PQconninfoFree(PQconninfoOption *connOptions)
PQconninfoOption * PQconninfo(PGconn *conn)
char * PQerrorMessage(const PGconn *conn)
ConnStatusType PQstatus(const PGconn *conn)
int PQbackendPID(const PGconn *conn)
char * PQport(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 * PQescapeIdentifier(PGconn *conn, const char *str, size_t len)
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)
char * PQescapeLiteral(PGconn *conn, const char *str, size_t len)
int PQconsumeInput(PGconn *conn)
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
int PQsendQuery(PGconn *conn, const char *query)
int PQisBusy(PGconn *conn)
char * PQresultErrorField(const PGresult *res, int fieldcode)
int PQnfields(const PGresult *res)
PGresult * PQgetResult(PGconn *conn)
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
static void libpqsrv_disconnect(PGconn *conn)
static PGconn * libpqsrv_connect_params(const char *const *keywords, const char *const *values, int expand_dbname, uint32 wait_event_info)
Assert(fmt[strlen(fmt) - 1] !='\n')
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 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 char * libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname, bool temporary, bool two_phase, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
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 WalReceiverConn * libpqrcv_connect(const char *conninfo, bool logical, bool must_use_password, const char *appname, char **err)
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
const char * GetDatabaseEncodingName(void)
void MemoryContextReset(MemoryContext context)
char * pchomp(const char *in)
char * pstrdup(const char *in)
void pfree(void *pointer)
void * palloc0(Size size)
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,...)
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)