/*-------------------------------------------------------------------------
*
+ * pg_backup_db.c
*
-*-------------------------------------------------------------------------
+ * Implements the basic DB functions used by the archiver.
+ *
+ * IDENTIFICATION
+ * src/bin/pg_dump/pg_backup_db.c
+ *
+ *-------------------------------------------------------------------------
*/
-#include <unistd.h> /* for getopt() */
-#include <ctype.h>
-
-#include "postgres.h"
+#include "pg_backup_db.h"
+#include "dumputils.h"
+#include <unistd.h>
+#include <ctype.h>
#ifdef HAVE_TERMIOS_H
#include <termios.h>
#endif
-#include "access/attnum.h"
-#include "access/htup.h"
-#include "catalog/pg_index.h"
-#include "catalog/pg_language.h"
-#include "catalog/pg_trigger.h"
-#include "catalog/pg_type.h"
-
-#include "libpq-fe.h"
-#include <libpq/libpq-fs.h>
-#ifndef HAVE_STRDUP
-#include "strdup.h"
-#endif
-#include "pg_dump.h"
-#include "pg_backup.h"
-#include "pg_backup_archiver.h"
-#include "pg_backup_db.h"
-
-static const char *progname = "Archiver(db)";
+#define DB_MAX_ERR_STMT 128
-static void _prompt_for_password(char *username, char *password);
-static void _check_database_version(ArchiveHandle *AH, bool ignoreVersion);
-static PGconn* _connectDB(ArchiveHandle *AH, const char* newdbname, char *newUser);
-static int _executeSqlCommand(ArchiveHandle* AH, PGconn *conn, PQExpBuffer qry, char *desc);
+/* translator: this is a module name */
+static const char *modulename = gettext_noop("archiver (db)");
+static void _check_database_version(ArchiveHandle *AH);
+static PGconn *_connectDB(ArchiveHandle *AH, const char *newdbname, const char *newUser);
+static void notice_processor(void *arg, const char *message);
-static void
-_prompt_for_password(char *username, char *password)
+static int
+_parse_version(const char *versionString)
{
- char buf[512];
- int length;
-
-#ifdef HAVE_TERMIOS_H
- struct termios t_orig,
- t;
-#endif
+ int v;
- /*
- * Allow for forcing a specific username
- */
- if (strlen(username) == 0)
- {
- fprintf(stderr, "Username: ");
- fflush(stderr);
- fgets(username, 100, stdin);
- length = strlen(username);
- /* skip rest of the line */
- if (length > 0 && username[length - 1] != '\n')
- {
- do
- {
- fgets(buf, 512, stdin);
- } while (buf[strlen(buf) - 1] != '\n');
- }
- if (length > 0 && username[length - 1] == '\n')
- username[length - 1] = '\0';
- }
+ v = parse_version(versionString);
+ if (v < 0)
+ exit_horribly(modulename, "could not parse version string \"%s\"\n", versionString);
-#ifdef HAVE_TERMIOS_H
- tcgetattr(0, &t);
- t_orig = t;
- t.c_lflag &= ~ECHO;
- tcsetattr(0, TCSADRAIN, &t);
-#endif
- fprintf(stderr, "Password: ");
- fflush(stderr);
- fgets(password, 100, stdin);
-#ifdef HAVE_TERMIOS_H
- tcsetattr(0, TCSADRAIN, &t_orig);
-#endif
-
- length = strlen(password);
- /* skip rest of the line */
- if (length > 0 && password[length - 1] != '\n')
- {
- do
- {
- fgets(buf, 512, stdin);
- } while (buf[strlen(buf) - 1] != '\n');
- }
- if (length > 0 && password[length - 1] == '\n')
- password[length - 1] = '\0';
-
- fprintf(stderr, "\n\n");
+ return v;
}
-
static void
-_check_database_version(ArchiveHandle *AH, bool ignoreVersion)
+_check_database_version(ArchiveHandle *AH)
{
- PGresult *res;
- double myversion;
+ int myversion;
const char *remoteversion_str;
- double remoteversion;
- PGconn *conn = AH->connection;
+ int remoteversion;
- myversion = strtod(PG_VERSION, NULL);
- res = PQexec(conn, "SELECT version()");
- if (!res ||
- PQresultStatus(res) != PGRES_TUPLES_OK ||
- PQntuples(res) != 1)
+ myversion = _parse_version(PG_VERSION);
- die_horribly(AH, "check_database_version(): command failed. "
- "Explanation from backend: '%s'.\n", PQerrorMessage(conn));
+ remoteversion_str = PQparameterStatus(AH->connection, "server_version");
+ if (!remoteversion_str)
+ exit_horribly(modulename, "could not get server_version from libpq\n");
- remoteversion_str = PQgetvalue(res, 0, 0);
- remoteversion = strtod(remoteversion_str + 11, NULL);
- if (myversion != remoteversion)
- {
- fprintf(stderr, "Database version: %s\n%s version: %s\n",
- remoteversion_str, progname, PG_VERSION);
- if (ignoreVersion)
- fprintf(stderr, "Proceeding despite version mismatch.\n");
- else
- die_horribly(AH, "Aborting because of version mismatch.\n"
- "Use --ignore-version if you think it's safe to proceed anyway.\n");
- }
- PQclear(res);
-}
-
-/*
- * Check if a given user is a superuser.
- */
-int UserIsSuperuser(ArchiveHandle *AH, char* user)
-{
- PQExpBuffer qry = createPQExpBuffer();
- PGresult *res;
- int i_usesuper;
- int ntups;
- int isSuper;
+ remoteversion = _parse_version(remoteversion_str);
- /* Get the superuser setting */
- appendPQExpBuffer(qry, "select usesuper from pg_user where usename = '%s'", user);
- res = PQexec(AH->connection, qry->data);
+ AH->public.remoteVersionStr = pg_strdup(remoteversion_str);
+ AH->public.remoteVersion = remoteversion;
+ if (!AH->archiveRemoteVersion)
+ AH->archiveRemoteVersion = AH->public.remoteVersionStr;
- if (!res)
- die_horribly(AH, "%s: null result checking superuser status of %s.\n",
- progname, user);
-
- if (PQresultStatus(res) != PGRES_TUPLES_OK)
- die_horribly(AH, "%s: Could not check superuser status of %s. Explanation from backend: %s\n",
- progname, user, PQerrorMessage(AH->connection));
-
- ntups = PQntuples(res);
-
- if (ntups == 0)
- isSuper = 0;
- else
+ if (myversion != remoteversion
+ && (remoteversion < AH->public.minRemoteVersion ||
+ remoteversion > AH->public.maxRemoteVersion))
{
- i_usesuper = PQfnumber(res, "usesuper");
- isSuper = (strcmp(PQgetvalue(res, 0, i_usesuper), "t") == 0);
+ write_msg(NULL, "server version: %s; %s version: %s\n",
+ remoteversion_str, progname, PG_VERSION);
+ exit_horribly(NULL, "aborting because of server version mismatch\n");
}
- PQclear(res);
-
- return isSuper;
-}
-
-int ConnectedUserIsSuperuser(ArchiveHandle *AH)
-{
- return UserIsSuperuser(AH, PQuser(AH->connection));
-}
-
-char* ConnectedUser(ArchiveHandle *AH)
-{
- return PQuser(AH->connection);
}
/*
- * Reconnect the DB associated with the archive handle
+ * Reconnect to the server. If dbname is not NULL, use that database,
+ * else the one associated with the archive handle. If username is
+ * not NULL, use that user name, else the one from the handle. If
+ * both the database and the user match the existing connection already,
+ * nothing will be done.
+ *
+ * Returns 1 in any case.
*/
-int ReconnectDatabase(ArchiveHandle *AH, const char* newdbname, char *newUser)
+int
+ReconnectToServer(ArchiveHandle *AH, const char *dbname, const char *username)
{
- PGconn *newConn;
- char *dbname;
+ PGconn *newConn;
+ const char *newdbname;
+ const char *newusername;
+
+ if (!dbname)
+ newdbname = PQdb(AH->connection);
+ else
+ newdbname = dbname;
- if (!newdbname || (strcmp(newdbname, "-") == 0) )
- dbname = PQdb(AH->connection);
+ if (!username)
+ newusername = PQuser(AH->connection);
else
- dbname = (char*)newdbname;
+ newusername = username;
/* Let's see if the request is already satisfied */
- if (strcmp(PQuser(AH->connection), newUser) == 0 && strcmp(newdbname, PQdb(AH->connection)) == 0)
+ if (strcmp(newdbname, PQdb(AH->connection)) == 0 &&
+ strcmp(newusername, PQuser(AH->connection)) == 0)
return 1;
- newConn = _connectDB(AH, dbname, newUser);
+ newConn = _connectDB(AH, newdbname, newusername);
PQfinish(AH->connection);
AH->connection = newConn;
- strcpy(AH->username, newUser);
return 1;
}
/*
* Connect to the db again.
+ *
+ * Note: it's not really all that sensible to use a single-entry password
+ * cache if the username keeps changing. In current usage, however, the
+ * username never does change, so one savedPassword is sufficient. We do
+ * update the cache on the off chance that the password has changed since the
+ * start of the run.
*/
-static PGconn* _connectDB(ArchiveHandle *AH, const char* reqdb, char *requser)
+static PGconn *
+_connectDB(ArchiveHandle *AH, const char *reqdb, const char *requser)
{
- int need_pass;
- PGconn *newConn;
- char password[100];
- char *pwparam = NULL;
- int badPwd = 0;
- int noPwd = 0;
- char *newdb;
- char *newuser;
-
- if (!reqdb || (strcmp(reqdb, "-") == 0) )
+ PGconn *newConn;
+ const char *newdb;
+ const char *newuser;
+ char *password = AH->savedPassword;
+ bool new_pass;
+
+ if (!reqdb)
newdb = PQdb(AH->connection);
else
- newdb = (char*)reqdb;
+ newdb = reqdb;
- if (!requser || (strlen(requser) == 0))
+ if (!requser || strlen(requser) == 0)
newuser = PQuser(AH->connection);
else
- newuser = (char*)requser;
+ newuser = requser;
- ahlog(AH, 1, "Connecting to %s as %s\n", newdb, newuser);
+ ahlog(AH, 1, "connecting to database \"%s\" as user \"%s\"\n",
+ newdb, newuser);
+
+ if (AH->promptPassword == TRI_YES && password == NULL)
+ {
+ password = simple_prompt("Password: ", 100, false);
+ if (password == NULL)
+ exit_horribly(modulename, "out of memory\n");
+ }
do
{
- need_pass = false;
- newConn = PQsetdbLogin(PQhost(AH->connection), PQport(AH->connection),
- NULL, NULL, newdb,
- newuser, pwparam);
+#define PARAMS_ARRAY_SIZE 7
+ const char **keywords = pg_malloc(PARAMS_ARRAY_SIZE * sizeof(*keywords));
+ const char **values = pg_malloc(PARAMS_ARRAY_SIZE * sizeof(*values));
+
+ keywords[0] = "host";
+ values[0] = PQhost(AH->connection);
+ keywords[1] = "port";
+ values[1] = PQport(AH->connection);
+ keywords[2] = "user";
+ values[2] = newuser;
+ keywords[3] = "password";
+ values[3] = password;
+ keywords[4] = "dbname";
+ values[4] = newdb;
+ keywords[5] = "fallback_application_name";
+ values[5] = progname;
+ keywords[6] = NULL;
+ values[6] = NULL;
+
+ new_pass = false;
+ newConn = PQconnectdbParams(keywords, values, true);
+
+ free(keywords);
+ free(values);
+
if (!newConn)
- die_horribly(AH, "%s: Failed to reconnect (PQsetdbLogin failed).\n", progname);
+ exit_horribly(modulename, "failed to reconnect to database\n");
if (PQstatus(newConn) == CONNECTION_BAD)
{
- noPwd = (strcmp(PQerrorMessage(newConn), "fe_sendauth: no password supplied\n") == 0);
- badPwd = (strncmp(PQerrorMessage(newConn), "Password authentication failed for user", 39)
- == 0);
+ if (!PQconnectionNeedsPassword(newConn))
+ exit_horribly(modulename, "could not reconnect to database: %s",
+ PQerrorMessage(newConn));
+ PQfinish(newConn);
- if (noPwd || badPwd)
- {
+ if (password)
+ fprintf(stderr, "Password incorrect\n");
- if (badPwd)
- fprintf(stderr, "Password incorrect\n");
+ fprintf(stderr, "Connecting to %s as %s\n",
+ newdb, newuser);
- fprintf(stderr, "Connecting to %s as %s\n", PQdb(AH->connection), newuser);
+ if (password)
+ free(password);
- need_pass = true;
- _prompt_for_password(newuser, password);
- pwparam = password;
- }
+ if (AH->promptPassword != TRI_NO)
+ password = simple_prompt("Password: ", 100, false);
else
- die_horribly(AH, "%s: Could not reconnect. %s\n", progname, PQerrorMessage(newConn));
+ exit_horribly(modulename, "connection needs password\n");
+
+ if (password == NULL)
+ exit_horribly(modulename, "out of memory\n");
+ new_pass = true;
}
+ } while (new_pass);
+
+ AH->savedPassword = password;
- } while (need_pass);
+ /* check for version mismatch */
+ _check_database_version(AH);
+
+ PQsetNoticeProcessor(newConn, notice_processor, NULL);
return newConn;
}
-PGconn* ConnectDatabase(Archive *AHX,
- const char* dbname,
- const char* pghost,
- const char* pgport,
- const int reqPwd,
- const int ignoreVersion)
+/*
+ * Make a database connection with the given parameters. The
+ * connection handle is returned, the parameters are stored in AHX.
+ * An interactive password prompt is automatically issued if required.
+ *
+ * Note: it's not really all that sensible to use a single-entry password
+ * cache if the username keeps changing. In current usage, however, the
+ * username never does change, so one savedPassword is sufficient.
+ */
+void
+ConnectDatabase(Archive *AHX,
+ const char *dbname,
+ const char *pghost,
+ const char *pgport,
+ const char *username,
+ enum trivalue prompt_password)
{
- ArchiveHandle *AH = (ArchiveHandle*)AHX;
- char connect_string[512] = "";
- char tmp_string[128];
- char password[100];
+ ArchiveHandle *AH = (ArchiveHandle *) AHX;
+ char *password = AH->savedPassword;
+ bool new_pass;
if (AH->connection)
- die_horribly(AH, "%s: already connected to database\n", progname);
-
- if (!dbname && !(dbname = getenv("PGDATABASE")) )
- die_horribly(AH, "%s: no database name specified\n", progname);
-
- AH->dbname = strdup(dbname);
+ exit_horribly(modulename, "already connected to a database\n");
- if (pghost != NULL)
+ if (prompt_password == TRI_YES && password == NULL)
{
- AH->pghost = strdup(pghost);
- sprintf(tmp_string, "host=%s ", AH->pghost);
- strcat(connect_string, tmp_string);
+ password = simple_prompt("Password: ", 100, false);
+ if (password == NULL)
+ exit_horribly(modulename, "out of memory\n");
}
- else
- AH->pghost = NULL;
+ AH->promptPassword = prompt_password;
- if (pgport != NULL)
+ /*
+ * Start the connection. Loop until we have a password if requested by
+ * backend.
+ */
+ do
{
- AH->pgport = strdup(pgport);
- sprintf(tmp_string, "port=%s ", AH->pgport);
- strcat(connect_string, tmp_string);
- }
- else
- AH->pgport = NULL;
+#define PARAMS_ARRAY_SIZE 7
+ const char **keywords = pg_malloc(PARAMS_ARRAY_SIZE * sizeof(*keywords));
+ const char **values = pg_malloc(PARAMS_ARRAY_SIZE * sizeof(*values));
+
+ keywords[0] = "host";
+ values[0] = pghost;
+ keywords[1] = "port";
+ values[1] = pgport;
+ keywords[2] = "user";
+ values[2] = username;
+ keywords[3] = "password";
+ values[3] = password;
+ keywords[4] = "dbname";
+ values[4] = dbname;
+ keywords[5] = "fallback_application_name";
+ values[5] = progname;
+ keywords[6] = NULL;
+ values[6] = NULL;
+
+ new_pass = false;
+ AH->connection = PQconnectdbParams(keywords, values, true);
+
+ free(keywords);
+ free(values);
+
+ if (!AH->connection)
+ exit_horribly(modulename, "failed to connect to database\n");
+
+ if (PQstatus(AH->connection) == CONNECTION_BAD &&
+ PQconnectionNeedsPassword(AH->connection) &&
+ password == NULL &&
+ prompt_password != TRI_NO)
+ {
+ PQfinish(AH->connection);
+ password = simple_prompt("Password: ", 100, false);
+ if (password == NULL)
+ exit_horribly(modulename, "out of memory\n");
+ new_pass = true;
+ }
+ } while (new_pass);
- sprintf(tmp_string, "dbname=%s ", AH->dbname);
- strcat(connect_string, tmp_string);
-
- if (reqPwd)
- {
- AH->username[0] = '\0';
- _prompt_for_password(AH->username, password);
- strcat(connect_string, "authtype=password ");
- sprintf(tmp_string, "user=%s ", AH->username);
- strcat(connect_string, tmp_string);
- sprintf(tmp_string, "password=%s ", password);
- strcat(connect_string, tmp_string);
- MemSet(tmp_string, 0, sizeof(tmp_string));
- MemSet(password, 0, sizeof(password));
- }
- AH->connection = PQconnectdb(connect_string);
- MemSet(connect_string, 0, sizeof(connect_string));
+ AH->savedPassword = password;
/* check to see that the backend connection was successfully made */
if (PQstatus(AH->connection) == CONNECTION_BAD)
- die_horribly(AH, "Connection to database '%s' failed.\n%s\n",
- AH->dbname, PQerrorMessage(AH->connection));
+ exit_horribly(modulename, "connection to database \"%s\" failed: %s",
+ PQdb(AH->connection), PQerrorMessage(AH->connection));
/* check for version mismatch */
- _check_database_version(AH, ignoreVersion);
+ _check_database_version(AH);
- /*
- * AH->currUser = PQuser(AH->connection);
- *
- * Removed because it prevented an initial \connect
- * when dumping to SQL in pg_dump.
- */
-
- return AH->connection;
+ PQsetNoticeProcessor(AH->connection, notice_processor, NULL);
}
-/* Public interface */
-/* Convenience function to send a query. Monitors result to handle COPY statements */
-int ExecuteSqlCommand(ArchiveHandle* AH, PQExpBuffer qry, char *desc)
+void
+DisconnectDatabase(Archive *AHX)
{
- return _executeSqlCommand(AH, AH->connection, qry, desc);
+ ArchiveHandle *AH = (ArchiveHandle *) AHX;
+
+ PQfinish(AH->connection); /* noop if AH->connection is NULL */
+ AH->connection = NULL;
}
-/*
- * Handle command execution. This is used to execute a command on more than one connection,
- * but the 'pgCopyIn' setting assumes the COPY commands are ONLY executed on the primary
- * setting...an error will be raised otherwise.
- */
-static int _executeSqlCommand(ArchiveHandle* AH, PGconn *conn, PQExpBuffer qry, char *desc)
+PGconn *
+GetConnection(Archive *AHX)
{
- PGresult *res;
+ ArchiveHandle *AH = (ArchiveHandle *) AHX;
- /* fprintf(stderr, "Executing: '%s'\n\n", qry->data); */
- res = PQexec(conn, qry->data);
- if (!res)
- die_horribly(AH, "%s: %s. No result from backend.\n", progname, desc);
+ return AH->connection;
+}
- if (PQresultStatus(res) != PGRES_COMMAND_OK && PQresultStatus(res) != PGRES_TUPLES_OK)
- {
- if (PQresultStatus(res) == PGRES_COPY_IN)
- {
- if (conn != AH->connection)
- die_horribly(AH, "%s: COPY command execute in non-primary connection.\n", progname);
+static void
+notice_processor(void *arg, const char *message)
+{
+ write_msg(NULL, "%s", message);
+}
- AH->pgCopyIn = 1;
- }
- else
- die_horribly(AH, "%s: %s. Code = %d. Explanation from backend: '%s'.\n",
- progname, desc, PQresultStatus(res), PQerrorMessage(AH->connection));
- }
+/* Like exit_horribly(), but with a complaint about a particular query. */
+static void
+die_on_query_failure(ArchiveHandle *AH, const char *modulename, const char *query)
+{
+ write_msg(modulename, "query failed: %s",
+ PQerrorMessage(AH->connection));
+ exit_horribly(modulename, "query was: %s\n", query);
+}
- PQclear(res);
+void
+ExecuteSqlStatement(Archive *AHX, const char *query)
+{
+ ArchiveHandle *AH = (ArchiveHandle *) AHX;
+ PGresult *res;
- return strlen(qry->data);
+ res = PQexec(AH->connection, query);
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ die_on_query_failure(AH, modulename, query);
+ PQclear(res);
}
-/* Convenience function to send one or more queries. Monitors result to handle COPY statements */
-int ExecuteSqlCommandBuf(ArchiveHandle* AH, void *qryv, int bufLen)
+PGresult *
+ExecuteSqlQuery(Archive *AHX, const char *query, ExecStatusType status)
{
- int loc;
- int pos = 0;
- int sPos = 0;
- char *qry = (char*)qryv;
- int isEnd = 0;
- char *eos = qry + bufLen;
+ ArchiveHandle *AH = (ArchiveHandle *) AHX;
+ PGresult *res;
- /* fprintf(stderr, "\n\n*****\n Buffer:\n\n%s\n*******************\n\n", qry); */
+ res = PQexec(AH->connection, query);
+ if (PQresultStatus(res) != status)
+ die_on_query_failure(AH, modulename, query);
+ return res;
+}
- /* If we're in COPY IN mode, then just break it into lines and send... */
- if (AH->pgCopyIn) {
- for(;;) {
+/*
+ * Convenience function to send a query.
+ * Monitors result to detect COPY statements
+ */
+static void
+ExecuteSqlCommand(ArchiveHandle *AH, const char *qry, const char *desc)
+{
+ PGconn *conn = AH->connection;
+ PGresult *res;
+ char errStmt[DB_MAX_ERR_STMT];
- /* Find a lf */
- loc = strcspn(&qry[pos], "\n") + pos;
- pos = 0;
+#ifdef NOT_USED
+ fprintf(stderr, "Executing: '%s'\n\n", qry);
+#endif
+ res = PQexec(conn, qry);
- /* If no match, then wait */
- if (loc >= (eos - qry)) /* None found */
+ switch (PQresultStatus(res))
+ {
+ case PGRES_COMMAND_OK:
+ case PGRES_TUPLES_OK:
+ case PGRES_EMPTY_QUERY:
+ /* A-OK */
+ break;
+ case PGRES_COPY_IN:
+ /* Assume this is an expected result */
+ AH->pgCopyIn = true;
+ break;
+ default:
+ /* trouble */
+ strncpy(errStmt, qry, DB_MAX_ERR_STMT);
+ if (errStmt[DB_MAX_ERR_STMT - 1] != '\0')
{
- appendBinaryPQExpBuffer(AH->pgCopyBuf, qry, (eos - qry));
- break;
- };
+ errStmt[DB_MAX_ERR_STMT - 4] = '.';
+ errStmt[DB_MAX_ERR_STMT - 3] = '.';
+ errStmt[DB_MAX_ERR_STMT - 2] = '.';
+ errStmt[DB_MAX_ERR_STMT - 1] = '\0';
+ }
+ warn_or_exit_horribly(AH, modulename, "%s: %s Command was: %s\n",
+ desc, PQerrorMessage(conn), errStmt);
+ break;
+ }
- /* fprintf(stderr, "Found cr at %d, prev char was %c, next was %c\n", loc, qry[loc-1], qry[loc+1]); */
-
- /* Count the number of preceding slashes */
- sPos = loc;
- while (sPos > 0 && qry[sPos-1] == '\\')
- sPos--;
+ PQclear(res);
+}
- sPos = loc - sPos;
- /* If an odd number of preceding slashes, then \n was escaped
- * so set the next search pos, and restart (if any left).
- */
- if ((sPos & 1) == 1)
- {
- /* fprintf(stderr, "cr was escaped\n"); */
- pos = loc + 1;
- if (pos >= (eos - qry))
- {
- appendBinaryPQExpBuffer(AH->pgCopyBuf, qry, (eos - qry));
- break;
- }
- }
- else
- {
- /* We got a good cr */
- qry[loc] = '\0';
- appendPQExpBuffer(AH->pgCopyBuf, "%s\n", qry);
- qry += loc + 1;
- isEnd = (strcmp(AH->pgCopyBuf->data, "\\.\n") == 0);
+/*
+ * Process non-COPY table data (that is, INSERT commands).
+ *
+ * The commands have been run together as one long string for compressibility,
+ * and we are receiving them in bufferloads with arbitrary boundaries, so we
+ * have to locate command boundaries and save partial commands across calls.
+ * All state must be kept in AH->sqlparse, not in local variables of this
+ * routine. We assume that AH->sqlparse was filled with zeroes when created.
+ *
+ * We have to lex the data to the extent of identifying literals and quoted
+ * identifiers, so that we can recognize statement-terminating semicolons.
+ * We assume that INSERT data will not contain SQL comments, E'' literals,
+ * or dollar-quoted strings, so this is much simpler than a full SQL lexer.
+ */
+static void
+ExecuteInsertCommands(ArchiveHandle *AH, const char *buf, size_t bufLen)
+{
+ const char *qry = buf;
+ const char *eos = buf + bufLen;
- /* fprintf(stderr, "Sending '%s' via COPY (at end = %d)\n\n", AH->pgCopyBuf->data, isEnd); */
-
- PQputline(AH->connection, AH->pgCopyBuf->data);
+ /* initialize command buffer if first time through */
+ if (AH->sqlparse.curCmd == NULL)
+ AH->sqlparse.curCmd = createPQExpBuffer();
- resetPQExpBuffer(AH->pgCopyBuf);
+ for (; qry < eos; qry++)
+ {
+ char ch = *qry;
- /* fprintf(stderr, "Buffer is '%s'\n", AH->pgCopyBuf->data); */
+ /* For neatness, we skip any newlines between commands */
+ if (!(ch == '\n' && AH->sqlparse.curCmd->len == 0))
+ appendPQExpBufferChar(AH->sqlparse.curCmd, ch);
- if(isEnd) {
- PQendcopy(AH->connection);
- AH->pgCopyIn = 0;
- break;
+ switch (AH->sqlparse.state)
+ {
+ case SQL_SCAN: /* Default state == 0, set in _allocAH */
+ if (ch == ';')
+ {
+ /*
+ * We've found the end of a statement. Send it and reset
+ * the buffer.
+ */
+ ExecuteSqlCommand(AH, AH->sqlparse.curCmd->data,
+ "could not execute query");
+ resetPQExpBuffer(AH->sqlparse.curCmd);
+ }
+ else if (ch == '\'')
+ {
+ AH->sqlparse.state = SQL_IN_SINGLE_QUOTE;
+ AH->sqlparse.backSlash = false;
+ }
+ else if (ch == '"')
+ {
+ AH->sqlparse.state = SQL_IN_DOUBLE_QUOTE;
}
-
- }
-
- /* Make sure we're not past the original buffer end */
- if (qry >= eos)
break;
- }
- }
-
- /* We may have finished Copy In, and have a non-empty buffer */
- if (!AH->pgCopyIn) {
-
- /*
- * The following is a mini state machine to assess then of of an SQL statement.
- * It really only needs to parse good SQL, or at least that's the theory...
- * End-of-statement is assumed to be an unquoted, un commented semi-colon.
- */
-
- /* fprintf(stderr, "Buffer at start is: '%s'\n\n", AH->sqlBuf->data); */
-
- for(pos=0; pos < (eos - qry); pos++)
- {
- appendPQExpBufferChar(AH->sqlBuf, qry[pos]);
- /* fprintf(stderr, " %c",qry[pos]); */
-
- switch (AH->sqlparse.state) {
-
- case SQL_SCAN: /* Default state == 0, set in _allocAH */
-
- if (qry[pos] == ';')
- {
- /* Send It & reset the buffer */
- /* fprintf(stderr, " sending: '%s'\n\n", AH->sqlBuf->data); */
- ExecuteSqlCommand(AH, AH->sqlBuf, "Could not execute query");
- resetPQExpBuffer(AH->sqlBuf);
- AH->sqlparse.lastChar = '\0';
- }
- else
- {
- if (qry[pos] == '"' || qry[pos] == '\'')
- {
- /* fprintf(stderr,"[startquote]\n"); */
- AH->sqlparse.state = SQL_IN_QUOTE;
- AH->sqlparse.quoteChar = qry[pos];
- AH->sqlparse.backSlash = 0;
- }
- else if (qry[pos] == '-' && AH->sqlparse.lastChar == '-')
- {
- AH->sqlparse.state = SQL_IN_SQL_COMMENT;
- }
- else if (qry[pos] == '*' && AH->sqlparse.lastChar == '/')
- {
- AH->sqlparse.state = SQL_IN_EXT_COMMENT;
- }
- AH->sqlparse.lastChar = qry[pos];
- }
-
- break;
-
- case SQL_IN_SQL_COMMENT:
-
- if (qry[pos] == '\n')
- AH->sqlparse.state = SQL_SCAN;
- break;
-
- case SQL_IN_EXT_COMMENT:
-
- if (AH->sqlparse.lastChar == '*' && qry[pos] == '/')
- AH->sqlparse.state = SQL_SCAN;
- break;
-
- case SQL_IN_QUOTE:
-
- if (!AH->sqlparse.backSlash && AH->sqlparse.quoteChar == qry[pos])
- {
- /* fprintf(stderr,"[endquote]\n"); */
- AH->sqlparse.state = SQL_SCAN;
- }
- else
- {
-
- if (qry[pos] == '\\')
- {
- if (AH->sqlparse.lastChar == '\\')
- AH->sqlparse.backSlash = !AH->sqlparse.backSlash;
- else
- AH->sqlparse.backSlash = 1;
- } else {
- AH->sqlparse.backSlash = 0;
- }
- }
- break;
+ case SQL_IN_SINGLE_QUOTE:
+ /* We needn't handle '' specially */
+ if (ch == '\'' && !AH->sqlparse.backSlash)
+ AH->sqlparse.state = SQL_SCAN;
+ else if (ch == '\\' && !AH->public.std_strings)
+ AH->sqlparse.backSlash = !AH->sqlparse.backSlash;
+ else
+ AH->sqlparse.backSlash = false;
+ break;
- }
- AH->sqlparse.lastChar = qry[pos];
- /* fprintf(stderr, "\n"); */
+ case SQL_IN_DOUBLE_QUOTE:
+ /* We needn't handle "" specially */
+ if (ch == '"')
+ AH->sqlparse.state = SQL_SCAN;
+ break;
}
-
}
-
- return 1;
}
-void FixupBlobRefs(ArchiveHandle *AH, char *tablename)
-{
- PQExpBuffer tblQry = createPQExpBuffer();
- PGresult *res, *uRes;
- int i, n;
- char *attr;
-
- for(i=0 ; i < strlen(tablename) ; i++)
- tablename[i] = tolower(tablename[i]);
-
- if (strcmp(tablename, BLOB_XREF_TABLE) == 0)
- return;
- appendPQExpBuffer(tblQry, "SELECT a.attname FROM pg_class c, pg_attribute a, pg_type t "
- " WHERE a.attnum > 0 AND a.attrelid = c.oid AND a.atttypid = t.oid "
- " AND t.typname = 'oid' AND c.relname = '%s';", tablename);
-
- res = PQexec(AH->blobConnection, tblQry->data);
- if (!res)
- die_horribly(AH, "%s: could not find OID attrs of %s. Explanation from backend '%s'\n",
- progname, tablename, PQerrorMessage(AH->connection));
-
- if ((n = PQntuples(res)) == 0) {
- /* We're done */
- ahlog(AH, 1, "No OID attributes in table %s\n", tablename);
- PQclear(res);
- return;
+/*
+ * Implement ahwrite() for direct-to-DB restore
+ */
+int
+ExecuteSqlCommandBuf(ArchiveHandle *AH, const char *buf, size_t bufLen)
+{
+ if (AH->outputKind == OUTPUT_COPYDATA)
+ {
+ /*
+ * COPY data.
+ *
+ * We drop the data on the floor if libpq has failed to enter COPY
+ * mode; this allows us to behave reasonably when trying to continue
+ * after an error in a COPY command.
+ */
+ if (AH->pgCopyIn &&
+ PQputCopyData(AH->connection, buf, bufLen) <= 0)
+ exit_horribly(modulename, "error returned by PQputCopyData: %s",
+ PQerrorMessage(AH->connection));
}
-
- for (i = 0 ; i < n ; i++)
+ else if (AH->outputKind == OUTPUT_OTHERDATA)
{
- attr = PQgetvalue(res, i, 0);
-
- ahlog(AH, 1, " - %s.%s\n", tablename, attr);
-
- resetPQExpBuffer(tblQry);
-
/*
- * We should use coalesce here (rather than 'exists'), but it seems to
- * be broken in 7.0.2 (weird optimizer strategy)
+ * Table data expressed as INSERT commands.
*/
- appendPQExpBuffer(tblQry, "UPDATE \"%s\" SET \"%s\" = ",tablename, attr);
- appendPQExpBuffer(tblQry, " (SELECT x.newOid FROM \"%s\" x WHERE x.oldOid = \"%s\".\"%s\")",
- BLOB_XREF_TABLE, tablename, attr);
- appendPQExpBuffer(tblQry, " where exists"
- "(select * from %s x where x.oldOid = \"%s\".\"%s\");",
- BLOB_XREF_TABLE, tablename, attr);
-
- ahlog(AH, 10, " - sql:\n%s\n", tblQry->data);
-
- uRes = PQexec(AH->blobConnection, tblQry->data);
- if (!uRes)
- die_horribly(AH, "%s: could not update attr %s of table %s. Explanation from backend '%s'\n",
- progname, attr, tablename, PQerrorMessage(AH->blobConnection));
-
- if ( PQresultStatus(uRes) != PGRES_COMMAND_OK )
- die_horribly(AH, "%s: error while updating attr %s of table %s (result = %d)."
- " Explanation from backend '%s'\n",
- progname, attr, tablename, PQresultStatus(uRes),
- PQerrorMessage(AH->blobConnection));
-
- PQclear(uRes);
+ ExecuteInsertCommands(AH, buf, bufLen);
}
+ else
+ {
+ /*
+ * General SQL commands; we assume that commands will not be split
+ * across calls.
+ *
+ * In most cases the data passed to us will be a null-terminated
+ * string, but if it's not, we have to add a trailing null.
+ */
+ if (buf[bufLen] == '\0')
+ ExecuteSqlCommand(AH, buf, "could not execute query");
+ else
+ {
+ char *str = (char *) pg_malloc(bufLen + 1);
- PQclear(res);
+ memcpy(str, buf, bufLen);
+ str[bufLen] = '\0';
+ ExecuteSqlCommand(AH, str, "could not execute query");
+ free(str);
+ }
+ }
+ return 1;
}
-/**********
- * Convenient SQL calls
- **********/
-void CreateBlobXrefTable(ArchiveHandle* AH)
+/*
+ * Terminate a COPY operation during direct-to-DB restore
+ */
+void
+EndDBCopyMode(ArchiveHandle *AH, TocEntry *te)
{
- PQExpBuffer qry = createPQExpBuffer();
-
- /* IF we don't have a BLOB connection, then create one */
- if (!AH->blobConnection)
+ if (AH->pgCopyIn)
{
- AH->blobConnection = _connectDB(AH, NULL, NULL);
- }
-
- ahlog(AH, 1, "Creating table for BLOBS xrefs\n");
+ PGresult *res;
-/*
- appendPQExpBuffer(qry, "Create Temporary Table %s(oldOid oid, newOid oid);", BLOB_XREF_TABLE);
-*/
- appendPQExpBuffer(qry, "Create Table %s(oldOid oid, newOid oid);", BLOB_XREF_TABLE);
+ if (PQputCopyEnd(AH->connection, NULL) <= 0)
+ exit_horribly(modulename, "error returned by PQputCopyEnd: %s",
+ PQerrorMessage(AH->connection));
- _executeSqlCommand(AH, AH->blobConnection, qry, "can not create BLOB xref table '" BLOB_XREF_TABLE "'");
-
- resetPQExpBuffer(qry);
+ /* Check command status and return to normal libpq state */
+ res = PQgetResult(AH->connection);
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ warn_or_exit_horribly(AH, modulename, "COPY failed for table \"%s\": %s",
+ te->tag, PQerrorMessage(AH->connection));
+ PQclear(res);
- appendPQExpBuffer(qry, "Create Unique Index %s_ix on %s(oldOid)", BLOB_XREF_TABLE, BLOB_XREF_TABLE);
- _executeSqlCommand(AH, AH->blobConnection, qry, "can not create index on BLOB xref table '" BLOB_XREF_TABLE "'");
+ AH->pgCopyIn = false;
+ }
}
-void InsertBlobXref(ArchiveHandle* AH, int old, int new)
+void
+StartTransaction(ArchiveHandle *AH)
{
- PQExpBuffer qry = createPQExpBuffer();
-
- appendPQExpBuffer(qry, "Insert Into %s(oldOid, newOid) Values (%d, %d);", BLOB_XREF_TABLE, old, new);
-
- _executeSqlCommand(AH, AH->blobConnection, qry, "can not create BLOB xref entry");
+ ExecuteSqlCommand(AH, "BEGIN", "could not start database transaction");
}
-void StartTransaction(ArchiveHandle* AH)
+void
+CommitTransaction(ArchiveHandle *AH)
{
- PQExpBuffer qry = createPQExpBuffer();
-
- appendPQExpBuffer(qry, "Begin;");
-
- ExecuteSqlCommand(AH, qry, "can not start database transaction");
+ ExecuteSqlCommand(AH, "COMMIT", "could not commit database transaction");
}
-void CommitTransaction(ArchiveHandle* AH)
+void
+DropBlobIfExists(ArchiveHandle *AH, Oid oid)
{
- PQExpBuffer qry = createPQExpBuffer();
-
- appendPQExpBuffer(qry, "Commit;");
-
- ExecuteSqlCommand(AH, qry, "can not commit database transaction");
+ /*
+ * If we are not restoring to a direct database connection, we have to
+ * guess about how to detect whether the blob exists. Assume new-style.
+ */
+ if (AH->connection == NULL ||
+ PQserverVersion(AH->connection) >= 90000)
+ {
+ ahprintf(AH,
+ "SELECT pg_catalog.lo_unlink(oid) "
+ "FROM pg_catalog.pg_largeobject_metadata "
+ "WHERE oid = '%u';\n",
+ oid);
+ }
+ else
+ {
+ /* Restoring to pre-9.0 server, so do it the old way */
+ ahprintf(AH,
+ "SELECT CASE WHEN EXISTS("
+ "SELECT 1 FROM pg_catalog.pg_largeobject WHERE loid = '%u'"
+ ") THEN pg_catalog.lo_unlink('%u') END;\n",
+ oid, oid);
+ }
}
-
-
projects / postgresql / blobdiff