fix pglogical protocol between decoder and receiver

Author: knizhnik

contrib/multimaster/pglogical_proto.c

@@ -131,17 +131,7 @@ pglogical_write_insert(StringInfo out, PGLogicalOutputData *data,
 {
     PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
     if (!mm->isLocal) { 
-        uint8 flags = 0;
-        
         pq_sendbyte(out, 'I');		/* action INSERT */
-        
-        /* send the flags field */
-        pq_sendbyte(out, flags);
-        
-        /* use Oid as relation identifier */
-        pq_sendint(out, RelationGetRelid(rel), 4);
-        
-        pq_sendbyte(out, 'N');		/* new tuple follows */
         pglogical_write_tuple(out, data, rel, newtuple);
     }
 }
@@ -155,15 +145,7 @@ pglogical_write_update(StringInfo out, PGLogicalOutputData *data,
 {
     PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
     if (!mm->isLocal) { 
-        uint8 flags = 0;
         pq_sendbyte(out, 'U');		/* action UPDATE */
-        
-        /* send the flags field */
-        pq_sendbyte(out, flags);
-        
-        /* use Oid as relation identifier */
-        pq_sendint(out, RelationGetRelid(rel), 4);
-        
         /* FIXME support whole tuple (O tuple type) */
         if (oldtuple != NULL)
         {
@@ -184,18 +166,7 @@ pglogical_write_delete(StringInfo out, PGLogicalOutputData *data,
 {
     PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
     if (!mm->isLocal) { 
-        uint8 flags = 0;
-
         pq_sendbyte(out, 'D');		/* action DELETE */
-
-        /* send the flags field */
-        pq_sendbyte(out, flags);
-        
-        /* use Oid as relation identifier */
-        pq_sendint(out, RelationGetRelid(rel), 4);
-        
-        /* FIXME support whole tuple (O tuple type) */
-        pq_sendbyte(out, 'K');	/* old key follows */
         pglogical_write_tuple(out, data, rel, oldtuple);
     }
 }
@@ -207,20 +178,6 @@ pglogical_write_delete(StringInfo out, PGLogicalOutputData *data,
 static void
 write_startup_message(StringInfo out, List *msg)
 {
-#if 0
-	ListCell *lc;
-
-	pq_sendbyte(out, 'S');	/* message type field */
-	pq_sendbyte(out, 1); 	/* startup message version */
-	foreach (lc, msg)
-	{
-		DefElem *param = (DefElem*)lfirst(lc);
-		Assert(IsA(param->arg, String) && strVal(param->arg) != NULL);
-		/* null-terminated key and value pairs, in client_encoding */
-		pq_sendstring(out, param->defname);
-		pq_sendstring(out, strVal(param->arg));
-	}
-#endif
 }
 
 /*
@@ -289,11 +246,10 @@ pglogical_write_tuple(StringInfo out, PGLogicalOutputData *data,
 		transfer_type = decide_datum_transfer(att, typclass,
 											  data->allow_internal_basetypes,
 											  data->allow_binary_basetypes);
-
+        pq_sendbyte(out, transfer_type);
 		switch (transfer_type)
 		{
-			case 'i':
-				pq_sendbyte(out, 'i');	/* internal-format binary data follows */
+			case 'b':	/* internal-format binary data follows */
 
 				/* pass by value */
 				if (att->attbyval)
@@ -338,13 +294,11 @@ pglogical_write_tuple(StringInfo out, PGLogicalOutputData *data,
 
 				break;
 
-			case 'b':
+			case 's': /* binary send/recv data follows */
 				{
 					bytea	   *outputbytes;
 					int			len;
 
-					pq_sendbyte(out, 'b');	/* binary send/recv data follows */
-
 					outputbytes = OidSendFunctionCall(typclass->typsend,
 													  values[i]);
 
@@ -360,8 +314,6 @@ pglogical_write_tuple(StringInfo out, PGLogicalOutputData *data,
 					char   	   *outputstr;
 					int			len;
 
-					pq_sendbyte(out, 't');	/* 'text' data follows */
-
 					outputstr =	OidOutputFunctionCall(typclass->typoutput,
 													  values[i]);
 					len = strlen(outputstr) + 1;
@@ -391,7 +343,7 @@ decide_datum_transfer(Form_pg_attribute att, Form_pg_type typclass,
 		att->atttypid < FirstNormalObjectId &&
 		typclass->typelem == InvalidOid)
 	{
-		return 'i';
+		return 'b';
 	}
 	/*
 	 * Use send/recv, if allowed, if the type is plain or builtin.
@@ -404,7 +356,7 @@ decide_datum_transfer(Form_pg_attribute att, Form_pg_type typclass,
 			 (att->atttypid < FirstNormalObjectId || typclass->typtype != 'c') &&
 			 (att->atttypid < FirstNormalObjectId || typclass->typelem == InvalidOid))
 	{
-		return 'b';
+		return 's';
 	}
 
 	return 't';

contrib/multimaster/pglogical_receiver.c

@@ -66,9 +66,9 @@ static void UserTableUpdateIndexes(EState *estate, TupleTableSlot *slot);
 
 static void process_remote_begin(StringInfo s);
 static void process_remote_commit(StringInfo s);
-static void process_remote_insert(StringInfo s);
-static void process_remote_update(StringInfo s);
-static void process_remote_delete(StringInfo s);
+static void process_remote_insert(StringInfo s, Relation rel);
+static void process_remote_update(StringInfo s, Relation rel);
+static void process_remote_delete(StringInfo s, Relation rel);
 
 /*
  * Search the index 'idxrel' for a tuple identified by 'skey' in 'rel'.
@@ -346,7 +346,7 @@ read_tuple_parts(StringInfo s, Relation rel, TupleData *tup)
 	memset(tup->isnull, 1, sizeof(tup->isnull));
 	memset(tup->changed, 1, sizeof(tup->changed));
 
-	rnatts = pq_getmsgint(s, 4);
+	rnatts = pq_getmsgint(s, 2);
 
 	if (desc->natts != rnatts)
 		elog(ERROR, "tuple natts mismatch, %u vs %u", desc->natts, rnatts);
@@ -370,7 +370,6 @@ read_tuple_parts(StringInfo s, Relation rel, TupleData *tup)
 				tup->isnull[i] = true;
 				tup->changed[i] = false;
 				tup->values[i] = 0xdeadbeef; /* make bad usage more obvious */
-
 				break;
 
 			case 'b': /* binary format */
@@ -463,25 +462,16 @@ process_remote_commit(StringInfo s)
 }
 
 static void
-process_remote_insert(StringInfo s)
+process_remote_insert(StringInfo s, Relation rel)
 {
-	char		action;
 	EState	   *estate;
 	TupleData   new_tuple;
 	TupleTableSlot *newslot;
 	TupleTableSlot *oldslot;
-	Relation	rel;
 	ResultRelInfo *relinfo;
 	ScanKey	   *index_keys;
 	int			i;
 
-	rel = read_rel(s, RowExclusiveLock);
-
-	action = pq_getmsgbyte(s);
-	if (action != 'N')
-		elog(ERROR, "expected new tuple but got %d",
-			 action);
-
 	estate = create_rel_estate(rel);
 	newslot = ExecInitExtraTupleSlot(estate);
 	oldslot = ExecInitExtraTupleSlot(estate);
@@ -565,7 +555,7 @@ process_remote_insert(StringInfo s)
 }
 
 static void
-process_remote_update(StringInfo s)
+process_remote_update(StringInfo s, Relation rel)
 {
 	char		action;
 	EState	   *estate;
@@ -576,13 +566,10 @@ process_remote_update(StringInfo s)
 	TupleData   old_tuple;
 	TupleData   new_tuple;
 	Oid			idxoid;
-	Relation	rel;
 	Relation	idxrel;
 	ScanKeyData skey[INDEX_MAX_KEYS];
 	HeapTuple	remote_tuple = NULL;
 
-	rel = read_rel(s, RowExclusiveLock);
-
 	action = pq_getmsgbyte(s);
 
 	/* old key present, identifying key changed */
@@ -690,32 +677,16 @@ process_remote_update(StringInfo s)
 }
 
 static void
-process_remote_delete(StringInfo s)
+process_remote_delete(StringInfo s, Relation rel)
 {
-	char		action;
 	EState	   *estate;
 	TupleData   oldtup;
 	TupleTableSlot *oldslot;
 	Oid			idxoid;
-	Relation	rel;
 	Relation	idxrel;
 	ScanKeyData skey[INDEX_MAX_KEYS];
 	bool		found_old;
 
-	rel = read_rel(s, RowExclusiveLock);
-
-	action = pq_getmsgbyte(s);
-
-	if (action != 'K' && action != 'E')
-		elog(ERROR, "expected action K or E got %c", action);
-
-	if (action == 'E')
-	{
-		elog(WARNING, "got delete without pkey");
-		heap_close(rel, NoLock);
-		return;
-	}
-
 	estate = create_rel_estate(rel);
 	oldslot = ExecInitExtraTupleSlot(estate);
 	ExecSetSlotDescriptor(oldslot, RelationGetDescr(rel));
@@ -784,6 +755,7 @@ process_remote_delete(StringInfo s)
 void MMExecutor(int id, void* work, size_t size)
 {
     StringInfoData s;
+    Relation rel = NULL;
     initStringInfo(&s);
     s.data = work;
     s.len = size;
@@ -796,26 +768,29 @@ void MMExecutor(int id, void* work, size_t size)
 
             switch (action) {
                 /* BEGIN */
-              case 'B':
+            case 'B':
                 process_remote_begin(&s);
                 continue;
                 /* COMMIT */
-              case 'C':
+            case 'C':
                 process_remote_commit(&s);
                 break;
                 /* INSERT */
-              case 'I':
-                process_remote_insert(&s);
+            case 'I':
+                process_remote_insert(&s, rel);
                 continue;
                 /* UPDATE */
-              case 'U':
-                process_remote_update(&s);
+            case 'U':
+                process_remote_update(&s, rel);
                 continue;
                 /* DELETE */
-              case 'D':
-                process_remote_delete(&s);
+            case 'D':
+                process_remote_delete(&s, rel);
+                continue;
+            case 'R':
+                rel = read_rel(&s, RowExclusiveLock);
                 continue;
-              default:
+            default:
                 elog(ERROR, "unknown action of type %c", action);
             }        
             break;