New DDL replication mechanism

Author: knizhnik

multimaster--1.0.sql

@@ -1,19 +1,20 @@
 -- complain if script is sourced in psql, rather than via CREATE EXTENSION
 \echo Use "CREATE EXTENSION multimaster" to load this file. \quit
 
-CREATE FUNCTION mtm_start_replication() RETURNS void
+CREATE FUNCTION mtm.start_replication() RETURNS void
 AS 'MODULE_PATHNAME','mtm_start_replication'
 LANGUAGE C;
 
-CREATE FUNCTION mtm_stop_replication() RETURNS void
+CREATE FUNCTION mtm.stop_replication() RETURNS void
 AS 'MODULE_PATHNAME','mtm_stop_replication'
 LANGUAGE C;
 
-CREATE FUNCTION mtm_drop_node(node integer, drop_slot bool default false) RETURNS void
+CREATE FUNCTION mtm.drop_node(node integer, drop_slot bool default false) RETURNS void
 AS 'MODULE_PATHNAME','mtm_drop_node'
 LANGUAGE C;
 
-CREATE FUNCTION mtm_get_snapshot() RETURNS bigint
+CREATE FUNCTION mtm.get_snapshot() RETURNS bigint
 AS 'MODULE_PATHNAME','mtm_get_snapshot'
 LANGUAGE C;
 
+CREATE TABLE IF NOT EXISTS mtm.ddl_log (issued timestamp with time zone not null, query text);

multimaster.c

@@ -48,6 +48,9 @@
 #include "replication/slot.h"
 #include "port/atomics.h"
 #include "tcop/utility.h"
+#include "nodes/makefuncs.h"
+#include "access/htup_details.h"
+#include "catalog/indexing.h"
 
 #include "multimaster.h"
 
@@ -865,7 +868,7 @@ mtm_drop_node(PG_FUNCTION_ARGS)
 		dtm->nNodes -= 1;
 		if (!IsTransactionBlock())
 		{
-			MtmBroadcastUtilityStmt(psprintf("select mtm_drop_node(%d,%s)", nodeId, dropSlot ? "true" : "false"), true);
+			MtmBroadcastUtilityStmt(psprintf("select multimaster.drop_node(%d,%s)", nodeId, dropSlot ? "true" : "false"), true);
 		}
 		if (dropSlot) 
 		{
@@ -880,7 +883,7 @@ mtm_get_snapshot(PG_FUNCTION_ARGS)
 {
 	PG_RETURN_INT64(dtmTx.snapshot);
 }
-	
+
 /*
  * Execute statement with specified parameters and check its result
  */
@@ -926,7 +929,7 @@ static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError)
 					failedNode = i;
 					do { 
 						PQfinish(conns[i]);
-					} while (--i >= 0);				
+					} while (--i >= 0);                             
 					elog(ERROR, "Failed to establish connection '%s' to node %d", conn_str, failedNode);
 				}
 			}
@@ -935,7 +938,7 @@ static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError)
 		i += 1;
 	}
 	Assert(i == MtmNodes);
-	
+    
 	for (i = 0; i < MtmNodes; i++) 
 	{ 
 		if (conns[i]) 
@@ -972,7 +975,7 @@ static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError)
 				failedNode = i;
 			}
 		}
-	}			
+	}                       
 	for (i = 0; i < MtmNodes; i++) 
 	{ 
 		if (conns[i])
@@ -986,6 +989,48 @@ static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError)
 	}
 }
 
+static void MtmProcessDDLCommand(char const* queryString)
+{
+	RangeVar   *rv;
+	Relation	rel;
+	TupleDesc	tupDesc;
+	HeapTuple	tup;
+	Datum		values[Natts_mtm_ddl_log];
+	bool		nulls[Natts_mtm_ddl_log];
+	TimestampTz ts = GetCurrentTimestamp();
+
+	rv = makeRangeVar(MULTIMASTER_SCHEMA_NAME, MULTIMASTER_DDL_TABLE, -1);
+	rel = heap_openrv_extended(rv, RowExclusiveLock, true);
+
+	if (rel == NULL) { 
+		return;
+	}
+		
+	tupDesc = RelationGetDescr(rel);
+
+	/* Form a tuple. */
+	memset(nulls, false, sizeof(nulls));
+
+	values[Anum_mtm_ddl_log_issued - 1] = TimestampTzGetDatum(ts);
+	values[Anum_mtm_ddl_log_query - 1] = CStringGetTextDatum(queryString);
+
+	tup = heap_form_tuple(tupDesc, values, nulls);
+
+	/* Insert the tuple to the catalog. */
+	simple_heap_insert(rel, tup);
+
+	/* Update the indexes. */
+	CatalogUpdateIndexes(rel, tup);
+
+	/* Cleanup. */
+	heap_freetuple(tup);
+	heap_close(rel, RowExclusiveLock);
+
+	elog(WARNING, "Replicate command: '%s'", queryString);
+
+	dtmTx.containsDML = true;
+}
+
 static void MtmProcessUtility(Node *parsetree, const char *queryString,
 							 ProcessUtilityContext context, ParamListInfo params,
 							 DestReceiver *dest, char *completionTag)
@@ -1013,22 +1058,18 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 			skipCommand = false;			
 			break;
 	}
-	if (skipCommand || IsTransactionBlock()) { 
-		if (PreviousProcessUtilityHook != NULL)
-		{
-			PreviousProcessUtilityHook(parsetree, queryString, context,
-									   params, dest, completionTag);
-		}
-		else
-		{
-			standard_ProcessUtility(parsetree, queryString, context,
-									params, dest, completionTag);
-		}
-		if (!skipCommand) {
-			dtmTx.isDistributed = false;
-		}
-	} else { 		
-		MtmBroadcastUtilityStmt(queryString, false);
+	if (!skipCommand && !dtmTx.isReplicated) {
+		MtmProcessDDLCommand(queryString);
+	}
+	if (PreviousProcessUtilityHook != NULL)
+	{
+		PreviousProcessUtilityHook(parsetree, queryString, context,
+								   params, dest, completionTag);
+	}
+	else
+	{
+		standard_ProcessUtility(parsetree, queryString, context,
+								params, dest, completionTag);
 	}
 }
 

multimaster.control

@@ -1,4 +1,5 @@
 comment = 'Multimaster'
 default_version = '1.0'
 module_pathname = '$libdir/multimaster'
-relocatable = true
+schema = mtm
+relocatable = false

multimaster.h

@@ -5,7 +5,7 @@
 #include "bgwpool.h"
 
 #define MTM_INFO(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__) 
-#define MTM_TRACE(fmt, ...)
+#define MTM_TRACE(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__) 
 #define MTM_TUPLE_TRACE(fmt, ...)
 /*
 #define MTM_TRACE(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__) 
@@ -14,7 +14,14 @@
 
 #define BIT_SET(mask, bit) ((mask) & ((int64)1 << (bit)))
 
-#define MULTIMASTER_NAME "mmts"
+#define MULTIMASTER_NAME           "mtm"
+#define MULTIMASTER_SCHEMA_NAME    "mtm"
+#define MULTIMASTER_DDL_TABLE      "ddl_log"
+
+#define Natts_mtm_ddl_log 2
+#define Anum_mtm_ddl_log_issued		1
+#define Anum_mtm_ddl_log_query		2
+
 
 typedef uint64 csn_t; /* commit serial number */
 #define INVALID_CSN  ((csn_t)-1)

pglogical_apply.c

@@ -472,13 +472,14 @@ process_remote_commit(StringInfo s)
 static void
 process_remote_insert(StringInfo s, Relation rel)
 {
-	EState	   *estate;
-	TupleData   new_tuple;
+	EState *estate;
+	TupleData new_tuple;
 	TupleTableSlot *newslot;
 	TupleTableSlot *oldslot;
 	ResultRelInfo *relinfo;
-	ScanKey	   *index_keys;
-	int			i;
+	ScanKey	*index_keys;
+	char* relname = RelationGetRelationName(rel);
+	int	i;
 
 	estate = create_rel_estate(rel);
 	newslot = ExecInitExtraTupleSlot(estate);
@@ -560,6 +561,18 @@ process_remote_insert(StringInfo s, Relation rel)
     FreeExecutorState(estate);
 
 	CommandCounterIncrement();
+
+	if (strcmp(relname, MULTIMASTER_DDL_TABLE) == 0) { 
+		char* ddl = TextDatumGetCString(new_tuple.values[Anum_mtm_ddl_log_query-1]);
+		int rc;
+		SPI_connect();
+		rc = SPI_execute(ddl, false, 0);
+        SPI_finish();
+		if (rc != SPI_OK_UTILITY) { 
+			elog(ERROR, "Failed to execute utility statement %s", ddl);
+		}
+	}
+
 }
 
 static void

tests/dtmbench.cpp

@@ -180,28 +180,20 @@ void* writer(void* arg)
 void initializeDatabase()
 {
 	connection conn(cfg.connections[0]);
-	printf("creating extension\n");
-	{
-		nontransaction txn(conn);
-		exec(txn, "drop extension if exists multimaster");
-		exec(txn, "create extension multimaster");
-	}
-	printf("extension created\n");
-
-	printf("creating table t\n");
+    time_t start = getCurrentTime();
+	printf("Creating database schema...\n");
 	{
 		nontransaction txn(conn);
 		exec(txn, "drop table if exists t");
 		exec(txn, "create table t(u int primary key, v int)");
 	}
-	printf("table t created\n");
-	printf("inserting stuff into t\n");
+	printf("Populating data...\n");
 	{
 		work txn(conn);
 		exec(txn, "insert into t (select generate_series(0,%d), %d)", cfg.nAccounts-1, 0);
 		txn.commit();
 	}
-	printf("stuff inserted\n");
+	printf("Initialization completed in %f seconds\n", (start  - getCurrentTime())/100000.0);
 }
 
 int main (int argc, char* argv[])

tests/perf.results

@@ -117,3 +117,24 @@ Bench started at Пн. февр. 15 17:26:11 MSK 2016
 astro5:{tps:96460.088384, transactions:1000000, selects:2000000, updates:0, aborts:0, abort_percent: 0, readers:0, writers:200, update_percent:0, accounts:500000, iterations:5000, hosts:3}
 Bench finished at Пн. февр. 15 17:26:22 MSK 2016
 Bench started at Пн. февр. 15 17:26:41 MSK 2016
+Bench started at Пн. февр. 15 17:58:14 MSK 2016
+astro5:{tps:93430.358394, transactions:1250000, selects:2500000, updates:0, aborts:0, abort_percent: 0, readers:0, writers:250, update_percent:0, accounts:500000, iterations:5000, hosts:3}
+Bench finished at Пн. февр. 15 17:58:28 MSK 2016
+Bench started at Пн. февр. 15 17:59:11 MSK 2016
+astro5:{tps:81893.902409, transactions:1500000, selects:3000000, updates:0, aborts:0, abort_percent: 0, readers:0, writers:300, update_percent:0, accounts:500000, iterations:5000, hosts:3}
+Bench finished at Пн. февр. 15 17:59:29 MSK 2016
+Bench started at Пн. февр. 15 17:59:59 MSK 2016
+astro5:{tps:105707.597142, transactions:1000000, selects:2000000, updates:0, aborts:0, abort_percent: 0, readers:0, writers:200, update_percent:0, accounts:500000, iterations:5000, hosts:3}
+Bench finished at Пн. февр. 15 18:00:09 MSK 2016
+Bench started at Пн. февр. 15 18:00:54 MSK 2016
+astro5:{tps:92668.464039, transactions:1250000, selects:2500000, updates:0, aborts:0, abort_percent: 0, readers:0, writers:250, update_percent:0, accounts:500000, iterations:5000, hosts:3}
+Bench finished at Пн. февр. 15 18:01:08 MSK 2016
+Bench started at Пн. февр. 15 18:06:22 MSK 2016
+astro5:{tps:121069.442298, transactions:125000000, selects:250000000, updates:0, aborts:0, abort_percent: 0, readers:0, writers:250, update_percent:0, accounts:500000, iterations:500000, hosts:3}
+Bench finished at Пн. февр. 15 18:23:35 MSK 2016
+Bench started at Пн. февр. 15 18:24:11 MSK 2016
+astro5:{tps:122202.228254, transactions:100000000, selects:200000000, updates:0, aborts:0, abort_percent: 0, readers:0, writers:200, update_percent:0, accounts:500000, iterations:500000, hosts:3}
+Bench finished at Пн. февр. 15 18:37:50 MSK 2016
+Bench started at Пн. февр. 15 18:44:02 MSK 2016
+astro5:{tps:121774.204222, transactions:100000000, selects:200000000, updates:0, aborts:0, abort_percent: 0, readers:0, writers:200, update_percent:0, accounts:500000, iterations:500000, hosts:3}
+Bench finished at Пн. февр. 15 18:57:44 MSK 2016

tests/perf.yml

@@ -46,7 +46,7 @@
   - name: run transfers
     shell: >
       ~/pg_cluster/install/bin/dtmbench {{connections}}
-      -w {{ nconns }} -r 0 -n 5000 -a 500000 -p {{ up }} |
+      -w {{ nconns }} -r 0 -n 500000 -a 500000 -p {{ up }} |
       tee -a perf.results |
       sed "s/^/`hostname`:/"
     register: transfers_result

tests/reinit-mm.sh

@@ -10,14 +10,15 @@ sep=""
 for ((i=1;i<=n_nodes;i++))
 do    
     port=$((5431+i))
-    conn_str="$conn_str${sep}dbname=postgres host=127.0.0.1 port=$port sslmode=disable"
+    conn_str="$conn_str${sep}dbname=postgres host=localhost port=$port sslmode=disable"
     sep=","
     initdb node$i
 done
 
-echo Start DTM
-~/postgres_cluster/contrib/arbiter/bin/arbiter -r 0.0.0.0:5431 -i 0 -d dtm 2> dtm.log &
-sleep 2
+#echo Start DTM
+#~/postgres_cluster/contrib/arbiter/bin/arbiter -r 0.0.0.0:5431 -i 0 -d dtm 2> dtm.log &
+#sleep 2
+echo "Starting nodes..."
 
 echo Start nodes
 for ((i=1;i<=n_nodes;i++))
@@ -31,7 +32,13 @@ do
 done
 
 sleep 5
-echo Initialize database schema
-psql postgres -f init.sql
+echo "Create multimaster extension..."
+
+for ((i=1;i<=n_nodes;i++))
+do
+    port=$((5431+i))
+	psql postgres -p $port -c "create extension multimaster"
+done
+
 
 echo Done