Use pglogical plugin in multimaster

Author: knizhnik

contrib/multimaster/Makefile

@@ -1,6 +1,6 @@
 MODULE_big = multimaster
-OBJS = multimaster.o receiver_raw.o decoder_raw.o libdtm.o bytebuf.o bgwpool.o sockhub/sockhub.o
-
+OBJS = multimaster.o libdtm.o bytebuf.o bgwpool.o sockhub/sockhub.o pglogical_output.o pglogical_proto.o pglogical_receiver.o
+#OBJS = multimaster.o receiver_raw.o decoder_raw.o libdtm.o bytebuf.o bgwpool.o sockhub/sockhub.o
 EXTENSION = multimaster
 DATA = multimaster--1.0.sql
 

contrib/multimaster/multimaster.c

@@ -28,7 +28,6 @@
 #include "access/xlog.h"
 #include "storage/proc.h"
 #include "storage/procarray.h"
-#include "executor/spi.h"
 #include "executor/executor.h"
 #include "access/twophase.h"
 #include "utils/guc.h"
@@ -104,7 +103,6 @@ static void DtmShmemStartup(void);
 static void DtmBackgroundWorker(Datum arg);
 
 static void MMMarkTransAsLocal(TransactionId xid);
-static void MMExecutor(int id, void* work, size_t size);
 static BgwPool* MMPoolConstructor(void);
 
 static shmem_startup_hook_type prev_shmem_startup_hook;
@@ -1185,48 +1183,7 @@ MMExecutorFinish(QueryDesc *queryDesc)
     {
         standard_ExecutorFinish(queryDesc);
     }
-}
-        
-static void MMExecutor(int id, void* work, size_t size)
-{
-    TransactionId xid = *(TransactionId*)work;
-    char* stmts = (char*)work + 4;
-    bool finished = false;
-
-    MMJoinTransaction(xid);
-
-    SetCurrentStatementStartTimestamp();               
-    StartTransactionCommand();
-    SPI_connect();
-    PushActiveSnapshot(GetTransactionSnapshot());
-
-    PG_TRY();
-    {
-        int rc = SPI_execute(stmts, false, 0);
-        SPI_finish();
-        PopActiveSnapshot();
-        finished = true;
-        if (rc != SPI_OK_INSERT && rc != SPI_OK_UPDATE && rc != SPI_OK_DELETE) {
-            ereport(LOG, (errmsg("Executor %d: failed to apply transaction %u",
-                                 id, xid)));
-            AbortCurrentTransaction();
-        } else { 
-            CommitTransactionCommand();
-        }
-    }
-    PG_CATCH();
-    {
-        FlushErrorState();
-        if (!finished) {
-            SPI_finish();
-            if (ActiveSnapshotSet()) { 
-                PopActiveSnapshot();
-            }
-        }
-        AbortCurrentTransaction();
-    }
-    PG_END_TRY();
-}
+}        
 
 extern void MMExecute(void* work, int size)
 {

contrib/multimaster/multimaster.h

@@ -13,6 +13,7 @@ extern void MMJoinTransaction(TransactionId xid);
 extern bool MMIsLocalTransaction(TransactionId xid);
 extern void MMReceiverStarted(void);
 extern void MMExecute(void* work, int size);
+extern void MMExecutor(int id, void* work, size_t size);
 
 extern char* MMDatabaseName;
 

contrib/multimaster/pglogical_config.h

@@ -0,0 +1,55 @@
+#ifndef PG_LOGICAL_CONFIG_H
+#define PG_LOGICAL_CONFIG_H
+
+#ifndef PG_VERSION_NUM
+#error <postgres.h> must be included first
+#endif
+
+inline static bool
+server_float4_byval(void)
+{
+#ifdef USE_FLOAT4_BYVAL
+	return true;
+#else
+	return false;
+#endif
+}
+
+inline static bool
+server_float8_byval(void)
+{
+#ifdef USE_FLOAT8_BYVAL
+	return true;
+#else
+	return false;
+#endif
+}
+
+inline static bool
+server_integer_datetimes(void)
+{
+#ifdef USE_INTEGER_DATETIMES
+	return true;
+#else
+	return false;
+#endif
+}
+
+inline static bool
+server_bigendian(void)
+{
+#ifdef WORDS_BIGENDIAN
+	return true;
+#else
+	return false;
+#endif
+}
+
+typedef struct List List;
+typedef struct PGLogicalOutputData PGLogicalOutputData;
+
+extern int process_parameters(List *options, PGLogicalOutputData *data);
+
+extern List * prepare_startup_message(PGLogicalOutputData *data);
+
+#endif

contrib/multimaster/pglogical_hooks.h

@@ -0,0 +1,22 @@
+#ifndef PGLOGICAL_HOOKS_H
+#define PGLOGICAL_HOOKS_H
+
+#include "replication/reorderbuffer.h"
+
+/* public interface for hooks */
+#include "pglogical_output/hooks.h"
+
+extern void load_hooks(PGLogicalOutputData *data);
+
+extern void call_startup_hook(PGLogicalOutputData *data, List *plugin_params);
+
+extern void call_shutdown_hook(PGLogicalOutputData *data);
+
+extern bool call_row_filter_hook(PGLogicalOutputData *data,
+		ReorderBufferTXN *txn, Relation rel, ReorderBufferChange *change);
+
+extern bool call_txn_filter_hook(PGLogicalOutputData *data,
+		RepOriginId txn_origin);
+
+
+#endif