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 "storage/lock.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 568 of file subscriptioncmds.c.
◆ appendQuotedLiteral
| #define appendQuotedLiteral | ( | b, | |
| s | |||
| ) | appendQuotedString(b, s, '\'') |
Definition at line 569 of file subscriptioncmds.c.
◆ IsSet
Definition at line 88 of file subscriptioncmds.c.
◆ SUBOPT_BINARY
| #define SUBOPT_BINARY 0x00000080 |
Definition at line 73 of file subscriptioncmds.c.
◆ SUBOPT_CONFLICT_LOG_DEST
| #define SUBOPT_CONFLICT_LOG_DEST 0x00080000 |
Definition at line 85 of file subscriptioncmds.c.
◆ SUBOPT_CONNECT
| #define SUBOPT_CONNECT 0x00000001 |
Definition at line 66 of file subscriptioncmds.c.
◆ SUBOPT_COPY_DATA
| #define SUBOPT_COPY_DATA 0x00000010 |
Definition at line 70 of file subscriptioncmds.c.
◆ SUBOPT_CREATE_SLOT
| #define SUBOPT_CREATE_SLOT 0x00000004 |
Definition at line 68 of file subscriptioncmds.c.
◆ SUBOPT_DISABLE_ON_ERR
| #define SUBOPT_DISABLE_ON_ERR 0x00000400 |
Definition at line 76 of file subscriptioncmds.c.
◆ SUBOPT_ENABLED
| #define SUBOPT_ENABLED 0x00000002 |
Definition at line 67 of file subscriptioncmds.c.
◆ SUBOPT_FAILOVER
| #define SUBOPT_FAILOVER 0x00002000 |
Definition at line 79 of file subscriptioncmds.c.
◆ SUBOPT_LSN
| #define SUBOPT_LSN 0x00020000 |
Definition at line 83 of file subscriptioncmds.c.
◆ SUBOPT_MAX_RETENTION_DURATION
| #define SUBOPT_MAX_RETENTION_DURATION 0x00008000 |
Definition at line 81 of file subscriptioncmds.c.
◆ SUBOPT_ORIGIN
| #define SUBOPT_ORIGIN 0x00040000 |
Definition at line 84 of file subscriptioncmds.c.
◆ SUBOPT_PASSWORD_REQUIRED
| #define SUBOPT_PASSWORD_REQUIRED 0x00000800 |
Definition at line 77 of file subscriptioncmds.c.
◆ SUBOPT_REFRESH
| #define SUBOPT_REFRESH 0x00000040 |
Definition at line 72 of file subscriptioncmds.c.
◆ SUBOPT_RETAIN_DEAD_TUPLES
| #define SUBOPT_RETAIN_DEAD_TUPLES 0x00004000 |
Definition at line 80 of file subscriptioncmds.c.
◆ SUBOPT_RUN_AS_OWNER
| #define SUBOPT_RUN_AS_OWNER 0x00001000 |
Definition at line 78 of file subscriptioncmds.c.
◆ SUBOPT_SLOT_NAME
| #define SUBOPT_SLOT_NAME 0x00000008 |
Definition at line 69 of file subscriptioncmds.c.
◆ SUBOPT_STREAMING
| #define SUBOPT_STREAMING 0x00000100 |
Definition at line 74 of file subscriptioncmds.c.
◆ SUBOPT_SYNCHRONOUS_COMMIT
| #define SUBOPT_SYNCHRONOUS_COMMIT 0x00000020 |
Definition at line 71 of file subscriptioncmds.c.
◆ SUBOPT_TWOPHASE_COMMIT
| #define SUBOPT_TWOPHASE_COMMIT 0x00000200 |
Definition at line 75 of file subscriptioncmds.c.
◆ SUBOPT_WAL_RECEIVER_TIMEOUT
| #define SUBOPT_WAL_RECEIVER_TIMEOUT 0x00010000 |
Definition at line 82 of file subscriptioncmds.c.
Typedef Documentation
◆ PublicationRelKind
◆ SubOpts
Function Documentation
◆ alter_sub_conflict_log_dest()
|
static |
Definition at line 1549 of file subscriptioncmds.c.
1552{
1557
1560
1562 {
1563 /* There is a conflict log table already. */
1565 {
1567 sub->conflictlogrelid);
1569 }
1570 }
1571 else
1572 {
1573 /* There was no previous conflict log table. */
1575 {
1578
1581
1582 /*
1583 * Establish an internal dependency between the conflict log table
1584 * and the subscription. For details refer comments in
1585 * CreateSubscription function.
1586 */
1590 }
1591 }
1592
1593 *conflicttablerelid = relid;
1595}
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:2502
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 1601 of file subscriptioncmds.c.
1603{
1604 Relation rel;
1610 Oid subid;
1620 char *origin;
1621 Subscription *sub;
1625
1627
1628 /* Fetch the existing tuple. */
1631
1636 stmt->subname)));
1637
1639 subid = form->oid;
1640
1641 /* must be owner */
1644 stmt->subname);
1645
1646 /* parse and check options */
1648 {
1653 SUBOPT_DISABLE_ON_ERR |
1654 SUBOPT_PASSWORD_REQUIRED |
1656 SUBOPT_RETAIN_DEAD_TUPLES |
1658 SUBOPT_WAL_RECEIVER_TIMEOUT |
1659 SUBOPT_ORIGIN |
1660 SUBOPT_CONFLICT_LOG_DEST);
1661 break;
1662
1665 break;
1666
1669 break;
1670
1674 break;
1675
1678 break;
1679
1682 break;
1683
1684 default:
1685 supported_opts = 0;
1686 break;
1687 }
1688
1691
1692 /*
1693 * Ensure that ALTER SUBSCRIPTION commands that could be used to fix a
1694 * broken connection or prepare to drop a broken subscription don't
1695 * attempt to construct the conninfo. Otherwise, we might encounter the
1696 * error the user is trying to fix.
1697 *
1698 * Specifically, ALTER SUBSCRIPTION DISABLE, ALTER SUBSCRIPTION SERVER,
1699 * ALTER SUBSCRIPTION CONNECTION, or ALTER SUBSCRIPTION SET
1700 * (slot_name=NONE).
1701 *
1702 * NB: if the user specifies multiple SET options, then we may still need
1703 * to construct conninfo even if slot_name is set to NONE.
1704 */
1706 {
1709 }
1712 {
1714 }
1716 {
1717 /* ... SET (slot_name = NONE) with no other options */
1720 }
1721
1722 /*
1723 * Skip ACL checks on the subscription's foreign server, if any. If
1724 * changing the server (or replacing it with a raw connection), then the
1725 * old one will be removed anyway. If changing something unrelated,
1726 * there's no need to do an additional ACL check here; that will be done
1727 * by the subscription worker.
1728 */
1730
1732 origin = sub->origin;
1735
1736 /*
1737 * Don't allow non-superuser modification of a subscription with
1738 * password_required=false.
1739 */
1744 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
1745
1746 /* Lock the subscription so nobody else can do anything with it. */
1748
1749 /* Form a new tuple. */
1753
1755
1757 {
1759 {
1761 {
1762 /*
1763 * The subscription must be disabled to allow slot_name as
1764 * 'none', otherwise, the apply worker will repeatedly try
1765 * to stream the data using that slot_name which neither
1766 * exists on the publisher nor the user will be allowed to
1767 * create it.
1768 */
1773 "slot_name = NONE")));
1774
1778 else
1781 }
1782
1784 {
1788 }
1789
1791 {
1795 }
1796
1798 {
1802 }
1803
1805 {
1809 = true;
1810 }
1811
1813 {
1814 /* Non-superuser may not disable password_required. */
1819 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
1820
1824 = true;
1825 }
1826
1828 {
1832 }
1833
1835 {
1836 /*
1837 * We need to update both the slot and the subscription
1838 * for the two_phase option. We can enable the two_phase
1839 * option for a slot only once the initial data
1840 * synchronization is done. This is to avoid missing some
1841 * data as explained in comments atop worker.c.
1842 */
1844
1846 isTopLevel);
1847
1848 /*
1849 * Modifying the two_phase slot option requires a slot
1850 * lookup by slot name, so changing the slot name at the
1851 * same time is not allowed.
1852 */
1858
1859 /*
1860 * Note that workers may still survive even if the
1861 * subscription has been disabled.
1862 *
1863 * Ensure workers have already been exited to avoid
1864 * getting prepared transactions while we are disabling
1865 * the two_phase option. Otherwise, the changes of an
1866 * already prepared transaction can be replicated again
1867 * along with its corresponding commit, leading to
1868 * duplicate data or errors.
1869 */
1875
1876 /*
1877 * two_phase cannot be disabled if there are any
1878 * uncommitted prepared transactions present otherwise it
1879 * can lead to duplicate data or errors as explained in
1880 * the comment above.
1881 */
1884 LookupGXactBySubid(subid))
1889
1890 /* Change system catalog accordingly */
1896 }
1897
1899 {
1900 /*
1901 * Similar to the two_phase case above, we need to update
1902 * the failover option for both the slot and the
1903 * subscription.
1904 */
1906
1908 isTopLevel);
1909
1913 }
1914
1916 {
1920
1921 /*
1922 * Update the retention status only if there's a change in
1923 * the retain_dead_tuples option value.
1924 *
1925 * Automatically marking retention as active when
1926 * retain_dead_tuples is enabled may not always be ideal,
1927 * especially if retention was previously stopped and the
1928 * user toggles retain_dead_tuples without adjusting the
1929 * publisher workload. However, this behavior provides a
1930 * convenient way for users to manually refresh the
1931 * retention status. Since retention will be stopped again
1932 * unless the publisher workload is reduced, this approach
1933 * is acceptable for now.
1934 */
1936 {
1940
1942 }
1943
1945
1946 /*
1947 * Workers may continue running even after the
1948 * subscription has been disabled.
1949 *
1950 * To prevent race conditions (as described in
1951 * CheckAlterSubOption()), ensure that all worker
1952 * processes have already exited before proceeding.
1953 */
1959
1960 /*
1961 * Notify the launcher to manage the replication slot for
1962 * conflict detection. This ensures that replication slot
1963 * is efficiently handled (created, updated, or dropped)
1964 * in response to any configuration changes.
1965 */
1966 ApplyLauncherWakeupAtCommit();
1967
1970 }
1971
1973 {
1977
1979 }
1980
1981 /*
1982 * Ensure that system configuration parameters are set
1983 * appropriately to support retain_dead_tuples and
1984 * max_retention_duration.
1985 */
1989 retain_dead_tuples,
1990 retention_active,
1991 (max_retention > 0));
1992
1994 {
1998
1999 /*
2000 * Check if changes from different origins may be received
2001 * from the publisher when the origin is changed to ANY
2002 * and retain_dead_tuples is enabled. Use |= so that we
2003 * don't clear the flag already set when
2004 * retain_dead_tuples was changed in the same command.
2005 */
2008
2009 origin = opts.origin;
2010 }
2011
2013 {
2017 }
2018
2020 {
2023
2025 {
2028
2032
2034 old_dest,
2035 opts.conflictlogdest,
2036 &relid);
2038 {
2040 ObjectIdGetDatum(relid);
2042 true;
2043 }
2044 }
2045 }
2046
2048 break;
2049 }
2050
2052 {
2054
2059
2060 /*
2061 * Check track_commit_timestamp only when enabling the
2062 * subscription in case it was disabled after creation. See
2063 * comments atop CheckSubDeadTupleRetention() for details.
2064 */
2068
2072
2074 ApplyLauncherWakeupAtCommit();
2075
2077
2078 /*
2079 * The subscription might be initially created with
2080 * connect=false and retain_dead_tuples=true, meaning the
2081 * remote server's status may not be checked. Ensure this
2082 * check is conducted now.
2083 */
2085 break;
2086 }
2087
2089 {
2093
2094 /*
2095 * Remove what was there before, either another foreign server
2096 * or a connection string.
2097 */
2099 {
2101 DEPENDENCY_NORMAL,
2103 }
2104 else
2105 {
2108 }
2109
2110 /*
2111 * Check that the subscription owner has USAGE privileges on
2112 * the server.
2113 */
2116 new_server->serverid,
2123 new_server->servername));
2124
2125 /* make sure a user mapping exists */
2127
2129 new_server);
2130
2131 /* Load the library providing us libpq calls. */
2133 /* Check the connection info string. */
2136
2139
2142
2144 }
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 /* remove reference to foreign server and dependencies, if present */
2157 {
2159 DEPENDENCY_NORMAL,
2161
2164 }
2165
2167
2168 /* Load the library providing us libpq calls. */
2170 /* Check the connection info string. */
2173
2178
2179 /*
2180 * Since the remote server configuration might have changed,
2181 * perform a check to ensure it permits enabling
2182 * retain_dead_tuples.
2183 */
2185 break;
2186
2188 {
2192
2194
2195 /* Refresh if user asked us to. */
2197 {
2203
2204 /*
2205 * See ALTER_SUBSCRIPTION_REFRESH_PUBLICATION for details
2206 * why this is not allowed.
2207 */
2211 errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when two_phase is enabled"),
2212 errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION with refresh = false, or with copy_data = false, or use DROP/CREATE SUBSCRIPTION.")));
2213
2215
2216 /* Make sure refresh sees the new list of publications. */
2218
2220 stmt->publication);
2221 }
2222
2223 break;
2224 }
2225
2228 {
2231
2236
2238
2239 /* Refresh if user asked us to. */
2241 {
2242 /* We only need to validate user specified publications. */
2244
2249 /* translator: %s is an SQL ALTER command */
2251 isadd ?
2252 "ALTER SUBSCRIPTION ... ADD PUBLICATION ... WITH (refresh = false)" :
2253 "ALTER SUBSCRIPTION ... DROP PUBLICATION ... WITH (refresh = false)")));
2254
2255 /*
2256 * See ALTER_SUBSCRIPTION_REFRESH_PUBLICATION for details
2257 * why this is not allowed.
2258 */
2262 errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when two_phase is enabled"),
2263 /* translator: %s is an SQL ALTER command */
2264 errhint("Use %s with refresh = false, or with copy_data = false, or use DROP/CREATE SUBSCRIPTION.",
2265 isadd ?
2266 "ALTER SUBSCRIPTION ... ADD PUBLICATION" :
2267 "ALTER SUBSCRIPTION ... DROP PUBLICATION")));
2268
2270
2271 /* Refresh the new list of publications. */
2273
2275 validate_publications);
2276 }
2277
2278 break;
2279 }
2280
2282 {
2287 "ALTER SUBSCRIPTION ... REFRESH PUBLICATION")));
2288
2289 /*
2290 * The subscription option "two_phase" requires that
2291 * replication has passed the initial table synchronization
2292 * phase before the two_phase becomes properly enabled.
2293 *
2294 * But, having reached this two-phase commit "enabled" state
2295 * we must not allow any subsequent table initialization to
2296 * occur. So the ALTER SUBSCRIPTION ... REFRESH PUBLICATION is
2297 * disallowed when the user had requested two_phase = on mode.
2298 *
2299 * The exception to this restriction is when copy_data =
2300 * false, because when copy_data is false the tablesync will
2301 * start already in READY state and will exit directly without
2302 * doing anything.
2303 *
2304 * For more details see comments atop worker.c.
2305 */
2309 errmsg("ALTER SUBSCRIPTION ... REFRESH PUBLICATION with copy_data is not allowed when two_phase is enabled"),
2310 errhint("Use ALTER SUBSCRIPTION ... REFRESH PUBLICATION with copy_data = false, or use DROP/CREATE SUBSCRIPTION.")));
2311
2313
2315
2316 break;
2317 }
2318
2320 {
2325 "ALTER SUBSCRIPTION ... REFRESH SEQUENCES"));
2326
2327 AlterSubscription_refresh_seq(sub);
2328
2329 break;
2330 }
2331
2333 {
2334 /* ALTER SUBSCRIPTION ... SKIP supports only LSN option */
2336
2337 /*
2338 * If the user sets subskiplsn, we do a sanity check to make
2339 * sure that the specified LSN is a probable value.
2340 */
2342 {
2345 XLogRecPtr remote_lsn;
2346
2351
2352 /* Check the given LSN is at least a future LSN */
2358 LSN_FORMAT_ARGS(remote_lsn))));
2359 }
2360
2363
2365 break;
2366 }
2367
2368 default:
2370 stmt->kind);
2371 }
2372
2373 /* Update the catalog if needed. */
2375 {
2377 replaces);
2378
2380
2382 }
2383
2384 /*
2385 * Try to acquire the connection necessary either for modifying the slot
2386 * or for checking if the remote server permits enabling
2387 * retain_dead_tuples.
2388 *
2389 * This has to be at the end because otherwise if there is an error while
2390 * doing the database operations we won't be able to rollback altered
2391 * slot.
2392 */
2394 {
2398
2400
2401 /* Load the library providing us libpq calls. */
2403
2404 /*
2405 * Try to connect to the publisher, using the new connection string if
2406 * available.
2407 */
2411 &err);
2417
2418 PG_TRY();
2419 {
2422
2425 sub->name);
2426
2431 }
2432 PG_FINALLY();
2433 {
2435 }
2436 PG_END_TRY();
2437 }
2438
2440
2442
2443 /* Wake up related replication workers to handle this change quickly. */
2444 LogicalRepWorkersWakeupAtCommit(subid);
2445
2447}
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:6342
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:4608
@ ALTER_SUBSCRIPTION_DROP_PUBLICATION
Definition parsenodes.h:4607
@ ALTER_SUBSCRIPTION_REFRESH_SEQUENCES
Definition parsenodes.h:4609
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:95
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:1446
static Datum publicationListToArray(List *publist)
Definition subscriptioncmds.c:640
static bool alter_sub_conflict_log_dest(Subscription *sub, ConflictLogDest oldlogdest, ConflictLogDest newlogdest, Oid *conflicttablerelid)
Definition subscriptioncmds.c:1549
static void AlterSubscription_refresh(Subscription *sub, bool copy_data, List *validate_publications)
Definition subscriptioncmds.c:1090
static void parse_subscription_options(ParseState *pstate, List *stmt_options, uint32 supported_opts, SubOpts *opts)
Definition subscriptioncmds.c:162
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:3402
static void AlterSubscription_refresh_seq(Subscription *sub)
Definition subscriptioncmds.c:1361
static List * merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname)
Definition subscriptioncmds.c:3674
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:3076
static void check_pub_dead_tuple_retention(WalReceiverConn *wrconn)
Definition subscriptioncmds.c:3341
#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 1090 of file subscriptioncmds.c.
1092{
1102 int off;
1107 {
1108 Oid relid;
1110 } SubRemoveRels;
1111
1114
1115 /* Load the library providing us libpq calls. */
1117
1118 /* Try to connect to the publisher. */
1127
1128 PG_TRY();
1129 {
1132
1133 /* Get the relation list from publisher. */
1135
1136 /* Get local relation list. */
1139
1140 /*
1141 * Build qsorted arrays of local table oids and sequence oids for
1142 * faster lookup. This can potentially contain all tables and
1143 * sequences in the database so speed of lookup is important.
1144 *
1145 * We do not yet know the exact count of tables and sequences, so we
1146 * allocate separate arrays for table OIDs and sequence OIDs based on
1147 * the total number of relations (subrel_count).
1148 */
1152 {
1154
1157 else
1159 }
1160
1165 sub->name);
1166
1169 copy_data, sub->origin,
1171 sub->name);
1172
1173 /*
1174 * Walk over the remote relations and try to match them to locally
1175 * known relations. If the relation is not known locally create a new
1176 * state for it.
1177 *
1178 * Also builds array of local oids of remote relations for the next
1179 * step.
1180 */
1181 off = 0;
1183
1185 {
1187 Oid relid;
1188 char relkind;
1189
1191 relkind = get_rel_relkind(relid);
1192
1193 /* Check for supported relkind. */
1196
1197 pubrel_local_oids[off++] = relid;
1198
1203 {
1211 }
1212 }
1213
1214 /*
1215 * Next remove state for tables we should not care about anymore using
1216 * the data we collected above
1217 */
1219
1221 {
1223
1226 {
1228 XLogRecPtr statelsn;
1230
1231 /*
1232 * Lock pg_subscription_rel with AccessExclusiveLock to
1233 * prevent any race conditions with the apply worker
1234 * re-launching workers at the same time this code is trying
1235 * to remove those tables.
1236 *
1237 * Even if new worker for this particular rel is restarted it
1238 * won't be able to make any progress as we hold exclusive
1239 * lock on pg_subscription_rel till the transaction end. It
1240 * will simply exit as there is no corresponding rel entry.
1241 *
1242 * This locking also ensures that the state of rels won't
1243 * change till we are done with this refresh operation.
1244 */
1245 if (!rel)
1247
1248 /* Last known rel state. */
1250
1252
1253 remove_rel->relid = relid;
1255
1257
1259
1260 /*
1261 * For READY state, we would have already dropped the
1262 * tablesync origin.
1263 */
1265 {
1267
1268 /*
1269 * Drop the tablesync's origin tracking if exists.
1270 *
1271 * It is possible that the origin is not yet created for
1272 * tablesync worker, this can happen for the states before
1273 * SUBREL_STATE_DATASYNC. The tablesync worker or apply
1274 * worker can also concurrently try to drop the origin and
1275 * by this time the origin might be already removed. For
1276 * these reasons, passing missing_ok = true.
1277 */
1281 }
1282
1286 get_rel_name(relid),
1287 sub->name)));
1288 }
1289 }
1290
1291 /*
1292 * Drop the tablesync slots associated with removed tables. This has
1293 * to be at the end because otherwise if there is an error while doing
1294 * the database operations we won't be able to rollback dropped slots.
1295 */
1297 {
1300 {
1302
1303 /*
1304 * For READY/SYNCDONE states we know the tablesync slot has
1305 * already been dropped by the tablesync worker.
1306 *
1307 * For other states, there is no certainty, maybe the slot
1308 * does not exist yet. Also, if we fail after removing some of
1309 * the slots, next time, it will again try to drop already
1310 * dropped slots and fail. For these reasons, we allow
1311 * missing_ok = true for the drop.
1312 */
1316 }
1317 }
1318
1319 /*
1320 * Next remove state for sequences we should not care about anymore
1321 * using the data we collected above
1322 */
1324 {
1326
1329 {
1330 /*
1331 * This locking ensures that the state of rels won't change
1332 * till we are done with this refresh operation.
1333 */
1334 if (!rel)
1336
1338
1342 get_rel_name(relid),
1343 sub->name));
1344 }
1345 }
1346 }
1347 PG_FINALLY();
1348 {
1350 }
1351 PG_END_TRY();
1352
1353 if (rel)
1355}
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:506
char GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn)
Definition pg_subscription.c:456
void AddSubscriptionRelState(Oid subid, Oid relid, char state, XLogRecPtr sublsn, bool retain_lock)
Definition pg_subscription.c:336
List * GetSubscriptionRelations(Oid subid, bool tables, bool sequences, bool not_ready)
Definition pg_subscription.c:634
Definition subscriptioncmds.c:124
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:3224
static void check_publications(WalReceiverConn *wrconn, List *publications)
Definition subscriptioncmds.c:575
static List * fetch_relation_list(WalReceiverConn *wrconn, List *publications)
Definition subscriptioncmds.c:3466
void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok)
Definition subscriptioncmds.c:2840
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 1361 of file subscriptioncmds.c.
1362{
1367
1368 /* Load the library providing us libpq calls. */
1370
1371 /* Try to connect to the publisher. */
1380
1381 /* The publisher connection is only needed for the origin check. */
1382 PG_TRY();
1383 {
1386 }
1387 PG_FINALLY();
1388 {
1390 }
1391 PG_END_TRY();
1392
1393 /*
1394 * Reset the sequences to INIT so they get re-synchronized with the latest
1395 * publisher values.
1396 *
1397 * A sequence sync worker may already be running. If it has fetched a
1398 * sequence's value from the publisher but not yet marked it READY, it
1399 * must not be allowed to complete that update, as it would overwrite the
1400 * reset below with a stale value and silently lose this refresh request.
1401 * So we stop any running sequence sync worker before resetting the
1402 * states.
1403 *
1404 * This is race-free because AlterSubscription() already holds
1405 * AccessExclusiveLock on the subscription object. That lock blocks a
1406 * running worker's update of sequence state to READY, see
1407 * UpdateSubscriptionRelState() which takes AccessShareLock on the object.
1408 * It also blocks any worker the apply worker re-launches, because a new
1409 * worker takes AccessShareLock on the object before it reads
1410 * pg_subscription_rel, see InitializeLogRepWorker(). Such a worker cannot
1411 * act on the states until we commit, by which time they are reset to INIT
1412 * and it will sync the latest values.
1413 */
1414#ifdef USE_ASSERT_CHECKING
1415 {
1416 LOCKTAG tag;
1417
1420 }
1421#endif
1422
1424
1425 /* Reset every local sequence of this subscription to INIT. */
1428 {
1430
1436 get_rel_name(relid),
1437 sub->name));
1438 }
1439}
bool LockHeldByMe(const LOCKTAG *locktag, LOCKMODE lockmode, bool orstronger)
Definition lock.c:640
#define SET_LOCKTAG_OBJECT(locktag, dboid, classoid, objoid, objsubid)
Definition locktag.h:162
void UpdateSubscriptionRelState(Oid subid, Oid relid, char state, XLogRecPtr sublsn, bool already_locked)
Definition pg_subscription.c:390
Definition locktag.h:65
References AccessExclusiveLock, Assert, 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(), InvalidOid, InvalidXLogRecPtr, load_file(), LockHeldByMe(), logicalrep_worker_stop(), Subscription::name, Subscription::oid, Subscription::origin, Subscription::ownersuperuser, Subscription::passwordrequired, PG_END_TRY, PG_FINALLY, PG_TRY, Subscription::publications, SET_LOCKTAG_OBJECT, UpdateSubscriptionRelState(), walrcv_connect, walrcv_disconnect, WORKERTYPE_SEQUENCESYNC, and wrconn.
Referenced by AlterSubscription().
◆ AlterSubscriptionOwner()
| ObjectAddress AlterSubscriptionOwner | ( | const char * | name, |
| Oid | newOwnerId | ||
| ) |
Definition at line 2986 of file subscriptioncmds.c.
2987{
2988 Oid subid;
2990 Relation rel;
2991 ObjectAddress address;
2993
2995
2998
3003
3005 subid = form->oid;
3006
3008
3010
3012
3014
3015 return address;
3016}
static void AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
Definition subscriptioncmds.c:2897
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 2897 of file subscriptioncmds.c.
2898{
2901
2903
2904 /* Must only alter subscriptions belonging to the current database. */
2906
2908 return;
2909
2913
2914 /*
2915 * Don't allow non-superuser modification of a subscription with
2916 * password_required=false.
2917 */
2922 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
2923
2924 /* Must be able to become new owner */
2926
2927 /*
2928 * current owner must have CREATE on database
2929 *
2930 * This is consistent with how ALTER SCHEMA ... OWNER TO works, but some
2931 * other object types behave differently (e.g. you can't give a table to a
2932 * user who lacks CREATE privileges on a schema).
2933 */
2939
2940 /*
2941 * If the subscription uses a server, check that the new owner has USAGE
2942 * privileges on the server and that a user mapping exists. Note: does not
2943 * re-check the resulting connection string.
2944 */
2946 {
2948
2955 server->servername));
2956
2957 /* make sure a user mapping exists */
2959 }
2960
2963
2964 /* Update owner of the conflict log table if it exists. */
2967 AccessExclusiveLock);
2968
2969 /* Update owner dependency reference */
2971 form->oid,
2972 newOwnerId);
2973
2975 form->oid, 0);
2976
2977 /* Wake up related background processes to handle this change quickly. */
2978 ApplyLauncherWakeupAtCommit();
2980}
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:16777
References AccessExclusiveLock, ACL_CREATE, ACL_USAGE, aclcheck_error(), ACLCHECK_NOT_OWNER, ACLCHECK_OK, ApplyLauncherWakeupAtCommit(), Assert, 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 3022 of file subscriptioncmds.c.
3023{
3025 Relation rel;
3027
3029
3031
3036
3038
3039 /*
3040 * Don't process subscriptions belonging to other databases. While
3041 * pg_subscription is a shared catalog, subscriptions refer to db-local
3042 * objects which exist only in the database identified by subdbid.
3043 */
3046
3048
3050}
References AlterSubscriptionOwner_internal(), ereport, errcode(), errmsg, ERROR, fb(), Form_pg_subscription, GETSTRUCT(), heap_freetuple(), HeapTupleIsValid, MyDatabaseId, ObjectIdGetDatum(), RowExclusiveLock, SearchSysCacheCopy1, table_close(), and table_open().
Referenced by shdepReassignOwned_Owner().
◆ appendQuotedString()
Definition at line 554 of file subscriptioncmds.c.
555{
558 {
560
564 }
566}
References appendStringInfoChar(), buf, and str.
◆ check_duplicates_in_publist()
Definition at line 3634 of file subscriptioncmds.c.
3635{
3636 ListCell *cell;
3638
3640 {
3643
3645 {
3647
3649 break;
3650
3655 pname)));
3656 }
3657
3658 if (datums)
3660 }
3661}
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 3341 of file subscriptioncmds.c.
3342{
3343 WalRcvExecResult *res;
3345 TupleTableSlot *slot;
3346 bool isnull;
3348
3352 errmsg("cannot enable retain_dead_tuples if the publisher is running a version earlier than PostgreSQL 19"));
3353
3355
3360 res->err)));
3361
3365
3367
3372
3373 ExecDropSingleTupleTableSlot(slot);
3374
3375 walrcv_clear_result(res);
3376}
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 575 of file subscriptioncmds.c.
576{
577 WalRcvExecResult *res;
578 StringInfoData cmd;
579 TupleTableSlot *slot;
582
583 initStringInfo(&cmd);
585 " pg_catalog.pg_publication t WHERE\n"
586 " t.pubname IN (");
589
592
596 res->err));
597
599
600 /* Process publication(s). */
603 {
604 char *pubname;
605 bool isnull;
606
608 Assert(!isnull);
609
610 /* Delete the publication present in publisher from the list. */
612 ExecClearTuple(slot);
613 }
614
615 ExecDropSingleTupleTableSlot(slot);
616
617 walrcv_clear_result(res);
618
620 {
621 /* Prepare the list of non-existent publication(s) for error message. */
623
625
630 "publications %s do not exist on the publisher",
632 pubnames.data));
633 }
634}
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 3224 of file subscriptioncmds.c.
3228{
3229 WalRcvExecResult *res;
3230 StringInfoData cmd;
3231 TupleTableSlot *slot;
3234
3235 /*
3236 * Enable sequence synchronization checks only when origin is 'none' , to
3237 * ensure that sequence data from other origins is not inadvertently
3238 * copied. This check is necessary if the publisher is running PG19 or
3239 * later, where logical replication sequence synchronization is supported.
3240 */
3243 return;
3244
3245 initStringInfo(&cmd);
3246 appendStringInfoString(&cmd,
3247 "SELECT DISTINCT P.pubname AS pubname\n"
3248 "FROM pg_publication P,\n"
3249 " LATERAL pg_get_publication_sequences(P.pubname) GPS\n"
3250 " JOIN pg_subscription_rel PS ON (GPS.relid = PS.srrelid),\n"
3251 " pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)\n"
3252 "WHERE C.oid = GPS.relid AND P.pubname IN (");
3253
3256
3257 /*
3258 * In case of ALTER SUBSCRIPTION ... REFRESH PUBLICATION,
3259 * subrel_local_oids contains the list of relations that are already
3260 * present on the subscriber. This check should be skipped as these will
3261 * not be re-synced.
3262 */
3264 {
3270
3271 appendStringInfo(&cmd,
3272 "AND NOT (N.nspname = %s AND C.relname = %s)\n",
3274
3277 }
3278
3281
3286 res->err)));
3287
3288 /* Process publications. */
3291 {
3292 char *pubname;
3293 bool isnull;
3294
3296 Assert(!isnull);
3297
3298 ExecClearTuple(slot);
3300 }
3301
3302 /*
3303 * Log a warning if the publisher has subscribed to the same sequence from
3304 * some other publisher. We cannot know the origin of sequences data
3305 * during the initial sync.
3306 */
3308 {
3310
3311 /* Prepare the list of publication(s) for warning message. */
3314
3317 errmsg("subscription \"%s\" requested copy_data with origin = NONE but might copy data that had a different origin",
3318 subname),
3319 errdetail_plural("The subscription subscribes to a publication (%s) that contains sequences that are written to by other subscriptions.",
3320 "The subscription subscribes to publications (%s) that contain sequences that are written to by other subscriptions.",
3322 errhint("Verify that initial data copied from the publisher sequences did not come from other origins."));
3323 }
3324
3325 ExecDropSingleTupleTableSlot(slot);
3326
3327 walrcv_clear_result(res);
3328}
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 3076 of file subscriptioncmds.c.
3080{
3081 WalRcvExecResult *res;
3082 StringInfoData cmd;
3083 TupleTableSlot *slot;
3091
3092 /*
3093 * Enable retain_dead_tuples checks only when origin is set to 'any',
3094 * since with origin='none' only local changes are replicated to the
3095 * subscriber.
3096 */
3098
3099 /*
3100 * Enable table synchronization checks only when origin is 'none', to
3101 * ensure that data from other origins is not inadvertently copied.
3102 */
3104
3105 /* retain_dead_tuples and table sync checks occur separately */
3107
3108 /* Return if no checks are required */
3110 return;
3111
3112 initStringInfo(&cmd);
3113 appendStringInfoString(&cmd,
3114 "SELECT DISTINCT P.pubname AS pubname\n"
3115 "FROM pg_publication P,\n"
3116 " LATERAL pg_get_publication_tables(P.pubname) GPT\n"
3117 " JOIN pg_subscription_rel PS ON (GPT.relid = PS.srrelid OR"
3118 " GPT.relid IN (SELECT relid FROM pg_partition_ancestors(PS.srrelid) UNION"
3119 " SELECT relid FROM pg_partition_tree(PS.srrelid))),\n"
3120 " pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)\n"
3121 "WHERE C.oid = GPT.relid AND P.pubname IN (");
3124
3125 /*
3126 * In case of ALTER SUBSCRIPTION ... REFRESH PUBLICATION,
3127 * subrel_local_oids contains the list of relation oids that are already
3128 * present on the subscriber. This check should be skipped for these
3129 * tables if checking for table sync scenario. However, when handling the
3130 * retain_dead_tuples scenario, ensure all tables are checked, as some
3131 * existing tables may now include changes from other origins due to newly
3132 * created subscriptions on the publisher.
3133 */
3135 {
3137 {
3143
3146
3149 }
3150 }
3151
3154
3159 res->err)));
3160
3161 /* Process publications. */
3164 {
3165 char *pubname;
3166 bool isnull;
3167
3169 Assert(!isnull);
3170
3171 ExecClearTuple(slot);
3173 }
3174
3175 /*
3176 * Log a warning if the publisher has subscribed to the same table from
3177 * some other publisher. We cannot know the origin of data during the
3178 * initial sync. Data origins can be found only from the WAL by looking at
3179 * the origin id.
3180 *
3181 * XXX: For simplicity, we don't check whether the table has any data or
3182 * not. If the table doesn't have any data then we don't need to
3183 * distinguish between data having origin and data not having origin so we
3184 * can avoid logging a warning for table sync scenario.
3185 */
3187 {
3189
3190 /* Prepare the list of publication(s) for warning message. */
3193
3197 errmsg("subscription \"%s\" requested copy_data with origin = NONE but might copy data that had a different origin",
3198 subname),
3199 errdetail_plural("The subscription subscribes to a publication (%s) that contains tables that are written to by other subscriptions.",
3200 "The subscription subscribes to publications (%s) that contain tables that are written to by other subscriptions.",
3202 errhint("Verify that initial data copied from the publisher tables did not come from other origins."));
3203 else
3206 errmsg("subscription \"%s\" enabled retain_dead_tuples but might not reliably detect conflicts for changes from different origins",
3207 subname),
3208 errdetail_plural("The subscription subscribes to a publication (%s) that contains tables that are written to by other subscriptions.",
3209 "The subscription subscribes to publications (%s) that contain tables that are written to by other subscriptions.",
3212 }
3213
3214 ExecDropSingleTupleTableSlot(slot);
3215
3216 walrcv_clear_result(res);
3217}
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 1446 of file subscriptioncmds.c.
1448{
1452
1453 /*
1454 * Altering the retain_dead_tuples option does not update the slot on the
1455 * publisher.
1456 */
1458
1459 /*
1460 * Do not allow changing the option if the subscription is enabled. This
1461 * is because both failover and two_phase options of the slot on the
1462 * publisher cannot be modified if the slot is currently acquired by the
1463 * existing walsender.
1464 *
1465 * Note that two_phase is enabled (aka changed from 'false' to 'true') on
1466 * the publisher by the existing walsender, so we could have allowed that
1467 * even when the subscription is enabled. But we kept this restriction for
1468 * the sake of consistency and simplicity.
1469 *
1470 * Additionally, do not allow changing the retain_dead_tuples option when
1471 * the subscription is enabled to prevent race conditions arising from the
1472 * new option value being acknowledged asynchronously by the launcher and
1473 * apply workers.
1474 *
1475 * Without the restriction, a race condition may arise when a user
1476 * disables and immediately re-enables the retain_dead_tuples option. In
1477 * this case, the launcher might drop the slot upon noticing the disabled
1478 * action, while the apply worker may keep maintaining
1479 * oldest_nonremovable_xid without noticing the option change. During this
1480 * period, a transaction ID wraparound could falsely make this ID appear
1481 * as if it originates from the future w.r.t the transaction ID stored in
1482 * the slot maintained by launcher.
1483 *
1484 * Similarly, if the user enables retain_dead_tuples concurrently with the
1485 * launcher starting the worker, the apply worker may start calculating
1486 * oldest_nonremovable_xid before the launcher notices the enable action.
1487 * Consequently, the launcher may update slot.xmin to a newer value than
1488 * that maintained by the worker. In subsequent cycles, upon integrating
1489 * the worker's oldest_nonremovable_xid, the launcher might detect a
1490 * retreat in the calculated xmin, necessitating additional handling.
1491 *
1492 * XXX To address the above race conditions, we can define
1493 * oldest_nonremovable_xid as FullTransactionId and adds the check to
1494 * disallow retreating the conflict slot's xmin. For now, we kept the
1495 * implementation simple by disallowing change to the retain_dead_tuples,
1496 * but in the future we can change this after some more analysis.
1497 *
1498 * Note that we could restrict only the enabling of retain_dead_tuples to
1499 * avoid the race conditions described above, but we maintain the
1500 * restriction for both enable and disable operations for the sake of
1501 * consistency.
1502 */
1507 option)));
1508
1510 {
1511 StringInfoData cmd;
1512
1513 /*
1514 * A valid slot must be associated with the subscription for us to
1515 * modify any of the slot's properties.
1516 */
1521 option)));
1522
1523 /* The changed option of the slot can't be rolled back. */
1524 initStringInfo(&cmd);
1526
1529 }
1530}
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 3402 of file subscriptioncmds.c.
3406{
3409
3411 {
3415 errmsg("\"wal_level\" is insufficient to create the replication slot required by retain_dead_tuples"),
3417
3423 "track_commit_timestamp"));
3424
3428 errmsg("deleted rows to detect conflicts would not be removed until the subscription is enabled"),
3431 "retain_dead_tuples") : 0);
3432 }
3434 {
3438 }
3439}
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 2458 of file subscriptioncmds.c.
2459{
2461 ForeignServer *server;
2462
2464
2466
2470 {
2471 /*
2472 * Unable to generate connection string because permissions on the
2473 * foreign server have been removed. Follow the same logic as an
2474 * unusable subconninfo (which will result in an ERROR later unless
2475 * slot_name = NONE).
2476 */
2477 *err = psprintf(_("subscription owner \"%s\" does not have permission on foreign server \"%s\""),
2479 server->servername);
2481 }
2482
2484}
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 670 of file subscriptioncmds.c.
672{
673 Relation rel;
675 Oid subid;
680 Oid serverid;
681 char *conninfo;
683 List *publications;
688
689 /*
690 * Parse and check options.
691 *
692 * Connection and publication should not be specified here.
693 */
703 SUBOPT_CONFLICT_LOG_DEST);
705
706 /*
707 * Since creating a replication slot is not transactional, rolling back
708 * the transaction leaves the created replication slot. So we cannot run
709 * CREATE SUBSCRIPTION inside a transaction block if creating a
710 * replication slot.
711 */
714
715 /*
716 * We don't want to allow unprivileged users to be able to trigger
717 * attempts to access arbitrary network destinations, so require the user
718 * to have been specifically authorized to create subscriptions.
719 */
725 "pg_create_subscription")));
726
727 /*
728 * Since a subscription is a database object, we also check for CREATE
729 * permission on the database.
730 */
732 owner, ACL_CREATE);
736
737 /*
738 * Non-superusers are required to set a password for authentication, and
739 * that password must be used by the target server, but the superuser can
740 * exempt a subscription from this requirement.
741 */
746 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
747
748 /*
749 * If built with appropriate switch, whine when regression-testing
750 * conventions for subscription names are violated.
751 */
752#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
754 elog(WARNING, "subscriptions created by regression test cases should have names starting with \"regress_\"");
755#endif
756
758
759 /* Check if name is used */
763 {
767 stmt->subname)));
768 }
769
770 /*
771 * Ensure that system configuration parameters are set appropriately to
772 * support retain_dead_tuples and max_retention_duration.
773 */
776 (opts.maxretention > 0));
777
781
782 /* The default for synchronous_commit of subscriptions is off. */
785
786 /*
787 * The default for wal_receiver_timeout of subscriptions is -1, which
788 * means the value is inherited from the server configuration, command
789 * line, or role/database settings.
790 */
793
794 /* Load the library providing us libpq calls. */
796
798 {
799 ForeignServer *server;
800
802 conninfo = NULL;
803
808
809 /* make sure a user mapping exists */
811
812 serverid = server->serverid;
813 conninfo = ForeignServerConnectionString(owner, server);
814 }
815 else
816 {
818
819 serverid = InvalidOid;
820 conninfo = stmt->conninfo;
821 }
822
823 /* Check the connection info string. */
825
826 publications = stmt->publication;
827
828 /* Everything ok, form a new tuple. */
831
833 Anum_pg_subscription_oid);
860 CStringGetTextDatum(conninfo);
861 else
866 else
873 publicationListToArray(publications);
876
879
880 /*
881 * We create the conflict log table here, if required, so that its
882 * relation OID can be stored when inserting the pg_subscription tuple
883 * below.
884 */
887
888 /* Store table OID in the catalog. */
891
893
894 /* Insert tuple into catalog. */
897
899
901
903 {
905
907
910 }
911
912 /*
913 * Establish an internal dependency between the conflict log table and the
914 * subscription.
915 *
916 * We use DEPENDENCY_INTERNAL to signify that the table's lifecycle is
917 * strictly tied to the subscription, similar to how a TOAST table relates
918 * to its main table or a sequence relates to an identity column.
919 *
920 * This ensures the conflict log table is automatically reaped during a
921 * DROP SUBSCRIPTION via performDeletion().
922 */
924 {
926
929 }
930
931 /*
932 * A replication origin is currently created for all subscriptions,
933 * including those that only contain sequences or are otherwise empty.
934 *
935 * XXX: While this is technically unnecessary, optimizing it would require
936 * additional logic to skip origin creation during DDL operations and
937 * apply workers initialization, and to handle origin creation dynamically
938 * when tables are added to the subscription. It is not clear whether
939 * preventing creation of origins is worth additional complexity.
940 */
943
944 /*
945 * Connect to remote side to execute requested commands and fetch table
946 * and sequence info.
947 */
949 {
953
954 /* Try to connect to the publisher. */
963
964 PG_TRY();
965 {
969
972 opts.copy_data,
978
981
982 /*
983 * Set sync state based on if we were asked to do data copy or
984 * not.
985 */
987
988 /*
989 * Build local relation status info. Relations are for both tables
990 * and sequences from the publisher.
991 */
993
995 {
996 Oid relid;
997 char relkind;
999
1001 relkind = get_rel_relkind(relid);
1002
1003 /* Check for supported relkind. */
1009 }
1010
1011 /*
1012 * If requested, create permanent slot for the subscription. We
1013 * won't use the initial snapshot for anything, so no need to
1014 * export it.
1015 *
1016 * XXX: Similar to origins, it is not clear whether preventing the
1017 * slot creation for empty and sequence-only subscriptions is
1018 * worth additional complexity.
1019 */
1021 {
1023
1025
1026 /*
1027 * Even if two_phase is set, don't create the slot with
1028 * two-phase enabled. Will enable it once all the tables are
1029 * synced and ready. This avoids race-conditions like prepared
1030 * transactions being skipped due to changes not being applied
1031 * due to checks in should_apply_changes_for_rel() when
1032 * tablesync for the corresponding tables are in progress. See
1033 * comments atop worker.c.
1034 *
1035 * Note that if tables were specified but copy_data is false
1036 * then it is safe to enable two_phase up-front because those
1037 * tables are already initially in READY state. When the
1038 * subscription has no tables, we leave the twophase state as
1039 * PENDING, to allow ALTER SUBSCRIPTION ... REFRESH
1040 * PUBLICATION to work.
1041 */
1044
1047
1050
1053 opts.slot_name)));
1054 }
1055 }
1056 PG_FINALLY();
1057 {
1059 }
1060 PG_END_TRY();
1061 }
1062 else
1065 errhint("To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.")));
1066
1068
1069 pgstat_create_subscription(subid);
1070
1071 /*
1072 * Notify the launcher to start the apply worker if the subscription is
1073 * enabled, or to create the conflict detection slot if retain_dead_tuples
1074 * is enabled.
1075 *
1076 * Creating the conflict detection slot is essential even when the
1077 * subscription is not enabled. This ensures that dead tuples are
1078 * retained, which is necessary for accurately identifying the type of
1079 * conflict during replication.
1080 */
1082 ApplyLauncherWakeupAtCommit();
1083
1085
1087}
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 3733 of file subscriptioncmds.c.
3734{
3735 /*
3736 * If no parameter value given, assume "true" is meant.
3737 */
3740
3741 /*
3742 * Allow 0, 1, "false", "true", "off", "on" or "parallel".
3743 */
3745 {
3748 {
3749 case 0:
3751 case 1:
3753 default:
3754 /* otherwise, error out below */
3755 break;
3756 }
3757 break;
3758 default:
3759 {
3761
3762 /*
3763 * The set of strings accepted here should match up with the
3764 * grammar's opt_boolean_or_string production.
3765 */
3774 }
3775 break;
3776 }
3777
3781 def->defname)));
3783}
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 2502 of file subscriptioncmds.c.
2503{
2504 /* Drop any dependent conflict log table */
2506 {
2507 ObjectAddress object;
2509
2513 subconflictlogrelid);
2514
2515 /*
2516 * By using PERFORM_DELETION_SKIP_ORIGINAL, we ensure that only the
2517 * conflict log table is deleted while the subscription remains.
2518 */
2521 PERFORM_DELETION_INTERNAL |
2523
2527 subname));
2528 }
2529}
void performDeletion(const ObjectAddress *object, DropBehavior behavior, int flags)
Definition dependency.c:279
char * get_qualified_objname(Oid nspid, char *objname)
Definition lsyscache.c:3720
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 2535 of file subscriptioncmds.c.
2536{
2537 Relation rel;
2540 Oid subid;
2545 Datum datum;
2546 bool isnull;
2549 char *slotname;
2558
2559 /*
2560 * The launcher may concurrently start a new worker for this subscription.
2561 * During initialization, the worker checks for subscription validity and
2562 * exits if the subscription has already been dropped. See
2563 * InitializeLogRepWorker.
2564 */
2566
2569
2571 {
2573
2578 stmt->subname)));
2579 else
2582 stmt->subname)));
2583
2584 return;
2585 }
2586
2588 Anum_pg_subscription_subconninfo, &isnull);
2589 if (!isnull)
2590 subconninfo = TextDatumGetCString(datum);
2591
2593 subid = form->oid;
2598
2599 /* must be owner */
2602 stmt->subname);
2603
2604 /* DROP hook for the subscription being removed */
2606
2607 /*
2608 * Lock the subscription so nobody else can do anything with it (including
2609 * the replication workers).
2610 */
2612
2613 /* Get subname */
2615 Anum_pg_subscription_subname);
2617
2618 /* Get slotname */
2620 Anum_pg_subscription_subslotname, &isnull);
2621 if (!isnull)
2623 else
2624 slotname = NULL;
2625
2626 /*
2627 * Since dropping a replication slot is not transactional, the replication
2628 * slot stays dropped even if the transaction rolls back. So we cannot
2629 * run DROP SUBSCRIPTION inside a transaction block if dropping the
2630 * replication slot. Also, in this case, we report a message for dropping
2631 * the subscription to the cumulative stats system.
2632 *
2633 * XXX The command name should really be something like "DROP SUBSCRIPTION
2634 * of a subscription that is associated with a replication slot", but we
2635 * don't have the proper facilities for that.
2636 */
2637 if (slotname)
2639
2642
2643 /* Remove the tuple from catalog. */
2645
2647
2648 /*
2649 * Stop all the subscription workers immediately.
2650 *
2651 * This is necessary if we are dropping the replication slot, so that the
2652 * slot becomes accessible.
2653 *
2654 * It is also necessary if the subscription is disabled and was disabled
2655 * in the same transaction. Then the workers haven't seen the disabling
2656 * yet and will still be running, leading to hangs later when we want to
2657 * drop the replication origin. If the subscription was disabled before
2658 * this transaction, then there shouldn't be any workers left, so this
2659 * won't make a difference.
2660 *
2661 * New workers won't be started because we hold an exclusive lock on the
2662 * subscription till the end of the transaction.
2663 */
2666 {
2668
2670 }
2672
2673 /*
2674 * Remove the no-longer-useful entry in the launcher's table of apply
2675 * worker start times.
2676 *
2677 * If this transaction rolls back, the launcher might restart a failed
2678 * apply worker before wal_retrieve_retry_interval milliseconds have
2679 * elapsed, but that's pretty harmless.
2680 */
2681 ApplyLauncherForgetWorkerStartTime(subid);
2682
2683 /*
2684 * Cleanup of tablesync replication origins.
2685 *
2686 * Any READY-state relations would already have dealt with clean-ups.
2687 *
2688 * Note that the state can't change because we have already stopped both
2689 * the apply and tablesync workers and they can't restart because of
2690 * exclusive lock on the subscription.
2691 */
2694 {
2697
2698 /* Only cleanup resources of tablesync workers */
2700 continue;
2701
2702 /*
2703 * Drop the tablesync's origin tracking if exists.
2704 *
2705 * It is possible that the origin is not yet created for tablesync
2706 * worker so passing missing_ok = true. This can happen for the states
2707 * before SUBREL_STATE_DATASYNC.
2708 */
2712 }
2713
2714 /* Drop subscription's conflict log table */
2716
2717 /* Clean up dependencies */
2720
2721 /* Remove any associated relation synchronization states. */
2723
2724 /* Remove the origin tracking if exists. */
2727
2728 /*
2729 * Tell the cumulative stats system that the subscription is getting
2730 * dropped.
2731 */
2732 pgstat_drop_subscription(subid);
2733
2734 /*
2735 * If there is no slot associated with the subscription, we can finish
2736 * here.
2737 */
2739 {
2741 return;
2742 }
2743
2744 /*
2745 * Try to acquire the connection necessary for dropping slots.
2746 *
2747 * Note: If the slotname is NONE/NULL then we allow the command to finish
2748 * and users need to manually cleanup the apply and tablesync worker slots
2749 * later.
2750 *
2751 * This has to be at the end because otherwise if there is an error while
2752 * doing the database operations we won't be able to rollback dropped
2753 * slot.
2754 */
2756
2759 else
2760 conninfo = subconninfo;
2761
2762 if (conninfo)
2765
2767 {
2768 if (!slotname)
2769 {
2770 /* be tidy */
2773 return;
2774 }
2775 else
2776 {
2778 }
2779 }
2780
2781 PG_TRY();
2782 {
2784 {
2787
2788 /* Only cleanup resources of tablesync workers */
2790 continue;
2791
2792 /*
2793 * Drop the tablesync slots associated with removed tables.
2794 *
2795 * For SYNCDONE/READY states, the tablesync slot is known to have
2796 * already been dropped by the tablesync worker.
2797 *
2798 * For other states, there is no certainty, maybe the slot does
2799 * not exist yet. Also, if we fail after removing some of the
2800 * slots, next time, it will again try to drop already dropped
2801 * slots and fail. For these reasons, we allow missing_ok = true
2802 * for the drop.
2803 */
2805 {
2807
2811 }
2812 }
2813
2815
2816 /*
2817 * If there is a slot associated with the subscription, then drop the
2818 * replication slot at the publisher.
2819 */
2820 if (slotname)
2822 }
2823 PG_FINALLY();
2824 {
2826 }
2827 PG_END_TRY();
2828
2830}
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:2458
static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err)
Definition subscriptioncmds.c:3590
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 3466 of file subscriptioncmds.c.
3467{
3468 WalRcvExecResult *res;
3469 StringInfoData cmd;
3470 TupleTableSlot *slot;
3476 StringInfoData pub_names;
3477
3478 initStringInfo(&cmd);
3479 initStringInfo(&pub_names);
3480
3481 /* Build the pub_names comma-separated string. */
3483
3484 /* Get the list of relations from the publisher */
3486 {
3488
3489 /*
3490 * From version 16, we allowed passing multiple publications to the
3491 * function pg_get_publication_tables. This helped to filter out the
3492 * partition table whose ancestor is also published in this
3493 * publication array.
3494 *
3495 * Join pg_get_publication_tables with pg_publication to exclude
3496 * non-existing publications.
3497 *
3498 * Note that attrs are always stored in sorted order so we don't need
3499 * to worry if different publications have specified them in a
3500 * different order. See pub_collist_validate.
3501 */
3503 " FROM pg_class c\n"
3504 " JOIN pg_namespace n ON n.oid = c.relnamespace\n"
3505 " JOIN ( SELECT (pg_get_publication_tables(VARIADIC array_agg(pubname::text))).*\n"
3506 " FROM pg_publication\n"
3507 " WHERE pubname IN ( %s )) AS gpt\n"
3508 " ON gpt.relid = c.oid\n",
3509 pub_names.data);
3510
3511 /* From version 19, inclusion of sequences in the target is supported */
3513 appendStringInfo(&cmd,
3514 "UNION ALL\n"
3515 " SELECT DISTINCT s.schemaname, s.sequencename, " CppAsString2(RELKIND_SEQUENCE) "::\"char\" AS relkind, NULL::int2vector AS attrs\n"
3516 " FROM pg_catalog.pg_publication_sequences s\n"
3517 " WHERE s.pubname IN ( %s )",
3518 pub_names.data);
3519 }
3520 else
3521 {
3523 appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename, " CppAsString2(RELKIND_RELATION) "::\"char\" AS relkind \n");
3524
3525 /* Get column lists for each relation if the publisher supports it */
3528
3530 " WHERE t.pubname IN ( %s )",
3531 pub_names.data);
3532 }
3533
3535
3538
3543 res->err)));
3544
3545 /* Process tables. */
3548 {
3549 char *nspname;
3551 bool isnull;
3552 char relkind;
3554
3556 Assert(!isnull);
3558 Assert(!isnull);
3560 Assert(!isnull);
3561
3563 relinfo->relkind = relkind;
3564
3566 check_columnlist &&
3571 nspname, relname));
3572 else
3574
3575 ExecClearTuple(slot);
3576 }
3577 ExecDropSingleTupleTableSlot(slot);
3578
3579 walrcv_clear_result(res);
3580
3582}
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:3445
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 3445 of file subscriptioncmds.c.
3446{
3448 {
3450 return true;
3451 }
3452
3453 return false;
3454}
References equal(), fb(), and foreach_ptr.
Referenced by fetch_relation_list().
◆ merge_publications()
|
static |
Definition at line 3674 of file subscriptioncmds.c.
3675{
3677
3679
3681
3683 {
3686 bool found = false;
3687
3689 {
3691
3693 {
3694 found = true;
3700 else
3702
3703 break;
3704 }
3705 }
3706
3714 }
3715
3716 /*
3717 * XXX Probably no strong reason for this, but for now it's to make ALTER
3718 * SUBSCRIPTION ... DROP PUBLICATION consistent with SET PUBLICATION.
3719 */
3724
3726}
static void check_duplicates_in_publist(List *publist, Datum *datums)
Definition subscriptioncmds.c:3634
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 162 of file subscriptioncmds.c.
164{
166
167 /* Start out with cleared opts. */
169
170 /* caller must expect some option */
172
173 /* If connect option is supported, these others also need to be. */
176 SUBOPT_COPY_DATA));
177
178 /* Set default values for the supported options. */
206 opts->maxretention = 0;
211
212 /* Parse options */
214 {
216
219 {
222
225 }
228 {
231
234 }
237 {
240
243 }
246 {
249
252
253 /* Setting slot_name = NONE is treated as no slot name. */
256 else
258 }
261 {
264
267 }
270 {
273
276
277 /* Test if the given value is valid for synchronous_commit GUC. */
280 false, 0, false);
281 }
284 {
287
290 }
293 {
296
299 }
302 {
305
308 }
311 {
314
317 }
320 {
323
326 }
329 {
332
335 }
338 {
341
344 }
347 {
350
353 }
356 {
359
362 }
365 {
368
371
376 }
379 {
382
385
386 /*
387 * Even though the "origin" parameter allows only "none" and "any"
388 * values, it is implemented as a string type so that the
389 * parameter can be extended in future versions to support
390 * filtering using origin names specified by the user.
391 */
393
399 }
402 {
404 XLogRecPtr lsn;
405
408
409 /* Setting lsn = NONE is treated as resetting LSN */
411 lsn = InvalidXLogRecPtr;
412 else
413 {
414 /* Parse the argument as LSN */
417
422 }
423
425 opts->lsn = lsn;
426 }
429 {
432
435
438
439 /*
440 * Test if the given value is valid for wal_receiver_timeout GUC.
441 * Skip this test if the value is -1, since -1 is allowed for the
442 * wal_receiver_timeout subscription option, but not for the GUC
443 * itself.
444 */
449 false, 0, false);
450 }
453 {
455
458
462 }
463 else
467 }
468
469 /*
470 * We've been explicitly asked to not connect, that requires some
471 * additional processing.
472 */
474 {
475 /* Check for incompatible options from the user. */
480 /*- translator: both %s are strings of the form "option = value" */
482 "connect = false", "enabled = true")));
483
489 "connect = false", "create_slot = true")));
490
496 "connect = false", "copy_data = true")));
497
498 /* Change the defaults of other options. */
502 }
503
504 /*
505 * Do additional checking for disallowed combination when slot_name = NONE
506 * was used.
507 */
510 {
512 {
516 /*- translator: both %s are strings of the form "option = value" */
518 "slot_name = NONE", "enabled = true")));
519 else
522 /*- translator: both %s are strings of the form "option = value" */
524 "slot_name = NONE", "enabled = false")));
525 }
526
528 {
532 /*- translator: both %s are strings of the form "option = value" */
534 "slot_name = NONE", "create_slot = true")));
535 else
538 /*- translator: both %s are strings of the form "option = value" */
540 "slot_name = NONE", "create_slot = false")));
541 }
542 }
543}
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:859
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 640 of file subscriptioncmds.c.
641{
642 ArrayType *arr;
643 Datum *datums;
644 MemoryContext memcxt;
646
647 /* Create memory context for temporary allocations. */
649 "publicationListToArray to array",
650 ALLOCSET_DEFAULT_SIZES);
652
654
656
658
660
661 MemoryContextDelete(memcxt);
662
664}
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 2840 of file subscriptioncmds.c.
2841{
2842 StringInfoData cmd;
2843
2845
2847
2848 initStringInfo(&cmd);
2850 appendQuotedIdentifier(&cmd, slotname);
2852
2853 PG_TRY();
2854 {
2855 WalRcvExecResult *res;
2856
2858
2860 {
2861 /* NOTICE. Success. */
2864 slotname)));
2865 }
2867 missing_ok &&
2869 {
2870 /* LOG. Error, but missing_ok = true. */
2873 slotname, res->err)));
2874 }
2875 else
2876 {
2877 /* ERROR. */
2881 slotname, res->err)));
2882 }
2883
2884 walrcv_clear_result(res);
2885 }
2886 PG_FINALLY();
2887 {
2889 }
2890 PG_END_TRY();
2891}
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 3590 of file subscriptioncmds.c.
3591{
3593
3595 {
3598
3599 /* Only cleanup resources of tablesync workers */
3601 continue;
3602
3603 /*
3604 * Caller needs to ensure that relstate doesn't change underneath us.
3605 * See DropSubscription where we get the relstates.
3606 */
3608 {
3610
3614 syncslotname);
3615 }
3616 }
3617
3621 slotname, err),
3622 /* translator: %s is an SQL ALTER command */
3623 errhint("Use %s to disable the subscription, and then use %s to disassociate it from the slot.",
3624 "ALTER SUBSCRIPTION ... DISABLE",
3625 "ALTER SUBSCRIPTION ... SET (slot_name = NONE)")));
3626}
References elog, ereport, err(), errcode(), errhint(), errmsg, ERROR, fb(), lfirst, NAMEDATALEN, OidIsValid, SubscriptionRelState::relid, ReplicationSlotNameForTablesync(), SubscriptionRelState::state, and WARNING.
Referenced by DropSubscription().
