Add walsender messages compression

Author: knizhnik

src/backend/replication/walsender.c

@@ -60,12 +60,15 @@
 #include "catalog/pg_type.h"
 #include "commands/dbcommands.h"
 #include "commands/defrem.h"
+#include "common/pg_lzcompress.h"
+#include "common/fe_memutils.h"
 #include "funcapi.h"
 #include "libpq/libpq.h"
 #include "libpq/pqformat.h"
 #include "miscadmin.h"
 #include "nodes/replnodes.h"
 #include "pgstat.h"
+#include "port/pg_bswap.h"
 #include "replication/basebackup.h"
 #include "replication/decode.h"
 #include "replication/logical.h"
@@ -95,7 +98,9 @@
 #include "utils/timestamp.h"
 
 #define ERROR_MESSAGE_SIZE 1024
-
+#define MIN_COMPRESSION_RATIO 2
+#define MIN_COMPRESSION_SIZE  32
+#define MSG_COMPRESS_HDR_SIZE 4
 /*
  * Shared structure to repair a corrupted block.
  *
@@ -160,6 +165,9 @@ int			wal_sender_timeout = 60 * 1000; /* maximum time to send one WAL
 											 * data message */
 bool		log_replication_commands = false;
 
+
+bool         wal_sender_compression = false;
+
 /*
  * State for WalSndWakeupRequest
  */
@@ -1192,7 +1200,29 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
 	TimestampTz	now;
 
 	/* output previously gathered data in a CopyData packet */
-	pq_putmessage_noblock('d', ctx->out->data, ctx->out->len);
+	if (wal_sender_compression && ctx->out->len > MIN_COMPRESSION_SIZE)
+	{
+		static char* lzbuf;
+		static int lzbuf_size;
+		int zip_len;
+		int raw_len = ctx->out->len;
+		int max_zip_len = PGLZ_MAX_OUTPUT(raw_len) + MSG_COMPRESS_HDR_SIZE;
+		if (max_zip_len > lzbuf_size) {
+			lzbuf_size = lzbuf_size*2 > max_zip_len ? lzbuf_size*2 : max_zip_len;
+			lzbuf = realloc(lzbuf, lzbuf_size);
+		}
+		zip_len = pglz_compress(ctx->out->data, raw_len, lzbuf + MSG_COMPRESS_HDR_SIZE, PGLZ_strategy_default);
+		if (zip_len > 0 && zip_len < raw_len/MIN_COMPRESSION_RATIO)
+		{
+			uint32 net_size = htonl(raw_len);
+			memcpy(lzbuf, &net_size, sizeof(net_size));
+			pq_putmessage_noblock('z', lzbuf, zip_len + MSG_COMPRESS_HDR_SIZE);
+			elog(LOG, "WRITE compressed message with length %d raw_len=%d", zip_len, raw_len);
+		}
+		else
+			pq_putmessage_noblock('d', ctx->out->data, raw_len);
+	} else
+		pq_putmessage_noblock('d', ctx->out->data, ctx->out->len);
 
 	/*
 	 * Fill the send timestamp last, so that it is taken as late as possible.

src/backend/utils/misc/guc.c

@@ -1076,6 +1076,16 @@ static struct config_bool ConfigureNamesBool[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"wal_sender_compression", PGC_SUSET, WAL_SETTINGS,
+			gettext_noop("Compress logical replicastion messages."),
+			NULL
+		},
+		&wal_sender_compression,
+		false,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"log_checkpoints", PGC_SIGHUP, LOGGING_WHAT,
 			gettext_noop("Logs each checkpoint."),

src/include/replication/walsender.h

@@ -41,6 +41,7 @@ extern bool wake_wal_senders;
 extern PGDLLIMPORT int	max_wal_senders;
 extern int	wal_sender_timeout;
 extern bool log_replication_commands;
+extern bool wal_sender_compression;
 
 extern void InitWalSender(void);
 extern bool exec_replication_command(const char *query_string);

src/interfaces/libpq/fe-protocol3.c

@@ -33,13 +33,15 @@
 #include <arpa/inet.h>
 #endif
 
+#include "common/pg_lzcompress.h"
+#include "utils/elog.h"
 
 /*
  * This macro lists the backend message types that could be "long" (more
  * than a couple of kilobytes).
  */
 #define VALID_LONG_MESSAGE_TYPE(id) \
-	((id) == 'T' || (id) == 'D' || (id) == 'd' || (id) == 'V' || \
+	((id) == 'T' || (id) == 'D' || (id) == 'd' || (id) == 'z' || (id) == 'V' || \
 	 (id) == 'E' || (id) == 'N' || (id) == 'A')
 
 
@@ -389,6 +391,7 @@ pqParseInput3(PGconn *conn)
 					conn->copy_already_done = 0;
 					break;
 				case 'd':		/* Copy Data */
+				case 'z':		/* Copy Compressed Data */
 
 					/*
 					 * If we see Copy Data, just silently drop it.  This would
@@ -1533,7 +1536,7 @@ getReadyForQuery(PGconn *conn)
  * message available, -1 if end of copy, -2 if error.
  */
 static int
-getCopyDataMessage(PGconn *conn)
+getCopyDataMessage(PGconn *conn, char* kind)
 {
 	char		id;
 	int			msgLength;
@@ -1584,6 +1587,7 @@ getCopyDataMessage(PGconn *conn)
 		 * completeness.)  Otherwise, if it's anything except Copy Data,
 		 * report end-of-copy.
 		 */
+		*kind = id;
 		switch (id)
 		{
 			case 'A':			/* NOTIFY */
@@ -1599,6 +1603,7 @@ getCopyDataMessage(PGconn *conn)
 					return 0;
 				break;
 			case 'd':			/* Copy Data, pass it back to caller */
+			case 'z':			/* Copy Compressed Data, pass it back to caller */
 				return msgLength;
 			case 'c':
 
@@ -1641,6 +1646,7 @@ int
 pqGetCopyData3(PGconn *conn, char **buffer, int async)
 {
 	int			msgLength;
+	char        id;
 
 	for (;;)
 	{
@@ -1649,7 +1655,7 @@ pqGetCopyData3(PGconn *conn, char **buffer, int async)
 		 * callers, we keep returning 0 until the next message is fully
 		 * available, even if it is not Copy Data.
 		 */
-		msgLength = getCopyDataMessage(conn);
+		msgLength = getCopyDataMessage(conn, &id);
 		if (msgLength < 0)
 			return msgLength;	/* end-of-copy or error */
 		if (msgLength == 0)
@@ -1671,20 +1677,55 @@ pqGetCopyData3(PGconn *conn, char **buffer, int async)
 		msgLength -= 4;
 		if (msgLength > 0)
 		{
-			*buffer = (char *) malloc(msgLength + 1);
-			if (*buffer == NULL)
-			{
-				printfPQExpBuffer(&conn->errorMessage,
-								  libpq_gettext("out of memory\n"));
-				return -2;
+			if (id == 'z') { /* compressed data */
+				int zip_len = msgLength -= 4;
+				int raw_len;
+				int len;
+
+				Assert(zip_len > 0);
+
+			    if (pqGetInt(&raw_len, 4, conn))
+					return -2;
+
+				*buffer = (char *) malloc(raw_len + 1);
+				if (*buffer == NULL)
+				{
+					printfPQExpBuffer(&conn->errorMessage,
+									  libpq_gettext("out of memory\n"));
+					return -2;
+				}
+				len = pglz_decompress(&conn->inBuffer[conn->inCursor], zip_len, *buffer, raw_len);
+				if (len != raw_len)
+				{
+					printfPQExpBuffer(&conn->errorMessage,
+									  libpq_gettext("decompress error\n"));
+					return -2;
+				}
+				(*buffer)[raw_len] = '\0';	/* Add terminating null */
+
+
+				/* Mark message consumed */
+				conn->inStart = conn->inCursor + msgLength;
+
+				return raw_len;
 			}
-			memcpy(*buffer, &conn->inBuffer[conn->inCursor], msgLength);
-			(*buffer)[msgLength] = '\0';	/* Add terminating null */
+			else
+			{
+				*buffer = (char *) malloc(msgLength + 1);
+				if (*buffer == NULL)
+				{
+					printfPQExpBuffer(&conn->errorMessage,
+									  libpq_gettext("out of memory\n"));
+					return -2;
+				}
+				memcpy(*buffer, &conn->inBuffer[conn->inCursor], msgLength);
+				(*buffer)[msgLength] = '\0';	/* Add terminating null */
 
-			/* Mark message consumed */
-			conn->inStart = conn->inCursor + msgLength;
+				/* Mark message consumed */
+				conn->inStart = conn->inCursor + msgLength;
 
-			return msgLength;
+				return msgLength;
+			}
 		}
 
 		/* Empty, so drop it and loop around for another */
@@ -1754,6 +1795,7 @@ pqGetlineAsync3(PGconn *conn, char *buffer, int bufsize)
 {
 	int			msgLength;
 	int			avail;
+	char        id;
 
 	if (conn->asyncStatus != PGASYNC_COPY_OUT
 		&& conn->asyncStatus != PGASYNC_COPY_BOTH)
@@ -1765,7 +1807,7 @@ pqGetlineAsync3(PGconn *conn, char *buffer, int bufsize)
 	 * even if it is not Copy Data.  This should keep PQendcopy from blocking.
 	 * (Note: unlike pqGetCopyData3, we do not change asyncStatus here.)
 	 */
-	msgLength = getCopyDataMessage(conn);
+	msgLength = getCopyDataMessage(conn, &id);
 	if (msgLength < 0)
 		return -1;				/* end-of-copy or error */
 	if (msgLength == 0)