farrokhi/pg_repack
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Fix ownership bug.

Browse Source 
New toast table, toast index, and toast type should not have
been owned by the executor of pg_reorg, but by the original owner.
This commit is contained in:
Takahiro Itagaki 2009-05-14 08:19:25 +00:00
parent 0c659ed31f
commit 9a8f2e9c33
7 changed files with 424 additions and 246 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

114
bin/pg_reorg.c
View File

@ -8,18 +8,16 @@
* @brief Client Modules * @brief Client Modules
*/ */
#define PROGRAM_VERSION "1.0.4" const char *PROGRAM_VERSION = "1.0.4";
#define PROGRAM_URL "http://reorg.projects.postgresql.org/" const char *PROGRAM_URL = "http://reorg.projects.postgresql.org/";
#define PROGRAM_EMAIL "reorg-general@lists.pgfoundry.org" const char *PROGRAM_EMAIL = "reorg-general@lists.pgfoundry.org";
#include "pgut/pgut.h" #include "pgut/pgut.h"
#include "pqexpbuffer.h"
#include <string.h> #include <string.h>
#include <stdlib.h> #include <stdlib.h>
#include <unistd.h> #include <unistd.h>
#define EXITCODE_HELP 2
#define APPLY_COUNT 1000 #define APPLY_COUNT 1000
#define SQL_XID_SNAPSHOT_80300 \ #define SQL_XID_SNAPSHOT_80300 \
@ -37,12 +35,12 @@
" AND pid <> pg_backend_pid() AND transactionid = ANY($1) LIMIT 1" " AND pid <> pg_backend_pid() AND transactionid = ANY($1) LIMIT 1"
#define SQL_XID_SNAPSHOT \ #define SQL_XID_SNAPSHOT \
(PQserverVersion(current_conn) >= 80300 \ (PQserverVersion(connection) >= 80300 \
? SQL_XID_SNAPSHOT_80300 \ ? SQL_XID_SNAPSHOT_80300 \
: SQL_XID_SNAPSHOT_80200) : SQL_XID_SNAPSHOT_80200)
#define SQL_XID_ALIVE \ #define SQL_XID_ALIVE \
(PQserverVersion(current_conn) >= 80300 \ (PQserverVersion(connection) >= 80300 \
? SQL_XID_ALIVE_80300 \ ? SQL_XID_ALIVE_80300 \
: SQL_XID_ALIVE_80200) : SQL_XID_ALIVE_80200)
@ -80,7 +78,7 @@ typedef struct reorg_index
} reorg_index; } reorg_index;
static void reorg_all_databases(const char *orderby); static void reorg_all_databases(const char *orderby);
static pqbool reorg_one_database(const char *orderby, const char *table); static bool reorg_one_database(const char *orderby, const char *table);
static void reorg_one_table(const reorg_table *table, const char *orderby); static void reorg_one_table(const reorg_table *table, const char *orderby);
static char *getstr(PGresult *res, int row, int col); static char *getstr(PGresult *res, int row, int col);
@ -89,14 +87,14 @@ static Oid getoid(PGresult *res, int row, int col);
#define SQLSTATE_INVALID_SCHEMA_NAME "3F000" #define SQLSTATE_INVALID_SCHEMA_NAME "3F000"
#define SQLSTATE_LOCK_NOT_AVAILABLE "55P03" #define SQLSTATE_LOCK_NOT_AVAILABLE "55P03"
static pqbool sqlstate_equals(PGresult *res, const char *state) static bool sqlstate_equals(PGresult *res, const char *state)
{ {
return strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), state) == 0; return strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), state) == 0;
} }
static pqbool echo = false; static bool echo = false;
static pqbool verbose = false; static bool verbose = false;
static pqbool quiet = false; static bool quiet = false;
/* /*
* The table begin re-organized. If not null, we need to cleanup temp * The table begin re-organized. If not null, we need to cleanup temp
@ -125,11 +123,11 @@ const struct option pgut_longopts[] = {
{NULL, 0, NULL, 0} {NULL, 0, NULL, 0}
}; };
pqbool alldb = false; bool alldb = false;
const char *table = NULL; const char *table = NULL;
const char *orderby = NULL; const char *orderby = NULL;
pqbool bool
pgut_argument(int c, const char *arg) pgut_argument(int c, const char *arg)
{ {
switch (c) switch (c)
@ -147,13 +145,13 @@ pgut_argument(int c, const char *arg)
alldb = true; alldb = true;
break; break;
case 't': case 't':
table = arg; assign_option(&table, c, arg);
break; break;
case 'n': case 'n':
orderby = ""; assign_option(&orderby, c, "");
break; break;
case 'o': case 'o':
orderby = arg; assign_option(&orderby, c, arg);
break; break;
default: default:
return false; return false;
@ -164,30 +162,18 @@ pgut_argument(int c, const char *arg)
int int
main(int argc, char *argv[]) main(int argc, char *argv[])
{ {
int exitcode; parse_options(argc, argv);
exitcode = pgut_getopt(argc, argv);
if (exitcode)
return exitcode;
if (alldb) if (alldb)
{ {
if (table) if (table)
{ elog(ERROR, "cannot reorg a specific table in all databases");
fprintf(stderr, "%s: cannot reorg a specific table in all databases\n",
progname);
exit(1);
}
reorg_all_databases(orderby); reorg_all_databases(orderby);
} }
else else
{ {
if (!reorg_one_database(orderby, table)) if (!reorg_one_database(orderby, table))
{ elog(ERROR, "%s is not installed", PROGRAM_NAME);
fprintf(stderr, "ERROR: %s is not installed\n", progname);
return 1;
}
} }
return 0; return 0;
@ -209,13 +195,13 @@ reorg_all_databases(const char *orderby)
for (i = 0; i < PQntuples(result); i++) for (i = 0; i < PQntuples(result); i++)
{ {
pqbool ret; bool ret;
dbname = PQgetvalue(result, i, 0); dbname = PQgetvalue(result, i, 0);
if (!quiet) if (!quiet)
{ {
printf("%s: reorg database \"%s\"", progname, dbname); printf("%s: reorg database \"%s\"", PROGRAM_NAME, dbname);
fflush(stdout); fflush(stdout);
} }
@ -256,10 +242,10 @@ getoid(PGresult *res, int row, int col)
/* /*
* Call reorg_one_table for the target table or each table in a database. * Call reorg_one_table for the target table or each table in a database.
*/ */
static pqbool static bool
reorg_one_database(const char *orderby, const char *table) reorg_one_database(const char *orderby, const char *table)
{ {
pqbool ret = true; bool ret = true;
PGresult *res; PGresult *res;
int i; int i;
int num; int num;
@ -301,7 +287,7 @@ reorg_one_database(const char *orderby, const char *table)
else else
{ {
/* exit otherwise */ /* exit otherwise */
printf("%s", PQerrorMessage(current_conn)); printf("%s", PQerrorMessage(connection));
PQclear(res); PQclear(res);
exit(1); exit(1);
} }
@ -324,10 +310,7 @@ reorg_one_database(const char *orderby, const char *table)
table.ckid = getoid(res, i, c++); table.ckid = getoid(res, i, c++);
if (table.pkid == 0) if (table.pkid == 0)
{ elog(ERROR, "relation \"%s\" has no primary key", table.target_name);
fprintf(stderr, "ERROR: relation \"%s\" has no primary key\n", table.target_name);
exit(1);
}
table.create_pktype = getstr(res, i, c++); table.create_pktype = getstr(res, i, c++);
table.create_log = getstr(res, i, c++); table.create_log = getstr(res, i, c++);
@ -343,10 +326,7 @@ reorg_one_database(const char *orderby, const char *table)
{ {
/* CLUSTER mode */ /* CLUSTER mode */
if (ckey == NULL) if (ckey == NULL)
{ elog(ERROR, "relation \"%s\" has no cluster key", table.target_name);
fprintf(stderr, "ERROR: relation \"%s\" has no cluster key\n", table.target_name);
exit(1);
}
appendPQExpBuffer(&sql, "%s ORDER BY %s", create_table, ckey); appendPQExpBuffer(&sql, "%s ORDER BY %s", create_table, ckey);
table.create_table = sql.data; table.create_table = sql.data;
} }
@ -455,11 +435,7 @@ reorg_one_table(const reorg_table *table, const char *orderby)
" WHERE tgrelid = $1 AND tgname >= 'z_reorg_trigger' LIMIT 1", " WHERE tgrelid = $1 AND tgname >= 'z_reorg_trigger' LIMIT 1",
1, params); 1, params);
if (PQntuples(res) > 0) if (PQntuples(res) > 0)
{ elog(ERROR, "trigger conflicted for %s", table->target_name);
fprintf(stderr, "%s: trigger conflicted for %s\n",
progname, table->target_name);
exit(1);
}
command(table->create_pktype, 0, NULL); command(table->create_pktype, 0, NULL);
command(table->create_log, 0, NULL); command(table->create_log, 0, NULL);
@ -482,7 +458,7 @@ reorg_one_table(const reorg_table *table, const char *orderby)
command("BEGIN ISOLATION LEVEL SERIALIZABLE", 0, NULL); command("BEGIN ISOLATION LEVEL SERIALIZABLE", 0, NULL);
/* SET work_mem = maintenance_work_mem */ /* SET work_mem = maintenance_work_mem */
command("SELECT set_config('work_mem', current_setting('maintenance_work_mem'), true)", 0, NULL); command("SELECT set_config('work_mem', current_setting('maintenance_work_mem'), true)", 0, NULL);
if (PQserverVersion(current_conn) >= 80300 && orderby && !orderby[0]) if (PQserverVersion(connection) >= 80300 && orderby && !orderby[0])
command("SET LOCAL synchronize_seqscans = off", 0, NULL); command("SET LOCAL synchronize_seqscans = off", 0, NULL);
res = execute(SQL_XID_SNAPSHOT, 0, NULL); res = execute(SQL_XID_SNAPSHOT, 0, NULL);
vxid = strdup(PQgetvalue(res, 0, 0)); vxid = strdup(PQgetvalue(res, 0, 0));
@ -578,7 +554,7 @@ reorg_one_table(const reorg_table *table, const char *orderby)
else else
{ {
/* exit otherwise */ /* exit otherwise */
printf("%s", PQerrorMessage(current_conn)); printf("%s", PQerrorMessage(connection));
PQclear(res); PQclear(res);
exit(1); exit(1);
} }
@ -606,7 +582,7 @@ reorg_one_table(const reorg_table *table, const char *orderby)
} }
void void
pgut_cleanup(pqbool fatal) pgut_cleanup(bool fatal)
{ {
if (fatal) if (fatal)
{ {
@ -622,11 +598,11 @@ pgut_cleanup(pqbool fatal)
return; /* no needs to cleanup */ return; /* no needs to cleanup */
/* Rollback current transaction */ /* Rollback current transaction */
if (current_conn) if (connection)
command("ROLLBACK", 0, NULL); command("ROLLBACK", 0, NULL);
/* Try reconnection if not available. */ /* Try reconnection if not available. */
if (PQstatus(current_conn) != CONNECTION_OK) if (PQstatus(connection) != CONNECTION_OK)
reconnect(); reconnect();
/* do cleanup */ /* do cleanup */
@ -636,7 +612,7 @@ pgut_cleanup(pqbool fatal)
} }
} }
int void
pgut_help(void) pgut_help(void)
{ {
fprintf(stderr, fprintf(stderr,
@ -645,33 +621,11 @@ pgut_help(void)
" %s [OPTION]... [DBNAME]\n" " %s [OPTION]... [DBNAME]\n"
"\nOptions:\n" "\nOptions:\n"
" -a, --all reorg all databases\n" " -a, --all reorg all databases\n"
" -d, --dbname=DBNAME database to reorg\n"
" -t, --table=TABLE reorg specific table only\n" " -t, --table=TABLE reorg specific table only\n"
" -n, --no-order do vacuum full instead of cluster\n" " -n, --no-order do vacuum full instead of cluster\n"
" -o, --order-by=columns order by columns instead of cluster keys\n" " -o, --order-by=columns order by columns instead of cluster keys\n"
" -e, --echo show the commands being sent to the server\n" " -e, --echo show the commands being sent to the server\n"
" -q, --quiet don't write any messages\n" " -q, --quiet don't write any messages\n"
" -v, --verbose display detailed information during processing\n" " -v, --verbose display detailed information during processing\n",
" --help show this help, then exit\n" PROGRAM_NAME, PROGRAM_NAME);
" --version output version information, then exit\n"
"\nConnection options:\n"
" -h, --host=HOSTNAME database server host or socket directory\n"
" -p, --port=PORT database server port\n"
" -U, --username=USERNAME user name to connect as\n"
" -W, --password force password prompt\n",
progname, progname);
#ifdef PROGRAM_URL
fprintf(stderr,"\nRead the website for details. <" PROGRAM_URL ">\n");
#endif
#ifdef PROGRAM_EMAIL
fprintf(stderr,"\nReport bugs to <" PROGRAM_EMAIL ">.\n");
#endif
return EXITCODE_HELP;
}
int
pgut_version(void)
{
fprintf(stderr, "%s %s\n", progname, PROGRAM_VERSION);
return EXITCODE_HELP;
} }

246
bin/pgut/pgut.c
View File

@ -1,29 +1,34 @@
/* /*-------------------------------------------------------------------------
*
* pgut.c * pgut.c
* *
* Copyright (c) 2009, NIPPON TELEGRAPH AND TELEPHONE CORPORATION * Copyright (c) 2009, NIPPON TELEGRAPH AND TELEPHONE CORPORATION
*
*-------------------------------------------------------------------------
*/ */
#include "pgut.h"
#include "postgres_fe.h" #include "postgres_fe.h"
#include "libpq/pqsignal.h" #include "libpq/pqsignal.h"
#include <unistd.h> #include <unistd.h>
const char *progname = NULL; #include "pgut.h"
const char *dbname = NULL;
char *host = NULL;
char *port = NULL;
char *username = NULL;
pqbool password = false;
/* Interrupted by SIGINT (Ctrl+C) ? */ const char *PROGRAM_NAME = NULL;
pqbool interrupted = false;
const char *dbname = NULL;
const char *host = NULL;
const char *port = NULL;
const char *username = NULL;
bool password = false;
/* Database connections */ /* Database connections */
PGconn *current_conn = NULL; PGconn *connection = NULL;
static PGcancel *volatile cancel_conn = NULL; static PGcancel *volatile cancel_conn = NULL;
/* Interrupted by SIGINT (Ctrl+C) ? */
static bool interrupted = false;
/* Connection routines */ /* Connection routines */
static void init_cancel_handler(void); static void init_cancel_handler(void);
static void on_before_exec(PGconn *conn); static void on_before_exec(PGconn *conn);
@ -31,7 +36,8 @@ static void on_after_exec(void);
static void on_interrupt(void); static void on_interrupt(void);
static void on_cleanup(void); static void on_cleanup(void);
static void exit_or_abort(int exitcode); static void exit_or_abort(int exitcode);
const char *get_user_name(const char *progname); static void help(void);
static const char *get_user_name(const char *PROGRAM_NAME);
const char default_optstring[] = "d:h:p:U:W"; const char default_optstring[] = "d:h:p:U:W";
@ -45,6 +51,9 @@ const struct option default_longopts[] =
{NULL, 0, NULL, 0} {NULL, 0, NULL, 0}
}; };
static const char *optstring = NULL;
static const struct option *longopts = NULL;;
static const char * static const char *
merge_optstring(const char *opts) merge_optstring(const char *opts)
{ {
@ -83,29 +92,31 @@ merge_longopts(const struct option *opts)
return result; return result;
} }
int void
pgut_getopt(int argc, char **argv) parse_options(int argc, char **argv)
{ {
const char *optstring;
const struct option *longopts;
int c; int c;
int optindex = 0; int optindex = 0;
progname = get_progname(argv[0]); PROGRAM_NAME = get_progname(argv[0]);
set_pglocale_pgservice(argv[0], "pgscripts"); set_pglocale_pgservice(argv[0], "pgscripts");
/* /* Help message and version are handled at first. */
* Help message and version are handled at first.
*/
if (argc > 1) if (argc > 1)
{ {
if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0) if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
return pgut_help(); {
help();
exit_or_abort(HELP);
}
if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-V") == 0) if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-V") == 0)
return pgut_version(); {
fprintf(stderr, "%s %s\n", PROGRAM_NAME, PROGRAM_VERSION);
exit_or_abort(HELP);
}
} }
/* Merge default and user options. */
optstring = merge_optstring(pgut_optstring); optstring = merge_optstring(pgut_optstring);
longopts = merge_longopts(pgut_longopts); longopts = merge_longopts(pgut_longopts);
@ -113,26 +124,26 @@ pgut_getopt(int argc, char **argv)
{ {
switch (c) switch (c)
{ {
case 'd':
assign_option(&dbname, c, optarg);
break;
case 'h': case 'h':
host = optarg; assign_option(&host, c, optarg);
break; break;
case 'p': case 'p':
port = optarg; assign_option(&port, c, optarg);
break; break;
case 'U': case 'U':
username = optarg; assign_option(&username, c, optarg);
break; break;
case 'W': case 'W':
password = true; password = true;
break; break;
case 'd':
dbname = optarg;
break;
default: default:
if (!pgut_argument(c, optarg)) if (!pgut_argument(c, optarg))
{ {
fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname); fprintf(stderr, "Try \"%s --help\" for more information.\n", PROGRAM_NAME);
exit_or_abort(EXITCODE_ERROR); exit_or_abort(ERROR);
} }
break; break;
} }
@ -142,10 +153,10 @@ pgut_getopt(int argc, char **argv)
{ {
if (!pgut_argument(0, argv[optind])) if (!pgut_argument(0, argv[optind]))
{ {
fprintf(stderr, _("%s: too many command-line arguments (first is \"%s\")\n"), fprintf(stderr, "%s: too many command-line arguments (first is \"%s\")\n",
progname, argv[optind]); PROGRAM_NAME, argv[optind]);
fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname); fprintf(stderr, "Try \"%s --help\" for more information.\n", PROGRAM_NAME);
exit_or_abort(EXITCODE_ERROR); exit_or_abort(ERROR);
} }
} }
@ -155,9 +166,28 @@ pgut_getopt(int argc, char **argv)
(void) (dbname || (void) (dbname ||
(dbname = getenv("PGDATABASE")) || (dbname = getenv("PGDATABASE")) ||
(dbname = getenv("PGUSER")) || (dbname = getenv("PGUSER")) ||
(dbname = get_user_name(progname))); (dbname = get_user_name(PROGRAM_NAME)));
}
return 0; bool
assign_option(const char **value, int c, const char *arg)
{
if (*value != NULL)
{
const struct option *opt;
for (opt = longopts; opt->name; opt++)
{
if (opt->val == c)
break;
}
if (opt->name)
elog(ERROR, "option -%c(--%s) should be specified only once", c, opt->name);
else
elog(ERROR, "option -%c should be specified only once", c);
return false;
}
*value = arg;
return true;
} }
void void
@ -182,11 +212,7 @@ reconnect(void)
conn = PQsetdbLogin(host, port, NULL, NULL, dbname, username, pwd); conn = PQsetdbLogin(host, port, NULL, NULL, dbname, username, pwd);
if (!conn) if (!conn)
{ elog(ERROR, "could not connect to database %s", dbname);
fprintf(stderr, _("%s: could not connect to database %s\n"),
progname, dbname);
exit_or_abort(EXITCODE_ERROR);
}
if (PQstatus(conn) == CONNECTION_BAD && if (PQstatus(conn) == CONNECTION_BAD &&
#if PG_VERSION_NUM >= 80300 #if PG_VERSION_NUM >= 80300
@ -207,22 +233,19 @@ reconnect(void)
/* check to see that the backend connection was successfully made */ /* check to see that the backend connection was successfully made */
if (PQstatus(conn) == CONNECTION_BAD) if (PQstatus(conn) == CONNECTION_BAD)
{ elog(ERROR, "could not connect to database %s: %s",
fprintf(stderr, _("%s: could not connect to database %s: %s"), dbname, PQerrorMessage(conn));
progname, dbname, PQerrorMessage(conn));
exit_or_abort(EXITCODE_ERROR);
}
current_conn = conn; connection = conn;
} }
void void
disconnect(void) disconnect(void)
{ {
if (current_conn) if (connection)
{ {
PQfinish(current_conn); PQfinish(connection);
current_conn = NULL; connection = NULL;
} }
} }
@ -231,11 +254,17 @@ execute_nothrow(const char *query, int nParams, const char **params)
{ {
PGresult *res; PGresult *res;
on_before_exec(current_conn); if (interrupted)
{
interrupted = false;
elog(ERROR, "%s: interrupted", PROGRAM_NAME);
}
on_before_exec(connection);
if (nParams == 0) if (nParams == 0)
res = PQexec(current_conn, query); res = PQexec(connection, query);
else else
res = PQexecParams(current_conn, query, nParams, NULL, params, NULL, NULL, 0); res = PQexecParams(connection, query, nParams, NULL, params, NULL, NULL, 0);
on_after_exec(); on_after_exec();
return res; return res;
@ -247,27 +276,16 @@ execute_nothrow(const char *query, int nParams, const char **params)
PGresult * PGresult *
execute(const char *query, int nParams, const char **params) execute(const char *query, int nParams, const char **params)
{ {
if (interrupted) PGresult *res = execute_nothrow(query, nParams, params);
{
interrupted = false;
fprintf(stderr, _("%s: interrupted\n"), progname);
}
else
{
PGresult *res = execute_nothrow(query, nParams, params);
if (PQresultStatus(res) == PGRES_TUPLES_OK || if (PQresultStatus(res) == PGRES_TUPLES_OK ||
PQresultStatus(res) == PGRES_COMMAND_OK) PQresultStatus(res) == PGRES_COMMAND_OK)
return res; return res;
fprintf(stderr, _("%s: query failed: %s"), fprintf(stderr, "%s: query failed: %s", PROGRAM_NAME, PQerrorMessage(connection));
progname, PQerrorMessage(current_conn)); fprintf(stderr, "%s: query was: %s\n", PROGRAM_NAME, query);
fprintf(stderr, _("%s: query was: %s\n"), PQclear(res);
progname, query); exit_or_abort(ERROR);
PQclear(res);
}
exit_or_abort(EXITCODE_ERROR);
return NULL; /* keep compiler quiet */ return NULL; /* keep compiler quiet */
} }
@ -281,6 +299,48 @@ command(const char *query, int nParams, const char **params)
PQclear(res); PQclear(res);
} }
/*
* elog - log to stderr and exit if ERROR or FATAL
*/
void
elog(int elevel, const char *fmt, ...)
{
va_list args;
switch (elevel)
{
case LOG:
fputs("LOG: ", stderr);
break;
case INFO:
fputs("INFO: ", stderr);
break;
case NOTICE:
fputs("NOTICE: ", stderr);
break;
case WARNING:
fputs("WARNING: ", stderr);
break;
case ERROR:
fputs("ERROR: ", stderr);
break;
case FATAL:
fputs("FATAL: ", stderr);
break;
case PANIC:
fputs("PANIC: ", stderr);
break;
}
va_start(args, fmt);
vfprintf(stderr, fmt, args);
va_end(args);
fputc('\n', stderr);
fflush(stderr);
if (elevel > 0)
exit_or_abort(elevel);
}
#ifdef WIN32 #ifdef WIN32
static CRITICAL_SECTION cancelConnLock; static CRITICAL_SECTION cancelConnLock;
@ -357,12 +417,12 @@ on_interrupt(void)
/* Send QueryCancel if we are processing a database query */ /* Send QueryCancel if we are processing a database query */
if (cancel_conn != NULL && PQcancel(cancel_conn, errbuf, sizeof(errbuf))) if (cancel_conn != NULL && PQcancel(cancel_conn, errbuf, sizeof(errbuf)))
fprintf(stderr, _("Cancel request sent\n")); fprintf(stderr, "Cancel request sent\n");
errno = save_errno; /* just in case the write changed it */ errno = save_errno; /* just in case the write changed it */
} }
static pqbool in_cleanup = false; static bool in_cleanup = false;
static void static void
on_cleanup(void) on_cleanup(void)
@ -388,33 +448,45 @@ exit_or_abort(int exitcode)
} }
} }
static void help(void)
{
pgut_help();
fprintf(stderr, "\nConnection options:\n");
fprintf(stderr, " -d, --dbname=DBNAME database to connect\n");
fprintf(stderr, " -h, --host=HOSTNAME database server host or socket directory\n");
fprintf(stderr, " -p, --port=PORT database server port\n");
fprintf(stderr, " -U, --username=USERNAME user name to connect as\n");
fprintf(stderr, " -W, --password force password prompt\n");
fprintf(stderr, "\nGeneric Options:\n");
fprintf(stderr, " --help show this help, then exit\n");
fprintf(stderr, " --version output version information, then exit\n\n");
if (PROGRAM_URL)
fprintf(stderr, "Read the website for details. <%s>\n", PROGRAM_URL);
if (PROGRAM_EMAIL)
fprintf(stderr, "Report bugs to <%s>.\n", PROGRAM_EMAIL);
}
/* /*
* Returns the current user name. * Returns the current user name.
*/ */
const char * static const char *
get_user_name(const char *progname) get_user_name(const char *PROGRAM_NAME)
{ {
#ifndef WIN32 #ifndef WIN32
struct passwd *pw; struct passwd *pw;
pw = getpwuid(geteuid()); pw = getpwuid(geteuid());
if (!pw) if (!pw)
{ elog(ERROR, "%s: could not obtain information about current user: %s",
fprintf(stderr, _("%s: could not obtain information about current user: %s\n"), PROGRAM_NAME, strerror(errno));
progname, strerror(errno));
exit_or_abort(EXITCODE_ERROR);
}
return pw->pw_name; return pw->pw_name;
#else #else
static char username[128]; /* remains after function exit */ static char username[128]; /* remains after function exit */
DWORD len = sizeof(username) - 1; DWORD len = sizeof(username) - 1;
if (!GetUserName(username, &len)) if (!GetUserName(username, &len))
{ elog(ERROR, "%s: could not get current user name: %s",
fprintf(stderr, _("%s: could not get current user name: %s\n"), PROGRAM_NAME, strerror(errno));
progname, strerror(errno));
exit_or_abort(EXITCODE_ERROR);
}
return username; return username;
#endif #endif
} }

97
bin/pgut/pgut.h
View File

@ -1,56 +1,61 @@
/* /*-------------------------------------------------------------------------
*
* pgut.h * pgut.h
* *
* Copyright (c) 2009, NIPPON TELEGRAPH AND TELEPHONE CORPORATION * Copyright (c) 2009, NIPPON TELEGRAPH AND TELEPHONE CORPORATION
*
*-------------------------------------------------------------------------
*/ */
#ifndef PGUT_H #ifndef PGUT_H
#define PGUT_H #define PGUT_H
#include "libpq-fe.h" #include "libpq-fe.h"
#include "pqexpbuffer.h"
#include <getopt.h> #include <getopt.h>
#if !defined(C_H) && !defined(__cplusplus)
#ifndef bool
typedef char bool;
#endif
#ifndef true
#define true ((bool) 1)
#endif
#ifndef false
#define false ((bool) 0)
#endif
#endif
/* /*
* pgut client variables and functions * pgut client variables and functions
*/ */
extern const char *pgut_optstring; extern const char *pgut_optstring;
extern const struct option pgut_longopts[]; extern const struct option pgut_longopts[];
extern pqbool pgut_argument(int c, const char *arg); extern bool pgut_argument(int c, const char *arg);
extern int pgut_help(void); extern void pgut_help(void);
extern int pgut_version(void); extern void pgut_cleanup(bool fatal);
extern void pgut_cleanup(pqbool fatal);
/*
* exit codes
*/
#define EXITCODE_OK 0 /**< normal exit */
#define EXITCODE_ERROR 1 /**< normal error */
#define EXITCODE_HELP 2 /**< help and version mode */
#define EXITCODE_FATAL 3 /**< fatal error */
/* /*
* pgut framework variables and functions * pgut framework variables and functions
*/ */
#ifndef true extern const char *PROGRAM_NAME;
#define true 1 extern const char *PROGRAM_VERSION;
#endif extern const char *PROGRAM_URL;
#ifndef false extern const char *PROGRAM_EMAIL;
#define false 0
#endif
extern const char *progname;
extern const char *dbname; extern const char *dbname;
extern char *host; extern const char *host;
extern char *port; extern const char *port;
extern char *username; extern const char *username;
extern pqbool password; extern bool password;
extern pqbool interrupted;
extern PGconn *current_conn;
extern int pgut_getopt(int argc, char **argv); extern PGconn *connection;
extern void parse_options(int argc, char **argv);
extern bool assign_option(const char **value, int c, const char *arg);
extern void reconnect(void); extern void reconnect(void);
extern void disconnect(void); extern void disconnect(void);
@ -62,4 +67,40 @@ extern void command(const char *query, int nParams, const char **params);
extern unsigned int sleep(unsigned int seconds); extern unsigned int sleep(unsigned int seconds);
#endif #endif
/*
* elog
*/
#define LOG (-4)
#define INFO (-3)
#define NOTICE (-2)
#define WARNING (-1)
#define ERROR 1
#define HELP 2
#define FATAL 3
#define PANIC 4
#undef elog
extern void
elog(int elevel, const char *fmt, ...)
__attribute__((format(printf, 2, 3)));
/*
* StringInfo
*/
#define StringInfoData PQExpBufferData
#define StringInfo PQExpBuffer
#define makeStringInfo createPQExpBuffer
#define initStringInfo initPQExpBuffer
#define freeStringInfo destroyPQExpBuffer
#define termStringInfo termPQExpBuffer
#define resetStringInfo resetPQExpBuffer
#define enlargeStringInfo enlargePQExpBuffer
/*
#define printfPQExpBuffer = resetStringInfo + appendStringInfo
*/
#define appendStringInfo appendPQExpBuffer
#define appendStringInfoString appendPQExpBufferStr
#define appendStringInfoChar appendPQExpBufferChar
#define appendBinaryStringInfo appendBinaryPQExpBuffer
#endif /* PGUT_H */ #endif /* PGUT_H */

2
lib/Makefile
View File

@ -3,7 +3,7 @@
# #
# Copyright (c) 2008-2009, NIPPON TELEGRAPH AND TELEPHONE CORPORATION # Copyright (c) 2008-2009, NIPPON TELEGRAPH AND TELEPHONE CORPORATION
# #
SRCS = reorg.c SRCS = reorg.c pgut/pgut-be.c
OBJS = $(SRCS:.c=.o) OBJS = $(SRCS:.c=.o)
MODULE_big = pg_reorg MODULE_big = pg_reorg
DATA_built = pg_reorg.sql DATA_built = pg_reorg.sql

61
lib/pgut/pgut-be.c Executable file
View File

@ -0,0 +1,61 @@
/*-------------------------------------------------------------------------
*
* pgut-be.c
*
* Copyright (c) 2009, NIPPON TELEGRAPH AND TELEPHONE CORPORATION
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "pgut-be.h"
#if PG_VERSION_NUM < 80400
char *
text_to_cstring(const text *t)
{
text *tunpacked = pg_detoast_datum_packed((struct varlena *) t);
int len = VARSIZE_ANY_EXHDR(tunpacked);
char *result;
result = (char *) palloc(len + 1);
memcpy(result, VARDATA_ANY(tunpacked), len);
result[len] = '\0';
if (tunpacked != t)
pfree(tunpacked);
return result;
}
text *
cstring_to_text(const char *s)
{
int len = strlen(s);
text *result = palloc(len + VARHDRSZ);
SET_VARSIZE(result, len + VARHDRSZ);
memcpy(VARDATA(result), s, len);
return result;
}
int
SPI_execute_with_args(const char *src,
int nargs, Oid *argtypes,
Datum *values, const char *nulls,
bool read_only, long tcount)
{
SPIPlanPtr plan;
int ret;
plan = SPI_prepare(src, nargs, argtypes);
if (plan == NULL)
return SPI_result;
ret = SPI_execute_plan(plan, values, nulls, read_only, tcount);
SPI_freeplan(plan);
return ret;
}
#endif

72
lib/pgut/pgut-be.h Executable file
View File

@ -0,0 +1,72 @@
/*-------------------------------------------------------------------------
*
* pgut-be.h
*
* Copyright (c) 2009, NIPPON TELEGRAPH AND TELEPHONE CORPORATION
*
*-------------------------------------------------------------------------
*/
#ifndef PGUT_BE_H
#define PGUT_BE_H
#include "executor/spi.h"
#if PG_VERSION_NUM < 80300
#define PGDLLIMPORT DLLIMPORT
#define SK_BT_DESC 0 /* Always ASC */
#define SK_BT_NULLS_FIRST 0 /* Always NULLS LAST */
#define MaxHeapTupleSize MaxTupleSize
#define PG_GETARG_TEXT_PP(n) PG_GETARG_TEXT_P(n)
#define VARSIZE_ANY_EXHDR(v) (VARSIZE(v) - VARHDRSZ)
#define VARDATA_ANY(v) VARDATA(v)
#define SET_VARSIZE(v, sz) (VARATT_SIZEP(v) = (sz))
#define pg_detoast_datum_packed(v) pg_detoast_datum(v)
#define DatumGetTextPP(v) DatumGetTextP(v)
#define ItemIdIsNormal(v) ItemIdIsUsed(v)
#define IndexBuildHeapScan(heap, index, info, sync, callback, state) \
IndexBuildHeapScan((heap), (index), (info), (callback), (state))
#define planner_rt_fetch(rti, root) \
rt_fetch(rti, (root)->parse->rtable)
#define heap_sync(rel) ((void)0)
#define ItemIdIsDead(itemId) ItemIdDeleted(itemId)
#define GetCurrentCommandId(used) GetCurrentCommandId()
#define stringToQualifiedNameList(str) \
stringToQualifiedNameList((str), "pg_bulkload")
#define setNewRelfilenode(rel, xid) \
setNewRelfilenode((rel))
#define PageAddItem(page, item, size, offnum, overwrite, is_heap) \
PageAddItem((page), (item), (size), (offnum), LP_USED)
typedef void *SPIPlanPtr;
#endif
#if PG_VERSION_NUM < 80400
#define MAIN_FORKNUM 0
#define HEAP_INSERT_SKIP_WAL 0x0001
#define HEAP_INSERT_SKIP_FSM 0x0002
#define relpath(rnode, forknum) relpath((rnode))
#define smgrimmedsync(reln, forknum) smgrimmedsync((reln))
#define smgrread(reln, forknum, blocknum, buffer) \
smgrread((reln), (blocknum), (buffer))
#define mdclose(reln, forknum) mdclose((reln))
#define heap_insert(relation, tup, cid, options, bistate) \
heap_insert((relation), (tup), (cid), true, true)
#define GetBulkInsertState() (NULL)
#define FreeBulkInsertState(bistate) ((void)0)
typedef void *BulkInsertState;
extern char *text_to_cstring(const text *t);
extern text *cstring_to_text(const char *s);
extern int SPI_execute_with_args(const char *src, int nargs, Oid *argtypes,
Datum *values, const char *nulls, bool read_only, long tcount);
#endif
#endif /* PGUT_BE_H */

78
lib/reorg.c
View File

@ -21,20 +21,15 @@
#include "catalog/pg_type.h" #include "catalog/pg_type.h"
#include "commands/tablecmds.h" #include "commands/tablecmds.h"
#include "commands/trigger.h" #include "commands/trigger.h"
#include "executor/spi.h"
#include "miscadmin.h" #include "miscadmin.h"
#include "utils/builtins.h" #include "utils/builtins.h"
#include "utils/lsyscache.h" #include "utils/lsyscache.h"
#include "utils/relcache.h" #include "utils/relcache.h"
#include "utils/syscache.h" #include "utils/syscache.h"
PG_MODULE_MAGIC; #include "pgut/pgut-be.h"
#if PG_VERSION_NUM < 80300 PG_MODULE_MAGIC;
#define SET_VARSIZE(PTR, len) (VARATT_SIZEP((PTR)) = (len))
#define PGDLLIMPORT DLLIMPORT
typedef void *SPIPlanPtr;
#endif
Datum reorg_trigger(PG_FUNCTION_ARGS); Datum reorg_trigger(PG_FUNCTION_ARGS);
Datum reorg_apply(PG_FUNCTION_ARGS); Datum reorg_apply(PG_FUNCTION_ARGS);
@ -72,10 +67,6 @@ must_be_superuser(const char *func)
} }
#if PG_VERSION_NUM < 80400 #if PG_VERSION_NUM < 80400
static int SPI_execute_with_args(const char *src, int nargs, Oid *argtypes,
Datum *values, const char *nulls,
bool read_only, long tcount);
static text *cstring_to_text(const char * s);
static void RenameRelationInternal(Oid myrelid, const char *newrelname, Oid namespaceId); static void RenameRelationInternal(Oid myrelid, const char *newrelname, Oid namespaceId);
#endif #endif
@ -205,7 +196,7 @@ reorg_apply(PG_FUNCTION_ARGS)
values_peek[0] = Int32GetDatum(NUMBER_OF_PROCESSING); values_peek[0] = Int32GetDatum(NUMBER_OF_PROCESSING);
else else
values_peek[0] = Int32GetDatum(Min(count - n, NUMBER_OF_PROCESSING)); values_peek[0] = Int32GetDatum(Min(count - n, NUMBER_OF_PROCESSING));
reorg_execp(plan_peek, values_peek, nulls_peek, SPI_OK_SELECT); reorg_execp(plan_peek, values_peek, nulls_peek, SPI_OK_SELECT);
if (SPI_processed <= 0) if (SPI_processed <= 0)
break; break;
@ -222,7 +213,7 @@ reorg_apply(PG_FUNCTION_ARGS)
{ {
HeapTuple tuple; HeapTuple tuple;
bool isnull; bool isnull;
tuple = tuptable->vals[i]; tuple = tuptable->vals[i];
values[0] = SPI_getbinval(tuple, desc, 1, &isnull); values[0] = SPI_getbinval(tuple, desc, 1, &isnull);
nulls[0] = ' '; nulls[0] = ' ';
@ -230,7 +221,7 @@ reorg_apply(PG_FUNCTION_ARGS)
nulls[1] = (isnull ? 'n' : ' '); nulls[1] = (isnull ? 'n' : ' ');
values[2] = SPI_getbinval(tuple, desc, 3, &isnull); values[2] = SPI_getbinval(tuple, desc, 3, &isnull);
nulls[2] = (isnull ? 'n' : ' '); nulls[2] = (isnull ? 'n' : ' ');
if (nulls[1] == 'n') if (nulls[1] == 'n')
{ {
/* INSERT */ /* INSERT */
@ -500,6 +491,8 @@ reorg_swap(PG_FUNCTION_ARGS)
Oid oid2; Oid oid2;
Oid reltoastrelid2; Oid reltoastrelid2;
Oid reltoastidxid2; Oid reltoastidxid2;
Oid owner1;
Oid owner2;
/* authority check */ /* authority check */
must_be_superuser("reorg_swap"); must_be_superuser("reorg_swap");
@ -510,10 +503,12 @@ reorg_swap(PG_FUNCTION_ARGS)
/* swap relfilenode and dependencies for tables. */ /* swap relfilenode and dependencies for tables. */
values[0] = ObjectIdGetDatum(oid); values[0] = ObjectIdGetDatum(oid);
reorg_execd( reorg_execd(
"SELECT X.oid, X.reltoastrelid, TX.reltoastidxid," "SELECT X.reltoastrelid, TX.reltoastidxid, X.relowner,"
" Y.oid, Y.reltoastrelid, TY.reltoastidxid" " Y.oid, Y.reltoastrelid, TY.reltoastidxid, Y.relowner"
" FROM pg_class X LEFT JOIN pg_class TX ON X.reltoastrelid = TX.oid," " FROM pg_catalog.pg_class X LEFT JOIN pg_catalog.pg_class TX"
" pg_class Y LEFT JOIN pg_class TY ON Y.reltoastrelid = TY.oid" " ON X.reltoastrelid = TX.oid,"
" pg_catalog.pg_class Y LEFT JOIN pg_catalog.pg_class TY"
" ON Y.reltoastrelid = TY.oid"
" WHERE X.oid = $1" " WHERE X.oid = $1"
" AND Y.oid = ('reorg.table_' || X.oid)::regclass", " AND Y.oid = ('reorg.table_' || X.oid)::regclass",
1, argtypes, values, nulls, SPI_OK_SELECT); 1, argtypes, values, nulls, SPI_OK_SELECT);
@ -527,11 +522,13 @@ reorg_swap(PG_FUNCTION_ARGS)
tuple = tuptable->vals[0]; tuple = tuptable->vals[0];
reltoastrelid1 = getoid(tuple, desc, 2); reltoastrelid1 = getoid(tuple, desc, 1);
reltoastidxid1 = getoid(tuple, desc, 3); reltoastidxid1 = getoid(tuple, desc, 2);
owner1 = getoid(tuple, desc, 3);
oid2 = getoid(tuple, desc, 4); oid2 = getoid(tuple, desc, 4);
reltoastrelid2 = getoid(tuple, desc, 5); reltoastrelid2 = getoid(tuple, desc, 5);
reltoastidxid2 = getoid(tuple, desc, 6); reltoastidxid2 = getoid(tuple, desc, 6);
owner2 = getoid(tuple, desc, 7);
/* should be all-or-nothing */ /* should be all-or-nothing */
if ((reltoastrelid1 == InvalidOid || reltoastidxid1 == InvalidOid || if ((reltoastrelid1 == InvalidOid || reltoastidxid1 == InvalidOid ||
@ -543,6 +540,14 @@ reorg_swap(PG_FUNCTION_ARGS)
reltoastrelid1, reltoastidxid1, reltoastrelid2, reltoastidxid2); reltoastrelid1, reltoastidxid1, reltoastrelid2, reltoastidxid2);
} }
/* change owner of new relation to original owner */
if (owner1 != owner2)
{
ATExecChangeOwner(oid2, owner1, true);
CommandCounterIncrement();
}
/* swap heap and index files */
swap_heap_or_index_files(oid, oid2); swap_heap_or_index_files(oid, oid2);
CommandCounterIncrement(); CommandCounterIncrement();
@ -550,7 +555,9 @@ reorg_swap(PG_FUNCTION_ARGS)
values[0] = ObjectIdGetDatum(oid); values[0] = ObjectIdGetDatum(oid);
reorg_execd( reorg_execd(
"SELECT X.oid, Y.oid" "SELECT X.oid, Y.oid"
" FROM pg_index I, pg_class X, pg_class Y" " FROM pg_catalog.pg_index I,"
" pg_catalog.pg_class X,"
" pg_catalog.pg_class Y"
" WHERE I.indrelid = $1" " WHERE I.indrelid = $1"
" AND I.indexrelid = X.oid" " AND I.indexrelid = X.oid"
" AND Y.oid = ('reorg.index_' || X.oid)::regclass", " AND Y.oid = ('reorg.index_' || X.oid)::regclass",
@ -897,35 +904,6 @@ swap_heap_or_index_files(Oid r1, Oid r2)
extern PGDLLIMPORT bool allowSystemTableMods; extern PGDLLIMPORT bool allowSystemTableMods;
static int
SPI_execute_with_args(const char *src,
int nargs, Oid *argtypes,
Datum *values, const char *nulls,
bool read_only, long tcount)
{
SPIPlanPtr plan;
int ret;
plan = SPI_prepare(src, nargs, argtypes);
if (plan == NULL)
return SPI_result;
ret = SPI_execute_plan(plan, values, nulls, read_only, tcount);
SPI_freeplan(plan);
return ret;
}
static text *
cstring_to_text(const char * s)
{
int len = strlen(s);
text *result = palloc(len + VARHDRSZ);
SET_VARSIZE(result, len + VARHDRSZ);
memcpy(VARDATA(result), s, len);
return result;
}
static void static void
RenameRelationInternal(Oid myrelid, const char *newrelname, Oid namespaceId) RenameRelationInternal(Oid myrelid, const char *newrelname, Oid namespaceId)
{ {