53 bool logical,
const char *appname,
58 char **sender_host,
int *sender_port);
64 char **content,
int *
len);
116 elog(
ERROR,
"libpqwalreceiver already loaded");
141 keys[++
i] =
"replication";
142 vals[
i] = logical ?
"database" :
"true";
149 keys[++
i] =
"dbname";
150 vals[
i] =
"replication";
152 keys[++
i] =
"fallback_application_name";
157 keys[++
i] =
"client_encoding";
168 keys[++
i] =
"options";
169 vals[
i] =
"-c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3";
240 (
errmsg(
"could not clear search path: %s",
246 conn->logical = logical;
264 char *errcopy = err ?
pstrdup(err) :
"out of memory";
268 (
errcode(ERRCODE_SYNTAX_ERROR),
269 errmsg(
"invalid connection string syntax: %s", errcopy)));
292 if (conn_opts == NULL)
294 (
errcode(ERRCODE_OUT_OF_MEMORY),
295 errmsg(
"could not parse connection string: %s",
296 _(
"out of memory"))));
299 for (conn_opt = conn_opts; conn_opt->
keyword != NULL; conn_opt++)
304 if (strchr(conn_opt->
dispchar,
'D') ||
305 conn_opt->
val == NULL ||
306 conn_opt->
val[0] ==
'\0')
310 obfuscate = strchr(conn_opt->
dispchar,
'*') != NULL;
313 buf.len == 0 ?
"" :
" ",
315 obfuscate ?
"********" : conn_opt->
val);
340 if (ret && strlen(ret) != 0)
344 if (ret && strlen(ret) != 0)
345 *sender_port = atoi(ret);
367 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
368 errmsg(
"could not receive database system identifier and timeline ID from "
369 "the primary server: %s",
379 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
380 errmsg(
"invalid response from primary server"),
381 errdetail(
"Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
382 ntuples, nfields, 3, 1)));
388 return primary_sysid;
441 char *pubnames_literal;
446 options->proto.logical.proto_version);
448 if (
options->proto.logical.streaming &&
452 if (
options->proto.logical.twophase &&
456 pubnames =
options->proto.logical.publication_names;
460 (
errcode(ERRCODE_OUT_OF_MEMORY),
461 errmsg(
"could not start WAL streaming: %s",
464 strlen(pubnames_str));
465 if (!pubnames_literal)
467 (
errcode(ERRCODE_OUT_OF_MEMORY),
468 errmsg(
"could not start WAL streaming: %s",
474 if (
options->proto.logical.binary &&
482 options->proto.physical.startpointTLI);
497 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
498 errmsg(
"could not start WAL streaming: %s",
521 (
errcode(ERRCODE_CONNECTION_FAILURE),
522 errmsg(
"could not send end-of-streaming message to primary: %s",
544 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
545 errmsg(
"unexpected result set after end-of-streaming")));
559 (
errcode(ERRCODE_CONNECTION_FAILURE),
560 errmsg(
"error while shutting down streaming COPY: %s",
569 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
570 errmsg(
"error reading result of streaming command: %s",
578 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
579 errmsg(
"unexpected result after CommandComplete: %s",
589 char **content,
int *
len)
599 snprintf(cmd,
sizeof(cmd),
"TIMELINE_HISTORY %u", tli);
605 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
606 errmsg(
"could not receive timeline history file from "
607 "the primary server: %s",
617 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
618 errmsg(
"invalid response from primary server"),
619 errdetail(
"Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
741 if (
conn->recvBuf != NULL)
768 if (
conn->recvBuf != NULL)
770 conn->recvBuf = NULL;
779 (
errcode(ERRCODE_CONNECTION_FAILURE),
780 errmsg(
"could not receive data from WAL stream: %s",
816 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
817 errmsg(
"unexpected result after CommandComplete: %s",
832 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
833 errmsg(
"could not receive data from WAL stream: %s",
839 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
840 errmsg(
"could not receive data from WAL stream: %s",
844 *buffer =
conn->recvBuf;
859 (
errcode(ERRCODE_CONNECTION_FAILURE),
860 errmsg(
"could not send data to WAL stream: %s",
877 int use_new_options_syntax;
891 if (use_new_options_syntax)
896 if (use_new_options_syntax)
902 if (use_new_options_syntax)
904 switch (snapshot_action)
919 switch (snapshot_action)
933 if (use_new_options_syntax)
938 if (use_new_options_syntax)
951 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
952 errmsg(
"could not create replication slot \"%s\": %s",
984 const int nRetTypes,
const Oid *retTypes)
995 if (nfields != nRetTypes)
997 (
errcode(ERRCODE_PROTOCOL_VIOLATION),
998 errmsg(
"invalid query response"),
999 errdetail(
"Expected %d fields, got %d fields.",
1000 nRetTypes, nfields)));
1006 for (coln = 0; coln < nRetTypes; coln++)
1008 PQfname(pgres, coln), retTypes[coln], -1, 0);
1017 "libpqrcv query result context",
1021 for (tupn = 0; tupn <
PQntuples(pgres); tupn++)
1033 for (coln = 0; coln < nfields; coln++)
1060 const int nRetTypes,
const Oid *retTypes)
1064 char *diag_sqlstate;
1068 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1069 errmsg(
"the query interface requires a database connection")));
1100 walres->
err =
_(
"empty query");
1106 walres->
err =
_(
"unexpected pipeline mode");
1146 foreach(lc, strings)
#define ALWAYS_SECURE_SEARCH_PATH_SQL
static void PGresult * res
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,...)
HeapTuple BuildTupleFromCStrings(AttInMetadata *attinmeta, char **values)
AttInMetadata * TupleDescGetAttInMetadata(TupleDesc tupdesc)
int PQserverVersion(const PGconn *conn)
PQconninfoOption * PQconninfoParse(const char *conninfo, char **errmsg)
PGconn * PQconnectStartParams(const char *const *keywords, const char *const *values, int expand_dbname)
char * PQhost(const PGconn *conn)
PostgresPollingStatusType PQconnectPoll(PGconn *conn)
void PQconninfoFree(PQconninfoOption *connOptions)
PQconninfoOption * PQconninfo(PGconn *conn)
char * PQerrorMessage(const PGconn *conn)
ConnStatusType PQstatus(const PGconn *conn)
void PQfinish(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)
void PQclear(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
#define WL_SOCKET_CONNECTED
#define WL_SOCKET_WRITEABLE
PostgresPollingStatusType
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 void static void status(const char *fmt,...) pg_attribute_printf(1
#define CStringGetDatum(X)
void initPQExpBuffer(PQExpBuffer str)
void appendPQExpBuffer(PQExpBuffer str, const char *fmt,...)
void termPQExpBuffer(PQExpBuffer str)
#define PQExpBufferDataBroken(buf)
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
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)