Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * pg_publication.c
4 : : * publication C API manipulation
5 : : *
6 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : * IDENTIFICATION
10 : : * src/backend/catalog/pg_publication.c
11 : : *
12 : : *-------------------------------------------------------------------------
13 : : */
14 : :
15 : : #include "postgres.h"
16 : :
17 : : #include "access/genam.h"
18 : : #include "access/heapam.h"
19 : : #include "access/htup_details.h"
20 : : #include "access/tableam.h"
21 : : #include "catalog/catalog.h"
22 : : #include "catalog/dependency.h"
23 : : #include "catalog/indexing.h"
24 : : #include "catalog/namespace.h"
25 : : #include "catalog/objectaddress.h"
26 : : #include "catalog/partition.h"
27 : : #include "catalog/pg_inherits.h"
28 : : #include "catalog/pg_namespace.h"
29 : : #include "catalog/pg_publication.h"
30 : : #include "catalog/pg_publication_namespace.h"
31 : : #include "catalog/pg_publication_rel.h"
32 : : #include "catalog/pg_type.h"
33 : : #include "commands/publicationcmds.h"
34 : : #include "funcapi.h"
35 : : #include "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
56 : 133 : check_publication_add_relation(Relation targetrel)
57 : : {
58 : : /* Must be a regular or partitioned table */
59 [ + + + + ]: 133 : if (RelationGetForm(targetrel)->relkind != RELKIND_RELATION &&
60 : 17 : RelationGetForm(targetrel)->relkind != RELKIND_PARTITIONED_TABLE)
61 [ + - + - ]: 2 : ereport(ERROR,
62 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
63 : : errmsg("cannot add relation \"%s\" to publication",
64 : : RelationGetRelationName(targetrel)),
65 : : errdetail_relkind_not_supported(RelationGetForm(targetrel)->relkind)));
66 : :
67 : : /* Can't be system table */
68 [ + + ]: 131 : if (IsCatalogRelation(targetrel))
69 [ + - + - ]: 1 : ereport(ERROR,
70 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
71 : : errmsg("cannot add relation \"%s\" to publication",
72 : : RelationGetRelationName(targetrel)),
73 : : errdetail("This operation is not supported for system tables.")));
74 : :
75 : : /* UNLOGGED and TEMP relations cannot be part of publication. */
76 [ + + ]: 130 : if (targetrel->rd_rel->relpersistence == RELPERSISTENCE_TEMP)
77 [ + - + - ]: 1 : ereport(ERROR,
78 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
79 : : errmsg("cannot add relation \"%s\" to publication",
80 : : RelationGetRelationName(targetrel)),
81 : : errdetail("This operation is not supported for temporary tables.")));
82 [ + + ]: 129 : else if (targetrel->rd_rel->relpersistence == RELPERSISTENCE_UNLOGGED)
83 [ + - + - ]: 1 : ereport(ERROR,
84 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
85 : : errmsg("cannot add relation \"%s\" to publication",
86 : : RelationGetRelationName(targetrel)),
87 : : errdetail("This operation is not supported for unlogged tables.")));
88 : 128 : }
89 : :
90 : : /*
91 : : * Check if schema can be in given publication and throw appropriate error if
92 : : * not.
93 : : */
94 : : static void
95 : 37 : check_publication_add_schema(Oid schemaid)
96 : : {
97 : : /* Can't be system namespace */
98 [ + + ]: 37 : if (IsCatalogNamespace(schemaid) || IsToastNamespace(schemaid))
99 [ + - + - ]: 1 : ereport(ERROR,
100 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
101 : : errmsg("cannot add schema \"%s\" to publication",
102 : : get_namespace_name(schemaid)),
103 : : errdetail("This operation is not supported for system schemas.")));
104 : :
105 : : /* Can't be temporary namespace */
106 [ + - ]: 36 : if (isAnyTempNamespace(schemaid))
107 [ # # # # ]: 0 : ereport(ERROR,
108 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
109 : : errmsg("cannot add schema \"%s\" to publication",
110 : : get_namespace_name(schemaid)),
111 : : errdetail("Temporary schemas cannot be replicated.")));
112 : 36 : }
113 : :
114 : : /*
115 : : * Returns if relation represented by oid and Form_pg_class entry
116 : : * is publishable.
117 : : *
118 : : * Does same checks as check_publication_add_relation() above except for
119 : : * RELKIND_SEQUENCE, but does not need relation to be opened and also does
120 : : * not throw errors. Here, the additional check is to support ALL SEQUENCES
121 : : * publication.
122 : : *
123 : : * XXX This also excludes all tables with relid < FirstNormalObjectId,
124 : : * ie all tables created during initdb. This mainly affects the preinstalled
125 : : * information_schema. IsCatalogRelationOid() only excludes tables with
126 : : * relid < FirstUnpinnedObjectId, making that test rather redundant,
127 : : * but really we should get rid of the FirstNormalObjectId test not
128 : : * IsCatalogRelationOid. We can't do so today because we don't want
129 : : * information_schema tables to be considered publishable; but this test
130 : : * is really inadequate for that, since the information_schema could be
131 : : * dropped and reloaded and then it'll be considered publishable. The best
132 : : * long-term solution may be to add a "relispublishable" bool to pg_class,
133 : : * and depend on that instead of OID checks.
134 : : */
135 : : static bool
136 : 8966 : is_publishable_class(Oid relid, Form_pg_class reltuple)
137 : : {
138 [ + + ]: 17932 : return (reltuple->relkind == RELKIND_RELATION ||
139 [ + + ]: 1997 : reltuple->relkind == RELKIND_PARTITIONED_TABLE ||
140 : 1775 : reltuple->relkind == RELKIND_SEQUENCE) &&
141 [ + + ]: 8966 : !IsCatalogRelationOid(relid) &&
142 [ + + ]: 7513 : reltuple->relpersistence == RELPERSISTENCE_PERMANENT &&
143 : 7102 : relid >= FirstNormalObjectId;
144 : : }
145 : :
146 : : /*
147 : : * Another variant of is_publishable_class(), taking a Relation.
148 : : */
149 : : bool
150 : 3743 : is_publishable_relation(Relation rel)
151 : : {
152 : 3743 : return is_publishable_class(RelationGetRelid(rel), rel->rd_rel);
153 : : }
154 : :
155 : : /*
156 : : * SQL-callable variant of the above
157 : : *
158 : : * This returns null when the relation does not exist. This is intended to be
159 : : * used for example in psql to avoid gratuitous errors when there are
160 : : * concurrent catalog changes.
161 : : */
162 : : Datum
163 : 1044 : pg_relation_is_publishable(PG_FUNCTION_ARGS)
164 : : {
165 : 1044 : Oid relid = PG_GETARG_OID(0);
166 : 1044 : HeapTuple tuple;
167 : 1044 : bool result;
168 : :
169 : 1044 : tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
170 [ + - ]: 1044 : if (!HeapTupleIsValid(tuple))
171 : 0 : PG_RETURN_NULL();
172 : 1044 : result = is_publishable_class(relid, (Form_pg_class) GETSTRUCT(tuple));
173 : 1044 : ReleaseSysCache(tuple);
174 : 1044 : PG_RETURN_BOOL(result);
175 : 1044 : }
176 : :
177 : : /*
178 : : * Returns true if the ancestor is in the list of published relations.
179 : : * Otherwise, returns false.
180 : : */
181 : : static bool
182 : 6 : is_ancestor_member_tableinfos(Oid ancestor, List *table_infos)
183 : : {
184 : 6 : ListCell *lc;
185 : :
186 [ + - + + : 13 : foreach(lc, table_infos)
+ + + + ]
187 : : {
188 : 7 : Oid relid = ((published_rel *) lfirst(lc))->relid;
189 : :
190 [ + + ]: 7 : if (relid == ancestor)
191 : 4 : return true;
192 [ + + ]: 7 : }
193 : :
194 : 2 : return false;
195 : 6 : }
196 : :
197 : : /*
198 : : * Filter out the partitions whose parent tables are also present in the list.
199 : : */
200 : : static void
201 : 4 : filter_partitions(List *table_infos)
202 : : {
203 : 4 : ListCell *lc;
204 : :
205 [ + - + + : 12 : foreach(lc, table_infos)
+ + ]
206 : : {
207 : 8 : bool skip = false;
208 : 8 : List *ancestors = NIL;
209 : 8 : ListCell *lc2;
210 : 8 : published_rel *table_info = (published_rel *) lfirst(lc);
211 : :
212 [ + + ]: 8 : if (get_rel_relispartition(table_info->relid))
213 : 6 : ancestors = get_partition_ancestors(table_info->relid);
214 : :
215 [ + + + + : 14 : foreach(lc2, ancestors)
+ + ]
216 : : {
217 : 6 : Oid ancestor = lfirst_oid(lc2);
218 : :
219 [ + + ]: 6 : if (is_ancestor_member_tableinfos(ancestor, table_infos))
220 : : {
221 : 4 : skip = true;
222 : 4 : break;
223 : : }
224 [ + + ]: 6 : }
225 : :
226 [ + + ]: 8 : if (skip)
227 : 4 : table_infos = foreach_delete_current(table_infos, lc);
228 : 8 : }
229 : 4 : }
230 : :
231 : : /*
232 : : * Returns true if any schema is associated with the publication, false if no
233 : : * schema is associated with the publication.
234 : : */
235 : : bool
236 : 36 : is_schema_publication(Oid pubid)
237 : : {
238 : 36 : Relation pubschsrel;
239 : 36 : ScanKeyData scankey;
240 : 36 : SysScanDesc scan;
241 : 36 : HeapTuple tup;
242 : 36 : bool result = false;
243 : :
244 : 36 : pubschsrel = table_open(PublicationNamespaceRelationId, AccessShareLock);
245 : 36 : ScanKeyInit(&scankey,
246 : : Anum_pg_publication_namespace_pnpubid,
247 : : BTEqualStrategyNumber, F_OIDEQ,
248 : 36 : ObjectIdGetDatum(pubid));
249 : :
250 : 36 : scan = systable_beginscan(pubschsrel,
251 : : PublicationNamespacePnnspidPnpubidIndexId,
252 : : true, NULL, 1, &scankey);
253 : 36 : tup = systable_getnext(scan);
254 : 36 : result = HeapTupleIsValid(tup);
255 : :
256 : 36 : systable_endscan(scan);
257 : 36 : table_close(pubschsrel, AccessShareLock);
258 : :
259 : 72 : return result;
260 : 36 : }
261 : :
262 : : /*
263 : : * Returns true if the relation has column list associated with the
264 : : * publication, false otherwise.
265 : : *
266 : : * If a column list is found, the corresponding bitmap is returned through the
267 : : * cols parameter, if provided. The bitmap is constructed within the given
268 : : * memory context (mcxt).
269 : : */
270 : : bool
271 : 74 : check_and_fetch_column_list(Publication *pub, Oid relid, MemoryContext mcxt,
272 : : Bitmapset **cols)
273 : : {
274 : 74 : HeapTuple cftuple;
275 : 74 : bool found = false;
276 : :
277 [ + + ]: 74 : if (pub->alltables)
278 : 5 : return false;
279 : :
280 : 69 : cftuple = SearchSysCache2(PUBLICATIONRELMAP,
281 : 69 : ObjectIdGetDatum(relid),
282 : 69 : ObjectIdGetDatum(pub->oid));
283 [ + + ]: 69 : if (HeapTupleIsValid(cftuple))
284 : : {
285 : 63 : Datum cfdatum;
286 : 63 : bool isnull;
287 : :
288 : : /* Lookup the column list attribute. */
289 : 63 : cfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, cftuple,
290 : : Anum_pg_publication_rel_prattrs, &isnull);
291 : :
292 : : /* Was a column list found? */
293 [ + + ]: 63 : if (!isnull)
294 : : {
295 : : /* Build the column list bitmap in the given memory context. */
296 [ - + ]: 34 : if (cols)
297 : 34 : *cols = pub_collist_to_bitmapset(*cols, cfdatum, mcxt);
298 : :
299 : 34 : found = true;
300 : 34 : }
301 : :
302 : 63 : ReleaseSysCache(cftuple);
303 : 63 : }
304 : :
305 : 69 : return found;
306 : 74 : }
307 : :
308 : : /*
309 : : * Gets the relations based on the publication partition option for a specified
310 : : * relation.
311 : : */
312 : : List *
313 : 461 : GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
314 : : Oid relid)
315 : : {
316 [ + + + + ]: 461 : if (get_rel_relkind(relid) == RELKIND_PARTITIONED_TABLE &&
317 : 163 : pub_partopt != PUBLICATION_PART_ROOT)
318 : : {
319 : 152 : List *all_parts = find_all_inheritors(relid, NoLock,
320 : : NULL);
321 : :
322 [ + + ]: 152 : if (pub_partopt == PUBLICATION_PART_ALL)
323 : 151 : result = list_concat(result, all_parts);
324 [ + - ]: 1 : else if (pub_partopt == PUBLICATION_PART_LEAF)
325 : : {
326 : 1 : ListCell *lc;
327 : :
328 [ + - + + : 3 : foreach(lc, all_parts)
+ + ]
329 : : {
330 : 2 : Oid partOid = lfirst_oid(lc);
331 : :
332 [ + + ]: 2 : if (get_rel_relkind(partOid) != RELKIND_PARTITIONED_TABLE)
333 : 1 : result = lappend_oid(result, partOid);
334 : 2 : }
335 : 1 : }
336 : : else
337 : 0 : Assert(false);
338 : 152 : }
339 : : else
340 : 309 : result = lappend_oid(result, relid);
341 : :
342 : 461 : return result;
343 : : }
344 : :
345 : : /*
346 : : * Returns the relid of the topmost ancestor that is published via this
347 : : * publication if any and set its ancestor level to ancestor_level,
348 : : * otherwise returns InvalidOid.
349 : : *
350 : : * The ancestor_level value allows us to compare the results for multiple
351 : : * publications, and decide which value is higher up.
352 : : *
353 : : * Note that the list of ancestors should be ordered such that the topmost
354 : : * ancestor is at the end of the list.
355 : : */
356 : : Oid
357 : 31 : GetTopMostAncestorInPublication(Oid puboid, List *ancestors, int *ancestor_level)
358 : : {
359 : 31 : ListCell *lc;
360 : 31 : Oid topmost_relid = InvalidOid;
361 : 31 : int level = 0;
362 : :
363 : : /*
364 : : * Find the "topmost" ancestor that is in this publication.
365 : : */
366 [ + - + + : 62 : foreach(lc, ancestors)
+ + ]
367 : : {
368 : 31 : Oid ancestor = lfirst_oid(lc);
369 : 31 : List *apubids = GetRelationPublications(ancestor);
370 : 31 : List *aschemaPubids = NIL;
371 : :
372 : 31 : level++;
373 : :
374 [ + - ]: 31 : if (list_member_oid(apubids, puboid))
375 : : {
376 : 31 : topmost_relid = ancestor;
377 : :
378 [ + - ]: 31 : if (ancestor_level)
379 : 0 : *ancestor_level = level;
380 : 31 : }
381 : : else
382 : : {
383 : 0 : aschemaPubids = GetSchemaPublications(get_rel_namespace(ancestor));
384 [ # # ]: 0 : if (list_member_oid(aschemaPubids, puboid))
385 : : {
386 : 0 : topmost_relid = ancestor;
387 : :
388 [ # # ]: 0 : if (ancestor_level)
389 : 0 : *ancestor_level = level;
390 : 0 : }
391 : : }
392 : :
393 : 31 : list_free(apubids);
394 : 31 : list_free(aschemaPubids);
395 : 31 : }
396 : :
397 : 62 : return topmost_relid;
398 : 31 : }
399 : :
400 : : /*
401 : : * attnumstoint2vector
402 : : * Convert a Bitmapset of AttrNumbers into an int2vector.
403 : : *
404 : : * AttrNumber numbers are 0-based, i.e., not offset by
405 : : * FirstLowInvalidHeapAttributeNumber.
406 : : */
407 : : static int2vector *
408 : 42 : attnumstoint2vector(Bitmapset *attrs)
409 : : {
410 : 42 : int2vector *result;
411 : 42 : int n = bms_num_members(attrs);
412 : 42 : int i = -1;
413 : 42 : int j = 0;
414 : :
415 : 42 : result = buildint2vector(NULL, n);
416 : :
417 [ + + ]: 113 : while ((i = bms_next_member(attrs, i)) >= 0)
418 : : {
419 [ + - ]: 71 : Assert(i <= PG_INT16_MAX);
420 : :
421 : 71 : result->values[j++] = (int16) i;
422 : : }
423 : :
424 : 84 : return result;
425 : 42 : }
426 : :
427 : : /*
428 : : * Insert new publication / relation mapping.
429 : : */
430 : : ObjectAddress
431 : 127 : publication_add_relation(Oid pubid, PublicationRelInfo *pri,
432 : : bool if_not_exists)
433 : : {
434 : 127 : Relation rel;
435 : 127 : HeapTuple tup;
436 : 127 : Datum values[Natts_pg_publication_rel];
437 : 127 : bool nulls[Natts_pg_publication_rel];
438 : 127 : Relation targetrel = pri->relation;
439 : 127 : Oid relid = RelationGetRelid(targetrel);
440 : 127 : Oid pubreloid;
441 : 127 : Bitmapset *attnums;
442 : 127 : Publication *pub = GetPublication(pubid);
443 : 127 : ObjectAddress myself,
444 : : referenced;
445 : 127 : List *relids = NIL;
446 : 127 : int i;
447 : :
448 : 127 : rel = table_open(PublicationRelRelationId, RowExclusiveLock);
449 : :
450 : : /*
451 : : * Check for duplicates. Note that this does not really prevent
452 : : * duplicates, it's here just to provide nicer error message in common
453 : : * case. The real protection is the unique key on the catalog.
454 : : */
455 [ + + ]: 127 : if (SearchSysCacheExists2(PUBLICATIONRELMAP, ObjectIdGetDatum(relid),
456 : : ObjectIdGetDatum(pubid)))
457 : : {
458 : 3 : table_close(rel, RowExclusiveLock);
459 : :
460 [ + + ]: 3 : if (if_not_exists)
461 : 2 : return InvalidObjectAddress;
462 : :
463 [ + - + - ]: 1 : ereport(ERROR,
464 : : (errcode(ERRCODE_DUPLICATE_OBJECT),
465 : : errmsg("relation \"%s\" is already member of publication \"%s\"",
466 : : RelationGetRelationName(targetrel), pub->name)));
467 : 0 : }
468 : :
469 : 124 : check_publication_add_relation(targetrel);
470 : :
471 : : /* Validate and translate column names into a Bitmapset of attnums. */
472 : 124 : attnums = pub_collist_validate(pri->relation, pri->columns);
473 : :
474 : : /* Form a tuple. */
475 : 124 : memset(values, 0, sizeof(values));
476 : 124 : memset(nulls, false, sizeof(nulls));
477 : :
478 : 124 : pubreloid = GetNewOidWithIndex(rel, PublicationRelObjectIndexId,
479 : : Anum_pg_publication_rel_oid);
480 : 124 : values[Anum_pg_publication_rel_oid - 1] = ObjectIdGetDatum(pubreloid);
481 : 124 : values[Anum_pg_publication_rel_prpubid - 1] =
482 : 124 : ObjectIdGetDatum(pubid);
483 : 124 : values[Anum_pg_publication_rel_prrelid - 1] =
484 : 124 : ObjectIdGetDatum(relid);
485 : :
486 : : /* Add qualifications, if available */
487 [ + + ]: 124 : if (pri->whereClause != NULL)
488 : 46 : values[Anum_pg_publication_rel_prqual - 1] = CStringGetTextDatum(nodeToString(pri->whereClause));
489 : : else
490 : 78 : nulls[Anum_pg_publication_rel_prqual - 1] = true;
491 : :
492 : : /* Add column list, if available */
493 [ + + ]: 124 : if (pri->columns)
494 : 42 : values[Anum_pg_publication_rel_prattrs - 1] = PointerGetDatum(attnumstoint2vector(attnums));
495 : : else
496 : 82 : nulls[Anum_pg_publication_rel_prattrs - 1] = true;
497 : :
498 : 124 : tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
499 : :
500 : : /* Insert tuple into catalog. */
501 : 124 : CatalogTupleInsert(rel, tup);
502 : 124 : heap_freetuple(tup);
503 : :
504 : : /* Register dependencies as needed */
505 : 124 : ObjectAddressSet(myself, PublicationRelRelationId, pubreloid);
506 : :
507 : : /* Add dependency on the publication */
508 : 124 : ObjectAddressSet(referenced, PublicationRelationId, pubid);
509 : 124 : recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
510 : :
511 : : /* Add dependency on the relation */
512 : 124 : ObjectAddressSet(referenced, RelationRelationId, relid);
513 : 124 : recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
514 : :
515 : : /* Add dependency on the objects mentioned in the qualifications */
516 [ + + ]: 124 : if (pri->whereClause)
517 : 46 : recordDependencyOnSingleRelExpr(&myself, pri->whereClause, relid,
518 : : DEPENDENCY_NORMAL, DEPENDENCY_NORMAL,
519 : : false);
520 : :
521 : : /* Add dependency on the columns, if any are listed */
522 : 124 : i = -1;
523 [ + + ]: 195 : while ((i = bms_next_member(attnums, i)) >= 0)
524 : : {
525 : 71 : ObjectAddressSubSet(referenced, RelationRelationId, relid, i);
526 : 71 : recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
527 : : }
528 : :
529 : : /* Close the table. */
530 : 124 : table_close(rel, RowExclusiveLock);
531 : :
532 : : /*
533 : : * Invalidate relcache so that publication info is rebuilt.
534 : : *
535 : : * For the partitioned tables, we must invalidate all partitions contained
536 : : * in the respective partition hierarchies, not just the one explicitly
537 : : * mentioned in the publication. This is required because we implicitly
538 : : * publish the child tables when the parent table is published.
539 : : */
540 : 248 : relids = GetPubPartitionOptionRelations(relids, PUBLICATION_PART_ALL,
541 : 124 : relid);
542 : :
543 : 124 : InvalidatePublicationRels(relids);
544 : :
545 : 124 : return myself;
546 : 126 : }
547 : :
548 : : /*
549 : : * pub_collist_validate
550 : : * Process and validate the 'columns' list and ensure the columns are all
551 : : * valid to use for a publication. Checks for and raises an ERROR for
552 : : * any unknown columns, system columns, duplicate columns, or virtual
553 : : * generated columns.
554 : : *
555 : : * Looks up each column's attnum and returns a 0-based Bitmapset of the
556 : : * corresponding attnums.
557 : : */
558 : : Bitmapset *
559 : 192 : pub_collist_validate(Relation targetrel, List *columns)
560 : : {
561 : 192 : Bitmapset *set = NULL;
562 : 192 : ListCell *lc;
563 : 192 : TupleDesc tupdesc = RelationGetDescr(targetrel);
564 : :
565 [ + + + + : 307 : foreach(lc, columns)
+ + ]
566 : : {
567 : 121 : char *colname = strVal(lfirst(lc));
568 : 121 : AttrNumber attnum = get_attnum(RelationGetRelid(targetrel), colname);
569 : :
570 [ + + ]: 121 : if (attnum == InvalidAttrNumber)
571 [ + - + - ]: 1 : ereport(ERROR,
572 : : errcode(ERRCODE_UNDEFINED_COLUMN),
573 : : errmsg("column \"%s\" of relation \"%s\" does not exist",
574 : : colname, RelationGetRelationName(targetrel)));
575 : :
576 [ + + ]: 120 : if (!AttrNumberIsForUserDefinedAttr(attnum))
577 [ + - + - ]: 2 : ereport(ERROR,
578 : : errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
579 : : errmsg("cannot use system column \"%s\" in publication column list",
580 : : colname));
581 : :
582 [ + + ]: 118 : if (TupleDescAttr(tupdesc, attnum - 1)->attgenerated == ATTRIBUTE_GENERATED_VIRTUAL)
583 [ + - + - ]: 1 : ereport(ERROR,
584 : : errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
585 : : errmsg("cannot use virtual generated column \"%s\" in publication column list",
586 : : colname));
587 : :
588 [ + + ]: 117 : if (bms_is_member(attnum, set))
589 [ + - + - ]: 2 : ereport(ERROR,
590 : : errcode(ERRCODE_DUPLICATE_OBJECT),
591 : : errmsg("duplicate column \"%s\" in publication column list",
592 : : colname));
593 : :
594 : 115 : set = bms_add_member(set, attnum);
595 : 115 : }
596 : :
597 : 372 : return set;
598 : 186 : }
599 : :
600 : : /*
601 : : * Transform a column list (represented by an array Datum) to a bitmapset.
602 : : *
603 : : * If columns isn't NULL, add the column numbers to that set.
604 : : *
605 : : * If mcxt isn't NULL, build the bitmapset in that context.
606 : : */
607 : : Bitmapset *
608 : 56 : pub_collist_to_bitmapset(Bitmapset *columns, Datum pubcols, MemoryContext mcxt)
609 : : {
610 : 56 : Bitmapset *result = columns;
611 : 56 : ArrayType *arr;
612 : 56 : int nelems;
613 : 56 : int16 *elems;
614 : 56 : MemoryContext oldcxt = NULL;
615 : :
616 : 56 : arr = DatumGetArrayTypeP(pubcols);
617 : 56 : nelems = ARR_DIMS(arr)[0];
618 [ - + ]: 56 : elems = (int16 *) ARR_DATA_PTR(arr);
619 : :
620 : : /* If a memory context was specified, switch to it. */
621 [ + - ]: 56 : if (mcxt)
622 : 0 : oldcxt = MemoryContextSwitchTo(mcxt);
623 : :
624 [ + + ]: 151 : for (int i = 0; i < nelems; i++)
625 : 95 : result = bms_add_member(result, elems[i]);
626 : :
627 [ + - ]: 56 : if (mcxt)
628 : 0 : MemoryContextSwitchTo(oldcxt);
629 : :
630 : 112 : return result;
631 : 56 : }
632 : :
633 : : /*
634 : : * Returns a bitmap representing the columns of the specified table.
635 : : *
636 : : * Generated columns are included if include_gencols_type is
637 : : * PUBLISH_GENCOLS_STORED.
638 : : */
639 : : Bitmapset *
640 : 0 : pub_form_cols_map(Relation relation, PublishGencolsType include_gencols_type)
641 : : {
642 : 0 : Bitmapset *result = NULL;
643 : 0 : TupleDesc desc = RelationGetDescr(relation);
644 : :
645 [ # # ]: 0 : for (int i = 0; i < desc->natts; i++)
646 : : {
647 : 0 : Form_pg_attribute att = TupleDescAttr(desc, i);
648 : :
649 [ # # ]: 0 : if (att->attisdropped)
650 : 0 : continue;
651 : :
652 [ # # ]: 0 : if (att->attgenerated)
653 : : {
654 : : /* We only support replication of STORED generated cols. */
655 [ # # ]: 0 : if (att->attgenerated != ATTRIBUTE_GENERATED_STORED)
656 : 0 : continue;
657 : :
658 : : /* User hasn't requested to replicate STORED generated cols. */
659 [ # # ]: 0 : if (include_gencols_type != PUBLISH_GENCOLS_STORED)
660 : 0 : continue;
661 : 0 : }
662 : :
663 : 0 : result = bms_add_member(result, att->attnum);
664 [ # # # ]: 0 : }
665 : :
666 : 0 : return result;
667 : 0 : }
668 : :
669 : : /*
670 : : * Insert new publication / schema mapping.
671 : : */
672 : : ObjectAddress
673 : 39 : publication_add_schema(Oid pubid, Oid schemaid, bool if_not_exists)
674 : : {
675 : 39 : Relation rel;
676 : 39 : HeapTuple tup;
677 : 39 : Datum values[Natts_pg_publication_namespace];
678 : 39 : bool nulls[Natts_pg_publication_namespace];
679 : 39 : Oid psschid;
680 : 39 : Publication *pub = GetPublication(pubid);
681 : 39 : List *schemaRels = NIL;
682 : 39 : ObjectAddress myself,
683 : : referenced;
684 : :
685 : 39 : rel = table_open(PublicationNamespaceRelationId, RowExclusiveLock);
686 : :
687 : : /*
688 : : * Check for duplicates. Note that this does not really prevent
689 : : * duplicates, it's here just to provide nicer error message in common
690 : : * case. The real protection is the unique key on the catalog.
691 : : */
692 [ + + ]: 39 : if (SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
693 : : ObjectIdGetDatum(schemaid),
694 : : ObjectIdGetDatum(pubid)))
695 : : {
696 : 3 : table_close(rel, RowExclusiveLock);
697 : :
698 [ + + ]: 3 : if (if_not_exists)
699 : 2 : return InvalidObjectAddress;
700 : :
701 [ + - + - ]: 1 : ereport(ERROR,
702 : : (errcode(ERRCODE_DUPLICATE_OBJECT),
703 : : errmsg("schema \"%s\" is already member of publication \"%s\"",
704 : : get_namespace_name(schemaid), pub->name)));
705 : 0 : }
706 : :
707 : 36 : check_publication_add_schema(schemaid);
708 : :
709 : : /* Form a tuple */
710 : 36 : memset(values, 0, sizeof(values));
711 : 36 : memset(nulls, false, sizeof(nulls));
712 : :
713 : 36 : psschid = GetNewOidWithIndex(rel, PublicationNamespaceObjectIndexId,
714 : : Anum_pg_publication_namespace_oid);
715 : 36 : values[Anum_pg_publication_namespace_oid - 1] = ObjectIdGetDatum(psschid);
716 : 36 : values[Anum_pg_publication_namespace_pnpubid - 1] =
717 : 36 : ObjectIdGetDatum(pubid);
718 : 36 : values[Anum_pg_publication_namespace_pnnspid - 1] =
719 : 36 : ObjectIdGetDatum(schemaid);
720 : :
721 : 36 : tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
722 : :
723 : : /* Insert tuple into catalog */
724 : 36 : CatalogTupleInsert(rel, tup);
725 : 36 : heap_freetuple(tup);
726 : :
727 : 36 : ObjectAddressSet(myself, PublicationNamespaceRelationId, psschid);
728 : :
729 : : /* Add dependency on the publication */
730 : 36 : ObjectAddressSet(referenced, PublicationRelationId, pubid);
731 : 36 : recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
732 : :
733 : : /* Add dependency on the schema */
734 : 36 : ObjectAddressSet(referenced, NamespaceRelationId, schemaid);
735 : 36 : recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
736 : :
737 : : /* Close the table */
738 : 36 : table_close(rel, RowExclusiveLock);
739 : :
740 : : /*
741 : : * Invalidate relcache so that publication info is rebuilt. See
742 : : * publication_add_relation for why we need to consider all the
743 : : * partitions.
744 : : */
745 : 36 : schemaRels = GetSchemaPublicationRelations(schemaid,
746 : : PUBLICATION_PART_ALL);
747 : 36 : InvalidatePublicationRels(schemaRels);
748 : :
749 : 36 : return myself;
750 : 38 : }
751 : :
752 : : /* Gets list of publication oids for a relation */
753 : : List *
754 : 1327 : GetRelationPublications(Oid relid)
755 : : {
756 : 1327 : List *result = NIL;
757 : 1327 : CatCList *pubrellist;
758 : 1327 : int i;
759 : :
760 : : /* Find all publications associated with the relation. */
761 : 1327 : pubrellist = SearchSysCacheList1(PUBLICATIONRELMAP,
762 : : ObjectIdGetDatum(relid));
763 [ + + ]: 1425 : for (i = 0; i < pubrellist->n_members; i++)
764 : : {
765 : 98 : HeapTuple tup = &pubrellist->members[i]->tuple;
766 : 98 : Oid pubid = ((Form_pg_publication_rel) GETSTRUCT(tup))->prpubid;
767 : :
768 : 98 : result = lappend_oid(result, pubid);
769 : 98 : }
770 : :
771 : 1327 : ReleaseSysCacheList(pubrellist);
772 : :
773 : 2654 : return result;
774 : 1327 : }
775 : :
776 : : /*
777 : : * Gets list of relation oids for a publication.
778 : : *
779 : : * This should only be used FOR TABLE publications, the FOR ALL TABLES/SEQUENCES
780 : : * should use GetAllPublicationRelations().
781 : : */
782 : : List *
783 : 103 : GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt)
784 : : {
785 : 103 : List *result;
786 : 103 : Relation pubrelsrel;
787 : 103 : ScanKeyData scankey;
788 : 103 : SysScanDesc scan;
789 : 103 : HeapTuple tup;
790 : :
791 : : /* Find all relations associated with the publication. */
792 : 103 : pubrelsrel = table_open(PublicationRelRelationId, AccessShareLock);
793 : :
794 : 103 : ScanKeyInit(&scankey,
795 : : Anum_pg_publication_rel_prpubid,
796 : : BTEqualStrategyNumber, F_OIDEQ,
797 : 103 : ObjectIdGetDatum(pubid));
798 : :
799 : 103 : scan = systable_beginscan(pubrelsrel, PublicationRelPrpubidIndexId,
800 : : true, NULL, 1, &scankey);
801 : :
802 : 103 : result = NIL;
803 [ + + ]: 195 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
804 : : {
805 : 92 : Form_pg_publication_rel pubrel;
806 : :
807 : 92 : pubrel = (Form_pg_publication_rel) GETSTRUCT(tup);
808 : 184 : result = GetPubPartitionOptionRelations(result, pub_partopt,
809 : 92 : pubrel->prrelid);
810 : 92 : }
811 : :
812 : 103 : systable_endscan(scan);
813 : 103 : table_close(pubrelsrel, AccessShareLock);
814 : :
815 : : /* Now sort and de-duplicate the result list */
816 : 103 : list_sort(result, list_oid_cmp);
817 : 103 : list_deduplicate_oid(result);
818 : :
819 : 206 : return result;
820 : 103 : }
821 : :
822 : : /*
823 : : * Gets list of publication oids for publications marked as FOR ALL TABLES.
824 : : */
825 : : List *
826 : 856 : GetAllTablesPublications(void)
827 : : {
828 : 856 : List *result;
829 : 856 : Relation rel;
830 : 856 : ScanKeyData scankey;
831 : 856 : SysScanDesc scan;
832 : 856 : HeapTuple tup;
833 : :
834 : : /* Find all publications that are marked as for all tables. */
835 : 856 : rel = table_open(PublicationRelationId, AccessShareLock);
836 : :
837 : 856 : ScanKeyInit(&scankey,
838 : : Anum_pg_publication_puballtables,
839 : : BTEqualStrategyNumber, F_BOOLEQ,
840 : 856 : BoolGetDatum(true));
841 : :
842 : 856 : scan = systable_beginscan(rel, InvalidOid, false,
843 : : NULL, 1, &scankey);
844 : :
845 : 856 : result = NIL;
846 [ + + ]: 861 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
847 : : {
848 : 5 : Oid oid = ((Form_pg_publication) GETSTRUCT(tup))->oid;
849 : :
850 : 5 : result = lappend_oid(result, oid);
851 : 5 : }
852 : :
853 : 856 : systable_endscan(scan);
854 : 856 : table_close(rel, AccessShareLock);
855 : :
856 : 1712 : return result;
857 : 856 : }
858 : :
859 : : /*
860 : : * Gets list of all relations published by FOR ALL TABLES/SEQUENCES
861 : : * publication(s).
862 : : *
863 : : * If the publication publishes partition changes via their respective root
864 : : * partitioned tables, we must exclude partitions in favor of including the
865 : : * root partitioned tables. This is not applicable to FOR ALL SEQUENCES
866 : : * publication.
867 : : */
868 : : List *
869 : 0 : GetAllPublicationRelations(char relkind, bool pubviaroot)
870 : : {
871 : 0 : Relation classRel;
872 : 0 : ScanKeyData key[1];
873 : 0 : TableScanDesc scan;
874 : 0 : HeapTuple tuple;
875 : 0 : List *result = NIL;
876 : :
877 [ # # # # ]: 0 : Assert(!(relkind == RELKIND_SEQUENCE && pubviaroot));
878 : :
879 : 0 : classRel = table_open(RelationRelationId, AccessShareLock);
880 : :
881 : 0 : ScanKeyInit(&key[0],
882 : : Anum_pg_class_relkind,
883 : : BTEqualStrategyNumber, F_CHAREQ,
884 : 0 : CharGetDatum(relkind));
885 : :
886 : 0 : scan = table_beginscan_catalog(classRel, 1, key);
887 : :
888 [ # # ]: 0 : while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
889 : : {
890 : 0 : Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
891 : 0 : Oid relid = relForm->oid;
892 : :
893 [ # # # # ]: 0 : if (is_publishable_class(relid, relForm) &&
894 [ # # ]: 0 : !(relForm->relispartition && pubviaroot))
895 : 0 : result = lappend_oid(result, relid);
896 : 0 : }
897 : :
898 : 0 : table_endscan(scan);
899 : :
900 [ # # ]: 0 : if (pubviaroot)
901 : : {
902 : 0 : ScanKeyInit(&key[0],
903 : : Anum_pg_class_relkind,
904 : : BTEqualStrategyNumber, F_CHAREQ,
905 : 0 : CharGetDatum(RELKIND_PARTITIONED_TABLE));
906 : :
907 : 0 : scan = table_beginscan_catalog(classRel, 1, key);
908 : :
909 [ # # ]: 0 : while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
910 : : {
911 : 0 : Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
912 : 0 : Oid relid = relForm->oid;
913 : :
914 [ # # # # ]: 0 : if (is_publishable_class(relid, relForm) &&
915 : 0 : !relForm->relispartition)
916 : 0 : result = lappend_oid(result, relid);
917 : 0 : }
918 : :
919 : 0 : table_endscan(scan);
920 : 0 : }
921 : :
922 : 0 : table_close(classRel, AccessShareLock);
923 : 0 : return result;
924 : 0 : }
925 : :
926 : : /*
927 : : * Gets the list of schema oids for a publication.
928 : : *
929 : : * This should only be used FOR TABLES IN SCHEMA publications.
930 : : */
931 : : List *
932 : 90 : GetPublicationSchemas(Oid pubid)
933 : : {
934 : 90 : List *result = NIL;
935 : 90 : Relation pubschsrel;
936 : 90 : ScanKeyData scankey;
937 : 90 : SysScanDesc scan;
938 : 90 : HeapTuple tup;
939 : :
940 : : /* Find all schemas associated with the publication */
941 : 90 : pubschsrel = table_open(PublicationNamespaceRelationId, AccessShareLock);
942 : :
943 : 90 : ScanKeyInit(&scankey,
944 : : Anum_pg_publication_namespace_pnpubid,
945 : : BTEqualStrategyNumber, F_OIDEQ,
946 : 90 : ObjectIdGetDatum(pubid));
947 : :
948 : 90 : scan = systable_beginscan(pubschsrel,
949 : : PublicationNamespacePnnspidPnpubidIndexId,
950 : : true, NULL, 1, &scankey);
951 [ + + ]: 98 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
952 : : {
953 : 8 : Form_pg_publication_namespace pubsch;
954 : :
955 : 8 : pubsch = (Form_pg_publication_namespace) GETSTRUCT(tup);
956 : :
957 : 8 : result = lappend_oid(result, pubsch->pnnspid);
958 : 8 : }
959 : :
960 : 90 : systable_endscan(scan);
961 : 90 : table_close(pubschsrel, AccessShareLock);
962 : :
963 : 180 : return result;
964 : 90 : }
965 : :
966 : : /*
967 : : * Gets the list of publication oids associated with a specified schema.
968 : : */
969 : : List *
970 : 1272 : GetSchemaPublications(Oid schemaid)
971 : : {
972 : 1272 : List *result = NIL;
973 : 1272 : CatCList *pubschlist;
974 : 1272 : int i;
975 : :
976 : : /* Find all publications associated with the schema */
977 : 1272 : pubschlist = SearchSysCacheList1(PUBLICATIONNAMESPACEMAP,
978 : : ObjectIdGetDatum(schemaid));
979 [ + + ]: 1277 : for (i = 0; i < pubschlist->n_members; i++)
980 : : {
981 : 5 : HeapTuple tup = &pubschlist->members[i]->tuple;
982 : 5 : Oid pubid = ((Form_pg_publication_namespace) GETSTRUCT(tup))->pnpubid;
983 : :
984 : 5 : result = lappend_oid(result, pubid);
985 : 5 : }
986 : :
987 : 1272 : ReleaseSysCacheList(pubschlist);
988 : :
989 : 2544 : return result;
990 : 1272 : }
991 : :
992 : : /*
993 : : * Get the list of publishable relation oids for a specified schema.
994 : : */
995 : : List *
996 : 70 : GetSchemaPublicationRelations(Oid schemaid, PublicationPartOpt pub_partopt)
997 : : {
998 : 70 : Relation classRel;
999 : 70 : ScanKeyData key[1];
1000 : 70 : TableScanDesc scan;
1001 : 70 : HeapTuple tuple;
1002 : 70 : List *result = NIL;
1003 : :
1004 [ + - ]: 70 : Assert(OidIsValid(schemaid));
1005 : :
1006 : 70 : classRel = table_open(RelationRelationId, AccessShareLock);
1007 : :
1008 : 140 : ScanKeyInit(&key[0],
1009 : : Anum_pg_class_relnamespace,
1010 : : BTEqualStrategyNumber, F_OIDEQ,
1011 : 70 : ObjectIdGetDatum(schemaid));
1012 : :
1013 : : /* get all the relations present in the specified schema */
1014 : 70 : scan = table_beginscan_catalog(classRel, 1, key);
1015 [ + + ]: 4249 : while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
1016 : : {
1017 : 4179 : Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
1018 : 4179 : Oid relid = relForm->oid;
1019 : 4179 : char relkind;
1020 : :
1021 [ + + ]: 4179 : if (!is_publishable_class(relid, relForm))
1022 : 1435 : continue;
1023 : :
1024 : 2744 : relkind = get_rel_relkind(relid);
1025 [ + + ]: 2744 : if (relkind == RELKIND_RELATION)
1026 : 2360 : result = lappend_oid(result, relid);
1027 [ + + ]: 384 : else if (relkind == RELKIND_PARTITIONED_TABLE)
1028 : : {
1029 : 120 : List *partitionrels = NIL;
1030 : :
1031 : : /*
1032 : : * It is quite possible that some of the partitions are in a
1033 : : * different schema than the parent table, so we need to get such
1034 : : * partitions separately.
1035 : : */
1036 : 240 : partitionrels = GetPubPartitionOptionRelations(partitionrels,
1037 : 120 : pub_partopt,
1038 : 120 : relForm->oid);
1039 : 120 : result = list_concat_unique_oid(result, partitionrels);
1040 : 120 : }
1041 [ - + + ]: 4179 : }
1042 : :
1043 : 70 : table_endscan(scan);
1044 : 70 : table_close(classRel, AccessShareLock);
1045 : 140 : return result;
1046 : 70 : }
1047 : :
1048 : : /*
1049 : : * Gets the list of all relations published by FOR TABLES IN SCHEMA
1050 : : * publication.
1051 : : */
1052 : : List *
1053 : 24 : GetAllSchemaPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt)
1054 : : {
1055 : 24 : List *result = NIL;
1056 : 24 : List *pubschemalist = GetPublicationSchemas(pubid);
1057 : 24 : ListCell *cell;
1058 : :
1059 [ + + + + : 27 : foreach(cell, pubschemalist)
+ + ]
1060 : : {
1061 : 3 : Oid schemaid = lfirst_oid(cell);
1062 : 3 : List *schemaRels = NIL;
1063 : :
1064 : 3 : schemaRels = GetSchemaPublicationRelations(schemaid, pub_partopt);
1065 : 3 : result = list_concat(result, schemaRels);
1066 : 3 : }
1067 : :
1068 : 48 : return result;
1069 : 24 : }
1070 : :
1071 : : /*
1072 : : * Get publication using oid
1073 : : *
1074 : : * The Publication struct and its data are palloc'ed here.
1075 : : */
1076 : : Publication *
1077 : 268 : GetPublication(Oid pubid)
1078 : : {
1079 : 268 : HeapTuple tup;
1080 : 268 : Publication *pub;
1081 : 268 : Form_pg_publication pubform;
1082 : :
1083 : 268 : tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
1084 [ + - ]: 268 : if (!HeapTupleIsValid(tup))
1085 [ # # # # ]: 0 : elog(ERROR, "cache lookup failed for publication %u", pubid);
1086 : :
1087 : 268 : pubform = (Form_pg_publication) GETSTRUCT(tup);
1088 : :
1089 : 268 : pub = palloc_object(Publication);
1090 : 268 : pub->oid = pubid;
1091 : 268 : pub->name = pstrdup(NameStr(pubform->pubname));
1092 : 268 : pub->alltables = pubform->puballtables;
1093 : 268 : pub->allsequences = pubform->puballsequences;
1094 : 268 : pub->pubactions.pubinsert = pubform->pubinsert;
1095 : 268 : pub->pubactions.pubupdate = pubform->pubupdate;
1096 : 268 : pub->pubactions.pubdelete = pubform->pubdelete;
1097 : 268 : pub->pubactions.pubtruncate = pubform->pubtruncate;
1098 : 268 : pub->pubviaroot = pubform->pubviaroot;
1099 : 268 : pub->pubgencols_type = pubform->pubgencols;
1100 : :
1101 : 268 : ReleaseSysCache(tup);
1102 : :
1103 : 536 : return pub;
1104 : 268 : }
1105 : :
1106 : : /*
1107 : : * Get Publication using name.
1108 : : */
1109 : : Publication *
1110 : 11 : GetPublicationByName(const char *pubname, bool missing_ok)
1111 : : {
1112 : 11 : Oid oid;
1113 : :
1114 : 11 : oid = get_publication_oid(pubname, missing_ok);
1115 : :
1116 [ + - ]: 11 : return OidIsValid(oid) ? GetPublication(oid) : NULL;
1117 : 11 : }
1118 : :
1119 : : /*
1120 : : * Get information of the tables in the given publication array.
1121 : : *
1122 : : * Returns pubid, relid, column list, row filter for each table.
1123 : : */
1124 : : Datum
1125 : 14 : pg_get_publication_tables(PG_FUNCTION_ARGS)
1126 : : {
1127 : : #define NUM_PUBLICATION_TABLES_ELEM 4
1128 : 14 : FuncCallContext *funcctx;
1129 : 14 : List *table_infos = NIL;
1130 : :
1131 : : /* stuff done only on the first call of the function */
1132 [ + + ]: 14 : if (SRF_IS_FIRSTCALL())
1133 : : {
1134 : 7 : TupleDesc tupdesc;
1135 : 7 : MemoryContext oldcontext;
1136 : 7 : ArrayType *arr;
1137 : 7 : Datum *elems;
1138 : 7 : int nelems,
1139 : : i;
1140 : 7 : bool viaroot = false;
1141 : :
1142 : : /* create a function context for cross-call persistence */
1143 : 7 : funcctx = SRF_FIRSTCALL_INIT();
1144 : :
1145 : : /* switch to memory context appropriate for multiple function calls */
1146 : 7 : oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
1147 : :
1148 : : /*
1149 : : * Deconstruct the parameter into elements where each element is a
1150 : : * publication name.
1151 : : */
1152 : 7 : arr = PG_GETARG_ARRAYTYPE_P(0);
1153 : 7 : deconstruct_array_builtin(arr, TEXTOID, &elems, NULL, &nelems);
1154 : :
1155 : : /* Get Oids of tables from each publication. */
1156 [ + + ]: 14 : for (i = 0; i < nelems; i++)
1157 : : {
1158 : 7 : Publication *pub_elem;
1159 : 7 : List *pub_elem_tables = NIL;
1160 : 7 : ListCell *lc;
1161 : :
1162 : 7 : pub_elem = GetPublicationByName(TextDatumGetCString(elems[i]), false);
1163 : :
1164 : : /*
1165 : : * Publications support partitioned tables. If
1166 : : * publish_via_partition_root is false, all changes are replicated
1167 : : * using leaf partition identity and schema, so we only need
1168 : : * those. Otherwise, get the partitioned table itself.
1169 : : */
1170 [ - + ]: 7 : if (pub_elem->alltables)
1171 : 0 : pub_elem_tables = GetAllPublicationRelations(RELKIND_RELATION,
1172 : 0 : pub_elem->pubviaroot);
1173 : : else
1174 : : {
1175 : 7 : List *relids,
1176 : : *schemarelids;
1177 : :
1178 : 14 : relids = GetPublicationRelations(pub_elem->oid,
1179 : 7 : pub_elem->pubviaroot ?
1180 : : PUBLICATION_PART_ROOT :
1181 : : PUBLICATION_PART_LEAF);
1182 : 14 : schemarelids = GetAllSchemaPublicationRelations(pub_elem->oid,
1183 : 7 : pub_elem->pubviaroot ?
1184 : : PUBLICATION_PART_ROOT :
1185 : : PUBLICATION_PART_LEAF);
1186 : 7 : pub_elem_tables = list_concat_unique_oid(relids, schemarelids);
1187 : 7 : }
1188 : :
1189 : : /*
1190 : : * Record the published table and the corresponding publication so
1191 : : * that we can get row filters and column lists later.
1192 : : *
1193 : : * When a table is published by multiple publications, to obtain
1194 : : * all row filters and column lists, the structure related to this
1195 : : * table will be recorded multiple times.
1196 : : */
1197 [ + - + + : 18 : foreach(lc, pub_elem_tables)
+ + ]
1198 : : {
1199 : 11 : published_rel *table_info = palloc_object(published_rel);
1200 : :
1201 : 11 : table_info->relid = lfirst_oid(lc);
1202 : 11 : table_info->pubid = pub_elem->oid;
1203 : 11 : table_infos = lappend(table_infos, table_info);
1204 : 11 : }
1205 : :
1206 : : /* At least one publication is using publish_via_partition_root. */
1207 [ + + ]: 7 : if (pub_elem->pubviaroot)
1208 : 4 : viaroot = true;
1209 : 7 : }
1210 : :
1211 : : /*
1212 : : * If the publication publishes partition changes via their respective
1213 : : * root partitioned tables, we must exclude partitions in favor of
1214 : : * including the root partitioned tables. Otherwise, the function
1215 : : * could return both the child and parent tables which could cause
1216 : : * data of the child table to be double-published on the subscriber
1217 : : * side.
1218 : : */
1219 [ + + ]: 7 : if (viaroot)
1220 : 4 : filter_partitions(table_infos);
1221 : :
1222 : : /* Construct a tuple descriptor for the result rows. */
1223 : 7 : tupdesc = CreateTemplateTupleDesc(NUM_PUBLICATION_TABLES_ELEM);
1224 : 7 : TupleDescInitEntry(tupdesc, (AttrNumber) 1, "pubid",
1225 : : OIDOID, -1, 0);
1226 : 7 : TupleDescInitEntry(tupdesc, (AttrNumber) 2, "relid",
1227 : : OIDOID, -1, 0);
1228 : 7 : TupleDescInitEntry(tupdesc, (AttrNumber) 3, "attrs",
1229 : : INT2VECTOROID, -1, 0);
1230 : 7 : TupleDescInitEntry(tupdesc, (AttrNumber) 4, "qual",
1231 : : PG_NODE_TREEOID, -1, 0);
1232 : :
1233 : 7 : funcctx->tuple_desc = BlessTupleDesc(tupdesc);
1234 : 7 : funcctx->user_fctx = table_infos;
1235 : :
1236 : 7 : MemoryContextSwitchTo(oldcontext);
1237 : 7 : }
1238 : :
1239 : : /* stuff done on every call of the function */
1240 : 14 : funcctx = SRF_PERCALL_SETUP();
1241 : 14 : table_infos = (List *) funcctx->user_fctx;
1242 : :
1243 [ + + ]: 14 : if (funcctx->call_cntr < list_length(table_infos))
1244 : : {
1245 : 7 : HeapTuple pubtuple = NULL;
1246 : 7 : HeapTuple rettuple;
1247 : 7 : Publication *pub;
1248 : 7 : published_rel *table_info = (published_rel *) list_nth(table_infos, funcctx->call_cntr);
1249 : 7 : Oid relid = table_info->relid;
1250 : 7 : Oid schemaid = get_rel_namespace(relid);
1251 : 7 : Datum values[NUM_PUBLICATION_TABLES_ELEM] = {0};
1252 : 7 : bool nulls[NUM_PUBLICATION_TABLES_ELEM] = {0};
1253 : :
1254 : : /*
1255 : : * Form tuple with appropriate data.
1256 : : */
1257 : :
1258 : 7 : pub = GetPublication(table_info->pubid);
1259 : :
1260 : 7 : values[0] = ObjectIdGetDatum(pub->oid);
1261 : 7 : values[1] = ObjectIdGetDatum(relid);
1262 : :
1263 : : /*
1264 : : * We don't consider row filters or column lists for FOR ALL TABLES or
1265 : : * FOR TABLES IN SCHEMA publications.
1266 : : */
1267 [ + - + + ]: 7 : if (!pub->alltables &&
1268 : 7 : !SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
1269 : : ObjectIdGetDatum(schemaid),
1270 : : ObjectIdGetDatum(pub->oid)))
1271 : 4 : pubtuple = SearchSysCacheCopy2(PUBLICATIONRELMAP,
1272 : : ObjectIdGetDatum(relid),
1273 : : ObjectIdGetDatum(pub->oid));
1274 : :
1275 [ + + ]: 7 : if (HeapTupleIsValid(pubtuple))
1276 : : {
1277 : : /* Lookup the column list attribute. */
1278 : 8 : values[2] = SysCacheGetAttr(PUBLICATIONRELMAP, pubtuple,
1279 : : Anum_pg_publication_rel_prattrs,
1280 : 4 : &(nulls[2]));
1281 : :
1282 : : /* Null indicates no filter. */
1283 : 8 : values[3] = SysCacheGetAttr(PUBLICATIONRELMAP, pubtuple,
1284 : : Anum_pg_publication_rel_prqual,
1285 : 4 : &(nulls[3]));
1286 : 4 : }
1287 : : else
1288 : : {
1289 : 3 : nulls[2] = true;
1290 : 3 : nulls[3] = true;
1291 : : }
1292 : :
1293 : : /* Show all columns when the column list is not specified. */
1294 [ - + ]: 7 : if (nulls[2])
1295 : : {
1296 : 7 : Relation rel = table_open(relid, AccessShareLock);
1297 : 7 : int nattnums = 0;
1298 : 7 : int16 *attnums;
1299 : 7 : TupleDesc desc = RelationGetDescr(rel);
1300 : 7 : int i;
1301 : :
1302 : 7 : attnums = palloc_array(int16, desc->natts);
1303 : :
1304 [ + + ]: 14 : for (i = 0; i < desc->natts; i++)
1305 : : {
1306 : 7 : Form_pg_attribute att = TupleDescAttr(desc, i);
1307 : :
1308 [ - + ]: 7 : if (att->attisdropped)
1309 : 0 : continue;
1310 : :
1311 [ + - ]: 7 : if (att->attgenerated)
1312 : : {
1313 : : /* We only support replication of STORED generated cols. */
1314 [ # # ]: 0 : if (att->attgenerated != ATTRIBUTE_GENERATED_STORED)
1315 : 0 : continue;
1316 : :
1317 : : /*
1318 : : * User hasn't requested to replicate STORED generated
1319 : : * cols.
1320 : : */
1321 [ # # ]: 0 : if (pub->pubgencols_type != PUBLISH_GENCOLS_STORED)
1322 : 0 : continue;
1323 : 0 : }
1324 : :
1325 : 7 : attnums[nattnums++] = att->attnum;
1326 [ - + ]: 7 : }
1327 : :
1328 [ - + ]: 7 : if (nattnums > 0)
1329 : : {
1330 : 7 : values[2] = PointerGetDatum(buildint2vector(attnums, nattnums));
1331 : 7 : nulls[2] = false;
1332 : 7 : }
1333 : :
1334 : 7 : table_close(rel, AccessShareLock);
1335 : 7 : }
1336 : :
1337 : 7 : rettuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
1338 : :
1339 : 7 : SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(rettuple));
1340 [ + - ]: 7 : }
1341 : :
1342 [ + - ]: 7 : SRF_RETURN_DONE(funcctx);
1343 : 14 : }
1344 : :
1345 : : /*
1346 : : * Returns Oids of sequences in a publication.
1347 : : */
1348 : : Datum
1349 : 0 : pg_get_publication_sequences(PG_FUNCTION_ARGS)
1350 : : {
1351 : 0 : FuncCallContext *funcctx;
1352 : 0 : List *sequences = NIL;
1353 : :
1354 : : /* stuff done only on the first call of the function */
1355 [ # # ]: 0 : if (SRF_IS_FIRSTCALL())
1356 : : {
1357 : 0 : char *pubname = text_to_cstring(PG_GETARG_TEXT_PP(0));
1358 : 0 : Publication *publication;
1359 : 0 : MemoryContext oldcontext;
1360 : :
1361 : : /* create a function context for cross-call persistence */
1362 : 0 : funcctx = SRF_FIRSTCALL_INIT();
1363 : :
1364 : : /* switch to memory context appropriate for multiple function calls */
1365 : 0 : oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
1366 : :
1367 : 0 : publication = GetPublicationByName(pubname, false);
1368 : :
1369 [ # # ]: 0 : if (publication->allsequences)
1370 : 0 : sequences = GetAllPublicationRelations(RELKIND_SEQUENCE, false);
1371 : :
1372 : 0 : funcctx->user_fctx = sequences;
1373 : :
1374 : 0 : MemoryContextSwitchTo(oldcontext);
1375 : 0 : }
1376 : :
1377 : : /* stuff done on every call of the function */
1378 : 0 : funcctx = SRF_PERCALL_SETUP();
1379 : 0 : sequences = (List *) funcctx->user_fctx;
1380 : :
1381 [ # # ]: 0 : if (funcctx->call_cntr < list_length(sequences))
1382 : : {
1383 : 0 : Oid relid = list_nth_oid(sequences, funcctx->call_cntr);
1384 : :
1385 : 0 : SRF_RETURN_NEXT(funcctx, ObjectIdGetDatum(relid));
1386 [ # # ]: 0 : }
1387 : :
1388 [ # # ]: 0 : SRF_RETURN_DONE(funcctx);
1389 [ # # ]: 0 : }
|
