From fd9c91e45b6a28fad202b6ca1b976f9fd94696c1 Mon Sep 17 00:00:00 2001 From: Dilip Kumar Date: Mon, 6 Jul 2026 12:15:36 +0530 Subject: [PATCH v68] Implement the conflict insertion infrastructure for the conflict log table 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 along with the corresponding local state at the time of the conflict. 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. 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. This avoids modifying the exposed table schema and dealing with upgrade handling when multi-row conflict resolution is introduced in the future. The JSON array uses the following structured format: [ { "xid": "1001", "commit_ts": "2025-12-25 10:00:00+05:30", "origin": "node_1", "tuple": {"id": 1, "val": "old_data"} }, ... ] Example of querying the structured conflict data: SELECT remote_xid, relname, remote_origin, local_conflicts[1] ->> 'xid' AS local_xid, local_conflicts[1] ->> 'tuple' AS local_tuple FROM pg_conflict.pg_conflict_log_16396; remote_xid | relname | remote_origin | local_xid | local_tuple ------------+----------+---------------+-----------+--------------------- 760 | test | pg_16406 | 771 | {"a":1,"b":10} 765 | conf_tab | pg_16406 | 775 | {"a":2,"b":2,"c":2} --- doc/src/sgml/logical-replication.sgml | 691 +++++++++++++-------- doc/src/sgml/ref/alter_subscription.sgml | 3 +- doc/src/sgml/ref/create_subscription.sgml | 22 +- src/backend/replication/logical/conflict.c | 578 +++++++++++++++-- src/backend/replication/logical/worker.c | 15 +- src/include/replication/conflict.h | 2 + src/include/replication/worker_internal.h | 4 + src/include/utils/rel.h | 13 +- src/test/subscription/t/030_origin.pl | 4 +- src/test/subscription/t/035_conflicts.pl | 38 +- 10 files changed, 1037 insertions(+), 333 deletions(-) diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 36298cacb75..8acef1a6118 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -2033,120 +2033,262 @@ 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 a 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. Note that 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 always reported exclusively to the server log. + - - 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 a 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. + + + + + 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 + . + + + + 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. + + + + 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. + + + remote_commit_ts + timestamptz + The remote commit timestamp of the remote transaction. + + + remote_origin + text + The origin of the remote transaction. + + + replica_identity_full + boolean + Indicates whether replica_identity represents a full tuple (true) or key values of a replica identity index (false). This is NULL when replica_identity is not logged. + + + replica_identity + json + The JSON representation of the replica identity key values or full tuple. + + + remote_tuple + json + The JSON representation of the incoming remote row that caused + the conflict. + + + 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), origin (origin), + and the full local row image (tuple). + + + + +
+ + + The conflicting row data, including the incoming remote row (remote_tuple) + and the associated local conflict details (local_conflicts), is stored in + JSON formats for flexible querying and analysis. + + + + + 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 +2301,185 @@ 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. + + + + + 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. - conflict_type is the type of conflict that occurred - (e.g., insert_exists, update_exists). + 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. - - - - - - - + + + + - - 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. - + + 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. - + + 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: + + 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 +2593,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 25a81e2e62a..ad417b012b3 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 always reported to the server log. diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index f0a6ed20c60..9130b4c0047 100644 --- a/src/backend/replication/logical/conflict.c +++ b/src/backend/replication/logical/conflict.c @@ -16,17 +16,22 @@ #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/lsyscache.h" +#include "utils/pg_lsn.h" /* * String representations for the supported conflict logging destinations. @@ -63,6 +68,11 @@ typedef struct ConflictLogColumnDef * 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}, @@ -81,6 +91,19 @@ static const ConflictLogColumnDef ConflictLogSchema[] = { #define NUM_CONFLICT_ATTRS ((AttrNumber) lengthof(ConflictLogSchema)) +/* + * 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}, + {.attname = "tuple", .atttypid = JSONOID} +}; + +#define NUM_LOCAL_CONFLICT_ATTRS lengthof(LocalConflictSchema) + static const char *const ConflictTypeNames[] = { [CT_INSERT_EXISTS] = "insert_exists", [CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs", @@ -108,8 +131,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 tuple_table_slot_to_json_datum(TupleTableSlot *slot); +static Datum tuple_table_slot_to_indextup_json(EState *estate, + Relation localrel, + Oid replica_index, + TupleTableSlot *slot); +static TupleDesc build_conflict_tupledesc(void); +static Datum build_local_conflicts_json_array(List *conflicttuples); +static HeapTuple prepare_conflict_log_tuple(EState *estate, Relation rel, + Relation conflictlogrel, + ConflictType conflict_type, + TupleTableSlot *searchslot, + List *conflicttuples, + TupleTableSlot *remoteslot); +static void insert_conflict_log_tuple(Relation conflictlogrel, HeapTuple tuple); /* * Builds the TupleDesc for the conflict log table. @@ -279,29 +320,114 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, TupleTableSlot *remoteslot, List *conflicttuples) { Relation localrel = relinfo->ri_RelationDesc; - StringInfoData err_detail; + ConflictLogDest dest; + Relation conflictlogrel = NULL; + bool log_dest_table = false; + bool log_dest_logfile = false; - 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); + /* + * 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. + */ + if (elevel < ERROR) + { + conflictlogrel = GetConflictLogDestAndTable(&dest); + log_dest_table = CONFLICTS_LOGGED_TO_TABLE(dest); + log_dest_logfile = CONFLICTS_LOGGED_TO_LOG(dest); - pgstat_report_subscription_conflict(MySubscription->oid, type); + /* + * If a conflict log table was requested but it has been dropped + * concurrently (e.g. a concurrent ALTER SUBSCRIPTION changed + * conflict_log_destination), GetConflictLogDestAndTable() returned + * NULL. Fall back to logging to the server log so that the conflict + * is not lost. + */ + if (log_dest_table && conflictlogrel == NULL) + { + log_dest_table = false; + log_dest_logfile = true; + } + } + else + { + /* ERROR-level conflicts always go to the server log with full detail */ + log_dest_logfile = true; + } - ereport(elevel, - errcode_apply_conflict(type), - errmsg("conflict detected on relation \"%s.%s\": conflict=%s", - get_namespace_name(RelationGetNamespace(localrel)), - RelationGetRelationName(localrel), - ConflictTypeNames[type]), - errdetail_internal("%s", err_detail.data)); + /* + * 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 (log_dest_logfile) + { + StringInfoData err_detail; + + 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); + + /* Standard reporting with full internal details. */ + ereport(elevel, + errcode_apply_conflict(type), + errmsg("conflict detected on relation \"%s\": conflict=%s", + RelationGetQualifiedRelationName(localrel), + ConflictTypeNames[type]), + errdetail_internal("%s", err_detail.data)); + } + else if (log_dest_table) + { + /* + * Not logging conflict details to the server log; report the conflict + * but omit raw tuple data since it is captured in the conflict log + * table. + */ + ereport(elevel, + errcode_apply_conflict(type), + errmsg("conflict detected on relation \"%s\": conflict=%s", + RelationGetQualifiedRelationName(localrel), + ConflictTypeNames[type]), + errdetail("Conflict details are logged to the conflict log table: %s", + RelationGetRelationName(conflictlogrel))); + } + + /* + * Insert into the conflict log table for LOG-level conflicts if + * requested. + */ + if (log_dest_table) + { + HeapTuple tuple; + + Assert(conflictlogrel != NULL); + Assert(elevel < ERROR); + + tuple = prepare_conflict_log_tuple(estate, + relinfo->ri_RelationDesc, + conflictlogrel, + type, + searchslot, + conflicttuples, + remoteslot); + + insert_conflict_log_tuple(conflictlogrel, tuple); + table_close(conflictlogrel, NoLock); + + heap_freetuple(tuple); + } } /* @@ -335,6 +461,61 @@ InitConflictIndexes(ResultRelInfo *relInfo) relInfo->ri_onConflictArbiterIndexes = uniqueIndexes; } +/* + * GetConflictLogDestAndTable + * + * Fetches conflict logging metadata from the cached MySubscription pointer. + * Sets the destination enum in *log_dest and, if a table is one of the + * destinations, opens and returns the relation handle for the conflict log + * table. + * + * The table is opened with try_table_open(), so NULL is returned if the + * conflict log table has been dropped concurrently (e.g. by an ALTER + * SUBSCRIPTION that changed conflict_log_destination). Callers must treat a + * NULL result for a table destination as "table unavailable" and fall back to + * server-log reporting rather than failing. + */ +Relation +GetConflictLogDestAndTable(ConflictLogDest *log_dest) +{ + Oid conflictlogrelid; + + /* + * 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(): the table may have been dropped concurrently by + * an ALTER SUBSCRIPTION that changed conflict_log_destination. Returning + * NULL lets the caller fall back to the server log instead of failing. + */ + return try_table_open(conflictlogrelid, RowExclusiveLock); +} + +/* + * insert_conflict_log_tuple + * + * Insert conflict log tuple into the conflict log table. + */ +static void +insert_conflict_log_tuple(Relation conflictlogrel, HeapTuple tuple) +{ + Assert(tuple != NULL); + + heap_insert(conflictlogrel, tuple, + GetCurrentCommandId(true), 0, NULL); +} + /* * Add SQLSTATE error code to the current conflict report. */ @@ -768,6 +949,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. @@ -783,41 +998,320 @@ 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); - /* - * 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)) + build_index_datums_from_slot(estate, localrel, slot, indexDesc, values, + isnull); + + index_value = BuildIndexValueDescription(indexDesc, values, isnull); + + index_close(indexDesc, NoLock); + + return index_value; +} + +/* + * tuple_table_slot_to_json_datum + * + * Helper function to convert a TupleTableSlot to JSON. + */ +static Datum +tuple_table_slot_to_json_datum(TupleTableSlot *slot) +{ + HeapTuple tuple; + Datum datum; + Datum json; + + Assert(slot != NULL); + + tuple = ExecCopySlotHeapTuple(slot); + datum = heap_copy_tuple_as_datum(tuple, slot->tts_tupleDescriptor); + + json = DirectFunctionCall1(row_to_json, datum); + heap_freetuple(tuple); + + return json; +} + +/* + * tuple_table_slot_to_indextup_json + * + * Fetch replica identity key from the tuple table slot and convert into a + * JSON datum. + */ +static Datum +tuple_table_slot_to_indextup_json(EState *estate, Relation localrel, + Oid indexid, TupleTableSlot *slot) +{ + Relation indexDesc; + Datum values[INDEX_MAX_KEYS]; + bool isnull[INDEX_MAX_KEYS]; + HeapTuple tuple; + TupleDesc tupdesc; + 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); + tupdesc = CreateTupleDescCopy(RelationGetDescr(indexDesc)); + + /* Bless the tupdesc so it can be looked up by row_to_json. */ + BlessTupleDesc(tupdesc); + + /* Form the replica identity tuple. */ + tuple = heap_form_tuple(tupdesc, values, isnull); + datum = heap_copy_tuple_as_datum(tuple, tupdesc); + + heap_freetuple(tuple); + FreeTupleDesc(tupdesc); + index_close(indexDesc, NoLock); + + /* Convert to a JSON datum. */ + return DirectFunctionCall1(row_to_json, datum); +} + +/* + * build_conflict_tupledesc + * + * Build and bless a tuple descriptor for the conflict log table based on the + * predefined LocalConflictSchema. + */ +static TupleDesc +build_conflict_tupledesc(void) +{ + static TupleDesc cached_tupdesc = NULL; + + if (cached_tupdesc == NULL) { - tableslot = table_slot_create(localrel, &estate->es_tupleTable); - tableslot = ExecCopySlot(tableslot, slot); + 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); } - /* - * Initialize ecxt_scantuple for potential use in FormIndexDatum when - * index expressions are present. - */ - GetPerTupleExprContext(estate)->ecxt_scantuple = tableslot; + return cached_tupdesc; +} - /* - * The values/nulls arrays passed to BuildIndexValueDescription should be - * the results of FormIndexDatum, which are the "raw" input to the index - * AM. - */ - FormIndexDatum(BuildIndexInfo(indexDesc), tableslot, estate, values, isnull); +/* + * Builds the local conflicts JSON array column from the list of + * ConflictTupleInfo objects. + * + * Example output structure: + * [ { "xid": "1001", "commit_ts": "...", "origin": "...", "tuple": {...} }, ... ] + */ +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; - index_value = BuildIndexValueDescription(indexDesc, values, isnull); + /* Build local conflicts tuple descriptor. */ + tupdesc = build_conflict_tupledesc(); - index_close(indexDesc, NoLock); + num_conflicts = list_length(conflicttuples); + json_datum_array = palloc_array(Datum, num_conflicts); - return index_value; + /* 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; + + /* Convert conflicting tuple to JSON datum. */ + if (conflicttuple->slot) + values[attno++] = tuple_table_slot_to_json_datum(conflicttuple->slot); + 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; +} + +/* + * prepare_conflict_log_tuple + * + * This routine prepares a tuple detailing a conflict encountered during + * logical replication to be inserted into the conflict log table. + */ +static HeapTuple +prepare_conflict_log_tuple(EState *estate, Relation rel, + Relation conflictlogrel, + ConflictType conflict_type, + TupleTableSlot *searchslot, + List *conflicttuples, + TupleTableSlot *remoteslot) +{ + Datum values[NUM_CONFLICT_ATTRS] = {0}; + bool nulls[NUM_CONFLICT_ATTRS] = {0}; + int attno; + char *remote_origin = NULL; + HeapTuple tuple; + + /* 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; + + if (!TupIsNull(searchslot)) + { + Oid replica_index = GetRelationIdentityOrPK(rel); + + /* + * If the table has a valid replica identity index, build the index + * JSON datum from key value. Otherwise, construct it from the + * complete tuple in REPLICA IDENTITY FULL cases. + */ + if (OidIsValid(replica_index)) + { + values[attno++] = BoolGetDatum(false); + values[attno++] = tuple_table_slot_to_indextup_json(estate, rel, + replica_index, + searchslot); + } + else + { + values[attno++] = BoolGetDatum(true); + values[attno++] = tuple_table_slot_to_json_datum(searchslot); + } + } + else + { + nulls[attno++] = true; + nulls[attno++] = true; + } + + if (!TupIsNull(remoteslot)) + values[attno++] = tuple_table_slot_to_json_datum(remoteslot); + else + nulls[attno++] = true; + + values[attno] = build_local_conflicts_json_array(conflicttuples); + + Assert(attno + 1 == NUM_CONFLICT_ATTRS); + + tuple = heap_form_tuple(RelationGetDescr(conflictlogrel), values, nulls); + + return tuple; } diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 92ea1d0df24..689002dcfec 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -488,7 +488,9 @@ static bool MySubscriptionValid = false; static List *on_commit_wakeup_workers_subids = NIL; bool in_remote_transaction = false; -static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr; +XLogRecPtr remote_final_lsn = InvalidXLogRecPtr; +TransactionId remote_xid = InvalidTransactionId; +TimestampTz remote_commit_ts = 0; /* fields valid only when processing streamed transaction */ static bool in_streamed_transaction = false; @@ -1237,6 +1239,8 @@ apply_handle_begin(StringInfo s) set_apply_error_context_xact(begin_data.xid, begin_data.final_lsn); remote_final_lsn = begin_data.final_lsn; + remote_commit_ts = begin_data.committime; + remote_xid = begin_data.xid; maybe_start_skipping_changes(begin_data.final_lsn); @@ -1297,6 +1301,8 @@ apply_handle_begin_prepare(StringInfo s) set_apply_error_context_xact(begin_data.xid, begin_data.prepare_lsn); remote_final_lsn = begin_data.prepare_lsn; + remote_xid = begin_data.xid; + remote_commit_ts = 0; maybe_start_skipping_changes(begin_data.prepare_lsn); @@ -1768,6 +1774,10 @@ apply_handle_stream_start(StringInfo s) (errcode(ERRCODE_PROTOCOL_VIOLATION), errmsg_internal("invalid transaction ID in streamed replication transaction"))); + remote_xid = stream_xid; + remote_final_lsn = InvalidXLogRecPtr; + remote_commit_ts = 0; + set_apply_error_context_xact(stream_xid, InvalidXLogRecPtr); /* Try to allocate a worker for the streaming transaction. */ @@ -2429,6 +2439,9 @@ apply_handle_stream_commit(StringInfo s) { case TRANS_LEADER_APPLY: + /* Set remote_commit_ts for conflict logging. */ + remote_commit_ts = commit_data.committime; + /* * The transaction has been serialized to file, so replay all the * spooled operations. diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h index 326570ed639..8b66b1876e1 100644 --- a/src/include/replication/conflict.h +++ b/src/include/replication/conflict.h @@ -12,6 +12,7 @@ #include "access/xlogdefs.h" #include "datatype/timestamp.h" #include "nodes/pg_list.h" +#include "utils/relcache.h" /* Avoid including execnodes.h here */ typedef struct EState EState; @@ -115,4 +116,5 @@ extern void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, TupleTableSlot *remoteslot, List *conflicttuples); extern void InitConflictIndexes(ResultRelInfo *relInfo); +extern Relation GetConflictLogDestAndTable(ConflictLogDest *log_dest); #endif diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 88cb7c1e252..45d381d5000 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -256,6 +256,10 @@ extern PGDLLIMPORT bool InitializingApplyWorker; extern PGDLLIMPORT List *table_states_not_ready; +extern XLogRecPtr remote_final_lsn; +extern TimestampTz remote_commit_ts; +extern TransactionId remote_xid; + 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 89c159b133f..13bece6a4ef 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -713,16 +713,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/subscription/t/030_origin.pl b/src/test/subscription/t/030_origin.pl index 6bc6b7874c2..5f4d00bdd33 100644 --- a/src/test/subscription/t/030_origin.pl +++ b/src/test/subscription/t/030_origin.pl @@ -166,7 +166,7 @@ is($result, qq(32), 'The node_A data replicated to node_B'); $node_C->safe_psql('postgres', "UPDATE $tab SET a = 33 WHERE a = 32;"); $node_B->wait_for_log( - qr/conflict detected on relation "public.$tab_unquoted": conflict=update_origin_differs.*\n.*DETAIL:.* Updating the row that was modified by a different origin ".*" in transaction [0-9]+ at .*: local row \(32\), remote row \(33\), replica identity \(a\)=\(32\)./ + qr/conflict detected on relation "public.$tab": conflict=update_origin_differs.*\n.*DETAIL:.* Updating the row that was modified by a different origin ".*" in transaction [0-9]+ at .*: local row \(32\), remote row \(33\), replica identity \(a\)=\(32\)./ ); $node_B->safe_psql('postgres', "DELETE FROM $tab;"); @@ -182,7 +182,7 @@ is($result, qq(33), 'The node_A data replicated to node_B'); $node_C->safe_psql('postgres', "DELETE FROM $tab WHERE a = 33;"); $node_B->wait_for_log( - qr/conflict detected on relation "public.$tab_unquoted": conflict=delete_origin_differs.*\n.*DETAIL:.* Deleting the row that was modified by a different origin ".*" in transaction [0-9]+ at .*: local row \(33\), replica identity \(a\)=\(33\).*/ + qr/conflict detected on relation "public.$tab": conflict=delete_origin_differs.*\n.*DETAIL:.* Deleting the row that was modified by a different origin ".*" in transaction [0-9]+ at .*: local row \(33\), replica identity \(a\)=\(33\).*/ ); # The remaining tests no longer test conflict detection. diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl index 3910a49c0aa..5efef6614ef 100644 --- a/src/test/subscription/t/035_conflicts.pl +++ b/src/test/subscription/t/035_conflicts.pl @@ -84,6 +84,17 @@ $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 +125,11 @@ $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 +199,7 @@ $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 +209,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 +332,13 @@ 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'); + $log_location = -s $node_A->logfile; $node_A->safe_psql('postgres', "ALTER SUBSCRIPTION $subname_AB ENABLE;"); @@ -329,6 +352,13 @@ 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'); + # Remember the next transaction ID to be assigned my $next_xid = $node_A->safe_psql('postgres', "SELECT txid_current() + 1;"); @@ -674,10 +704,6 @@ 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( -- 2.49.0