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