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