From 586af8e027932dfc31e1fa40f136b2bade6c5133 Mon Sep 17 00:00:00 2001 From: Dilip Kumar Date: Wed, 17 Dec 2025 11:53:47 +0530 Subject: [PATCH v22 1/3] 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_destination=('log'/'table'/'all') option in the CREATE SUBSCRIPTION command. If the user chooses to enable logging to a table (by selecting 'table' or 'all'), an internal logging table named conflict_log_table_ is automatically created within a dedicated, system-managed namespace named pg_conflict. This table is linked to the subscription via an internal dependency, ensuring it is automatically dropped when the subscription is removed. The conflict details, including the original and remote tuples, are stored in JSON columns, providing a flexible format to accommodate different table schemas. 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. --- src/backend/catalog/catalog.c | 27 +- src/backend/catalog/heap.c | 3 +- src/backend/catalog/namespace.c | 8 +- src/backend/catalog/pg_publication.c | 14 +- src/backend/catalog/pg_subscription.c | 7 + src/backend/commands/subscriptioncmds.c | 239 ++++++++++- src/backend/commands/tablecmds.c | 6 +- src/backend/executor/execMain.c | 18 + src/bin/psql/describe.c | 16 +- src/bin/psql/tab-complete.in.c | 8 +- src/include/catalog/catalog.h | 2 + src/include/catalog/pg_namespace.dat | 3 + src/include/catalog/pg_subscription.h | 11 + src/include/commands/subscriptioncmds.h | 3 + src/include/replication/conflict.h | 55 +++ src/test/regress/expected/subscription.out | 444 +++++++++++++++++---- src/test/regress/sql/subscription.sql | 217 ++++++++++ src/tools/pgindent/typedefs.list | 2 + 18 files changed, 981 insertions(+), 102 deletions(-) diff --git a/src/backend/catalog/catalog.c b/src/backend/catalog/catalog.c index 7be49032934..d438dc682ec 100644 --- a/src/backend/catalog/catalog.c +++ b/src/backend/catalog/catalog.c @@ -86,7 +86,8 @@ bool IsSystemClass(Oid relid, Form_pg_class reltuple) { /* IsCatalogRelationOid is a bit faster, so test that first */ - return (IsCatalogRelationOid(relid) || IsToastClass(reltuple)); + return (IsCatalogRelationOid(relid) || IsToastClass(reltuple) + || IsConflictClass(reltuple)); } /* @@ -230,6 +231,18 @@ IsToastClass(Form_pg_class reltuple) return IsToastNamespace(relnamespace); } +/* + * IsConflictClass - Check if the given pg_class tuple belongs to the conflict + * namespace. + */ +bool +IsConflictClass(Form_pg_class reltuple) +{ + Oid relnamespace = reltuple->relnamespace; + + return IsConflictNamespace(relnamespace); +} + /* * IsCatalogNamespace * True iff namespace is pg_catalog. @@ -264,6 +277,18 @@ IsToastNamespace(Oid namespaceId) isTempToastNamespace(namespaceId); } +/* + * IsConflictNamespace + * True iff namespace is pg_conflict. + * + * Does not perform any catalog accesses. + */ +bool +IsConflictNamespace(Oid namespaceId) +{ + return namespaceId == PG_CONFLICT_NAMESPACE; +} + /* * IsReservedName diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c index 606434823cf..10dadf378a4 100644 --- a/src/backend/catalog/heap.c +++ b/src/backend/catalog/heap.c @@ -314,7 +314,8 @@ heap_create(const char *relname, */ if (!allow_system_table_mods && ((IsCatalogNamespace(relnamespace) && relkind != RELKIND_INDEX) || - IsToastNamespace(relnamespace)) && + IsToastNamespace(relnamespace) || + IsConflictNamespace(relnamespace)) && IsNormalProcessingMode()) ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), diff --git a/src/backend/catalog/namespace.c b/src/backend/catalog/namespace.c index c3b79a2ba48..400292fd06b 100644 --- a/src/backend/catalog/namespace.c +++ b/src/backend/catalog/namespace.c @@ -3523,7 +3523,7 @@ LookupCreationNamespace(const char *nspname) * * We complain if either the old or new namespaces is a temporary schema * (or temporary toast schema), or if either the old or new namespaces is the - * TOAST schema. + * TOAST schema or the CONFLICT schema. */ void CheckSetNamespace(Oid oldNspOid, Oid nspOid) @@ -3539,6 +3539,12 @@ CheckSetNamespace(Oid oldNspOid, Oid nspOid) ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("cannot move objects into or out of TOAST schema"))); + + /* similarly for CONFLICT schema */ + if (nspOid == PG_CONFLICT_NAMESPACE || oldNspOid == PG_CONFLICT_NAMESPACE) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot move objects into or out of CONFLICT schema"))); } /* diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index 9a4791c573e..a33c33efe0d 100644 --- a/src/backend/catalog/pg_publication.c +++ b/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 (IsConflictNamespace(RelationGetNamespace(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."))); } /* @@ -95,7 +105,8 @@ static void check_publication_add_schema(Oid schemaid) { /* Can't be system namespace */ - if (IsCatalogNamespace(schemaid) || IsToastNamespace(schemaid)) + if (IsCatalogNamespace(schemaid) || IsToastNamespace(schemaid) || + IsConflictNamespace(schemaid)) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("cannot add schema \"%s\" to publication", @@ -139,6 +150,7 @@ is_publishable_class(Oid relid, Form_pg_class reltuple) reltuple->relkind == RELKIND_PARTITIONED_TABLE || reltuple->relkind == RELKIND_SEQUENCE) && !IsCatalogRelationOid(relid) && + !IsConflictClass(reltuple) && reltuple->relpersistence == RELPERSISTENCE_PERMANENT && relid >= FirstNormalObjectId; } diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c index 2b103245290..285a598497d 100644 --- a/src/backend/catalog/pg_subscription.c +++ b/src/backend/catalog/pg_subscription.c @@ -106,6 +106,7 @@ GetSubscription(Oid subid, bool missing_ok) sub->retaindeadtuples = subform->subretaindeadtuples; sub->maxretention = subform->submaxretention; sub->retentionactive = subform->subretentionactive; + sub->conflictlogrelid = subform->subconflictlogrelid; /* Get conninfo */ datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID, @@ -141,6 +142,12 @@ GetSubscription(Oid subid, bool missing_ok) Anum_pg_subscription_suborigin); sub->origin = TextDatumGetCString(datum); + /* Get conflict log destination */ + datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID, + tup, + Anum_pg_subscription_subconflictlogdest); + sub->conflictlogdest = TextDatumGetCString(datum); + /* Is the subscription owner a superuser? */ sub->ownersuperuser = superuser_arg(sub->owner); diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index d6674f20fc2..de1977f81ef 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -15,25 +15,31 @@ #include "postgres.h" #include "access/commit_ts.h" +#include "access/heapam.h" #include "access/htup_details.h" #include "access/table.h" #include "access/twophase.h" #include "access/xact.h" #include "catalog/catalog.h" #include "catalog/dependency.h" +#include "catalog/heap.h" #include "catalog/indexing.h" #include "catalog/namespace.h" #include "catalog/objectaccess.h" #include "catalog/objectaddress.h" +#include "catalog/pg_am_d.h" #include "catalog/pg_authid_d.h" #include "catalog/pg_database_d.h" +#include "catalog/pg_namespace.h" #include "catalog/pg_subscription.h" #include "catalog/pg_subscription_rel.h" #include "catalog/pg_type.h" +#include "commands/comment.h" #include "commands/defrem.h" #include "commands/event_trigger.h" #include "commands/subscriptioncmds.h" #include "executor/executor.h" +#include "funcapi.h" #include "miscadmin.h" #include "nodes/makefuncs.h" #include "pgstat.h" @@ -51,6 +57,7 @@ #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/pg_lsn.h" +#include "utils/regproc.h" #include "utils/syscache.h" /* @@ -75,6 +82,7 @@ #define SUBOPT_MAX_RETENTION_DURATION 0x00008000 #define SUBOPT_LSN 0x00010000 #define SUBOPT_ORIGIN 0x00020000 +#define SUBOPT_CONFLICT_LOG_DEST 0x00040000 /* check if the 'val' has 'bits' set */ #define IsSet(val, bits) (((val) & (bits)) == (bits)) @@ -103,6 +111,7 @@ typedef struct SubOpts bool retaindeadtuples; int32 maxretention; char *origin; + ConflictLogDest conflictlogdest; XLogRecPtr lsn; } SubOpts; @@ -135,7 +144,7 @@ static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err); static void CheckAlterSubOption(Subscription *sub, const char *option, bool slot_needs_update, bool isTopLevel); - +static Oid create_conflict_log_table(Oid subid, char *subname); /* * Common option parsing function for CREATE and ALTER SUBSCRIPTION commands. @@ -191,6 +200,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->maxretention = 0; if (IsSet(supported_opts, SUBOPT_ORIGIN)) opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY); + if (IsSet(supported_opts, SUBOPT_CONFLICT_LOG_DEST)) + opts->conflictlogdest = CONFLICT_LOG_DEST_LOG; /* Parse options */ foreach(lc, stmt_options) @@ -402,6 +413,18 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->specified_opts |= SUBOPT_LSN; opts->lsn = lsn; } + else if (IsSet(supported_opts, SUBOPT_CONFLICT_LOG_DEST) && + strcmp(defel->defname, "conflict_log_destination") == 0) + { + char *val; + + if (IsSet(opts->specified_opts, SUBOPT_CONFLICT_LOG_DEST)) + errorConflictingDefElem(defel, pstate); + + val = defGetString(defel); + opts->conflictlogdest = GetLogDestination(val); + opts->specified_opts |= SUBOPT_CONFLICT_LOG_DEST; + } else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -599,6 +622,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, bits32 supported_opts; SubOpts opts = {0}; AclResult aclresult; + Oid logrelid = InvalidOid; /* * Parse and check options. @@ -612,7 +636,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED | SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_RETAIN_DEAD_TUPLES | - SUBOPT_MAX_RETENTION_DURATION | SUBOPT_ORIGIN); + SUBOPT_MAX_RETENTION_DURATION | SUBOPT_ORIGIN | + SUBOPT_CONFLICT_LOG_DEST); parse_subscription_options(pstate, stmt->options, supported_opts, &opts); /* @@ -747,6 +772,18 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, values[Anum_pg_subscription_suborigin - 1] = CStringGetTextDatum(opts.origin); + /* Always set the destination, default will be 'log'. */ + values[Anum_pg_subscription_subconflictlogdest - 1] = + CStringGetTextDatum(ConflictLogDestNames[opts.conflictlogdest]); + + /* If logging to a table is required, physically create the table. */ + if (IsSet(opts.conflictlogdest, CONFLICT_LOG_DEST_TABLE)) + logrelid = create_conflict_log_table(subid, stmt->subname); + + /* Store table OID in the catalog. */ + values[Anum_pg_subscription_subconflictlogrelid - 1] = + ObjectIdGetDatum(logrelid); + tup = heap_form_tuple(RelationGetDescr(rel), values, nulls); /* Insert tuple into catalog. */ @@ -1410,7 +1447,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_RETAIN_DEAD_TUPLES | SUBOPT_MAX_RETENTION_DURATION | - SUBOPT_ORIGIN); + SUBOPT_ORIGIN | + SUBOPT_CONFLICT_LOG_DEST); parse_subscription_options(pstate, stmt->options, supported_opts, &opts); @@ -1665,6 +1703,63 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, origin = opts.origin; } + if (IsSet(opts.specified_opts, SUBOPT_CONFLICT_LOG_DEST)) + { + ConflictLogDest old_dest = + GetLogDestination(sub->conflictlogdest); + + if (opts.conflictlogdest != old_dest) + { + bool want_table = IsSet(opts.conflictlogdest, + CONFLICT_LOG_DEST_TABLE); + bool has_oldtable = + IsSet(old_dest, CONFLICT_LOG_DEST_TABLE); + + values[Anum_pg_subscription_subconflictlogdest - 1] = + CStringGetTextDatum(ConflictLogDestNames[opts.conflictlogdest]); + replaces[Anum_pg_subscription_subconflictlogdest - 1] = true; + + if (want_table && !has_oldtable) + { + Oid relid; + + relid = create_conflict_log_table(subid, sub->name); + + values[Anum_pg_subscription_subconflictlogrelid - 1] = + ObjectIdGetDatum(relid); + replaces[Anum_pg_subscription_subconflictlogrelid - 1] = + true; + } + else if (!want_table && has_oldtable) + { + ObjectAddress object; + + /* + * Conflict log tables are recorded as internal + * dependencies of the subscription. Drop the + * table if it is not required anymore to avoid + * stale or orphaned relations. + * + * XXX: At present, only conflict log tables are + * managed this way. In future if we introduce + * additional internal dependencies, we may need + * a targeted deletion to avoid deletion of any + * other objects. + */ + ObjectAddressSet(object, SubscriptionRelationId, + subid); + performDeletion(&object, DROP_CASCADE, + PERFORM_DELETION_INTERNAL | + PERFORM_DELETION_SKIP_ORIGINAL); + + values[Anum_pg_subscription_subconflictlogrelid - 1] = + ObjectIdGetDatum(InvalidOid); + replaces[Anum_pg_subscription_subconflictlogrelid - 1] = + true; + } + } + } + update_tuple = true; break; } @@ -2027,6 +2122,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) Form_pg_subscription form; List *rstates; bool must_use_password; + ObjectAddress object; /* * The launcher may concurrently start a new worker for this subscription. @@ -2184,6 +2280,19 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) /* Clean up dependencies */ deleteSharedDependencyRecordsFor(SubscriptionRelationId, subid, 0); + /* + * Conflict log tables are recorded as internal dependencies of the + * subscription. We must drop the dependent objects before the + * subscription itself is removed. By using + * PERFORM_DELETION_SKIP_ORIGINAL, we ensure that only the conflict log + * table is reaped while the subscription remains for the final deletion + * step. + */ + ObjectAddressSet(object, SubscriptionRelationId, subid); + performDeletion(&object, DROP_CASCADE, + PERFORM_DELETION_INTERNAL | + PERFORM_DELETION_SKIP_ORIGINAL); + /* Remove any associated relation synchronization states. */ RemoveSubscriptionRel(subid, InvalidOid); @@ -3190,3 +3299,127 @@ defGetStreamingMode(DefElem *def) def->defname))); return LOGICALREP_STREAM_OFF; /* keep compiler quiet */ } + +/* + * Builds the TupleDesc for the conflict log table. + */ +static TupleDesc +create_conflict_log_table_tupdesc(void) +{ + TupleDesc tupdesc; + + tupdesc = CreateTemplateTupleDesc(MAX_CONFLICT_ATTR_NUM); + + for (int i = 0; i < MAX_CONFLICT_ATTR_NUM; i++) + { + Oid type_oid = ConflictLogSchema[i].atttypid; + + /* + * Special handling for the JSON array type for proper + * TupleDescInitEntry call. + */ + if (type_oid == JSONARRAYOID) + type_oid = get_array_type(JSONOID); + + TupleDescInitEntry(tupdesc, i + 1, + ConflictLogSchema[i].attname, + type_oid, + -1, 0); + } + + return BlessTupleDesc(tupdesc); +} + +/* + * Create a structured conflict log table for a subscription. + * + * The table is created within the system-managed 'pg_conflict' namespace. + * The table name is generated automatically using the subscription's OID + * (e.g., "pg_conflict_") to ensure uniqueness within the cluster and + * to avoid collisions during subscription renames. + */ +static Oid +create_conflict_log_table(Oid subid, char *subname) +{ + TupleDesc tupdesc; + Oid relid; + ObjectAddress myself; + ObjectAddress subaddr; + char relname[NAMEDATALEN]; + + snprintf(relname, NAMEDATALEN, "pg_conflict_%u", subid); + + /* There can not be an existing table with the same name. */ + Assert(!OidIsValid(get_relname_relid(relname, PG_CONFLICT_NAMESPACE))); + + /* Build the tuple descriptor for the new table. */ + tupdesc = create_conflict_log_table_tupdesc(); + + /* Create conflict log table. */ + relid = heap_create_with_catalog(relname, + PG_CONFLICT_NAMESPACE, + 0, /* tablespace */ + InvalidOid, /* relid */ + InvalidOid, /* reltypeid */ + InvalidOid, /* reloftypeid */ + GetUserId(), + HEAP_TABLE_AM_OID, + tupdesc, + NIL, + RELKIND_RELATION, + RELPERSISTENCE_PERMANENT, + false, /* shared_relation */ + false, /* mapped_relation */ + ONCOMMIT_NOOP, + (Datum) 0, /* reloptions */ + false, /* use_user_acl */ + true, /* allow_system_table_mods */ + true, /* is_internal */ + InvalidOid, /* relrewrite */ + NULL); /* typaddress */ + + /* + * Establish an internal dependency between the conflict log table and + * the subscription. + * + * We use DEPENDENCY_INTERNAL to signify that the table's lifecycle is + * strictly tied to the subscription, similar to how a TOAST table relates + * to its main table or a sequence relates to an identity column. + * + * This ensures the conflict log table is automatically reaped during a + * DROP SUBSCRIPTION via performDeletion(). + */ + ObjectAddressSet(myself, RelationRelationId, relid); + ObjectAddressSet(subaddr, SubscriptionRelationId, subid); + recordDependencyOn(&myself, &subaddr, DEPENDENCY_INTERNAL); + + /* Release tuple descriptor memory. */ + FreeTupleDesc(tupdesc); + + return relid; +} + +/* + * GetLogDestination + * + * Convert string to enum by comparing against standardized labels. + */ +ConflictLogDest +GetLogDestination(const char *dest) +{ + /* Empty string or NULL defaults to LOG. */ + if (dest == NULL || dest[0] == '\0' || pg_strcasecmp(dest, "log") == 0) + return CONFLICT_LOG_DEST_LOG; + + if (pg_strcasecmp(dest, "table") == 0) + return CONFLICT_LOG_DEST_TABLE; + + if (pg_strcasecmp(dest, "all") == 0) + return CONFLICT_LOG_DEST_ALL; + + /* Unrecognized string. */ + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("unrecognized conflict_log_destination value: \"%s\"", dest), + errhint("Valid values are \"log\", \"table\", and \"all\"."))); +} diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index f976c0e5c7e..4bf39f359d7 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -2398,9 +2398,11 @@ truncate_check_rel(Oid relid, Form_pg_class reltuple) * pg_largeobject and pg_largeobject_metadata to be truncated as part of * pg_upgrade, because we need to change its relfilenode to match the old * cluster, and allowing a TRUNCATE command to be executed is the easiest - * way of doing that. + * way of doing that. We also allow TRUNCATE on the conflict log tables, + * to permit users to manually prune these logs to manage disk space. */ - if (!allowSystemTableMods && IsSystemClass(relid, reltuple) + if (!allowSystemTableMods && IsSystemClass(relid, reltuple) && + !IsConflictClass(reltuple) && (!IsBinaryUpgrade || (relid != LargeObjectRelationId && relid != LargeObjectMetadataRelationId))) diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index ca14cdabdd0..ff67a594d39 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -1166,6 +1166,24 @@ CheckValidResultRel(ResultRelInfo *resultRelInfo, CmdType operation, RelationGetRelationName(resultRel)))); break; } + + /* + * Conflict logging tables (CLT) are managed by the system to record + * replication conflicts. We allow DELETE to permit users to manually prune + * or truncate these logs, but manual data insertion or modification + * (INSERT, UPDATE, MERGE) is prohibited to maintain the integrity of the + * system-generated logs. + */ + if (IsConflictNamespace(RelationGetNamespace(resultRel)) && + operation != CMD_DELETE) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("cannot execute %s on conflict logging table \"%s\"", + (operation == CMD_INSERT ? "INSERT" : + operation == CMD_UPDATE ? "UPDATE" : + operation == CMD_MERGE ? "MERGE" : "this operation"), + RelationGetRelationName(resultRel)), + errdetail("Conflict logging tables are system-managed and only support cleanup via DELETE or TRUNCATE."))); } /* diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c index 3584c4e1428..7262541bcbd 100644 --- a/src/bin/psql/describe.c +++ b/src/bin/psql/describe.c @@ -6806,7 +6806,7 @@ describeSubscriptions(const char *pattern, bool verbose) printQueryOpt myopt = pset.popt; static const bool translate_columns[] = {false, false, false, false, false, false, false, false, false, false, false, false, false, false, - false, false, false, false}; + false, false, false, false, false, false}; if (pset.sversion < 100000) { @@ -6900,6 +6900,20 @@ describeSubscriptions(const char *pattern, bool verbose) appendPQExpBuffer(&buf, ", subskiplsn AS \"%s\"\n", gettext_noop("Skip LSN")); + + /* Conflict log destination is supported in v19 and higher */ + if (pset.sversion >= 190000) + { + appendPQExpBuffer(&buf, + ", subconflictlogdest AS \"%s\"\n", + gettext_noop("Conflict log destination")); + + appendPQExpBuffer(&buf, + ", (CASE WHEN subconflictlogdest IN ('table', 'all') " + " THEN 'pg_conflict.pg_conflict_' || oid " + " ELSE '-' END) AS \"%s\"\n", + gettext_noop("Conflict log table")); + } } /* Only display subscriptions in current database. */ diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c index 8b91bc00062..12eee8a0d43 100644 --- a/src/bin/psql/tab-complete.in.c +++ b/src/bin/psql/tab-complete.in.c @@ -2344,8 +2344,8 @@ match_previous_words(int pattern_id, COMPLETE_WITH("(", "PUBLICATION"); /* ALTER SUBSCRIPTION SET ( */ else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "SET", "(")) - COMPLETE_WITH("binary", "disable_on_error", "failover", - "max_retention_duration", "origin", + COMPLETE_WITH("binary", "conflict_log_destination", "disable_on_error", + "failover", "max_retention_duration", "origin", "password_required", "retain_dead_tuples", "run_as_owner", "slot_name", "streaming", "synchronous_commit", "two_phase"); @@ -3860,8 +3860,8 @@ match_previous_words(int pattern_id, COMPLETE_WITH("WITH ("); /* Complete "CREATE SUBSCRIPTION ... WITH ( " */ else if (Matches("CREATE", "SUBSCRIPTION", MatchAnyN, "WITH", "(")) - COMPLETE_WITH("binary", "connect", "copy_data", "create_slot", - "disable_on_error", "enabled", "failover", + COMPLETE_WITH("binary", "conflict_log_destination", "connect", "copy_data", + "create_slot", "disable_on_error", "enabled", "failover", "max_retention_duration", "origin", "password_required", "retain_dead_tuples", "run_as_owner", "slot_name", "streaming", diff --git a/src/include/catalog/catalog.h b/src/include/catalog/catalog.h index a9d6e8ea986..8193229f2e2 100644 --- a/src/include/catalog/catalog.h +++ b/src/include/catalog/catalog.h @@ -25,6 +25,7 @@ extern bool IsInplaceUpdateRelation(Relation relation); extern bool IsSystemClass(Oid relid, Form_pg_class reltuple); extern bool IsToastClass(Form_pg_class reltuple); +extern bool IsConflictClass(Form_pg_class reltuple); extern bool IsCatalogRelationOid(Oid relid); extern bool IsCatalogTextUniqueIndexOid(Oid relid); @@ -32,6 +33,7 @@ extern bool IsInplaceUpdateOid(Oid relid); extern bool IsCatalogNamespace(Oid namespaceId); extern bool IsToastNamespace(Oid namespaceId); +extern bool IsConflictNamespace(Oid namespaceId); extern bool IsReservedName(const char *name); diff --git a/src/include/catalog/pg_namespace.dat b/src/include/catalog/pg_namespace.dat index 3075e142c73..b45cb9383a8 100644 --- a/src/include/catalog/pg_namespace.dat +++ b/src/include/catalog/pg_namespace.dat @@ -18,6 +18,9 @@ { oid => '99', oid_symbol => 'PG_TOAST_NAMESPACE', descr => 'reserved schema for TOAST tables', nspname => 'pg_toast', nspacl => '_null_' }, +{ oid => '1382', oid_symbol => 'PG_CONFLICT_NAMESPACE', + descr => 'reserved schema for subscription-specific conflict log tables', + nspname => 'pg_conflict', nspacl => '_null_' }, # update dumpNamespace() if changing this descr { oid => '2200', oid_symbol => 'PG_PUBLIC_NAMESPACE', descr => 'standard public schema', diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h index f3571d2bfcf..76a4638b389 100644 --- a/src/include/catalog/pg_subscription.h +++ b/src/include/catalog/pg_subscription.h @@ -90,6 +90,7 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW * exceeded max_retention_duration, when * defined */ + Oid subconflictlogrelid; /* Relid of the conflict log table. */ #ifdef CATALOG_VARLEN /* variable-length fields start here */ /* Connection string to the publisher */ text subconninfo BKI_FORCE_NOT_NULL; @@ -103,6 +104,14 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW /* List of publications subscribed to */ text subpublications[1] BKI_FORCE_NOT_NULL; + /* + * Strategy for logging replication conflicts: + * 'log' - server log only, + * 'table' - conflict log table only, + * 'all' - both log and table. + */ + text subconflictlogdest; + /* Only publish data originating from the specified origin */ text suborigin BKI_DEFAULT(LOGICALREP_ORIGIN_ANY); #endif @@ -152,12 +161,14 @@ typedef struct Subscription * and the retention duration has not * exceeded max_retention_duration, when * defined */ + Oid conflictlogrelid; /* conflict log table Oid */ char *conninfo; /* Connection string to the publisher */ char *slotname; /* Name of the replication slot */ char *synccommit; /* Synchronous commit setting for worker */ List *publications; /* List of publication names to subscribe to */ char *origin; /* Only publish data originating from the * specified origin */ + char *conflictlogdest; /* Conflict log destination */ } Subscription; #ifdef EXPOSE_TO_CLIENT_CODE diff --git a/src/include/commands/subscriptioncmds.h b/src/include/commands/subscriptioncmds.h index 63504232a14..a895127f8fe 100644 --- a/src/include/commands/subscriptioncmds.h +++ b/src/include/commands/subscriptioncmds.h @@ -17,6 +17,7 @@ #include "catalog/objectaddress.h" #include "parser/parse_node.h" +#include "replication/conflict.h" extern ObjectAddress CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, bool isTopLevel); @@ -36,4 +37,6 @@ extern void CheckSubDeadTupleRetention(bool check_guc, bool sub_disabled, bool retention_active, bool max_retention_set); +extern ConflictLogDest GetLogDestination(const char *dest); + #endif /* SUBSCRIPTIONCMDS_H */ diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h index d538274637f..4e4f59bb453 100644 --- a/src/include/replication/conflict.h +++ b/src/include/replication/conflict.h @@ -10,6 +10,7 @@ #define CONFLICT_H #include "access/xlogdefs.h" +#include "catalog/pg_type.h" #include "nodes/pg_list.h" #include "utils/timestamp.h" @@ -79,6 +80,60 @@ typedef struct ConflictTupleInfo * conflicting local row occurred */ } ConflictTupleInfo; +/* + * Conflict log destination types. + * + * These values are defined as bitmask flags to allow for multiple simultaneous + * logging destinations (e.g., logging to both system logs and a table). + * Internally, we use these for bitwise comparisons (IsSet), but the string + * representation is stored in pg_subscription.subconflictlogdest. + */ +typedef enum ConflictLogDest +{ + /* Log conflicts to the server logs */ + CONFLICT_LOG_DEST_LOG = 1 << 0, /* 0x01 */ + + /* Log conflicts to an internally managed conflict log table */ + CONFLICT_LOG_DEST_TABLE = 1 << 1, /* 0x02 */ + + /* Convenience bitmask for all supported destinations */ + CONFLICT_LOG_DEST_ALL = (CONFLICT_LOG_DEST_LOG | CONFLICT_LOG_DEST_TABLE) +} ConflictLogDest; + +/* + * Array mapping for converting internal enum to string. + */ +static const char *const ConflictLogDestNames[] = { + [CONFLICT_LOG_DEST_LOG] = "log", + [CONFLICT_LOG_DEST_TABLE] = "table", + [CONFLICT_LOG_DEST_ALL] = "all" +}; + +/* Structure to hold metadata for one column of the conflict log table */ +typedef struct ConflictLogColumnDef +{ + const char *attname; /* Column name */ + Oid atttypid; /* Data type OID */ +} ConflictLogColumnDef; + +/* The single source of truth for the conflict log table schema */ +static const ConflictLogColumnDef ConflictLogSchema[] = +{ + { .attname = "relid", .atttypid = OIDOID }, + { .attname = "schemaname", .atttypid = TEXTOID }, + { .attname = "relname", .atttypid = TEXTOID }, + { .attname = "conflict_type", .atttypid = TEXTOID }, + { .attname = "remote_xid", .atttypid = XIDOID }, + { .attname = "remote_commit_lsn",.atttypid = LSNOID }, + { .attname = "remote_commit_ts", .atttypid = TIMESTAMPTZOID }, + { .attname = "remote_origin", .atttypid = TEXTOID }, + { .attname = "replica_identity", .atttypid = JSONOID }, + { .attname = "remote_tuple", .atttypid = JSONOID }, + { .attname = "local_conflicts", .atttypid = JSONARRAYOID } +}; + +#define MAX_CONFLICT_ATTR_NUM lengthof(ConflictLogSchema) + extern bool GetTupleTransactionInfo(TupleTableSlot *localslot, TransactionId *xmin, RepOriginId *localorigin, diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index b3eccd8afe3..b94bcc3cc23 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -116,18 +116,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ regress_testsub4 - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN -------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) ALTER SUBSCRIPTION regress_testsub4 SET (origin = any); \dRs+ regress_testsub4 - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN -------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) DROP SUBSCRIPTION regress_testsub3; @@ -145,10 +145,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar'; ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false); @@ -157,10 +157,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname'); ALTER SUBSCRIPTION regress_testsub SET (password_required = false); ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+------------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | f | 0 | f | off | dbname=regress_doesnotexist2 | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+------------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | f | 0 | f | off | dbname=regress_doesnotexist2 | 0/00000000 | log | - (1 row) ALTER SUBSCRIPTION regress_testsub SET (password_required = true); @@ -176,10 +176,10 @@ ERROR: unrecognized subscription parameter: "create_slot" -- ok ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345'); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+------------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist2 | 0/00012345 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+------------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist2 | 0/00012345 | log | - (1 row) -- ok - with lsn = NONE @@ -188,10 +188,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE); ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0'); ERROR: invalid WAL location (LSN): 0/0 \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+------------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist2 | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+------------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist2 | 0/00000000 | log | - (1 row) BEGIN; @@ -223,10 +223,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar); ERROR: invalid value for parameter "synchronous_commit": "foobar" HINT: Available values: local, remote_write, remote_apply, on, off. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+------------------------------+------------ - regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | f | 0 | f | local | dbname=regress_doesnotexist2 | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+------------------------------+------------+--------------------------+-------------------- + regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | f | 0 | f | local | dbname=regress_doesnotexist2 | 0/00000000 | log | - (1 row) -- rename back to keep the rest simple @@ -255,19 +255,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) ALTER SUBSCRIPTION regress_testsub SET (binary = false); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) DROP SUBSCRIPTION regress_testsub; @@ -279,27 +279,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) ALTER SUBSCRIPTION regress_testsub SET (streaming = false); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) -- fail - publication already exists @@ -314,10 +314,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false); ERROR: publication "testpub1" is already in subscription "regress_testsub" \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) -- fail - publication used more than once @@ -332,10 +332,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub" -- ok - delete publications ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) DROP SUBSCRIPTION regress_testsub; @@ -371,19 +371,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) -- we can alter streaming when two_phase enabled ALTER SUBSCRIPTION regress_testsub SET (streaming = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -393,10 +393,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -409,18 +409,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -433,10 +433,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -450,19 +450,19 @@ NOTICE: max_retention_duration is ineffective when retain_dead_tuples is disabl WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 1000 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 1000 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) -- ok ALTER SUBSCRIPTION regress_testsub SET (max_retention_duration = 0); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------ - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Skip LSN | Conflict log destination | Conflict log table +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------+--------------------------+-------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | f | off | dbname=regress_doesnotexist | 0/00000000 | log | - (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -517,6 +517,274 @@ COMMIT; -- ok, owning it is enough for this stuff ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); DROP SUBSCRIPTION regress_testsub; +-- +-- CONFLICT LOG DESTINATION TESTS +-- +SET SESSION AUTHORIZATION 'regress_subscription_user'; +-- fail - unrecognized parameter value +CREATE SUBSCRIPTION regress_conflict_fail CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'invalid'); +ERROR: unrecognized conflict_log_destination value: "invalid" +HINT: Valid values are "log", "table", and "all". +-- verify subconflictlogdest is 'log' and relid is 0 (InvalidOid) for default case +CREATE SUBSCRIPTION regress_conflict_log_default CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false); +WARNING: subscription was created, but is not connected +HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. +SELECT subname, subconflictlogdest, subconflictlogrelid +FROM pg_subscription WHERE subname = 'regress_conflict_log_default'; + subname | subconflictlogdest | subconflictlogrelid +------------------------------+--------------------+--------------------- + regress_conflict_log_default | log | 0 +(1 row) + +-- verify empty string defaults to 'log' +CREATE SUBSCRIPTION regress_conflict_empty_str CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = ''); +WARNING: subscription was created, but is not connected +HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. +SELECT subname, subconflictlogdest, subconflictlogrelid +FROM pg_subscription WHERE subname = 'regress_conflict_empty_str'; + subname | subconflictlogdest | subconflictlogrelid +----------------------------+--------------------+--------------------- + regress_conflict_empty_str | log | 0 +(1 row) + +-- this should generate an internal conflict log table named pg_conflict_$subid$ +CREATE SUBSCRIPTION regress_conflict_test1 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'table'); +WARNING: subscription was created, but is not connected +HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. +-- check metadata in pg_subscription: destination should be 'table' and relid valid +SELECT subname, subconflictlogdest, subconflictlogrelid > 0 AS has_relid +FROM pg_subscription WHERE subname = 'regress_conflict_test1'; + subname | subconflictlogdest | has_relid +------------------------+--------------------+----------- + regress_conflict_test1 | table | t +(1 row) + +-- verify the physical table exists, its OID matches subconflictlogrelid, +-- and it is located in the 'pg_conflict' namespace +SELECT n.nspname, (c.oid = s.subconflictlogrelid) AS "oid_matches" +FROM pg_class c +JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid +JOIN pg_namespace n ON c.relnamespace = n.oid +WHERE s.subname = 'regress_conflict_test1'; + nspname | oid_matches +-------------+------------- + pg_conflict | t +(1 row) + +-- check if the conflict log table has the correct schema +SELECT a.attnum, a.attname +FROM pg_attribute a +JOIN pg_class c ON a.attrelid = c.oid +JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid +WHERE s.subname = 'regress_conflict_test1' AND a.attnum > 0 + ORDER BY a.attnum; + attnum | attname +--------+------------------- + 1 | relid + 2 | schemaname + 3 | relname + 4 | conflict_type + 5 | remote_xid + 6 | remote_commit_lsn + 7 | remote_commit_ts + 8 | remote_origin + 9 | replica_identity + 10 | remote_tuple + 11 | local_conflicts +(11 rows) + +-- +-- ALTER SUBSCRIPTION - conflict_log_destination state transitions +-- +-- These tests verify the transition logic between different logging +-- destinations, ensuring conflict log tables are created or dropped as +-- expected +-- +-- transition from 'log' to 'all' +-- a new internal conflict log table should be created +CREATE SUBSCRIPTION regress_conflict_test2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'log'); +WARNING: subscription was created, but is not connected +HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. +ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'all'); +-- verify metadata after ALTER (destination should be 'all') +SELECT subname, subconflictlogdest, subconflictlogrelid > 0 AS has_relid +FROM pg_subscription WHERE subname = 'regress_conflict_test2'; + subname | subconflictlogdest | has_relid +------------------------+--------------------+----------- + regress_conflict_test2 | all | t +(1 row) + +-- transition from 'all' to 'table' +-- should NOT drop the table, only change destination string +SELECT subconflictlogrelid AS old_relid FROM pg_subscription WHERE subname = 'regress_conflict_test2' \gset +ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'table'); +SELECT subconflictlogdest, subconflictlogrelid = :old_relid AS relid_unchanged +FROM pg_subscription WHERE subname = 'regress_conflict_test2'; + subconflictlogdest | relid_unchanged +--------------------+----------------- + table | t +(1 row) + +-- transition from 'table' to 'log' +-- should drop the table and clear relid +ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'log'); +SELECT subconflictlogdest, subconflictlogrelid +FROM pg_subscription WHERE subname = 'regress_conflict_test2'; + subconflictlogdest | subconflictlogrelid +--------------------+--------------------- + log | 0 +(1 row) + +-- verify the physical table is gone +SELECT count(*) +FROM pg_class c +JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid +WHERE s.subname = 'regress_conflict_test2'; + count +------- + 0 +(1 row) + +-- +-- PUBLICATION: Verify conflict log tables are not publishable +-- +-- pg_relation_is_publishable should return false for internal conflict log +-- tables to prevent them from being accidentally included in publications +-- +SELECT n.nspname, pg_relation_is_publishable(c.oid) +FROM pg_class c +JOIN pg_namespace n ON c.relnamespace = n.oid +JOIN pg_subscription s ON s.subconflictlogrelid = c.oid +WHERE s.subname = 'regress_conflict_test1'; + nspname | pg_relation_is_publishable +-------------+---------------------------- + pg_conflict | f +(1 row) + +-- +-- Table Protection and Lifecycle Management +-- +-- These tests verify that: +-- Manual DROP TABLE is disallowed +-- DROP SUBSCRIPTION automatically reaps the table +-- +-- re-enable table logging for verification +ALTER SUBSCRIPTION regress_conflict_test1 SET (conflict_log_destination = 'table'); +-- We use a DO block with dynamic SQL because the internal conflict log table +-- name contains the subscription OID, which is non-deterministic. This +-- approach allows us to attempt the DROP and capture the expected error +-- without hard-coding a specific OID in the expected output +-- fail - drop table not allowed due to internal dependency +DO $$ +BEGIN + EXECUTE 'DROP TABLE ' || (SELECT 'pg_conflict.pg_conflict_' || oid FROM pg_subscription WHERE subname = 'regress_conflict_test1'); +EXCEPTION WHEN insufficient_privilege THEN + RAISE NOTICE 'captured expected error: insufficient_privilege'; +END $$; +NOTICE: captured expected error: insufficient_privilege +-- CLEANUP: DROP SUBSCRIPTION reaps the table +ALTER SUBSCRIPTION regress_conflict_test1 DISABLE; +ALTER SUBSCRIPTION regress_conflict_test1 SET (slot_name = NONE); +-- Verify the table OID for reap check +SELECT 'pg_conflict_' || oid AS internal_tablename FROM pg_subscription WHERE subname = 'regress_conflict_test1' \gset +DROP SUBSCRIPTION regress_conflict_test1; +-- should return NULL, meaning the conflict log table was reaped via dependency +SELECT to_regclass(:'internal_tablename'); + to_regclass +------------- + +(1 row) + +-- +-- Additional Namespace and Table Protection Tests +-- +-- Setup: Ensure we have a subscription with a conflict log table +CREATE SUBSCRIPTION regress_conflict_protection_test CONNECTION 'dbname=regress_doesnotexist' + PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'table'); +WARNING: subscription was created, but is not connected +HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. +-- Trying to ALTER the internal conflict log table +-- This should fail because the table is system-managed +-- As mentioned in previous test cases, we use a DO block to hide dynamic OIDs +DO $$ +DECLARE + tab_name text; +BEGIN + SELECT 'pg_conflict.' || relname INTO tab_name + FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid + WHERE s.subname = 'regress_conflict_protection_test'; + + RAISE NOTICE 'Attempting ALTER TABLE on internal conflict log table'; + EXECUTE 'ALTER TABLE ' || tab_name || ' ADD COLUMN extra_info text'; +EXCEPTION WHEN insufficient_privilege THEN + RAISE NOTICE 'captured expected error: insufficient_privilege during ALTER'; +END $$; +NOTICE: Attempting ALTER TABLE on internal conflict log table +NOTICE: captured expected error: insufficient_privilege during ALTER +-- Test Manual INSERT on conflict log table +-- This should fail because the table is system-managed +-- Hiding the OID in the error message by catching the exception +DO $$ +DECLARE + tab_name text; +BEGIN + SELECT 'pg_conflict.' || relname INTO tab_name + FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid + WHERE s.subname = 'regress_conflict_protection_test'; + + EXECUTE 'INSERT INTO ' || tab_name || ' (relname) VALUES (''mytest'')'; +EXCEPTION WHEN insufficient_privilege THEN + RAISE NOTICE 'captured expected error: insufficient_privilege during INSERT'; +END $$; +NOTICE: captured expected error: insufficient_privilege during INSERT +-- Test Manual UPDATE on conflict log table +-- This should fail because the table is system-managed +-- Hiding the OID in the error message by catching the exception +DO $$ +DECLARE + tab_name text; +BEGIN + SELECT 'pg_conflict.' || relname INTO tab_name + FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid + WHERE s.subname = 'regress_conflict_protection_test'; + + EXECUTE 'UPDATE ' || tab_name || ' SET relname = ''mytest'' '; +EXCEPTION WHEN insufficient_privilege THEN + RAISE NOTICE 'captured expected error: insufficient_privilege during UPDATE'; +END $$; +NOTICE: captured expected error: insufficient_privilege during UPDATE +-- Trying to perform TRUNCATE/DELETE on the internal conflict log table +-- This should be allowed so that user can perform cleanup +SELECT 'pg_conflict.' || relname AS conflict_tab +FROM pg_class c +JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid +WHERE s.subname = 'regress_conflict_protection_test' \gset +TRUNCATE :conflict_tab; +DELETE FROM :conflict_tab; +-- Trying to create a new table manually in the pg_conflict namespace +-- This should fail as the namespace is reserved for conflict logs tables +CREATE TABLE pg_conflict.manual_table (id int); +ERROR: permission denied to create "pg_conflict.manual_table" +DETAIL: System catalog modifications are currently disallowed. +-- Moving an existing table into the pg_conflict namespace +-- Users should not be able to move their own tables within this namespace +CREATE TABLE public.test_move (id int); +ALTER TABLE public.test_move SET SCHEMA pg_conflict; +ERROR: cannot move objects into or out of CONFLICT schema +DROP TABLE public.test_move; +-- Clean up remaining test subscription +ALTER SUBSCRIPTION regress_conflict_log_default DISABLE; +ALTER SUBSCRIPTION regress_conflict_log_default SET (slot_name = NONE); +DROP SUBSCRIPTION regress_conflict_log_default; +ALTER SUBSCRIPTION regress_conflict_empty_str DISABLE; +ALTER SUBSCRIPTION regress_conflict_empty_str SET (slot_name = NONE); +DROP SUBSCRIPTION regress_conflict_empty_str; +ALTER SUBSCRIPTION regress_conflict_test2 DISABLE; +ALTER SUBSCRIPTION regress_conflict_test2 SET (slot_name = NONE); +DROP SUBSCRIPTION regress_conflict_test2; +ALTER SUBSCRIPTION regress_conflict_protection_test DISABLE; +ALTER SUBSCRIPTION regress_conflict_protection_test SET (slot_name = NONE); +DROP SUBSCRIPTION regress_conflict_protection_test; RESET SESSION AUTHORIZATION; DROP ROLE regress_subscription_user; DROP ROLE regress_subscription_user2; diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index ef0c298d2df..d2934478392 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -365,6 +365,223 @@ COMMIT; ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); DROP SUBSCRIPTION regress_testsub; +-- +-- CONFLICT LOG DESTINATION TESTS +-- + +SET SESSION AUTHORIZATION 'regress_subscription_user'; + +-- fail - unrecognized parameter value +CREATE SUBSCRIPTION regress_conflict_fail CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'invalid'); + +-- verify subconflictlogdest is 'log' and relid is 0 (InvalidOid) for default case +CREATE SUBSCRIPTION regress_conflict_log_default CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false); +SELECT subname, subconflictlogdest, subconflictlogrelid +FROM pg_subscription WHERE subname = 'regress_conflict_log_default'; + +-- verify empty string defaults to 'log' +CREATE SUBSCRIPTION regress_conflict_empty_str CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = ''); +SELECT subname, subconflictlogdest, subconflictlogrelid +FROM pg_subscription WHERE subname = 'regress_conflict_empty_str'; + +-- this should generate an internal conflict log table named pg_conflict_$subid$ +CREATE SUBSCRIPTION regress_conflict_test1 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'table'); + +-- check metadata in pg_subscription: destination should be 'table' and relid valid +SELECT subname, subconflictlogdest, subconflictlogrelid > 0 AS has_relid +FROM pg_subscription WHERE subname = 'regress_conflict_test1'; + +-- verify the physical table exists, its OID matches subconflictlogrelid, +-- and it is located in the 'pg_conflict' namespace +SELECT n.nspname, (c.oid = s.subconflictlogrelid) AS "oid_matches" +FROM pg_class c +JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid +JOIN pg_namespace n ON c.relnamespace = n.oid +WHERE s.subname = 'regress_conflict_test1'; + +-- check if the conflict log table has the correct schema +SELECT a.attnum, a.attname +FROM pg_attribute a +JOIN pg_class c ON a.attrelid = c.oid +JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid +WHERE s.subname = 'regress_conflict_test1' AND a.attnum > 0 + ORDER BY a.attnum; + +-- +-- ALTER SUBSCRIPTION - conflict_log_destination state transitions +-- +-- These tests verify the transition logic between different logging +-- destinations, ensuring conflict log tables are created or dropped as +-- expected +-- +-- transition from 'log' to 'all' +-- a new internal conflict log table should be created +CREATE SUBSCRIPTION regress_conflict_test2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'log'); +ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'all'); + +-- verify metadata after ALTER (destination should be 'all') +SELECT subname, subconflictlogdest, subconflictlogrelid > 0 AS has_relid +FROM pg_subscription WHERE subname = 'regress_conflict_test2'; + +-- transition from 'all' to 'table' +-- should NOT drop the table, only change destination string +SELECT subconflictlogrelid AS old_relid FROM pg_subscription WHERE subname = 'regress_conflict_test2' \gset +ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'table'); +SELECT subconflictlogdest, subconflictlogrelid = :old_relid AS relid_unchanged +FROM pg_subscription WHERE subname = 'regress_conflict_test2'; + +-- transition from 'table' to 'log' +-- should drop the table and clear relid +ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'log'); +SELECT subconflictlogdest, subconflictlogrelid +FROM pg_subscription WHERE subname = 'regress_conflict_test2'; + +-- verify the physical table is gone +SELECT count(*) +FROM pg_class c +JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid +WHERE s.subname = 'regress_conflict_test2'; + +-- +-- PUBLICATION: Verify conflict log tables are not publishable +-- +-- pg_relation_is_publishable should return false for internal conflict log +-- tables to prevent them from being accidentally included in publications +-- +SELECT n.nspname, pg_relation_is_publishable(c.oid) +FROM pg_class c +JOIN pg_namespace n ON c.relnamespace = n.oid +JOIN pg_subscription s ON s.subconflictlogrelid = c.oid +WHERE s.subname = 'regress_conflict_test1'; + +-- +-- Table Protection and Lifecycle Management +-- +-- These tests verify that: +-- Manual DROP TABLE is disallowed +-- DROP SUBSCRIPTION automatically reaps the table +-- +-- re-enable table logging for verification +ALTER SUBSCRIPTION regress_conflict_test1 SET (conflict_log_destination = 'table'); + +-- We use a DO block with dynamic SQL because the internal conflict log table +-- name contains the subscription OID, which is non-deterministic. This +-- approach allows us to attempt the DROP and capture the expected error +-- without hard-coding a specific OID in the expected output + +-- fail - drop table not allowed due to internal dependency +DO $$ +BEGIN + EXECUTE 'DROP TABLE ' || (SELECT 'pg_conflict.pg_conflict_' || oid FROM pg_subscription WHERE subname = 'regress_conflict_test1'); +EXCEPTION WHEN insufficient_privilege THEN + RAISE NOTICE 'captured expected error: insufficient_privilege'; +END $$; + +-- CLEANUP: DROP SUBSCRIPTION reaps the table +ALTER SUBSCRIPTION regress_conflict_test1 DISABLE; +ALTER SUBSCRIPTION regress_conflict_test1 SET (slot_name = NONE); + +-- Verify the table OID for reap check +SELECT 'pg_conflict_' || oid AS internal_tablename FROM pg_subscription WHERE subname = 'regress_conflict_test1' \gset + +DROP SUBSCRIPTION regress_conflict_test1; + +-- should return NULL, meaning the conflict log table was reaped via dependency +SELECT to_regclass(:'internal_tablename'); + +-- +-- Additional Namespace and Table Protection Tests +-- + +-- Setup: Ensure we have a subscription with a conflict log table +CREATE SUBSCRIPTION regress_conflict_protection_test CONNECTION 'dbname=regress_doesnotexist' + PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'table'); + +-- Trying to ALTER the internal conflict log table +-- This should fail because the table is system-managed +-- As mentioned in previous test cases, we use a DO block to hide dynamic OIDs +DO $$ +DECLARE + tab_name text; +BEGIN + SELECT 'pg_conflict.' || relname INTO tab_name + FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid + WHERE s.subname = 'regress_conflict_protection_test'; + + RAISE NOTICE 'Attempting ALTER TABLE on internal conflict log table'; + EXECUTE 'ALTER TABLE ' || tab_name || ' ADD COLUMN extra_info text'; +EXCEPTION WHEN insufficient_privilege THEN + RAISE NOTICE 'captured expected error: insufficient_privilege during ALTER'; +END $$; + +-- Test Manual INSERT on conflict log table +-- This should fail because the table is system-managed +-- Hiding the OID in the error message by catching the exception +DO $$ +DECLARE + tab_name text; +BEGIN + SELECT 'pg_conflict.' || relname INTO tab_name + FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid + WHERE s.subname = 'regress_conflict_protection_test'; + + EXECUTE 'INSERT INTO ' || tab_name || ' (relname) VALUES (''mytest'')'; +EXCEPTION WHEN insufficient_privilege THEN + RAISE NOTICE 'captured expected error: insufficient_privilege during INSERT'; +END $$; + +-- Test Manual UPDATE on conflict log table +-- This should fail because the table is system-managed +-- Hiding the OID in the error message by catching the exception +DO $$ +DECLARE + tab_name text; +BEGIN + SELECT 'pg_conflict.' || relname INTO tab_name + FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid + WHERE s.subname = 'regress_conflict_protection_test'; + + EXECUTE 'UPDATE ' || tab_name || ' SET relname = ''mytest'' '; +EXCEPTION WHEN insufficient_privilege THEN + RAISE NOTICE 'captured expected error: insufficient_privilege during UPDATE'; +END $$; + +-- Trying to perform TRUNCATE/DELETE on the internal conflict log table +-- This should be allowed so that user can perform cleanup +SELECT 'pg_conflict.' || relname AS conflict_tab +FROM pg_class c +JOIN pg_subscription s ON c.relname = 'pg_conflict_' || s.oid +WHERE s.subname = 'regress_conflict_protection_test' \gset +TRUNCATE :conflict_tab; +DELETE FROM :conflict_tab; + +-- Trying to create a new table manually in the pg_conflict namespace +-- This should fail as the namespace is reserved for conflict logs tables +CREATE TABLE pg_conflict.manual_table (id int); + +-- Moving an existing table into the pg_conflict namespace +-- Users should not be able to move their own tables within this namespace +CREATE TABLE public.test_move (id int); +ALTER TABLE public.test_move SET SCHEMA pg_conflict; +DROP TABLE public.test_move; + +-- Clean up remaining test subscription +ALTER SUBSCRIPTION regress_conflict_log_default DISABLE; +ALTER SUBSCRIPTION regress_conflict_log_default SET (slot_name = NONE); +DROP SUBSCRIPTION regress_conflict_log_default; + +ALTER SUBSCRIPTION regress_conflict_empty_str DISABLE; +ALTER SUBSCRIPTION regress_conflict_empty_str SET (slot_name = NONE); +DROP SUBSCRIPTION regress_conflict_empty_str; + +ALTER SUBSCRIPTION regress_conflict_test2 DISABLE; +ALTER SUBSCRIPTION regress_conflict_test2 SET (slot_name = NONE); +DROP SUBSCRIPTION regress_conflict_test2; + +ALTER SUBSCRIPTION regress_conflict_protection_test DISABLE; +ALTER SUBSCRIPTION regress_conflict_protection_test SET (slot_name = NONE); +DROP SUBSCRIPTION regress_conflict_protection_test; + RESET SESSION AUTHORIZATION; DROP ROLE regress_subscription_user; DROP ROLE regress_subscription_user2; diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 14dec2d49c1..60644734ed6 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -501,6 +501,8 @@ ConditionVariableMinimallyPadded ConditionalStack ConfigData ConfigVariable +ConflictLogColumnDef +ConflictLogDest ConflictTupleInfo ConflictType ConnCacheEntry -- 2.49.0