PGPRO-731 # Table copy fixes

Author: knizhnik

multimaster.c

@@ -4422,7 +4422,8 @@ Datum mtm_broadcast_table(PG_FUNCTION_ARGS)
 	MtmCopyRequest copy;
 	copy.sourceTable = PG_GETARG_OID(0);
 	copy.targetNodes = PG_GETARG_INT64(1);
-	LogLogicalMessage("B", (char*)&copy, sizeof(copy), false);
+	LogLogicalMessage("B", (char*)&copy, sizeof(copy), true);
+	MtmTx.containsDML = true;
 	PG_RETURN_VOID();
 }
 
@@ -4431,7 +4432,8 @@ Datum mtm_copy_table(PG_FUNCTION_ARGS)
 	MtmCopyRequest copy;
 	copy.sourceTable = PG_GETARG_OID(0);
 	copy.targetNodes = (nodemask_t)1 << (PG_GETARG_INT32(1) - 1);
-	LogLogicalMessage("B", (char*)&copy, sizeof(copy), false);
+	LogLogicalMessage("B", (char*)&copy, sizeof(copy), true);
+	MtmTx.containsDML = true;
 	PG_RETURN_VOID();
 }
 

pglogical_apply.c

@@ -368,49 +368,6 @@ process_remote_begin(StringInfo s)
 	return true;
 }
 
-static void 
-process_broadcast_table(StringInfo s)
-{
-	Relation rel;
-	char ch;
-	EState *estate;
-	TupleData new_tuple;
-	TupleTableSlot *newslot;
-	TupleTableSlot *oldslot;
-	HeapTuple tup;
-
-	StartTransactionCommand();
-
-    ch = pq_getmsgbyte(s);
-	Assert(ch == 'R');
-	rel = read_rel(s, AccessExclusiveLock);
-
-	heap_truncate_one_rel(rel);
-
-	estate = create_rel_estate(rel);
-	newslot = ExecInitExtraTupleSlot(estate);
-	oldslot = ExecInitExtraTupleSlot(estate);
-	ExecSetSlotDescriptor(newslot, RelationGetDescr(rel));
-	ExecSetSlotDescriptor(oldslot, RelationGetDescr(rel));
-
-	ExecOpenIndices(estate->es_result_relation_info, false);
-	
-	while (s->cursor != s->len) { 
-		read_tuple_parts(s, rel, &new_tuple);
-		tup = heap_form_tuple(RelationGetDescr(rel),
-							  new_tuple.values, new_tuple.isnull);
-		ExecStoreTuple(tup, newslot, InvalidBuffer, true);
-		simple_heap_insert(rel, newslot->tts_tuple);
-		UserTableUpdateOpenIndexes(estate, newslot);
-	}
-
-	ExecCloseIndices(estate->es_result_relation_info);
-    ExecResetTupleTable(estate->es_tupleTable, true); 
-    FreeExecutorState(estate);
-
-	CommitTransactionCommand();
-}
-
 static bool
 process_remote_message(StringInfo s)
 {
@@ -421,12 +378,6 @@ process_remote_message(StringInfo s)
 
 	switch (action)
 	{
-  	    case 'B':
-		{
-			process_broadcast_table(s);
-			standalone = true;
-			break;
-		}
 		case 'C':
 		{
 			MTM_LOG1("%d: Executing non-tx utility statement %s", MyProcPid, messageBody);
@@ -1206,6 +1157,9 @@ void MtmExecutor(void* work, size_t size)
 				s.len = save_len;
 				break;
 			}
+		    case '0':
+			    heap_truncate_one_rel(rel);
+				break;
 			case 'M':
 			{
   			    close_rel(rel);

pglogical_output.c

@@ -81,14 +81,14 @@ static bool startup_message_sent = false;
 
 #define OUTPUT_BUFFER_SIZE (16*1024*1024) 
 
-static void MtmOutputPluginWrite(LogicalDecodingContext *ctx, bool last_write, bool flush)
+void MtmOutputPluginWrite(LogicalDecodingContext *ctx, bool last_write, bool flush)
 {
 	if (flush) {
 		OutputPluginWrite(ctx, last_write);
 	}
 }
 
-static void MtmOutputPluginPrepareWrite(LogicalDecodingContext *ctx, bool last_write, bool flush)
+void MtmOutputPluginPrepareWrite(LogicalDecodingContext *ctx, bool last_write, bool flush)
 {
 	if (!ctx->prepared_write) { 
 		OutputPluginPrepareWrite(ctx, last_write);
@@ -557,7 +557,7 @@ pg_decode_message(LogicalDecodingContext *ctx,
 	PGLogicalOutputData* data = (PGLogicalOutputData*)ctx->output_plugin_private;
 
 	MtmOutputPluginPrepareWrite(ctx, true, !transactional);
-	data->api->write_message(ctx->out, data, prefix, sz, message);
+	data->api->write_message(ctx->out, ctx, prefix, sz, message);
 	MtmOutputPluginWrite(ctx, true, !transactional);
 }
 

pglogical_output.h

@@ -102,4 +102,7 @@ typedef struct PGLogicalTupleData
 	bool	changed[MaxTupleAttributeNumber];
 } PGLogicalTupleData;
 
+extern void MtmOutputPluginWrite(LogicalDecodingContext *ctx, bool last_write, bool flush);
+extern void MtmOutputPluginPrepareWrite(LogicalDecodingContext *ctx, bool last_write, bool flush);
+
 #endif /* PG_LOGICAL_OUTPUT_H */

pglogical_proto.c

@@ -168,38 +168,34 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 	}
 }
 
-static void pglogical_broadcast_table(StringInfo out, PGLogicalOutputData *data, MtmCopyRequest* copy)
+static void pglogical_broadcast_table(StringInfo out, LogicalDecodingContext *ctx, MtmCopyRequest* copy)
 {
 	if (BIT_CHECK(copy->targetNodes, MtmReplicationNodeId-1)) { 
 		HeapScanDesc scandesc;
 		HeapTuple	 tuple;
 		Relation     rel;
 
-		StartTransactionCommand();
-
 		rel = heap_open(copy->sourceTable, ShareLock);
 
-		pq_sendbyte(out, 'M');
-		pq_sendbyte(out, 'B');
-		pq_sendint(out, sizeof(*copy), 4);
-		pq_sendbytes(out, (char*)copy, sizeof(*copy));
-		
-		pglogical_write_rel(out, data, rel);
+		pglogical_write_rel(out, ctx->output_plugin_private, rel);
+
+		pq_sendbyte(out, '0');
 
 		scandesc = heap_beginscan(rel, GetTransactionSnapshot(), 0, NULL);
 		while ((tuple = heap_getnext(scandesc, ForwardScanDirection)) != NULL)
 		{
-			pglogical_write_tuple(out, data, rel, tuple);
+			MtmOutputPluginPrepareWrite(ctx, false, false);
+			pq_sendbyte(out, 'I');		/* action INSERT */
+			pglogical_write_tuple(out, ctx->output_plugin_private, rel, tuple);
+			MtmOutputPluginWrite(ctx, false, false);
 		}
 		heap_endscan(scandesc);
 		heap_close(rel, ShareLock);
-
-		CommitTransactionCommand();
 	}
 }
 
 static void
-pglogical_write_message(StringInfo out, PGLogicalOutputData *data,
+pglogical_write_message(StringInfo out, LogicalDecodingContext *ctx,
 						const char *prefix, Size sz, const char *message)
 {
 	MtmLastRelId = InvalidOid;
@@ -231,7 +227,7 @@ pglogical_write_message(StringInfo out, PGLogicalOutputData *data,
 		 */
 		return;
 	  case 'B':
-		pglogical_broadcast_table(out, data, (MtmCopyRequest*)message);
+		pglogical_broadcast_table(out, ctx, (MtmCopyRequest*)message);
 		return;
 	}
 	pq_sendbyte(out, 'M');

pglogical_proto.h

@@ -21,7 +21,7 @@ typedef void (*pglogical_write_rel_fn)(StringInfo out, struct PGLogicalOutputDat
 
 typedef void (*pglogical_write_begin_fn)(StringInfo out, struct PGLogicalOutputData *data,
 							 ReorderBufferTXN *txn);
-typedef void (*pglogical_write_message_fn)(StringInfo out, struct PGLogicalOutputData *data,
+typedef void (*pglogical_write_message_fn)(StringInfo out, LogicalDecodingContext *ctx,
 					const char *prefix, Size sz, const char *message);
 typedef void (*pglogical_write_commit_fn)(StringInfo out, struct PGLogicalOutputData *data,
 							 ReorderBufferTXN *txn, XLogRecPtr commit_lsn);

pglogical_receiver.c

@@ -552,9 +552,9 @@ pglogical_receiver_main(Datum main_arg)
 						MtmSpillToFile(spill_file, buf.data, buf.used);
 						ByteBufferReset(&buf);
 					}
-					if (stmt[0] == 'Z' || (stmt[0] == 'M' && (stmt[1] == 'L' || stmt[1] == 'A' || stmt[1] == 'B' || stmt[1] == 'C'))) {
+					if (stmt[0] == 'Z' || (stmt[0] == 'M' && (stmt[1] == 'L' || stmt[1] == 'A' || stmt[1] == 'C'))) {
 						MTM_LOG3("Process '%c' message from %d", stmt[1], nodeId);
-						if (stmt[0] == 'M' && (stmt[1] == 'B' || stmt[1] == 'C')) { /* concurrent DDL should be executed by parallel workers */
+						if (stmt[0] == 'M' && stmt[1] == 'C') { /* concurrent DDL should be executed by parallel workers */
 							MtmExecute(stmt, msg_len);
 						} else {
 							MtmExecutor(stmt, msg_len); /* all other messages can be processed by receiver itself */