Merge branch 'master' into uniq-arbiter

Author: kvap

contrib/multimaster/multimaster.c

@@ -123,6 +123,7 @@ static TransactionId DtmNextXid;
 static SnapshotData DtmSnapshot = { HeapTupleSatisfiesMVCC };
 static bool DtmHasGlobalSnapshot;
 static int DtmLocalXidReserve;
+static CommandId DtmCurcid;
 static Snapshot DtmLastSnapshot;
 static TransactionManager DtmTM = {
 	DtmGetTransactionStatus,
@@ -630,11 +631,12 @@ static Snapshot DtmGetSnapshot(Snapshot snapshot)
 {
 	if (TransactionIdIsValid(DtmNextXid) && snapshot != &CatalogSnapshotData)
 	{
-		if (!DtmHasGlobalSnapshot) {
+		if (!DtmHasGlobalSnapshot && (snapshot != DtmLastSnapshot || DtmCurcid != GetCurrentCommandId(false))) {
 			DtmGlobalGetSnapshot(DtmNextXid, &DtmSnapshot, &dtm->minXid);
         }
 		DtmLastSnapshot = snapshot;
 		DtmMergeWithGlobalSnapshot(snapshot);
+		DtmCurcid = snapshot->curcid;
 		if (!IsolationUsesXactSnapshot())
 		{
 			/* Use single global snapshot during all transaction for repeatable read isolation level,

contrib/pg_dtm/pg_dtm.c

@@ -107,6 +107,7 @@ static SnapshotData DtmSnapshot = { HeapTupleSatisfiesMVCC };
 static bool DtmHasGlobalSnapshot;
 static bool DtmGlobalXidAssigned;
 static int DtmLocalXidReserve;
+static CommandId DtmCurcid;
 static Snapshot DtmLastSnapshot;
 static TransactionManager DtmTM = {
 	DtmGetTransactionStatus,
@@ -600,7 +601,6 @@ DtmGetNewTransactionId(bool isSubXact)
 	return xid;
 }
 
-
 static Snapshot DtmGetSnapshot(Snapshot snapshot)
 {
 	if (DtmGlobalXidAssigned)
@@ -612,12 +612,13 @@ static Snapshot DtmGetSnapshot(Snapshot snapshot)
 		return PgGetSnapshotData(snapshot);
 	}
 	if (TransactionIdIsValid(DtmNextXid) && snapshot != &CatalogSnapshotData)
-	{
-		if (!DtmHasGlobalSnapshot) { 
+	{		
+		if (!DtmHasGlobalSnapshot && (snapshot != DtmLastSnapshot || DtmCurcid != GetCurrentCommandId(false))) {
 			ArbiterGetSnapshot(DtmNextXid, &DtmSnapshot, &dtm->minXid);
 		}
 		DtmLastSnapshot = snapshot;
 		DtmMergeWithGlobalSnapshot(snapshot);
+		DtmCurcid = snapshot->curcid;
 		if (!IsolationUsesXactSnapshot())
 		{
 			/* Use single global snapshot during all transaction for repeatable read isolation level,

src/backend/commands/indexcmds.c

@@ -368,7 +368,7 @@ AlterIndex(Oid indexRelationId, IndexStmt *stmt)
         SPI_cursor_fetch(portal, true, 1);
         if (!SPI_processed) { 
 			break;
-		}
+		}										
 		tuple = SPI_tuptable->vals[0];
 		tupleid = &tuple->t_data->t_ctid;
 		ExecStoreTuple(tuple, slot, InvalidBuffer, false);

src/bin/insbench/insbench.cpp

@@ -98,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);
@@ -109,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();
 	}
@@ -124,10 +126,11 @@ void* indexUpdater(void* arg)
 		sleep(cfg.indexUpdateInterval);
 		printf("Alter indexes\n");
 		time_t now = getCurrentTime();
+		time_t limit = cfg.useSystemTime ? now : currTimestamp;
 		{
 			work txn(con);
 			for (int i = 0; i < cfg.nIndexes; i++) { 
-				exec(txn, "alter index idx%d where pk<%lu", i, cfg.useSystemTime ? now : currTimestamp);
+				exec(txn, "alter index idx%d where pk<%lu", i, limit);
 			}
 			txn.commit();
 		}
@@ -149,18 +152,6 @@ void initializeDatabase()
 	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)");
-	if (!cfg.noPK) { 
-		exec(txn, "create index pk on t(pk)");
-	}
-	for (int i = 0; i < cfg.nIndexes; i++) { 
-		if (cfg.indexUpdateInterval == 0)  { 
-			exec(txn, "create index idx%d on t(k%d)", i, i+1);
-		} else if (cfg.useSystemTime) { 
-			exec(txn, "create index idx%d on t(k%d) where pk<%ld", i, i+1, now);
-		} else { 
-			exec(txn, "create index idx%d on t(k%d) where pk<%ld", i, i+1, 0);
-		}
-	}
 
 	if (cfg.initialSize)
 	{
@@ -184,9 +175,26 @@ void initializeDatabase()
 		    txn.prepared("insert")(cfg.initialSize)(cfg.initialSize-1).exec();
 			currTimestamp = cfg.initialSize;
 	    }
-		txn.exec("vacuum analyze");
+	}
+	if (!cfg.noPK) { 
+		exec(txn, "create index pk on t(pk)");
+	}
+	for (int i = 0; i < cfg.nIndexes; i++) { 
+		if (cfg.indexUpdateInterval == 0)  { 
+			exec(txn, "create index idx%d on t(k%d)", i, i+1);
+		} else if (cfg.useSystemTime) { 
+			exec(txn, "create index idx%d on t(k%d) where pk<%ld", i, i+1, now);
+		} else { 
+			exec(txn, "create index idx%d on t(k%d) where pk<%ld", i, i+1, currTimestamp);
+		}
 	}
 	txn.commit();
+	{
+		nontransaction txn(con);
+		txn.exec("vacuum analyze");
+		sleep(2);
+	}
+	printf("Database intialized\n");
 }
 
 
@@ -234,7 +242,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");
@@ -257,10 +266,11 @@ int main (int argc, char* argv[])
     for (int i = 0; i < cfg.nInserters; i++) { 
         inserters[i].wait();
     }    
+    time_t elapsed = getCurrentTime() - start;
+
     running = false;
 	bgw.wait();
 
-    time_t elapsed = getCurrentTime() - start;
 
     printf(
         "{\"tps\":%f, \"index_updates\":%d, \"max_update_time\":%ld, \"avg_update_time\":%f,"

src/bin/insbench/run.sh

@@ -0,0 +1,6 @@
+echo Insert with 1 index
+./insbench -c "dbname=postgres host=localhost port=5432 sslmode=disable" -x 0
+echo Insert with 9 indexex
+./insbench -c "dbname=postgres host=localhost port=5432 sslmode=disable" -x 8
+echo Insert with 9 partial indexes
+./insbench -c "dbname=postgres host=localhost port=5432 sslmode=disable" -x 8 -u 1