Revert "Add walsender messages compression"

This reverts commit 765d063.

Author: kelvich

src/backend/replication/walsender.c

@@ -60,15 +60,12 @@
 #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"
@@ -98,9 +95,7 @@
 #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.
  *
@@ -165,9 +160,6 @@ 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
  */
@@ -1200,29 +1192,7 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
 	TimestampTz	now;
 
 	/* output previously gathered data in a CopyData packet */
-	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);
+	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,16 +1076,6 @@ 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,7 +41,6 @@ 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,15 +33,13 @@
 #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) == 'z' || (id) == 'V' || \
+	((id) == 'T' || (id) == 'D' || (id) == 'd' || (id) == 'V' || \
 	 (id) == 'E' || (id) == 'N' || (id) == 'A')
 
 
@@ -391,7 +389,6 @@ 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
@@ -1536,7 +1533,7 @@ getReadyForQuery(PGconn *conn)
  * message available, -1 if end of copy, -2 if error.
  */
 static int
-getCopyDataMessage(PGconn *conn, char* kind)
+getCopyDataMessage(PGconn *conn)
 {
 	char		id;
 	int			msgLength;
@@ -1587,7 +1584,6 @@ getCopyDataMessage(PGconn *conn, char* kind)
 		 * completeness.)  Otherwise, if it's anything except Copy Data,
 		 * report end-of-copy.
 		 */
-		*kind = id;
 		switch (id)
 		{
 			case 'A':			/* NOTIFY */
@@ -1603,7 +1599,6 @@ getCopyDataMessage(PGconn *conn, char* kind)
 					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':
 
@@ -1646,7 +1641,6 @@ int
 pqGetCopyData3(PGconn *conn, char **buffer, int async)
 {
 	int			msgLength;
-	char        id;
 
 	for (;;)
 	{
@@ -1655,7 +1649,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, &id);
+		msgLength = getCopyDataMessage(conn);
 		if (msgLength < 0)
 			return msgLength;	/* end-of-copy or error */
 		if (msgLength == 0)
@@ -1677,55 +1671,20 @@ pqGetCopyData3(PGconn *conn, char **buffer, int async)
 		msgLength -= 4;
 		if (msgLength > 0)
 		{
-			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;
-			}
-			else
+			*buffer = (char *) malloc(msgLength + 1);
+			if (*buffer == NULL)
 			{
-				*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 */
+				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 */
@@ -1795,7 +1754,6 @@ pqGetlineAsync3(PGconn *conn, char *buffer, int bufsize)
 {
 	int			msgLength;
 	int			avail;
-	char        id;
 
 	if (conn->asyncStatus != PGASYNC_COPY_OUT
 		&& conn->asyncStatus != PGASYNC_COPY_BOTH)
@@ -1807,7 +1765,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, &id);
+	msgLength = getCopyDataMessage(conn);
 	if (msgLength < 0)
 		return -1;				/* end-of-copy or error */
 	if (msgLength == 0)