2017-01-19 18:00:00 +01:00
|
|
|
/*-------------------------------------------------------------------------
|
|
|
|
*
|
|
|
|
* pg_subscription.c
|
|
|
|
* replication subscriptions
|
|
|
|
*
|
2017-01-25 18:32:05 +01:00
|
|
|
* Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group
|
|
|
|
* Portions Copyright (c) 1994, Regents of the University of California
|
2017-01-19 18:00:00 +01:00
|
|
|
*
|
|
|
|
* IDENTIFICATION
|
|
|
|
* src/backend/catalog/pg_subscription.c
|
|
|
|
*
|
|
|
|
*-------------------------------------------------------------------------
|
|
|
|
*/
|
|
|
|
|
|
|
|
#include "postgres.h"
|
|
|
|
|
|
|
|
#include "miscadmin.h"
|
|
|
|
|
|
|
|
#include "access/genam.h"
|
|
|
|
#include "access/heapam.h"
|
|
|
|
#include "access/htup_details.h"
|
2017-03-23 13:36:36 +01:00
|
|
|
#include "access/xact.h"
|
2017-01-19 18:00:00 +01:00
|
|
|
|
2017-03-23 13:36:36 +01:00
|
|
|
#include "catalog/indexing.h"
|
2017-01-19 18:00:00 +01:00
|
|
|
#include "catalog/pg_type.h"
|
|
|
|
#include "catalog/pg_subscription.h"
|
2017-03-23 13:36:36 +01:00
|
|
|
#include "catalog/pg_subscription_rel.h"
|
2017-01-19 18:00:00 +01:00
|
|
|
|
|
|
|
#include "nodes/makefuncs.h"
|
|
|
|
|
|
|
|
#include "utils/array.h"
|
|
|
|
#include "utils/builtins.h"
|
|
|
|
#include "utils/fmgroids.h"
|
2017-03-23 13:36:36 +01:00
|
|
|
#include "utils/pg_lsn.h"
|
|
|
|
#include "utils/rel.h"
|
2017-01-19 18:00:00 +01:00
|
|
|
#include "utils/syscache.h"
|
|
|
|
|
|
|
|
|
|
|
|
static List *textarray_to_stringlist(ArrayType *textarray);
|
|
|
|
|
|
|
|
/*
|
|
|
|
* Fetch the subscription from the syscache.
|
|
|
|
*/
|
|
|
|
Subscription *
|
|
|
|
GetSubscription(Oid subid, bool missing_ok)
|
|
|
|
{
|
2017-05-17 22:31:56 +02:00
|
|
|
HeapTuple tup;
|
|
|
|
Subscription *sub;
|
|
|
|
Form_pg_subscription subform;
|
|
|
|
Datum datum;
|
|
|
|
bool isnull;
|
2017-01-19 18:00:00 +01:00
|
|
|
|
|
|
|
tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
|
|
|
|
|
|
|
|
if (!HeapTupleIsValid(tup))
|
|
|
|
{
|
|
|
|
if (missing_ok)
|
|
|
|
return NULL;
|
|
|
|
|
|
|
|
elog(ERROR, "cache lookup failed for subscription %u", subid);
|
|
|
|
}
|
|
|
|
|
|
|
|
subform = (Form_pg_subscription) GETSTRUCT(tup);
|
|
|
|
|
|
|
|
sub = (Subscription *) palloc(sizeof(Subscription));
|
|
|
|
sub->oid = subid;
|
|
|
|
sub->dbid = subform->subdbid;
|
|
|
|
sub->name = pstrdup(NameStr(subform->subname));
|
|
|
|
sub->owner = subform->subowner;
|
|
|
|
sub->enabled = subform->subenabled;
|
|
|
|
|
|
|
|
/* Get conninfo */
|
|
|
|
datum = SysCacheGetAttr(SUBSCRIPTIONOID,
|
|
|
|
tup,
|
|
|
|
Anum_pg_subscription_subconninfo,
|
|
|
|
&isnull);
|
|
|
|
Assert(!isnull);
|
2017-04-14 18:54:09 +02:00
|
|
|
sub->conninfo = TextDatumGetCString(datum);
|
2017-01-19 18:00:00 +01:00
|
|
|
|
|
|
|
/* Get slotname */
|
|
|
|
datum = SysCacheGetAttr(SUBSCRIPTIONOID,
|
|
|
|
tup,
|
|
|
|
Anum_pg_subscription_subslotname,
|
|
|
|
&isnull);
|
2017-05-09 16:20:42 +02:00
|
|
|
if (!isnull)
|
|
|
|
sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
|
|
|
|
else
|
|
|
|
sub->slotname = NULL;
|
2017-01-19 18:00:00 +01:00
|
|
|
|
2017-04-14 19:58:46 +02:00
|
|
|
/* Get synccommit */
|
|
|
|
datum = SysCacheGetAttr(SUBSCRIPTIONOID,
|
|
|
|
tup,
|
|
|
|
Anum_pg_subscription_subsynccommit,
|
|
|
|
&isnull);
|
|
|
|
Assert(!isnull);
|
|
|
|
sub->synccommit = TextDatumGetCString(datum);
|
|
|
|
|
2017-01-19 18:00:00 +01:00
|
|
|
/* Get publications */
|
|
|
|
datum = SysCacheGetAttr(SUBSCRIPTIONOID,
|
|
|
|
tup,
|
|
|
|
Anum_pg_subscription_subpublications,
|
|
|
|
&isnull);
|
|
|
|
Assert(!isnull);
|
|
|
|
sub->publications = textarray_to_stringlist(DatumGetArrayTypeP(datum));
|
|
|
|
|
|
|
|
ReleaseSysCache(tup);
|
|
|
|
|
|
|
|
return sub;
|
|
|
|
}
|
|
|
|
|
|
|
|
/*
|
|
|
|
* Return number of subscriptions defined in given database.
|
|
|
|
* Used by dropdb() to check if database can indeed be dropped.
|
|
|
|
*/
|
|
|
|
int
|
|
|
|
CountDBSubscriptions(Oid dbid)
|
|
|
|
{
|
2017-05-17 22:31:56 +02:00
|
|
|
int nsubs = 0;
|
|
|
|
Relation rel;
|
|
|
|
ScanKeyData scankey;
|
|
|
|
SysScanDesc scan;
|
|
|
|
HeapTuple tup;
|
2017-01-19 18:00:00 +01:00
|
|
|
|
|
|
|
rel = heap_open(SubscriptionRelationId, RowExclusiveLock);
|
|
|
|
|
|
|
|
ScanKeyInit(&scankey,
|
|
|
|
Anum_pg_subscription_subdbid,
|
|
|
|
BTEqualStrategyNumber, F_OIDEQ,
|
|
|
|
ObjectIdGetDatum(dbid));
|
|
|
|
|
|
|
|
scan = systable_beginscan(rel, InvalidOid, false,
|
|
|
|
NULL, 1, &scankey);
|
|
|
|
|
|
|
|
while (HeapTupleIsValid(tup = systable_getnext(scan)))
|
|
|
|
nsubs++;
|
|
|
|
|
|
|
|
systable_endscan(scan);
|
|
|
|
|
|
|
|
heap_close(rel, NoLock);
|
|
|
|
|
|
|
|
return nsubs;
|
|
|
|
}
|
|
|
|
|
|
|
|
/*
|
|
|
|
* Free memory allocated by subscription struct.
|
|
|
|
*/
|
|
|
|
void
|
|
|
|
FreeSubscription(Subscription *sub)
|
|
|
|
{
|
|
|
|
pfree(sub->name);
|
|
|
|
pfree(sub->conninfo);
|
2017-05-09 16:20:42 +02:00
|
|
|
if (sub->slotname)
|
|
|
|
pfree(sub->slotname);
|
2017-01-19 18:00:00 +01:00
|
|
|
list_free_deep(sub->publications);
|
|
|
|
pfree(sub);
|
|
|
|
}
|
|
|
|
|
|
|
|
/*
|
|
|
|
* get_subscription_oid - given a subscription name, look up the OID
|
|
|
|
*
|
|
|
|
* If missing_ok is false, throw an error if name not found. If true, just
|
|
|
|
* return InvalidOid.
|
|
|
|
*/
|
|
|
|
Oid
|
|
|
|
get_subscription_oid(const char *subname, bool missing_ok)
|
|
|
|
{
|
|
|
|
Oid oid;
|
|
|
|
|
|
|
|
oid = GetSysCacheOid2(SUBSCRIPTIONNAME, MyDatabaseId,
|
|
|
|
CStringGetDatum(subname));
|
|
|
|
if (!OidIsValid(oid) && !missing_ok)
|
|
|
|
ereport(ERROR,
|
|
|
|
(errcode(ERRCODE_UNDEFINED_OBJECT),
|
|
|
|
errmsg("subscription \"%s\" does not exist", subname)));
|
|
|
|
return oid;
|
|
|
|
}
|
|
|
|
|
|
|
|
/*
|
|
|
|
* get_subscription_name - given a subscription OID, look up the name
|
|
|
|
*/
|
|
|
|
char *
|
|
|
|
get_subscription_name(Oid subid)
|
|
|
|
{
|
2017-05-17 22:31:56 +02:00
|
|
|
HeapTuple tup;
|
|
|
|
char *subname;
|
2017-01-19 18:00:00 +01:00
|
|
|
Form_pg_subscription subform;
|
|
|
|
|
|
|
|
tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
|
|
|
|
|
|
|
|
if (!HeapTupleIsValid(tup))
|
|
|
|
elog(ERROR, "cache lookup failed for subscription %u", subid);
|
|
|
|
|
|
|
|
subform = (Form_pg_subscription) GETSTRUCT(tup);
|
|
|
|
subname = pstrdup(NameStr(subform->subname));
|
|
|
|
|
|
|
|
ReleaseSysCache(tup);
|
|
|
|
|
|
|
|
return subname;
|
|
|
|
}
|
|
|
|
|
|
|
|
/*
|
|
|
|
* Convert text array to list of strings.
|
|
|
|
*
|
|
|
|
* Note: the resulting list of strings is pallocated here.
|
|
|
|
*/
|
|
|
|
static List *
|
|
|
|
textarray_to_stringlist(ArrayType *textarray)
|
|
|
|
{
|
2017-05-17 22:31:56 +02:00
|
|
|
Datum *elems;
|
|
|
|
int nelems,
|
|
|
|
i;
|
|
|
|
List *res = NIL;
|
2017-01-19 18:00:00 +01:00
|
|
|
|
|
|
|
deconstruct_array(textarray,
|
|
|
|
TEXTOID, -1, false, 'i',
|
|
|
|
&elems, NULL, &nelems);
|
|
|
|
|
|
|
|
if (nelems == 0)
|
|
|
|
return NIL;
|
|
|
|
|
|
|
|
for (i = 0; i < nelems; i++)
|
2017-04-14 18:54:09 +02:00
|
|
|
res = lappend(res, makeString(TextDatumGetCString(elems[i])));
|
2017-01-19 18:00:00 +01:00
|
|
|
|
|
|
|
return res;
|
|
|
|
}
|
2017-03-23 13:36:36 +01:00
|
|
|
|
|
|
|
/*
|
|
|
|
* Set the state of a subscription table.
|
2017-04-10 21:08:14 +02:00
|
|
|
*
|
2017-06-07 19:49:14 +02:00
|
|
|
* If update_only is true and the record for given table doesn't exist, do
|
|
|
|
* nothing. This can be used to avoid inserting a new record that was deleted
|
|
|
|
* by someone else. Generally, subscription DDL commands should use false,
|
|
|
|
* workers should use true.
|
|
|
|
*
|
2017-04-10 21:08:14 +02:00
|
|
|
* The insert-or-update logic in this function is not concurrency safe so it
|
|
|
|
* might raise an error in rare circumstances. But if we took a stronger lock
|
|
|
|
* such as ShareRowExclusiveLock, we would risk more deadlocks.
|
2017-03-23 13:36:36 +01:00
|
|
|
*/
|
|
|
|
Oid
|
|
|
|
SetSubscriptionRelState(Oid subid, Oid relid, char state,
|
2017-06-07 19:49:14 +02:00
|
|
|
XLogRecPtr sublsn, bool update_only)
|
2017-03-23 13:36:36 +01:00
|
|
|
{
|
|
|
|
Relation rel;
|
|
|
|
HeapTuple tup;
|
2017-06-07 19:49:14 +02:00
|
|
|
Oid subrelid = InvalidOid;
|
2017-03-23 13:36:36 +01:00
|
|
|
bool nulls[Natts_pg_subscription_rel];
|
|
|
|
Datum values[Natts_pg_subscription_rel];
|
|
|
|
|
2017-04-10 21:08:14 +02:00
|
|
|
rel = heap_open(SubscriptionRelRelationId, RowExclusiveLock);
|
2017-03-23 13:36:36 +01:00
|
|
|
|
|
|
|
/* Try finding existing mapping. */
|
|
|
|
tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
|
|
|
|
ObjectIdGetDatum(relid),
|
|
|
|
ObjectIdGetDatum(subid));
|
|
|
|
|
|
|
|
/*
|
2017-05-17 22:31:56 +02:00
|
|
|
* If the record for given table does not exist yet create new record,
|
|
|
|
* otherwise update the existing one.
|
2017-03-23 13:36:36 +01:00
|
|
|
*/
|
2017-06-07 19:49:14 +02:00
|
|
|
if (!HeapTupleIsValid(tup) && !update_only)
|
2017-03-23 13:36:36 +01:00
|
|
|
{
|
|
|
|
/* Form the tuple. */
|
|
|
|
memset(values, 0, sizeof(values));
|
|
|
|
memset(nulls, false, sizeof(nulls));
|
|
|
|
values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
|
|
|
|
values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
|
|
|
|
values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
|
|
|
|
if (sublsn != InvalidXLogRecPtr)
|
|
|
|
values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
|
|
|
|
else
|
|
|
|
nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
|
|
|
|
|
|
|
|
tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
|
|
|
|
|
|
|
|
/* Insert tuple into catalog. */
|
|
|
|
subrelid = CatalogTupleInsert(rel, tup);
|
|
|
|
|
|
|
|
heap_freetuple(tup);
|
|
|
|
}
|
2017-06-07 19:49:14 +02:00
|
|
|
else if (HeapTupleIsValid(tup))
|
2017-03-23 13:36:36 +01:00
|
|
|
{
|
|
|
|
bool replaces[Natts_pg_subscription_rel];
|
|
|
|
|
|
|
|
/* Update the tuple. */
|
|
|
|
memset(values, 0, sizeof(values));
|
|
|
|
memset(nulls, false, sizeof(nulls));
|
|
|
|
memset(replaces, false, sizeof(replaces));
|
|
|
|
|
|
|
|
replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
|
|
|
|
values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
|
|
|
|
|
|
|
|
replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
|
|
|
|
if (sublsn != InvalidXLogRecPtr)
|
|
|
|
values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
|
|
|
|
else
|
|
|
|
nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
|
|
|
|
|
|
|
|
tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
|
|
|
|
replaces);
|
|
|
|
|
|
|
|
/* Update the catalog. */
|
|
|
|
CatalogTupleUpdate(rel, &tup->t_self, tup);
|
|
|
|
|
|
|
|
subrelid = HeapTupleGetOid(tup);
|
|
|
|
}
|
|
|
|
|
|
|
|
/* Cleanup. */
|
|
|
|
heap_close(rel, NoLock);
|
|
|
|
|
|
|
|
return subrelid;
|
|
|
|
}
|
|
|
|
|
|
|
|
/*
|
|
|
|
* Get state of subscription table.
|
|
|
|
*
|
|
|
|
* Returns SUBREL_STATE_UNKNOWN when not found and missing_ok is true.
|
|
|
|
*/
|
|
|
|
char
|
|
|
|
GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn,
|
|
|
|
bool missing_ok)
|
|
|
|
{
|
|
|
|
Relation rel;
|
|
|
|
HeapTuple tup;
|
|
|
|
char substate;
|
|
|
|
bool isnull;
|
|
|
|
Datum d;
|
|
|
|
|
|
|
|
rel = heap_open(SubscriptionRelRelationId, AccessShareLock);
|
|
|
|
|
|
|
|
/* Try finding the mapping. */
|
|
|
|
tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
|
|
|
|
ObjectIdGetDatum(relid),
|
|
|
|
ObjectIdGetDatum(subid));
|
|
|
|
|
|
|
|
if (!HeapTupleIsValid(tup))
|
|
|
|
{
|
|
|
|
if (missing_ok)
|
|
|
|
{
|
|
|
|
heap_close(rel, AccessShareLock);
|
|
|
|
*sublsn = InvalidXLogRecPtr;
|
|
|
|
return SUBREL_STATE_UNKNOWN;
|
|
|
|
}
|
|
|
|
|
|
|
|
elog(ERROR, "subscription table %u in subscription %u does not exist",
|
|
|
|
relid, subid);
|
|
|
|
}
|
|
|
|
|
|
|
|
/* Get the state. */
|
|
|
|
d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
|
|
|
|
Anum_pg_subscription_rel_srsubstate, &isnull);
|
|
|
|
Assert(!isnull);
|
|
|
|
substate = DatumGetChar(d);
|
|
|
|
d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
|
|
|
|
Anum_pg_subscription_rel_srsublsn, &isnull);
|
|
|
|
if (isnull)
|
|
|
|
*sublsn = InvalidXLogRecPtr;
|
|
|
|
else
|
|
|
|
*sublsn = DatumGetLSN(d);
|
|
|
|
|
|
|
|
/* Cleanup */
|
|
|
|
ReleaseSysCache(tup);
|
|
|
|
heap_close(rel, AccessShareLock);
|
|
|
|
|
|
|
|
return substate;
|
|
|
|
}
|
|
|
|
|
|
|
|
/*
|
|
|
|
* Drop subscription relation mapping. These can be for a particular
|
|
|
|
* subscription, or for a particular relation, or both.
|
|
|
|
*/
|
|
|
|
void
|
|
|
|
RemoveSubscriptionRel(Oid subid, Oid relid)
|
|
|
|
{
|
|
|
|
Relation rel;
|
|
|
|
HeapScanDesc scan;
|
|
|
|
ScanKeyData skey[2];
|
|
|
|
HeapTuple tup;
|
|
|
|
int nkeys = 0;
|
|
|
|
|
2017-04-10 21:08:14 +02:00
|
|
|
rel = heap_open(SubscriptionRelRelationId, RowExclusiveLock);
|
2017-03-23 13:36:36 +01:00
|
|
|
|
|
|
|
if (OidIsValid(subid))
|
|
|
|
{
|
|
|
|
ScanKeyInit(&skey[nkeys++],
|
|
|
|
Anum_pg_subscription_rel_srsubid,
|
|
|
|
BTEqualStrategyNumber,
|
|
|
|
F_OIDEQ,
|
|
|
|
ObjectIdGetDatum(subid));
|
|
|
|
}
|
|
|
|
|
|
|
|
if (OidIsValid(relid))
|
|
|
|
{
|
|
|
|
ScanKeyInit(&skey[nkeys++],
|
|
|
|
Anum_pg_subscription_rel_srrelid,
|
|
|
|
BTEqualStrategyNumber,
|
|
|
|
F_OIDEQ,
|
|
|
|
ObjectIdGetDatum(relid));
|
|
|
|
}
|
|
|
|
|
|
|
|
/* Do the search and delete what we found. */
|
|
|
|
scan = heap_beginscan_catalog(rel, nkeys, skey);
|
|
|
|
while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
|
|
|
|
{
|
2017-06-14 16:26:46 +02:00
|
|
|
CatalogTupleDelete(rel, &tup->t_self);
|
2017-03-23 13:36:36 +01:00
|
|
|
}
|
|
|
|
heap_endscan(scan);
|
|
|
|
|
2017-04-10 21:08:14 +02:00
|
|
|
heap_close(rel, RowExclusiveLock);
|
2017-03-23 13:36:36 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
/*
|
|
|
|
* Get all relations for subscription.
|
|
|
|
*
|
2017-04-26 18:04:44 +02:00
|
|
|
* Returned list is palloc'ed in current memory context.
|
2017-03-23 13:36:36 +01:00
|
|
|
*/
|
|
|
|
List *
|
|
|
|
GetSubscriptionRelations(Oid subid)
|
|
|
|
{
|
|
|
|
List *res = NIL;
|
|
|
|
Relation rel;
|
|
|
|
HeapTuple tup;
|
|
|
|
int nkeys = 0;
|
2017-05-17 22:31:56 +02:00
|
|
|
ScanKeyData skey[2];
|
|
|
|
SysScanDesc scan;
|
2017-03-23 13:36:36 +01:00
|
|
|
|
|
|
|
rel = heap_open(SubscriptionRelRelationId, AccessShareLock);
|
|
|
|
|
|
|
|
ScanKeyInit(&skey[nkeys++],
|
|
|
|
Anum_pg_subscription_rel_srsubid,
|
|
|
|
BTEqualStrategyNumber, F_OIDEQ,
|
|
|
|
ObjectIdGetDatum(subid));
|
|
|
|
|
|
|
|
scan = systable_beginscan(rel, InvalidOid, false,
|
|
|
|
NULL, nkeys, skey);
|
|
|
|
|
|
|
|
while (HeapTupleIsValid(tup = systable_getnext(scan)))
|
|
|
|
{
|
2017-05-17 22:31:56 +02:00
|
|
|
Form_pg_subscription_rel subrel;
|
|
|
|
SubscriptionRelState *relstate;
|
2017-03-23 13:36:36 +01:00
|
|
|
|
|
|
|
subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
|
|
|
|
|
2017-05-17 22:31:56 +02:00
|
|
|
relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
|
2017-03-23 13:36:36 +01:00
|
|
|
relstate->relid = subrel->srrelid;
|
|
|
|
relstate->state = subrel->srsubstate;
|
|
|
|
relstate->lsn = subrel->srsublsn;
|
|
|
|
|
|
|
|
res = lappend(res, relstate);
|
|
|
|
}
|
|
|
|
|
|
|
|
/* Cleanup */
|
|
|
|
systable_endscan(scan);
|
|
|
|
heap_close(rel, AccessShareLock);
|
|
|
|
|
|
|
|
return res;
|
|
|
|
}
|
|
|
|
|
|
|
|
/*
|
|
|
|
* Get all relations for subscription that are not in a ready state.
|
|
|
|
*
|
2017-04-26 18:04:44 +02:00
|
|
|
* Returned list is palloc'ed in current memory context.
|
2017-03-23 13:36:36 +01:00
|
|
|
*/
|
|
|
|
List *
|
|
|
|
GetSubscriptionNotReadyRelations(Oid subid)
|
|
|
|
{
|
|
|
|
List *res = NIL;
|
|
|
|
Relation rel;
|
|
|
|
HeapTuple tup;
|
|
|
|
int nkeys = 0;
|
2017-05-17 22:31:56 +02:00
|
|
|
ScanKeyData skey[2];
|
|
|
|
SysScanDesc scan;
|
2017-03-23 13:36:36 +01:00
|
|
|
|
|
|
|
rel = heap_open(SubscriptionRelRelationId, AccessShareLock);
|
|
|
|
|
|
|
|
ScanKeyInit(&skey[nkeys++],
|
|
|
|
Anum_pg_subscription_rel_srsubid,
|
|
|
|
BTEqualStrategyNumber, F_OIDEQ,
|
|
|
|
ObjectIdGetDatum(subid));
|
|
|
|
|
|
|
|
ScanKeyInit(&skey[nkeys++],
|
|
|
|
Anum_pg_subscription_rel_srsubstate,
|
|
|
|
BTEqualStrategyNumber, F_CHARNE,
|
|
|
|
CharGetDatum(SUBREL_STATE_READY));
|
|
|
|
|
|
|
|
scan = systable_beginscan(rel, InvalidOid, false,
|
|
|
|
NULL, nkeys, skey);
|
|
|
|
|
|
|
|
while (HeapTupleIsValid(tup = systable_getnext(scan)))
|
|
|
|
{
|
2017-05-17 22:31:56 +02:00
|
|
|
Form_pg_subscription_rel subrel;
|
|
|
|
SubscriptionRelState *relstate;
|
2017-03-23 13:36:36 +01:00
|
|
|
|
|
|
|
subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
|
|
|
|
|
2017-05-17 22:31:56 +02:00
|
|
|
relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
|
2017-03-23 13:36:36 +01:00
|
|
|
relstate->relid = subrel->srrelid;
|
|
|
|
relstate->state = subrel->srsubstate;
|
|
|
|
relstate->lsn = subrel->srsublsn;
|
|
|
|
|
|
|
|
res = lappend(res, relstate);
|
|
|
|
}
|
|
|
|
|
|
|
|
/* Cleanup */
|
|
|
|
systable_endscan(scan);
|
|
|
|
heap_close(rel, AccessShareLock);
|
|
|
|
|
|
|
|
return res;
|
|
|
|
}
|