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