PGPRO-692: Compress/Decompress WAL files during archive-push/archive-get

Author: za-arthur

src/archive.c

@@ -18,7 +18,7 @@
  * --wal-file-path %p --wal-file-name %f', to move backups into arclog_path.
  * Where archlog_path is $BACKUP_PATH/wal/system_id.
  * Currently it just copies wal files to the new location.
- * TODO: Planned options: compress, list the arclog content,
+ * TODO: Planned options: list the arclog content,
  * compute and validate checksums.
  */
 int
@@ -64,7 +64,8 @@ do_archive_push(char *wal_file_path, char *wal_file_name)
 	if (access(backup_wal_file_path, F_OK) != -1)
 		elog(ERROR, "file '%s', already exists.", backup_wal_file_path);
 
-	copy_wal_file(absolute_wal_file_path, backup_wal_file_path);
+	push_wal_file(absolute_wal_file_path, backup_wal_file_path,
+				  compress_shortcut);
 	elog(INFO, "pg_probackup archive-push completed successfully");
 
 	return 0;
@@ -96,8 +97,9 @@ do_archive_get(char *wal_file_path, char *wal_file_name)
 	join_path_components(absolute_wal_file_path, current_dir, wal_file_path);
 	join_path_components(backup_wal_file_path, arclog_path, wal_file_name);
 
-	elog(INFO, "pg_probackup archive-get from %s to %s", backup_wal_file_path, absolute_wal_file_path);
-	copy_wal_file(backup_wal_file_path, absolute_wal_file_path);
+	elog(INFO, "pg_probackup archive-get from %s to %s",
+		 backup_wal_file_path, absolute_wal_file_path);
+	get_wal_file(backup_wal_file_path, absolute_wal_file_path);
 	elog(INFO, "pg_probackup archive-get completed successfully");
 
 	return 0;

src/backup.c

@@ -1350,11 +1350,16 @@ wait_wal_lsn(XLogRecPtr lsn, bool wait_prev_segment)
 	TimeLineID	tli;
 	XLogSegNo	targetSegNo;
 	char		wal_dir[MAXPGPATH],
-				wal_segment_full_path[MAXPGPATH];
+				wal_segment_path[MAXPGPATH];
 	char		wal_segment[MAXFNAMELEN];
+	bool		file_exists = false;
 	uint32		try_count = 0,
 				timeout;
 
+#ifdef HAVE_LIBZ
+	char		gz_wal_segment_path[MAXPGPATH];
+#endif
+
 	tli = get_current_timeline(false);
 
 	/* Compute the name of the WAL file containig requested LSN */
@@ -1367,14 +1372,14 @@ wait_wal_lsn(XLogRecPtr lsn, bool wait_prev_segment)
 	{
 		pgBackupGetPath2(&current, wal_dir, lengthof(wal_dir),
 						 DATABASE_DIR, PG_XLOG_DIR);
-		join_path_components(wal_segment_full_path, wal_dir, wal_segment);
+		join_path_components(wal_segment_path, wal_dir, wal_segment);
 
 		timeout = (uint32) checkpoint_timeout();
 		timeout = timeout + timeout * 0.1;
 	}
 	else
 	{
-		join_path_components(wal_segment_full_path, arclog_path, wal_segment);
+		join_path_components(wal_segment_path, arclog_path, wal_segment);
 		timeout = archive_timeout;
 	}
 
@@ -1383,14 +1388,33 @@ wait_wal_lsn(XLogRecPtr lsn, bool wait_prev_segment)
 	else
 		elog(LOG, "Looking for LSN: %X/%X in segment: %s", (uint32) (lsn >> 32), (uint32) lsn, wal_segment);
 
+#ifdef HAVE_LIBZ
+	snprintf(gz_wal_segment_path, sizeof(gz_wal_segment_path), "%s.gz",
+			 wal_segment_path);
+#endif
+
 	/* Wait until target LSN is archived or streamed */
 	while (true)
 	{
-		bool		file_exists = fileExists(wal_segment_full_path);
+		if (!file_exists)
+		{
+			file_exists = fileExists(wal_segment_path);
+
+			/* Try to find compressed WAL file */
+			if (!file_exists)
+			{
+#ifdef HAVE_LIBZ
+				file_exists = fileExists(gz_wal_segment_path);
+				if (file_exists)
+					elog(LOG, "Found compressed WAL segment: %s", wal_segment_path);
+#endif
+			}
+			else
+				elog(LOG, "Found WAL segment: %s", wal_segment_path);
+		}
 
 		if (file_exists)
 		{
-			elog(LOG, "Found segment: %s", wal_segment);
 			/* Do not check LSN for previous WAL segment */
 			if (wait_prev_segment)
 				return;
@@ -1409,18 +1433,18 @@ wait_wal_lsn(XLogRecPtr lsn, bool wait_prev_segment)
 
 		sleep(1);
 		if (interrupted)
-			elog(ERROR, "interrupted during waiting for WAL archiving");
+			elog(ERROR, "Interrupted during waiting for WAL archiving");
 		try_count++;
 
 		/* Inform user if WAL segment is absent in first attempt */
 		if (try_count == 1)
 		{
 			if (wait_prev_segment)
-				elog(INFO, "wait for WAL segment %s to be archived",
-					 wal_segment_full_path);
+				elog(INFO, "Wait for WAL segment %s to be archived",
+					 wal_segment_path);
 			else
-				elog(INFO, "wait for LSN %X/%X in archived WAL segment %s",
-					 (uint32) (lsn >> 32), (uint32) lsn, wal_segment_full_path);
+				elog(INFO, "Wait for LSN %X/%X in archived WAL segment %s",
+					 (uint32) (lsn >> 32), (uint32) lsn, wal_segment_path);
 		}
 
 		if (timeout > 0 && try_count > timeout)
@@ -1432,7 +1456,7 @@ wait_wal_lsn(XLogRecPtr lsn, bool wait_prev_segment)
 			/* If WAL segment doesn't exist or we wait for previous segment */
 			else
 				elog(ERROR,
-					 "switched WAL segment %s could not be archived in %d seconds",
+					 "Switched WAL segment %s could not be archived in %d seconds",
 					 wal_segment, timeout);
 		}
 	}