Set logical decoding hooks

Author: knizhnik

multimaster.c

@@ -53,6 +53,7 @@
 #include "nodes/makefuncs.h"
 #include "access/htup_details.h"
 #include "catalog/indexing.h"
+#include "pglogical_output/hooks.h"
 
 #include "multimaster.h"
 #include "ddd.h"
@@ -161,6 +162,7 @@ bool  MtmDoReplication;
 char* MtmDatabaseName;
 
 int   MtmNodeId;
+int   MtmReplicationNodeId;
 int   MtmArbiterPort;
 int   MtmNodes;
 int   MtmConnectAttempts;
@@ -1637,6 +1639,27 @@ void MtmDropNode(int nodeId, bool dropSlot)
 	}
 }
 
+static void 
+MtmReplicationShutdownHook(struct PGLogicalShutdownHookArgs* args)
+{
+	MtmOnNodeDisconnect(MtmReplicationNodeId);
+}
+
+static bool 
+MtmReplicationTxnFilterHook(struct PGLogicalTxnFilterArgs* args)
+{
+	elog(WARNING, "MtmReplicationTxnFilterHook: args->origin_id=%d, MtmReplicationNodeId=%d", args->origin_id, MtmReplicationNodeId);
+	return args->origin_id == InvalidRepOriginId || MtmIsRecoveredNode(MtmReplicationNodeId);
+}
+
+void MtmSetupReplicationHooks(struct PGLogicalHooks* hooks)
+{
+	hooks->shutdown_hook = MtmReplicationShutdownHook;
+	hooks->txn_filter_hook = MtmReplicationTxnFilterHook;
+}
+
+	
+
 /*
  * -------------------------------------------
  * SQL API functions
@@ -1988,16 +2011,42 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 		case T_ClosePortalStmt:
 		case T_FetchStmt:
 		case T_DoStmt:
+		case T_CreateTableSpaceStmt:
+		case T_DropTableSpaceStmt:
+		case T_AlterTableSpaceOptionsStmt:
+		case T_TruncateStmt:
+		case T_CommentStmt: /* XXX: we could replicate these */;
 		case T_CopyStmt:
 		case T_PrepareStmt:
 		case T_ExecuteStmt:
+		case T_DeallocateStmt:
+		case T_GrantStmt: /* XXX: we could replicate some of these these */;
+		case T_GrantRoleStmt:
+		case T_AlterDatabaseStmt:
+		case T_AlterDatabaseSetStmt:
 		case T_NotifyStmt:
 		case T_ListenStmt:
 		case T_UnlistenStmt:
 		case T_LoadStmt:
+		case T_ClusterStmt: /* XXX: we could replicate these */;
+		case T_VacuumStmt:
+		case T_ExplainStmt:
+		case T_AlterSystemStmt:
 		case T_VariableSetStmt:
 		case T_VariableShowStmt:
-			skipCommand = true;
+		case T_DiscardStmt:
+		case T_CreateEventTrigStmt:
+		case T_AlterEventTrigStmt:
+		case T_CreateRoleStmt:
+		case T_AlterRoleStmt:
+		case T_AlterRoleSetStmt:
+		case T_DropRoleStmt:
+		case T_ReassignOwnedStmt:
+		case T_LockStmt:
+		case T_ConstraintsSetStmt:
+		case T_CheckPointStmt:
+		case T_ReindexStmt:
+		    skipCommand = true;
 			break;
 	    default:
 			skipCommand = false;

multimaster.h

@@ -5,6 +5,8 @@
 #include "bgwpool.h"
 #include "bkb.h"
 
+#include "pglogical_output/hooks.h"
+
 #define MTM_TUPLE_TRACE(fmt, ...)
 /*
 #define MTM_INFO(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__) 
@@ -149,6 +151,7 @@ extern char const* const MtmNodeStatusMnem[];
 extern MtmState* Mtm;
 
 extern int   MtmNodeId;
+extern int   MtmReplicationNodeId;
 extern int   MtmNodes;
 extern int   MtmArbiterPort;
 extern char* MtmDatabaseName;
@@ -192,5 +195,5 @@ extern bool  MtmIsRecoveredNode(int nodeId);
 extern void  MtmRefreshClusterStatus(bool nowait);
 extern void  MtmSwitchClusterMode(MtmNodeStatus mode);
 extern void  MtmUpdateNodeConnectionInfo(MtmConnectionInfo* conn, char const* connStr);
-
+extern void  MtmSetupReplicationHooks(struct PGLogicalHooks* hooks);
 #endif

pglogical_hooks.c

@@ -106,6 +106,12 @@ load_hooks(PGLogicalOutputData *data)
 				data->hooks.row_filter_hook,
 				data->hooks.txn_filter_hook,
 				data->hooks.hooks_private_data);
+	} 
+	else if (data->api->setup_hooks) 
+	{ 
+		old_ctxt = MemoryContextSwitchTo(data->hooks_mctxt);
+		(*data->api->setup_hooks)(&data->hooks);
+		MemoryContextSwitchTo(old_ctxt);
 	}
 
 	if (txn_started)

pglogical_output.c

@@ -343,7 +343,7 @@ pg_decode_startup(LogicalDecodingContext * ctx, OutputPluginOptions *opt,
 			data->forward_changeset_origins = false;
 		}
 
-		if (data->hooks_setup_funcname != NIL)
+		if (data->hooks_setup_funcname != NIL || data->api->setup_hooks)
 		{
 
 			data->hooks_mctxt = AllocSetContextCreate(ctx->context,

pglogical_proto.c

@@ -37,12 +37,7 @@
 
 #include "multimaster.h"
 
-typedef struct PGLogicalProtoMM
-{
-    PGLogicalProtoAPI api;
-	int   nodeId;
-    bool  isLocal;
-} PGLogicalProtoMM;
+static bool MtmIsFilteredTxn;
 
 static void pglogical_write_rel(StringInfo out, PGLogicalOutputData *data, Relation rel);
 
@@ -72,30 +67,31 @@ static char decide_datum_transfer(Form_pg_attribute att,
 static void
 pglogical_write_rel(StringInfo out, PGLogicalOutputData *data, Relation rel)
 {
-    PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
-    if (!mm->isLocal) { 
-        const char *nspname;
-        uint8		nspnamelen;
-        const char *relname;
-        uint8		relnamelen;
-        
-        pq_sendbyte(out, 'R');		/* sending RELATION */
-        
-        nspname = get_namespace_name(rel->rd_rel->relnamespace);
-        if (nspname == NULL)
-            elog(ERROR, "cache lookup failed for namespace %u",
-                 rel->rd_rel->relnamespace);
-        nspnamelen = strlen(nspname) + 1;
-        
-        relname = NameStr(rel->rd_rel->relname);
-        relnamelen = strlen(relname) + 1;
-        
-        pq_sendbyte(out, nspnamelen);		/* schema name length */
-        pq_sendbytes(out, nspname, nspnamelen);
-        
-        pq_sendbyte(out, relnamelen);		/* table name length */
-        pq_sendbytes(out, relname, relnamelen);
-    }
+	const char *nspname;
+	uint8		nspnamelen;
+	const char *relname;
+	uint8		relnamelen;
+		
+    if (MtmIsFilteredTxn) { 
+		return;
+	}
+	
+	pq_sendbyte(out, 'R');		/* sending RELATION */
+    
+	nspname = get_namespace_name(rel->rd_rel->relnamespace);
+	if (nspname == NULL)
+		elog(ERROR, "cache lookup failed for namespace %u",
+			 rel->rd_rel->relnamespace);
+	nspnamelen = strlen(nspname) + 1;
+	
+	relname = NameStr(rel->rd_rel->relname);
+	relnamelen = strlen(relname) + 1;
+    
+	pq_sendbyte(out, nspnamelen);		/* schema name length */
+	pq_sendbytes(out, nspname, nspnamelen);
+    
+	pq_sendbyte(out, relnamelen);		/* table name length */
+	pq_sendbytes(out, relname, relnamelen);
 }
 
 /*
@@ -105,21 +101,19 @@ static void
 pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 					  ReorderBufferTXN *txn)
 {
-    PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
+	bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
 	csn_t csn = MtmTransactionSnapshot(txn->xid);
-	bool isRecovery = MtmIsRecoveredNode(mm->nodeId);
 	MTM_TRACE("pglogical_write_begin %d CSN=%ld\n", txn->xid, csn);
-    if (csn == INVALID_CSN && !isRecovery) {
-		//Assert(txn->origin_id != InvalidRepOriginId);
-        mm->isLocal = true; 
-    } else { 
-        mm->isLocal = false;        
-		//Assert(txn->origin_id == InvalidRepOriginId || isRecovery);
-        pq_sendbyte(out, 'B');		/* BEGIN */
+	
+	if (csn == INVALID_CSN && !isRecovery) { 
+		MtmIsFilteredTxn = true;
+	} else { 
+		pq_sendbyte(out, 'B');		/* BEGIN */
 		pq_sendint(out, MtmNodeId, 4);
 		pq_sendint(out, isRecovery ? InvalidTransactionId : txn->xid, 4);
-        pq_sendint64(out, csn);
-    }
+		pq_sendint64(out, csn);
+		MtmIsFilteredTxn = false;
+	}
 }
 
 /*
@@ -129,9 +123,11 @@ static void
 pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 					   ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
 {
-    PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
     uint8 flags = 0;
 
+    if (MtmIsFilteredTxn) { 
+		return;
+	}
     if (txn->xact_action == XLOG_XACT_COMMIT) 
     	flags = PGLOGICAL_COMMIT;
 	else if (txn->xact_action == XLOG_XACT_PREPARE)
@@ -143,18 +139,19 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 	else
     	Assert(false);
 
-
+#if 0
 	if (flags == PGLOGICAL_COMMIT || flags == PGLOGICAL_PREPARE) { 
 		if (mm->isLocal) { 
 			return;
 		}
 	} else { 
 		csn_t csn = MtmTransactionSnapshot(txn->xid);
-		bool isRecovery = MtmIsRecoveredNode(mm->nodeId);
+		bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
 		if (csn == INVALID_CSN && !isRecovery) {
 			return;
 		}
 	}
+#endif
 
     pq_sendbyte(out, 'C');		/* sending COMMIT */
 
@@ -185,11 +182,10 @@ static void
 pglogical_write_insert(StringInfo out, PGLogicalOutputData *data,
 						Relation rel, HeapTuple newtuple)
 {
-    PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
-    if (!mm->isLocal) { 
-        pq_sendbyte(out, 'I');		/* action INSERT */
-        pglogical_write_tuple(out, data, rel, newtuple);
-    }
+    if (!MtmIsFilteredTxn) { 
+		pq_sendbyte(out, 'I');		/* action INSERT */
+		pglogical_write_tuple(out, data, rel, newtuple);
+	}
 }
 
 /*
@@ -199,32 +195,31 @@ static void
 pglogical_write_update(StringInfo out, PGLogicalOutputData *data,
 						Relation rel, HeapTuple oldtuple, HeapTuple newtuple)
 {
-    PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
-    if (!mm->isLocal) { 
-        pq_sendbyte(out, 'U');		/* action UPDATE */
-        /* FIXME support whole tuple (O tuple type) */
-        if (oldtuple != NULL)
-        {
-            pq_sendbyte(out, 'K');	/* old key follows */
-            pglogical_write_tuple(out, data, rel, oldtuple);
-	    }
-
-        pq_sendbyte(out, 'N');		/* new tuple follows */
-        pglogical_write_tuple(out, data, rel, newtuple);
-    }
+    if (!MtmIsFilteredTxn) { 
+		pq_sendbyte(out, 'U');		/* action UPDATE */
+		/* FIXME support whole tuple (O tuple type) */
+		if (oldtuple != NULL)
+		{
+			pq_sendbyte(out, 'K');	/* old key follows */
+			pglogical_write_tuple(out, data, rel, oldtuple);
+		}
+		
+		pq_sendbyte(out, 'N');		/* new tuple follows */
+		pglogical_write_tuple(out, data, rel, newtuple);
+	}
 }
+	
 /*
  * Write DELETE to the output stream.
  */
 static void
 pglogical_write_delete(StringInfo out, PGLogicalOutputData *data,
 						Relation rel, HeapTuple oldtuple)
 {
-    PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
-    if (!mm->isLocal) { 
-        pq_sendbyte(out, 'D');		/* action DELETE */
-        pglogical_write_tuple(out, data, rel, oldtuple);
-    }
+    if (!MtmIsFilteredTxn) {
+		pq_sendbyte(out, 'D');		/* action DELETE */
+		pglogical_write_tuple(out, data, rel, oldtuple);
+	}
 }
 
 /*
@@ -422,16 +417,16 @@ decide_datum_transfer(Form_pg_attribute att, Form_pg_type typclass,
 PGLogicalProtoAPI *
 pglogical_init_api(PGLogicalProtoType typ)
 {
-	PGLogicalProtoMM* pmm = palloc0(sizeof(PGLogicalProtoMM));
-    PGLogicalProtoAPI* res = &pmm->api;
-    pmm->isLocal = false;
-	sscanf(MyReplicationSlot->data.name.data, MULTIMASTER_SLOT_PATTERN, &pmm->nodeId);
+    PGLogicalProtoAPI* res = palloc0(sizeof(PGLogicalProtoAPI));
+	sscanf(MyReplicationSlot->data.name.data, MULTIMASTER_SLOT_PATTERN, &MtmReplicationNodeId);
+	elog(WARNING, "%d: PRGLOGICAL init API for slot %s node %d", MyProcPid, MyReplicationSlot->data.name.data, MtmReplicationNodeId);
     res->write_rel = pglogical_write_rel;
     res->write_begin = pglogical_write_begin;
     res->write_commit = pglogical_write_commit;
     res->write_insert = pglogical_write_insert;
     res->write_update = pglogical_write_update;
     res->write_delete = pglogical_write_delete;
+	res->setup_hooks = MtmSetupReplicationHooks;
     res->write_startup_message = write_startup_message;
     return res;
 }

pglogical_proto.h

@@ -33,6 +33,8 @@ typedef void (*pglogical_write_delete_fn)(StringInfo out, PGLogicalOutputData *d
 
 typedef void (*write_startup_message_fn)(StringInfo out, List *msg);
 
+typedef void (*pglogical_setup_hooks_fn)(struct PGLogicalHooks* hooks);
+
 typedef struct PGLogicalProtoAPI
 {
 	pglogical_write_rel_fn		write_rel;
@@ -42,7 +44,8 @@ typedef struct PGLogicalProtoAPI
 	pglogical_write_insert_fn	write_insert;
 	pglogical_write_update_fn	write_update;
 	pglogical_write_delete_fn	write_delete;
-	write_startup_message_fn	write_startup_message;
+	pglogical_setup_hooks_fn    setup_hooks;
+	write_startup_message_fn	write_startup_message;	
 } PGLogicalProtoAPI;