database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-12-21 05:21:08 +03:00
Code Activity

psql: Add support for pipelines

Browse Source 
With \bind, \parse, \bind_named and \close, it is possible to issue
queries from psql using the extended protocol.  However, it was not
possible to send these queries using libpq's pipeline mode.  This
feature has two advantages:
- Testing.  Pipeline tests were only possible with pgbench, using TAP
tests.  It now becomes possible to have more SQL tests that are able to
stress the backend with pipelines and extended queries.  More tests will
be added in a follow-up commit that were discussed on some other
threads.  Some external projects in the community had to implement their
own facility to work around this limitation.
- Emulation of custom workloads, with more control over the actions
taken by a client with libpq APIs.  It is possible to emulate more
workload patterns to bottleneck the backend with the extended query
protocol.

This patch adds six new meta-commands to be able to control pipelines:
* \startpipeline starts a new pipeline.  All extended queries are queued
until the end of the pipeline are reached or a sync request is sent and
processed.
* \endpipeline ends an existing pipeline.  All queued commands are sent
to the server and all responses are processed by psql.
* \syncpipeline queues a synchronisation request, without flushing the
commands to the server, equivalent of PQsendPipelineSync().
* \flush, equivalent of PQflush().
* \flushrequest, equivalent of PQsendFlushRequest()
* \getresults reads the server's results for the queries in a pipeline.
Unsent data is automatically pushed when \getresults is called.  It is
possible to control the number of results read in a single meta-command
execution with an optional parameter, 0 means that all the results
should be read.

Author: Anthonin Bonnefoy <anthonin.bonnefoy@datadoghq.com>
Reviewed-by: Jelte Fennema-Nio <postgres@jeltef.nl>
Reviewed-by: Kirill Reshke <reshkekirill@gmail.com>
Discussion: https://postgr.es/m/CAO6_XqroE7JuMEm1sWz55rp9fAYX2JwmcP_3m_v51vnOFdsLiQ@mail.gmail.com
This commit is contained in:
Michael Paquier
2025-02-21 11:19:59 +09:00
parent 40af897eb7
commit 41625ab8ea
11 changed files with 1497 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

170
src/bin/psql/command.c
View File

@@ -90,9 +90,12 @@ static backslashResult exec_command_else(PsqlScanState scan_state, ConditionalSt
PQExpBuffer query_buf);
static backslashResult exec_command_endif(PsqlScanState scan_state, ConditionalStack cstack,
PQExpBuffer query_buf);
static backslashResult exec_command_endpipeline(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_encoding(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_errverbose(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_f(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_flush(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_flushrequest(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_g(PsqlScanState scan_state, bool active_branch,
const char *cmd);
static backslashResult process_command_g_options(char *first_option,
@@ -103,6 +106,7 @@ static backslashResult exec_command_gdesc(PsqlScanState scan_state, bool active_
static backslashResult exec_command_getenv(PsqlScanState scan_state, bool active_branch,
const char *cmd);
static backslashResult exec_command_gexec(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_getresults(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_gset(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_help(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_html(PsqlScanState scan_state, bool active_branch);
@@ -132,6 +136,8 @@ static backslashResult exec_command_setenv(PsqlScanState scan_state, bool active
const char *cmd);
static backslashResult exec_command_sf_sv(PsqlScanState scan_state, bool active_branch,
const char *cmd, bool is_func);
static backslashResult exec_command_startpipeline(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_syncpipeline(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_t(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_T(PsqlScanState scan_state, bool active_branch);
static backslashResult exec_command_timing(PsqlScanState scan_state, bool active_branch);
@@ -351,18 +357,26 @@ exec_command(const char *cmd,
status = exec_command_else(scan_state, cstack, query_buf);
else if (strcmp(cmd, "endif") == 0)
status = exec_command_endif(scan_state, cstack, query_buf);
else if (strcmp(cmd, "endpipeline") == 0)
status = exec_command_endpipeline(scan_state, active_branch);
else if (strcmp(cmd, "encoding") == 0)
status = exec_command_encoding(scan_state, active_branch);
else if (strcmp(cmd, "errverbose") == 0)
status = exec_command_errverbose(scan_state, active_branch);
else if (strcmp(cmd, "f") == 0)
status = exec_command_f(scan_state, active_branch);
else if (strcmp(cmd, "flush") == 0)
status = exec_command_flush(scan_state, active_branch);
else if (strcmp(cmd, "flushrequest") == 0)
status = exec_command_flushrequest(scan_state, active_branch);
else if (strcmp(cmd, "g") == 0 || strcmp(cmd, "gx") == 0)
status = exec_command_g(scan_state, active_branch, cmd);
else if (strcmp(cmd, "gdesc") == 0)
status = exec_command_gdesc(scan_state, active_branch);
else if (strcmp(cmd, "getenv") == 0)
status = exec_command_getenv(scan_state, active_branch, cmd);
else if (strcmp(cmd, "getresults") == 0)
status = exec_command_getresults(scan_state, active_branch);
else if (strcmp(cmd, "gexec") == 0)
status = exec_command_gexec(scan_state, active_branch);
else if (strcmp(cmd, "gset") == 0)
@@ -411,6 +425,10 @@ exec_command(const char *cmd,
status = exec_command_sf_sv(scan_state, active_branch, cmd, true);
else if (strcmp(cmd, "sv") == 0 || strcmp(cmd, "sv+") == 0)
status = exec_command_sf_sv(scan_state, active_branch, cmd, false);
else if (strcmp(cmd, "startpipeline") == 0)
status = exec_command_startpipeline(scan_state, active_branch);
else if (strcmp(cmd, "syncpipeline") == 0)
status = exec_command_syncpipeline(scan_state, active_branch);
else if (strcmp(cmd, "t") == 0)
status = exec_command_t(scan_state, active_branch);
else if (strcmp(cmd, "T") == 0)
@@ -1515,6 +1533,44 @@ exec_command_f(PsqlScanState scan_state, bool active_branch)
return success ? PSQL_CMD_SKIP_LINE : PSQL_CMD_ERROR;
}
/*
* \flush -- call PQflush() on the connection
*/
static backslashResult
exec_command_flush(PsqlScanState scan_state, bool active_branch)
{
backslashResult status = PSQL_CMD_SKIP_LINE;
if (active_branch)
{
pset.send_mode = PSQL_SEND_FLUSH;
status = PSQL_CMD_SEND;
}
else
ignore_slash_options(scan_state);
return status;
}
/*
* \flushrequest -- call PQsendFlushRequest() on the connection
*/
static backslashResult
exec_command_flushrequest(PsqlScanState scan_state, bool active_branch)
{
backslashResult status = PSQL_CMD_SKIP_LINE;
if (active_branch)
{
pset.send_mode = PSQL_SEND_FLUSH_REQUEST;
status = PSQL_CMD_SEND;
}
else
ignore_slash_options(scan_state);
return status;
}
/*
* \g [(pset-option[=pset-value] ...)] [filename/shell-command]
* \gx [(pset-option[=pset-value] ...)] [filename/shell-command]
@@ -1550,6 +1606,14 @@ exec_command_g(PsqlScanState scan_state, bool active_branch, const char *cmd)
if (status == PSQL_CMD_SKIP_LINE && active_branch)
{
if (strcmp(cmd, "gx") == 0 &&
PQpipelineStatus(pset.db) != PQ_PIPELINE_OFF)
{
pg_log_error("\\gx not allowed in pipeline mode");
clean_extended_state();
return PSQL_CMD_ERROR;
}
if (!fname)
pset.gfname = NULL;
else
@@ -1703,6 +1767,42 @@ exec_command_getenv(PsqlScanState scan_state, bool active_branch,
return success ? PSQL_CMD_SKIP_LINE : PSQL_CMD_ERROR;
}
/*
* \getresults -- read results
*/
static backslashResult
exec_command_getresults(PsqlScanState scan_state, bool active_branch)
{
backslashResult status = PSQL_CMD_SKIP_LINE;
if (active_branch)
{
char *opt;
int num_results;
pset.send_mode = PSQL_SEND_GET_RESULTS;
status = PSQL_CMD_SEND;
opt = psql_scan_slash_option(scan_state, OT_NORMAL, NULL, false);
pset.requested_results = 0;
if (opt != NULL)
{
num_results = atoi(opt);
if (num_results < 0)
{
pg_log_error("\\getresults: invalid number of requested results");
return PSQL_CMD_SKIP_LINE;
}
pset.requested_results = num_results;
}
}
else
ignore_slash_options(scan_state);
return status;
}
/*
* \gexec -- send query and execute each field of result
*/
@@ -1713,6 +1813,12 @@ exec_command_gexec(PsqlScanState scan_state, bool active_branch)
if (active_branch)
{
if (PQpipelineStatus(pset.db) != PQ_PIPELINE_OFF)
{
pg_log_error("\\gexec not allowed in pipeline mode");
clean_extended_state();
return PSQL_CMD_ERROR;
}
pset.gexec_flag = true;
status = PSQL_CMD_SEND;
}
@@ -1733,6 +1839,13 @@ exec_command_gset(PsqlScanState scan_state, bool active_branch)
char *prefix = psql_scan_slash_option(scan_state,
OT_NORMAL, NULL, false);
if (PQpipelineStatus(pset.db) != PQ_PIPELINE_OFF)
{
pg_log_error("\\gset not allowed in pipeline mode");
clean_extended_state();
return PSQL_CMD_ERROR;
}
if (prefix)
pset.gset_prefix = prefix;
else
@@ -2718,6 +2831,63 @@ exec_command_sf_sv(PsqlScanState scan_state, bool active_branch,
return status;
}
/*
* \startpipeline -- enter pipeline mode
*/
static backslashResult
exec_command_startpipeline(PsqlScanState scan_state, bool active_branch)
{
backslashResult status = PSQL_CMD_SKIP_LINE;
if (active_branch)
{
pset.send_mode = PSQL_SEND_START_PIPELINE_MODE;
status = PSQL_CMD_SEND;
}
else
ignore_slash_options(scan_state);
return status;
}
/*
* \syncpipeline -- send a sync message to an active pipeline
*/
static backslashResult
exec_command_syncpipeline(PsqlScanState scan_state, bool active_branch)
{
backslashResult status = PSQL_CMD_SKIP_LINE;
if (active_branch)
{
pset.send_mode = PSQL_SEND_PIPELINE_SYNC;
status = PSQL_CMD_SEND;
}
else
ignore_slash_options(scan_state);
return status;
}
/*
* \endpipeline -- end pipeline mode
*/
static backslashResult
exec_command_endpipeline(PsqlScanState scan_state, bool active_branch)
{
backslashResult status = PSQL_CMD_SKIP_LINE;
if (active_branch)
{
pset.send_mode = PSQL_SEND_END_PIPELINE_MODE;
status = PSQL_CMD_SEND;
}
else
ignore_slash_options(scan_state);
return status;
}
/*
* \t -- turn off table headers and row count
*/