LCOV - code coverage report
Current view: top level - src/backend/catalog - pg_subscription.c (source / functions) Hit Total Coverage
Test: PostgreSQL 18devel Lines: 170 179 95.0 %
Date: 2024-07-18 17:11:54 Functions: 11 11 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * pg_subscription.c
       4             :  *      replication subscriptions
       5             :  *
       6             :  * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
       7             :  * Portions Copyright (c) 1994, Regents of the University of California
       8             :  *
       9             :  * IDENTIFICATION
      10             :  *      src/backend/catalog/pg_subscription.c
      11             :  *
      12             :  *-------------------------------------------------------------------------
      13             :  */
      14             : 
      15             : #include "postgres.h"
      16             : 
      17             : #include "access/genam.h"
      18             : #include "access/heapam.h"
      19             : #include "access/htup_details.h"
      20             : #include "access/tableam.h"
      21             : #include "catalog/indexing.h"
      22             : #include "catalog/pg_subscription.h"
      23             : #include "catalog/pg_subscription_rel.h"
      24             : #include "catalog/pg_type.h"
      25             : #include "miscadmin.h"
      26             : #include "storage/lmgr.h"
      27             : #include "utils/array.h"
      28             : #include "utils/builtins.h"
      29             : #include "utils/fmgroids.h"
      30             : #include "utils/lsyscache.h"
      31             : #include "utils/pg_lsn.h"
      32             : #include "utils/rel.h"
      33             : #include "utils/syscache.h"
      34             : 
      35             : static List *textarray_to_stringlist(ArrayType *textarray);
      36             : 
      37             : /*
      38             :  * Fetch the subscription from the syscache.
      39             :  */
      40             : Subscription *
      41        1292 : GetSubscription(Oid subid, bool missing_ok)
      42             : {
      43             :     HeapTuple   tup;
      44             :     Subscription *sub;
      45             :     Form_pg_subscription subform;
      46             :     Datum       datum;
      47             :     bool        isnull;
      48             : 
      49        1292 :     tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
      50             : 
      51        1290 :     if (!HeapTupleIsValid(tup))
      52             :     {
      53           0 :         if (missing_ok)
      54           0 :             return NULL;
      55             : 
      56           0 :         elog(ERROR, "cache lookup failed for subscription %u", subid);
      57             :     }
      58             : 
      59        1290 :     subform = (Form_pg_subscription) GETSTRUCT(tup);
      60             : 
      61        1290 :     sub = (Subscription *) palloc(sizeof(Subscription));
      62        1290 :     sub->oid = subid;
      63        1290 :     sub->dbid = subform->subdbid;
      64        1290 :     sub->skiplsn = subform->subskiplsn;
      65        1290 :     sub->name = pstrdup(NameStr(subform->subname));
      66        1290 :     sub->owner = subform->subowner;
      67        1290 :     sub->enabled = subform->subenabled;
      68        1290 :     sub->binary = subform->subbinary;
      69        1290 :     sub->stream = subform->substream;
      70        1290 :     sub->twophasestate = subform->subtwophasestate;
      71        1290 :     sub->disableonerr = subform->subdisableonerr;
      72        1290 :     sub->passwordrequired = subform->subpasswordrequired;
      73        1290 :     sub->runasowner = subform->subrunasowner;
      74        1290 :     sub->failover = subform->subfailover;
      75             : 
      76             :     /* Get conninfo */
      77        1290 :     datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
      78             :                                    tup,
      79             :                                    Anum_pg_subscription_subconninfo);
      80        1290 :     sub->conninfo = TextDatumGetCString(datum);
      81             : 
      82             :     /* Get slotname */
      83        1290 :     datum = SysCacheGetAttr(SUBSCRIPTIONOID,
      84             :                             tup,
      85             :                             Anum_pg_subscription_subslotname,
      86             :                             &isnull);
      87        1290 :     if (!isnull)
      88        1224 :         sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
      89             :     else
      90          66 :         sub->slotname = NULL;
      91             : 
      92             :     /* Get synccommit */
      93        1290 :     datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
      94             :                                    tup,
      95             :                                    Anum_pg_subscription_subsynccommit);
      96        1290 :     sub->synccommit = TextDatumGetCString(datum);
      97             : 
      98             :     /* Get publications */
      99        1290 :     datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
     100             :                                    tup,
     101             :                                    Anum_pg_subscription_subpublications);
     102        1290 :     sub->publications = textarray_to_stringlist(DatumGetArrayTypeP(datum));
     103             : 
     104             :     /* Get origin */
     105        1290 :     datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
     106             :                                    tup,
     107             :                                    Anum_pg_subscription_suborigin);
     108        1290 :     sub->origin = TextDatumGetCString(datum);
     109             : 
     110             :     /* Is the subscription owner a superuser? */
     111        1290 :     sub->ownersuperuser = superuser_arg(sub->owner);
     112             : 
     113        1290 :     ReleaseSysCache(tup);
     114             : 
     115        1290 :     return sub;
     116             : }
     117             : 
     118             : /*
     119             :  * Return number of subscriptions defined in given database.
     120             :  * Used by dropdb() to check if database can indeed be dropped.
     121             :  */
     122             : int
     123          66 : CountDBSubscriptions(Oid dbid)
     124             : {
     125          66 :     int         nsubs = 0;
     126             :     Relation    rel;
     127             :     ScanKeyData scankey;
     128             :     SysScanDesc scan;
     129             :     HeapTuple   tup;
     130             : 
     131          66 :     rel = table_open(SubscriptionRelationId, RowExclusiveLock);
     132             : 
     133          66 :     ScanKeyInit(&scankey,
     134             :                 Anum_pg_subscription_subdbid,
     135             :                 BTEqualStrategyNumber, F_OIDEQ,
     136             :                 ObjectIdGetDatum(dbid));
     137             : 
     138          66 :     scan = systable_beginscan(rel, InvalidOid, false,
     139             :                               NULL, 1, &scankey);
     140             : 
     141          66 :     while (HeapTupleIsValid(tup = systable_getnext(scan)))
     142           0 :         nsubs++;
     143             : 
     144          66 :     systable_endscan(scan);
     145             : 
     146          66 :     table_close(rel, NoLock);
     147             : 
     148          66 :     return nsubs;
     149             : }
     150             : 
     151             : /*
     152             :  * Free memory allocated by subscription struct.
     153             :  */
     154             : void
     155          66 : FreeSubscription(Subscription *sub)
     156             : {
     157          66 :     pfree(sub->name);
     158          66 :     pfree(sub->conninfo);
     159          66 :     if (sub->slotname)
     160          66 :         pfree(sub->slotname);
     161          66 :     list_free_deep(sub->publications);
     162          66 :     pfree(sub);
     163          66 : }
     164             : 
     165             : /*
     166             :  * Disable the given subscription.
     167             :  */
     168             : void
     169           8 : DisableSubscription(Oid subid)
     170             : {
     171             :     Relation    rel;
     172             :     bool        nulls[Natts_pg_subscription];
     173             :     bool        replaces[Natts_pg_subscription];
     174             :     Datum       values[Natts_pg_subscription];
     175             :     HeapTuple   tup;
     176             : 
     177             :     /* Look up the subscription in the catalog */
     178           8 :     rel = table_open(SubscriptionRelationId, RowExclusiveLock);
     179           8 :     tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
     180             : 
     181           8 :     if (!HeapTupleIsValid(tup))
     182           0 :         elog(ERROR, "cache lookup failed for subscription %u", subid);
     183             : 
     184           8 :     LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
     185             : 
     186             :     /* Form a new tuple. */
     187           8 :     memset(values, 0, sizeof(values));
     188           8 :     memset(nulls, false, sizeof(nulls));
     189           8 :     memset(replaces, false, sizeof(replaces));
     190             : 
     191             :     /* Set the subscription to disabled. */
     192           8 :     values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(false);
     193           8 :     replaces[Anum_pg_subscription_subenabled - 1] = true;
     194             : 
     195             :     /* Update the catalog */
     196           8 :     tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
     197             :                             replaces);
     198           8 :     CatalogTupleUpdate(rel, &tup->t_self, tup);
     199           8 :     heap_freetuple(tup);
     200             : 
     201           8 :     table_close(rel, NoLock);
     202           8 : }
     203             : 
     204             : /*
     205             :  * Convert text array to list of strings.
     206             :  *
     207             :  * Note: the resulting list of strings is pallocated here.
     208             :  */
     209             : static List *
     210        1290 : textarray_to_stringlist(ArrayType *textarray)
     211             : {
     212             :     Datum      *elems;
     213             :     int         nelems,
     214             :                 i;
     215        1290 :     List       *res = NIL;
     216             : 
     217        1290 :     deconstruct_array_builtin(textarray, TEXTOID, &elems, NULL, &nelems);
     218             : 
     219        1290 :     if (nelems == 0)
     220           0 :         return NIL;
     221             : 
     222        3240 :     for (i = 0; i < nelems; i++)
     223        1950 :         res = lappend(res, makeString(TextDatumGetCString(elems[i])));
     224             : 
     225        1290 :     return res;
     226             : }
     227             : 
     228             : /*
     229             :  * Add new state record for a subscription table.
     230             :  *
     231             :  * If retain_lock is true, then don't release the locks taken in this function.
     232             :  * We normally release the locks at the end of transaction but in binary-upgrade
     233             :  * mode, we expect to release those immediately.
     234             :  */
     235             : void
     236         350 : AddSubscriptionRelState(Oid subid, Oid relid, char state,
     237             :                         XLogRecPtr sublsn, bool retain_lock)
     238             : {
     239             :     Relation    rel;
     240             :     HeapTuple   tup;
     241             :     bool        nulls[Natts_pg_subscription_rel];
     242             :     Datum       values[Natts_pg_subscription_rel];
     243             : 
     244         350 :     LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
     245             : 
     246         350 :     rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
     247             : 
     248             :     /* Try finding existing mapping. */
     249         350 :     tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
     250             :                               ObjectIdGetDatum(relid),
     251             :                               ObjectIdGetDatum(subid));
     252         350 :     if (HeapTupleIsValid(tup))
     253           0 :         elog(ERROR, "subscription table %u in subscription %u already exists",
     254             :              relid, subid);
     255             : 
     256             :     /* Form the tuple. */
     257         350 :     memset(values, 0, sizeof(values));
     258         350 :     memset(nulls, false, sizeof(nulls));
     259         350 :     values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
     260         350 :     values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
     261         350 :     values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
     262         350 :     if (sublsn != InvalidXLogRecPtr)
     263           2 :         values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
     264             :     else
     265         348 :         nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
     266             : 
     267         350 :     tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
     268             : 
     269             :     /* Insert tuple into catalog. */
     270         350 :     CatalogTupleInsert(rel, tup);
     271             : 
     272         350 :     heap_freetuple(tup);
     273             : 
     274             :     /* Cleanup. */
     275         350 :     if (retain_lock)
     276             :     {
     277         346 :         table_close(rel, NoLock);
     278             :     }
     279             :     else
     280             :     {
     281           4 :         table_close(rel, RowExclusiveLock);
     282           4 :         UnlockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
     283             :     }
     284         350 : }
     285             : 
     286             : /*
     287             :  * Update the state of a subscription table.
     288             :  */
     289             : void
     290        1296 : UpdateSubscriptionRelState(Oid subid, Oid relid, char state,
     291             :                            XLogRecPtr sublsn)
     292             : {
     293             :     Relation    rel;
     294             :     HeapTuple   tup;
     295             :     bool        nulls[Natts_pg_subscription_rel];
     296             :     Datum       values[Natts_pg_subscription_rel];
     297             :     bool        replaces[Natts_pg_subscription_rel];
     298             : 
     299        1296 :     LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
     300             : 
     301        1296 :     rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
     302             : 
     303             :     /* Try finding existing mapping. */
     304        1296 :     tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
     305             :                               ObjectIdGetDatum(relid),
     306             :                               ObjectIdGetDatum(subid));
     307        1296 :     if (!HeapTupleIsValid(tup))
     308           0 :         elog(ERROR, "subscription table %u in subscription %u does not exist",
     309             :              relid, subid);
     310             : 
     311             :     /* Update the tuple. */
     312        1296 :     memset(values, 0, sizeof(values));
     313        1296 :     memset(nulls, false, sizeof(nulls));
     314        1296 :     memset(replaces, false, sizeof(replaces));
     315             : 
     316        1296 :     replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
     317        1296 :     values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
     318             : 
     319        1296 :     replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
     320        1296 :     if (sublsn != InvalidXLogRecPtr)
     321         636 :         values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
     322             :     else
     323         660 :         nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
     324             : 
     325        1296 :     tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
     326             :                             replaces);
     327             : 
     328             :     /* Update the catalog. */
     329        1296 :     CatalogTupleUpdate(rel, &tup->t_self, tup);
     330             : 
     331             :     /* Cleanup. */
     332        1296 :     table_close(rel, NoLock);
     333        1296 : }
     334             : 
     335             : /*
     336             :  * Get state of subscription table.
     337             :  *
     338             :  * Returns SUBREL_STATE_UNKNOWN when the table is not in the subscription.
     339             :  */
     340             : char
     341        2034 : GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn)
     342             : {
     343             :     HeapTuple   tup;
     344             :     char        substate;
     345             :     bool        isnull;
     346             :     Datum       d;
     347             :     Relation    rel;
     348             : 
     349             :     /*
     350             :      * This is to avoid the race condition with AlterSubscription which tries
     351             :      * to remove this relstate.
     352             :      */
     353        2034 :     rel = table_open(SubscriptionRelRelationId, AccessShareLock);
     354             : 
     355             :     /* Try finding the mapping. */
     356        2034 :     tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
     357             :                           ObjectIdGetDatum(relid),
     358             :                           ObjectIdGetDatum(subid));
     359             : 
     360        2034 :     if (!HeapTupleIsValid(tup))
     361             :     {
     362          50 :         table_close(rel, AccessShareLock);
     363          50 :         *sublsn = InvalidXLogRecPtr;
     364          50 :         return SUBREL_STATE_UNKNOWN;
     365             :     }
     366             : 
     367             :     /* Get the state. */
     368        1984 :     substate = ((Form_pg_subscription_rel) GETSTRUCT(tup))->srsubstate;
     369             : 
     370             :     /* Get the LSN */
     371        1984 :     d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
     372             :                         Anum_pg_subscription_rel_srsublsn, &isnull);
     373        1984 :     if (isnull)
     374        1072 :         *sublsn = InvalidXLogRecPtr;
     375             :     else
     376         912 :         *sublsn = DatumGetLSN(d);
     377             : 
     378             :     /* Cleanup */
     379        1984 :     ReleaseSysCache(tup);
     380             : 
     381        1984 :     table_close(rel, AccessShareLock);
     382             : 
     383        1984 :     return substate;
     384             : }
     385             : 
     386             : /*
     387             :  * Drop subscription relation mapping. These can be for a particular
     388             :  * subscription, or for a particular relation, or both.
     389             :  */
     390             : void
     391       42694 : RemoveSubscriptionRel(Oid subid, Oid relid)
     392             : {
     393             :     Relation    rel;
     394             :     TableScanDesc scan;
     395             :     ScanKeyData skey[2];
     396             :     HeapTuple   tup;
     397       42694 :     int         nkeys = 0;
     398             : 
     399       42694 :     rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
     400             : 
     401       42694 :     if (OidIsValid(subid))
     402             :     {
     403         212 :         ScanKeyInit(&skey[nkeys++],
     404             :                     Anum_pg_subscription_rel_srsubid,
     405             :                     BTEqualStrategyNumber,
     406             :                     F_OIDEQ,
     407             :                     ObjectIdGetDatum(subid));
     408             :     }
     409             : 
     410       42694 :     if (OidIsValid(relid))
     411             :     {
     412       42518 :         ScanKeyInit(&skey[nkeys++],
     413             :                     Anum_pg_subscription_rel_srrelid,
     414             :                     BTEqualStrategyNumber,
     415             :                     F_OIDEQ,
     416             :                     ObjectIdGetDatum(relid));
     417             :     }
     418             : 
     419             :     /* Do the search and delete what we found. */
     420       42694 :     scan = table_beginscan_catalog(rel, nkeys, skey);
     421       42866 :     while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
     422             :     {
     423             :         Form_pg_subscription_rel subrel;
     424             : 
     425         172 :         subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
     426             : 
     427             :         /*
     428             :          * We don't allow to drop the relation mapping when the table
     429             :          * synchronization is in progress unless the caller updates the
     430             :          * corresponding subscription as well. This is to ensure that we don't
     431             :          * leave tablesync slots or origins in the system when the
     432             :          * corresponding table is dropped.
     433             :          */
     434         172 :         if (!OidIsValid(subid) && subrel->srsubstate != SUBREL_STATE_READY)
     435             :         {
     436           0 :             ereport(ERROR,
     437             :                     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
     438             :                      errmsg("could not drop relation mapping for subscription \"%s\"",
     439             :                             get_subscription_name(subrel->srsubid, false)),
     440             :                      errdetail("Table synchronization for relation \"%s\" is in progress and is in state \"%c\".",
     441             :                                get_rel_name(relid), subrel->srsubstate),
     442             : 
     443             :             /*
     444             :              * translator: first %s is a SQL ALTER command and second %s is a
     445             :              * SQL DROP command
     446             :              */
     447             :                      errhint("Use %s to enable subscription if not already enabled or use %s to drop the subscription.",
     448             :                              "ALTER SUBSCRIPTION ... ENABLE",
     449             :                              "DROP SUBSCRIPTION ...")));
     450             :         }
     451             : 
     452         172 :         CatalogTupleDelete(rel, &tup->t_self);
     453             :     }
     454       42694 :     table_endscan(scan);
     455             : 
     456       42694 :     table_close(rel, RowExclusiveLock);
     457       42694 : }
     458             : 
     459             : /*
     460             :  * Does the subscription have any relations?
     461             :  *
     462             :  * Use this function only to know true/false, and when you have no need for the
     463             :  * List returned by GetSubscriptionRelations.
     464             :  */
     465             : bool
     466         384 : HasSubscriptionRelations(Oid subid)
     467             : {
     468             :     Relation    rel;
     469             :     ScanKeyData skey[1];
     470             :     SysScanDesc scan;
     471             :     bool        has_subrels;
     472             : 
     473         384 :     rel = table_open(SubscriptionRelRelationId, AccessShareLock);
     474             : 
     475         384 :     ScanKeyInit(&skey[0],
     476             :                 Anum_pg_subscription_rel_srsubid,
     477             :                 BTEqualStrategyNumber, F_OIDEQ,
     478             :                 ObjectIdGetDatum(subid));
     479             : 
     480         384 :     scan = systable_beginscan(rel, InvalidOid, false,
     481             :                               NULL, 1, skey);
     482             : 
     483             :     /* If even a single tuple exists then the subscription has tables. */
     484         384 :     has_subrels = HeapTupleIsValid(systable_getnext(scan));
     485             : 
     486             :     /* Cleanup */
     487         384 :     systable_endscan(scan);
     488         384 :     table_close(rel, AccessShareLock);
     489             : 
     490         384 :     return has_subrels;
     491             : }
     492             : 
     493             : /*
     494             :  * Get the relations for the subscription.
     495             :  *
     496             :  * If not_ready is true, return only the relations that are not in a ready
     497             :  * state, otherwise return all the relations of the subscription.  The
     498             :  * returned list is palloc'ed in the current memory context.
     499             :  */
     500             : List *
     501        1652 : GetSubscriptionRelations(Oid subid, bool not_ready)
     502             : {
     503        1652 :     List       *res = NIL;
     504             :     Relation    rel;
     505             :     HeapTuple   tup;
     506        1652 :     int         nkeys = 0;
     507             :     ScanKeyData skey[2];
     508             :     SysScanDesc scan;
     509             : 
     510        1652 :     rel = table_open(SubscriptionRelRelationId, AccessShareLock);
     511             : 
     512        1652 :     ScanKeyInit(&skey[nkeys++],
     513             :                 Anum_pg_subscription_rel_srsubid,
     514             :                 BTEqualStrategyNumber, F_OIDEQ,
     515             :                 ObjectIdGetDatum(subid));
     516             : 
     517        1652 :     if (not_ready)
     518        1596 :         ScanKeyInit(&skey[nkeys++],
     519             :                     Anum_pg_subscription_rel_srsubstate,
     520             :                     BTEqualStrategyNumber, F_CHARNE,
     521             :                     CharGetDatum(SUBREL_STATE_READY));
     522             : 
     523        1652 :     scan = systable_beginscan(rel, InvalidOid, false,
     524             :                               NULL, nkeys, skey);
     525             : 
     526        3942 :     while (HeapTupleIsValid(tup = systable_getnext(scan)))
     527             :     {
     528             :         Form_pg_subscription_rel subrel;
     529             :         SubscriptionRelState *relstate;
     530             :         Datum       d;
     531             :         bool        isnull;
     532             : 
     533        2290 :         subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
     534             : 
     535        2290 :         relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
     536        2290 :         relstate->relid = subrel->srrelid;
     537        2290 :         relstate->state = subrel->srsubstate;
     538        2290 :         d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
     539             :                             Anum_pg_subscription_rel_srsublsn, &isnull);
     540        2290 :         if (isnull)
     541        1808 :             relstate->lsn = InvalidXLogRecPtr;
     542             :         else
     543         482 :             relstate->lsn = DatumGetLSN(d);
     544             : 
     545        2290 :         res = lappend(res, relstate);
     546             :     }
     547             : 
     548             :     /* Cleanup */
     549        1652 :     systable_endscan(scan);
     550        1652 :     table_close(rel, AccessShareLock);
     551             : 
     552        1652 :     return res;
     553             : }

Generated by: LCOV version 1.14