agora inbox for [email protected]help / color / mirror / Atom feed
[PATCH v37 2/2] Add libpq pipeline mode support to pgbench 31+ messages / 3 participants [nested] [flat]
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v37 2/2] Add libpq pipeline mode support to pgbench @ 2021-03-15 18:07 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Alvaro Herrera @ 2021-03-15 18:07 UTC (permalink / raw) Author: Daniel V�rit� <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/bin/pgbench/pgbench.c | 128 +++++++++++++++++-- src/bin/pgbench/t/001_pgbench_with_server.pl | 61 +++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index f6a214669c..ba7b35d83c 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -395,10 +395,11 @@ typedef enum * * CSTATE_START_COMMAND starts the execution of a command. On a SQL * command, the command is sent to the server, and we move to - * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set, - * and we enter the CSTATE_SLEEP state to wait for it to expire. Other - * meta-commands are executed immediately. If the command about to start - * is actually beyond the end of the script, advance to CSTATE_END_TX. + * CSTATE_WAIT_RESULT state unless in pipeline mode. On a \sleep + * meta-command, the timer is set, and we enter the CSTATE_SLEEP state to + * wait for it to expire. Other meta-commands are executed immediately. If + * the command about to start is actually beyond the end of the script, + * advance to CSTATE_END_TX. * * CSTATE_WAIT_RESULT waits until we get a result set back from the server * for the current command. @@ -530,7 +531,9 @@ typedef enum MetaCommand META_IF, /* \if */ META_ELIF, /* \elif */ META_ELSE, /* \else */ - META_ENDIF /* \endif */ + META_ENDIF, /* \endif */ + META_STARTPIPELINE, /* \startpipeline */ + META_ENDPIPELINE /* \endpipeline */ } MetaCommand; typedef enum QueryMode @@ -2568,6 +2571,10 @@ getMetaCommand(const char *cmd) mc = META_GSET; else if (pg_strcasecmp(cmd, "aset") == 0) mc = META_ASET; + else if (pg_strcasecmp(cmd, "startpipeline") == 0) + mc = META_STARTPIPELINE; + else if (pg_strcasecmp(cmd, "endpipeline") == 0) + mc = META_ENDPIPELINE; else mc = META_NONE; return mc; @@ -2757,11 +2764,25 @@ sendCommand(CState *st, Command *command) 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 (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; } @@ -2805,8 +2826,10 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) * varprefix should be set only with \gset or \aset, and SQL commands do * not need it. */ +#if 0 Assert((meta == META_NONE && varprefix == NULL) || ((meta == META_GSET || meta == META_ASET) && varprefix != NULL)); +#endif res = PQgetResult(st->con); @@ -2874,6 +2897,13 @@ readCommandResponse(CState *st, MetaCommand meta, char *varprefix) /* otherwise the result is simply thrown away by PQclear below */ break; + case PGRES_PIPELINE_SYNC: + pg_log_debug("client %d pipeline ending", st->id); + if (PQexitPipelineMode(st->con) != 1) + pg_log_error("client %d failed to exit pipeline mode: %s", st->id, + PQerrorMessage(st->con)); + break; + default: /* anything else is unexpected */ pg_log_error("client %d script %d aborted in command %d query %d: %s", @@ -3127,13 +3157,36 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) /* Execute the command */ if (command->type == SQL_COMMAND) { + /* disallow \aset and \gset in pipeline mode */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + if (command->meta == META_GSET) + { + commandFailed(st, "gset", "\\gset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + else if (command->meta == META_ASET) + { + commandFailed(st, "aset", "\\aset is not allowed in pipeline mode"); + st->state = CSTATE_ABORTED; + break; + } + } + if (!sendCommand(st, command)) { commandFailed(st, "SQL", "SQL command send failed"); st->state = CSTATE_ABORTED; } else - st->state = CSTATE_WAIT_RESULT; + { + /* Wait for results, unless in pipeline mode */ + if (PQpipelineStatus(st->con) == PQ_PIPELINE_OFF) + st->state = CSTATE_WAIT_RESULT; + else + st->state = CSTATE_END_COMMAND; + } } else if (command->type == META_COMMAND) { @@ -3273,7 +3326,15 @@ advanceConnectionState(TState *thread, CState *st, StatsData *agg) if (readCommandResponse(st, sql_script[st->use_file].commands[st->command]->meta, sql_script[st->use_file].commands[st->command]->varprefix)) - st->state = CSTATE_END_COMMAND; + { + /* + * outside of pipeline mode: stop reading results. + * pipeline mode: continue reading results until an + * end-of-pipeline response. + */ + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + st->state = CSTATE_END_COMMAND; + } else st->state = CSTATE_ABORTED; break; @@ -3516,6 +3577,45 @@ executeMetaCommand(CState *st, pg_time_usec_t *now) return CSTATE_ABORTED; } } + else if (command->meta == META_STARTPIPELINE) + { + /* + * In pipeline mode, we use a workflow based on libpq pipeline + * functions. + */ + if (querymode == QUERY_SIMPLE) + { + commandFailed(st, "startpipeline", "cannot use pipeline mode with the simple query protocol"); + return CSTATE_ABORTED; + } + + if (PQpipelineStatus(st->con) != PQ_PIPELINE_OFF) + { + commandFailed(st, "startpipeline", "already in pipeline mode"); + return CSTATE_ABORTED; + } + if (PQenterPipelineMode(st->con) == 0) + { + commandFailed(st, "startpipeline", "failed to enter pipeline mode"); + return CSTATE_ABORTED; + } + } + else if (command->meta == META_ENDPIPELINE) + { + if (PQpipelineStatus(st->con) != PQ_PIPELINE_ON) + { + commandFailed(st, "endpipeline", "not in pipeline mode"); + return CSTATE_ABORTED; + } + if (!PQpipelineSync(st->con)) + { + commandFailed(st, "endpipeline", "failed to send a pipeline sync"); + return CSTATE_ABORTED; + } + /* Now wait for the PGRES_PIPELINE_SYNC and exit pipeline mode there */ + /* collect pending results before getting out of pipeline mode */ + return CSTATE_WAIT_RESULT; + } /* * executing the expression or shell command might have taken a @@ -4725,7 +4825,9 @@ process_backslash_command(PsqlScanState sstate, const char *source) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], "missing command", NULL, -1); } - else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF) + else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF || + my_command->meta == META_STARTPIPELINE || + my_command->meta == META_ENDPIPELINE) { if (my_command->argc != 1) syntax_error(source, lineno, my_command->first_line, my_command->argv[0], diff --git a/src/bin/pgbench/t/001_pgbench_with_server.pl b/src/bin/pgbench/t/001_pgbench_with_server.pl index daffc18e52..d07b36faa7 100644 --- a/src/bin/pgbench/t/001_pgbench_with_server.pl +++ b/src/bin/pgbench/t/001_pgbench_with_server.pl @@ -755,6 +755,67 @@ pgbench( } }); +# Working \startpipeline +pgbench( + '-t 1 -n -M extended', + 0, + [ qr{type: .*/001_pgbench_pipeline}, qr{processed: 1/1} ], + [], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline' => q{ +-- test startpipeline +\startpipeline +} . "select 1;\n" x 10 . q{ +\endpipeline +} + }); + +# Try \startpipeline twice +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{already in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_2' => q{ +-- startpipeline twice +\startpipeline +\startpipeline +} + }); + +# Try to end a pipeline that hasn't started +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{not in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_3' => q{ +-- pipeline not started +\endpipeline +} + }); + +# Try \gset in pipeline mode +pgbench( + '-t 1 -n -M extended', + 2, + [], + [qr{gset is not allowed in pipeline mode}], + 'pgbench startpipeline command', + { + '001_pgbench_pipeline_gset' => q{ +\startpipeline +select 1 \gset f +\endpipeline +} + }); + + # trigger many expression errors my @errors = ( -- 2.20.1 --BXVAT5kNtrzKuDFl-- ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH 3/4] Reindent and wrap between added sect1 tags @ 2022-02-22 02:21 Chapman Flack <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Chapman Flack @ 2022-02-22 02:21 UTC (permalink / raw) A whitespace-only commit. --- doc/src/sgml/plhandler.sgml | 638 ++++++++++++++++++++++---------------------- 1 file changed, 321 insertions(+), 317 deletions(-) diff --git a/doc/src/sgml/plhandler.sgml b/doc/src/sgml/plhandler.sgml index 95e8515..003a6c9 100644 --- a/doc/src/sgml/plhandler.sgml +++ b/doc/src/sgml/plhandler.sgml @@ -57,385 +57,389 @@ <sect1 id="plhandler-call"> <title>Call handler function</title> - <para> - Every routine defined with a language name other than - <literal>internal</literal> (as defined in <xref linkend="xfunc-internal"/>) - or <literal>c</literal> (<xref linkend="xfunc-c"/>) will be called by - invoking the procedural language's call handler. - <footnote> - <para> - This is true even of routines with language name <literal>sql</literal>, - though, as a special case, that call handler has no entry in the system - catalogs. - </para> - </footnote> - It is the responsibility of - the call handler to execute the routine in a meaningful way, such - as by interpreting the supplied source text. - </para> - - <para> - The call handler is a <quote>normal</quote> user-defined function that must - be declared to <productname>PostgreSQL</productname> as taking no arguments - and returning the type <type>language_handler</type>. This - special pseudo-type identifies the function as a call handler and - prevents it from being called directly in SQL commands. - This handler must not require a call handler of its own, which makes - the predefined languages <literal>internal</literal> or <literal>c</literal> - the only choices for the handler's own declaration. - Typically, it will be a loadable function in language - <literal>c</literal>, as described in <xref linkend="xfunc-c"/>. - <footnote> - <para> - It may be implemented in a language other than C, as long as it can be - built into a loadable object with compatible calling conventions. - </para> - </footnote> - </para> - - <para> - The call handler is called in the same way as any other function: - It receives a pointer to a - <structname>FunctionCallInfoBaseData</structname> <type>struct</type> containing - argument values and information about the called routine, and it - is expected to return a <type>Datum</type> result (and possibly - set the <structfield>isnull</structfield> field of the - <structname>FunctionCallInfoBaseData</structname> structure, if it wishes - to return an SQL null result). The difference between a call - handler and an ordinary callee is that the - <structfield>flinfo->fn_oid</structfield> field of the - <structname>FunctionCallInfoBaseData</structname> structure will contain - the OID of the actual routine to be called, not of the call - handler itself. The call handler must use this field to determine - which routine to execute. Also, the passed argument list has - been set up according to the declaration of the target routine, - not of the call handler. - </para> - - <para> - It's up to the call handler to fetch the routine's defining - <classname>pg_proc</classname> row from the system catalog cache - to determine what to execute, what parameter and return types are expected, - and so on. - The <literal>AS</literal> clause from the - <command>CREATE FUNCTION</command> or <command>CREATE PROCEDURE</command> - command for the routine will be found - in the <literal>prosrc</literal> column of the - <classname>pg_proc</classname> row. This is commonly source - text in the procedural language, but in theory it could be something else, - such as a path name to a file, or anything else that tells the call handler - what to do in detail. - </para> - - <para> - The handler may also examine the passed - <structname>FunctionCallInfoBaseData</structname> structure for information - on the context of the call. If the procedural language will support - returning sets, the structure may contain a pointer to a - <structname>ReturnSetInfo</structname> structure for use as described in - <xref linkend="xfunc-c-return-set"/>. If the language will support triggers - or event triggers, the structure may hold a pointer to one of the structures - described in <xref linkend="trigger-interface"/> or - <xref linkend="event-trigger-interface"/>, and the procedural language - should provide some way for the called function to use the information - they carry. - </para> - - <sect2 id="plhandler-call-args"> - <title>Parameter and return type resolution</title> - <para> - A routine's statically-declared parameter types (and, for a function, - return type) are found in the <literal>proargtypes</literal> and - <literal>prorettype</literal> columns of the <classname>pg_proc</classname> - row. - If a routine has <literal>OUT</literal> parameters, those types are - included in the <literal>proallargtypes</literal> column, and their names - in <literal>proargnames</literal>. - Convenience functions declared in <filename>funcapi.h</filename> are - available for extracting that information. + Every routine defined with a language name other than + <literal>internal</literal> (as defined in + <xref linkend="xfunc-internal"/>) + or <literal>c</literal> (<xref linkend="xfunc-c"/>) will be called by + invoking the procedural language's call handler. + <footnote> + <para> + This is true even of routines with language name <literal>sql</literal>, + though, as a special case, that call handler has no entry in the system + catalogs. + </para> + </footnote> + It is the responsibility of + the call handler to execute the routine in a meaningful way, such + as by interpreting the supplied source text. </para> <para> - The statically-declared types may include polymorphic types that need - to be resolved according to the actual types present at the call site, - as described in <xref linkend="xfunc-c-polymorphism"/>. + The call handler is a <quote>normal</quote> user-defined function that must + be declared to <productname>PostgreSQL</productname> as taking no arguments + and returning the type <type>language_handler</type>. This + special pseudo-type identifies the function as a call handler and + prevents it from being called directly in SQL commands. + This handler must not require a call handler of its own, which makes + the predefined languages <literal>internal</literal> or + <literal>c</literal> the only choices for the handler's own declaration. + Typically, it will be a loadable function in language + <literal>c</literal>, as described in <xref linkend="xfunc-c"/>. + <footnote> + <para> + It may be implemented in a language other than C, as long as it can be + built into a loadable object with compatible calling conventions. + </para> + </footnote> </para> - </sect2> - - <sect2 id="plhandler-call-types"> - <title>Mapping to procedural language types</title> + <para> + The call handler is called in the same way as any other function: + It receives a pointer to a + <structname>FunctionCallInfoBaseData</structname> <type>struct</type> + containing argument values and information about the called routine, and it + is expected to return a <type>Datum</type> result (and possibly + set the <structfield>isnull</structfield> field of the + <structname>FunctionCallInfoBaseData</structname> structure, if it wishes + to return an SQL null result). The difference between a call + handler and an ordinary callee is that the + <structfield>flinfo->fn_oid</structfield> field of the + <structname>FunctionCallInfoBaseData</structname> structure will contain + the OID of the actual routine to be called, not of the call + handler itself. The call handler must use this field to determine + which routine to execute. Also, the passed argument list has + been set up according to the declaration of the target routine, + not of the call handler. + </para> <para> - Once the <productname>PostgreSQL</productname> types of any parameters - and results have been resolved, the handler must determine how it will - map their values to and from suitable types that exist in the procedural - language. + It's up to the call handler to fetch the routine's defining + <classname>pg_proc</classname> row from the system catalog cache + to determine what to execute, what parameter and return types are expected, + and so on. + The <literal>AS</literal> clause from the + <command>CREATE FUNCTION</command> or <command>CREATE PROCEDURE</command> + command for the routine will be found + in the <literal>prosrc</literal> column of the + <classname>pg_proc</classname> row. This is commonly source + text in the procedural language, but in theory it could be something else, + such as a path name to a file, or anything else that tells the call handler + what to do in detail. </para> <para> - The designer of a procedural language will typically document what types - will be supported and how they will be mapped, which could be as simple - as using every type's text input/output format to map it to the target - language's string type, or could directly map many types to corresponding - ones the target language provides. The handler function will implement - those rules. + The handler may also examine the passed + <structname>FunctionCallInfoBaseData</structname> structure for information + on the context of the call. If the procedural language will support + returning sets, the structure may contain a pointer to a + <structname>ReturnSetInfo</structname> structure for use as described in + <xref linkend="xfunc-c-return-set"/>. If the language will support triggers + or event triggers, the structure may hold a pointer to one of + the structures described in <xref linkend="trigger-interface"/> or + <xref linkend="event-trigger-interface"/>, and the procedural language + should provide some way for the called function to use the information + they carry. </para> - <sect3 id="plhandler-call-types-transform"> - <title>Type transforms</title> + <sect2 id="plhandler-call-args"> + <title>Parameter and return type resolution</title> <para> - Because <productname>PostgreSQL</productname> is extensible, and - an extension can easily supply new types, a procedural language handler - may encounter types it has no predefined mappings for, or only an awkward - default mapping such as to a text string. A procedural language can be - designed so that its type mappings are also extensible, and an extension - can add mappings between new <productname>PostgreSQL</productname> types - and suitable types in the target language. + A routine's statically-declared parameter types (and, for a function, + return type) are found in the <literal>proargtypes</literal> and + <literal>prorettype</literal> columns of the + <classname>pg_proc</classname> row. + If a routine has <literal>OUT</literal> parameters, those types are + included in the <literal>proallargtypes</literal> column, and their names + in <literal>proargnames</literal>. + Convenience functions declared in <filename>funcapi.h</filename> are + available for extracting that information. </para> <para> - One mechanism <productname>PostgreSQL</productname> provides that may be - used for that purpose is <xref linkend="sql-createtransform"/>. - The command associates a <productname>PostgreSQL</productname> type and - a specific procedural language with a pair of functions to handle the - mapping of that type to a corresponding procedural language type and back. + The statically-declared types may include polymorphic types that need + to be resolved according to the actual types present at the call site, + as described in <xref linkend="xfunc-c-polymorphism"/>. </para> - <para> - For a procedural language to support transforms, its call handler is - responsible for consulting the <literal>protrftypes</literal> column of - a routine's <classname>pg_proc</classname> row to determine which types - should have transforms applied. - A convenience function <function>get_call_trftypes</function> is - available. - The call handler must then resolve the <quote>from SQL</quote> function - for each affected parameter type, and the <quote>to SQL</quote> function - for any affected result. - It may use the <function>get_transform_fromsql</function> and - <function>get_transform_tosql</function> functions for that. - </para> + </sect2> - <para> - The handler must then apply the proper <quote>from SQL</quote> functions - to all affected inputs (including elements within array or composite - types) and, after calling the target routine, apply the proper - <quote>to SQL</quote> functions similarly to any results. - If the target routine might interact with the database using SPI, - the handler may arrange for the requested transforms to be applied - in those operations as well. - </para> + <sect2 id="plhandler-call-types"> + <title>Mapping to procedural language types</title> <para> - Because the procedural language implementation, and not - <productname>PostgreSQL</productname> itself, is responsible for calling - the transform functions, it is free to define what it will pass as the - parameter to each function (declared as <type>internal</type> for both), - and how it will interpret the result (also declared <type>internal</type>) - of the <quote>from SQL</quote> function. Effectively, each procedural - language's implementation defines the API that must be adhered to - by any author of transforms for that language. + Once the <productname>PostgreSQL</productname> types of any parameters + and results have been resolved, the handler must determine how it will + map their values to and from suitable types that exist in the procedural + language. </para> <para> - A procedural language might impose limits on where and how it will apply - transforms (such as on array or domain types). The - <function>get_transform_fromsql</function> and - <function>get_transform_tosql</function> functions mentioned above - consider each type only shallowly, and will not, for example, return - a transform function for a domain type if only its base type was listed in - the <literal>TRANSFORM</literal> clause. - If a procedural language's call handler does not implement transforms - at all, no <literal>TRANSFORM</literal> clause will have any effect - for routines declared in that language. - The language's validator function can be used to give immediate feedback - if a routine is declared with <literal>TRANSFORM</literal> clauses - the implementation cannot support. + The designer of a procedural language will typically document what types + will be supported and how they will be mapped, which could be as simple + as using every type's text input/output format to map it to the target + language's string type, or could directly map many types to corresponding + ones the target language provides. The handler function will implement + those rules. </para> - </sect3> - - </sect2> - - <sect2 id="plhandler-call-cache"> - <title>Caching resolved routine information</title> + <sect3 id="plhandler-call-types-transform"> + <title>Type transforms</title> + + <para> + Because <productname>PostgreSQL</productname> is extensible, and + an extension can easily supply new types, a procedural language handler + may encounter types it has no predefined mappings for, or only an awkward + default mapping such as to a text string. A procedural language can be + designed so that its type mappings are also extensible, and an extension + can add mappings between new <productname>PostgreSQL</productname> types + and suitable types in the target language. + </para> + + <para> + One mechanism <productname>PostgreSQL</productname> provides that may be + used for that purpose is <xref linkend="sql-createtransform"/>. + The command associates a <productname>PostgreSQL</productname> type and + a specific procedural language with a pair of functions to handle the + mapping of that type to a corresponding procedural language type + and back. + </para> + + <para> + For a procedural language to support transforms, its call handler is + responsible for consulting the <literal>protrftypes</literal> column of + a routine's <classname>pg_proc</classname> row to determine which types + should have transforms applied. + A convenience function <function>get_call_trftypes</function> is + available. + The call handler must then resolve the <quote>from SQL</quote> function + for each affected parameter type, and the <quote>to SQL</quote> function + for any affected result. + It may use the <function>get_transform_fromsql</function> and + <function>get_transform_tosql</function> functions for that. + </para> + + <para> + The handler must then apply the proper <quote>from SQL</quote> functions + to all affected inputs (including elements within array or composite + types) and, after calling the target routine, apply the proper + <quote>to SQL</quote> functions similarly to any results. + If the target routine might interact with the database using SPI, + the handler may arrange for the requested transforms to be applied + in those operations as well. + </para> + + <para> + Because the procedural language implementation, and not + <productname>PostgreSQL</productname> itself, is responsible for calling + the transform functions, it is free to define what it will pass as the + parameter to each function (declared as <type>internal</type> for both), + and how it will interpret the result (also declared + <type>internal</type>) of the <quote>from SQL</quote> function. + Effectively, each procedural language's implementation defines the API + that must be adhered to by any author of transforms for that language. + </para> + + <para> + A procedural language might impose limits on where and how it will apply + transforms (such as on array or domain types). The + <function>get_transform_fromsql</function> and + <function>get_transform_tosql</function> functions mentioned above + consider each type only shallowly, and will not, for example, return + a transform function for a domain type if only its base type was listed + in the <literal>TRANSFORM</literal> clause. + If a procedural language's call handler does not implement transforms + at all, no <literal>TRANSFORM</literal> clause will have any effect + for routines declared in that language. + The language's validator function can be used to give immediate feedback + if a routine is declared with <literal>TRANSFORM</literal> clauses + the implementation cannot support. + </para> + + </sect3> + + </sect2> + + <sect2 id="plhandler-call-cache"> + <title>Caching resolved routine information</title> - <para> - Often, the same routine is called many times per SQL statement. - A call handler can avoid repeated lookups of information about the - called routine by using the - <structfield>flinfo->fn_extra</structfield> field. This will - initially be <symbol>NULL</symbol>, but can be set by the call handler to point at - information about the called routine. On subsequent calls, if - <structfield>flinfo->fn_extra</structfield> is already non-<symbol>NULL</symbol> - then it can be used and the information lookup step skipped. The - call handler must make sure that - <structfield>flinfo->fn_extra</structfield> is made to point at - memory that will live at least until the end of the current query, - since an <structname>FmgrInfo</structname> data structure could be - kept that long. One way to do this is to allocate the extra data - in the memory context specified by - <structfield>flinfo->fn_mcxt</structfield>; such data will - normally have the same lifespan as the - <structname>FmgrInfo</structname> itself. But the handler could - also choose to use a longer-lived memory context so that it can cache - routine definition information across queries. - </para> + <para> + Often, the same routine is called many times per SQL statement. + A call handler can avoid repeated lookups of information about the + called routine by using the + <structfield>flinfo->fn_extra</structfield> field. This will + initially be <symbol>NULL</symbol>, but can be set by the call handler + to point at information about the called routine. On subsequent calls, + if <structfield>flinfo->fn_extra</structfield> is already + non-<symbol>NULL</symbol> then it can be used and the information lookup + step skipped. The call handler must make sure that + <structfield>flinfo->fn_extra</structfield> is made to point at + memory that will live at least until the end of the current query, + since an <structname>FmgrInfo</structname> data structure could be + kept that long. One way to do this is to allocate the extra data + in the memory context specified by + <structfield>flinfo->fn_mcxt</structfield>; such data will + normally have the same lifespan as the + <structname>FmgrInfo</structname> itself. But the handler could + also choose to use a longer-lived memory context so that it can cache + routine definition information across queries. + </para> - <para> - If the handler supports returning sets, and uses the ValuePerCall mode - helper macros described in <xref linkend="xfunc-c-return-set"/>, it must - not use <structfield>fn_extra</structfield> during set-returning calls. - The helper macros use that field for their own purposes. After - <literal>SRF_FIRSTCALL_INIT</literal> has been called, the field will point - to a <structname>FuncCallContext</structname> structure, which has - a <structfield>user_fctx</structfield> field that can be used similarly, - but only through the sequence of calls returning one set result. - </para> + <para> + If the handler supports returning sets, and uses the ValuePerCall mode + helper macros described in <xref linkend="xfunc-c-return-set"/>, it must + not use <structfield>fn_extra</structfield> during set-returning calls. + The helper macros use that field for their own purposes. After + <literal>SRF_FIRSTCALL_INIT</literal> has been called, the field will point + to a <structname>FuncCallContext</structname> structure, which has + a <structfield>user_fctx</structfield> field that can be used similarly, + but only through the sequence of calls returning one set result. + </para> - </sect2> + </sect2> </sect1> <sect1 id="plhandler-validator"> <title>Validator function</title> - <para> - If a validator is provided by a procedural language, it - must be declared as a function taking a single parameter of type - <type>oid</type>. The validator's result is ignored, so it is customarily - declared to return <type>void</type>. - The validator itself may be written in any procedural language able to - receive an <type>oid</type>-typed parameter and query system catalogs. - </para> + <para> + If a validator is provided by a procedural language, it + must be declared as a function taking a single parameter of type + <type>oid</type>. The validator's result is ignored, so it is customarily + declared to return <type>void</type>. + The validator itself may be written in any procedural language able to + receive an <type>oid</type>-typed parameter and query system catalogs. + </para> - <para> - The validator will be called at - the end of a <command>CREATE FUNCTION</command> or - <command>CREATE PROCEDURE</command> command that has created - or updated a routine written in the procedural language. - The passed-in OID is the OID of the routine's <classname>pg_proc</classname> - row. The validator must fetch this row in the usual way, and do - whatever checking is appropriate. - </para> + <para> + The validator will be called at + the end of a <command>CREATE FUNCTION</command> or + <command>CREATE PROCEDURE</command> command that has created + or updated a routine written in the procedural language. + The passed-in OID is the OID of the routine's + <classname>pg_proc</classname> row. The validator must fetch this row + in the usual way, and do whatever checking is appropriate. + </para> - <para> - First, call <function>CheckFunctionValidatorAccess()</function> to diagnose - explicit calls to the validator that the user could not achieve through - <command>CREATE FUNCTION</command> or <command>CREATE PROCEDURE</command>. - Typical checks then include verifying - that the routine's argument and result types are supported by the - language, and that the routine's body is syntactically correct - in the language. If the validator finds the routine to be okay, - it should just return. If it finds an error, it should report that - via the normal <function>ereport()</function> error reporting mechanism. - Throwing an error will force a transaction rollback and thus prevent - the incorrect routine definition from being committed. - </para> + <para> + First, call <function>CheckFunctionValidatorAccess()</function> + to diagnose explicit calls to the validator that the user could not achieve + through <command>CREATE FUNCTION</command> or + <command>CREATE PROCEDURE</command>. + Typical checks then include verifying + that the routine's argument and result types are supported by the + language, and that the routine's body is syntactically correct + in the language. If the validator finds the routine to be okay, + it should just return. If it finds an error, it should report that + via the normal <function>ereport()</function> error reporting mechanism. + Throwing an error will force a transaction rollback and thus prevent + the incorrect routine definition from being committed. + </para> - <para> - Validator functions should typically honor the <xref - linkend="guc-check-function-bodies"/> parameter: if it is turned off then - any expensive or context-sensitive checking should be skipped. If the - language provides for code execution at compilation time, the validator - must suppress checks that would induce such execution. In particular, - this parameter is turned off by <application>pg_dump</application> so that it can - load procedural language routines without worrying about side effects or - dependencies of the routine bodies on other database objects. - (Because of this requirement, the call handler should avoid - assuming that the validator has fully checked the routine. The point - of having a validator is not to let the call handler omit checks, but - to notify the user immediately if there are obvious errors in a - <command>CREATE FUNCTION</command> or <command>CREATE PROCEDURE</command> - command.) - </para> + <para> + Validator functions should typically honor the <xref + linkend="guc-check-function-bodies"/> parameter: if it is turned off then + any expensive or context-sensitive checking should be skipped. If the + language provides for code execution at compilation time, the validator + must suppress checks that would induce such execution. In particular, + this parameter is turned off by <application>pg_dump</application> so that + it can load procedural language routines without worrying about + side effects or dependencies of the routine bodies on other database + objects. + (Because of this requirement, the call handler should avoid + assuming that the validator has fully checked the routine. The point + of having a validator is not to let the call handler omit checks, but + to notify the user immediately if there are obvious errors in a + <command>CREATE FUNCTION</command> or <command>CREATE PROCEDURE</command> + command.) + </para> - <para> - While the choice of exactly what to check is mostly left to the - discretion of the validator function, note that the core - <command>CREATE FUNCTION</command> and <command>CREATE PROCEDURE</command> - code only executes <literal>SET</literal> clauses - attached to a routine when <varname>check_function_bodies</varname> is on. - Therefore, checks whose results might be affected by GUC parameters - definitely should be skipped when <varname>check_function_bodies</varname> is - off, to avoid false failures when reloading a dump. - </para> + <para> + While the choice of exactly what to check is mostly left to the + discretion of the validator function, note that the core + <command>CREATE FUNCTION</command> and <command>CREATE PROCEDURE</command> + code only executes <literal>SET</literal> clauses + attached to a routine when <varname>check_function_bodies</varname> is on. + Therefore, checks whose results might be affected by GUC parameters + definitely should be skipped when <varname>check_function_bodies</varname> + is off, to avoid false failures when reloading a dump. + </para> - <para> - If a language's call handler does not apply parameter and return type - transforms, then no <literal>TRANSFORM</literal> clause in a routine - declaration will have any effect. To provide immediate feedback if a - declaration contains such a clause, the validator can report a suitable - error whenever the <literal>protrftypes</literal> column of the routine's - <classname>pg_proc</classname> row is non-null. - </para> + <para> + If a language's call handler does not apply parameter and return type + transforms, then no <literal>TRANSFORM</literal> clause in a routine + declaration will have any effect. To provide immediate feedback if a + declaration contains such a clause, the validator can report a suitable + error whenever the <literal>protrftypes</literal> column of the routine's + <classname>pg_proc</classname> row is non-null. + </para> </sect1> <sect1 id="plhandler-inline"> <title>Inline handler function</title> - <para> - If this handler is provided by a procedural language, it - must be declared as a function taking a single parameter of type - <type>internal</type>, which will be a pointer - to an <structname>InlineCodeBlock</structname> struct when the handler - is called. The result is ignored, so the return type is customarily - declared as <type>void</type>. - The inline handler itself may be written in any procedural language that - permits declaring an <type>internal</type> parameter with a suitable - language binding for accessing it as an - <structname>InlineCodeBlock</structname> struct. - </para> + <para> + If this handler is provided by a procedural language, it + must be declared as a function taking a single parameter of type + <type>internal</type>, which will be a pointer + to an <structname>InlineCodeBlock</structname> struct when the handler + is called. The result is ignored, so the return type is customarily + declared as <type>void</type>. + The inline handler itself may be written in any procedural language that + permits declaring an <type>internal</type> parameter with a suitable + language binding for accessing it as an + <structname>InlineCodeBlock</structname> struct. + </para> - <para> - The inline handler - will be called when a <command>DO</command> statement is executed specifying - the procedural language. The <structname>InlineCodeBlock</structname> - struct contains information - about the <command>DO</command> statement's parameters, in particular the - text of the anonymous code block to be executed. - It also contains the OID of the intended procedural language and whether - that procedural language is declared as <literal>TRUSTED</literal>, useful - if a single inline handler is supporting more than one procedural language. - The inline handler should execute the code block and return. - </para> + <para> + The inline handler + will be called when a <command>DO</command> statement is executed + specifying the procedural language. + The <structname>InlineCodeBlock</structname> struct contains information + about the <command>DO</command> statement's parameters, in particular the + text of the anonymous code block to be executed. + It also contains the OID of the intended procedural language and whether + that procedural language is declared as <literal>TRUSTED</literal>, useful + if a single inline handler is supporting more than one procedural language. + The inline handler should execute the code block and return. + </para> </sect1> <sect1 id="plhandler-packaging"> <title>Packaging the language handlers</title> - <para> - It's recommended that you wrap all these function declarations, - as well as the <command>CREATE LANGUAGE</command> command itself, into - an <firstterm>extension</firstterm> so that a simple <command>CREATE EXTENSION</command> - command is sufficient to install the language. See - <xref linkend="extend-extensions"/> for information about writing - extensions. - </para> + <para> + It's recommended that you wrap all these function declarations, + as well as the <command>CREATE LANGUAGE</command> command itself, into + an <firstterm>extension</firstterm> so that a simple + <command>CREATE EXTENSION</command> command is sufficient to install + the language. See <xref linkend="extend-extensions"/> for information + about writing extensions. + </para> </sect1> <sect1 id="plhandler-examples"> <title>Example code</title> - <para> - A template for a procedural-language handler written as a C extension is - provided in <literal>src/test/modules/plsample</literal>. This is a - working sample demonstrating one way to create a procedural-language - handler, process parameters, and return a value. - </para> + <para> + A template for a procedural-language handler written as a C extension is + provided in <literal>src/test/modules/plsample</literal>. This is a + working sample demonstrating one way to create a procedural-language + handler, process parameters, and return a value. + </para> - <para> - The procedural languages included in the standard distribution - are good references when trying to write your own language handler. - Look into the <filename>src/pl</filename> subdirectory of the source tree. - The <xref linkend="sql-createlanguage"/> - reference page also has some useful details. - </para> + <para> + The procedural languages included in the standard distribution + are good references when trying to write your own language handler. + Look into the <filename>src/pl</filename> subdirectory of the source tree. + The <xref linkend="sql-createlanguage"/> + reference page also has some useful details. + </para> </sect1> -- 2.7.3 --------------070002080407000808050907 Content-Type: text/x-patch; name="4.patch" Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename="4.patch" ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v2 04/15] heapam: Use exclusive lock on old page in CLUSTER @ 2024-10-17 15:45 Andres Freund <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Andres Freund @ 2024-10-17 15:45 UTC (permalink / raw) To be able to guarantee that we can set the hint bit, acquire an exclusive lock on the old buffer. We need the hint bits to be set as otherwise reform_and_rewrite_tuple() -> rewrite_heap_tuple() -> heap_freeze_tuple() will get confused. It'd be better if we somehow coulda void setting hint bits on the old page. One reason to use VACUUM FULL are very bloated tables - rewriting most of the old table before during VACUUM FULL doesn't. Author: Reviewed-by: Discussion: https://postgr.es/m/ Backpatch: --- src/backend/access/heap/heapam_handler.c | 13 ++++++++++++- src/backend/access/heap/heapam_visibility.c | 7 +++++++ src/backend/storage/buffer/bufmgr.c | 2 +- 3 files changed, 20 insertions(+), 2 deletions(-) diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index a8d95e0f1c1..009601445aa 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -833,7 +833,18 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, tuple = ExecFetchSlotHeapTuple(slot, false, NULL); buf = hslot->buffer; - LockBuffer(buf, BUFFER_LOCK_SHARE); + /* + * To be able to guarantee that we can set the hint bit, acquire an + * exclusive lock on the old buffer. We need the hint bits to be set + * as otherwise reform_and_rewrite_tuple() -> rewrite_heap_tuple() -> + * heap_freeze_tuple() will get confused. + * + * It'd be better if we somehow could avoid setting hint bits on the + * old page. One reason to use VACUUM FULL are very bloated tables - + * rewriting most of the old table before during VACUUM FULL doesn't + * exactly help... + */ + LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE); switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf)) { diff --git a/src/backend/access/heap/heapam_visibility.c b/src/backend/access/heap/heapam_visibility.c index 057b88767db..b7aa8bb7a52 100644 --- a/src/backend/access/heap/heapam_visibility.c +++ b/src/backend/access/heap/heapam_visibility.c @@ -141,6 +141,13 @@ void HeapTupleSetHintBits(HeapTupleHeader tuple, Buffer buffer, uint16 infomask, TransactionId xid) { + /* + * The uses from heapam.c rely on being able to perform the hint bit + * updates, which can only be guaranteed if we are holding an exclusive + * lock on the buffer - which all callers are doing. + */ + Assert(BufferIsLocal(buffer) || BufferLockHeldByMe(buffer, BUFFER_LOCK_EXCLUSIVE)); + SetHintBits(tuple, buffer, infomask, xid); } diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c index 88d18e85d64..f56f7f9f441 100644 --- a/src/backend/storage/buffer/bufmgr.c +++ b/src/backend/storage/buffer/bufmgr.c @@ -5527,7 +5527,7 @@ BufferLockHeldByMe(Buffer buffer, int mode) { Assert(false); pg_unreachable(); - lwmode = LW_EXCLUSIVE; /* assuage compiler */ + lwmode = LW_EXCLUSIVE; /* assuage compiler */ } return LWLockHeldByMeInMode(BufferDescriptorGetContentLock(buf), lwmode); -- 2.45.2.746.g06e570c0df.dirty --ftvuw4gllm5odk5g Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v2-0005-heapam-Only-set-tuple-s-block-once-per-page-in-pa.patch" ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v1 04/14] heapam: Use exclusive lock on old page in CLUSTER @ 2024-10-17 15:45 Andres Freund <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Andres Freund @ 2024-10-17 15:45 UTC (permalink / raw) To be able to guarantee that we can set the hint bit, acquire an exclusive lock on the old buffer. We need the hint bits to be set as otherwise reform_and_rewrite_tuple() -> rewrite_heap_tuple() -> heap_freeze_tuple() will get confused. It'd be better if we somehow coulda void setting hint bits on the old page. One reason to use VACUUM FULL are very bloated tables - rewriting most of the old table before during VACUUM FULL doesn't. Author: Reviewed-by: Discussion: https://postgr.es/m/ Backpatch: --- src/backend/access/heap/heapam_handler.c | 13 ++++++++++++- src/backend/access/heap/heapam_visibility.c | 7 +++++++ src/backend/storage/buffer/bufmgr.c | 2 +- 3 files changed, 20 insertions(+), 2 deletions(-) diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index a8d95e0f1c1..009601445aa 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -833,7 +833,18 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, tuple = ExecFetchSlotHeapTuple(slot, false, NULL); buf = hslot->buffer; - LockBuffer(buf, BUFFER_LOCK_SHARE); + /* + * To be able to guarantee that we can set the hint bit, acquire an + * exclusive lock on the old buffer. We need the hint bits to be set + * as otherwise reform_and_rewrite_tuple() -> rewrite_heap_tuple() -> + * heap_freeze_tuple() will get confused. + * + * It'd be better if we somehow could avoid setting hint bits on the + * old page. One reason to use VACUUM FULL are very bloated tables - + * rewriting most of the old table before during VACUUM FULL doesn't + * exactly help... + */ + LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE); switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf)) { diff --git a/src/backend/access/heap/heapam_visibility.c b/src/backend/access/heap/heapam_visibility.c index 057b88767db..b7aa8bb7a52 100644 --- a/src/backend/access/heap/heapam_visibility.c +++ b/src/backend/access/heap/heapam_visibility.c @@ -141,6 +141,13 @@ void HeapTupleSetHintBits(HeapTupleHeader tuple, Buffer buffer, uint16 infomask, TransactionId xid) { + /* + * The uses from heapam.c rely on being able to perform the hint bit + * updates, which can only be guaranteed if we are holding an exclusive + * lock on the buffer - which all callers are doing. + */ + Assert(BufferIsLocal(buffer) || BufferLockHeldByMe(buffer, BUFFER_LOCK_EXCLUSIVE)); + SetHintBits(tuple, buffer, infomask, xid); } diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c index 88d18e85d64..f56f7f9f441 100644 --- a/src/backend/storage/buffer/bufmgr.c +++ b/src/backend/storage/buffer/bufmgr.c @@ -5527,7 +5527,7 @@ BufferLockHeldByMe(Buffer buffer, int mode) { Assert(false); pg_unreachable(); - lwmode = LW_EXCLUSIVE; /* assuage compiler */ + lwmode = LW_EXCLUSIVE; /* assuage compiler */ } return LWLockHeldByMeInMode(BufferDescriptorGetContentLock(buf), lwmode); -- 2.46.0.519.g2e7b89e038 --gpbnw6zzq3spol54 Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v1-0005-heapam-Only-set-tuple-s-block-once-per-page-in-pa.patch" ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v9 02/10] heapam: Use exclusive lock on old page in CLUSTER @ 2025-01-26 20:18 Andres Freund <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Andres Freund @ 2025-01-26 20:18 UTC (permalink / raw) To be able to guarantee that we can set the hint bit, acquire an exclusive lock on the old buffer. This is required as a future commit will only allow hint bits to be set with a new lock level, which is acquired as-needed in a non-blocking fashion. We need the hint bits, set in heapam_relation_copy_for_cluster() -> HeapTupleSatisfiesVacuum(), to be set, as otherwise reform_and_rewrite_tuple() -> rewrite_heap_tuple() will get confused. Specifically, rewrite_heap_tuple() checks for HEAP_XMAX_INVALID in the old tuple to determine whether to check the old-to-new mapping hash table. It'd be better if we somehow could avoid setting hint bits on the old page. A common reason to use VACUUM FULL are very bloated tables - rewriting most of the old table before during VACUUM FULL doesn't exactly help. Author: Reviewed-by: Discussion: https://postgr.es/m/ Backpatch: --- src/backend/access/heap/heapam_handler.c | 16 +++++++++++++++- src/backend/access/heap/heapam_visibility.c | 7 +++++++ src/backend/access/heap/rewriteheap.c | 3 +++ 3 files changed, 25 insertions(+), 1 deletion(-) diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index 09a456e9966..505faaaf9d2 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -837,7 +837,21 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, tuple = ExecFetchSlotHeapTuple(slot, false, NULL); buf = hslot->buffer; - LockBuffer(buf, BUFFER_LOCK_SHARE); + /* + * To be able to guarantee that we can set the hint bit, acquire an + * exclusive lock on the old buffer. We need the hint bits, set in + * heapam_relation_copy_for_cluster() -> HeapTupleSatisfiesVacuum(), + * to be set, as otherwise reform_and_rewrite_tuple() -> + * rewrite_heap_tuple() will get confused. Specifically, + * rewrite_heap_tuple() checks for HEAP_XMAX_INVALID in the old tuple + * to determine whether to check the old-to-new mapping hash table. + * + * It'd be better if we somehow could avoid setting hint bits on the + * old page. One reason to use VACUUM FULL are very bloated tables - + * rewriting most of the old table before during VACUUM FULL doesn't + * exactly help... + */ + LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE); switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf)) { diff --git a/src/backend/access/heap/heapam_visibility.c b/src/backend/access/heap/heapam_visibility.c index 05e70b7d92a..9a034d5c9e8 100644 --- a/src/backend/access/heap/heapam_visibility.c +++ b/src/backend/access/heap/heapam_visibility.c @@ -141,6 +141,13 @@ void HeapTupleSetHintBits(HeapTupleHeader tuple, Buffer buffer, uint16 infomask, TransactionId xid) { + /* + * The uses from heapam.c rely on being able to perform the hint bit + * updates, which can only be guaranteed if we are holding an exclusive + * lock on the buffer - which all callers are doing. + */ + Assert(BufferIsLockedByMeInMode(buffer, BUFFER_LOCK_EXCLUSIVE)); + SetHintBits(tuple, buffer, infomask, xid); } diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c index bae3a2da77a..77fd48eb59e 100644 --- a/src/backend/access/heap/rewriteheap.c +++ b/src/backend/access/heap/rewriteheap.c @@ -382,6 +382,9 @@ rewrite_heap_tuple(RewriteState state, /* * If the tuple has been updated, check the old-to-new mapping hash table. + * + * Note that this check relies on the HeapTupleSatisfiesVacuum() in + * heapam_relation_copy_for_cluster() to have set hint bits. */ if (!((old_tuple->t_data->t_infomask & HEAP_XMAX_INVALID) || HeapTupleHeaderIsOnlyLocked(old_tuple->t_data)) && -- 2.48.1.76.g4e746b1a31.dirty --rkiyqpij3ajqn7ww Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v9-0003-heapam-Add-batch-mode-mvcc-check-and-use-it-in-pa.patch" ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v6 10/14] heapam: Use exclusive lock on old page in CLUSTER @ 2025-01-26 20:18 Andres Freund <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Andres Freund @ 2025-01-26 20:18 UTC (permalink / raw) To be able to guarantee that we can set the hint bit, acquire an exclusive lock on the old buffer. We need the hint bits to be set as otherwise reform_and_rewrite_tuple() -> rewrite_heap_tuple() -> heap_freeze_tuple() will get confused. It'd be better if we somehow could avoid setting hint bits on the old page. A commonreason to use VACUUM FULL are very bloated tables - rewriting most of the old table before during VACUUM FULL doesn't exactly help. Author: Reviewed-by: Discussion: https://postgr.es/m/ Backpatch: --- src/backend/access/heap/heapam_handler.c | 13 ++++++++++++- src/backend/access/heap/heapam_visibility.c | 7 +++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index bcbac844bb6..f84254f0737 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -837,7 +837,18 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, tuple = ExecFetchSlotHeapTuple(slot, false, NULL); buf = hslot->buffer; - LockBuffer(buf, BUFFER_LOCK_SHARE); + /* + * To be able to guarantee that we can set the hint bit, acquire an + * exclusive lock on the old buffer. We need the hint bits to be set + * as otherwise reform_and_rewrite_tuple() -> rewrite_heap_tuple() -> + * heap_freeze_tuple() will get confused. + * + * It'd be better if we somehow could avoid setting hint bits on the + * old page. One reason to use VACUUM FULL are very bloated tables - + * rewriting most of the old table before during VACUUM FULL doesn't + * exactly help... + */ + LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE); switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf)) { diff --git a/src/backend/access/heap/heapam_visibility.c b/src/backend/access/heap/heapam_visibility.c index 4fefcbca5f5..762538a2040 100644 --- a/src/backend/access/heap/heapam_visibility.c +++ b/src/backend/access/heap/heapam_visibility.c @@ -141,6 +141,13 @@ void HeapTupleSetHintBits(HeapTupleHeader tuple, Buffer buffer, uint16 infomask, TransactionId xid) { + /* + * The uses from heapam.c rely on being able to perform the hint bit + * updates, which can only be guaranteed if we are holding an exclusive + * lock on the buffer - which all callers are doing. + */ + Assert(BufferIsLockedByMeInMode(buffer, BUFFER_LOCK_EXCLUSIVE)); + SetHintBits(tuple, buffer, infomask, xid); } -- 2.48.1.76.g4e746b1a31.dirty --mhhc7c45w3dmihtp Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v6-0011-heapam-Add-batch-mode-mvcc-check-and-use-it-in-pa.patch" ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v7 09/15] heapam: Use exclusive lock on old page in CLUSTER @ 2025-01-26 20:18 Andres Freund <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Andres Freund @ 2025-01-26 20:18 UTC (permalink / raw) To be able to guarantee that we can set the hint bit, acquire an exclusive lock on the old buffer. We need the hint bits to be set as otherwise reform_and_rewrite_tuple() -> rewrite_heap_tuple() -> heap_freeze_tuple() will get confused. It'd be better if we somehow could avoid setting hint bits on the old page. A commonreason to use VACUUM FULL are very bloated tables - rewriting most of the old table before during VACUUM FULL doesn't exactly help. Author: Reviewed-by: Discussion: https://postgr.es/m/ Backpatch: --- src/backend/access/heap/heapam_handler.c | 13 ++++++++++++- src/backend/access/heap/heapam_visibility.c | 7 +++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index bcbac844bb6..f84254f0737 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -837,7 +837,18 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, tuple = ExecFetchSlotHeapTuple(slot, false, NULL); buf = hslot->buffer; - LockBuffer(buf, BUFFER_LOCK_SHARE); + /* + * To be able to guarantee that we can set the hint bit, acquire an + * exclusive lock on the old buffer. We need the hint bits to be set + * as otherwise reform_and_rewrite_tuple() -> rewrite_heap_tuple() -> + * heap_freeze_tuple() will get confused. + * + * It'd be better if we somehow could avoid setting hint bits on the + * old page. One reason to use VACUUM FULL are very bloated tables - + * rewriting most of the old table before during VACUUM FULL doesn't + * exactly help... + */ + LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE); switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf)) { diff --git a/src/backend/access/heap/heapam_visibility.c b/src/backend/access/heap/heapam_visibility.c index 4fefcbca5f5..762538a2040 100644 --- a/src/backend/access/heap/heapam_visibility.c +++ b/src/backend/access/heap/heapam_visibility.c @@ -141,6 +141,13 @@ void HeapTupleSetHintBits(HeapTupleHeader tuple, Buffer buffer, uint16 infomask, TransactionId xid) { + /* + * The uses from heapam.c rely on being able to perform the hint bit + * updates, which can only be guaranteed if we are holding an exclusive + * lock on the buffer - which all callers are doing. + */ + Assert(BufferIsLockedByMeInMode(buffer, BUFFER_LOCK_EXCLUSIVE)); + SetHintBits(tuple, buffer, infomask, xid); } -- 2.48.1.76.g4e746b1a31.dirty --kad4mm4awdo7v2ki Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v7-0010-heapam-Add-batch-mode-mvcc-check-and-use-it-in-pa.patch" ^ permalink raw reply [nested|flat] 31+ messages in thread
* [PATCH v8 04/10] heapam: Use exclusive lock on old page in CLUSTER @ 2025-01-26 20:18 Andres Freund <[email protected]> 0 siblings, 0 replies; 31+ messages in thread From: Andres Freund @ 2025-01-26 20:18 UTC (permalink / raw) To be able to guarantee that we can set the hint bit, acquire an exclusive lock on the old buffer. We need the hint bits to be set as otherwise reform_and_rewrite_tuple() -> rewrite_heap_tuple() -> heap_freeze_tuple() will get confused. It'd be better if we somehow could avoid setting hint bits on the old page. A commonreason to use VACUUM FULL are very bloated tables - rewriting most of the old table before during VACUUM FULL doesn't exactly help. Author: Reviewed-by: Discussion: https://postgr.es/m/ Backpatch: --- src/backend/access/heap/heapam_handler.c | 13 ++++++++++++- src/backend/access/heap/heapam_visibility.c | 7 +++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index dd4fe6bf62f..0c684786382 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -837,7 +837,18 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, tuple = ExecFetchSlotHeapTuple(slot, false, NULL); buf = hslot->buffer; - LockBuffer(buf, BUFFER_LOCK_SHARE); + /* + * To be able to guarantee that we can set the hint bit, acquire an + * exclusive lock on the old buffer. We need the hint bits to be set + * as otherwise reform_and_rewrite_tuple() -> rewrite_heap_tuple() -> + * heap_freeze_tuple() will get confused. + * + * It'd be better if we somehow could avoid setting hint bits on the + * old page. One reason to use VACUUM FULL are very bloated tables - + * rewriting most of the old table before during VACUUM FULL doesn't + * exactly help... + */ + LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE); switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf)) { diff --git a/src/backend/access/heap/heapam_visibility.c b/src/backend/access/heap/heapam_visibility.c index bf899c2d2c6..debf5d56b95 100644 --- a/src/backend/access/heap/heapam_visibility.c +++ b/src/backend/access/heap/heapam_visibility.c @@ -141,6 +141,13 @@ void HeapTupleSetHintBits(HeapTupleHeader tuple, Buffer buffer, uint16 infomask, TransactionId xid) { + /* + * The uses from heapam.c rely on being able to perform the hint bit + * updates, which can only be guaranteed if we are holding an exclusive + * lock on the buffer - which all callers are doing. + */ + Assert(BufferIsLockedByMeInMode(buffer, BUFFER_LOCK_EXCLUSIVE)); + SetHintBits(tuple, buffer, infomask, xid); } -- 2.48.1.76.g4e746b1a31.dirty --lbbj3oq7cjvyhmbk Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v8-0005-heapam-Add-batch-mode-mvcc-check-and-use-it-in-pa.patch" ^ permalink raw reply [nested|flat] 31+ messages in thread
end of thread, other threads:[~2025-01-26 20:18 UTC | newest] Thread overview: 31+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2021-03-15 18:07 [PATCH v37 2/2] Add libpq pipeline mode support to pgbench Alvaro Herrera <[email protected]> 2022-02-22 02:21 [PATCH 3/4] Reindent and wrap between added sect1 tags Chapman Flack <[email protected]> 2024-10-17 15:45 [PATCH v2 04/15] heapam: Use exclusive lock on old page in CLUSTER Andres Freund <[email protected]> 2024-10-17 15:45 [PATCH v1 04/14] heapam: Use exclusive lock on old page in CLUSTER Andres Freund <[email protected]> 2025-01-26 20:18 [PATCH v9 02/10] heapam: Use exclusive lock on old page in CLUSTER Andres Freund <[email protected]> 2025-01-26 20:18 [PATCH v6 10/14] heapam: Use exclusive lock on old page in CLUSTER Andres Freund <[email protected]> 2025-01-26 20:18 [PATCH v7 09/15] heapam: Use exclusive lock on old page in CLUSTER Andres Freund <[email protected]> 2025-01-26 20:18 [PATCH v8 04/10] heapam: Use exclusive lock on old page in CLUSTER Andres Freund <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox