public inbox for [email protected]help / color / mirror / Atom feed
[PATCH 1/6] Add progress-reported components for COPY progress reporting 2+ 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; 2+ 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] 2+ messages in thread
* Re: [BUG] Logical replication failure "ERROR: could not map filenode "base/13237/442428" to relation OID" with catalog modifying txns @ 2022-02-21 10:29 Drouvot, Bertrand <[email protected]> 0 siblings, 0 replies; 2+ messages in thread From: Drouvot, Bertrand @ 2022-02-21 10:29 UTC (permalink / raw) To: Kyotaro Horiguchi <[email protected]>; [email protected]; [email protected]; +Cc: [email protected]; [email protected] Hi, On 10/19/21 8:43 AM, Kyotaro Horiguchi wrote: > At Tue, 19 Oct 2021 02:45:24 +0000, "[email protected]" <[email protected]> wrote in >> On Thursday, October 14, 2021 11:21 AM Kyotaro Horiguchi <[email protected]> wrote: >>> It adds a call to ReorderBufferAssignChild but usually subtransactions are >>> assigned to top level elsewherae. Addition to that >>> ReorderBufferCommitChild() called just later does the same thing. We are >>> adding the third call to the same function, which looks a bit odd. >> It can be odd. However, we >> have a check at the top of ReorderBufferAssignChild >> to judge if the sub transaction is already associated or not >> and skip the processings if it is. > My question was why do we need to make the extra call to > ReorerBufferCommitChild when XACT_XINFO_HAS_INVALS in spite of the > existing call to the same fuction that unconditionally made. It > doesn't cost so much but also it's not free. > >>> And I'm not sure it is wise to mark all subtransactions as "catalog changed" >>> always when the top transaction is XACT_XINFO_HAS_INVALS. >> In order to avoid this, >> can't we have a new flag (for example, in reorderbuffer struct) to check >> if we start decoding from RUNNING_XACTS, which is similar to the first patch of [1] >> and use it at DecodeCommit ? This still leads to some extra specific codes added >> to DecodeCommit and this solution becomes a bit similar to other previous patches though. > If it is somehow wrong in any sense that we add subtransactions in > SnapBuildProcessRunningXacts (for example, we should avoid relaxing > the assertion condition.), I think we would go another way. Otherwise > we don't even need that additional flag. (But Sawadasan's recent PoC > also needs that relaxation.) > > ASAICS, and unless I'm missing something (that odds are rtlatively > high:p), we need the specially added subransactions only for the > transactions that were running at passing the first RUNNING_XACTS, > becuase otherwise (substantial) subtransactions are assigned to > toplevel by the first record of the subtransaction. > > Before reaching consistency, DecodeCommit feeds the subtransactions to > ReorderBufferForget individually so the subtransactions are not needed > to be assigned to the top transaction at all. Since the > subtransactions added by the first RUNNING_XACT are processed that > way, we don't need in the first place to call ReorderBufferCommitChild > for such subtransactions. > >> [1] - https://www.postgresql.org/message-id/81D0D8B0-E7C4-4999-B616-1E5004DBDCD2%40amazon.com > regards. > > -- > Kyotaro Horiguchi > NTT Open Source Software Center > Just rebased (minor change in the contrib/test_decoding/Makefile) the last POC version linked to the CF entry as it was failing the CF bot. Thanks Bertrand diff --git a/contrib/test_decoding/Makefile b/contrib/test_decoding/Makefile index 36929dd97d..05c0c5a2f8 100644 --- a/contrib/test_decoding/Makefile +++ b/contrib/test_decoding/Makefile @@ -9,7 +9,7 @@ REGRESS = ddl xact rewrite toast permissions decoding_in_xact \ sequence ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \ oldest_xmin snapshot_transfer subxact_without_top concurrent_stream \ - twophase_snapshot slot_creation_error + twophase_snapshot slot_creation_error catalog_change_snapshot REGRESS_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf ISOLATION_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf diff --git a/contrib/test_decoding/expected/catalog_change_snapshot.out b/contrib/test_decoding/expected/catalog_change_snapshot.out new file mode 100644 index 0000000000..bc142fc384 --- /dev/null +++ b/contrib/test_decoding/expected/catalog_change_snapshot.out @@ -0,0 +1,44 @@ +Parsed test spec with 2 sessions + +starting permutation: s0_init s0_begin s0_savepoint s0_truncate s1_checkpoint s1_get_changes s0_commit s0_begin s0_insert s1_checkpoint s1_get_changes s0_commit s1_get_changes +step s0_init: SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); +?column? +-------- +init +(1 row) + +step s0_begin: BEGIN; +step s0_savepoint: SAVEPOINT sp1; +step s0_truncate: TRUNCATE tbl1; +step s1_checkpoint: CHECKPOINT; +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0'); +data +---- +(0 rows) + +step s0_commit: COMMIT; +step s0_begin: BEGIN; +step s0_insert: INSERT INTO tbl1 VALUES (1); +step s1_checkpoint: CHECKPOINT; +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0'); +data +--------------------------------------- +BEGIN +table public.tbl1: TRUNCATE: (no-flags) +COMMIT +(3 rows) + +step s0_commit: COMMIT; +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0'); +data +------------------------------------------------------------- +BEGIN +table public.tbl1: INSERT: val1[integer]:1 val2[integer]:null +COMMIT +(3 rows) + +?column? +-------- +stop +(1 row) + diff --git a/contrib/test_decoding/specs/catalog_change_snapshot.spec b/contrib/test_decoding/specs/catalog_change_snapshot.spec new file mode 100644 index 0000000000..43c0b64289 --- /dev/null +++ b/contrib/test_decoding/specs/catalog_change_snapshot.spec @@ -0,0 +1,32 @@ +setup +{ + DROP TABLE IF EXISTS tbl1; + CREATE TABLE tbl1 (val1 integer, val2 integer); +} + +teardown +{ + DROP TABLE tbl1; + SELECT 'stop' FROM pg_drop_replication_slot('isolation_slot'); +} + +session "s0" +setup { SET synchronous_commit=on; } +step "s0_init" { SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); } +step "s0_begin" { BEGIN; } +step "s0_savepoint" { SAVEPOINT sp1; } +step "s0_truncate" { TRUNCATE tbl1; } +step "s0_insert" { INSERT INTO tbl1 VALUES (1); } +step "s0_commit" { COMMIT; } + +session "s1" +setup { SET synchronous_commit=on; } +step "s1_checkpoint" { CHECKPOINT; } +step "s1_get_changes" { SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0'); } + +# For the transaction that TRUNCATEd the table tbl1, the last decoding decodes +# only its COMMIT record, because it starts from the RUNNING_XACT record emitted +# during the second checkpoint execution. This transaction must be marked as +# containing catalog changes during decoding the COMMIT record and the decoding +# of the INSERT record must read the pg_class with the correct historic snapshot. +permutation "s0_init" "s0_begin" "s0_savepoint" "s0_truncate" "s1_checkpoint" "s1_get_changes" "s0_commit" "s0_begin" "s0_insert" "s1_checkpoint" "s1_get_changes" "s0_commit" "s1_get_changes" diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 18cf931822..ade48bd71e 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -629,6 +629,25 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, commit_time = parsed->origin_timestamp; } + /* + * Mark the top transaction and its subtransactions as containing catalog + * changes, if the commit record has invalidation message. This is necessary + * for the case where we decode only the commit record of the transaction + * that actually has done catalog changes. + */ + if (parsed->xinfo & XACT_XINFO_HAS_INVALS) + { + ReorderBufferXidSetCatalogChanges(ctx->reorder, xid, buf->origptr); + + for (int i = 0; i < parsed->nsubxacts; i++) + { + ReorderBufferAssignChild(ctx->reorder, xid, parsed->subxacts[i], + buf->origptr); + ReorderBufferXidSetCatalogChanges(ctx->reorder, parsed->subxacts[i], + buf->origptr); + } + } + SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid, parsed->nsubxacts, parsed->subxacts); Attachments: [text/plain] poc_mark_all_txns_catalog_change.patch (5.2K, ../../[email protected]/2-poc_mark_all_txns_catalog_change.patch) download | inline diff: diff --git a/contrib/test_decoding/Makefile b/contrib/test_decoding/Makefile index 36929dd97d..05c0c5a2f8 100644 --- a/contrib/test_decoding/Makefile +++ b/contrib/test_decoding/Makefile @@ -9,7 +9,7 @@ REGRESS = ddl xact rewrite toast permissions decoding_in_xact \ sequence ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \ oldest_xmin snapshot_transfer subxact_without_top concurrent_stream \ - twophase_snapshot slot_creation_error + twophase_snapshot slot_creation_error catalog_change_snapshot REGRESS_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf ISOLATION_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf diff --git a/contrib/test_decoding/expected/catalog_change_snapshot.out b/contrib/test_decoding/expected/catalog_change_snapshot.out new file mode 100644 index 0000000000..bc142fc384 --- /dev/null +++ b/contrib/test_decoding/expected/catalog_change_snapshot.out @@ -0,0 +1,44 @@ +Parsed test spec with 2 sessions + +starting permutation: s0_init s0_begin s0_savepoint s0_truncate s1_checkpoint s1_get_changes s0_commit s0_begin s0_insert s1_checkpoint s1_get_changes s0_commit s1_get_changes +step s0_init: SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); +?column? +-------- +init +(1 row) + +step s0_begin: BEGIN; +step s0_savepoint: SAVEPOINT sp1; +step s0_truncate: TRUNCATE tbl1; +step s1_checkpoint: CHECKPOINT; +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0'); +data +---- +(0 rows) + +step s0_commit: COMMIT; +step s0_begin: BEGIN; +step s0_insert: INSERT INTO tbl1 VALUES (1); +step s1_checkpoint: CHECKPOINT; +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0'); +data +--------------------------------------- +BEGIN +table public.tbl1: TRUNCATE: (no-flags) +COMMIT +(3 rows) + +step s0_commit: COMMIT; +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0'); +data +------------------------------------------------------------- +BEGIN +table public.tbl1: INSERT: val1[integer]:1 val2[integer]:null +COMMIT +(3 rows) + +?column? +-------- +stop +(1 row) + diff --git a/contrib/test_decoding/specs/catalog_change_snapshot.spec b/contrib/test_decoding/specs/catalog_change_snapshot.spec new file mode 100644 index 0000000000..43c0b64289 --- /dev/null +++ b/contrib/test_decoding/specs/catalog_change_snapshot.spec @@ -0,0 +1,32 @@ +setup +{ + DROP TABLE IF EXISTS tbl1; + CREATE TABLE tbl1 (val1 integer, val2 integer); +} + +teardown +{ + DROP TABLE tbl1; + SELECT 'stop' FROM pg_drop_replication_slot('isolation_slot'); +} + +session "s0" +setup { SET synchronous_commit=on; } +step "s0_init" { SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); } +step "s0_begin" { BEGIN; } +step "s0_savepoint" { SAVEPOINT sp1; } +step "s0_truncate" { TRUNCATE tbl1; } +step "s0_insert" { INSERT INTO tbl1 VALUES (1); } +step "s0_commit" { COMMIT; } + +session "s1" +setup { SET synchronous_commit=on; } +step "s1_checkpoint" { CHECKPOINT; } +step "s1_get_changes" { SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0'); } + +# For the transaction that TRUNCATEd the table tbl1, the last decoding decodes +# only its COMMIT record, because it starts from the RUNNING_XACT record emitted +# during the second checkpoint execution. This transaction must be marked as +# containing catalog changes during decoding the COMMIT record and the decoding +# of the INSERT record must read the pg_class with the correct historic snapshot. +permutation "s0_init" "s0_begin" "s0_savepoint" "s0_truncate" "s1_checkpoint" "s1_get_changes" "s0_commit" "s0_begin" "s0_insert" "s1_checkpoint" "s1_get_changes" "s0_commit" "s1_get_changes" diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 18cf931822..ade48bd71e 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -629,6 +629,25 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, commit_time = parsed->origin_timestamp; } + /* + * Mark the top transaction and its subtransactions as containing catalog + * changes, if the commit record has invalidation message. This is necessary + * for the case where we decode only the commit record of the transaction + * that actually has done catalog changes. + */ + if (parsed->xinfo & XACT_XINFO_HAS_INVALS) + { + ReorderBufferXidSetCatalogChanges(ctx->reorder, xid, buf->origptr); + + for (int i = 0; i < parsed->nsubxacts; i++) + { + ReorderBufferAssignChild(ctx->reorder, xid, parsed->subxacts[i], + buf->origptr); + ReorderBufferXidSetCatalogChanges(ctx->reorder, parsed->subxacts[i], + buf->origptr); + } + } + SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid, parsed->nsubxacts, parsed->subxacts); ^ permalink raw reply [nested|flat] 2+ messages in thread
end of thread, other threads:[~2022-02-21 10:29 UTC | newest] Thread overview: 2+ 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]> 2022-02-21 10:29 Re: [BUG] Logical replication failure "ERROR: could not map filenode "base/13237/442428" to relation OID" with catalog modifying txns Drouvot, Bertrand <[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