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