From b93872bb6505bc4e8c2233394b1a5854c3599e3c Mon Sep 17 00:00:00 2001 From: Amit Kapila Date: Mon, 2 Nov 2026 15:00:36 +0530 Subject: [PATCH v88] Implement the conflict insertion infrastructure for the conflict log table. Populate the conflict log table whenever a logical replication conflict is detected. Each row records the remote transaction (xid, final LSN, commit timestamp), the replica identity key of the conflicting row, and metadata of the local conflicting tuples. Only resolved (LOG-level) conflicts are recorded in the table. Conflicts that raise an ERROR (such as unique constraint violations) halt replication and abort the transaction, so they are logged exclusively to the server log. Conflict log tables are also excluded from logical decoding. To avoid serializing full tuples that could exceed the 1GB varlena limit and halt the apply worker, the 'remote_tuple' column is removed from the table, 'local_conflicts' records only transaction metadata, and tables using REPLICA IDENTITY FULL set 'replica_identity_full' to true with 'replica_identity' set to NULL. replica_identity: When the table has a replica identity index or primary key, this column stores a JSON object keyed by column name. A value whose output exceeds 64kB is replaced by a marker recording its length ({"omitted":true,"length":...}), and 'has_omitted_values' is set to true. Values are omitted rather than truncated so that join predicates against the table never silently give wrong answers. local_conflicts: This column is an array of JSON objects (json[]), so that conflicts involving several local tuples can be recorded in the future. Each element has the format: [ { "xid": "1001", "commit_ts": "2025-12-25 10:00:00+05:30", "origin": "node_1" }, ... ] Example: SELECT remote_xid, relname, apply_origin, replica_identity, local_conflicts[1] ->> 'xid' AS local_xid FROM pg_conflict.pg_conflict_log_16396; remote_xid | relname | apply_origin | replica_identity | lo ------------+----------+---------------+------------------+----------- 760 | test | pg_16406 | {"a":"1"} | 77 765 | conf_tab | pg_16406 | {"a":"2"} | 775 With two_phase enabled, a prepared transaction that recorded a conflict holds a lock on the conflict log table until the apply worker finishes needing a stronger lock on the table (such as TRUNCATE) can block the worker's next conflict insert and stall replication without being detected as a deadlock. Avoid this by refusing DROP SUBSCRIPTION and the change of conflict_log_destination to 'log' while prepared transactions having the apply worker error out, or wait only for a bounded time, instead of waiting indefinitely for the lock on the conflict log table. Author: Dilip Kumar Author: Nisha Moond Author: Zhijie Hou Reviewed-by: shveta malik Reviewed-by: Amit Kapila Reviewed-by: vignesh C Reviewed-by: shihao zhong Reviewed-by: Kedar Anavardekar Reviewed-by: Masahiko Sawada Discussion: https://postgr.es/m/CAFiTN-u5D5o_AGNbHRZHaOqAMWkxLf+hSk_r9X3gv6HbLOB5+g@mail.gmail.com --- doc/src/sgml/logical-replication.sgml | 813 ++++++++++++++------- doc/src/sgml/ref/alter_subscription.sgml | 12 +- doc/src/sgml/ref/create_subscription.sgml | 22 +- doc/src/sgml/ref/drop_subscription.sgml | 16 +- src/backend/commands/subscriptioncmds.c | 125 +++- src/backend/executor/execReplication.c | 2 +- src/backend/replication/logical/conflict.c | 739 ++++++++++++++++++- src/backend/replication/logical/worker.c | 108 ++- src/include/replication/conflict.h | 2 + src/include/replication/worker_internal.h | 5 + src/include/utils/rel.h | 13 +- src/test/regress/expected/subscription.out | 13 +- src/test/regress/sql/subscription.sql | 7 + src/test/subscription/t/035_conflicts.pl | 416 ++++++++++- 14 files changed, 1920 insertions(+), 373 deletions(-) diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 4701a3d9d18..0d5e7098930 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -2033,120 +2033,134 @@ Included in publications: operations will simply be skipped. - - Additional logging is triggered, and the conflict statistics are collected (displayed in the - pg_stat_subscription_stats view) - in the following conflict cases: - - - insert_exists - - - Inserting a row that violates a NOT DEFERRABLE - unique constraint. Note that to log the origin and commit - timestamp details of the conflicting key, - track_commit_timestamp - should be enabled on the subscriber. In this case, an error will be - raised until the conflict is resolved manually. - - - - - update_origin_differs - - - Updating a row that was previously modified by another origin. - Note that this conflict can only be detected when - track_commit_timestamp - is enabled on the subscriber. Currently, the update is always applied - regardless of the origin of the local row. - - - - - update_exists - - - The updated value of a row violates a NOT DEFERRABLE - unique constraint. Note that to log the origin and commit - timestamp details of the conflicting key, - track_commit_timestamp - should be enabled on the subscriber. In this case, an error will be - raised until the conflict is resolved manually. Note that when updating a - partitioned table, if the updated row value satisfies another partition - constraint resulting in the row being inserted into a new partition, the - insert_exists conflict may arise if the new row - violates a NOT DEFERRABLE unique constraint. - - - - - update_deleted - - - The tuple to be updated was concurrently deleted by another origin. The - update will simply be skipped in this scenario. Note that this conflict - can only be detected when - track_commit_timestamp - and retain_dead_tuples - are enabled. Note that if a tuple cannot be found due to the table being - truncated, only an update_missing conflict will - arise. Additionally, if the tuple was deleted by the same origin, an - update_missing conflict will arise. - - - - - update_missing - - - The row to be updated was not found. The update will simply be - skipped in this scenario. - - - - - delete_origin_differs - - - Deleting a row that was previously modified by another origin. Note that - this conflict can only be detected when - track_commit_timestamp - is enabled on the subscriber. Currently, the delete is always applied - regardless of the origin of the local row. - - - - - delete_missing - - - The row to be deleted was not found. The delete will simply be - skipped in this scenario. - - - - - multiple_unique_conflicts - - - Inserting or updating a row violates multiple - NOT DEFERRABLE unique constraints. Note that to log - the origin and commit timestamp details of conflicting keys, ensure - that track_commit_timestamp - is enabled on the subscriber. In this case, an error will be raised until - the conflict is resolved manually. - - - - - Note that there are other conflict scenarios, such as exclusion constraint - violations. Currently, we do not provide additional details for them in the - log. - + + Conflict logging + + Conflict logging is triggered, and conflict statistics are collected + (displayed in the + pg_stat_subscription_stats + view) for the conflict cases described below. Depending on the value of + conflict_log_destination, + the conflict information is written to the server log, a conflict log table, + or both. + - - The log format for logical replication conflicts is as follows: + + The supported conflict cases are: + + + insert_exists + + + Inserting a row that violates a NOT DEFERRABLE + unique constraint. Note that to log the origin and commit + timestamp details of the conflicting key, + track_commit_timestamp + should be enabled on the subscriber. In this case, an error will be + raised until the conflict is resolved manually. + + + + + update_origin_differs + + + Updating a row that was previously modified by another origin. + Note that this conflict can only be detected when + track_commit_timestamp + is enabled on the subscriber. Currently, the update is always applied + regardless of the origin of the local row. + + + + + update_exists + + + The updated value of a row violates a NOT DEFERRABLE + unique constraint. Note that to log the origin and commit + timestamp details of the conflicting key, + track_commit_timestamp + should be enabled on the subscriber. In this case, an error will be + raised until the conflict is resolved manually. Note that when updating a + partitioned table, if the updated row value satisfies another partition + constraint resulting in the row being inserted into a new partition, the + insert_exists conflict may arise if the new row + violates a NOT DEFERRABLE unique constraint. + + + + + update_deleted + + + The tuple to be updated was concurrently deleted by another origin. The + update will simply be skipped in this scenario. Note that this conflict + can only be detected when + track_commit_timestamp + and retain_dead_tuples + are enabled. Note that if a tuple cannot be found due to the table being + truncated, only an update_missing conflict will + arise. Additionally, if the tuple was deleted by the same origin, an + update_missing conflict will arise. + + + + + update_missing + + + The row to be updated was not found. The update will simply be + skipped in this scenario. + + + + + delete_origin_differs + + + Deleting a row that was previously modified by another origin. Note that + this conflict can only be detected when + track_commit_timestamp + is enabled on the subscriber. Currently, the delete is always applied + regardless of the origin of the local row. + + + + + delete_missing + + + The row to be deleted was not found. The delete will simply be + skipped in this scenario. + + + + + multiple_unique_conflicts + + + Inserting or updating a row violates multiple + NOT DEFERRABLE unique constraints. Note that to log + the origin and commit timestamp details of conflicting keys, ensure + that track_commit_timestamp + is enabled on the subscriber. In this case, an error will be raised until + the conflict is resolved manually. + + + + + Note that there are other conflict scenarios, such as exclusion constraint + violations. Currently, we do not provide additional details for them in the + log. + + + + File-based logging + + If the conflict_log_destination + parameter is set to log or all, conflicts + are logged to the server log using the following format: LOG: conflict detected on relation "schemaname.tablename": conflict=conflict_type DETAIL: detailed_explanation[: detail_values [, ... ]]. @@ -2159,182 +2173,437 @@ DETAIL: detailed_explanation[: replica identity {(column_name , ...)=(column_value , ...) | full (column_name , ...)=(column_value , ...)} - The log provides the following information: - - - LOG - + The log provides the following information: + + + LOG + + + + + schemaname.tablename + identifies the local relation involved in the conflict. + + + + + conflict_type is the type of conflict that occurred + (e.g., insert_exists, update_exists). + + + + + + + + DETAIL + - schemaname.tablename - identifies the local relation involved in the conflict. + detailed_explanation includes + the origin, transaction ID, and commit timestamp of the transaction that + modified the local row, if available. - conflict_type is the type of conflict that occurred - (e.g., insert_exists, update_exists). + The key section includes the key values of the local + row that violated a unique constraint for + insert_exists, update_exists or + multiple_unique_conflicts conflicts. + + + + + The local row section includes the local row if its + origin differs from the remote row for + update_origin_differs or delete_origin_differs + conflicts, or if the key value conflicts with the remote row for + insert_exists, update_exists or + multiple_unique_conflicts conflicts. + + + + + The remote row section includes the new row from + the remote insert or update operation that caused the conflict. Note that + for an update operation, the column value of the new row will be null + if the value is unchanged and toasted. + + + + + The replica identity section includes the replica + identity key values that were used to search for the existing local + row to be updated or deleted. This may include the full row value + if the local relation is marked with + REPLICA IDENTITY FULL. + + + + + column_name is the column name. + For local row, remote row, and + replica identity full cases, column names are + logged only if the user lacks the privilege to access all columns of + the table. If column names are present, they appear in the same order + as the corresponding column values. + + + + + column_value is the column value. + The large column values are truncated to 64 bytes. + + + + + Note that in case of multiple_unique_conflicts conflict, + multiple detailed_explanation + and detail_values lines + will be generated, each detailing the conflict information associated + with distinct unique constraints. - + + + + - - DETAIL - - - - - detailed_explanation includes - the origin, transaction ID, and commit timestamp of the transaction that - modified the local row, if available. - - - - - The key section includes the key values of the local - row that violated a unique constraint for - insert_exists, update_exists or - multiple_unique_conflicts conflicts. - - - - - The local row section includes the local row if its - origin differs from the remote row for - update_origin_differs or delete_origin_differs - conflicts, or if the key value conflicts with the remote row for - insert_exists, update_exists or - multiple_unique_conflicts conflicts. - - - - - The remote row section includes the new row from - the remote insert or update operation that caused the conflict. Note that - for an update operation, the column value of the new row will be null - if the value is unchanged and toasted. - - - - - The replica identity section includes the replica - identity key values that were used to search for the existing local - row to be updated or deleted. This may include the full row value - if the local relation is marked with - REPLICA IDENTITY FULL. - - - - - column_name is the column name. - For local row, remote row, and - replica identity full cases, column names are - logged only if the user lacks the privilege to access all columns of - the table. If column names are present, they appear in the same order - as the corresponding column values. - - - - - column_value is the column value. - The large column values are truncated to 64 bytes. - - - - - Note that in case of multiple_unique_conflicts conflict, - multiple detailed_explanation - and detail_values lines - will be generated, each detailing the conflict information associated - with distinct unique - constraints. - - - - - - - + + Table-based logging + + If the conflict_log_destination + parameter is set to table or all, + a dedicated conflict log table will be automatically created. This table is + created in the pg_conflict namespace. The name of the + conflict log table is + pg_conflict_log_<subid>. The predefined + schema of this table is detailed in + . + - - Logical replication operations are performed with the privileges of the role - which owns the subscription. Permissions failures on target tables will - cause replication conflicts, as will enabled - row-level security on target tables - that the subscription owner is subject to, without regard to whether any - policy would ordinarily reject the INSERT, - UPDATE, DELETE or - TRUNCATE which is being replicated. This restriction on - row-level security may be lifted in a future version of - PostgreSQL. - + + + Only resolved conflicts (logged at LOG level where the + apply transaction continues) are recorded in the conflict log table. + Conflicts that raise an ERROR halt replication and are + reported to the server log alone. + + - - A conflict that produces an error will stop the replication; it must be - resolved manually by the user. Details about the conflict can be found in - the subscriber's server log. - + + The conflict log table serves as an audit trail for resolved conflicts + where replication continues and the transaction successfully commits + (such as update_origin_differs, + update_missing, update_deleted, + delete_origin_differs, and + delete_missing). In these cases, local data may diverge + or be overwritten, making a queryable audit log valuable for DBAs to + understand post-hoc why a row changed or was skipped. + - - The resolution can be done either by changing data or permissions on the subscriber so - that it does not conflict with the incoming change or by skipping the - transaction that conflicts with the existing data. When a conflict produces - an error, the replication won't proceed, and the logical replication worker will - emit the following kind of message to the subscriber's server log: + + In contrast, conflicts raised at ERROR level (such as + insert_exists, update_exists, and + multiple_unique_conflicts) halt replication and abort the + entire apply transaction. Because the transaction rolls back and no data + modifications survive on the subscriber, no data divergence occurs. These + errors are always reported to the server log (regardless of the + conflict_log_destination setting) so that monitoring and + alerting systems can capture the replication failure. + + + + Conflict Log Table Schema + + + + Column + Type + Description + + + + + relid + oid + The OID of the local table where the conflict occurred. + + + schemaname + text + The schema name of the conflicting table. + + + relname + text + The name of the conflicting table. + + + conflict_type + text + The type of conflict that occurred (e.g., update_origin_differs). + + + remote_xid + xid + The remote transaction ID that caused the conflict. + + + remote_commit_lsn + pg_lsn + The commit LSN of the remote transaction, or the prepare LSN if the conflict occurs in a transaction that is prepared on the subscriber (see two_phase), because the changes are applied at prepare time before COMMIT PREPARED arrives. This is NULL if the conflict is handled by a parallel apply worker, because in-progress transactions are streamed and applied before the remote commit or prepare LSN is known. + + + remote_commit_ts + timestamptz + The remote commit timestamp of the remote transaction. This is NULL if the conflict is handled by a parallel apply worker, because in-progress transactions are streamed and applied before the remote commit timestamp is known. It is also NULL if the conflict occurs in a transaction that is prepared on the subscriber (see two_phase), because the commit timestamp is known only when COMMIT PREPARED arrives. + + + apply_origin + text + The local replication origin under which the change was applied. This is pg_<subid> for the apply worker, or pg_<subid>_<relid> for a tablesync worker. + + + replica_identity_full + boolean + True if the conflicting local row was located using the full tuple, rather than the replica identity key of the local table. + + + replica_identity + json + The replica identity key values identifying the conflicting row, + as a JSON object keyed by column name; see below for + how the values are represented. This is NULL when + replica_identity_full is true. + + + local_conflicts + json[] + + An array of JSON objects representing the state of existing local + row(s) that caused the conflict. Each object includes the local + transaction ID (xid), commit timestamp + (commit_ts), and the name of the replication + origin that last modified the row (origin). + The origin field is NULL if + the transaction had no replication origin set or if that origin has + since been dropped. The column itself is NULL + if there are no local conflicting rows (e.g., in + update_missing and + delete_missing conflicts). + + + + has_omitted_values + boolean + Indicates whether any value in + replica_identity was too large to be recorded and + was replaced by an omitted marker. + + + +
+ + + Every value in replica_identity is written as a + JSON string containing the same text a + SELECT of that column would show. Values are quoted + whatever the column's type, so numbers, booleans and jsonb + documents all appear as strings: + +CREATE TABLE t (id int, tag text, doc jsonb, PRIMARY KEY (id, tag, doc)); + +SELECT replica_identity FROM pg_conflict.pg_conflict_log_16392; + replica_identity +------------------------------------------------- + {"id":"1","tag":"abc","doc":"{\"x\": 1}"} + + + + + Extracting a value with the ->> operator yields a + form that can be cast back to the column's type, so a recorded replica + identity can be used to locate the row it identifies: + +SELECT t.* FROM pg_conflict.pg_conflict_log_16392 c, t + WHERE t.id = (c.replica_identity->>'id')::int + AND t.tag = c.replica_identity->>'tag' + AND t.doc = (c.replica_identity->>'doc')::jsonb; + + Because container values such as jsonb, arrays and composite + types are recorded as strings rather than as nested + JSON, looking inside one requires an explicit cast: + +SELECT replica_identity->'doc'->>'x' FROM ...; -- NULL +SELECT (replica_identity->>'doc')::jsonb->>'x' FROM ...; -- 1 + + + + + The apply worker's , + , + , + and + settings can affect how a + key value is written. If these settings differ between conflicts, the + same key can be recorded in different forms, and comparing the recorded + text may miss some of them. For settings that affect only the + representation, such as and + , cast the value back to the + column's type to compare equivalent forms. Here are examples for a + bytea key column b and a + regclass[] key column r: + +SELECT * FROM ... WHERE replica_identity->>'b' = '\x0102'; -- may miss rows +SELECT * FROM ... WHERE (replica_identity->>'b')::bytea = '\x0102'; -- matches all +SELECT * FROM ... WHERE replica_identity->>'r' = '{pg_class}'; -- may miss rows +SELECT * FROM ... WHERE (replica_identity->>'r')::regclass[] = '{pg_class}'; -- matches all + + + + + and + affect how values are written and + interpreted. When casting such recorded values, use the same settings + that the apply worker used when recording the conflict; otherwise, the + original value may not be restored correctly. For example, + 02/01/2025 means January 2, 2025 with + DMY, but February 1, 2025 with MDY. + For intervals, the hours in -1 02:00:00 are positive + with postgres, but negative with + sql_standard. For floating-point keys, keep + at 1 or higher when recording; + zero or negative values round the output and can prevent exact + restoration. + + + + If a key value's text representation is larger than 64kB, it is not + recorded. In its place the conflict log stores a small object marking it + as omitted and giving its length in bytes, and sets + has_omitted_values for that row: + + {"id":"1","tag":"abc","doc":{"omitted":true,"length":70000}} + + Large values are dropped rather than shortened, because a shortened value + would still look like a real one and would give wrong answers when + compared or joined against the table. + + + + Use the flag to find rows with a missing value, without having to search + the json column: + +SELECT * FROM pg_conflict.pg_conflict_log_16392 WHERE has_omitted_values; + + A value omitted this way cannot be recovered from the conflict log table. + + + + Some limit is needed because replica_identity is a + json value, which cannot exceed 1GB, and the text of a value + can be much longer than the value itself. However, the 64kB limit is + applied only after a key value's text representation has been produced, + so it limits how much text is kept in the conflict log table but does not + stop that text from exceeding 1GB in the first place. Normally this + cannot happen, because the publisher had to produce the same text, within + the same 1GB limit, in order to send the value. That does not hold when, + for example, the key column has a different data type on the subscriber, + the subscription uses + binary + mode so the value is not sent as text, or the subscriber's + or + setting differs from the + publisher's. In such cases the text can exceed 1GB, and the apply worker + fails with an error when it tries to log the conflict. + + + + + + Notes + + Logical replication operations are performed with the privileges of the role + which owns the subscription. Permissions failures on target tables will + cause replication conflicts, as will enabled + row-level security on target tables + that the subscription owner is subject to, without regard to whether any + policy would ordinarily reject the INSERT, + UPDATE, DELETE or + TRUNCATE which is being replicated. This restriction on + row-level security may be lifted in a future version of + PostgreSQL. + + + + A conflict that produces an error will stop the replication; it must be + resolved manually by the user. Details about the conflict can be found in + the subscriber's server log. + + + + The resolution can be done either by changing data or permissions on the subscriber so + that it does not conflict with the incoming change or by skipping the + transaction that conflicts with the existing data. When a conflict produces + an error, the replication won't proceed, and the logical replication worker will + emit the following kind of message to the subscriber's server log: ERROR: conflict detected on relation "public.test": conflict=insert_exists DETAIL: Could not apply remote change: remote row (1, 'remote'). Key already exists in unique index "test_pkey", modified locally in transaction 800 at 2026-01-16 18:15:25.652759+09: key (c)=(1), local row (1, 'local'). CONTEXT: processing remote data for replication origin "pg_16395" during "INSERT" for replication target relation "public.test" in transaction 725 finished at 0/014C0378 - The LSN of the transaction that contains the change violating the constraint and - the replication origin name can be found from the server log (LSN 0/014C0378 and - replication origin pg_16395 in the above case). The - transaction that produced the conflict can be skipped by using - ALTER SUBSCRIPTION ... SKIP - with the finish LSN - (i.e., LSN 0/014C0378). The finish LSN could be an LSN at which the transaction - is committed or prepared on the publisher. Alternatively, the transaction can - also be skipped by calling the - pg_replication_origin_advance() function. - Before using this function, the subscription needs to be disabled temporarily - either by - ALTER SUBSCRIPTION ... DISABLE or, the - subscription can be used with the - disable_on_error - option. Then, you can use pg_replication_origin_advance() - function with the node_name (i.e., pg_16395) - and the next LSN of the finish LSN (i.e., 0/014C0379). The current position of - origins can be seen in the - pg_replication_origin_status system view. - Please note that skipping the whole transaction includes skipping changes that - might not violate any constraint. This can easily make the subscriber - inconsistent. - The additional details regarding conflicting rows, such as their origin and - commit timestamp can be seen in the DETAIL line of the - log. But note that this information is only available when - track_commit_timestamp - is enabled on the subscriber. Users can use this information to decide - whether to retain the local change or adopt the remote alteration. For - instance, the DETAIL line in the above log indicates that - the existing row was modified locally. Users can manually perform a - remote-change-win. - - - - When the - streaming - mode is parallel, the finish LSN of failed transactions - may not be logged. In that case, it may be necessary to change the streaming - mode to on or off and cause the same - conflicts again so the finish LSN of the failed transaction will be written - to the server log. For the usage of finish LSN, please refer to ALTER SUBSCRIPTION ... - SKIP. - + The LSN of the transaction that contains the change violating the constraint and + the replication origin name can be found from the server log (LSN 0/014C0378 and + replication origin pg_16395 in the above case). The + transaction that produced the conflict can be skipped by using + ALTER SUBSCRIPTION ... SKIP + with the finish LSN + (i.e., LSN 0/014C0378). The finish LSN could be an LSN at which the transaction + is committed or prepared on the publisher. Alternatively, the transaction can + also be skipped by calling the + pg_replication_origin_advance() function. + Before using this function, the subscription needs to be disabled temporarily + either by + ALTER SUBSCRIPTION ... DISABLE or, the + subscription can be used with the + disable_on_error + option. Then, you can use pg_replication_origin_advance() + function with the node_name (i.e., pg_16395) + and the next LSN of the finish LSN (i.e., 0/014C0379). The current position of + origins can be seen in the + pg_replication_origin_status system view. + Please note that skipping the whole transaction includes skipping changes that + might not violate any constraint. This can easily make the subscriber + inconsistent. + The additional details regarding conflicting rows, such as their origin and + commit timestamp can be seen in the DETAIL line of the + log. But note that this information is only available when + track_commit_timestamp + is enabled on the subscriber. Users can use this information to decide + whether to retain the local change or adopt the remote alteration. For + instance, the DETAIL line in the above log indicates that + the existing row was modified locally. Users can manually perform a + remote-change-win. + + + + When the + streaming + mode is parallel, the finish LSN of failed transactions + may not be logged. In that case, it may be necessary to change the streaming + mode to on or off and cause the same + conflicts again so the finish LSN of the failed transaction will be written + to the server log. For the usage of finish LSN, please refer to ALTER SUBSCRIPTION ... + SKIP. + + @@ -2448,6 +2717,14 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER key or replica identity defined for it. + + + + Conflict log tables (see conflict_log_destination parameter) + are never published, even when using FOR ALL TABLES in a + publication. + + diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 545264e8a0a..070a0d05d58 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -366,7 +366,17 @@ ALTER SUBSCRIPTION name RENAME TO < When the conflict_log_destination parameter is set to table or all, the system automatically creates the conflict log table. Conversely, if the destination is changed to - log, the conflict log table is automatically dropped. + log, logging to the table stops and the conflict log + table is automatically dropped. + + + + When two_phase + is enabled, changing conflict_log_destination + from table or all to + log is only allowed when the subscription is disabled + and there are no pending prepared transactions, because such transactions + may hold a lock on the conflict log table being dropped. diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml index 6248c228a64..0824d6e3dea 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -272,8 +272,10 @@ CREATE SUBSCRIPTION subscription_namelog - Conflict details are recorded in the server log. - This is the default behavior. + Conflict details are recorded in the server log. This is + the default behavior. See + + for details. @@ -283,15 +285,20 @@ CREATE SUBSCRIPTION subscription_name The system automatically creates a structured table named pg_conflict_log_<subid> in the - pg_conflict schema. This allows for easy - querying and analysis of conflicts. + pg_conflict schema to record resolved + conflicts. Conflicts raised at ERROR level + (which halt replication and abort the transaction) are not + recorded in the table; they are always reported to the server log. + See + for details. The conflict log table is strictly tied to the lifecycle of the subscription or the conflict_log_destination setting. If the subscription is dropped, or if the destination is changed to - log, the table is dropped. + log, the table and all its recorded conflict data are + permanently deleted. @@ -300,8 +307,9 @@ CREATE SUBSCRIPTION subscription_nameall - This is equivalent to configuring both destinations log - and table together. + Records resolved conflict details to both the server log and the + conflict log table. Conflicts raised at ERROR level + are reported to the server log alone. diff --git a/doc/src/sgml/ref/drop_subscription.sgml b/doc/src/sgml/ref/drop_subscription.sgml index e4c646fcf2a..909f6d1bf43 100644 --- a/doc/src/sgml/ref/drop_subscription.sgml +++ b/doc/src/sgml/ref/drop_subscription.sgml @@ -118,11 +118,19 @@ DROP SUBSCRIPTION [ IF EXISTS ] name - If a conflict log table exists for the subscription (that is, when - + If conflict_log_destination is set to table - or all), DROP SUBSCRIPTION automatically drops - the associated conflict log table. + or all, the subscription has an associated conflict log + table, which DROP SUBSCRIPTION drops along with the + subscription. However, DROP SUBSCRIPTION cannot be + executed while the subscription has prepared transactions (see + two_phase), + because such a transaction may hold a lock on the conflict log table that is + released only when the apply worker processes the corresponding + COMMIT PREPARED or ROLLBACK PREPARED. + Resolve these transactions first (see + pg_prepared_xacts) + before dropping the subscription. diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 36bb20a1f6e..282d45ad2aa 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1495,8 +1495,8 @@ AlterSubscription_refresh_seq(Subscription *sub, char *conninfo) } /* - * Common checks for altering failover, two_phase, and retain_dead_tuples - * options. + * Common checks for altering failover, two_phase, retain_dead_tuples, and + * conflict_log_destination options. */ static void CheckAlterSubOption(Subscription *sub, const char *option, @@ -1504,13 +1504,17 @@ CheckAlterSubOption(Subscription *sub, const char *option, { Assert(strcmp(option, "failover") == 0 || strcmp(option, "two_phase") == 0 || - strcmp(option, "retain_dead_tuples") == 0); + strcmp(option, "retain_dead_tuples") == 0 || + strcmp(option, "conflict_log_destination") == 0); /* - * Altering the retain_dead_tuples option does not update the slot on the - * publisher. + * Only the failover and two_phase options change the slot on the + * publisher. Altering retain_dead_tuples or conflict_log_destination does + * not. */ - Assert(!slot_needs_update || strcmp(option, "retain_dead_tuples") != 0); + if (slot_needs_update) + Assert(strcmp(option, "failover") == 0 || + strcmp(option, "two_phase") == 0); /* * Do not allow changing the option if the subscription is enabled. This @@ -1555,6 +1559,10 @@ CheckAlterSubOption(Subscription *sub, const char *option, * avoid the race conditions described above, but we maintain the * restriction for both enable and disable operations for the sake of * consistency. + * + * Also, when two_phase is enabled, do not allow changing the + * conflict_log_destination option to 'log' when the subscription is + * enabled; see AlterSubscription() and DropSubscription() for details. */ if (sub->enabled) ereport(ERROR, @@ -2103,6 +2111,57 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, bool update_relid; Oid relid = InvalidOid; + /* + * Similar to DropSubscription(), when two_phase is + * enabled, changing to 'log' drops the conflict log + * table, which can be blocked forever if a prepared + * transaction of the subscription holds a lock on it. + * So allow it only when the subscription is disabled, + * no worker is running, and there are no prepared + * transactions. + * + * XXX Another option would be to allow this on an + * enabled subscription by restarting its workers. We + * chose to require the subscription to be disabled + * instead, following the precedent of existing + * options such as two_phase and retain_dead_tuples + * (see CheckAlterSubOption()). + */ + if (CONFLICTS_LOGGED_TO_TABLE(old_dest) && + !CONFLICTS_LOGGED_TO_TABLE(opts.conflictlogdest) && + sub->twophasestate != LOGICALREP_TWOPHASE_STATE_DISABLED) + { + CheckAlterSubOption(sub, "conflict_log_destination", + false, isTopLevel); + + /* + * Note that workers may still survive even if the + * subscription has been disabled. + * + * Ensure workers have already been exited to + * avoid getting prepared transactions while we + * are dropping the conflict log table. + */ + if (logicalrep_workers_find(subid, true, true)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot alter option \"%s\" when logical replication worker is still running", + "conflict_log_destination"), + errhint("Try again after some time."))); + + /* + * Disallow the operation when a prepared + * transaction exists. + */ + if (LookupGXactBySubid(subid)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot alter option \"%s\" when prepared transactions exist", + "conflict_log_destination"), + errdetail("Dropping the conflict log table requires a lock that a prepared transaction of this subscription may be holding."), + errhint("Resolve these transactions and try again."))); + } + values[Anum_pg_subscription_subconflictlogdest - 1] = CStringGetTextDatum(ConflictLogDestNames[opts.conflictlogdest]); replaces[Anum_pg_subscription_subconflictlogdest - 1] = true; @@ -2613,6 +2672,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) Oid subowner; Oid subserver; Oid subconflictlogrelid; + char subtwophasestate; char *subconninfo = NULL; Datum datum; bool isnull; @@ -2666,6 +2726,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) subowner = form->subowner; subserver = form->subserver; subconflictlogrelid = form->subconflictlogrelid; + subtwophasestate = form->subtwophasestate; must_use_password = !superuser_arg(subowner) && form->subpasswordrequired; /* must be owner */ @@ -2742,6 +2803,25 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) } list_free(subworkers); + /* + * Disallow dropping a subscription that has a conflict log table while it + * has a prepared transaction, as we cannot tell whether that transaction + * holds a lock on the table. If it does, dropping the table would wait for + * the lock, but only an apply worker can finish the prepared transaction + * and none can run while we hold the subscription lock. The lock manager + * cannot detect this wait cycle, so the DROP would hang until canceled or + * until the prepared transaction is resolved manually. + */ + if (OidIsValid(subconflictlogrelid) && + subtwophasestate != LOGICALREP_TWOPHASE_STATE_DISABLED && + LookupGXactBySubid(subid)) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot drop subscription \"%s\" when prepared transactions exist", + subname), + errdetail("A prepared transaction of the subscription may hold a lock on the subscription's conflict log table."), + errhint("Resolve these transactions and try again.")); + /* * Remove the no-longer-useful entry in the launcher's table of apply * worker start times. @@ -2970,6 +3050,7 @@ AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId) { Form_pg_subscription form; AclResult aclresult; + HeapTuple curtup = tup; form = (Form_pg_subscription) GETSTRUCT(tup); @@ -3020,14 +3101,35 @@ AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId) if (OidIsValid(form->subserver)) GetUserMappingExtended(newOwnerId, form->subserver, WARNING); - form->subowner = newOwnerId; - CatalogTupleUpdate(rel, &tup->t_self, tup); - - /* Update owner of the conflict log table if it exists. */ + /* + * Update owner of the conflict log table first, if it exists, before + * updating the subscription's row. Changing its owner can wait for a + * prepared transaction of this subscription that holds a lock on it. If + * we had already updated the row, the apply worker could wait for us when + * it updates the row itself, e.g. to disable the subscription after an + * error, while we wait for the prepared transaction to be resolved by the + * apply worker. + */ if (OidIsValid(form->subconflictlogrelid)) + { + Oid subid = form->oid; + ATExecChangeOwner(form->subconflictlogrelid, newOwnerId, true, AccessExclusiveLock); + /* + * The row may have been updated while we waited for the lock, so + * update its current version. + */ + curtup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid)); + if (!HeapTupleIsValid(curtup)) + elog(ERROR, "cache lookup failed for subscription %u", subid); + form = (Form_pg_subscription) GETSTRUCT(curtup); + } + + form->subowner = newOwnerId; + CatalogTupleUpdate(rel, &curtup->t_self, curtup); + /* Update owner dependency reference */ changeDependencyOnOwner(SubscriptionRelationId, form->oid, @@ -3039,6 +3141,9 @@ AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId) /* Wake up related background processes to handle this change quickly. */ ApplyLauncherWakeupAtCommit(); LogicalRepWorkersWakeupAtCommit(form->oid); + + if (curtup != tup) + heap_freetuple(curtup); } /* diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index 13644328164..1fc68b0ecef 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -876,7 +876,7 @@ CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate, if (conflicttuples) ReportApplyConflict(estate, resultRelInfo, ERROR, list_length(conflicttuples) > 1 ? CT_MULTIPLE_UNIQUE_CONFLICTS : type, - searchslot, remoteslot, conflicttuples); + NULL, searchslot, remoteslot, conflicttuples); } /* diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index 90431ffce98..3a05f85b56e 100644 --- a/src/backend/replication/logical/conflict.c +++ b/src/backend/replication/logical/conflict.c @@ -16,17 +16,26 @@ #include "access/commit_ts.h" #include "access/genam.h" +#include "access/heapam.h" #include "access/tableam.h" +#include "access/twophase.h" #include "catalog/heap.h" #include "catalog/pg_am.h" #include "catalog/pg_namespace.h" #include "catalog/toasting.h" #include "executor/executor.h" +#include "funcapi.h" #include "pgstat.h" #include "replication/conflict.h" #include "replication/worker_internal.h" #include "storage/lmgr.h" +#include "storage/proc.h" +#include "utils/array.h" +#include "utils/builtins.h" +#include "utils/guc.h" +#include "utils/json.h" #include "utils/lsyscache.h" +#include "utils/pg_lsn.h" /* * String representations for the supported conflict logging destinations. @@ -56,13 +65,18 @@ typedef struct ConflictLogColumnDef * type OID; the table is created in this column order by * create_conflict_log_table(). * - * The tuple/key columns (replica_identity, remote_tuple, local_conflicts) are - * typed json rather than jsonb on purpose: they hold an exact audit snapshot - * of the applied tuples and replica identity, and json preserves the verbatim + * The replica_identity and local_conflicts columns are typed json rather than + * jsonb on purpose: they hold an exact audit snapshot of the replica identity + * key values and local conflict metadata, and json preserves the verbatim * representation whereas jsonb would normalize it. Indexing them (jsonb's main * advantage) wouldn't help anyway, as the conflict log is looked up by its * scalar columns (relid, conflict_type, commit timestamp) while these json * columns are per-conflict payload to inspect, not search keys. + * + * 'local_conflicts' is typed as an array of JSON objects (json[]), not a + * single json object, so that a future conflict type needing to record + * multiple local rows for one remote operation doesn't require a + * backward-incompatible schema change. */ static const ConflictLogColumnDef ConflictLogSchema[] = { {.attname = "relid", .atttypid = OIDOID}, @@ -72,15 +86,40 @@ static const ConflictLogColumnDef ConflictLogSchema[] = { {.attname = "remote_xid", .atttypid = XIDOID}, {.attname = "remote_commit_lsn", .atttypid = LSNOID}, {.attname = "remote_commit_ts", .atttypid = TIMESTAMPTZOID}, - {.attname = "remote_origin", .atttypid = TEXTOID}, + {.attname = "apply_origin", .atttypid = TEXTOID}, {.attname = "replica_identity_full", .atttypid = BOOLOID}, {.attname = "replica_identity", .atttypid = JSONOID}, - {.attname = "remote_tuple", .atttypid = JSONOID}, - {.attname = "local_conflicts", .atttypid = JSONARRAYOID} + {.attname = "local_conflicts", .atttypid = JSONARRAYOID}, + {.attname = "has_omitted_values", .atttypid = BOOLOID} }; #define NUM_CONFLICT_ATTRS ((AttrNumber) lengthof(ConflictLogSchema)) +/* + * Largest replica identity value recorded verbatim; anything larger is replaced + * by a marker recording its length. See build_index_key_json(). + * + * The cap applies to the length of the type's text output before JSON escaping. + * Since escape_json() expands each byte at most sixfold (to \uXXXX), and an + * index has at most INDEX_MAX_KEYS (32) key columns, the worst-case JSON output + * is 64kB * 6 * 32 = 12MB, comfortably inside the 1GB limit on a json value. + * Ordinary keys are orders of magnitude below the cap, so in practice nothing + * is ever omitted. + */ +#define CONFLICT_MAX_VALUE_SIZE (64 * 1024) + +/* + * Schema for the elements within the 'local_conflicts' JSON array. + */ +static const ConflictLogColumnDef LocalConflictSchema[] = +{ + {.attname = "xid", .atttypid = XIDOID}, + {.attname = "commit_ts", .atttypid = TIMESTAMPTZOID}, + {.attname = "origin", .atttypid = TEXTOID} +}; + +#define NUM_LOCAL_CONFLICT_ATTRS lengthof(LocalConflictSchema) + static const char *const ConflictTypeNames[] = { [CT_INSERT_EXISTS] = "insert_exists", [CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs", @@ -92,6 +131,8 @@ static const char *const ConflictTypeNames[] = { [CT_MULTIPLE_UNIQUE_CONFLICTS] = "multiple_unique_conflicts" }; +static Relation get_conflictlog_dest_and_table(int elevel, + ConflictLogDest *log_dest); static int errcode_apply_conflict(ConflictType type); static void errdetail_apply_conflict(EState *estate, ResultRelInfo *relinfo, @@ -108,8 +149,26 @@ static void get_tuple_desc(EState *estate, ResultRelInfo *relinfo, TupleTableSlot *remoteslot, char **remote_desc, TupleTableSlot *searchslot, char **search_desc, Oid indexoid); +static void build_index_datums_from_slot(EState *estate, Relation localrel, + TupleTableSlot *slot, + Relation indexDesc, Datum *values, + bool *isnull); static char *build_index_value_desc(EState *estate, Relation localrel, TupleTableSlot *slot, Oid indexoid); +static Datum build_index_key_json(EState *estate, + Relation localrel, + Oid replica_index, + TupleTableSlot *slot, + bool *omitted); +static TupleDesc build_local_conflicts_tupledesc(void); +static Datum build_local_conflicts_json_array(List *conflicttuples); +static void insert_conflict_log_tuple(EState *estate, Relation rel, + Relation conflictlogrel, + ConflictType conflict_type, + Oid searchindexoid, + TupleTableSlot *searchslot, + List *conflicttuples); +static void lock_conflictlog_relation(Oid relid); /* * Builds the TupleDesc for the conflict log table. @@ -262,6 +321,11 @@ GetTupleTransactionInfo(TupleTableSlot *localslot, TransactionId *xmin, * This function is used to report a conflict while applying replication * changes. * + * 'relmapentry' is the relation map entry that was used to search the local + * row, or NULL for ERROR-level insert/update conflicts. If the search used + * the replica identity or primary key index, the caller must have locked + * it, which the search itself does, see FindReplTupleInLocalRel(). + * * 'searchslot' should contain the tuple used to search the local row to be * updated or deleted. * @@ -275,25 +339,71 @@ GetTupleTransactionInfo(TupleTableSlot *localslot, TransactionId *xmin, */ void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, - ConflictType type, TupleTableSlot *searchslot, + ConflictType type, LogicalRepRelMapEntry *relmapentry, + TupleTableSlot *searchslot, TupleTableSlot *remoteslot, List *conflicttuples) { Relation localrel = relinfo->ri_RelationDesc; + ConflictLogDest dest; + Oid searchindexoid = InvalidOid; + Relation conflictlogrel; StringInfoData err_detail; + pgstat_report_subscription_conflict(MySubscription->oid, type); + + /* + * Use the replica identity or primary key index that the search actually + * used, which the caller has locked, rather than looking it up again. + * Concurrent DDL could have changed which index that is since the search, + * see comments in FindReplTupleInLocalRel(). + */ + if (relmapentry && relmapentry->idxisreplident) + searchindexoid = relmapentry->localindexoid; + + conflictlogrel = get_conflictlog_dest_and_table(elevel, &dest); + initStringInfo(&err_detail); - /* Form errdetail message by combining conflicting tuples information. */ - foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples) - errdetail_apply_conflict(estate, relinfo, type, searchslot, - conflicttuple->slot, remoteslot, - conflicttuple->indexoid, - conflicttuple->xmin, - conflicttuple->origin, - conflicttuple->ts, - &err_detail); + /* + * Report the conflict to the server log. When the server log is one of + * the destinations (or for ERROR-level conflicts), emit the full details. + * Otherwise (table-only for LOG-level conflicts), emit a shorter message + * noting that the details are captured in the conflict log table. + */ + if (CONFLICTS_LOGGED_TO_LOG(dest)) + { + /* Form errdetail message by combining conflicting tuples information. */ + if (conflicttuples != NIL) + { + foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples) + errdetail_apply_conflict(estate, relinfo, type, searchslot, + conflicttuple->slot, remoteslot, + conflicttuple->indexoid, + conflicttuple->xmin, + conflicttuple->origin, + conflicttuple->ts, + &err_detail); + } + else + { + errdetail_apply_conflict(estate, relinfo, type, searchslot, + NULL, remoteslot, + InvalidOid, + InvalidTransactionId, + InvalidReplOriginId, + 0, + &err_detail); + } + } + else + { + Assert(conflictlogrel); - pgstat_report_subscription_conflict(MySubscription->oid, type); + appendStringInfo(&err_detail, + _("Conflict details are logged to the conflict log table: %s.%s."), + get_namespace_name(RelationGetNamespace(conflictlogrel)), + RelationGetRelationName(conflictlogrel)); + } ereport(elevel, errcode_apply_conflict(type), @@ -302,6 +412,22 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, RelationGetRelationName(localrel), ConflictTypeNames[type]), errdetail_internal("%s", err_detail.data)); + + /* + * Insert into the conflict log table for LOG-level conflicts if + * requested. + */ + if (conflictlogrel) + { + insert_conflict_log_tuple(estate, + relinfo->ri_RelationDesc, + conflictlogrel, + type, + searchindexoid, + searchslot, + conflicttuples); + table_close(conflictlogrel, NoLock); + } } /* @@ -335,6 +461,137 @@ InitConflictIndexes(ResultRelInfo *relInfo) relInfo->ri_onConflictArbiterIndexes = uniqueIndexes; } +/* + * get_conflictlog_dest_and_table + * + * Decides where a conflict of the given elevel is to be reported, based on + * the subscription's conflict_log_destination. Sets *log_dest to the + * effective destination and, if that includes the conflict log table, returns + * the table opened for insertion. Otherwise returns NULL. + * + * Only LOG-level conflicts (i.e. resolved conflicts where the transaction + * continues) are recorded in the conflict log table. ERROR-level conflicts + * halt replication and abort the transaction, so they are always reported + * exclusively to the server log. + * + * With two_phase enabled, this may error out instead of waiting indefinitely + * for the lock on the conflict log table or its TOAST table, see + * lock_conflictlog_relation(). + */ +static Relation +get_conflictlog_dest_and_table(int elevel, ConflictLogDest *log_dest) +{ + Relation conflictlogrel; + Oid conflictlogrelid; + + if (elevel >= ERROR) + { + *log_dest = CONFLICT_LOG_DEST_LOG; + return NULL; + } + + /* + * Convert the text log destination to the internal enum. MySubscription + * already contains the data from pg_subscription. + */ + *log_dest = GetConflictLogDest(MySubscription->conflictlogdest); + + /* Quick exit if a conflict log table was not requested. */ + if (!CONFLICTS_LOGGED_TO_TABLE(*log_dest)) + return NULL; + + conflictlogrelid = MySubscription->conflictlogrelid; + + Assert(OidIsValid(conflictlogrelid)); + + lock_conflictlog_relation(conflictlogrelid); + + /* + * Use try_table_open() because the table may have been dropped + * concurrently by an ALTER SUBSCRIPTION that changed + * conflict_log_destination. In that case, fall back to the server log + * instead of failing. + */ + conflictlogrel = try_table_open(conflictlogrelid, NoLock); + if (conflictlogrel == NULL) + { + UnlockRelationOid(conflictlogrelid, RowExclusiveLock); + *log_dest = CONFLICT_LOG_DEST_LOG; + return NULL; + } + + /* + * Lock the TOAST table too. A command like REINDEX TABLE can be run + * directly on it without taking any lock on the conflict log table, so + * locking only the conflict log table is not enough. Such a command could + * wait for a prepared transaction of this subscription, while the apply + * worker waits behind it and can never finish that transaction. + */ + if (OidIsValid(conflictlogrel->rd_rel->reltoastrelid)) + lock_conflictlog_relation(conflictlogrel->rd_rel->reltoastrelid); + + return conflictlogrel; +} + +/* + * Lock the conflict log table or its TOAST table in RowExclusiveLock mode. + * + * With two_phase enabled, a prepared transaction of this subscription that + * recorded a conflict holds a lock on the relation until we apply its COMMIT + * PREPARED or ROLLBACK PREPARED. A command that needs a stronger lock, such + * as TRUNCATE, waits for that transaction, and if we waited indefinitely + * behind the command we could never apply the COMMIT PREPARED or ROLLBACK + * PREPARED that releases the lock. + * + * Therefore, with two_phase enabled, don't wait indefinitely. If the lock is + * not immediately available, error out if this subscription has any prepared + * transactions, as one of them may hold the lock. Otherwise, wait with + * lock_timeout set to DeadlockTimeout + 5s, so that ProcSleep() can run + * DeadLockCheck() and interrupt a blocking autovacuum truncation, and so that + * the wait ends if a parallel apply worker prepares a transaction meanwhile. + * If lock_timeout is already configured, just wait and rely on it. + * + * Without two_phase, no prepared transaction of this subscription can exist + * or appear while we run, as two_phase only takes effect after the workers + * restart, so wait for the lock as usual. + */ +static void +lock_conflictlog_relation(Oid relid) +{ + int save_nestlevel; + char buf[32]; + + if (MySubscription->twophasestate != LOGICALREP_TWOPHASE_STATE_ENABLED || + LockTimeout > 0) + { + LockRelationOid(relid, RowExclusiveLock); + return; + } + + if (ConditionalLockRelationOid(relid, RowExclusiveLock)) + return; + + if (LookupGXactBySubid(MySubscription->oid)) + ereport(ERROR, + errcode(ERRCODE_LOCK_NOT_AVAILABLE), + errmsg("could not obtain lock on conflict log table of subscription \"%s\"", + MySubscription->name), + errdetail("A prepared transaction of this subscription may hold a lock on its conflict log table, and other session is waiting on it."), + errhint("If a session is waiting for that lock, cancel it and let the apply worker retry.")); + + save_nestlevel = NewGUCNestLevel(); + + snprintf(buf, sizeof(buf), "%d", + Min(DeadlockTimeout, INT_MAX - 5000) + 5000); + (void) set_config_option("lock_timeout", buf, + PGC_USERSET, PGC_S_SESSION, + GUC_ACTION_SAVE, true, 0, false); + + LockRelationOid(relid, RowExclusiveLock); + + AtEOXact_GUC(true, save_nestlevel); +} + /* * Add SQLSTATE error code to the current conflict report. */ @@ -772,6 +1029,38 @@ get_tuple_desc(EState *estate, ResultRelInfo *relinfo, ConflictType type, } } +/* + * Helper function to extract the "raw" index key Datums and their null flags + * from a TupleTableSlot, given an already open index descriptor. + */ +static void +build_index_datums_from_slot(EState *estate, Relation localrel, + TupleTableSlot *slot, + Relation indexDesc, Datum *values, + bool *isnull) +{ + TupleTableSlot *tableslot = slot; + + /* + * FormIndexDatum needs an EState whose per-tuple econtext's scantuple + * points to the slot holding the tuple, to evaluate any index + * expressions. It also requires a heap tuple slot, so copy a virtual slot + * into one, created in the EState's tuple table so that it is freed along + * with it. + */ + if (TTS_IS_VIRTUAL(slot)) + { + tableslot = table_slot_create(localrel, &estate->es_tupleTable); + tableslot = ExecCopySlot(tableslot, slot); + } + + GetPerTupleExprContext(estate)->ecxt_scantuple = tableslot; + + /* Form the index datums */ + FormIndexDatum(BuildIndexInfo(indexDesc), tableslot, estate, values, + isnull); +} + /* * Helper functions to construct a string describing the contents of an index * entry. See BuildIndexValueDescription for details. @@ -787,41 +1076,423 @@ build_index_value_desc(EState *estate, Relation localrel, TupleTableSlot *slot, Relation indexDesc; Datum values[INDEX_MAX_KEYS]; bool isnull[INDEX_MAX_KEYS]; - TupleTableSlot *tableslot = slot; - if (!tableslot) + if (!slot) return NULL; Assert(CheckRelationOidLockedByMe(indexoid, RowExclusiveLock, true)); indexDesc = index_open(indexoid, NoLock); + build_index_datums_from_slot(estate, localrel, slot, indexDesc, values, + isnull); + + index_value = BuildIndexValueDescription(indexDesc, values, isnull); + + index_close(indexDesc, NoLock); + + return index_value; +} + +/* + * build_index_key_json + * + * Fetch replica identity key from the tuple table slot and convert into a + * JSON datum. + * + * The caller must ensure that the index with the OID 'indexid' is locked so + * that we can safely fetch and construct the replica identity key values. + * See comments atop ReportApplyConflict() for how this holds. + */ +static Datum +build_index_key_json(EState *estate, Relation localrel, + Oid indexid, TupleTableSlot *slot, + bool *omitted) +{ + Relation indexDesc; + TupleDesc reltupdesc = RelationGetDescr(localrel); + Datum values[INDEX_MAX_KEYS]; + bool isnull[INDEX_MAX_KEYS]; + StringInfoData result; + int indnkeyatts; + Datum datum; + + Assert(slot != NULL); + + Assert(CheckRelationOidLockedByMe(indexid, RowExclusiveLock, true)); + + indexDesc = index_open(indexid, NoLock); + + build_index_datums_from_slot(estate, localrel, slot, indexDesc, values, + isnull); + /* - * If the slot is a virtual slot, copy it into a heap tuple slot as - * FormIndexDatum only works with heap tuple slots. + * Build the JSON object here rather than passing the values to + * row_to_json(). For a type at or above FirstNormalObjectId, + * json_categorize_type() would use a user-defined cast to json if one + * exists. Nothing else in apply calls that cast, so its output is + * unrelated to the size of the value we received, and a small key could + * produce a json value over the 1GB limit and fail the apply worker. + * Rendering each value with its type's output function keeps us to the + * representation the publisher sent (when the column types and output + * settings match, see below), and matches what the server log does in + * BuildIndexValueDescription(). + * + * Only the key attributes are recorded. FormIndexDatum() above also fills + * in any non-key (INCLUDE) columns, but those are not part of the replica + * identity. */ - if (TTS_IS_VIRTUAL(slot)) + indnkeyatts = IndexRelationGetNumberOfKeyAttributes(indexDesc); + + initStringInfo(&result); + appendStringInfoChar(&result, '{'); + + for (int i = 0; i < indnkeyatts; i++) { - tableslot = table_slot_create(localrel, &estate->es_tupleTable); - tableslot = ExecCopySlot(tableslot, slot); + Form_pg_attribute att; + AttrNumber attnum; + + /* + * Take the column name from the table, not the index descriptor. The + * index keeps the name it had at creation, which goes stale after a + * column rename and can be wrong for repeated columns such as UNIQUE + * (a, a). Using the table's names keeps these consistent with the + * server log. + * + * Replica identity and primary key indexes have no expression + * columns, so every key maps to a table column. + */ + attnum = indexDesc->rd_index->indkey.values[i]; + Assert(attnum > 0); + + att = TupleDescAttr(reltupdesc, attnum - 1); + + if (i > 0) + appendStringInfoChar(&result, ','); + + escape_json(&result, NameStr(att->attname)); + appendStringInfoChar(&result, ':'); + + if (isnull[i]) + appendStringInfoString(&result, "null"); + else + { + Oid outfuncoid; + bool typisvarlena; + char *outputstr; + Size outputlen; + + /* + * See BuildIndexValueDescription() for why we used rd_opcintype + * instead index's tupleDesc. + */ + getTypeOutputInfo(indexDesc->rd_opcintype[i], &outfuncoid, &typisvarlena); + + /* + * XXX The cap below is checked only after the value is rendered. + * The publisher rendered the same value to send it, which bounds + * its length, unless the column types differ, subscription option + * binary = true, or the apply worker's bytea_output or + * quote_all_identifiers differs. Then the render can exceed 1GB + * and error out the apply worker. + */ + outputstr = OidOutputFunctionCall(outfuncoid, values[i]); + outputlen = strlen(outputstr); + + /* + * Don't record a value larger than CONFLICT_MAX_VALUE_SIZE; put a + * marker recording its length in its place. + * + * XXX The marker deliberately keeps no prefix of the value. For + * arrays, composites and similar types the leading bytes are + * mostly punctuation and say little about the value. + */ + if (outputlen > CONFLICT_MAX_VALUE_SIZE) + { + /* A marker is an object, so it is appended unescaped. */ + appendStringInfo(&result, + "{\"omitted\":true,\"length\":" UINT64_FORMAT "}", + (uint64) outputlen); + *omitted = true; + pfree(outputstr); + continue; + } + + escape_json(&result, outputstr); + pfree(outputstr); + } + } + + appendStringInfoChar(&result, '}'); + + index_close(indexDesc, NoLock); + + datum = PointerGetDatum(cstring_to_text_with_len(result.data, result.len)); + pfree(result.data); + + return datum; +} + +/* + * build_local_conflicts_tupledesc + * + * Build and bless a tuple descriptor for the conflict log table based on the + * predefined LocalConflictSchema. + */ +static TupleDesc +build_local_conflicts_tupledesc(void) +{ + static TupleDesc cached_tupdesc = NULL; + + if (cached_tupdesc == NULL) + { + MemoryContext oldcxt; + + oldcxt = MemoryContextSwitchTo(CacheMemoryContext); + + cached_tupdesc = CreateTemplateTupleDesc(NUM_LOCAL_CONFLICT_ATTRS); + + for (int i = 0; i < NUM_LOCAL_CONFLICT_ATTRS; i++) + TupleDescInitEntry(cached_tupdesc, + (AttrNumber) (i + 1), + LocalConflictSchema[i].attname, + LocalConflictSchema[i].atttypid, + -1, 0); + + TupleDescFinalize(cached_tupdesc); + + /* + * Bless once so it can be used as a RECORD type (e.g. for row_to_json + * or other record-based operations). + */ + BlessTupleDesc(cached_tupdesc); + + MemoryContextSwitchTo(oldcxt); + } + + return cached_tupdesc; +} + +/* + * Builds the local conflicts JSON array column from the list of + * ConflictTupleInfo objects. + * + * Example output structure: + * [ { "xid": "1001", "commit_ts": "...", "origin": "..." }, ... ] + */ +static Datum +build_local_conflicts_json_array(List *conflicttuples) +{ + Datum *json_datum_array; + Datum json_array_datum; + int num_conflicts; + int i = 0; + int16 typlen; + bool typbyval; + char typalign; + TupleDesc tupdesc; + + Assert(conflicttuples != NIL); + + /* Build local conflicts tuple descriptor. */ + tupdesc = build_local_conflicts_tupledesc(); + + num_conflicts = list_length(conflicttuples); + json_datum_array = palloc_array(Datum, num_conflicts); + + /* Process local conflict tuple list and prepare an array of JSON. */ + foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples) + { + Datum values[NUM_LOCAL_CONFLICT_ATTRS] = {0}; + bool nulls[NUM_LOCAL_CONFLICT_ATTRS] = {0}; + char *origin_name = NULL; + HeapTuple tuple; + Datum datum; + Datum json_datum; + int attno; + + attno = 0; + if (TransactionIdIsValid(conflicttuple->xmin)) + values[attno++] = TransactionIdGetDatum(conflicttuple->xmin); + else + nulls[attno++] = true; + + if (conflicttuple->ts) + values[attno++] = TimestampTzGetDatum(conflicttuple->ts); + else + nulls[attno++] = true; + + if (conflicttuple->origin != InvalidReplOriginId) + replorigin_by_oid(conflicttuple->origin, true, &origin_name); + + /* + * Set NULL if origin name for the tuple is InvalidReplOriginId or not + * found. + */ + if (origin_name != NULL) + values[attno++] = CStringGetTextDatum(origin_name); + else + nulls[attno++] = true; + + Assert(attno == NUM_LOCAL_CONFLICT_ATTRS); + + tuple = heap_form_tuple(tupdesc, values, nulls); + + datum = heap_copy_tuple_as_datum(tuple, tupdesc); + + /* + * Build the higher level JSON datum in format described in function + * header. + * + * Unlike build_index_key_json(), using row_to_json() is safe here + * because the object's fields are all built-in types (xid, + * timestamptz, text) below FirstNormalObjectId, so + * json_categorize_type() never invokes a user CREATE CAST (t AS + * json), and the values are small and bounded. + */ + json_datum = DirectFunctionCall1(row_to_json, datum); + + /* Done with the temporary tuple. */ + heap_freetuple(tuple); + + /* Add to the array element. */ + json_datum_array[i++] = json_datum; } + Assert(i == num_conflicts); + + /* Construct the JSON array Datum. */ + get_typlenbyvalalign(JSONOID, &typlen, &typbyval, &typalign); + json_array_datum = PointerGetDatum(construct_array(json_datum_array, + num_conflicts, + JSONOID, + typlen, + typbyval, + typalign)); + pfree(json_datum_array); + + return json_array_datum; +} + +/* + * insert_conflict_log_tuple + * + * Prepares and inserts a tuple detailing a conflict encountered during + * logical replication into the conflict log table. + */ +static void +insert_conflict_log_tuple(EState *estate, Relation rel, + Relation conflictlogrel, + ConflictType conflict_type, + Oid searchindexoid, + TupleTableSlot *searchslot, + List *conflicttuples) +{ + Datum values[NUM_CONFLICT_ATTRS] = {0}; + bool nulls[NUM_CONFLICT_ATTRS] = {0}; + int attno; + char *apply_origin = NULL; + TransactionId remote_xid; + XLogRecPtr remote_final_lsn; + TimestampTz remote_commit_ts; + bool omitted = false; + Oid replica_index; + HeapTuple tuple; + + Assert(conflictlogrel != NULL); + + /* Fetch the information of the remote transaction being applied. */ + GetRemoteTransactionInfoForConflict(&remote_xid, &remote_final_lsn, + &remote_commit_ts); + + /* Populate the values and nulls arrays. */ + attno = 0; + values[attno++] = ObjectIdGetDatum(RelationGetRelid(rel)); + + values[attno++] = + CStringGetTextDatum(get_namespace_name(RelationGetNamespace(rel))); + + values[attno++] = CStringGetTextDatum(RelationGetRelationName(rel)); + + values[attno++] = CStringGetTextDatum(ConflictTypeNames[conflict_type]); + /* - * Initialize ecxt_scantuple for potential use in FormIndexDatum when - * index expressions are present. + * The remote xid is always valid here: a conflict is only detected while + * applying a row change, which is preceded by a BEGIN, BEGIN PREPARE or + * STREAM START that recorded a valid top-level xid. */ - GetPerTupleExprContext(estate)->ecxt_scantuple = tableslot; + Assert(TransactionIdIsValid(remote_xid)); + values[attno++] = TransactionIdGetDatum(remote_xid); + + if (XLogRecPtrIsValid(remote_final_lsn)) + values[attno++] = LSNGetDatum(remote_final_lsn); + else + nulls[attno++] = true; + + if (remote_commit_ts > 0) + values[attno++] = TimestampTzGetDatum(remote_commit_ts); + else + nulls[attno++] = true; + + if (replorigin_xact_state.origin != InvalidReplOriginId) + replorigin_by_oid(replorigin_xact_state.origin, true, &apply_origin); + + if (apply_origin != NULL) + values[attno++] = CStringGetTextDatum(apply_origin); + else + nulls[attno++] = true; /* - * The values/nulls arrays passed to BuildIndexValueDescription should be - * the results of FormIndexDatum, which are the "raw" input to the index - * AM. + * All conflicts logged to the table are LOG-level update or delete + * conflicts, which always have a searchslot. Insert conflicts, which + * don't, are ERROR-level and never reach here. */ - FormIndexDatum(BuildIndexInfo(indexDesc), tableslot, estate, values, isnull); + Assert(!TupIsNull(searchslot)); - index_value = BuildIndexValueDescription(indexDesc, values, isnull); + replica_index = searchindexoid; - index_close(indexDesc, NoLock); + /* + * If the search used a replica identity index, build the index JSON datum + * from key value. Otherwise, in REPLICA IDENTITY FULL cases, set + * replica_identity_full to true and leave replica_identity NULL to avoid + * serializing full tuples that could exceed memory allocation limits. + */ + if (OidIsValid(replica_index)) + { + values[attno++] = BoolGetDatum(false); + values[attno++] = build_index_key_json(estate, rel, + replica_index, + searchslot, + &omitted); + } + else + { + values[attno++] = BoolGetDatum(true); + nulls[attno++] = true; + } - return index_value; + /* + * In update_missing and delete_missing conflicts, there are no local + * conflicting rows, so set local_conflicts to NULL. + */ + if (conflicttuples != NIL) + values[attno++] = build_local_conflicts_json_array(conflicttuples); + else + nulls[attno++] = true; + + values[attno] = BoolGetDatum(omitted); + + Assert(attno + 1 == NUM_CONFLICT_ATTRS); + + tuple = heap_form_tuple(RelationGetDescr(conflictlogrel), values, nulls); + + /* + * The conflict log table itself is never logically decoded, but its TOAST + * table would be. heap_insert() passes HEAP_INSERT_NO_LOGICAL on to the + * TOAST inserts, so toasted values are not decoded either. + */ + heap_insert(conflictlogrel, tuple, + GetCurrentCommandId(true), HEAP_INSERT_NO_LOGICAL, NULL); + + heap_freetuple(tuple); } diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 44d735cdc55..7135abe842e 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -329,10 +329,13 @@ typedef struct ApplyExecutionData * Context describing the remote transaction whose changes are currently * being applied, and the change within it. * - * The remote transaction information (remote_xid and finish_lsn) is set when - * the transaction's changes begin to be applied. finish_lsn is invalid when - * the final LSN of the remote transaction is not yet known (e.g. while - * streaming an in-progress transaction). + * The remote transaction information (remote_xid, finish_lsn, and commit_ts) + * is set when the transaction's changes begin to be applied. finish_lsn is + * invalid when the final LSN of the remote transaction is not yet known (e.g. + * while streaming an in-progress transaction). Similarly, commit_ts is 0 when + * the remote commit timestamp is not yet known; for a streamed transaction it + * is filled in only if and when the leader apply worker processes the commit + * record, so it remains 0 for conflicts handled by a parallel apply worker. * * The remaining fields describe the individual change being applied and are * used only for error context reporting. @@ -346,6 +349,7 @@ typedef struct ApplyRemoteCtx int remote_attnum; /* -1 if invalid */ TransactionId remote_xid; XLogRecPtr finish_lsn; + TimestampTz commit_ts; /* remote commit ts; 0 if unknown */ char *origin_name; } ApplyRemoteCtx; @@ -480,6 +484,7 @@ static ApplyRemoteCtx remote_ctx = .remote_attnum = -1, .remote_xid = InvalidTransactionId, .finish_lsn = InvalidXLogRecPtr, + .commit_ts = 0, .origin_name = NULL, }; @@ -636,7 +641,9 @@ static void stop_skipping_changes(void); static void clear_subscription_skip_lsn(XLogRecPtr finish_lsn); /* Functions to maintain the context of the remote transaction being applied */ -static inline void set_remote_transaction_info(TransactionId xid, XLogRecPtr lsn); +static inline void set_remote_transaction_info(TransactionId xid, + XLogRecPtr lsn, + TimestampTz commit_ts); static inline void reset_apply_remote_context(void); static TransApplyAction get_transaction_apply_action(TransactionId xid, @@ -1252,7 +1259,8 @@ apply_handle_begin(StringInfo s) Assert(!TransactionIdIsValid(stream_xid)); logicalrep_read_begin(s, &begin_data); - set_remote_transaction_info(begin_data.xid, begin_data.final_lsn); + set_remote_transaction_info(begin_data.xid, begin_data.final_lsn, + begin_data.committime); maybe_start_skipping_changes(begin_data.final_lsn); @@ -1310,7 +1318,12 @@ apply_handle_begin_prepare(StringInfo s) Assert(!TransactionIdIsValid(stream_xid)); logicalrep_read_begin_prepare(s, &begin_data); - set_remote_transaction_info(begin_data.xid, begin_data.prepare_lsn); + + /* + * The commit timestamp is known only when the commit record arrives, so + * it is set to 0 for now. + */ + set_remote_transaction_info(begin_data.xid, begin_data.prepare_lsn, 0); maybe_start_skipping_changes(begin_data.prepare_lsn); @@ -1440,7 +1453,8 @@ apply_handle_commit_prepared(StringInfo s) char gid[GIDSIZE]; logicalrep_read_commit_prepared(s, &prepare_data); - set_remote_transaction_info(prepare_data.xid, prepare_data.commit_lsn); + set_remote_transaction_info(prepare_data.xid, prepare_data.commit_lsn, + prepare_data.commit_time); /* Compute GID for two_phase transactions. */ TwoPhaseTransactionGid(MySubscription->oid, prepare_data.xid, @@ -1492,7 +1506,8 @@ apply_handle_rollback_prepared(StringInfo s) char gid[GIDSIZE]; logicalrep_read_rollback_prepared(s, &rollback_data); - set_remote_transaction_info(rollback_data.xid, rollback_data.rollback_end_lsn); + set_remote_transaction_info(rollback_data.xid, + rollback_data.rollback_end_lsn, 0); /* Compute GID for two_phase transactions. */ TwoPhaseTransactionGid(MySubscription->oid, rollback_data.xid, @@ -1568,7 +1583,7 @@ apply_handle_stream_prepare(StringInfo s) errmsg_internal("tablesync worker received a STREAM PREPARE message"))); logicalrep_read_stream_prepare(s, &prepare_data); - set_remote_transaction_info(prepare_data.xid, prepare_data.prepare_lsn); + set_remote_transaction_info(prepare_data.xid, prepare_data.prepare_lsn, 0); apply_action = get_transaction_apply_action(prepare_data.xid, &winfo); @@ -1783,10 +1798,10 @@ apply_handle_stream_start(StringInfo s) errmsg_internal("invalid transaction ID in streamed replication transaction"))); /* - * The final LSN of the streamed transaction is known only when its commit - * record arrives. + * The final LSN and commit timestamp of the streamed transaction are + * known only when its commit record arrives. */ - set_remote_transaction_info(stream_xid, InvalidXLogRecPtr); + set_remote_transaction_info(stream_xid, InvalidXLogRecPtr, 0); /* Try to allocate a worker for the streaming transaction. */ if (first_segment) @@ -2135,7 +2150,7 @@ apply_handle_stream_abort(StringInfo s) * when a top-level transaction aborts, and a subxid only when a * subtransaction rolls back. See set_remote_transaction_info(). */ - set_remote_transaction_info(subxid, abort_data.abort_lsn); + set_remote_transaction_info(subxid, abort_data.abort_lsn, 0); apply_action = get_transaction_apply_action(xid, &winfo); @@ -2443,7 +2458,8 @@ apply_handle_stream_commit(StringInfo s) errmsg_internal("STREAM COMMIT message without STREAM STOP"))); xid = logicalrep_read_stream_commit(s, &commit_data); - set_remote_transaction_info(xid, commit_data.commit_lsn); + set_remote_transaction_info(xid, commit_data.commit_lsn, + commit_data.committime); apply_action = get_transaction_apply_action(xid, &winfo); @@ -2995,7 +3011,7 @@ apply_handle_update_internal(ApplyExecutionData *edata, conflicttuple.slot = localslot; ReportApplyConflict(estate, relinfo, LOG, CT_UPDATE_ORIGIN_DIFFERS, - remoteslot, newslot, + relmapentry, remoteslot, newslot, list_make1(&conflicttuple)); } @@ -3036,10 +3052,11 @@ apply_handle_update_internal(ApplyExecutionData *edata, /* * The tuple to be updated could not be found or was deleted. Do - * nothing except for emitting a log message. + * nothing except for reporting the conflict. */ - ReportApplyConflict(estate, relinfo, LOG, type, remoteslot, newslot, - list_make1(&conflicttuple)); + ReportApplyConflict(estate, relinfo, LOG, type, relmapentry, + remoteslot, newslot, + type == CT_UPDATE_DELETED ? list_make1(&conflicttuple) : NIL); } /* Cleanup. */ @@ -3182,7 +3199,7 @@ apply_handle_delete_internal(ApplyExecutionData *edata, { conflicttuple.slot = localslot; ReportApplyConflict(estate, relinfo, LOG, CT_DELETE_ORIGIN_DIFFERS, - remoteslot, NULL, + relmapentry, remoteslot, NULL, list_make1(&conflicttuple)); } @@ -3196,10 +3213,10 @@ apply_handle_delete_internal(ApplyExecutionData *edata, { /* * The tuple to be deleted could not be found. Do nothing except for - * emitting a log message. + * reporting the conflict. */ ReportApplyConflict(estate, relinfo, LOG, CT_DELETE_MISSING, - remoteslot, NULL, list_make1(&conflicttuple)); + relmapentry, remoteslot, NULL, NIL); } /* Cleanup. */ @@ -3556,11 +3573,12 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, /* * The tuple to be updated could not be found or was - * deleted. Do nothing except for emitting a log message. + * deleted. Do nothing except for reporting the conflict. */ ReportApplyConflict(estate, partrelinfo, LOG, - type, remoteslot_part, newslot, - list_make1(&conflicttuple)); + type, part_entry, + remoteslot_part, newslot, + type == CT_UPDATE_DELETED ? list_make1(&conflicttuple) : NIL); return; } @@ -3583,7 +3601,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, conflicttuple.slot = localslot; ReportApplyConflict(estate, partrelinfo, LOG, CT_UPDATE_ORIGIN_DIFFERS, - remoteslot_part, newslot, + part_entry, remoteslot_part, newslot, list_make1(&conflicttuple)); } @@ -6420,13 +6438,45 @@ apply_error_callback(void *arg) * subtransaction rolls back, and the top-level xid otherwise. Nothing else * observes a subxid recorded this way, because no change is applied between a * STREAM ABORT and the STREAM START or STREAM COMMIT/PREPARE that follows it, - * and each of those calls this again with the transaction's own values. + * and each of those calls this again with the transaction's own values. The + * assertion in GetRemoteTransactionInfoForConflict() enforces that for + * conflict logging. + * + * Callers pass the remote commit timestamp when it is known (from the BEGIN + * or STREAM COMMIT message) and 0 otherwise. */ static inline void -set_remote_transaction_info(TransactionId xid, XLogRecPtr lsn) +set_remote_transaction_info(TransactionId xid, XLogRecPtr lsn, + TimestampTz commit_ts) { remote_ctx.remote_xid = xid; remote_ctx.finish_lsn = lsn; + remote_ctx.commit_ts = commit_ts; +} + +/* + * Return information identifying the remote transaction currently being + * applied, for use by conflict logging (see insert_conflict_log_tuple() in + * conflict.c). See set_remote_transaction_info(). + */ +void +GetRemoteTransactionInfoForConflict(TransactionId *remote_xid, + XLogRecPtr *finish_lsn, + TimestampTz *commit_ts) +{ + /* + * Conflicts are only detected while applying a row change, which + * guarantees the recorded values describe the top-level transaction. In + * particular they are never the subxid and abort LSN recorded by a + * subtransaction abort. See set_remote_transaction_info(). + */ + Assert(remote_ctx.command == LOGICAL_REP_MSG_INSERT || + remote_ctx.command == LOGICAL_REP_MSG_UPDATE || + remote_ctx.command == LOGICAL_REP_MSG_DELETE); + + *remote_xid = remote_ctx.remote_xid; + *finish_lsn = remote_ctx.finish_lsn; + *commit_ts = remote_ctx.commit_ts; } /* Reset all information of the remote transaction context */ @@ -6436,7 +6486,7 @@ reset_apply_remote_context(void) remote_ctx.command = 0; remote_ctx.rel = NULL; remote_ctx.remote_attnum = -1; - set_remote_transaction_info(InvalidTransactionId, InvalidXLogRecPtr); + set_remote_transaction_info(InvalidTransactionId, InvalidXLogRecPtr, 0); } /* diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h index 326570ed639..5643052a133 100644 --- a/src/include/replication/conflict.h +++ b/src/include/replication/conflict.h @@ -15,6 +15,7 @@ /* Avoid including execnodes.h here */ typedef struct EState EState; +typedef struct LogicalRepRelMapEntry LogicalRepRelMapEntry; typedef struct ResultRelInfo ResultRelInfo; typedef struct TupleTableSlot TupleTableSlot; @@ -111,6 +112,7 @@ extern bool GetTupleTransactionInfo(TupleTableSlot *localslot, TimestampTz *localts); extern void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, ConflictType type, + LogicalRepRelMapEntry *relmapentry, TupleTableSlot *searchslot, TupleTableSlot *remoteslot, List *conflicttuples); diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 88cb7c1e252..e2e9ca1f708 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -256,6 +256,11 @@ extern PGDLLIMPORT bool InitializingApplyWorker; extern PGDLLIMPORT List *table_states_not_ready; +/* Remote transaction information of the transaction being applied */ +extern void GetRemoteTransactionInfoForConflict(TransactionId *remote_xid, + XLogRecPtr *finish_lsn, + TimestampTz *commit_ts); + extern void logicalrep_worker_attach(int slot); extern LogicalRepWorker *logicalrep_worker_find(LogicalRepWorkerType wtype, Oid subid, Oid relid, diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h index 7b8c2b1e362..25ee2842996 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -741,16 +741,19 @@ RelationCloseSmgr(Relation relation) * * We don't log information for unlogged tables (since they don't WAL log * anyway), for foreign tables (since they don't WAL log, either), - * and for system tables (their content is hard to make sense of, and - * it would complicate decoding slightly for little gain). Note that we *do* - * log information for user defined catalog tables since they presumably are - * interesting to the user... + * for system tables (their content is hard to make sense of, and + * it would complicate decoding slightly for little gain), and for conflict + * log tables in the pg_conflict namespace (which are system-managed tables + * used internally on subscribers). Note that we *do* log information for + * user defined catalog tables since they presumably are interesting to the + * user... */ #define RelationIsLogicallyLogged(relation) \ (XLogLogicalInfoActive() && \ RelationNeedsWAL(relation) && \ (relation)->rd_rel->relkind != RELKIND_FOREIGN_TABLE && \ - !IsCatalogRelation(relation)) + !IsCatalogRelation(relation) && \ + !IsConflictLogTableNamespace(RelationGetNamespace(relation))) /* routines in utils/cache/relcache.c */ extern void RelationIncrementReferenceCount(Relation rel); diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index 681dc66dca6..fed9c4a89b0 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -701,11 +701,11 @@ WHERE s.subname = 'regress_conflict_test1' AND a.attnum > 0 5 | remote_xid 6 | remote_commit_lsn 7 | remote_commit_ts - 8 | remote_origin + 8 | apply_origin 9 | replica_identity_full 10 | replica_identity - 11 | remote_tuple - 12 | local_conflicts + 11 | local_conflicts + 12 | has_omitted_values (12 rows) -- Changing the subscription owner should also update the owner @@ -775,6 +775,13 @@ FROM pg_subscription WHERE subname = 'regress_conflict_test2'; table | t (1 row) +-- fail - transition from 'table' to 'log' with two_phase requires a disabled subscription +BEGIN; +ALTER SUBSCRIPTION regress_conflict_test2 SET (two_phase = true); +ALTER SUBSCRIPTION regress_conflict_test2 ENABLE; +ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'log'); +ERROR: cannot set option "conflict_log_destination" for enabled subscription +ROLLBACK; -- transition from 'table' to 'log' -- should drop the table and clear subconflictlogrelid ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'log'); diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index cfeebaf9302..89c9a32af06 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -586,6 +586,13 @@ ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'table SELECT subconflictlogdest, subconflictlogrelid = :old_relid AS relid_unchanged FROM pg_subscription WHERE subname = 'regress_conflict_test2'; +-- fail - transition from 'table' to 'log' with two_phase requires a disabled subscription +BEGIN; +ALTER SUBSCRIPTION regress_conflict_test2 SET (two_phase = true); +ALTER SUBSCRIPTION regress_conflict_test2 ENABLE; +ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'log'); +ROLLBACK; + -- transition from 'table' to 'log' -- should drop the table and clear subconflictlogrelid ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'log'); diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl index 48ce95d9d13..56415a46659 100644 --- a/src/test/subscription/t/035_conflicts.pl +++ b/src/test/subscription/t/035_conflicts.pl @@ -11,14 +11,19 @@ use Test::More; # Setup ############################### -# Create a publisher node +# Create a publisher node. Prepared transactions are needed by the two_phase +# tests at the end. my $node_publisher = PostgreSQL::Test::Cluster->new('publisher'); $node_publisher->init(allows_streaming => 'logical'); +$node_publisher->append_conf('postgresql.conf', + "max_prepared_transactions = 10"); $node_publisher->start; # Create a subscriber node my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber'); $node_subscriber->init(allows_streaming => 'logical'); +$node_subscriber->append_conf('postgresql.conf', + "max_prepared_transactions = 10"); $node_subscriber->start; # Create a table on publisher @@ -84,6 +89,19 @@ $node_subscriber->wait_for_log( .*Key already exists in unique index \"conf_tab_c_key\", modified in transaction .*: key \(c\)=\(4\), local row \(4, 4, 4\)./, $log_offset); +# Verify the contents of the Conflict Log Table (CLT) +# ERROR-level conflicts halt replication and are reported only to server logs, +# so the CLT should remain empty. +my $subid = $node_subscriber->safe_psql('postgres', + "SELECT oid FROM pg_subscription WHERE subname = 'sub_tab';"); +my $clt = "pg_conflict.pg_conflict_log_$subid"; + +my $conflict_count = + $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM $clt;"); +is($conflict_count, '0', + 'Verified multiple_unique_conflicts is not logged into conflict log table' +); + pass('multiple_unique_conflicts detected during insert'); # Truncate table to get rid of the error @@ -114,6 +132,13 @@ $node_subscriber->wait_for_log( .*Key already exists in unique index \"conf_tab_c_key\", modified in transaction .*: key \(c\)=\(8\), local row \(8, 8, 8\)./, $log_offset); +# Verify that the CLT still does not contain ERROR-level conflicts +$conflict_count = + $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM $clt;"); +is($conflict_count, '0', + 'Verified multiple_unique_conflicts during UPDATE is not logged into conflict log table' +); + pass('multiple_unique_conflicts detected during update'); # Truncate table to get rid of the error @@ -183,7 +208,8 @@ $node_B->safe_psql( CREATE SUBSCRIPTION $subname_BA CONNECTION '$node_A_connstr application_name=$subname_BA' PUBLICATION tap_pub_A - WITH (origin = none, retain_dead_tuples = true)"); + WITH (origin = none, retain_dead_tuples = true, conflict_log_destination = 'all')" +); # node_B (pub) -> node_A (sub) my $node_B_connstr = $node_B->connstr . ' dbname=postgres'; @@ -193,7 +219,7 @@ $node_A->safe_psql( CREATE SUBSCRIPTION $subname_AB CONNECTION '$node_B_connstr application_name=$subname_AB' PUBLICATION tap_pub_B - WITH (origin = none, copy_data = off)"); + WITH (origin = none, copy_data = off, conflict_log_destination = 'all')"); # Wait for initial table sync to finish $node_A->wait_for_subscription_sync($node_B, $subname_AB); @@ -316,6 +342,20 @@ like( .*DETAIL:.* Deleting the row that was modified locally in transaction [0-9]+ at .*: local row \(1, 3\), replica identity \(a\)=\(1\)./, 'delete target row was modified in tab'); +my $subid_BA = $node_B->safe_psql('postgres', + "SELECT oid FROM pg_subscription WHERE subname = '$subname_BA';"); +my $clt_BA = "pg_conflict.pg_conflict_log_$subid_BA"; +my $clt_check_ba = $node_B->poll_query_until('postgres', + "SELECT count(*) > 0 FROM $clt_BA WHERE conflict_type = 'delete_origin_differs';" +); +is($clt_check_ba, 1, 'delete_origin_differs logged into CLT on Node B'); + +my $clt_row_ba = $node_B->safe_psql('postgres', + "SELECT replica_identity_full, replica_identity::text, (local_conflicts[1]->>'xid') IS NOT NULL FROM $clt_BA WHERE conflict_type = 'delete_origin_differs';" +); +is($clt_row_ba, 'f|{"a":"1"}|t', + 'delete_origin_differs records RI key columns and local conflict xid'); + $log_location = -s $node_A->logfile; $node_A->safe_psql('postgres', "ALTER SUBSCRIPTION $subname_AB ENABLE;"); @@ -329,6 +369,20 @@ like( .*The row to be updated was deleted locally in transaction [0-9]+ at .*/, 'update target row was deleted in tab'); +my $subid_AB = $node_A->safe_psql('postgres', + "SELECT oid FROM pg_subscription WHERE subname = '$subname_AB';"); +my $clt_AB = "pg_conflict.pg_conflict_log_$subid_AB"; +my $clt_check_ab = $node_A->poll_query_until('postgres', + "SELECT count(*) > 0 FROM $clt_AB WHERE conflict_type = 'update_deleted';" +); +is($clt_check_ab, 1, 'update_deleted logged into CLT on Node A'); + +my $clt_row_ab = $node_A->safe_psql('postgres', + "SELECT replica_identity_full, replica_identity::text, (local_conflicts[1]->>'xid') IS NOT NULL FROM $clt_AB WHERE conflict_type = 'update_deleted';" +); +is($clt_row_ab, 'f|{"a":"1"}|t', + 'update_deleted records RI key columns and local conflict xid'); + # Remember the next transaction ID to be assigned my $next_xid = $node_A->safe_psql('postgres', "SELECT txid_current() + 1;"); @@ -376,6 +430,139 @@ like( .*The row to be updated was deleted locally in transaction [0-9]+ at .*/, 'update target row was deleted in tab'); +my $clt_row_ab_full = $node_A->safe_psql('postgres', + "SELECT replica_identity_full, replica_identity IS NULL, (local_conflicts[1]->>'xid') IS NOT NULL FROM $clt_AB WHERE replica_identity_full = true;" +); +is($clt_row_ab_full, 't|t|t', + 'update_deleted with REPLICA IDENTITY FULL sets replica_identity_full=true and replica_identity=NULL' +); + +############################################################################### +# Check that a user-defined CAST (t AS json) is not used when recording the +# replica identity. +# +# Each key value is formatted with its data type's output function, as in the +# server log; a user-defined cast to json is never consulted. Calling it would +# execute user code that the apply worker otherwise never calls, and its output +# could be arbitrarily large, unrelated to the size of the original value. Even +# a tiny key could thus produce a json value exceeding the 1GB column limit and +# cause the apply worker to fail. +############################################################################### + +my $cast_ddl = qq{ + CREATE TYPE ri_color AS ENUM ('red', 'green'); + CREATE FUNCTION ri_color_to_json(ri_color) RETURNS json + AS \$\$ SELECT '"cast-was-used"'::json \$\$ LANGUAGE sql IMMUTABLE; + CREATE CAST (ri_color AS json) WITH FUNCTION ri_color_to_json(ri_color); + CREATE TABLE tab_cast (a ri_color PRIMARY KEY, b int); +}; + +$node_A->safe_psql('postgres', $cast_ddl); +$node_B->safe_psql('postgres', $cast_ddl); + +$node_A->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_A ADD TABLE tab_cast"); +$node_B->safe_psql('postgres', + "ALTER SUBSCRIPTION $subname_BA REFRESH PUBLICATION"); +$node_B->wait_for_subscription_sync($node_A, $subname_BA); + +# Replicate a row, remove it locally, then delete it on node_A so that the +# delete cannot find its target and the replica identity gets recorded. +$node_A->safe_psql('postgres', "INSERT INTO tab_cast VALUES ('red', 1)"); +$node_A->wait_for_catchup($subname_BA); +$node_B->safe_psql('postgres', "DELETE FROM tab_cast"); +$node_A->safe_psql('postgres', "DELETE FROM tab_cast WHERE a = 'red'"); +$node_A->wait_for_catchup($subname_BA); + +my $clt_check_cast = $node_B->poll_query_until('postgres', + "SELECT count(*) > 0 FROM $clt_BA WHERE relname = 'tab_cast';"); +is($clt_check_cast, 1, + 'delete_missing on tab_cast logged into CLT on Node B'); + +my $clt_row_cast = $node_B->safe_psql('postgres', + "SELECT replica_identity::text FROM $clt_BA WHERE relname = 'tab_cast';"); +is($clt_row_cast, '{"a":"red"}', + 'replica identity is rendered by the type output function, not by a cast to json' +); + +# Restore tap_pub_A to publishing only 'tab'. Later tests subscribe to it from +# another database, which does not have tab_cast. +$node_A->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_A DROP TABLE tab_cast"); +$node_B->safe_psql('postgres', + "ALTER SUBSCRIPTION $subname_BA REFRESH PUBLICATION"); + +my $cast_cleanup = q{ + DROP TABLE tab_cast; + DROP TYPE ri_color CASCADE; +}; + +$node_A->safe_psql('postgres', $cast_cleanup); +$node_B->safe_psql('postgres', $cast_cleanup); + +############################################################################### +# Check that a replica identity value too large to record is replaced by a +# marker rather than stored, and that has_omitted_values flags the row. +# +# A value is omitted rather than truncated, because a partial value in a +# queryable table would silently give wrong answers to equality and join +# conditions. +############################################################################### + +$node_A->safe_psql('postgres', + "CREATE TABLE tab_big (a text PRIMARY KEY, b int)"); +$node_B->safe_psql('postgres', + "CREATE TABLE tab_big (a text PRIMARY KEY, b int)"); + +$node_A->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_A ADD TABLE tab_big"); +$node_B->safe_psql('postgres', + "ALTER SUBSCRIPTION $subname_BA REFRESH PUBLICATION"); +$node_B->wait_for_subscription_sync($node_A, $subname_BA); + +# One key at the cap (64kB), recorded verbatim, and one over it, omitted. +$node_A->safe_psql( + 'postgres', qq{ + INSERT INTO tab_big VALUES (repeat('s', 65536), 1); + INSERT INTO tab_big VALUES (repeat('L', 65537), 2); +}); +$node_A->wait_for_catchup($subname_BA); +$node_B->safe_psql('postgres', "DELETE FROM tab_big"); +$node_A->safe_psql('postgres', "DELETE FROM tab_big"); +$node_A->wait_for_catchup($subname_BA); + +my $clt_check_big = $node_B->poll_query_until('postgres', + "SELECT count(*) = 2 FROM $clt_BA WHERE relname = 'tab_big';"); +is($clt_check_big, 1, + 'both delete_missing conflicts on tab_big logged into CLT'); + +my $clt_row_small = $node_B->safe_psql( + 'postgres', + "SELECT has_omitted_values, length(replica_identity->>'a') + FROM $clt_BA + WHERE relname = 'tab_big' AND NOT has_omitted_values;"); +is($clt_row_small, 'f|65536', + 'a replica identity value at the cap is recorded verbatim'); + +my $clt_row_big = $node_B->safe_psql( + 'postgres', + "SELECT has_omitted_values, + replica_identity->'a'->>'omitted', + replica_identity->'a'->>'length' + FROM $clt_BA + WHERE relname = 'tab_big' AND has_omitted_values;"); +is($clt_row_big, 't|true|65537', + 'an oversized replica identity value is replaced by a marker recording its length' +); + +# Restore tap_pub_A to publishing only 'tab', as the later tests expect. +$node_A->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_A DROP TABLE tab_big"); +$node_B->safe_psql('postgres', + "ALTER SUBSCRIPTION $subname_BA REFRESH PUBLICATION"); +$node_A->safe_psql('postgres', "DROP TABLE tab_big"); +$node_B->safe_psql('postgres', "DROP TABLE tab_big"); + ############################################################################### # Ensure that a deferrable primary key is not used to match deleted tuples in # a sequential table scan. Such a key cannot serve as a replica identity, so @@ -450,6 +637,10 @@ $node_B->safe_psql('postgres', "DROP TABLE tab_defer"); # same bounding range. Without a recheck, the deleted row would be accepted # as the update target and the conflict would be misreported as update_deleted # instead of update_missing. +# +# The conflict is also checked in the conflict log table, because the key is a +# multirange stored by a GiST opclass whose opckeytype (anyrange) differs from +# the type of the value. ############################################################################### # Publisher table holding the row that will be updated. @@ -515,6 +706,14 @@ like( ok( $logfile !~ /conflict detected on relation "public.tab_mr": conflict=update_deleted/, 'deleted row sharing the bounding range is not reported as update_deleted'); +# The replica identity key must be recorded in a form that matches the value. +is( $node_A->safe_psql( + 'postgres', + "SELECT replica_identity::text FROM $clt_AB + WHERE relname = 'tab_mr' AND conflict_type = 'update_missing';"), + '{"id":"{[1,5)}","valid_at":"[2000-01-01,2010-01-01)"}', + 'replica identity of a multirange key recorded in the conflict log table'); + # Clean up $holder->query_safe('COMMIT'); ok($holder->quit, 'release the pinned transaction'); @@ -816,7 +1015,8 @@ $node_B->safe_psql('dbb', "CREATE TABLE tab (a int PRIMARY KEY, b int)"); # Hold a transaction with an assigned transaction ID open in dbb, pinning its # oldest active transaction ID. my $dbb_session = $node_B->background_psql('dbb'); -$dbb_session->query_safe(q{ +$dbb_session->query_safe( + q{ BEGIN; SELECT txid_current(); }); @@ -835,7 +1035,8 @@ ok( $node_B->poll_query_until( # Create the second retention subscription in dbb. The launcher must # re-initialize the slot's xmin before launching dbb's apply worker. my $subname_BA2 = 'tap_sub_b_a2'; -$node_B->safe_psql('dbb', +$node_B->safe_psql( + 'dbb', "CREATE SUBSCRIPTION $subname_BA2 CONNECTION '$node_A_connstr application_name=$subname_BA2' PUBLICATION tap_pub_A @@ -890,16 +1091,13 @@ ok( $node_A->poll_query_until( # A conflict log table is system-managed and cannot be altered directly, so # moving it to another tablespace must be rejected. ############################################################################### -my $subid = $node_subscriber->safe_psql('postgres', - "SELECT oid FROM pg_subscription WHERE subname = 'sub_tab';"); -my $clt = "pg_conflict.pg_conflict_log_$subid"; - (undef, undef, $stderr) = $node_subscriber->psql('postgres', "ALTER TABLE $clt SET TABLESPACE pg_default"); like( $stderr, qr/cannot alter conflict log table "pg_conflict_log_\d+"/, - "moving a conflict log table with ALTER TABLE SET TABLESPACE is rejected"); + "moving a conflict log table with ALTER TABLE SET TABLESPACE is rejected" +); ############################################################################### # ALTER TABLE ALL IN TABLESPACE must skip conflict log tables, the same way it @@ -907,7 +1105,8 @@ like( # so the bulk move only touches the objects created here. ############################################################################### $node_subscriber->safe_psql('postgres', "CREATE DATABASE clt_ts_test"); -$node_subscriber->safe_psql('clt_ts_test', +$node_subscriber->safe_psql( + 'clt_ts_test', "CREATE SUBSCRIPTION sub_ts_test CONNECTION 'dbname=nonexistent' PUBLICATION pub @@ -951,4 +1150,199 @@ $node_subscriber->safe_psql('clt_ts_test', "DROP SUBSCRIPTION sub_ts_test"); $node_subscriber->safe_psql('postgres', "DROP DATABASE clt_ts_test"); $node_subscriber->safe_psql('postgres', "DROP TABLESPACE backup_space"); +############################################################################### +# Check that the conflict log table can't be dropped by ALTER SUBSCRIPTION ... +# SET (conflict_log_destination = 'log') unless the subscription is disabled +# and has no prepared transactions, nor by DROP SUBSCRIPTION while prepared +# transactions exist, and that the apply worker doesn't wait for the conflict +# log table lock behind such a command. +############################################################################### +$node_publisher->safe_psql( + 'postgres', qq[ + CREATE TABLE tab_2pc (a int PRIMARY KEY); + CREATE PUBLICATION pub_2pc FOR TABLE tab_2pc; +]); +$node_subscriber->safe_psql('postgres', + "CREATE TABLE tab_2pc (a int PRIMARY KEY)"); + +$node_subscriber->append_conf('postgresql.conf', "deadlock_timeout = 10ms"); +$node_subscriber->reload; + +my $appname_2pc = 'sub_2pc'; +$node_subscriber->safe_psql( + 'postgres', + "CREATE SUBSCRIPTION sub_2pc + CONNECTION '$publisher_connstr application_name=$appname_2pc' + PUBLICATION pub_2pc + WITH (two_phase = true, conflict_log_destination = 'table')"); +$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_2pc); +$node_subscriber->poll_query_until('postgres', + "SELECT subtwophasestate = 'e' FROM pg_subscription WHERE subname = 'sub_2pc'" +) or die "Timed out while waiting for two_phase to be enabled"; + +my $clt_2pc = $node_subscriber->safe_psql('postgres', + "SELECT subconflictlogrelid::regclass FROM pg_subscription WHERE subname = 'sub_2pc'" +); + +# Replicate rows, then remove them locally so that the remote DELETEs of +# them raise delete_missing conflicts. +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_2pc VALUES (0), (1), (2)"); +$node_publisher->wait_for_catchup($appname_2pc); +$node_subscriber->safe_psql('postgres', "DELETE FROM tab_2pc"); + +# When no prepared transaction exists, the apply worker waits for a conflicting +# lock on the conflict log table past deadlock_timeout without erroring out, +# and records the conflict once the lock is released. +my $log_offset_2pc = -s $node_subscriber->logfile; +my $lock_session = $node_subscriber->background_psql('postgres'); +$lock_session->query_safe( + qq[ + BEGIN; + LOCK TABLE $clt_2pc IN ACCESS EXCLUSIVE MODE; +]); + +$node_publisher->safe_psql('postgres', "DELETE FROM tab_2pc WHERE a = 0"); +$node_subscriber->wait_for_log( + qr/still waiting for RowExclusiveLock on relation/, + $log_offset_2pc); +unlike( + slurp_file($node_subscriber->logfile, $log_offset_2pc), + qr/ERROR: could not obtain lock on conflict log table of subscription "sub_2pc"/, + 'apply worker does not error out on conflict log table lock when no prepared transactions exist' +); + +$lock_session->query_safe("ROLLBACK;"); +$lock_session->quit; +$node_publisher->wait_for_catchup($appname_2pc); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM $clt_2pc"), + '1', + 'conflict is recorded after conflict log table lock is released'); +$node_subscriber->safe_psql('postgres', "TRUNCATE $clt_2pc"); + +$node_publisher->safe_psql( + 'postgres', qq[ + BEGIN; + DELETE FROM tab_2pc WHERE a = 1; + PREPARE TRANSACTION 'gx_2pc'; +]); +$node_publisher->wait_for_catchup($appname_2pc); + +# The prepared transaction on the subscriber holds the lock taken to record +# the conflict. +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_locks + WHERE relation = '$clt_2pc'::regclass AND pid IS NULL AND granted"), + '1', + 'prepared transaction holds a lock on the conflict log table'); + +# A command needing a stronger lock on the conflict log table, such as +# TRUNCATE, waits for the prepared transaction. Check that the apply worker +# errors out on its next conflict instead of waiting behind the command, and +# applies the change once the command is cancelled. +$log_offset_2pc = -s $node_subscriber->logfile; +my $truncate_session = + $node_subscriber->background_psql('postgres', on_error_stop => 0); +my $truncate_pid = $truncate_session->query_safe("SELECT pg_backend_pid()"); +$truncate_session->query_until( + qr/truncate_started/, qq[ + \\echo truncate_started + TRUNCATE $clt_2pc; +]); +$node_subscriber->poll_query_until( + 'postgres', + "SELECT count(*) = 1 FROM pg_locks + WHERE relation = '$clt_2pc'::regclass + AND mode = 'AccessExclusiveLock' AND NOT granted" +) or die "Timed out while waiting for TRUNCATE to wait for the lock"; + +$node_publisher->safe_psql('postgres', "DELETE FROM tab_2pc WHERE a = 2"); +$node_subscriber->wait_for_log( + qr/ERROR: could not obtain lock on conflict log table of subscription "sub_2pc"/, + $log_offset_2pc); +pass( + 'apply worker errors out when the conflict log table lock is not available with two_phase enabled' +); + +$node_subscriber->safe_psql('postgres', + "SELECT pg_cancel_backend($truncate_pid)"); +$truncate_session->quit; +$node_publisher->wait_for_catchup($appname_2pc); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM $clt_2pc"), + '1', + 'conflict is recorded after the waiting command is cancelled'); + +# Fail - the subscription is enabled +($cmdret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "ALTER SUBSCRIPTION sub_2pc SET (conflict_log_destination = 'log')"); +like( + $stderr, + qr/ERROR: cannot set option "conflict_log_destination" for enabled subscription/, + "changing conflict_log_destination to 'log' is not allowed for enabled subscription" +); + +# Disable the subscription and wait for the apply worker to stop +$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub_2pc DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'sub_2pc' AND pid IS NOT NULL" +) or die "Timed out while waiting for the apply worker to stop"; + +# Fail - prepared transactions exist +($cmdret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "ALTER SUBSCRIPTION sub_2pc SET (conflict_log_destination = 'log')"); +like( + $stderr, + qr/ERROR: cannot alter option "conflict_log_destination" when prepared transactions exist/, + "changing conflict_log_destination to 'log' is not allowed when prepared transactions exist" +); + +($cmdret, $stdout, $stderr) = + $node_subscriber->psql('postgres', "DROP SUBSCRIPTION sub_2pc"); +like( + $stderr, + qr/ERROR: cannot drop subscription "sub_2pc" when prepared transactions exist/, + "dropping subscription with a conflict log table is not allowed when prepared transactions exist" +); + +# Resolve the prepared transaction: commit it on the publisher and let the +# apply worker apply COMMIT PREPARED. +$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'gx_2pc'"); +$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub_2pc ENABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_prepared_xacts") + or die + "Timed out while waiting for the prepared transaction to be resolved"; +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT count(*) FROM $clt_2pc WHERE conflict_type = 'delete_missing'" + ), + '2', + 'conflict recorded by the prepared transaction is committed'); + +# With the subscription disabled and no prepared transactions, the conflict +# log table can be dropped. +$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub_2pc DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'sub_2pc' AND pid IS NOT NULL" +) or die "Timed out while waiting for the apply worker to stop"; +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION sub_2pc SET (conflict_log_destination = 'log')"); +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT subconflictlogrelid FROM pg_subscription WHERE subname = 'sub_2pc'" + ), + '0', + "conflict_log_destination is changed to 'log' for disabled subscription without prepared transactions" +); + +# Clean up +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION sub_2pc"); +$node_subscriber->safe_psql('postgres', "DROP TABLE tab_2pc"); +$node_publisher->safe_psql( + 'postgres', qq[ + DROP PUBLICATION pub_2pc; + DROP TABLE tab_2pc; +]); + done_testing(); -- 2.54.0