Re: Logical Replication WIP

Hi,

+ <sect1 id="catalog-pg-publication-rel">
+  <title><structname>pg_publication_rel</structname></title>
+
+  <indexterm zone="catalog-pg-publication-rel">
+   <primary>pg_publication_rel</primary>
+  </indexterm>
+
+  <para>
+   The <structname>pg_publication_rel</structname> catalog contains
+   mapping between tables and publications in the database. This is many to
+   many mapping.
+  </para>

I wonder if we shouldn't abstract this a bit away from relations to
allow other objects to be exported to. Could structure it a bit more
like pg_depend.


+ALTER PUBLICATION <replaceable class="PARAMETER">name</replaceable> [ [ WITH ] <replaceable
class="PARAMETER">option</replaceable>[ ... ] ]
 
+
+<phrase>where <replaceable class="PARAMETER">option</replaceable> can be:</phrase>
+
+      PuBLISH_INSERT | NOPuBLISH_INSERT
+    | PuBLISH_UPDATE | NOPuBLISH_UPDATE
+    | PuBLISH_DELETE | NOPuBLISH_DELETE

That's odd casing.


+   <varlistentry>
+    <term><literal>PuBLISH_INSERT</literal></term>
+    <term><literal>NOPuBLISH_INSERT</literal></term>
+    <term><literal>PuBLISH_UPDATE</literal></term>
+    <term><literal>NOPuBLISH_UPDATE</literal></term>
+    <term><literal>PuBLISH_DELETE</literal></term>
+    <term><literal>NOPuBLISH_DELETE</literal></term>

More odd casing.

+   <varlistentry>
+    <term><literal>FOR TABLE</literal></term>
+    <listitem>
+     <para>
+      Specifies optional list of tables to add to the publication.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term><literal>FOR TABLE ALL IN SCHEMA</literal></term>
+    <listitem>
+     <para>
+      Specifies optional schema for which all logged tables will be added to
+      publication.
+     </para>
+    </listitem>
+   </varlistentry>

"FOR TABLE ALL IN SCHEMA" sounds weird.


+  <para>
+   This operation does not reserve any resources on the server. It only
+   defines grouping and filtering logic for future subscribers.
+  </para>

That's strictly speaking not true, maybe rephrase a bit?

+/*
+ * Check if relation can be in given publication and throws appropriate
+ * error if not.
+ */
+static void
+check_publication_add_relation(Relation targetrel)
+{
+    /* Must be table */
+    if (RelationGetForm(targetrel)->relkind != RELKIND_RELATION)
+        ereport(ERROR,
+                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                 errmsg("only tables can be added to publication"),
+                 errdetail("%s is not a table",
+                           RelationGetRelationName(targetrel))));
+
+    /* Can't be system table */
+    if (IsCatalogRelation(targetrel))
+        ereport(ERROR,
+                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                 errmsg("only user tables can be added to publication"),
+                 errdetail("%s is a system table",
+                           RelationGetRelationName(targetrel))));
+
+    /* UNLOGGED and TEMP relations cannot be part of publication. */
+    if (!RelationNeedsWAL(targetrel))
+        ereport(ERROR,
+                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                 errmsg("UNLOGGED and TEMP relations cannot be replicated")));
+}

This probably means we need a check in the ALTER TABLE ... SET UNLOGGED
path.


+/*
+ * Returns if relation represented by oid and Form_pg_class entry
+ * is publishable.
+ *
+ * Does same checks as the above, but does not need relation to be opened
+ * and also does not throw errors.
+ */
+static bool
+is_publishable_class(Oid relid, Form_pg_class reltuple)
+{
+    return reltuple->relkind == RELKIND_RELATION &&
+        !IsCatalogClass(relid, reltuple) &&
+        reltuple->relpersistence == RELPERSISTENCE_PERMANENT &&
+        /* XXX needed to exclude information_schema tables */
+        relid >= FirstNormalObjectId;
+}

Shouldn't that be IsCatalogRelation() instead?


+CREATE VIEW pg_publication_tables AS
+    SELECT
+        P.pubname AS pubname,
+        N.nspname AS schemaname,
+        C.relname AS tablename
+    FROM pg_publication P, pg_class C
+         JOIN pg_namespace N ON (N.oid = C.relnamespace)
+    WHERE C.relkind = 'r'
+      AND C.oid IN (SELECT relid FROM pg_get_publication_tables(P.pubname));

That's going to be quite inefficient if you filter by table... Might be
better to do that via the underlying table.


+/*
+ * Create new publication.
+ * TODO ACL check
+ */

Hm?

+ObjectAddress
+CreatePublication(CreatePublicationStmt *stmt)
+{
+    check_replication_permissions();

+
+/*
+ * Drop publication by OID
+ */
+void
+DropPublicationById(Oid pubid)
+
+/*
+ * Remove relation from publication by mapping OID.
+ */
+void
+RemovePublicationRelById(Oid proid)
+{

Permission checks?

+}

Hm. Neither of these does dependency checking, wonder if that can be
argued to be problematic.


+/*
+ * Gather Relations based o provided by RangeVar list.
+ * The gathered tables are locked in ShareUpdateExclusiveLock mode.
+ */

s/o/on/.  Not sure if gather is the best name.

+static List *
+GatherTableList(List *tables)


+/*
+ * Close all relations in the list.
+ */
+static void
+CloseTables(List *rels)

Shouldn't that be CloseTableList() based on the preceding function's naming?

+
+/*
+ * Add listed tables to the publication.
+ */
+static void
+PublicationAddTables(Oid pubid, List *rels, bool if_not_exists,
+                     AlterPublicationStmt *stmt)
+{
+    ListCell       *lc;
+
+    Assert(!stmt || !stmt->for_all_tables);
+
+    foreach(lc, rels)
+    {
+        Relation    rel = (Relation) lfirst(lc);
+        ObjectAddress    obj;
+
+        obj = publication_add_relation(pubid, rel, if_not_exists);
+        if (stmt)
+            EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress,
+                                             (Node *) stmt);
+    }
+}
+
+/*
+ * Remove listed tables to the publication.
+ */

s/to/from/

+static void
+PublicationDropTables(Oid pubid, List *rels, bool missing_ok)
+{
+    ObjectAddress    obj;
+    ListCell       *lc;
+    Oid                prid;
+
+    foreach(lc, rels)
+    {
+        Relation    rel = (Relation) lfirst(lc);
+        Oid            relid = RelationGetRelid(rel);
+
+        prid = GetSysCacheOid2(PUBLICATIONRELMAP, ObjectIdGetDatum(relid),
+                               ObjectIdGetDatum(pubid));
+        if (!OidIsValid(prid))
+        {
+            if (missing_ok)
+                continue;
+
+            ereport(ERROR,
+                    (errcode(ERRCODE_UNDEFINED_OBJECT),
+                     errmsg("relation \"%s\" is not part of the publication",
+                            RelationGetRelationName(rel))));
+        }
+
+        ObjectAddressSet(obj, PublicationRelRelationId, prid);
+        performDeletion(&obj, DROP_CASCADE, 0);
+    }
+}


/*
+ * Check if command can be executed with current replica identity.
+ */
+static void
+CheckCmdReplicaIdentity(Relation rel, CmdType cmd)
+{
+    PublicationActions *pubactions;
+
+    /* We only need to do checks for UPDATE and DELETE. */
+    if (cmd != CMD_UPDATE && cmd != CMD_DELETE)
+        return;
+
+    /* If relation has replica identity we are always good. */
+    if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
+        OidIsValid(RelationGetReplicaIndex(rel)))
+        return;
+
+    /*
+     * This is either UPDATE OR DELETE and there is no replica identity.
+     *
+     * Check if the table publishes UPDATES or DELETES.
+     */
+    pubactions = GetRelationPublicationActions(rel);
+    if (pubactions->pubupdate || pubactions->pubdelete)

I think that leads to spurious errors. Consider a DELETE with a
publication that replicates updates but not deletes.

+        ereport(ERROR,
+                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+                 errmsg("cannot update table \"%s\" because it does not have replica identity and publishes updates",
+                        RelationGetRelationName(rel)),
+                 errhint("To enable updating the table, provide set REPLICA IDENTITY using ALTER TABLE.")));
+}

"provide set"


+publication_opt_item:
+            IDENT
+                {
+                    /*
+                     * We handle identifiers that aren't parser keywords with
+                     * the following special-case codes, to avoid bloating the
+                     * size of the main parser.
+                     */
+                    if (strcmp($1, "publish_insert") == 0)
+                        $$ = makeDefElem("publish_insert",
+                                         (Node *)makeInteger(TRUE), @1);
+                    else if (strcmp($1, "nopublish_insert") == 0)
+                        $$ = makeDefElem("publish_insert",
+                                         (Node *)makeInteger(FALSE), @1);
+                    else if (strcmp($1, "publish_update") == 0)
+                        $$ = makeDefElem("publish_update",
+                                         (Node *)makeInteger(TRUE), @1);
+                    else if (strcmp($1, "nopublish_update") == 0)
+                        $$ = makeDefElem("publish_update",
+                                         (Node *)makeInteger(FALSE), @1);
+                    else if (strcmp($1, "publish_delete") == 0)
+                        $$ = makeDefElem("publish_delete",
+                                         (Node *)makeInteger(TRUE), @1);
+                    else if (strcmp($1, "nopublish_delete") == 0)
+                        $$ = makeDefElem("publish_delete",
+                                         (Node *)makeInteger(FALSE), @1);
+                    else
+                        ereport(ERROR,
+                                (errcode(ERRCODE_SYNTAX_ERROR),
+                                 errmsg("unrecognized publication option \"%s\"", $1),
+                                     parser_errposition(@1)));
+                }
+        ;

I still would very much like to move this outside of gram.y and just use
IDENTs here. Like how COPY options are handled.


+/*
+ * Get publication actions for list of publication oids.
+ */
+struct PublicationActions *
+GetRelationPublicationActions(Relation relation)

API description and function name/parameters don't quite match.



+CATALOG(pg_publication,6104)
+{
+    NameData    pubname;            /* name of the publication */
+
+    /*
+     * indicates that this is special publication which should encompass
+     * all tables in the database (except for the unlogged and temp ones)
+     */
+    bool        puballtables;
+
+    /* true if inserts are published */
+    bool        pubinsert;
+
+    /* true if updates are published */
+    bool        pubupdate;
+
+    /* true if deletes are published */
+    bool        pubdelete;
+
+} FormData_pg_publication;

Shouldn't this have an owner?   I also wonder if we want an easier to
extend form of pubinsert/update/delete (say to add pubddl, pubtruncate,
pub ... without changing the schema).


+/* ----------------
+ *        pg_publication_rel definition.  cpp turns this into
+ *        typedef struct FormData_pg_publication_rel
+ *
+ * ----------------
+ */
+#define PublicationRelRelationId                6106
+
+CATALOG(pg_publication_rel,6106)
+{
+    Oid        prpubid;                /* Oid of the publication */
+    Oid        prrelid;                /* Oid of the relation */
+} FormData_pg_publication_rel;

To me it seems like a good idea to have objclassid/objsubid here.


Regards,

Andres