Merge branch 'PGPROEE10_MULTIMASTER_stable' into PGPROEE10_MULTIMASTER

Author: kelvich

configure

@@ -2788,7 +2788,7 @@ fi
 
 
 
-PGPRO_VERSION="$PACKAGE_VERSION.1"
+PGPRO_VERSION="$PACKAGE_VERSION.3"
 PGPRO_PACKAGE_NAME="PostgresPro"
 PGPRO_EDITION="enterprise"
 PGPRO_EDN="ent"

configure.in

@@ -39,7 +39,7 @@ PGAC_ARG_REQ(with, extra-version, [STRING], [append STRING to version],
              [PG_VERSION="$PACKAGE_VERSION$withval"],
              [PG_VERSION="$PACKAGE_VERSION"])
 
-PGPRO_VERSION="$PACKAGE_VERSION.1"
+PGPRO_VERSION="$PACKAGE_VERSION.3"
 PGPRO_PACKAGE_NAME="PostgresPro"
 PGPRO_EDITION="enterprise"
 PGPRO_EDN="ent"

contrib/mchar/Makefile

@@ -4,7 +4,10 @@ OBJS = mchar_io.o mchar_proc.o mchar_op.o mchar_recode.o \
 EXTENSION = mchar
 DATA = mchar--1.0.sql
 DOCS = README.mchar
-REGRESS = init mchar mvarchar mm like compat
+REGRESS = init mchar mvarchar like
+ifndef NO_LOCALE
+REGRESS += mm compat
+endif
 PGFIELDDESC = "mchar - mchar type implementation"
 
 ifdef USE_PGXS

contrib/mchar/expected/mm.out

@@ -803,3 +803,44 @@ SELECT * FROM a, c WHERE mvarchar255 = mchar2;
 
 DROP TABLE a;
 DROP TABLE c;
+select * from (values ('е'::mchar),('ё'),('еа'),('еб'),('ее'),('еж'),('ёа'),('ёб'),('ёё'),('ёж'),('ёе'),('её')) z order by 1;
+ column1 
+---------
+ е
+ еа
+ еб
+ ее
+ её
+ еж
+ ё
+ ёа
+ ёб
+ ёе
+ ёё
+ ёж
+(12 rows)
+
+select mvarchar_icase_cmp('ё', 'е');
+ mvarchar_icase_cmp 
+--------------------
+                  1
+(1 row)
+
+select mvarchar_icase_cmp('Ё', 'Е');
+ mvarchar_icase_cmp 
+--------------------
+                  1
+(1 row)
+
+select mvarchar_icase_cmp('й', 'и');
+ mvarchar_icase_cmp 
+--------------------
+                  1
+(1 row)
+
+select mvarchar_icase_cmp('Й', 'И');
+ mvarchar_icase_cmp 
+--------------------
+                  1
+(1 row)
+

contrib/mchar/mchar_recode.c

@@ -31,6 +31,7 @@ createUObjs() {
 			elog(ERROR,"ICU ucol_open returns %d (%s)", err,  u_errorName(err));
 		}
 
+		/* UCOL_PRIMARY doesn't distinguish И & Й, Е & Ё */
 		ucol_setStrength( colCaseInsensitive, UCOL_SECONDARY );
 	}
 
@@ -114,29 +115,49 @@ FillWhiteSpace( UChar *dst, int n ) {
 int 
 UCharCaseCompare(UChar * a, int alen, UChar *b, int blen) {
 	int len = Min(alen, blen);
-	int res;
+	int i, res;
 
 	createUObjs();
 
-	res = (int)ucol_strcoll( colCaseInsensitive,
-							  a, len,
-							  b, len);
-	if ( res == 0 && alen != blen )
+	/*
+	 * Preventing any influence of following characters to
+	 * current one, try
+	 * select * from (values ('е'::mchar),('ё'),('еа'),('еб'),('ее'),('еж'),
+	 *						 ('ёа'),('ёб'),('ёё'),('ёж'),('ёе'),('её')) z
+	 * order by 1;
+	 */
+	for (i=0; i<len; i++)
+	{
+		res = (int)ucol_strcoll(colCaseInsensitive,
+								a + i, 1,
+								b + i, 1);
+		if (res)
+			return res;
+	}
+
+	if (alen != blen)
 		return (alen > blen) ? 1 : - 1;
-	return res;
+	return 0;
 }
 
 int 
 UCharCompare(UChar * a, int alen, UChar *b, int blen) {
 	int len = Min(alen, blen);
-	int res;
+	int i, res;
 
 	createUObjs();
 
-	res =  (int)ucol_strcoll( colCaseSensitive,
-							  a, len,
-							  b, len);
-	if ( res == 0 && alen != blen )
+	/* see above */
+	for (i=0; i<len; i++)
+	{
+		res = (int)ucol_strcoll(colCaseSensitive,
+								a + i, 1,
+								b + i, 1);
+		if (res)
+			return res;
+	}
+
+	if (alen != blen)
 		return (alen > blen) ? 1 : - 1;
-	return res;
+	return 0;
 }

contrib/mchar/sql/mm.sql

@@ -183,3 +183,8 @@ SELECT * FROM a, c WHERE mvarchar255 = mchar2;
 DROP TABLE a;
 DROP TABLE c;
 
+select * from (values ('е'::mchar),('ё'),('еа'),('еб'),('ее'),('еж'),('ёа'),('ёб'),('ёё'),('ёж'),('ёе'),('её')) z order by 1;
+select mvarchar_icase_cmp('ё', 'е');
+select mvarchar_icase_cmp('Ё', 'Е');
+select mvarchar_icase_cmp('й', 'и');
+select mvarchar_icase_cmp('Й', 'И');

contrib/mmts/multimaster--1.0.sql

@@ -84,7 +84,7 @@ CREATE FUNCTION mtm.make_table_local(relation regclass) RETURNS void
 AS 'MODULE_PATHNAME','mtm_make_table_local'
 LANGUAGE C;
 
-CREATE FUNCTION mtm.broadcast_table(srcTable regclass, dstNodesMask bigint) RETURNS void
+CREATE FUNCTION mtm.broadcast_table(srcTable regclass) RETURNS void
 AS 'MODULE_PATHNAME','mtm_broadcast_table'
 LANGUAGE C;
 

contrib/mmts/multimaster.c

@@ -602,8 +602,9 @@ csn_t MtmDistributedTransactionSnapshot(TransactionId xid, int nodeId, nodemask_
 		MtmTransState* ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
 		if (ts != NULL) {
 			*participantsMask = ts->participantsMask;
-			/* If node is disables, then we are in a process of recovery of this node */
-			snapshot = ts->snapshot;
+			if (!ts->isLocal)
+				snapshot = ts->snapshot;
+			// /* If node is disables, then we are in a process of recovery of this node */
 			// if (!ts->isLocal && BIT_CHECK(ts->participantsMask|Mtm->disabledNodeMask, nodeId-1)) {
 			// 	snapshot = ts->snapshot;
 			// 	Assert(ts->gtid.node == MtmNodeId || MtmIsRecoverySession);
@@ -4458,7 +4459,7 @@ Datum mtm_broadcast_table(PG_FUNCTION_ARGS)
 {
 	MtmCopyRequest copy;
 	copy.sourceTable = PG_GETARG_OID(0);
-	copy.targetNodes = PG_GETARG_INT64(1);
+	copy.targetNodes = ~Mtm->disabledNodeMask;
 	LogLogicalMessage("B", (char*)&copy, sizeof(copy), true);
 	MtmTx.containsDML = true;
 	PG_RETURN_VOID();
@@ -5186,7 +5187,8 @@ static void MtmProcessUtility(PlannedStmt *pstmt,
 		case T_CreateSeqStmt:
 			{
 				CreateSeqStmt *stmt = (CreateSeqStmt *) parsetree;
-				AdjustCreateSequence(stmt->options);
+				if (!MtmVolksWagenMode)
+					AdjustCreateSequence(stmt->options);
 			}
 			break;
 

contrib/mmts/pglogical_receiver.c

@@ -705,6 +705,7 @@ pglogical_receiver_main(Datum main_arg)
 		continue;
 
 	  OnError:
+		ByteBufferReset(&buf);
 		PQfinish(conn);
 		MtmReleaseRecoverySlot(nodeId);
 		MtmSleep(RECEIVER_SUSPEND_TIMEOUT);

contrib/mmts/state.c

@@ -88,14 +88,15 @@ MtmSetClusterStatus(MtmNodeStatus status)
 	 */
 	if (status == MTM_ONLINE)
 	{
-		if (!Mtm->refereeGrant && MtmRefereeReadSaved() > 0)
+		int saved_winner_node_id = MtmRefereeReadSaved();
+		if (!Mtm->refereeGrant && saved_winner_node_id > 0)
 		{
 			/*
 			 * We booted after being with refereeGrant,
 			 * but now have ordinary majority.
 			 */
 			MtmPollStatusOfPreparedTransactions(true);
-			MtmRefereeClearWinner();
+			Mtm->refereeWinnerId = saved_winner_node_id;
 		}
 	}
 
@@ -783,7 +784,7 @@ MtmRefereeClearWinner(void)
 		CommitTransactionCommand();
 		if (rc < 0)
 		{
-			MTM_ELOG(WARNING, "Failed to save referee decision, proceeding anyway");
+			MTM_ELOG(WARNING, "Failed to clean referee decision");
 			return false;
 		}
 	}

contrib/mmts/tests2/lib/test_helper.py

@@ -95,11 +95,6 @@ def performFailure(self, failure, wait=0, node_wait_for_commit=-1, node_wait_for
             self.client.get_aggregates(clean=False)
             self.client.stop()
 
-        if node_wait_for_commit >= 0:
-            self.awaitCommit(node_wait_for_commit)
-        else:
-            time.sleep(TEST_RECOVERY_TIME)
-
         if node_wait_for_online != None:
             self.awaitOnline(node_wait_for_online)
         else:
@@ -109,6 +104,11 @@ def performFailure(self, failure, wait=0, node_wait_for_commit=-1, node_wait_for
             self.client.bgrun()
             time.sleep(3)
 
+        if node_wait_for_commit >= 0:
+            self.awaitCommit(node_wait_for_commit)
+        else:
+            time.sleep(TEST_RECOVERY_TIME)
+
         aggs = self.client.get_aggregates()
 
         return (aggs_failure, aggs)

contrib/mmts/tests2/test_recovery.py

@@ -11,60 +11,7 @@
 
 from lib.bank_client import MtmClient
 from lib.failure_injector import *
-
-TEST_WARMING_TIME = 5
-TEST_DURATION = 10
-TEST_RECOVERY_TIME = 30
-TEST_SETUP_TIME = 20
-TEST_STOP_DELAY = 5
-
-class TestHelper(object):
-
-    def assertIsolation(self, aggs):
-        isolated = True
-        for conn_id, agg in enumerate(aggs):
-            isolated = isolated and agg['sumtotal']['isolation'] == 0
-        if not isolated:
-            raise AssertionError('Isolation failure')
-
-    def assertCommits(self, aggs):
-        commits = True
-        for conn_id, agg in enumerate(aggs):
-            commits = commits and 'commit' in agg['transfer']['finish']
-        if not commits:
-            print('No commits during aggregation interval')
-            # time.sleep(100000)
-            raise AssertionError('No commits during aggregation interval')
-
-    def assertNoCommits(self, aggs):
-        commits = True
-        for conn_id, agg in enumerate(aggs):
-            commits = commits and 'commit' in agg['transfer']['finish']
-        if commits:
-            raise AssertionError('There are commits during aggregation interval')
-
-    def performFailure(self, failure):
-
-        time.sleep(TEST_WARMING_TIME)
-         
-        print('Simulate failure at ',datetime.datetime.utcnow())
-
-        failure.start()
-
-        self.client.clean_aggregates()
-        time.sleep(TEST_DURATION)
-        aggs_failure = self.client.get_aggregates()
-
-        failure.stop()
-
-        print('Eliminate failure at ',datetime.datetime.utcnow())
-
-        self.client.clean_aggregates()
-        time.sleep(TEST_RECOVERY_TIME)
-        aggs = self.client.get_aggregates()
-
-        return (aggs_failure, aggs)
-
+from lib.test_helper import *
 
 class RecoveryTest(unittest.TestCase, TestHelper):
 
@@ -124,7 +71,8 @@ def test_normal_operations(self):
     def test_node_partition(self):
         print('### test_node_partition ###')
 
-        aggs_failure, aggs = self.performFailure(SingleNodePartition('node3'))
+        aggs_failure, aggs = self.performFailure(SingleNodePartition('node3'),
+            node_wait_for_online="dbname=regression user=postgres host=127.0.0.1 port=15434", stop_load=True)
 
         self.assertCommits(aggs_failure[:2])
         self.assertNoCommits(aggs_failure[2:])
@@ -148,7 +96,8 @@ def test_edge_partition(self):
     def test_node_restart(self):
         print('### test_node_restart ###')
 
-        aggs_failure, aggs = self.performFailure(RestartNode('node3'))
+        aggs_failure, aggs = self.performFailure(RestartNode('node3'),
+            node_wait_for_online="dbname=regression user=postgres host=127.0.0.1 port=15434", stop_load=True)
 
         self.assertCommits(aggs_failure[:2])
         self.assertNoCommits(aggs_failure[2:])
@@ -160,7 +109,8 @@ def test_node_restart(self):
     def test_node_crash(self):
         print('### test_node_crash ###')
 
-        aggs_failure, aggs = self.performFailure(CrashRecoverNode('node3'))
+        aggs_failure, aggs = self.performFailure(CrashRecoverNode('node3'),
+            node_wait_for_online="dbname=regression user=postgres host=127.0.0.1 port=15434", stop_load=True)
 
         self.assertCommits(aggs_failure[:2])
         self.assertNoCommits(aggs_failure[2:])
@@ -172,7 +122,8 @@ def test_node_crash(self):
     def test_node_bicrash(self):
         print('### test_node_bicrash ###')
 
-        aggs_failure, aggs = self.performFailure(CrashRecoverNode('node3'))
+        aggs_failure, aggs = self.performFailure(CrashRecoverNode('node3'),
+            node_wait_for_commit=2, stop_load=True)
 
         self.assertCommits(aggs_failure[:2])
         self.assertNoCommits(aggs_failure[2:])
@@ -181,10 +132,11 @@ def test_node_bicrash(self):
         self.assertCommits(aggs)
         self.assertIsolation(aggs)
 
-        aggs_failure, aggs = self.performFailure(CrashRecoverNode('node3'))
+        aggs_failure, aggs = self.performFailure(CrashRecoverNode('node1'),
+            node_wait_for_commit=0, stop_load=True)
 
-        self.assertCommits(aggs_failure[:2])
-        self.assertNoCommits(aggs_failure[2:])
+        self.assertNoCommits(aggs_failure[0:1])  # [1]
+        self.assertCommits(aggs_failure[1:]) # [2, 3]
         self.assertIsolation(aggs_failure)
 
         self.assertCommits(aggs)