Re: pgbench and timestamps (bounced)

diff --git a/doc/src/sgml/ref/pgbench.sgml b/doc/src/sgml/ref/pgbench.sgml
index 8e728dc094..7436210fd4 100644
--- a/doc/src/sgml/ref/pgbench.sgml
+++ b/doc/src/sgml/ref/pgbench.sgml
@@ -998,15 +998,14 @@ pgbench <optional> <replaceable>options</replaceable> </optional> <replaceable>d
   <para>
    There is a simple variable-substitution facility for script files.
    Variable names must consist of letters (including non-Latin letters),
-   digits, and underscores.
+   digits (but not on the first character), and underscores.
    Variables can be set by the command-line <option>-D</option> option,
    explained above, or by the meta commands explained below.
    In addition to any variables preset by <option>-D</option> command-line options,
    there are a few variables that are preset automatically, listed in
    <xref linkend="pgbench-automatic-variables"/>. A value specified for these
    variables using <option>-D</option> takes precedence over the automatic presets.
-   Once set, a variable's
-   value can be inserted into a SQL command by writing
+   Once set, a variable's value can be inserted into a SQL command by writing
    <literal>:</literal><replaceable>variablename</replaceable>.  When running more than
    one client session, each session has its own set of variables.
    <application>pgbench</application> supports up to 255 variable uses in one
diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c
index 08a5947a9e..09ccf05db5 100644
--- a/src/bin/pgbench/pgbench.c
+++ b/src/bin/pgbench/pgbench.c
@@ -119,12 +119,24 @@ typedef int pthread_attr_t;
 
 static int	pthread_create(pthread_t *thread, pthread_attr_t *attr, void *(*start_routine) (void *), void *arg);
 static int	pthread_join(pthread_t th, void **thread_return);
+
+typedef HANDLE pthread_mutex_t;
+static void pthread_mutex_init(pthread_mutex_t *pm, void *unused);
+static void pthread_mutex_lock(pthread_mutex_t *pm);
+static void pthread_mutex_unlock(pthread_mutex_t *pm);
+static void pthread_mutex_destroy(pthread_mutex_t *pm);
+
 #elif defined(ENABLE_THREAD_SAFETY)
 /* Use platform-dependent pthread capability */
 #include <pthread.h>
 #else
 /* No threads implementation, use none (-j 1) */
 #define pthread_t void *
+#define pthread_mutex_t void *
+#define pthread_mutex_init(m, p)
+#define pthread_mutex_lock(m)
+#define pthread_mutex_unlock(m)
+#define pthread_mutex_destroy(m)
 #endif
 
 
@@ -422,7 +434,7 @@ typedef struct
 	instr_time	txn_begin;		/* used for measuring schedule lag times */
 	instr_time	stmt_begin;		/* used for measuring statement latencies */
 
-	bool		prepared[MAX_SCRIPTS];	/* whether client prepared the script */
+	bool*		prepared[MAX_SCRIPTS];	/* whether client prepared the script commands */
 
 	/* per client collected stats */
 	int64		cnt;			/* client transaction count, for -t */
@@ -472,7 +484,7 @@ typedef struct
  */
 #define MAX_ARGS		256
 
-typedef enum MetaCommand
+typedef enum Meta
 {
 	META_NONE,					/* not a known meta-command */
 	META_SET,					/* \set */
@@ -503,6 +515,8 @@ static const char *QUERYMODE[] = {"simple", "extended", "prepared"};
  *
  * lines		The raw, possibly multi-line command text.  Variable substitution
  *				not applied.
+ * substituted	Whether command	:-variables were substituted for
+ *				QUERY_EXTENDED and QUERY_PREPARED
  * first_line	A short, single-line extract of 'lines', for error reporting.
  * type			SQL_COMMAND or META_COMMAND
  * meta			The type of meta-command, with META_NONE/GSET/ASET if command
@@ -517,10 +531,12 @@ static const char *QUERYMODE[] = {"simple", "extended", "prepared"};
  * aset			do gset on all possible queries of a combined query (\;).
  * expr			Parsed expression, if needed.
  * stats		Time spent in this command.
+ * mutex		For delayed initializations of SQL commands.
  */
 typedef struct Command
 {
 	PQExpBufferData lines;
+	bool		substituted;
 	char	   *first_line;
 	int			type;
 	MetaCommand meta;
@@ -529,6 +545,7 @@ typedef struct Command
 	char	   *varprefix;
 	PgBenchExpr *expr;
 	SimpleStats stats;
+	pthread_mutex_t	mutex;
 } Command;
 
 typedef struct ParsedScript
@@ -613,6 +630,7 @@ static void clear_socket_set(socket_set *sa);
 static void add_socket_to_set(socket_set *sa, int fd, int idx);
 static int	wait_on_socket_set(socket_set *sa, int64 usecs);
 static bool socket_has_input(socket_set *sa, int fd, int idx);
+static bool makeVariablesParameters(CState *st, Command *cmd);
 
 
 /* callback functions for our flex lexer */
@@ -1273,7 +1291,7 @@ lookupVariable(CState *st, char *name)
 
 /* Get the value of a variable, in string form; returns NULL if unknown */
 static char *
-getVariable(CState *st, char *name)
+getVariable(CState *st, const char *name, bool *isnull)
 {
 	Variable   *var;
 	char		stringform[64];
@@ -1282,6 +1300,9 @@ getVariable(CState *st, char *name)
 	if (var == NULL)
 		return NULL;			/* not found */
 
+	if (isnull != NULL)
+		*isnull = var->value.type == PGBT_NULL;
+
 	if (var->svalue)
 		return var->svalue;		/* we have it in string form */
 
@@ -1377,6 +1398,9 @@ makeVariableValue(Variable *var)
  * "src/bin/pgbench/exprscan.l".  Also see parseVariable(), below.
  *
  * Note: this static function is copied from "src/bin/psql/variables.c"
+ * but changed to disallow variable names starting with a figure.
+ *
+ * FIXME does this change needs to be reverted?
  */
 static bool
 valid_variable_name(const char *name)
@@ -1387,6 +1411,15 @@ valid_variable_name(const char *name)
 	if (*ptr == '\0')
 		return false;
 
+	/* must not start with [0-9] */
+	if (IS_HIGHBIT_SET(*ptr) ||
+		strchr("ABCDEFGHIJKLMNOPQRSTUVWXYZ" "abcdefghijklmnopqrstuvwxyz"
+			   "_", *ptr) != NULL)
+		ptr++;
+	else
+		return false;
+
+	/* remainder characters can include [0-9] */
 	while (*ptr)
 	{
 		if (IS_HIGHBIT_SET(*ptr) ||
@@ -1502,6 +1535,16 @@ putVariableInt(CState *st, const char *context, char *name, int64 value)
 	return putVariableValue(st, context, name, &val);
 }
 
+/* Assign variable to NULL, return false on bad name */
+static bool
+putVariableNull(CState *st, const char *context, char *name)
+{
+	PgBenchValue val;
+
+	setNullValue(&val);
+	return putVariableValue(st, context, name, &val);
+}
+
 /*
  * Parse a possible variable reference (:varname).
  *
@@ -1516,14 +1559,22 @@ parseVariable(const char *sql, int *eaten)
 	int			i = 0;
 	char	   *name;
 
-	do
-	{
+	/* skip ':' */
+	i++;
+
+	/* first char of name must be valid but not [0-9] */
+	if (IS_HIGHBIT_SET(sql[i]) ||
+		strchr("ABCDEFGHIJKLMNOPQRSTUVWXYZ" "abcdefghijklmnopqrstuvwxyz"
+			   "_", sql[i]) != NULL)
+		i++;
+	else
+		return NULL;
+
+	/* then proceed to check other chars */
+	while (IS_HIGHBIT_SET(sql[i]) ||
+		   strchr("ABCDEFGHIJKLMNOPQRSTUVWXYZ" "abcdefghijklmnopqrstuvwxyz"
+				  "_0123456789", sql[i]) != NULL)
 		i++;
-	} while (IS_HIGHBIT_SET(sql[i]) ||
-			 strchr("ABCDEFGHIJKLMNOPQRSTUVWXYZ" "abcdefghijklmnopqrstuvwxyz"
-					"_0123456789", sql[i]) != NULL);
-	if (i == 1)
-		return NULL;			/* no valid variable name chars */
 
 	name = pg_malloc(i);
 	memcpy(name, &sql[1], i - 1);
@@ -1575,7 +1626,7 @@ assignVariables(CState *st, char *sql)
 			continue;
 		}
 
-		val = getVariable(st, name);
+		val = getVariable(st, name, NULL);
 		free(name);
 		if (val == NULL)
 		{
@@ -1592,10 +1643,12 @@ assignVariables(CState *st, char *sql)
 static void
 getQueryParams(CState *st, const Command *command, const char **params)
 {
-	int			i;
-
-	for (i = 0; i < command->argc - 1; i++)
-		params[i] = getVariable(st, command->argv[i + 1]);
+	for (int i = 0; i < command->argc - 1; i++)
+	{
+		bool	isnull;
+		char   *sval = getVariable(st, command->argv[i + 1], &isnull);
+		params[i] = isnull ? NULL : sval;
+	}
 }
 
 static char *
@@ -2535,7 +2588,7 @@ runShellCommand(CState *st, char *variable, char **argv, int argc)
 		{
 			arg = argv[i] + 1;	/* a string literal starting with colons */
 		}
-		else if ((arg = getVariable(st, argv[i] + 1)) == NULL)
+		else if ((arg = getVariable(st, argv[i] + 1, NULL)) == NULL)
 		{
 			pg_log_error("%s: undefined variable \"%s\"", argv[0], argv[i]);
 			return false;
@@ -2656,13 +2709,29 @@ sendCommand(CState *st, Command *command)
 	}
 	else if (querymode == QUERY_EXTENDED)
 	{
-		const char *sql = command->argv[0];
 		const char *params[MAX_ARGS];
 
+		/* do not try the mutex unless it will be necessary */
+		if (!command->substituted)
+		{
+			pthread_mutex_lock(&command->mutex);
+			if (!command->substituted)
+			{
+				if (!makeVariablesParameters(st, command))
+				{
+					pg_log_fatal("client %d script %s command %d pgbench variable substitution failed",
+								 st->id, sql_script[st->use_file].desc, st->command);
+					exit(1);
+				}
+				command->substituted = true;
+			}
+			pthread_mutex_unlock(&command->mutex);
+		}
+
 		getQueryParams(st, command, params);
 
-		pg_log_debug("client %d sending %s", st->id, sql);
-		r = PQsendQueryParams(st->con, sql, command->argc - 1,
+		pg_log_debug("client %d sending %s", st->id, command->argv[0]);
+		r = PQsendQueryParams(st->con, command->argv[0], command->argc - 1,
 							  NULL, params, NULL, NULL, 0);
 	}
 	else if (querymode == QUERY_PREPARED)
@@ -2670,31 +2739,42 @@ sendCommand(CState *st, Command *command)
 		char		name[MAX_PREPARE_NAME];
 		const char *params[MAX_ARGS];
 
-		if (!st->prepared[st->use_file])
+		/* do not try the mutex unless it will be necessary */
+		if (!command->substituted)
 		{
-			int			j;
-			Command   **commands = sql_script[st->use_file].commands;
-
-			for (j = 0; commands[j] != NULL; j++)
+			pthread_mutex_lock(&command->mutex);
+			if (!command->substituted)
 			{
-				PGresult   *res;
-				char		name[MAX_PREPARE_NAME];
-
-				if (commands[j]->type != SQL_COMMAND)
-					continue;
-				preparedStatementName(name, st->use_file, j);
-				res = PQprepare(st->con, name,
-								commands[j]->argv[0], commands[j]->argc - 1, NULL);
-				if (PQresultStatus(res) != PGRES_COMMAND_OK)
-					pg_log_error("%s", PQerrorMessage(st->con));
-				PQclear(res);
+				if (!makeVariablesParameters(st, command))
+				{
+					pg_log_fatal("client %d script %s command %d pgbench variable substitution failed",
+								 st->id, sql_script[st->use_file].desc, st->command);
+					exit(1);
+				}
+				command->substituted = true;
 			}
-			st->prepared[st->use_file] = true;
+			pthread_mutex_unlock(&command->mutex);
 		}
 
-		getQueryParams(st, command, params);
 		preparedStatementName(name, st->use_file, st->command);
 
+		/* each client must prepare each sql command once */
+		if (!st->prepared[st->use_file][st->command])
+		{
+			PGresult   *res;
+
+			res = PQprepare(st->con, name,
+							command->argv[0], command->argc - 1, NULL);
+			if (PQresultStatus(res) != PGRES_COMMAND_OK)
+				pg_log_error("%s", PQerrorMessage(st->con));
+
+			PQclear(res);
+
+			st->prepared[st->use_file][st->command] = true;
+		}
+
+		getQueryParams(st, command, params);
+
 		pg_log_debug("client %d sending %s", st->id, name);
 		r = PQsendQueryPrepared(st->con, name, command->argc - 1,
 								params, NULL, NULL, 0);
@@ -2785,7 +2865,9 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix)
 							varname = psprintf("%s%s", varprefix, varname);
 
 						/* store last row result as a string */
-						if (!putVariable(st, meta == META_ASET ? "aset" : "gset", varname,
+						if (PQgetisnull(res, ntuples - 1, fld) ?
+							!putVariableNull(st, meta == META_ASET ? "aset" : "gset", varname) :
+							!putVariable(st, meta == META_ASET ? "aset" : "gset", varname,
 										 PQgetvalue(res, ntuples - 1, fld)))
 						{
 							/* internal error */
@@ -2848,7 +2930,7 @@ evaluateSleep(CState *st, int argc, char **argv, int *usecs)
 
 	if (*argv[1] == ':')
 	{
-		if ((var = getVariable(st, argv[1] + 1)) == NULL)
+		if ((var = getVariable(st, argv[1] + 1, NULL)) == NULL)
 		{
 			pg_log_error("%s: undefined variable \"%s\"", argv[0], argv[1] + 1);
 			return false;
@@ -4301,11 +4383,15 @@ GetTableInfo(PGconn *con, bool scale_given)
 }
 
 /*
- * Replace :param with $n throughout the command's SQL text, which
+ * Roughly replace :param with $n throughout the command's SQL text, which
  * is a modifiable string in cmd->lines.
+ *
+ * Wherever :whatever is found, it is tried as a possible a variable name.
+ *
+ * FIXME should it do minimal lexing (in s/d quotes, ...)?
  */
 static bool
-parseQuery(Command *cmd)
+makeVariablesParameters(CState *st, Command *cmd)
 {
 	char	   *sql,
 			   *p;
@@ -4317,15 +4403,23 @@ parseQuery(Command *cmd)
 	{
 		char		var[13];
 		char	   *name;
+		char	   *val;
 		int			eaten;
 
 		name = parseVariable(p, &eaten);
 		if (name == NULL)
 		{
 			while (*p == ':')
-			{
 				p++;
-			}
+			continue;
+		}
+
+		/* skip if variable is undefined? */
+		val = getVariable(st, name, NULL);
+		if (val == NULL)
+		{
+			pg_free(name);
+			p++;
 			continue;
 		}
 
@@ -4449,6 +4543,7 @@ create_sql_command(PQExpBuffer buf, const char *source)
 	my_command = (Command *) pg_malloc(sizeof(Command));
 	initPQExpBuffer(&my_command->lines);
 	appendPQExpBufferStr(&my_command->lines, p);
+	my_command->substituted = false;
 	my_command->first_line = NULL;	/* this is set later */
 	my_command->type = SQL_COMMAND;
 	my_command->meta = META_NONE;
@@ -4456,6 +4551,7 @@ create_sql_command(PQExpBuffer buf, const char *source)
 	memset(my_command->argv, 0, sizeof(my_command->argv));
 	my_command->varprefix = NULL;	/* allocated later, if needed */
 	my_command->expr = NULL;
+	pthread_mutex_init(&my_command->mutex, NULL);
 	initSimpleStats(&my_command->stats);
 
 	return my_command;
@@ -4496,20 +4592,11 @@ postprocess_sql_command(Command *my_command)
 	buffer[strcspn(buffer, "\n\r")] = '\0';
 	my_command->first_line = pg_strdup(buffer);
 
-	/* parse query if necessary */
-	switch (querymode)
+	/* adjust argv[0] on simple query */
+	if (querymode == QUERY_SIMPLE)
 	{
-		case QUERY_SIMPLE:
-			my_command->argv[0] = my_command->lines.data;
-			my_command->argc++;
-			break;
-		case QUERY_EXTENDED:
-		case QUERY_PREPARED:
-			if (!parseQuery(my_command))
-				exit(1);
-			break;
-		default:
-			exit(1);
+		my_command->argv[0] = my_command->lines.data;
+		my_command->argc++;
 	}
 }
 
@@ -4893,6 +4980,15 @@ ParseScript(const char *script, const char *desc, int weight)
 	psql_scan_destroy(sstate);
 }
 
+static int
+number_of_commands(ParsedScript *ps)
+{
+	int i = 0;
+	while (ps->commands[i] != NULL)
+		i++;
+	return i + 1;
+}
+
 /*
  * Read the entire contents of file fd, and return it in a malloc'd buffer.
  *
@@ -6018,6 +6114,9 @@ main(int argc, char **argv)
 	{
 		state[i].cstack = conditional_stack_create();
 		initRandomState(&state[i].cs_func_rs);
+		for (int s = 0; s < num_scripts; s++)
+			state[i].prepared[s] =
+				pg_malloc0(sizeof(bool) * number_of_commands(&sql_script[s]));
 	}
 
 	pg_log_debug("pghost: %s pgport: %s nclients: %d %s: %d dbName: %s",
@@ -6809,4 +6908,27 @@ pthread_join(pthread_t th, void **thread_return)
 	return 0;
 }
 
+static void
+pthread_mutex_init(pthread_mutex_t *pm, void *unused)
+{
+	*pm = CreateMutex(NULL, FALSE, NULL)
+}
+
+static void
+pthread_mutex_lock(pthread_mutex_t *pm)
+{
+	WaitForSingleObject(*pm, INFINITE);
+}
+
+static void
+pthread_mutex_unlock(pthread_mutex_t *pm)
+{
+	ReleaseMutex(*pm);
+}
+
+static void
+pthread_mutex_destroy(pthread_mutex_t *pm)
+{
+	CloseHandle(*pm);
+}
 #endif							/* WIN32 */
diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl
index 52009c3524..986ab1bfa7 100644
--- a/src/bin/pgbench/t/001_pgbench_with_server.pl
+++ b/src/bin/pgbench/t/001_pgbench_with_server.pl
@@ -318,7 +318,7 @@ pgbench(
 	],
 	'server parameter logging',
 	{
-		'001_param_2' => q{select '1' as one \gset
+		'001_param_2' => q{select '1' as one, null as two \gset
 SELECT 1 / (random() / 2)::int, :one::int, :two::int;
 }
 	});
@@ -360,7 +360,7 @@ pgbench(
 	],
 	'server parameter logging',
 	{
-		'001_param_4' => q{select '1' as one \gset
+		'001_param_4' => q{select '1' as one, null as two \gset
 SELECT 1 / (random() / 2)::int, :one::int, :two::int;
 }
 	});
@@ -467,6 +467,8 @@ pgbench(
 		qr{command=98.: int 5432\b},                    # :random_seed
 		qr{command=99.: int -9223372036854775808\b},    # min int
 		qr{command=100.: int 9223372036854775807\b},    # max int
+		qr{command=103.: boolean true\b},
+		qr{command=105.: boolean true\b},
 	],
 	'pgbench expressions',
 	{
@@ -594,6 +596,13 @@ SELECT :v0, :v1, :v2, :v3;
 -- minint constant parsing
 \set min debug(-9223372036854775808)
 \set max debug(-(:min + 1))
+-- null handling
+\set n0 null
+SELECT NULL AS n1, 1 AS one \gset
+\set n2 debug(:n0 is null and :n1 is null and :one is not null)
+-- undefined variables are not substituted
+SELECT ':no_such_variable' = ':' || 'no_such_variable' AS eq \gset
+\set ok debug(:eq)
 }
 	});