worker_spi_procedure.patch

diff --git a/src/test/modules/worker_spi/worker_spi.c b/src/test/modules/worker_spi/worker_spi.c
index d0acef2652..0b9e4e9335 100644
--- a/src/test/modules/worker_spi/worker_spi.c
+++ b/src/test/modules/worker_spi/worker_spi.c
@@ -108,8 +108,20 @@ initialize_worker_spi(worktable *table)
 						 "		type text CHECK (type IN ('total', 'delta')), "
 						 "		value	integer)"
 						 "CREATE UNIQUE INDEX \"%s_unique_total\" ON \"%s\" (type) "
-						 "WHERE type = 'total'",
-						 table->schema, table->name, table->name, table->name);
+						 "WHERE type = 'total'; "
+						 "CREATE PROCEDURE \"%s\".\"%s_proc\"() AS $$ "
+						 "DECLARE "
+						 "  i INTEGER; "
+						 "BEGIN "
+						 "  FOR i IN 1..10 "
+						 "  LOOP "
+						 "    INSERT INTO \"%s\".\"%s\" VALUES ('delta', i); "
+						 "    COMMIT; "
+						 "  END LOOP; "
+						 "END; "
+						 "$$ LANGUAGE plpgsql; ",
+						 table->schema, table->name, table->name, table->name,
+						 table->schema, table->name, table->schema, table->name);
 
 		/* set statement start time */
 		SetCurrentStatementStartTimestamp();
@@ -168,20 +180,8 @@ worker_spi_main(Datum main_arg)
 	table->name = quote_identifier(table->name);
 
 	initStringInfo(&buf);
-	appendStringInfo(&buf,
-					 "WITH deleted AS (DELETE "
-					 "FROM %s.%s "
-					 "WHERE type = 'delta' RETURNING value), "
-					 "total AS (SELECT coalesce(sum(value), 0) as sum "
-					 "FROM deleted) "
-					 "UPDATE %s.%s "
-					 "SET value = %s.value + total.sum "
-					 "FROM total WHERE type = 'total' "
-					 "RETURNING %s.value",
-					 table->schema, table->name,
-					 table->schema, table->name,
-					 table->name,
-					 table->name);
+
+	appendStringInfo(&buf, "CALL \"%s\".\"%s_proc\"()", table->schema, table->name);
 
 	/*
 	 * Main loop: do this until SIGTERM is received and processed by
@@ -232,7 +232,7 @@ worker_spi_main(Datum main_arg)
 		 */
 		SetCurrentStatementStartTimestamp();
 		StartTransactionCommand();
-		SPI_connect();
+		SPI_connect_ext(SPI_OPT_NONATOMIC);
 		PushActiveSnapshot(GetTransactionSnapshot());
 		debug_query_string = buf.data;
 		pgstat_report_activity(STATE_RUNNING, buf.data);
@@ -240,30 +240,15 @@ worker_spi_main(Datum main_arg)
 		/* We can now execute queries via SPI */
 		ret = SPI_execute(buf.data, false, 0);
 
-		if (ret != SPI_OK_UPDATE_RETURNING)
-			elog(FATAL, "cannot select from table %s.%s: error code %d",
-				 table->schema, table->name, ret);
-
-		if (SPI_processed > 0)
-		{
-			bool		isnull;
-			int32		val;
-
-			val = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[0],
-											  SPI_tuptable->tupdesc,
-											  1, &isnull));
-			if (!isnull)
-				elog(LOG, "%s: count in %s.%s is now %d",
-					 MyBgworkerEntry->bgw_name,
-					 table->schema, table->name, val);
-		}
+		if (ret != SPI_OK_UTILITY)
+			elog(FATAL, "failed to call procedure");
 
 		/*
 		 * And finish our transaction.
 		 */
-		SPI_finish();
 		PopActiveSnapshot();
 		CommitTransactionCommand();
+		SPI_finish();
 		debug_query_string = NULL;
 		pgstat_report_stat(false);
 		pgstat_report_activity(STATE_IDLE, NULL);