TLA Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * publicationcmds.c
4 : * publication 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/commands/publicationcmds.c
11 : *
12 : *-------------------------------------------------------------------------
13 : */
14 :
15 : #include "postgres.h"
16 :
17 : #include "access/genam.h"
18 : #include "access/htup_details.h"
19 : #include "access/table.h"
20 : #include "access/xact.h"
21 : #include "catalog/catalog.h"
22 : #include "catalog/indexing.h"
23 : #include "catalog/namespace.h"
24 : #include "catalog/objectaccess.h"
25 : #include "catalog/objectaddress.h"
26 : #include "catalog/partition.h"
27 : #include "catalog/pg_database.h"
28 : #include "catalog/pg_inherits.h"
29 : #include "catalog/pg_namespace.h"
30 : #include "catalog/pg_proc.h"
31 : #include "catalog/pg_publication.h"
32 : #include "catalog/pg_publication_namespace.h"
33 : #include "catalog/pg_publication_rel.h"
34 : #include "catalog/pg_type.h"
35 : #include "commands/dbcommands.h"
36 : #include "commands/defrem.h"
37 : #include "commands/event_trigger.h"
38 : #include "commands/publicationcmds.h"
39 : #include "funcapi.h"
40 : #include "miscadmin.h"
41 : #include "nodes/nodeFuncs.h"
42 : #include "parser/parse_clause.h"
43 : #include "parser/parse_collate.h"
44 : #include "parser/parse_relation.h"
45 : #include "storage/lmgr.h"
46 : #include "utils/acl.h"
47 : #include "utils/array.h"
48 : #include "utils/builtins.h"
49 : #include "utils/catcache.h"
50 : #include "utils/fmgroids.h"
51 : #include "utils/inval.h"
52 : #include "utils/lsyscache.h"
53 : #include "utils/rel.h"
54 : #include "utils/syscache.h"
55 : #include "utils/varlena.h"
56 :
57 :
58 : /*
59 : * Information used to validate the columns in the row filter expression. See
60 : * contain_invalid_rfcolumn_walker for details.
61 : */
62 : typedef struct rf_context
63 : {
64 : Bitmapset *bms_replident; /* bitset of replica identity columns */
65 : bool pubviaroot; /* true if we are validating the parent
66 : * relation's row filter */
67 : Oid relid; /* relid of the relation */
68 : Oid parentid; /* relid of the parent relation */
69 : } rf_context;
70 :
71 : static List *OpenTableList(List *tables);
72 : static void CloseTableList(List *rels);
73 : static void LockSchemaList(List *schemalist);
74 : static void PublicationAddTables(Oid pubid, List *rels, bool if_not_exists,
75 : AlterPublicationStmt *stmt);
76 : static void PublicationDropTables(Oid pubid, List *rels, bool missing_ok);
77 : static void PublicationAddSchemas(Oid pubid, List *schemas, bool if_not_exists,
78 : AlterPublicationStmt *stmt);
79 : static void PublicationDropSchemas(Oid pubid, List *schemas, bool missing_ok);
80 :
81 :
82 : static void
83 GIC 374 : parse_publication_options(ParseState *pstate,
84 ECB : List *options,
85 : bool *publish_given,
86 : PublicationActions *pubactions,
87 : bool *publish_via_partition_root_given,
88 : bool *publish_via_partition_root)
89 : {
90 : ListCell *lc;
91 :
92 GIC 374 : *publish_given = false;
93 CBC 374 : *publish_via_partition_root_given = false;
94 ECB :
95 : /* defaults */
96 GIC 374 : pubactions->pubinsert = true;
97 CBC 374 : pubactions->pubupdate = true;
98 374 : pubactions->pubdelete = true;
99 374 : pubactions->pubtruncate = true;
100 374 : *publish_via_partition_root = false;
101 ECB :
102 : /* Parse options */
103 GIC 509 : foreach(lc, options)
104 ECB : {
105 GIC 144 : DefElem *defel = (DefElem *) lfirst(lc);
106 ECB :
107 GIC 144 : if (strcmp(defel->defname, "publish") == 0)
108 ECB : {
109 : char *publish;
110 : List *publish_list;
111 : ListCell *lc2;
112 :
113 GIC 60 : if (*publish_given)
114 LBC 0 : errorConflictingDefElem(defel, pstate);
115 EUB :
116 : /*
117 : * If publish option was given only the explicitly listed actions
118 : * should be published.
119 : */
120 GIC 60 : pubactions->pubinsert = false;
121 CBC 60 : pubactions->pubupdate = false;
122 60 : pubactions->pubdelete = false;
123 60 : pubactions->pubtruncate = false;
124 ECB :
125 GIC 60 : *publish_given = true;
126 CBC 60 : publish = defGetString(defel);
127 ECB :
128 GIC 60 : if (!SplitIdentifierString(publish, ',', &publish_list))
129 LBC 0 : ereport(ERROR,
130 EUB : (errcode(ERRCODE_SYNTAX_ERROR),
131 : errmsg("invalid list syntax in parameter \"%s\"",
132 : "publish")));
133 :
134 : /* Process the option list. */
135 GNC 130 : foreach(lc2, publish_list)
136 ECB : {
137 GNC 73 : char *publish_opt = (char *) lfirst(lc2);
138 ECB :
139 GIC 73 : if (strcmp(publish_opt, "insert") == 0)
140 CBC 53 : pubactions->pubinsert = true;
141 20 : else if (strcmp(publish_opt, "update") == 0)
142 9 : pubactions->pubupdate = true;
143 11 : else if (strcmp(publish_opt, "delete") == 0)
144 4 : pubactions->pubdelete = true;
145 7 : else if (strcmp(publish_opt, "truncate") == 0)
146 4 : pubactions->pubtruncate = true;
147 ECB : else
148 GIC 3 : ereport(ERROR,
149 ECB : (errcode(ERRCODE_SYNTAX_ERROR),
150 : errmsg("unrecognized value for publication option \"%s\": \"%s\"",
151 : "publish", publish_opt)));
152 : }
153 : }
154 GIC 84 : else if (strcmp(defel->defname, "publish_via_partition_root") == 0)
155 ECB : {
156 GIC 81 : if (*publish_via_partition_root_given)
157 CBC 3 : errorConflictingDefElem(defel, pstate);
158 78 : *publish_via_partition_root_given = true;
159 78 : *publish_via_partition_root = defGetBoolean(defel);
160 ECB : }
161 : else
162 GIC 3 : ereport(ERROR,
163 ECB : (errcode(ERRCODE_SYNTAX_ERROR),
164 : errmsg("unrecognized publication parameter: \"%s\"", defel->defname)));
165 : }
166 GIC 365 : }
167 ECB :
168 : /*
169 : * Convert the PublicationObjSpecType list into schema oid list and
170 : * PublicationTable list.
171 : */
172 : static void
173 GIC 737 : ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate,
174 ECB : List **rels, List **schemas)
175 : {
176 : ListCell *cell;
177 : PublicationObjSpec *pubobj;
178 :
179 GIC 737 : if (!pubobjspec_list)
180 CBC 39 : return;
181 ECB :
182 GIC 1477 : foreach(cell, pubobjspec_list)
183 ECB : {
184 : Oid schemaid;
185 : List *search_path;
186 :
187 GIC 797 : pubobj = (PublicationObjSpec *) lfirst(cell);
188 ECB :
189 GIC 797 : switch (pubobj->pubobjtype)
190 ECB : {
191 GIC 619 : case PUBLICATIONOBJ_TABLE:
192 CBC 619 : *rels = lappend(*rels, pubobj->pubtable);
193 619 : break;
194 166 : case PUBLICATIONOBJ_TABLES_IN_SCHEMA:
195 166 : schemaid = get_namespace_oid(pubobj->name, false);
196 ECB :
197 : /* Filter out duplicates if user specifies "sch1, sch1" */
198 GIC 151 : *schemas = list_append_unique_oid(*schemas, schemaid);
199 CBC 151 : break;
200 12 : case PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA:
201 12 : search_path = fetch_search_path(false);
202 12 : if (search_path == NIL) /* nothing valid in search_path? */
203 3 : ereport(ERROR,
204 ECB : errcode(ERRCODE_UNDEFINED_SCHEMA),
205 : errmsg("no schema has been selected for CURRENT_SCHEMA"));
206 :
207 GIC 9 : schemaid = linitial_oid(search_path);
208 CBC 9 : list_free(search_path);
209 ECB :
210 : /* Filter out duplicates if user specifies "sch1, sch1" */
211 GIC 9 : *schemas = list_append_unique_oid(*schemas, schemaid);
212 CBC 9 : break;
213 LBC 0 : default:
214 EUB : /* shouldn't happen */
215 UIC 0 : elog(ERROR, "invalid publication object type %d", pubobj->pubobjtype);
216 EUB : break;
217 : }
218 : }
219 : }
220 :
221 : /*
222 : * Returns true if any of the columns used in the row filter WHERE expression is
223 : * not part of REPLICA IDENTITY, false otherwise.
224 : */
225 : static bool
226 GIC 129 : contain_invalid_rfcolumn_walker(Node *node, rf_context *context)
227 ECB : {
228 GIC 129 : if (node == NULL)
229 LBC 0 : return false;
230 EUB :
231 GIC 129 : if (IsA(node, Var))
232 ECB : {
233 GIC 52 : Var *var = (Var *) node;
234 CBC 52 : AttrNumber attnum = var->varattno;
235 ECB :
236 : /*
237 : * If pubviaroot is true, we are validating the row filter of the
238 : * parent table, but the bitmap contains the replica identity
239 : * information of the child table. So, get the column number of the
240 : * child table as parent and child column order could be different.
241 : */
242 GIC 52 : if (context->pubviaroot)
243 ECB : {
244 GIC 8 : char *colname = get_attname(context->parentid, attnum, false);
245 ECB :
246 GIC 8 : attnum = get_attnum(context->relid, colname);
247 ECB : }
248 :
249 GIC 52 : if (!bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
250 CBC 52 : context->bms_replident))
251 30 : return true;
252 ECB : }
253 :
254 GIC 99 : return expression_tree_walker(node, contain_invalid_rfcolumn_walker,
255 ECB : (void *) context);
256 : }
257 :
258 : /*
259 : * Check if all columns referenced in the filter expression are part of the
260 : * REPLICA IDENTITY index or not.
261 : *
262 : * Returns true if any invalid column is found.
263 : */
264 : bool
265 GIC 321 : pub_rf_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
266 ECB : bool pubviaroot)
267 : {
268 : HeapTuple rftuple;
269 GIC 321 : Oid relid = RelationGetRelid(relation);
270 CBC 321 : Oid publish_as_relid = RelationGetRelid(relation);
271 321 : bool result = false;
272 ECB : Datum rfdatum;
273 : bool rfisnull;
274 :
275 : /*
276 : * FULL means all columns are in the REPLICA IDENTITY, so all columns are
277 : * allowed in the row filter and we can skip the validation.
278 : */
279 GIC 321 : if (relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
280 CBC 72 : return false;
281 ECB :
282 : /*
283 : * For a partition, if pubviaroot is true, find the topmost ancestor that
284 : * is published via this publication as we need to use its row filter
285 : * expression to filter the partition's changes.
286 : *
287 : * Note that even though the row filter used is for an ancestor, the
288 : * REPLICA IDENTITY used will be for the actual child table.
289 : */
290 GIC 249 : if (pubviaroot && relation->rd_rel->relispartition)
291 ECB : {
292 : publish_as_relid
293 GIC 53 : = GetTopMostAncestorInPublication(pubid, ancestors, NULL);
294 ECB :
295 GIC 53 : if (!OidIsValid(publish_as_relid))
296 CBC 3 : publish_as_relid = relid;
297 ECB : }
298 :
299 GIC 249 : rftuple = SearchSysCache2(PUBLICATIONRELMAP,
300 ECB : ObjectIdGetDatum(publish_as_relid),
301 : ObjectIdGetDatum(pubid));
302 :
303 GIC 249 : if (!HeapTupleIsValid(rftuple))
304 CBC 28 : return false;
305 ECB :
306 GIC 221 : rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
307 ECB : Anum_pg_publication_rel_prqual,
308 : &rfisnull);
309 :
310 GIC 221 : if (!rfisnull)
311 ECB : {
312 GIC 51 : rf_context context = {0};
313 ECB : Node *rfnode;
314 GIC 51 : Bitmapset *bms = NULL;
315 ECB :
316 GIC 51 : context.pubviaroot = pubviaroot;
317 CBC 51 : context.parentid = publish_as_relid;
318 51 : context.relid = relid;
319 ECB :
320 : /* Remember columns that are part of the REPLICA IDENTITY */
321 GIC 51 : bms = RelationGetIndexAttrBitmap(relation,
322 ECB : INDEX_ATTR_BITMAP_IDENTITY_KEY);
323 :
324 GIC 51 : context.bms_replident = bms;
325 CBC 51 : rfnode = stringToNode(TextDatumGetCString(rfdatum));
326 51 : result = contain_invalid_rfcolumn_walker(rfnode, &context);
327 ECB : }
328 :
329 GIC 221 : ReleaseSysCache(rftuple);
330 ECB :
331 GIC 221 : return result;
332 ECB : }
333 :
334 : /*
335 : * Check if all columns referenced in the REPLICA IDENTITY are covered by
336 : * the column list.
337 : *
338 : * Returns true if any replica identity column is not covered by column list.
339 : */
340 : bool
341 GIC 321 : pub_collist_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
342 ECB : bool pubviaroot)
343 : {
344 : HeapTuple tuple;
345 GIC 321 : Oid relid = RelationGetRelid(relation);
346 CBC 321 : Oid publish_as_relid = RelationGetRelid(relation);
347 321 : bool result = false;
348 ECB : Datum datum;
349 : bool isnull;
350 :
351 : /*
352 : * For a partition, if pubviaroot is true, find the topmost ancestor that
353 : * is published via this publication as we need to use its column list for
354 : * the changes.
355 : *
356 : * Note that even though the column list used is for an ancestor, the
357 : * REPLICA IDENTITY used will be for the actual child table.
358 : */
359 GIC 321 : if (pubviaroot && relation->rd_rel->relispartition)
360 ECB : {
361 GIC 64 : publish_as_relid = GetTopMostAncestorInPublication(pubid, ancestors, NULL);
362 ECB :
363 GIC 64 : if (!OidIsValid(publish_as_relid))
364 CBC 3 : publish_as_relid = relid;
365 ECB : }
366 :
367 GIC 321 : tuple = SearchSysCache2(PUBLICATIONRELMAP,
368 ECB : ObjectIdGetDatum(publish_as_relid),
369 : ObjectIdGetDatum(pubid));
370 :
371 GIC 321 : if (!HeapTupleIsValid(tuple))
372 CBC 32 : return false;
373 ECB :
374 GIC 289 : datum = SysCacheGetAttr(PUBLICATIONRELMAP, tuple,
375 ECB : Anum_pg_publication_rel_prattrs,
376 : &isnull);
377 :
378 GIC 289 : if (!isnull)
379 ECB : {
380 : int x;
381 : Bitmapset *idattrs;
382 GIC 115 : Bitmapset *columns = NULL;
383 ECB :
384 : /* With REPLICA IDENTITY FULL, no column list is allowed. */
385 GIC 115 : if (relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
386 CBC 18 : result = true;
387 ECB :
388 : /* Transform the column list datum to a bitmapset. */
389 GIC 115 : columns = pub_collist_to_bitmapset(NULL, datum, NULL);
390 ECB :
391 : /* Remember columns that are part of the REPLICA IDENTITY */
392 GIC 115 : idattrs = RelationGetIndexAttrBitmap(relation,
393 ECB : INDEX_ATTR_BITMAP_IDENTITY_KEY);
394 :
395 : /*
396 : * Attnums in the bitmap returned by RelationGetIndexAttrBitmap are
397 : * offset (to handle system columns the usual way), while column list
398 : * does not use offset, so we can't do bms_is_subset(). Instead, we
399 : * have to loop over the idattrs and check all of them are in the
400 : * list.
401 : */
402 GIC 115 : x = -1;
403 CBC 175 : while ((x = bms_next_member(idattrs, x)) >= 0)
404 ECB : {
405 GIC 96 : AttrNumber attnum = (x + FirstLowInvalidHeapAttributeNumber);
406 ECB :
407 : /*
408 : * If pubviaroot is true, we are validating the column list of the
409 : * parent table, but the bitmap contains the replica identity
410 : * information of the child table. The parent/child attnums may
411 : * not match, so translate them to the parent - get the attname
412 : * from the child, and look it up in the parent.
413 : */
414 GIC 96 : if (pubviaroot)
415 ECB : {
416 : /* attribute name in the child table */
417 GIC 40 : char *colname = get_attname(relid, attnum, false);
418 ECB :
419 : /*
420 : * Determine the attnum for the attribute name in parent (we
421 : * are using the column list defined on the parent).
422 : */
423 GIC 40 : attnum = get_attnum(publish_as_relid, colname);
424 ECB : }
425 :
426 : /* replica identity column, not covered by the column list */
427 GIC 96 : if (!bms_is_member(attnum, columns))
428 ECB : {
429 GIC 36 : result = true;
430 CBC 36 : break;
431 ECB : }
432 : }
433 :
434 GIC 115 : bms_free(idattrs);
435 CBC 115 : bms_free(columns);
436 ECB : }
437 :
438 GIC 289 : ReleaseSysCache(tuple);
439 ECB :
440 GIC 289 : return result;
441 ECB : }
442 :
443 : /* check_functions_in_node callback */
444 : static bool
445 GIC 198 : contain_mutable_or_user_functions_checker(Oid func_id, void *context)
446 ECB : {
447 GIC 198 : return (func_volatile(func_id) != PROVOLATILE_IMMUTABLE ||
448 ECB : func_id >= FirstNormalObjectId);
449 : }
450 :
451 : /*
452 : * The row filter walker checks if the row filter expression is a "simple
453 : * expression".
454 : *
455 : * It allows only simple or compound expressions such as:
456 : * - (Var Op Const)
457 : * - (Var Op Var)
458 : * - (Var Op Const) AND/OR (Var Op Const)
459 : * - etc
460 : * (where Var is a column of the table this filter belongs to)
461 : *
462 : * The simple expression has the following restrictions:
463 : * - User-defined operators are not allowed;
464 : * - User-defined functions are not allowed;
465 : * - User-defined types are not allowed;
466 : * - User-defined collations are not allowed;
467 : * - Non-immutable built-in functions are not allowed;
468 : * - System columns are not allowed.
469 : *
470 : * NOTES
471 : *
472 : * We don't allow user-defined functions/operators/types/collations because
473 : * (a) if a user drops a user-defined object used in a row filter expression or
474 : * if there is any other error while using it, the logical decoding
475 : * infrastructure won't be able to recover from such an error even if the
476 : * object is recreated again because a historic snapshot is used to evaluate
477 : * the row filter;
478 : * (b) a user-defined function can be used to access tables that could have
479 : * unpleasant results because a historic snapshot is used. That's why only
480 : * immutable built-in functions are allowed in row filter expressions.
481 : *
482 : * We don't allow system columns because currently, we don't have that
483 : * information in the tuple passed to downstream. Also, as we don't replicate
484 : * those to subscribers, there doesn't seem to be a need for a filter on those
485 : * columns.
486 : *
487 : * We can allow other node types after more analysis and testing.
488 : */
489 : static bool
490 GIC 664 : check_simple_rowfilter_expr_walker(Node *node, ParseState *pstate)
491 ECB : {
492 GIC 664 : char *errdetail_msg = NULL;
493 ECB :
494 GIC 664 : if (node == NULL)
495 CBC 3 : return false;
496 ECB :
497 GIC 661 : switch (nodeTag(node))
498 ECB : {
499 GIC 178 : case T_Var:
500 ECB : /* System columns are not allowed. */
501 GIC 178 : if (((Var *) node)->varattno < InvalidAttrNumber)
502 CBC 3 : errdetail_msg = _("System columns are not allowed.");
503 178 : break;
504 176 : case T_OpExpr:
505 ECB : case T_DistinctExpr:
506 : case T_NullIfExpr:
507 : /* OK, except user-defined operators are not allowed. */
508 GIC 176 : if (((OpExpr *) node)->opno >= FirstNormalObjectId)
509 CBC 3 : errdetail_msg = _("User-defined operators are not allowed.");
510 176 : break;
511 3 : case T_ScalarArrayOpExpr:
512 ECB : /* OK, except user-defined operators are not allowed. */
513 GIC 3 : if (((ScalarArrayOpExpr *) node)->opno >= FirstNormalObjectId)
514 LBC 0 : errdetail_msg = _("User-defined operators are not allowed.");
515 EUB :
516 : /*
517 : * We don't need to check the hashfuncid and negfuncid of
518 : * ScalarArrayOpExpr as those functions are only built for a
519 : * subquery.
520 : */
521 GIC 3 : break;
522 CBC 3 : case T_RowCompareExpr:
523 ECB : {
524 : ListCell *opid;
525 :
526 : /* OK, except user-defined operators are not allowed. */
527 GIC 9 : foreach(opid, ((RowCompareExpr *) node)->opnos)
528 ECB : {
529 GIC 6 : if (lfirst_oid(opid) >= FirstNormalObjectId)
530 ECB : {
531 UIC 0 : errdetail_msg = _("User-defined operators are not allowed.");
532 UBC 0 : break;
533 EUB : }
534 : }
535 : }
536 GIC 3 : break;
537 CBC 298 : case T_Const:
538 ECB : case T_FuncExpr:
539 : case T_BoolExpr:
540 : case T_RelabelType:
541 : case T_CollateExpr:
542 : case T_CaseExpr:
543 : case T_CaseTestExpr:
544 : case T_ArrayExpr:
545 : case T_RowExpr:
546 : case T_CoalesceExpr:
547 : case T_MinMaxExpr:
548 : case T_XmlExpr:
549 : case T_NullTest:
550 : case T_BooleanTest:
551 : case T_List:
552 : /* OK, supported */
553 GIC 298 : break;
554 CBC 3 : default:
555 3 : errdetail_msg = _("Only columns, constants, built-in operators, built-in data types, built-in collations, and immutable built-in functions are allowed.");
556 3 : break;
557 ECB : }
558 :
559 : /*
560 : * For all the supported nodes, if we haven't already found a problem,
561 : * check the types, functions, and collations used in it. We check List
562 : * by walking through each element.
563 : */
564 GIC 661 : if (!errdetail_msg && !IsA(node, List))
565 ECB : {
566 GIC 625 : if (exprType(node) >= FirstNormalObjectId)
567 CBC 3 : errdetail_msg = _("User-defined types are not allowed.");
568 622 : else if (check_functions_in_node(node, contain_mutable_or_user_functions_checker,
569 ECB : (void *) pstate))
570 GIC 6 : errdetail_msg = _("User-defined or built-in mutable functions are not allowed.");
571 CBC 1232 : else if (exprCollation(node) >= FirstNormalObjectId ||
572 616 : exprInputCollation(node) >= FirstNormalObjectId)
573 3 : errdetail_msg = _("User-defined collations are not allowed.");
574 ECB : }
575 :
576 : /*
577 : * If we found a problem in this node, throw error now. Otherwise keep
578 : * going.
579 : */
580 GIC 661 : if (errdetail_msg)
581 CBC 21 : ereport(ERROR,
582 ECB : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
583 : errmsg("invalid publication WHERE expression"),
584 : errdetail_internal("%s", errdetail_msg),
585 : parser_errposition(pstate, exprLocation(node))));
586 :
587 GIC 640 : return expression_tree_walker(node, check_simple_rowfilter_expr_walker,
588 ECB : (void *) pstate);
589 : }
590 :
591 : /*
592 : * Check if the row filter expression is a "simple expression".
593 : *
594 : * See check_simple_rowfilter_expr_walker for details.
595 : */
596 : static bool
597 GIC 172 : check_simple_rowfilter_expr(Node *node, ParseState *pstate)
598 ECB : {
599 GIC 172 : return check_simple_rowfilter_expr_walker(node, pstate);
600 ECB : }
601 :
602 : /*
603 : * Transform the publication WHERE expression for all the relations in the list,
604 : * ensuring it is coerced to boolean and necessary collation information is
605 : * added if required, and add a new nsitem/RTE for the associated relation to
606 : * the ParseState's namespace list.
607 : *
608 : * Also check the publication row filter expression and throw an error if
609 : * anything not permitted or unexpected is encountered.
610 : */
611 : static void
612 GIC 511 : TransformPubWhereClauses(List *tables, const char *queryString,
613 ECB : bool pubviaroot)
614 : {
615 : ListCell *lc;
616 :
617 GIC 1022 : foreach(lc, tables)
618 ECB : {
619 : ParseNamespaceItem *nsitem;
620 GIC 541 : Node *whereclause = NULL;
621 ECB : ParseState *pstate;
622 GIC 541 : PublicationRelInfo *pri = (PublicationRelInfo *) lfirst(lc);
623 ECB :
624 GIC 541 : if (pri->whereClause == NULL)
625 CBC 360 : continue;
626 ECB :
627 : /*
628 : * If the publication doesn't publish changes via the root partitioned
629 : * table, the partition's row filter will be used. So disallow using
630 : * WHERE clause on partitioned table in this case.
631 : */
632 GIC 181 : if (!pubviaroot &&
633 CBC 170 : pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
634 3 : ereport(ERROR,
635 ECB : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
636 : errmsg("cannot use publication WHERE clause for relation \"%s\"",
637 : RelationGetRelationName(pri->relation)),
638 : errdetail("WHERE clause cannot be used for a partitioned table when %s is false.",
639 : "publish_via_partition_root")));
640 :
641 : /*
642 : * A fresh pstate is required so that we only have "this" table in its
643 : * rangetable
644 : */
645 GIC 178 : pstate = make_parsestate(NULL);
646 CBC 178 : pstate->p_sourcetext = queryString;
647 178 : nsitem = addRangeTableEntryForRelation(pstate, pri->relation,
648 ECB : AccessShareLock, NULL,
649 : false, false);
650 GIC 178 : addNSItemToQuery(pstate, nsitem, false, true, true);
651 ECB :
652 GIC 178 : whereclause = transformWhereClause(pstate,
653 CBC 178 : copyObject(pri->whereClause),
654 ECB : EXPR_KIND_WHERE,
655 : "PUBLICATION WHERE");
656 :
657 : /* Fix up collation information */
658 GIC 172 : assign_expr_collations(pstate, whereclause);
659 ECB :
660 : /*
661 : * We allow only simple expressions in row filters. See
662 : * check_simple_rowfilter_expr_walker.
663 : */
664 GIC 172 : check_simple_rowfilter_expr(whereclause, pstate);
665 ECB :
666 GIC 151 : free_parsestate(pstate);
667 ECB :
668 GIC 151 : pri->whereClause = whereclause;
669 ECB : }
670 GIC 481 : }
671 ECB :
672 :
673 : /*
674 : * Given a list of tables that are going to be added to a publication,
675 : * verify that they fulfill the necessary preconditions, namely: no tables
676 : * have a column list if any schema is published; and partitioned tables do
677 : * not have column lists if publish_via_partition_root is not set.
678 : *
679 : * 'publish_schema' indicates that the publication contains any TABLES IN
680 : * SCHEMA elements (newly added in this command, or preexisting).
681 : * 'pubviaroot' is the value of publish_via_partition_root.
682 : */
683 : static void
684 GIC 481 : CheckPubRelationColumnList(char *pubname, List *tables,
685 ECB : bool publish_schema, bool pubviaroot)
686 : {
687 : ListCell *lc;
688 :
689 GIC 977 : foreach(lc, tables)
690 ECB : {
691 GIC 511 : PublicationRelInfo *pri = (PublicationRelInfo *) lfirst(lc);
692 ECB :
693 GIC 511 : if (pri->columns == NIL)
694 CBC 348 : continue;
695 ECB :
696 : /*
697 : * Disallow specifying column list if any schema is in the
698 : * publication.
699 : *
700 : * XXX We could instead just forbid the case when the publication
701 : * tries to publish the table with a column list and a schema for that
702 : * table. However, if we do that then we need a restriction during
703 : * ALTER TABLE ... SET SCHEMA to prevent such a case which doesn't
704 : * seem to be a good idea.
705 : */
706 GIC 163 : if (publish_schema)
707 CBC 12 : ereport(ERROR,
708 ECB : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
709 : errmsg("cannot use column list for relation \"%s.%s\" in publication \"%s\"",
710 : get_namespace_name(RelationGetNamespace(pri->relation)),
711 : RelationGetRelationName(pri->relation), pubname),
712 : errdetail("Column lists cannot be specified in publications containing FOR TABLES IN SCHEMA elements."));
713 :
714 : /*
715 : * If the publication doesn't publish changes via the root partitioned
716 : * table, the partition's column list will be used. So disallow using
717 : * a column list on the partitioned table in this case.
718 : */
719 GIC 151 : if (!pubviaroot &&
720 CBC 113 : pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
721 3 : ereport(ERROR,
722 ECB : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
723 : errmsg("cannot use column list for relation \"%s.%s\" in publication \"%s\"",
724 : get_namespace_name(RelationGetNamespace(pri->relation)),
725 : RelationGetRelationName(pri->relation), pubname),
726 : errdetail("Column lists cannot be specified for partitioned tables when %s is false.",
727 : "publish_via_partition_root")));
728 : }
729 GIC 466 : }
730 ECB :
731 : /*
732 : * Create new publication.
733 : */
734 : ObjectAddress
735 GIC 325 : CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
736 ECB : {
737 : Relation rel;
738 : ObjectAddress myself;
739 : Oid puboid;
740 : bool nulls[Natts_pg_publication];
741 : Datum values[Natts_pg_publication];
742 : HeapTuple tup;
743 : bool publish_given;
744 : PublicationActions pubactions;
745 : bool publish_via_partition_root_given;
746 : bool publish_via_partition_root;
747 : AclResult aclresult;
748 GIC 325 : List *relations = NIL;
749 CBC 325 : List *schemaidlist = NIL;
750 ECB :
751 : /* must have CREATE privilege on database */
752 GNC 325 : aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId, GetUserId(), ACL_CREATE);
753 CBC 325 : if (aclresult != ACLCHECK_OK)
754 3 : aclcheck_error(aclresult, OBJECT_DATABASE,
755 3 : get_database_name(MyDatabaseId));
756 ECB :
757 : /* FOR ALL TABLES requires superuser */
758 GIC 322 : if (stmt->for_all_tables && !superuser())
759 LBC 0 : ereport(ERROR,
760 EUB : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
761 : errmsg("must be superuser to create FOR ALL TABLES publication")));
762 :
763 GIC 322 : rel = table_open(PublicationRelationId, RowExclusiveLock);
764 ECB :
765 : /* Check if name is used */
766 GIC 322 : puboid = GetSysCacheOid1(PUBLICATIONNAME, Anum_pg_publication_oid,
767 ECB : CStringGetDatum(stmt->pubname));
768 GIC 322 : if (OidIsValid(puboid))
769 CBC 3 : ereport(ERROR,
770 ECB : (errcode(ERRCODE_DUPLICATE_OBJECT),
771 : errmsg("publication \"%s\" already exists",
772 : stmt->pubname)));
773 :
774 : /* Form a tuple. */
775 GIC 319 : memset(values, 0, sizeof(values));
776 CBC 319 : memset(nulls, false, sizeof(nulls));
777 ECB :
778 GIC 319 : values[Anum_pg_publication_pubname - 1] =
779 CBC 319 : DirectFunctionCall1(namein, CStringGetDatum(stmt->pubname));
780 319 : values[Anum_pg_publication_pubowner - 1] = ObjectIdGetDatum(GetUserId());
781 ECB :
782 GIC 319 : parse_publication_options(pstate,
783 ECB : stmt->options,
784 : &publish_given, &pubactions,
785 : &publish_via_partition_root_given,
786 : &publish_via_partition_root);
787 :
788 GIC 310 : puboid = GetNewOidWithIndex(rel, PublicationObjectIndexId,
789 ECB : Anum_pg_publication_oid);
790 GIC 310 : values[Anum_pg_publication_oid - 1] = ObjectIdGetDatum(puboid);
791 CBC 310 : values[Anum_pg_publication_puballtables - 1] =
792 310 : BoolGetDatum(stmt->for_all_tables);
793 310 : values[Anum_pg_publication_pubinsert - 1] =
794 310 : BoolGetDatum(pubactions.pubinsert);
795 310 : values[Anum_pg_publication_pubupdate - 1] =
796 310 : BoolGetDatum(pubactions.pubupdate);
797 310 : values[Anum_pg_publication_pubdelete - 1] =
798 310 : BoolGetDatum(pubactions.pubdelete);
799 310 : values[Anum_pg_publication_pubtruncate - 1] =
800 310 : BoolGetDatum(pubactions.pubtruncate);
801 310 : values[Anum_pg_publication_pubviaroot - 1] =
802 310 : BoolGetDatum(publish_via_partition_root);
803 ECB :
804 GIC 310 : tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
805 ECB :
806 : /* Insert tuple into catalog. */
807 GIC 310 : CatalogTupleInsert(rel, tup);
808 CBC 310 : heap_freetuple(tup);
809 ECB :
810 GIC 310 : recordDependencyOnOwner(PublicationRelationId, puboid, GetUserId());
811 ECB :
812 GIC 310 : ObjectAddressSet(myself, PublicationRelationId, puboid);
813 ECB :
814 : /* Make the changes visible. */
815 GIC 310 : CommandCounterIncrement();
816 ECB :
817 : /* Associate objects with the publication. */
818 GIC 310 : if (stmt->for_all_tables)
819 ECB : {
820 : /* Invalidate relcache so that publication info is rebuilt. */
821 GIC 24 : CacheInvalidateRelcacheAll();
822 ECB : }
823 : else
824 : {
825 GIC 286 : ObjectsInPublicationToOids(stmt->pubobjects, pstate, &relations,
826 ECB : &schemaidlist);
827 :
828 : /* FOR TABLES IN SCHEMA requires superuser */
829 GIC 277 : if (schemaidlist != NIL && !superuser())
830 CBC 3 : ereport(ERROR,
831 ECB : errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
832 : errmsg("must be superuser to create FOR TABLES IN SCHEMA publication"));
833 :
834 GNC 274 : if (relations != NIL)
835 ECB : {
836 : List *rels;
837 :
838 GIC 181 : rels = OpenTableList(relations);
839 CBC 169 : TransformPubWhereClauses(rels, pstate->p_sourcetext,
840 ECB : publish_via_partition_root);
841 :
842 GIC 157 : CheckPubRelationColumnList(stmt->pubname, rels,
843 ECB : schemaidlist != NIL,
844 : publish_via_partition_root);
845 :
846 GIC 154 : PublicationAddTables(puboid, rels, true, NULL);
847 CBC 141 : CloseTableList(rels);
848 ECB : }
849 :
850 GNC 234 : if (schemaidlist != NIL)
851 ECB : {
852 : /*
853 : * Schema lock is held until the publication is created to prevent
854 : * concurrent schema deletion.
855 : */
856 GIC 67 : LockSchemaList(schemaidlist);
857 CBC 67 : PublicationAddSchemas(puboid, schemaidlist, true, NULL);
858 ECB : }
859 : }
860 :
861 GIC 255 : table_close(rel, RowExclusiveLock);
862 ECB :
863 GIC 255 : InvokeObjectPostCreateHook(PublicationRelationId, puboid, 0);
864 ECB :
865 GIC 255 : if (wal_level != WAL_LEVEL_LOGICAL)
866 CBC 153 : ereport(WARNING,
867 ECB : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
868 : errmsg("wal_level is insufficient to publish logical changes"),
869 : errhint("Set wal_level to \"logical\" before creating subscriptions.")));
870 :
871 GIC 255 : return myself;
872 ECB : }
873 :
874 : /*
875 : * Change options of a publication.
876 : */
877 : static void
878 GIC 55 : AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
879 ECB : Relation rel, HeapTuple tup)
880 : {
881 : bool nulls[Natts_pg_publication];
882 : bool replaces[Natts_pg_publication];
883 : Datum values[Natts_pg_publication];
884 : bool publish_given;
885 : PublicationActions pubactions;
886 : bool publish_via_partition_root_given;
887 : bool publish_via_partition_root;
888 : ObjectAddress obj;
889 : Form_pg_publication pubform;
890 GIC 55 : List *root_relids = NIL;
891 ECB : ListCell *lc;
892 :
893 GIC 55 : parse_publication_options(pstate,
894 ECB : stmt->options,
895 : &publish_given, &pubactions,
896 : &publish_via_partition_root_given,
897 : &publish_via_partition_root);
898 :
899 GIC 55 : pubform = (Form_pg_publication) GETSTRUCT(tup);
900 ECB :
901 : /*
902 : * If the publication doesn't publish changes via the root partitioned
903 : * table, the partition's row filter and column list will be used. So
904 : * disallow using WHERE clause and column lists on partitioned table in
905 : * this case.
906 : */
907 GIC 55 : if (!pubform->puballtables && publish_via_partition_root_given &&
908 CBC 40 : !publish_via_partition_root)
909 ECB : {
910 : /*
911 : * Lock the publication so nobody else can do anything with it. This
912 : * prevents concurrent alter to add partitioned table(s) with WHERE
913 : * clause(s) and/or column lists which we don't allow when not
914 : * publishing via root.
915 : */
916 GIC 24 : LockDatabaseObject(PublicationRelationId, pubform->oid, 0,
917 ECB : AccessShareLock);
918 :
919 GIC 24 : root_relids = GetPublicationRelations(pubform->oid,
920 ECB : PUBLICATION_PART_ROOT);
921 :
922 GIC 42 : foreach(lc, root_relids)
923 ECB : {
924 GIC 24 : Oid relid = lfirst_oid(lc);
925 ECB : HeapTuple rftuple;
926 : char relkind;
927 : char *relname;
928 : bool has_rowfilter;
929 : bool has_collist;
930 :
931 : /*
932 : * Beware: we don't have lock on the relations, so cope silently
933 : * with the cache lookups returning NULL.
934 : */
935 :
936 GIC 24 : rftuple = SearchSysCache2(PUBLICATIONRELMAP,
937 ECB : ObjectIdGetDatum(relid),
938 : ObjectIdGetDatum(pubform->oid));
939 GIC 24 : if (!HeapTupleIsValid(rftuple))
940 LBC 0 : continue;
941 GBC 24 : has_rowfilter = !heap_attisnull(rftuple, Anum_pg_publication_rel_prqual, NULL);
942 CBC 24 : has_collist = !heap_attisnull(rftuple, Anum_pg_publication_rel_prattrs, NULL);
943 24 : if (!has_rowfilter && !has_collist)
944 ECB : {
945 GIC 6 : ReleaseSysCache(rftuple);
946 CBC 6 : continue;
947 ECB : }
948 :
949 GIC 18 : relkind = get_rel_relkind(relid);
950 CBC 18 : if (relkind != RELKIND_PARTITIONED_TABLE)
951 ECB : {
952 GIC 12 : ReleaseSysCache(rftuple);
953 CBC 12 : continue;
954 ECB : }
955 GIC 6 : relname = get_rel_name(relid);
956 CBC 6 : if (relname == NULL) /* table concurrently dropped */
957 ECB : {
958 UIC 0 : ReleaseSysCache(rftuple);
959 UBC 0 : continue;
960 EUB : }
961 :
962 GIC 6 : if (has_rowfilter)
963 CBC 3 : ereport(ERROR,
964 ECB : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
965 : errmsg("cannot set parameter \"%s\" to false for publication \"%s\"",
966 : "publish_via_partition_root",
967 : stmt->pubname),
968 : errdetail("The publication contains a WHERE clause for partitioned table \"%s\", which is not allowed when \"%s\" is false.",
969 : relname, "publish_via_partition_root")));
970 GIC 3 : Assert(has_collist);
971 CBC 3 : ereport(ERROR,
972 ECB : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
973 : errmsg("cannot set parameter \"%s\" to false for publication \"%s\"",
974 : "publish_via_partition_root",
975 : stmt->pubname),
976 : errdetail("The publication contains a column list for partitioned table \"%s\", which is not allowed when \"%s\" is false.",
977 : relname, "publish_via_partition_root")));
978 : }
979 : }
980 :
981 : /* Everything ok, form a new tuple. */
982 GIC 49 : memset(values, 0, sizeof(values));
983 CBC 49 : memset(nulls, false, sizeof(nulls));
984 49 : memset(replaces, false, sizeof(replaces));
985 ECB :
986 GIC 49 : if (publish_given)
987 ECB : {
988 GIC 14 : values[Anum_pg_publication_pubinsert - 1] = BoolGetDatum(pubactions.pubinsert);
989 CBC 14 : replaces[Anum_pg_publication_pubinsert - 1] = true;
990 ECB :
991 GIC 14 : values[Anum_pg_publication_pubupdate - 1] = BoolGetDatum(pubactions.pubupdate);
992 CBC 14 : replaces[Anum_pg_publication_pubupdate - 1] = true;
993 ECB :
994 GIC 14 : values[Anum_pg_publication_pubdelete - 1] = BoolGetDatum(pubactions.pubdelete);
995 CBC 14 : replaces[Anum_pg_publication_pubdelete - 1] = true;
996 ECB :
997 GIC 14 : values[Anum_pg_publication_pubtruncate - 1] = BoolGetDatum(pubactions.pubtruncate);
998 CBC 14 : replaces[Anum_pg_publication_pubtruncate - 1] = true;
999 ECB : }
1000 :
1001 GIC 49 : if (publish_via_partition_root_given)
1002 ECB : {
1003 GIC 35 : values[Anum_pg_publication_pubviaroot - 1] = BoolGetDatum(publish_via_partition_root);
1004 CBC 35 : replaces[Anum_pg_publication_pubviaroot - 1] = true;
1005 ECB : }
1006 :
1007 GIC 49 : tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
1008 ECB : replaces);
1009 :
1010 : /* Update the catalog. */
1011 GIC 49 : CatalogTupleUpdate(rel, &tup->t_self, tup);
1012 ECB :
1013 GIC 49 : CommandCounterIncrement();
1014 ECB :
1015 GIC 49 : pubform = (Form_pg_publication) GETSTRUCT(tup);
1016 ECB :
1017 : /* Invalidate the relcache. */
1018 GIC 49 : if (pubform->puballtables)
1019 ECB : {
1020 GIC 4 : CacheInvalidateRelcacheAll();
1021 ECB : }
1022 : else
1023 : {
1024 GIC 45 : List *relids = NIL;
1025 CBC 45 : List *schemarelids = NIL;
1026 ECB :
1027 : /*
1028 : * For any partitioned tables contained in the publication, we must
1029 : * invalidate all partitions contained in the respective partition
1030 : * trees, not just those explicitly mentioned in the publication.
1031 : */
1032 GIC 45 : if (root_relids == NIL)
1033 CBC 27 : relids = GetPublicationRelations(pubform->oid,
1034 ECB : PUBLICATION_PART_ALL);
1035 : else
1036 : {
1037 : /*
1038 : * We already got tables explicitly mentioned in the publication.
1039 : * Now get all partitions for the partitioned table in the list.
1040 : */
1041 GIC 36 : foreach(lc, root_relids)
1042 CBC 18 : relids = GetPubPartitionOptionRelations(relids,
1043 ECB : PUBLICATION_PART_ALL,
1044 : lfirst_oid(lc));
1045 : }
1046 :
1047 GIC 45 : schemarelids = GetAllSchemaPublicationRelations(pubform->oid,
1048 ECB : PUBLICATION_PART_ALL);
1049 GIC 45 : relids = list_concat_unique_oid(relids, schemarelids);
1050 ECB :
1051 GIC 45 : InvalidatePublicationRels(relids);
1052 ECB : }
1053 :
1054 GIC 49 : ObjectAddressSet(obj, PublicationRelationId, pubform->oid);
1055 CBC 49 : EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress,
1056 ECB : (Node *) stmt);
1057 :
1058 GIC 49 : InvokeObjectPostAlterHook(PublicationRelationId, pubform->oid, 0);
1059 CBC 49 : }
1060 ECB :
1061 : /*
1062 : * Invalidate the relations.
1063 : */
1064 : void
1065 GIC 1071 : InvalidatePublicationRels(List *relids)
1066 ECB : {
1067 : /*
1068 : * We don't want to send too many individual messages, at some point it's
1069 : * cheaper to just reset whole relcache.
1070 : */
1071 GIC 1071 : if (list_length(relids) < MAX_RELCACHE_INVAL_MSGS)
1072 ECB : {
1073 : ListCell *lc;
1074 :
1075 GIC 9421 : foreach(lc, relids)
1076 CBC 8350 : CacheInvalidateRelcacheByRelid(lfirst_oid(lc));
1077 ECB : }
1078 : else
1079 UIC 0 : CacheInvalidateRelcacheAll();
1080 GBC 1071 : }
1081 ECB :
1082 : /*
1083 : * Add or remove table to/from publication.
1084 : */
1085 : static void
1086 GIC 421 : AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
1087 ECB : List *tables, const char *queryString,
1088 : bool publish_schema)
1089 : {
1090 GIC 421 : List *rels = NIL;
1091 CBC 421 : Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
1092 421 : Oid pubid = pubform->oid;
1093 ECB :
1094 : /*
1095 : * Nothing to do if no objects, except in SET: for that it is quite
1096 : * possible that user has not specified any tables in which case we need
1097 : * to remove all the existing tables.
1098 : */
1099 GIC 421 : if (!tables && stmt->action != AP_SetObjects)
1100 CBC 33 : return;
1101 ECB :
1102 GIC 388 : rels = OpenTableList(tables);
1103 ECB :
1104 GIC 388 : if (stmt->action == AP_AddObjects)
1105 ECB : {
1106 GIC 131 : TransformPubWhereClauses(rels, queryString, pubform->pubviaroot);
1107 ECB :
1108 GIC 122 : publish_schema |= is_schema_publication(pubid);
1109 ECB :
1110 GIC 122 : CheckPubRelationColumnList(stmt->pubname, rels, publish_schema,
1111 CBC 122 : pubform->pubviaroot);
1112 ECB :
1113 GIC 116 : PublicationAddTables(pubid, rels, false, stmt);
1114 ECB : }
1115 GIC 257 : else if (stmt->action == AP_DropObjects)
1116 CBC 46 : PublicationDropTables(pubid, rels, false);
1117 ECB : else /* AP_SetObjects */
1118 : {
1119 GIC 211 : List *oldrelids = GetPublicationRelations(pubid,
1120 ECB : PUBLICATION_PART_ROOT);
1121 GIC 211 : List *delrels = NIL;
1122 ECB : ListCell *oldlc;
1123 :
1124 GIC 211 : TransformPubWhereClauses(rels, queryString, pubform->pubviaroot);
1125 ECB :
1126 GIC 202 : CheckPubRelationColumnList(stmt->pubname, rels, publish_schema,
1127 CBC 202 : pubform->pubviaroot);
1128 ECB :
1129 : /*
1130 : * To recreate the relation list for the publication, look for
1131 : * existing relations that do not need to be dropped.
1132 : */
1133 GIC 385 : foreach(oldlc, oldrelids)
1134 ECB : {
1135 GIC 189 : Oid oldrelid = lfirst_oid(oldlc);
1136 ECB : ListCell *newlc;
1137 : PublicationRelInfo *oldrel;
1138 GIC 189 : bool found = false;
1139 ECB : HeapTuple rftuple;
1140 GIC 189 : Node *oldrelwhereclause = NULL;
1141 CBC 189 : Bitmapset *oldcolumns = NULL;
1142 ECB :
1143 : /* look up the cache for the old relmap */
1144 GIC 189 : rftuple = SearchSysCache2(PUBLICATIONRELMAP,
1145 ECB : ObjectIdGetDatum(oldrelid),
1146 : ObjectIdGetDatum(pubid));
1147 :
1148 : /*
1149 : * See if the existing relation currently has a WHERE clause or a
1150 : * column list. We need to compare those too.
1151 : */
1152 GIC 189 : if (HeapTupleIsValid(rftuple))
1153 ECB : {
1154 GIC 189 : bool isnull = true;
1155 ECB : Datum whereClauseDatum;
1156 : Datum columnListDatum;
1157 :
1158 : /* Load the WHERE clause for this table. */
1159 GIC 189 : whereClauseDatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
1160 ECB : Anum_pg_publication_rel_prqual,
1161 : &isnull);
1162 GIC 189 : if (!isnull)
1163 CBC 105 : oldrelwhereclause = stringToNode(TextDatumGetCString(whereClauseDatum));
1164 ECB :
1165 : /* Transform the int2vector column list to a bitmap. */
1166 GIC 189 : columnListDatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
1167 ECB : Anum_pg_publication_rel_prattrs,
1168 : &isnull);
1169 :
1170 GIC 189 : if (!isnull)
1171 CBC 62 : oldcolumns = pub_collist_to_bitmapset(NULL, columnListDatum, NULL);
1172 ECB :
1173 GIC 189 : ReleaseSysCache(rftuple);
1174 ECB : }
1175 :
1176 GIC 371 : foreach(newlc, rels)
1177 ECB : {
1178 : PublicationRelInfo *newpubrel;
1179 : Oid newrelid;
1180 GIC 190 : Bitmapset *newcolumns = NULL;
1181 ECB :
1182 GIC 190 : newpubrel = (PublicationRelInfo *) lfirst(newlc);
1183 CBC 190 : newrelid = RelationGetRelid(newpubrel->relation);
1184 ECB :
1185 : /*
1186 : * If the new publication has column list, transform it to a
1187 : * bitmap too.
1188 : */
1189 GIC 190 : if (newpubrel->columns)
1190 ECB : {
1191 : ListCell *lc;
1192 :
1193 GIC 177 : foreach(lc, newpubrel->columns)
1194 ECB : {
1195 GIC 109 : char *colname = strVal(lfirst(lc));
1196 CBC 109 : AttrNumber attnum = get_attnum(newrelid, colname);
1197 ECB :
1198 GIC 109 : newcolumns = bms_add_member(newcolumns, attnum);
1199 ECB : }
1200 : }
1201 :
1202 : /*
1203 : * Check if any of the new set of relations matches with the
1204 : * existing relations in the publication. Additionally, if the
1205 : * relation has an associated WHERE clause, check the WHERE
1206 : * expressions also match. Same for the column list. Drop the
1207 : * rest.
1208 : */
1209 GIC 190 : if (RelationGetRelid(newpubrel->relation) == oldrelid)
1210 ECB : {
1211 GIC 130 : if (equal(oldrelwhereclause, newpubrel->whereClause) &&
1212 CBC 34 : bms_equal(oldcolumns, newcolumns))
1213 ECB : {
1214 GIC 8 : found = true;
1215 CBC 8 : break;
1216 ECB : }
1217 : }
1218 : }
1219 :
1220 : /*
1221 : * Add the non-matched relations to a list so that they can be
1222 : * dropped.
1223 : */
1224 GIC 189 : if (!found)
1225 ECB : {
1226 GIC 181 : oldrel = palloc(sizeof(PublicationRelInfo));
1227 CBC 181 : oldrel->whereClause = NULL;
1228 181 : oldrel->columns = NIL;
1229 181 : oldrel->relation = table_open(oldrelid,
1230 ECB : ShareUpdateExclusiveLock);
1231 GIC 181 : delrels = lappend(delrels, oldrel);
1232 ECB : }
1233 : }
1234 :
1235 : /* And drop them. */
1236 GIC 196 : PublicationDropTables(pubid, delrels, true);
1237 ECB :
1238 : /*
1239 : * Don't bother calculating the difference for adding, we'll catch and
1240 : * skip existing ones when doing catalog update.
1241 : */
1242 GIC 196 : PublicationAddTables(pubid, rels, true, stmt);
1243 ECB :
1244 GIC 196 : CloseTableList(delrels);
1245 ECB : }
1246 :
1247 GIC 331 : CloseTableList(rels);
1248 ECB : }
1249 :
1250 : /*
1251 : * Alter the publication schemas.
1252 : *
1253 : * Add or remove schemas to/from publication.
1254 : */
1255 : static void
1256 GIC 364 : AlterPublicationSchemas(AlterPublicationStmt *stmt,
1257 ECB : HeapTuple tup, List *schemaidlist)
1258 : {
1259 GIC 364 : Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
1260 ECB :
1261 : /*
1262 : * Nothing to do if no objects, except in SET: for that it is quite
1263 : * possible that user has not specified any schemas in which case we need
1264 : * to remove all the existing schemas.
1265 : */
1266 GIC 364 : if (!schemaidlist && stmt->action != AP_SetObjects)
1267 CBC 135 : return;
1268 ECB :
1269 : /*
1270 : * Schema lock is held until the publication is altered to prevent
1271 : * concurrent schema deletion.
1272 : */
1273 GIC 229 : LockSchemaList(schemaidlist);
1274 CBC 229 : if (stmt->action == AP_AddObjects)
1275 ECB : {
1276 : ListCell *lc;
1277 : List *reloids;
1278 :
1279 GIC 14 : reloids = GetPublicationRelations(pubform->oid, PUBLICATION_PART_ROOT);
1280 ECB :
1281 GIC 18 : foreach(lc, reloids)
1282 ECB : {
1283 : HeapTuple coltuple;
1284 :
1285 GIC 7 : coltuple = SearchSysCache2(PUBLICATIONRELMAP,
1286 ECB : ObjectIdGetDatum(lfirst_oid(lc)),
1287 : ObjectIdGetDatum(pubform->oid));
1288 :
1289 GIC 7 : if (!HeapTupleIsValid(coltuple))
1290 LBC 0 : continue;
1291 EUB :
1292 : /*
1293 : * Disallow adding schema if column list is already part of the
1294 : * publication. See CheckPubRelationColumnList.
1295 : */
1296 GIC 7 : if (!heap_attisnull(coltuple, Anum_pg_publication_rel_prattrs, NULL))
1297 CBC 3 : ereport(ERROR,
1298 ECB : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1299 : errmsg("cannot add schema to publication \"%s\"",
1300 : stmt->pubname),
1301 : errdetail("Schemas cannot be added if any tables that specify a column list are already part of the publication."));
1302 :
1303 GIC 4 : ReleaseSysCache(coltuple);
1304 ECB : }
1305 :
1306 GIC 11 : PublicationAddSchemas(pubform->oid, schemaidlist, false, stmt);
1307 ECB : }
1308 GIC 215 : else if (stmt->action == AP_DropObjects)
1309 CBC 19 : PublicationDropSchemas(pubform->oid, schemaidlist, false);
1310 ECB : else /* AP_SetObjects */
1311 : {
1312 GIC 196 : List *oldschemaids = GetPublicationSchemas(pubform->oid);
1313 CBC 196 : List *delschemas = NIL;
1314 ECB :
1315 : /* Identify which schemas should be dropped */
1316 GIC 196 : delschemas = list_difference_oid(oldschemaids, schemaidlist);
1317 ECB :
1318 : /*
1319 : * Schema lock is held until the publication is altered to prevent
1320 : * concurrent schema deletion.
1321 : */
1322 GIC 196 : LockSchemaList(delschemas);
1323 ECB :
1324 : /* And drop them */
1325 GIC 196 : PublicationDropSchemas(pubform->oid, delschemas, true);
1326 ECB :
1327 : /*
1328 : * Don't bother calculating the difference for adding, we'll catch and
1329 : * skip existing ones when doing catalog update.
1330 : */
1331 GIC 196 : PublicationAddSchemas(pubform->oid, schemaidlist, true, stmt);
1332 ECB : }
1333 : }
1334 :
1335 : /*
1336 : * Check if relations and schemas can be in a given publication and throw
1337 : * appropriate error if not.
1338 : */
1339 : static void
1340 GIC 442 : CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup,
1341 ECB : List *tables, List *schemaidlist)
1342 : {
1343 GIC 442 : Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
1344 ECB :
1345 GIC 442 : if ((stmt->action == AP_AddObjects || stmt->action == AP_SetObjects) &&
1346 CBC 47 : schemaidlist && !superuser())
1347 3 : ereport(ERROR,
1348 ECB : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1349 : errmsg("must be superuser to add or set schemas")));
1350 :
1351 : /*
1352 : * Check that user is allowed to manipulate the publication tables in
1353 : * schema
1354 : */
1355 GIC 439 : if (schemaidlist && pubform->puballtables)
1356 CBC 9 : ereport(ERROR,
1357 ECB : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1358 : errmsg("publication \"%s\" is defined as FOR ALL TABLES",
1359 : NameStr(pubform->pubname)),
1360 : errdetail("Schemas cannot be added to or dropped from FOR ALL TABLES publications.")));
1361 :
1362 : /* Check that user is allowed to manipulate the publication tables. */
1363 GIC 430 : if (tables && pubform->puballtables)
1364 CBC 9 : ereport(ERROR,
1365 ECB : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1366 : errmsg("publication \"%s\" is defined as FOR ALL TABLES",
1367 : NameStr(pubform->pubname)),
1368 : errdetail("Tables cannot be added to or dropped from FOR ALL TABLES publications.")));
1369 GIC 421 : }
1370 ECB :
1371 : /*
1372 : * Alter the existing publication.
1373 : *
1374 : * This is dispatcher function for AlterPublicationOptions,
1375 : * AlterPublicationSchemas and AlterPublicationTables.
1376 : */
1377 : void
1378 GIC 506 : AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt)
1379 ECB : {
1380 : Relation rel;
1381 : HeapTuple tup;
1382 : Form_pg_publication pubform;
1383 :
1384 GIC 506 : rel = table_open(PublicationRelationId, RowExclusiveLock);
1385 ECB :
1386 GIC 506 : tup = SearchSysCacheCopy1(PUBLICATIONNAME,
1387 ECB : CStringGetDatum(stmt->pubname));
1388 :
1389 GIC 506 : if (!HeapTupleIsValid(tup))
1390 LBC 0 : ereport(ERROR,
1391 EUB : (errcode(ERRCODE_UNDEFINED_OBJECT),
1392 : errmsg("publication \"%s\" does not exist",
1393 : stmt->pubname)));
1394 :
1395 GIC 506 : pubform = (Form_pg_publication) GETSTRUCT(tup);
1396 ECB :
1397 : /* must be owner */
1398 GNC 506 : if (!object_ownercheck(PublicationRelationId, pubform->oid, GetUserId()))
1399 LBC 0 : aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_PUBLICATION,
1400 UBC 0 : stmt->pubname);
1401 EUB :
1402 GIC 506 : if (stmt->options)
1403 CBC 55 : AlterPublicationOptions(pstate, stmt, rel, tup);
1404 ECB : else
1405 : {
1406 GIC 451 : List *relations = NIL;
1407 CBC 451 : List *schemaidlist = NIL;
1408 451 : Oid pubid = pubform->oid;
1409 ECB :
1410 GIC 451 : ObjectsInPublicationToOids(stmt->pubobjects, pstate, &relations,
1411 ECB : &schemaidlist);
1412 :
1413 GIC 442 : CheckAlterPublication(stmt, tup, relations, schemaidlist);
1414 ECB :
1415 GIC 421 : heap_freetuple(tup);
1416 ECB :
1417 : /* Lock the publication so nobody else can do anything with it. */
1418 GIC 421 : LockDatabaseObject(PublicationRelationId, pubid, 0,
1419 ECB : AccessExclusiveLock);
1420 :
1421 : /*
1422 : * It is possible that by the time we acquire the lock on publication,
1423 : * concurrent DDL has removed it. We can test this by checking the
1424 : * existence of publication. We get the tuple again to avoid the risk
1425 : * of any publication option getting changed.
1426 : */
1427 GIC 421 : tup = SearchSysCacheCopy1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
1428 CBC 421 : if (!HeapTupleIsValid(tup))
1429 LBC 0 : ereport(ERROR,
1430 EUB : errcode(ERRCODE_UNDEFINED_OBJECT),
1431 : errmsg("publication \"%s\" does not exist",
1432 : stmt->pubname));
1433 :
1434 GIC 421 : AlterPublicationTables(stmt, tup, relations, pstate->p_sourcetext,
1435 ECB : schemaidlist != NIL);
1436 GIC 364 : AlterPublicationSchemas(stmt, tup, schemaidlist);
1437 ECB : }
1438 :
1439 : /* Cleanup. */
1440 GIC 404 : heap_freetuple(tup);
1441 CBC 404 : table_close(rel, RowExclusiveLock);
1442 404 : }
1443 ECB :
1444 : /*
1445 : * Remove relation from publication by mapping OID.
1446 : */
1447 : void
1448 GIC 374 : RemovePublicationRelById(Oid proid)
1449 ECB : {
1450 : Relation rel;
1451 : HeapTuple tup;
1452 : Form_pg_publication_rel pubrel;
1453 GIC 374 : List *relids = NIL;
1454 ECB :
1455 GIC 374 : rel = table_open(PublicationRelRelationId, RowExclusiveLock);
1456 ECB :
1457 GIC 374 : tup = SearchSysCache1(PUBLICATIONREL, ObjectIdGetDatum(proid));
1458 ECB :
1459 GIC 374 : if (!HeapTupleIsValid(tup))
1460 LBC 0 : elog(ERROR, "cache lookup failed for publication table %u",
1461 EUB : proid);
1462 :
1463 GIC 374 : pubrel = (Form_pg_publication_rel) GETSTRUCT(tup);
1464 ECB :
1465 : /*
1466 : * Invalidate relcache so that publication info is rebuilt.
1467 : *
1468 : * For the partitioned tables, we must invalidate all partitions contained
1469 : * in the respective partition hierarchies, not just the one explicitly
1470 : * mentioned in the publication. This is required because we implicitly
1471 : * publish the child tables when the parent table is published.
1472 : */
1473 GIC 374 : relids = GetPubPartitionOptionRelations(relids, PUBLICATION_PART_ALL,
1474 ECB : pubrel->prrelid);
1475 :
1476 GIC 374 : InvalidatePublicationRels(relids);
1477 ECB :
1478 GIC 374 : CatalogTupleDelete(rel, &tup->t_self);
1479 ECB :
1480 GIC 374 : ReleaseSysCache(tup);
1481 ECB :
1482 GIC 374 : table_close(rel, RowExclusiveLock);
1483 CBC 374 : }
1484 ECB :
1485 : /*
1486 : * Remove the publication by mapping OID.
1487 : */
1488 : void
1489 GIC 178 : RemovePublicationById(Oid pubid)
1490 ECB : {
1491 : Relation rel;
1492 : HeapTuple tup;
1493 : Form_pg_publication pubform;
1494 :
1495 GIC 178 : rel = table_open(PublicationRelationId, RowExclusiveLock);
1496 ECB :
1497 GIC 178 : tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
1498 CBC 178 : if (!HeapTupleIsValid(tup))
1499 LBC 0 : elog(ERROR, "cache lookup failed for publication %u", pubid);
1500 EUB :
1501 GIC 178 : pubform = (Form_pg_publication) GETSTRUCT(tup);
1502 ECB :
1503 : /* Invalidate relcache so that publication info is rebuilt. */
1504 GIC 178 : if (pubform->puballtables)
1505 CBC 10 : CacheInvalidateRelcacheAll();
1506 ECB :
1507 GIC 178 : CatalogTupleDelete(rel, &tup->t_self);
1508 ECB :
1509 GIC 178 : ReleaseSysCache(tup);
1510 ECB :
1511 GIC 178 : table_close(rel, RowExclusiveLock);
1512 CBC 178 : }
1513 ECB :
1514 : /*
1515 : * Remove schema from publication by mapping OID.
1516 : */
1517 : void
1518 GIC 96 : RemovePublicationSchemaById(Oid psoid)
1519 ECB : {
1520 : Relation rel;
1521 : HeapTuple tup;
1522 GIC 96 : List *schemaRels = NIL;
1523 ECB : Form_pg_publication_namespace pubsch;
1524 :
1525 GIC 96 : rel = table_open(PublicationNamespaceRelationId, RowExclusiveLock);
1526 ECB :
1527 GIC 96 : tup = SearchSysCache1(PUBLICATIONNAMESPACE, ObjectIdGetDatum(psoid));
1528 ECB :
1529 GIC 96 : if (!HeapTupleIsValid(tup))
1530 LBC 0 : elog(ERROR, "cache lookup failed for publication schema %u", psoid);
1531 EUB :
1532 GIC 96 : pubsch = (Form_pg_publication_namespace) GETSTRUCT(tup);
1533 ECB :
1534 : /*
1535 : * Invalidate relcache so that publication info is rebuilt. See
1536 : * RemovePublicationRelById for why we need to consider all the
1537 : * partitions.
1538 : */
1539 GIC 96 : schemaRels = GetSchemaPublicationRelations(pubsch->pnnspid,
1540 ECB : PUBLICATION_PART_ALL);
1541 GIC 96 : InvalidatePublicationRels(schemaRels);
1542 ECB :
1543 GIC 96 : CatalogTupleDelete(rel, &tup->t_self);
1544 ECB :
1545 GIC 96 : ReleaseSysCache(tup);
1546 ECB :
1547 GIC 96 : table_close(rel, RowExclusiveLock);
1548 CBC 96 : }
1549 ECB :
1550 : /*
1551 : * Open relations specified by a PublicationTable list.
1552 : * The returned tables are locked in ShareUpdateExclusiveLock mode in order to
1553 : * add them to a publication.
1554 : */
1555 : static List *
1556 GIC 569 : OpenTableList(List *tables)
1557 ECB : {
1558 GIC 569 : List *relids = NIL;
1559 CBC 569 : List *rels = NIL;
1560 ECB : ListCell *lc;
1561 GIC 569 : List *relids_with_rf = NIL;
1562 CBC 569 : List *relids_with_collist = NIL;
1563 ECB :
1564 : /*
1565 : * Open, share-lock, and check all the explicitly-specified relations
1566 : */
1567 GIC 1167 : foreach(lc, tables)
1568 ECB : {
1569 GIC 610 : PublicationTable *t = lfirst_node(PublicationTable, lc);
1570 CBC 610 : bool recurse = t->relation->inh;
1571 ECB : Relation rel;
1572 : Oid myrelid;
1573 : PublicationRelInfo *pub_rel;
1574 :
1575 : /* Allow query cancel in case this takes a long time */
1576 GIC 610 : CHECK_FOR_INTERRUPTS();
1577 ECB :
1578 GIC 610 : rel = table_openrv(t->relation, ShareUpdateExclusiveLock);
1579 CBC 610 : myrelid = RelationGetRelid(rel);
1580 ECB :
1581 : /*
1582 : * Filter out duplicates if user specifies "foo, foo".
1583 : *
1584 : * Note that this algorithm is known to not be very efficient (O(N^2))
1585 : * but given that it only works on list of tables given to us by user
1586 : * it's deemed acceptable.
1587 : */
1588 GIC 610 : if (list_member_oid(relids, myrelid))
1589 ECB : {
1590 : /* Disallow duplicate tables if there are any with row filters. */
1591 GIC 12 : if (t->whereClause || list_member_oid(relids_with_rf, myrelid))
1592 CBC 6 : ereport(ERROR,
1593 ECB : (errcode(ERRCODE_DUPLICATE_OBJECT),
1594 : errmsg("conflicting or redundant WHERE clauses for table \"%s\"",
1595 : RelationGetRelationName(rel))));
1596 :
1597 : /* Disallow duplicate tables if there are any with column lists. */
1598 GIC 6 : if (t->columns || list_member_oid(relids_with_collist, myrelid))
1599 CBC 6 : ereport(ERROR,
1600 ECB : (errcode(ERRCODE_DUPLICATE_OBJECT),
1601 : errmsg("conflicting or redundant column lists for table \"%s\"",
1602 : RelationGetRelationName(rel))));
1603 :
1604 UIC 0 : table_close(rel, ShareUpdateExclusiveLock);
1605 UBC 0 : continue;
1606 EUB : }
1607 :
1608 GIC 598 : pub_rel = palloc(sizeof(PublicationRelInfo));
1609 CBC 598 : pub_rel->relation = rel;
1610 598 : pub_rel->whereClause = t->whereClause;
1611 598 : pub_rel->columns = t->columns;
1612 598 : rels = lappend(rels, pub_rel);
1613 598 : relids = lappend_oid(relids, myrelid);
1614 ECB :
1615 GIC 598 : if (t->whereClause)
1616 CBC 186 : relids_with_rf = lappend_oid(relids_with_rf, myrelid);
1617 ECB :
1618 GIC 598 : if (t->columns)
1619 CBC 166 : relids_with_collist = lappend_oid(relids_with_collist, myrelid);
1620 ECB :
1621 : /*
1622 : * Add children of this rel, if requested, so that they too are added
1623 : * to the publication. A partitioned table can't have any inheritance
1624 : * children other than its partitions, which need not be explicitly
1625 : * added to the publication.
1626 : */
1627 GIC 598 : if (recurse && rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
1628 ECB : {
1629 : List *children;
1630 : ListCell *child;
1631 :
1632 GIC 507 : children = find_all_inheritors(myrelid, ShareUpdateExclusiveLock,
1633 ECB : NULL);
1634 :
1635 GIC 1018 : foreach(child, children)
1636 ECB : {
1637 GIC 511 : Oid childrelid = lfirst_oid(child);
1638 ECB :
1639 : /* Allow query cancel in case this takes a long time */
1640 GIC 511 : CHECK_FOR_INTERRUPTS();
1641 ECB :
1642 : /*
1643 : * Skip duplicates if user specified both parent and child
1644 : * tables.
1645 : */
1646 GIC 511 : if (list_member_oid(relids, childrelid))
1647 ECB : {
1648 : /*
1649 : * We don't allow to specify row filter for both parent
1650 : * and child table at the same time as it is not very
1651 : * clear which one should be given preference.
1652 : */
1653 GIC 507 : if (childrelid != myrelid &&
1654 LBC 0 : (t->whereClause || list_member_oid(relids_with_rf, childrelid)))
1655 UBC 0 : ereport(ERROR,
1656 EUB : (errcode(ERRCODE_DUPLICATE_OBJECT),
1657 : errmsg("conflicting or redundant WHERE clauses for table \"%s\"",
1658 : RelationGetRelationName(rel))));
1659 :
1660 : /*
1661 : * We don't allow to specify column list for both parent
1662 : * and child table at the same time as it is not very
1663 : * clear which one should be given preference.
1664 : */
1665 GIC 507 : if (childrelid != myrelid &&
1666 LBC 0 : (t->columns || list_member_oid(relids_with_collist, childrelid)))
1667 UBC 0 : ereport(ERROR,
1668 EUB : (errcode(ERRCODE_DUPLICATE_OBJECT),
1669 : errmsg("conflicting or redundant column lists for table \"%s\"",
1670 : RelationGetRelationName(rel))));
1671 :
1672 GIC 507 : continue;
1673 ECB : }
1674 :
1675 : /* find_all_inheritors already got lock */
1676 GIC 4 : rel = table_open(childrelid, NoLock);
1677 CBC 4 : pub_rel = palloc(sizeof(PublicationRelInfo));
1678 4 : pub_rel->relation = rel;
1679 ECB : /* child inherits WHERE clause from parent */
1680 GIC 4 : pub_rel->whereClause = t->whereClause;
1681 ECB :
1682 : /* child inherits column list from parent */
1683 GIC 4 : pub_rel->columns = t->columns;
1684 CBC 4 : rels = lappend(rels, pub_rel);
1685 4 : relids = lappend_oid(relids, childrelid);
1686 ECB :
1687 GIC 4 : if (t->whereClause)
1688 CBC 1 : relids_with_rf = lappend_oid(relids_with_rf, childrelid);
1689 ECB :
1690 GIC 4 : if (t->columns)
1691 LBC 0 : relids_with_collist = lappend_oid(relids_with_collist, childrelid);
1692 EUB : }
1693 : }
1694 : }
1695 :
1696 GIC 557 : list_free(relids);
1697 CBC 557 : list_free(relids_with_rf);
1698 ECB :
1699 GIC 557 : return rels;
1700 ECB : }
1701 :
1702 : /*
1703 : * Close all relations in the list.
1704 : */
1705 : static void
1706 GIC 668 : CloseTableList(List *rels)
1707 ECB : {
1708 : ListCell *lc;
1709 :
1710 GIC 1354 : foreach(lc, rels)
1711 ECB : {
1712 : PublicationRelInfo *pub_rel;
1713 :
1714 GIC 686 : pub_rel = (PublicationRelInfo *) lfirst(lc);
1715 CBC 686 : table_close(pub_rel->relation, NoLock);
1716 ECB : }
1717 :
1718 GIC 668 : list_free_deep(rels);
1719 CBC 668 : }
1720 ECB :
1721 : /*
1722 : * Lock the schemas specified in the schema list in AccessShareLock mode in
1723 : * order to prevent concurrent schema deletion.
1724 : */
1725 : static void
1726 GIC 492 : LockSchemaList(List *schemalist)
1727 ECB : {
1728 : ListCell *lc;
1729 :
1730 GIC 634 : foreach(lc, schemalist)
1731 ECB : {
1732 GIC 142 : Oid schemaid = lfirst_oid(lc);
1733 ECB :
1734 : /* Allow query cancel in case this takes a long time */
1735 GIC 142 : CHECK_FOR_INTERRUPTS();
1736 CBC 142 : LockDatabaseObject(NamespaceRelationId, schemaid, 0, AccessShareLock);
1737 ECB :
1738 : /*
1739 : * It is possible that by the time we acquire the lock on schema,
1740 : * concurrent DDL has removed it. We can test this by checking the
1741 : * existence of schema.
1742 : */
1743 GIC 142 : if (!SearchSysCacheExists1(NAMESPACEOID, ObjectIdGetDatum(schemaid)))
1744 LBC 0 : ereport(ERROR,
1745 EUB : errcode(ERRCODE_UNDEFINED_SCHEMA),
1746 : errmsg("schema with OID %u does not exist", schemaid));
1747 : }
1748 GIC 492 : }
1749 ECB :
1750 : /*
1751 : * Add listed tables to the publication.
1752 : */
1753 : static void
1754 GIC 466 : PublicationAddTables(Oid pubid, List *rels, bool if_not_exists,
1755 ECB : AlterPublicationStmt *stmt)
1756 : {
1757 : ListCell *lc;
1758 :
1759 GIC 466 : Assert(!stmt || !stmt->for_all_tables);
1760 ECB :
1761 GIC 931 : foreach(lc, rels)
1762 ECB : {
1763 GIC 496 : PublicationRelInfo *pub_rel = (PublicationRelInfo *) lfirst(lc);
1764 CBC 496 : Relation rel = pub_rel->relation;
1765 ECB : ObjectAddress obj;
1766 :
1767 : /* Must be owner of the table or superuser. */
1768 GNC 496 : if (!object_ownercheck(RelationRelationId, RelationGetRelid(rel), GetUserId()))
1769 CBC 3 : aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(rel->rd_rel->relkind),
1770 3 : RelationGetRelationName(rel));
1771 ECB :
1772 GIC 493 : obj = publication_add_relation(pubid, pub_rel, if_not_exists);
1773 CBC 465 : if (stmt)
1774 ECB : {
1775 GIC 293 : EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress,
1776 ECB : (Node *) stmt);
1777 :
1778 GIC 293 : InvokeObjectPostCreateHook(PublicationRelRelationId,
1779 ECB : obj.objectId, 0);
1780 : }
1781 : }
1782 GIC 435 : }
1783 ECB :
1784 : /*
1785 : * Remove listed tables from the publication.
1786 : */
1787 : static void
1788 GIC 242 : PublicationDropTables(Oid pubid, List *rels, bool missing_ok)
1789 ECB : {
1790 : ObjectAddress obj;
1791 : ListCell *lc;
1792 : Oid prid;
1793 :
1794 GIC 463 : foreach(lc, rels)
1795 ECB : {
1796 GIC 230 : PublicationRelInfo *pubrel = (PublicationRelInfo *) lfirst(lc);
1797 CBC 230 : Relation rel = pubrel->relation;
1798 230 : Oid relid = RelationGetRelid(rel);
1799 ECB :
1800 GIC 230 : if (pubrel->columns)
1801 LBC 0 : ereport(ERROR,
1802 EUB : errcode(ERRCODE_SYNTAX_ERROR),
1803 : errmsg("column list must not be specified in ALTER PUBLICATION ... DROP"));
1804 :
1805 GIC 230 : prid = GetSysCacheOid2(PUBLICATIONRELMAP, Anum_pg_publication_rel_oid,
1806 ECB : ObjectIdGetDatum(relid),
1807 : ObjectIdGetDatum(pubid));
1808 GIC 230 : if (!OidIsValid(prid))
1809 ECB : {
1810 GIC 6 : if (missing_ok)
1811 LBC 0 : continue;
1812 EUB :
1813 GIC 6 : ereport(ERROR,
1814 ECB : (errcode(ERRCODE_UNDEFINED_OBJECT),
1815 : errmsg("relation \"%s\" is not part of the publication",
1816 : RelationGetRelationName(rel))));
1817 : }
1818 :
1819 GIC 224 : if (pubrel->whereClause)
1820 CBC 3 : ereport(ERROR,
1821 ECB : (errcode(ERRCODE_SYNTAX_ERROR),
1822 : errmsg("cannot use a WHERE clause when removing a table from a publication")));
1823 :
1824 GIC 221 : ObjectAddressSet(obj, PublicationRelRelationId, prid);
1825 CBC 221 : performDeletion(&obj, DROP_CASCADE, 0);
1826 ECB : }
1827 GIC 233 : }
1828 ECB :
1829 : /*
1830 : * Add listed schemas to the publication.
1831 : */
1832 : static void
1833 GIC 274 : PublicationAddSchemas(Oid pubid, List *schemas, bool if_not_exists,
1834 ECB : AlterPublicationStmt *stmt)
1835 : {
1836 : ListCell *lc;
1837 :
1838 GIC 274 : Assert(!stmt || !stmt->for_all_tables);
1839 ECB :
1840 GIC 379 : foreach(lc, schemas)
1841 ECB : {
1842 GIC 111 : Oid schemaid = lfirst_oid(lc);
1843 ECB : ObjectAddress obj;
1844 :
1845 GIC 111 : obj = publication_add_schema(pubid, schemaid, if_not_exists);
1846 CBC 105 : if (stmt)
1847 ECB : {
1848 GIC 29 : EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress,
1849 ECB : (Node *) stmt);
1850 :
1851 GIC 29 : InvokeObjectPostCreateHook(PublicationNamespaceRelationId,
1852 ECB : obj.objectId, 0);
1853 : }
1854 : }
1855 GIC 268 : }
1856 ECB :
1857 : /*
1858 : * Remove listed schemas from the publication.
1859 : */
1860 : static void
1861 GIC 215 : PublicationDropSchemas(Oid pubid, List *schemas, bool missing_ok)
1862 ECB : {
1863 : ObjectAddress obj;
1864 : ListCell *lc;
1865 : Oid psid;
1866 :
1867 GIC 240 : foreach(lc, schemas)
1868 ECB : {
1869 GIC 28 : Oid schemaid = lfirst_oid(lc);
1870 ECB :
1871 GIC 28 : psid = GetSysCacheOid2(PUBLICATIONNAMESPACEMAP,
1872 ECB : Anum_pg_publication_namespace_oid,
1873 : ObjectIdGetDatum(schemaid),
1874 : ObjectIdGetDatum(pubid));
1875 GIC 28 : if (!OidIsValid(psid))
1876 ECB : {
1877 GIC 3 : if (missing_ok)
1878 LBC 0 : continue;
1879 EUB :
1880 GIC 3 : ereport(ERROR,
1881 ECB : (errcode(ERRCODE_UNDEFINED_OBJECT),
1882 : errmsg("tables from schema \"%s\" are not part of the publication",
1883 : get_namespace_name(schemaid))));
1884 : }
1885 :
1886 GIC 25 : ObjectAddressSet(obj, PublicationNamespaceRelationId, psid);
1887 CBC 25 : performDeletion(&obj, DROP_CASCADE, 0);
1888 ECB : }
1889 GIC 212 : }
1890 ECB :
1891 : /*
1892 : * Internal workhorse for changing a publication owner
1893 : */
1894 : static void
1895 GIC 12 : AlterPublicationOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
1896 ECB : {
1897 : Form_pg_publication form;
1898 :
1899 GIC 12 : form = (Form_pg_publication) GETSTRUCT(tup);
1900 ECB :
1901 GIC 12 : if (form->pubowner == newOwnerId)
1902 LBC 0 : return;
1903 EUB :
1904 GIC 12 : if (!superuser())
1905 ECB : {
1906 : AclResult aclresult;
1907 :
1908 : /* Must be owner */
1909 GNC 6 : if (!object_ownercheck(PublicationRelationId, form->oid, GetUserId()))
1910 LBC 0 : aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_PUBLICATION,
1911 UBC 0 : NameStr(form->pubname));
1912 EUB :
1913 : /* Must be able to become new owner */
1914 GNC 6 : check_can_set_role(GetUserId(), newOwnerId);
1915 ECB :
1916 : /* New owner must have CREATE privilege on database */
1917 GNC 6 : aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId, newOwnerId, ACL_CREATE);
1918 CBC 6 : if (aclresult != ACLCHECK_OK)
1919 LBC 0 : aclcheck_error(aclresult, OBJECT_DATABASE,
1920 UBC 0 : get_database_name(MyDatabaseId));
1921 EUB :
1922 GIC 6 : if (form->puballtables && !superuser_arg(newOwnerId))
1923 LBC 0 : ereport(ERROR,
1924 EUB : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1925 : errmsg("permission denied to change owner of publication \"%s\"",
1926 : NameStr(form->pubname)),
1927 : errhint("The owner of a FOR ALL TABLES publication must be a superuser.")));
1928 :
1929 GIC 6 : if (!superuser_arg(newOwnerId) && is_schema_publication(form->oid))
1930 CBC 3 : ereport(ERROR,
1931 ECB : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1932 : errmsg("permission denied to change owner of publication \"%s\"",
1933 : NameStr(form->pubname)),
1934 : errhint("The owner of a FOR TABLES IN SCHEMA publication must be a superuser.")));
1935 : }
1936 :
1937 GIC 9 : form->pubowner = newOwnerId;
1938 CBC 9 : CatalogTupleUpdate(rel, &tup->t_self, tup);
1939 ECB :
1940 : /* Update owner dependency reference */
1941 GIC 9 : changeDependencyOnOwner(PublicationRelationId,
1942 ECB : form->oid,
1943 : newOwnerId);
1944 :
1945 GIC 9 : InvokeObjectPostAlterHook(PublicationRelationId,
1946 ECB : form->oid, 0);
1947 : }
1948 :
1949 : /*
1950 : * Change publication owner -- by name
1951 : */
1952 : ObjectAddress
1953 GIC 12 : AlterPublicationOwner(const char *name, Oid newOwnerId)
1954 ECB : {
1955 : Oid subid;
1956 : HeapTuple tup;
1957 : Relation rel;
1958 : ObjectAddress address;
1959 : Form_pg_publication pubform;
1960 :
1961 GIC 12 : rel = table_open(PublicationRelationId, RowExclusiveLock);
1962 ECB :
1963 GIC 12 : tup = SearchSysCacheCopy1(PUBLICATIONNAME, CStringGetDatum(name));
1964 ECB :
1965 GIC 12 : if (!HeapTupleIsValid(tup))
1966 LBC 0 : ereport(ERROR,
1967 EUB : (errcode(ERRCODE_UNDEFINED_OBJECT),
1968 : errmsg("publication \"%s\" does not exist", name)));
1969 :
1970 GIC 12 : pubform = (Form_pg_publication) GETSTRUCT(tup);
1971 CBC 12 : subid = pubform->oid;
1972 ECB :
1973 GIC 12 : AlterPublicationOwner_internal(rel, tup, newOwnerId);
1974 ECB :
1975 GIC 9 : ObjectAddressSet(address, PublicationRelationId, subid);
1976 ECB :
1977 GIC 9 : heap_freetuple(tup);
1978 ECB :
1979 GIC 9 : table_close(rel, RowExclusiveLock);
1980 ECB :
1981 GIC 9 : return address;
1982 ECB : }
1983 :
1984 : /*
1985 : * Change publication owner -- by OID
1986 : */
1987 : void
1988 UIC 0 : AlterPublicationOwner_oid(Oid subid, Oid newOwnerId)
1989 EUB : {
1990 : HeapTuple tup;
1991 : Relation rel;
1992 :
1993 UIC 0 : rel = table_open(PublicationRelationId, RowExclusiveLock);
1994 EUB :
1995 UIC 0 : tup = SearchSysCacheCopy1(PUBLICATIONOID, ObjectIdGetDatum(subid));
1996 EUB :
1997 UIC 0 : if (!HeapTupleIsValid(tup))
1998 UBC 0 : ereport(ERROR,
1999 EUB : (errcode(ERRCODE_UNDEFINED_OBJECT),
2000 : errmsg("publication with OID %u does not exist", subid)));
2001 :
2002 UIC 0 : AlterPublicationOwner_internal(rel, tup, newOwnerId);
2003 EUB :
2004 UIC 0 : heap_freetuple(tup);
2005 EUB :
2006 UIC 0 : table_close(rel, RowExclusiveLock);
2007 UBC 0 : }
|
