Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * pg_publication.c
4 : : * publication C API manipulation
5 : : *
6 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : * IDENTIFICATION
10 : : * src/backend/catalog/pg_publication.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/catalog.h"
22 : : #include "catalog/dependency.h"
23 : : #include "catalog/indexing.h"
24 : : #include "catalog/namespace.h"
25 : : #include "catalog/objectaddress.h"
26 : : #include "catalog/partition.h"
27 : : #include "catalog/pg_inherits.h"
28 : : #include "catalog/pg_namespace.h"
29 : : #include "catalog/pg_publication.h"
30 : : #include "catalog/pg_publication_namespace.h"
31 : : #include "catalog/pg_publication_rel.h"
32 : : #include "catalog/pg_type.h"
33 : : #include "commands/publicationcmds.h"
34 : : #include "funcapi.h"
35 : : #include "miscadmin.h"
36 : : #include "utils/array.h"
37 : : #include "utils/builtins.h"
38 : : #include "utils/catcache.h"
39 : : #include "utils/fmgroids.h"
40 : : #include "utils/lsyscache.h"
41 : : #include "utils/rel.h"
42 : : #include "utils/syscache.h"
43 : :
44 : : /* Records association between publication and published table */
45 : : typedef struct
46 : : {
47 : : Oid relid; /* OID of published table */
48 : : Oid pubid; /* OID of publication that publishes this
49 : : * table. */
50 : : } published_rel;
51 : :
52 : : /*
53 : : * Check if relation can be in given publication and throws appropriate
54 : : * error if not.
55 : : */
56 : : static void
57 : 812 : check_publication_add_relation(PublicationRelInfo *pri)
58 : : {
59 : 812 : Relation targetrel = pri->relation;
60 : : const char *relname;
61 : : const char *errormsg;
62 : :
63 [ + + ]: 812 : if (pri->except)
64 : : {
65 : 81 : relname = RelationGetQualifiedRelationName(targetrel);
66 : 81 : errormsg = gettext_noop("cannot specify relation \"%s\" in the publication EXCEPT clause");
67 : : }
68 : : else
69 : : {
70 : 731 : relname = RelationGetRelationName(targetrel);
71 : 731 : errormsg = gettext_noop("cannot add relation \"%s\" to publication");
72 : : }
73 : :
74 : : /* If in EXCEPT clause, must be root partitioned table */
75 [ + + + + ]: 812 : if (pri->except && targetrel->rd_rel->relispartition)
76 [ + - ]: 4 : ereport(ERROR,
77 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
78 : : errmsg(errormsg, relname),
79 : : errdetail("This operation is not supported for individual partitions.")));
80 : :
81 : : /* Must be a regular or partitioned table */
82 [ + + ]: 808 : if (RelationGetForm(targetrel)->relkind != RELKIND_RELATION &&
83 [ + + ]: 115 : RelationGetForm(targetrel)->relkind != RELKIND_PARTITIONED_TABLE)
84 [ + - ]: 9 : ereport(ERROR,
85 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
86 : : errmsg(errormsg, relname),
87 : : errdetail_relkind_not_supported(RelationGetForm(targetrel)->relkind)));
88 : :
89 : : /* Can't be system table */
90 [ + + ]: 799 : if (IsCatalogRelation(targetrel))
91 [ + - ]: 4 : ereport(ERROR,
92 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
93 : : errmsg(errormsg, relname),
94 : : errdetail("This operation is not supported for system tables.")));
95 : :
96 : : /* Can't be conflict log table */
97 [ + + ]: 795 : if (IsConflictLogTableNamespace(RelationGetNamespace(targetrel)))
98 [ + - ]: 4 : ereport(ERROR,
99 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
100 : : errmsg(errormsg, relname),
101 : : errdetail("This operation is not supported for conflict log tables.")));
102 : :
103 : : /* UNLOGGED and TEMP relations cannot be part of publication. */
104 [ + + ]: 791 : if (targetrel->rd_rel->relpersistence == RELPERSISTENCE_TEMP)
105 [ + - ]: 4 : ereport(ERROR,
106 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
107 : : errmsg(errormsg, relname),
108 : : errdetail("This operation is not supported for temporary tables.")));
109 [ + + ]: 787 : else if (targetrel->rd_rel->relpersistence == RELPERSISTENCE_UNLOGGED)
110 [ + - ]: 4 : ereport(ERROR,
111 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
112 : : errmsg(errormsg, relname),
113 : : errdetail("This operation is not supported for unlogged tables.")));
114 : 783 : }
115 : :
116 : : /*
117 : : * Check if schema can be in given publication and throw appropriate error if
118 : : * not.
119 : : */
120 : : static void
121 : 164 : check_publication_add_schema(Oid schemaid)
122 : : {
123 : : /* Can't be system namespace */
124 [ + + + - : 324 : if (IsCatalogNamespace(schemaid) || IsToastNamespace(schemaid) ||
- + ]
125 : 160 : IsConflictLogTableNamespace(schemaid))
126 [ + - ]: 4 : ereport(ERROR,
127 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
128 : : errmsg("cannot add schema \"%s\" to publication",
129 : : get_namespace_name(schemaid)),
130 : : errdetail("This operation is not supported for system schemas.")));
131 : :
132 : : /* Can't be temporary namespace */
133 [ - + ]: 160 : if (isAnyTempNamespace(schemaid))
134 [ # # ]: 0 : ereport(ERROR,
135 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
136 : : errmsg("cannot add schema \"%s\" to publication",
137 : : get_namespace_name(schemaid)),
138 : : errdetail("Temporary schemas cannot be replicated.")));
139 : 160 : }
140 : :
141 : : /*
142 : : * Returns if relation represented by oid and Form_pg_class entry
143 : : * is publishable.
144 : : *
145 : : * Does same checks as check_publication_add_relation() above except for
146 : : * RELKIND_SEQUENCE, but does not need relation to be opened and also does
147 : : * not throw errors. Here, the additional check is to support ALL SEQUENCES
148 : : * publication.
149 : : *
150 : : * XXX This also excludes all tables with relid < FirstNormalObjectId,
151 : : * ie all tables created during initdb. This mainly affects the preinstalled
152 : : * information_schema. IsCatalogRelationOid() only excludes tables with
153 : : * relid < FirstUnpinnedObjectId, making that test rather redundant,
154 : : * but really we should get rid of the FirstNormalObjectId test not
155 : : * IsCatalogRelationOid. We can't do so today because we don't want
156 : : * information_schema tables to be considered publishable; but this test
157 : : * is really inadequate for that, since the information_schema could be
158 : : * dropped and reloaded and then it'll be considered publishable. The best
159 : : * long-term solution may be to add a "relispublishable" bool to pg_class,
160 : : * and depend on that instead of OID checks. IsConflictLogTableClass()
161 : : * excludes tables in conflict schema.
162 : : */
163 : : static bool
164 : 323884 : is_publishable_class(Oid relid, Form_pg_class reltuple)
165 : : {
166 : 332203 : return (reltuple->relkind == RELKIND_RELATION ||
167 [ + + ]: 8319 : reltuple->relkind == RELKIND_PARTITIONED_TABLE ||
168 [ + + ]: 7296 : reltuple->relkind == RELKIND_SEQUENCE) &&
169 [ + + ]: 317959 : !IsCatalogRelationOid(relid) &&
170 [ + + ]: 312608 : !IsConflictLogTableClass(reltuple) &&
171 [ + + + + : 647768 : reltuple->relpersistence == RELPERSISTENCE_PERMANENT &&
+ + ]
172 : : relid >= FirstNormalObjectId;
173 : : }
174 : :
175 : : /*
176 : : * Another variant of is_publishable_class(), taking a Relation.
177 : : */
178 : : bool
179 : 298270 : is_publishable_relation(Relation rel)
180 : : {
181 : 298270 : return is_publishable_class(RelationGetRelid(rel), rel->rd_rel);
182 : : }
183 : :
184 : : /*
185 : : * Similar to is_publishable_class() but checks whether the given OID
186 : : * is a publishable "table" or not.
187 : : */
188 : : static bool
189 : 551 : is_publishable_table(Oid tableoid)
190 : : {
191 : : HeapTuple tuple;
192 : : Form_pg_class relform;
193 : :
194 : 551 : tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(tableoid));
195 [ + + ]: 551 : if (!HeapTupleIsValid(tuple))
196 : 4 : return false;
197 : :
198 : 547 : relform = (Form_pg_class) GETSTRUCT(tuple);
199 : :
200 : : /*
201 : : * is_publishable_class() includes sequences, so we need to explicitly
202 : : * check the relkind to filter them out here.
203 : : */
204 [ + - + + ]: 1094 : if (relform->relkind != RELKIND_SEQUENCE &&
205 : 547 : is_publishable_class(tableoid, relform))
206 : : {
207 : 543 : ReleaseSysCache(tuple);
208 : 543 : return true;
209 : : }
210 : :
211 : 4 : ReleaseSysCache(tuple);
212 : 4 : return false;
213 : : }
214 : :
215 : : /*
216 : : * SQL-callable variant of the above
217 : : *
218 : : * This returns null when the relation does not exist. This is intended to be
219 : : * used for example in psql to avoid gratuitous errors when there are
220 : : * concurrent catalog changes.
221 : : */
222 : : Datum
223 : 4356 : pg_relation_is_publishable(PG_FUNCTION_ARGS)
224 : : {
225 : 4356 : Oid relid = PG_GETARG_OID(0);
226 : : HeapTuple tuple;
227 : : bool result;
228 : :
229 : 4356 : tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
230 [ - + ]: 4356 : if (!HeapTupleIsValid(tuple))
231 : 0 : PG_RETURN_NULL();
232 : 4356 : result = is_publishable_class(relid, (Form_pg_class) GETSTRUCT(tuple));
233 : 4356 : ReleaseSysCache(tuple);
234 : 4356 : PG_RETURN_BOOL(result);
235 : : }
236 : :
237 : : /*
238 : : * Returns true if the ancestor is in the list of published relations.
239 : : * Otherwise, returns false.
240 : : */
241 : : static bool
242 : 61 : is_ancestor_member_tableinfos(Oid ancestor, List *table_infos)
243 : : {
244 : : ListCell *lc;
245 : :
246 [ + - + + : 272 : foreach(lc, table_infos)
+ + ]
247 : : {
248 : 241 : Oid relid = ((published_rel *) lfirst(lc))->relid;
249 : :
250 [ + + ]: 241 : if (relid == ancestor)
251 : 30 : return true;
252 : : }
253 : :
254 : 31 : return false;
255 : : }
256 : :
257 : : /*
258 : : * Filter out the partitions whose parent tables are also present in the list.
259 : : */
260 : : static void
261 : 168 : filter_partitions(List *table_infos)
262 : : {
263 : : ListCell *lc;
264 : :
265 [ + + + + : 397 : foreach(lc, table_infos)
+ + ]
266 : : {
267 : 229 : bool skip = false;
268 : 229 : List *ancestors = NIL;
269 : : ListCell *lc2;
270 : 229 : published_rel *table_info = (published_rel *) lfirst(lc);
271 : :
272 [ + + ]: 229 : if (get_rel_relispartition(table_info->relid))
273 : 61 : ancestors = get_partition_ancestors(table_info->relid);
274 : :
275 [ + + + + : 260 : foreach(lc2, ancestors)
+ + ]
276 : : {
277 : 61 : Oid ancestor = lfirst_oid(lc2);
278 : :
279 [ + + ]: 61 : if (is_ancestor_member_tableinfos(ancestor, table_infos))
280 : : {
281 : 30 : skip = true;
282 : 30 : break;
283 : : }
284 : : }
285 : :
286 [ + + ]: 229 : if (skip)
287 : 30 : table_infos = foreach_delete_current(table_infos, lc);
288 : : }
289 : 168 : }
290 : :
291 : : /*
292 : : * Returns true if any schema is associated with the publication, false if no
293 : : * schema is associated with the publication.
294 : : */
295 : : bool
296 : 215 : is_schema_publication(Oid pubid)
297 : : {
298 : : Relation pubschsrel;
299 : : ScanKeyData scankey;
300 : : SysScanDesc scan;
301 : : HeapTuple tup;
302 : 215 : bool result = false;
303 : :
304 : 215 : pubschsrel = table_open(PublicationNamespaceRelationId, AccessShareLock);
305 : 215 : ScanKeyInit(&scankey,
306 : : Anum_pg_publication_namespace_pnpubid,
307 : : BTEqualStrategyNumber, F_OIDEQ,
308 : : ObjectIdGetDatum(pubid));
309 : :
310 : 215 : scan = systable_beginscan(pubschsrel,
311 : : PublicationNamespacePnnspidPnpubidIndexId,
312 : : true, NULL, 1, &scankey);
313 : 215 : tup = systable_getnext(scan);
314 : 215 : result = HeapTupleIsValid(tup);
315 : :
316 : 215 : systable_endscan(scan);
317 : 215 : table_close(pubschsrel, AccessShareLock);
318 : :
319 : 215 : return result;
320 : : }
321 : :
322 : : /*
323 : : * Returns true if the publication has explicitly included relation (i.e.,
324 : : * not marked as EXCEPT).
325 : : */
326 : : bool
327 : 49 : is_table_publication(Oid pubid)
328 : : {
329 : : Relation pubrelsrel;
330 : : ScanKeyData scankey;
331 : : SysScanDesc scan;
332 : : HeapTuple tup;
333 : 49 : bool result = false;
334 : :
335 : 49 : pubrelsrel = table_open(PublicationRelRelationId, AccessShareLock);
336 : 49 : ScanKeyInit(&scankey,
337 : : Anum_pg_publication_rel_prpubid,
338 : : BTEqualStrategyNumber, F_OIDEQ,
339 : : ObjectIdGetDatum(pubid));
340 : :
341 : 49 : scan = systable_beginscan(pubrelsrel,
342 : : PublicationRelPrpubidIndexId,
343 : : true, NULL, 1, &scankey);
344 : 49 : tup = systable_getnext(scan);
345 [ + + ]: 49 : if (HeapTupleIsValid(tup))
346 : : {
347 : : Form_pg_publication_rel pubrel;
348 : :
349 : 25 : pubrel = (Form_pg_publication_rel) GETSTRUCT(tup);
350 : :
351 : : /*
352 : : * For any publication, pg_publication_rel contains either only EXCEPT
353 : : * entries or only explicitly included tables. Therefore, examining
354 : : * the first tuple is sufficient to determine table inclusion.
355 : : */
356 : 25 : result = !pubrel->prexcept;
357 : : }
358 : :
359 : 49 : systable_endscan(scan);
360 : 49 : table_close(pubrelsrel, AccessShareLock);
361 : :
362 : 49 : return result;
363 : : }
364 : :
365 : : /*
366 : : * Returns true if the relation has column list associated with the
367 : : * publication, false otherwise.
368 : : *
369 : : * If a column list is found, the corresponding bitmap is returned through the
370 : : * cols parameter, if provided. The bitmap is constructed within the given
371 : : * memory context (mcxt).
372 : : */
373 : : bool
374 : 947 : check_and_fetch_column_list(Publication *pub, Oid relid, MemoryContext mcxt,
375 : : Bitmapset **cols)
376 : : {
377 : : HeapTuple cftuple;
378 : 947 : bool found = false;
379 : :
380 [ + + ]: 947 : if (pub->alltables)
381 : 220 : return false;
382 : :
383 : 727 : cftuple = SearchSysCache2(PUBLICATIONRELMAP,
384 : : ObjectIdGetDatum(relid),
385 : : ObjectIdGetDatum(pub->oid));
386 [ + + ]: 727 : if (HeapTupleIsValid(cftuple))
387 : : {
388 : : Datum cfdatum;
389 : : bool isnull;
390 : :
391 : : /* Lookup the column list attribute. */
392 : 669 : cfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, cftuple,
393 : : Anum_pg_publication_rel_prattrs, &isnull);
394 : :
395 : : /* Was a column list found? */
396 [ + + ]: 669 : if (!isnull)
397 : : {
398 : : /* Build the column list bitmap in the given memory context. */
399 [ + + ]: 191 : if (cols)
400 : 188 : *cols = pub_collist_to_bitmapset(*cols, cfdatum, mcxt);
401 : :
402 : 191 : found = true;
403 : : }
404 : :
405 : 669 : ReleaseSysCache(cftuple);
406 : : }
407 : :
408 : 727 : return found;
409 : : }
410 : :
411 : : /*
412 : : * Gets the relations based on the publication partition option for a specified
413 : : * relation.
414 : : */
415 : : List *
416 : 2498 : GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
417 : : Oid relid)
418 : : {
419 [ + + + + ]: 2498 : if (get_rel_relkind(relid) == RELKIND_PARTITIONED_TABLE &&
420 : : pub_partopt != PUBLICATION_PART_ROOT)
421 : 701 : {
422 : 701 : List *all_parts = find_all_inheritors(relid, NoLock,
423 : : NULL);
424 : :
425 [ + + ]: 701 : if (pub_partopt == PUBLICATION_PART_ALL)
426 : 682 : result = list_concat(result, all_parts);
427 [ + - ]: 19 : else if (pub_partopt == PUBLICATION_PART_LEAF)
428 : : {
429 : : ListCell *lc;
430 : :
431 [ + - + + : 69 : foreach(lc, all_parts)
+ + ]
432 : : {
433 : 50 : Oid partOid = lfirst_oid(lc);
434 : :
435 [ + + ]: 50 : if (get_rel_relkind(partOid) != RELKIND_PARTITIONED_TABLE)
436 : 30 : result = lappend_oid(result, partOid);
437 : : }
438 : : }
439 : : else
440 : : Assert(false);
441 : : }
442 : : else
443 : 1797 : result = lappend_oid(result, relid);
444 : :
445 : 2498 : return result;
446 : : }
447 : :
448 : : /*
449 : : * Returns the relid of the topmost ancestor that is published via this
450 : : * publication if any and set its ancestor level to ancestor_level,
451 : : * otherwise returns InvalidOid.
452 : : *
453 : : * The ancestor_level value allows us to compare the results for multiple
454 : : * publications, and decide which value is higher up.
455 : : *
456 : : * Note that the list of ancestors should be ordered such that the topmost
457 : : * ancestor is at the end of the list.
458 : : */
459 : : Oid
460 : 428 : GetTopMostAncestorInPublication(Oid puboid, List *ancestors, int *ancestor_level)
461 : : {
462 : : ListCell *lc;
463 : 428 : Oid topmost_relid = InvalidOid;
464 : 428 : int level = 0;
465 : :
466 : : /*
467 : : * Find the "topmost" ancestor that is in this publication.
468 : : */
469 [ + - + + : 868 : foreach(lc, ancestors)
+ + ]
470 : : {
471 : 440 : Oid ancestor = lfirst_oid(lc);
472 : 440 : List *apubids = GetRelationIncludedPublications(ancestor);
473 : 440 : List *aschemaPubids = NIL;
474 : :
475 : 440 : level++;
476 : :
477 [ + + ]: 440 : if (list_member_oid(apubids, puboid))
478 : : {
479 : 223 : topmost_relid = ancestor;
480 : :
481 [ + + ]: 223 : if (ancestor_level)
482 : 43 : *ancestor_level = level;
483 : : }
484 : : else
485 : : {
486 : 217 : aschemaPubids = GetSchemaPublications(get_rel_namespace(ancestor));
487 [ + + ]: 217 : if (list_member_oid(aschemaPubids, puboid))
488 : : {
489 : 13 : topmost_relid = ancestor;
490 : :
491 [ + + ]: 13 : if (ancestor_level)
492 : 5 : *ancestor_level = level;
493 : : }
494 : : }
495 : :
496 : 440 : list_free(apubids);
497 : 440 : list_free(aschemaPubids);
498 : : }
499 : :
500 : 428 : return topmost_relid;
501 : : }
502 : :
503 : : /*
504 : : * attnumstoint2vector
505 : : * Convert a Bitmapset of AttrNumbers into an int2vector.
506 : : *
507 : : * AttrNumber numbers are 0-based, i.e., not offset by
508 : : * FirstLowInvalidHeapAttributeNumber.
509 : : */
510 : : static int2vector *
511 : 212 : attnumstoint2vector(Bitmapset *attrs)
512 : : {
513 : : int2vector *result;
514 : 212 : int n = bms_num_members(attrs);
515 : 212 : int i = -1;
516 : 212 : int j = 0;
517 : :
518 : 212 : result = buildint2vector(NULL, n);
519 : :
520 [ + + ]: 577 : while ((i = bms_next_member(attrs, i)) >= 0)
521 : : {
522 : : Assert(i <= PG_INT16_MAX);
523 : :
524 : 365 : result->values[j++] = (int16) i;
525 : : }
526 : :
527 : 212 : return result;
528 : : }
529 : :
530 : : /*
531 : : * Insert new publication / relation mapping.
532 : : */
533 : : ObjectAddress
534 : 834 : publication_add_relation(Oid pubid, PublicationRelInfo *pri,
535 : : bool if_not_exists, AlterPublicationStmt *alter_stmt)
536 : : {
537 : : Relation rel;
538 : : HeapTuple tup;
539 : : Datum values[Natts_pg_publication_rel];
540 : : bool nulls[Natts_pg_publication_rel];
541 : 834 : Relation targetrel = pri->relation;
542 : 834 : Oid relid = RelationGetRelid(targetrel);
543 : : Oid pubreloid;
544 : : Bitmapset *attnums;
545 : 834 : Publication *pub = GetPublication(pubid);
546 : : ObjectAddress myself,
547 : : referenced;
548 : 834 : List *relids = NIL;
549 : : int i;
550 : : bool inval_except_table;
551 : :
552 : 834 : rel = table_open(PublicationRelRelationId, RowExclusiveLock);
553 : :
554 : : /*
555 : : * Check for duplicates. Note that this does not really prevent
556 : : * duplicates, it's here just to provide nicer error message in common
557 : : * case. The real protection is the unique key on the catalog.
558 : : */
559 [ + + ]: 834 : if (SearchSysCacheExists2(PUBLICATIONRELMAP, ObjectIdGetDatum(relid),
560 : : ObjectIdGetDatum(pubid)))
561 : : {
562 : 22 : table_close(rel, RowExclusiveLock);
563 : :
564 [ + + ]: 22 : if (if_not_exists)
565 : 18 : return InvalidObjectAddress;
566 : :
567 [ + - ]: 4 : ereport(ERROR,
568 : : (errcode(ERRCODE_DUPLICATE_OBJECT),
569 : : errmsg("relation \"%s\" is already a member of publication \"%s\"",
570 : : RelationGetRelationName(targetrel), pub->name)));
571 : : }
572 : :
573 : 812 : check_publication_add_relation(pri);
574 : :
575 : : /* Validate and translate column names into a Bitmapset of attnums. */
576 : 783 : attnums = pub_collist_validate(pri->relation, pri->columns);
577 : :
578 : : /* Form a tuple. */
579 : 767 : memset(values, 0, sizeof(values));
580 : 767 : memset(nulls, false, sizeof(nulls));
581 : :
582 : 767 : pubreloid = GetNewOidWithIndex(rel, PublicationRelObjectIndexId,
583 : : Anum_pg_publication_rel_oid);
584 : 767 : values[Anum_pg_publication_rel_oid - 1] = ObjectIdGetDatum(pubreloid);
585 : 767 : values[Anum_pg_publication_rel_prpubid - 1] =
586 : 767 : ObjectIdGetDatum(pubid);
587 : 767 : values[Anum_pg_publication_rel_prrelid - 1] =
588 : 767 : ObjectIdGetDatum(relid);
589 : 767 : values[Anum_pg_publication_rel_prexcept - 1] =
590 : 767 : BoolGetDatum(pri->except);
591 : :
592 : : /* Add qualifications, if available */
593 [ + + ]: 767 : if (pri->whereClause != NULL)
594 : 220 : values[Anum_pg_publication_rel_prqual - 1] = CStringGetTextDatum(nodeToString(pri->whereClause));
595 : : else
596 : 547 : nulls[Anum_pg_publication_rel_prqual - 1] = true;
597 : :
598 : : /* Add column list, if available */
599 [ + + ]: 767 : if (pri->columns)
600 : 212 : values[Anum_pg_publication_rel_prattrs - 1] = PointerGetDatum(attnumstoint2vector(attnums));
601 : : else
602 : 555 : nulls[Anum_pg_publication_rel_prattrs - 1] = true;
603 : :
604 : 767 : tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
605 : :
606 : : /* Insert tuple into catalog. */
607 : 767 : CatalogTupleInsert(rel, tup);
608 : 767 : heap_freetuple(tup);
609 : :
610 : : /* Register dependencies as needed */
611 : 767 : ObjectAddressSet(myself, PublicationRelRelationId, pubreloid);
612 : :
613 : : /* Add dependency on the publication */
614 : 767 : ObjectAddressSet(referenced, PublicationRelationId, pubid);
615 : 767 : recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
616 : :
617 : : /* Add dependency on the relation */
618 : 767 : ObjectAddressSet(referenced, RelationRelationId, relid);
619 : 767 : recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
620 : :
621 : : /* Add dependency on the objects mentioned in the qualifications */
622 [ + + ]: 767 : if (pri->whereClause)
623 : : {
624 : 220 : CheckUsageOnTypesInSingleRelExpr(pri->whereClause, relid, GetUserId());
625 : 220 : recordDependencyOnSingleRelExpr(&myself, pri->whereClause, relid,
626 : : DEPENDENCY_NORMAL, DEPENDENCY_NORMAL,
627 : : false);
628 : : }
629 : :
630 : : /* Add dependency on the columns, if any are listed */
631 : 767 : i = -1;
632 [ + + ]: 1132 : while ((i = bms_next_member(attnums, i)) >= 0)
633 : : {
634 : 365 : ObjectAddressSubSet(referenced, RelationRelationId, relid, i);
635 : 365 : recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
636 : : }
637 : :
638 : : /* Close the table. */
639 : 767 : table_close(rel, RowExclusiveLock);
640 : :
641 : : /*
642 : : * Determine whether EXCEPT tables require explicit relcache invalidation.
643 : : *
644 : : * For CREATE PUBLICATION with EXCEPT tables, invalidation is skipped
645 : : * here, as CreatePublication() function invalidates all relations as part
646 : : * of defining a FOR ALL TABLES publication.
647 : : *
648 : : * For ALTER PUBLICATION, invalidation is needed only when adding an
649 : : * EXCEPT table to a publication already marked as ALL TABLES. For
650 : : * publications that were originally empty or defined as ALL SEQUENCES and
651 : : * are being converted to ALL TABLES, invalidation is skipped here, as
652 : : * AlterPublicationAllFlags() function invalidates all relations while
653 : : * marking the publication as ALL TABLES publication.
654 : : */
655 [ + + + + ]: 776 : inval_except_table = (alter_stmt != NULL) && pub->alltables &&
656 [ + - + - ]: 9 : (alter_stmt->for_all_tables && pri->except);
657 : :
658 [ + + + + ]: 767 : if (!pri->except || inval_except_table)
659 : : {
660 : : /*
661 : : * Invalidate relcache so that publication info is rebuilt.
662 : : *
663 : : * For the partitioned tables, we must invalidate all partitions
664 : : * contained in the respective partition hierarchies, not just the one
665 : : * explicitly mentioned in the publication. This is required because
666 : : * we implicitly publish the child tables when the parent table is
667 : : * published.
668 : : */
669 : 699 : relids = GetPubPartitionOptionRelations(relids, PUBLICATION_PART_ALL,
670 : : relid);
671 : :
672 : 699 : InvalidatePublicationRels(relids);
673 : : }
674 : :
675 : 767 : return myself;
676 : : }
677 : :
678 : : /*
679 : : * pub_collist_validate
680 : : * Process and validate the 'columns' list and ensure the columns are all
681 : : * valid to use for a publication. Checks for and raises an ERROR for
682 : : * any unknown columns, system columns, duplicate columns, or virtual
683 : : * generated columns.
684 : : *
685 : : * Looks up each column's attnum and returns a 0-based Bitmapset of the
686 : : * corresponding attnums.
687 : : */
688 : : Bitmapset *
689 : 1073 : pub_collist_validate(Relation targetrel, List *columns)
690 : : {
691 : 1073 : Bitmapset *set = NULL;
692 : : ListCell *lc;
693 : 1073 : TupleDesc tupdesc = RelationGetDescr(targetrel);
694 : :
695 [ + + + + : 1618 : foreach(lc, columns)
+ + ]
696 : : {
697 : 569 : char *colname = strVal(lfirst(lc));
698 : 569 : AttrNumber attnum = get_attnum(RelationGetRelid(targetrel), colname);
699 : :
700 [ + + ]: 569 : if (attnum == InvalidAttrNumber)
701 [ + - ]: 4 : ereport(ERROR,
702 : : errcode(ERRCODE_UNDEFINED_COLUMN),
703 : : errmsg("column \"%s\" of relation \"%s\" does not exist",
704 : : colname, RelationGetRelationName(targetrel)));
705 : :
706 [ + + ]: 565 : if (!AttrNumberIsForUserDefinedAttr(attnum))
707 [ + - ]: 8 : ereport(ERROR,
708 : : errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
709 : : errmsg("cannot use system column \"%s\" in publication column list",
710 : : colname));
711 : :
712 [ + + ]: 557 : if (TupleDescAttr(tupdesc, attnum - 1)->attgenerated == ATTRIBUTE_GENERATED_VIRTUAL)
713 [ + - ]: 4 : ereport(ERROR,
714 : : errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
715 : : errmsg("cannot use virtual generated column \"%s\" in publication column list",
716 : : colname));
717 : :
718 [ + + ]: 553 : if (bms_is_member(attnum, set))
719 [ + - ]: 8 : ereport(ERROR,
720 : : errcode(ERRCODE_DUPLICATE_OBJECT),
721 : : errmsg("duplicate column \"%s\" in publication column list",
722 : : colname));
723 : :
724 : 545 : set = bms_add_member(set, attnum);
725 : : }
726 : :
727 : 1049 : return set;
728 : : }
729 : :
730 : : /*
731 : : * Transform a column list (represented by an array Datum) to a bitmapset.
732 : : *
733 : : * If columns isn't NULL, add the column numbers to that set.
734 : : *
735 : : * If mcxt isn't NULL, build the bitmapset in that context.
736 : : */
737 : : Bitmapset *
738 : 278 : pub_collist_to_bitmapset(Bitmapset *columns, Datum pubcols, MemoryContext mcxt)
739 : : {
740 : 278 : Bitmapset *result = columns;
741 : : ArrayType *arr;
742 : : int nelems;
743 : : int16 *elems;
744 : 278 : MemoryContext oldcxt = NULL;
745 : :
746 : 278 : arr = DatumGetArrayTypeP(pubcols);
747 : 278 : nelems = ARR_DIMS(arr)[0];
748 [ - + ]: 278 : elems = (int16 *) ARR_DATA_PTR(arr);
749 : :
750 : : /* If a memory context was specified, switch to it. */
751 [ + + ]: 278 : if (mcxt)
752 : 39 : oldcxt = MemoryContextSwitchTo(mcxt);
753 : :
754 [ + + ]: 764 : for (int i = 0; i < nelems; i++)
755 : 486 : result = bms_add_member(result, elems[i]);
756 : :
757 [ + + ]: 278 : if (mcxt)
758 : 39 : MemoryContextSwitchTo(oldcxt);
759 : :
760 : 278 : return result;
761 : : }
762 : :
763 : : /*
764 : : * Returns a bitmap representing the columns of the specified table.
765 : : *
766 : : * Generated columns are included if include_gencols_type is
767 : : * PUBLISH_GENCOLS_STORED.
768 : : */
769 : : Bitmapset *
770 : 9 : pub_form_cols_map(Relation relation, PublishGencolsType include_gencols_type)
771 : : {
772 : 9 : Bitmapset *result = NULL;
773 : 9 : TupleDesc desc = RelationGetDescr(relation);
774 : :
775 [ + + ]: 30 : for (int i = 0; i < desc->natts; i++)
776 : : {
777 : 21 : Form_pg_attribute att = TupleDescAttr(desc, i);
778 : :
779 [ + + ]: 21 : if (att->attisdropped)
780 : 1 : continue;
781 : :
782 [ + + ]: 20 : if (att->attgenerated)
783 : : {
784 : : /* We only support replication of STORED generated cols. */
785 [ + + ]: 2 : if (att->attgenerated != ATTRIBUTE_GENERATED_STORED)
786 : 1 : continue;
787 : :
788 : : /* User hasn't requested to replicate STORED generated cols. */
789 [ + - ]: 1 : if (include_gencols_type != PUBLISH_GENCOLS_STORED)
790 : 1 : continue;
791 : : }
792 : :
793 : 18 : result = bms_add_member(result, att->attnum);
794 : : }
795 : :
796 : 9 : return result;
797 : : }
798 : :
799 : : /*
800 : : * Insert new publication / schema mapping.
801 : : */
802 : : ObjectAddress
803 : 176 : publication_add_schema(Oid pubid, Oid schemaid, bool if_not_exists)
804 : : {
805 : : Relation rel;
806 : : HeapTuple tup;
807 : : Datum values[Natts_pg_publication_namespace];
808 : : bool nulls[Natts_pg_publication_namespace];
809 : : Oid psschid;
810 : 176 : Publication *pub = GetPublication(pubid);
811 : 176 : List *schemaRels = NIL;
812 : : ObjectAddress myself,
813 : : referenced;
814 : :
815 : 176 : rel = table_open(PublicationNamespaceRelationId, RowExclusiveLock);
816 : :
817 : : /*
818 : : * Check for duplicates. Note that this does not really prevent
819 : : * duplicates, it's here just to provide nicer error message in common
820 : : * case. The real protection is the unique key on the catalog.
821 : : */
822 [ + + ]: 176 : if (SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
823 : : ObjectIdGetDatum(schemaid),
824 : : ObjectIdGetDatum(pubid)))
825 : : {
826 : 12 : table_close(rel, RowExclusiveLock);
827 : :
828 [ + + ]: 12 : if (if_not_exists)
829 : 8 : return InvalidObjectAddress;
830 : :
831 [ + - ]: 4 : ereport(ERROR,
832 : : (errcode(ERRCODE_DUPLICATE_OBJECT),
833 : : errmsg("schema \"%s\" is already a member of publication \"%s\"",
834 : : get_namespace_name(schemaid), pub->name)));
835 : : }
836 : :
837 : 164 : check_publication_add_schema(schemaid);
838 : :
839 : : /* Form a tuple */
840 : 160 : memset(values, 0, sizeof(values));
841 : 160 : memset(nulls, false, sizeof(nulls));
842 : :
843 : 160 : psschid = GetNewOidWithIndex(rel, PublicationNamespaceObjectIndexId,
844 : : Anum_pg_publication_namespace_oid);
845 : 160 : values[Anum_pg_publication_namespace_oid - 1] = ObjectIdGetDatum(psschid);
846 : 160 : values[Anum_pg_publication_namespace_pnpubid - 1] =
847 : 160 : ObjectIdGetDatum(pubid);
848 : 160 : values[Anum_pg_publication_namespace_pnnspid - 1] =
849 : 160 : ObjectIdGetDatum(schemaid);
850 : :
851 : 160 : tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
852 : :
853 : : /* Insert tuple into catalog */
854 : 160 : CatalogTupleInsert(rel, tup);
855 : 160 : heap_freetuple(tup);
856 : :
857 : 160 : ObjectAddressSet(myself, PublicationNamespaceRelationId, psschid);
858 : :
859 : : /* Add dependency on the publication */
860 : 160 : ObjectAddressSet(referenced, PublicationRelationId, pubid);
861 : 160 : recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
862 : :
863 : : /* Add dependency on the schema */
864 : 160 : ObjectAddressSet(referenced, NamespaceRelationId, schemaid);
865 : 160 : recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
866 : :
867 : : /* Close the table */
868 : 160 : table_close(rel, RowExclusiveLock);
869 : :
870 : : /*
871 : : * Invalidate relcache so that publication info is rebuilt. See
872 : : * publication_add_relation for why we need to consider all the
873 : : * partitions.
874 : : */
875 : 160 : schemaRels = GetSchemaPublicationRelations(schemaid,
876 : : PUBLICATION_PART_ALL);
877 : 160 : InvalidatePublicationRels(schemaRels);
878 : :
879 : 160 : return myself;
880 : : }
881 : :
882 : : /*
883 : : * Internal function to get the list of publication oids for a relation.
884 : : *
885 : : * If except_flag is true, returns the list of publication that specified the
886 : : * relation in the EXCEPT clause; otherwise, returns the list of publications
887 : : * in which relation is included.
888 : : */
889 : : static List *
890 : 16970 : get_relation_publications(Oid relid, bool except_flag)
891 : : {
892 : 16970 : List *result = NIL;
893 : : CatCList *pubrellist;
894 : :
895 : : /* Find all publications associated with the relation. */
896 : 16970 : pubrellist = SearchSysCacheList1(PUBLICATIONRELMAP,
897 : : ObjectIdGetDatum(relid));
898 [ + + ]: 18641 : for (int i = 0; i < pubrellist->n_members; i++)
899 : : {
900 : 1671 : HeapTuple tup = &pubrellist->members[i]->tuple;
901 : 1671 : Form_pg_publication_rel pubrel = (Form_pg_publication_rel) GETSTRUCT(tup);
902 : 1671 : Oid pubid = pubrel->prpubid;
903 : :
904 [ + + ]: 1671 : if (pubrel->prexcept == except_flag)
905 : 1198 : result = lappend_oid(result, pubid);
906 : : }
907 : :
908 : 16970 : ReleaseSysCacheList(pubrellist);
909 : :
910 : 16970 : return result;
911 : : }
912 : :
913 : : /*
914 : : * Gets list of publication oids for a relation.
915 : : */
916 : : List *
917 : 9155 : GetRelationIncludedPublications(Oid relid)
918 : : {
919 : 9155 : return get_relation_publications(relid, false);
920 : : }
921 : :
922 : : /*
923 : : * Gets list of publication oids which has relation in the EXCEPT clause.
924 : : */
925 : : List *
926 : 7815 : GetRelationExcludedPublications(Oid relid)
927 : : {
928 : 7815 : return get_relation_publications(relid, true);
929 : : }
930 : :
931 : : /*
932 : : * Internal function to get the list of relation oids for a publication.
933 : : *
934 : : * If except_flag is true, returns the list of relations specified in the
935 : : * EXCEPT clause of the publication; otherwise, returns the list of relations
936 : : * included in the publication.
937 : : */
938 : : static List *
939 : 680 : get_publication_relations(Oid pubid, PublicationPartOpt pub_partopt,
940 : : bool except_flag)
941 : : {
942 : : List *result;
943 : : Relation pubrelsrel;
944 : : ScanKeyData scankey;
945 : : SysScanDesc scan;
946 : : HeapTuple tup;
947 : :
948 : : /* Find all relations associated with the publication. */
949 : 680 : pubrelsrel = table_open(PublicationRelRelationId, AccessShareLock);
950 : :
951 : 680 : ScanKeyInit(&scankey,
952 : : Anum_pg_publication_rel_prpubid,
953 : : BTEqualStrategyNumber, F_OIDEQ,
954 : : ObjectIdGetDatum(pubid));
955 : :
956 : 680 : scan = systable_beginscan(pubrelsrel, PublicationRelPrpubidIndexId,
957 : : true, NULL, 1, &scankey);
958 : :
959 : 680 : result = NIL;
960 [ + + ]: 1998 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
961 : : {
962 : : Form_pg_publication_rel pubrel;
963 : :
964 : 638 : pubrel = (Form_pg_publication_rel) GETSTRUCT(tup);
965 : :
966 [ - + ]: 638 : if (except_flag == pubrel->prexcept)
967 : 638 : result = GetPubPartitionOptionRelations(result, pub_partopt,
968 : : pubrel->prrelid);
969 : : }
970 : :
971 : 680 : systable_endscan(scan);
972 : 680 : table_close(pubrelsrel, AccessShareLock);
973 : :
974 : : /* Now sort and de-duplicate the result list */
975 : 680 : list_sort(result, list_oid_cmp);
976 : 680 : list_deduplicate_oid(result);
977 : :
978 : 680 : return result;
979 : : }
980 : :
981 : : /*
982 : : * Gets list of relation oids that are associated with a publication.
983 : : *
984 : : * This should only be used FOR TABLE publications, the FOR ALL TABLES/SEQUENCES
985 : : * should use GetAllPublicationRelations().
986 : : */
987 : : List *
988 : 610 : GetIncludedPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt)
989 : : {
990 : : Assert(!GetPublication(pubid)->alltables);
991 : :
992 : 610 : return get_publication_relations(pubid, pub_partopt, false);
993 : : }
994 : :
995 : : /*
996 : : * Gets list of table oids that were specified in the EXCEPT clause for a
997 : : * publication.
998 : : *
999 : : * This should only be used FOR ALL TABLES publications.
1000 : : */
1001 : : List *
1002 : 70 : GetExcludedPublicationTables(Oid pubid, PublicationPartOpt pub_partopt)
1003 : : {
1004 : : Assert(GetPublication(pubid)->alltables);
1005 : :
1006 : 70 : return get_publication_relations(pubid, pub_partopt, true);
1007 : : }
1008 : :
1009 : : /*
1010 : : * Gets list of publication oids for publications marked as FOR ALL TABLES.
1011 : : */
1012 : : List *
1013 : 6113 : GetAllTablesPublications(void)
1014 : : {
1015 : : List *result;
1016 : : Relation rel;
1017 : : ScanKeyData scankey;
1018 : : SysScanDesc scan;
1019 : : HeapTuple tup;
1020 : :
1021 : : /* Find all publications that are marked as for all tables. */
1022 : 6113 : rel = table_open(PublicationRelationId, AccessShareLock);
1023 : :
1024 : 6113 : ScanKeyInit(&scankey,
1025 : : Anum_pg_publication_puballtables,
1026 : : BTEqualStrategyNumber, F_BOOLEQ,
1027 : : BoolGetDatum(true));
1028 : :
1029 : 6113 : scan = systable_beginscan(rel, InvalidOid, false,
1030 : : NULL, 1, &scankey);
1031 : :
1032 : 6113 : result = NIL;
1033 [ + + ]: 6243 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
1034 : : {
1035 : 130 : Oid oid = ((Form_pg_publication) GETSTRUCT(tup))->oid;
1036 : :
1037 : 130 : result = lappend_oid(result, oid);
1038 : : }
1039 : :
1040 : 6113 : systable_endscan(scan);
1041 : 6113 : table_close(rel, AccessShareLock);
1042 : :
1043 : 6113 : return result;
1044 : : }
1045 : :
1046 : : /*
1047 : : * Gets list of all relations published by FOR ALL TABLES/SEQUENCES
1048 : : * publication.
1049 : : *
1050 : : * If the publication publishes partition changes via their respective root
1051 : : * partitioned tables, we must exclude partitions in favor of including the
1052 : : * root partitioned tables. This is not applicable to FOR ALL SEQUENCES
1053 : : * publication.
1054 : : *
1055 : : * For a FOR ALL TABLES publication, the returned list excludes tables mentioned
1056 : : * in the EXCEPT clause.
1057 : : */
1058 : : List *
1059 : 55 : GetAllPublicationRelations(Oid pubid, char relkind, bool pubviaroot)
1060 : : {
1061 : : Relation classRel;
1062 : : ScanKeyData key[1];
1063 : : TableScanDesc scan;
1064 : : HeapTuple tuple;
1065 : 55 : List *result = NIL;
1066 : 55 : List *exceptlist = NIL;
1067 : :
1068 : : Assert(!(relkind == RELKIND_SEQUENCE && pubviaroot));
1069 : :
1070 : : /* EXCEPT filtering applies only to relations, not sequences */
1071 [ + + ]: 55 : if (relkind == RELKIND_RELATION)
1072 : 49 : exceptlist = GetExcludedPublicationTables(pubid, pubviaroot ?
1073 : 49 : PUBLICATION_PART_ROOT :
1074 : : PUBLICATION_PART_LEAF);
1075 : :
1076 : 55 : classRel = table_open(RelationRelationId, AccessShareLock);
1077 : :
1078 : 55 : ScanKeyInit(&key[0],
1079 : : Anum_pg_class_relkind,
1080 : : BTEqualStrategyNumber, F_CHAREQ,
1081 : : CharGetDatum(relkind));
1082 : :
1083 : 55 : scan = table_beginscan_catalog(classRel, 1, key);
1084 : :
1085 [ + + ]: 3769 : while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
1086 : : {
1087 : 3714 : Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
1088 : 3714 : Oid relid = relForm->oid;
1089 : :
1090 [ + + ]: 3714 : if (is_publishable_class(relid, relForm) &&
1091 [ + + + + ]: 135 : !(relForm->relispartition && pubviaroot) &&
1092 [ + + ]: 112 : !list_member_oid(exceptlist, relid))
1093 : 104 : result = lappend_oid(result, relid);
1094 : : }
1095 : :
1096 : 55 : table_endscan(scan);
1097 : :
1098 [ + + ]: 55 : if (pubviaroot)
1099 : : {
1100 : 4 : ScanKeyInit(&key[0],
1101 : : Anum_pg_class_relkind,
1102 : : BTEqualStrategyNumber, F_CHAREQ,
1103 : : CharGetDatum(RELKIND_PARTITIONED_TABLE));
1104 : :
1105 : 4 : scan = table_beginscan_catalog(classRel, 1, key);
1106 : :
1107 [ + + ]: 21 : while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
1108 : : {
1109 : 17 : Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
1110 : 17 : Oid relid = relForm->oid;
1111 : :
1112 [ + - ]: 17 : if (is_publishable_class(relid, relForm) &&
1113 [ + + ]: 17 : !relForm->relispartition &&
1114 [ + + ]: 13 : !list_member_oid(exceptlist, relid))
1115 : 12 : result = lappend_oid(result, relid);
1116 : : }
1117 : :
1118 : 4 : table_endscan(scan);
1119 : : }
1120 : :
1121 : 55 : table_close(classRel, AccessShareLock);
1122 : 55 : return result;
1123 : : }
1124 : :
1125 : : /*
1126 : : * Gets the list of schema oids for a publication.
1127 : : *
1128 : : * This should only be used FOR TABLES IN SCHEMA publications.
1129 : : */
1130 : : List *
1131 : 576 : GetPublicationSchemas(Oid pubid)
1132 : : {
1133 : 576 : List *result = NIL;
1134 : : Relation pubschsrel;
1135 : : ScanKeyData scankey;
1136 : : SysScanDesc scan;
1137 : : HeapTuple tup;
1138 : :
1139 : : /* Find all schemas associated with the publication */
1140 : 576 : pubschsrel = table_open(PublicationNamespaceRelationId, AccessShareLock);
1141 : :
1142 : 576 : ScanKeyInit(&scankey,
1143 : : Anum_pg_publication_namespace_pnpubid,
1144 : : BTEqualStrategyNumber, F_OIDEQ,
1145 : : ObjectIdGetDatum(pubid));
1146 : :
1147 : 576 : scan = systable_beginscan(pubschsrel,
1148 : : PublicationNamespacePnnspidPnpubidIndexId,
1149 : : true, NULL, 1, &scankey);
1150 [ + + ]: 617 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
1151 : : {
1152 : : Form_pg_publication_namespace pubsch;
1153 : :
1154 : 41 : pubsch = (Form_pg_publication_namespace) GETSTRUCT(tup);
1155 : :
1156 : 41 : result = lappend_oid(result, pubsch->pnnspid);
1157 : : }
1158 : :
1159 : 576 : systable_endscan(scan);
1160 : 576 : table_close(pubschsrel, AccessShareLock);
1161 : :
1162 : 576 : return result;
1163 : : }
1164 : :
1165 : : /*
1166 : : * Gets the list of publication oids associated with a specified schema.
1167 : : */
1168 : : List *
1169 : 8811 : GetSchemaPublications(Oid schemaid)
1170 : : {
1171 : 8811 : List *result = NIL;
1172 : : CatCList *pubschlist;
1173 : : int i;
1174 : :
1175 : : /* Find all publications associated with the schema */
1176 : 8811 : pubschlist = SearchSysCacheList1(PUBLICATIONNAMESPACEMAP,
1177 : : ObjectIdGetDatum(schemaid));
1178 [ + + ]: 8890 : for (i = 0; i < pubschlist->n_members; i++)
1179 : : {
1180 : 79 : HeapTuple tup = &pubschlist->members[i]->tuple;
1181 : 79 : Oid pubid = ((Form_pg_publication_namespace) GETSTRUCT(tup))->pnpubid;
1182 : :
1183 : 79 : result = lappend_oid(result, pubid);
1184 : : }
1185 : :
1186 : 8811 : ReleaseSysCacheList(pubschlist);
1187 : :
1188 : 8811 : return result;
1189 : : }
1190 : :
1191 : : /*
1192 : : * Get the list of publishable relation oids for a specified schema.
1193 : : */
1194 : : List *
1195 : 313 : GetSchemaPublicationRelations(Oid schemaid, PublicationPartOpt pub_partopt)
1196 : : {
1197 : : Relation classRel;
1198 : : ScanKeyData key[1];
1199 : : TableScanDesc scan;
1200 : : HeapTuple tuple;
1201 : 313 : List *result = NIL;
1202 : :
1203 : : Assert(OidIsValid(schemaid));
1204 : :
1205 : 313 : classRel = table_open(RelationRelationId, AccessShareLock);
1206 : :
1207 : 313 : ScanKeyInit(&key[0],
1208 : : Anum_pg_class_relnamespace,
1209 : : BTEqualStrategyNumber, F_OIDEQ,
1210 : : ObjectIdGetDatum(schemaid));
1211 : :
1212 : : /* get all the relations present in the specified schema */
1213 : 313 : scan = table_beginscan_catalog(classRel, 1, key);
1214 [ + + ]: 17293 : while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
1215 : : {
1216 : 16980 : Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
1217 : 16980 : Oid relid = relForm->oid;
1218 : : char relkind;
1219 : :
1220 [ + + ]: 16980 : if (!is_publishable_class(relid, relForm))
1221 : 5871 : continue;
1222 : :
1223 : 11109 : relkind = get_rel_relkind(relid);
1224 [ + + ]: 11109 : if (relkind == RELKIND_RELATION)
1225 : 9556 : result = lappend_oid(result, relid);
1226 [ + + ]: 1553 : else if (relkind == RELKIND_PARTITIONED_TABLE)
1227 : : {
1228 : 495 : List *partitionrels = NIL;
1229 : :
1230 : : /*
1231 : : * It is quite possible that some of the partitions are in a
1232 : : * different schema than the parent table, so we need to get such
1233 : : * partitions separately.
1234 : : */
1235 : 495 : partitionrels = GetPubPartitionOptionRelations(partitionrels,
1236 : : pub_partopt,
1237 : : relForm->oid);
1238 : 495 : result = list_concat_unique_oid(result, partitionrels);
1239 : : }
1240 : : }
1241 : :
1242 : 313 : table_endscan(scan);
1243 : 313 : table_close(classRel, AccessShareLock);
1244 : 313 : return result;
1245 : : }
1246 : :
1247 : : /*
1248 : : * Gets the list of all relations published by FOR TABLES IN SCHEMA
1249 : : * publication.
1250 : : */
1251 : : List *
1252 : 283 : GetAllSchemaPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt)
1253 : : {
1254 : 283 : List *result = NIL;
1255 : 283 : List *pubschemalist = GetPublicationSchemas(pubid);
1256 : : ListCell *cell;
1257 : :
1258 [ + + + + : 304 : foreach(cell, pubschemalist)
+ + ]
1259 : : {
1260 : 21 : Oid schemaid = lfirst_oid(cell);
1261 : 21 : List *schemaRels = NIL;
1262 : :
1263 : 21 : schemaRels = GetSchemaPublicationRelations(schemaid, pub_partopt);
1264 : 21 : result = list_concat(result, schemaRels);
1265 : : }
1266 : :
1267 : 283 : return result;
1268 : : }
1269 : :
1270 : : /*
1271 : : * Get publication using oid
1272 : : *
1273 : : * The Publication struct and its data are palloc'ed here.
1274 : : */
1275 : : Publication *
1276 : 4195 : GetPublication(Oid pubid)
1277 : : {
1278 : : HeapTuple tup;
1279 : : Publication *pub;
1280 : : Form_pg_publication pubform;
1281 : :
1282 : 4195 : tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
1283 [ - + ]: 4195 : if (!HeapTupleIsValid(tup))
1284 [ # # ]: 0 : elog(ERROR, "cache lookup failed for publication %u", pubid);
1285 : :
1286 : 4195 : pubform = (Form_pg_publication) GETSTRUCT(tup);
1287 : :
1288 : 4195 : pub = palloc_object(Publication);
1289 : 4195 : pub->oid = pubid;
1290 : 4195 : pub->name = pstrdup(NameStr(pubform->pubname));
1291 : 4195 : pub->alltables = pubform->puballtables;
1292 : 4195 : pub->allsequences = pubform->puballsequences;
1293 : 4195 : pub->pubactions.pubinsert = pubform->pubinsert;
1294 : 4195 : pub->pubactions.pubupdate = pubform->pubupdate;
1295 : 4195 : pub->pubactions.pubdelete = pubform->pubdelete;
1296 : 4195 : pub->pubactions.pubtruncate = pubform->pubtruncate;
1297 : 4195 : pub->pubviaroot = pubform->pubviaroot;
1298 : 4195 : pub->pubgencols_type = pubform->pubgencols;
1299 : :
1300 : 4195 : ReleaseSysCache(tup);
1301 : :
1302 : 4195 : return pub;
1303 : : }
1304 : :
1305 : : /*
1306 : : * Get Publication using name.
1307 : : */
1308 : : Publication *
1309 : 1783 : GetPublicationByName(const char *pubname, bool missing_ok)
1310 : : {
1311 : : Oid oid;
1312 : :
1313 : 1783 : oid = get_publication_oid(pubname, missing_ok);
1314 : :
1315 [ + + ]: 1783 : return OidIsValid(oid) ? GetPublication(oid) : NULL;
1316 : : }
1317 : :
1318 : : /*
1319 : : * A helper function for pg_get_publication_tables() to check whether the
1320 : : * table with the given relid is published in the specified publication.
1321 : : *
1322 : : * This function evaluates the effective published OID based on the
1323 : : * publish_via_partition_root setting, rather than just checking catalog entries
1324 : : * (e.g., pg_publication_rel). For instance, when publish_via_partition_root is
1325 : : * false, it returns false for a parent partitioned table and returns true
1326 : : * for its leaf partitions, even if the parent is the one explicitly added
1327 : : * to the publication.
1328 : : *
1329 : : * For performance reasons, this function avoids the overhead of constructing
1330 : : * the complete list of published tables during the evaluation. It can execute
1331 : : * quickly even when the publication contains a large number of relations.
1332 : : *
1333 : : * Note: this leaks memory for the ancestors list into the current memory
1334 : : * context.
1335 : : */
1336 : : static bool
1337 : 987 : is_table_publishable_in_publication(Oid relid, Publication *pub)
1338 : : {
1339 : : bool relispartition;
1340 : 987 : List *ancestors = NIL;
1341 : :
1342 : : /*
1343 : : * For non-pubviaroot publications, a partitioned table is never the
1344 : : * effective published OID; only its leaf partitions can be.
1345 : : */
1346 [ + + + + ]: 987 : if (!pub->pubviaroot && get_rel_relkind(relid) == RELKIND_PARTITIONED_TABLE)
1347 : 86 : return false;
1348 : :
1349 : 901 : relispartition = get_rel_relispartition(relid);
1350 : :
1351 [ + + ]: 901 : if (relispartition)
1352 : 172 : ancestors = get_partition_ancestors(relid);
1353 : :
1354 [ + + ]: 901 : if (pub->alltables)
1355 : : {
1356 : : /*
1357 : : * ALL TABLES with pubviaroot includes only regular tables or top-most
1358 : : * partitioned tables -- never child partitions.
1359 : : */
1360 [ + + + + ]: 202 : if (pub->pubviaroot && relispartition)
1361 : 12 : return false;
1362 : :
1363 : : /*
1364 : : * For ALL TABLES publications, the table is published unless it
1365 : : * appears in the EXCEPT clause. Only the top-most can appear in the
1366 : : * EXCEPT clause, so exclusion must be evaluated at the top-most
1367 : : * ancestor if it has. These publications store only EXCEPT'ed tables
1368 : : * in pg_publication_rel, so checking existence is sufficient.
1369 : : *
1370 : : * Note that this existence check below would incorrectly return true
1371 : : * (published) for partitions when pubviaroot is enabled; however,
1372 : : * that case is already caught and returned false by the above check.
1373 : : */
1374 [ + + ]: 190 : return !SearchSysCacheExists2(PUBLICATIONRELMAP,
1375 : : ObjectIdGetDatum(ancestors
1376 : : ? llast_oid(ancestors) : relid),
1377 : : ObjectIdGetDatum(pub->oid));
1378 : : }
1379 : :
1380 : : /*
1381 : : * Non-ALL-TABLE publication cases.
1382 : : *
1383 : : * A table is published if it (or a containing schema) was explicitly
1384 : : * added, or if it is a partition whose ancestor was added.
1385 : : */
1386 : :
1387 : : /*
1388 : : * If an ancestor is published, the partition's status depends on
1389 : : * publish_via_partition_root value.
1390 : : *
1391 : : * If it's true, the ancestor's relation OID is the effective published
1392 : : * OID, so the partition itself should be excluded (return false).
1393 : : *
1394 : : * If it's false, the partition is covered by its ancestor's presence in
1395 : : * the publication, it should be included (return true).
1396 : : */
1397 [ + + + + ]: 841 : if (relispartition &&
1398 : 142 : OidIsValid(GetTopMostAncestorInPublication(pub->oid, ancestors, NULL)))
1399 : 44 : return !pub->pubviaroot;
1400 : :
1401 : : /*
1402 : : * Check whether the table is explicitly published via pg_publication_rel
1403 : : * or pg_publication_namespace.
1404 : : */
1405 : 655 : return (SearchSysCacheExists2(PUBLICATIONRELMAP,
1406 : : ObjectIdGetDatum(relid),
1407 [ + + + + ]: 1025 : ObjectIdGetDatum(pub->oid)) ||
1408 : 370 : SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
1409 : : ObjectIdGetDatum(get_rel_namespace(relid)),
1410 : : ObjectIdGetDatum(pub->oid)));
1411 : : }
1412 : :
1413 : : /*
1414 : : * Helper function to get information of the tables in the given
1415 : : * publication(s).
1416 : : *
1417 : : * If filter_by_relid is true, only the row(s) for target_relid is returned;
1418 : : * if target_relid does not exist or is not part of the publications, zero
1419 : : * rows are returned. If filter_by_relid is false, rows for all tables
1420 : : * within the specified publications are returned and target_relid is
1421 : : * ignored.
1422 : : *
1423 : : * Returns pubid, relid, column list, and row filter for each table.
1424 : : */
1425 : : static Datum
1426 : 1601 : pg_get_publication_tables(FunctionCallInfo fcinfo, ArrayType *pubnames,
1427 : : Oid target_relid, bool filter_by_relid,
1428 : : bool pub_missing_ok)
1429 : : {
1430 : : #define NUM_PUBLICATION_TABLES_ELEM 4
1431 : :
1432 : : /*
1433 : : * State carried across SRF calls. We track the index ourselves instead of
1434 : : * using funcctx->call_cntr, so that concurrently dropped tables can be
1435 : : * skipped without emitting a row.
1436 : : */
1437 : : typedef struct
1438 : : {
1439 : : List *table_infos; /* list of published_rel */
1440 : : int curr_idx; /* current index into table_infos */
1441 : : } publication_tables_state;
1442 : :
1443 : : FuncCallContext *funcctx;
1444 : 1601 : publication_tables_state *ptstate = NULL;
1445 : :
1446 : : /* stuff done only on the first call of the function */
1447 [ + + ]: 1601 : if (SRF_IS_FIRSTCALL())
1448 : : {
1449 : : TupleDesc tupdesc;
1450 : : MemoryContext oldcontext;
1451 : 755 : List *table_infos = NIL;
1452 : : Datum *elems;
1453 : : int nelems,
1454 : : i;
1455 : 755 : bool viaroot = false;
1456 : :
1457 : : /* create a function context for cross-call persistence */
1458 : 755 : funcctx = SRF_FIRSTCALL_INIT();
1459 : :
1460 : : /*
1461 : : * Preliminary check if the specified table can be published in the
1462 : : * first place. If not, we can return early without checking the given
1463 : : * publications and the table.
1464 : : */
1465 [ + + + + ]: 755 : if (filter_by_relid && !is_publishable_table(target_relid))
1466 : 8 : SRF_RETURN_DONE(funcctx);
1467 : :
1468 : : /* switch to memory context appropriate for multiple function calls */
1469 : 747 : oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
1470 : :
1471 : : /*
1472 : : * Deconstruct the parameter into elements where each element is a
1473 : : * publication name.
1474 : : */
1475 : 747 : deconstruct_array_builtin(pubnames, TEXTOID, &elems, NULL, &nelems);
1476 : :
1477 : : /* Get Oids of tables from each publication. */
1478 [ + + ]: 1992 : for (i = 0; i < nelems; i++)
1479 : : {
1480 : : Publication *pub_elem;
1481 : 1245 : List *pub_elem_tables = NIL;
1482 : : ListCell *lc;
1483 : :
1484 : 1245 : pub_elem = GetPublicationByName(TextDatumGetCString(elems[i]),
1485 : : pub_missing_ok);
1486 : :
1487 [ + + ]: 1245 : if (pub_elem == NULL)
1488 : 6 : continue;
1489 : :
1490 [ + + ]: 1239 : if (filter_by_relid)
1491 : : {
1492 : : /* Check if the given table is published for the publication */
1493 [ + + ]: 987 : if (is_table_publishable_in_publication(target_relid, pub_elem))
1494 : : {
1495 : 501 : pub_elem_tables = list_make1_oid(target_relid);
1496 : : }
1497 : : }
1498 : : else
1499 : : {
1500 : : /*
1501 : : * Publications support partitioned tables. If
1502 : : * publish_via_partition_root is false, all changes are
1503 : : * replicated using leaf partition identity and schema, so we
1504 : : * only need those. Otherwise, get the partitioned table
1505 : : * itself.
1506 : : */
1507 [ + + ]: 252 : if (pub_elem->alltables)
1508 : 49 : pub_elem_tables = GetAllPublicationRelations(pub_elem->oid,
1509 : : RELKIND_RELATION,
1510 : 49 : pub_elem->pubviaroot);
1511 : : else
1512 : : {
1513 : : List *relids,
1514 : : *schemarelids;
1515 : :
1516 : 203 : relids = GetIncludedPublicationRelations(pub_elem->oid,
1517 : 203 : pub_elem->pubviaroot ?
1518 : 203 : PUBLICATION_PART_ROOT :
1519 : : PUBLICATION_PART_LEAF);
1520 : 203 : schemarelids = GetAllSchemaPublicationRelations(pub_elem->oid,
1521 : 203 : pub_elem->pubviaroot ?
1522 : 203 : PUBLICATION_PART_ROOT :
1523 : : PUBLICATION_PART_LEAF);
1524 : 203 : pub_elem_tables = list_concat_unique_oid(relids, schemarelids);
1525 : : }
1526 : : }
1527 : :
1528 : : /*
1529 : : * Record the published table and the corresponding publication so
1530 : : * that we can get row filters and column lists later.
1531 : : *
1532 : : * When a table is published by multiple publications, to obtain
1533 : : * all row filters and column lists, the structure related to this
1534 : : * table will be recorded multiple times.
1535 : : */
1536 [ + + + + : 2116 : foreach(lc, pub_elem_tables)
+ + ]
1537 : : {
1538 : 877 : published_rel *table_info = palloc_object(published_rel);
1539 : :
1540 : 877 : table_info->relid = lfirst_oid(lc);
1541 : 877 : table_info->pubid = pub_elem->oid;
1542 : 877 : table_infos = lappend(table_infos, table_info);
1543 : : }
1544 : :
1545 : : /* At least one publication is using publish_via_partition_root. */
1546 [ + + ]: 1239 : if (pub_elem->pubviaroot)
1547 : 248 : viaroot = true;
1548 : : }
1549 : :
1550 : : /*
1551 : : * If the publication publishes partition changes via their respective
1552 : : * root partitioned tables, we must exclude partitions in favor of
1553 : : * including the root partitioned tables. Otherwise, the function
1554 : : * could return both the child and parent tables which could cause
1555 : : * data of the child table to be double-published on the subscriber
1556 : : * side.
1557 : : */
1558 [ + + ]: 747 : if (viaroot)
1559 : 168 : filter_partitions(table_infos);
1560 : :
1561 : : /* Construct a tuple descriptor for the result rows. */
1562 : 747 : tupdesc = CreateTemplateTupleDesc(NUM_PUBLICATION_TABLES_ELEM);
1563 : 747 : TupleDescInitEntry(tupdesc, (AttrNumber) 1, "pubid",
1564 : : OIDOID, -1, 0);
1565 : 747 : TupleDescInitEntry(tupdesc, (AttrNumber) 2, "relid",
1566 : : OIDOID, -1, 0);
1567 : 747 : TupleDescInitEntry(tupdesc, (AttrNumber) 3, "attrs",
1568 : : INT2VECTOROID, -1, 0);
1569 : 747 : TupleDescInitEntry(tupdesc, (AttrNumber) 4, "qual",
1570 : : PG_NODE_TREEOID, -1, 0);
1571 : :
1572 : 747 : TupleDescFinalize(tupdesc);
1573 : 747 : funcctx->tuple_desc = BlessTupleDesc(tupdesc);
1574 : :
1575 : : /* Store the state to be used across SRF calls. */
1576 : 747 : ptstate = palloc_object(publication_tables_state);
1577 : 747 : ptstate->table_infos = table_infos;
1578 : 747 : ptstate->curr_idx = 0;
1579 : 747 : funcctx->user_fctx = ptstate;
1580 : :
1581 : 747 : MemoryContextSwitchTo(oldcontext);
1582 : : }
1583 : :
1584 : : /* stuff done on every call of the function */
1585 : 1593 : funcctx = SRF_PERCALL_SETUP();
1586 : 1593 : ptstate = (publication_tables_state *) funcctx->user_fctx;
1587 : :
1588 [ + + ]: 1594 : while (ptstate->curr_idx < list_length(ptstate->table_infos))
1589 : : {
1590 : 847 : HeapTuple pubtuple = NULL;
1591 : : HeapTuple rettuple;
1592 : : Publication *pub;
1593 : 847 : published_rel *table_info = (published_rel *) list_nth(ptstate->table_infos,
1594 : : ptstate->curr_idx);
1595 : 847 : Oid relid = table_info->relid;
1596 : : Relation rel;
1597 : : Oid schemaid;
1598 : 847 : Datum values[NUM_PUBLICATION_TABLES_ELEM] = {0};
1599 : 847 : bool nulls[NUM_PUBLICATION_TABLES_ELEM] = {0};
1600 : :
1601 : : /* Advance the index for the next call. */
1602 : 847 : ptstate->curr_idx++;
1603 : :
1604 : : /*
1605 : : * The table OIDs were collected earlier, so a table may have been
1606 : : * dropped before we get here. try_table_open() returns NULL if it is
1607 : : * already gone, in which case we skip it; such tables are simply
1608 : : * absent from the result set, which is the expected point-in-time
1609 : : * behavior.
1610 : : */
1611 : 847 : rel = try_table_open(relid, AccessShareLock);
1612 [ + + ]: 847 : if (rel == NULL)
1613 : 1 : continue;
1614 : :
1615 : : /*
1616 : : * Form tuple with appropriate data.
1617 : : */
1618 : :
1619 : 846 : pub = GetPublication(table_info->pubid);
1620 : :
1621 : 846 : values[0] = ObjectIdGetDatum(pub->oid);
1622 : 846 : values[1] = ObjectIdGetDatum(relid);
1623 : :
1624 : : /*
1625 : : * We don't consider row filters or column lists for FOR ALL TABLES or
1626 : : * FOR TABLES IN SCHEMA publications.
1627 : : */
1628 : 846 : schemaid = RelationGetNamespace(rel);
1629 [ + + ]: 846 : if (!pub->alltables &&
1630 [ + + ]: 575 : !SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
1631 : : ObjectIdGetDatum(schemaid),
1632 : : ObjectIdGetDatum(pub->oid)))
1633 : 538 : pubtuple = SearchSysCacheCopy2(PUBLICATIONRELMAP,
1634 : : ObjectIdGetDatum(relid),
1635 : : ObjectIdGetDatum(pub->oid));
1636 : :
1637 [ + + ]: 846 : if (HeapTupleIsValid(pubtuple))
1638 : : {
1639 : : /* Lookup the column list attribute. */
1640 : 499 : values[2] = SysCacheGetAttr(PUBLICATIONRELMAP, pubtuple,
1641 : : Anum_pg_publication_rel_prattrs,
1642 : : &(nulls[2]));
1643 : :
1644 : : /* Null indicates no filter. */
1645 : 499 : values[3] = SysCacheGetAttr(PUBLICATIONRELMAP, pubtuple,
1646 : : Anum_pg_publication_rel_prqual,
1647 : : &(nulls[3]));
1648 : : }
1649 : : else
1650 : : {
1651 : 347 : nulls[2] = true;
1652 : 347 : nulls[3] = true;
1653 : : }
1654 : :
1655 : : /* Show all columns when the column list is not specified. */
1656 [ + + ]: 846 : if (nulls[2])
1657 : : {
1658 : 750 : int nattnums = 0;
1659 : : int16 *attnums;
1660 : 750 : TupleDesc desc = RelationGetDescr(rel);
1661 : : int i;
1662 : :
1663 : 750 : attnums = palloc_array(int16, desc->natts);
1664 : :
1665 [ + + ]: 1982 : for (i = 0; i < desc->natts; i++)
1666 : : {
1667 : 1232 : Form_pg_attribute att = TupleDescAttr(desc, i);
1668 : :
1669 [ + + ]: 1232 : if (att->attisdropped)
1670 : 4 : continue;
1671 : :
1672 [ + + ]: 1228 : if (att->attgenerated)
1673 : : {
1674 : : /* We only support replication of STORED generated cols. */
1675 [ + + ]: 22 : if (att->attgenerated != ATTRIBUTE_GENERATED_STORED)
1676 : 10 : continue;
1677 : :
1678 : : /*
1679 : : * User hasn't requested to replicate STORED generated
1680 : : * cols.
1681 : : */
1682 [ + + ]: 12 : if (pub->pubgencols_type != PUBLISH_GENCOLS_STORED)
1683 : 9 : continue;
1684 : : }
1685 : :
1686 : 1209 : attnums[nattnums++] = att->attnum;
1687 : : }
1688 : :
1689 [ + + ]: 750 : if (nattnums > 0)
1690 : : {
1691 : 746 : values[2] = PointerGetDatum(buildint2vector(attnums, nattnums));
1692 : 746 : nulls[2] = false;
1693 : : }
1694 : : }
1695 : :
1696 : 846 : table_close(rel, AccessShareLock);
1697 : :
1698 : 846 : rettuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
1699 : :
1700 : 846 : SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(rettuple));
1701 : : }
1702 : :
1703 : 747 : SRF_RETURN_DONE(funcctx);
1704 : : }
1705 : :
1706 : : Datum
1707 : 549 : pg_get_publication_tables_a(PG_FUNCTION_ARGS)
1708 : : {
1709 : : /*
1710 : : * Get information for all tables in the given publications.
1711 : : * filter_by_relid is false so all tables are returned; pub_missing_ok is
1712 : : * false for backward compatibility.
1713 : : */
1714 : 549 : return pg_get_publication_tables(fcinfo, PG_GETARG_ARRAYTYPE_P(0),
1715 : : InvalidOid, false, false);
1716 : : }
1717 : :
1718 : : Datum
1719 : 1052 : pg_get_publication_tables_b(PG_FUNCTION_ARGS)
1720 : : {
1721 : : /*
1722 : : * Get information for the specified table in the given publications. The
1723 : : * SQL-level function is declared STRICT, so target_relid is guaranteed to
1724 : : * be non-NULL here.
1725 : : */
1726 : 1052 : return pg_get_publication_tables(fcinfo, PG_GETARG_ARRAYTYPE_P(0),
1727 : : PG_GETARG_OID(1), true, true);
1728 : : }
1729 : :
1730 : : /*
1731 : : * Returns Oids of sequences in a publication.
1732 : : */
1733 : : Datum
1734 : 234 : pg_get_publication_sequences(PG_FUNCTION_ARGS)
1735 : : {
1736 : : FuncCallContext *funcctx;
1737 : 234 : List *sequences = NIL;
1738 : :
1739 : : /* stuff done only on the first call of the function */
1740 [ + + ]: 234 : if (SRF_IS_FIRSTCALL())
1741 : : {
1742 : 217 : char *pubname = text_to_cstring(PG_GETARG_TEXT_PP(0));
1743 : : Publication *publication;
1744 : : MemoryContext oldcontext;
1745 : :
1746 : : /* create a function context for cross-call persistence */
1747 : 217 : funcctx = SRF_FIRSTCALL_INIT();
1748 : :
1749 : : /* switch to memory context appropriate for multiple function calls */
1750 : 217 : oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
1751 : :
1752 : 217 : publication = GetPublicationByName(pubname, false);
1753 : :
1754 [ + + ]: 217 : if (publication->allsequences)
1755 : 6 : sequences = GetAllPublicationRelations(publication->oid,
1756 : : RELKIND_SEQUENCE,
1757 : : false);
1758 : :
1759 : 217 : funcctx->user_fctx = sequences;
1760 : :
1761 : 217 : MemoryContextSwitchTo(oldcontext);
1762 : : }
1763 : :
1764 : : /* stuff done on every call of the function */
1765 : 234 : funcctx = SRF_PERCALL_SETUP();
1766 : 234 : sequences = (List *) funcctx->user_fctx;
1767 : :
1768 [ + + ]: 234 : if (funcctx->call_cntr < list_length(sequences))
1769 : : {
1770 : 17 : Oid relid = list_nth_oid(sequences, funcctx->call_cntr);
1771 : :
1772 : 17 : SRF_RETURN_NEXT(funcctx, ObjectIdGetDatum(relid));
1773 : : }
1774 : :
1775 : 217 : SRF_RETURN_DONE(funcctx);
1776 : : }
|
