Use user from multimaster connstrings in bgwpool

Author: kelvich

contrib/mmts/bgwpool.c

@@ -25,7 +25,7 @@ static void BgwPoolMainLoop(Datum arg)
     void* work;
 
     BackgroundWorkerUnblockSignals();
-	BackgroundWorkerInitializeConnection(pool->dbname, "stas");
+	BackgroundWorkerInitializeConnection(pool->dbname, pool->dbuser);
 
     while(true) { 
         PGSemaphoreLock(&pool->available);
@@ -63,7 +63,7 @@ static void BgwPoolMainLoop(Datum arg)
     }
 }
 
-void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname, size_t queueSize, size_t nWorkers)
+void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname,  char const* dbuser, size_t queueSize, size_t nWorkers)
 {
     pool->queue = (char*)ShmemAlloc(queueSize);
     pool->executor = executor;
@@ -80,7 +80,8 @@ void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname, si
     pool->pending = 0;
 	pool->nWorkers = nWorkers;
 	pool->lastPeakTime = 0;
-    strcpy(pool->dbname, dbname);
+	strncpy(pool->dbname, dbname, MAX_DBNAME_LEN);
+	strncpy(pool->dbuser, dbuser, MAX_DBUSER_LEN);
 }
 
 timestamp_t BgwGetLastPeekTime(BgwPool* pool)

contrib/mmts/bgwpool.h

@@ -10,6 +10,7 @@ typedef void(*BgwPoolExecutor)(int id, void* work, size_t size);
 typedef uint64 timestamp_t;
 
 #define MAX_DBNAME_LEN 30
+#define MAX_DBUSER_LEN 30
 #define MULTIMASTER_BGW_RESTART_TIMEOUT 1 /* seconds */
 
 extern timestamp_t MtmGetSystemTime(void);   /* non-adjusted current system time */
@@ -30,14 +31,15 @@ typedef struct
 	time_t lastPeakTime;
     bool   producerBlocked;
     char   dbname[MAX_DBNAME_LEN];
+	char   dbuser[MAX_DBUSER_LEN];
     char*  queue;
 } BgwPool;
 
 typedef BgwPool*(*BgwPoolConstructor)(void);
 
 extern void BgwPoolStart(int nWorkers, BgwPoolConstructor constructor);
 
-extern void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname, size_t queueSize, size_t nWorkers);
+extern void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname, char const* dbuser, size_t queueSize, size_t nWorkers);
 
 extern void BgwPoolExecute(BgwPool* pool, void* work, size_t size);
 

contrib/mmts/multimaster.c

@@ -197,6 +197,7 @@ char const* const MtmNodeStatusMnem[] =
 
 bool  MtmDoReplication;
 char* MtmDatabaseName;
+char* MtmDatabaseUser;
 char* MtmUtilityStmt = NULL;
 
 int   MtmNodes;
@@ -1718,7 +1719,7 @@ static void MtmInitialize()
 		PGSemaphoreCreate(&Mtm->votingSemaphore);
 		PGSemaphoreReset(&Mtm->votingSemaphore);
 		SpinLockInit(&Mtm->spinlock);
-        BgwPoolInit(&Mtm->pool, MtmExecutor, MtmDatabaseName, MtmQueueSize, MtmWorkers);
+		BgwPoolInit(&Mtm->pool, MtmExecutor, MtmDatabaseName, MtmDatabaseUser, MtmQueueSize, MtmWorkers);
 		RegisterXactCallback(MtmXactCallback, NULL);
 		MtmTx.snapshot = INVALID_CSN;
 		MtmTx.xid = InvalidTransactionId;		
@@ -1805,19 +1806,31 @@ static void MtmSplitConnStrs(void)
 
 		MtmUpdateNodeConnectionInfo(&MtmConnections[i], connStr);
 
-		if (i+1 == MtmNodeId) { 
-			char* dbName = strstr(connStr, "dbname=");
+		if (i+1 == MtmNodeId) {
+			char* dbName = strstr(connStr, "dbname="); // XXX: shoud we care about string 'itisnotdbname=xxx'?
+			char* dbUser = strstr(connStr, "user=");
 			char* end;
 			size_t len;
-			if (dbName == NULL) { 
-				elog(ERROR, "Database not specified in connection string: '%s'", connStr);
-			}
+
+			if (dbName == NULL)
+				elog(ERROR, "Database is not specified in connection string: '%s'", connStr);
+
+			if (dbUser == NULL)
+				elog(ERROR, "Database user is not specified in connection string: '%s'", connStr);
+
 			dbName += 7;
 			for (end = dbName; *end != ' ' && *end != '\0'; end++);
 			len = end - dbName;
 			MtmDatabaseName = (char*)palloc(len + 1);
 			memcpy(MtmDatabaseName, dbName, len);
 			MtmDatabaseName[len] = '\0';
+
+			dbUser += 5;
+			for (end = dbUser; *end != ' ' && *end != '\0'; end++);
+			len = end - dbUser;
+			MtmDatabaseUser = (char*)palloc(len + 1);
+			memcpy(MtmDatabaseUser, dbUser, len);
+			MtmDatabaseUser[len] = '\0';
 		}
 		connStr = p + 1;
     }
@@ -3287,6 +3300,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 		case T_LockStmt:
 		case T_CheckPointStmt:
 		case T_ReindexStmt:
+		case T_RefreshMatViewStmt:
 		    skipCommand = true;
 			break;
 

contrib/mmts/tests/reinit-mm.sh

@@ -18,7 +18,7 @@ for ((i=1;i<=n_nodes;i++))
 do    
     port=$((5431 + i))
 	raft_port=$((6665 + i))
-    conn_str="$conn_str${sep}dbname=regression host=localhost port=$port sslmode=disable"
+    conn_str="$conn_str${sep}dbname=regression user=stas host=localhost port=$port sslmode=disable"
 	raft_conn_str="$raft_conn_str${sep}${i}:localhost:$raft_port"
     sep=","
     initdb node$i