database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-06-26 12:21:12 +03:00
Code Activity

Fix statistics reporting in logical replication workers

Browse Source 
This new arrangement ensures that statistics are reported right after
commit of transactions.  The previous arrangement didn't get this quite
right and could lead to assertion failures.

Author: Petr Jelinek <petr.jelinek@2ndquadrant.com>
Reported-by: Erik Rijkers <er@xs4all.nl>
This commit is contained in:
Peter Eisentraut
2017-05-08 12:07:59 -04:00
parent b6576e5914
commit 9a591c1bcc
2 changed files with 15 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/backend/replication/logical/tablesync.c
View File

@ -274,6 +274,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
static List *table_states = NIL; static List *table_states = NIL;
static HTAB *last_start_times = NULL; static HTAB *last_start_times = NULL;
ListCell *lc; ListCell *lc;
bool started_tx = false;
Assert(!IsTransactionState()); Assert(!IsTransactionState());
@ -290,6 +291,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
table_states = NIL; table_states = NIL;
StartTransactionCommand(); StartTransactionCommand();
started_tx = true;
/* Fetch all non-ready tables. */ /* Fetch all non-ready tables. */
rstates = GetSubscriptionNotReadyRelations(MySubscription->oid); rstates = GetSubscriptionNotReadyRelations(MySubscription->oid);
@ -304,8 +306,6 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
} }
MemoryContextSwitchTo(oldctx); MemoryContextSwitchTo(oldctx);
CommitTransactionCommand();
table_states_valid = true; table_states_valid = true;
} }
@ -350,11 +350,14 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
{ {
rstate->state = SUBREL_STATE_READY; rstate->state = SUBREL_STATE_READY;
rstate->lsn = current_lsn; rstate->lsn = current_lsn;
StartTransactionCommand(); if (!started_tx)
{
StartTransactionCommand();
started_tx = true;
}
SetSubscriptionRelState(MyLogicalRepWorker->subid, SetSubscriptionRelState(MyLogicalRepWorker->subid,
rstate->relid, rstate->state, rstate->relid, rstate->state,
rstate->lsn); rstate->lsn);
CommitTransactionCommand();
} }
} }
else 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,
MyLogicalRepWorker->relstate_lsn); MyLogicalRepWorker->relstate_lsn);
CommitTransactionCommand(); CommitTransactionCommand();
pgstat_report_stat(false);
/* /*
* We want to do the table data sync in single * We want to do the table data sync in single

2
src/backend/replication/logical/worker.c
View File

@ -453,6 +453,7 @@ apply_handle_commit(StringInfo s)
replorigin_session_origin_timestamp = commit_data.committime; replorigin_session_origin_timestamp = commit_data.committime;
CommitTransactionCommand(); CommitTransactionCommand();
pgstat_report_stat(false);
store_flush_position(commit_data.end_lsn); store_flush_position(commit_data.end_lsn);
} }
@ -462,7 +463,6 @@ apply_handle_commit(StringInfo s)
/* Process any tables that are being synchronized in parallel. */ /* Process any tables that are being synchronized in parallel. */
process_syncing_tables(commit_data.end_lsn); process_syncing_tables(commit_data.end_lsn);
pgstat_report_stat(false);
pgstat_report_activity(STATE_IDLE, NULL); pgstat_report_activity(STATE_IDLE, NULL);
} }