From 64bf6cd66478e5b5b710d404b726c4147a98a660 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Josef=20=C5=A0im=C3=A1nek?= Date: Mon, 8 Jun 2020 02:00:53 +0200 Subject: [PATCH 1/3] Initial work on COPY progress. --- src/backend/commands/copy.c | 13 ++++++++++--- src/backend/utils/adt/pgstatfuncs.c | 2 ++ src/include/commands/progress.h | 3 +++ src/include/pgstat.h | 3 ++- 4 files changed, 17 insertions(+), 4 deletions(-) diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index 6b1fd6d4cca6..bef551770862 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -29,6 +29,7 @@ #include "catalog/pg_type.h" #include "commands/copy.h" #include "commands/defrem.h" +#include "commands/progress.h" #include "commands/trigger.h" #include "executor/execPartition.h" #include "executor/executor.h" @@ -45,6 +46,7 @@ #include "parser/parse_collate.h" #include "parser/parse_expr.h" #include "parser/parse_relation.h" +#include "pgstat.h" #include "port/pg_bswap.h" #include "rewrite/rewriteHandler.h" #include "storage/fd.h" @@ -1751,6 +1753,9 @@ BeginCopy(ParseState *pstate, cstate->copy_dest = COPY_FILE; /* default */ + pgstat_progress_start_command(PROGRESS_COMMAND_COPY, queryRelId); + pgstat_progress_update_param(PROGRESS_COPY_PROCESSED,0); + MemoryContextSwitchTo(oldcontext); return cstate; @@ -1810,6 +1815,8 @@ EndCopy(CopyState cstate) cstate->filename))); } + pgstat_progress_end_command(); + MemoryContextDelete(cstate->copycontext); pfree(cstate); } @@ -2122,7 +2129,7 @@ CopyTo(CopyState cstate) /* Format and send the data */ CopyOneRowTo(cstate, slot); - processed++; + pgstat_progress_update_param(PROGRESS_COPY_PROCESSED, ++processed); } ExecDropSingleTupleTableSlot(slot); @@ -3261,7 +3268,7 @@ CopyFrom(CopyState cstate) * or FDW; this is the same definition used by nodeModifyTable.c * for counting tuples inserted by an INSERT command. */ - processed++; + pgstat_progress_update_param(PROGRESS_COPY_PROCESSED, ++processed); } } @@ -5114,7 +5121,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self) /* Send the data */ CopyOneRowTo(cstate, slot); - myState->processed++; + pgstat_progress_update_param(PROGRESS_COPY_PROCESSED, ++myState->processed); return true; } diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c index 2aff739466ff..b740eef7c102 100644 --- a/src/backend/utils/adt/pgstatfuncs.c +++ b/src/backend/utils/adt/pgstatfuncs.c @@ -494,6 +494,8 @@ pg_stat_get_progress_info(PG_FUNCTION_ARGS) cmdtype = PROGRESS_COMMAND_CREATE_INDEX; else if (pg_strcasecmp(cmd, "BASEBACKUP") == 0) cmdtype = PROGRESS_COMMAND_BASEBACKUP; + else if (pg_strcasecmp(cmd, "COPY") == 0) + cmdtype = PROGRESS_COMMAND_COPY; else ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h index 36b073e67757..2d72f12a75b5 100644 --- a/src/include/commands/progress.h +++ b/src/include/commands/progress.h @@ -133,4 +133,7 @@ #define PROGRESS_BASEBACKUP_PHASE_WAIT_WAL_ARCHIVE 4 #define PROGRESS_BASEBACKUP_PHASE_TRANSFER_WAL 5 +/* Commands of PROGRESS_CLUSTER */ +#define PROGRESS_COPY_PROCESSED 0 + #endif diff --git a/src/include/pgstat.h b/src/include/pgstat.h index 13872013823e..c338a1a5ba00 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -995,7 +995,8 @@ typedef enum ProgressCommandType PROGRESS_COMMAND_ANALYZE, PROGRESS_COMMAND_CLUSTER, PROGRESS_COMMAND_CREATE_INDEX, - PROGRESS_COMMAND_BASEBACKUP + PROGRESS_COMMAND_BASEBACKUP, + PROGRESS_COMMAND_COPY } ProgressCommandType; #define PGSTAT_NUM_PROGRESS_PARAM 20 From f9cb4fe9b3fc223cb3174044ea00edbb96675138 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Josef=20=C5=A0im=C3=A1nek?= Date: Sun, 14 Jun 2020 02:46:04 +0200 Subject: [PATCH 2/3] Enhance copy progress with more info. - add pg_stat_progress_copy system view --- src/backend/catalog/system_views.sql | 14 ++++++++++++++ src/backend/commands/copy.c | 16 +++++++++++----- src/include/commands/progress.h | 8 ++++++-- src/test/regress/expected/rules.out | 15 +++++++++++++++ 4 files changed, 46 insertions(+), 7 deletions(-) diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 5314e9348fa7..7bc429ce299e 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1093,6 +1093,20 @@ CREATE VIEW pg_stat_progress_basebackup AS S.param5 AS tablespaces_streamed FROM pg_stat_get_progress_info('BASEBACKUP') AS S; +CREATE VIEW pg_stat_progress_copy AS + SELECT + S.pid AS pid, S.datid AS datid, D.datname AS datname, + S.relid AS relid, + CASE S.param1 WHEN 0 THEN 'TO' + WHEN 1 THEN 'FROM' + END as direction, + CAST (S.param2::integer AS bool) AS file, + CAST (S.param3::integer AS bool) AS program, + S.param4 AS lines_processed, + S.param5 AS file_bytes_processed + FROM pg_stat_get_progress_info('COPY') AS S + LEFT JOIN pg_database D ON S.datid = D.oid; + CREATE VIEW pg_user_mappings AS SELECT U.oid AS umid, diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index bef551770862..a733b6e1c676 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -565,6 +565,7 @@ CopySendEndOfRow(CopyState cstate) (errcode_for_file_access(), errmsg("could not write to COPY file: %m"))); } + pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, ftell(cstate->copy_file)); break; case COPY_OLD_FE: /* The FE/BE protocol uses \n as newline for all platforms */ @@ -617,6 +618,7 @@ CopyGetData(CopyState cstate, void *databuf, int minread, int maxread) { case COPY_FILE: bytesread = fread(databuf, 1, maxread, cstate->copy_file); + pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, ftell(cstate->copy_file)); if (ferror(cstate->copy_file)) ereport(ERROR, (errcode_for_file_access(), @@ -1753,8 +1755,12 @@ BeginCopy(ParseState *pstate, cstate->copy_dest = COPY_FILE; /* default */ - pgstat_progress_start_command(PROGRESS_COMMAND_COPY, queryRelId); - pgstat_progress_update_param(PROGRESS_COPY_PROCESSED,0); + pgstat_progress_start_command(PROGRESS_COMMAND_COPY, cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid); + + pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, 0); + pgstat_progress_update_param(PROGRESS_COPY_IS_FROM, (int) cstate->is_copy_from); + pgstat_progress_update_param(PROGRESS_COPY_IS_FILE, (int) cstate->copy_dest == COPY_FILE); + pgstat_progress_update_param(PROGRESS_COPY_IS_PROGRAM, (int) cstate->is_program); MemoryContextSwitchTo(oldcontext); @@ -2129,7 +2135,7 @@ CopyTo(CopyState cstate) /* Format and send the data */ CopyOneRowTo(cstate, slot); - pgstat_progress_update_param(PROGRESS_COPY_PROCESSED, ++processed); + pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed); } ExecDropSingleTupleTableSlot(slot); @@ -3268,7 +3274,7 @@ CopyFrom(CopyState cstate) * or FDW; this is the same definition used by nodeModifyTable.c * for counting tuples inserted by an INSERT command. */ - pgstat_progress_update_param(PROGRESS_COPY_PROCESSED, ++processed); + pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed); } } @@ -5121,7 +5127,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self) /* Send the data */ CopyOneRowTo(cstate, slot); - pgstat_progress_update_param(PROGRESS_COPY_PROCESSED, ++myState->processed); + pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed); return true; } diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h index 2d72f12a75b5..3947222e6f61 100644 --- a/src/include/commands/progress.h +++ b/src/include/commands/progress.h @@ -133,7 +133,11 @@ #define PROGRESS_BASEBACKUP_PHASE_WAIT_WAL_ARCHIVE 4 #define PROGRESS_BASEBACKUP_PHASE_TRANSFER_WAL 5 -/* Commands of PROGRESS_CLUSTER */ -#define PROGRESS_COPY_PROCESSED 0 +/* Commands of PROGRESS_COPY */ +#define PROGRESS_COPY_IS_FROM 0 +#define PROGRESS_COPY_IS_FILE 1 +#define PROGRESS_COPY_IS_PROGRAM 2 +#define PROGRESS_COPY_LINES_PROCESSED 3 +#define PROGRESS_COPY_BYTES_PROCESSED 4 #endif diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index b813e322153d..dac5da4e6dd9 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1925,6 +1925,21 @@ pg_stat_progress_cluster| SELECT s.pid, s.param8 AS index_rebuild_count FROM (pg_stat_get_progress_info('CLUSTER'::text) s(pid, datid, relid, param1, param2, param3, param4, param5, param6, param7, param8, param9, param10, param11, param12, param13, param14, param15, param16, param17, param18, param19, param20) LEFT JOIN pg_database d ON ((s.datid = d.oid))); +pg_stat_progress_copy| SELECT s.pid, + s.datid, + d.datname, + s.relid, + CASE s.param1 + WHEN 0 THEN 'TO'::text + WHEN 1 THEN 'FROM'::text + ELSE NULL::text + END AS direction, + ((s.param2)::integer)::boolean AS file, + ((s.param3)::integer)::boolean AS program, + s.param4 AS lines_processed, + s.param5 AS file_bytes_processed + FROM (pg_stat_get_progress_info('COPY'::text) s(pid, datid, relid, param1, param2, param3, param4, param5, param6, param7, param8, param9, param10, param11, param12, param13, param14, param15, param16, param17, param18, param19, param20) + LEFT JOIN pg_database d ON ((s.datid = d.oid))); pg_stat_progress_create_index| SELECT s.pid, s.datid, d.datname, From 5b1ddddf0c0a83606757b762f996e9c40469a831 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Josef=20=C5=A0im=C3=A1nek?= Date: Sun, 21 Jun 2020 13:15:57 +0200 Subject: [PATCH 3/3] Support STDIN, STDOUT. Add docs. --- doc/src/sgml/monitoring.sgml | 119 +++++++++++++++++++++++++++ src/backend/catalog/system_views.sql | 2 +- src/backend/commands/copy.c | 34 ++++++-- src/test/regress/expected/rules.out | 2 +- 4 files changed, 149 insertions(+), 8 deletions(-) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index dfa9d0d6410c..d4a9fa9e2f2f 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -390,6 +390,13 @@ postgres 27093 0.0 0.0 30096 2752 ? Ss 11:34 0:00 postgres: ser + + pg_stat_progress_copypg_stat_progress_copy + One row for each backend running COPY, showing current progress. + See . + + + @@ -5013,6 +5020,7 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid, which support progress reporting are ANALYZE, CLUSTER, CREATE INDEX, VACUUM, + COPY and (i.e., replication command that issues to take a base backup). @@ -6142,6 +6150,117 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid, + + + COPY Progress Reporting + + + Whenever COPY is running, the + pg_stat_copy_progress view will contain one row + for each backend that is currently running COPY command. + The table bellow describes the information that will be reported and provide + information how to interpret it. + + + + <structname>pg_stat_progress_copy</structname> View + + + + + Column Type + + + Description + + + + + + + + pid integer + + + Process ID of backend. + + + + + + datid text + + + OID of the database to which this backend is connected. + + + + + + datname name + + + Name of the database to which this backend is connected. + + + + + + relid oid + + + OID of the table on which the COPY command is executed. It is set to 0 if SELECT query is provided. + + + + + + direction text + + + Can be one of TO or FROM. + + + + + + file boolean + + + true if filename argument is used. Notice STDIN and STDOUT are considered file as well. + + + + + + program boolean + + + true if PROGRAM option is used. + + + + + + line_processed bigint + + + Number of lines already processed by COPY command. + + + + + + bytes_processed bigint + + + Number of bytes already processed by COPY command. + + + + +
+ diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 7bc429ce299e..53b38ad70cfc 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1103,7 +1103,7 @@ CREATE VIEW pg_stat_progress_copy AS CAST (S.param2::integer AS bool) AS file, CAST (S.param3::integer AS bool) AS program, S.param4 AS lines_processed, - S.param5 AS file_bytes_processed + S.param5 AS bytes_processed FROM pg_stat_get_progress_info('COPY') AS S LEFT JOIN pg_database D ON S.datid = D.oid; diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index a733b6e1c676..1eccc7721fe5 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -225,6 +225,7 @@ typedef struct CopyStateData char *raw_buf; int raw_buf_index; /* next byte to process */ int raw_buf_len; /* total # of bytes stored */ + uint64 bytes_processed; /* total # of bytes processed, used for progress reporting */ } CopyStateData; /* DestReceiver for COPY (query) TO */ @@ -393,6 +394,7 @@ static void CopySendInt32(CopyState cstate, int32 val); static bool CopyGetInt32(CopyState cstate, int32 *val); static void CopySendInt16(CopyState cstate, int16 val); static bool CopyGetInt16(CopyState cstate, int16 *val); +static int CopyUpdateBytesProgress(CopyState cstate, int newbytesprocessed); /* @@ -500,18 +502,19 @@ SendCopyEnd(CopyState cstate) static void CopySendData(CopyState cstate, const void *databuf, int datasize) { - appendBinaryStringInfo(cstate->fe_msgbuf, databuf, datasize); + appendBinaryStringInfo(cstate->fe_msgbuf, databuf, CopyUpdateBytesProgress(cstate, datasize)); } static void CopySendString(CopyState cstate, const char *str) { - appendBinaryStringInfo(cstate->fe_msgbuf, str, strlen(str)); + appendBinaryStringInfo(cstate->fe_msgbuf, str, CopyUpdateBytesProgress(cstate, strlen(str))); } static void CopySendChar(CopyState cstate, char c) { + CopyUpdateBytesProgress(cstate, 1); appendStringInfoCharMacro(cstate->fe_msgbuf, c); } @@ -565,7 +568,6 @@ CopySendEndOfRow(CopyState cstate) (errcode_for_file_access(), errmsg("could not write to COPY file: %m"))); } - pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, ftell(cstate->copy_file)); break; case COPY_OLD_FE: /* The FE/BE protocol uses \n as newline for all platforms */ @@ -618,7 +620,6 @@ CopyGetData(CopyState cstate, void *databuf, int minread, int maxread) { case COPY_FILE: bytesread = fread(databuf, 1, maxread, cstate->copy_file); - pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, ftell(cstate->copy_file)); if (ferror(cstate->copy_file)) ereport(ERROR, (errcode_for_file_access(), @@ -713,6 +714,8 @@ CopyGetData(CopyState cstate, void *databuf, int minread, int maxread) break; } + CopyUpdateBytesProgress(cstate, bytesread); + return bytesread; } @@ -1757,7 +1760,8 @@ BeginCopy(ParseState *pstate, pgstat_progress_start_command(PROGRESS_COMMAND_COPY, cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid); - pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, 0); + /* initialize progress */ + cstate->bytes_processed = 0; pgstat_progress_update_param(PROGRESS_COPY_IS_FROM, (int) cstate->is_copy_from); pgstat_progress_update_param(PROGRESS_COPY_IS_FILE, (int) cstate->copy_dest == COPY_FILE); pgstat_progress_update_param(PROGRESS_COPY_IS_PROGRAM, (int) cstate->is_program); @@ -2135,6 +2139,8 @@ CopyTo(CopyState cstate) /* Format and send the data */ CopyOneRowTo(cstate, slot); + + /* Increment processed lines counter and update progress */ pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed); } @@ -3272,7 +3278,8 @@ CopyFrom(CopyState cstate) /* * We count only tuples not suppressed by a BEFORE INSERT trigger * or FDW; this is the same definition used by nodeModifyTable.c - * for counting tuples inserted by an INSERT command. + * for counting tuples inserted by an INSERT command. Update + * progress as well. */ pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed); } @@ -5127,6 +5134,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self) /* Send the data */ CopyOneRowTo(cstate, slot); + /* Increment processed lines counter and update progress */ pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed); return true; @@ -5169,3 +5177,17 @@ CreateCopyDestReceiver(void) return (DestReceiver *) self; } + +/* + * CopyUpdateBytesProgress --- increment bytes_processed + * on CopyState, updates progress and returns amount + * of processed bytes back; + */ + +static int +CopyUpdateBytesProgress(CopyState cstate, int newbytesprocessed) +{ + cstate->bytes_processed += newbytesprocessed; + pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, cstate->bytes_processed); + return newbytesprocessed; +} diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index dac5da4e6dd9..c04f7b94c3f0 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1937,7 +1937,7 @@ pg_stat_progress_copy| SELECT s.pid, ((s.param2)::integer)::boolean AS file, ((s.param3)::integer)::boolean AS program, s.param4 AS lines_processed, - s.param5 AS file_bytes_processed + s.param5 AS bytes_processed FROM (pg_stat_get_progress_info('COPY'::text) s(pid, datid, relid, param1, param2, param3, param4, param5, param6, param7, param8, param9, param10, param11, param12, param13, param14, param15, param16, param17, param18, param19, param20) LEFT JOIN pg_database d ON ((s.datid = d.oid))); pg_stat_progress_create_index| SELECT s.pid,