Loading...
Searching...
No Matches
subscriptioncmds.c File Reference
#include "postgres.h"#include "access/commit_ts.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/indexing.h"#include "catalog/namespace.h"#include "catalog/objectaccess.h"#include "catalog/objectaddress.h"#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 "catalog/pg_user_mapping.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"#include "nodes/makefuncs.h"#include "pgstat.h"#include "replication/logicallauncher.h"#include "replication/logicalworker.h"#include "replication/origin.h"#include "replication/slot.h"#include "replication/walreceiver.h"#include "replication/walsender.h"#include "replication/worker_internal.h"#include "storage/lmgr.h"#include "utils/acl.h"#include "utils/builtins.h"#include "utils/guc.h"#include "utils/lsyscache.h"#include "utils/memutils.h"#include "utils/pg_lsn.h"#include "utils/syscache.h"
Include dependency graph for subscriptioncmds.c:
Go to the source code of this file.
Data Structures | |
| struct | SubOpts |
| struct | PublicationRelKind |
Macros | |
| #define | SUBOPT_CONNECT 0x00000001 |
| #define | SUBOPT_ENABLED 0x00000002 |
| #define | SUBOPT_CREATE_SLOT 0x00000004 |
| #define | SUBOPT_SLOT_NAME 0x00000008 |
| #define | SUBOPT_COPY_DATA 0x00000010 |
| #define | SUBOPT_SYNCHRONOUS_COMMIT 0x00000020 |
| #define | SUBOPT_REFRESH 0x00000040 |
| #define | SUBOPT_BINARY 0x00000080 |
| #define | SUBOPT_STREAMING 0x00000100 |
| #define | SUBOPT_TWOPHASE_COMMIT 0x00000200 |
| #define | SUBOPT_DISABLE_ON_ERR 0x00000400 |
| #define | SUBOPT_PASSWORD_REQUIRED 0x00000800 |
| #define | SUBOPT_RUN_AS_OWNER 0x00001000 |
| #define | SUBOPT_FAILOVER 0x00002000 |
| #define | SUBOPT_RETAIN_DEAD_TUPLES 0x00004000 |
| #define | SUBOPT_MAX_RETENTION_DURATION 0x00008000 |
| #define | SUBOPT_WAL_RECEIVER_TIMEOUT 0x00010000 |
| #define | SUBOPT_LSN 0x00020000 |
| #define | SUBOPT_ORIGIN 0x00040000 |
| #define | SUBOPT_CONFLICT_LOG_DEST 0x00080000 |
| #define | IsSet(val, bits) (((val) & (bits)) == (bits)) |
| #define | appendQuotedIdentifier(b, s) appendQuotedString(b, s, '"') |
| #define | appendQuotedLiteral(b, s) appendQuotedString(b, s, '\'') |
Typedefs | |
| typedef struct SubOpts | SubOpts |
| typedef struct PublicationRelKind | PublicationRelKind |
Macro Definition Documentation
◆ appendQuotedIdentifier
| #define appendQuotedIdentifier | ( | b, | |
| s | |||
| ) | appendQuotedString(b, s, '"') |
Definition at line 567 of file subscriptioncmds.c.
◆ appendQuotedLiteral
| #define appendQuotedLiteral | ( | b, | |
| s | |||
| ) | appendQuotedString(b, s, '\'') |
Definition at line 568 of file subscriptioncmds.c.
◆ IsSet
Definition at line 87 of file subscriptioncmds.c.
◆ SUBOPT_BINARY
| #define SUBOPT_BINARY 0x00000080 |
Definition at line 72 of file subscriptioncmds.c.
◆ SUBOPT_CONFLICT_LOG_DEST
| #define SUBOPT_CONFLICT_LOG_DEST 0x00080000 |
Definition at line 84 of file subscriptioncmds.c.
◆ SUBOPT_CONNECT
| #define SUBOPT_CONNECT 0x00000001 |
Definition at line 65 of file subscriptioncmds.c.
◆ SUBOPT_COPY_DATA
| #define SUBOPT_COPY_DATA 0x00000010 |
Definition at line 69 of file subscriptioncmds.c.
◆ SUBOPT_CREATE_SLOT
| #define SUBOPT_CREATE_SLOT 0x00000004 |
Definition at line 67 of file subscriptioncmds.c.
◆ SUBOPT_DISABLE_ON_ERR
| #define SUBOPT_DISABLE_ON_ERR 0x00000400 |
Definition at line 75 of file subscriptioncmds.c.
◆ SUBOPT_ENABLED
| #define SUBOPT_ENABLED 0x00000002 |
Definition at line 66 of file subscriptioncmds.c.
◆ SUBOPT_FAILOVER
| #define SUBOPT_FAILOVER 0x00002000 |
Definition at line 78 of file subscriptioncmds.c.
◆ SUBOPT_LSN
| #define SUBOPT_LSN 0x00020000 |
Definition at line 82 of file subscriptioncmds.c.
◆ SUBOPT_MAX_RETENTION_DURATION
| #define SUBOPT_MAX_RETENTION_DURATION 0x00008000 |
Definition at line 80 of file subscriptioncmds.c.
◆ SUBOPT_ORIGIN
| #define SUBOPT_ORIGIN 0x00040000 |
Definition at line 83 of file subscriptioncmds.c.
◆ SUBOPT_PASSWORD_REQUIRED
| #define SUBOPT_PASSWORD_REQUIRED 0x00000800 |
Definition at line 76 of file subscriptioncmds.c.
◆ SUBOPT_REFRESH
| #define SUBOPT_REFRESH 0x00000040 |
Definition at line 71 of file subscriptioncmds.c.
◆ SUBOPT_RETAIN_DEAD_TUPLES
| #define SUBOPT_RETAIN_DEAD_TUPLES 0x00004000 |
Definition at line 79 of file subscriptioncmds.c.
◆ SUBOPT_RUN_AS_OWNER
| #define SUBOPT_RUN_AS_OWNER 0x00001000 |
Definition at line 77 of file subscriptioncmds.c.
◆ SUBOPT_SLOT_NAME
| #define SUBOPT_SLOT_NAME 0x00000008 |
Definition at line 68 of file subscriptioncmds.c.
◆ SUBOPT_STREAMING
| #define SUBOPT_STREAMING 0x00000100 |
Definition at line 73 of file subscriptioncmds.c.
◆ SUBOPT_SYNCHRONOUS_COMMIT
| #define SUBOPT_SYNCHRONOUS_COMMIT 0x00000020 |
Definition at line 70 of file subscriptioncmds.c.
◆ SUBOPT_TWOPHASE_COMMIT
| #define SUBOPT_TWOPHASE_COMMIT 0x00000200 |
Definition at line 74 of file subscriptioncmds.c.
◆ SUBOPT_WAL_RECEIVER_TIMEOUT
| #define SUBOPT_WAL_RECEIVER_TIMEOUT 0x00010000 |
Definition at line 81 of file subscriptioncmds.c.
Typedef Documentation
◆ PublicationRelKind
◆ SubOpts
Function Documentation
◆ alter_sub_conflict_log_dest()
|
static |
Definition at line 1516 of file subscriptioncmds.c.
1519{
1524
1527
1529 {
1530 /* There is a conflict log table already. */
1532 {
1534 sub->conflictlogrelid);
1536 }
1537 }
1538 else
1539 {
1540 /* There was no previous conflict log table. */
1542 {
1545
1548
1549 /*
1550 * Establish an internal dependency between the conflict log table
1551 * and the subscription. For details refer comments in
1552 * CreateSubscription function.
1553 */
1557 }
1558 }
1559
1560 *conflicttablerelid = relid;
1562}
Oid create_conflict_log_table(Oid subid, char *subname, Oid subowner)
Definition conflict.c:147
void recordDependencyOn(const ObjectAddress *depender, const ObjectAddress *referenced, DependencyType behavior)
Definition pg_depend.c:51
Definition objectaddress.h:26
static void drop_sub_conflict_log_table(Oid subid, char *subname, Oid subconflictlogrelid)
Definition subscriptioncmds.c:2469
References Subscription::conflictlogrelid, CONFLICTS_LOGGED_TO_TABLE, create_conflict_log_table(), DEPENDENCY_INTERNAL, drop_sub_conflict_log_table(), fb(), InvalidOid, Subscription::name, ObjectAddressSet, Subscription::oid, Subscription::owner, and recordDependencyOn().
Referenced by AlterSubscription().
◆ AlterSubscription()
| ObjectAddress AlterSubscription | ( | ParseState * | pstate, |
| AlterSubscriptionStmt * | stmt, | ||
| bool | isTopLevel | ||
| ) |
Definition at line 1568 of file subscriptioncmds.c.
1570{
1571 Relation rel;
1577 Oid subid;
1587 char *origin;
1588 Subscription *sub;
1592
1594
1595 /* Fetch the existing tuple. */
1598
1603 stmt->subname)));
1604
1606 subid = form->oid;
1607
1608 /* must be owner */
1611 stmt->subname);
1612
1613 /* parse and check options */
1615 {
1620 SUBOPT_DISABLE_ON_ERR |
1621 SUBOPT_PASSWORD_REQUIRED |
1623 SUBOPT_RETAIN_DEAD_TUPLES |
1625 SUBOPT_WAL_RECEIVER_TIMEOUT |
1626 SUBOPT_ORIGIN |
1627 SUBOPT_CONFLICT_LOG_DEST);
1628 break;
1629
1632 break;
1633
1636 break;
1637
1641 break;
1642
1645 break;
1646
1649 break;
1650
1651 default:
1652 supported_opts = 0;
1653 break;
1654 }
1655
1658
1659 /*
1660 * Ensure that ALTER SUBSCRIPTION commands that could be used to fix a
1661 * broken connection or prepare to drop a broken subscription don't
1662 * attempt to construct the conninfo. Otherwise, we might encounter the
1663 * error the user is trying to fix.
1664 *
1665 * Specifically, ALTER SUBSCRIPTION DISABLE, ALTER SUBSCRIPTION SERVER,
1666 * ALTER SUBSCRIPTION CONNECTION, or ALTER SUBSCRIPTION SET
1667 * (slot_name=NONE).
1668 *
1669 * NB: if the user specifies multiple SET options, then we may still need
1670 * to construct conninfo even if slot_name is set to NONE.
1671 */
1673 {
1676 }
1679 {
1681 }
1683 {
1684 /* ... SET (slot_name = NONE) with no other options */
1687 }
1688
1689 /*
1690 * Skip ACL checks on the subscription's foreign server, if any. If
1691 * changing the server (or replacing it with a raw connection), then the
1692 * old one will be removed anyway. If changing something unrelated,
1693 * there's no need to do an additional ACL check here; that will be done
1694 * by the subscription worker.
1695 */
1697
1699 origin = sub->origin;
1702
1703 /*
1704 * Don't allow non-superuser modification of a subscription with
1705 * password_required=false.
1706 */
1711 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
1712
1713 /* Lock the subscription so nobody else can do anything with it. */
1715
1716 /* Form a new tuple. */
1720
1722
1724 {
1726 {
1728 {
1729 /*
1730 * The subscription must be disabled to allow slot_name as
1731 * 'none', otherwise, the apply worker will repeatedly try
1732 * to stream the data using that slot_name which neither
1733 * exists on the publisher nor the user will be allowed to
1734 * create it.
1735 */
1740 "slot_name = NONE")));
1741
1745 else
1748 }
1749
1751 {
1755 }
1756
1758 {
1762 }
1763
1765 {
1769 }
1770
1772 {
1776 = true;
1777 }
1778
1780 {
1781 /* Non-superuser may not disable password_required. */
1786 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
1787
1791 = true;
1792 }
1793
1795 {
1799 }
1800
1802 {
1803 /*
1804 * We need to update both the slot and the subscription
1805 * for the two_phase option. We can enable the two_phase
1806 * option for a slot only once the initial data
1807 * synchronization is done. This is to avoid missing some
1808 * data as explained in comments atop worker.c.
1809 */
1811
1813 isTopLevel);
1814
1815 /*
1816 * Modifying the two_phase slot option requires a slot
1817 * lookup by slot name, so changing the slot name at the
1818 * same time is not allowed.
1819 */
1825
1826 /*
1827 * Note that workers may still survive even if the
1828 * subscription has been disabled.
1829 *
1830 * Ensure workers have already been exited to avoid
1831 * getting prepared transactions while we are disabling
1832 * the two_phase option. Otherwise, the changes of an
1833 * already prepared transaction can be replicated again
1834 * along with its corresponding commit, leading to
1835 * duplicate data or errors.
1836 */
1842
1843 /*
1844 * two_phase cannot be disabled if there are any
1845 * uncommitted prepared transactions present otherwise it
1846 * can lead to duplicate data or errors as explained in
1847 * the comment above.
1848 */
1851 LookupGXactBySubid(subid))
1856
1857 /* Change system catalog accordingly */
1863 }
1864
1866 {
1867 /*
1868 * Similar to the two_phase case above, we need to update
1869 * the failover option for both the slot and the
1870 * subscription.
1871 */
1873
1875 isTopLevel);
1876
1880 }
1881
1883 {
1887
1888 /*
1889 * Update the retention status only if there's a change in
1890 * the retain_dead_tuples option value.
1891 *
1892 * Automatically marking retention as active when
1893 * retain_dead_tuples is enabled may not always be ideal,
1894 * especially if retention was previously stopped and the
1895 * user toggles retain_dead_tuples without adjusting the
1896 * publisher workload. However, this behavior provides a
1897 * convenient way for users to manually refresh the
1898 * retention status. Since retention will be stopped again
1899 * unless the publisher workload is reduced, this approach
1900 * is acceptable for now.
1901 */
1903 {
1907
1909 }
1910
1912
1913 /*
1914 * Workers may continue running even after the
1915 * subscription has been disabled.
1916 *
1917 * To prevent race conditions (as described in
1918 * CheckAlterSubOption()), ensure that all worker
1919 * processes have already exited before proceeding.
1920 */
1926
1927 /*
1928 * Notify the launcher to manage the replication slot for
1929 * conflict detection. This ensures that replication slot
1930 * is efficiently handled (created, updated, or dropped)
1931 * in response to any configuration changes.
1932 */
1933 ApplyLauncherWakeupAtCommit();
1934
1937 }
1938
1940 {
1944
1946 }
1947
1948 /*
1949 * Ensure that system configuration parameters are set
1950 * appropriately to support retain_dead_tuples and
1951 * max_retention_duration.
1952 */
1956 retain_dead_tuples,
1957 retention_active,
1958 (max_retention > 0));
1959
1961 {
1965
1966 /*
1967 * Check if changes from different origins may be received
1968 * from the publisher when the origin is changed to ANY
1969 * and retain_dead_tuples is enabled. Use |= so that we
1970 * don't clear the flag already set when
1971 * retain_dead_tuples was changed in the same command.
1972 */
1975
1976 origin = opts.origin;
1977 }
1978
1980 {
1984 }
1985
1987 {
1990
1992 {
1995
1999
2001 old_dest,
2002 opts.conflictlogdest,
2003 &relid);
2005 {
2007 ObjectIdGetDatum(relid);
2009 true;
2010 }
2011 }
2012 }
2013
2015 break;
2016 }
2017
2019 {
2021
2026
2027 /*
2028 * Check track_commit_timestamp only when enabling the
2029 * subscription in case it was disabled after creation. See
2030 * comments atop CheckSubDeadTupleRetention() for details.
2031 */
2035
2039
2041 ApplyLauncherWakeupAtCommit();
2042
2044
2045 /*
2046 * The subscription might be initially created with
2047 * connect=false and retain_dead_tuples=true, meaning the
2048 * remote server's status may not be checked. Ensure this
2049 * check is conducted now.
2050 */
2052 break;
2053 }
2054
2056 {
2060
2061 /*
2062 * Remove what was there before, either another foreign server
2063 * or a connection string.
2064 */
2066 {
2068 DEPENDENCY_NORMAL,
2070 }
2071 else
2072 {
2075 }
2076
2077 /*
2078 * Check that the subscription owner has USAGE privileges on
2079 * the server.
2080 */
2083 new_server->serverid,
2090 new_server->servername));
2091
2092 /* make sure a user mapping exists */
2094
2096 new_server);
2097
2098 /* Load the library providing us libpq calls. */
2100 /* Check the connection info string. */
2103
2106
2109
2111 }
2112
2113 /*
2114 * Since the remote server configuration might have changed,
2115 * perform a check to ensure it permits enabling
2116 * retain_dead_tuples.
2117 */
2119 break;
2120
2122 /* remove reference to foreign server and dependencies, if present */
2124 {
2126 DEPENDENCY_NORMAL,
2128
2131 }
2132
2134
2135 /* Load the library providing us libpq calls. */
2137 /* Check the connection info string. */
2140
2145
2146 /*
2147 * Since the remote server configuration might have changed,
2148 * perform a check to ensure it permits enabling
2149 * retain_dead_tuples.
2150 */
2152 break;
2153
2155 {
2159
2161
2162 /* Refresh if user asked us to. */
2164 {
2170
2171 /*
2172 * See ALTER_SUBSCRIPTION_REFRESH_PUBLICATION for details
2173 * why this is not allowed.
2174 */
2178 errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when two_phase is enabled"),
2179 errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION with refresh = false, or with copy_data = false, or use DROP/CREATE SUBSCRIPTION.")));
2180
2182
2183 /* Make sure refresh sees the new list of publications. */
2185
2187 stmt->publication);
2188 }
2189
2190 break;
2191 }
2192
2195 {
2198
2203
2205
2206 /* Refresh if user asked us to. */
2208 {
2209 /* We only need to validate user specified publications. */
2211
2216 /* translator: %s is an SQL ALTER command */
2218 isadd ?
2219 "ALTER SUBSCRIPTION ... ADD PUBLICATION ... WITH (refresh = false)" :
2220 "ALTER SUBSCRIPTION ... DROP PUBLICATION ... WITH (refresh = false)")));
2221
2222 /*
2223 * See ALTER_SUBSCRIPTION_REFRESH_PUBLICATION for details
2224 * why this is not allowed.
2225 */
2229 errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when two_phase is enabled"),
2230 /* translator: %s is an SQL ALTER command */
2231 errhint("Use %s with refresh = false, or with copy_data = false, or use DROP/CREATE SUBSCRIPTION.",
2232 isadd ?
2233 "ALTER SUBSCRIPTION ... ADD PUBLICATION" :
2234 "ALTER SUBSCRIPTION ... DROP PUBLICATION")));
2235
2237
2238 /* Refresh the new list of publications. */
2240
2242 validate_publications);
2243 }
2244
2245 break;
2246 }
2247
2249 {
2254 "ALTER SUBSCRIPTION ... REFRESH PUBLICATION")));
2255
2256 /*
2257 * The subscription option "two_phase" requires that
2258 * replication has passed the initial table synchronization
2259 * phase before the two_phase becomes properly enabled.
2260 *
2261 * But, having reached this two-phase commit "enabled" state
2262 * we must not allow any subsequent table initialization to
2263 * occur. So the ALTER SUBSCRIPTION ... REFRESH PUBLICATION is
2264 * disallowed when the user had requested two_phase = on mode.
2265 *
2266 * The exception to this restriction is when copy_data =
2267 * false, because when copy_data is false the tablesync will
2268 * start already in READY state and will exit directly without
2269 * doing anything.
2270 *
2271 * For more details see comments atop worker.c.
2272 */
2276 errmsg("ALTER SUBSCRIPTION ... REFRESH PUBLICATION with copy_data is not allowed when two_phase is enabled"),
2277 errhint("Use ALTER SUBSCRIPTION ... REFRESH PUBLICATION with copy_data = false, or use DROP/CREATE SUBSCRIPTION.")));
2278
2280
2282
2283 break;
2284 }
2285
2287 {
2292 "ALTER SUBSCRIPTION ... REFRESH SEQUENCES"));
2293
2294 AlterSubscription_refresh_seq(sub);
2295
2296 break;
2297 }
2298
2300 {
2301 /* ALTER SUBSCRIPTION ... SKIP supports only LSN option */
2303
2304 /*
2305 * If the user sets subskiplsn, we do a sanity check to make
2306 * sure that the specified LSN is a probable value.
2307 */
2309 {
2312 XLogRecPtr remote_lsn;
2313
2318
2319 /* Check the given LSN is at least a future LSN */
2325 LSN_FORMAT_ARGS(remote_lsn))));
2326 }
2327
2330
2332 break;
2333 }
2334
2335 default:
2337 stmt->kind);
2338 }
2339
2340 /* Update the catalog if needed. */
2342 {
2344 replaces);
2345
2347
2349 }
2350
2351 /*
2352 * Try to acquire the connection necessary either for modifying the slot
2353 * or for checking if the remote server permits enabling
2354 * retain_dead_tuples.
2355 *
2356 * This has to be at the end because otherwise if there is an error while
2357 * doing the database operations we won't be able to rollback altered
2358 * slot.
2359 */
2361 {
2365
2367
2368 /* Load the library providing us libpq calls. */
2370
2371 /*
2372 * Try to connect to the publisher, using the new connection string if
2373 * available.
2374 */
2378 &err);
2384
2385 PG_TRY();
2386 {
2389
2392 sub->name);
2393
2398 }
2399 PG_FINALLY();
2400 {
2402 }
2403 PG_END_TRY();
2404 }
2405
2407
2409
2410 /* Wake up related replication workers to handle this change quickly. */
2411 LogicalRepWorkersWakeupAtCommit(subid);
2412
2414}
void aclcheck_error(AclResult aclerr, ObjectType objtype, const char *objectname)
Definition aclchk.c:2672
AclResult object_aclcheck(Oid classid, Oid objectid, Oid roleid, AclMode mode)
Definition aclchk.c:3902
bool object_ownercheck(Oid classid, Oid objectid, Oid roleid)
Definition aclchk.c:4156
void LogicalRepWorkersWakeupAtCommit(Oid subid)
Definition worker.c:6334
void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid, char *originname, Size szoriginname)
Definition worker.c:648
int errhint(const char *fmt,...) pg_attribute_printf(1
ForeignServer * GetForeignServerByName(const char *srvname, bool missing_ok)
Definition foreign.c:185
char * ForeignServerConnectionString(Oid userid, ForeignServer *server)
Definition foreign.c:202
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, const Datum *replValues, const bool *replIsnull, const bool *doReplace)
Definition heaptuple.c:1118
void CatalogTupleUpdate(Relation heapRel, const ItemPointerData *otid, HeapTuple tup)
Definition indexing.c:313
List * logicalrep_workers_find(Oid subid, bool only_running, bool acquire_lock)
Definition launcher.c:303
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition lmgr.c:1088
#define InvokeObjectPostAlterHook(classId, objectId, subId)
Definition objectaccess.h:197
ReplOriginId replorigin_by_name(const char *roname, bool missing_ok)
Definition origin.c:243
XLogRecPtr replorigin_get_progress(ReplOriginId node, bool flush)
Definition origin.c:1057
@ ALTER_SUBSCRIPTION_REFRESH_PUBLICATION
Definition parsenodes.h:4567
@ ALTER_SUBSCRIPTION_DROP_PUBLICATION
Definition parsenodes.h:4566
@ ALTER_SUBSCRIPTION_REFRESH_SEQUENCES
Definition parsenodes.h:4568
long deleteDependencyRecordsForSpecific(Oid classId, Oid objectId, char deptype, Oid refclassId, Oid refobjectId)
Definition pg_depend.c:411
Subscription * GetSubscription(Oid subid, bool missing_ok, bool conninfo_needed, bool conninfo_aclcheck)
Definition pg_subscription.c:88
END_CATALOG_STRUCT typedef FormData_pg_subscription * Form_pg_subscription
Definition pg_subscription.h:129
Definition foreign.h:36
Definition htup.h:63
Definition pg_list.h:54
Definition rel.h:56
Definition subscriptioncmds.c:94
Definition pg_subscription.h:140
Definition libpqwalreceiver.c:46
static void CheckAlterSubOption(Subscription *sub, const char *option, bool slot_needs_update, bool isTopLevel)
Definition subscriptioncmds.c:1413
static Datum publicationListToArray(List *publist)
Definition subscriptioncmds.c:639
static bool alter_sub_conflict_log_dest(Subscription *sub, ConflictLogDest oldlogdest, ConflictLogDest newlogdest, Oid *conflicttablerelid)
Definition subscriptioncmds.c:1516
static void AlterSubscription_refresh(Subscription *sub, bool copy_data, List *validate_publications)
Definition subscriptioncmds.c:1089
static void parse_subscription_options(ParseState *pstate, List *stmt_options, uint32 supported_opts, SubOpts *opts)
Definition subscriptioncmds.c:161
void CheckSubDeadTupleRetention(bool check_guc, bool sub_disabled, int elevel_for_sub_disabled, bool retain_dead_tuples, bool retention_active, bool max_retention_set)
Definition subscriptioncmds.c:3357
static void AlterSubscription_refresh_seq(Subscription *sub)
Definition subscriptioncmds.c:1360
static List * merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname)
Definition subscriptioncmds.c:3629
static void check_publications_origin_tables(WalReceiverConn *wrconn, List *publications, bool copydata, bool retain_dead_tuples, char *origin, Oid *subrel_local_oids, int subrel_count, char *subname)
Definition subscriptioncmds.c:3031
static void check_pub_dead_tuple_retention(WalReceiverConn *wrconn)
Definition subscriptioncmds.c:3296
#define walrcv_connect(conninfo, replication, logical, must_use_password, appname, err)
Definition walreceiver.h:436
#define walrcv_check_conninfo(conninfo, must_use_password)
Definition walreceiver.h:438
#define walrcv_alter_slot(conn, slotname, failover, two_phase)
Definition walreceiver.h:462
void PreventInTransactionBlock(bool isTopLevel, const char *stmtType)
Definition xact.c:3701
References AccessExclusiveLock, ACL_USAGE, aclcheck_error(), ACLCHECK_NOT_OWNER, ACLCHECK_OK, alter_sub_conflict_log_dest(), ALTER_SUBSCRIPTION_ADD_PUBLICATION, ALTER_SUBSCRIPTION_CONNECTION, ALTER_SUBSCRIPTION_DROP_PUBLICATION, ALTER_SUBSCRIPTION_ENABLED, ALTER_SUBSCRIPTION_OPTIONS, ALTER_SUBSCRIPTION_REFRESH_PUBLICATION, ALTER_SUBSCRIPTION_REFRESH_SEQUENCES, ALTER_SUBSCRIPTION_SERVER, ALTER_SUBSCRIPTION_SET_PUBLICATION, ALTER_SUBSCRIPTION_SKIP, AlterSubscription_refresh(), AlterSubscription_refresh_seq(), ApplyLauncherWakeupAtCommit(), Assert, BoolGetDatum(), CatalogTupleUpdate(), CharGetDatum(), check_pub_dead_tuple_retention(), check_publications_origin_tables(), CheckAlterSubOption(), CheckSubDeadTupleRetention(), Subscription::conflictlogdest, ConflictLogDestNames, Subscription::conninfo, CStringGetDatum(), CStringGetTextDatum, deleteDependencyRecordsForSpecific(), DEPENDENCY_NORMAL, DirectFunctionCall1, elog, Subscription::enabled, ereport, err(), errcode(), errhint(), errmsg, ERROR, fb(), ForeignServerConnectionString(), Form_pg_subscription, GetConflictLogDest(), GetForeignServerByName(), GETSTRUCT(), GetSubscription(), GetUserId(), GetUserMapping(), GetUserNameFromId(), heap_freetuple(), heap_modify_tuple(), HeapTupleIsValid, Int32GetDatum(), InvalidOid, InvokeObjectPostAlterHook, IsSet, load_file(), LockSharedObject(), logicalrep_workers_find(), LogicalRepWorkersWakeupAtCommit(), LookupGXactBySubid(), LSN_FORMAT_ARGS, LSNGetDatum(), Subscription::maxretention, merge_publications(), MyDatabaseId, Subscription::name, NAMEDATALEN, namein(), NOTICE, object_aclcheck(), object_ownercheck(), OBJECT_SUBSCRIPTION, ObjectAddressSet, ObjectIdGetDatum(), opts, Subscription::origin, Subscription::ownersuperuser, parse_subscription_options(), Subscription::passwordrequired, PG_END_TRY, PG_FINALLY, pg_strcasecmp(), PG_TRY, PreventInTransactionBlock(), publicationListToArray(), Subscription::publications, recordDependencyOn(), RelationGetDescr, ReplicationOriginNameForLogicalRep(), replorigin_by_name(), replorigin_get_progress(), Subscription::retaindeadtuples, Subscription::retentionactive, RowExclusiveLock, SearchSysCacheCopy2, Subscription::slotname, stmt, SUBOPT_BINARY, SUBOPT_CONFLICT_LOG_DEST, SUBOPT_COPY_DATA, SUBOPT_DISABLE_ON_ERR, SUBOPT_ENABLED, SUBOPT_FAILOVER, SUBOPT_LSN, SUBOPT_MAX_RETENTION_DURATION, SUBOPT_ORIGIN, SUBOPT_PASSWORD_REQUIRED, SUBOPT_REFRESH, SUBOPT_RETAIN_DEAD_TUPLES, SUBOPT_RUN_AS_OWNER, SUBOPT_SLOT_NAME, SUBOPT_STREAMING, SUBOPT_SYNCHRONOUS_COMMIT, SUBOPT_TWOPHASE_COMMIT, SUBOPT_WAL_RECEIVER_TIMEOUT, superuser(), table_close(), table_open(), Subscription::twophasestate, values, walrcv_alter_slot, walrcv_check_conninfo, walrcv_connect, walrcv_disconnect, WARNING, wrconn, and XLogRecPtrIsValid.
Referenced by ProcessUtilitySlow().
◆ AlterSubscription_refresh()
|
static |
Definition at line 1089 of file subscriptioncmds.c.
1091{
1101 int off;
1106 {
1107 Oid relid;
1109 } SubRemoveRels;
1110
1113
1114 /* Load the library providing us libpq calls. */
1116
1117 /* Try to connect to the publisher. */
1126
1127 PG_TRY();
1128 {
1131
1132 /* Get the relation list from publisher. */
1134
1135 /* Get local relation list. */
1138
1139 /*
1140 * Build qsorted arrays of local table oids and sequence oids for
1141 * faster lookup. This can potentially contain all tables and
1142 * sequences in the database so speed of lookup is important.
1143 *
1144 * We do not yet know the exact count of tables and sequences, so we
1145 * allocate separate arrays for table OIDs and sequence OIDs based on
1146 * the total number of relations (subrel_count).
1147 */
1151 {
1153
1156 else
1158 }
1159
1164 sub->name);
1165
1168 copy_data, sub->origin,
1170 sub->name);
1171
1172 /*
1173 * Walk over the remote relations and try to match them to locally
1174 * known relations. If the relation is not known locally create a new
1175 * state for it.
1176 *
1177 * Also builds array of local oids of remote relations for the next
1178 * step.
1179 */
1180 off = 0;
1182
1184 {
1186 Oid relid;
1187 char relkind;
1188
1190 relkind = get_rel_relkind(relid);
1191
1192 /* Check for supported relkind. */
1195
1196 pubrel_local_oids[off++] = relid;
1197
1202 {
1210 }
1211 }
1212
1213 /*
1214 * Next remove state for tables we should not care about anymore using
1215 * the data we collected above
1216 */
1218
1220 {
1222
1225 {
1227 XLogRecPtr statelsn;
1229
1230 /*
1231 * Lock pg_subscription_rel with AccessExclusiveLock to
1232 * prevent any race conditions with the apply worker
1233 * re-launching workers at the same time this code is trying
1234 * to remove those tables.
1235 *
1236 * Even if new worker for this particular rel is restarted it
1237 * won't be able to make any progress as we hold exclusive
1238 * lock on pg_subscription_rel till the transaction end. It
1239 * will simply exit as there is no corresponding rel entry.
1240 *
1241 * This locking also ensures that the state of rels won't
1242 * change till we are done with this refresh operation.
1243 */
1244 if (!rel)
1246
1247 /* Last known rel state. */
1249
1251
1252 remove_rel->relid = relid;
1254
1256
1258
1259 /*
1260 * For READY state, we would have already dropped the
1261 * tablesync origin.
1262 */
1264 {
1266
1267 /*
1268 * Drop the tablesync's origin tracking if exists.
1269 *
1270 * It is possible that the origin is not yet created for
1271 * tablesync worker, this can happen for the states before
1272 * SUBREL_STATE_DATASYNC. The tablesync worker or apply
1273 * worker can also concurrently try to drop the origin and
1274 * by this time the origin might be already removed. For
1275 * these reasons, passing missing_ok = true.
1276 */
1280 }
1281
1285 get_rel_name(relid),
1286 sub->name)));
1287 }
1288 }
1289
1290 /*
1291 * Drop the tablesync slots associated with removed tables. This has
1292 * to be at the end because otherwise if there is an error while doing
1293 * the database operations we won't be able to rollback dropped slots.
1294 */
1296 {
1299 {
1301
1302 /*
1303 * For READY/SYNCDONE states we know the tablesync slot has
1304 * already been dropped by the tablesync worker.
1305 *
1306 * For other states, there is no certainty, maybe the slot
1307 * does not exist yet. Also, if we fail after removing some of
1308 * the slots, next time, it will again try to drop already
1309 * dropped slots and fail. For these reasons, we allow
1310 * missing_ok = true for the drop.
1311 */
1315 }
1316 }
1317
1318 /*
1319 * Next remove state for sequences we should not care about anymore
1320 * using the data we collected above
1321 */
1323 {
1325
1328 {
1329 /*
1330 * This locking ensures that the state of rels won't change
1331 * till we are done with this refresh operation.
1332 */
1333 if (!rel)
1335
1337
1341 get_rel_name(relid),
1342 sub->name));
1343 }
1344 }
1345 }
1346 PG_FINALLY();
1347 {
1349 }
1350 PG_END_TRY();
1351
1352 if (rel)
1354}
int int errmsg_internal(const char *fmt,...) pg_attribute_printf(1
void CheckSubscriptionRelkind(char localrelkind, char remoterelkind, const char *nspname, const char *relname)
Definition execReplication.c:1140
void logicalrep_worker_stop(LogicalRepWorkerType wtype, Oid subid, Oid relid)
Definition launcher.c:662
void replorigin_drop_by_name(const char *name, bool missing_ok, bool nowait)
Definition origin.c:459
void RemoveSubscriptionRel(Oid subid, Oid relid)
Definition pg_subscription.c:497
char GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn)
Definition pg_subscription.c:447
void AddSubscriptionRelState(Oid subid, Oid relid, char state, XLogRecPtr sublsn, bool retain_lock)
Definition pg_subscription.c:327
List * GetSubscriptionRelations(Oid subid, bool tables, bool sequences, bool not_ready)
Definition pg_subscription.c:625
Definition subscriptioncmds.c:123
Definition primnodes.h:74
Definition pg_subscription_rel.h:83
Definition regguts.h:331
static void check_publications_origin_sequences(WalReceiverConn *wrconn, List *publications, bool copydata, char *origin, Oid *subrel_local_oids, int subrel_count, char *subname)
Definition subscriptioncmds.c:3179
static void check_publications(WalReceiverConn *wrconn, List *publications)
Definition subscriptioncmds.c:574
static List * fetch_relation_list(WalReceiverConn *wrconn, List *publications)
Definition subscriptioncmds.c:3421
void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok)
Definition subscriptioncmds.c:2807
void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, Size szslot)
Definition tablesync.c:1236
Definition pg_list.h:46
References AccessExclusiveLock, AccessShareLock, AddSubscriptionRelState(), check_publications(), check_publications_origin_sequences(), check_publications_origin_tables(), CheckSubscriptionRelkind(), Subscription::conninfo, DEBUG1, ereport, err(), errcode(), errmsg, errmsg_internal(), ERROR, fb(), fetch_relation_list(), foreach_ptr, get_namespace_name(), get_rel_name(), get_rel_namespace(), get_rel_relkind(), GetSubscriptionRelations(), GetSubscriptionRelState(), InvalidXLogRecPtr, lappend(), lfirst, list_length(), load_file(), logicalrep_worker_stop(), Subscription::name, NAMEDATALEN, NIL, NoLock, Subscription::oid, oid_cmp(), Subscription::origin, Subscription::ownersuperuser, palloc(), palloc_object, Subscription::passwordrequired, PG_END_TRY, PG_FINALLY, PG_TRY, Subscription::publications, qsort, RangeVarGetRelid, SubscriptionRelState::relid, RangeVar::relname, RemoveSubscriptionRel(), ReplicationOriginNameForLogicalRep(), ReplicationSlotDropAtPubNode(), ReplicationSlotNameForTablesync(), replorigin_drop_by_name(), Subscription::retaindeadtuples, RangeVar::schemaname, table_close(), table_open(), walrcv_connect, walrcv_disconnect, WORKERTYPE_TABLESYNC, and wrconn.
Referenced by AlterSubscription().
◆ AlterSubscription_refresh_seq()
|
static |
Definition at line 1360 of file subscriptioncmds.c.
1361{
1365
1366 /* Load the library providing us libpq calls. */
1368
1369 /* Try to connect to the publisher. */
1378
1379 PG_TRY();
1380 {
1382
1385
1386 /* Get local sequence list. */
1389 {
1391
1397 get_rel_name(relid),
1398 sub->name));
1399 }
1400 }
1401 PG_FINALLY();
1402 {
1404 }
1405 PG_END_TRY();
1406}
void UpdateSubscriptionRelState(Oid subid, Oid relid, char state, XLogRecPtr sublsn, bool already_locked)
Definition pg_subscription.c:381
References check_publications_origin_sequences(), Subscription::conninfo, DEBUG1, ereport, err(), errcode(), errmsg, errmsg_internal(), ERROR, fb(), foreach_ptr, get_namespace_name(), get_rel_name(), get_rel_namespace(), GetSubscriptionRelations(), InvalidXLogRecPtr, load_file(), Subscription::name, Subscription::oid, Subscription::origin, Subscription::ownersuperuser, Subscription::passwordrequired, PG_END_TRY, PG_FINALLY, PG_TRY, Subscription::publications, UpdateSubscriptionRelState(), walrcv_connect, walrcv_disconnect, and wrconn.
Referenced by AlterSubscription().
◆ AlterSubscriptionOwner()
| ObjectAddress AlterSubscriptionOwner | ( | const char * | name, |
| Oid | newOwnerId | ||
| ) |
Definition at line 2950 of file subscriptioncmds.c.
2951{
2952 Oid subid;
2954 Relation rel;
2955 ObjectAddress address;
2957
2959
2962
2967
2969 subid = form->oid;
2970
2972
2974
2976
2978
2979 return address;
2980}
static void AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
Definition subscriptioncmds.c:2864
References AlterSubscriptionOwner_internal(), CStringGetDatum(), ereport, errcode(), errmsg, ERROR, fb(), Form_pg_subscription, GETSTRUCT(), heap_freetuple(), HeapTupleIsValid, MyDatabaseId, name, ObjectAddressSet, ObjectIdGetDatum(), RowExclusiveLock, SearchSysCacheCopy2, table_close(), and table_open().
Referenced by ExecAlterOwnerStmt().
◆ AlterSubscriptionOwner_internal()
Definition at line 2864 of file subscriptioncmds.c.
2865{
2868
2870
2872 return;
2873
2877
2878 /*
2879 * Don't allow non-superuser modification of a subscription with
2880 * password_required=false.
2881 */
2886 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
2887
2888 /* Must be able to become new owner */
2890
2891 /*
2892 * current owner must have CREATE on database
2893 *
2894 * This is consistent with how ALTER SCHEMA ... OWNER TO works, but some
2895 * other object types behave differently (e.g. you can't give a table to a
2896 * user who lacks CREATE privileges on a schema).
2897 */
2903
2904 /*
2905 * If the subscription uses a server, check that the new owner has USAGE
2906 * privileges on the server and that a user mapping exists. Note: does not
2907 * re-check the resulting connection string.
2908 */
2910 {
2912
2919 server->servername));
2920
2921 /* make sure a user mapping exists */
2923 }
2924
2927
2928 /* Update owner of the conflict log table if it exists. */
2931 AccessExclusiveLock);
2932
2933 /* Update owner dependency reference */
2935 form->oid,
2936 newOwnerId);
2937
2939 form->oid, 0);
2940
2941 /* Wake up related background processes to handle this change quickly. */
2942 ApplyLauncherWakeupAtCommit();
2944}
void changeDependencyOnOwner(Oid classId, Oid objectId, Oid newOwnerId)
Definition pg_shdepend.c:316
void ATExecChangeOwner(Oid relationOid, Oid newOwnerId, bool recursing, LOCKMODE lockmode)
Definition tablecmds.c:16601
References AccessExclusiveLock, ACL_CREATE, ACL_USAGE, aclcheck_error(), ACLCHECK_NOT_OWNER, ACLCHECK_OK, ApplyLauncherWakeupAtCommit(), ATExecChangeOwner(), CatalogTupleUpdate(), changeDependencyOnOwner(), check_can_set_role(), ereport, errcode(), errhint(), errmsg, ERROR, fb(), Form_pg_subscription, get_database_name(), GetForeignServer(), GETSTRUCT(), GetUserId(), GetUserMapping(), GetUserNameFromId(), InvokeObjectPostAlterHook, LogicalRepWorkersWakeupAtCommit(), MyDatabaseId, NameStr, object_aclcheck(), OBJECT_DATABASE, object_ownercheck(), OBJECT_SUBSCRIPTION, OidIsValid, ForeignServer::serverid, ForeignServer::servername, and superuser().
Referenced by AlterSubscriptionOwner(), and AlterSubscriptionOwner_oid().
◆ AlterSubscriptionOwner_oid()
Definition at line 2986 of file subscriptioncmds.c.
References AlterSubscriptionOwner_internal(), ereport, errcode(), errmsg, ERROR, fb(), heap_freetuple(), HeapTupleIsValid, ObjectIdGetDatum(), RowExclusiveLock, SearchSysCacheCopy1, table_close(), and table_open().
Referenced by shdepReassignOwned_Owner().
◆ appendQuotedString()
Definition at line 553 of file subscriptioncmds.c.
554{
557 {
559
563 }
565}
References appendStringInfoChar(), buf, and str.
◆ check_duplicates_in_publist()
Definition at line 3589 of file subscriptioncmds.c.
3590{
3591 ListCell *cell;
3593
3595 {
3598
3600 {
3602
3604 break;
3605
3610 pname)));
3611 }
3612
3613 if (datums)
3615 }
3616}
References CStringGetTextDatum, ereport, errcode(), ERRCODE_DUPLICATE_OBJECT, errmsg, ERROR, fb(), j, lfirst, name, and strVal.
Referenced by merge_publications().
◆ check_pub_dead_tuple_retention()
|
static |
Definition at line 3296 of file subscriptioncmds.c.
3297{
3298 WalRcvExecResult *res;
3300 TupleTableSlot *slot;
3301 bool isnull;
3303
3307 errmsg("cannot enable retain_dead_tuples if the publisher is running a version earlier than PostgreSQL 19"));
3308
3310
3315 res->err)));
3316
3320
3322
3327
3328 ExecDropSingleTupleTableSlot(slot);
3329
3330 walrcv_clear_result(res);
3331}
TupleTableSlot * MakeSingleTupleTableSlot(TupleDesc tupdesc, const TupleTableSlotOps *tts_ops)
Definition execTuples.c:1504
void ExecDropSingleTupleTableSlot(TupleTableSlot *slot)
Definition execTuples.c:1520
Definition tuptable.h:121
Definition walreceiver.h:220
bool tuplestore_gettupleslot(Tuplestorestate *state, bool forward, bool copy, TupleTableSlot *slot)
Definition tuplestore.c:1131
static Datum slot_getattr(TupleTableSlot *slot, int attnum, bool *isnull)
Definition tuptable.h:417
static void walrcv_clear_result(WalRcvExecResult *walres)
Definition walreceiver.h:472
References DatumGetBool(), elog, ereport, WalRcvExecResult::err, errcode(), errmsg, ERROR, ExecDropSingleTupleTableSlot(), fb(), MakeSingleTupleTableSlot(), slot_getattr(), WalRcvExecResult::status, TTSOpsMinimalTuple, WalRcvExecResult::tupledesc, WalRcvExecResult::tuplestore, tuplestore_gettupleslot(), walrcv_clear_result(), walrcv_exec, WALRCV_OK_TUPLES, walrcv_server_version, and wrconn.
Referenced by AlterSubscription().
◆ check_publications()
|
static |
Definition at line 574 of file subscriptioncmds.c.
575{
576 WalRcvExecResult *res;
577 StringInfoData cmd;
578 TupleTableSlot *slot;
581
582 initStringInfo(&cmd);
584 " pg_catalog.pg_publication t WHERE\n"
585 " t.pubname IN (");
588
591
595 res->err));
596
598
599 /* Process publication(s). */
602 {
603 char *pubname;
604 bool isnull;
605
607 Assert(!isnull);
608
609 /* Delete the publication present in publisher from the list. */
611 ExecClearTuple(slot);
612 }
613
614 ExecDropSingleTupleTableSlot(slot);
615
616 walrcv_clear_result(res);
617
619 {
620 /* Prepare the list of non-existent publication(s) for error message. */
622
624
629 "publications %s do not exist on the publisher",
631 pubnames.data));
632 }
633}
int int int errmsg_plural(const char *fmt_singular, const char *fmt_plural, unsigned long n,...) pg_attribute_printf(1
void GetPublicationsStr(List *publications, StringInfo dest, bool quote_literal)
Definition pg_subscription.c:50
void appendStringInfoString(StringInfo str, const char *s)
Definition stringinfo.c:230
Definition stringinfo.h:47
Referenced by AlterSubscription_refresh().
◆ check_publications_origin_sequences()
|
static |
Definition at line 3179 of file subscriptioncmds.c.
3183{
3184 WalRcvExecResult *res;
3185 StringInfoData cmd;
3186 TupleTableSlot *slot;
3189
3190 /*
3191 * Enable sequence synchronization checks only when origin is 'none' , to
3192 * ensure that sequence data from other origins is not inadvertently
3193 * copied. This check is necessary if the publisher is running PG19 or
3194 * later, where logical replication sequence synchronization is supported.
3195 */
3198 return;
3199
3200 initStringInfo(&cmd);
3201 appendStringInfoString(&cmd,
3202 "SELECT DISTINCT P.pubname AS pubname\n"
3203 "FROM pg_publication P,\n"
3204 " LATERAL pg_get_publication_sequences(P.pubname) GPS\n"
3205 " JOIN pg_subscription_rel PS ON (GPS.relid = PS.srrelid),\n"
3206 " pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)\n"
3207 "WHERE C.oid = GPS.relid AND P.pubname IN (");
3208
3211
3212 /*
3213 * In case of ALTER SUBSCRIPTION ... REFRESH PUBLICATION,
3214 * subrel_local_oids contains the list of relations that are already
3215 * present on the subscriber. This check should be skipped as these will
3216 * not be re-synced.
3217 */
3219 {
3225
3226 appendStringInfo(&cmd,
3227 "AND NOT (N.nspname = %s AND C.relname = %s)\n",
3229
3232 }
3233
3236
3241 res->err)));
3242
3243 /* Process publications. */
3246 {
3247 char *pubname;
3248 bool isnull;
3249
3251 Assert(!isnull);
3252
3253 ExecClearTuple(slot);
3255 }
3256
3257 /*
3258 * Log a warning if the publisher has subscribed to the same sequence from
3259 * some other publisher. We cannot know the origin of sequences data
3260 * during the initial sync.
3261 */
3263 {
3265
3266 /* Prepare the list of publication(s) for warning message. */
3269
3272 errmsg("subscription \"%s\" requested copy_data with origin = NONE but might copy data that had a different origin",
3273 subname),
3274 errdetail_plural("The subscription subscribes to a publication (%s) that contains sequences that are written to by other subscriptions.",
3275 "The subscription subscribes to publications (%s) that contain sequences that are written to by other subscriptions.",
3277 errhint("Verify that initial data copied from the publisher sequences did not come from other origins."));
3278 }
3279
3280 ExecDropSingleTupleTableSlot(slot);
3281
3282 walrcv_clear_result(res);
3283}
int errdetail_plural(const char *fmt_singular, const char *fmt_plural, unsigned long n,...) pg_attribute_printf(1
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition stringinfo.c:145
References appendStringInfo(), appendStringInfoString(), Assert, StringInfoData::data, ereport, WalRcvExecResult::err, errcode(), errdetail_plural(), errhint(), errmsg, ERROR, ExecClearTuple(), ExecDropSingleTupleTableSlot(), fb(), get_namespace_name(), get_rel_name(), get_rel_namespace(), GetPublicationsStr(), i, initStringInfo(), list_append_unique(), list_length(), MakeSingleTupleTableSlot(), makeString(), NIL, pfree(), pg_strcasecmp(), quote_literal_cstr(), slot_getattr(), WalRcvExecResult::status, subname, TextDatumGetCString, TTSOpsMinimalTuple, WalRcvExecResult::tupledesc, WalRcvExecResult::tuplestore, tuplestore_gettupleslot(), walrcv_clear_result(), walrcv_exec, WALRCV_OK_TUPLES, walrcv_server_version, WARNING, and wrconn.
Referenced by AlterSubscription_refresh(), and AlterSubscription_refresh_seq().
◆ check_publications_origin_tables()
|
static |
Definition at line 3031 of file subscriptioncmds.c.
3035{
3036 WalRcvExecResult *res;
3037 StringInfoData cmd;
3038 TupleTableSlot *slot;
3046
3047 /*
3048 * Enable retain_dead_tuples checks only when origin is set to 'any',
3049 * since with origin='none' only local changes are replicated to the
3050 * subscriber.
3051 */
3053
3054 /*
3055 * Enable table synchronization checks only when origin is 'none', to
3056 * ensure that data from other origins is not inadvertently copied.
3057 */
3059
3060 /* retain_dead_tuples and table sync checks occur separately */
3062
3063 /* Return if no checks are required */
3065 return;
3066
3067 initStringInfo(&cmd);
3068 appendStringInfoString(&cmd,
3069 "SELECT DISTINCT P.pubname AS pubname\n"
3070 "FROM pg_publication P,\n"
3071 " LATERAL pg_get_publication_tables(P.pubname) GPT\n"
3072 " JOIN pg_subscription_rel PS ON (GPT.relid = PS.srrelid OR"
3073 " GPT.relid IN (SELECT relid FROM pg_partition_ancestors(PS.srrelid) UNION"
3074 " SELECT relid FROM pg_partition_tree(PS.srrelid))),\n"
3075 " pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)\n"
3076 "WHERE C.oid = GPT.relid AND P.pubname IN (");
3079
3080 /*
3081 * In case of ALTER SUBSCRIPTION ... REFRESH PUBLICATION,
3082 * subrel_local_oids contains the list of relation oids that are already
3083 * present on the subscriber. This check should be skipped for these
3084 * tables if checking for table sync scenario. However, when handling the
3085 * retain_dead_tuples scenario, ensure all tables are checked, as some
3086 * existing tables may now include changes from other origins due to newly
3087 * created subscriptions on the publisher.
3088 */
3090 {
3092 {
3098
3101
3104 }
3105 }
3106
3109
3114 res->err)));
3115
3116 /* Process publications. */
3119 {
3120 char *pubname;
3121 bool isnull;
3122
3124 Assert(!isnull);
3125
3126 ExecClearTuple(slot);
3128 }
3129
3130 /*
3131 * Log a warning if the publisher has subscribed to the same table from
3132 * some other publisher. We cannot know the origin of data during the
3133 * initial sync. Data origins can be found only from the WAL by looking at
3134 * the origin id.
3135 *
3136 * XXX: For simplicity, we don't check whether the table has any data or
3137 * not. If the table doesn't have any data then we don't need to
3138 * distinguish between data having origin and data not having origin so we
3139 * can avoid logging a warning for table sync scenario.
3140 */
3142 {
3144
3145 /* Prepare the list of publication(s) for warning message. */
3148
3152 errmsg("subscription \"%s\" requested copy_data with origin = NONE but might copy data that had a different origin",
3153 subname),
3154 errdetail_plural("The subscription subscribes to a publication (%s) that contains tables that are written to by other subscriptions.",
3155 "The subscription subscribes to publications (%s) that contain tables that are written to by other subscriptions.",
3157 errhint("Verify that initial data copied from the publisher tables did not come from other origins."));
3158 else
3161 errmsg("subscription \"%s\" enabled retain_dead_tuples but might not reliably detect conflicts for changes from different origins",
3162 subname),
3163 errdetail_plural("The subscription subscribes to a publication (%s) that contains tables that are written to by other subscriptions.",
3164 "The subscription subscribes to publications (%s) that contain tables that are written to by other subscriptions.",
3167 }
3168
3169 ExecDropSingleTupleTableSlot(slot);
3170
3171 walrcv_clear_result(res);
3172}
References appendStringInfo(), appendStringInfoString(), Assert, StringInfoData::data, ereport, WalRcvExecResult::err, errcode(), errdetail_plural(), errhint(), errmsg, ERROR, ExecClearTuple(), ExecDropSingleTupleTableSlot(), fb(), get_namespace_name(), get_rel_name(), get_rel_namespace(), GetPublicationsStr(), i, initStringInfo(), list_append_unique(), list_length(), MakeSingleTupleTableSlot(), makeString(), NIL, pfree(), pg_strcasecmp(), quote_literal_cstr(), slot_getattr(), WalRcvExecResult::status, subname, TextDatumGetCString, TTSOpsMinimalTuple, WalRcvExecResult::tupledesc, WalRcvExecResult::tuplestore, tuplestore_gettupleslot(), walrcv_clear_result(), walrcv_exec, WALRCV_OK_TUPLES, WARNING, and wrconn.
Referenced by AlterSubscription(), and AlterSubscription_refresh().
◆ CheckAlterSubOption()
|
static |
Definition at line 1413 of file subscriptioncmds.c.
1415{
1419
1420 /*
1421 * Altering the retain_dead_tuples option does not update the slot on the
1422 * publisher.
1423 */
1425
1426 /*
1427 * Do not allow changing the option if the subscription is enabled. This
1428 * is because both failover and two_phase options of the slot on the
1429 * publisher cannot be modified if the slot is currently acquired by the
1430 * existing walsender.
1431 *
1432 * Note that two_phase is enabled (aka changed from 'false' to 'true') on
1433 * the publisher by the existing walsender, so we could have allowed that
1434 * even when the subscription is enabled. But we kept this restriction for
1435 * the sake of consistency and simplicity.
1436 *
1437 * Additionally, do not allow changing the retain_dead_tuples option when
1438 * the subscription is enabled to prevent race conditions arising from the
1439 * new option value being acknowledged asynchronously by the launcher and
1440 * apply workers.
1441 *
1442 * Without the restriction, a race condition may arise when a user
1443 * disables and immediately re-enables the retain_dead_tuples option. In
1444 * this case, the launcher might drop the slot upon noticing the disabled
1445 * action, while the apply worker may keep maintaining
1446 * oldest_nonremovable_xid without noticing the option change. During this
1447 * period, a transaction ID wraparound could falsely make this ID appear
1448 * as if it originates from the future w.r.t the transaction ID stored in
1449 * the slot maintained by launcher.
1450 *
1451 * Similarly, if the user enables retain_dead_tuples concurrently with the
1452 * launcher starting the worker, the apply worker may start calculating
1453 * oldest_nonremovable_xid before the launcher notices the enable action.
1454 * Consequently, the launcher may update slot.xmin to a newer value than
1455 * that maintained by the worker. In subsequent cycles, upon integrating
1456 * the worker's oldest_nonremovable_xid, the launcher might detect a
1457 * retreat in the calculated xmin, necessitating additional handling.
1458 *
1459 * XXX To address the above race conditions, we can define
1460 * oldest_nonremovable_xid as FullTransactionId and adds the check to
1461 * disallow retreating the conflict slot's xmin. For now, we kept the
1462 * implementation simple by disallowing change to the retain_dead_tuples,
1463 * but in the future we can change this after some more analysis.
1464 *
1465 * Note that we could restrict only the enabling of retain_dead_tuples to
1466 * avoid the race conditions described above, but we maintain the
1467 * restriction for both enable and disable operations for the sake of
1468 * consistency.
1469 */
1474 option)));
1475
1477 {
1478 StringInfoData cmd;
1479
1480 /*
1481 * A valid slot must be associated with the subscription for us to
1482 * modify any of the slot's properties.
1483 */
1488 option)));
1489
1490 /* The changed option of the slot can't be rolled back. */
1491 initStringInfo(&cmd);
1493
1496 }
1497}
Definition getopt_long.h:18
References appendStringInfo(), Assert, StringInfoData::data, Subscription::enabled, ereport, errcode(), errmsg, ERROR, fb(), initStringInfo(), pfree(), PreventInTransactionBlock(), and Subscription::slotname.
Referenced by AlterSubscription().
◆ CheckSubDeadTupleRetention()
| void CheckSubDeadTupleRetention | ( | bool | check_guc, |
| bool | sub_disabled, | ||
| int | elevel_for_sub_disabled, | ||
| bool | retain_dead_tuples, | ||
| bool | retention_active, | ||
| bool | max_retention_set | ||
| ) |
Definition at line 3357 of file subscriptioncmds.c.
3361{
3364
3366 {
3370 errmsg("\"wal_level\" is insufficient to create the replication slot required by retain_dead_tuples"),
3372
3378 "track_commit_timestamp"));
3379
3383 errmsg("deleted rows to detect conflicts would not be removed until the subscription is enabled"),
3386 "retain_dead_tuples") : 0);
3387 }
3389 {
3393 }
3394}
References Assert, ereport, errcode(), errhint(), errmsg, ERROR, fb(), NOTICE, track_commit_timestamp, wal_level, WAL_LEVEL_REPLICA, and WARNING.
Referenced by AlterSubscription(), and DisableSubscriptionAndExit().
◆ construct_subserver_conninfo()
Definition at line 2425 of file subscriptioncmds.c.
2426{
2428 ForeignServer *server;
2429
2431
2433
2437 {
2438 /*
2439 * Unable to generate connection string because permissions on the
2440 * foreign server have been removed. Follow the same logic as an
2441 * unusable subconninfo (which will result in an ERROR later unless
2442 * slot_name = NONE).
2443 */
2444 *err = psprintf(_("subscription owner \"%s\" does not have permission on foreign server \"%s\""),
2446 server->servername);
2448 }
2449
2451}
References _, ACL_USAGE, ACLCHECK_OK, err(), fb(), ForeignServerConnectionString(), GetForeignServer(), GetUserNameFromId(), object_aclcheck(), psprintf(), and ForeignServer::servername.
Referenced by DropSubscription().
◆ CreateSubscription()
| ObjectAddress CreateSubscription | ( | ParseState * | pstate, |
| CreateSubscriptionStmt * | stmt, | ||
| bool | isTopLevel | ||
| ) |
Definition at line 669 of file subscriptioncmds.c.
671{
672 Relation rel;
674 Oid subid;
679 Oid serverid;
680 char *conninfo;
682 List *publications;
687
688 /*
689 * Parse and check options.
690 *
691 * Connection and publication should not be specified here.
692 */
702 SUBOPT_CONFLICT_LOG_DEST);
704
705 /*
706 * Since creating a replication slot is not transactional, rolling back
707 * the transaction leaves the created replication slot. So we cannot run
708 * CREATE SUBSCRIPTION inside a transaction block if creating a
709 * replication slot.
710 */
713
714 /*
715 * We don't want to allow unprivileged users to be able to trigger
716 * attempts to access arbitrary network destinations, so require the user
717 * to have been specifically authorized to create subscriptions.
718 */
724 "pg_create_subscription")));
725
726 /*
727 * Since a subscription is a database object, we also check for CREATE
728 * permission on the database.
729 */
731 owner, ACL_CREATE);
735
736 /*
737 * Non-superusers are required to set a password for authentication, and
738 * that password must be used by the target server, but the superuser can
739 * exempt a subscription from this requirement.
740 */
745 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
746
747 /*
748 * If built with appropriate switch, whine when regression-testing
749 * conventions for subscription names are violated.
750 */
751#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
753 elog(WARNING, "subscriptions created by regression test cases should have names starting with \"regress_\"");
754#endif
755
757
758 /* Check if name is used */
762 {
766 stmt->subname)));
767 }
768
769 /*
770 * Ensure that system configuration parameters are set appropriately to
771 * support retain_dead_tuples and max_retention_duration.
772 */
775 (opts.maxretention > 0));
776
780
781 /* The default for synchronous_commit of subscriptions is off. */
784
785 /*
786 * The default for wal_receiver_timeout of subscriptions is -1, which
787 * means the value is inherited from the server configuration, command
788 * line, or role/database settings.
789 */
792
793 /* Load the library providing us libpq calls. */
795
797 {
798 ForeignServer *server;
799
801 conninfo = NULL;
802
807
808 /* make sure a user mapping exists */
810
811 serverid = server->serverid;
812 conninfo = ForeignServerConnectionString(owner, server);
813 }
814 else
815 {
817
818 serverid = InvalidOid;
819 conninfo = stmt->conninfo;
820 }
821
822 /* Check the connection info string. */
824
825 publications = stmt->publication;
826
827 /* Everything ok, form a new tuple. */
830
832 Anum_pg_subscription_oid);
859 CStringGetTextDatum(conninfo);
860 else
865 else
872 publicationListToArray(publications);
875
878
879 /*
880 * We create the conflict log table here, if required, so that its
881 * relation OID can be stored when inserting the pg_subscription tuple
882 * below.
883 */
886
887 /* Store table OID in the catalog. */
890
892
893 /* Insert tuple into catalog. */
896
898
900
902 {
904
906
909 }
910
911 /*
912 * Establish an internal dependency between the conflict log table and the
913 * subscription.
914 *
915 * We use DEPENDENCY_INTERNAL to signify that the table's lifecycle is
916 * strictly tied to the subscription, similar to how a TOAST table relates
917 * to its main table or a sequence relates to an identity column.
918 *
919 * This ensures the conflict log table is automatically reaped during a
920 * DROP SUBSCRIPTION via performDeletion().
921 */
923 {
925
928 }
929
930 /*
931 * A replication origin is currently created for all subscriptions,
932 * including those that only contain sequences or are otherwise empty.
933 *
934 * XXX: While this is technically unnecessary, optimizing it would require
935 * additional logic to skip origin creation during DDL operations and
936 * apply workers initialization, and to handle origin creation dynamically
937 * when tables are added to the subscription. It is not clear whether
938 * preventing creation of origins is worth additional complexity.
939 */
942
943 /*
944 * Connect to remote side to execute requested commands and fetch table
945 * and sequence info.
946 */
948 {
952
953 /* Try to connect to the publisher. */
962
963 PG_TRY();
964 {
968
971 opts.copy_data,
977
980
981 /*
982 * Set sync state based on if we were asked to do data copy or
983 * not.
984 */
986
987 /*
988 * Build local relation status info. Relations are for both tables
989 * and sequences from the publisher.
990 */
992
994 {
995 Oid relid;
996 char relkind;
998
1000 relkind = get_rel_relkind(relid);
1001
1002 /* Check for supported relkind. */
1008 }
1009
1010 /*
1011 * If requested, create permanent slot for the subscription. We
1012 * won't use the initial snapshot for anything, so no need to
1013 * export it.
1014 *
1015 * XXX: Similar to origins, it is not clear whether preventing the
1016 * slot creation for empty and sequence-only subscriptions is
1017 * worth additional complexity.
1018 */
1020 {
1022
1024
1025 /*
1026 * Even if two_phase is set, don't create the slot with
1027 * two-phase enabled. Will enable it once all the tables are
1028 * synced and ready. This avoids race-conditions like prepared
1029 * transactions being skipped due to changes not being applied
1030 * due to checks in should_apply_changes_for_rel() when
1031 * tablesync for the corresponding tables are in progress. See
1032 * comments atop worker.c.
1033 *
1034 * Note that if tables were specified but copy_data is false
1035 * then it is safe to enable two_phase up-front because those
1036 * tables are already initially in READY state. When the
1037 * subscription has no tables, we leave the twophase state as
1038 * PENDING, to allow ALTER SUBSCRIPTION ... REFRESH
1039 * PUBLICATION to work.
1040 */
1043
1046
1049
1052 opts.slot_name)));
1053 }
1054 }
1055 PG_FINALLY();
1056 {
1058 }
1059 PG_END_TRY();
1060 }
1061 else
1064 errhint("To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.")));
1065
1067
1068 pgstat_create_subscription(subid);
1069
1070 /*
1071 * Notify the launcher to start the apply worker if the subscription is
1072 * enabled, or to create the conflict detection slot if retain_dead_tuples
1073 * is enabled.
1074 *
1075 * Creating the conflict detection slot is essential even when the
1076 * subscription is not enabled. This ensures that dead tuples are
1077 * retained, which is necessary for accurately identifying the type of
1078 * conflict during replication.
1079 */
1081 ApplyLauncherWakeupAtCommit();
1082
1084
1086}
Oid GetNewOidWithIndex(Relation relation, Oid indexId, AttrNumber oidcolumn)
Definition catalog.c:475
int errdetail(const char *fmt,...) pg_attribute_printf(1
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, const Datum *values, const bool *isnull)
Definition heaptuple.c:1025
#define InvokeObjectPostCreateHook(classId, objectId, subId)
Definition objectaccess.h:173
void recordDependencyOnOwner(Oid classId, Oid objectId, Oid owner)
Definition pg_shdepend.c:168
void pgstat_create_subscription(Oid subid)
Definition pgstat_subscription.c:78
#define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn)
Definition walreceiver.h:460
Referenced by ProcessUtilitySlow().
◆ defGetStreamingMode()
Definition at line 3688 of file subscriptioncmds.c.
3689{
3690 /*
3691 * If no parameter value given, assume "true" is meant.
3692 */
3695
3696 /*
3697 * Allow 0, 1, "false", "true", "off", "on" or "parallel".
3698 */
3700 {
3703 {
3704 case 0:
3706 case 1:
3708 default:
3709 /* otherwise, error out below */
3710 break;
3711 }
3712 break;
3713 default:
3714 {
3716
3717 /*
3718 * The set of strings accepted here should match up with the
3719 * grammar's opt_boolean_or_string production.
3720 */
3729 }
3730 break;
3731 }
3732
3736 def->defname)));
3738}
References DefElem::arg, defGetString(), DefElem::defname, ereport, errcode(), errmsg, ERROR, fb(), intVal, nodeTag, and pg_strcasecmp().
Referenced by parse_output_parameters(), and parse_subscription_options().
◆ drop_sub_conflict_log_table()
|
static |
Definition at line 2469 of file subscriptioncmds.c.
2470{
2471 /* Drop any dependent conflict log table */
2473 {
2474 ObjectAddress object;
2476
2480 subconflictlogrelid);
2481
2482 /*
2483 * By using PERFORM_DELETION_SKIP_ORIGINAL, we ensure that only the
2484 * conflict log table is deleted while the subscription remains.
2485 */
2488 PERFORM_DELETION_INTERNAL |
2490
2494 subname));
2495 }
2496}
void performDeletion(const ObjectAddress *object, DropBehavior behavior, int flags)
Definition dependency.c:279
char * get_qualified_objname(Oid nspid, char *objname)
Definition lsyscache.c:3712
References DROP_CASCADE, elog, ereport, errmsg, ERROR, fb(), get_qualified_objname(), get_rel_name(), NOTICE, ObjectAddressSet, OidIsValid, PERFORM_DELETION_INTERNAL, PERFORM_DELETION_SKIP_ORIGINAL, performDeletion(), subconflictlogrelid, and subname.
Referenced by alter_sub_conflict_log_dest(), and DropSubscription().
◆ DropSubscription()
| void DropSubscription | ( | DropSubscriptionStmt * | stmt, |
| bool | isTopLevel | ||
| ) |
Definition at line 2502 of file subscriptioncmds.c.
2503{
2504 Relation rel;
2507 Oid subid;
2512 Datum datum;
2513 bool isnull;
2516 char *slotname;
2525
2526 /*
2527 * The launcher may concurrently start a new worker for this subscription.
2528 * During initialization, the worker checks for subscription validity and
2529 * exits if the subscription has already been dropped. See
2530 * InitializeLogRepWorker.
2531 */
2533
2536
2538 {
2540
2545 stmt->subname)));
2546 else
2549 stmt->subname)));
2550
2551 return;
2552 }
2553
2555 Anum_pg_subscription_subconninfo, &isnull);
2556 if (!isnull)
2557 subconninfo = TextDatumGetCString(datum);
2558
2560 subid = form->oid;
2565
2566 /* must be owner */
2569 stmt->subname);
2570
2571 /* DROP hook for the subscription being removed */
2573
2574 /*
2575 * Lock the subscription so nobody else can do anything with it (including
2576 * the replication workers).
2577 */
2579
2580 /* Get subname */
2582 Anum_pg_subscription_subname);
2584
2585 /* Get slotname */
2587 Anum_pg_subscription_subslotname, &isnull);
2588 if (!isnull)
2590 else
2591 slotname = NULL;
2592
2593 /*
2594 * Since dropping a replication slot is not transactional, the replication
2595 * slot stays dropped even if the transaction rolls back. So we cannot
2596 * run DROP SUBSCRIPTION inside a transaction block if dropping the
2597 * replication slot. Also, in this case, we report a message for dropping
2598 * the subscription to the cumulative stats system.
2599 *
2600 * XXX The command name should really be something like "DROP SUBSCRIPTION
2601 * of a subscription that is associated with a replication slot", but we
2602 * don't have the proper facilities for that.
2603 */
2604 if (slotname)
2606
2609
2610 /* Remove the tuple from catalog. */
2612
2614
2615 /*
2616 * Stop all the subscription workers immediately.
2617 *
2618 * This is necessary if we are dropping the replication slot, so that the
2619 * slot becomes accessible.
2620 *
2621 * It is also necessary if the subscription is disabled and was disabled
2622 * in the same transaction. Then the workers haven't seen the disabling
2623 * yet and will still be running, leading to hangs later when we want to
2624 * drop the replication origin. If the subscription was disabled before
2625 * this transaction, then there shouldn't be any workers left, so this
2626 * won't make a difference.
2627 *
2628 * New workers won't be started because we hold an exclusive lock on the
2629 * subscription till the end of the transaction.
2630 */
2633 {
2635
2637 }
2639
2640 /*
2641 * Remove the no-longer-useful entry in the launcher's table of apply
2642 * worker start times.
2643 *
2644 * If this transaction rolls back, the launcher might restart a failed
2645 * apply worker before wal_retrieve_retry_interval milliseconds have
2646 * elapsed, but that's pretty harmless.
2647 */
2648 ApplyLauncherForgetWorkerStartTime(subid);
2649
2650 /*
2651 * Cleanup of tablesync replication origins.
2652 *
2653 * Any READY-state relations would already have dealt with clean-ups.
2654 *
2655 * Note that the state can't change because we have already stopped both
2656 * the apply and tablesync workers and they can't restart because of
2657 * exclusive lock on the subscription.
2658 */
2661 {
2664
2665 /* Only cleanup resources of tablesync workers */
2667 continue;
2668
2669 /*
2670 * Drop the tablesync's origin tracking if exists.
2671 *
2672 * It is possible that the origin is not yet created for tablesync
2673 * worker so passing missing_ok = true. This can happen for the states
2674 * before SUBREL_STATE_DATASYNC.
2675 */
2679 }
2680
2681 /* Drop subscription's conflict log table */
2683
2684 /* Clean up dependencies */
2687
2688 /* Remove any associated relation synchronization states. */
2690
2691 /* Remove the origin tracking if exists. */
2694
2695 /*
2696 * Tell the cumulative stats system that the subscription is getting
2697 * dropped.
2698 */
2699 pgstat_drop_subscription(subid);
2700
2701 /*
2702 * If there is no slot associated with the subscription, we can finish
2703 * here.
2704 */
2706 {
2708 return;
2709 }
2710
2711 /*
2712 * Try to acquire the connection necessary for dropping slots.
2713 *
2714 * Note: If the slotname is NONE/NULL then we allow the command to finish
2715 * and users need to manually cleanup the apply and tablesync worker slots
2716 * later.
2717 *
2718 * This has to be at the end because otherwise if there is an error while
2719 * doing the database operations we won't be able to rollback dropped
2720 * slot.
2721 */
2723
2726 else
2727 conninfo = subconninfo;
2728
2729 if (conninfo)
2732
2734 {
2735 if (!slotname)
2736 {
2737 /* be tidy */
2740 return;
2741 }
2742 else
2743 {
2745 }
2746 }
2747
2748 PG_TRY();
2749 {
2751 {
2754
2755 /* Only cleanup resources of tablesync workers */
2757 continue;
2758
2759 /*
2760 * Drop the tablesync slots associated with removed tables.
2761 *
2762 * For SYNCDONE/READY states, the tablesync slot is known to have
2763 * already been dropped by the tablesync worker.
2764 *
2765 * For other states, there is no certainty, maybe the slot does
2766 * not exist yet. Also, if we fail after removing some of the
2767 * slots, next time, it will again try to drop already dropped
2768 * slots and fail. For these reasons, we allow missing_ok = true
2769 * for the drop.
2770 */
2772 {
2774
2778 }
2779 }
2780
2782
2783 /*
2784 * If there is a slot associated with the subscription, then drop the
2785 * replication slot at the publisher.
2786 */
2787 if (slotname)
2789 }
2790 PG_FINALLY();
2791 {
2793 }
2794 PG_END_TRY();
2795
2797}
void EventTriggerSQLDropAddObject(const ObjectAddress *object, bool original, bool normal)
Definition event_trigger.c:1287
void CatalogTupleDelete(Relation heapRel, const ItemPointerData *tid)
Definition indexing.c:365
void ApplyLauncherForgetWorkerStartTime(Oid subid)
Definition launcher.c:1155
#define InvokeObjectDropHook(classId, objectId, subId)
Definition objectaccess.h:182
long deleteDependencyRecordsFor(Oid classId, Oid objectId, bool skipExtensionDeps)
Definition pg_depend.c:314
void deleteSharedDependencyRecordsFor(Oid classId, Oid objectId, int32 objectSubId)
Definition pg_shdepend.c:1047
Definition worker_internal.h:38
static char * construct_subserver_conninfo(Oid subserver, Oid subowner, char **err)
Definition subscriptioncmds.c:2425
static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err)
Definition subscriptioncmds.c:3545
HeapTuple SearchSysCache2(SysCacheIdentifier cacheId, Datum key1, Datum key2)
Definition syscache.c:231
Datum SysCacheGetAttrNotNull(SysCacheIdentifier cacheId, HeapTuple tup, AttrNumber attributeNumber)
Definition syscache.c:626
Datum SysCacheGetAttr(SysCacheIdentifier cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition syscache.c:596
References AccessExclusiveLock, aclcheck_error(), ACLCHECK_NOT_OWNER, ApplyLauncherForgetWorkerStartTime(), CatalogTupleDelete(), construct_subserver_conninfo(), CStringGetDatum(), DatumGetName(), deleteDependencyRecordsFor(), deleteSharedDependencyRecordsFor(), drop_sub_conflict_log_table(), ereport, err(), errcode(), errmsg, ERROR, EventTriggerSQLDropAddObject(), fb(), Form_pg_subscription, GETSTRUCT(), GetSubscriptionRelations(), GetUserId(), HeapTupleIsValid, InvalidOid, InvokeObjectDropHook, lfirst, list_free(), load_file(), LockSharedObject(), logicalrep_worker_stop(), logicalrep_workers_find(), MyDatabaseId, NAMEDATALEN, NameStr, NIL, NoLock, NOTICE, object_ownercheck(), OBJECT_SUBSCRIPTION, ObjectAddressSet, ObjectIdGetDatum(), OidIsValid, PG_END_TRY, PG_FINALLY, PG_TRY, pgstat_drop_subscription(), PreventInTransactionBlock(), pstrdup(), ReleaseSysCache(), SubscriptionRelState::relid, LogicalRepWorker::relid, RemoveSubscriptionRel(), ReplicationOriginNameForLogicalRep(), ReplicationSlotDropAtPubNode(), ReplicationSlotNameForTablesync(), replorigin_drop_by_name(), ReportSlotConnectionError(), RowExclusiveLock, SearchSysCache2(), SubscriptionRelState::state, stmt, subconflictlogrelid, LogicalRepWorker::subid, subname, superuser_arg(), SysCacheGetAttr(), SysCacheGetAttrNotNull(), table_close(), table_open(), TextDatumGetCString, LogicalRepWorker::type, walrcv_connect, walrcv_disconnect, and wrconn.
Referenced by ProcessUtilitySlow().
◆ fetch_relation_list()
|
static |
Definition at line 3421 of file subscriptioncmds.c.
3422{
3423 WalRcvExecResult *res;
3424 StringInfoData cmd;
3425 TupleTableSlot *slot;
3431 StringInfoData pub_names;
3432
3433 initStringInfo(&cmd);
3434 initStringInfo(&pub_names);
3435
3436 /* Build the pub_names comma-separated string. */
3438
3439 /* Get the list of relations from the publisher */
3441 {
3443
3444 /*
3445 * From version 16, we allowed passing multiple publications to the
3446 * function pg_get_publication_tables. This helped to filter out the
3447 * partition table whose ancestor is also published in this
3448 * publication array.
3449 *
3450 * Join pg_get_publication_tables with pg_publication to exclude
3451 * non-existing publications.
3452 *
3453 * Note that attrs are always stored in sorted order so we don't need
3454 * to worry if different publications have specified them in a
3455 * different order. See pub_collist_validate.
3456 */
3458 " FROM pg_class c\n"
3459 " JOIN pg_namespace n ON n.oid = c.relnamespace\n"
3460 " JOIN ( SELECT (pg_get_publication_tables(VARIADIC array_agg(pubname::text))).*\n"
3461 " FROM pg_publication\n"
3462 " WHERE pubname IN ( %s )) AS gpt\n"
3463 " ON gpt.relid = c.oid\n",
3464 pub_names.data);
3465
3466 /* From version 19, inclusion of sequences in the target is supported */
3468 appendStringInfo(&cmd,
3469 "UNION ALL\n"
3470 " SELECT DISTINCT s.schemaname, s.sequencename, " CppAsString2(RELKIND_SEQUENCE) "::\"char\" AS relkind, NULL::int2vector AS attrs\n"
3471 " FROM pg_catalog.pg_publication_sequences s\n"
3472 " WHERE s.pubname IN ( %s )",
3473 pub_names.data);
3474 }
3475 else
3476 {
3478 appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename, " CppAsString2(RELKIND_RELATION) "::\"char\" AS relkind \n");
3479
3480 /* Get column lists for each relation if the publisher supports it */
3483
3485 " WHERE t.pubname IN ( %s )",
3486 pub_names.data);
3487 }
3488
3490
3493
3498 res->err)));
3499
3500 /* Process tables. */
3503 {
3504 char *nspname;
3506 bool isnull;
3507 char relkind;
3509
3511 Assert(!isnull);
3513 Assert(!isnull);
3515 Assert(!isnull);
3516
3518 relinfo->relkind = relkind;
3519
3521 check_columnlist &&
3526 nspname, relname));
3527 else
3529
3530 ExecClearTuple(slot);
3531 }
3532 ExecDropSingleTupleTableSlot(slot);
3533
3534 walrcv_clear_result(res);
3535
3537}
RangeVar * makeRangeVar(char *schemaname, char *relname, int location)
Definition makefuncs.c:473
static bool list_member_rangevar(const List *list, RangeVar *rv)
Definition subscriptioncmds.c:3400
References appendStringInfo(), appendStringInfoString(), Assert, CppAsString2, StringInfoData::data, DatumGetChar(), ereport, WalRcvExecResult::err, errcode(), errmsg, ERROR, ExecClearTuple(), ExecDropSingleTupleTableSlot(), fb(), GetPublicationsStr(), initStringInfo(), InvalidOid, lappend(), list_member_rangevar(), makeRangeVar(), MakeSingleTupleTableSlot(), NIL, palloc_object, pfree(), relname, server_version, slot_getattr(), WalRcvExecResult::status, TextDatumGetCString, TTSOpsMinimalTuple, WalRcvExecResult::tupledesc, WalRcvExecResult::tuplestore, tuplestore_gettupleslot(), walrcv_clear_result(), walrcv_exec, WALRCV_OK_TUPLES, walrcv_server_version, and wrconn.
Referenced by AlterSubscription_refresh().
◆ list_member_rangevar()
Definition at line 3400 of file subscriptioncmds.c.
3401{
3403 {
3405 return true;
3406 }
3407
3408 return false;
3409}
References equal(), fb(), and foreach_ptr.
Referenced by fetch_relation_list().
◆ merge_publications()
|
static |
Definition at line 3629 of file subscriptioncmds.c.
3630{
3632
3634
3636
3638 {
3641 bool found = false;
3642
3644 {
3646
3648 {
3649 found = true;
3655 else
3657
3658 break;
3659 }
3660 }
3661
3669 }
3670
3671 /*
3672 * XXX Probably no strong reason for this, but for now it's to make ALTER
3673 * SUBSCRIPTION ... DROP PUBLICATION consistent with SET PUBLICATION.
3674 */
3679
3681}
static void check_duplicates_in_publist(List *publist, Datum *datums)
Definition subscriptioncmds.c:3589
References check_duplicates_in_publist(), ereport, errcode(), ERRCODE_DUPLICATE_OBJECT, errmsg, ERROR, fb(), foreach_delete_current, lappend(), lfirst, list_copy(), makeString(), name, strVal, and subname.
Referenced by AlterSubscription().
◆ parse_subscription_options()
|
static |
Definition at line 161 of file subscriptioncmds.c.
163{
165
166 /* Start out with cleared opts. */
168
169 /* caller must expect some option */
171
172 /* If connect option is supported, these others also need to be. */
175 SUBOPT_COPY_DATA));
176
177 /* Set default values for the supported options. */
205 opts->maxretention = 0;
210
211 /* Parse options */
213 {
215
218 {
221
224 }
227 {
230
233 }
236 {
239
242 }
245 {
248
251
252 /* Setting slot_name = NONE is treated as no slot name. */
255 else
257 }
260 {
263
266 }
269 {
272
275
276 /* Test if the given value is valid for synchronous_commit GUC. */
279 false, 0, false);
280 }
283 {
286
289 }
292 {
295
298 }
301 {
304
307 }
310 {
313
316 }
319 {
322
325 }
328 {
331
334 }
337 {
340
343 }
346 {
349
352 }
355 {
358
361 }
364 {
367
370
375 }
378 {
381
384
385 /*
386 * Even though the "origin" parameter allows only "none" and "any"
387 * values, it is implemented as a string type so that the
388 * parameter can be extended in future versions to support
389 * filtering using origin names specified by the user.
390 */
392
398 }
401 {
403 XLogRecPtr lsn;
404
407
408 /* Setting lsn = NONE is treated as resetting LSN */
410 lsn = InvalidXLogRecPtr;
411 else
412 {
413 /* Parse the argument as LSN */
416
421 }
422
424 opts->lsn = lsn;
425 }
428 {
431
434
437
438 /*
439 * Test if the given value is valid for wal_receiver_timeout GUC.
440 * Skip this test if the value is -1, since -1 is allowed for the
441 * wal_receiver_timeout subscription option, but not for the GUC
442 * itself.
443 */
448 false, 0, false);
449 }
452 {
454
457
461 }
462 else
466 }
467
468 /*
469 * We've been explicitly asked to not connect, that requires some
470 * additional processing.
471 */
473 {
474 /* Check for incompatible options from the user. */
479 /*- translator: both %s are strings of the form "option = value" */
481 "connect = false", "enabled = true")));
482
488 "connect = false", "create_slot = true")));
489
495 "connect = false", "copy_data = true")));
496
497 /* Change the defaults of other options. */
501 }
502
503 /*
504 * Do additional checking for disallowed combination when slot_name = NONE
505 * was used.
506 */
509 {
511 {
515 /*- translator: both %s are strings of the form "option = value" */
517 "slot_name = NONE", "enabled = true")));
518 else
521 /*- translator: both %s are strings of the form "option = value" */
523 "slot_name = NONE", "enabled = false")));
524 }
525
527 {
531 /*- translator: both %s are strings of the form "option = value" */
533 "slot_name = NONE", "create_slot = true")));
534 else
537 /*- translator: both %s are strings of the form "option = value" */
539 "slot_name = NONE", "create_slot = false")));
540 }
541 }
542}
void errorConflictingDefElem(DefElem *defel, ParseState *pstate)
Definition define.c:370
bool parse_int(const char *value, int *result, int flags, const char **hintmsg)
Definition guc.c:2775
int set_config_option(const char *name, const char *value, GucContext context, GucSource source, GucAction action, bool changeVal, int elevel, bool is_reload)
Definition guc.c:3248
bool ReplicationSlotValidateName(const char *name, bool allow_reserved_name, int elevel)
Definition slot.c:265
Definition parsenodes.h:860
References Assert, CONFLICT_LOG_DEST_LOG, CStringGetDatum(), DatumGetLSN(), defGetBoolean(), defGetInt32(), defGetStreamingMode(), defGetString(), DirectFunctionCall1, ereport, errcode(), errmsg, ERROR, errorConflictingDefElem(), fb(), GetConflictLogDest(), GUC_ACTION_SET, InvalidXLogRecPtr, IsSet, lfirst, opts, parse_int(), pfree(), pg_lsn_in(), pg_strcasecmp(), PGC_BACKEND, PGC_S_TEST, pstrdup(), ReplicationSlotValidateName(), set_config_option(), SUBOPT_BINARY, SUBOPT_CONFLICT_LOG_DEST, SUBOPT_CONNECT, SUBOPT_COPY_DATA, SUBOPT_CREATE_SLOT, SUBOPT_DISABLE_ON_ERR, SUBOPT_ENABLED, SUBOPT_FAILOVER, SUBOPT_LSN, SUBOPT_MAX_RETENTION_DURATION, SUBOPT_ORIGIN, SUBOPT_PASSWORD_REQUIRED, SUBOPT_REFRESH, SUBOPT_RETAIN_DEAD_TUPLES, SUBOPT_RUN_AS_OWNER, SUBOPT_SLOT_NAME, SUBOPT_STREAMING, SUBOPT_SYNCHRONOUS_COMMIT, SUBOPT_TWOPHASE_COMMIT, SUBOPT_WAL_RECEIVER_TIMEOUT, val, and XLogRecPtrIsValid.
Referenced by AlterSubscription().
◆ publicationListToArray()
Definition at line 639 of file subscriptioncmds.c.
640{
641 ArrayType *arr;
642 Datum *datums;
643 MemoryContext memcxt;
645
646 /* Create memory context for temporary allocations. */
648 "publicationListToArray to array",
649 ALLOCSET_DEFAULT_SIZES);
651
653
655
657
659
660 MemoryContextDelete(memcxt);
661
663}
ArrayType * construct_array_builtin(Datum *elems, int nelems, Oid elmtype)
Definition arrayfuncs.c:3387
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition palloc.h:138
Definition array.h:93
Definition memnodes.h:118
Referenced by AlterSubscription().
◆ ReplicationSlotDropAtPubNode()
| void ReplicationSlotDropAtPubNode | ( | WalReceiverConn * | wrconn, |
| char * | slotname, | ||
| bool | missing_ok | ||
| ) |
Definition at line 2807 of file subscriptioncmds.c.
2808{
2809 StringInfoData cmd;
2810
2812
2814
2815 initStringInfo(&cmd);
2817 appendQuotedIdentifier(&cmd, slotname);
2819
2820 PG_TRY();
2821 {
2822 WalRcvExecResult *res;
2823
2825
2827 {
2828 /* NOTICE. Success. */
2831 slotname)));
2832 }
2834 missing_ok &&
2836 {
2837 /* LOG. Error, but missing_ok = true. */
2840 slotname, res->err)));
2841 }
2842 else
2843 {
2844 /* ERROR. */
2848 slotname, res->err)));
2849 }
2850
2851 walrcv_clear_result(res);
2852 }
2853 PG_FINALLY();
2854 {
2856 }
2857 PG_END_TRY();
2858}
References appendQuotedIdentifier, appendStringInfoString(), Assert, StringInfoData::data, ereport, WalRcvExecResult::err, errcode(), errmsg, ERROR, fb(), initStringInfo(), load_file(), LOG, NOTICE, pfree(), PG_END_TRY, PG_FINALLY, PG_TRY, WalRcvExecResult::sqlstate, WalRcvExecResult::status, walrcv_clear_result(), WALRCV_ERROR, walrcv_exec, WALRCV_OK_COMMAND, and wrconn.
Referenced by AlterSubscription_refresh(), DropSubscription(), LogicalRepSyncTableStart(), and ProcessSyncingTablesForSync().
◆ ReportSlotConnectionError()
|
static |
Definition at line 3545 of file subscriptioncmds.c.
3546{
3548
3550 {
3553
3554 /* Only cleanup resources of tablesync workers */
3556 continue;
3557
3558 /*
3559 * Caller needs to ensure that relstate doesn't change underneath us.
3560 * See DropSubscription where we get the relstates.
3561 */
3563 {
3565
3569 syncslotname);
3570 }
3571 }
3572
3576 slotname, err),
3577 /* translator: %s is an SQL ALTER command */
3578 errhint("Use %s to disable the subscription, and then use %s to disassociate it from the slot.",
3579 "ALTER SUBSCRIPTION ... DISABLE",
3580 "ALTER SUBSCRIPTION ... SET (slot_name = NONE)")));
3581}
References elog, ereport, err(), errcode(), errhint(), errmsg, ERROR, fb(), lfirst, NAMEDATALEN, OidIsValid, SubscriptionRelState::relid, ReplicationSlotNameForTablesync(), SubscriptionRelState::state, and WARNING.
Referenced by DropSubscription().
