v2-0001-Support-for-extracting-gzip-compressed-archive.patch

From 826a1cbb639afb7e10a20955d3ec64b1bab1fa80 Mon Sep 17 00:00:00 2001
From: Dipesh Pandit <dipesh.pandit@enterprisedb.com>
Date: Thu, 20 Jan 2022 16:38:36 +0530
Subject: [PATCH 1/2] Support for extracting gzip compressed archive

pg_basebackup can support server side compression using gzip. In
order to support plain format backup with option '-Fp' we need to
add support for decompressing the compressed blocks at client. This
patch addresses the extraction of gzip compressed blocks at client.
---
 src/bin/pg_basebackup/Makefile          |   1 +
 src/bin/pg_basebackup/bbstreamer.h      |   1 +
 src/bin/pg_basebackup/bbstreamer_file.c | 182 ---------------
 src/bin/pg_basebackup/bbstreamer_gzip.c | 377 ++++++++++++++++++++++++++++++++
 src/bin/pg_basebackup/pg_basebackup.c   |  43 +++-
 5 files changed, 419 insertions(+), 185 deletions(-)
 create mode 100644 src/bin/pg_basebackup/bbstreamer_gzip.c

diff --git a/src/bin/pg_basebackup/Makefile b/src/bin/pg_basebackup/Makefile
index 5b18851..78d96c6 100644
--- a/src/bin/pg_basebackup/Makefile
+++ b/src/bin/pg_basebackup/Makefile
@@ -38,6 +38,7 @@ OBJS = \
 BBOBJS = \
 	pg_basebackup.o \
 	bbstreamer_file.o \
+	bbstreamer_gzip.o \
 	bbstreamer_inject.o \
 	bbstreamer_tar.o
 
diff --git a/src/bin/pg_basebackup/bbstreamer.h b/src/bin/pg_basebackup/bbstreamer.h
index fc88b50..270b0df 100644
--- a/src/bin/pg_basebackup/bbstreamer.h
+++ b/src/bin/pg_basebackup/bbstreamer.h
@@ -205,6 +205,7 @@ extern bbstreamer *bbstreamer_extractor_new(const char *basepath,
 											const char *(*link_map) (const char *),
 											void (*report_output_file) (const char *));
 
+extern bbstreamer *bbstreamer_gzip_extractor_new(bbstreamer *next);
 extern bbstreamer *bbstreamer_tar_parser_new(bbstreamer *next);
 extern bbstreamer *bbstreamer_tar_terminator_new(bbstreamer *next);
 extern bbstreamer *bbstreamer_tar_archiver_new(bbstreamer *next);
diff --git a/src/bin/pg_basebackup/bbstreamer_file.c b/src/bin/pg_basebackup/bbstreamer_file.c
index 77ca222..d721f87 100644
--- a/src/bin/pg_basebackup/bbstreamer_file.c
+++ b/src/bin/pg_basebackup/bbstreamer_file.c
@@ -11,10 +11,6 @@
 
 #include "postgres_fe.h"
 
-#ifdef HAVE_LIBZ
-#include <zlib.h>
-#endif
-
 #include <unistd.h>
 
 #include "bbstreamer.h"
@@ -30,15 +26,6 @@ typedef struct bbstreamer_plain_writer
 	bool		should_close_file;
 } bbstreamer_plain_writer;
 
-#ifdef HAVE_LIBZ
-typedef struct bbstreamer_gzip_writer
-{
-	bbstreamer	base;
-	char	   *pathname;
-	gzFile		gzfile;
-}			bbstreamer_gzip_writer;
-#endif
-
 typedef struct bbstreamer_extractor
 {
 	bbstreamer	base;
@@ -62,22 +49,6 @@ const bbstreamer_ops bbstreamer_plain_writer_ops = {
 	.free = bbstreamer_plain_writer_free
 };
 
-#ifdef HAVE_LIBZ
-static void bbstreamer_gzip_writer_content(bbstreamer *streamer,
-										   bbstreamer_member *member,
-										   const char *data, int len,
-										   bbstreamer_archive_context context);
-static void bbstreamer_gzip_writer_finalize(bbstreamer *streamer);
-static void bbstreamer_gzip_writer_free(bbstreamer *streamer);
-static const char *get_gz_error(gzFile gzf);
-
-const bbstreamer_ops bbstreamer_gzip_writer_ops = {
-	.content = bbstreamer_gzip_writer_content,
-	.finalize = bbstreamer_gzip_writer_finalize,
-	.free = bbstreamer_gzip_writer_free
-};
-#endif
-
 static void bbstreamer_extractor_content(bbstreamer *streamer,
 										 bbstreamer_member *member,
 										 const char *data, int len,
@@ -196,159 +167,6 @@ bbstreamer_plain_writer_free(bbstreamer *streamer)
 }
 
 /*
- * Create a bbstreamer that just compresses data using gzip, and then writes
- * it to a file.
- *
- * As in the case of bbstreamer_plain_writer_new, pathname is always used
- * for error reporting purposes; if file is NULL, it is also the opened and
- * closed so that the data may be written there.
- */
-bbstreamer *
-bbstreamer_gzip_writer_new(char *pathname, FILE *file, int compresslevel)
-{
-#ifdef HAVE_LIBZ
-	bbstreamer_gzip_writer *streamer;
-
-	streamer = palloc0(sizeof(bbstreamer_gzip_writer));
-	*((const bbstreamer_ops **) &streamer->base.bbs_ops) =
-		&bbstreamer_gzip_writer_ops;
-
-	streamer->pathname = pstrdup(pathname);
-
-	if (file == NULL)
-	{
-		streamer->gzfile = gzopen(pathname, "wb");
-		if (streamer->gzfile == NULL)
-		{
-			pg_log_error("could not create compressed file \"%s\": %m",
-						 pathname);
-			exit(1);
-		}
-	}
-	else
-	{
-		int			fd = dup(fileno(file));
-
-		if (fd < 0)
-		{
-			pg_log_error("could not duplicate stdout: %m");
-			exit(1);
-		}
-
-		streamer->gzfile = gzdopen(fd, "wb");
-		if (streamer->gzfile == NULL)
-		{
-			pg_log_error("could not open output file: %m");
-			exit(1);
-		}
-	}
-
-	if (gzsetparams(streamer->gzfile, compresslevel,
-					Z_DEFAULT_STRATEGY) != Z_OK)
-	{
-		pg_log_error("could not set compression level %d: %s",
-					 compresslevel, get_gz_error(streamer->gzfile));
-		exit(1);
-	}
-
-	return &streamer->base;
-#else
-	pg_log_error("this build does not support compression");
-	exit(1);
-#endif
-}
-
-#ifdef HAVE_LIBZ
-/*
- * Write archive content to gzip file.
- */
-static void
-bbstreamer_gzip_writer_content(bbstreamer *streamer,
-							   bbstreamer_member *member, const char *data,
-							   int len, bbstreamer_archive_context context)
-{
-	bbstreamer_gzip_writer *mystreamer;
-
-	mystreamer = (bbstreamer_gzip_writer *) streamer;
-
-	if (len == 0)
-		return;
-
-	errno = 0;
-	if (gzwrite(mystreamer->gzfile, data, len) != len)
-	{
-		/* if write didn't set errno, assume problem is no disk space */
-		if (errno == 0)
-			errno = ENOSPC;
-		pg_log_error("could not write to compressed file \"%s\": %s",
-					 mystreamer->pathname, get_gz_error(mystreamer->gzfile));
-		exit(1);
-	}
-}
-
-/*
- * End-of-archive processing when writing to a gzip file consists of just
- * calling gzclose.
- *
- * It makes no difference whether we opened the file or the caller did it,
- * because libz provides no way of avoiding a close on the underling file
- * handle. Notice, however, that bbstreamer_gzip_writer_new() uses dup() to
- * work around this issue, so that the behavior from the caller's viewpoint
- * is the same as for bbstreamer_plain_writer.
- */
-static void
-bbstreamer_gzip_writer_finalize(bbstreamer *streamer)
-{
-	bbstreamer_gzip_writer *mystreamer;
-
-	mystreamer = (bbstreamer_gzip_writer *) streamer;
-
-	errno = 0;					/* in case gzclose() doesn't set it */
-	if (gzclose(mystreamer->gzfile) != 0)
-	{
-		pg_log_error("could not close compressed file \"%s\": %m",
-					 mystreamer->pathname);
-		exit(1);
-	}
-
-	mystreamer->gzfile = NULL;
-}
-
-/*
- * Free memory associated with this bbstreamer.
- */
-static void
-bbstreamer_gzip_writer_free(bbstreamer *streamer)
-{
-	bbstreamer_gzip_writer *mystreamer;
-
-	mystreamer = (bbstreamer_gzip_writer *) streamer;
-
-	Assert(mystreamer->base.bbs_next == NULL);
-	Assert(mystreamer->gzfile == NULL);
-
-	pfree(mystreamer->pathname);
-	pfree(mystreamer);
-}
-
-/*
- * Helper function for libz error reporting.
- */
-static const char *
-get_gz_error(gzFile gzf)
-{
-	int			errnum;
-	const char *errmsg;
-
-	errmsg = gzerror(gzf, &errnum);
-	if (errnum == Z_ERRNO)
-		return strerror(errno);
-	else
-		return errmsg;
-}
-#endif
-
-/*
  * Create a bbstreamer that extracts an archive.
  *
  * All pathnames in the archive are interpreted relative to basepath.
diff --git a/src/bin/pg_basebackup/bbstreamer_gzip.c b/src/bin/pg_basebackup/bbstreamer_gzip.c
new file mode 100644
index 0000000..c144a73
--- /dev/null
+++ b/src/bin/pg_basebackup/bbstreamer_gzip.c
@@ -0,0 +1,377 @@
+/*-------------------------------------------------------------------------
+ *
+ * bbstreamer_gzip.c
+ *
+ * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		  src/bin/pg_basebackup/bbstreamer_gzip.c
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres_fe.h"
+
+#ifdef HAVE_LIBZ
+#include <zlib.h>
+#endif
+
+#include "bbstreamer.h"
+#include "common/logging.h"
+#include "common/file_perm.h"
+#include "common/string.h"
+
+#ifdef HAVE_LIBZ
+typedef struct bbstreamer_gzip_writer
+{
+	bbstreamer	base;
+	char	   *pathname;
+	gzFile		gzfile;
+}			bbstreamer_gzip_writer;
+
+typedef struct bbstreamer_gzip_extractor
+{
+	bbstreamer	base;
+	z_stream	zstream;
+	size_t		bytes_written;
+} bbstreamer_gzip_extractor;
+
+static void bbstreamer_gzip_writer_content(bbstreamer *streamer,
+										   bbstreamer_member *member,
+										   const char *data, int len,
+										   bbstreamer_archive_context context);
+static void bbstreamer_gzip_writer_finalize(bbstreamer *streamer);
+static void bbstreamer_gzip_writer_free(bbstreamer *streamer);
+static const char *get_gz_error(gzFile gzf);
+
+const bbstreamer_ops bbstreamer_gzip_writer_ops = {
+	.content = bbstreamer_gzip_writer_content,
+	.finalize = bbstreamer_gzip_writer_finalize,
+	.free = bbstreamer_gzip_writer_free
+};
+
+static void bbstreamer_gzip_extractor_content(bbstreamer *streamer,
+											  bbstreamer_member *member,
+											  const char *data, int len,
+											  bbstreamer_archive_context context);
+static void bbstreamer_gzip_extractor_finalize(bbstreamer *streamer);
+static void bbstreamer_gzip_extractor_free(bbstreamer *streamer);
+static void *gzip_palloc(void *opaque, unsigned items, unsigned size);
+static void gzip_pfree(void *opaque, void *address);
+
+const bbstreamer_ops bbstreamer_gzip_extractor_ops = {
+	.content = bbstreamer_gzip_extractor_content,
+	.finalize = bbstreamer_gzip_extractor_finalize,
+	.free = bbstreamer_gzip_extractor_free
+};
+#endif
+
+/*
+ * Create a bbstreamer that just compresses data using gzip, and then writes
+ * it to a file.
+ *
+ * As in the case of bbstreamer_plain_writer_new, pathname is always used
+ * for error reporting purposes; if file is NULL, it is also the opened and
+ * closed so that the data may be written there.
+ */
+bbstreamer *
+bbstreamer_gzip_writer_new(char *pathname, FILE *file, int compresslevel)
+{
+#ifdef HAVE_LIBZ
+	bbstreamer_gzip_writer *streamer;
+
+	streamer = palloc0(sizeof(bbstreamer_gzip_writer));
+	*((const bbstreamer_ops **) &streamer->base.bbs_ops) =
+		&bbstreamer_gzip_writer_ops;
+
+	streamer->pathname = pstrdup(pathname);
+
+	if (file == NULL)
+	{
+		streamer->gzfile = gzopen(pathname, "wb");
+		if (streamer->gzfile == NULL)
+		{
+			pg_log_error("could not create compressed file \"%s\": %m",
+						 pathname);
+			exit(1);
+		}
+	}
+	else
+	{
+		int			fd = dup(fileno(file));
+
+		if (fd < 0)
+		{
+			pg_log_error("could not duplicate stdout: %m");
+			exit(1);
+		}
+
+		streamer->gzfile = gzdopen(fd, "wb");
+		if (streamer->gzfile == NULL)
+		{
+			pg_log_error("could not open output file: %m");
+			exit(1);
+		}
+	}
+
+	if (gzsetparams(streamer->gzfile, compresslevel,
+					Z_DEFAULT_STRATEGY) != Z_OK)
+	{
+		pg_log_error("could not set compression level %d: %s",
+					 compresslevel, get_gz_error(streamer->gzfile));
+		exit(1);
+	}
+
+	return &streamer->base;
+#else
+	pg_log_error("this build does not support compression");
+	exit(1);
+#endif
+}
+
+#ifdef HAVE_LIBZ
+/*
+ * Write archive content to gzip file.
+ */
+static void
+bbstreamer_gzip_writer_content(bbstreamer *streamer,
+							   bbstreamer_member *member, const char *data,
+							   int len, bbstreamer_archive_context context)
+{
+	bbstreamer_gzip_writer *mystreamer;
+
+	mystreamer = (bbstreamer_gzip_writer *) streamer;
+
+	if (len == 0)
+		return;
+
+	errno = 0;
+	if (gzwrite(mystreamer->gzfile, data, len) != len)
+	{
+		/* if write didn't set errno, assume problem is no disk space */
+		if (errno == 0)
+			errno = ENOSPC;
+		pg_log_error("could not write to compressed file \"%s\": %s",
+					 mystreamer->pathname, get_gz_error(mystreamer->gzfile));
+		exit(1);
+	}
+}
+
+/*
+ * End-of-archive processing when writing to a gzip file consists of just
+ * calling gzclose.
+ *
+ * It makes no difference whether we opened the file or the caller did it,
+ * because libz provides no way of avoiding a close on the underling file
+ * handle. Notice, however, that bbstreamer_gzip_writer_new() uses dup() to
+ * work around this issue, so that the behavior from the caller's viewpoint
+ * is the same as for bbstreamer_plain_writer.
+ */
+static void
+bbstreamer_gzip_writer_finalize(bbstreamer *streamer)
+{
+	bbstreamer_gzip_writer *mystreamer;
+
+	mystreamer = (bbstreamer_gzip_writer *) streamer;
+
+	errno = 0;					/* in case gzclose() doesn't set it */
+	if (gzclose(mystreamer->gzfile) != 0)
+	{
+		pg_log_error("could not close compressed file \"%s\": %m",
+					 mystreamer->pathname);
+		exit(1);
+	}
+
+	mystreamer->gzfile = NULL;
+}
+
+/*
+ * Free memory associated with this bbstreamer.
+ */
+static void
+bbstreamer_gzip_writer_free(bbstreamer *streamer)
+{
+	bbstreamer_gzip_writer *mystreamer;
+
+	mystreamer = (bbstreamer_gzip_writer *) streamer;
+
+	Assert(mystreamer->base.bbs_next == NULL);
+	Assert(mystreamer->gzfile == NULL);
+
+	pfree(mystreamer->pathname);
+	pfree(mystreamer);
+}
+
+/*
+ * Helper function for libz error reporting.
+ */
+static const char *
+get_gz_error(gzFile gzf)
+{
+	int			errnum;
+	const char *errmsg;
+
+	errmsg = gzerror(gzf, &errnum);
+	if (errnum == Z_ERRNO)
+		return strerror(errno);
+	else
+		return errmsg;
+}
+#endif
+
+/*
+ * Create a new base backup streamer that performs decompression of gzip
+ * compressed blocks.
+ */
+bbstreamer *
+bbstreamer_gzip_extractor_new(bbstreamer *next)
+{
+#ifdef HAVE_LIBZ
+	bbstreamer_gzip_extractor	*streamer;
+	z_stream *zs;
+
+	Assert(next != NULL);
+
+	streamer = palloc0(sizeof(bbstreamer_gzip_extractor));
+	*((const bbstreamer_ops **) &streamer->base.bbs_ops) =
+		&bbstreamer_gzip_extractor_ops;
+
+	streamer->base.bbs_next = next;
+	initStringInfo(&streamer->base.bbs_buffer);
+
+	/* Initialize internal stream state for decompression */
+	zs = &streamer->zstream;
+	zs->zalloc = gzip_palloc;
+	zs->zfree = gzip_pfree;
+	zs->next_out = (uint8 *) streamer->base.bbs_buffer.data;
+	zs->avail_out = streamer->base.bbs_buffer.maxlen;
+
+	/*
+	 * Data compression was initialized using deflateInit2 to request a gzip
+	 * header. Similarly, we are using inflateInit2 to initialize data
+	 * decompression.
+	 *
+	 * Per the documentation of inflateInit2, the second argument is
+	 * "windowBits" and it's value must be greater than or equal to the value
+	 * provided while compressing the data, so we are using the maximum
+	 * possible value for safety.
+	 */
+	if (inflateInit2(zs, 15 + 16) != Z_OK)
+	{
+		pg_log_error("could not initialize compression library");
+		exit(1);
+
+	}
+
+	return &streamer->base;
+#else
+	pg_log_error("this build does not support compression");
+	exit(1);
+#endif
+}
+
+#ifdef HAVE_LIBZ
+/*
+ * Decompress the input data to output buffer until we ran out of the input
+ * data. Each time the output buffer is full invoke bbstreamer_content to pass
+ * on the decompressed data to next streamer.
+ */
+static void
+bbstreamer_gzip_extractor_content(bbstreamer *streamer,
+								  bbstreamer_member *member,
+								  const char *data, int len,
+								  bbstreamer_archive_context context)
+{
+	bbstreamer_gzip_extractor *mystreamer = (bbstreamer_gzip_extractor *) streamer;
+	z_stream *zs = &mystreamer->zstream;
+
+
+	zs->next_in = (uint8 *) data;
+	zs->avail_in = len;
+
+	/* Process the current chunk */
+	while (zs->avail_in > 0)
+	{
+		int res;
+
+		Assert(mystreamer->bytes_written < mystreamer->base.bbs_buffer.maxlen);
+
+		zs->next_out = (uint8 *)
+			mystreamer->base.bbs_buffer.data + mystreamer->bytes_written;
+		zs->avail_out = mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written;
+
+		/*
+		 * Decompresses data starting at zs->next_in and update zs->next_in
+		 * and zs->avail_in, generate output data starting at zs->next_out
+		 * and update zs->next_out and zs->avail_out accordingly.
+		 */
+		res = inflate(zs, Z_NO_FLUSH);
+
+		if (res == Z_STREAM_ERROR)
+			pg_log_error("could not decompress data: %s", zs->msg);
+
+		mystreamer->bytes_written = mystreamer->base.bbs_buffer.maxlen - zs->avail_out;
+
+		/* If output buffer is full then pass on the content to next streamer */
+		if (mystreamer->bytes_written >= mystreamer->base.bbs_buffer.maxlen)
+		{
+			bbstreamer_content(mystreamer->base.bbs_next, member,
+							   mystreamer->base.bbs_buffer.data,
+							   mystreamer->base.bbs_buffer.maxlen, context);
+			mystreamer->bytes_written = 0;
+		}
+	}
+}
+
+/*
+ * End-of-stream processing.
+ */
+static void
+bbstreamer_gzip_extractor_finalize(bbstreamer *streamer)
+{
+	bbstreamer_gzip_extractor *mystreamer = (bbstreamer_gzip_extractor *) streamer;
+
+	/*
+	 * End of the stream, if there is some pending data in output buffers then
+	 * we must forward it to next streamer.
+	 */
+	bbstreamer_content(mystreamer->base.bbs_next, NULL,
+					   mystreamer->base.bbs_buffer.data,
+					   mystreamer->base.bbs_buffer.maxlen,
+					   BBSTREAMER_UNKNOWN);
+
+	bbstreamer_finalize(mystreamer->base.bbs_next);
+}
+
+/*
+ * Free memory.
+ */
+static void
+bbstreamer_gzip_extractor_free(bbstreamer *streamer)
+{
+	bbstreamer_gzip_extractor *mystreamer = (bbstreamer_gzip_extractor *) streamer;
+
+	bbstreamer_free(mystreamer->base.bbs_next);
+	pfree(mystreamer->base.bbs_buffer.data);
+	pfree(streamer);
+}
+
+/*
+ * Wrapper function to adjust the signature of palloc to match what libz
+ * expects.
+ */
+static void *
+gzip_palloc(void *opaque, unsigned items, unsigned size)
+{
+	return palloc(items * size);
+}
+
+/*
+ * Wrapper function to adjust the signature of pfree to match what libz
+ * expects.
+ */
+static void
+gzip_pfree(void *opaque, void *address)
+{
+	pfree(address);
+}
+#endif
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index 6ee49a5..d43eb4b 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -111,6 +111,12 @@ typedef enum
 	STREAM_WAL
 } IncludeWal;
 
+typedef enum
+{
+	BACKUP_COMPRESSION_NONE,
+	BACKUP_COMPRESSION_GZIP
+} compression_type;
+
 /* Global options */
 static char *basedir = NULL;
 static TablespaceList tablespace_dirs = {NULL, NULL};
@@ -173,6 +179,10 @@ static int	has_xlogendptr = 0;
 static volatile LONG has_xlogendptr = 0;
 #endif
 
+/* Server side compression method and compression level */
+static compression_type	server_compression_type = BACKUP_COMPRESSION_NONE;
+static int				server_compression_level = 0;
+
 /* Contents of configuration file to be generated */
 static PQExpBuffer recoveryconfcontents = NULL;
 
@@ -1002,7 +1012,8 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
 	bbstreamer *streamer;
 	bbstreamer *manifest_inject_streamer = NULL;
 	bool		inject_manifest;
-	bool		is_tar;
+	bool		is_tar,
+				is_tar_gz;
 	bool		must_parse_archive;
 	int			archive_name_len = strlen(archive_name);
 
@@ -1017,6 +1028,10 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
 	is_tar = (archive_name_len > 4 &&
 			  strcmp(archive_name + archive_name_len - 4, ".tar") == 0);
 
+	/* Is this a gzip archive? */
+	is_tar_gz = (archive_name_len > 8 &&
+				 strcmp(archive_name + archive_name_len - 3, ".gz") == 0);
+
 	/*
 	 * We have to parse the archive if (1) we're suppose to extract it, or if
 	 * (2) we need to inject backup_manifest or recovery configuration into it.
@@ -1025,8 +1040,8 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
 	must_parse_archive = (format == 'p' || inject_manifest ||
 		(spclocation == NULL && writerecoveryconf));
 
-	/* At present, we only know how to parse tar archives. */
-	if (must_parse_archive && !is_tar)
+	/* At present, we only know how to parse tar and gzip archives. */
+	if (must_parse_archive && !is_tar && !is_tar_gz)
 	{
 		pg_log_error("unable to parse archive: %s", archive_name);
 		pg_log_info("only tar archives can be parsed");
@@ -1136,6 +1151,13 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
 	else if (expect_unterminated_tarfile)
 		streamer = bbstreamer_tar_terminator_new(streamer);
 
+	/*
+	 * Extract the gzip compressed archive using a gzip extractor and then
+	 * forward it to next streamer.
+	 */
+	if (format == 'p' && server_compression_type == BACKUP_COMPRESSION_GZIP)
+		streamer = bbstreamer_gzip_extractor_new(streamer);
+
 	/* Return the results. */
 	*manifest_inject_streamer_p = manifest_inject_streamer;
 	return streamer;
@@ -2448,6 +2470,21 @@ main(int argc, char **argv)
 		exit(1);
 	}
 
+	if (server_compression != NULL)
+	{
+		if (strcmp(server_compression, "gzip") == 0)
+			server_compression_type = BACKUP_COMPRESSION_GZIP;
+		else if (strlen(server_compression) == 5 &&
+				strncmp(server_compression, "gzip", 4) == 0 &&
+				server_compression[4] >= '1' && server_compression[4] <= '9')
+		{
+			server_compression_type = BACKUP_COMPRESSION_GZIP;
+			server_compression_level = server_compression[4] - '0';
+		}
+	}
+	else
+		server_compression_type = BACKUP_COMPRESSION_NONE;
+
 	/*
 	 * Compression doesn't make sense unless tar format is in use.
 	 */
-- 
1.8.3.1