appendStringInfo(buf, " relmap db %u", msg->rm.dbId);
else if (msg->id == SHAREDINVALSNAPSHOT_ID)
appendStringInfo(buf, " snapshot %u", msg->sn.relId);
+ else if (msg->id == SHAREDINVALRELSYNC_ID)
+ appendStringInfo(buf, " relsync %u", msg->rs.relid);
else
appendStringInfo(buf, " unrecognized id %d", msg->id);
}
InvokeObjectPostAlterHook(classId, objectId, 0);
+ /* Do post catalog-update tasks */
+ if (classId == PublicationRelationId)
+ {
+ Form_pg_publication pub = (Form_pg_publication) GETSTRUCT(oldtup);
+
+ /*
+ * Invalidate relsynccache entries.
+ *
+ * Unlike ALTER PUBLICATION ADD/SET/DROP commands, renaming a
+ * publication does not impact the publication status of tables. So,
+ * we don't need to invalidate relcache to rebuild the rd_pubdesc.
+ * Instead, we invalidate only the relsyncache.
+ */
+ InvalidatePubRelSyncCache(pub->oid, pub->puballtables);
+ }
+
/* Release memory */
pfree(values);
pfree(nulls);
return *invalid_column_list || *invalid_gen_col;
}
+/*
+ * Invalidate entries in the RelationSyncCache for relations included in the
+ * specified publication, either via FOR TABLE or FOR TABLES IN SCHEMA.
+ *
+ * If 'puballtables' is true, invalidate all cache entries.
+ */
+void
+InvalidatePubRelSyncCache(Oid pubid, bool puballtables)
+{
+ if (puballtables)
+ {
+ CacheInvalidateRelSyncAll();
+ }
+ else
+ {
+ List *relids = NIL;
+ List *schemarelids = NIL;
+
+ /*
+ * For partitioned tables, we must invalidate all partitions and
+ * itself. WAL records for INSERT/UPDATE/DELETE specify leaf tables as
+ * a target. However, WAL records for TRUNCATE specify both a root and
+ * its leaves.
+ */
+ relids = GetPublicationRelations(pubid,
+ PUBLICATION_PART_ALL);
+ schemarelids = GetAllSchemaPublicationRelations(pubid,
+ PUBLICATION_PART_ALL);
+
+ relids = list_concat_unique_oid(relids, schemarelids);
+
+ /* Invalidate the relsyncache */
+ foreach_oid(relid, relids)
+ CacheInvalidateRelSync(relid);
+ }
+
+ return;
+}
+
/* check_functions_in_node callback */
static bool
contain_mutable_or_user_functions_checker(Oid func_id, void *context)
CacheRegisterSyscacheCallback(PUBLICATIONOID,
publication_invalidation_cb,
(Datum) 0);
+ CacheRegisterRelSyncCallback(rel_sync_cache_relation_cb,
+ (Datum) 0);
publication_callback_registered = true;
}
publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue)
{
publications_valid = false;
-
- /*
- * Also invalidate per-relation cache so that next time the filtering info
- * is checked it will be updated with the new publication settings.
- */
- rel_sync_cache_publication_cb(arg, cacheid, hashvalue);
}
/*
#define MAX_SYSCACHE_CALLBACKS 64
#define MAX_RELCACHE_CALLBACKS 10
+#define MAX_RELSYNC_CALLBACKS 10
static struct SYSCACHECALLBACK
{
static int relcache_callback_count = 0;
+static struct RELSYNCCALLBACK
+{
+ RelSyncCallbackFunction function;
+ Datum arg;
+} relsync_callback_list[MAX_RELSYNC_CALLBACKS];
+
+static int relsync_callback_count = 0;
+
+
/* ----------------------------------------------------------------
* Invalidation subgroup support functions
* ----------------------------------------------------------------
AddInvalidationMessage(group, RelCacheMsgs, &msg);
}
+/*
+ * Add a relsync inval entry
+ *
+ * We put these into the relcache subgroup for simplicity. This message is the
+ * same as AddRelcacheInvalidationMessage() except that it is for
+ * RelationSyncCache maintained by decoding plugin pgoutput.
+ */
+static void
+AddRelsyncInvalidationMessage(InvalidationMsgsGroup *group,
+ Oid dbId, Oid relId)
+{
+ SharedInvalidationMessage msg;
+
+ /* Don't add a duplicate item. */
+ ProcessMessageSubGroup(group, RelCacheMsgs,
+ if (msg->rc.id == SHAREDINVALRELSYNC_ID &&
+ (msg->rc.relId == relId ||
+ msg->rc.relId == InvalidOid))
+ return);
+
+ /* OK, add the item */
+ msg.rc.id = SHAREDINVALRELSYNC_ID;
+ msg.rc.dbId = dbId;
+ msg.rc.relId = relId;
+ /* check AddCatcacheInvalidationMessage() for an explanation */
+ VALGRIND_MAKE_MEM_DEFINED(&msg, sizeof(msg));
+
+ AddInvalidationMessage(group, RelCacheMsgs, &msg);
+}
+
/*
* Add a snapshot inval entry
*
info->RelcacheInitFileInval = true;
}
+/*
+ * RegisterRelsyncInvalidation
+ *
+ * As above, but register a relsynccache invalidation event.
+ */
+static void
+RegisterRelsyncInvalidation(InvalidationInfo *info, Oid dbId, Oid relId)
+{
+ AddRelsyncInvalidationMessage(&info->CurrentCmdInvalidMsgs, dbId, relId);
+}
+
/*
* RegisterSnapshotInvalidation
*
ccitem->function(ccitem->arg, InvalidOid);
}
+
+ for (i = 0; i < relsync_callback_count; i++)
+ {
+ struct RELSYNCCALLBACK *ccitem = relsync_callback_list + i;
+
+ ccitem->function(ccitem->arg, InvalidOid);
+ }
}
/*
else if (msg->sn.dbId == MyDatabaseId)
InvalidateCatalogSnapshot();
}
+ else if (msg->id == SHAREDINVALRELSYNC_ID)
+ {
+ /* We only care about our own database */
+ if (msg->rs.dbId == MyDatabaseId)
+ CallRelSyncCallbacks(msg->rs.relid);
+ }
else
elog(FATAL, "unrecognized SI message ID: %d", msg->id);
}
ReleaseSysCache(tup);
}
+/*
+ * CacheInvalidateRelSync
+ * Register invalidation of the cache in logical decoding output plugin
+ * for a database.
+ *
+ * This type of invalidation message is used for the specific purpose of output
+ * plugins. Processes which do not decode WALs would do nothing even when it
+ * receives the message.
+ */
+void
+CacheInvalidateRelSync(Oid relid)
+{
+ RegisterRelsyncInvalidation(PrepareInvalidationState(),
+ MyDatabaseId, relid);
+}
+
+/*
+ * CacheInvalidateRelSyncAll
+ * Register invalidation of the whole cache in logical decoding output
+ * plugin.
+ */
+void
+CacheInvalidateRelSyncAll(void)
+{
+ CacheInvalidateRelSync(InvalidOid);
+}
/*
* CacheInvalidateSmgr
++relcache_callback_count;
}
+/*
+ * CacheRegisterRelSyncCallback
+ * Register the specified function to be called for all future
+ * relsynccache invalidation events.
+ *
+ * This function is intended to be call from the logical decoding output
+ * plugins.
+ */
+void
+CacheRegisterRelSyncCallback(RelSyncCallbackFunction func,
+ Datum arg)
+{
+ if (relsync_callback_count >= MAX_RELSYNC_CALLBACKS)
+ elog(FATAL, "out of relsync_callback_list slots");
+
+ relsync_callback_list[relsync_callback_count].function = func;
+ relsync_callback_list[relsync_callback_count].arg = arg;
+
+ ++relsync_callback_count;
+}
+
/*
* CallSyscacheCallbacks
*
}
}
+/*
+ * CallSyscacheCallbacks
+ */
+void
+CallRelSyncCallbacks(Oid relid)
+{
+ for (int i = 0; i < relsync_callback_count; i++)
+ {
+ struct RELSYNCCALLBACK *ccitem = relsync_callback_list + i;
+
+ ccitem->function(ccitem->arg, relid);
+ }
+}
+
/*
* LogLogicalInvalidations
*
char pubgencols_type,
bool *invalid_column_list,
bool *invalid_gen_col);
+extern void InvalidatePubRelSyncCache(Oid pubid, bool puballtables);
#endif /* PUBLICATIONCMDS_H */
/*
* For cache-invalidation debugging, define DISCARD_CACHES_ENABLED to enable
- * use of the debug_discard_caches GUC to aggressively flush syscache/relcache
- * entries whenever it's possible to deliver invalidations. See
- * AcceptInvalidationMessages() in src/backend/utils/cache/inval.c for
- * details.
+ * use of the debug_discard_caches GUC to aggressively flush
+ * syscache/relcache/relsynccache entries whenever it's possible to deliver
+ * invalidations. See AcceptInvalidationMessages() in
+ * src/backend/utils/cache/inval.c for details.
*
* USE_ASSERT_CHECKING builds default to enabling this. It's possible to use
* DISCARD_CACHES_ENABLED without a cassert build and the implied
* * invalidate an smgr cache entry for a specific physical relation
* * invalidate the mapped-relation mapping for a given database
* * invalidate any saved snapshot that might be used to scan a given relation
+ * * invalidate a RelationSyncCache entry for a specific relation
* More types could be added if needed. The message type is identified by
* the first "int8" field of the message struct. Zero or positive means a
* specific-catcache inval message (and also serves as the catcache ID field).
* catcache inval messages must be generated for each of its caches, since
* the hash keys will generally be different.
*
- * Catcache, relcache, and snapshot invalidations are transactional, and so
- * are sent to other backends upon commit. Internally to the generating
- * backend, they are also processed at CommandCounterIncrement so that later
- * commands in the same transaction see the new state. The generating backend
- * also has to process them at abort, to flush out any cache state it's loaded
- * from no-longer-valid entries.
+ * Catcache, relcache, relsynccache, and snapshot invalidations are
+ * transactional, and so are sent to other backends upon commit. Internally
+ * to the generating backend, they are also processed at
+ * CommandCounterIncrement so that later commands in the same transaction see
+ * the new state. The generating backend also has to process them at abort,
+ * to flush out any cache state it's loaded from no-longer-valid entries.
*
* smgr and relation mapping invalidations are non-transactional: they are
* sent immediately when the underlying file change is made.
Oid relId; /* relation ID */
} SharedInvalSnapshotMsg;
+#define SHAREDINVALRELSYNC_ID (-6)
+
+typedef struct
+{
+ int8 id; /* type field --- must be first */
+ Oid dbId; /* database ID */
+ Oid relid; /* relation ID, or 0 if whole
+ * RelationSyncCache */
+} SharedInvalRelSyncMsg;
+
typedef union
{
int8 id; /* type field --- must be first */
SharedInvalSmgrMsg sm;
SharedInvalRelmapMsg rm;
SharedInvalSnapshotMsg sn;
+ SharedInvalRelSyncMsg rs;
} SharedInvalidationMessage;
typedef void (*SyscacheCallbackFunction) (Datum arg, int cacheid, uint32 hashvalue);
typedef void (*RelcacheCallbackFunction) (Datum arg, Oid relid);
+typedef void (*RelSyncCallbackFunction) (Datum arg, Oid relid);
extern void AcceptInvalidationMessages(void);
extern void CacheInvalidateRelcacheByRelid(Oid relid);
+extern void CacheInvalidateRelSync(Oid relid);
+
+extern void CacheInvalidateRelSyncAll(void);
+
extern void CacheInvalidateSmgr(RelFileLocatorBackend rlocator);
extern void CacheInvalidateRelmap(Oid databaseId);
extern void CacheRegisterRelcacheCallback(RelcacheCallbackFunction func,
Datum arg);
+extern void CacheRegisterRelSyncCallback(RelSyncCallbackFunction func,
+ Datum arg);
+
extern void CallSyscacheCallbacks(int cacheid, uint32 hashvalue);
+extern void CallRelSyncCallbacks(Oid relid);
+
extern void InvalidateSystemCaches(void);
extern void InvalidateSystemCachesExtended(bool debug_discard);
"Alter subscription set publication throws warning for non-existent publication"
);
+# Cleanup
+$node_publisher->safe_psql('postgres', qq[
+ DROP PUBLICATION mypub;
+ SELECT pg_drop_replication_slot('mysub');
+]);
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION mysub1");
+
+#
+# Test ALTER PUBLICATION RENAME command during the replication
+#
+
+# Test function for swaping name of publications
+sub test_swap
+{
+ my ($table_name, $pubname, $appname) = @_;
+
+ # Confirms tuples can be replicated
+ $node_publisher->safe_psql('postgres', "INSERT INTO $table_name VALUES (1);");
+ $node_publisher->wait_for_catchup($appname);
+ my $result =
+ $node_subscriber->safe_psql('postgres', "SELECT a FROM $table_name");
+ is($result, qq(1), 'check replication worked well before renaming a publication');
+
+ # Swap the name of publications; $pubname <-> pub_empty
+ $node_publisher->safe_psql('postgres', qq[
+ ALTER PUBLICATION $pubname RENAME TO tap_pub_tmp;
+ ALTER PUBLICATION pub_empty RENAME TO $pubname;
+ ALTER PUBLICATION tap_pub_tmp RENAME TO pub_empty;
+ ]);
+
+ # Insert the data again
+ $node_publisher->safe_psql('postgres', "INSERT INTO $table_name VALUES (2);");
+ $node_publisher->wait_for_catchup($appname);
+
+ # Confirms the second tuple won't be replicated because $pubname does not
+ # contains relations anymore.
+ $result =
+ $node_subscriber->safe_psql('postgres', "SELECT a FROM $table_name ORDER BY a");
+ is($result, qq(1),
+ 'check the tuple inserted after the RENAME was not replicated');
+
+ # Restore the name of publications because it can be called several times
+ $node_publisher->safe_psql('postgres', qq[
+ ALTER PUBLICATION $pubname RENAME TO tap_pub_tmp;
+ ALTER PUBLICATION pub_empty RENAME TO $pubname;
+ ALTER PUBLICATION tap_pub_tmp RENAME TO pub_empty;
+ ]);
+}
+
+# Create another table
+$ddl = "CREATE TABLE test2 (a int, b text);";
+$node_publisher->safe_psql('postgres', $ddl);
+$node_subscriber->safe_psql('postgres', $ddl);
+
+# Create publications and a subscription
+$node_publisher->safe_psql('postgres', qq[
+ CREATE PUBLICATION pub_empty;
+ CREATE PUBLICATION pub_for_tab FOR TABLE test1;
+ CREATE PUBLICATION pub_for_all_tables FOR ALL TABLES;
+]);
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION pub_for_tab WITH (copy_data = off)"
+);
+
+# Confirms RENAME command works well for a publication
+test_swap('test1', 'pub_for_tab', 'tap_sub');
+
+# Switches a publication which includes all tables
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub SET PUBLICATION pub_for_all_tables WITH (refresh = true, copy_data = false);"
+);
+
+# Confirms RENAME command works well for ALL TABLES publication
+test_swap('test2', 'pub_for_all_tables', 'tap_sub');
+
+# Cleanup
+$node_publisher->safe_psql('postgres', qq[
+ DROP PUBLICATION pub_empty, pub_for_tab, pub_for_all_tables;
+ DROP TABLE test1, test2;
+]);
+$node_subscriber->safe_psql('postgres', qq[
+ DROP SUBSCRIPTION tap_sub;
+ DROP TABLE test1, test2;
+]);
+
$node_subscriber->stop;
$node_publisher->stop;
projects / postgresql.git / commitdiff