Avoid redundand sending of relation to replicas

Author: knizhnik

pglogical_apply.c

@@ -62,6 +62,7 @@ typedef struct TupleData
 	bool		changed[MaxTupleAttributeNumber];
 } TupleData;
 
+
 static Relation read_rel(StringInfo s, LOCKMODE mode);
 static void read_tuple_parts(StringInfo s, Relation rel, TupleData *tup);
 static EState* create_rel_estate(Relation rel);
@@ -581,6 +582,15 @@ read_tuple_parts(StringInfo s, Relation rel, TupleData *tup)
 	}
 }
 
+static void
+close_rel(Relation rel)
+{
+	if (rel != NULL) 
+	{
+		heap_close(rel, NoLock);	   
+	} 		
+}
+
 static Relation 
 read_rel(StringInfo s, LOCKMODE mode)
 {
@@ -823,7 +833,6 @@ process_remote_insert(StringInfo s, Relation rel)
 		MtmMakeRelationLocal(RelationGetRelid(rel));
 	}
 
-    heap_close(rel, NoLock);
     ExecResetTupleTable(estate->es_tupleTable, true);
     FreeExecutorState(estate);
 
@@ -944,7 +953,6 @@ process_remote_update(StringInfo s, Relation rel)
 
 	/* release locks upon commit */
 	index_close(idxrel, NoLock);
-	heap_close(rel, NoLock);
 
 	ExecResetTupleTable(estate->es_tupleTable, true);
 	FreeExecutorState(estate);
@@ -1019,7 +1027,6 @@ process_remote_delete(StringInfo s, Relation rel)
 	PopActiveSnapshot();
 
 	index_close(idxrel, NoLock);
-	heap_close(rel, NoLock);
 
 	ExecResetTupleTable(estate->es_tupleTable, true);
 	FreeExecutorState(estate);
@@ -1071,6 +1078,7 @@ void MtmExecutor(void* work, size_t size)
 				}
                 /* COMMIT */
             case 'C':
+  			    close_rel(rel);
                 process_remote_commit(&s);
                 break;
                 /* INSERT */
@@ -1086,6 +1094,7 @@ void MtmExecutor(void* work, size_t size)
                 process_remote_delete(&s, rel);
                 continue;
             case 'R':
+  			    close_rel(rel);
                 rel = read_rel(&s, RowExclusiveLock);
                 continue;
 			case 'F':

pglogical_proto.c

@@ -43,6 +43,7 @@ static bool MtmIsFilteredTxn;
 static TransactionId MtmCurrentXid;
 static bool DDLInProgress = false;
 static Oid MtmSenderTID; /* transaction identifier for WAL sender */
+static Oid MtmLastRelId; /* last relation ID sent to the receiver in this transaction */
 
 static void pglogical_write_rel(StringInfo out, PGLogicalOutputData *data, Relation rel);
 
@@ -95,6 +96,11 @@ pglogical_write_rel(StringInfo out, PGLogicalOutputData *data, Relation rel)
 
 	relid = RelationGetRelid(rel);
 
+	if (relid == MtmLastRelId) { 
+		return;
+	}
+	MtmLastRelId = relid;
+
 	pq_sendbyte(out, 'R');		/* sending RELATION */	
 	pq_sendint(out, relid, sizeof relid); /* use Oid as relation identifier */
 
@@ -143,6 +149,7 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 			pglogical_relid_map_reset();
 			MtmSenderTID += 1; /* skip InvalidOid */
 		}
+		MtmLastRelId = InvalidOid;
 		MtmCurrentXid = txn->xid;
 		MtmIsFilteredTxn = false;
 		MTM_LOG3("%d: pglogical_write_begin XID=%d node=%d CSN=%lld recovery=%d restart_decoding_lsn=%llx first_lsn=%llx end_lsn=%llx confirmed_flush=%llx",