52 bool logical,
const char *appname,
57 char **sender_host,
int *sender_port);
63 char **content,
int *
len);
115 elog(
ERROR,
"libpqwalreceiver already loaded");
139 keys[++
i] =
"replication";
140 vals[
i] = logical ?
"database" :
"true";
147 keys[++
i] =
"dbname";
148 vals[
i] =
"replication";
150 keys[++
i] =
"fallback_application_name";
155 keys[++
i] =
"client_encoding";
166 keys[++
i] =
"options";
167 vals[
i] =
"-c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3";
181 goto bad_connection_errmsg;
199 conn->logical = logical;
204 bad_connection_errmsg:
231 (
errcode(ERRCODE_SYNTAX_ERROR),
232 errmsg(
"invalid connection string syntax: %s", errcopy)));
255 if (conn_opts == NULL)
257 (
errcode(ERRCODE_OUT_OF_MEMORY),
258 errmsg(
"could not parse connection string: %s",
259 _(
"out of memory"))));
262 for (conn_opt = conn_opts; conn_opt->
keyword != NULL; conn_opt++)
267 if (strchr(conn_opt->
dispchar,
'D') ||
268 conn_opt->
val == NULL ||
269 conn_opt->
val[0] ==
'\0')
273 obfuscate = strchr(conn_opt->
dispchar,
'*') != NULL;
276 buf.len == 0 ?
"" :
" ",
278 obfuscate ?
"********" : conn_opt->
val);
303 if (ret && strlen(ret) != 0)
307 if (ret && strlen(ret) != 0)
308 *sender_port = atoi(ret);
330 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
331 errmsg(
"could not receive database system identifier and timeline ID from "
332 "the primary server: %s",
342 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
343 errmsg(
"invalid response from primary server"),
344 errdetail(
"Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
345 ntuples, nfields, 3, 1)));
351 return primary_sysid;
404 char *pubnames_literal;
409 options->proto.logical.proto_version);
411 if (
options->proto.logical.streaming_str)
413 options->proto.logical.streaming_str);
415 if (
options->proto.logical.twophase &&
419 if (
options->proto.logical.origin &&
422 options->proto.logical.origin);
424 pubnames =
options->proto.logical.publication_names;
428 (
errcode(ERRCODE_OUT_OF_MEMORY),
429 errmsg(
"could not start WAL streaming: %s",
432 strlen(pubnames_str));
433 if (!pubnames_literal)
435 (
errcode(ERRCODE_OUT_OF_MEMORY),
436 errmsg(
"could not start WAL streaming: %s",
442 if (
options->proto.logical.binary &&
450 options->proto.physical.startpointTLI);
465 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
466 errmsg(
"could not start WAL streaming: %s",
489 (
errcode(ERRCODE_CONNECTION_FAILURE),
490 errmsg(
"could not send end-of-streaming message to primary: %s",
512 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
513 errmsg(
"unexpected result set after end-of-streaming")));
527 (
errcode(ERRCODE_CONNECTION_FAILURE),
528 errmsg(
"error while shutting down streaming COPY: %s",
537 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
538 errmsg(
"error reading result of streaming command: %s",
546 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
547 errmsg(
"unexpected result after CommandComplete: %s",
557 char **content,
int *
len)
567 snprintf(cmd,
sizeof(cmd),
"TIMELINE_HISTORY %u", tli);
573 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
574 errmsg(
"could not receive timeline history file from "
575 "the primary server: %s",
585 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
586 errmsg(
"invalid response from primary server"),
587 errdetail(
"Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
736 conn->recvBuf = NULL;
745 (
errcode(ERRCODE_CONNECTION_FAILURE),
746 errmsg(
"could not receive data from WAL stream: %s",
782 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
783 errmsg(
"unexpected result after CommandComplete: %s",
798 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
799 errmsg(
"could not receive data from WAL stream: %s",
805 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
806 errmsg(
"could not receive data from WAL stream: %s",
810 *buffer =
conn->recvBuf;
825 (
errcode(ERRCODE_CONNECTION_FAILURE),
826 errmsg(
"could not send data to WAL stream: %s",
843 int use_new_options_syntax;
857 if (use_new_options_syntax)
862 if (use_new_options_syntax)
868 if (use_new_options_syntax)
870 switch (snapshot_action)
885 switch (snapshot_action)
899 if (use_new_options_syntax)
904 if (use_new_options_syntax)
917 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
918 errmsg(
"could not create replication slot \"%s\": %s",
950 const int nRetTypes,
const Oid *retTypes)
961 if (nfields != nRetTypes)
963 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
964 errmsg(
"invalid query response"),
965 errdetail(
"Expected %d fields, got %d fields.",
966 nRetTypes, nfields)));
972 for (coln = 0; coln < nRetTypes; coln++)
974 PQfname(pgres, coln), retTypes[coln], -1, 0);
983 "libpqrcv query result context",
987 for (tupn = 0; tupn <
PQntuples(pgres); tupn++)
999 for (coln = 0; coln < nfields; coln++)
1026 const int nRetTypes,
const Oid *retTypes)
1030 char *diag_sqlstate;
1034 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1035 errmsg(
"the query interface requires a database connection")));
1066 walres->
err =
_(
"empty query");
1072 walres->
err =
_(
"unexpected pipeline mode");
1112 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 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)
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 WalReceiverConn * libpqrcv_connect(const char *conninfo, bool logical, const char *appname, char **err)
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 void libpqrcv_check_conninfo(const char *conninfo)
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 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)
@ WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE
@ WAIT_EVENT_LIBPQWALRECEIVER_CONNECT
void ProcessWalRcvInterrupts(void)
WalReceiverFunctionsType * WalReceiverFunctions
#define LSN_FORMAT_ARGS(lsn)