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 5314e9348fa7..53b38ad70cfc 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 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 6b1fd6d4cca6..1eccc7721fe5 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"
@@ -223,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 */
@@ -391,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);
/*
@@ -498,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);
}
@@ -709,6 +714,8 @@ CopyGetData(CopyState cstate, void *databuf, int minread, int maxread)
break;
}
+ CopyUpdateBytesProgress(cstate, bytesread);
+
return bytesread;
}
@@ -1751,6 +1758,14 @@ BeginCopy(ParseState *pstate,
cstate->copy_dest = COPY_FILE; /* default */
+ pgstat_progress_start_command(PROGRESS_COMMAND_COPY, cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+
+ /* 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);
+
MemoryContextSwitchTo(oldcontext);
return cstate;
@@ -1810,6 +1825,8 @@ EndCopy(CopyState cstate)
cstate->filename)));
}
+ pgstat_progress_end_command();
+
MemoryContextDelete(cstate->copycontext);
pfree(cstate);
}
@@ -2122,7 +2139,9 @@ CopyTo(CopyState cstate)
/* Format and send the data */
CopyOneRowTo(cstate, slot);
- processed++;
+
+ /* Increment processed lines counter and update progress */
+ pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -3259,9 +3278,10 @@ 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.
*/
- processed++;
+ pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
}
}
@@ -5114,7 +5134,8 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
/* Send the data */
CopyOneRowTo(cstate, slot);
- myState->processed++;
+ /* Increment processed lines counter and update progress */
+ pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
return true;
}
@@ -5156,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/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..3947222e6f61 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -133,4 +133,11 @@
#define PROGRESS_BASEBACKUP_PHASE_WAIT_WAL_ARCHIVE 4
#define PROGRESS_BASEBACKUP_PHASE_TRANSFER_WAL 5
+/* 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/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
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index b813e322153d..c04f7b94c3f0 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 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,