Re: Logical Replication WIP

On 2016-12-10 08:48:55 +0100, Petr Jelinek wrote:

> diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
> new file mode 100644
> index 0000000..e3560b7
> --- /dev/null
> +++ b/src/backend/catalog/pg_publication.c
> +
> +Datum pg_get_publication_tables(PG_FUNCTION_ARGS);

Don't we usually put these in a header?

> +/*
> + * Insert new publication / relation mapping.
> + */
> +ObjectAddress
> +publication_add_relation(Oid pubid, Relation targetrel,
> + bool if_not_exists)
> +{
> + Relation rel;
> + HeapTuple tup;
> + Datum values[Natts_pg_publication_rel];
> + bool nulls[Natts_pg_publication_rel];
> + Oid relid = RelationGetRelid(targetrel);
> + Oid prrelid;
> + Publication *pub = GetPublication(pubid);
> + ObjectAddress myself,
> + referenced;
> +
> + rel = heap_open(PublicationRelRelationId, RowExclusiveLock);
> +
> + /* Check for duplicates */

Maybe mention that that check is racy, but a unique index protects
against the race?

> + /* Insert tuple into catalog. */
> + prrelid = simple_heap_insert(rel, tup);
> + CatalogUpdateIndexes(rel, tup);
> + heap_freetuple(tup);
> +
> + ObjectAddressSet(myself, PublicationRelRelationId, prrelid);
> +
> + /* Add dependency on the publication */
> + ObjectAddressSet(referenced, PublicationRelationId, pubid);
> + recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
> +
> + /* Add dependency on the relation */
> + ObjectAddressSet(referenced, RelationRelationId, relid);
> + recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
> +
> + /* Close the table. */
> + heap_close(rel, RowExclusiveLock);

I'm not quite sure abou the policy, but shouldn't we invoke
InvokeObjectPostCreateHook etc here?

> +/*
> + * Gets list of relation oids for a publication.
> + *
> + * This should only be used for normal publications, the FOR ALL TABLES
> + * should use GetAllTablesPublicationRelations().
> + */
> +List *
> +GetPublicationRelations(Oid pubid)
> +{
> + List *result;
> + Relation pubrelsrel;
> + ScanKeyData scankey;
> + SysScanDesc scan;
> + HeapTuple tup;
> +
> + /* Find all publications associated with the relation. */
> + pubrelsrel = heap_open(PublicationRelRelationId, AccessShareLock);
> +
> + ScanKeyInit(&scankey,
> + Anum_pg_publication_rel_prpubid,
> + BTEqualStrategyNumber, F_OIDEQ,
> + ObjectIdGetDatum(pubid));
> +
> + scan = systable_beginscan(pubrelsrel, PublicationRelMapIndexId, true,
> + NULL, 1, &scankey);
> +
> + result = NIL;
> + while (HeapTupleIsValid(tup = systable_getnext(scan)))
> + {
> + Form_pg_publication_rel pubrel;
> +
> + pubrel = (Form_pg_publication_rel) GETSTRUCT(tup);
> +
> + result = lappend_oid(result, pubrel->prrelid);
> + }
> +
> + systable_endscan(scan);
> + heap_close(pubrelsrel, NoLock);

In other parts of this you drop the lock, but not here?

> + heap_close(rel, NoLock);
> +
> + return result;
> +}

and here.

> +/*
> + * Gets list of all relation published by FOR ALL TABLES publication(s).
> + */
> +List *
> +GetAllTablesPublicationRelations(void)
> +{
> + Relation classRel;
> + ScanKeyData key[1];
> + HeapScanDesc scan;
> + HeapTuple tuple;
> + List *result = NIL;
> +
> + classRel = heap_open(RelationRelationId, AccessShareLock);

> + heap_endscan(scan);
> + heap_close(classRel, AccessShareLock);
> +
> + return result;
> +}

but here.

Btw, why are matviews not publishable?

> +/*
> + * Get Publication using name.
> + */
> +Publication *
> +GetPublicationByName(const char *pubname, bool missing_ok)
> +{
> + Oid oid;
> +
> + oid = GetSysCacheOid1(PUBLICATIONNAME, CStringGetDatum(pubname));
> + if (!OidIsValid(oid))
> + {
> + if (missing_ok)
> + return NULL;
> +
> + ereport(ERROR,
> + (errcode(ERRCODE_UNDEFINED_OBJECT),
> + errmsg("publication \"%s\" does not exist", pubname)));
> + }
> +
> + return GetPublication(oid);
> +}

That's racy... Also, shouldn't we specify for how to deal with the
returned memory for Publication * returning methods?

> diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
> new file mode 100644
> index 0000000..954b2bd
> --- /dev/null
> +++ b/src/backend/commands/publicationcmds.c
> @@ -0,0 +1,613 @@

> +/*
> + * Create new publication.
> + */
> +ObjectAddress
> +CreatePublication(CreatePublicationStmt *stmt)
> +{
> + Relation rel;

> +
> + values[Anum_pg_publication_puballtables - 1] =
> + BoolGetDatum(stmt->for_all_tables);
> + values[Anum_pg_publication_pubinsert - 1] =
> + BoolGetDatum(publish_insert);
> + values[Anum_pg_publication_pubupdate - 1] =
> + BoolGetDatum(publish_update);
> + values[Anum_pg_publication_pubdelete - 1] =
> + BoolGetDatum(publish_delete);

I remain convinced that a different representation would be
better. There'll be more options over time (truncate, DDL at least).

> +static void
> +AlterPublicationOptions(AlterPublicationStmt *stmt, Relation rel,
> + HeapTuple tup)
> +{
> + bool publish_insert_given;
> + bool publish_update_given;
> + bool publish_delete_given;
> + bool publish_insert;
> + bool publish_update;
> + bool publish_delete;
> + ObjectAddress obj;
> +
> + parse_publication_options(stmt->options,
> + &publish_insert_given, &publish_insert,
> + &publish_update_given, &publish_update,
> + &publish_delete_given, &publish_delete);

You could pass it a struct instead...

> +static List *
> +OpenTableList(List *tables)
> +{
> + List *relids = NIL;
> + List *rels = NIL;
> + ListCell *lc;
> +
> + /*
> + * Open, share-lock, and check all the explicitly-specified relations
> + */
> + foreach(lc, tables)
> + {
> + RangeVar *rv = lfirst(lc);
> + Relation rel;
> + bool recurse = interpretInhOption(rv->inhOpt);
> + Oid myrelid;
> +
> + rel = heap_openrv(rv, ShareUpdateExclusiveLock);
> + myrelid = RelationGetRelid(rel);
> + /* filter out duplicates when user specifies "foo, foo" */
> + if (list_member_oid(relids, myrelid))
> + {
> + heap_close(rel, ShareUpdateExclusiveLock);
> + continue;
> + }

This is a quadratic algorithm - that could bite us... Not sure if we
need to care. If we want to fix it, one approach owuld be to use
RangeVarGetRelid() instead, and then do a qsort/deduplicate before
actually opening the relations.

>
> -def_elem: ColLabel '=' def_arg
> +def_elem: def_key '=' def_arg
> {
> $$ = makeDefElem($1, (Node *) $3, @1);
> }
> - | ColLabel
> + | def_key
> {
> $$ = makeDefElem($1, NULL, @1);
> }
> ;

> +def_key:
> + ColLabel { $$ = $1; }
> + | ColLabel ColLabel { $$ = psprintf("%s %s", $1, $2); }
> + ;
> +

Not quite sure what this is about? Doesn't that change the accepted
syntax in a bunch of places?

> @@ -2337,6 +2338,8 @@ RelationDestroyRelation(Relation relation, bool remember_tupdesc)
> bms_free(relation->rd_indexattr);
> bms_free(relation->rd_keyattr);
> bms_free(relation->rd_idattr);
> + if (relation->rd_pubactions)
> + pfree(relation->rd_pubactions);
> if (relation->rd_options)
> pfree(relation->rd_options);
> if (relation->rd_indextuple)
> @@ -4992,6 +4995,67 @@ RelationGetExclusionInfo(Relation indexRelation,
> MemoryContextSwitchTo(oldcxt);
> }
>
> +/*
> + * Get publication actions for the given relation.
> + */
> +struct PublicationActions *
> +GetRelationPublicationActions(Relation relation)
> +{
> + List *puboids;
> + ListCell *lc;
> + MemoryContext oldcxt;
> + PublicationActions *pubactions = palloc0(sizeof(PublicationActions));
> +
> + if (relation->rd_pubactions)
> + return memcpy(pubactions, relation->rd_pubactions,
> + sizeof(PublicationActions));
> +
> + /* Fetch the publication membership info. */
> + puboids = GetRelationPublications(RelationGetRelid(relation));
> + puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
> +
> + foreach(lc, puboids)
> + {
> + Oid pubid = lfirst_oid(lc);
> + HeapTuple tup;
> + Form_pg_publication pubform;
> +
> + tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
> +
> + if (!HeapTupleIsValid(tup))
> + elog(ERROR, "cache lookup failed for publication %u", pubid);
> +
> + pubform = (Form_pg_publication) GETSTRUCT(tup);
> +
> + pubactions->pubinsert |= pubform->pubinsert;
> + pubactions->pubupdate |= pubform->pubupdate;
> + pubactions->pubdelete |= pubform->pubdelete;
> +
> + ReleaseSysCache(tup);
> +
> + /*
> + * If we know everything is replicated, there is no point to check
> + * for other publications.
> + */
> + if (pubactions->pubinsert && pubactions->pubupdate &&
> + pubactions->pubdelete)
> + break;
> + }
> +
> + if (relation->rd_pubactions)
> + {
> + pfree(relation->rd_pubactions);
> + relation->rd_pubactions = NULL;
> + }
> +
> + /* Now save copy of the actions in the relcache entry. */
> + oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
> + relation->rd_pubactions = palloc(sizeof(PublicationActions));
> + memcpy(relation->rd_pubactions, pubactions, sizeof(PublicationActions));
> + MemoryContextSwitchTo(oldcxt);
> +
> + return pubactions;
> +}

Hm. Do we actually have enough cache invalidation support to make this
cached version correct? I haven't seen anything in that regard? Seems
to mean that all changes to an ALL TABLES publication need to do a
global relcache invalidation?

- Andres