32 #define ERRCODE_DUPLICATE_OBJECT "42710"
39 #define MINIMUM_VERSION_FOR_SHOW_CMD 100000
44 #define MINIMUM_VERSION_FOR_GROUP_ACCESS 110000
68 const char **keywords;
87 if (conn_opts == NULL)
90 for (conn_opt = conn_opts; conn_opt->
keyword != NULL; conn_opt++)
92 if (conn_opt->
val != NULL && conn_opt->
val[0] !=
'\0')
96 keywords =
pg_malloc0((argcount + 1) *
sizeof(*keywords));
103 keywords[
i] =
"dbname";
107 for (conn_opt = conn_opts; conn_opt->
keyword != NULL; conn_opt++)
109 if (conn_opt->
val != NULL && conn_opt->
val[0] !=
'\0')
119 keywords =
pg_malloc0((argcount + 1) *
sizeof(*keywords));
121 keywords[
i] =
"dbname";
126 keywords[
i] =
"replication";
129 keywords[
i] =
"fallback_application_name";
135 keywords[
i] =
"host";
141 keywords[
i] =
"user";
147 keywords[
i] =
"port";
162 need_password =
false;
168 keywords[
i] =
"password";
188 pg_fatal(
"could not connect to server");
196 need_password =
true;
199 while (need_password);
245 pg_log_error(
"could not determine server setting for integer_datetimes");
250 if (strcmp(tmpparam,
"on") != 0)
252 pg_log_error(
"integer_datetimes compile flag does not match server");
295 pg_log_error(
"could not send replication command \"%s\": %s",
303 pg_log_error(
"could not fetch WAL segment size: got %d rows and %d fields, expected %d rows and %d or more fields",
311 if (sscanf(
PQgetvalue(
res, 0, 0),
"%d%2s", &xlog_val, xlog_unit) != 2)
321 if (strcmp(xlog_unit,
"MB") == 0)
322 multiplier = 1024 * 1024;
323 else if (strcmp(xlog_unit,
"GB") == 0)
324 multiplier = 1024 * 1024 * 1024;
332 "remote server reported invalid WAL segment size (%d bytes)",
369 pg_log_error(
"could not send replication command \"%s\": %s",
377 pg_log_error(
"could not fetch group access flag: got %d rows and %d fields, expected %d rows and %d or more fields",
386 pg_log_error(
"group access flag could not be parsed: %s",
421 pg_log_error(
"could not send replication command \"%s\": %s",
429 pg_log_error(
"could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields",
441 if (starttli != NULL)
449 pg_log_error(
"could not parse write-ahead log location \"%s\"",
455 *
startpos = ((uint64) hi) << 32 | lo;
466 pg_log_error(
"could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields",
498 *restart_lsn = lsn_loc;
500 *restart_tli = tli_loc;
509 pg_log_error(
"could not send replication command \"%s\": %s",
518 pg_log_error(
"could not read replication slot \"%s\": got %d rows and %d fields, expected %d rows and %d fields",
530 pg_log_error(
"replication slot \"%s\" does not exist", slot_name);
541 pg_log_error(
"expected a physical replication slot, got type \"%s\" instead",
555 pg_log_error(
"could not parse restart_lsn \"%s\" for replication slot \"%s\"",
560 lsn_loc = ((uint64) hi) << 32 | lo;
571 *restart_lsn = lsn_loc;
573 *restart_tli = tli_loc;
584 bool is_temporary,
bool is_physical,
bool reserve_wal,
594 (!is_physical &&
plugin != NULL));
596 Assert(slot_name != NULL);
608 if (use_new_option_syntax)
625 if (use_new_option_syntax)
627 "SNAPSHOT",
"nothing");
630 "NOEXPORT_SNAPSHOT");
633 if (use_new_option_syntax)
636 if (query->
data[query->
len - 1] ==
'(')
639 query->
data[query->
len] =
'\0';
661 pg_log_error(
"could not send replication command \"%s\": %s",
672 pg_log_error(
"could not create replication slot \"%s\": got %d rows and %d fields, expected %d rows and %d fields",
696 Assert(slot_name != NULL);
706 pg_log_error(
"could not send replication command \"%s\": %s",
716 pg_log_error(
"could not drop replication slot \"%s\": got %d rows and %d fields, expected %d rows and %d fields",
743 if (
buf->len > 0 &&
buf->data[
buf->len - 1] !=
'(')
745 if (use_new_option_syntax)
762 char *option_name,
char *option_value)
766 if (option_value != NULL)
768 size_t length = strlen(option_value);
769 char *escaped_value =
palloc(1 + 2 * length);
773 pfree(escaped_value);
785 char *option_name,
int32 option_value)
817 long *secs,
int *microsecs)
844 return (diff >= msec * INT64CONST(1000));
855 memcpy(
buf, &n64,
sizeof(n64));
866 memcpy(&n64,
buf,
sizeof(n64));
static Datum values[MAXATTR]
#define ngettext(s, p, n)
#define ALWAYS_SECURE_SEARCH_PATH_SQL
#define POSTGRES_EPOCH_JDATE
static void PGresult * res
const char * PQparameterStatus(const PGconn *conn, const char *paramName)
int PQserverVersion(const PGconn *conn)
PQconninfoOption * PQconninfoParse(const char *conninfo, char **errmsg)
PGconn * PQconnectdbParams(const char *const *keywords, const char *const *values, int expand_dbname)
void PQconninfoFree(PQconninfoOption *connOptions)
int PQconnectionNeedsPassword(const PGconn *conn)
char * PQerrorMessage(const PGconn *conn)
ConnStatusType PQstatus(const PGconn *conn)
void PQfinish(PGconn *conn)
size_t PQescapeStringConn(PGconn *conn, char *to, const char *from, size_t length, int *error)
ExecStatusType PQresultStatus(const PGresult *res)
int PQntuples(const PGresult *res)
PGresult * PQexec(PGconn *conn, const char *query)
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
char * PQresultErrorField(const PGresult *res, int fieldcode)
int PQnfields(const PGresult *res)
void * pg_malloc0(size_t size)
char * pg_strdup(const char *in)
void SetDataDirectoryCreatePerm(int dataDirMode)
Assert(fmt[strlen(fmt) - 1] !='\n')
#define pg_log_error(...)
#define pg_log_error_detail(...)
void pfree(void *pointer)
#define DEFAULT_XLOG_SEG_SIZE
static bool slot_exists_ok
static const char * plugin
static XLogRecPtr startpos
PQExpBuffer createPQExpBuffer(void)
void appendPQExpBuffer(PQExpBuffer str, const char *fmt,...)
void destroyPQExpBuffer(PQExpBuffer str)
void appendPQExpBufferChar(PQExpBuffer str, char ch)
void appendPQExpBufferStr(PQExpBuffer str, const char *data)
char * simple_prompt(const char *prompt, bool echo)
void AppendIntegerCommandOption(PQExpBuffer buf, bool use_new_option_syntax, char *option_name, int32 option_value)
bool RetrieveWalSegSize(PGconn *conn)
#define ERRCODE_DUPLICATE_OBJECT
#define MINIMUM_VERSION_FOR_SHOW_CMD
int64 fe_recvint64(char *buf)
bool CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin, bool is_temporary, bool is_physical, bool reserve_wal, bool slot_exists_ok, bool two_phase)
TimestampTz feGetCurrentTimestamp(void)
#define MINIMUM_VERSION_FOR_GROUP_ACCESS
void feTimestampDifference(TimestampTz start_time, TimestampTz stop_time, long *secs, int *microsecs)
void AppendPlainCommandOption(PQExpBuffer buf, bool use_new_option_syntax, char *option_name)
void AppendStringCommandOption(PQExpBuffer buf, bool use_new_option_syntax, char *option_name, char *option_value)
void fe_sendint64(int64 i, char *buf)
PGconn * GetConnection(void)
bool feTimestampDifferenceExceeds(TimestampTz start_time, TimestampTz stop_time, int msec)
bool GetSlotInformation(PGconn *conn, const char *slot_name, XLogRecPtr *restart_lsn, TimeLineID *restart_tli)
static bool RetrieveDataDirCreatePerm(PGconn *conn)
bool DropReplicationSlot(PGconn *conn, const char *slot_name)
bool RunIdentifySystem(PGconn *conn, char **sysid, TimeLineID *starttli, XLogRecPtr *startpos, char **db_name)
int gettimeofday(struct timeval *tp, void *tzp)
#define IsValidWalSegSize(size)
#define InvalidXLogRecPtr