Subject: Re: pgbench: using prepared BEGIN statement in a pipeline could cause an error

From: Kyotaro Horiguchi <[email-masked]>
Date: Thu, 27 Jan 2022 17:50:25 +0900 (JST)

At Tue, 16 Nov 2021 02:26:43 +0900, Yugo NAGATA <[email-masked]> wrote in 
> Thank you for pointing it out!
> I attached the updated patch.

I think we want more elabolative comment for the new place of
preparing as you mentioned in the first mail.

At Fri, 16 Jul 2021 15:30:13 +0900, Yugo NAGATA <[email-masked]> wrote in 
> One way to avoid these errors is to send Parse messages before
> pipeline mode starts.

-- 
Kyotaro Horiguchi
NTT Open Source Software Center

---

From: Yugo NAGATA <[email-masked]>
Date: Tue, 1 Mar 2022 15:55:59 +0900

Hi Horiguchi-san,

On Thu, 27 Jan 2022 17:50:25 +0900 (JST)

> preparing as you mentioned in the first mail.

Thank you for your suggestion.

I added comments on the prepareCommands() call as in the updated patch.

Regards,
Yugo Nagata

Yugo NAGATA <[email-masked]>
diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c
index f166a77e3a..c5893eabe4 100644
--- a/src/bin/pgbench/pgbench.c
+++ b/src/bin/pgbench/pgbench.c
@@ -2832,6 +2832,30 @@ chooseScript(TState *thread)
 	return i - 1;
 }
 
+/* Prepare SQL commands in the chosen script */
+static void
+prepareCommands(CState *st)
+{
+	int			j;
+	Command   **commands = sql_script[st->use_file].commands;
+
+	for (j = 0; commands[j] != NULL; j++)
+	{
+		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);
+	}
+	st->prepared[st->use_file] = true;
+}
+
 /* Send a SQL command, using the chosen querymode */
 static bool
 sendCommand(CState *st, Command *command)
@@ -2865,42 +2889,6 @@ sendCommand(CState *st, Command *command)
 		char		name[MAX_PREPARE_NAME];
 		const char *params[MAX_ARGS];
 
-		if (!st->prepared[st->use_file])
-		{
-			int			j;
-			Command   **commands = sql_script[st->use_file].commands;
-
-			for (j = 0; commands[j] != NULL; j++)
-			{
-				PGresult   *res;
-				char		name[MAX_PREPARE_NAME];
-
-				if (commands[j]->type != SQL_COMMAND)
-					continue;
-				preparedStatementName(name, st->use_file, j);
-				if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF)
-				{
-					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);
-				}
-				else
-				{
-					/*
-					 * In pipeline mode, we use asynchronous functions. If a
-					 * server-side error occurs, it will be processed later
-					 * among the other results.
-					 */
-					if (!PQsendPrepare(st->con, name,
-									   commands[j]->argv[0], commands[j]->argc - 1, NULL))
-						pg_log_error("%s", PQerrorMessage(st->con));
-				}
-			}
-			st->prepared[st->use_file] = true;
-		}
-
 		getQueryParams(st, command, params);
 		preparedStatementName(name, st->use_file, st->command);
 
@@ -3172,6 +3160,20 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg)
 					memset(st->prepared, 0, sizeof(st->prepared));
 				}
 
+				/*
+				 * Prepare SQL commands if needed.
+				 *
+				 * We should send Parse messages before executing meta commands
+				 * especially /startpipeline. If a Parse message is sent in
+				 * pipeline mode, a transaction starts before BEGIN is sent, and
+				 * it could be a problem. For example, "BEGIN ISOLATION LEVEL
+				 * SERIALIZABLE" is sent after a transaction starts, the error
+				 * "ERROR:  SET TRANSACTION ISOLATION LEVEL must be called before any query"
+				 * occurs.
+				 */
+				if (querymode == QUERY_PREPARED && !st->prepared[st->use_file])
+					prepareCommands(st);
+
 				/* record transaction start time */
 				st->txn_begin = now;
 
diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl
index f1341092fe..7e4ec728e8 100644
--- a/src/bin/pgbench/t/001_pgbench_with_server.pl
+++ b/src/bin/pgbench/t/001_pgbench_with_server.pl
@@ -837,6 +837,23 @@ select 1 \gset f
 }
 	});
 
+# Working \startpipeline in prepared query mode with serializable
+$node->pgbench(
+	'-t 1 -n -M prepared',
+	0,
+	[ qr{type: .*/001_pgbench_pipeline_serializable}, qr{actually processed: 1/1} ],
+	[],
+	'working \startpipeline with serializable',
+	{
+		'001_pgbench_pipeline_serializable' => q{
+-- test startpipeline with serializable
+\startpipeline
+BEGIN ISOLATION LEVEL SERIALIZABLE;
+} . "select 1;\n" x 10 . q{
+END;
+\endpipeline
+}
+	});
 
 # trigger many expression errors
 my @errors = (