Add configurable conflict log table for Logical Replication

This patch adds a feature to provide a structured, queryable record of all
logical replication conflicts. The current approach of logging conflicts as
plain text in the server logs makes it difficult to query, analyze, and
use for external monitoring and automation.

This patch addresses these limitations by introducing a configurable
conflict_log_table option in the CREATE SUBSCRIPTION command. Key design
decisions include:

User-Managed Table: The conflict log is stored in a user-managed table
rather than a system catalog.

Structured Data: Conflict details, including the original and remote tuples,
are stored in JSON columns, providing a flexible format to accommodate different
table schemas.

Comprehensive Information: The log table captures essential attributes such as
local and remote transaction IDs, LSNs, commit timestamps, and conflict type,
providing a complete record for post-mortem analysis.

This feature will make logical replication conflicts easier to monitor and manage,
significantly improving the overall resilience and operability of replication setups.

The conflict log tables will not be included in a publication, even if the publication
is configured to include ALL TABLES or ALL TABLES IN SCHEMA.

Note: A single remote tuple may conflict with multiple local tuples when conflict type
is CT_MULTIPLE_UNIQUE_CONFLICTS, so for handling this case we create a single row in
conflict log table with respect to each remote conflict tuple even if it conflicts with
multiple local tuples and we store the multiple conflict tuples as a single JSON array
element in format as
[ { "xid": "1001", "commit_ts": "...", "origin": "...", "tuple": {...} }, ... ]
We can extract the elements of the local tuples from the conflict log table row
as given in below example.

SELECT remote_xid, relname, remote_origin, local_conflicts[1] ->> 'xid' AS local_xid,
       local_conflicts[1] ->> 'tuple' AS local_tuple
FROM myschema.conflict_log_history2;

 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}

Author: Dilip Kumar

src/backend/catalog/pg_publication.c

@@ -31,6 +31,7 @@
 #include "catalog/pg_publication_rel.h"
 #include "catalog/pg_type.h"
 #include "commands/publicationcmds.h"
+#include "commands/subscriptioncmds.h"
 #include "funcapi.h"
 #include "utils/array.h"
 #include "utils/builtins.h"
@@ -85,6 +86,15 @@ check_publication_add_relation(Relation targetrel)
 				 errmsg("cannot add relation \"%s\" to publication",
 						RelationGetRelationName(targetrel)),
 				 errdetail("This operation is not supported for unlogged tables.")));
+
+	/* Can't be conflict log table */
+	if (IsConflictLogTable(RelationGetRelid(targetrel)))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("cannot add relation \"%s.%s\" to publication",
+						get_namespace_name(RelationGetNamespace(targetrel)),
+						RelationGetRelationName(targetrel)),
+				 errdetail("This operation is not supported for conflict log tables.")));
 }
 
 /*
@@ -145,6 +155,13 @@ is_publishable_class(Oid relid, Form_pg_class reltuple)
 
 /*
  * Another variant of is_publishable_class(), taking a Relation.
+ *
+ * Note: Conflict log tables are not publishable.  However, we intentionally
+ * skip this check here because this function is called for every change and
+ * performing this check during every change publication is costly.  To ensure
+ * unpublishable entries are ignored without incurring performance overhead,
+ * tuples inserted into the conflict log table uses the HEAP_INSERT_NO_LOGICAL
+ * flag.  This allows the decoding layer to bypass these entries automatically.
  */
 bool
 is_publishable_relation(Relation rel)
@@ -169,7 +186,10 @@ pg_relation_is_publishable(PG_FUNCTION_ARGS)
 	tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
 	if (!HeapTupleIsValid(tuple))
 		PG_RETURN_NULL();
-	result = is_publishable_class(relid, (Form_pg_class) GETSTRUCT(tuple));
+
+	/* Subscription conflict log tables are not published */
+	result = is_publishable_class(relid, (Form_pg_class) GETSTRUCT(tuple)) &&
+			 !IsConflictLogTable(relid);
 	ReleaseSysCache(tuple);
 	PG_RETURN_BOOL(result);
 }
@@ -890,7 +910,9 @@ GetAllPublicationRelations(char relkind, bool pubviaroot)
 		Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
 		Oid			relid = relForm->oid;
 
+		/* Subscription conflict log tables are not published */
 		if (is_publishable_class(relid, relForm) &&
+			!IsConflictLogTable(relid) &&
 			!(relForm->relispartition && pubviaroot))
 			result = lappend_oid(result, relid);
 	}
@@ -1018,7 +1040,7 @@ GetSchemaPublicationRelations(Oid schemaid, PublicationPartOpt pub_partopt)
 		Oid			relid = relForm->oid;
 		char		relkind;
 
-		if (!is_publishable_class(relid, relForm))
+		if (!is_publishable_class(relid, relForm) || IsConflictLogTable(relid))
 			continue;
 
 		relkind = get_rel_relkind(relid);