Merge branch 'master' of https://github.com/postgrespro/postgres_cluster

Author: knizhnik

contrib/pg_dtm/dtmd/src/raft.c

@@ -662,12 +662,13 @@ void raft_handle_message(raft_t *r, raft_msg_t *m) {
 	}
 }
 
+static char buf[1024];
+
 raft_msg_t *raft_recv_message(raft_t *r) {
 	struct sockaddr_in addr;
 	unsigned int addrlen = sizeof(addr);
 
 	//try to receive some data, this is a blocking call
-	char buf[1024];
 	raft_msg_t *m = (raft_msg_t *)buf;
 	int recved = recvfrom(
 		r->sock, buf, sizeof(buf), 0,

src/backend/commands/indexcmds.c

@@ -310,8 +310,8 @@ AlterIndex(Oid indexRelationId, IndexStmt *stmt)
 
 	/* Open the target index relation */
 	/*	indexRelation = index_open(indexRelationId, RowExclusiveLock); */
-	//indexRelation = index_open(indexRelationId, ShareUpdateExclusiveLock);
-	indexRelation = index_open(indexRelationId, AccessShareLock);
+	indexRelation = index_open(indexRelationId, ShareUpdateExclusiveLock);
+	/* indexRelation = index_open(indexRelationId, AccessShareLock); */
 	namespaceId = RelationGetNamespace(indexRelation);
 
 	pg_index = heap_open(IndexRelationId, RowExclusiveLock);

src/backend/tcop/utility.c

@@ -1244,7 +1244,7 @@ ProcessUtilitySlow(Node *parsetree,
 					 * eventually be needed here, so the lockmode calculation
 					 * needs to match what DefineIndex() does.
 					 */
-					lockmode = stmt->is_alter ? AccessShareLock : stmt->concurrent ? ShareUpdateExclusiveLock
+					lockmode = stmt->is_alter || stmt->concurrent ? ShareUpdateExclusiveLock
 						: ShareLock;
 					relid =
 						RangeVarGetRelidExtended(stmt->relation, lockmode,

src/bin/insbench/insbench.cpp

@@ -41,11 +41,13 @@ struct config
     int nIndexes;
     int nIterations;
 	int transactionSize;
+	int initialSize;
 	bool useSystemTime;
 	bool noPK;
     string connection;
 
     config() {
+		initialSize = 1000000;
 		indexUpdateInterval = 0;
         nInserters = 1;
 		nIndexes = 8;
@@ -96,7 +98,8 @@ void* inserter(void* arg)
 	} else {
 		con.prepare("insert", "insert into t (select generate_series($1::integer,$2::integer),ceil(random()*1000000000),ceil(random()*1000000000),ceil(random()*1000000000),ceil(random()*1000000000),ceil(random()*1000000000),ceil(random()*1000000000),ceil(random()*1000000000),ceil(random()*1000000000))");
 	}
-	
+	time_t curr = currTimestamp;
+
     for (int i = 0; i < cfg.nIterations; i++)
     { 
 		work txn(con);
@@ -107,8 +110,9 @@ void* inserter(void* arg)
 		        txn.prepared("insert")(getCurrentTime())(random())(random())(random())(random())(random())(random())(random())(random()).exec();
 	        }
 	    } else { 
-			currTimestamp = i*cfg.transactionSize;
-		    txn.prepared("insert")(i*cfg.transactionSize)((i+1)*cfg.transactionSize-1).exec();
+		    txn.prepared("insert")(curr)(curr+cfg.transactionSize-1).exec();
+			curr += cfg.transactionSize;
+			currTimestamp = curr;
 	    }
 		txn.commit();
 	}
@@ -142,8 +146,8 @@ void* indexUpdater(void* arg)
 
 void initializeDatabase()
 {
-    connection conn(cfg.connection);
-	work txn(conn);
+    connection con(cfg.connection);
+	work txn(con);
 	time_t now = getCurrentTime();
 	exec(txn, "drop table if exists t");
 	exec(txn, "create table t (pk bigint, k1 bigint, k2 bigint, k3 bigint, k4 bigint, k5 bigint, k6 bigint, k7 bigint, k8 bigint)");
@@ -159,6 +163,31 @@ void initializeDatabase()
 			exec(txn, "create index idx%d on t(k%d) where pk<%ld", i, i+1, 0);
 		}
 	}
+
+	if (cfg.initialSize)
+	{
+		if (cfg.useSystemTime) 
+		{
+#if PQXX_VERSION_MAJOR >= 4
+			con.prepare("insert", "insert into t values ($1,$2,$3,$4,$5,$6,$7,$8,$9)");
+#else
+			con.prepare("insert", "insert into t values ($1,$2,$3,$4,$5,$6,$7,$8,$9)")("bigint")("bigint")("bigint")("bigint")("bigint")("bigint")("bigint")("bigint")("bigint");
+#endif
+		} else {
+			con.prepare("insert", "insert into t (select generate_series($1::integer,$2::integer),ceil(random()*1000000000),ceil(random()*1000000000),ceil(random()*1000000000),ceil(random()*1000000000),ceil(random()*1000000000),ceil(random()*1000000000),ceil(random()*1000000000),ceil(random()*1000000000))");
+		}
+		if (cfg.useSystemTime) 
+		{
+		    for (int i = 0; i < cfg.initialSize; i++) 
+			{ 
+		        txn.prepared("insert")(getCurrentTime())(random())(random())(random())(random())(random())(random())(random())(random()).exec();
+	        }
+	    } else { 
+		    txn.prepared("insert")(cfg.initialSize)(cfg.initialSize-1).exec();
+			currTimestamp = cfg.initialSize;
+	    }
+		txn.exec("vacuum analyze");
+	}
 	txn.commit();
 }
 
@@ -185,9 +214,12 @@ int main (int argc, char* argv[])
             case 'n':
                 cfg.nIterations = atoi(argv[++i]);
                 continue;
-            case 'i':
+            case 'x':
                 cfg.nIndexes = atoi(argv[++i]);
                 continue;
+            case 'i':
+                cfg.initialSize = atoi(argv[++i]);
+                continue;
             case 'c':
                 cfg.connection = string(argv[++i]);
                 continue;
@@ -204,7 +236,8 @@ int main (int argc, char* argv[])
                "\t-w N\tnumber of inserters (1)\n"
                "\t-u N\tindex update interval (0)\n"
                "\t-n N\tnumber of iterations (10000)\n"
-               "\t-i N\tnumber of indexes (8)\n"
+               "\t-x N\tnumber of indexes (8)\n"
+               "\t-i N\tinitial table size (1000000)\n"
                "\t-q\tuse system time and libpq\n"
                "\t-p\tno primary key\n"
                "\t-c STR\tdatabase connection string\n");