diff options
Diffstat (limited to 'src/backend/replication/logical/tablesync.c')
| -rw-r--r-- | src/backend/replication/logical/tablesync.c | 18 |
1 files changed, 14 insertions, 4 deletions
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 0823000f001..7e51076b376 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -274,6 +274,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) static List *table_states = NIL; static HTAB *last_start_times = NULL; ListCell *lc; + bool started_tx = false; Assert(!IsTransactionState()); @@ -290,6 +291,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) table_states = NIL; StartTransactionCommand(); + started_tx = true; /* Fetch all non-ready tables. */ rstates = GetSubscriptionNotReadyRelations(MySubscription->oid); @@ -304,8 +306,6 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) } MemoryContextSwitchTo(oldctx); - CommitTransactionCommand(); - table_states_valid = true; } @@ -350,11 +350,14 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) { rstate->state = SUBREL_STATE_READY; rstate->lsn = current_lsn; - StartTransactionCommand(); + if (!started_tx) + { + StartTransactionCommand(); + started_tx = true; + } SetSubscriptionRelState(MyLogicalRepWorker->subid, rstate->relid, rstate->state, rstate->lsn); - CommitTransactionCommand(); } } else @@ -457,6 +460,12 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) } } } + + if (started_tx) + { + CommitTransactionCommand(); + pgstat_report_stat(false); + } } /* @@ -806,6 +815,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos) MyLogicalRepWorker->relstate, MyLogicalRepWorker->relstate_lsn); CommitTransactionCommand(); + pgstat_report_stat(false); /* * We want to do the table data sync in single |