</para>
</sect2>
+ <sect2 id="ddl-schemas-conflict">
+ <title>The Conflict Schema</title>
+
+ <indexterm zone="ddl-schemas-conflict">
+ <primary>conflict</primary>
+ <secondary>schema</secondary>
+ </indexterm>
+
+ <para>
+ The <literal>pg_conflict</literal> schema (sometimes referred to as the
+ <emphasis>conflict schema</emphasis>) contains system-managed conflict
+ log tables used for logical replication conflict tracking.
+ These tables are created and maintained by the system and are not intended for
+ direct user manipulation. Unlike <literal>pg_catalog</literal>, the
+ <literal>pg_conflict</literal> schema is not implicitly included in the
+ search path, so objects within it must be referenced explicitly or by
+ adjusting the search path.
+ </para>
+ </sect2>
+
<sect2 id="ddl-schemas-patterns">
<title>Usage Patterns</title>
</glossdef>
</glossentry>
+ <glossentry id="glossary-conflict-schema">
+ <glossterm>Conflict schema</glossterm>
+ <glossdef>
+ <para>
+ The <literal>pg_conflict</literal> schema that contains system-managed
+ conflict log tables for logical replication. These tables are created
+ and maintained automatically by the system and are not intended for
+ direct user manipulation. See <xref linkend="ddl-schemas-conflict"/>.
+ </para>
+ </glossdef>
+ </glossentry>
+
<glossentry id="glossary-connection">
<glossterm>Connection</glossterm>
<glossdef>
option of <command>CREATE SUBSCRIPTION</command> for details.
</para>
+ <para>
+ <link linkend="logical-replication-conflicts">Conflicts</link> that occur
+ during replication are, by default, logged as plain text in the server log,
+ which can make automated monitoring and analysis difficult. The
+ <command>CREATE SUBSCRIPTION</command> command provides the
+ <link linkend="sql-createsubscription-params-with-conflict-log-destination">
+ <literal>conflict_log_destination</literal></link> option to record detailed
+ conflict information in a structured, queryable format. When this parameter
+ is set to <literal>table</literal> or <literal>all</literal>, the system
+ automatically manages a dedicated <firstterm>conflict log table</firstterm>,
+ which is created and dropped along with the subscription. This significantly
+ improves post-mortem analysis and operational visibility of the replication
+ setup.
+ </para>
+
<sect2 id="logical-replication-subscription-slot">
<title>Logical Replication Slot Management</title>
<link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>,
<link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>,
<link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>,
- <link linkend="sql-createsubscription-params-with-max-retention-duration"><literal>max_retention_duration</literal></link>, and
- <link linkend="sql-createsubscription-params-with-wal-receiver-timeout"><literal>wal_receiver_timeout</literal></link>.
+ <link linkend="sql-createsubscription-params-with-max-retention-duration"><literal>max_retention_duration</literal></link>,
+ <link linkend="sql-createsubscription-params-with-wal-receiver-timeout"><literal>wal_receiver_timeout</literal></link>, and
+ <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>.
Only a superuser can set <literal>password_required = false</literal>.
</para>
<quote><literal>pg_conflict_detection</literal></quote>, created to retain
dead tuples for conflict detection, will be dropped.
</para>
+
+ <para>
+ When the <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>
+ parameter is set to <literal>table</literal> or <literal>all</literal>, the system
+ automatically creates the conflict log table. Conversely, if the destination is changed to
+ <literal>log</literal>, the conflict log table is automatically dropped.
+ </para>
</listitem>
</varlistentry>
</listitem>
</varlistentry>
+ <varlistentry id="sql-createsubscription-params-with-conflict-log-destination">
+ <term><literal>conflict_log_destination</literal> (<type>enum</type>)</term>
+ <listitem>
+ <para>
+ Specifies the destination for recording logical replication conflicts.
+ The available values are:
+ <variablelist>
+ <varlistentry>
+ <term><literal>log</literal></term>
+ <listitem>
+ <para>
+ Conflict details are recorded in the server log.
+ This is the default behavior.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry>
+ <term><literal>table</literal></term>
+ <listitem>
+ <para>
+ The system automatically creates a structured table named
+ <literal>pg_conflict_log_<subid></literal> in the
+ <literal>pg_conflict</literal> schema. This allows for easy
+ querying and analysis of conflicts.
+ </para>
+ <caution>
+ <para>
+ The conflict log table is strictly tied to the lifecycle of the
+ subscription or the <literal>conflict_log_destination</literal> setting. If
+ the subscription is dropped, or if the destination is changed to
+ <literal>log</literal>, the table is <emphasis>dropped</emphasis>.
+ </para>
+ </caution>
+ </listitem>
+ </varlistentry>
+ <varlistentry>
+ <term><literal>all</literal></term>
+ <listitem>
+ <para>
+ This is equivalent to configuring both destinations <literal>log</literal>
+ and <literal>table</literal> together.
+ </para>
+ </listitem>
+ </varlistentry>
+ </variablelist>
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry id="sql-createsubscription-params-with-copy-data">
<term><literal>copy_data</literal> (<type>boolean</type>)</term>
<listitem>
If a subscription is associated with a replication slot, then <command>DROP
SUBSCRIPTION</command> cannot be executed inside a transaction block.
</para>
+
+ <para>
+ If a conflict log table exists for the subscription (that is, when
+ <link linkend="sql-createsubscription-params-with-conflict-log-destination">
+ <literal>conflict_log_destination</literal></link> is set to <literal>table</literal>
+ or <literal>all</literal>), <command>DROP SUBSCRIPTION</command> automatically drops
+ the associated conflict log table.
+ </para>
</refsect1>
<refsect1>
classForm = (Form_pg_class) GETSTRUCT(tuple);
- /*
- * Deny anyone permission to update a system catalog unless
- * pg_authid.rolsuper is set.
- *
- * As of 7.4 we have some updatable system views; those shouldn't be
- * protected in this way. Assume the view rules can take care of
- * themselves. ACL_USAGE is if we ever have system sequences.
- */
- if ((mask & (ACL_INSERT | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE | ACL_USAGE)) &&
- IsSystemClass(table_oid, classForm) &&
- classForm->relkind != RELKIND_VIEW &&
- !superuser_arg(roleid))
- mask &= ~(ACL_INSERT | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE | ACL_USAGE);
-
- /*
- * Otherwise, superusers bypass all permission-checking.
- */
- if (superuser_arg(roleid))
+ if (!superuser_arg(roleid))
{
+ if (mask & (ACL_INSERT | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE | ACL_USAGE))
+ {
+ if (IsConflictLogTableClass(classForm))
+ {
+ /*
+ * For conflict log tables, allow non-superusers to perform
+ * DELETE and TRUNCATE for cleanup and maintenance, while
+ * still restricting INSERT, UPDATE, and USAGE.
+ */
+ mask &= ~(ACL_INSERT | ACL_UPDATE | ACL_USAGE);
+ }
+ else if (IsSystemClass(table_oid, classForm) &&
+ classForm->relkind != RELKIND_VIEW)
+ {
+ /*
+ * Deny anyone permission to update a system catalog unless
+ * pg_authid.rolsuper is set.
+ *
+ * As of 7.4 we have some updatable system views; those
+ * shouldn't be protected in this way. Assume the view rules
+ * can take care of themselves. ACL_USAGE is if we ever have
+ * system sequences.
+ */
+ mask &= ~(ACL_INSERT | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE |
+ ACL_USAGE);
+ }
+ }
+ }
+ else
+ {
+ /* Superusers bypass all permission-checking. */
ReleaseSysCache(tuple);
return mask;
}
Acl *acl;
Oid ownerId;
+ /*
+ * Disallow creation in the conflict schema for everyone, including
+ * superusers, unless in binary-upgrade mode.
+ */
+ if (!IsBinaryUpgrade && (mask & ACL_CREATE) &&
+ IsConflictLogTableNamespace(nsp_oid))
+ mask &= ~ACL_CREATE;
+
/* Superusers bypass all permission checking. */
if (superuser_arg(roleid))
return mask;
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) ||
+ IsConflictLogTableClass(reltuple));
}
/*
return IsToastNamespace(relnamespace);
}
+/*
+ * IsConflictLogTableClass
+ * True iff pg_class tuple represents a Conflict Log Table.
+ *
+ * Does not perform any catalog accesses.
+ */
+bool
+IsConflictLogTableClass(Form_pg_class reltuple)
+{
+ Oid relnamespace = reltuple->relnamespace;
+
+ return IsConflictLogTableNamespace(relnamespace);
+}
+
/*
* IsCatalogNamespace
* True iff namespace is pg_catalog.
isTempToastNamespace(namespaceId);
}
+/*
+ * IsConflictLogTableNamespace
+ * True iff namespace is pg_conflict.
+ *
+ * Does not perform any catalog accesses.
+ */
+bool
+IsConflictLogTableNamespace(Oid namespaceId)
+{
+ return namespaceId == PG_CONFLICT_NAMESPACE;
+}
/*
* IsReservedName
errmsg(errormsg, relname),
errdetail("This operation is not supported for system tables.")));
+ /* Can't be conflict log table */
+ if (IsConflictLogTableNamespace(RelationGetNamespace(targetrel)))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg(errormsg, relname),
+ errdetail("This operation is not supported for conflict log tables.")));
+
/* UNLOGGED and TEMP relations cannot be part of publication. */
if (targetrel->rd_rel->relpersistence == RELPERSISTENCE_TEMP)
ereport(ERROR,
check_publication_add_schema(Oid schemaid)
{
/* Can't be system namespace */
- if (IsCatalogNamespace(schemaid) || IsToastNamespace(schemaid))
+ if (IsCatalogNamespace(schemaid) || IsToastNamespace(schemaid) ||
+ IsConflictLogTableNamespace(schemaid))
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("cannot add schema \"%s\" to publication",
* is really inadequate for that, since the information_schema could be
* dropped and reloaded and then it'll be considered publishable. The best
* long-term solution may be to add a "relispublishable" bool to pg_class,
- * and depend on that instead of OID checks.
+ * and depend on that instead of OID checks. IsConflictLogTableClass()
+ * excludes tables in conflict schema.
*/
static bool
is_publishable_class(Oid relid, Form_pg_class reltuple)
reltuple->relkind == RELKIND_PARTITIONED_TABLE ||
reltuple->relkind == RELKIND_SEQUENCE) &&
!IsCatalogRelationOid(relid) &&
+ !IsConflictLogTableClass(reltuple) &&
reltuple->relpersistence == RELPERSISTENCE_PERMANENT &&
relid >= FirstNormalObjectId;
}
sub->retaindeadtuples = subform->subretaindeadtuples;
sub->maxretention = subform->submaxretention;
sub->retentionactive = subform->subretentionactive;
+ sub->conflictlogrelid = subform->subconflictlogrelid;
if (conninfo_needed)
{
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);
subbinary, substream, subtwophasestate, subdisableonerr,
subpasswordrequired, subrunasowner, subfailover,
subretaindeadtuples, submaxretention, subretentionactive,
- subserver, subslotname, subsynccommit, subwalrcvtimeout,
- subpublications, suborigin)
+ subserver, subconflictlogrelid, subconflictlogdest, subslotname,
+ subsynccommit, subwalrcvtimeout, subpublications, suborigin)
ON pg_subscription TO public;
CREATE VIEW pg_stat_subscription_stats AS
return; /* woops, concurrently dropped; no permissions
* check */
- /* Currently, we only allow plain tables or views to be locked */
+ /*
+ * Note: Conflict log tables are deliberately NOT blocked here, even
+ * though other direct DDL on them is rejected elsewhere. pg_dump relies
+ * on being able to take an ACCESS SHARE lock on these tables to safely
+ * dump their definitions during a binary upgrade, so we permit LOCK
+ * commands on them and treat them like ordinary tables here. It's true
+ * that a strong lock (ShareLock or above) on such a table would conflict
+ * with the RowExclusiveLock taken by the apply worker's inserts and could
+ * stall conflict logging as well as the apply worker for as long as it is
+ * held. But locking a system-managed conflict log table is an unusual
+ * thing to do, and it doesn't seem worth the trouble of filtering by lock
+ * mode here.
+ */
if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE &&
relkind != RELKIND_VIEW)
ereport(ERROR,
relkind != RELKIND_VIEW)
continue;
+ /*
+ * Conflict log tables only support SELECT, DELETE, and TRUNCATE.
+ * A direct LOCK on them is permitted solely so that pg_dump can
+ * lock them during a binary upgrade; locking one indirectly by
+ * locking a view over it is not needed for that, so skip it here
+ * rather than locking it.
+ */
+ if (IsConflictLogTableNamespace(get_rel_namespace(relid)))
+ continue;
+
/*
* We might be dealing with a self-referential view. If so, we
* can just stop recursing, since we already locked it.
if (!object_ownercheck(RelationRelationId, relid, GetUserId()))
aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(get_rel_relkind(relid)), rv->relname);
+ /*
+ * Conflict log tables are used internally for logical replication
+ * conflict logging and should not be modified directly, as it could
+ * disrupt conflict logging.
+ */
+ if (IsConflictLogTableClass(classform))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot create policy on conflict log table \"%s\"",
+ rv->relname),
+ errdetail("Conflict log tables are system-managed tables for logical replication conflicts.")));
+
/* No system table modifications unless explicitly allowed. */
if (!allowSystemTableMods && IsSystemClass(relid, classform))
ereport(ERROR,
aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(rel->rd_rel->relkind),
RelationGetRelationName(rel));
+ /*
+ * Conflict log tables are system-managed tables used internally for
+ * logical replication conflict logging. Unlike user tables, they are
+ * not expected to have complex query usage, so to keep things simple,
+ * user-defined extended statistics are not required or supported at
+ * present.
+ */
+ if (IsConflictLogTableClass(rel->rd_rel))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot create statistics on conflict log table \"%s\"",
+ RelationGetRelationName(rel)),
+ errdetail("Conflict log tables are system-managed tables for logical replication conflicts.")));
+
/* Creating statistics on system catalogs is not allowed */
if (!allowSystemTableMods && IsSystemRelation(rel))
ereport(ERROR,
#include "catalog/pg_authid_d.h"
#include "catalog/pg_database_d.h"
#include "catalog/pg_foreign_server.h"
+#include "catalog/pg_namespace.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_type.h"
#include "commands/defrem.h"
#include "commands/event_trigger.h"
#include "commands/subscriptioncmds.h"
+#include "commands/tablecmds.h"
#include "executor/executor.h"
#include "foreign/foreign.h"
#include "miscadmin.h"
#define SUBOPT_WAL_RECEIVER_TIMEOUT 0x00010000
#define SUBOPT_LSN 0x00020000
#define SUBOPT_ORIGIN 0x00040000
+#define SUBOPT_CONFLICT_LOG_DEST 0x00080000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
bool retaindeadtuples;
int32 maxretention;
char *origin;
+ ConflictLogDest conflictlogdest;
XLogRecPtr lsn;
char *wal_receiver_timeout;
} SubOpts;
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 bool alter_sub_conflict_log_dest(Subscription *sub,
+ ConflictLogDest oldlogdest,
+ ConflictLogDest newlogdest,
+ Oid *conflicttablerelid);
+static void drop_sub_conflict_log_table(Oid subid, char *subname,
+ Oid subconflictlogrelid);
/*
* Common option parsing function for CREATE and ALTER SUBSCRIPTION commands.
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)
PGC_BACKEND, PGC_S_TEST, GUC_ACTION_SET,
false, 0, false);
}
+ 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 = GetConflictLogDest(val);
+ opts->specified_opts |= SUBOPT_CONFLICT_LOG_DEST;
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
uint32 supported_opts;
SubOpts opts = {0};
AclResult aclresult;
+ Oid logrelid = InvalidOid;
/*
* Parse and check options.
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
SUBOPT_RETAIN_DEAD_TUPLES |
SUBOPT_MAX_RETENTION_DURATION |
- SUBOPT_WAL_RECEIVER_TIMEOUT | SUBOPT_ORIGIN);
+ SUBOPT_WAL_RECEIVER_TIMEOUT | SUBOPT_ORIGIN |
+ SUBOPT_CONFLICT_LOG_DEST);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
values[Anum_pg_subscription_suborigin - 1] =
CStringGetTextDatum(opts.origin);
+ values[Anum_pg_subscription_subconflictlogdest - 1] =
+ CStringGetTextDatum(ConflictLogDestNames[opts.conflictlogdest]);
+
+ /*
+ * We create the conflict log table here, if required, so that its
+ * relation OID can be stored when inserting the pg_subscription tuple
+ * below.
+ */
+ if (CONFLICTS_LOGGED_TO_TABLE(opts.conflictlogdest))
+ logrelid = create_conflict_log_table(subid, stmt->subname, owner);
+
+ /* 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. */
recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
}
+ /*
+ * 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().
+ */
+ if (OidIsValid(logrelid))
+ {
+ ObjectAddress cltaddr;
+
+ ObjectAddressSet(cltaddr, RelationRelationId, logrelid);
+ recordDependencyOn(&cltaddr, &myself, DEPENDENCY_INTERNAL);
+ }
+
/*
* A replication origin is currently created for all subscriptions,
* including those that only contain sequences or are otherwise empty.
}
}
+/*
+ * alter_sub_conflict_log_dest
+ *
+ * When the subscription's 'conflict_log_destination' is changed, update the
+ * conflict log table if required.
+ *
+ * If the new destination no longer requires a conflict log table, the existing
+ * conflict log table associated with the subscription is removed via internal
+ * dependency cleanup to prevent orphaned relations.
+ *
+ * On success, *conflicttablerelid is set to the OID of the conflict log table
+ * that was created or validated, or to InvalidOid if no table is required.
+ *
+ * Returns true if the subscription's conflict log table reference must be
+ * updated as a result of the destination change; false otherwise.
+ */
+static bool
+alter_sub_conflict_log_dest(Subscription *sub, ConflictLogDest oldlogdest,
+ ConflictLogDest newlogdest,
+ Oid *conflicttablerelid)
+{
+ bool want_table;
+ bool has_oldtable;
+ bool update_relid = false;
+ Oid relid = InvalidOid;
+
+ want_table = CONFLICTS_LOGGED_TO_TABLE(newlogdest);
+ has_oldtable = CONFLICTS_LOGGED_TO_TABLE(oldlogdest);
+
+ if (has_oldtable)
+ {
+ /* There is a conflict log table already. */
+ if (!want_table)
+ {
+ drop_sub_conflict_log_table(sub->oid, sub->name,
+ sub->conflictlogrelid);
+ update_relid = true;
+ }
+ }
+ else
+ {
+ /* There was no previous conflict log table. */
+ if (want_table)
+ {
+ ObjectAddress cltaddr;
+ ObjectAddress subobj;
+
+ relid = create_conflict_log_table(sub->oid, sub->name, sub->owner);
+ update_relid = true;
+
+ /*
+ * Establish an internal dependency between the conflict log table
+ * and the subscription. For details refer comments in
+ * CreateSubscription function.
+ */
+ ObjectAddressSet(cltaddr, RelationRelationId, relid);
+ ObjectAddressSet(subobj, SubscriptionRelationId, sub->oid);
+ recordDependencyOn(&cltaddr, &subobj, DEPENDENCY_INTERNAL);
+ }
+ }
+
+ *conflicttablerelid = relid;
+ return update_relid;
+}
+
/*
* Alter the existing subscription.
*/
SUBOPT_RETAIN_DEAD_TUPLES |
SUBOPT_MAX_RETENTION_DURATION |
SUBOPT_WAL_RECEIVER_TIMEOUT |
- SUBOPT_ORIGIN);
+ SUBOPT_ORIGIN |
+ SUBOPT_CONFLICT_LOG_DEST);
break;
case ALTER_SUBSCRIPTION_ENABLED:
replaces[Anum_pg_subscription_subwalrcvtimeout - 1] = true;
}
+ if (IsSet(opts.specified_opts, SUBOPT_CONFLICT_LOG_DEST))
+ {
+ ConflictLogDest old_dest =
+ GetConflictLogDest(sub->conflictlogdest);
+
+ if (opts.conflictlogdest != old_dest)
+ {
+ bool update_relid;
+ Oid relid = InvalidOid;
+
+ values[Anum_pg_subscription_subconflictlogdest - 1] =
+ CStringGetTextDatum(ConflictLogDestNames[opts.conflictlogdest]);
+ replaces[Anum_pg_subscription_subconflictlogdest - 1] = true;
+
+ update_relid = alter_sub_conflict_log_dest(sub,
+ old_dest,
+ opts.conflictlogdest,
+ &relid);
+ if (update_relid)
+ {
+ values[Anum_pg_subscription_subconflictlogrelid - 1] =
+ ObjectIdGetDatum(relid);
+ replaces[Anum_pg_subscription_subconflictlogrelid - 1] =
+ true;
+ }
+ }
+ }
+
update_tuple = true;
break;
}
return ForeignServerConnectionString(subowner, server);
}
+/*
+ * Drop subscription's conflict log table
+ *
+ * The conflict log table is registered as an internal dependency of the
+ * subscription. This function removes the dependency by performing a
+ * cascading deletion on the subscription object, which in turn drops the
+ * associated conflict log table.
+ *
+ * This is used to clean up conflict log tables that are no longer required,
+ * preventing accumulation of stale or orphaned relations.
+ *
+ * NOTE:
+ * Only conflict log tables are currently managed via this internal dependency
+ * mechanism.
+ */
+static void
+drop_sub_conflict_log_table(Oid subid, char *subname, Oid subconflictlogrelid)
+{
+ /* Drop any dependent conflict log table */
+ if (OidIsValid(subconflictlogrelid))
+ {
+ ObjectAddress object;
+ char *conflictrelname;
+
+ conflictrelname = get_rel_name(subconflictlogrelid);
+ if (conflictrelname == NULL)
+ elog(ERROR, "cache lookup failed for relation %u",
+ subconflictlogrelid);
+
+ /*
+ * By using PERFORM_DELETION_SKIP_ORIGINAL, we ensure that only the
+ * conflict log table is deleted while the subscription remains.
+ */
+ ObjectAddressSet(object, SubscriptionRelationId, subid);
+ performDeletion(&object, DROP_CASCADE,
+ PERFORM_DELETION_INTERNAL |
+ PERFORM_DELETION_SKIP_ORIGINAL);
+
+ ereport(NOTICE,
+ errmsg("dropped conflict log table \"%s\" for subscription \"%s\"",
+ get_qualified_objname(PG_CONFLICT_NAMESPACE, conflictrelname),
+ subname));
+ }
+}
+
/*
* Drop a subscription
*/
Oid subid;
Oid subowner;
Oid subserver;
+ Oid subconflictlogrelid;
char *subconninfo = NULL;
Datum datum;
bool isnull;
subid = form->oid;
subowner = form->subowner;
subserver = form->subserver;
+ subconflictlogrelid = form->subconflictlogrelid;
must_use_password = !superuser_arg(subowner) && form->subpasswordrequired;
/* must be owner */
replorigin_drop_by_name(originname, true, false);
}
+ /* Drop subscription's conflict log table */
+ drop_sub_conflict_log_table(subid, subname, subconflictlogrelid);
+
/* Clean up dependencies */
deleteDependencyRecordsFor(SubscriptionRelationId, subid, false);
deleteSharedDependencyRecordsFor(SubscriptionRelationId, subid, 0);
form->subowner = newOwnerId;
CatalogTupleUpdate(rel, &tup->t_self, tup);
+ /* Update owner of the conflict log table if it exists. */
+ if (OidIsValid(form->subconflictlogrelid))
+ ATExecChangeOwner(form->subconflictlogrelid, newOwnerId, true,
+ AccessExclusiveLock);
+
/* Update owner dependency reference */
changeDependencyOnOwner(SubscriptionRelationId,
form->oid,
* 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 conflict data to manage disk space.
*/
- if (!allowSystemTableMods && IsSystemClass(relid, reltuple)
+ if (!allowSystemTableMods && IsSystemClass(relid, reltuple) &&
+ !IsConflictLogTableClass(reltuple)
&& (!IsBinaryUpgrade ||
(relid != LargeObjectRelationId &&
relid != LargeObjectMetadataRelationId)))
errmsg("cannot inherit from partition \"%s\"",
RelationGetRelationName(relation))));
+ /*
+ * Conflict log tables are managed by the system for logical
+ * replication and should not be used as parent tables, as inheritance
+ * could interfere with the logging behavior.
+ */
+ if (IsConflictLogTableNamespace(relation->rd_rel->relnamespace))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot inherit from conflict log table \"%s\"",
+ RelationGetRelationName(relation)),
+ errdetail("Conflict log tables are system-managed tables for logical replication conflicts.")));
+
if (relation->rd_rel->relkind != RELKIND_RELATION &&
relation->rd_rel->relkind != RELKIND_FOREIGN_TABLE &&
relation->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
if (!object_ownercheck(RelationRelationId, myrelid, GetUserId()))
aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(get_rel_relkind(myrelid)),
NameStr(classform->relname));
+
+ /*
+ * Conflict log tables are used internally for logical replication
+ * conflict logging and should not be modified directly, as it could
+ * disrupt conflict logging.
+ */
+ if (IsConflictLogTableClass(classform))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot rename columns of conflict log table \"%s\"",
+ NameStr(classform->relname)),
+ errdetail("Conflict log tables are system-managed tables for logical replication conflicts.")));
+
if (!allowSystemTableMods && IsSystemClass(myrelid, classform))
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(rel->rd_rel->relkind),
RelationGetRelationName(rel));
+ /*
+ * Conflict log tables are used internally for logical replication
+ * conflict logging and should not be altered directly, as it could
+ * disrupt conflict logging. Direct ALTER commands are already rejected
+ * during relation lookup in RangeVarCallbackForAlterRelation(), and
+ * AlterTableMoveAll() skips these tables, so a conflict log table does
+ * not normally reach here; this check guards any internal caller that
+ * arrives via AlterTableInternal().
+ */
+ if (IsConflictLogTableClass(rel->rd_rel))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot alter conflict log table \"%s\"",
+ RelationGetRelationName(rel)),
+ errdetail("Conflict log tables are system-managed tables for logical replication conflicts.")));
+
if (!allowSystemTableMods && IsSystemRelation(rel))
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("referenced relation \"%s\" is not a table",
RelationGetRelationName(pkrel))));
+ /*
+ * Conflict log tables are used internally for logical replication
+ * conflict logging and should not be referenced by foreign keys, as it
+ * could disrupt conflict logging.
+ */
+ if (IsConflictLogTableClass(pkrel->rd_rel))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot reference conflict log table \"%s\"",
+ RelationGetRelationName(pkrel)),
+ errdetail("Conflict log tables are system-managed tables for logical replication conflicts.")));
+
if (!allowSystemTableMods && IsSystemRelation(pkrel))
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
* really wishes to do so, they can issue the individual ALTER
* commands directly.
*
- * Also, explicitly avoid any shared tables, temp tables, or TOAST
- * (TOAST will be moved with the main table).
+ * Also, explicitly avoid any shared tables, temp tables, TOAST (TOAST
+ * will be moved with the main table).
+ *
+ * Conflict log tables are system-managed for logical replication and
+ * cannot be altered directly, so skip them as well; otherwise a
+ * single such table in the source tablespace would abort the whole
+ * bulk move.
*/
if (IsCatalogNamespace(relForm->relnamespace) ||
relForm->relisshared ||
isAnyTempNamespace(relForm->relnamespace) ||
- IsToastNamespace(relForm->relnamespace))
+ IsToastNamespace(relForm->relnamespace) ||
+ IsConflictLogTableNamespace(relForm->relnamespace))
continue;
/* Only move the object type requested */
aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(get_rel_relkind(relId)),
relation->relname);
+ /*
+ * Conflict log tables are used internally for logical replication
+ * conflict logging and should not be modified directly, as it could
+ * disrupt conflict logging.
+ */
+ if (IsConflictLogTableClass((Form_pg_class) GETSTRUCT(tuple)))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot change conflict log table \"%s\"",
+ relation->relname),
+ errdetail("Conflict log tables are system-managed tables for logical replication conflicts.")));
+
if (!allowSystemTableMods &&
IsSystemClass(relId, (Form_pg_class) GETSTRUCT(tuple)))
ereport(ERROR,
if (!object_ownercheck(RelationRelationId, relid, GetUserId()))
aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(get_rel_relkind(relid)), rv->relname);
+ /*
+ * Conflict log tables are used internally for logical replication
+ * conflict logging and should not be altered directly, as it could
+ * disrupt conflict logging.
+ */
+ if (IsConflictLogTableClass(classform))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot alter conflict log table \"%s\"",
+ rv->relname),
+ errdetail("Conflict log tables are system-managed tables for logical replication conflicts.")));
+
/* No system table modifications unless explicitly allowed. */
if (!allowSystemTableMods && IsSystemClass(relid, classform))
ereport(ERROR,
RelationGetRelationName(rel)),
errdetail_relkind_not_supported(rel->rd_rel->relkind)));
+ /*
+ * Conflict log tables are used internally for logical replication
+ * conflict logging and should not have triggers, as it could disrupt
+ * conflict logging.
+ */
+ if (IsConflictLogTableClass(rel->rd_rel))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot create trigger on conflict log table \"%s\"",
+ RelationGetRelationName(rel)),
+ errdetail("Conflict log tables are system-managed tables for logical replication conflicts.")));
+
if (!allowSystemTableMods && IsSystemRelation(rel))
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
/* you must own the table to rename one of its triggers */
if (!object_ownercheck(RelationRelationId, relid, GetUserId()))
aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(get_rel_relkind(relid)), rv->relname);
+
+ /*
+ * Conflict log tables are used internally for logical replication
+ * conflict logging and should not have triggers, as it could disrupt
+ * conflict logging.
+ */
+ if (IsConflictLogTableClass(form))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot rename trigger on conflict log table \"%s\"",
+ rv->relname),
+ errdetail("Conflict log tables are system-managed tables for logical replication conflicts.")));
+
if (!allowSystemTableMods && IsSystemClass(relid, form))
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
RelationGetRelationName(resultRel))));
break;
}
+
+ /*
+ * Conflict log tables are managed by the system to record logical
+ * replication conflicts. We allow DELETE and TRUNCATE to permit users to
+ * manually prune these logs, but manual data insertion or modification
+ * (INSERT, UPDATE, MERGE) is prohibited to maintain the integrity of the
+ * system-generated logs.
+ *
+ * Since TRUNCATE is handled as a separate utility command, we only need
+ * to explicitly permit CMD_DELETE here.
+ */
+ if (IsConflictLogTableNamespace(RelationGetNamespace(resultRel)) &&
+ operation != CMD_DELETE)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot modify or insert data into conflict log table \"%s\"",
+ RelationGetRelationName(resultRel)),
+ errdetail("Conflict log tables are system-managed and only support cleanup using DELETE or TRUNCATE.")));
}
/*
RelationGetRelationName(rel))));
break;
}
+
+ /*
+ * Conflict log tables are managed by the system to record logical
+ * replication conflicts.
+ */
+ if (IsConflictLogTableNamespace(RelationGetNamespace(rel)))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot lock rows in the conflict log table \"%s\"",
+ RelationGetRelationName(rel))));
}
/*
#include "access/commit_ts.h"
#include "access/genam.h"
#include "access/tableam.h"
+#include "catalog/heap.h"
+#include "catalog/pg_am.h"
+#include "catalog/pg_namespace.h"
+#include "catalog/toasting.h"
#include "executor/executor.h"
#include "pgstat.h"
#include "replication/conflict.h"
#include "storage/lmgr.h"
#include "utils/lsyscache.h"
+/*
+ * String representations for the supported conflict logging destinations.
+ */
+const char *const ConflictLogDestNames[] = {
+ [CONFLICT_LOG_DEST_LOG] = "log",
+ [CONFLICT_LOG_DEST_TABLE] = "table",
+ [CONFLICT_LOG_DEST_ALL] = "all"
+};
+
+StaticAssertDecl(lengthof(ConflictLogDestNames) == CONFLICT_LOG_DEST_ALL + 1,
+ "ConflictLogDestNames length mismatch");
+
+
+/* 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;
+
+/*
+ * Schema definition for conflict log tables.
+ *
+ * Defines the fixed schema of the per-subscription conflict log table created
+ * in the pg_conflict namespace. Each entry specifies the column name and its
+ * type OID; the table is created in this column order by
+ * create_conflict_log_table().
+ *
+ * The tuple/key columns (replica_identity, remote_tuple, local_conflicts) are
+ * typed json rather than jsonb on purpose: they hold an exact audit snapshot
+ * of the applied tuples and replica identity, and json preserves the verbatim
+ * representation whereas jsonb would normalize it. Indexing them (jsonb's main
+ * advantage) wouldn't help anyway, as the conflict log is looked up by its
+ * scalar columns (relid, conflict_type, commit timestamp) while these json
+ * columns are per-conflict payload to inspect, not search keys.
+ */
+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_full", .atttypid = BOOLOID},
+ {.attname = "replica_identity", .atttypid = JSONOID},
+ {.attname = "remote_tuple", .atttypid = JSONOID},
+ {.attname = "local_conflicts", .atttypid = JSONARRAYOID}
+};
+
+#define NUM_CONFLICT_ATTRS lengthof(ConflictLogSchema)
+
static const char *const ConflictTypeNames[] = {
[CT_INSERT_EXISTS] = "insert_exists",
[CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs",
static char *build_index_value_desc(EState *estate, Relation localrel,
TupleTableSlot *slot, Oid indexoid);
+/*
+ * Builds the TupleDesc for the conflict log table.
+ */
+static TupleDesc
+create_conflict_log_table_tupdesc(void)
+{
+ TupleDesc tupdesc;
+
+ tupdesc = CreateTemplateTupleDesc(NUM_CONFLICT_ATTRS);
+
+ for (int i = 0; i < NUM_CONFLICT_ATTRS; i++)
+ TupleDescInitEntry(tupdesc, i + 1,
+ ConflictLogSchema[i].attname,
+ ConflictLogSchema[i].atttypid,
+ -1, 0);
+
+ TupleDescFinalize(tupdesc);
+
+ return tupdesc;
+}
+
+/*
+ * Create a structured conflict log table for a subscription.
+ *
+ * The table is created within the system-managed 'pg_conflict' namespace to
+ * prevent users from manually dropping or altering it. This also prevents
+ * accidental name collisions with user-created tables with the same name.
+ *
+ * The table name is generated automatically using the subscription's OID
+ * (e.g., "pg_conflict_log_<subid>") to ensure uniqueness within the
+ * cluster and to avoid collisions during subscription renames.
+ */
+Oid
+create_conflict_log_table(Oid subid, char *subname, Oid subowner)
+{
+ TupleDesc tupdesc;
+ Oid relid;
+ char relname[NAMEDATALEN];
+
+ snprintf(relname, NAMEDATALEN, "pg_conflict_log_%u", subid);
+
+ /* 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 */
+ subowner,
+ 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 */
+ false, /* allow_system_table_mods */
+ true, /* is_internal */
+ InvalidOid, /* relrewrite */
+ NULL); /* typaddress */
+ Assert(OidIsValid(relid));
+
+ /* Release tuple descriptor memory. */
+ FreeTupleDesc(tupdesc);
+
+ /*
+ * We must bump the command counter to make the newly-created relation
+ * tuple visible for opening.
+ */
+ CommandCounterIncrement();
+
+ /*
+ * Create a TOAST table for the conflict log to support out-of-line
+ * storage of large json data.
+ */
+ NewRelationCreateToastTable(relid, (Datum) 0);
+
+ ereport(NOTICE,
+ (errmsg("created conflict log table \"%s\" for subscription \"%s\"",
+ get_qualified_objname(PG_CONFLICT_NAMESPACE, relname),
+ subname)));
+
+ return relid;
+}
+
+/*
+ * Convert the string representation of a conflict logging destination to its
+ * corresponding enum value.
+ */
+ConflictLogDest
+GetConflictLogDest(const char *dest)
+{
+ /* NULL defaults to LOG. */
+ if (dest == NULL || 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\".")));
+}
+
/*
* Get the xmin and commit timestamp data (origin and timestamp) associated
* with the provided local row.
RelationGetRelationName(event_relation)),
errdetail_relkind_not_supported(event_relation->rd_rel->relkind)));
+ /*
+ * Conflict log tables are used internally for logical replication
+ * conflict logging and should not have rules, as it could disrupt
+ * conflict logging.
+ */
+ if (IsConflictLogTableClass(event_relation->rd_rel))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("conflict log table \"%s\" cannot have rules",
+ RelationGetRelationName(event_relation)),
+ errdetail("Conflict log tables are system-managed tables for logical replication conflicts.")));
+
if (!allowSystemTableMods && IsSystemRelation(event_relation))
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("relation \"%s\" cannot have rules", rv->relname),
errdetail_relkind_not_supported(form->relkind)));
+ /*
+ * Conflict log tables are used internally for logical replication
+ * conflict logging and should not have rules, as it could disrupt
+ * conflict logging.
+ */
+ if (IsConflictLogTableClass(form))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("conflict log table \"%s\" cannot have rules",
+ rv->relname),
+ errdetail("Conflict log tables are system-managed tables for logical replication conflicts.")));
+
if (!allowSystemTableMods && IsSystemClass(relid, form))
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
" AND relacl IS NULL;\n\n",
escape_quotes(username));
PG_CMD_PUTS("GRANT USAGE ON SCHEMA pg_catalog, public TO PUBLIC;\n\n");
+
+ /*
+ * Allow non-superuser subscription owners to access their associated
+ * conflict log tables in the pg_conflict schema.
+ */
+ PG_CMD_PUTS("GRANT USAGE ON SCHEMA pg_conflict TO PUBLIC;\n\n");
PG_CMD_PUTS("REVOKE ALL ON pg_largeobject FROM PUBLIC;\n\n");
PG_CMD_PUTS("INSERT INTO pg_init_privs "
" (objoid, classoid, objsubid, initprivs, privtype)"
COMPLETE_WITH("(", "PUBLICATION");
/* ALTER SUBSCRIPTION <name> 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",
COMPLETE_WITH("WITH (");
/* Complete "CREATE SUBSCRIPTION <name> ... WITH ( <opt>" */
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",
extern bool IsSystemClass(Oid relid, Form_pg_class reltuple);
extern bool IsToastClass(Form_pg_class reltuple);
+extern bool IsConflictLogTableClass(Form_pg_class reltuple);
extern bool IsCatalogRelationOid(Oid relid);
extern bool IsCatalogTextUniqueIndexOid(Oid relid);
extern bool IsCatalogNamespace(Oid namespaceId);
extern bool IsToastNamespace(Oid namespaceId);
+extern bool IsConflictLogTableNamespace(Oid namespaceId);
extern bool IsReservedName(const char *name);
*/
/* yyyymmddN */
-#define CATALOG_VERSION_NO 202607021
+#define CATALOG_VERSION_NO 202607022
#endif
{ oid => '99', oid_symbol => 'PG_TOAST_NAMESPACE',
descr => 'reserved schema for TOAST tables',
nspname => 'pg_toast', nspacl => '_null_' },
+{ oid => '8988', 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',
Oid subserver BKI_LOOKUP_OPT(pg_foreign_server); /* If connection uses
* server */
+ Oid subconflictlogrelid; /* Relid of the conflict log table. */
#ifdef CATALOG_VARLEN /* variable-length fields start here */
+
+ /*
+ * Strategy for logging replication conflicts: 'log' - server log only,
+ * 'table' - conflict log table only, 'all' - both log and table.
+ */
+ text subconflictlogdest BKI_FORCE_NOT_NULL;
+
/* Connection string to the publisher */
text subconninfo; /* Set if connecting with connection string */
* 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
* conflicting local row occurred */
} ConflictTupleInfo;
+/*
+ * Defines where logical replication conflict details are recorded.
+ *
+ * While stored as a text-based array/string in
+ * pg_subscription.subconflictlogdest for user readability and extensibility,
+ * we map these to an internal enum to allow for efficient checks.
+ */
+typedef enum ConflictLogDest
+{
+ CONFLICT_LOG_DEST_LOG = 0, /* Emit to server logs */
+ CONFLICT_LOG_DEST_TABLE, /* Insert into the conflict log table */
+ CONFLICT_LOG_DEST_ALL /* Both log and table */
+} ConflictLogDest;
+
+#define CONFLICTS_LOGGED_TO_TABLE(dest) \
+ ((dest == CONFLICT_LOG_DEST_TABLE) || (dest == CONFLICT_LOG_DEST_ALL))
+#define CONFLICTS_LOGGED_TO_LOG(dest) \
+ ((dest == CONFLICT_LOG_DEST_LOG) || (dest == CONFLICT_LOG_DEST_ALL))
+
+/*
+ * Array mapping for converting internal enum to string.
+ */
+extern PGDLLIMPORT const char *const ConflictLogDestNames[];
+
+extern Oid create_conflict_log_table(Oid subid, char *subname, Oid subowner);
+extern ConflictLogDest GetConflictLogDest(const char *dest);
extern bool GetTupleTransactionInfo(TupleTableSlot *localslot,
TransactionId *xmin,
ReplOriginId *localorigin,
-- 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';
+SET client_min_messages = WARNING;
+-- 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 subconflictlogrelid 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)
+
+-- fail - empty string parameter value
+CREATE SUBSCRIPTION regress_conflict_empty_str CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = '');
+ERROR: unrecognized conflict_log_destination value: ""
+HINT: Valid values are "log", "table", and "all".
+-- this should generate a conflict log table named pg_conflict_log_$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 subconflictlogrelid 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_log_' || 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_log_' || 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_full
+ 10 | replica_identity
+ 11 | remote_tuple
+ 12 | local_conflicts
+(12 rows)
+
+-- Changing the subscription owner should also update the owner
+-- of the associated conflict log table.
+ALTER SUBSCRIPTION regress_conflict_test1 OWNER TO regress_subscription_user2;
+SELECT pg_catalog.pg_get_userbyid(c.relowner) AS owner
+FROM pg_catalog.pg_class c
+JOIN pg_catalog.pg_subscription s
+ ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test1';
+ owner
+----------------------------
+ regress_subscription_user2
+(1 row)
+
+-- Verify that a non-superuser subscription owner can truncate,
+-- delete from, and select from the associated conflict log table.
+SET ROLE 'regress_subscription_user2';
+SELECT format('%I.%I', n.nspname, c.relname) AS conflict_log_table
+FROM pg_catalog.pg_class c
+JOIN pg_catalog.pg_namespace n
+ ON n.oid = c.relnamespace
+JOIN pg_catalog.pg_subscription s
+ ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test1'
+\gset
+TRUNCATE TABLE :conflict_log_table;
+DELETE FROM :conflict_log_table;
+SELECT COUNT(*) FROM :conflict_log_table;
+ count
+-------
+ 0
+(1 row)
+
+RESET ROLE;
+-- Restore the original subscription owner.
+ALTER SUBSCRIPTION regress_conflict_test1 OWNER TO regress_subscription_user;
+--
+-- 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 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 subconflictlogrelid
+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_log_' || 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 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');
+-- The conflict log table name contains the subscription OID, which is
+-- non-deterministic. Capture it into a psql variable and report only the
+-- SQLSTATE, so the expected output does not depend on the OID.
+-- fail - drop table not allowed due to internal dependency
+SET client_min_messages = NOTICE;
+SELECT 'pg_conflict.pg_conflict_log_' || oid AS clt1
+ FROM pg_subscription WHERE subname = 'regress_conflict_test1' \gset
+\set VERBOSITY sqlstate
+DROP TABLE :clt1;
+ERROR: 42501
+\set VERBOSITY default
+-- 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.pg_conflict_log_' || oid AS internal_tablename FROM pg_subscription WHERE subname = 'regress_conflict_test1' \gset
+SET client_min_messages = WARNING;
+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.
+-- The conflict log table is system-managed; its name contains the
+-- subscription OID, which is non-deterministic. Capture the name into a psql
+-- variable and report only the SQLSTATE for the operations that must be
+-- rejected, so the expected output stays free of the dynamic OID. Every
+-- statement in the VERBOSITY-sqlstate block below must fail with 42809
+-- (wrong_object_type), except adding the table to a publication, which fails
+-- with 22023 (invalid_parameter_value).
+SELECT 'pg_conflict.' || relname AS clt
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+ WHERE s.subname = 'regress_conflict_protection_test' \gset
+-- Trigger function used by the CREATE TRIGGER check below.
+CREATE FUNCTION public.dummy_trigger_func() RETURNS trigger AS $$
+BEGIN
+ RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+\set VERBOSITY sqlstate
+ALTER TABLE :clt ADD COLUMN extra_info text;
+ERROR: 42809
+INSERT INTO :clt (relname) VALUES ('mytest');
+ERROR: 42809
+UPDATE :clt SET relname = 'mytest';
+ERROR: 42809
+CREATE POLICY p1 ON :clt USING (true);
+ERROR: 42809
+CREATE STATISTICS s1 ON relname, schemaname FROM :clt;
+ERROR: 42809
+CREATE TABLE public.conflict_child () INHERITS (:clt);
+ERROR: 42809
+ALTER TABLE :clt RENAME COLUMN relname TO new_relname;
+ERROR: 42809
+CREATE TABLE public.conflict_fk (relname text REFERENCES :clt(relname));
+ERROR: 42809
+ALTER TABLE :clt OWNER TO regress_subscription_user_dummy;
+ERROR: 42809
+ALTER TABLE :clt SET SCHEMA public;
+ERROR: 42809
+CREATE TRIGGER t1 BEFORE INSERT ON :clt FOR EACH ROW EXECUTE FUNCTION public.dummy_trigger_func();
+ERROR: 42809
+ALTER TRIGGER non_existent_trigger ON :clt RENAME TO new_trigger;
+ERROR: 42809
+CREATE RULE r1 AS ON INSERT TO :clt DO INSTEAD NOTHING;
+ERROR: 42809
+ALTER RULE non_existent_rule ON :clt RENAME TO new_rule;
+ERROR: 42809
+CREATE INDEX idx1 ON :clt (relname);
+ERROR: 42809
+SELECT 1 FROM :clt FOR UPDATE;
+ERROR: 42809
+CREATE PUBLICATION testpub_for_clt FOR TABLE :clt;
+ERROR: 22023
+\set VERBOSITY default
+-- Clean up the trigger function used above.
+DROP FUNCTION public.dummy_trigger_func();
+-- TRUNCATE and DELETE are allowed so that users can prune the conflict log.
+TRUNCATE :clt;
+DELETE FROM :clt;
+-- Creating a table directly in the pg_conflict namespace is rejected for
+-- everyone (the schema is reserved for conflict log tables).
+CREATE TABLE pg_conflict.manual_table (id int);
+ERROR: permission denied for schema pg_conflict
+LINE 1: CREATE TABLE pg_conflict.manual_table (id int);
+ ^
+-- Moving a user table into the pg_conflict namespace is likewise rejected.
+CREATE TABLE public.test_move (id int);
+ALTER TABLE public.test_move SET SCHEMA pg_conflict;
+ERROR: permission denied for schema pg_conflict
+DROP TABLE public.test_move;
+SET client_min_messages = WARNING;
+-- 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_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;
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
+--
+-- CONFLICT LOG DESTINATION TESTS
+--
+
+SET SESSION AUTHORIZATION 'regress_subscription_user';
+
+SET client_min_messages = WARNING;
+
+-- 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 subconflictlogrelid 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';
+
+-- fail - empty string parameter value
+CREATE SUBSCRIPTION regress_conflict_empty_str CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = '');
+
+-- this should generate a conflict log table named pg_conflict_log_$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 subconflictlogrelid 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_log_' || 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_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test1' AND a.attnum > 0
+ ORDER BY a.attnum;
+
+-- Changing the subscription owner should also update the owner
+-- of the associated conflict log table.
+ALTER SUBSCRIPTION regress_conflict_test1 OWNER TO regress_subscription_user2;
+SELECT pg_catalog.pg_get_userbyid(c.relowner) AS owner
+FROM pg_catalog.pg_class c
+JOIN pg_catalog.pg_subscription s
+ ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test1';
+
+-- Verify that a non-superuser subscription owner can truncate,
+-- delete from, and select from the associated conflict log table.
+SET ROLE 'regress_subscription_user2';
+
+SELECT format('%I.%I', n.nspname, c.relname) AS conflict_log_table
+FROM pg_catalog.pg_class c
+JOIN pg_catalog.pg_namespace n
+ ON n.oid = c.relnamespace
+JOIN pg_catalog.pg_subscription s
+ ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test1'
+\gset
+
+TRUNCATE TABLE :conflict_log_table;
+DELETE FROM :conflict_log_table;
+SELECT COUNT(*) FROM :conflict_log_table;
+
+RESET ROLE;
+
+-- Restore the original subscription owner.
+ALTER SUBSCRIPTION regress_conflict_test1 OWNER TO regress_subscription_user;
+
+--
+-- 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 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 subconflictlogrelid
+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_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test2';
+
+--
+-- PUBLICATION: Verify conflict log tables are not publishable
+--
+-- pg_relation_is_publishable should return false for 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');
+
+-- The conflict log table name contains the subscription OID, which is
+-- non-deterministic. Capture it into a psql variable and report only the
+-- SQLSTATE, so the expected output does not depend on the OID.
+
+-- fail - drop table not allowed due to internal dependency
+SET client_min_messages = NOTICE;
+SELECT 'pg_conflict.pg_conflict_log_' || oid AS clt1
+ FROM pg_subscription WHERE subname = 'regress_conflict_test1' \gset
+\set VERBOSITY sqlstate
+DROP TABLE :clt1;
+\set VERBOSITY default
+
+-- 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.pg_conflict_log_' || oid AS internal_tablename FROM pg_subscription WHERE subname = 'regress_conflict_test1' \gset
+
+SET client_min_messages = WARNING;
+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');
+
+-- The conflict log table is system-managed; its name contains the
+-- subscription OID, which is non-deterministic. Capture the name into a psql
+-- variable and report only the SQLSTATE for the operations that must be
+-- rejected, so the expected output stays free of the dynamic OID. Every
+-- statement in the VERBOSITY-sqlstate block below must fail with 42809
+-- (wrong_object_type), except adding the table to a publication, which fails
+-- with 22023 (invalid_parameter_value).
+SELECT 'pg_conflict.' || relname AS clt
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+ WHERE s.subname = 'regress_conflict_protection_test' \gset
+
+-- Trigger function used by the CREATE TRIGGER check below.
+CREATE FUNCTION public.dummy_trigger_func() RETURNS trigger AS $$
+BEGIN
+ RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+\set VERBOSITY sqlstate
+ALTER TABLE :clt ADD COLUMN extra_info text;
+INSERT INTO :clt (relname) VALUES ('mytest');
+UPDATE :clt SET relname = 'mytest';
+CREATE POLICY p1 ON :clt USING (true);
+CREATE STATISTICS s1 ON relname, schemaname FROM :clt;
+CREATE TABLE public.conflict_child () INHERITS (:clt);
+ALTER TABLE :clt RENAME COLUMN relname TO new_relname;
+CREATE TABLE public.conflict_fk (relname text REFERENCES :clt(relname));
+ALTER TABLE :clt OWNER TO regress_subscription_user_dummy;
+ALTER TABLE :clt SET SCHEMA public;
+CREATE TRIGGER t1 BEFORE INSERT ON :clt FOR EACH ROW EXECUTE FUNCTION public.dummy_trigger_func();
+ALTER TRIGGER non_existent_trigger ON :clt RENAME TO new_trigger;
+CREATE RULE r1 AS ON INSERT TO :clt DO INSTEAD NOTHING;
+ALTER RULE non_existent_rule ON :clt RENAME TO new_rule;
+CREATE INDEX idx1 ON :clt (relname);
+SELECT 1 FROM :clt FOR UPDATE;
+CREATE PUBLICATION testpub_for_clt FOR TABLE :clt;
+\set VERBOSITY default
+
+-- Clean up the trigger function used above.
+DROP FUNCTION public.dummy_trigger_func();
+
+-- TRUNCATE and DELETE are allowed so that users can prune the conflict log.
+TRUNCATE :clt;
+DELETE FROM :clt;
+
+-- Creating a table directly in the pg_conflict namespace is rejected for
+-- everyone (the schema is reserved for conflict log tables).
+CREATE TABLE pg_conflict.manual_table (id int);
+
+-- Moving a user table into the pg_conflict namespace is likewise rejected.
+CREATE TABLE public.test_move (id int);
+ALTER TABLE public.test_move SET SCHEMA pg_conflict;
+DROP TABLE public.test_move;
+
+
+SET client_min_messages = WARNING;
+
+-- 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_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;
'postgres',
"CREATE SUBSCRIPTION sub_tab
CONNECTION '$publisher_connstr application_name=$appname'
- PUBLICATION pub_tab;");
+ PUBLICATION pub_tab WITH (conflict_log_destination=all)");
# Wait for initial table sync to finish
$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
),
"the slot 'pg_conflict_detection' has been dropped on Node A");
+###############################################################################
+# A conflict log table is system-managed and cannot be altered directly, so
+# moving it to another tablespace must be rejected.
+###############################################################################
+my $subid = $node_subscriber->safe_psql('postgres',
+ "SELECT oid FROM pg_subscription WHERE subname = 'sub_tab';");
+my $clt = "pg_conflict.pg_conflict_log_$subid";
+
+(undef, undef, $stderr) = $node_subscriber->psql('postgres',
+ "ALTER TABLE $clt SET TABLESPACE pg_default");
+like(
+ $stderr,
+ qr/cannot alter conflict log table "pg_conflict_log_\d+"/,
+ "moving a conflict log table with ALTER TABLE SET TABLESPACE is rejected");
+
+###############################################################################
+# ALTER TABLE ALL IN TABLESPACE must skip conflict log tables, the same way it
+# skips catalog and TOAST tables, instead of failing. Use an isolated database
+# so the bulk move only touches the objects created here.
+###############################################################################
+$node_subscriber->safe_psql('postgres', "CREATE DATABASE clt_ts_test");
+$node_subscriber->safe_psql('clt_ts_test',
+ "CREATE SUBSCRIPTION sub_ts_test
+ CONNECTION 'dbname=nonexistent'
+ PUBLICATION pub
+ WITH (connect=false, conflict_log_destination='table')");
+
+# A plain user table that should be moved, alongside the CLT that must not be.
+$node_subscriber->safe_psql('clt_ts_test', "CREATE TABLE user_tbl (i int)");
+
+# Create a tablespace backed by a directory inside the data dir.
+my $ts_dir = $node_subscriber->data_dir . '/backup_space';
+mkdir($ts_dir)
+ or die "could not create tablespace directory $ts_dir: $!";
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLESPACE backup_space LOCATION '$ts_dir'");
+
+# The bulk move succeeds: the user table is relocated while the CLT is skipped.
+$node_subscriber->safe_psql('clt_ts_test',
+ "ALTER TABLE ALL IN TABLESPACE pg_default SET TABLESPACE backup_space");
+
+is( $node_subscriber->safe_psql(
+ 'clt_ts_test',
+ "SELECT reltablespace <> 0 FROM pg_class WHERE relname = 'user_tbl'"),
+ 't',
+ "ALTER TABLE ALL IN TABLESPACE moves an ordinary user table");
+
+is( $node_subscriber->safe_psql(
+ 'clt_ts_test',
+ "SELECT count(*) FROM pg_class c JOIN pg_subscription s
+ ON c.relname = 'pg_conflict_log_' || s.oid
+ WHERE s.subname = 'sub_ts_test' AND c.reltablespace <> 0"),
+ '0',
+ "ALTER TABLE ALL IN TABLESPACE skips the conflict log table");
+
+# Cleanup. The subscription has no real publisher connection, so detach its
+# slot before dropping it.
+$node_subscriber->safe_psql('clt_ts_test',
+ "ALTER SUBSCRIPTION sub_ts_test DISABLE");
+$node_subscriber->safe_psql('clt_ts_test',
+ "ALTER SUBSCRIPTION sub_ts_test SET (slot_name = NONE)");
+$node_subscriber->safe_psql('clt_ts_test', "DROP SUBSCRIPTION sub_ts_test");
+$node_subscriber->safe_psql('postgres', "DROP DATABASE clt_ts_test");
+$node_subscriber->safe_psql('postgres', "DROP TABLESPACE backup_space");
+
done_testing();
ConditionalStackData
ConfigData
ConfigVariable
+ConflictLogColumnDef
+ConflictLogDest
ConflictTupleInfo
ConflictType
ConnCacheEntry
projects / thirdparty / postgresql.git / commitdiff
