streamutil.h File Reference

#include "access/xlogdefs.h"
#include "datatype/timestamp.h"
#include "libpq-fe.h"
#include "pqexpbuffer.h"

Include dependency graph for streamutil.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Functions
PGconn *	GetConnection (void)
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, bool failover)
bool	DropReplicationSlot (PGconn *conn, const char *slot_name)
bool	RunIdentifySystem (PGconn *conn, char **sysid, TimeLineID *starttli, XLogRecPtr *startpos, char **db_name)
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	AppendIntegerCommandOption (PQExpBuffer buf, bool use_new_option_syntax, char *option_name, int32 option_value)
bool	GetSlotInformation (PGconn *conn, const char *slot_name, XLogRecPtr *restart_lsn, TimeLineID *restart_tli)
bool	RetrieveWalSegSize (PGconn *conn)
TimestampTz	feGetCurrentTimestamp (void)
void	feTimestampDifference (TimestampTz start_time, TimestampTz stop_time, long *secs, int *microsecs)
bool	feTimestampDifferenceExceeds (TimestampTz start_time, TimestampTz stop_time, int msec)
void	fe_sendint64 (int64 i, char *buf)
int64	fe_recvint64 (char *buf)

Variables
const char *	progname
char *	connection_string
char *	dbhost
char *	dbuser
char *	dbport
char *	dbname
int	dbgetpassword
int	WalSegSz
PGconn *	conn

Function Documentation

AppendIntegerCommandOption()

void AppendIntegerCommandOption	(	PQExpBuffer	buf,
		bool	use_new_option_syntax,
		char *	option_name,
		int32	option_value
	)

Definition at line 790 of file streamutil.c.

{
    AppendPlainCommandOption(buf, use_new_option_syntax, option_name);
 
    appendPQExpBuffer(buf, " %d", option_value);
}

References AppendPlainCommandOption(), appendPQExpBuffer(), and buf.

Referenced by BaseBackup().

AppendPlainCommandOption()

void AppendPlainCommandOption	(	PQExpBuffer	buf,
		bool	use_new_option_syntax,
		char *	option_name
	)

Definition at line 746 of file streamutil.c.

{
    if (buf->len > 0 && buf->data[buf->len - 1] != '(')
    {
        if (use_new_option_syntax)
            appendPQExpBufferStr(buf, ", ");
        else
            appendPQExpBufferChar(buf, ' ');
    }
 
    appendPQExpBuffer(buf, " %s", option_name);
}

References appendPQExpBuffer(), appendPQExpBufferChar(), appendPQExpBufferStr(), and buf.

Referenced by AppendIntegerCommandOption(), AppendStringCommandOption(), BaseBackup(), and CreateReplicationSlot().

AppendStringCommandOption()

void AppendStringCommandOption	(	PQExpBuffer	buf,
		bool	use_new_option_syntax,
		char *	option_name,
		char *	option_value
	)

Definition at line 767 of file streamutil.c.

{
    AppendPlainCommandOption(buf, use_new_option_syntax, option_name);
 
    if (option_value != NULL)
    {
        size_t      length = strlen(option_value);
        char       *escaped_value = palloc(1 + 2 * length);
 
        PQescapeStringConn(conn, escaped_value, option_value, length, NULL);
        appendPQExpBuffer(buf, " '%s'", escaped_value);
        pfree(escaped_value);
    }
}

References AppendPlainCommandOption(), appendPQExpBuffer(), buf, conn, palloc(), pfree(), and PQescapeStringConn().

Referenced by BaseBackup(), and CreateReplicationSlot().

CreateReplicationSlot()

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,
		bool	failover
	)

Definition at line 584 of file streamutil.c.

{
    PQExpBuffer query;
    PGresult   *res;
    bool        use_new_option_syntax = (PQserverVersion(conn) >= 150000);
 
    query = createPQExpBuffer();
 
    Assert((is_physical && plugin == NULL) ||
           (!is_physical && plugin != NULL));
    Assert(!(two_phase && is_physical));
    Assert(!(failover && is_physical));
    Assert(slot_name != NULL);
 
    /* Build base portion of query */
    appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\"", slot_name);
    if (is_temporary)
        appendPQExpBufferStr(query, " TEMPORARY");
    if (is_physical)
        appendPQExpBufferStr(query, " PHYSICAL");
    else
        appendPQExpBuffer(query, " LOGICAL \"%s\"", plugin);
 
    /* Add any requested options */
    if (use_new_option_syntax)
        appendPQExpBufferStr(query, " (");
    if (is_physical)
    {
        if (reserve_wal)
            AppendPlainCommandOption(query, use_new_option_syntax,
                                     "RESERVE_WAL");
    }
    else
    {
        if (failover && PQserverVersion(conn) >= 170000)
            AppendPlainCommandOption(query, use_new_option_syntax,
                                     "FAILOVER");
 
        if (two_phase && PQserverVersion(conn) >= 150000)
            AppendPlainCommandOption(query, use_new_option_syntax,
                                     "TWO_PHASE");
 
        if (PQserverVersion(conn) >= 100000)
        {
            /* pg_recvlogical doesn't use an exported snapshot, so suppress */
            if (use_new_option_syntax)
                AppendStringCommandOption(query, use_new_option_syntax,
                                          "SNAPSHOT", "nothing");
            else
                AppendPlainCommandOption(query, use_new_option_syntax,
                                         "NOEXPORT_SNAPSHOT");
        }
    }
    if (use_new_option_syntax)
    {
        /* Suppress option list if it would be empty, otherwise terminate */
        if (query->data[query->len - 1] == '(')
        {
            query->len -= 2;
            query->data[query->len] = '\0';
        }
        else
            appendPQExpBufferChar(query, ')');
    }
 
    /* Now run the query */
    res = PQexec(conn, query->data);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        const char *sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
 
        if (slot_exists_ok &&
            sqlstate &&
            strcmp(sqlstate, ERRCODE_DUPLICATE_OBJECT) == 0)
        {
            destroyPQExpBuffer(query);
            PQclear(res);
            return true;
        }
        else
        {
            pg_log_error("could not send replication command \"%s\": %s",
                         query->data, PQerrorMessage(conn));
 
            destroyPQExpBuffer(query);
            PQclear(res);
            return false;
        }
    }
 
    if (PQntuples(res) != 1 || PQnfields(res) != 4)
    {
        pg_log_error("could not create replication slot \"%s\": got %d rows and %d fields, expected %d rows and %d fields",
                     slot_name,
                     PQntuples(res), PQnfields(res), 1, 4);
 
        destroyPQExpBuffer(query);
        PQclear(res);
        return false;
    }
 
    destroyPQExpBuffer(query);
    PQclear(res);
    return true;
}

References AppendPlainCommandOption(), appendPQExpBuffer(), appendPQExpBufferChar(), appendPQExpBufferStr(), AppendStringCommandOption(), Assert(), conn, createPQExpBuffer(), PQExpBufferData::data, destroyPQExpBuffer(), ERRCODE_DUPLICATE_OBJECT, failover, PQExpBufferData::len, PG_DIAG_SQLSTATE, pg_log_error, PGRES_TUPLES_OK, plugin, PQclear(), PQerrorMessage(), PQexec(), PQnfields(), PQntuples(), PQresultErrorField(), PQresultStatus(), PQserverVersion(), slot_exists_ok, and two_phase.

DropReplicationSlot()

bool DropReplicationSlot	(	PGconn *	conn,
		const char *	slot_name
	)

Definition at line 697 of file streamutil.c.

{
    PQExpBuffer query;
    PGresult   *res;
 
    Assert(slot_name != NULL);
 
    query = createPQExpBuffer();
 
    /* Build query */
    appendPQExpBuffer(query, "DROP_REPLICATION_SLOT \"%s\"",
                      slot_name);
    res = PQexec(conn, query->data);
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
    {
        pg_log_error("could not send replication command \"%s\": %s",
                     query->data, PQerrorMessage(conn));
 
        destroyPQExpBuffer(query);
        PQclear(res);
        return false;
    }
 
    if (PQntuples(res) != 0 || PQnfields(res) != 0)
    {
        pg_log_error("could not drop replication slot \"%s\": got %d rows and %d fields, expected %d rows and %d fields",
                     slot_name,
                     PQntuples(res), PQnfields(res), 0, 0);
 
        destroyPQExpBuffer(query);
        PQclear(res);
        return false;
    }
 
    destroyPQExpBuffer(query);
    PQclear(res);
    return true;
}

References appendPQExpBuffer(), Assert(), conn, createPQExpBuffer(), PQExpBufferData::data, destroyPQExpBuffer(), pg_log_error, PGRES_COMMAND_OK, PQclear(), PQerrorMessage(), PQexec(), PQnfields(), PQntuples(), and PQresultStatus().

fe_recvint64()

int64 fe_recvint64

(

char *

buf

)

Definition at line 868 of file streamutil.c.

{
    uint64      n64;
 
    memcpy(&n64, buf, sizeof(n64));
 
    return pg_ntoh64(n64);
}

References buf, and pg_ntoh64.

Referenced by ProcessXLogDataMsg(), and StreamLogicalLog().

fe_sendint64()

void fe_sendint64	(	int64	i,
		char *	buf
	)

Definition at line 857 of file streamutil.c.

{
    uint64      n64 = pg_hton64(i);
 
    memcpy(buf, &n64, sizeof(n64));
}

References buf, i, and pg_hton64.

Referenced by sendFeedback().

feGetCurrentTimestamp()

TimestampTz feGetCurrentTimestamp

(

void

)

Definition at line 803 of file streamutil.c.

{
    TimestampTz result;
    struct timeval tp;
 
    gettimeofday(&tp, NULL);
 
    result = (TimestampTz) tp.tv_sec -
        ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY);
    result = (result * USECS_PER_SEC) + tp.tv_usec;
 
    return result;
}

References gettimeofday(), POSTGRES_EPOCH_JDATE, SECS_PER_DAY, UNIX_EPOCH_JDATE, and USECS_PER_SEC.

Referenced by flushAndSendFeedback(), HandleCopyStream(), ProcessKeepaliveMsg(), and StreamLogicalLog().

feTimestampDifference()

void feTimestampDifference	(	TimestampTz	start_time,
		TimestampTz	stop_time,
		long *	secs,
		int *	microsecs
	)

Definition at line 822 of file streamutil.c.

{
    TimestampTz diff = stop_time - start_time;
 
    if (diff <= 0)
    {
        *secs = 0;
        *microsecs = 0;
    }
    else
    {
        *secs = (long) (diff / USECS_PER_SEC);
        *microsecs = (int) (diff % USECS_PER_SEC);
    }
}

References start_time, and USECS_PER_SEC.

Referenced by CalculateCopyStreamSleeptime(), and StreamLogicalLog().

feTimestampDifferenceExceeds()

bool feTimestampDifferenceExceeds	(	TimestampTz	start_time,
		TimestampTz	stop_time,
		int	msec
	)

Definition at line 844 of file streamutil.c.

{
    TimestampTz diff = stop_time - start_time;
 
    return (diff >= msec * INT64CONST(1000));
}

References INT64CONST, and start_time.

Referenced by HandleCopyStream(), and StreamLogicalLog().

GetConnection()

PGconn * GetConnection

(

void

)

Definition at line 60 of file streamutil.c.

{
    PGconn     *tmpconn;
    int         argcount = 7;   /* dbname, replication, fallback_app_name,
                                 * host, user, port, password */
    int         i;
    const char **keywords;
    const char **values;
    const char *tmpparam;
    bool        need_password;
    PQconninfoOption *conn_opts = NULL;
    PQconninfoOption *conn_opt;
    char       *err_msg = NULL;
 
    /*
     * pg_recvlogical uses dbname only; others use connection_string only.
     * (Note: both variables will be NULL if there's no command line options.)
     */
    Assert(dbname == NULL || connection_string == NULL);
 
    /*
     * Merge the connection info inputs given in form of connection string,
     * options and default values (dbname=replication, replication=true, etc.)
     */
    i = 0;
    if (connection_string)
    {
        conn_opts = PQconninfoParse(connection_string, &err_msg);
        if (conn_opts == NULL)
            pg_fatal("%s", err_msg);
 
        for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
        {
            if (conn_opt->val != NULL && conn_opt->val[0] != '\0')
                argcount++;
        }
 
        keywords = pg_malloc0((argcount + 1) * sizeof(*keywords));
        values = pg_malloc0((argcount + 1) * sizeof(*values));
 
        /*
         * Set dbname here already, so it can be overridden by a dbname in the
         * connection string.
         */
        keywords[i] = "dbname";
        values[i] = "replication";
        i++;
 
        for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
        {
            if (conn_opt->val != NULL && conn_opt->val[0] != '\0')
            {
                keywords[i] = conn_opt->keyword;
                values[i] = conn_opt->val;
                i++;
            }
        }
    }
    else
    {
        keywords = pg_malloc0((argcount + 1) * sizeof(*keywords));
        values = pg_malloc0((argcount + 1) * sizeof(*values));
        keywords[i] = "dbname";
        values[i] = (dbname == NULL) ? "replication" : dbname;
        i++;
    }
 
    keywords[i] = "replication";
    values[i] = (dbname == NULL) ? "true" : "database";
    i++;
    keywords[i] = "fallback_application_name";
    values[i] = progname;
    i++;
 
    if (dbhost)
    {
        keywords[i] = "host";
        values[i] = dbhost;
        i++;
    }
    if (dbuser)
    {
        keywords[i] = "user";
        values[i] = dbuser;
        i++;
    }
    if (dbport)
    {
        keywords[i] = "port";
        values[i] = dbport;
        i++;
    }
 
    /* If -W was given, force prompt for password, but only the first time */
    need_password = (dbgetpassword == 1 && !password);
 
    do
    {
        /* Get a new password if appropriate */
        if (need_password)
        {
            free(password);
            password = simple_prompt("Password: ", false);
            need_password = false;
        }
 
        /* Use (or reuse, on a subsequent connection) password if we have it */
        if (password)
        {
            keywords[i] = "password";
            values[i] = password;
        }
        else
        {
            keywords[i] = NULL;
            values[i] = NULL;
        }
 
        /*
         * Only expand dbname when we did not already parse the argument as a
         * connection string ourselves.
         */
        tmpconn = PQconnectdbParams(keywords, values, !connection_string);
 
        /*
         * If there is too little memory even to allocate the PGconn object
         * and PQconnectdbParams returns NULL, we call exit(1) directly.
         */
        if (!tmpconn)
            pg_fatal("could not connect to server");
 
        /* If we need a password and -w wasn't given, loop back and get one */
        if (PQstatus(tmpconn) == CONNECTION_BAD &&
            PQconnectionNeedsPassword(tmpconn) &&
            dbgetpassword != -1)
        {
            PQfinish(tmpconn);
            need_password = true;
        }
    }
    while (need_password);
 
    if (PQstatus(tmpconn) != CONNECTION_OK)
    {
        pg_log_error("%s", PQerrorMessage(tmpconn));
        PQfinish(tmpconn);
        free(values);
        free(keywords);
        PQconninfoFree(conn_opts);
        return NULL;
    }
 
    /* Connection ok! */
    free(values);
    free(keywords);
    PQconninfoFree(conn_opts);
 
    /*
     * Set always-secure search path, so malicious users can't get control.
     * The capacity to run normal SQL queries was added in PostgreSQL 10, so
     * the search path cannot be changed (by us or attackers) on earlier
     * versions.
     */
    if (dbname != NULL && PQserverVersion(tmpconn) >= 100000)
    {
        PGresult   *res;
 
        res = PQexec(tmpconn, ALWAYS_SECURE_SEARCH_PATH_SQL);
        if (PQresultStatus(res) != PGRES_TUPLES_OK)
        {
            pg_log_error("could not clear \"search_path\": %s",
                         PQerrorMessage(tmpconn));
            PQclear(res);
            PQfinish(tmpconn);
            exit(1);
        }
        PQclear(res);
    }
 
    /*
     * Ensure we have the same value of integer_datetimes (now always "on") as
     * the server we are connecting to.
     */
    tmpparam = PQparameterStatus(tmpconn, "integer_datetimes");
    if (!tmpparam)
    {
        pg_log_error("could not determine server setting for \"integer_datetimes\"");
        PQfinish(tmpconn);
        exit(1);
    }
 
    if (strcmp(tmpparam, "on") != 0)
    {
        pg_log_error("\"integer_datetimes\" compile flag does not match server");
        PQfinish(tmpconn);
        exit(1);
    }
 
    /*
     * Retrieve the source data directory mode and use it to construct a umask
     * for creating directories and files.
     */
    if (!RetrieveDataDirCreatePerm(tmpconn))
    {
        PQfinish(tmpconn);
        exit(1);
    }
 
    return tmpconn;
}

References ALWAYS_SECURE_SEARCH_PATH_SQL, Assert(), CONNECTION_BAD, CONNECTION_OK, connection_string, dbgetpassword, dbhost, dbname, dbport, dbuser, free, i, _PQconninfoOption::keyword, keywords, password, pg_fatal, pg_log_error, pg_malloc0(), PGRES_TUPLES_OK, PQclear(), PQconnectdbParams(), PQconnectionNeedsPassword(), PQconninfoFree(), PQconninfoParse(), PQerrorMessage(), PQexec(), PQfinish(), PQparameterStatus(), PQresultStatus(), PQserverVersion(), PQstatus(), progname, RetrieveDataDirCreatePerm(), simple_prompt(), _PQconninfoOption::val, and values.

GetSlotInformation()

bool GetSlotInformation	(	PGconn *	conn,
		const char *	slot_name,
		XLogRecPtr *	restart_lsn,
		TimeLineID *	restart_tli
	)

Definition at line 490 of file streamutil.c.

{
    PGresult   *res;
    PQExpBuffer query;
    XLogRecPtr  lsn_loc = InvalidXLogRecPtr;
    TimeLineID  tli_loc = 0;
 
    if (restart_lsn)
        *restart_lsn = lsn_loc;
    if (restart_tli)
        *restart_tli = tli_loc;
 
    query = createPQExpBuffer();
    appendPQExpBuffer(query, "READ_REPLICATION_SLOT %s", slot_name);
    res = PQexec(conn, query->data);
    destroyPQExpBuffer(query);
 
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        pg_log_error("could not send replication command \"%s\": %s",
                     "READ_REPLICATION_SLOT", PQerrorMessage(conn));
        PQclear(res);
        return false;
    }
 
    /* The command should always return precisely one tuple and three fields */
    if (PQntuples(res) != 1 || PQnfields(res) != 3)
    {
        pg_log_error("could not read replication slot \"%s\": got %d rows and %d fields, expected %d rows and %d fields",
                     slot_name, PQntuples(res), PQnfields(res), 1, 3);
        PQclear(res);
        return false;
    }
 
    /*
     * When the slot doesn't exist, the command returns a tuple with NULL
     * values.  This checks only the slot type field.
     */
    if (PQgetisnull(res, 0, 0))
    {
        pg_log_error("replication slot \"%s\" does not exist", slot_name);
        PQclear(res);
        return false;
    }
 
    /*
     * Note that this cannot happen as READ_REPLICATION_SLOT supports only
     * physical slots, but play it safe.
     */
    if (strcmp(PQgetvalue(res, 0, 0), "physical") != 0)
    {
        pg_log_error("expected a physical replication slot, got type \"%s\" instead",
                     PQgetvalue(res, 0, 0));
        PQclear(res);
        return false;
    }
 
    /* restart LSN */
    if (!PQgetisnull(res, 0, 1))
    {
        uint32      hi,
                    lo;
 
        if (sscanf(PQgetvalue(res, 0, 1), "%X/%X", &hi, &lo) != 2)
        {
            pg_log_error("could not parse restart_lsn \"%s\" for replication slot \"%s\"",
                         PQgetvalue(res, 0, 1), slot_name);
            PQclear(res);
            return false;
        }
        lsn_loc = ((uint64) hi) << 32 | lo;
    }
 
    /* current TLI */
    if (!PQgetisnull(res, 0, 2))
        tli_loc = (TimeLineID) atoll(PQgetvalue(res, 0, 2));
 
    PQclear(res);
 
    /* Assign results if requested */
    if (restart_lsn)
        *restart_lsn = lsn_loc;
    if (restart_tli)
        *restart_tli = tli_loc;
 
    return true;
}

References appendPQExpBuffer(), conn, createPQExpBuffer(), PQExpBufferData::data, destroyPQExpBuffer(), InvalidXLogRecPtr, pg_log_error, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQexec(), PQgetisnull(), PQgetvalue(), PQnfields(), PQntuples(), and PQresultStatus().

Referenced by StreamLog().

RetrieveWalSegSize()

bool RetrieveWalSegSize

(

PGconn *

conn

)

Definition at line 276 of file streamutil.c.

{
    PGresult   *res;
    char        xlog_unit[3];
    int         xlog_val,
                multiplier = 1;
 
    /* check connection existence */
    Assert(conn != NULL);
 
    /* for previous versions set the default xlog seg size */
    if (PQserverVersion(conn) < MINIMUM_VERSION_FOR_SHOW_CMD)
    {
        WalSegSz = DEFAULT_XLOG_SEG_SIZE;
        return true;
    }
 
    res = PQexec(conn, "SHOW wal_segment_size");
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        pg_log_error("could not send replication command \"%s\": %s",
                     "SHOW wal_segment_size", PQerrorMessage(conn));
 
        PQclear(res);
        return false;
    }
    if (PQntuples(res) != 1 || PQnfields(res) < 1)
    {
        pg_log_error("could not fetch WAL segment size: got %d rows and %d fields, expected %d rows and %d or more fields",
                     PQntuples(res), PQnfields(res), 1, 1);
 
        PQclear(res);
        return false;
    }
 
    /* fetch xlog value and unit from the result */
    if (sscanf(PQgetvalue(res, 0, 0), "%d%2s", &xlog_val, xlog_unit) != 2)
    {
        pg_log_error("WAL segment size could not be parsed");
        PQclear(res);
        return false;
    }
 
    PQclear(res);
 
    /* set the multiplier based on unit to convert xlog_val to bytes */
    if (strcmp(xlog_unit, "MB") == 0)
        multiplier = 1024 * 1024;
    else if (strcmp(xlog_unit, "GB") == 0)
        multiplier = 1024 * 1024 * 1024;
 
    /* convert and set WalSegSz */
    WalSegSz = xlog_val * multiplier;
 
    if (!IsValidWalSegSize(WalSegSz))
    {
        pg_log_error(ngettext("remote server reported invalid WAL segment size (%d byte)",
                              "remote server reported invalid WAL segment size (%d bytes)",
                              WalSegSz),
                     WalSegSz);
        pg_log_error_detail("The WAL segment size must be a power of two between 1 MB and 1 GB.");
        return false;
    }
 
    return true;
}

References Assert(), conn, DEFAULT_XLOG_SEG_SIZE, IsValidWalSegSize, MINIMUM_VERSION_FOR_SHOW_CMD, ngettext, pg_log_error, pg_log_error_detail, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQexec(), PQgetvalue(), PQnfields(), PQntuples(), PQresultStatus(), PQserverVersion(), and WalSegSz.

Referenced by main().

RunIdentifySystem()

bool RunIdentifySystem	(	PGconn *	conn,
		char **	sysid,
		TimeLineID *	starttli,
		XLogRecPtr *	startpos,
		char **	db_name
	)

Definition at line 409 of file streamutil.c.

{
    PGresult   *res;
    uint32      hi,
                lo;
 
    /* Check connection existence */
    Assert(conn != NULL);
 
    res = PQexec(conn, "IDENTIFY_SYSTEM");
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        pg_log_error("could not send replication command \"%s\": %s",
                     "IDENTIFY_SYSTEM", PQerrorMessage(conn));
 
        PQclear(res);
        return false;
    }
    if (PQntuples(res) != 1 || PQnfields(res) < 3)
    {
        pg_log_error("could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields",
                     PQntuples(res), PQnfields(res), 1, 3);
 
        PQclear(res);
        return false;
    }
 
    /* Get system identifier */
    if (sysid != NULL)
        *sysid = pg_strdup(PQgetvalue(res, 0, 0));
 
    /* Get timeline ID to start streaming from */
    if (starttli != NULL)
        *starttli = atoi(PQgetvalue(res, 0, 1));
 
    /* Get LSN start position if necessary */
    if (startpos != NULL)
    {
        if (sscanf(PQgetvalue(res, 0, 2), "%X/%X", &hi, &lo) != 2)
        {
            pg_log_error("could not parse write-ahead log location \"%s\"",
                         PQgetvalue(res, 0, 2));
 
            PQclear(res);
            return false;
        }
        *startpos = ((uint64) hi) << 32 | lo;
    }
 
    /* Get database name, only available in 9.4 and newer versions */
    if (db_name != NULL)
    {
        *db_name = NULL;
        if (PQserverVersion(conn) >= 90400)
        {
            if (PQnfields(res) < 4)
            {
                pg_log_error("could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields",
                             PQntuples(res), PQnfields(res), 1, 4);
 
                PQclear(res);
                return false;
            }
            if (!PQgetisnull(res, 0, 3))
                *db_name = pg_strdup(PQgetvalue(res, 0, 3));
        }
    }
 
    PQclear(res);
    return true;
}

References Assert(), conn, pg_log_error, pg_strdup(), PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQexec(), PQgetisnull(), PQgetvalue(), PQnfields(), PQntuples(), PQresultStatus(), PQserverVersion(), and startpos.

Referenced by BaseBackup(), main(), ReceiveXlogStream(), and StreamLog().

Variable Documentation

conn

PGconn* conn

extern

Definition at line 52 of file streamutil.c.

Referenced by add_client_identification(), append_db_pattern_cte(), append_rel_pattern_filtered_cte(), append_rel_pattern_raw_cte(), appendQualifiedRelation(), AppendStringCommandOption(), appendStringLiteralConn(), applyRemoteGucs(), BaseBackup(), build_client_final_message(), build_client_first_message(), build_startup_packet(), buildShSecLabels(), canChangeResultMode(), check_and_drop_existing_subscriptions(), check_and_drop_publications(), check_expected_areq(), check_for_pg_role_prefix(), check_for_prepared_transactions(), check_is_install_user(), check_issuer(), check_loadable_libraries(), check_new_cluster_logical_replication_slots(), check_new_cluster_subscription_configuration(), check_old_cluster_subscription_state(), check_prepare_conn(), check_publisher(), check_subscriber(), CheckCopyStreamStop(), CheckServerVersionForStreaming(), cleanup_objects_atexit(), cleanup_user_oauth_flow(), clear_allowed_sasl_mechs(), client_initial_response(), close_cursor(), cluster_all_databases(), cluster_one_database(), compile_database_list(), compile_relation_list_one_db(), configure_remote_session(), confirm_query_canceled_impl(), connect_database(), connect_pg_server(), ConnectDatabase(), connectDatabase(), connectFailureMessage(), connection_failed(), connectMaintenanceDatabase(), connectNoDelay(), connectOptions1(), connectToServer(), copy_connection(), CopyStreamPoll(), CopyStreamReceive(), create_cursor(), create_logical_replication_slot(), create_logical_replication_slots(), create_publication(), create_subscription(), CreateReplicationSlot(), dblink_cancel_query(), dblink_close(), dblink_connect(), dblink_disconnect(), dblink_error_message(), dblink_exec(), dblink_fetch(), dblink_get_conn(), dblink_get_notify(), dblink_is_busy(), dblink_open(), dblink_record_internal(), dblink_res_error(), dblink_res_internalerror(), dblink_security_check(), dblink_send_query(), disconnect_atexit(), disconnect_database(), disconnectDatabase(), do_sql_command(), do_sql_command_begin(), do_sql_command_end(), doConnect(), drop_existing_subscriptions(), drop_failover_replication_slots(), drop_primary_replication_slot(), drop_publication(), drop_replication_slot(), dropDBs(), DropReplicationSlot(), dropRoles(), dropTablespaces(), dumpDatabase(), dumpDatabaseConfig(), dumpDatabases(), dumpLOs(), dumpRoleGUCPrivs(), dumpRoleMembership(), dumpRoles(), dumpTableData_copy(), dumpTablespaces(), dumpUserConfig(), ecpg_raise_backend(), emitHostIdentityInfo(), emitShSecLabels(), enable_subscription(), escape_append_literal(), escape_fmt_id(), escape_identifier(), escape_literal(), escape_string_conn(), estimate_path_cost_size(), executeCommand(), executeMaintenanceCommand(), executeQuery(), executeQueryOrDie(), ExecuteSqlCommand(), exit_nicely(), expand_dbname_patterns(), exportFile(), fetch_more_data(), fill_allowed_sasl_mechs(), fillPGconn(), flushAndSendFeedback(), free_async_ctx(), freePGconn(), gen_reindex_command(), generate_object_name(), get_db_conn(), get_db_infos(), get_dbnames_list_to_restore(), get_parallel_tabidx_list(), get_parallel_tables_list(), get_primary_sysid(), get_publisher_databases(), get_remote_estimate(), get_subscription_count(), get_tablespace_paths(), get_template0_info(), getAnotherTuple(), getBackendKeyData(), getCopyDataMessage(), getCopyResult(), getCopyStart(), getHostaddr(), getNotify(), getParamDescriptions(), getParameterStatus(), getReadyForQuery(), getRowDescriptions(), GetSlotInformation(), gss_read(), handle_oauth_sasl_error(), handleCopyIn(), handleCopyOut(), HandleCopyStream(), HandleEndOfCopyStream(), handleSyncLoss(), importFile(), index_of_allowed_sasl_mech(), init_allowed_encryption_methods(), init_libpq_conn(), init_libpq_source(), initialize_curl(), initialize_SSL(), internal_ping(), issuer_from_well_known_uri(), libpq_append_conn_error(), libpq_fetch_file(), libpq_get_current_wal_insert_lsn(), libpq_prng_init(), libpq_traverse_files(), libpqrcv_alter_slot(), libpqrcv_connect(), libpqrcv_create_slot(), libpqrcv_disconnect(), libpqrcv_endstreaming(), libpqrcv_exec(), libpqrcv_get_backend_pid(), libpqrcv_get_conninfo(), libpqrcv_get_senderinfo(), libpqrcv_identify_system(), libpqrcv_readtimelinehistoryfile(), libpqrcv_receive(), libpqrcv_send(), libpqrcv_server_version(), libpqrcv_startstreaming(), libpqsrv_cancel(), libpqsrv_connect(), libpqsrv_connect_internal(), libpqsrv_connect_params(), libpqsrv_disconnect(), libpqsrv_exec(), libpqsrv_exec_params(), libpqsrv_get_result(), libpqsrv_get_result_last(), lo_close(), lo_creat(), lo_create(), lo_export(), lo_import(), lo_import_internal(), lo_import_with_oid(), lo_initialize(), lo_lseek(), lo_lseek64(), lo_open(), lo_read(), lo_tell(), lo_tell64(), lo_truncate(), lo_truncate64(), lo_unlink(), lo_write(), main(), makeAlterConfigCommand(), materializeResult(), my_truncate(), oauth_exchange(), oauth_init(), old_9_6_invalidate_hash_indexes(), open_client_SSL(), openssl_verify_peer_name_matches_certificate_ip(), openssl_verify_peer_name_matches_certificate_name(), overwrite(), ParallelSlotsAdoptConn(), ParallelSlotsTerminate(), parseInput(), pg_fe_cleanup_oauth_flow(), pg_fe_run_oauth_flow(), pg_fe_run_oauth_flow_impl(), pg_fe_sendauth(), pg_GSS_error(), pg_GSS_load_servicename(), pg_GSS_read(), pg_GSS_write(), pg_password_sendauth(), pg_SASL_continue(), pg_SASL_init(), pgconn_bio_ctrl(), pgconn_bio_read(), pgfdw_cancel_query(), pgfdw_cancel_query_begin(), pgfdw_cancel_query_end(), pgfdw_conn_check(), pgfdw_exec_cleanup_query(), pgfdw_exec_cleanup_query_begin(), pgfdw_exec_cleanup_query_end(), pgfdw_exec_query(), pgfdw_get_cleanup_result(), pgfdw_get_result(), pgfdw_report_error(), pgfdw_security_check(), pgpassfileWarning(), pgtls_close(), pgtls_get_peer_certificate_hash(), pgtls_open_client(), pgtls_read(), pgtls_read_pending(), pgtls_verify_peer_name_matches_certificate_guts(), pgtls_write(), pickout(), postgresAcquireSampleRowsFunc(), postgresAnalyzeForeignTable(), postgresExecForeignTruncate(), postgresGetAnalyzeInfoForForeignTable(), postgresImportForeignSchema(), pq_verify_peer_name_matches_certificate(), pq_verify_peer_name_matches_certificate_ip(), pq_verify_peer_name_matches_certificate_name(), pqAllocCmdQueueEntry(), pqAppendCmdQueueEntry(), PQbackendPID(), pqBuildStartupPacket3(), PQcancelCreate(), PQcancelPoll(), PQchangePassword(), pqCheckInBufferSpace(), pqCheckOutBufferSpace(), pqClearAsyncResult(), pqClearOAuthToken(), PQclientEncoding(), pqClosePGconn(), PQclosePortal(), PQclosePrepared(), pqCommandQueueAdvance(), PQconnectdb(), pqConnectDBComplete(), PQconnectdbParams(), pqConnectDBStart(), PQconnectionNeedsPassword(), PQconnectionUsedGSSAPI(), PQconnectionUsedPassword(), pqConnectOptions2(), PQconnectPoll(), PQconnectStart(), PQconnectStartParams(), PQconninfo(), PQconsumeInput(), PQdb(), PQdefaultSSLKeyPassHook_OpenSSL(), PQdescribePortal(), PQdescribePrepared(), pqDropConnection(), pqDropServerData(), PQencryptPasswordConn(), PQendcopy(), pqEndcopy3(), PQenterPipelineMode(), PQerrorMessage(), PQescapeByteaConn(), PQescapeByteaInternal(), PQescapeIdentifier(), PQescapeInternal(), PQescapeLiteral(), PQescapeStringConn(), PQescapeStringInternal(), PQexec(), PQexecFinish(), PQexecParams(), PQexecPrepared(), PQexecStart(), PQexitPipelineMode(), PQfinish(), PQfireResultCreateEvents(), PQflush(), pqFlush(), PQfn(), PQfullProtocolVersion(), pqFunctionCall3(), pqGetc(), PQgetCancel(), PQgetCopyData(), pqGetCopyData3(), pqGetErrorNotice3(), PQgetgssctx(), pqGetInt(), PQgetline(), pqGetline3(), PQgetlineAsync(), pqGetlineAsync3(), pqGetnchar(), pqGetNegotiateProtocolVersion3(), PQgetResult(), pqGets(), pqGets_append(), pqGets_internal(), PQgetssl(), PQgssEncInUse(), PQhost(), PQhostaddr(), PQinstanceData(), PQisBusy(), PQisnonblocking(), pqMakeEmptyPGconn(), PQmakeEmptyPGresult(), PQnotifies(), PQoptions(), pqPacketSend(), PQparameterStatus(), pqParseDone(), pqParseInput3(), pqParseIntParam(), pqParseProtocolVersion(), PQpass(), PQping(), PQpingParams(), pqPipelineFlush(), pqPipelineProcessQueue(), PQpipelineStatus(), PQpipelineSync(), pqPipelineSyncInternal(), PQport(), PQprepare(), pqPrepareAsyncResult(), PQprotocolVersion(), pqPutc(), PQputCopyData(), PQputCopyEnd(), pqPutInt(), PQputline(), pqPutMsgBytes(), pqPutMsgEnd(), pqPutMsgStart(), PQputnbytes(), pqPutnchar(), pqPuts(), pqReadData(), pqReadReady(), pqRecycleCmdQueueEntry(), PQregisterEventProc(), pqReleaseConnHosts(), PQrequestCancel(), PQreset(), PQresetPoll(), PQresetStart(), pqRowProcessor(), pqSaveErrorResult(), pqSaveParameterStatus(), pqSaveWriteError(), pqsecure_close(), pqsecure_open_client(), pqsecure_open_gss(), pqsecure_raw_read(), pqsecure_raw_write(), pqsecure_read(), pqsecure_write(), PQsendClosePortal(), PQsendClosePrepared(), PQsendDescribePortal(), PQsendDescribePrepared(), PQsendFlushRequest(), PQsendPipelineSync(), PQsendPrepare(), PQsendQuery(), PQsendQueryContinue(), PQsendQueryGuts(), PQsendQueryInternal(), PQsendQueryParams(), PQsendQueryPrepared(), PQsendQueryStart(), pqSendSome(), PQsendTypedCommand(), PQserverVersion(), PQservice(), PQsetChunkedRowsMode(), PQsetClientEncoding(), PQsetdbLogin(), PQsetErrorContextVisibility(), PQsetErrorVerbosity(), PQsetInstanceData(), PQsetnonblocking(), PQsetNoticeProcessor(), PQsetNoticeReceiver(), PQsetSingleRowMode(), PQsetTraceFlags(), pqSkipnchar(), PQsocket(), pqSocketCheck(), PQssl_passwd_cb(), PQsslAttribute(), PQsslAttributeNames(), PQsslInUse(), PQsslStruct(), PQstatus(), PQtrace(), pqTraceOutputCharResponse(), pqTraceOutputMessage(), pqTraceOutputNoTypeByteMessage(), PQtransactionStatus(), PQtty(), PQuntrace(), PQuser(), pqWait(), pqWaitTimed(), pqWriteReady(), prepareToTerminate(), process_global_sql_commands(), process_result(), ProcessKeepaliveMsg(), processSQLNamePattern(), ProcessXLogDataMsg(), prohibit_crossdb_refs(), prompt_user(), read_server_final_message(), read_server_first_message(), ReceiveArchiveStream(), ReceiveBackupManifest(), ReceiveBackupManifestInMemory(), ReceiveCopyData(), ReceiveTarFile(), ReceiveXlogStream(), reindex_all_databases(), reindex_one_database(), release_conn_addrinfo(), restore_all_databases(), retrieve_objects(), RetrieveDataDirCreatePerm(), RetrieveWalSegSize(), run_permutation(), run_reindex_command(), run_simple_command(), run_simple_query(), run_user_oauth_flow(), run_vacuum_command(), RunIdentifySystem(), scram_exchange(), scram_init(), select_next_encryption_method(), send_cancellable_query_impl(), sendFeedback(), sendTerminateConn(), server_is_in_recovery(), set_archive_cancel_info(), set_frozenxids(), set_replication_progress(), SetCancelConn(), setKeepalivesCount(), setKeepalivesIdle(), setKeepalivesInterval(), setTCPUserTimeout(), setup_connection(), setup_oauth_parameters(), setup_publisher(), setup_recovery(), setup_subscriber(), setup_token_request(), sql_conn(), sql_exec(), sql_exec_dumpalldbs(), sql_exec_dumpalltables(), sql_exec_dumpalltbspc(), sql_exec_searchtables(), ssl_set_pgconn_bio(), start_device_authz(), start_postmaster(), start_token_request(), StartLogStreamer(), store_conn_addrinfo(), storeQueryResult(), StreamLog(), StreamLogicalLog(), stringlist_to_identifierstr(), TableCommandResultHandler(), test_cancel(), test_disallowed_in_pipeline(), test_multi_pipelines(), test_nosync(), test_pipeline_abort(), test_pipeline_idle(), test_pipelined_insert(), test_prepared(), test_protocol_version(), test_simple_pipeline(), test_singlerowmode(), test_transaction(), test_uniqviol(), try_complete_step(), useKeepalives(), vacuum_all_databases(), vacuum_one_database(), vacuumlo(), verify_btree_slot_handler(), verify_heap_slot_handler(), wait_for_end_recovery(), wait_on_slots(), and wait_until_connected().

connection_string

char* connection_string

extern

Definition at line 45 of file streamutil.c.

Referenced by BaseBackup(), ConnectDatabase(), GetConnection(), and main().

dbgetpassword

int dbgetpassword

extern

Definition at line 50 of file streamutil.c.

Referenced by GetConnection(), and main().

dbhost

char* dbhost

extern

Definition at line 46 of file streamutil.c.

Referenced by GetConnection(), and main().

dbname

char* dbname

extern

Definition at line 49 of file streamutil.c.

Referenced by _reconnectToDB(), AlterDatabaseOwner(), appendPsqlMetaConnect(), AutoVacWorkerMain(), BackgroundWorkerInitializeConnection(), check_and_drop_existing_subscriptions(), check_db(), CheckAndGetDbnameFromConninfo(), cluster_all_databases(), concat_conninfo_dbname(), connect_slot(), ConnectDatabase(), conninfo_uri_parse_options(), convert_database_name(), create_database(), createdb(), do_connect(), drop_database_if_exists(), drop_existing_subscriptions(), drop_publication(), dropdb(), dropDBs(), dumpDatabaseConfig(), dumpDatabases(), ECPGconnect(), find_matching_idle_slot(), GenerateRecoveryConfig(), get_base_conninfo(), get_database_oid(), get_dbname_oid_list_from_mfile(), get_publisher_databases(), GetConnection(), GetDatabaseTuple(), GetDbnameFromConnectionOptions(), IdentifySystem(), InitPostgres(), libpqrcv_get_dbname_from_conninfo(), log_status_format(), main(), map_sql_catalog_to_xmlschema_types(), map_sql_schema_to_xmlschema_types(), movedb(), ParallelSlotsGetIdle(), passwordFromFile(), PostgresMain(), PostgresSingleUserMain(), process_pgfdw_appname(), process_postgres_switches(), prohibit_crossdb_refs(), ReconnectToServer(), reindex_all_databases(), ReplSlotSyncWorkerMain(), runPgDump(), and set_replication_progress().

dbport

char* dbport

extern

Definition at line 48 of file streamutil.c.

Referenced by GetConnection(), and main().

dbuser

char* dbuser

extern

Definition at line 47 of file streamutil.c.

Referenced by GetConnection(), and main().

progname

const char* progname

extern

Definition at line 44 of file main.c.

WalSegSz

int WalSegSz

extern

Definition at line 32 of file streamutil.c.