Skip to content

Commit 4563ca7

Browse files
vbwagnervbwagner
committed
Merge branch 'PGPROEE9_6_MULTIMASTER' into PGPROEE9_6
2 parents f28c2b3 + 39b73d0 commit 4563ca7

File tree

2 files changed

+37
-29
lines changed

2 files changed

+37
-29
lines changed

‎contrib/mmts/multimaster--1.0.sql

Lines changed: 6 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,7 @@ CREATE TABLE IF NOT EXISTS mtm.local_tables(rel_schema text, rel_name text, prim
8787
CREATE OR REPLACE FUNCTION mtm.alter_sequences() RETURNS boolean AS
8888
$$
8989
DECLARE
90-
seq_class pg_class%rowtype;
90+
seq_class record;
9191
seq_tuple record;
9292
node_id int;
9393
max_nodes int;
@@ -97,25 +97,20 @@ BEGIN
9797
select current_setting('multimaster.max_nodes') into max_nodes;
9898
select id, "allNodes" into node_id from mtm.get_cluster_state();
9999
FOR seq_class IN
100-
SELECT * FROM pg_class WHERE pg_class.relkind = 'S'
100+
SELECT '"' || ns.nspname || '"."' || seq.relname || '"' as seqname FROM pg_namespace ns,pg_class seq WHERE seq.relkind = 'S' and seq.relnamespace=ns.oid
101101
LOOP
102-
BEGIN
103-
EXECUTE 'select * from ' || seq_class.relname || ';' INTO seq_tuple;
102+
EXECUTE 'select * from ' || seq_class.seqname INTO seq_tuple;
104103
IF seq_tuple.increment_by != max_nodes THEN
105104
altered := true;
106105
RAISE NOTICE 'Altering step for sequence % to %.', seq_tuple.sequence_name, max_nodes;
107-
EXECUTE 'ALTER SEQUENCE ' || seq_class.relname || ' INCREMENT BY ' || max_nodes || ';';
106+
EXECUTE 'ALTER SEQUENCE ' || seq_class.seqname || ' INCREMENT BY ' || max_nodes || ';';
108107
END IF;
109108
IF (seq_tuple.last_value % max_nodes) != node_id THEN
110109
altered := true;
111110
new_start := (seq_tuple.last_value / max_nodes + 1)*max_nodes + node_id;
112111
RAISE NOTICE 'Altering start for sequence % to %.', seq_tuple.sequence_name, new_start;
113-
EXECUTE 'ALTER SEQUENCE ' || seq_class.relname || ' RESTART WITH ' || new_start || ';';
112+
EXECUTE 'ALTER SEQUENCE ' || seq_class.seqname || ' RESTART WITH ' || new_start || ';';
114113
END IF;
115-
EXCEPTION
116-
WHEN OTHERS THEN
117-
RAISE NOTICE 'Failed to alter sequence %s', seq_class.relname;
118-
END;
119114
END LOOP;
120115
IF altered = false THEN
121116
RAISE NOTICE 'All found sequnces have proper params.';
@@ -125,4 +120,4 @@ END
125120
$$
126121
LANGUAGE plpgsql;
127122

128-
select mtm.alter_sequences();
123+
-- select mtm.alter_sequences();

‎contrib/mmts/multimaster.c

Lines changed: 31 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,7 @@
6262
#include "access/htup_details.h"
6363
#include "catalog/indexing.h"
6464
#include "catalog/namespace.h"
65+
#include "catalog/pg_constraint_fn.h"
6566
#include "pglogical_output/hooks.h"
6667
#include "parser/analyze.h"
6768
#include "parser/parse_relation.h"
@@ -2143,7 +2144,9 @@ static void
21432144
MtmLockCluster(void)
21442145
{
21452146
timestamp_t delay = MIN_WAIT_TIMEOUT;
2146-
Assert(!MtmClusterLocked);
2147+
if (MtmClusterLocked) {
2148+
MtmUnlockCluster();
2149+
}
21472150
MtmLock(LW_EXCLUSIVE);
21482151
if (BIT_CHECK(Mtm->originLockNodeMask, MtmNodeId-1)) {
21492152
elog(ERROR, "There is already pending exclusive lock");
@@ -4876,8 +4879,8 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
48764879
bool skipCommand = false;
48774880
bool executed = false;
48784881

4879-
MTM_LOG3("%d: Process utility statement tag=%d, context=%d, issubtrans=%d, query=%s",
4880-
MyProcPid, nodeTag(parsetree), context, IsSubTransaction(), queryString);
4882+
MTM_LOG2("%d: Process utility statement tag=%d, context=%d, issubtrans=%d, creating_extension=%d, query=%s",
4883+
MyProcPid, nodeTag(parsetree), context, IsSubTransaction(), creating_extension, queryString);
48814884
switch (nodeTag(parsetree))
48824885
{
48834886
case T_TransactionStmt:
@@ -5130,24 +5133,14 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
51305133
break;
51315134
}
51325135

5133-
/* XXX: dirty. Clear on new tx */
5134-
/* Some "black magic here":( We want to avoid redundant execution of utility statement by ProcessUtilitySlow (which is done with PROCESS_UTILITY_SUBCOMMAND).
5135-
* But if we allow only PROCESS_UTILITY_TOPLEVEL context, then we will not replicated DDL inside dynamic queries in plpgsql functions (see https://jira.postgrespro.ru/browse/CORE-526).
5136-
* If we disable only PROCESS_UTILITY_SUBCOMMAND, then we will get problems with "create extension" which is executed also in PROCESS_UTILITY_QUERY context.
5137-
* So workaround at this moment is to treat extension as special case.
5138-
* TODO: try to find right solution and rewrite this dummy check.
5139-
*/
5140-
if (!skipCommand && (context == PROCESS_UTILITY_TOPLEVEL || (context == PROCESS_UTILITY_QUERY && !creating_extension) || MtmUtilityProcessedInXid != GetCurrentTransactionId()))
5141-
MtmUtilityProcessedInXid = InvalidTransactionId;
5142-
5143-
if (!skipCommand && !MtmTx.isReplicated && (MtmUtilityProcessedInXid == InvalidTransactionId)) {
5136+
if (!skipCommand && !MtmTx.isReplicated && (context == PROCESS_UTILITY_TOPLEVEL || MtmUtilityProcessedInXid != GetCurrentTransactionId()))
5137+
{
51445138
MtmUtilityProcessedInXid = GetCurrentTransactionId();
5145-
5146-
if (context == PROCESS_UTILITY_TOPLEVEL)
5139+
if (context == PROCESS_UTILITY_TOPLEVEL) {
51475140
MtmProcessDDLCommand(queryString, true);
5148-
else
5141+
} else {
51495142
MtmProcessDDLCommand(ActivePortal->sourceText, true);
5150-
5143+
}
51515144
executed = true;
51525145
}
51535146

@@ -5177,6 +5170,26 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
51775170
{
51785171
MtmFinishDDLCommand();
51795172
}
5173+
if (nodeTag(parsetree) == T_CreateStmt)
5174+
{
5175+
CreateStmt* create = (CreateStmt*)parsetree;
5176+
Oid relid = RangeVarGetRelid(create->relation, NoLock, true);
5177+
if (relid != InvalidOid) {
5178+
Oid constraint_oid;
5179+
Bitmapset* pk = get_primary_key_attnos(relid, true, &constraint_oid);
5180+
if (pk == NULL && !MtmVolksWagenMode) {
5181+
elog(WARNING,
5182+
MtmIgnoreTablesWithoutPk
5183+
? "Table %s.%s without primary will not be replicated"
5184+
: "Updates and deletes of table %s.%s without primary will not be replicated",
5185+
create->relation->schemaname ? create->relation->schemaname : "public",
5186+
create->relation->relname);
5187+
}
5188+
}
5189+
}
5190+
if (context == PROCESS_UTILITY_TOPLEVEL) {
5191+
MtmUtilityProcessedInXid = InvalidTransactionId;
5192+
}
51805193
}
51815194

51825195
static void

0 commit comments

Comments
 (0)