public inbox for [email protected]
help / color / mirror / Atom feed[PATCH 1/6] Add progress-reported components for COPY progress reporting
23+ messages / 2 participants
[nested] [flat]
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* [PATCH 1/6] Add progress-reported components for COPY progress reporting
@ 2021-02-12 13:06 Matthias van de Meent <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Matthias van de Meent @ 2021-02-12 13:06 UTC (permalink / raw)
The command, io target and excluded tuple count (by COPY ... FROM ... s'
WHERE -clause) are now reported in the pg_stat_progress_copy view.
Additionally, the column lines_processed is renamed to tuples_processed to
disambiguate the meaning of this column in cases of CSV and BINARY copies and
to stay consistent with regards to names in the pg_stat_progress_*-views.
Of special interest is the reporting of io_target, with which we can
distinguish logical replications' initial table synchronization workers'
progress without having to join the pg_stat_activity view.
---
doc/src/sgml/monitoring.sgml | 37 ++++++++++++++++++++++++++--
src/backend/catalog/system_views.sql | 11 ++++++++-
src/backend/commands/copyfrom.c | 30 +++++++++++++++++++++-
src/backend/commands/copyto.c | 29 ++++++++++++++++++++--
src/include/commands/progress.h | 15 ++++++++++-
src/test/regress/expected/rules.out | 15 ++++++++++-
6 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c602ee4427..3c39c82f1a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -6544,6 +6544,29 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>command</structfield> <type>text</type>
+ </para>
+ <para>
+ The command that is running: <literal>COPY FROM</literal>, or
+ <literal>COPY TO</literal>.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>io_target</structfield> <type>text</type>
+ </para>
+ <para>
+ The io target that the data is read from or written to:
+ <literal>FILE</literal>, <literal>PROGRAM</literal>,
+ <literal>STDIO</literal> (for COPY FROM STDIN and COPY TO STDOUT),
+ or <literal>CALLBACK</literal> (used in the table synchronization
+ background worker).
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>bytes_processed</structfield> <type>bigint</type>
@@ -6565,10 +6588,20 @@ SELECT pg_stat_get_backend_pid(s.backendid) AS pid,
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>lines_processed</structfield> <type>bigint</type>
+ <structfield>tuples_processed</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of tuples already processed by <command>COPY</command> command.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>tuples_excluded</structfield> <type>bigint</type>
</para>
<para>
- Number of lines already processed by <command>COPY</command> command.
+ Number of tuples not processed because they were excluded by the
+ <command>WHERE</command> clause of the <command>COPY</command> command.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fa58afd9d7..6a3ac47b85 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1129,9 +1129,18 @@ 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.param5 WHEN 1 THEN 'COPY FROM'
+ WHEN 2 THEN 'COPY TO'
+ END AS command,
+ CASE S.param6 WHEN 1 THEN 'FILE'
+ WHEN 2 THEN 'PROGRAM'
+ WHEN 3 THEN 'STDIO'
+ WHEN 4 THEN 'CALLBACK'
+ END AS io_target,
S.param1 AS bytes_processed,
S.param2 AS bytes_total,
- S.param3 AS lines_processed
+ S.param3 AS tuples_processed,
+ S.param4 AS tuples_excluded
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/copyfrom.c b/src/backend/commands/copyfrom.c
index 796ca7b3f7..c3610eb67e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -540,6 +540,7 @@ CopyFrom(CopyFromState cstate)
CopyInsertMethod insertMethod;
CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
+ uint64 excluded = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
bool leafpart_use_multi_insert = false;
@@ -869,7 +870,11 @@ CopyFrom(CopyFromState cstate)
econtext->ecxt_scantuple = myslot;
/* Skip items that don't match COPY's WHERE clause */
if (!ExecQual(cstate->qualexpr, econtext))
+ {
+ /* Report that this tuple was filtered out by the WHERE clause */
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED, ++excluded);
continue;
+ }
}
/* Determine the partition to insert the tuple into */
@@ -1107,7 +1112,7 @@ CopyFrom(CopyFromState cstate)
* for counting tuples inserted by an INSERT command. Update
* progress of the COPY command as well.
*/
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
}
@@ -1424,6 +1429,8 @@ BeginCopyFrom(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ pgstat_progress_update_param(PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_FROM);
+
cstate->bytes_processed = 0;
/* We keep those variables in cstate. */
@@ -1501,6 +1508,27 @@ BeginCopyFrom(ParseState *pstate,
ReceiveCopyBinaryHeader(cstate);
}
+ {
+ int64 io_target;
+ switch (cstate->copy_src)
+ {
+ case COPY_FILE:
+ if (is_program)
+ io_target = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ io_target = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ io_target = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ case COPY_CALLBACK:
+ io_target = PROGRESS_COPY_IO_TARGET_CALLBACK;
+ break;
+ }
+ pgstat_progress_update_param(PROGRESS_COPY_IO_TARGET, io_target);
+ }
+
/* create workspace for CopyReadAttributes results */
if (!cstate->opts.binary)
{
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index e04ec1e331..42c4a828df 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -772,6 +772,31 @@ BeginCopyTo(ParseState *pstate,
/* initialize progress */
pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
+ {
+ const int progress_index[] = {
+ PROGRESS_COPY_COMMAND,
+ PROGRESS_COPY_IO_TARGET
+ };
+ int64 progress_vals[] = {
+ PROGRESS_COPY_COMMAND_TO,
+ 0
+ };
+ switch (cstate->copy_dest)
+ {
+ case COPY_FILE:
+ if (is_program)
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_PROGRAM;
+ else
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_FILE;
+ break;
+ case COPY_OLD_FE:
+ case COPY_NEW_FE:
+ progress_vals[1] = PROGRESS_COPY_IO_TARGET_STDIO;
+ break;
+ }
+ pgstat_progress_update_multi_param(2, progress_index, progress_vals);
+ }
+
cstate->bytes_processed = 0;
MemoryContextSwitchTo(oldcontext);
@@ -954,7 +979,7 @@ CopyTo(CopyToState cstate)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++processed);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1321,7 +1346,7 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
CopyOneRowTo(cstate, slot);
/* Increment amount of processed tuples and update the progress */
- pgstat_progress_update_param(PROGRESS_COPY_LINES_PROCESSED, ++myState->processed);
+ pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, ++myState->processed);
return true;
}
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index 95ec5d02e9..e003217554 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -136,6 +136,19 @@
/* Commands of PROGRESS_COPY */
#define PROGRESS_COPY_BYTES_PROCESSED 0
#define PROGRESS_COPY_BYTES_TOTAL 1
-#define PROGRESS_COPY_LINES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_PROCESSED 2
+#define PROGRESS_COPY_TUPLES_EXCLUDED 3
+#define PROGRESS_COPY_COMMAND 4
+#define PROGRESS_COPY_IO_TARGET 5
+
+/* Commands of PROGRESS_COPY_COMMAND */
+#define PROGRESS_COPY_COMMAND_FROM 1
+#define PROGRESS_COPY_COMMAND_TO 2
+
+/* Types of PROGRESS_COPY_INOUT_TYPE */
+#define PROGRESS_COPY_IO_TARGET_FILE 1
+#define PROGRESS_COPY_IO_TARGET_PROGRAM 2
+#define PROGRESS_COPY_IO_TARGET_STDIO 3
+#define PROGRESS_COPY_IO_TARGET_CALLBACK 4
#endif
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 10a1f34ebc..3c6776429d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1949,9 +1949,22 @@ pg_stat_progress_copy| SELECT s.pid,
s.datid,
d.datname,
s.relid,
+ CASE s.param5
+ WHEN 1 THEN 'COPY FROM'::text
+ WHEN 2 THEN 'COPY TO'::text
+ ELSE NULL::text
+ END AS command,
+ CASE s.param6
+ WHEN 1 THEN 'FILE'::text
+ WHEN 2 THEN 'PROGRAM'::text
+ WHEN 3 THEN 'STDIO'::text
+ WHEN 4 THEN 'CALLBACK'::text
+ ELSE NULL::text
+ END AS io_target,
s.param1 AS bytes_processed,
s.param2 AS bytes_total,
- s.param3 AS lines_processed
+ s.param3 AS tuples_processed,
+ s.param4 AS tuples_excluded
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,
--
2.26.2
--------------DA8D60BF026F7ACE69A9AEE5
Content-Type: text/x-patch; charset=UTF-8;
name="v7-0002-0001-review.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename="v7-0002-0001-review.patch"
^ permalink raw reply [nested|flat] 23+ messages in thread
* Re: Memoize ANTI and SEMI JOIN inner
@ 2025-04-04 13:49 Andrei Lepikhov <[email protected]>
0 siblings, 0 replies; 23+ messages in thread
From: Andrei Lepikhov @ 2025-04-04 13:49 UTC (permalink / raw)
To: Richard Guo <[email protected]>; David Rowley <[email protected]>; +Cc: pgsql-hackers
On 4/1/25 09:18, Richard Guo wrote:
> On Mon, Mar 31, 2025 at 7:33 PM Andrei Lepikhov <[email protected]> wrote:
>> and I don't get the case. As I see, ANTI/SEMI join just transforms to
>> the regular join and it is still not the case. May you be more specific?
>
> Upthread, you said that a qual contained in ppi_clauses will also be
> included in extra->restrictlist. I provided this example to show that
> this is not true. And it seems to me that this discrepancy makes the
> check I mentioned earlier not reliable in all cases. As I explained
> earlier, this check could pass even if a restriction clause isn't
> parameterized, as long as another join clause, which doesn't belong to
> the current join, is included in ppi_clauses. This is not correct.
Ok, keep aside the question of why we discuss it here.
After second thought, I think the current check is correct at the
moment. Looking into the inner_unique proving machinery, I see that the
inner may be decided to be unique if:
1. It has a unique index on joining columns - in the case of NestLoop,
these clauses inevitably go down as parameters.
2. degenerate case like x=1 where x is unique column - not our case,
because x=1 goes down
3. DISTINCT clause - in this case, we can't transform the sublink at
early stages, and using the already planned subquery --> upper rel
clause will not be pushed to the unique inner.
So, I see nothing criminal in this clause - it may be a little more
specific to be sure we will be safe in case of improvements in the
inner_unique proof logic or subquery parameterisation, but that's it. If
my analysis is correct, we only need to change the comment.
As usual, it would be nice to see an example of incorrect behaviour if
I'm wrong.
--
regards, Andrei Lepikhov
^ permalink raw reply [nested|flat] 23+ messages in thread
end of thread, other threads:[~2025-04-04 13:49 UTC | newest]
Thread overview: 23+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2021-02-12 13:06 [PATCH 1/6] Add progress-reported components for COPY progress reporting Matthias van de Meent <[email protected]>
2025-04-04 13:49 Re: Memoize ANTI and SEMI JOIN inner Andrei Lepikhov <[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