From c012d67f3cc2ae48440518113b384f4860011113 Mon Sep 17 00:00:00 2001 From: Dilip Kumar Date: Sun, 4 Oct 2026 10:59:02 +0530 Subject: [PATCH v82 1/3] Implement the conflict insertion infrastructure for the conflict log table MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This patch introduces the core logic to populate the conflict log table whenever a logical replication conflict is detected. It captures the remote transaction details, the replica identity key identifying the conflicting row, and the corresponding local conflicting tuple metadata. Only resolved (LOG-level) conflicts are recorded in the conflict log 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. To avoid serializing full tuples that could exceed PostgreSQL's 1GB varlena limit and halt the apply worker, the 'remote_tuple' column is removed from the conflict log table schema, '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 Column: When the table has a replica identity index or primary key, 'replica_identity' stores a JSON object keyed by column name. Rather than calling row_to_json()—which honors user-defined CAST (t AS json) functions whose output could exceed the 1GB limit—each key value is rendered as a JSON string using its type's output function (via rd_opcintype), matching the representation used in the server log. If a key value's output exceeds 64kB (CONFLICT_MAX_VALUE_SIZE), it is replaced by a marker object recording its length ({"omitted":true,"length":...}) and the new 'has_omitted_values' column is set to true. Values are omitted rather than truncated so that equality and join predicates against the conflict log table never silently produce wrong answers. Local Conflicts Column: The 'local_conflicts' column is typed as an array of JSON objects (json[]). Although currently recorded conflicts involve a single local tuple, the column type is preserved as a JSON array for future-proofing when multi-row conflict resolution is introduced. Each element has the following format: [ { "xid": "1001", "commit_ts": "2025-12-25 10:00:00+05:30", "origin": "node_1" }, ... ] Example of querying the structured conflict data: SELECT remote_xid, relname, remote_origin, replica_identity, local_conflicts[1] ->> 'xid' AS local_xid FROM pg_conflict.pg_conflict_log_16396; remote_xid | relname | remote_origin | replica_identity | local_xid ------------+----------+---------------+------------------+----------- 760 | test | pg_16406 | {"a":"1"} | 771 765 | conf_tab | pg_16406 | {"a":"2"} | 775 The remote transaction details (xid, final LSN, commit timestamp) recorded in the conflict log table are fetched from the apply worker's remote transaction context through GetRemoteTransactionInfoForConflict() instead of new global variables. --- doc/src/sgml/logical-replication.sgml | 756 +++++++++++++-------- doc/src/sgml/ref/alter_subscription.sgml | 3 +- doc/src/sgml/ref/create_subscription.sgml | 22 +- src/backend/replication/logical/conflict.c | 596 +++++++++++++++- src/backend/replication/logical/worker.c | 90 ++- src/include/replication/worker_internal.h | 5 + src/include/utils/rel.h | 13 +- src/test/regress/expected/subscription.out | 4 +- src/test/subscription/t/035_conflicts.pl | 252 ++++++- 9 files changed, 1394 insertions(+), 347 deletions(-) diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 4701a3d9d18..0d88cbfd4fd 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,380 @@ 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. + + + + + 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. - conflict_type is the type of conflict that occurred - (e.g., insert_exists, update_exists). + 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 final LSN 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 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. + + + remote_origin + text + The origin of the remote transaction. + + + 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 or + when replica identity is not applicable. + + + 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 origin (origin). + This 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 + + + + + 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. + + + + + + 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 +2660,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..d1c49c2c7c4 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -366,7 +366,8 @@ 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. 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/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index 90431ffce98..efc398286de 100644 --- a/src/backend/replication/logical/conflict.c +++ b/src/backend/replication/logical/conflict.c @@ -16,17 +16,23 @@ #include "access/commit_ts.h" #include "access/genam.h" +#include "access/heapam.h" #include "access/tableam.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 "utils/array.h" +#include "utils/builtins.h" +#include "utils/json.h" #include "utils/lsyscache.h" +#include "utils/pg_lsn.h" /* * String representations for the supported conflict logging destinations. @@ -56,13 +62,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}, @@ -75,12 +86,37 @@ static const ConflictLogColumnDef ConflictLogSchema[] = { {.attname = "remote_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 +128,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 +146,24 @@ 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, + TupleTableSlot *searchslot, + List *conflicttuples); /* * Builds the TupleDesc for the conflict log table. @@ -279,21 +333,54 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, TupleTableSlot *remoteslot, List *conflicttuples) { Relation localrel = relinfo->ri_RelationDesc; + ConflictLogDest dest; + Relation conflictlogrel; StringInfoData err_detail; - initStringInfo(&err_detail); + pgstat_report_subscription_conflict(MySubscription->oid, type); - /* 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); + conflictlogrel = get_conflictlog_dest_and_table(elevel, &dest); - pgstat_report_subscription_conflict(MySubscription->oid, type); + initStringInfo(&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 + { + 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 +389,21 @@ 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, + searchslot, + conflicttuples); + table_close(conflictlogrel, NoLock); + } } /* @@ -335,6 +437,59 @@ 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. + */ +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)); + + /* + * 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, RowExclusiveLock); + + if (!conflictlogrel) + *log_dest = CONFLICT_LOG_DEST_LOG; + + return conflictlogrel; +} + /* * Add SQLSTATE error code to the current conflict report. */ @@ -772,6 +927,40 @@ 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. + * This is the reusable core logic. + */ +static void +build_index_datums_from_slot(EState *estate, Relation localrel, + TupleTableSlot *slot, + Relation indexDesc, Datum *values, + bool *isnull) +{ + TupleTableSlot *tableslot = slot; + + /* + * If the slot is a virtual slot, copy it into a heap tuple slot as + * FormIndexDatum only works with heap tuple slots. + */ + if (TTS_IS_VIRTUAL(slot)) + { + /* Slot is created within the EState's tuple table */ + tableslot = table_slot_create(localrel, &estate->es_tupleTable); + tableslot = ExecCopySlot(tableslot, slot); + } + + /* + * Initialize ecxt_scantuple for potential use in FormIndexDatum + */ + 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 +976,382 @@ 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. + */ +static Datum +build_index_key_json(EState *estate, Relation localrel, + Oid indexid, TupleTableSlot *slot, + bool *omitted) +{ + Relation indexDesc; + TupleDesc indexTupDesc; + 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); + indexTupDesc = RelationGetDescr(indexDesc); + + 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 handing the values to + * row_to_json(), and render each of them with its type's output function. + * + * row_to_json() would go through json_categorize_type(), which honours a + * user CREATE CAST (t AS json) for any type at or above + * FirstNormalObjectId. That cast is a function no other part of apply + * ever calls -- the publisher sends the value using the type's output + * function, the index compares it with the opclass, and neither asks for + * a json representation -- so its result is unrelated to the size of the + * value we received, and a tiny key could render a json value above the + * 1GB limit and error out the apply worker. Using the output function + * keeps us to the same representation the publisher already produced, as + * 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 = TupleDescAttr(indexTupDesc, i); + + 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); + 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. + */ + 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. + */ + 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, + TupleTableSlot *searchslot, + List *conflicttuples) +{ + Datum values[NUM_CONFLICT_ATTRS] = {0}; + bool nulls[NUM_CONFLICT_ATTRS] = {0}; + int attno; + char *remote_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]); + + if (TransactionIdIsValid(remote_xid)) + values[attno++] = TransactionIdGetDatum(remote_xid); + else + nulls[attno++] = true; + + 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, &remote_origin); + + if (remote_origin != NULL) + values[attno++] = CStringGetTextDatum(remote_origin); + else + nulls[attno++] = true; + /* - * Initialize ecxt_scantuple for potential use in FormIndexDatum when - * index expressions are present. + * 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. */ - GetPerTupleExprContext(estate)->ecxt_scantuple = tableslot; + Assert(!TupIsNull(searchslot)); + + replica_index = GetRelationIdentityOrPK(rel); /* - * The values/nulls arrays passed to BuildIndexValueDescription should be - * the results of FormIndexDatum, which are the "raw" input to the index - * AM. + * If the table has a valid 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. */ - FormIndexDatum(BuildIndexInfo(indexDesc), tableslot, estate, values, isnull); + 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; + } - index_value = BuildIndexValueDescription(indexDesc, values, isnull); + /* + * 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; - index_close(indexDesc, NoLock); + values[attno] = BoolGetDatum(omitted); - return index_value; + Assert(attno + 1 == NUM_CONFLICT_ATTRS); + + tuple = heap_form_tuple(RelationGetDescr(conflictlogrel), values, nulls); + + 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..c46afc2f55b 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); @@ -3039,7 +3055,7 @@ apply_handle_update_internal(ApplyExecutionData *edata, * nothing except for emitting a log message. */ ReportApplyConflict(estate, relinfo, LOG, type, remoteslot, newslot, - list_make1(&conflicttuple)); + type == CT_UPDATE_DELETED ? list_make1(&conflicttuple) : NIL); } /* Cleanup. */ @@ -3199,7 +3215,7 @@ apply_handle_delete_internal(ApplyExecutionData *edata, * emitting a log message. */ ReportApplyConflict(estate, relinfo, LOG, CT_DELETE_MISSING, - remoteslot, NULL, list_make1(&conflicttuple)); + remoteslot, NULL, NIL); } /* Cleanup. */ @@ -3560,7 +3576,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, */ ReportApplyConflict(estate, partrelinfo, LOG, type, remoteslot_part, newslot, - list_make1(&conflicttuple)); + type == CT_UPDATE_DELETED ? list_make1(&conflicttuple) : NIL); return; } @@ -6420,13 +6436,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 +6484,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/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..b678b7fe007 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -704,8 +704,8 @@ WHERE s.subname = 'regress_conflict_test1' AND a.attnum > 0 8 | remote_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 diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl index 80177b214c0..55125dd65a2 100644 --- a/src/test/subscription/t/035_conflicts.pl +++ b/src/test/subscription/t/035_conflicts.pl @@ -84,6 +84,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 +127,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 +203,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 +214,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 +337,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 +364,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 +425,189 @@ 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 consulted when recording +# the replica identity. +# +# Each key value is rendered with its type's output function, as the server log +# does, so a cast cannot substitute its own representation. This matters because +# such a cast is a function no other part of apply calls, and its output bears no +# relation to the size of the value received, so a tiny key could otherwise +# render a json value past the 1GB limit and error out the apply worker. +############################################################################### + +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"); + +############################################################################### +# Check that when an index opclass stores a type (opckeytype) different from +# the table column's type (such as GiST multirange_ops, whose opckeytype is +# anyrange), the table column's output function is used when recording the +# replica identity. +############################################################################### + +my $temporal_ddl = qq{ + CREATE TABLE tab_temporal ( + id int4range, + valid_at datemultirange, + PRIMARY KEY (id, valid_at WITHOUT OVERLAPS) + ); +}; + +$node_A->safe_psql('postgres', $temporal_ddl); +$node_B->safe_psql('postgres', $temporal_ddl); + +$node_A->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_A ADD TABLE tab_temporal"); +$node_B->safe_psql('postgres', + "ALTER SUBSCRIPTION $subname_BA REFRESH PUBLICATION"); +$node_B->wait_for_subscription_sync($node_A, $subname_BA); + +$node_A->safe_psql('postgres', + "INSERT INTO tab_temporal VALUES ('[1,2)', '{[2020-01-01,2021-01-01)}')"); +$node_A->wait_for_catchup($subname_BA); +$node_B->safe_psql('postgres', "DELETE FROM tab_temporal"); +$node_A->safe_psql('postgres', "DELETE FROM tab_temporal"); +$node_A->wait_for_catchup($subname_BA); + +my $clt_check_temporal = $node_B->poll_query_until('postgres', + "SELECT count(*) > 0 FROM $clt_BA WHERE relname = 'tab_temporal';"); +is($clt_check_temporal, 1, + 'delete_missing on tab_temporal logged into CLT on Node B'); + +my $clt_row_temporal = $node_B->safe_psql('postgres', + "SELECT replica_identity::text FROM $clt_BA WHERE relname = 'tab_temporal';" +); +is( $clt_row_temporal, + '{"id":"[1,2)","valid_at":"{[2020-01-01,2021-01-01)}"}', + 'replica identity uses table column type rather than index opckeytype'); + +# 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_temporal"); +$node_B->safe_psql('postgres', + "ALTER SUBSCRIPTION $subname_BA REFRESH PUBLICATION"); +$node_A->safe_psql('postgres', "DROP TABLE tab_temporal"); +$node_B->safe_psql('postgres', "DROP TABLE tab_temporal"); + ############################################################################### # 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 @@ -730,7 +962,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(); }); @@ -749,7 +982,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 @@ -804,16 +1038,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 @@ -821,7 +1052,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 -- 2.54.0 (Apple Git-157)