PostgreSQL Source Code: src/backend/commands/publicationcmds.c Source File

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

AltStyle によって変換されたページ (->オリジナル) /