Re: Logical Replication WIP

(continuing, uh, a bit happier)

On 2016-09-09 00:59:26 +0200, Petr Jelinek wrote:

> +/*
> + * Relcache invalidation callback for our relation map cache.
> + */
> +static void
> +logicalreprelmap_invalidate_cb(Datum arg, Oid reloid)
> +{
> + LogicalRepRelMapEntry *entry;
> +
> + /* Just to be sure. */
> + if (LogicalRepRelMap == NULL)
> + return;
> +
> + if (reloid != InvalidOid)
> + {
> + HASH_SEQ_STATUS status;
> +
> + hash_seq_init(&status, LogicalRepRelMap);
> +
> + /* TODO, use inverse lookup hastable? */

*hashtable

> + while ((entry = (LogicalRepRelMapEntry *) hash_seq_search(&status)) != NULL)
> + {
> + if (entry->reloid == reloid)
> + entry->reloid = InvalidOid;

can't we break here?

> +/*
> + * Initialize the relation map cache.
> + */
> +static void
> +remoterelmap_init(void)
> +{
> + HASHCTL ctl;
> +
> + /* Make sure we've initialized CacheMemoryContext. */
> + if (CacheMemoryContext == NULL)
> + CreateCacheMemoryContext();
> +
> + /* Initialize the hash table. */
> + MemSet(&ctl, 0, sizeof(ctl));
> + ctl.keysize = sizeof(uint32);
> + ctl.entrysize = sizeof(LogicalRepRelMapEntry);
> + ctl.hcxt = CacheMemoryContext;

Wonder if this (and similar code earlier) should try to do everything in
a sub-context of CacheMemoryContext instead. That'd make some issues
easier to track down.

> +/*
> + * Open the local relation associated with the remote one.
> + */
> +static LogicalRepRelMapEntry *
> +logicalreprel_open(uint32 remoteid, LOCKMODE lockmode)
> +{
> + LogicalRepRelMapEntry *entry;
> + bool found;
> +
> + if (LogicalRepRelMap == NULL)
> + remoterelmap_init();
> +
> + /* Search for existing entry. */
> + entry = hash_search(LogicalRepRelMap, (void *) &remoteid,
> + HASH_FIND, &found);
> +
> + if (!found)
> + elog(FATAL, "cache lookup failed for remote relation %u",
> + remoteid);
> +
> + /* Need to update the local cache? */
> + if (!OidIsValid(entry->reloid))
> + {
> + Oid nspid;
> + Oid relid;
> + int i;
> + TupleDesc desc;
> + LogicalRepRelation *remoterel;
> +
> + remoterel = &entry->remoterel;
> +
> + nspid = LookupExplicitNamespace(remoterel->nspname, false);
> + if (!OidIsValid(nspid))
> + ereport(FATAL,
> + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
> + errmsg("the logical replication target %s not found",
> + quote_qualified_identifier(remoterel->nspname,
remoterel->relname))));
> + relid = get_relname_relid(remoterel->relname, nspid);
> + if (!OidIsValid(relid))
> + ereport(FATAL,
> + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
> + errmsg("the logical replication target %s not found",
> + quote_qualified_identifier(remoterel->nspname,
> + remoterel->relname))));
> +
> + entry->rel = heap_open(relid, lockmode);

This seems rather racy. I think this really instead needs something akin
to RangeVarGetRelidExtended().

> +/*
> + * Executor state preparation for evaluation of constraint expressions,
> + * indexes and triggers.
> + *
> + * This is based on similar code in copy.c
> + */
> +static EState *
> +create_estate_for_relation(LogicalRepRelMapEntry *rel)
> +{
> + EState *estate;
> + ResultRelInfo *resultRelInfo;
> + RangeTblEntry *rte;
> +
> + estate = CreateExecutorState();
> +
> + rte = makeNode(RangeTblEntry);
> + rte->rtekind = RTE_RELATION;
> + rte->relid = RelationGetRelid(rel->rel);
> + rte->relkind = rel->rel->rd_rel->relkind;
> + estate->es_range_table = list_make1(rte);
> +
> + resultRelInfo = makeNode(ResultRelInfo);
> + InitResultRelInfo(resultRelInfo, rel->rel, 1, 0);
> +
> + estate->es_result_relations = resultRelInfo;
> + estate->es_num_result_relations = 1;
> + estate->es_result_relation_info = resultRelInfo;
> +
> + /* Triggers might need a slot */
> + if (resultRelInfo->ri_TrigDesc)
> + estate->es_trig_tuple_slot = ExecInitExtraTupleSlot(estate);
> +
> + return estate;
> +}

Ugh, we do this for every single change? That's pretty darn heavy.

> +/*
> + * Check if the local attribute is present in relation definition used
> + * by upstream and hence updated by the replication.
> + */
> +static bool
> +physatt_in_attmap(LogicalRepRelMapEntry *rel, int attid)
> +{
> + AttrNumber i;
> +
> + /* Fast path for tables that are same on upstream and downstream. */
> + if (attid < rel->remoterel.natts && rel->attmap[attid] == attid)
> + return true;
> +
> + /* Try to find the attribute in the map. */
> + for (i = 0; i < rel->remoterel.natts; i++)
> + if (rel->attmap[i] == attid)
> + return true;
> +
> + return false;
> +}

Shouldn't we rather try to keep an attribute map that always can map
remote attribute numbers to local ones? That doesn't seem hard on a
first blush? But I might be missing something here.

> +/*
> + * Executes default values for columns for which we can't map to remote
> + * relation columns.
> + *
> + * This allows us to support tables which have more columns on the downstream
> + * than on the upsttream.
> + */

Typo: upsttream.

> +static void
> +FillSlotDefaults(LogicalRepRelMapEntry *rel, EState *estate,
> + TupleTableSlot *slot)
> +{

Why is this using a different naming scheme?

> +/*
> + * Handle COMMIT message.
> + *
> + * TODO, support tracking of multiple origins
> + */
> +static void
> +handle_commit(StringInfo s)
> +{
> + XLogRecPtr commit_lsn;
> + XLogRecPtr end_lsn;
> + TimestampTz commit_time;
> +
> + logicalrep_read_commit(s, &commit_lsn, &end_lsn, &commit_time);

Perhaps this (and related routines) should rather be
LogicalRepCommitdata commit_data;
logicalrep_read_commit(s, &commit_data);
etc? That way the data can transparently be enhanced.

> + Assert(commit_lsn == replorigin_session_origin_lsn);
> + Assert(commit_time == replorigin_session_origin_timestamp);
> +
> + if (IsTransactionState())
> + {
> + FlushPosition *flushpos;
> +
> + CommitTransactionCommand();
> + MemoryContextSwitchTo(CacheMemoryContext);
> +
> + /* Track commit lsn */
> + flushpos = (FlushPosition *) palloc(sizeof(FlushPosition));
> + flushpos->local_end = XactLastCommitEnd;
> + flushpos->remote_end = end_lsn;
> +
> + dlist_push_tail(&lsn_mapping, &flushpos->node);
> + MemoryContextSwitchTo(ApplyContext);

Seems like it should be in a separate function.

> +/*
> + * Handle INSERT message.
> + */
> +static void
> +handle_insert(StringInfo s)
> +{
> + LogicalRepRelMapEntry *rel;
> + LogicalRepTupleData newtup;
> + LogicalRepRelId relid;
> + EState *estate;
> + TupleTableSlot *remoteslot;
> + MemoryContext oldctx;
> +
> + ensure_transaction();
> +
> + relid = logicalrep_read_insert(s, &newtup);
> + rel = logicalreprel_open(relid, RowExclusiveLock);
> +
> + /* Initialize the executor state. */
> + estate = create_estate_for_relation(rel);
> + remoteslot = ExecInitExtraTupleSlot(estate);
> + ExecSetSlotDescriptor(remoteslot, RelationGetDescr(rel->rel));

This seems incredibly expensive for replicating a lot of rows.

> + /* Process and store remote tuple in the slot */
> + oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
> + SlotStoreCStrings(remoteslot, newtup.values);
> + FillSlotDefaults(rel, estate, remoteslot);
> + MemoryContextSwitchTo(oldctx);
> +
> + PushActiveSnapshot(GetTransactionSnapshot());
> + ExecOpenIndices(estate->es_result_relation_info, false);
> +
> + ExecInsert(NULL, /* mtstate is only used for onconflict handling which we don't support atm */
> + remoteslot,
> + remoteslot,
> + NIL,
> + ONCONFLICT_NONE,
> + estate,
> + false);

I have *severe* doubts about just using the (newly) exposed functions
1:1 here.

> +/*
> + * Search the relation 'rel' for tuple using the replication index.
> + *
> + * If a matching tuple is found lock it with lockmode, fill the slot with its
> + * contents and return true, return false is returned otherwise.
> + */
> +static bool
> +tuple_find_by_replidx(Relation rel, LockTupleMode lockmode,
> + TupleTableSlot *searchslot, TupleTableSlot *slot)
> +{
> + HeapTuple scantuple;
> + ScanKeyData skey[INDEX_MAX_KEYS];
> + IndexScanDesc scan;
> + SnapshotData snap;
> + TransactionId xwait;
> + Oid idxoid;
> + Relation idxrel;
> + bool found;
> +
> + /* Open REPLICA IDENTITY index.*/
> + idxoid = RelationGetReplicaIndex(rel);
> + if (!OidIsValid(idxoid))
> + {
> + elog(ERROR, "could not find configured replica identity for table \"%s\"",
> + RelationGetRelationName(rel));
> + return false;
> + }
> + idxrel = index_open(idxoid, RowExclusiveLock);
> +
> + /* Start an index scan. */
> + InitDirtySnapshot(snap);
> + scan = index_beginscan(rel, idxrel, &snap,
> + RelationGetNumberOfAttributes(idxrel),
> + 0);
> +
> + /* Build scan key. */
> + build_replindex_scan_key(skey, rel, idxrel, searchslot);
> +
> +retry:
> + found = false;
> +
> + index_rescan(scan, skey, RelationGetNumberOfAttributes(idxrel), NULL, 0);
> +
> + /* Try to find the tuple */
> + if ((scantuple = index_getnext(scan, ForwardScanDirection)) != NULL)
> + {
> + found = true;
> + ExecStoreTuple(scantuple, slot, InvalidBuffer, false);
> + ExecMaterializeSlot(slot);
> +
> + xwait = TransactionIdIsValid(snap.xmin) ?
> + snap.xmin : snap.xmax;
> +
> + /*
> + * If the tuple is locked, wait for locking transaction to finish
> + * and retry.
> + */
> + if (TransactionIdIsValid(xwait))
> + {
> + XactLockTableWait(xwait, NULL, NULL, XLTW_None);
> + goto retry;
> + }
> + }

Hm. So we potentially find multiple tuples here, and lock all of
them. but then only use one for the update.

> +static List *
> +get_subscription_list(void)
> +{
> + List *res = NIL;
> + Relation rel;
> + HeapScanDesc scan;
> + HeapTuple tup;
> + MemoryContext resultcxt;
> +
> + /* This is the context that we will allocate our output data in */
> + resultcxt = CurrentMemoryContext;
> +
> + /*
> + * Start a transaction so we can access pg_database, and get a snapshot.
> + * We don't have a use for the snapshot itself, but we're interested in
> + * the secondary effect that it sets RecentGlobalXmin. (This is critical
> + * for anything that reads heap pages, because HOT may decide to prune
> + * them even if the process doesn't attempt to modify any tuples.)
> + */

> + StartTransactionCommand();
> + (void) GetTransactionSnapshot();
> +
> + rel = heap_open(SubscriptionRelationId, AccessShareLock);
> + scan = heap_beginscan_catalog(rel, 0, NULL);
> +
> + while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
> + {
> + Form_pg_subscription subform = (Form_pg_subscription) GETSTRUCT(tup);
> + Subscription *sub;
> + MemoryContext oldcxt;
> +
> + /*
> + * Allocate our results in the caller's context, not the
> + * transaction's. We do this inside the loop, and restore the original
> + * context at the end, so that leaky things like heap_getnext() are
> + * not called in a potentially long-lived context.
> + */
> + oldcxt = MemoryContextSwitchTo(resultcxt);
> +
> + sub = (Subscription *) palloc(sizeof(Subscription));
> + sub->oid = HeapTupleGetOid(tup);
> + sub->dbid = subform->subdbid;
> + sub->enabled = subform->subenabled;
> +
> + /* We don't fill fields we are not intereste in. */
> + sub->name = NULL;
> + sub->conninfo = NULL;
> + sub->slotname = NULL;
> + sub->publications = NIL;
> +
> + res = lappend(res, sub);
> + MemoryContextSwitchTo(oldcxt);
> + }
> +
> + heap_endscan(scan);
> + heap_close(rel, AccessShareLock);
> +
> + CommitTransactionCommand();

Hm. this doesn't seem quite right from a locking pov. What if, in the
middle of this, a new subscription is created?

> +void
> +logicalrep_worker_stop(LogicalRepWorker *worker)
> +{
> + Assert(LWLockHeldByMe(LogicalRepWorkerLock));
> +
> + /* Check that the worker is up and what we expect. */
> + if (!worker->proc)
> + return;
> + if (!IsBackendPid(worker->proc->pid))
> + return;
> +
> + /* Terminate the worker. */
> + kill(worker->proc->pid, SIGTERM);
> +
> + LWLockRelease(LogicalRepLauncherLock);
> +
> + /* Wait for it to detach. */
> + for (;;)
> + {
> + int rc = WaitLatch(&MyProc->procLatch,
> + WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
> + 1000L);
> +
> + /* emergency bailout if postmaster has died */
> + if (rc & WL_POSTMASTER_DEATH)
> + proc_exit(1);
> +
> + ResetLatch(&MyProc->procLatch);
> +
> + CHECK_FOR_INTERRUPTS();
> +
> + if (!worker->proc)
> + return;
> + }
> +}

indentation here seems scfrewed.

> +static void
> +xacthook_signal_launcher(XactEvent event, void *arg)
> +{
> + switch (event)
> + {
> + case XACT_EVENT_COMMIT:
> + if (xacthook_do_signal_launcher)
> + ApplyLauncherWakeup();
> + break;
> + default:
> + /* We're not interested in other tx events */
> + break;
> + }
> +}

> +void
> +ApplyLauncherWakeupOnCommit(void)
> +{
> + if (!xacthook_do_signal_launcher)
> + {
> + RegisterXactCallback(xacthook_signal_launcher, NULL);
> + xacthook_do_signal_launcher = true;
> + }
> +}

Hm. This seems like it really should be an AtCommit_* routine instead.
This also needs more docs.

Hadn't I previously read about always streaming data to disk first?

> @@ -0,0 +1,674 @@
> +/*-------------------------------------------------------------------------
> + * tablesync.c
> + * PostgreSQL logical replication
> + *
> + * Copyright (c) 2012-2016, PostgreSQL Global Development Group
> + *
> + * IDENTIFICATION
> + * src/backend/replication/logical/tablesync.c
> + *
> + * NOTES
> + * This file contains code for initial table data synchronization for
> + * logical replication.
> + *
> + * The initial data synchronization is done separately for each table,
> + * in separate apply worker that only fetches the initial snapshot data
> + * from the provider and then synchronizes the position in stream with
> + * the main apply worker.

Why? I guess that's because it allows to incrementally add tables, with
acceptable overhead.

> + * The stream position synchronization works in multiple steps.
> + * - sync finishes copy and sets table state as SYNCWAIT and waits
> + * for state to change in a loop
> + * - apply periodically checks unsynced tables for SYNCWAIT, when it
> + * appears it will compare its position in the stream with the
> + * SYNCWAIT position and decides to either set it to CATCHUP when
> + * the apply was infront (and wait for the sync to do the catchup),
> + * or set the state to SYNCDONE if the sync was infront or in case
> + * both sync and apply are at the same position it will set it to
> + * READY and stops tracking it

I'm not quite following here.

> + * - if the state was set to CATCHUP sync will read the stream and
> + * apply changes until it catches up to the specified stream
> + * position and then sets state to READY and signals apply that it
> + * can stop waiting and exits, if the state was set to something
> + * else than CATCHUP the sync process will simply end
> + * - if the state was set to SYNCDONE by apply, the apply will
> + * continue tracking the table until it reaches the SYNCDONE stream
> + * position at which point it sets state to READY and stops tracking
> + *
> + * Example flows look like this:
> + * - Apply is infront:
> + * sync:8 -> set SYNCWAIT
> + * apply:10 -> set CATCHUP
> + * sync:10 -> set ready
> + * exit
> + * apply:10
> + * stop tracking
> + * continue rep
> + * - Sync infront:
> + * sync:10
> + * set SYNCWAIT
> + * apply:8
> + * set SYNCDONE
> + * sync:10
> + * exit
> + * apply:10
> + * set READY
> + * stop tracking
> + * continue rep

This definitely needs to be expanded a bit. Where are we tracking how
far replication has progressed on individual tables? Are we creating new
slots for syncing? Is there any parallelism in syncing?

> +/*
> + * Exit routine for synchronization worker.
> + */
> +static void
> +finish_sync_worker(char *slotname)
> +{
> + LogicalRepWorker *worker;
> + RepOriginId originid;
> + MemoryContext oldctx = CurrentMemoryContext;
> +
> + /*
> + * Drop the replication slot on remote server.
> + * We want to continue even in the case that the slot on remote side
> + * is already gone. This means that we can leave slot on the remote
> + * side but that can happen for other reasons as well so we can't
> + * really protect against that.
> + */
> + PG_TRY();
> + {
> + wrcapi->drop_slot(wrchandle, slotname);
> + }
> + PG_CATCH();
> + {
> + MemoryContext ectx;
> + ErrorData *edata;
> +
> + ectx = MemoryContextSwitchTo(oldctx);
> + /* Save error info */
> + edata = CopyErrorData();
> + MemoryContextSwitchTo(ectx);
> + FlushErrorState();
> +
> + ereport(WARNING,
> + (errmsg("there was problem dropping the replication slot "
> + "\"%s\" on provider", slotname),
> + errdetail("The error was: %s", edata->message),
> + errhint("You may have to drop it manually")));
> + FreeErrorData(edata);

ISTM we really should rather return success/failure here, and not throw
an error inside the libpqwalreceiver stuff. I kind of wonder if we
actually can get rid of this indirection.

> + /* Find the main apply worker and signal it. */
> + LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
> + worker = logicalrep_worker_find(MyLogicalRepWorker->subid, InvalidOid);
> + if (worker && worker->proc)
> + SetLatch(&worker->proc->procLatch);
> + LWLockRelease(LogicalRepWorkerLock);

I'd rather do the SetLatch outside of the critical section.

> +static bool
> +wait_for_sync_status_change(TableState *tstate)
> +{
> + int rc;
> + char state = tstate->state;
> +
> + while (!got_SIGTERM)
> + {
> + StartTransactionCommand();
> + tstate->state = GetSubscriptionRelState(MyLogicalRepWorker->subid,
> + tstate->relid,
> + &tstate->lsn,
> + true);
> + CommitTransactionCommand();
> +
> + /* Status record was removed. */
> + if (tstate->state == SUBREL_STATE_UNKNOWN)
> + return false;
> +
> + if (tstate->state != state)
> + return true;
> +
> + rc = WaitLatch(&MyProc->procLatch,
> + WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
> + 10000L);
> +
> + /* emergency bailout if postmaster has died */
> + if (rc & WL_POSTMASTER_DEATH)
> + proc_exit(1);
> +
> + ResetLatch(&MyProc->procLatch);

broken indentation.

> +/*
> + * Read the state of the tables in the subscription and update our table
> + * state list.
> + */
> +static void
> +reread_sync_state(Oid relid)
> +{
> + dlist_mutable_iter iter;
> + Relation rel;
> + HeapTuple tup;
> + ScanKeyData skey[2];
> + HeapScanDesc scan;
> +
> + /* Clean the old list. */
> + dlist_foreach_modify(iter, &table_states)
> + {
> + TableState *tstate = dlist_container(TableState, node, iter.cur);
> +
> + dlist_delete(iter.cur);
> + pfree(tstate);
> + }
> +
> + /*
> + * Fetch all the subscription relation states that are not marked as
> + * ready and push them into our table state tracking list.
> + */
> + rel = heap_open(SubscriptionRelRelationId, RowExclusiveLock);
> +
> + ScanKeyInit(&skey[0],
> + Anum_pg_subscription_rel_subid,
> + BTEqualStrategyNumber, F_OIDEQ,
> + ObjectIdGetDatum(MyLogicalRepWorker->subid));
> +
> + if (OidIsValid(relid))
> + {
> + ScanKeyInit(&skey[1],
> + Anum_pg_subscription_rel_subrelid,
> + BTEqualStrategyNumber, F_OIDEQ,
> + ObjectIdGetDatum(relid));
> + }
> + else
> + {
> + ScanKeyInit(&skey[1],
> + Anum_pg_subscription_rel_substate,
> + BTEqualStrategyNumber, F_CHARNE,
> + CharGetDatum(SUBREL_STATE_READY));
> + }
> +
> + scan = heap_beginscan_catalog(rel, 2, skey);

Hm. So this is a seqscan. Shouldn't we make this use an index (depending
on which branch is taken above)?

> + while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
> + {
> + Form_pg_subscription_rel subrel;
> + TableState *tstate;
> + MemoryContext oldctx;
> +
> + subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
> +
> + /* Allocate the tracking info in a permament memory context. */

s/permament/permanent/

> +/*
> + * Handle table synchronization cooperation from the synchroniation
> + * worker.
> + */
> +static void
> +process_syncing_tables_sync(char *slotname, XLogRecPtr end_lsn)
> +{
> + TableState *tstate;
> + TimeLineID tli;
> +
> + Assert(!IsTransactionState());
> +
> + /*
> + * Synchronization workers don't keep track of all synchronization
> + * tables, they only care about their table.
> + */
> + if (!table_states_valid)
> + {
> + StartTransactionCommand();
> + reread_sync_state(MyLogicalRepWorker->relid);
> + CommitTransactionCommand();
> + }
> +
> + /* Somebody removed table underneath this worker, nothing more to do. */
> + if (dlist_is_empty(&table_states))
> + {
> + wrcapi->endstreaming(wrchandle, &tli);
> + finish_sync_worker(slotname);
> + }
> +
> + /* Check if we are done with catchup now. */
> + tstate = dlist_container(TableState, node, dlist_head_node(&table_states));
> + if (tstate->state == SUBREL_STATE_CATCHUP)
> + {
> + Assert(tstate->lsn != InvalidXLogRecPtr);
> +
> + if (tstate->lsn == end_lsn)
> + {
> + tstate->state = SUBREL_STATE_READY;
> + tstate->lsn = InvalidXLogRecPtr;
> + /* Update state of the synchronization. */
> + StartTransactionCommand();
> + SetSubscriptionRelState(MyLogicalRepWorker->subid,
> + tstate->relid, tstate->state,
> + tstate->lsn);
> + CommitTransactionCommand();
> +
> + wrcapi->endstreaming(wrchandle, &tli);
> + finish_sync_worker(slotname);
> + }
> + return;
> + }
> +}

The return inside the if is a bit weird. Makes one think it might be a
loop or such.

> +/*
> + * Handle table synchronization cooperation from the apply worker.
> + */
> +static void
> +process_syncing_tables_apply(char *slotname, XLogRecPtr end_lsn)
> +{
> + dlist_mutable_iter iter;
> +
> + Assert(!IsTransactionState());
> +
> + if (!table_states_valid)
> + {
> + StartTransactionCommand();
> + reread_sync_state(InvalidOid);
> + CommitTransactionCommand();
> + }

So this pattern is repeated a bunch of times, maybe we can encapsulate
that somewhat? Maybe like ensure_sync_state_valid() or such?

> + dlist_foreach_modify(iter, &table_states)
> + {
> + TableState *tstate = dlist_container(TableState, node, iter.cur);
> + bool start_worker;
> + LogicalRepWorker *worker;
> +
> + /*
> + * When the synchronization process is at the cachup phase we need

s/cachup/catchup/

> + * to ensure that we are not behind it (it's going to wait at this
> + * point for the change of state). Once we are infront or at the same
> + * position as the synchronization proccess we can signal it to
> + * finish the catchup.
> + */
> + if (tstate->state == SUBREL_STATE_SYNCWAIT)
> + {
> + if (end_lsn > tstate->lsn)
> + {
> + /*
> + * Apply is infront, tell sync to catchup. and wait until
> + * it does.
> + */
> + tstate->state = SUBREL_STATE_CATCHUP;
> + tstate->lsn = end_lsn;
> + StartTransactionCommand();
> + SetSubscriptionRelState(MyLogicalRepWorker->subid,
> + tstate->relid, tstate->state,
> + tstate->lsn);
> + CommitTransactionCommand();
> +
> + /* Signal the worker as it may be waiting for us. */
> + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
> + worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
> + tstate->relid);
> + if (worker && worker->proc)
> + SetLatch(&worker->proc->procLatch);
> + LWLockRelease(LogicalRepWorkerLock);

Different parts of this file use different lock level to set the
latch. Why?

> + if (wait_for_sync_status_change(tstate))
> + Assert(tstate->state == SUBREL_STATE_READY);
> + }
> + else
> + {
> + /*
> + * Apply is either behind in which case sync worker is done
> + * but apply needs to keep tracking the table until it
> + * catches up to where sync finished.
> + * Or apply and sync are at the same position in which case
> + * table can be switched to standard replication mode
> + * immediately.
> + */
> + if (end_lsn < tstate->lsn)
> + tstate->state = SUBREL_STATE_SYNCDONE;
> + else
> + tstate->state = SUBREL_STATE_READY;
> +

What I'm failing to understand is how this can be done under
concurrency. You probably thought about this, but it should really be
explained somewhere.

> + StartTransactionCommand();
> + SetSubscriptionRelState(MyLogicalRepWorker->subid,
> + tstate->relid, tstate->state,
> + tstate->lsn);
> + CommitTransactionCommand();
> +
> + /* Signal the worker as it may be waiting for us. */
> + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
> + worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
> + tstate->relid);
> + if (worker && worker->proc)
> + SetLatch(&worker->proc->procLatch);
> + LWLockRelease(LogicalRepWorkerLock);

Oh, and again, please set latches outside of the lock.

> + else if (tstate->state == SUBREL_STATE_SYNCDONE &&
> + end_lsn >= tstate->lsn)
> + {
> + /*
> + * Apply catched up to the position where table sync finished,
> + * mark the table as ready for normal replication.
> + */

Sentence needs to be rephrased a bit.

> + /*
> + * In case table is supposed to be synchronizing but the
> + * synchronization worker is not running, start it.
> + * Limit the number of launched workers here to one (for now).
> + */

Hm. That seems problematic for online upgrade type cases, we might never
be catch up that way...

> +/*
> + * Start syncing the table in the sync worker.
> + */
> +char *
> +LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
> +{
> + StringInfoData s;
> + TableState tstate;
> + MemoryContext oldctx;
> + char *slotname;
> +
> + /* Check the state of the table synchronization. */
> + StartTransactionCommand();
> + tstate.relid = MyLogicalRepWorker->relid;
> + tstate.state = GetSubscriptionRelState(MySubscription->oid, tstate.relid,
> + &tstate.lsn, false);
> +
> + /*
> + * Build unique slot name.
> + * TODO: protect against too long slot name.
> + */
> + oldctx = MemoryContextSwitchTo(CacheMemoryContext);
> + initStringInfo(&s);
> + appendStringInfo(&s, "%s_sync_%s", MySubscription->slotname,
> + get_rel_name(tstate.relid));
> + slotname = s.data;

Is this memory freed somewhere?

> + /*
> + * We want to do the table data sync in single
> + * transaction so do not close the transaction opened
> + * above.
> + * There will be no BEGIN or COMMIT messages coming via
> + * logical replication while the copy table command is
> + * running so start the transaction here.
> + * Note the memory context for data handling will still
> + * be done using ensure_transaction called by the insert
> + * handler.
> + */
> + StartTransactionCommand();
> +
> + /*
> + * Don't allow parallel access other than SELECT while
> + * the initial contents are being copied.
> + */
> + rel = heap_open(tstate.relid, ExclusiveLock);

Why do we want to allow access at all?

> @@ -87,6 +92,8 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
> cb->commit_cb = pgoutput_commit_txn;
> cb->filter_by_origin_cb = pgoutput_origin_filter;
> cb->shutdown_cb = pgoutput_shutdown;
> + cb->tuple_cb = pgoutput_tuple;
> + cb->list_tables_cb = pgoutput_list_tables;
> }

What are these new, and undocumented callbacks actually doing? And why
is this integrated into logical decoding?

> /*
> + * Handle LIST_TABLES command.
> + */
> +static void
> +SendTableList(ListTablesCmd *cmd)
> +{

Ugh.

I really dislike this kind of command. I think we should instead change
things around, allowing to issue normal SQL via the replication
command. We'll have to error out for running sql for non-database
connected replication connections, but that seems fine.

Andres