From c277577e33927e4eb62c334d87be3f16804257a6 Mon Sep 17 00:00:00 2001 From: Peter Smith Date: Wed, 2 Aug 2023 12:18:01 +1000 Subject: [PATCH v2] Switch on worker type --- src/backend/replication/logical/tablesync.c | 32 ++++++++++++++-------- src/backend/replication/logical/worker.c | 42 +++++++++++++++++------------ 2 files changed, 46 insertions(+), 28 deletions(-) diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index e359286..98ebbc9 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -647,18 +647,28 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) void process_syncing_tables(XLogRecPtr current_lsn) { - /* - * Skip for parallel apply workers because they only operate on tables - * that are in a READY state. See pa_can_start() and - * should_apply_changes_for_rel(). - */ - if (IsParallelApplyWorker()) - return; + switch (MyLogicalRepWorker->type) + { + case TYPE_TABLESYNC_WORKER: + process_syncing_tables_for_sync(current_lsn); + break; - if (IsTablesyncWorker()) - process_syncing_tables_for_sync(current_lsn); - else - process_syncing_tables_for_apply(current_lsn); + case TYPE_APPLY_WORKER: + process_syncing_tables_for_apply(current_lsn); + break; + + case TYPE_PARALLEL_APPLY_WORKER: + /* + * Skip for parallel apply workers because they only operate on tables + * that are in a READY state. See pa_can_start() and + * should_apply_changes_for_rel(). + */ + break; + + case TYPE_UNKNOWN: + Assert(false); + break; + } } /* diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index d165db6..dabfa48 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -486,25 +486,33 @@ ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid, static bool should_apply_changes_for_rel(LogicalRepRelMapEntry *rel) { - if (IsTablesyncWorker()) - return MyLogicalRepWorker->relid == rel->localreloid; - else if (IsParallelApplyWorker()) + switch (MyLogicalRepWorker->type) { - /* We don't synchronize rel's that are in unknown state. */ - if (rel->state != SUBREL_STATE_READY && - rel->state != SUBREL_STATE_UNKNOWN) - ereport(ERROR, - (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("logical replication parallel apply worker for subscription \"%s\" will stop", - MySubscription->name), - errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized."))); - - return rel->state == SUBREL_STATE_READY; + case TYPE_TABLESYNC_WORKER: + return MyLogicalRepWorker->relid == rel->localreloid; + + case TYPE_APPLY_WORKER: + return (rel->state == SUBREL_STATE_READY || + (rel->state == SUBREL_STATE_SYNCDONE && + rel->statelsn <= remote_final_lsn)); + + case TYPE_PARALLEL_APPLY_WORKER: + /* We don't synchronize rel's that are in unknown state. */ + if (rel->state != SUBREL_STATE_READY && + rel->state != SUBREL_STATE_UNKNOWN) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("logical replication parallel apply worker for subscription \"%s\" will stop", + MySubscription->name), + errdetail("Cannot handle streamed replication transactions using parallel apply workers until all tables have been synchronized."))); + return rel->state == SUBREL_STATE_READY; + + case TYPE_UNKNOWN: + Assert(false); + return false; } - else - return (rel->state == SUBREL_STATE_READY || - (rel->state == SUBREL_STATE_SYNCDONE && - rel->statelsn <= remote_final_lsn)); + + return false; /* dummy for compiler */ } /* -- 1.8.3.1