From 50dc48b442e2c7683fb0e2948cec1e9b30f08c81 Mon Sep 17 00:00:00 2001 From: Vignesh C Date: Mon, 5 Jan 2026 15:46:10 +0530 Subject: [PATCH v19 4/6] Add shared index for conflict log table lookup and allow explicit publication throuhg FOR TABLE publication Introduce a dedicated shared unique index on pg_subscription.subconflictlogrelid to make conflict log table detection efficient, and index-backed. Previously, IsConflictLogTable() relied on a full catalog scan of pg_subscription, which was inefficient. This change adds pg_subscription_conflictrel_index and marks it as a shared index, matching the shared pg_subscription table, and rewrites conflict log table detection to use an indexed systable scan. Additionally conflict tables can be replicated when the table is explicitly specified through a FOR TABLE publication. --- src/backend/catalog/catalog.c | 1 + src/backend/catalog/pg_publication.c | 20 +-------------- src/backend/catalog/pg_subscription.c | 23 +++++++++++++++++ src/backend/commands/subscriptioncmds.c | 28 +++++++++------------ src/backend/replication/logical/conflict.c | 4 +-- src/backend/replication/pgoutput/pgoutput.c | 20 ++++++++++++--- src/bin/psql/describe.c | 4 ++- src/include/catalog/pg_proc.dat | 7 ++++++ src/include/catalog/pg_subscription.h | 1 + src/test/regress/expected/subscription.out | 7 +++--- src/test/regress/sql/subscription.sql | 5 ++-- 11 files changed, 73 insertions(+), 47 deletions(-) diff --git a/src/backend/catalog/catalog.c b/src/backend/catalog/catalog.c index d438dc682ec..148a6ccf998 100644 --- a/src/backend/catalog/catalog.c +++ b/src/backend/catalog/catalog.c @@ -361,6 +361,7 @@ IsSharedRelation(Oid relationId) relationId == SharedSecLabelObjectIndexId || relationId == SubscriptionNameIndexId || relationId == SubscriptionObjectIndexId || + relationId == SubscriptionConflictrelIndexId || relationId == TablespaceNameIndexId || relationId == TablespaceOidIndexId) return true; diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index cb383a5ce04..fcd48166edf 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -86,15 +86,6 @@ 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."))); } /* @@ -155,13 +146,6 @@ 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) @@ -187,9 +171,7 @@ pg_relation_is_publishable(PG_FUNCTION_ARGS) if (!HeapTupleIsValid(tuple)) PG_RETURN_NULL(); - /* Subscription conflict log tables are not published */ - result = is_publishable_class(relid, (Form_pg_class) GETSTRUCT(tuple)) && - !IsConflictLogTable(relid); + result = is_publishable_class(relid, (Form_pg_class) GETSTRUCT(tuple)); ReleaseSysCache(tuple); PG_RETURN_BOOL(result); } diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c index 285a598497d..1a93824504c 100644 --- a/src/backend/catalog/pg_subscription.c +++ b/src/backend/catalog/pg_subscription.c @@ -22,6 +22,7 @@ #include "catalog/pg_subscription.h" #include "catalog/pg_subscription_rel.h" #include "catalog/pg_type.h" +#include "commands/subscriptioncmds.h" #include "miscadmin.h" #include "storage/lmgr.h" #include "utils/array.h" @@ -156,6 +157,28 @@ GetSubscription(Oid subid, bool missing_ok) return sub; } +/* + * pg_relation_is_conflict_log_table + * + * Returns true if the given relation OID is used as a conflict log table + * by any subscription, else returns false. + */ +Datum +pg_relation_is_conflict_log_table(PG_FUNCTION_ARGS) +{ + Oid relid = PG_GETARG_OID(0); + HeapTuple tuple; + bool result; + + tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid)); + if (!HeapTupleIsValid(tuple)) + PG_RETURN_NULL(); + + result = IsConflictLogTable(relid); + ReleaseSysCache(tuple); + PG_RETURN_BOOL(result); +} + /* * Return number of subscriptions defined in given database. * Used by dropdb() to check if database can indeed be dropped. diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 4b2d98c9ed2..49c504960db 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -53,6 +53,7 @@ #include "storage/lmgr.h" #include "utils/acl.h" #include "utils/builtins.h" +#include "utils/fmgroids.h" #include "utils/guc.h" #include "utils/lsyscache.h" #include "utils/memutils.h" @@ -3443,27 +3444,22 @@ bool IsConflictLogTable(Oid relid) { Relation rel; - TableScanDesc scan; - HeapTuple tup; - bool is_clt = false; + ScanKeyData scankey; + SysScanDesc scan; + bool is_clt; rel = table_open(SubscriptionRelationId, AccessShareLock); - scan = table_beginscan_catalog(rel, 0, NULL); + ScanKeyInit(&scankey, + Anum_pg_subscription_subconflictlogrelid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(relid)); - while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection))) - { - Form_pg_subscription subform = (Form_pg_subscription) GETSTRUCT(tup); + scan = systable_beginscan(rel, SubscriptionConflictrelIndexId, + true, NULL, 1, &scankey); - /* Direct Oid comparison from catalog */ - if (OidIsValid(subform->subconflictlogrelid) && - subform->subconflictlogrelid == relid) - { - is_clt = true; - break; - } - } + is_clt = HeapTupleIsValid(systable_getnext(scan)); - table_endscan(scan); + systable_endscan(scan); table_close(rel, AccessShareLock); return is_clt; diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index e23ff0b70cf..6fce652dbcb 100644 --- a/src/backend/replication/logical/conflict.c +++ b/src/backend/replication/logical/conflict.c @@ -302,13 +302,11 @@ GetConflictLogTableInfo(ConflictLogDest *log_dest) void InsertConflictLogTuple(Relation conflictlogrel) { - int options = HEAP_INSERT_NO_LOGICAL; - /* A valid tuple must be prepared and stored in MyLogicalRepWorker. */ Assert(MyLogicalRepWorker->conflict_log_tuple != NULL); heap_insert(conflictlogrel, MyLogicalRepWorker->conflict_log_tuple, - GetCurrentCommandId(true), options, NULL); + GetCurrentCommandId(true), 0, NULL); /* Free conflict log tuple. */ heap_freetuple(MyLogicalRepWorker->conflict_log_tuple); diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 458418a249d..7ab8c942652 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -2098,6 +2098,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) bool am_partition = get_rel_relispartition(relid); char relkind = get_rel_relkind(relid); List *rel_publications = NIL; + bool isconflictlogrel; /* Reload publications if needed before use. */ if (!publications_valid) @@ -2176,6 +2177,14 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) entry->estate = NULL; memset(entry->exprstate, 0, sizeof(entry->exprstate)); + /* + * Check whether this table is a conflict log table. If so, avoid + * publishing it via FOR ALL TABLES or FOR TABLES IN SCHEMA + * publications. However, they may still be published if explicitly + * added to a FOR TABLE publication for this table. + */ + isconflictlogrel = IsConflictLogTable(relid); + /* * Build publication cache. We can't use one provided by relcache as * relcache considers all publications that the given relation is in, @@ -2199,7 +2208,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) * If this is a FOR ALL TABLES publication, pick the partition * root and set the ancestor level accordingly. */ - if (pub->alltables) + if (pub->alltables && !isconflictlogrel) { publish = true; if (pub->pubviaroot && am_partition) @@ -2225,8 +2234,12 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) { Oid ancestor; int level; - List *ancestors = get_partition_ancestors(relid); + List *ancestors; + + /* Conflict log table cannot be a partition */ + Assert(isconflictlogrel == false); + ancestors = get_partition_ancestors(relid); ancestor = GetTopMostAncestorInPublication(pub->oid, ancestors, &level); @@ -2243,7 +2256,8 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) } if (list_member_oid(pubids, pub->oid) || - list_member_oid(schemaPubids, pub->oid) || + (list_member_oid(schemaPubids, pub->oid) && + !isconflictlogrel) || ancestor_published) publish = true; } diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c index 20f08e548ba..230a68892ae 100644 --- a/src/bin/psql/describe.c +++ b/src/bin/psql/describe.c @@ -3063,6 +3063,7 @@ describeOneTableDetails(const char *schemaname, " JOIN pg_catalog.pg_publication_namespace pn ON p.oid = pn.pnpubid\n" " JOIN pg_catalog.pg_class pc ON pc.relnamespace = pn.pnnspid\n" "WHERE pc.oid ='%s' and pg_catalog.pg_relation_is_publishable('%s')\n" + "AND NOT pg_catalog.pg_relation_is_conflict_log_table('%s'::oid)\n" "UNION\n" "SELECT pubname\n" " , pg_get_expr(pr.prqual, c.oid)\n" @@ -3082,8 +3083,9 @@ describeOneTableDetails(const char *schemaname, " , NULL\n" "FROM pg_catalog.pg_publication p\n" "WHERE p.puballtables AND pg_catalog.pg_relation_is_publishable('%s')\n" + "AND NOT pg_catalog.pg_relation_is_conflict_log_table('%s'::oid)\n" "ORDER BY 1;", - oid, oid, oid, oid); + oid, oid, oid, oid, oid, oid); } else { diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 7f481687afe..d99f8500ac5 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -12328,6 +12328,13 @@ prorettype => 'bool', proargtypes => 'regclass', prosrc => 'pg_relation_is_publishable' }, +# subscriptions +{ oid => '6123', + descr => 'returns whether a relation is a subscription conflict log table', + proname => 'pg_relation_is_conflict_log_table', provolatile => 's', + prorettype => 'bool', proargtypes => 'regclass', + prosrc => 'pg_relation_is_conflict_log_table' }, + # rls { oid => '3298', descr => 'row security for current context active on table by table oid', diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h index 4aa29ea15d4..3d220b2db8a 100644 --- a/src/include/catalog/pg_subscription.h +++ b/src/include/catalog/pg_subscription.h @@ -123,6 +123,7 @@ DECLARE_TOAST_WITH_MACRO(pg_subscription, 4183, 4184, PgSubscriptionToastTable, DECLARE_UNIQUE_INDEX_PKEY(pg_subscription_oid_index, 6114, SubscriptionObjectIndexId, pg_subscription, btree(oid oid_ops)); DECLARE_UNIQUE_INDEX(pg_subscription_subname_index, 6115, SubscriptionNameIndexId, pg_subscription, btree(subdbid oid_ops, subname name_ops)); +DECLARE_INDEX(pg_subscription_conflictrel_index, 6122, SubscriptionConflictrelIndexId, pg_subscription, btree(subconflictlogrelid oid_ops)); MAKE_SYSCACHE(SUBSCRIPTIONOID, pg_subscription_oid_index, 4); MAKE_SYSCACHE(SUBSCRIPTIONNAME, pg_subscription_subname_index, 4); diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index d5f8abe9325..84abbaa5a4a 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -643,13 +643,14 @@ EXCEPTION WHEN insufficient_privilege THEN RAISE NOTICE 'captured expected error: insufficient_privilege'; END $$; NOTICE: captured expected error: insufficient_privilege --- PUBLICATION: Verify internal tables are not publishable --- pg_relation_is_publishable should return false for internal conflict log tables +-- PUBLICATION: Verify internal tables are publishable +-- pg_relation_is_publishable should return true for internal conflict log +-- tables, as it can be published using TABLE publication. SELECT pg_relation_is_publishable(subconflictlogrelid) FROM pg_subscription WHERE subname = 'regress_conflict_test1'; pg_relation_is_publishable ---------------------------- - f + t (1 row) -- CLEANUP: Proper drop reaps the table diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index 6c7f358ffd2..83befa8722c 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -452,8 +452,9 @@ EXCEPTION WHEN insufficient_privilege THEN RAISE NOTICE 'captured expected error: insufficient_privilege'; END $$; --- PUBLICATION: Verify internal tables are not publishable --- pg_relation_is_publishable should return false for internal conflict log tables +-- PUBLICATION: Verify internal tables are publishable +-- pg_relation_is_publishable should return true for internal conflict log +-- tables, as it can be published using TABLE publication. SELECT pg_relation_is_publishable(subconflictlogrelid) FROM pg_subscription WHERE subname = 'regress_conflict_test1'; -- 2.43.0