Hi all,
We have enhanced pgbench so that it accepts a series of SQL commands
in a file(see attached patches against 8.0.3). This would make it
possible to test various sets of SQL commands. In the file it is
allowed to use a "meta command". Currently only "\setrandom" meta
command is allowed, which sets specified random number into a
variable. For example,
\setrandom aid 1 100000
will set a random number into variable "aid" between 1 and 10000.
A variable can be reffered to in an SQL command by adding ":" in front
of the the command name.
Here is an example SQL command file.
\setrandom aid 1 100000
\setrandom bid 1 1
\setrandom tid 1 10
\setrandom delta 1 10000
BEGIN
UPDATE accounts SET abalance = abalance + :delta WHERE aid = :aid
SELECT abalance FROM accounts WHERE aid = :aid
UPDATE tellers SET tbalance = tbalance + :delta WHERE tid = :tid
UPDATE branches SET bbalance = bbalance + :delta WHERE bid = :bid
INSERT INTO history (tid, bid, aid, delta, mtime) VALUES (:tid, :bid, :aid,
:delta, 'now')
END
This will execute virtually same SQL commands builtin pgbench.
To use the SQL command file, you can use "-f" option:
pgbench -f /foo/bar/sqlfile
I think the enhanced pgbench is quite usefull and I would like to
include in 8.1. Or should I keep it for 8.2?
--
SRA OSS, Inc. Japan
Tatsuo Ishii
*** pgbench/pgbench.c 2004-11-09 15:09:31.000000000 +0900
--- pgbench-new/pgbench.c 2005-09-27 14:31:34.000000000 +0900
***************
*** 41,46 ****
--- 41,49 ----
#include <sys/resource.h>
#endif /* ! WIN32 */
+ #include <ctype.h>
+ #include <search.h>
+
extern char *optarg;
extern int optind;
***************
*** 72,77 ****
--- 75,83 ----
#define ntellers 10
#define naccounts 100000
+ #define SQL_COMMAND 1
+ #define META_COMMAND 2
+
FILE *LOGFILE = NULL;
bool use_log; /* log transaction latencies to
a file */
***************
*** 91,96 ****
--- 97,108 ----
typedef struct
{
+ char *name;
+ char *value;
+ } Variable;
+
+ typedef struct
+ {
PGconn *con; /* connection handle to DB */
int id; /* client No. */
int state; /* state No. */
***************
*** 103,115 ****
int tid; /* teller id for this
transaction */
int delta;
int abalance;
struct timeval txn_begin; /* used for measuring latencies */
} CState;
static void
usage(void)
{
! fprintf(stderr, "usage: pgbench [-h hostname][-p port][-c nclients][-t
ntransactions][-s scaling_factor][-n][-C][-v][-S][-N][-l][-U login][-P
password][-d][dbname]\n");
fprintf(stderr, "(initialize mode): pgbench -i [-h hostname][-p
port][-s scaling_factor][-U login][-P password][-d][dbname]\n");
}
--- 115,137 ----
int tid; /* teller id for this
transaction */
int delta;
int abalance;
+ void *variables;
struct timeval txn_begin; /* used for measuring latencies */
} CState;
+ typedef struct
+ {
+ int type;
+ int argc;
+ char **argv;
+ } Command;
+
+ Command **commands = NULL;
+
static void
usage(void)
{
! fprintf(stderr, "usage: pgbench [-h hostname][-p port][-c nclients][-t
ntransactions][-s scaling_factor][-n][-C][-v][-S][-N][-f filename][-l][-U
login][-P password][-d][dbname]\n");
fprintf(stderr, "(initialize mode): pgbench -i [-h hostname][-p
port][-s scaling_factor][-U login][-P password][-d][dbname]\n");
}
***************
*** 190,195 ****
--- 212,326 ----
return (0); /* OK */
}
+ static int
+ compareVariables(const void *v1, const void *v2)
+ {
+ return strcmp(((Variable *)v1)->name, ((Variable *)v2)->name);
+ }
+
+ static char *
+ getVariable(CState * st, char *name)
+ {
+ Variable key = { name }, *var;
+
+ var = tfind(&key, &st->variables, compareVariables);
+ if (var != NULL)
+ return (*(Variable **)var)->value;
+ else
+ return NULL;
+ }
+
+ static int
+ putVariable(CState * st, char *name, char *value)
+ {
+ Variable key = { name }, *var;
+
+ var = tfind(&key, &st->variables, compareVariables);
+ if (var == NULL)
+ {
+ if ((var = malloc(sizeof(Variable))) == NULL)
+ return false;
+
+ var->name = NULL;
+ var->value = NULL;
+
+ if ((var->name = strdup(name)) == NULL
+ || (var->value = strdup(value)) == NULL
+ || tsearch(var, &st->variables, compareVariables) ==
NULL)
+ {
+ free(var->name);
+ free(var->value);
+ free(var);
+ return false;
+ }
+ }
+ else
+ {
+ free((*(Variable **)var)->value);
+ if (((*(Variable **)var)->value = strdup(value)) == NULL)
+ return false;
+ }
+
+ return true;
+ }
+
+ static char *
+ assignVariables(CState * st, char *sql)
+ {
+ int i, j;
+ char *p, *name, *val;
+ void *tmp;
+
+ i = 0;
+ while ((p = strchr(&sql[i], ':')) != NULL)
+ {
+ i = j = p - sql;
+ do
+ i++;
+ while (isalnum(sql[i]) != 0 || sql[i] == '_');
+ if (i == j + 1)
+ continue;
+
+ if ((name = strndup(&sql[j + 1], i - (j + 1))) == NULL)
+ return NULL;
+ val = getVariable(st, name);
+ free(name);
+ if (val == NULL)
+ continue;
+
+ if (strlen(val) > i - j)
+ {
+ tmp = realloc(sql, strlen(sql) - (i - j) + strlen(val)
+ 1);
+ if (tmp == NULL)
+ {
+ free(sql);
+ return NULL;
+ }
+ sql = tmp;
+ }
+
+ if (strlen(val) != i - j)
+ memmove(&sql[j + strlen(val)], &sql[i], strlen(&sql[i])
+ 1);
+
+ strncpy(&sql[j], val, strlen(val));
+
+ if (strlen(val) < i - j)
+ {
+ tmp = realloc(sql, strlen(sql) + 1);
+ if (tmp == NULL)
+ {
+ free(sql);
+ return NULL;
+ }
+ sql = tmp;
+ }
+
+ i = j + strlen(val);
+ }
+
+ return sql;
+ }
+
/* process a transaction */
static void
doOne(CState * state, int n, int debug, int ttype)
***************
*** 465,470 ****
--- 596,765 ----
}
}
+ static void
+ doCustom(CState * state, int n, int debug)
+ {
+ PGresult *res;
+ CState *st = &state[n];
+
+ if (st->listen)
+ { /* are we
receiver? */
+ if (commands[st->state]->type == SQL_COMMAND)
+ {
+ if (debug)
+ fprintf(stderr, "client %d receiving\n", n);
+ if (!PQconsumeInput(st->con))
+ { /*
there's something wrong */
+ fprintf(stderr, "Client %d aborted in state %d.
Probably the backend died while processing.\n", n, st->state);
+ remains--; /* I've aborted
*/
+ PQfinish(st->con);
+ st->con = NULL;
+ return;
+ }
+ if (PQisBusy(st->con))
+ return; /* don't have
the whole result yet */
+ }
+
+ /*
+ * transaction finished: record the time it took in the
+ * log
+ */
+ if (use_log && commands[st->state + 1] == NULL)
+ {
+ double diff;
+ struct timeval now;
+
+ gettimeofday(&now, NULL);
+ diff = (int) (now.tv_sec - st->txn_begin.tv_sec) *
1000000.0 +
+ (int) (now.tv_usec - st->txn_begin.tv_usec);
+
+ fprintf(LOGFILE, "%d %d %.0f\n", st->id, st->cnt, diff);
+ }
+
+ if (commands[st->state]->type == SQL_COMMAND)
+ {
+ res = PQgetResult(st->con);
+ if (strncasecmp(commands[st->state]->argv[0], "select",
6) != 0)
+ {
+ if (check(state, res, n, PGRES_COMMAND_OK))
+ return;
+ }
+ else
+ {
+ if (check(state, res, n, PGRES_TUPLES_OK))
+ return;
+ }
+ PQclear(res);
+ discard_response(st);
+ }
+
+ if (commands[st->state + 1] == NULL)
+ {
+ if (is_connect)
+ {
+ PQfinish(st->con);
+ st->con = NULL;
+ }
+
+ if (++st->cnt >= nxacts)
+ {
+ remains--; /* I'm done */
+ if (st->con != NULL)
+ {
+ PQfinish(st->con);
+ st->con = NULL;
+ }
+ return;
+ }
+ }
+
+ /* increment state counter */
+ st->state++;
+ if (commands[st->state] == NULL)
+ st->state = 0;
+ }
+
+ if (st->con == NULL)
+ {
+ if ((st->con = doConnect()) == NULL)
+ {
+ fprintf(stderr, "Client %d aborted in establishing
connection.\n",
+ n);
+ remains--; /* I've aborted */
+ PQfinish(st->con);
+ st->con = NULL;
+ return;
+ }
+ }
+
+ if (use_log && st->state == 0)
+ gettimeofday(&(st->txn_begin), NULL);
+
+ if (commands[st->state]->type == SQL_COMMAND)
+ {
+ char *sql;
+
+ if ((sql = strdup(commands[st->state]->argv[0])) == NULL
+ || (sql = assignVariables(st, sql)) == NULL)
+ {
+ fprintf(stderr, "out of memory\n");
+ st->ecnt++;
+ return;
+ }
+
+ if (debug)
+ fprintf(stderr, "client %d sending %s\n", n, sql);
+ if (PQsendQuery(st->con, sql) == 0)
+ {
+ if (debug)
+ fprintf(stderr, "PQsendQuery(%s)failed\n", sql);
+ st->ecnt++;
+ }
+ else
+ {
+ st->listen++; /* flags that should be
listened */
+ }
+ }
+ else if (commands[st->state]->type == META_COMMAND)
+ {
+ int argc = commands[st->state]->argc, i;
+ char **argv = commands[st->state]->argv;
+
+ if (debug)
+ {
+ fprintf(stderr, "client %d executing \\%s", n, argv[0]);
+ for (i = 1; i < argc; i++)
+ fprintf(stderr, " %s", argv[i]);
+ fprintf(stderr, "\n");
+ }
+
+ if (strcasecmp(argv[0], "setrandom") == 0)
+ {
+ char *val;
+
+ if ((val = malloc(strlen(argv[3]) + 1)) == NULL)
+ {
+ fprintf(stderr, "%s: out of memory\n", argv[0]);
+ st->ecnt++;
+ return;
+ }
+
+ sprintf(val, "%d", getrand(atoi(argv[2]),
atoi(argv[3])));
+
+ if (putVariable(st, argv[1], val) == false)
+ {
+ fprintf(stderr, "%s: out of memory\n", argv[0]);
+ free(val);
+ st->ecnt++;
+ return;
+ }
+
+ free(val);
+ st->listen++;
+ }
+ }
+ }
+
/* discard connections */
static void
disconnect_all(CState * state)
***************
*** 644,649 ****
--- 939,1098 ----
PQfinish(con);
}
+ static int
+ process_file(char *filename)
+ {
+ const char delim[] = " \f\n\r\t\v";
+
+ FILE *fd;
+ int lineno, i, j;
+ char buf[BUFSIZ], *p, *tok;
+ void *tmp;
+
+ if (strcmp(filename, "-") == 0)
+ fd = stdin;
+ else if ((fd = fopen(filename, "r")) == NULL)
+ {
+ fprintf(stderr, "%s: %s\n", strerror(errno), filename);
+ return false;
+ }
+
+ fprintf(stderr, "processing file...\n");
+
+ lineno = 1;
+ i = 0;
+ while (fgets(buf, sizeof(buf), fd) != NULL)
+ {
+ if ((p = strchr(buf, '\n')) != NULL)
+ *p = '\0';
+ p = buf;
+ while (isspace(*p))
+ p++;
+ if (*p == '\0' || strncmp(p, "--", 2) == 0)
+ {
+ lineno++;
+ continue;
+ }
+
+ if ((tmp = realloc(commands, sizeof(Command *) * (i + 1))) ==
NULL)
+ {
+ i--;
+ goto error;
+ }
+ commands = tmp;
+
+ if ((commands[i] = malloc(sizeof(Command))) == NULL)
+ goto error;
+
+ commands[i]->argv = NULL;
+ commands[i]->argc = 0;
+
+ if (*p == '\\')
+ {
+ commands[i]->type = META_COMMAND;
+
+ j = 0;
+ tok = strtok(++p, delim);
+ while (tok != NULL)
+ {
+ tmp = realloc(commands[i]->argv, sizeof(char *)
* (j + 1));
+ if (tmp == NULL)
+ goto error;
+ commands[i]->argv = tmp;
+
+ if ((commands[i]->argv[j] = strdup(tok)) ==
NULL)
+ goto error;
+
+ commands[i]->argc++;
+
+ j++;
+ tok = strtok(NULL, delim);
+ }
+
+ if (strcasecmp(commands[i]->argv[0], "setrandom") == 0)
+ {
+ int min, max;
+
+ if (commands[i]->argc < 4)
+ {
+ fprintf(stderr, "%s: %d: \\%s: missing
argument\n", filename, lineno, commands[i]->argv[0]);
+ goto error;
+ }
+
+ for (j = 4; j < commands[i]->argc; j++)
+ fprintf(stderr, "%s: %d: \\%s: extra
argument \"%s\" ignored\n", filename, lineno, commands[i]->argv[0],
commands[i]->argv[j]);
+
+ if ((min = atoi(commands[i]->argv[2])) < 0)
+ {
+ fprintf(stderr, "%s: %d: \\%s: invalid
minimum number %s\n", filename, lineno, commands[i]->argv[0],
commands[i]->argv[2]);
+ goto error;
+ }
+
+ if ((max = atoi(commands[i]->argv[3])) < min ||
max > RAND_MAX)
+ {
+ fprintf(stderr, "%s: %d: \\%s: invalid
maximum number %s\n", filename, lineno, commands[i]->argv[0],
commands[i]->argv[3]);
+ goto error;
+ }
+ }
+ else
+ {
+ fprintf(stderr, "%s: %d: invalid command
\\%s\n", filename, lineno, commands[i]->argv[0]);
+ goto error;
+ }
+ }
+ else
+ {
+ commands[i]->type = SQL_COMMAND;
+
+ if ((commands[i]->argv = malloc(sizeof(char *))) ==
NULL)
+ goto error;
+
+ if ((commands[i]->argv[0] = strdup(p)) == NULL)
+ goto error;
+
+ commands[i]->argc++;
+ }
+
+ i++;
+ lineno++;
+ }
+ fclose(fd);
+
+ if ((tmp = realloc(commands, sizeof(Command *) * (i + 1))) == NULL)
+ goto error;
+ commands = tmp;
+
+ commands[i] = NULL;
+
+ return true;
+
+ error:
+ if (errno == ENOMEM)
+ fprintf(stderr, "%s: %d: out of memory\n", filename, lineno);
+
+ fclose(fd);
+
+ if (commands == NULL)
+ return false;
+
+ while (i >= 0)
+ {
+ if (commands[i] != NULL)
+ {
+ for (j = 0; j < commands[i]->argc; j++)
+ free(commands[i]->argv[j]);
+
+ free(commands[i]->argv);
+ free(commands[i]);
+ }
+
+ i--;
+ }
+ free(commands);
+
+ return false;
+ }
+
/* print out results */
static void
printResults(
***************
*** 670,677 ****
s = "TPC-B (sort of)";
else if (ttype == 2)
s = "Update only accounts";
! else
s = "SELECT only";
printf("transaction type: %s\n", s);
printf("scaling factor: %d\n", tps);
--- 1119,1128 ----
s = "TPC-B (sort of)";
else if (ttype == 2)
s = "Update only accounts";
! else if (ttype == 1)
s = "SELECT only";
+ else
+ s = "Custom query";
printf("transaction type: %s\n", s);
printf("scaling factor: %d\n", tps);
***************
*** 695,700 ****
--- 1146,1152 ----
int ttype = 0; /* transaction type. 0:
TPC-B, 1: SELECT
* only, 2:
skip update of branches and
* tellers */
+ char *filename = NULL;
static CState *state; /* status of clients */
***************
*** 724,730 ****
else if ((env = getenv("PGUSER")) != NULL && *env != '\0')
login = env;
! while ((c = getopt(argc, argv, "ih:nvp:dc:t:s:U:P:CNSl")) != -1)
{
switch (c)
{
--- 1176,1182 ----
else if ((env = getenv("PGUSER")) != NULL && *env != '\0')
login = env;
! while ((c = getopt(argc, argv, "ih:nvp:dc:t:s:U:P:CNSlf:")) != -1)
{
switch (c)
{
***************
*** 806,811 ****
--- 1258,1267 ----
case 'l':
use_log = true;
break;
+ case 'f':
+ ttype = 3;
+ filename = optarg;
+ break;
default:
usage();
exit(1);
***************
*** 868,941 ****
exit(1);
}
! /*
! * get the scaling factor that should be same as count(*) from
! * branches...
! */
! res = PQexec(con, "select count(*) from branches");
! if (PQresultStatus(res) != PGRES_TUPLES_OK)
{
! fprintf(stderr, "%s", PQerrorMessage(con));
! exit(1);
! }
! tps = atoi(PQgetvalue(res, 0, 0));
! if (tps < 0)
! {
! fprintf(stderr, "count(*) from branches invalid (%d)\n", tps);
! exit(1);
}
! PQclear(res);
!
! if (!is_no_vacuum)
{
! fprintf(stderr, "starting vacuum...");
! res = PQexec(con, "vacuum branches");
! if (PQresultStatus(res) != PGRES_COMMAND_OK)
{
fprintf(stderr, "%s", PQerrorMessage(con));
exit(1);
}
! PQclear(res);
!
! res = PQexec(con, "vacuum tellers");
! if (PQresultStatus(res) != PGRES_COMMAND_OK)
{
! fprintf(stderr, "%s", PQerrorMessage(con));
exit(1);
}
PQclear(res);
! res = PQexec(con, "delete from history");
! if (PQresultStatus(res) != PGRES_COMMAND_OK)
{
! fprintf(stderr, "%s", PQerrorMessage(con));
! exit(1);
! }
! PQclear(res);
! res = PQexec(con, "vacuum history");
! if (PQresultStatus(res) != PGRES_COMMAND_OK)
! {
! fprintf(stderr, "%s", PQerrorMessage(con));
! exit(1);
! }
! PQclear(res);
! fprintf(stderr, "end.\n");
! if (is_full_vacuum)
! {
! fprintf(stderr, "starting full vacuum...");
! res = PQexec(con, "vacuum analyze accounts");
if (PQresultStatus(res) != PGRES_COMMAND_OK)
{
fprintf(stderr, "%s", PQerrorMessage(con));
exit(1);
}
PQclear(res);
fprintf(stderr, "end.\n");
}
}
- PQfinish(con);
/* set random seed */
gettimeofday(&tv1, NULL);
--- 1324,1406 ----
exit(1);
}
! if (ttype == 3)
{
! PQfinish(con);
! if (process_file(filename) == false)
! exit(1);
}
! else
{
! /*
! * get the scaling factor that should be same as count(*) from
! * branches...
! */
! res = PQexec(con, "select count(*) from branches");
! if (PQresultStatus(res) != PGRES_TUPLES_OK)
{
fprintf(stderr, "%s", PQerrorMessage(con));
exit(1);
}
! tps = atoi(PQgetvalue(res, 0, 0));
! if (tps < 0)
{
! fprintf(stderr, "count(*) from branches invalid
(%d)\n", tps);
exit(1);
}
PQclear(res);
! if (!is_no_vacuum)
{
! fprintf(stderr, "starting vacuum...");
! res = PQexec(con, "vacuum branches");
! if (PQresultStatus(res) != PGRES_COMMAND_OK)
! {
! fprintf(stderr, "%s", PQerrorMessage(con));
! exit(1);
! }
! PQclear(res);
! res = PQexec(con, "vacuum tellers");
! if (PQresultStatus(res) != PGRES_COMMAND_OK)
! {
! fprintf(stderr, "%s", PQerrorMessage(con));
! exit(1);
! }
! PQclear(res);
! res = PQexec(con, "delete from history");
! if (PQresultStatus(res) != PGRES_COMMAND_OK)
! {
! fprintf(stderr, "%s", PQerrorMessage(con));
! exit(1);
! }
! PQclear(res);
! res = PQexec(con, "vacuum history");
if (PQresultStatus(res) != PGRES_COMMAND_OK)
{
fprintf(stderr, "%s", PQerrorMessage(con));
exit(1);
}
PQclear(res);
+
fprintf(stderr, "end.\n");
+
+ if (is_full_vacuum)
+ {
+ fprintf(stderr, "starting full vacuum...");
+ res = PQexec(con, "vacuum analyze accounts");
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ {
+ fprintf(stderr, "%s",
PQerrorMessage(con));
+ exit(1);
+ }
+ PQclear(res);
+ fprintf(stderr, "end.\n");
+ }
}
+ PQfinish(con);
}
/* set random seed */
gettimeofday(&tv1, NULL);
***************
*** 965,970 ****
--- 1430,1437 ----
doOne(state, i, debug, ttype);
else if (ttype == 1)
doSelectOnly(state, i, debug);
+ else if (ttype == 3)
+ doCustom(state, i, debug);
}
for (;;)
***************
*** 982,997 ****
FD_ZERO(&input_mask);
! maxsock = 0;
for (i = 0; i < nclients; i++)
{
! if (state[i].con)
{
int sock =
PQsocket(state[i].con);
if (sock < 0)
{
- fprintf(stderr, "Client %d: PQsocket
failed\n", i);
disconnect_all(state);
exit(1);
}
--- 1449,1464 ----
FD_ZERO(&input_mask);
! maxsock = -1;
for (i = 0; i < nclients; i++)
{
! if (state[i].con &&
! (ttype != 3 || commands[state[i].state]->type
!= META_COMMAND))
{
int sock =
PQsocket(state[i].con);
if (sock < 0)
{
disconnect_all(state);
exit(1);
}
***************
*** 1001,1036 ****
}
}
! if ((nsocks = select(maxsock + 1, &input_mask, (fd_set *) NULL,
! (fd_set *) NULL, (struct
timeval *) NULL)) < 0)
{
! if (errno == EINTR)
! continue;
! /* must be something wrong */
! disconnect_all(state);
! fprintf(stderr, "select failed: %s\n", strerror(errno));
! exit(1);
! }
! else if (nsocks == 0)
! { /* timeout */
! fprintf(stderr, "select timeout\n");
! for (i = 0; i < nclients; i++)
{
! fprintf(stderr, "client %d:state %d cnt %d ecnt
%d listen %d\n",
! i, state[i].state,
state[i].cnt, state[i].ecnt, state[i].listen);
}
- exit(0);
}
/* ok, backend returns reply */
for (i = 0; i < nclients; i++)
{
! if (state[i].con && FD_ISSET(PQsocket(state[i].con),
&input_mask))
{
if (ttype == 0 || ttype == 2)
doOne(state, i, debug, ttype);
else if (ttype == 1)
doSelectOnly(state, i, debug);
}
}
}
--- 1468,1510 ----
}
}
! if (maxsock != -1)
{
! if ((nsocks = select(maxsock + 1, &input_mask, (fd_set
*) NULL,
! (fd_set *) NULL,
(struct timeval *) NULL)) < 0)
{
! if (errno == EINTR)
! continue;
! /* must be something wrong */
! disconnect_all(state);
! fprintf(stderr, "select failed: %s\n",
strerror(errno));
! exit(1);
! }
! else if (nsocks == 0)
! { /*
timeout */
! fprintf(stderr, "select timeout\n");
! for (i = 0; i < nclients; i++)
! {
! fprintf(stderr, "client %d:state %d cnt
%d ecnt %d listen %d\n",
! i, state[i].state,
state[i].cnt, state[i].ecnt, state[i].listen);
! }
! exit(0);
}
}
/* ok, backend returns reply */
for (i = 0; i < nclients; i++)
{
! if (state[i].con && (FD_ISSET(PQsocket(state[i].con),
&input_mask)
! || (ttype == 3
! &&
commands[state[i].state]->type == META_COMMAND)))
{
if (ttype == 0 || ttype == 2)
doOne(state, i, debug, ttype);
else if (ttype == 1)
doSelectOnly(state, i, debug);
+ else if (ttype == 3)
+ doCustom(state, i, debug);
}
}
}
---------------------------(end of broadcast)---------------------------
TIP 4: Have you searched our list archives?
http://archives.postgresql.org
