Fix raftable_wrapper

Author: knizhnik

contrib/mmts/multimaster.c

@@ -2354,7 +2354,7 @@ MtmDetectGlobalDeadLock(PGPROC* proc)
         ByteBufferFree(&buf);
 		for (i = 0; i < MtmNodes; i++) { 
 			if (i+1 != MtmNodeId && !BIT_CHECK(Mtm->disabledNodeMask, i)) { 
-				int size;
+				size_t size;
 				void* data = RaftableGet(psprintf("lock-graph-%d", i+1), &size, NULL, true);
 				if (data == NULL) { 
 					return true; /* If using Raftable is disabled */
@@ -2369,31 +2369,3 @@ MtmDetectGlobalDeadLock(PGPROC* proc)
 	}
     return hasDeadlock;
 }
-
-void* RaftableGet(char const* key, int* size, RaftableTimestamp* ts, bool nowait)
-{
-	size_t s;
-	char *value;
-
-	if (!MtmUseRaftable) return NULL;
-
-	Assert(ts == NULL); /* not implemented */
-	value = raftable_get(key, &s);
-	*size = s;
-	return value;
-}
-
-void  RaftableSet(char const* key, void const* value, int size, bool nowait)
-{
-	if (!MtmUseRaftable) return;
-
-	raftable_set(key, value, size, nowait ? 0 : -1);
-}
-
-bool  RaftableCAS(char const* key, char const* value, bool nowait)
-{
-	if (!MtmUseRaftable) return false;
-
-	Assert(false); /* not implemented */
-	return false;
-}

contrib/mmts/pglogical_output.c

@@ -480,17 +480,23 @@ pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 /*
  * Decide if the whole transaction with specific origin should be filtered out.
  */
+extern int MtmReplicationNodeId;
+
 static bool
 pg_decode_origin_filter(LogicalDecodingContext *ctx,
 						RepOriginId origin_id)
 {
 	PGLogicalOutputData *data = ctx->output_plugin_private;
 
-	if (!call_txn_filter_hook(data, origin_id))
+	if (!call_txn_filter_hook(data, origin_id)) { 
+		elog(WARNING, "Record with origin %d is not sent to node %d", origin_id, MtmReplicationNodeId);
 		return true;
+	}
 
-	if (!data->forward_changesets && origin_id != InvalidRepOriginId)
+	if (!data->forward_changesets && origin_id != InvalidRepOriginId) {
+		*(int*)0 = 0;
 		return true;
+	}
 
 	return false;
 }

contrib/mmts/raftable.c

@@ -19,20 +19,28 @@ static void RaftableResolve()
 /*
  * Raftable function proxies
  */
-void* RaftableGet(char const* key, int* size, RaftableTimestamp* ts, bool nowait)
+void* RaftableGet(char const* key, size_t* size, RaftableTimestamp* ts, bool nowait)
 {
 	if (!MtmUseRaftable) { 
 		return NULL;
 	}
 	RaftableResolve();
-	return (*raftable_get_impl)(key, size, nowait ? 0 : -1);
+	return (*raftable_get_impl)(key, size);
 }
 
 
-void RaftableSet(char const* key, void const* value, int size, bool nowait)
+void RaftableSet(char const* key, void const* value, size_t size, bool nowait)
 {
 	if (MtmUseRaftable) { 
 		RaftableResolve();
 		(*raftable_set_impl)(key, value, size, nowait ? 0 : -1);
 	}		
 }
+
+bool RaftableCAS(char const* key, char const* value, bool nowait)
+{
+	if (!MtmUseRaftable) return false;
+
+	Assert(false); /* not implemented */
+	return false;
+}

contrib/mmts/raftable_wrapper.h

@@ -15,13 +15,13 @@ typedef struct RaftableTimestamp {
  * If "ts" is not NULL, then it is assigned timestamp of last update of this value
  * If RAFT master is not accessible, then depending non value of "nowait" parameter, this funciton should either block until RAFT quorum is reached, either report error.
  */
-extern void* RaftableGet(char const* key, int* size, RaftableTimestamp* ts, bool nowait);
+extern void* RaftableGet(char const* key, size_t* size, RaftableTimestamp* ts, bool nowait);
 
 /*
  * Set new value for the specified key. IF value is NULL, then key should be deleted.
  * If RAFT master is not accessible, then depending non value of "nowait" parameter, this funciton should either block until RAFT quorum is reached, either report error.
  */
-extern void  RaftableSet(char const* key, void const* value, int size, bool nowait);
+extern void  RaftableSet(char const* key, void const* value, size_t size, bool nowait);
 
 /* 
  * If key doesn't exists or its value is not equal to the specified value then store this value and return true.
@@ -30,8 +30,8 @@ extern void  RaftableSet(char const* key, void const* value, int size, bool nowa
  */
 extern bool  RaftableCAS(char const* key, char const* value, bool nowait);
 
-typedef void* (*raftable_get_t)(char const* key, int* size, int timeout);
-typedef void (*raftable_set_t)(char const* key, void const* value, int size, int timeout);
+typedef void* (*raftable_get_t)(char const* key, size_t* size);
+typedef void (*raftable_set_t)(char const* key, void const* value, size_t size, int timeout_ms);
 
 extern bool MtmUseRaftable;